Merge efcd701 into 0a19242

ydb-platform · Oct 29, 2024 · bdf0e09 · bdf0e09
2 parents 0a19242 + efcd701
commit bdf0e09
Show file tree

Hide file tree

Showing 30 changed files with 2,455 additions and 490 deletions.
diff --git a/ydb/core/formats/arrow/serializer/native.cpp b/ydb/core/formats/arrow/serializer/native.cpp
@@ -113,9 +113,8 @@ NKikimr::TConclusion<std::shared_ptr<arrow::util::Codec>> TNativeSerializer::Bui
     const int levelMin = codec->minimum_compression_level();
     const int levelMax = codec->maximum_compression_level();
     if (levelDef < levelMin || levelMax < levelDef) {
-        return TConclusionStatus::Fail(
-            TStringBuilder() << "incorrect level for codec. have to be: [" << levelMin << ":" << levelMax << "]"
-        );
+        return TConclusionStatus::Fail(TStringBuilder() << "incorrect level for codec `" << arrow::util::Codec::GetCodecAsString(cType)
+                                                        << "`. have to be: [" << levelMin << ":" << levelMax << "]");
     }
     std::shared_ptr<arrow::util::Codec> codecPtr = std::move(NArrow::TStatusValidator::GetValid(arrow::util::Codec::Create(cType, levelDef)));
     return codecPtr;
@@ -182,7 +181,9 @@ NKikimr::TConclusionStatus TNativeSerializer::DoDeserializeFromProto(const NKiki
 void TNativeSerializer::DoSerializeToProto(NKikimrSchemeOp::TOlapColumn::TSerializer& proto) const {
     if (Options.codec) {
         proto.MutableArrowCompression()->SetCodec(NArrow::CompressionToProto(Options.codec->compression_type()));
-        proto.MutableArrowCompression()->SetLevel(Options.codec->compression_level());
+        if (arrow::util::Codec::SupportsCompressionLevel(Options.codec->compression_type())) {
+            proto.MutableArrowCompression()->SetLevel(Options.codec->compression_level());
+        }
     } else {
         proto.MutableArrowCompression()->SetCodec(NArrow::CompressionToProto(arrow::Compression::UNCOMPRESSED));
     }

diff --git a/ydb/core/formats/arrow/serializer/native.h b/ydb/core/formats/arrow/serializer/native.h
@@ -103,6 +103,20 @@ class TNativeSerializer: public ISerializer {
         Options.use_threads = false;
         Options.memory_pool = pool;
     }
+
+    arrow::Compression::type GetCodecType() const {
+        if (Options.codec) {
+            return Options.codec->compression_type();
+        }
+        return arrow::Compression::type::UNCOMPRESSED;
+    }
+
+    std::optional<i32> GetCodecLevel() const {
+        if (Options.codec && arrow::util::Codec::SupportsCompressionLevel(Options.codec->compression_type())) {
+            return Options.codec->compression_level();
+        }
+        return {};
+    }
 };
 
 }
diff --git a/ydb/core/formats/arrow/serializer/parsing.cpp b/ydb/core/formats/arrow/serializer/parsing.cpp
@@ -7,6 +7,18 @@ std::string CompressionToString(const arrow::Compression::type compression) {
     return arrow::util::Codec::GetCodecAsString(compression);
 }
 
+std::string CompressionToString(const NKikimrSchemeOp::EColumnCodec compression) {
+    switch (compression) {
+        case NKikimrSchemeOp::EColumnCodec::ColumnCodecPlain:
+            return "off";
+        case NKikimrSchemeOp::EColumnCodec::ColumnCodecZSTD:
+            return "zstd";
+        case NKikimrSchemeOp::EColumnCodec::ColumnCodecLZ4:
+            return "lz4";
+    }
+    return "";
+}
+
 std::optional<arrow::Compression::type> CompressionFromString(const std::string& compressionStr) {
     auto result = arrow::util::Codec::GetCompressionType(compressionStr);
     if (!result.ok()) {

diff --git a/ydb/core/formats/arrow/serializer/parsing.h b/ydb/core/formats/arrow/serializer/parsing.h
@@ -9,9 +9,9 @@
 namespace NKikimr::NArrow {
 
 std::string CompressionToString(const arrow::Compression::type compression);
+std::string CompressionToString(const NKikimrSchemeOp::EColumnCodec compression);
 std::optional<arrow::Compression::type> CompressionFromString(const std::string& compressionStr);
 
 NKikimrSchemeOp::EColumnCodec CompressionToProto(const arrow::Compression::type compression);
 std::optional<arrow::Compression::type> CompressionFromProto(const NKikimrSchemeOp::EColumnCodec compression);
-
 }