From f757a5d180f67bcd63b008023ff1f79033ddc37e Mon Sep 17 00:00:00 2001 From: TsukiokaKogane Date: Mon, 13 Jul 2026 22:20:14 +0800 Subject: [PATCH 1/2] [refactor](be) Unify read-time column resolution fix be ut fix core --- be/src/storage/segment/column_reader.cpp | 4 - be/src/storage/segment/column_reader.h | 21 +- .../storage/segment/column_reader_cache.cpp | 4 +- be/src/storage/segment/column_reader_cache.h | 6 +- .../segment/row_binlog_segment_writer.cpp | 6 +- be/src/storage/segment/segment.cpp | 171 ++++++------- be/src/storage/segment/segment.h | 25 +- be/src/storage/segment/segment_iterator.cpp | 104 ++------ be/src/storage/segment/segment_iterator.h | 3 +- .../predicate/block_column_predicate_test.cpp | 68 ----- .../segment/constant_column_iterator_test.cpp | 13 +- .../segment_iterator_expr_zonemap_test.cpp | 233 +++++++++++++++--- 12 files changed, 331 insertions(+), 327 deletions(-) diff --git a/be/src/storage/segment/column_reader.cpp b/be/src/storage/segment/column_reader.cpp index 25886d7515d2b8..d0ac73d2f2f535 100644 --- a/be/src/storage/segment/column_reader.cpp +++ b/be/src/storage/segment/column_reader.cpp @@ -256,10 +256,6 @@ bool ColumnReader::is_compaction_reader_type(ReaderType type) { Status ColumnReader::create(const ColumnReaderOptions& opts, const ColumnMetaPB& meta, uint64_t num_rows, const io::FileReaderSPtr& file_reader, std::shared_ptr* reader) { - if (opts.const_value.has_value()) { - *reader = std::make_shared(*opts.const_value); - return Status::OK(); - } if (is_scalar_type((FieldType)meta.type())) { std::shared_ptr reader_local( new ColumnReader(opts, meta, num_rows, file_reader)); diff --git a/be/src/storage/segment/column_reader.h b/be/src/storage/segment/column_reader.h index 777f8eddf0247b..d8812dd46aaf2c 100644 --- a/be/src/storage/segment/column_reader.h +++ b/be/src/storage/segment/column_reader.h @@ -26,7 +26,6 @@ #include // for uint32_t #include #include // for unique_ptr -#include #include #include #include @@ -95,12 +94,6 @@ struct ColumnReaderOptions { int be_exec_version = -1; TabletSchemaSPtr tablet_schema = nullptr; - - // When set, ColumnReader::create returns a ConstantColumnReader carrying this value instead - // of reading on-disk data. Used for read-time-filled constant columns (e.g. - // __DORIS_COMMIT_TSO_COL__) on a single-version segment, whose on-disk value is only a - // placeholder. The value is constant within a segment, so the resulting reader is cacheable. - std::optional const_value = std::nullopt; }; struct ColumnIteratorOptions { @@ -1020,10 +1013,8 @@ class DefaultValueColumnIterator : public ColumnIterator { ordinal_t _current_rowid = 0; }; -// Produces a column whose every row is the same constant Field value. -// Used for read-time-filled constant hidden columns (e.g. __DORIS_COMMIT_TSO_COL__), -// where the on-disk value is only a placeholder and the real value comes from the read -// context (StorageReadOptions). +// Produces a column whose every row is the same request-scoped constant Field value. +// It is created directly by Segment and never stored in ColumnReaderCache. class ConstantColumnIterator : public ColumnIterator { public: ConstantColumnIterator() = delete; @@ -1078,11 +1069,9 @@ class ConstantColumnIterator : public ColumnIterator { ordinal_t _current_rowid = 0; }; -// A ColumnReader that represents a single constant value for the whole segment instead of reading -// on-disk data. Used for read-time-filled constant columns (e.g. __DORIS_COMMIT_TSO_COL__) on a -// single-version segment, whose on-disk zonemap only reflects the placeholder. It advertises a -// single-value [v, v] zonemap so segment-level pruning matches against the real value, and produces -// a ConstantColumnIterator for data reads. +// A request-scoped ColumnReader that represents one constant value for the whole segment instead +// of reading on-disk data. It is never stored in ColumnReaderCache. It advertises a single-value +// [v, v] zonemap and produces a ConstantColumnIterator for data reads. class ConstantColumnReader : public ColumnReader { public: explicit ConstantColumnReader(Field value) : _value(std::move(value)) {} diff --git a/be/src/storage/segment/column_reader_cache.cpp b/be/src/storage/segment/column_reader_cache.cpp index 4b9d0cbe107dba..6b96f5d7b98629 100644 --- a/be/src/storage/segment/column_reader_cache.cpp +++ b/be/src/storage/segment/column_reader_cache.cpp @@ -96,8 +96,7 @@ std::map> ColumnReaderCache::get_availabl Status ColumnReaderCache::get_column_reader(int32_t col_uid, std::shared_ptr* column_reader, OlapReaderStatistics* stats, - const io::IOContext* source_io_ctx, - std::optional const_value) { + const io::IOContext* source_io_ctx) { // Attempt to find in cache if (auto cached = _lookup({col_uid, {}})) { *column_reader = cached; @@ -123,7 +122,6 @@ Status ColumnReaderCache::get_column_reader(int32_t col_uid, ColumnReaderOptions opts {.kept_in_memory = _tablet_schema->is_in_memory(), .be_exec_version = _be_exec_version, .tablet_schema = _tablet_schema}; - opts.const_value = std::move(const_value); std::shared_ptr reader; if ((FieldType)meta.type() == FieldType::OLAP_FIELD_TYPE_VARIANT) { diff --git a/be/src/storage/segment/column_reader_cache.h b/be/src/storage/segment/column_reader_cache.h index 2282b56c32ffff..91398b9c85485a 100644 --- a/be/src/storage/segment/column_reader_cache.h +++ b/be/src/storage/segment/column_reader_cache.h @@ -16,10 +16,7 @@ // under the License. #pragma once -#include - #include "agent/be_exec_version_manager.h" -#include "core/field.h" #include "io/fs/file_reader.h" #include "storage/segment/stream_reader.h" #include "storage/tablet/tablet_fwd.h" @@ -65,8 +62,7 @@ class ColumnReaderCache { // Get column reader by column unique id Status get_column_reader(int32_t col_uid, std::shared_ptr* column_reader, - OlapReaderStatistics* stats, const io::IOContext* io_ctx = nullptr, - std::optional const_value = std::nullopt); + OlapReaderStatistics* stats, const io::IOContext* io_ctx = nullptr); // Get column reader by column unique id and path(leaf node of variant's subcolumn) virtual Status get_path_column_reader(int32_t col_uid, PathInData relative_path, diff --git a/be/src/storage/segment/row_binlog_segment_writer.cpp b/be/src/storage/segment/row_binlog_segment_writer.cpp index c1102b318bb3b4..08afa0ad39f659 100644 --- a/be/src/storage/segment/row_binlog_segment_writer.cpp +++ b/be/src/storage/segment/row_binlog_segment_writer.cpp @@ -305,10 +305,8 @@ Status RowBinlogSegmentWriter::_fill_binlog_columns(size_t num_rows, { auto binlog_prefix_columns_guard = binlog_prefix_block.mutate_columns_scoped(); auto& binlog_prefix_columns = binlog_prefix_columns_guard.mutable_columns(); - // We can't get the real commit tso here (only known after publish). The tso column - // is replaced with the real commit_tso at read time - // (SegmentIterator::_update_tso_col_if_needed), so its on-disk value is never used. - // Write a NULL placeholder. + // We can't get the real commit tso here (only known after publish). Read paths use a + // request-scoped ConstantColumnReader for this column, so write a NULL placeholder. IColumn* ts_col_ptr = binlog_prefix_columns[0].get(); auto* ts_nullable_column = check_and_get_column(ts_col_ptr); DCHECK(ts_nullable_column != nullptr); diff --git a/be/src/storage/segment/segment.cpp b/be/src/storage/segment/segment.cpp index 9f41d84bc29510..12a4f837ecc524 100644 --- a/be/src/storage/segment/segment.cpp +++ b/be/src/storage/segment/segment.cpp @@ -26,6 +26,7 @@ #include #include #include +#include #include #include #include @@ -91,6 +92,39 @@ class InvertedIndexIterator; namespace { +// Call only after the segment's physical column existence has been established. +std::optional get_read_time_constant_value(int32_t col_uid, + const StorageReadOptions& read_options) { + if (read_options.tablet_schema == nullptr || + read_options.version.first != read_options.version.second) { + return std::nullopt; + } + + const auto& tablet_schema = *read_options.tablet_schema; + const auto matches_column = [&](int32_t column_idx) { + return column_idx >= 0 && tablet_schema.column(column_idx).unique_id() == col_uid; + }; + + if (matches_column(tablet_schema.version_col_idx())) { + return Field::create_field(read_options.version.second); + } + + const auto reader_type = read_options.io_ctx.reader_type; + const int64_t commit_tso = read_options.commit_tso.end_tso(); + if ((reader_type == ReaderType::READER_BINLOG || + reader_type == ReaderType::READER_BINLOG_COMPACTION) && + matches_column(tablet_schema.binlog_timestamp_col_idx())) { + DCHECK_EQ(read_options.commit_tso.start_tso(), commit_tso); + return Field::create_field(commit_tso == -1 ? 0 : commit_tso); + } + + if (commit_tso != -1 && matches_column(tablet_schema.commit_tso_col_idx())) { + return Field::create_field(commit_tso); + } + + return std::nullopt; +} + Status build_segment_zonemap_context(Segment* segment, const Schema& schema, const StorageReadOptions& read_options, const VExprContextSPtrs& conjuncts, ZoneMapEvalContext* ctx) { @@ -366,19 +400,6 @@ Status Segment::_open_index_file_reader() { return Status::OK(); } -bool Segment::is_tso_placeholder_col(int cid, const Schema& schema, - const StorageReadOptions& read_options) const { - if (read_options.version.first != read_options.version.second) { - return false; - } - if (read_options.io_ctx.reader_type != ReaderType::READER_BINLOG && - read_options.io_ctx.reader_type != ReaderType::READER_BINLOG_COMPACTION) { - return false; - } - // tso_col_idx() is -1 for non-binlog schemas, so this returns false there. - return cid == schema.tso_col_idx(); -} - Status Segment::new_iterator(SchemaSPtr schema, const StorageReadOptions& read_options, std::unique_ptr* iter) { if (read_options.runtime_state != nullptr) { @@ -396,17 +417,7 @@ Status Segment::new_iterator(SchemaSPtr schema, const StorageReadOptions& read_o } const TabletColumn& col = read_options.tablet_schema->column(column_id); std::shared_ptr reader; - // __DORIS_COMMIT_TSO_COL__ on a single-version segment stores a 0 placeholder on disk - // (replaced with the rowset's real commit_tso at read time). Its on-disk zonemap [0,0] - // must not drive segment-level pruning, so build a ConstantColumnReader carrying the real - // commit_tso to prune against the real value instead. - std::optional const_value; - if (read_options.version.first == read_options.version.second && - column_id == schema->commit_tso_col_idx() && read_options.commit_tso.end_tso() != -1) { - const_value = Field::create_field(read_options.commit_tso.end_tso()); - } - Status st = get_column_reader(col, &reader, read_options.stats, &read_options.io_ctx, - std::move(const_value)); + Status st = _get_column_reader_for_read(col, &reader, read_options); // not found in this segment, skip if (st.is()) { continue; @@ -417,28 +428,6 @@ Status Segment::new_iterator(SchemaSPtr schema, const StorageReadOptions& read_o if (!reader->has_zone_map()) { continue; } - // Placeholder tso column on a single-version binlog segment: its zonemap reflects the - // NULL placeholder (replaced with commit_tso at read time), so skip pruning by - // zonemap (min == max == commit_tso) and reuse the predicate's own zonemap matching: - // evaluate_and() returns false iff no value in [min, max] can satisfy the predicates, - // i.e. commit_tso fails them and the whole segment can be pruned. Predicates that don't - // support zonemap return true (conservative: not pruned, row-level eval handles them). - if (read_options.col_id_to_predicates.contains(column_id) && - is_tso_placeholder_col(column_id, *schema, read_options)) { - const Int64 commit_tso = - read_options.commit_tso.end_tso() == -1 ? 0 : read_options.commit_tso.end_tso(); - ZoneMap zone_map; - zone_map.min_value = Field::create_field(commit_tso); - zone_map.max_value = Field::create_field(commit_tso); - zone_map.has_not_null = true; - if (!entry.second->evaluate_and(zone_map)) { - // any condition not satisfied, return. - *iter = std::make_unique(*schema); - read_options.stats->filtered_segment_number++; - return Status::OK(); - } - continue; - } if (read_options.col_id_to_predicates.contains(column_id) && can_apply_predicate_safely(column_id, *schema, read_options.target_cast_type_for_variants, read_options)) { @@ -492,6 +481,12 @@ Status Segment::new_iterator(SchemaSPtr schema, const StorageReadOptions& read_o auto pruned = false; for (auto& it : _column_reader_cache->get_available_readers(false)) { const auto uid = it.first; + + // Do not use a physical placeholder zonemap for a runtime constant. + if (get_read_time_constant_value(uid, read_options).has_value()) { + continue; + } + const auto column_id = read_options.tablet_schema->field_index(uid); bool tmp_pruned = false; RETURN_IF_ERROR(it.second->prune_predicates_by_zone_map(pruned_predicates, column_id, @@ -898,36 +893,16 @@ Status Segment::new_column_iterator(const TabletColumn& tablet_column, int32_t unique_id = tablet_column.unique_id() >= 0 ? tablet_column.unique_id() : tablet_column.parent_unique_id(); - // If column meta for this uid is not found in this segment, use default iterator. - if (!_column_meta_accessor->has_column_uid(unique_id)) { + // Initialize the reader through the read-semantic entry. A physically missing column still + // uses its schema default, including hidden columns added by a later schema change. + std::shared_ptr reader; + Status st = _get_column_reader_for_read(unique_id, &reader, *opt); + if (st.is()) { RETURN_IF_ERROR(new_default_iterator(tablet_column, iter)); return Status::OK(); } - - // __DORIS_COMMIT_TSO_COL__ on a single-version segment stores a 0 placeholder on disk (its - // real value is the rowset's commit_tso, filled at read time). Pass the real commit_tso as a - // const value so the cache returns a ConstantColumnReader, whose iterator yields the real value - // on every read path (projection / predicate / MIN-MAX zone-map) instead of the placeholder 0. - // commit_tso == -1 means it is not assigned yet (before publish); keep the on-disk value then. - // The value is constant per segment (a segment belongs to a single rowset), so caching the - // ConstantColumnReader does not cross-pollute other queries. Some internal read paths (e.g. MOW - // partial-update row fetch) build a bare StorageReadOptions without tablet_schema, so guard it. - std::optional const_value; - if (opt->tablet_schema != nullptr && opt->version.first == opt->version.second && - opt->commit_tso.end_tso() != -1) { - int32_t tso_idx = opt->tablet_schema->commit_tso_col_idx(); - if (tso_idx != -1 && opt->tablet_schema->column(tso_idx).unique_id() == unique_id) { - const_value = Field::create_field(opt->commit_tso.end_tso()); - } - } - - // init iterator by unique id - std::shared_ptr reader; - RETURN_IF_ERROR(get_column_reader(unique_id, &reader, opt->stats, &opt->io_ctx, - std::move(const_value))); - if (reader == nullptr) { - return Status::InternalError("column reader is nullptr, unique_id={}", unique_id); - } + RETURN_IF_ERROR(st); + DCHECK(reader != nullptr); if (reader->get_meta_type() == FieldType::OLAP_FIELD_TYPE_VARIANT) { // if sparse_column_cache_ptr is nullptr, means the sparse column cache is not used PathToBinaryColumnCache* sparse_column_cache_ptr = nullptr; @@ -974,18 +949,16 @@ Status Segment::new_column_iterator(const TabletColumn& tablet_column, } Status Segment::get_column_reader(int32_t col_uid, std::shared_ptr* column_reader, - OlapReaderStatistics* stats, const io::IOContext* source_io_ctx, - std::optional const_value) { + OlapReaderStatistics* stats, const io::IOContext* source_io_ctx) { RETURN_IF_ERROR(_create_column_meta_once(stats, source_io_ctx)); SCOPED_RAW_TIMER(&stats->segment_create_column_readers_timer_ns); // The column is not in this segment, return nullptr - if (!_tablet_schema->has_column_unique_id(col_uid)) { + if (!_column_meta_accessor->has_column_uid(col_uid)) { *column_reader = nullptr; return Status::Error("column not found in segment, col_uid={}", col_uid); } - return _column_reader_cache->get_column_reader(col_uid, column_reader, stats, source_io_ctx, - std::move(const_value)); + return _column_reader_cache->get_column_reader(col_uid, column_reader, stats, source_io_ctx); } Status Segment::traverse_column_meta_pbs(const std::function& visitor) { @@ -999,13 +972,12 @@ Status Segment::traverse_column_meta_pbs(const std::function* column_reader, - OlapReaderStatistics* stats, const io::IOContext* source_io_ctx, - std::optional const_value) { + OlapReaderStatistics* stats, const io::IOContext* source_io_ctx) { RETURN_IF_ERROR(_create_column_meta_once(stats, source_io_ctx)); SCOPED_RAW_TIMER(&stats->segment_create_column_readers_timer_ns); int col_uid = col.unique_id() >= 0 ? col.unique_id() : col.parent_unique_id(); // The column is not in this segment, return nullptr - if (!_tablet_schema->has_column_unique_id(col_uid)) { + if (!_column_meta_accessor->has_column_uid(col_uid)) { *column_reader = nullptr; return Status::Error("column not found in segment, col_uid={}", col_uid); @@ -1015,8 +987,41 @@ Status Segment::get_column_reader(const TabletColumn& col, return _column_reader_cache->get_path_column_reader(col_uid, relative_path, column_reader, stats, nullptr, source_io_ctx); } - return _column_reader_cache->get_column_reader(col_uid, column_reader, stats, source_io_ctx, - std::move(const_value)); + return _column_reader_cache->get_column_reader(col_uid, column_reader, stats, source_io_ctx); +} + +Status Segment::_get_column_reader_for_read(int32_t col_uid, + std::shared_ptr* column_reader, + const StorageReadOptions& read_options) { + RETURN_IF_ERROR(_create_column_meta_once(read_options.stats, &read_options.io_ctx)); + SCOPED_RAW_TIMER(&read_options.stats->segment_create_column_readers_timer_ns); + + // A column added after this segment was written must use its schema default. Resolve runtime + // constants only after physical existence has been established. + if (!_column_meta_accessor->has_column_uid(col_uid)) { + *column_reader = nullptr; + return Status::Error("column not found in segment, col_uid={}", + col_uid); + } + + if (auto const_value = get_read_time_constant_value(col_uid, read_options); + const_value.has_value()) { + *column_reader = std::make_shared(std::move(*const_value)); + return Status::OK(); + } + + return _column_reader_cache->get_column_reader(col_uid, column_reader, read_options.stats, + &read_options.io_ctx); +} + +Status Segment::_get_column_reader_for_read(const TabletColumn& col, + std::shared_ptr* column_reader, + const StorageReadOptions& read_options) { + const int32_t col_uid = col.unique_id() >= 0 ? col.unique_id() : col.parent_unique_id(); + if (col.has_path_info()) { + return get_column_reader(col, column_reader, read_options.stats, &read_options.io_ctx); + } + return _get_column_reader_for_read(col_uid, column_reader, read_options); } Status Segment::new_index_iterator(const TabletColumn& tablet_column, const TabletIndex* index_meta, diff --git a/be/src/storage/segment/segment.h b/be/src/storage/segment/segment.h index c46f5dad268356..92b13419d1fd6f 100644 --- a/be/src/storage/segment/segment.h +++ b/be/src/storage/segment/segment.h @@ -25,7 +25,6 @@ #include #include #include // for unique_ptr -#include #include #include @@ -206,25 +205,15 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd } } - // The tso column (__DORIS_BINLOG_TSO__) is a NULL placeholder on disk on a - // single-version binlog segment, replaced with the real commit_tso at read time - // (SegmentIterator::_update_tso_col_if_needed). Its zonemap reflects the placeholder, so - // it must NOT drive zonemap pruning. Mirrors the guards of _update_tso_col_if_needed. - // Returns false for range (compaction) segments whose on-disk value is real. - bool is_tso_placeholder_col(int cid, const Schema& schema, - const StorageReadOptions& read_options) const; - const TabletSchemaSPtr& tablet_schema() const { return _tablet_schema; } - // get the column reader by tablet column, return NOT_FOUND if not found reader in this segment + // get the physical column reader by tablet column, return NOT_FOUND if not found Status get_column_reader(const TabletColumn& col, std::shared_ptr* column_reader, - OlapReaderStatistics* stats, const io::IOContext* io_ctx = nullptr, - std::optional const_value = std::nullopt); + OlapReaderStatistics* stats, const io::IOContext* io_ctx = nullptr); - // get the column reader by column unique id, return NOT_FOUND if not found reader in this segment + // get the physical column reader by column unique id, return NOT_FOUND if not found Status get_column_reader(int32_t col_uid, std::shared_ptr* column_reader, - OlapReaderStatistics* stats, const io::IOContext* io_ctx = nullptr, - std::optional const_value = std::nullopt); + OlapReaderStatistics* stats, const io::IOContext* io_ctx = nullptr); Status traverse_column_meta_pbs(const std::function& visitor); @@ -257,6 +246,12 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd const io::IOContext* io_ctx = nullptr); Status _load_pk_bloom_filter(OlapReaderStatistics* stats, const io::IOContext* io_ctx = nullptr); + Status _get_column_reader_for_read(const TabletColumn& col, + std::shared_ptr* column_reader, + const StorageReadOptions& read_options); + Status _get_column_reader_for_read(int32_t col_uid, + std::shared_ptr* column_reader, + const StorageReadOptions& read_options); Status _write_error_file(size_t file_size, size_t offset, size_t bytes_read, char* data, io::IOContext& io_ctx); diff --git a/be/src/storage/segment/segment_iterator.cpp b/be/src/storage/segment/segment_iterator.cpp index bc04ffa2aca276..574be47571b9f5 100644 --- a/be/src/storage/segment/segment_iterator.cpp +++ b/be/src/storage/segment/segment_iterator.cpp @@ -1132,12 +1132,6 @@ Status SegmentIterator::_get_row_ranges_from_conditions(RowRanges* condition_row _opts.target_cast_type_for_variants, _opts)) { continue; } - if (_segment->is_tso_placeholder_col(cid, *_schema, _opts)) { - // skip untrustworthy tso placeholder zonemap - // if possible already be pruned as a whole before, - // so just skip - continue; - } // do not check zonemap if predicate does not support zonemap if (!_opts.col_id_to_predicates.at(cid)->support_zonemap()) { VLOG_DEBUG << "skip zonemap for column " << cid; @@ -2427,35 +2421,9 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 return Status::OK(); } -void SegmentIterator::_replace_version_col_if_needed(const std::vector& column_ids, - size_t num_rows) { - // Only the rowset with single version need to replace the version column. - // Doris can't determine the version before publish_version finished, so - // we can't write data to __DORIS_VERSION_COL__ in segment writer, the value - // is 0 by default. - // So we need to replace the value to real version while reading. - if (_opts.version.first != _opts.version.second) { - return; - } - int32_t version_idx = _schema->version_col_idx(); - if (std::ranges::find(column_ids, version_idx) == column_ids.end()) { - return; - } - - const auto* column_desc = _schema->column(version_idx); - auto column = Schema::get_data_type_ptr(*column_desc)->create_column(); - DCHECK(_schema->column(version_idx)->type() == FieldType::OLAP_FIELD_TYPE_BIGINT); - auto* col_ptr = assert_cast(column.get()); - for (size_t j = 0; j < num_rows; j++) { - col_ptr->insert_value(_opts.version.second); - } - _current_return_columns[version_idx] = std::move(column); - VLOG_DEBUG << "replaced version column in segment iterator, version_col_idx:" << version_idx; -} - -void SegmentIterator::_update_tso_col_if_needed(const std::vector& column_ids, +void SegmentIterator::_update_lsn_col_if_needed(const std::vector& column_ids, size_t num_rows) { - // use physical time part of commit tso to replace timestamp col + // | commit tso(64) | auto-inc row_id(64) | if (_opts.version.first != _opts.version.second) { return; } @@ -2465,54 +2433,40 @@ void SegmentIterator::_update_tso_col_if_needed(const std::vector& col return; } - int32_t tso_col_idx = _schema->tso_col_idx(); - if (tso_col_idx < 0 || std::ranges::find(column_ids, tso_col_idx) == column_ids.end()) { + int32_t lsn_col_idx = _schema->lsn_col_idx(); + if (lsn_col_idx < 0 || std::ranges::find(column_ids, lsn_col_idx) == column_ids.end()) { return; } DCHECK_EQ(_opts.commit_tso.start_tso(), _opts.commit_tso.end_tso()); - Int64 commit_tso = _opts.commit_tso.end_tso() == -1 ? 0 : _opts.commit_tso.end_tso(); - - if (_is_pred_column[tso_col_idx]) { - // Nullable predicate column is represented as ColumnNullable(predicate_col) - if (auto* tso_nullable = check_and_get_column( - _current_return_columns[tso_col_idx].get())) { - _current_return_columns[tso_col_idx]->clear(); - auto value = commit_tso; - for (size_t j = 0; j < num_rows; j++) { - tso_nullable->get_nested_column_ptr()->insert_data( - reinterpret_cast(&value), 0); - tso_nullable->get_null_map_data().emplace_back(0); - } - return; - } + const Int64 commit_tso = _opts.commit_tso.end_tso() == -1 ? 0 : _opts.commit_tso.end_tso(); - auto* tso_column = assert_cast(_current_return_columns[tso_col_idx].get()); - tso_column->clear(); - auto value = commit_tso; + if (_is_pred_column[lsn_col_idx]) { + auto* lsn_column = assert_cast(_current_return_columns[lsn_col_idx].get()); + std::vector binlog_lsns; + binlog_lsns.reserve(num_rows); for (size_t j = 0; j < num_rows; j++) { - tso_column->insert_data(reinterpret_cast(&value), 0); + const Int128 row_id = lsn_column->get_data()[j]; + binlog_lsns.emplace_back(make_row_binlog_lsn(commit_tso, row_id)); + } + lsn_column->clear(); + for (const auto& binlog_lsn : binlog_lsns) { + lsn_column->insert_data(reinterpret_cast(&binlog_lsn), 0); } return; } - const auto* column_desc = _schema->column(tso_col_idx); + auto* lsn_column = assert_cast(_current_return_columns[lsn_col_idx].get()); + const auto* column_desc = _schema->column(lsn_col_idx); auto column = Schema::get_data_type_ptr(*column_desc)->create_column(); - DCHECK(column_desc->type() == FieldType::OLAP_FIELD_TYPE_BIGINT); + DCHECK(column_desc->type() == FieldType::OLAP_FIELD_TYPE_LARGEINT); + auto* col_ptr = assert_cast(column.get()); - if (auto* tso_nullable = check_and_get_column(column.get())) { - auto* col_ptr = assert_cast(&tso_nullable->get_nested_column()); - for (size_t j = 0; j < num_rows; j++) { - col_ptr->insert_value(commit_tso); - tso_nullable->get_null_map_data().emplace_back(0); - } - } else { - auto* col_ptr = assert_cast(column.get()); - for (size_t j = 0; j < num_rows; j++) { - col_ptr->insert_value(commit_tso); - } + for (size_t j = 0; j < num_rows; j++) { + const Int128 row_id = lsn_column->get_element(j); + col_ptr->insert_value(make_row_binlog_lsn(commit_tso, row_id)); } - _current_return_columns[tso_col_idx] = std::move(column); + _current_return_columns[lsn_col_idx] = std::move(column); } uint16_t SegmentIterator::_evaluate_vectorization_predicate(uint16_t* sel_rowid_idx, @@ -2921,8 +2875,7 @@ Status SegmentIterator::_next_batch_internal(Block* block) { _selected_size = 0; RETURN_IF_ERROR(_read_columns_by_index(nrows_read_limit, _selected_size)); - _replace_version_col_if_needed(_predicate_column_ids, _selected_size); - _update_tso_col_if_needed(_predicate_column_ids, _selected_size); + _update_lsn_col_if_needed(_predicate_column_ids, _selected_size); _opts.stats->blocks_load += 1; _opts.stats->raw_rows_read += _selected_size; @@ -2965,8 +2918,7 @@ Status SegmentIterator::_next_batch_internal(Block* block) { RETURN_IF_ERROR(_read_columns_by_rowids( _common_expr_column_ids, _block_rowids, _sel_rowid_idx.data(), _selected_size, &_current_return_columns, false, true)); - _replace_version_col_if_needed(_common_expr_column_ids, _selected_size); - _update_tso_col_if_needed(_common_expr_column_ids, _selected_size); + _update_lsn_col_if_needed(_common_expr_column_ids, _selected_size); RETURN_IF_ERROR(_process_columns(_common_expr_column_ids, block)); } @@ -2999,8 +2951,7 @@ Status SegmentIterator::_next_batch_internal(Block* block) { _non_predicate_columns, _block_rowids, _sel_rowid_idx.data(), _selected_size, &_current_return_columns, _opts.condition_cache_digest && !_find_condition_cache, false)); - _replace_version_col_if_needed(_non_predicate_columns, _selected_size); - _update_tso_col_if_needed(_non_predicate_columns, _selected_size); + _update_lsn_col_if_needed(_non_predicate_columns, _selected_size); } else { if (_opts.condition_cache_digest && !_find_condition_cache) { auto& condition_cache = *_condition_cache; @@ -3370,8 +3321,7 @@ Status SegmentIterator::_apply_expr_zonemap_to_row_ranges(const VExprContextSPtr continue; } std::shared_ptr reader; - Status st = - _segment->get_column_reader(*tablet_column, &reader, _opts.stats, &_opts.io_ctx); + Status st = _segment->_get_column_reader_for_read(*tablet_column, &reader, _opts); if (st.is()) { continue; } diff --git a/be/src/storage/segment/segment_iterator.h b/be/src/storage/segment/segment_iterator.h index 99e40f2a5d1214..eaa131dfbd7b0a 100644 --- a/be/src/storage/segment/segment_iterator.h +++ b/be/src/storage/segment/segment_iterator.h @@ -216,8 +216,7 @@ class SegmentIterator : public RowwiseIterator { [[nodiscard]] Status _read_columns(const std::vector& column_ids, MutableColumns& column_block, size_t nrows); [[nodiscard]] Status _read_columns_by_index(uint32_t nrows_read_limit, uint16_t& nrows_read); - void _replace_version_col_if_needed(const std::vector& column_ids, size_t num_rows); - void _update_tso_col_if_needed(const std::vector& column_ids, size_t num_rows); + void _update_lsn_col_if_needed(const std::vector& column_ids, size_t num_rows); Status _init_current_block(Block* block, std::vector& non_pred_vector, uint32_t nrows_read_limit); uint16_t _evaluate_vectorization_predicate(uint16_t* sel_rowid_idx, uint16_t selected_size); diff --git a/be/test/storage/predicate/block_column_predicate_test.cpp b/be/test/storage/predicate/block_column_predicate_test.cpp index 64d9fb244d6a7d..c43c8692b6a597 100644 --- a/be/test/storage/predicate/block_column_predicate_test.cpp +++ b/be/test/storage/predicate/block_column_predicate_test.cpp @@ -2702,72 +2702,4 @@ TEST_F(BlockColumnPredicateTest, COMBINED_PREDICATE) { } } -// Verifies the core algorithm of SegmentIterator::_can_prune_segment_by_tso: on a -// single-version binlog segment the tso column (__DORIS_BINLOG_TSO__) is replaced at -// read time with a constant commit_tso, so whole-segment pruning is decided by matching the -// tso predicates against a degenerate zonemap (min == max == commit_tso). evaluate_and() -// returns false iff commit_tso fails the predicates, i.e. the whole segment can be pruned. -TEST_F(BlockColumnPredicateTest, tso_whole_segment_prune_by_commit_tso) { - // The user-visible raw commit tso (~4.66e17). Physical time part is ~1.78e12. - constexpr int64_t kCommitTso = 466872251335573505L; - - auto make_tso_zone_map = [](int64_t commit_tso) { - return segment_v2::ZoneMap {.min_value = Field::create_field(commit_tso), - .max_value = Field::create_field(commit_tso), - .has_null = false, - .has_not_null = true}; - }; - // Mirrors _can_prune_segment_by_tso: returns true if the segment can be pruned. - auto can_prune = [&](const AndBlockColumnPredicate& preds) { - return !preds.evaluate_and(make_tso_zone_map(kCommitTso)); - }; - - auto make_gt = [](int64_t value) { - std::shared_ptr pred( - new ComparisonPredicateBase( - 0, "", Field::create_field(value))); - return SingleColumnBlockPredicate::create_unique(pred); - }; - auto make_le = [](int64_t value) { - std::shared_ptr pred( - new ComparisonPredicateBase( - 0, "", Field::create_field(value))); - return SingleColumnBlockPredicate::create_unique(pred); - }; - - // `> physical-time-magnitude` (between physical time and raw tso): commit_tso satisfies it, - // so the segment must NOT be pruned (this is exactly the historical mis-prune case). - { - AndBlockColumnPredicate preds; - preds.add_column_predicate(make_gt(1790000000000L)); - EXPECT_FALSE(can_prune(preds)); - } - // `> 0`: commit_tso satisfies it, segment kept. - { - AndBlockColumnPredicate preds; - preds.add_column_predicate(make_gt(0)); - EXPECT_FALSE(can_prune(preds)); - } - // `> upper-bound (> commit_tso)`: commit_tso fails it, whole segment can be pruned. - { - AndBlockColumnPredicate preds; - preds.add_column_predicate(make_gt(kCommitTso)); - EXPECT_TRUE(can_prune(preds)); - } - // Range `commit_tso < tso <= commit_tso+10`: commit_tso fails the lower bound, pruned. - { - AndBlockColumnPredicate preds; - preds.add_column_predicate(make_gt(kCommitTso)); - preds.add_column_predicate(make_le(kCommitTso + 10)); - EXPECT_TRUE(can_prune(preds)); - } - // Range `commit_tso-10 < tso <= commit_tso`: commit_tso satisfies both, kept. - { - AndBlockColumnPredicate preds; - preds.add_column_predicate(make_gt(kCommitTso - 10)); - preds.add_column_predicate(make_le(kCommitTso)); - EXPECT_FALSE(can_prune(preds)); - } -} - } // namespace doris diff --git a/be/test/storage/segment/constant_column_iterator_test.cpp b/be/test/storage/segment/constant_column_iterator_test.cpp index cde4063fd81754..c71b87013301ef 100644 --- a/be/test/storage/segment/constant_column_iterator_test.cpp +++ b/be/test/storage/segment/constant_column_iterator_test.cpp @@ -36,20 +36,15 @@ namespace doris { class ConstantColumnIteratorTest : public testing::Test {}; -TEST_F(ConstantColumnIteratorTest, ColumnReaderCreateWithConstValueReturnsConstantReader) { +TEST_F(ConstantColumnIteratorTest, ConstantColumnReaderExposesConstantValue) { const int64_t kValue = 8888; - ColumnReaderOptions opts; - opts.const_value = Field::create_field(kValue); - - std::shared_ptr reader; - auto st = ColumnReader::create(opts, ColumnMetaPB(), 3, io::FileReaderSPtr(), &reader); - ASSERT_TRUE(st.ok()) << st; - ASSERT_NE(nullptr, reader); + std::shared_ptr reader = + std::make_shared(Field::create_field(kValue)); EXPECT_TRUE(reader->has_zone_map()); EXPECT_EQ(FieldType::OLAP_FIELD_TYPE_BIGINT, reader->get_meta_type()); segment_v2::ZoneMap zone_map; - st = reader->get_segment_zone_map(&zone_map); + auto st = reader->get_segment_zone_map(&zone_map); ASSERT_TRUE(st.ok()) << st; EXPECT_EQ(kValue, zone_map.min_value.get()); EXPECT_EQ(kValue, zone_map.max_value.get()); diff --git a/be/test/storage/segment/segment_iterator_expr_zonemap_test.cpp b/be/test/storage/segment/segment_iterator_expr_zonemap_test.cpp index 172ab8fdafc1ad..8d6b5402261196 100644 --- a/be/test/storage/segment/segment_iterator_expr_zonemap_test.cpp +++ b/be/test/storage/segment/segment_iterator_expr_zonemap_test.cpp @@ -23,6 +23,7 @@ #include #include "core/assert_cast.h" +#include "core/column/column_nullable.h" #include "core/column/column_vector.h" #include "core/data_type/data_type_number.h" #include "core/field.h" @@ -50,7 +51,10 @@ namespace { constexpr auto kTestDir = "./ut_dir/segment_iterator_expr_zonemap_test"; constexpr int kNumRows = 8192; -constexpr int kCommitTsoRows = 8; +constexpr int kRuntimeColumnRows = 8; +constexpr int kVersionCid = 1; +constexpr int kBinlogTimestampCid = 2; +constexpr int kCommitTsoCid = 3; const RowsetId kRowsetId {.version = 1}; Field int_field(int32_t value) { @@ -105,10 +109,38 @@ TabletSchemaSPtr make_tablet_schema() { return tablet_schema; } -TabletSchemaSPtr make_commit_tso_tablet_schema() { +TabletColumnPtr make_runtime_bigint_column(int32_t id, const std::string& name, bool is_nullable, + const std::string& default_value = "") { + auto column = std::make_shared(); + column->set_unique_id(id); + column->set_name(name); + column->set_type(FieldType::OLAP_FIELD_TYPE_BIGINT); + column->set_is_key(false); + column->set_is_nullable(is_nullable); + column->set_length(8); + column->set_index_length(8); + column->set_aggregation_method(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE); + if (!default_value.empty()) { + column->set_default_value(default_value); + } + return column; +} + +TabletSchemaSPtr make_runtime_column_tablet_schema() { + auto tablet_schema = std::make_shared(); + tablet_schema->append_column(*create_int_key(0, false)); + tablet_schema->append_column(*make_runtime_bigint_column(kVersionCid, VERSION_COL, false, "0")); + tablet_schema->append_column( + *make_runtime_bigint_column(kBinlogTimestampCid, BINLOG_TIMESTAMP_COL, true)); + tablet_schema->append_column( + *make_runtime_bigint_column(kCommitTsoCid, COMMIT_TSO_COL, false, "0")); + tablet_schema->set_storage_page_size(4096); + return tablet_schema; +} + +TabletSchemaSPtr make_key_only_tablet_schema() { auto tablet_schema = std::make_shared(); tablet_schema->append_column(*create_int_key(0, false)); - tablet_schema->append_column(*create_commit_tso_column(1)); tablet_schema->set_storage_page_size(4096); return tablet_schema; } @@ -185,27 +217,33 @@ class SegmentIteratorExprZonemapTest : public testing::Test { ASSERT_EQ(kNumRows, (*segment)->num_rows()); } - void build_commit_tso_segment(std::shared_ptr* segment) { - const auto path = std::string(kTestDir) + "/commit_tso_segment.dat"; + void build_runtime_column_segment(std::shared_ptr* segment, + bool write_runtime_columns = true) { + const auto path = std::string(kTestDir) + "/runtime_column_segment.dat"; auto fs = io::global_local_filesystem(); io::FileWriterPtr file_writer; auto st = fs->create_file(path, &file_writer); ASSERT_TRUE(st.ok()) << st; + auto writer_schema = write_runtime_columns ? _tablet_schema : make_key_only_tablet_schema(); SegmentWriterOptions opts; opts.num_rows_per_block = 4; - TestSegmentWriter writer(file_writer.get(), 0, _tablet_schema, nullptr, nullptr, opts, + TestSegmentWriter writer(file_writer.get(), 0, writer_schema, nullptr, nullptr, opts, nullptr); st = writer.init(); ASSERT_TRUE(st.ok()) << st; RowCursor row; - std::vector fields(_tablet_schema->num_columns(), Field(PrimitiveType::TYPE_NULL)); - st = row.init_scan_key(_tablet_schema, std::move(fields)); + std::vector fields(writer_schema->num_columns(), Field(PrimitiveType::TYPE_NULL)); + st = row.init_scan_key(writer_schema, std::move(fields)); ASSERT_TRUE(st.ok()) << st; - for (int rid = 0; rid < kCommitTsoRows; ++rid) { + for (int rid = 0; rid < kRuntimeColumnRows; ++rid) { row.mutable_field(0) = int_field(rid); - row.mutable_field(1) = Field::create_field(0); + if (write_runtime_columns) { + row.mutable_field(kVersionCid) = Field::create_field(0); + row.mutable_field(kBinlogTimestampCid) = Field(); + row.mutable_field(kCommitTsoCid) = Field::create_field(0); + } st = writer.append_row(row); ASSERT_TRUE(st.ok()) << st; } @@ -220,7 +258,58 @@ class SegmentIteratorExprZonemapTest : public testing::Test { st = Segment::open(fs, path, 100, 0, kRowsetId, _tablet_schema, io::FileReaderOptions {}, segment); ASSERT_TRUE(st.ok()) << st; - ASSERT_EQ(kCommitTsoRows, (*segment)->num_rows()); + ASSERT_EQ(kRuntimeColumnRows, (*segment)->num_rows()); + } + + void read_column(const std::shared_ptr& segment, int32_t cid, + const StorageReadOptions& read_options, MutableColumnPtr* dst) { + ColumnIteratorUPtr iter; + auto st = segment->new_column_iterator(_tablet_schema->column(cid), &iter, &read_options); + ASSERT_TRUE(st.ok()) << st; + ASSERT_NE(nullptr, iter); + + auto file_reader = segment->file_reader(); + ColumnIteratorOptions iter_opts; + iter_opts.stats = &_stats; + iter_opts.file_reader = file_reader.get(); + iter_opts.io_ctx = read_options.io_ctx; + st = iter->init(iter_opts); + ASSERT_TRUE(st.ok()) << st; + st = iter->seek_to_ordinal(0); + ASSERT_TRUE(st.ok()) << st; + + *dst = Schema::get_data_type_ptr(_tablet_schema->column(cid))->create_column(); + size_t n = kRuntimeColumnRows; + bool has_null = false; + st = iter->next_batch(&n, *dst, &has_null); + ASSERT_TRUE(st.ok()) << st; + ASSERT_EQ(kRuntimeColumnRows, n); + ASSERT_EQ(kRuntimeColumnRows, (*dst)->size()); + } + + void expect_bigint_values(const MutableColumnPtr& column, int64_t expected) { + ASSERT_NE(nullptr, column.get()); + const IColumn* data_column = column.get(); + if (const auto* nullable = check_and_get_column(data_column)) { + for (size_t i = 0; i < nullable->size(); ++i) { + EXPECT_FALSE(nullable->is_null_at(i)); + } + data_column = &nullable->get_nested_column(); + } + const auto* bigint_column = check_and_get_column(data_column); + ASSERT_NE(nullptr, bigint_column); + for (size_t i = 0; i < bigint_column->size(); ++i) { + EXPECT_EQ(expected, bigint_column->get_element(i)); + } + } + + void expect_all_null(const MutableColumnPtr& column) { + ASSERT_NE(nullptr, column.get()); + const auto* nullable = check_and_get_column(column.get()); + ASSERT_NE(nullptr, nullable); + for (size_t i = 0; i < nullable->size(); ++i) { + EXPECT_TRUE(nullable->is_null_at(i)); + } } void prepare_expr_context(const VExprContextSPtr& expr_ctx) { @@ -303,52 +392,113 @@ TEST_F(SegmentIteratorExprZonemapTest, ApplyExprZonemapPrunesPageRowRanges) { EXPECT_EQ(kNumRows, row_ranges.to()); } -TEST_F(SegmentIteratorExprZonemapTest, NewColumnIteratorReadsCommitTsoFromReadOptions) { - constexpr int64_t kCommitTso = 466872251335573505L; - _tablet_schema = make_commit_tso_tablet_schema(); +TEST_F(SegmentIteratorExprZonemapTest, RuntimeColumnsUseCurrentReadOptions) { + constexpr int64_t kCommitTso1 = 466872251335573505L; + constexpr int64_t kCommitTso2 = kCommitTso1 + 100; + _tablet_schema = make_runtime_column_tablet_schema(); std::shared_ptr segment; - ASSERT_NO_FATAL_FAILURE(build_commit_tso_segment(&segment)); + ASSERT_NO_FATAL_FAILURE(build_runtime_column_segment(&segment)); StorageReadOptions read_options; read_options.stats = &_stats; read_options.tablet_schema = _tablet_schema; + read_options.version = Version(0, 0); + read_options.io_ctx.reader_type = ReaderType::READER_QUERY; + + // Before publish, physical readers may already be created and cached. + MutableColumnPtr version_column; + MutableColumnPtr binlog_timestamp_column; + MutableColumnPtr commit_tso_column; + ASSERT_NO_FATAL_FAILURE(read_column(segment, kVersionCid, read_options, &version_column)); + ASSERT_NO_FATAL_FAILURE( + read_column(segment, kBinlogTimestampCid, read_options, &binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(read_column(segment, kCommitTsoCid, read_options, &commit_tso_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(version_column, 0)); + ASSERT_NO_FATAL_FAILURE(expect_all_null(binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(commit_tso_column, 0)); + read_options.version = Version(7, 7); + read_options.commit_tso = TsoRange(kCommitTso1, kCommitTso1); + read_options.io_ctx.reader_type = ReaderType::READER_BINLOG; + ASSERT_NO_FATAL_FAILURE(read_column(segment, kVersionCid, read_options, &version_column)); + ASSERT_NO_FATAL_FAILURE( + read_column(segment, kBinlogTimestampCid, read_options, &binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(read_column(segment, kCommitTsoCid, read_options, &commit_tso_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(version_column, 7)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(binlog_timestamp_column, kCommitTso1)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(commit_tso_column, kCommitTso1)); + + // A second read must observe the new request values instead of a cached constant reader. + read_options.version = Version(9, 9); + read_options.commit_tso = TsoRange(kCommitTso2, kCommitTso2); + ASSERT_NO_FATAL_FAILURE(read_column(segment, kVersionCid, read_options, &version_column)); + ASSERT_NO_FATAL_FAILURE( + read_column(segment, kBinlogTimestampCid, read_options, &binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(read_column(segment, kCommitTsoCid, read_options, &commit_tso_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(version_column, 9)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(binlog_timestamp_column, kCommitTso2)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(commit_tso_column, kCommitTso2)); +} + +TEST_F(SegmentIteratorExprZonemapTest, RangeVersionUsesPhysicalValues) { + constexpr int64_t kCommitTso = 466872251335573505L; + _tablet_schema = make_runtime_column_tablet_schema(); + + std::shared_ptr segment; + ASSERT_NO_FATAL_FAILURE(build_runtime_column_segment(&segment)); + + StorageReadOptions read_options; + read_options.stats = &_stats; + read_options.tablet_schema = _tablet_schema; + read_options.version = Version(7, 9); read_options.commit_tso = TsoRange(kCommitTso, kCommitTso); - read_options.io_ctx.reader_type = ReaderType::READER_QUERY; + read_options.io_ctx.reader_type = ReaderType::READER_BINLOG; + + MutableColumnPtr version_column; + MutableColumnPtr binlog_timestamp_column; + MutableColumnPtr commit_tso_column; + ASSERT_NO_FATAL_FAILURE(read_column(segment, kVersionCid, read_options, &version_column)); + ASSERT_NO_FATAL_FAILURE( + read_column(segment, kBinlogTimestampCid, read_options, &binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(read_column(segment, kCommitTsoCid, read_options, &commit_tso_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(version_column, 0)); + ASSERT_NO_FATAL_FAILURE(expect_all_null(binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(commit_tso_column, 0)); +} - ColumnIteratorUPtr iter; - auto st = segment->new_column_iterator(_tablet_schema->column(1), &iter, &read_options); - ASSERT_TRUE(st.ok()) << st; - ASSERT_NE(nullptr, iter); +TEST_F(SegmentIteratorExprZonemapTest, MissingPhysicalColumnsUseSchemaDefaults) { + constexpr int64_t kCommitTso = 466872251335573505L; + _tablet_schema = make_runtime_column_tablet_schema(); - auto file_reader = segment->file_reader(); - ColumnIteratorOptions iter_opts; - iter_opts.stats = &_stats; - iter_opts.file_reader = file_reader.get(); - iter_opts.io_ctx.reader_type = ReaderType::READER_QUERY; - st = iter->init(iter_opts); - ASSERT_TRUE(st.ok()) << st; + std::shared_ptr segment; + ASSERT_NO_FATAL_FAILURE(build_runtime_column_segment(&segment, false)); - MutableColumnPtr dst = ColumnVector::create(); - size_t n = kCommitTsoRows; - bool has_null = true; - st = iter->next_batch(&n, dst, &has_null); - ASSERT_TRUE(st.ok()) << st; - ASSERT_FALSE(has_null); - ASSERT_EQ(kCommitTsoRows, dst->size()); - auto* col = assert_cast(dst.get()); - for (size_t i = 0; i < dst->size(); ++i) { - EXPECT_EQ(kCommitTso, col->get_element(i)); - } + StorageReadOptions read_options; + read_options.stats = &_stats; + read_options.tablet_schema = _tablet_schema; + read_options.version = Version(7, 7); + read_options.commit_tso = TsoRange(kCommitTso, kCommitTso); + read_options.io_ctx.reader_type = ReaderType::READER_BINLOG; + + MutableColumnPtr version_column; + MutableColumnPtr binlog_timestamp_column; + MutableColumnPtr commit_tso_column; + ASSERT_NO_FATAL_FAILURE(read_column(segment, kVersionCid, read_options, &version_column)); + ASSERT_NO_FATAL_FAILURE( + read_column(segment, kBinlogTimestampCid, read_options, &binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(read_column(segment, kCommitTsoCid, read_options, &commit_tso_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(version_column, 0)); + ASSERT_NO_FATAL_FAILURE(expect_all_null(binlog_timestamp_column)); + ASSERT_NO_FATAL_FAILURE(expect_bigint_values(commit_tso_column, 0)); } TEST_F(SegmentIteratorExprZonemapTest, NewIteratorPrunesCommitTsoByReadOptionValue) { constexpr int64_t kCommitTso = 466872251335573505L; - _tablet_schema = make_commit_tso_tablet_schema(); + _tablet_schema = make_runtime_column_tablet_schema(); std::shared_ptr segment; - ASSERT_NO_FATAL_FAILURE(build_commit_tso_segment(&segment)); + ASSERT_NO_FATAL_FAILURE(build_runtime_column_segment(&segment)); auto read_schema = make_read_schema(_tablet_schema); StorageReadOptions read_options; @@ -357,7 +507,8 @@ TEST_F(SegmentIteratorExprZonemapTest, NewIteratorPrunesCommitTsoByReadOptionVal read_options.version = Version(7, 7); read_options.commit_tso = TsoRange(kCommitTso, kCommitTso); read_options.io_ctx.reader_type = ReaderType::READER_QUERY; - read_options.col_id_to_predicates.emplace(1, make_commit_tso_gt_predicate(1, kCommitTso)); + read_options.col_id_to_predicates.emplace( + kCommitTsoCid, make_commit_tso_gt_predicate(kCommitTsoCid, kCommitTso)); std::unique_ptr iter; auto st = segment->new_iterator(read_schema, read_options, &iter); From 6bac554d9be91df5292fa2b9fbc1411fcf0395cd Mon Sep 17 00:00:00 2001 From: TsukiokaKogane Date: Wed, 15 Jul 2026 07:46:33 +0000 Subject: [PATCH 2/2] fix compile --- be/src/storage/segment/segment.cpp | 2 +- be/src/storage/segment/segment_iterator.cpp | 50 --------------------- be/src/storage/segment/segment_iterator.h | 1 - 3 files changed, 1 insertion(+), 52 deletions(-) diff --git a/be/src/storage/segment/segment.cpp b/be/src/storage/segment/segment.cpp index 12a4f837ecc524..0704c5f30f3b4c 100644 --- a/be/src/storage/segment/segment.cpp +++ b/be/src/storage/segment/segment.cpp @@ -113,7 +113,7 @@ std::optional get_read_time_constant_value(int32_t col_uid, const int64_t commit_tso = read_options.commit_tso.end_tso(); if ((reader_type == ReaderType::READER_BINLOG || reader_type == ReaderType::READER_BINLOG_COMPACTION) && - matches_column(tablet_schema.binlog_timestamp_col_idx())) { + matches_column(tablet_schema.binlog_tso_col_idx())) { DCHECK_EQ(read_options.commit_tso.start_tso(), commit_tso); return Field::create_field(commit_tso == -1 ? 0 : commit_tso); } diff --git a/be/src/storage/segment/segment_iterator.cpp b/be/src/storage/segment/segment_iterator.cpp index 574be47571b9f5..0b931a3e3ddca8 100644 --- a/be/src/storage/segment/segment_iterator.cpp +++ b/be/src/storage/segment/segment_iterator.cpp @@ -2421,53 +2421,6 @@ Status SegmentIterator::_read_columns_by_index(uint32_t nrows_read_limit, uint16 return Status::OK(); } -void SegmentIterator::_update_lsn_col_if_needed(const std::vector& column_ids, - size_t num_rows) { - // | commit tso(64) | auto-inc row_id(64) | - if (_opts.version.first != _opts.version.second) { - return; - } - - if (_opts.io_ctx.reader_type != ReaderType::READER_BINLOG && - _opts.io_ctx.reader_type != ReaderType::READER_BINLOG_COMPACTION) { - return; - } - - int32_t lsn_col_idx = _schema->lsn_col_idx(); - if (lsn_col_idx < 0 || std::ranges::find(column_ids, lsn_col_idx) == column_ids.end()) { - return; - } - - DCHECK_EQ(_opts.commit_tso.start_tso(), _opts.commit_tso.end_tso()); - const Int64 commit_tso = _opts.commit_tso.end_tso() == -1 ? 0 : _opts.commit_tso.end_tso(); - - if (_is_pred_column[lsn_col_idx]) { - auto* lsn_column = assert_cast(_current_return_columns[lsn_col_idx].get()); - std::vector binlog_lsns; - binlog_lsns.reserve(num_rows); - for (size_t j = 0; j < num_rows; j++) { - const Int128 row_id = lsn_column->get_data()[j]; - binlog_lsns.emplace_back(make_row_binlog_lsn(commit_tso, row_id)); - } - lsn_column->clear(); - for (const auto& binlog_lsn : binlog_lsns) { - lsn_column->insert_data(reinterpret_cast(&binlog_lsn), 0); - } - return; - } - - auto* lsn_column = assert_cast(_current_return_columns[lsn_col_idx].get()); - const auto* column_desc = _schema->column(lsn_col_idx); - auto column = Schema::get_data_type_ptr(*column_desc)->create_column(); - DCHECK(column_desc->type() == FieldType::OLAP_FIELD_TYPE_LARGEINT); - auto* col_ptr = assert_cast(column.get()); - - for (size_t j = 0; j < num_rows; j++) { - const Int128 row_id = lsn_column->get_element(j); - col_ptr->insert_value(make_row_binlog_lsn(commit_tso, row_id)); - } - _current_return_columns[lsn_col_idx] = std::move(column); -} uint16_t SegmentIterator::_evaluate_vectorization_predicate(uint16_t* sel_rowid_idx, uint16_t selected_size) { @@ -2875,7 +2828,6 @@ Status SegmentIterator::_next_batch_internal(Block* block) { _selected_size = 0; RETURN_IF_ERROR(_read_columns_by_index(nrows_read_limit, _selected_size)); - _update_lsn_col_if_needed(_predicate_column_ids, _selected_size); _opts.stats->blocks_load += 1; _opts.stats->raw_rows_read += _selected_size; @@ -2918,7 +2870,6 @@ Status SegmentIterator::_next_batch_internal(Block* block) { RETURN_IF_ERROR(_read_columns_by_rowids( _common_expr_column_ids, _block_rowids, _sel_rowid_idx.data(), _selected_size, &_current_return_columns, false, true)); - _update_lsn_col_if_needed(_common_expr_column_ids, _selected_size); RETURN_IF_ERROR(_process_columns(_common_expr_column_ids, block)); } @@ -2951,7 +2902,6 @@ Status SegmentIterator::_next_batch_internal(Block* block) { _non_predicate_columns, _block_rowids, _sel_rowid_idx.data(), _selected_size, &_current_return_columns, _opts.condition_cache_digest && !_find_condition_cache, false)); - _update_lsn_col_if_needed(_non_predicate_columns, _selected_size); } else { if (_opts.condition_cache_digest && !_find_condition_cache) { auto& condition_cache = *_condition_cache; diff --git a/be/src/storage/segment/segment_iterator.h b/be/src/storage/segment/segment_iterator.h index eaa131dfbd7b0a..6007187f42a193 100644 --- a/be/src/storage/segment/segment_iterator.h +++ b/be/src/storage/segment/segment_iterator.h @@ -216,7 +216,6 @@ class SegmentIterator : public RowwiseIterator { [[nodiscard]] Status _read_columns(const std::vector& column_ids, MutableColumns& column_block, size_t nrows); [[nodiscard]] Status _read_columns_by_index(uint32_t nrows_read_limit, uint16_t& nrows_read); - void _update_lsn_col_if_needed(const std::vector& column_ids, size_t num_rows); Status _init_current_block(Block* block, std::vector& non_pred_vector, uint32_t nrows_read_limit); uint16_t _evaluate_vectorization_predicate(uint16_t* sel_rowid_idx, uint16_t selected_size);