[feature](compression) Support per-column compression for non-cloud

be/src/storage/segment/column_writer.cpp

@@ -486,7 +486,10 @@ ScalarColumnWriter::~ScalarColumnWriter() {
 }
 
 Status ScalarColumnWriter::init() {
-    RETURN_IF_ERROR(get_block_compression_codec(_opts.meta->compression(), &_compress_codec));
+    RETURN_IF_ERROR(get_block_compression_codec(
+            _opts.meta->compression(),
+            _opts.meta->has_compression_level() ? _opts.meta->compression_level() : 0,
+            &_compress_codec));
 
     PageBuilder* page_builder = nullptr;
 

be/src/storage/segment/segment_writer.cpp

@@ -117,7 +117,14 @@ void SegmentWriter::init_column_meta(ColumnMetaPB* meta, uint32_t column_id,
     meta->set_type(int(column.type()));
     meta->set_length(column.length());
     meta->set_encoding(EncodingInfo::resolve_default_encoding(opts.storage_format, column));
-    meta->set_compression(_opts.compression_type);
+    if (column.has_compression()) {
+        meta->set_compression(column.compression());
+        if (column.compression_level() > 0) {
+            meta->set_compression_level(column.compression_level());
+        }
+    } else {
+        meta->set_compression(_opts.compression_type);
+    }
     meta->set_is_nullable(column.is_nullable());
     meta->set_default_value(column.default_value());
     meta->set_precision(column.precision());

be/src/storage/segment/vertical_segment_writer.cpp

@@ -134,7 +134,14 @@ void VerticalSegmentWriter::_init_column_meta(ColumnMetaPB* meta, uint32_t colum
     meta->set_type(int(column.type()));
     meta->set_length(cast_set<int32_t>(column.length()));
     meta->set_encoding(EncodingInfo::resolve_default_encoding(opts.storage_format, column));
-    meta->set_compression(_opts.compression_type);
+    if (column.has_compression()) {
+        meta->set_compression(column.compression());
+        if (column.compression_level() > 0) {
+            meta->set_compression_level(column.compression_level());
+        }
+    } else {
+        meta->set_compression(_opts.compression_type);
+    }
     meta->set_is_nullable(column.is_nullable());
     meta->set_default_value(column.default_value());
     meta->set_precision(column.precision());

be/src/storage/segment/vertical_segment_writer.h

@@ -199,6 +199,7 @@ class VerticalSegmentWriter {
 
 private:
     friend class ::doris::BlockAggregator;
+    friend class TestVerticalSegmentWriter;
     uint32_t _segment_id;
     TabletSchemaSPtr _tablet_schema;
     BaseTabletSPtr _tablet;

be/src/storage/tablet/tablet_meta.cpp

@@ -417,6 +417,34 @@ void TabletMeta::init_column_from_tcolumn(uint32_t unique_id, const TColumn& tco
     if (tcolumn.__isset.variant_enable_nested_group) {
         column->set_variant_enable_nested_group(tcolumn.variant_enable_nested_group);
     }
+    if (tcolumn.__isset.compression_type) {
+        // The raw cast below is only valid while TCompressionType (thrift) and
+        // CompressionTypePB (proto) stay numerically identical. Guard every value
+        // at compile time so a future reorder of either enum fails to build
+        // instead of silently writing a wrong compression tag into segments.
+        static_assert(static_cast<int>(TCompressionType::UNKNOWN_COMPRESSION) ==
+                      static_cast<int>(segment_v2::UNKNOWN_COMPRESSION));
+        static_assert(static_cast<int>(TCompressionType::DEFAULT_COMPRESSION) ==
+                      static_cast<int>(segment_v2::DEFAULT_COMPRESSION));
+        static_assert(static_cast<int>(TCompressionType::NO_COMPRESSION) ==
+                      static_cast<int>(segment_v2::NO_COMPRESSION));
+        static_assert(static_cast<int>(TCompressionType::SNAPPY) ==
+                      static_cast<int>(segment_v2::SNAPPY));
+        static_assert(static_cast<int>(TCompressionType::LZ4) == static_cast<int>(segment_v2::LZ4));
+        static_assert(static_cast<int>(TCompressionType::LZ4F) ==
+                      static_cast<int>(segment_v2::LZ4F));
+        static_assert(static_cast<int>(TCompressionType::ZLIB) ==
+                      static_cast<int>(segment_v2::ZLIB));
+        static_assert(static_cast<int>(TCompressionType::ZSTD) ==
+                      static_cast<int>(segment_v2::ZSTD));
+        static_assert(static_cast<int>(TCompressionType::LZ4HC) ==
+                      static_cast<int>(segment_v2::LZ4HC));
+        column->set_compression_type(
+                static_cast<segment_v2::CompressionTypePB>(tcolumn.compression_type));
+        if (tcolumn.__isset.compression_level && tcolumn.compression_level > 0) {
+            column->set_compression_level(tcolumn.compression_level);
+        }
+    }
 }
 
 void TabletMeta::init_schema_from_thrift(const TTabletSchema& tablet_schema,

be/src/storage/tablet/tablet_schema.cpp

@@ -555,6 +555,8 @@ void TabletColumn::init_from_pb(const ColumnPB& column) {
     if (column.has_pattern_type()) {
         _pattern_type = column.pattern_type();
     }
+    _compression = column.compression_type();
+    _compression_level = column.has_compression_level() ? column.compression_level() : 0;
 }
 
 TabletColumn TabletColumn::create_materialized_variant_column(const std::string& root,
@@ -641,6 +643,12 @@ void TabletColumn::to_schema_pb(ColumnPB* column) const {
     column->set_variant_doc_materialization_min_rows(_variant.doc_materialization_min_rows);
     column->set_variant_doc_hash_shard_count(_variant.doc_hash_shard_count);
     column->set_variant_enable_nested_group(_variant.enable_nested_group);
+    if (has_compression()) {
+        column->set_compression_type(_compression);
+        if (_compression_level > 0) {
+            column->set_compression_level(_compression_level);
+        }
+    }
 }
 
 void TabletColumn::add_sub_column(TabletColumn& sub_column) {

be/src/storage/tablet/tablet_schema.h

@@ -109,6 +109,9 @@ class TabletColumn : public MetadataAdder<TabletColumn> {
     void set_type(FieldType type) { _type = type; }
     bool is_key() const { return _is_key; }
     bool is_nullable() const { return _is_nullable; }
+    bool has_compression() const { return _compression != segment_v2::UNKNOWN_COMPRESSION; }
+    segment_v2::CompressionTypePB compression() const { return _compression; }
+    int compression_level() const { return _compression_level; }
     bool is_auto_increment() const { return _is_auto_increment; }
     bool is_seqeunce_col() const { return _col_name == SEQUENCE_COL; }
     bool is_on_update_current_timestamp() const { return _is_on_update_current_timestamp; }
@@ -302,6 +305,9 @@ class TabletColumn : public MetadataAdder<TabletColumn> {
     bool _has_default_value = false;
     std::string _default_value;
 
+    segment_v2::CompressionTypePB _compression = segment_v2::UNKNOWN_COMPRESSION;
+    int _compression_level = 0;
+
     bool _is_decimal = false;
     int32_t _precision = -1;
     int32_t _frac = -1;