Skip to content

[fix](be) Suppress no-op updates in min delta scans #66849

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
gavinchou wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[fix](be) Suppress no-op updates in min delta scans #66849

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
50e3ba3
[fix](be) Suppress no-op updates in min delta scans
gavinchou Aug 17, 2026
56a1341
[test](be) Cover complex min delta operation chains
gavinchou Aug 17, 2026
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
35 changes: 23 additions & 12 deletions be/src/exec/scan/olap_scanner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<uint32_t>(i));
Expand All @@ -496,16 +497,26 @@ Status OlapScanner::_init_tablet_reader_params(
add_return_column_if_absent(static_cast<uint32_t>(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<uint32_t>(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();

@github-actions github-actions Bot Aug 17, 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] Preflight unsupported types before widening the projection. If any value is BITMAP, HLL, QUANTILE_STATE, or another rejected shape, BlockReader permanently sets comparison completeness to false, so this optimization can never suppress a group. This loop nevertheless reads every unrequested AFTER and BEFORE value on every block; even a key-only MIN_DELTA query on a BITMAP table now pays the large extra I/O for no behavior change. When complete comparison is statically impossible, preserve the old narrow projection (and keep the runtime check conservative).

cid < tablet_schema->num_columns(); ++cid) {
add_return_column_if_absent(cid);

@github-actions github-actions Bot Aug 17, 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] Build this full projection without a linear search per column. add_return_column_if_absent() runs std::find over the growing return_columns vector, and this new loop calls it for all C physical binlog columns, so each tablet/range scanner now does roughly C^2/2 integer comparisons even for a narrow query. That is separate from the quadratic name lookup later in BlockReader. Please track selected dense column ids with a bitmap/set or append the remaining schema columns in one linear pass while preserving the required output order.

}
} 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<uint32_t>(before_idx));
}
}
}
}
Expand Down
66 changes: 63 additions & 3 deletions be/src/storage/iterator/block_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <algorithm>
#include <boost/iterator/iterator_facade.hpp>
#include <memory>
#include <numeric>
#include <ostream>
#include <string>

Expand All @@ -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"
Expand All @@ -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 &&

@github-actions github-actions Bot Aug 17, 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] Guard the concrete AGG_STATE representation before comparing. TYPE_AGG_STATE passes this allowlist, but DataTypeAggState creates aggregate-specific serialized columns: fixed states such as sum/count use ColumnFixedLengthObject (no compare_at override), while bitmap-style states use ColumnComplexType whose override is BE_TEST-only. Row-binlog schema construction permits AGG_STATE and wraps value cells nullable, so a non-NULL update reaches _min_delta_values_equal() and throws NOT_IMPLEMENTED_ERROR instead of conservatively retaining the UPDATE. Please inspect the actual serialized column capability or exclude AGG_STATE here.

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]);
Expand All @@ -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) {
Expand All @@ -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<bool> is_before_value_column(col_num, false);

@github-actions github-actions Bot Aug 17, 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] Build the name map once before resolving mirrors. MIN_DELTA now feeds every AFTER and BEFORE column into this block, but the loop below calls Block::get_position_by_name()--an O(C) scan--for each physical column. A table with V values has roughly 2V value/mirror columns, so every BlockReader performs O(C^2) string comparisons, repeated per tablet scanner. Please use Block::get_name_to_pos_map() once here and resolve every BEFORE name from it.

_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) {
Expand All @@ -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) {

@github-actions github-actions Bot Aug 17, 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] Pair BEFORE values by schema identity, not first matching name. Doris permits a user value named BEFORE__v, while row-binlog generation also gives v's mirror that name. Block::get_position_by_name() selects the earlier user AFTER column, so this loop builds a complete-looking but wrong graph. For two same-typed values v and BEFORE__v, updates (0,1)->(1,2)->(2,1) make all three erroneous comparisons equal and this PR emits SKIP even though v changed from 0 to 2. Please disambiguate generated names or resolve pairs by unique id/ordinal, and cover this collision end to end.

_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<uint32_t>(_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<int>(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;
Expand Down Expand Up @@ -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(

@github-actions github-actions Bot Aug 17, 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] Do not suppress an observable signed-zero update. FLOAT and DOUBLE pass the new gate, but CompareFloat (used by ColumnVector::compare_at) returns 0 for +0.0 versus -0.0. Doris preserves and displays both signs, and signbit(+0.0) / signbit(-0.0) returns false / true, so a valid MOW update between them changes row state. This comparison turns that update into SKIP and can leave an incremental consumer's derived state stale; nested floats inherit the same behavior. Please use state equality that preserves signed zero, or conservatively exclude floating shapes, and add an end-to-end MIN_DELTA test.

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;
Expand Down Expand Up @@ -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)) {

@github-actions github-actions Bot Aug 17, 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] Do not treat an unavailable BEFORE image as a real all-NULL row. build_before_block() fills every BEFORE value with NULL when no historical row exists. A delete sign for a missing key is still recorded as DELETE, and a later reinsertion after that tombstone becomes APPEND. If all value columns are NULL, DELETE-to-APPEND first yields UPDATE_BEFORE_AFTER, then this gate sees NULL equal to NULL and drops the net absent-to-present INSERT as SKIP. Compatibility placeholder BEFORE rows have the same ambiguity. Only suppress when the first BEFORE image is known valid, or conservatively retain ambiguous all-NULL images.

result = binlog::AggregateFunctionMinDelta::ResultType::SKIP;
}
switch (result) {
case binlog::AggregateFunctionMinDelta::ResultType::SKIP:
break;
Expand Down
9 changes: 9 additions & 0 deletions be/src/storage/iterator/block_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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<int> _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<std::pair<int, int>> _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;
};

Expand Down
Loading
Loading