Skip to content

feat(compression): support per-column compression for non-cloud #66169

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zwy991114 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat(compression): support per-column compression for non-cloud #66169

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions be/src/storage/rowset/segment_creator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,7 @@ Status SegmentFlusher::_create_segment_writer(std::unique_ptr<segment_v2::Segmen
writer_options.mow_ctx = _context.mow_context;
if (no_compression) {
writer_options.compression_type = NO_COMPRESSION;
writer_options.suppress_compression = true;
}

if (_context.write_binlog_opt().enable) {
Expand Down Expand Up @@ -187,6 +188,7 @@ Status SegmentFlusher::_create_segment_writer(
writer_options.mow_ctx = _context.mow_context;
if (no_compression) {
writer_options.compression_type = NO_COMPRESSION;
writer_options.suppress_compression = true;
}

writer = std::make_unique<segment_v2::VerticalSegmentWriter>(
Expand Down
5 changes: 4 additions & 1 deletion be/src/storage/segment/column_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_owned, &_compress_codec));

PageBuilder* page_builder = nullptr;

Expand Down
1 change: 1 addition & 0 deletions be/src/storage/segment/column_writer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,7 @@ class ScalarColumnWriter : public ColumnWriter {
std::vector<std::unique_ptr<Page>> _pages;
ordinal_t _first_rowid = 0;

std::unique_ptr<BlockCompressionCodec> _compress_codec_owned;
BlockCompressionCodec* _compress_codec;

std::unique_ptr<OrdinalIndexWriter> _ordinal_index_builder;
Expand Down
12 changes: 11 additions & 1 deletion be/src/storage/segment/segment_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,17 @@ 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);
// A per-column codec overrides the table-level default, but never when compression is
// being suppressed for this segment (small-segment NO_COMPRESSION), otherwise the
// sub-threshold optimization would be defeated for per-column-compressed columns.
if (column.has_compression() && !_opts.suppress_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());
Expand Down
4 changes: 4 additions & 0 deletions be/src/storage/segment/segment_writer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,10 @@ struct SegmentWriterOptions {
uint32_t max_rows_per_segment = UINT32_MAX;
bool enable_unique_key_merge_on_write = false;
CompressionTypePB compression_type = UNKNOWN_COMPRESSION;
// Set for the sub-threshold small-segment optimization, which forces NO_COMPRESSION for
// the whole segment. When true, per-column codec overrides are suppressed as well; a
// table-level NO_COMPRESSION default must NOT set this, so explicit per-column codecs win.
bool suppress_compression = false;

RowsetWriterContext* rowset_ctx = nullptr;
DataWriteType write_type = DataWriteType::TYPE_DEFAULT;
Expand Down
12 changes: 11 additions & 1 deletion be/src/storage/segment/vertical_segment_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,17 @@ 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);
// A per-column codec overrides the table-level default, but never when compression is
// being suppressed for this segment (small-segment NO_COMPRESSION), otherwise the
// sub-threshold optimization would be defeated for per-column-compressed columns.
if (column.has_compression() && !_opts.suppress_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());
Expand Down
5 changes: 5 additions & 0 deletions be/src/storage/segment/vertical_segment_writer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,10 @@ struct VerticalSegmentWriterOptions {
uint32_t num_rows_per_block = 1024;
bool enable_unique_key_merge_on_write = false;
CompressionTypePB compression_type = UNKNOWN_COMPRESSION;
// Set for the sub-threshold small-segment optimization, which forces NO_COMPRESSION for
// the whole segment. When true, per-column codec overrides are suppressed as well; a
// table-level NO_COMPRESSION default must NOT set this, so explicit per-column codecs win.
bool suppress_compression = false;

RowsetWriterContext* rowset_ctx = nullptr;
DataWriteType write_type = DataWriteType::TYPE_DEFAULT;
Expand Down Expand Up @@ -199,6 +203,7 @@ class VerticalSegmentWriter {

private:
friend class ::doris::BlockAggregator;
friend class TestVerticalSegmentWriter;
uint32_t _segment_id;
TabletSchemaSPtr _tablet_schema;
BaseTabletSPtr _tablet;
Expand Down
28 changes: 28 additions & 0 deletions be/src/storage/tablet/tablet_meta.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
15 changes: 15 additions & 0 deletions be/src/storage/tablet/tablet_schema.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -555,6 +555,15 @@ void TabletColumn::init_from_pb(const ColumnPB& column) {
if (column.has_pattern_type()) {
_pattern_type = column.pattern_type();
}
if (column.has_compression_type()) {
_has_compression = true;
_compression = column.compression_type();
_compression_level = column.has_compression_level() ? column.compression_level() : 0;
} else {
_has_compression = false;
_compression = segment_v2::UNKNOWN_COMPRESSION;
_compression_level = 0;
}
}

TabletColumn TabletColumn::create_materialized_variant_column(const std::string& root,
Expand Down Expand Up @@ -641,6 +650,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) {
Expand Down
7 changes: 7 additions & 0 deletions be/src/storage/tablet/tablet_schema.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 _has_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; }
Expand Down Expand Up @@ -302,6 +305,10 @@ class TabletColumn : public MetadataAdder<TabletColumn> {
bool _has_default_value = false;
std::string _default_value;

bool _has_compression = false;
segment_v2::CompressionTypePB _compression = segment_v2::UNKNOWN_COMPRESSION;
int _compression_level = 0;

bool _is_decimal = false;
int32_t _precision = -1;
int32_t _frac = -1;
Expand Down
42 changes: 40 additions & 2 deletions be/src/util/block_compression.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -580,6 +580,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());
Expand Down Expand Up @@ -663,6 +665,10 @@ class Lz4HCBlockCompression : public BlockCompressionCodec {
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<int>(_compression_level));
out = std::move(localCtx);
return Status::OK();
}
Expand Down Expand Up @@ -1077,6 +1083,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());
Expand Down Expand Up @@ -1123,9 +1131,8 @@ 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);
_compression_level);
if (ZSTD_isError(ret)) {
return Status::InvalidArgument("ZSTD_CCtx_setParameter compression level error: {}",
ZSTD_getErrorString(ZSTD_getErrorCode(ret)));
Expand Down Expand Up @@ -1262,6 +1269,7 @@ class ZstdBlockCompression : public BlockCompressionCodec {
}

private:
int _compression_level = ZSTD_CLEVEL_DEFAULT;
mutable std::mutex _ctx_c_mutex;
mutable std::vector<std::unique_ptr<CContext>> _ctx_c_pool;

Expand Down Expand Up @@ -1615,6 +1623,36 @@ Status get_block_compression_codec(segment_v2::CompressionTypePB type,
return Status::OK();
}

Status get_block_compression_codec(segment_v2::CompressionTypePB type, int level,
std::unique_ptr<BlockCompressionCodec>* codec_owned,
BlockCompressionCodec** codec) {
codec_owned->reset();
// 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: {
auto instance = std::make_unique<ZstdBlockCompression>(level);

@github-actions github-actions Bot Jul 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Avoid a private context pool per levelled column. ScalarColumnWriter owns this instance, and after the first page its ZstdBlockCompression/Lz4HCBlockCompression retains a native context plus reusable buffer until the whole segment clears its column writers. Wide schemas (and multiple open segments) therefore retain one heavyweight workspace per levelled column even though compression is serial; the old singleton pool reused contexts according to actual concurrency, and this memory is not included in the segment-size estimate. Please share/reset pools by codec+level (or release rather than pool per-column contexts) and add a wide-schema tracker test.

RETURN_IF_ERROR(instance->init());
*codec = instance.get();
*codec_owned = std::move(instance);
break;
}
case segment_v2::CompressionTypePB::LZ4HC: {
auto instance = std::make_unique<Lz4HCBlockCompression>(level);

@github-actions github-actions Bot Jul 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Balance native context allocation and destruction under the same tracker for these new short-lived instances. Both _acquire_compression_ctx() paths allocate the LZ4/ZSTD native context (and ZSTD later allocates workspace) under the current load/schema-change tracker, while the context/codec destructors switch to block_compression_mem_tracker() before freeing it. MemTrackerLimiter explicitly requires allocation and free to hit the same tracker; owned per-column codecs now repeat this mismatch at every segment teardown. Put creation/lazy allocation under the compression tracker as well, or keep the entire owned lifetime consistently task-local, and test tracker balance.

RETURN_IF_ERROR(instance->init());
*codec = instance.get();
*codec_owned = std::move(instance);
break;
}
default:
// types without a tunable level ignore it and use the singleton
return get_block_compression_codec(type, codec);
}
return Status::OK();
}

// this can only be used in hive text write
Status get_block_compression_codec(TFileCompressType::type type, BlockCompressionCodec** codec) {
switch (type) {
Expand Down
7 changes: 7 additions & 0 deletions be/src/util/block_compression.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,13 @@ 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 heap instance in `*codec_owned` (caller owns it). Otherwise `*codec_owned`
// is left empty and `*codec` points to a process-wide singleton (do not delete).
Status get_block_compression_codec(segment_v2::CompressionTypePB type, int level,
std::unique_ptr<BlockCompressionCodec>* codec_owned,
BlockCompressionCodec** codec);

Status get_block_compression_codec(tparquet::CompressionCodec::type parquet_codec,
BlockCompressionCodec** codec);

Expand Down
Loading
Loading