diff --git a/be/src/exec/scan/olap_scanner.cpp b/be/src/exec/scan/olap_scanner.cpp index 91aa9685f0c824..81840633178fa3 100644 --- a/be/src/exec/scan/olap_scanner.cpp +++ b/be/src/exec/scan/olap_scanner.cpp @@ -478,9 +478,10 @@ Status OlapScanner::_init_tablet_reader_params( // key-ordered merge. They must read every key column, every requested value column, the // binlog meta columns (tso / op) and their __BEFORE__ mirrors. APPEND_ONLY streams rows // as-is and stays on the plain projection paths below. + const bool is_min_delta_scan = + _tablet_reader_params.binlog_scan_type == TBinlogScanType::MIN_DELTA; const bool is_binlog_merge_scan = - _tablet_reader_params.binlog_scan_type == TBinlogScanType::MIN_DELTA || - _tablet_reader_params.binlog_scan_type == TBinlogScanType::DETAIL; + is_min_delta_scan || _tablet_reader_params.binlog_scan_type == TBinlogScanType::DETAIL; if (is_binlog_merge_scan) { for (size_t i = 0; i < tablet_schema->num_key_columns(); ++i) { add_return_column_if_absent(static_cast(i)); @@ -496,16 +497,26 @@ Status OlapScanner::_init_tablet_reader_params( add_return_column_if_absent(static_cast(op_idx)); } - for (auto cid : _return_columns) { - if (cid >= tablet_schema->num_key_columns()) { - const auto& col_name = tablet_schema->column(cid).name(); - std::string before_col_name; - before_col_name.append("__BEFORE__"); - before_col_name.append(col_name); - before_col_name.append("__"); - if (int32_t before_idx = tablet_schema->field_index(before_col_name); - before_idx >= 0) { - add_return_column_if_absent(static_cast(before_idx)); + if (is_min_delta_scan) { + // No-op UPDATE detection compares the complete row state at the two ends of the + // window. Read every AFTER/BEFORE value column even when SQL projects only a subset; + // BlockReader's return-column mapping keeps these comparison-only columns hidden. + for (uint32_t cid = tablet_schema->num_key_columns(); + cid < tablet_schema->num_columns(); ++cid) { + add_return_column_if_absent(cid); + } + } else { + for (auto cid : _return_columns) { + if (cid >= tablet_schema->num_key_columns()) { + const auto& col_name = tablet_schema->column(cid).name(); + std::string before_col_name; + before_col_name.append("__BEFORE__"); + before_col_name.append(col_name); + before_col_name.append("__"); + if (int32_t before_idx = tablet_schema->field_index(before_col_name); + before_idx >= 0) { + add_return_column_if_absent(static_cast(before_idx)); + } } } } diff --git a/be/src/storage/iterator/block_reader.cpp b/be/src/storage/iterator/block_reader.cpp index 16389c84a9ee02..bdd0440af5fae4 100644 --- a/be/src/storage/iterator/block_reader.cpp +++ b/be/src/storage/iterator/block_reader.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -37,6 +38,7 @@ #include "core/column/column_string.h" #include "core/column/column_vector.h" #include "core/data_type/data_type_number.h" +#include "core/data_type/primitive_type.h" #include "exprs/aggregate/aggregate_function_reader.h" #include "exprs/function_filter.h" #include "runtime/runtime_state.h" @@ -60,6 +62,18 @@ using namespace ErrorCode; static constexpr int32_t BLOCK_SIZE_CHECK_INTERVAL_ROWS = 64; +namespace { + +// IColumn::compare_at is not implemented in production for these internal/opaque column +// families. Row binlog currently rejects VARIANT, while the remaining types are kept here as a +// conservative guard so an old or malformed schema cannot turn a MIN_DELTA query into an error. +bool supports_min_delta_value_comparison(PrimitiveType type) { + return !is_var_len_object(type) && type != TYPE_VARIANT && type != TYPE_FIXED_LENGTH_OBJECT && + type != TYPE_BINARY && type != INVALID_TYPE; +} + +} // namespace + BlockReader::~BlockReader() { for (int i = 0; i < _agg_functions.size(); ++i) { _agg_functions[i]->destroy(_agg_places[i]); @@ -78,8 +92,9 @@ Status BlockReader::next_block_with_aggregation(Block* block, bool* eof) { } // Lazily resolves the positions of the binlog meta columns (tso / lsn / op) inside the -// merged source block, and builds _before_column_idx mapping each non-meta column to its -// __BEFORE__ mirror. The resolved positions are reused across blocks; if the column +// merged source block, builds _before_column_idx mapping each non-meta column to its +// __BEFORE__ mirror, and records the complete set of AFTER/BEFORE value pairs used by +// MIN_DELTA equality checks. The resolved positions are reused across blocks; if the column // layout changes (detected via _binlog_op_pos sanity check), they are re-resolved. Status BlockReader::_ensure_binlog_column_pos(const Block& src_block) { if (_binlog_column_pos_inited) { @@ -95,6 +110,10 @@ Status BlockReader::_ensure_binlog_column_pos(const Block& src_block) { const uint32_t col_num = src_block.columns(); _before_column_idx.resize(col_num); + std::iota(_before_column_idx.begin(), _before_column_idx.end(), 0); + std::vector is_before_value_column(col_num, false); + _min_delta_value_column_pairs.clear(); + _min_delta_value_comparison_complete = true; for (uint32_t i = 0; i < col_num; ++i) { const auto& name = src_block.get_by_position(i).name; if (name == BINLOG_TSO_COL) { @@ -106,7 +125,30 @@ Status BlockReader::_ensure_binlog_column_pos(const Block& src_block) { } else { std::string before_name = binlog::build_before_column_name(name); int tmp_idx = src_block.get_position_by_name(before_name); - _before_column_idx[i] = tmp_idx < 0 ? i : tmp_idx; + if (tmp_idx >= 0) { + _before_column_idx[i] = tmp_idx; + is_before_value_column[tmp_idx] = true; + if (i >= _tablet_schema->num_key_columns()) { + _min_delta_value_column_pairs.emplace_back(i, tmp_idx); + } + } + } + } + + // OlapScanner places the full key prefix first for MIN_DELTA/DETAIL scans. Everything after + // that prefix which is neither metadata nor a BEFORE mirror is an AFTER value and must have + // a type-compatible mirror before a no-op UPDATE can be suppressed. + for (uint32_t i = static_cast(_tablet_schema->num_key_columns()); i < col_num; ++i) { + if (_is_binlog_meta_column(i) || is_before_value_column[i]) { + continue; + } + int before_idx = _before_column_idx[i]; + const auto& after = src_block.get_by_position(i); + if (before_idx == static_cast(i) || + !after.type->equals(*src_block.get_by_position(before_idx).type) || + !supports_min_delta_value_comparison(after.type->get_primitive_type())) { + _min_delta_value_comparison_complete = false; + break; } } _binlog_column_pos_inited = true; @@ -165,6 +207,19 @@ int BlockReader::_resolve_source_column_index(int idx, bool use_before) const { return _before_column_idx[idx]; } +bool BlockReader::_min_delta_values_equal(size_t last_row) const { + if (!_min_delta_value_comparison_complete) { + return false; + } + for (const auto& [after_idx, before_idx] : _min_delta_value_column_pairs) { + if (_stored_data_columns[before_idx]->compare_at( + 0, last_row, *_stored_data_columns[after_idx], -1) != 0) { + return false; + } + } + return true; +} + void BlockReader::_init_pending_row_columns(const Block& block) { if (!_pending_row_columns.empty()) { return; @@ -255,6 +310,11 @@ Status BlockReader::_min_delta_next_block(Block* block, bool* eof) { auto first_op = _read_binlog_op(*_stored_data_columns[_binlog_op_pos], 0); auto last_op = _read_binlog_op(*_stored_data_columns[_binlog_op_pos], group_size - 1); auto result = binlog::AggregateFunctionMinDelta::calculate_result(first_op, last_op); + if (result == binlog::AggregateFunctionMinDelta::ResultType::UPDATE_BEFORE_AFTER && + binlog::is_valid_row_binlog_op(first_op) && binlog::is_valid_row_binlog_op(last_op) && + _min_delta_values_equal(group_size - 1)) { + result = binlog::AggregateFunctionMinDelta::ResultType::SKIP; + } switch (result) { case binlog::AggregateFunctionMinDelta::ResultType::SKIP: break; diff --git a/be/src/storage/iterator/block_reader.h b/be/src/storage/iterator/block_reader.h index f1107772b30db6..674ffb67a32891 100644 --- a/be/src/storage/iterator/block_reader.h +++ b/be/src/storage/iterator/block_reader.h @@ -89,6 +89,8 @@ class BlockReader final : public TabletReader { int _resolve_source_column_index(int idx, bool use_before) const; + bool _min_delta_values_equal(size_t last_row) const; + void _init_pending_row_columns(const Block& block); bool _emit_pending_row(MutableColumns& target_columns, size_t& output_row_count); @@ -179,6 +181,13 @@ class BlockReader final : public TabletReader { // column (or itself if no BEFORE mirror exists). Built lazily by _ensure_binlog_column_pos // and consulted via _resolve_source_column_index when emitting BEFORE rows. std::vector _before_column_idx; + // Physical AFTER/BEFORE column pairs used to compare the complete row image for MIN_DELTA. + // These include columns widened into the storage projection solely for comparison and are + // therefore independent of the SQL output projection. + std::vector> _min_delta_value_column_pairs; + // False when the source block does not carry a comparable BEFORE image for every value + // column. In that case MIN_DELTA retains UPDATE output conservatively. + bool _min_delta_value_comparison_complete = false; Arena _arena; }; diff --git a/be/test/storage/iterator/block_reader_change_next_block_test.cpp b/be/test/storage/iterator/block_reader_change_next_block_test.cpp index 0ed73e1287fa0a..7644fad6006705 100644 --- a/be/test/storage/iterator/block_reader_change_next_block_test.cpp +++ b/be/test/storage/iterator/block_reader_change_next_block_test.cpp @@ -36,6 +36,7 @@ #include #include +#include #include #include "common/config.h" @@ -79,6 +80,19 @@ struct Row { int64_t op; }; +TabletSchemaSPtr make_test_tablet_schema() { + auto schema = std::make_shared(); + TabletColumn key_column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, + FieldType::OLAP_FIELD_TYPE_BIGINT, false); + key_column.set_unique_id(0); + key_column.set_name("key"); + key_column.set_is_key(true); + key_column.set_length(sizeof(int64_t)); + key_column.set_index_length(sizeof(int64_t)); + schema->append_column(std::move(key_column)); + return schema; +} + std::shared_ptr make_source_block(const std::vector& rows) { auto block = std::make_shared(); auto type = std::make_shared(); @@ -139,7 +153,7 @@ class FakeLevelIterator : public VCollectIterator::LevelIterator { Status next(Block* /*block*/) override { return Status::Error(""); } - RowLocation current_row_location() override { return RowLocation(); } + RowLocation current_row_location() override { return {}; } Status current_block_row_locations(std::vector* /*loc*/) override { return Status::OK(); } @@ -161,9 +175,8 @@ void configure_reader(BlockReader& reader, std::shared_ptr source, size_t config::enable_adaptive_batch_size = false; reader._reader_context.batch_size = batch_size; - // The fake LevelIterator base ctor dereferences reader->tablet_schema(), so a - // schema must exist even though its contents are unused by these code paths. - reader._tablet_schema = std::make_shared(); + // The fake LevelIterator and MIN_DELTA value-pair discovery both need the key count. + reader._tablet_schema = make_test_tablet_schema(); // All 6 columns are "normal" columns and are returned in-place. reader._normal_columns_idx = {KEY_IDX, VAL_IDX, BEFORE_VAL_IDX, TSO_IDX, LSN_IDX, OP_IDX}; @@ -223,6 +236,16 @@ std::vector drain(BlockReader& reader, Status (BlockReader::*fn)(Block*, return result; } +void expect_out_rows(const std::vector& actual, const std::vector& expected) { + ASSERT_EQ(actual.size(), expected.size()); + for (size_t i = 0; i < expected.size(); ++i) { + SCOPED_TRACE(i); + EXPECT_EQ(actual[i].key, expected[i].key); + EXPECT_EQ(actual[i].val, expected[i].val); + EXPECT_EQ(actual[i].op, expected[i].op); + } +} + } // namespace class BlockReaderChangeNextBlockTest : public testing::Test { @@ -299,6 +322,239 @@ TEST_F(BlockReaderChangeNextBlockTest, MinDeltaUpdateBeforeAfter) { EXPECT_EQ(out[1].val, 30); // after value from the last op } +// A physical UPDATE whose complete BEFORE and AFTER row values are equal has no net delta. +TEST_F(BlockReaderChangeNextBlockTest, MinDeltaNoOpUpdateIsSkipped) { + auto source = make_source_block({ + {1, 20, 20, 1, 1, ROW_BINLOG_UPDATE}, + }); + BlockReader reader; + configure_reader(reader, source, 16); + + auto out = drain(reader, &BlockReader::_min_delta_next_block); + EXPECT_TRUE(out.empty()); +} + +// The comparison is between the first BEFORE and last AFTER values, so A -> B -> A also has no +// net delta even though neither individual row-binlog UPDATE is a no-op. +TEST_F(BlockReaderChangeNextBlockTest, MinDeltaUpdatesReturningToOriginalAreSkipped) { + auto source = make_source_block({ + {1, 20, 10, 1, 1, ROW_BINLOG_UPDATE}, + {1, 10, 20, 2, 2, ROW_BINLOG_UPDATE}, + }); + BlockReader reader; + configure_reader(reader, source, 16); + + auto out = drain(reader, &BlockReader::_min_delta_next_block); + EXPECT_TRUE(out.empty()); +} + +// Exercise long operation chains where intermediate rows repeatedly change existence and value. +// MIN_DELTA must preserve only the net state transition across the whole key window. +TEST_F(BlockReaderChangeNextBlockTest, MinDeltaComplexOperationChains) { + struct TestCase { + std::string_view name; + std::vector rows; + std::vector expected; + size_t batch_size; + }; + const std::vector test_cases = { + { + "insert_delete_reinsert_update_back_delete", + { + {1, 1, 0, 1, 1, ROW_BINLOG_APPEND}, + {1, 1, 1, 2, 2, ROW_BINLOG_DELETE}, + {1, 1, 0, 3, 3, ROW_BINLOG_APPEND}, + {1, 2, 1, 4, 4, ROW_BINLOG_UPDATE}, + {1, 3, 2, 5, 5, ROW_BINLOG_UPDATE}, + {1, 1, 3, 6, 6, ROW_BINLOG_UPDATE}, + {1, 1, 1, 7, 7, ROW_BINLOG_DELETE}, + }, + {}, + 1, + }, + { + "existing_row_delete_reinsert_and_return_to_original", + { + {1, 2, 1, 1, 1, ROW_BINLOG_UPDATE}, + {1, 2, 2, 2, 2, ROW_BINLOG_DELETE}, + {1, 2, 0, 3, 3, ROW_BINLOG_APPEND}, + {1, 3, 2, 4, 4, ROW_BINLOG_UPDATE}, + {1, 1, 3, 5, 5, ROW_BINLOG_UPDATE}, + }, + {}, + 2, + }, + { + "existing_row_delete_reinsert_and_finish_changed", + { + {1, 2, 1, 1, 1, ROW_BINLOG_UPDATE}, + {1, 2, 2, 2, 2, ROW_BINLOG_DELETE}, + {1, 2, 0, 3, 3, ROW_BINLOG_APPEND}, + {1, 3, 2, 4, 4, ROW_BINLOG_UPDATE}, + {1, 4, 3, 5, 5, ROW_BINLOG_UPDATE}, + }, + { + {1, 1, binlog::STREAM_CHANGE_UPDATE_BEFORE}, + {1, 4, binlog::STREAM_CHANGE_UPDATE_AFTER}, + }, + 1, + }, + { + "new_row_temporarily_deleted_but_finishes_present", + { + {1, 1, 0, 1, 1, ROW_BINLOG_APPEND}, + {1, 2, 1, 2, 2, ROW_BINLOG_UPDATE}, + {1, 2, 2, 3, 3, ROW_BINLOG_DELETE}, + {1, 5, 0, 4, 4, ROW_BINLOG_APPEND}, + {1, 6, 5, 5, 5, ROW_BINLOG_UPDATE}, + }, + { + {1, 6, binlog::STREAM_CHANGE_INSERT}, + }, + 1, + }, + { + "existing_row_temporarily_reinserted_but_finishes_deleted", + { + {1, 2, 1, 1, 1, ROW_BINLOG_UPDATE}, + {1, 2, 2, 2, 2, ROW_BINLOG_DELETE}, + {1, 3, 0, 3, 3, ROW_BINLOG_APPEND}, + {1, 4, 3, 4, 4, ROW_BINLOG_UPDATE}, + {1, 4, 4, 5, 5, ROW_BINLOG_DELETE}, + }, + { + {1, 1, binlog::STREAM_CHANGE_DELETE}, + }, + 1, + }, + { + "delete_reinsert_update_and_return_to_original", + { + {1, 1, 1, 1, 1, ROW_BINLOG_DELETE}, + {1, 1, 0, 2, 2, ROW_BINLOG_APPEND}, + {1, 2, 1, 3, 3, ROW_BINLOG_UPDATE}, + {1, 1, 2, 4, 4, ROW_BINLOG_UPDATE}, + }, + {}, + 1, + }, + }; + + for (const auto& test_case : test_cases) { + SCOPED_TRACE(test_case.name); + auto source = make_source_block(test_case.rows); + BlockReader reader; + configure_reader(reader, source, test_case.batch_size); + + auto out = drain(reader, &BlockReader::_min_delta_next_block); + expect_out_rows(out, test_case.expected); + } +} + +// Build a source block with a second value column that is present in the physical MIN_DELTA +// projection but absent from the SQL output projection. +struct TwoValueRow { + int64_t key; + int64_t val1; + int64_t val2; + int64_t before_val1; + int64_t before_val2; + int64_t tso; + int64_t lsn; + int64_t op; +}; + +std::shared_ptr make_two_value_source_block(const std::vector& rows) { + auto block = std::make_shared(); + auto type = std::make_shared(); + auto key_col = ColumnInt64::create(); + auto val1_col = ColumnInt64::create(); + auto val2_col = ColumnInt64::create(); + auto before_val1_col = ColumnInt64::create(); + auto before_val2_col = ColumnInt64::create(); + auto tso_col = ColumnInt64::create(); + auto lsn_col = ColumnInt64::create(); + auto op_col = ColumnInt64::create(); + for (const auto& row : rows) { + key_col->insert_value(row.key); + val1_col->insert_value(row.val1); + val2_col->insert_value(row.val2); + before_val1_col->insert_value(row.before_val1); + before_val2_col->insert_value(row.before_val2); + tso_col->insert_value(row.tso); + lsn_col->insert_value(row.lsn); + op_col->insert_value(row.op); + } + block->insert({std::move(key_col), type, "key"}); + block->insert({std::move(val1_col), type, "val"}); + block->insert({std::move(val2_col), type, "val2"}); + block->insert({std::move(before_val1_col), type, binlog::build_before_column_name("val")}); + block->insert({std::move(before_val2_col), type, binlog::build_before_column_name("val2")}); + block->insert({std::move(tso_col), type, BINLOG_TSO_COL}); + block->insert({std::move(lsn_col), type, BINLOG_LSN_COL}); + block->insert({std::move(op_col), type, BINLOG_OP_COL}); + return block; +} + +void configure_two_value_reader(BlockReader& reader, std::shared_ptr source, + size_t batch_size = 16) { + configure_reader(reader, source, batch_size); + // Return key/val1/before-val1/meta only. val2 and before-val2 remain available internally + // at physical positions 2 and 4, with no target output position. + reader._normal_columns_idx = {0, 1, 3, 5, 6, 7}; + reader._return_columns_loc = {0, 1, -1, 2, -1, 3, 4, 5}; +} + +TEST_F(BlockReaderChangeNextBlockTest, MinDeltaNoOpUpdateComparesAllValueColumns) { + auto source = make_two_value_source_block({ + {/*key=*/1, /*val1=*/20, /*val2=*/30, /*before_val1=*/20, /*before_val2=*/30, + /*tso=*/1, /*lsn=*/1, ROW_BINLOG_UPDATE}, + }); + BlockReader reader; + configure_two_value_reader(reader, source); + + auto out = drain(reader, &BlockReader::_min_delta_next_block); + EXPECT_TRUE(out.empty()); +} + +TEST_F(BlockReaderChangeNextBlockTest, MinDeltaRetainsChangeInUnprojectedValueColumn) { + auto source = make_two_value_source_block({ + {/*key=*/1, /*val1=*/20, /*val2=*/31, /*before_val1=*/20, /*before_val2=*/30, + /*tso=*/1, /*lsn=*/1, ROW_BINLOG_UPDATE}, + }); + BlockReader reader; + configure_two_value_reader(reader, source); + + auto out = drain(reader, &BlockReader::_min_delta_next_block); + ASSERT_EQ(out.size(), 2); + EXPECT_EQ(out[0].op, binlog::STREAM_CHANGE_UPDATE_BEFORE); + EXPECT_EQ(out[1].op, binlog::STREAM_CHANGE_UPDATE_AFTER); + EXPECT_EQ(out[0].val, 20); + EXPECT_EQ(out[1].val, 20); +} + +// key 1 changes both columns several times and returns to its complete original row image, so it +// disappears. key 2 returns only the projected value column to its original value while the hidden +// value column remains changed, so its UPDATE pair must survive. batch_size=1 also forces the pair +// through the pending-row path after the skipped key. +TEST_F(BlockReaderChangeNextBlockTest, MinDeltaComplexMultiColumnChains) { + auto source = make_two_value_source_block({ + {1, 11, 100, 10, 100, 1, 1, ROW_BINLOG_UPDATE}, + {1, 11, 101, 11, 100, 2, 2, ROW_BINLOG_UPDATE}, + {1, 10, 100, 11, 101, 3, 3, ROW_BINLOG_UPDATE}, + {2, 21, 200, 20, 200, 4, 4, ROW_BINLOG_UPDATE}, + {2, 20, 201, 21, 200, 5, 5, ROW_BINLOG_UPDATE}, + }); + BlockReader reader; + configure_two_value_reader(reader, source, /*batch_size=*/1); + + auto out = drain(reader, &BlockReader::_min_delta_next_block); + expect_out_rows(out, { + {2, 20, binlog::STREAM_CHANGE_UPDATE_BEFORE}, + {2, 20, binlog::STREAM_CHANGE_UPDATE_AFTER}, + }); +} + // Multiple distinct keys, each in its own group, are folded independently. TEST_F(BlockReaderChangeNextBlockTest, MinDeltaMultipleKeys) { auto source = make_source_block({