diff --git a/be/src/storage/segment/column_writer.cpp b/be/src/storage/segment/column_writer.cpp index 7d604897494f7d..bd434fbc29faf1 100644 --- a/be/src/storage/segment/column_writer.cpp +++ b/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; diff --git a/be/src/storage/segment/segment_writer.cpp b/be/src/storage/segment/segment_writer.cpp index 2afbf62fb34ac5..a11f7dbd9d2b1d 100644 --- a/be/src/storage/segment/segment_writer.cpp +++ b/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()); diff --git a/be/src/storage/segment/vertical_segment_writer.cpp b/be/src/storage/segment/vertical_segment_writer.cpp index ea9bae3a9266f2..3ecba29b13b8f6 100644 --- a/be/src/storage/segment/vertical_segment_writer.cpp +++ b/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(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()); diff --git a/be/src/storage/segment/vertical_segment_writer.h b/be/src/storage/segment/vertical_segment_writer.h index ba381ea20c705e..fd0ab2cb0d96c1 100644 --- a/be/src/storage/segment/vertical_segment_writer.h +++ b/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; diff --git a/be/src/storage/tablet/tablet_meta.cpp b/be/src/storage/tablet/tablet_meta.cpp index 557fe02743b178..86872096c4986e 100644 --- a/be/src/storage/tablet/tablet_meta.cpp +++ b/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(TCompressionType::UNKNOWN_COMPRESSION) == + static_cast(segment_v2::UNKNOWN_COMPRESSION)); + static_assert(static_cast(TCompressionType::DEFAULT_COMPRESSION) == + static_cast(segment_v2::DEFAULT_COMPRESSION)); + static_assert(static_cast(TCompressionType::NO_COMPRESSION) == + static_cast(segment_v2::NO_COMPRESSION)); + static_assert(static_cast(TCompressionType::SNAPPY) == + static_cast(segment_v2::SNAPPY)); + static_assert(static_cast(TCompressionType::LZ4) == static_cast(segment_v2::LZ4)); + static_assert(static_cast(TCompressionType::LZ4F) == + static_cast(segment_v2::LZ4F)); + static_assert(static_cast(TCompressionType::ZLIB) == + static_cast(segment_v2::ZLIB)); + static_assert(static_cast(TCompressionType::ZSTD) == + static_cast(segment_v2::ZSTD)); + static_assert(static_cast(TCompressionType::LZ4HC) == + static_cast(segment_v2::LZ4HC)); + column->set_compression_type( + static_cast(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, diff --git a/be/src/storage/tablet/tablet_schema.cpp b/be/src/storage/tablet/tablet_schema.cpp index 8dec9913f8815c..ccecd0b6300928 100644 --- a/be/src/storage/tablet/tablet_schema.cpp +++ b/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) { diff --git a/be/src/storage/tablet/tablet_schema.h b/be/src/storage/tablet/tablet_schema.h index caa7c9b54f6c42..334d294a4234df 100644 --- a/be/src/storage/tablet/tablet_schema.h +++ b/be/src/storage/tablet/tablet_schema.h @@ -109,6 +109,9 @@ class TabletColumn : public MetadataAdder { 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 { 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; diff --git a/be/src/util/block_compression.cpp b/be/src/util/block_compression.cpp index 47c67a011a1701..b1127483788203 100644 --- a/be/src/util/block_compression.cpp +++ b/be/src/util/block_compression.cpp @@ -47,6 +47,7 @@ #include #include #include +#include #include "absl/strings/substitute.h" #include "common/config.h" @@ -580,6 +581,8 @@ class Lz4HCBlockCompression : public BlockCompressionCodec { static Lz4HCBlockCompression s_instance; return &s_instance; } + Lz4HCBlockCompression() = default; + explicit Lz4HCBlockCompression(int level) : _compression_level(level) {} ~Lz4HCBlockCompression() { SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER( ExecEnv::GetInstance()->block_compression_mem_tracker()); @@ -659,10 +662,20 @@ class Lz4HCBlockCompression : public BlockCompressionCodec { if (localCtx.get() == nullptr) { return Status::InvalidArgument("new LZ4HC context error"); } - localCtx->ctx = LZ4_createStreamHC(); + // Allocate the native stream under the compression tracker so its + // creation and the destructor's LZ4_freeStreamHC() hit the same tracker. + { + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER( + ExecEnv::GetInstance()->block_compression_mem_tracker()); + localCtx->ctx = LZ4_createStreamHC(); + } if (localCtx->ctx == nullptr) { return Status::InvalidArgument("LZ4_createStreamHC error"); } + // A newly created stream defaults to the library's default level, so + // apply the requested level here; otherwise the first page compressed + // by this context would ignore the configured level. + LZ4_resetStreamHC_fast(localCtx->ctx, static_cast(_compression_level)); out = std::move(localCtx); return Status::OK(); } @@ -1077,6 +1090,8 @@ class ZstdBlockCompression : public BlockCompressionCodec { static ZstdBlockCompression s_instance; return &s_instance; } + ZstdBlockCompression() = default; + explicit ZstdBlockCompression(int level) : _compression_level(level) {} ~ZstdBlockCompression() { SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER( ExecEnv::GetInstance()->block_compression_mem_tracker()); @@ -1123,47 +1138,51 @@ class ZstdBlockCompression : public BlockCompressionCodec { compressed_buf.size = max_len; } - // set compression level to default 3 - auto ret = ZSTD_CCtx_setParameter(context->ctx, ZSTD_c_compressionLevel, - ZSTD_CLEVEL_DEFAULT); - if (ZSTD_isError(ret)) { - return Status::InvalidArgument("ZSTD_CCtx_setParameter compression level error: {}", - ZSTD_getErrorString(ZSTD_getErrorCode(ret))); - } - // set checksum flag to 1 - ret = ZSTD_CCtx_setParameter(context->ctx, ZSTD_c_checksumFlag, 1); - if (ZSTD_isError(ret)) { - return Status::InvalidArgument("ZSTD_CCtx_setParameter checksumFlag error: {}", - ZSTD_getErrorString(ZSTD_getErrorCode(ret))); - } - ZSTD_outBuffer out_buf = {compressed_buf.data, compressed_buf.size, 0}; + { + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER( + ExecEnv::GetInstance()->block_compression_mem_tracker()); + auto ret = ZSTD_CCtx_setParameter(context->ctx, ZSTD_c_compressionLevel, + _compression_level); + if (ZSTD_isError(ret)) { + return Status::InvalidArgument( + "ZSTD_CCtx_setParameter compression level error: {}", + ZSTD_getErrorString(ZSTD_getErrorCode(ret))); + } + // set checksum flag to 1 + ret = ZSTD_CCtx_setParameter(context->ctx, ZSTD_c_checksumFlag, 1); + if (ZSTD_isError(ret)) { + return Status::InvalidArgument("ZSTD_CCtx_setParameter checksumFlag error: {}", + ZSTD_getErrorString(ZSTD_getErrorCode(ret))); + } - for (size_t i = 0; i < inputs.size(); i++) { - ZSTD_inBuffer in_buf = {inputs[i].data, inputs[i].size, 0}; + for (size_t i = 0; i < inputs.size(); i++) { + ZSTD_inBuffer in_buf = {inputs[i].data, inputs[i].size, 0}; - bool last_input = (i == inputs.size() - 1); - auto mode = last_input ? ZSTD_e_end : ZSTD_e_continue; + bool last_input = (i == inputs.size() - 1); + auto mode = last_input ? ZSTD_e_end : ZSTD_e_continue; - bool finished = false; - do { - // do compress - ret = ZSTD_compressStream2(context->ctx, &out_buf, &in_buf, mode); + bool finished = false; + do { + // do compress + ret = ZSTD_compressStream2(context->ctx, &out_buf, &in_buf, mode); - if (ZSTD_isError(ret)) { - compress_failed = true; - return Status::InternalError("ZSTD_compressStream2 error: {}", - ZSTD_getErrorString(ZSTD_getErrorCode(ret))); - } + if (ZSTD_isError(ret)) { + compress_failed = true; + return Status::InternalError( + "ZSTD_compressStream2 error: {}", + ZSTD_getErrorString(ZSTD_getErrorCode(ret))); + } - // ret is ZSTD hint for needed output buffer size - if (ret > 0 && out_buf.pos == out_buf.size) { - compress_failed = true; - return Status::InternalError("ZSTD_compressStream2 output buffer full"); - } + // ret is ZSTD hint for needed output buffer size + if (ret > 0 && out_buf.pos == out_buf.size) { + compress_failed = true; + return Status::InternalError("ZSTD_compressStream2 output buffer full"); + } - finished = last_input ? (ret == 0) : (in_buf.pos == inputs[i].size); - } while (!finished); + finished = last_input ? (ret == 0) : (in_buf.pos == inputs[i].size); + } while (!finished); + } } // set compressed size for caller @@ -1215,7 +1234,13 @@ class ZstdBlockCompression : public BlockCompressionCodec { return Status::InvalidArgument("failed to new ZSTD CContext"); } //typedef LZ4F_cctx* LZ4F_compressionContext_t; - localCtx->ctx = ZSTD_createCCtx(); + // Allocate the native context under the compression tracker so its + // creation and the destructor's ZSTD_freeCCtx() hit the same tracker. + { + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER( + ExecEnv::GetInstance()->block_compression_mem_tracker()); + localCtx->ctx = ZSTD_createCCtx(); + } if (localCtx->ctx == nullptr) { return Status::InvalidArgument("Failed to create ZSTD compress ctx"); } @@ -1262,6 +1287,7 @@ class ZstdBlockCompression : public BlockCompressionCodec { } private: + int _compression_level = ZSTD_CLEVEL_DEFAULT; mutable std::mutex _ctx_c_mutex; mutable std::vector> _ctx_c_pool; @@ -1615,6 +1641,84 @@ Status get_block_compression_codec(segment_v2::CompressionTypePB type, return Status::OK(); } +// Process-wide registry of level-aware codecs, keyed by (type, level). All +// column writers that request the same codec+level share one instance, so its +// internal context pool is reused according to actual write concurrency rather +// than allocated once per column. Instances live for the process lifetime (like +// the type-only singletons above), so their native contexts are never torn down +// per segment. +namespace { +class LeveledCompressionCodecPool { +public: + static LeveledCompressionCodecPool& instance() { + static LeveledCompressionCodecPool s_instance; + return s_instance; + } + + Status get(segment_v2::CompressionTypePB type, int level, BlockCompressionCodec** codec) { + const int64_t key = (static_cast(type) << 32) | static_cast(level); + { + std::lock_guard l(_mutex); + auto it = _codecs.find(key); + if (it != _codecs.end()) { + *codec = it->second.get(); + return Status::OK(); + } + } + + // Build the instance outside the lock; init() may allocate native state. + std::unique_ptr instance; + switch (type) { + case segment_v2::CompressionTypePB::ZSTD: + instance = std::make_unique(level); + break; + case segment_v2::CompressionTypePB::LZ4HC: + instance = std::make_unique(level); + break; + default: + return Status::InternalError("compression type({}) is not level-aware", type); + } + RETURN_IF_ERROR(instance->init()); + + std::lock_guard l(_mutex); + // Another thread may have inserted the same key while we were building. + auto it = _codecs.try_emplace(key, std::move(instance)).first; + *codec = it->second.get(); + return Status::OK(); + } + + // Test hook: drop all pooled instances so a fresh test observes a clean pool. + void clear() { + std::lock_guard l(_mutex); + _codecs.clear(); + } + +private: + std::mutex _mutex; + std::unordered_map> _codecs; +}; +} // namespace + +Status get_block_compression_codec(segment_v2::CompressionTypePB type, int level, + BlockCompressionCodec** codec) { + // level <= 0 means "use codec default" -> fall back to the stateless singleton path. + if (level <= 0) { + return get_block_compression_codec(type, codec); + } + switch (type) { + case segment_v2::CompressionTypePB::ZSTD: + case segment_v2::CompressionTypePB::LZ4HC: + return LeveledCompressionCodecPool::instance().get(type, level, codec); + default: + // types without a tunable level ignore it and use the singleton + return get_block_compression_codec(type, codec); + } +} + +void clear_leveled_compression_codec_pool_for_test() { + LeveledCompressionCodecPool::instance().clear(); +} + // this can only be used in hive text write Status get_block_compression_codec(TFileCompressType::type type, BlockCompressionCodec** codec) { switch (type) { diff --git a/be/src/util/block_compression.h b/be/src/util/block_compression.h index 3b2a0197fa8105..23a8f04db2c24e 100644 --- a/be/src/util/block_compression.h +++ b/be/src/util/block_compression.h @@ -84,6 +84,16 @@ class BlockCompressionCodec { Status get_block_compression_codec(segment_v2::CompressionTypePB type, BlockCompressionCodec** codec); +// Level-aware variant. If `level` > 0 and `type` is a level-aware codec (ZSTD, LZ4HC), +// returns a process-wide instance shared by all callers that request the same +// (type, level) pair (do not delete). Otherwise `*codec` points to the type-only +// singleton. In both cases the returned codec is owned by the process, not the caller. +Status get_block_compression_codec(segment_v2::CompressionTypePB type, int level, + BlockCompressionCodec** codec); + +// Test-only: drops all pooled level-aware codec instances. +void clear_leveled_compression_codec_pool_for_test(); + Status get_block_compression_codec(tparquet::CompressionCodec::type parquet_codec, BlockCompressionCodec** codec); diff --git a/be/test/storage/segment/column_compression_roundtrip_test.cpp b/be/test/storage/segment/column_compression_roundtrip_test.cpp new file mode 100644 index 00000000000000..9f63efe5c39617 --- /dev/null +++ b/be/test/storage/segment/column_compression_roundtrip_test.cpp @@ -0,0 +1,267 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include +#include +#include + +#include "common/config.h" +#include "core/assert_cast.h" +#include "core/column/column_vector.h" +#include "io/fs/file_reader.h" +#include "io/fs/file_writer.h" +#include "io/fs/local_file_system.h" +#include "storage/olap_common.h" +#include "storage/segment/column_reader.h" +#include "storage/segment/column_writer.h" +#include "storage/tablet/tablet_schema.h" + +namespace doris::segment_v2 { + +static const std::string TEST_DIR = "./ut_dir/column_compression_roundtrip_test"; + +class ColumnCompressionRoundtripTest : public ::testing::Test { +protected: + void SetUp() override { + _old_disable_storage_page_cache = config::disable_storage_page_cache; + config::disable_storage_page_cache = true; + auto st = io::global_local_filesystem()->delete_directory(TEST_DIR); + ASSERT_TRUE(st.ok()) << st.to_string(); + st = io::global_local_filesystem()->create_directory(TEST_DIR); + ASSERT_TRUE(st.ok()) << st.to_string(); + } + + void TearDown() override { + EXPECT_TRUE(io::global_local_filesystem()->delete_directory(TEST_DIR).ok()); + config::disable_storage_page_cache = _old_disable_storage_page_cache; + } + +private: + bool _old_disable_storage_page_cache = false; +}; + +// Write `num_rows` INT values through a segment column configured with the given +// compression codec + level, reopen the file, read every value back, and assert +// the round-trip is lossless. This is an integration smoke test of the full +// per-column compression plumbing: the writer picks a level-aware codec, the +// codec/level is persisted into ColumnMetaPB, and the reader decompresses. +// +// NOTE: a lossless round-trip is level-INDEPENDENT -- decompression never needs +// the level, so this test cannot by itself prove the configured level reached +// the codec. The observable level effect (different level => different output) +// is guarded by BlockCompressionTest.DifferentLevelsProduceDifferentOutput at +// the codec layer and by SegmentBytesDifferWithLevel below at the segment layer. +// +// A tiny data_page_size forces many data pages so the codec is invoked once per +// page. +static void test_int_roundtrip(CompressionTypePB compression, int compression_level, + const std::string& test_name) { + const int32_t num_rows = 4096; + std::vector src(num_rows); + for (int32_t i = 0; i < num_rows; ++i) { + // Repetitive-but-varying data so compression actually kicks in. + src[i] = (i % 97) * 31 + (i / 97); + } + + ColumnMetaPB meta; + std::string fname = TEST_DIR + "/" + test_name; + auto fs = io::global_local_filesystem(); + + // ---- write ---- + { + io::FileWriterPtr file_writer; + Status st = fs->create_file(fname, &file_writer); + ASSERT_TRUE(st.ok()) << st.to_string(); + + ColumnWriterOptions writer_opts; + writer_opts.meta = &meta; + writer_opts.meta->set_column_id(0); + writer_opts.meta->set_unique_id(0); + writer_opts.meta->set_type(static_cast(FieldType::OLAP_FIELD_TYPE_INT)); + writer_opts.meta->set_length(0); + writer_opts.meta->set_encoding(PLAIN_ENCODING); + writer_opts.meta->set_compression(compression); + if (compression_level > 0) { + writer_opts.meta->set_compression_level(compression_level); + } + writer_opts.meta->set_is_nullable(false); + writer_opts.data_page_size = 128; + + TabletColumn column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_INT); + std::unique_ptr writer; + st = ColumnWriter::create(writer_opts, &column, file_writer.get(), &writer); + ASSERT_TRUE(st.ok()) << st.to_string(); + st = writer->init(); + ASSERT_TRUE(st.ok()) << st.to_string(); + + for (int32_t i = 0; i < num_rows; ++i) { + st = writer->append(false, &src[i]); + ASSERT_TRUE(st.ok()) << st.to_string(); + } + + ASSERT_TRUE(writer->finish().ok()); + ASSERT_TRUE(writer->write_data().ok()); + ASSERT_TRUE(writer->write_ordinal_index().ok()); + ASSERT_TRUE(file_writer->close().ok()); + } + + // Codec must be persisted so the reader can reconstruct the decompressor. + // (compression_level is round-tripped through ColumnMetaPB by the caller and + // consumed by the writer only; asserting it here would be tautological.) + ASSERT_EQ(meta.compression(), compression); + + // ---- read back ---- + io::FileReaderSPtr file_reader; + ASSERT_TRUE(fs->open_file(fname, &file_reader).ok()); + + ColumnReaderOptions reader_opts; + std::shared_ptr reader; + ASSERT_TRUE(ColumnReader::create(reader_opts, meta, num_rows, file_reader, &reader).ok()); + + TabletColumn read_column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_INT); + ColumnIteratorUPtr iter; + ASSERT_TRUE(reader->new_iterator(&iter, &read_column).ok()); + + ColumnIteratorOptions iter_opts; + OlapReaderStatistics stats; + iter_opts.stats = &stats; + iter_opts.file_reader = file_reader.get(); + ASSERT_TRUE(iter->init(iter_opts).ok()); + ASSERT_TRUE(iter->seek_to_ordinal(0).ok()); + + MutableColumnPtr dst = ColumnInt32::create(); + size_t total_read = 0; + while (total_read < static_cast(num_rows)) { + size_t rows_read = 1024; + bool has_null = false; + ASSERT_TRUE(iter->next_batch(&rows_read, dst, &has_null).ok()); + if (rows_read == 0) { + break; + } + total_read += rows_read; + } + ASSERT_EQ(total_read, static_cast(num_rows)); + + const auto& int_col = assert_cast(*dst); + ASSERT_EQ(int_col.size(), static_cast(num_rows)); + for (int32_t i = 0; i < num_rows; ++i) { + ASSERT_EQ(src[i], int_col.get_element(i)) + << "codec=" << compression << " level=" << compression_level << " idx=" << i; + } +} + +TEST_F(ColumnCompressionRoundtripTest, Lz4fNoLevel) { + test_int_roundtrip(CompressionTypePB::LZ4F, 0, "int_lz4f"); +} + +TEST_F(ColumnCompressionRoundtripTest, ZstdWithLevel) { + test_int_roundtrip(CompressionTypePB::ZSTD, 9, "int_zstd_l9"); +} + +TEST_F(ColumnCompressionRoundtripTest, ZstdMaxLevel) { + test_int_roundtrip(CompressionTypePB::ZSTD, 22, "int_zstd_l22"); +} + +TEST_F(ColumnCompressionRoundtripTest, Lz4hcWithLevel) { + test_int_roundtrip(CompressionTypePB::LZ4HC, 12, "int_lz4hc_l12"); +} + +TEST_F(ColumnCompressionRoundtripTest, ZstdDefaultLevelFallback) { + // level == 0 means "use codec default"; must still round-trip via the singleton. + test_int_roundtrip(CompressionTypePB::ZSTD, 0, "int_zstd_default"); +} + +TEST_F(ColumnCompressionRoundtripTest, SnappyNoLevel) { + test_int_roundtrip(CompressionTypePB::SNAPPY, 0, "int_snappy"); +} + +TEST_F(ColumnCompressionRoundtripTest, ZlibNoLevel) { + test_int_roundtrip(CompressionTypePB::ZLIB, 0, "int_zlib"); +} + +// Segment-layer level oracle. Write the SAME data through a real segment column +// at two ZSTD levels using a realistic page size, then compare the resulting +// on-disk bytes. This is the end-to-end guard that the writer forwards +// meta.compression_level() into the codec (column_writer.cpp): if the level were +// dropped, both writes would use the codec default and the files would be +// byte-identical. (We compare bytes, not sizes: ZSTD level is search effort, not +// a monotonic size bound, so a higher level is not guaranteed to be smaller.) +static std::string write_int_segment_bytes(int compression_level, int32_t num_rows, + const std::string& test_name) { + std::vector src(num_rows); + for (int32_t i = 0; i < num_rows; ++i) { + src[i] = ((i / 64) % 41) * 100 + (i % 7); + } + + ColumnMetaPB meta; + std::string fname = TEST_DIR + "/" + test_name; + auto fs = io::global_local_filesystem(); + + io::FileWriterPtr file_writer; + EXPECT_TRUE(fs->create_file(fname, &file_writer).ok()); + + ColumnWriterOptions writer_opts; + writer_opts.meta = &meta; + writer_opts.meta->set_column_id(0); + writer_opts.meta->set_unique_id(0); + writer_opts.meta->set_type(static_cast(FieldType::OLAP_FIELD_TYPE_INT)); + writer_opts.meta->set_length(0); + writer_opts.meta->set_encoding(PLAIN_ENCODING); + writer_opts.meta->set_compression(CompressionTypePB::ZSTD); + if (compression_level > 0) { + writer_opts.meta->set_compression_level(compression_level); + } + writer_opts.meta->set_is_nullable(false); + writer_opts.data_page_size = 1024 * 1024; + + TabletColumn column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_INT); + std::unique_ptr writer; + EXPECT_TRUE(ColumnWriter::create(writer_opts, &column, file_writer.get(), &writer).ok()); + EXPECT_TRUE(writer->init().ok()); + for (int32_t i = 0; i < num_rows; ++i) { + EXPECT_TRUE(writer->append(false, &src[i]).ok()); + } + EXPECT_TRUE(writer->finish().ok()); + EXPECT_TRUE(writer->write_data().ok()); + EXPECT_TRUE(writer->write_ordinal_index().ok()); + EXPECT_TRUE(file_writer->close().ok()); + + int64_t size = 0; + EXPECT_TRUE(fs->file_size(fname, &size).ok()); + EXPECT_GT(size, 0); + + io::FileReaderSPtr file_reader; + EXPECT_TRUE(fs->open_file(fname, &file_reader).ok()); + std::string bytes(static_cast(size), '\0'); + size_t bytes_read = 0; + EXPECT_TRUE(file_reader->read_at(0, Slice(bytes.data(), bytes.size()), &bytes_read).ok()); + EXPECT_EQ(bytes_read, static_cast(size)); + return bytes; +} + +TEST_F(ColumnCompressionRoundtripTest, SegmentBytesDifferWithLevel) { + const int32_t num_rows = 512 * 1024; // multi-page column + std::string low = write_int_segment_bytes(1, num_rows, "int_seg_zstd_l1"); + std::string high = write_int_segment_bytes(22, num_rows, "int_seg_zstd_l22"); + EXPECT_NE(low, high) << "ZSTD level 1 and level 22 produced byte-identical segments; " + << "the writer likely dropped the per-column compression level"; +} + +} // namespace doris::segment_v2 diff --git a/be/test/storage/segment/segment_writer_column_compression_test.cpp b/be/test/storage/segment/segment_writer_column_compression_test.cpp new file mode 100644 index 00000000000000..ae2c7e632aed55 --- /dev/null +++ b/be/test/storage/segment/segment_writer_column_compression_test.cpp @@ -0,0 +1,258 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include +#include +#include + +#include +#include + +#include "io/fs/local_file_system.h" +#include "storage/olap_common.h" +#include "storage/rowset/rowset_id_generator.h" +#include "storage/segment/column_writer.h" +#include "storage/segment/segment_writer.h" +#include "storage/segment/vertical_segment_writer.h" +#include "storage/tablet/tablet_schema.h" + +namespace doris::segment_v2 { + +// Both SegmentWriter and VerticalSegmentWriter build a ColumnMetaPB for every column via +// (respectively) init_column_meta() and _init_column_meta(). These tests verify that a +// per-column compression override on the TabletColumn is propagated into that ColumnMetaPB, +// on BOTH writers -- VerticalSegmentWriter is the default write path +// (config::enable_vertical_segment_writer defaults to true), so a regression there silently +// drops the feature. They also verify that the per-column codec takes priority over the +// writer-level compression setting. +// +// SegmentWriter::init_column_meta is public, so it is called directly. VerticalSegmentWriter's +// _init_column_meta is private, so this subclass (a declared friend in the production header, +// mirroring test_segment_writer.h's TestSegmentWriter) exposes it. +class TestVerticalSegmentWriter : public VerticalSegmentWriter { +public: + using VerticalSegmentWriter::VerticalSegmentWriter; + void build_meta(ColumnMetaPB* meta, const TabletColumn& column, + const ColumnWriterOptions& opts) { + _init_column_meta(meta, 0, column, opts); + } +}; + +static const std::string kSegmentDir = "./ut_dir/segment_writer_column_compression_test"; + +// Build a plain INT TabletColumn through the production ColumnPB path so the per-column +// compression override (when requested) is set exactly the way init_from_pb() does at runtime. +static TabletColumn make_int_column(bool has_compression, CompressionTypePB compression, + int compression_level) { + ColumnPB column_pb; + column_pb.set_unique_id(0); + column_pb.set_name("c0"); + column_pb.set_type("INT"); + column_pb.set_is_key(true); + column_pb.set_is_nullable(false); + column_pb.set_length(4); + if (has_compression) { + column_pb.set_compression_type(compression); + if (compression_level > 0) { + column_pb.set_compression_level(compression_level); + } + } + TabletColumn column; + column.init_from_pb(column_pb); + return column; +} + +static TabletSchemaSPtr make_schema(const TabletColumn& column) { + TabletSchemaSPtr schema = std::make_shared(); + schema->append_column(column); + schema->_keys_type = DUP_KEYS; + return schema; +} + +class SegmentWriterColumnCompressionTest : public testing::Test { +public: + void SetUp() override { + auto fs = io::global_local_filesystem(); + auto st = fs->delete_directory(kSegmentDir); + ASSERT_TRUE(st.ok() || st.is()) << st; + st = fs->create_directory(kSegmentDir); + ASSERT_TRUE(st.ok()) << st; + } + + void TearDown() override { + EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kSegmentDir).ok()); + } + + io::FileWriterPtr create_file_writer(size_t segment_id) { + RowsetId rowset_id; + rowset_id.init(1); + std::string filename = fmt::format("{}_{}.dat", rowset_id.to_string(), segment_id); + std::string path = fmt::format("{}/{}", kSegmentDir, filename); + io::FileWriterPtr file_writer; + auto st = io::global_local_filesystem()->create_file(path, &file_writer); + EXPECT_TRUE(st.ok()) << st; + return file_writer; + } +}; + +// --- SegmentWriter (legacy path) --- + +TEST_F(SegmentWriterColumnCompressionTest, SegmentWriterUsesPerColumnCodec) { + auto column = make_int_column(true, CompressionTypePB::ZSTD, 9); + auto schema = make_schema(column); + SegmentWriterOptions opts; + opts.compression_type = LZ4F; // table-level default, must be overridden per column + auto file_writer = create_file_writer(0); + SegmentWriter writer(file_writer.get(), 0, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.init_column_meta(&meta, 0, column, col_opts); + EXPECT_EQ(meta.compression(), CompressionTypePB::ZSTD); + ASSERT_TRUE(meta.has_compression_level()); + EXPECT_EQ(meta.compression_level(), 9); +} + +TEST_F(SegmentWriterColumnCompressionTest, SegmentWriterFallsBackToTableCodec) { + auto column = make_int_column(false, UNKNOWN_COMPRESSION, 0); + auto schema = make_schema(column); + SegmentWriterOptions opts; + opts.compression_type = LZ4F; + auto file_writer = create_file_writer(1); + SegmentWriter writer(file_writer.get(), 1, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.init_column_meta(&meta, 0, column, col_opts); + EXPECT_EQ(meta.compression(), LZ4F); + EXPECT_FALSE(meta.has_compression_level()); +} + +TEST_F(SegmentWriterColumnCompressionTest, UpgradeLegacyColumnPbInheritsTableCodec) { + ColumnPB legacy_pb; + legacy_pb.set_unique_id(0); + legacy_pb.set_name("c0"); + legacy_pb.set_type("INT"); + legacy_pb.set_is_key(true); + legacy_pb.set_is_nullable(false); + legacy_pb.set_length(4); + std::string serialized; + ASSERT_TRUE(legacy_pb.SerializeToString(&serialized)); + + ColumnPB upgraded_pb; + ASSERT_TRUE(upgraded_pb.ParseFromString(serialized)); + ASSERT_FALSE(upgraded_pb.has_compression_type()); + ASSERT_EQ(upgraded_pb.compression_type(), UNKNOWN_COMPRESSION); + + TabletColumn column(upgraded_pb); + auto schema = make_schema(column); + SegmentWriterOptions opts; + opts.compression_type = LZ4F; + auto file_writer = create_file_writer(2); + SegmentWriter writer(file_writer.get(), 2, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.init_column_meta(&meta, 0, column, col_opts); + EXPECT_EQ(meta.compression(), LZ4F); + EXPECT_FALSE(meta.has_compression_level()); +} + +TEST_F(SegmentWriterColumnCompressionTest, SegmentWriterColumnCodecOverridesTableNoCompression) { + auto column = make_int_column(true, CompressionTypePB::ZSTD, 9); + auto schema = make_schema(column); + SegmentWriterOptions opts; + opts.compression_type = NO_COMPRESSION; + auto file_writer = create_file_writer(2); + SegmentWriter writer(file_writer.get(), 2, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.init_column_meta(&meta, 0, column, col_opts); + EXPECT_EQ(meta.compression(), CompressionTypePB::ZSTD); + ASSERT_TRUE(meta.has_compression_level()); + EXPECT_EQ(meta.compression_level(), 9); +} + +// --- VerticalSegmentWriter (default path) --- + +TEST_F(SegmentWriterColumnCompressionTest, VerticalSegmentWriterUsesPerColumnCodec) { + auto column = make_int_column(true, CompressionTypePB::LZ4HC, 12); + auto schema = make_schema(column); + VerticalSegmentWriterOptions opts; + opts.compression_type = LZ4F; + auto file_writer = create_file_writer(3); + TestVerticalSegmentWriter writer(file_writer.get(), 3, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.build_meta(&meta, column, col_opts); + EXPECT_EQ(meta.compression(), CompressionTypePB::LZ4HC); + ASSERT_TRUE(meta.has_compression_level()); + EXPECT_EQ(meta.compression_level(), 12); +} + +TEST_F(SegmentWriterColumnCompressionTest, VerticalSegmentWriterFallsBackToTableCodec) { + auto column = make_int_column(false, UNKNOWN_COMPRESSION, 0); + auto schema = make_schema(column); + VerticalSegmentWriterOptions opts; + opts.compression_type = LZ4F; + auto file_writer = create_file_writer(4); + TestVerticalSegmentWriter writer(file_writer.get(), 4, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.build_meta(&meta, column, col_opts); + EXPECT_EQ(meta.compression(), LZ4F); + EXPECT_FALSE(meta.has_compression_level()); +} + +TEST_F(SegmentWriterColumnCompressionTest, + VerticalSegmentWriterColumnCodecOverridesTableNoCompression) { + auto column = make_int_column(true, CompressionTypePB::ZSTD, 9); + auto schema = make_schema(column); + VerticalSegmentWriterOptions opts; + opts.compression_type = NO_COMPRESSION; + auto file_writer = create_file_writer(5); + TestVerticalSegmentWriter writer(file_writer.get(), 5, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.build_meta(&meta, column, col_opts); + EXPECT_EQ(meta.compression(), CompressionTypePB::ZSTD); + ASSERT_TRUE(meta.has_compression_level()); + EXPECT_EQ(meta.compression_level(), 9); +} + +// A per-column codec with no explicit level must persist the codec but leave the level absent +// (0 => "use codec default"), on the default write path. +TEST_F(SegmentWriterColumnCompressionTest, VerticalSegmentWriterPerColumnCodecWithoutLevel) { + auto column = make_int_column(true, CompressionTypePB::ZSTD, 0); + auto schema = make_schema(column); + VerticalSegmentWriterOptions opts; + opts.compression_type = LZ4F; + auto file_writer = create_file_writer(6); + TestVerticalSegmentWriter writer(file_writer.get(), 6, schema, nullptr, nullptr, opts, nullptr); + + ColumnMetaPB meta; + ColumnWriterOptions col_opts; + writer.build_meta(&meta, column, col_opts); + EXPECT_EQ(meta.compression(), CompressionTypePB::ZSTD); + EXPECT_FALSE(meta.has_compression_level()); +} + +} // namespace doris::segment_v2 diff --git a/be/test/storage/tablet_schema_compression_test.cpp b/be/test/storage/tablet_schema_compression_test.cpp new file mode 100644 index 00000000000000..ee886302c7a746 --- /dev/null +++ b/be/test/storage/tablet_schema_compression_test.cpp @@ -0,0 +1,149 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include +#include +#include +#include +#include + +#include "storage/tablet/tablet_meta.h" +#include "storage/tablet/tablet_schema.h" + +namespace doris { + +TEST(TabletColumnCompressionTest, InitFromPbRoundTrip) { + ColumnPB pb; + pb.set_unique_id(1); + pb.set_name("c1"); + pb.set_type("INT"); + pb.set_compression_type(segment_v2::ZSTD); + pb.set_compression_level(9); + + TabletColumn col; + col.init_from_pb(pb); + ASSERT_TRUE(col.has_compression()); + ASSERT_EQ(col.compression(), segment_v2::ZSTD); + ASSERT_EQ(col.compression_level(), 9); + + ColumnPB out; + col.to_schema_pb(&out); + ASSERT_TRUE(out.has_compression_type()); + ASSERT_EQ(out.compression_type(), segment_v2::ZSTD); + ASSERT_EQ(out.compression_level(), 9); +} + +TEST(TabletColumnCompressionTest, InitFromPbNoOverride) { + ColumnPB pb; + pb.set_unique_id(1); + pb.set_name("c1"); + pb.set_type("INT"); + TabletColumn col; + col.init_from_pb(pb); + ASSERT_FALSE(col.has_compression()); + ASSERT_EQ(col.compression(), segment_v2::UNKNOWN_COMPRESSION); + + ColumnPB out; + col.to_schema_pb(&out); + ASSERT_FALSE(out.has_compression_type()); +} + +TEST(TabletColumnCompressionTest, InitFromThriftRoundTrip) { + TColumn tcolumn; + tcolumn.column_name = "c1"; + tcolumn.column_type.type = TPrimitiveType::INT; + tcolumn.__set_is_key(true); + tcolumn.__set_compression_type(static_cast(TCompressionType::ZSTD)); + tcolumn.__set_compression_level(9); + + TabletColumn col; + col.init_from_thrift(tcolumn); + ASSERT_TRUE(col.has_compression()); + ASSERT_EQ(col.compression(), segment_v2::ZSTD); + ASSERT_EQ(col.compression_level(), 9); + + ColumnPB out; + col.to_schema_pb(&out); + ASSERT_TRUE(out.has_compression_type()); + ASSERT_EQ(out.compression_type(), segment_v2::ZSTD); + ASSERT_EQ(out.compression_level(), 9); +} + +TEST(TabletColumnCompressionTest, InitFromThriftNoOverride) { + TColumn tcolumn; + tcolumn.column_name = "c1"; + tcolumn.column_type.type = TPrimitiveType::INT; + tcolumn.__set_is_key(true); + + TabletColumn col; + col.init_from_thrift(tcolumn); + ASSERT_FALSE(col.has_compression()); + + ColumnPB out; + col.to_schema_pb(&out); + ASSERT_FALSE(out.has_compression_type()); +} + +// The persisted (non-cloud) tablet meta is built by init_column_from_tcolumn, not by +// TabletColumn::init_from_thrift. Compaction reloads that persisted ColumnPB, so if the +// compression override is dropped here it silently reverts to the table default after the +// first compaction. Assert the persisted ColumnPB carries the per-column codec. +TEST(TabletColumnCompressionTest, InitColumnFromTColumnPersistsCompression) { + TColumn tcolumn; + tcolumn.column_name = "c1"; + tcolumn.column_type.type = TPrimitiveType::INT; + tcolumn.__set_is_key(true); + tcolumn.__set_compression_type(static_cast(TCompressionType::ZSTD)); + tcolumn.__set_compression_level(9); + + ColumnPB column; + TabletMeta::init_column_from_tcolumn(1, tcolumn, &column); + ASSERT_TRUE(column.has_compression_type()); + ASSERT_EQ(column.compression_type(), segment_v2::ZSTD); + ASSERT_EQ(column.compression_level(), 9); +} + +// A codec without an explicit level must persist the type but leave the level unset +// (level absent => codec default). +TEST(TabletColumnCompressionTest, InitColumnFromTColumnPersistsCompressionNoLevel) { + TColumn tcolumn; + tcolumn.column_name = "c1"; + tcolumn.column_type.type = TPrimitiveType::INT; + tcolumn.__set_is_key(true); + tcolumn.__set_compression_type(static_cast(TCompressionType::LZ4F)); + + ColumnPB column; + TabletMeta::init_column_from_tcolumn(1, tcolumn, &column); + ASSERT_TRUE(column.has_compression_type()); + ASSERT_EQ(column.compression_type(), segment_v2::LZ4F); + ASSERT_FALSE(column.has_compression_level()); +} + +TEST(TabletColumnCompressionTest, InitColumnFromTColumnNoOverride) { + TColumn tcolumn; + tcolumn.column_name = "c1"; + tcolumn.column_type.type = TPrimitiveType::INT; + tcolumn.__set_is_key(true); + + ColumnPB column; + TabletMeta::init_column_from_tcolumn(1, tcolumn, &column); + ASSERT_FALSE(column.has_compression_type()); + ASSERT_EQ(column.compression_type(), segment_v2::UNKNOWN_COMPRESSION); + ASSERT_FALSE(column.has_compression_level()); +} + +} // namespace doris diff --git a/be/test/util/block_compression_test.cpp b/be/test/util/block_compression_test.cpp index f430a0274e5f93..703a435dbc894a 100644 --- a/be/test/util/block_compression_test.cpp +++ b/be/test/util/block_compression_test.cpp @@ -25,6 +25,8 @@ #include #include "gtest/gtest_pred_impl.h" +#include "runtime/exec_env.h" +#include "runtime/memory/mem_tracker_limiter.h" #include "util/faststring.h" namespace doris { @@ -148,4 +150,161 @@ TEST_F(BlockCompressionTest, multi) { test_multi_slices(segment_v2::CompressionTypePB::ZSTD); } +TEST_F(BlockCompressionTest, GetCodecWithLevelDefaultReturnsSingleton) { + for (auto type : {segment_v2::ZSTD, segment_v2::LZ4HC}) { + BlockCompressionCodec* codec = nullptr; + ASSERT_TRUE(get_block_compression_codec(type, 0, &codec).ok()); + ASSERT_NE(codec, nullptr); + BlockCompressionCodec* singleton = nullptr; + ASSERT_TRUE(get_block_compression_codec(type, &singleton).ok()); + ASSERT_EQ(codec, singleton); + } +} + +TEST_F(BlockCompressionTest, GetCodecWithZstdLevelReturnsSharedInstance) { + BlockCompressionCodec* codec = nullptr; + ASSERT_TRUE(get_block_compression_codec(segment_v2::ZSTD, 9, &codec).ok()); + ASSERT_NE(codec, nullptr); + BlockCompressionCodec* singleton = nullptr; + ASSERT_TRUE(get_block_compression_codec(segment_v2::ZSTD, &singleton).ok()); + ASSERT_NE(codec, singleton); // leveled instance, not the type-only singleton + // round-trip compress/decompress works at this level + std::string in(4096, 'x'); + for (size_t i = 0; i < in.size(); ++i) in[i] = static_cast(i % 251); + faststring compressed; + ASSERT_TRUE(codec->compress(Slice(in), &compressed).ok()); + std::string out(in.size(), '\0'); + Slice out_slice(out); + ASSERT_TRUE(codec->decompress(Slice(compressed.data(), compressed.size()), &out_slice).ok()); + ASSERT_EQ(std::string(out_slice.data, out_slice.size), in); +} + +// The observable effect of compression level is the compressed byte stream: two +// distinct levels of the same codec must produce DIFFERENT output on data that +// is neither trivially nor maximally compressible. If the writer/codec dropped +// the requested level, every level would collapse to the codec default and the +// outputs would be byte-identical. This is the real guard for the level fix in +// block_compression.cpp (ZSTD_c_compressionLevel / LZ4_resetStreamHC_fast); a +// lossless round-trip cannot detect a dropped level because decompression is +// level-independent. (Note: higher level does NOT guarantee smaller output -- +// ZSTD level is search effort, not a monotonic size bound -- so we assert +// difference + valid round-trip, not an ordering.) +static std::string compress_at_level(segment_v2::CompressionTypePB type, int level, + const std::string& in) { + BlockCompressionCodec* codec = nullptr; + EXPECT_TRUE(get_block_compression_codec(type, level, &codec).ok()); + EXPECT_NE(codec, nullptr); + faststring compressed; + EXPECT_TRUE(codec->compress(Slice(in), &compressed).ok()); + // every level must still decompress losslessly + std::string out(in.size(), '\0'); + Slice out_slice(out); + EXPECT_TRUE(codec->decompress(Slice(compressed.data(), compressed.size()), &out_slice).ok()); + EXPECT_EQ(std::string(out_slice.data, out_slice.size), in); + return std::string(reinterpret_cast(compressed.data()), compressed.size()); +} + +TEST_F(BlockCompressionTest, DowngradeLegacyCodecReadsLeveledData) { + std::string in; + in.reserve(64 * 1024); + for (int i = 0; in.size() < 64 * 1024; ++i) { + in += "apache doris column compression "; + in += std::to_string(i % 1024); + } + + struct Case { + segment_v2::CompressionTypePB type; + int level; + }; + for (auto c : {Case {segment_v2::ZSTD, 9}, Case {segment_v2::LZ4HC, 9}}) { + std::string compressed = compress_at_level(c.type, c.level, in); + + BlockCompressionCodec* legacy_codec = nullptr; + ASSERT_TRUE(get_block_compression_codec(c.type, &legacy_codec).ok()); + ASSERT_NE(legacy_codec, nullptr); + std::string out(in.size(), '\0'); + Slice out_slice(out); + ASSERT_TRUE(legacy_codec->decompress(Slice(compressed), &out_slice).ok()); + EXPECT_EQ(std::string(out_slice.data, out_slice.size), in); + } +} + +TEST_F(BlockCompressionTest, DifferentLevelsProduceDifferentOutput) { + // Moderately compressible data: enough redundancy that the level changes the + // encoder's choices, but not so uniform that every level saturates to the + // same output. + std::string in; + in.reserve(256 * 1024); + const char* words[] = {"apache", "doris", "compression", "column", "segment", + "rowset", "vectorized", "pipeline", "storage", "codec"}; + for (int i = 0; in.size() < 256 * 1024; ++i) { + in += words[(i * 7) % 10]; + in += words[(i * 13) % 10]; + in += std::to_string(i % 512); + in.push_back(' '); + } + + // ZSTD: a low level and the max level must yield different byte streams. + std::string zstd_low = compress_at_level(segment_v2::ZSTD, 1, in); + std::string zstd_high = compress_at_level(segment_v2::ZSTD, 22, in); + EXPECT_NE(zstd_low, zstd_high) + << "ZSTD level 1 and level 22 produced identical output (level likely ignored)"; + + // LZ4HC: likewise across its level range. + std::string lz4hc_low = compress_at_level(segment_v2::LZ4HC, 1, in); + std::string lz4hc_high = compress_at_level(segment_v2::LZ4HC, 12, in); + EXPECT_NE(lz4hc_low, lz4hc_high) + << "LZ4HC level 1 and level 12 produced identical output (level likely ignored)"; +} + +// A wide schema opens many column writers at the same codec+level. They must all +// share a single pooled codec instance (rather than one heavyweight context pool +// per column), and repeatedly acquiring them must not accumulate memory on the +// block-compression tracker once the shared context pool is warm. +TEST_F(BlockCompressionTest, WideSchemaSharesLeveledInstanceAndTrackerIsBalanced) { + clear_leveled_compression_codec_pool_for_test(); + + constexpr int kColumns = 128; + auto tracker = ExecEnv::GetInstance()->block_compression_mem_tracker(); + + struct Case { + segment_v2::CompressionTypePB type; + int level; + }; + for (auto c : {Case {segment_v2::ZSTD, 9}, Case {segment_v2::LZ4HC, 9}}) { + // Every column asking for the same (type, level) gets the same instance. + BlockCompressionCodec* first = nullptr; + ASSERT_TRUE(get_block_compression_codec(c.type, c.level, &first).ok()); + ASSERT_NE(first, nullptr); + for (int i = 0; i < kColumns; ++i) { + BlockCompressionCodec* codec = nullptr; + ASSERT_TRUE(get_block_compression_codec(c.type, c.level, &codec).ok()); + ASSERT_EQ(codec, first) << "wide schema must share one leveled codec instance"; + } + + // A different level yields a distinct instance (keyed by codec+level). + BlockCompressionCodec* other_level = nullptr; + ASSERT_TRUE(get_block_compression_codec(c.type, c.level + 1, &other_level).ok()); + ASSERT_NE(other_level, first); + + // Drive many serial compressions across the shared instance; after the + // pool is warm the tracker must not keep growing (contexts are reused, + // and reusable buffers are released back, not retained per column). + std::string in(64 * 1024, '\0'); + for (size_t i = 0; i < in.size(); ++i) in[i] = static_cast((i * 7) % 251); + + auto compress_once = [&]() { + faststring compressed; + ASSERT_TRUE(first->compress(Slice(in), &compressed).ok()); + }; + compress_once(); // warm up: lazily allocates the single shared context + int64_t warm = tracker->consumption(); + for (int i = 0; i < kColumns; ++i) { + compress_once(); + } + int64_t after = tracker->consumption(); + EXPECT_EQ(after, warm) << "shared pool must not grow per column on the compression tracker"; + } +} + } // namespace doris diff --git a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java index 7cac33db8089bf..52813e127ee71f 100644 --- a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java +++ b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java @@ -24,6 +24,7 @@ import org.apache.doris.common.DdlException; import org.apache.doris.common.util.SqlUtils; import org.apache.doris.persist.gson.GsonPostProcessable; +import org.apache.doris.thrift.TCompressionType; import com.google.common.base.Strings; import com.google.common.collect.Lists; @@ -184,6 +185,11 @@ public static Column generateBeforeValueColumn(Column column) { @SerializedName(value = "clusterKeyId") private int clusterKeyId = -1; + @SerializedName(value = "compressionType") + private TCompressionType compressionType = null; + @SerializedName(value = "compressionLevel") + private int compressionLevel = -1; + private boolean isCompoundKey = false; @SerializedName(value = "hasOnUpdateDefaultValue") @@ -389,6 +395,8 @@ public Column(Column column) { this.clusterKeyId = column.getClusterKeyId(); this.generatedColumnInfo = column.generatedColumnInfo; this.sessionVariables = column.sessionVariables; + this.compressionType = column.compressionType; + this.compressionLevel = column.compressionLevel; } public void createChildrenColumn(Type type, Column column) { @@ -828,6 +836,23 @@ public int getClusterKeyId() { return clusterKeyId; } + public void setCompression(TCompressionType compressionType, int compressionLevel) { + this.compressionType = compressionType; + this.compressionLevel = compressionLevel; + } + + public TCompressionType getCompressionType() { + return compressionType; + } + + public int getCompressionLevel() { + return compressionLevel; + } + + public boolean hasCompressionOverride() { + return compressionType != null; + } + public String toSql() { return toSql(false, false); } @@ -877,6 +902,12 @@ public String toSql(boolean isUniqueTable, boolean isCompatible) { if (hasOnUpdateDefaultValue) { sb.append(" ON UPDATE ").append(defaultValue).append(""); } + if (compressionType != null) { + sb.append(" COMPRESSION ").append(compressionType.name()); + if (compressionLevel > 0) { + sb.append("(").append(compressionLevel).append(")"); + } + } if (StringUtils.isNotBlank(comment)) { sb.append(" COMMENT \"").append(getComment(true)).append("\""); } @@ -892,7 +923,7 @@ public String toString() { public int hashCode() { return Objects.hash(name, getDataType(), getStrLen(), getPrecision(), getScale(), aggregationType, isAggregationTypeImplicit, isKey, isAllowNull, isAutoInc, defaultValue, getComment(), children, visible, - realDefaultValue, clusterKeyId); + realDefaultValue, clusterKeyId, compressionType, compressionLevel); } @Override @@ -918,7 +949,9 @@ public boolean equals(Object obj) { && visible == other.visible && Objects.equals(children, other.children) && Objects.equals(realDefaultValue, other.realDefaultValue) - && clusterKeyId == other.clusterKeyId; + && clusterKeyId == other.clusterKeyId + && Objects.equals(compressionType, other.compressionType) + && compressionLevel == other.compressionLevel; } // distribution column compare only care about attrs which affect data, diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java index 233cead1bdce43..380693f3aa92b3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java @@ -186,6 +186,9 @@ private boolean processAddColumn(AddColumnOp addColumnOp, OlapTable olapTable, Map colUniqueIdSupplierMap) throws DdlException { Column column = addColumnOp.getColumn(); + if (column.hasCompressionOverride()) { + throw new DdlException("Per-column compression is not supported for ADD COLUMN"); + } ColumnPosition columnPos = addColumnOp.getColPos(); String targetIndexName = addColumnOp.getRollupName(); checkIndexExists(olapTable, targetIndexName); @@ -264,6 +267,9 @@ public boolean processAddColumns(AddColumnsOp addColumnsOp, OlapTable olapTable, Map> indexSchemaMap, boolean ignoreSameColumn, Map colUniqueIdSupplierMap) throws DdlException { List columns = addColumnsOp.getColumns(); + if (columns.stream().anyMatch(Column::hasCompressionOverride)) { + throw new DdlException("Per-column compression is not supported for ADD COLUMN"); + } String targetIndexName = addColumnsOp.getRollupName(); checkIndexExists(olapTable, targetIndexName); @@ -1022,6 +1028,10 @@ private boolean processModifyColumn(ModifyColumnOp modifyColumnOp, OlapTable ola && modColumn.getDataType() == PrimitiveType.VARIANT) { lightSchemaChange = olapTable.getEnableLightSchemaChange(); } + if (col.hasCompressionOverride() || modColumn.hasCompressionOverride()) { + throw new DdlException( + "Per-column compression is not supported for MODIFY COLUMN"); + } if (col.isClusterKey()) { throw new DdlException("Can not modify cluster key column: " + col.getName()); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToProtobuf.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToProtobuf.java index d9c17012a64d28..3081c691407f25 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToProtobuf.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToProtobuf.java @@ -21,10 +21,12 @@ import org.apache.doris.common.DdlException; import org.apache.doris.proto.OlapFile; import org.apache.doris.proto.OlapFile.PatternTypePB; +import org.apache.doris.thrift.TCompressionType; import com.google.common.collect.Lists; import com.google.common.collect.Sets; import com.google.protobuf.ByteString; +import doris.segment_v2.SegmentV2; import java.util.List; import java.util.Set; @@ -43,6 +45,12 @@ public static OlapFile.ColumnPB toPb(Column column, Set bfColumns, List< builder.setUniqueId(column.getUniqueId()); builder.setType(column.getDataType().toThrift().name()); builder.setIsKey(column.isKey()); + if (column.hasCompressionOverride()) { + builder.setCompressionType(toCompressionTypePb(column.getCompressionType())); + if (column.getCompressionLevel() > 0) { + builder.setCompressionLevel(column.getCompressionLevel()); + } + } if (column.getFieldPatternType() != null) { switch (column.getFieldPatternType()) { case MATCH_NAME: @@ -124,6 +132,31 @@ public static OlapFile.ColumnPB toPb(Column column, Set bfColumns, List< return builder.build(); } + private static SegmentV2.CompressionTypePB toCompressionTypePb(TCompressionType compressionType) { + switch (compressionType) { + case UNKNOWN_COMPRESSION: + return SegmentV2.CompressionTypePB.UNKNOWN_COMPRESSION; + case DEFAULT_COMPRESSION: + return SegmentV2.CompressionTypePB.DEFAULT_COMPRESSION; + case NO_COMPRESSION: + return SegmentV2.CompressionTypePB.NO_COMPRESSION; + case SNAPPY: + return SegmentV2.CompressionTypePB.SNAPPY; + case LZ4: + return SegmentV2.CompressionTypePB.LZ4; + case LZ4F: + return SegmentV2.CompressionTypePB.LZ4F; + case ZLIB: + return SegmentV2.CompressionTypePB.ZLIB; + case ZSTD: + return SegmentV2.CompressionTypePB.ZSTD; + case LZ4HC: + return SegmentV2.CompressionTypePB.LZ4HC; + default: + throw new IllegalArgumentException("Unknown compression type: " + compressionType); + } + } + public static void addChildren(Column column, OlapFile.ColumnPB.Builder builder) throws DdlException { if (column.getChildren() != null) { List childrenColumns = column.getChildren(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToThrift.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToThrift.java index 9ed324713b69b5..8558659b287be6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToThrift.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColumnToThrift.java @@ -93,6 +93,12 @@ public static TColumn toThrift(Column column) { // And when creating `TAlterMaterializedViewParam`, the `defineExpr` is certainly analyzed. // If we need to use `defineExpr` and call defineExpr.treeToThrift(), // make sure it is analyzed, or NPE will thrown. + if (column.hasCompressionOverride()) { + tColumn.setCompressionType(column.getCompressionType().getValue()); + if (column.getCompressionLevel() > 0) { + tColumn.setCompressionLevel(column.getCompressionLevel()); + } + } return tColumn; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java index 51dacb59e83110..f6feb3ceb5fe94 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java @@ -1108,6 +1108,7 @@ import org.apache.doris.resource.workloadschedpolicy.WorkloadConditionMeta; import org.apache.doris.statistics.AnalysisInfo; import org.apache.doris.system.NodeType; +import org.apache.doris.thrift.TCompressionType; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; @@ -4180,11 +4181,11 @@ public ColumnDefinition visitColumnDef(ColumnDefContext ctx) { Optional defaultValue = Optional.empty(); Optional onUpdateDefaultValue = Optional.empty(); if (ctx.DEFAULT() != null) { - if (ctx.INTEGER_VALUE() != null) { + if (ctx.defaultInteger != null) { if (ctx.SUBTRACT() == null) { - defaultValue = Optional.of(new DefaultValue(ctx.INTEGER_VALUE().getText())); + defaultValue = Optional.of(new DefaultValue(ctx.defaultInteger.getText())); } else { - defaultValue = Optional.of(new DefaultValue("-" + ctx.INTEGER_VALUE().getText())); + defaultValue = Optional.of(new DefaultValue("-" + ctx.defaultInteger.getText())); } } else if (ctx.DECIMAL_VALUE() != null) { if (ctx.SUBTRACT() == null) { @@ -4250,8 +4251,44 @@ public ColumnDefinition visitColumnDef(ColumnDefContext ctx) { Optional desc = ctx.generatedExpr != null ? Optional.of(new GeneratedColumnDesc(ctx.generatedExpr.getText(), getExpression(ctx.generatedExpr))) : Optional.empty(); - return new ColumnDefinition(colName, colType, isKey, aggType, nullableType, autoIncInitValue, defaultValue, - onUpdateDefaultValue, comment, ctx.comment != null, true, desc); + ColumnDefinition columnDef = new ColumnDefinition(colName, colType, isKey, aggType, nullableType, + autoIncInitValue, defaultValue, onUpdateDefaultValue, comment, ctx.comment != null, true, desc); + if (ctx.compressionType != null) { + TCompressionType compressionType; + switch (ctx.compressionType.getType()) { + case DorisParser.NO_COMPRESSION: + compressionType = TCompressionType.NO_COMPRESSION; + break; + case DorisParser.LZ4: + compressionType = TCompressionType.LZ4; + break; + case DorisParser.LZ4F: + compressionType = TCompressionType.LZ4F; + break; + case DorisParser.LZ4HC: + compressionType = TCompressionType.LZ4HC; + break; + case DorisParser.ZLIB: + compressionType = TCompressionType.ZLIB; + break; + case DorisParser.ZSTD: + compressionType = TCompressionType.ZSTD; + break; + case DorisParser.SNAPPY: + compressionType = TCompressionType.SNAPPY; + break; + default: + throw new AnalysisException("Unsupported compression type: " + ctx.compressionType.getText()); + } + try { + int compressionLevel = ctx.compressionLevel == null + ? -1 : Integer.parseInt(ctx.compressionLevel.getText()); + columnDef.setCompression(compressionType, compressionLevel); + } catch (NumberFormatException | org.apache.doris.common.AnalysisException e) { + throw new AnalysisException(e.getMessage(), e); + } + } + return columnDef; } @Override diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java index 6c386489139492..d651763cb4e3af 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/ColumnDefinition.java @@ -43,6 +43,7 @@ import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.ConnectContextUtil; import org.apache.doris.qe.SessionVariable; +import org.apache.doris.thrift.TCompressionType; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; @@ -77,6 +78,9 @@ public class ColumnDefinition { private Set generatedColumnsThatReferToThis = new HashSet<>(); // if add hidden column, must set enableAddHiddenColumn true private boolean enableAddHiddenColumn = false; + // per-column generic compression override; null / -1 means "inherit codec default" + private TCompressionType compressionType = null; + private int compressionLevel = -1; public ColumnDefinition(String name, DataType type, boolean isKey, AggregateType aggType, boolean isNullable, Optional defaultValue, String comment) { @@ -171,6 +175,39 @@ public ColumnDefinition(String name, DataType type, boolean isNullable, String c this(name, type, false, null, isNullable, Optional.empty(), comment); } + /** Set and validate the per-column compression algorithm and optional level. */ + public void setCompression(TCompressionType type, int level) throws org.apache.doris.common.AnalysisException { + if (level != -1) { + switch (type) { + case ZSTD: + if (level < 1 || level > 22) { + throw new org.apache.doris.common.AnalysisException( + "ZSTD compression level must be in [1, 22], got " + level); + } + break; + case LZ4HC: + if (level < 1 || level > 12) { + throw new org.apache.doris.common.AnalysisException( + "LZ4HC compression level must be in [1, 12], got " + level); + } + break; + default: + throw new org.apache.doris.common.AnalysisException( + "compression level is only supported for ZSTD and LZ4HC, not " + type); + } + } + this.compressionType = type; + this.compressionLevel = level; + } + + public TCompressionType getCompressionType() { + return compressionType; + } + + public int getCompressionLevel() { + return compressionLevel; + } + public String getName() { return name; } @@ -271,7 +308,7 @@ private String toSql(String columnNameSql, boolean includeComment) { sb.append("AUTO_INCREMENT "); sb.append("("); sb.append(autoIncInitValue); - sb.append(")"); + sb.append(") "); } if (defaultValue.isPresent()) { @@ -292,6 +329,13 @@ private String toSql(String columnNameSql, boolean includeComment) { sb.append("DEFAULT ").append("NULL").append(" "); } } + if (compressionType != null) { + sb.append("COMPRESSION ").append(compressionType.name()); + if (compressionLevel > 0) { + sb.append("(").append(compressionLevel).append(")"); + } + sb.append(" "); + } if (includeComment) { sb.append("COMMENT ").append(SqlLiteralUtils.quoteStringLiteral(getComment())); } @@ -381,6 +425,28 @@ private void validateInternal(boolean isOlap, Set keysSet, Set c } catch (Exception e) { throw new AnalysisException(e.getMessage(), e); } + if (compressionType != null && !isOlap) { + throw new AnalysisException( + "COMPRESSION is only supported for OLAP table columns, column: " + name); + } + // A per-column codec only stamps the top-level column meta. For ARRAY/MAP/STRUCT/VARIANT + // the bulk of the bytes live in child/sub-columns that never receive the override, so the + // requested codec would be silently dropped for that data. Reject it instead. + if (compressionType != null + && (type.isArrayType() || type.isMapType() || type.isStructType() + || type.isVariantType())) { + throw new AnalysisException( + "COMPRESSION is not supported for complex type columns (ARRAY/MAP/STRUCT/VARIANT)," + + " column: " + name); + } + // AGG_STATE serializes to a function-dependent physical layout; some functions (e.g. + // ARRAY_AGG/MAP_AGG) dispatch to ARRAY/MAP segment writers whose child/auxiliary metas + // never receive the override codec+level, so the state would be stored with an + // inconsistent policy. Reject it like the other complex types above. + if (compressionType != null && type.isAggStateType()) { + throw new AnalysisException( + "COMPRESSION is not supported for AGG_STATE columns, column: " + name); + } type.validateDataType(); type = updateCharacterTypeLength(type); if (type.isArrayType()) { @@ -605,6 +671,9 @@ public Column translateToCatalogStyle() { .orElse(null) ); column.setAggregationTypeImplicit(aggTypeImplicit); + if (compressionType != null) { + column.setCompression(compressionType, compressionLevel); + } return column; } @@ -625,6 +694,9 @@ public Column translateToCatalogStyleForSchemaChange() { column.setNullableSpecified(nullableSpecified); column.setCommentSpecified(commentSpecified); column.setAggregationTypeImplicit(aggTypeImplicit); + if (compressionType != null) { + column.setCompression(compressionType, compressionLevel); + } return column; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java index 7c910231f6a16c..e9c007daf93fb4 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java @@ -170,6 +170,25 @@ private void expectException(String alterStmt, String expectedErrorMsg) { } } + @Test + public void testRejectAddColumnWithCompression() { + expectException("ALTER TABLE test.sc_dup ADD COLUMN compression_v1 INT COMPRESSION ZSTD(9)", + "Per-column compression is not supported for ADD COLUMN"); + } + + @Test + public void testRejectAddColumnsWithCompression() { + expectException("ALTER TABLE test.sc_dup ADD COLUMN " + + "(compression_v1 INT, compression_v2 INT COMPRESSION ZSTD(9))", + "Per-column compression is not supported for ADD COLUMN"); + } + + @Test + public void testRejectModifyColumnWithCompression() { + expectException("ALTER TABLE test.sc_dup MODIFY COLUMN error_msg VARCHAR(1024) COMPRESSION ZSTD(9)", + "Per-column compression is not supported for MODIFY COLUMN"); + } + @Test public void testWithRowBinlogSchemaChangeNoHistoricalValue() throws Exception { String tableName = "binlog_no_hist"; diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/ColumnCompressionSqlTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/ColumnCompressionSqlTest.java new file mode 100644 index 00000000000000..73ec239009d935 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/ColumnCompressionSqlTest.java @@ -0,0 +1,101 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.catalog; + +import org.apache.doris.proto.OlapFile; +import org.apache.doris.thrift.TColumn; +import org.apache.doris.thrift.TCompressionType; + +import doris.segment_v2.SegmentV2; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +public class ColumnCompressionSqlTest { + @Test + public void testToSqlRendersCompression() { + Column c = new Column("c1", Type.INT, true, null, false, "compressed column", true); + c.setCompression(TCompressionType.ZSTD, 9); + String sql = c.toSql(); + Assertions.assertTrue(sql.contains("COMPRESSION ZSTD(9)")); + Assertions.assertTrue(sql.indexOf("COMPRESSION") < sql.indexOf("COMMENT")); + } + + @Test + public void testToSqlNoCompressionWhenUnset() { + Column c = new Column("c1", Type.INT, true, null, false, "", true); + Assertions.assertFalse(c.toSql().contains("COMPRESSION")); + } + + @Test + public void testToThriftSetsCompression() { + Column c = new Column("c1", Type.INT, true, null, false, "", true); + c.setCompression(TCompressionType.ZSTD, 9); + TColumn t = ColumnToThrift.toThrift(c); + Assertions.assertTrue(t.isSetCompressionType()); + Assertions.assertEquals(TCompressionType.ZSTD.getValue(), t.getCompressionType()); + Assertions.assertEquals(9, t.getCompressionLevel()); + } + + @Test + public void testToThriftNoCompressionWhenUnset() { + Column c = new Column("c1", Type.INT, true, null, false, "", true); + TColumn t = ColumnToThrift.toThrift(c); + Assertions.assertFalse(t.isSetCompressionType()); + } + + @Test + public void testToProtobufSetsCompression() throws Exception { + Column c = new Column("c1", Type.INT, true, null, false, "", true); + c.setCompression(TCompressionType.ZSTD, 9); + OlapFile.ColumnPB columnPb = ColumnToProtobuf.toPb( + c, Collections.emptySet(), Collections.emptyList()); + Assertions.assertTrue(columnPb.hasCompressionType()); + Assertions.assertEquals(SegmentV2.CompressionTypePB.ZSTD, columnPb.getCompressionType()); + Assertions.assertEquals(9, columnPb.getCompressionLevel()); + } + + @Test + public void testToProtobufNoCompressionWhenUnset() throws Exception { + Column c = new Column("c1", Type.INT, true, null, false, "", true); + OlapFile.ColumnPB columnPb = ColumnToProtobuf.toPb( + c, Collections.emptySet(), Collections.emptyList()); + Assertions.assertFalse(columnPb.hasCompressionType()); + Assertions.assertFalse(columnPb.hasCompressionLevel()); + } + + @Test + public void testEqualsIncludesCompression() { + Column base = new Column("c1", Type.INT, true, null, false, "", true); + base.setCompression(TCompressionType.ZSTD, 9); + + Column same = new Column("c1", Type.INT, true, null, false, "", true); + same.setCompression(TCompressionType.ZSTD, 9); + Assertions.assertEquals(base, same); + Assertions.assertEquals(base.hashCode(), same.hashCode()); + + Column differentType = new Column("c1", Type.INT, true, null, false, "", true); + differentType.setCompression(TCompressionType.LZ4HC, 9); + Assertions.assertNotEquals(base, differentType); + + Column differentLevel = new Column("c1", Type.INT, true, null, false, "", true); + differentLevel.setCompression(TCompressionType.ZSTD, 10); + Assertions.assertNotEquals(base, differentLevel); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ColumnCompressionTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ColumnCompressionTest.java new file mode 100644 index 00000000000000..e7c2c0643b0f1d --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ColumnCompressionTest.java @@ -0,0 +1,163 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans; + +import org.apache.doris.catalog.AggregateType; +import org.apache.doris.catalog.KeysType; +import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.Config; +import org.apache.doris.nereids.exceptions.ParseException; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.commands.info.ColumnDefinition; +import org.apache.doris.nereids.types.AggStateType; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.IntegerType; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.thrift.TCompressionType; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Sets; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Optional; + +public class ColumnCompressionTest { + @Test + public void testCompressionSyntax() { + ConnectContext connectContext = new ConnectContext(); + connectContext.setDatabase("test"); + connectContext.setThreadLocalInfo(); + try { + NereidsParser parser = new NereidsParser(); + Assertions.assertDoesNotThrow(() -> parser.parseSingle("CREATE TABLE test_compression (" + + "k INT, v VARCHAR(10) COMPRESSION ZSTD(9) COMMENT 'value') " + + "DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')")); + Assertions.assertThrows(ParseException.class, () -> parser.parseSingle("CREATE TABLE test_compression (" + + "k INT, v VARCHAR(10) COMPRESSION 'zstd:9') " + + "DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')")); + Assertions.assertThrows(ParseException.class, () -> parser.parseSingle("CREATE TABLE test_compression (" + + "k INT, v VARCHAR(10) COMMENT 'value' COMPRESSION ZSTD(9)) " + + "DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 " + + "PROPERTIES ('replication_num' = '1')")); + } finally { + ConnectContext.remove(); + } + } + + @Test + public void testLevelOnLz4Rejected() { + Assertions.assertThrows(AnalysisException.class, + () -> newColumn().setCompression(TCompressionType.LZ4, 5)); + } + + @Test + public void testZstdLevelOutOfRange() { + Assertions.assertThrows(AnalysisException.class, + () -> newColumn().setCompression(TCompressionType.ZSTD, 99)); + } + + @Test + public void testLz4hcLevelRange() throws Exception { + ColumnDefinition column = newColumn(); + column.setCompression(TCompressionType.LZ4HC, 12); + Assertions.assertEquals(TCompressionType.LZ4HC, column.getCompressionType()); + Assertions.assertEquals(12, column.getCompressionLevel()); + Assertions.assertThrows(AnalysisException.class, + () -> column.setCompression(TCompressionType.LZ4HC, 13)); + } + + @Test + public void testCompressionRejectedOnNonOlap() throws Exception { + ColumnDefinition col = new ColumnDefinition("col1", IntegerType.INSTANCE, false, AggregateType.NONE, + true, Optional.empty(), ""); + col.setCompression(TCompressionType.ZSTD, 9); + Assertions.assertThrows(org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> col.validate(false, Sets.newHashSet(), Sets.newHashSet(), false, KeysType.DUP_KEYS)); + } + + @Test + public void testCompressionAllowedOnOlap() throws Exception { + ColumnDefinition col = new ColumnDefinition("col1", IntegerType.INSTANCE, false, AggregateType.NONE, + true, Optional.empty(), ""); + col.setCompression(TCompressionType.ZSTD, 9); + Assertions.assertDoesNotThrow( + () -> col.validate(true, Sets.newHashSet(), Sets.newHashSet(), false, KeysType.DUP_KEYS)); + } + + @Test + public void testCompressionAllowedInCloudMode() throws Exception { + String previousCloudUniqueId = Config.cloud_unique_id; + Config.cloud_unique_id = "column-compression-test"; + try { + ColumnDefinition col = new ColumnDefinition("col1", IntegerType.INSTANCE, false, AggregateType.NONE, + true, Optional.empty(), ""); + col.setCompression(TCompressionType.ZSTD, 9); + Assertions.assertDoesNotThrow( + () -> col.validate(true, Sets.newHashSet(), Sets.newHashSet(), false, KeysType.DUP_KEYS)); + } finally { + Config.cloud_unique_id = previousCloudUniqueId; + } + } + + @Test + public void testCompressionRejectedOnComplexType() throws Exception { + ColumnDefinition col = new ColumnDefinition("col1", ArrayType.of(IntegerType.INSTANCE), false, + AggregateType.NONE, true, Optional.empty(), ""); + col.setCompression(TCompressionType.ZSTD, 9); + Assertions.assertThrows(org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> col.validate(true, Sets.newHashSet(), Sets.newHashSet(), false, KeysType.DUP_KEYS)); + } + + @Test + public void testCompressionRejectedOnAggState() throws Exception { + // AGG_STATE serializes to a function-dependent physical layout (ARRAY/MAP writers for + // array_agg/map_agg) whose child metas never receive the override, so COMPRESSION must be + // rejected for all AGG_STATE columns. + AggStateType aggState = new AggStateType("array_agg", + ImmutableList.of(IntegerType.INSTANCE), ImmutableList.of(true), true); + ColumnDefinition col = new ColumnDefinition("col1", aggState, false, + AggregateType.GENERIC, true, Optional.empty(), ""); + col.setCompression(TCompressionType.ZSTD, 9); + Assertions.assertThrows(org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> col.validate(true, Sets.newHashSet(), Sets.newHashSet(), false, KeysType.AGG_KEYS)); + } + + @Test + public void testToSqlRendersCompressionClause() throws Exception { + ColumnDefinition withLevel = new ColumnDefinition("c1", IntegerType.INSTANCE, false, + AggregateType.NONE, true, Optional.empty(), ""); + withLevel.setCompression(TCompressionType.ZSTD, 9); + Assertions.assertTrue(withLevel.toSql("`c1`").contains("COMPRESSION ZSTD(9)"), + "toSql should render COMPRESSION with level, got: " + withLevel.toSql("`c1`")); + + ColumnDefinition noLevel = new ColumnDefinition("c2", IntegerType.INSTANCE, false, + AggregateType.NONE, true, Optional.empty(), ""); + noLevel.setCompression(TCompressionType.ZSTD, -1); + String sql = noLevel.toSql("`c2`"); + Assertions.assertTrue(sql.contains("COMPRESSION ZSTD") && !sql.contains("ZSTD("), + "toSql should render COMPRESSION without level, got: " + noLevel.toSql("`c2`")); + } + + private ColumnDefinition newColumn() { + return new ColumnDefinition("col1", IntegerType.INSTANCE, false, AggregateType.NONE, + true, Optional.empty(), ""); + } +} diff --git a/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4 b/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4 index 446685892295e4..b096025ac4904a 100644 --- a/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4 +++ b/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4 @@ -146,6 +146,7 @@ COMMITTED: 'COMMITTED'; COMPACT: 'COMPACT'; COMPLETE: 'COMPLETE'; COMPRESS_TYPE: 'COMPRESS_TYPE'; +COMPRESSION: 'COMPRESSION'; COMPUTE: 'COMPUTE'; CONDITIONS: 'CONDITIONS'; CONFIG: 'CONFIG'; @@ -354,6 +355,9 @@ LOCATION: 'LOCATION'; LOCK: 'LOCK'; LOGICAL: 'LOGICAL'; LOW_PRIORITY: 'LOW_PRIORITY'; +LZ4: 'LZ4'; +LZ4F: 'LZ4F'; +LZ4HC: 'LZ4HC'; MANUAL: 'MANUAL'; MAP: 'MAP'; MAPPING: 'MAPPING'; @@ -396,6 +400,7 @@ NGRAM_BF: 'NGRAM_BF'; ANN: 'ANN'; NO: 'NO'; NONE: 'NONE'; +NO_COMPRESSION: 'NO_COMPRESSION'; NO_USE_MV: 'NO_USE_MV'; NON_NULLABLE: 'NON_NULLABLE'; NORMALIZER: 'NORMALIZER'; @@ -531,6 +536,7 @@ SKEW: 'SKEW'; SMALLINT: 'SMALLINT'; SNAPSHOT: 'SNAPSHOT'; SNAPSHOTS: 'SNAPSHOTS'; +SNAPPY: 'SNAPPY'; SONAME: 'SONAME'; SPLIT: 'SPLIT'; SQL: 'SQL'; @@ -631,6 +637,8 @@ WRITE: 'WRITE'; XOR: 'XOR'; YEAR: 'YEAR'; YEAR_MONTH: 'YEAR_MONTH'; +ZLIB: 'ZLIB'; +ZSTD: 'ZSTD'; //--DORIS-KEYWORD-LIST-END //============================ // End of the keywords list diff --git a/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 b/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 index b300b6185c8730..ef4d1eefee949a 100644 --- a/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 +++ b/fe/fe-sql-parser/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 @@ -1548,9 +1548,11 @@ columnDef ((GENERATED ALWAYS)? AS LEFT_PAREN generatedExpr=expression RIGHT_PAREN)? ((NOT)? nullable=NULL)? (AUTO_INCREMENT (LEFT_PAREN autoIncInitValue=number RIGHT_PAREN)?)? - (DEFAULT (nullValue=NULL | SUBTRACT? INTEGER_VALUE | SUBTRACT? DECIMAL_VALUE | PI | E | BITMAP_EMPTY | stringValue=STRING_LITERAL + (DEFAULT (nullValue=NULL | SUBTRACT? defaultInteger=INTEGER_VALUE | SUBTRACT? DECIMAL_VALUE | PI | E | BITMAP_EMPTY | stringValue=STRING_LITERAL | CURRENT_DATE | defaultTimestamp=CURRENT_TIMESTAMP (LEFT_PAREN defaultValuePrecision=number RIGHT_PAREN)?))? (ON UPDATE CURRENT_TIMESTAMP (LEFT_PAREN onUpdateValuePrecision=number RIGHT_PAREN)?)? + (COMPRESSION compressionType=(NO_COMPRESSION | LZ4 | LZ4F | LZ4HC | ZLIB | ZSTD | SNAPPY) + (LEFT_PAREN compressionLevel=INTEGER_VALUE RIGHT_PAREN)?)? (COMMENT comment=STRING_LITERAL)? ; @@ -2065,6 +2067,7 @@ nonReserved | COMMITTED | COMPACT | COMPLETE + | COMPRESSION | COMPRESS_TYPE | COMPUTE | CONDITIONS @@ -2203,6 +2206,9 @@ nonReserved | LOCATION | LOCK | LOGICAL + | LZ4 + | LZ4F + | LZ4HC | MANUAL | MAP | MAPPING @@ -2239,6 +2245,7 @@ nonReserved | NGRAM_BF | NO | NONE + | NO_COMPRESSION | NON_NULLABLE | NORMALIZER | NULLS @@ -2334,6 +2341,7 @@ nonReserved | SKEW | SNAPSHOT | SNAPSHOTS + | SNAPPY | SONAME | SPLIT | SQL @@ -2396,5 +2404,7 @@ nonReserved | WORK | YEAR | YEAR_MONTH + | ZLIB + | ZSTD //--DEFAULT-NON-RESERVED-END ; diff --git a/gensrc/proto/olap_file.proto b/gensrc/proto/olap_file.proto index 7f1c62a446100b..83dcbbeeacc357 100644 --- a/gensrc/proto/olap_file.proto +++ b/gensrc/proto/olap_file.proto @@ -422,6 +422,9 @@ message ColumnPB { // Number of buckets used to store doc map in variant doc mode. optional int32 variant_doc_hash_shard_count = 33 [default = 64]; optional bool variant_enable_nested_group = 34 [default = false]; + // per-column generic compression override; UNKNOWN_COMPRESSION => inherit table-level compression + optional segment_v2.CompressionTypePB compression_type = 35 [default = UNKNOWN_COMPRESSION]; + optional int32 compression_level = 36; } // Dictionary of Schema info, to reduce TabletSchemaCloudPB fdb kv size diff --git a/gensrc/proto/segment_v2.proto b/gensrc/proto/segment_v2.proto index 6fb6fa5fe037a5..c16a1cb098b33d 100644 --- a/gensrc/proto/segment_v2.proto +++ b/gensrc/proto/segment_v2.proto @@ -232,6 +232,8 @@ message ColumnMetaPB { optional uint64 uncompressed_data_bytes = 25; optional uint64 raw_data_bytes = 26; optional bool variant_enable_doc_mode = 27 [default = false]; + // per-column generic compression level (e.g. ZSTD/LZ4HC). Absent => codec default level. + optional int32 compression_level = 28; } // External column meta entry describing one top-level column's externalized diff --git a/gensrc/thrift/Descriptors.thrift b/gensrc/thrift/Descriptors.thrift index 46c84d2826632e..12f7be781398b5 100644 --- a/gensrc/thrift/Descriptors.thrift +++ b/gensrc/thrift/Descriptors.thrift @@ -100,6 +100,11 @@ struct TColumn { 27: optional i64 variant_doc_materialization_min_rows 28: optional i32 variant_doc_hash_shard_count 29: optional bool variant_enable_nested_group + // per-column generic compression override; i32 holds TCompressionType enum value + // (cannot reference TCompressionType directly: it lives in AgentService.thrift which + // already includes Descriptors.thrift, so a direct reference would create an include cycle) + 30: optional i32 compression_type + 31: optional i32 compression_level } struct TSlotDescriptor { diff --git a/regression-test/data/ddl_p0/test_column_compression.out b/regression-test/data/ddl_p0/test_column_compression.out new file mode 100644 index 00000000000000..d26434b49fdf5e --- /dev/null +++ b/regression-test/data/ddl_p0/test_column_compression.out @@ -0,0 +1,15 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !select_main -- +1 a hello world hello world +2 b the quick brown fox jumps +3 c lorem ipsum dolor sit amet + +-- !select_after_modify -- +1 hello world hello world +2 the quick brown fox jumps +3 lorem ipsum dolor sit amet + +-- !select_scalar -- +1 100 +2 200 + diff --git a/regression-test/suites/ddl_p0/test_column_compression.groovy b/regression-test/suites/ddl_p0/test_column_compression.groovy new file mode 100644 index 00000000000000..57ef2ae1df6ca4 --- /dev/null +++ b/regression-test/suites/ddl_p0/test_column_compression.groovy @@ -0,0 +1,92 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +suite("test_column_compression") { + sql "DROP TABLE IF EXISTS test_column_compression_tbl" + sql "DROP TABLE IF EXISTS test_column_compression_bad" + sql "DROP TABLE IF EXISTS test_column_compression_bad_complex" + sql "DROP TABLE IF EXISTS test_column_compression_scalar" + + // CREATE TABLE with a per-column codec on the heavy column, table default LZ4F + sql """ + CREATE TABLE test_column_compression_tbl ( + k INT, + v_default VARCHAR(64), + v_heavy VARCHAR(64) COMPRESSION ZSTD(9) + ) + DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num" = "1", "compression" = "lz4f") + """ + + // insert + read back correctness + sql """ INSERT INTO test_column_compression_tbl VALUES + (1, 'a', 'hello world hello world'), + (2, 'b', 'the quick brown fox jumps'), + (3, 'c', 'lorem ipsum dolor sit amet') """ + sql "sync" + order_qt_select_main "SELECT k, v_default, v_heavy FROM test_column_compression_tbl ORDER BY k" + order_qt_select_after_modify "SELECT k, v_heavy FROM test_column_compression_tbl ORDER BY k" + + test { + sql "ALTER TABLE test_column_compression_tbl ADD COLUMN v_added VARCHAR(64) COMPRESSION ZSTD(5)" + exception "Per-column compression is not supported for ADD COLUMN" + } + + test { + sql "ALTER TABLE test_column_compression_tbl MODIFY COLUMN v_heavy VARCHAR(64) COMPRESSION ZSTD(12)" + exception "Per-column compression is not supported for MODIFY COLUMN" + } + + // invalid: level on lz4 must be rejected at DDL time + test { + sql """ + CREATE TABLE test_column_compression_bad ( + k INT, + v VARCHAR(64) COMPRESSION LZ4(5) + ) DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + exception "level" + } + + // invalid: COMPRESSION on a complex-type column must be rejected -- the override only + // stamps the top-level column meta, so the element data would silently ignore it. + test { + sql """ + CREATE TABLE test_column_compression_bad_complex ( + k INT, + v ARRAY COMPRESSION ZSTD(9) + ) DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + exception "complex type" + } + + sql """ + CREATE TABLE test_column_compression_scalar ( + k INT, + v BIGINT COMPRESSION ZSTD(5) + ) + DUPLICATE KEY(k) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + sql "INSERT INTO test_column_compression_scalar VALUES (1, 100), (2, 200)" + sql "sync" + order_qt_select_scalar "SELECT k, v FROM test_column_compression_scalar ORDER BY k" +}