This is an automated email from the ASF dual-hosted git repository. mrhhsg pushed a commit to branch adaptive_batch_size in repository https://gitbox.apache.org/repos/asf/doris.git
commit e88ad3a5cd54226411e0496b4ed9e3687860950d Author: Hu Shenggang <[email protected]> AuthorDate: Tue Apr 7 18:05:59 2026 +0800 [fix](be) Fix adaptive batch size review issues ### What problem does this PR solve? Issue Number: close #N/A Related PR: #61535 Problem Summary: Address code review findings on the adaptive batch size feature: 1. `RuntimeState::block_max_rows()` now gates on `preferred_block_size_bytes() > 0`, so `SET preferred_block_size_bytes = 0` correctly restores legacy fixed-row behavior. Also fixes the `std::max` semantics that turned `preferred_block_size_rows` into a lower bound instead of upper bound, and fixes `size_t` to `int` cast overflow risk. 2. `TabletReader::batch_max_rows()` now also gates on `preferred_block_size_bytes > 0`. 3. `BetaRowsetReader` now seeds `block_row_max` from `preferred_block_size_rows` when adaptive sizing is active, so the OLAP scan path can grow beyond the legacy `batch_size` (1024) for narrow tables. `SegmentIterator::_initial_block_row_max` then correctly reflects the larger cap. 4. `VSortedRunMerger` single-run fast path now falls through to the general merge path when byte budget is active, preventing oversized output blocks. 5. Regression test: renamed `qt_` to `order_qt_` prefix per standard, removed post-test `drop table` statements (keep drop-before-create pattern per standard). 6. Unit tests updated to reflect the new gating semantics, including a new test case for the `preferred_block_size_bytes=0` fallback. ### Release note None ### Check List (For Author) - Test: Unit Test (71+110 tests passed including all adaptive batch size, sort merger, block reader, collect iterator, and operator tests) - Behavior changed: Yes - `preferred_block_size_bytes = 0` now correctly disables adaptive row cap expansion (previously still used preferred_block_size_rows=65535) - Does this need documentation: No Co-authored-by: Copilot <[email protected]> --- be/src/exec/sort/vsorted_run_merger.cpp | 5 ++++- be/src/runtime/runtime_state.h | 8 +++++--- be/src/storage/rowset/beta_rowset_reader.cpp | 10 +++++++++- be/src/storage/tablet/tablet_reader.h | 7 ++++--- .../iterator/block_reader_batch_max_rows_test.cpp | 16 ++++++++++++++++ .../adaptive_batch_size/adaptive_batch_size.out | 10 +++++----- .../adaptive_batch_size/adaptive_batch_size.groovy | 18 +++++------------- 7 files changed, 48 insertions(+), 26 deletions(-) diff --git a/be/src/exec/sort/vsorted_run_merger.cpp b/be/src/exec/sort/vsorted_run_merger.cpp index 91803716f08..88e139d201a 100644 --- a/be/src/exec/sort/vsorted_run_merger.cpp +++ b/be/src/exec/sort/vsorted_run_merger.cpp @@ -118,7 +118,10 @@ Status VSortedRunMerger::get_next(Block* output_block, bool* eos) { if (_priority_queue.empty()) { *eos = true; return Status::OK(); - } else if (_priority_queue.size() == 1) { + } else if (_priority_queue.size() == 1 && _block_max_bytes == 0) { + // Single-run fast path: swap the entire remaining supplier block into output. + // When byte budget is active, fall through to the general merge path which + // already respects _block_max_bytes. auto current = _priority_queue.top(); DCHECK(!current->eof()); DCHECK(current->block_ptr() != nullptr); diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h index f4f1740b482..d93998da69c 100644 --- a/be/src/runtime/runtime_state.h +++ b/be/src/runtime/runtime_state.h @@ -140,9 +140,11 @@ public: void set_desc_tbl(const DescriptorTbl* desc_tbl) { _desc_tbl = desc_tbl; } MOCK_FUNCTION int block_max_rows() const { - return config::enable_adaptive_batch_size - ? std::max(_query_options.batch_size, int(preferred_block_size_rows())) - : _query_options.batch_size; + if (config::enable_adaptive_batch_size && preferred_block_size_bytes() > 0) { + auto rows = preferred_block_size_rows(); + return static_cast<int>(std::min(rows, static_cast<size_t>(INT32_MAX))); + } + return _query_options.batch_size; } MOCK_FUNCTION size_t block_max_bytes() const { diff --git a/be/src/storage/rowset/beta_rowset_reader.cpp b/be/src/storage/rowset/beta_rowset_reader.cpp index 2e3196c14be..8d147c52a1c 100644 --- a/be/src/storage/rowset/beta_rowset_reader.cpp +++ b/be/src/storage/rowset/beta_rowset_reader.cpp @@ -97,7 +97,15 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context RETURN_IF_ERROR(_rowset->load()); // convert RowsetReaderContext to StorageReadOptions - _read_options.block_row_max = read_context->batch_size; + // When adaptive batch size is enabled and the byte budget is active, use the larger + // preferred_block_size_rows so the predictor can grow beyond the legacy batch_size. + if (config::enable_adaptive_batch_size && read_context->preferred_block_size_bytes > 0 && + read_context->preferred_block_size_rows > 0) { + _read_options.block_row_max = + cast_set<uint32_t>(read_context->preferred_block_size_rows); + } else { + _read_options.block_row_max = read_context->batch_size; + } _read_options.preferred_block_size_bytes = read_context->preferred_block_size_bytes; _read_options.preferred_block_size_rows = read_context->preferred_block_size_rows; _read_options.preferred_max_col_bytes = read_context->preferred_max_col_bytes; diff --git a/be/src/storage/tablet/tablet_reader.h b/be/src/storage/tablet/tablet_reader.h index 66dc90a580a..ef437bea00b 100644 --- a/be/src/storage/tablet/tablet_reader.h +++ b/be/src/storage/tablet/tablet_reader.h @@ -253,10 +253,11 @@ public: int batch_size() const { return _reader_context.batch_size; } // Return the effective maximum row count for output blocks. - // When adaptive batch sizing is enabled, prefer preferred_block_size_rows; - // otherwise fall back to the traditional batch_size. + // When adaptive batch sizing is enabled and the byte budget is active, + // prefer preferred_block_size_rows; otherwise fall back to the traditional batch_size. size_t batch_max_rows() const { - if (config::enable_adaptive_batch_size && _reader_context.preferred_block_size_rows > 0) { + if (config::enable_adaptive_batch_size && _reader_context.preferred_block_size_bytes > 0 && + _reader_context.preferred_block_size_rows > 0) { return _reader_context.preferred_block_size_rows; } return _reader_context.batch_size; diff --git a/be/test/storage/iterator/block_reader_batch_max_rows_test.cpp b/be/test/storage/iterator/block_reader_batch_max_rows_test.cpp index cebcceb6895..b93588fd11c 100644 --- a/be/test/storage/iterator/block_reader_batch_max_rows_test.cpp +++ b/be/test/storage/iterator/block_reader_batch_max_rows_test.cpp @@ -61,6 +61,7 @@ TEST_F(BlockReaderBatchMaxRowsTest, UsePreferredRowsWhenAdaptiveEnabled) { BlockReader reader; reader._reader_context.batch_size = 4096; reader._reader_context.preferred_block_size_rows = 8192; + reader._reader_context.preferred_block_size_bytes = 8388608; // byte budget must be active EXPECT_EQ(reader.batch_max_rows(), 8192); } @@ -71,7 +72,21 @@ TEST_F(BlockReaderBatchMaxRowsTest, FallbackWhenPreferredRowsIsZero) { BlockReader reader; reader._reader_context.batch_size = 4096; reader._reader_context.preferred_block_size_rows = 0; + reader._reader_context.preferred_block_size_bytes = 8388608; + + EXPECT_EQ(reader.batch_max_rows(), 4096); +} + +TEST_F(BlockReaderBatchMaxRowsTest, FallbackWhenByteBudgetIsZero) { + config::enable_adaptive_batch_size = true; + + BlockReader reader; + reader._reader_context.batch_size = 4096; + reader._reader_context.preferred_block_size_rows = 65535; + reader._reader_context.preferred_block_size_bytes = 0; + // When byte budget is zero, batch_max_rows falls back to batch_size + // regardless of preferred_block_size_rows. EXPECT_EQ(reader.batch_max_rows(), 4096); } @@ -80,6 +95,7 @@ TEST_F(BlockReaderBatchMaxRowsTest, DefaultPreferredRows) { BlockReader reader; reader._reader_context.batch_size = 4096; + reader._reader_context.preferred_block_size_bytes = 8388608; // byte budget must be active // default preferred_block_size_rows is 65535 EXPECT_EQ(reader.batch_max_rows(), 65535); diff --git a/regression-test/data/query_p0/adaptive_batch_size/adaptive_batch_size.out b/regression-test/data/query_p0/adaptive_batch_size/adaptive_batch_size.out index 5079227ce95..058b15b83f8 100644 --- a/regression-test/data/query_p0/adaptive_batch_size/adaptive_batch_size.out +++ b/regression-test/data/query_p0/adaptive_batch_size/adaptive_batch_size.out @@ -1,5 +1,5 @@ -- This file is automatically generated. You should know what you did if you want to edit this --- !wide -- +-- !order_wide -- 1 1000 1000 1000 2 1000 1000 1000 3 1000 1000 1000 @@ -51,10 +51,10 @@ 49 1000 1000 1000 50 1000 1000 1000 --- !narrow -- +-- !order_narrow -- 24995000 37492500 49990000 --- !agg -- +-- !order_agg -- 1 3 2 6 3 9 @@ -66,9 +66,9 @@ 9 27 10 30 --- !unique -- +-- !order_unique -- 3000 4498500 --- !flag -- +-- !order_flag -- 4950 diff --git a/regression-test/suites/query_p0/adaptive_batch_size/adaptive_batch_size.groovy b/regression-test/suites/query_p0/adaptive_batch_size/adaptive_batch_size.groovy index 19be1e62ca6..56b342b585f 100644 --- a/regression-test/suites/query_p0/adaptive_batch_size/adaptive_batch_size.groovy +++ b/regression-test/suites/query_p0/adaptive_batch_size/adaptive_batch_size.groovy @@ -73,7 +73,7 @@ suite("adaptive_batch_size") { set_adaptive(true) def res_enabled = sql "select id, length(c1) as l1, length(c2) as l2, length(c3) as l3 from abs_wide_table order by 1, 2, 3, 4" - qt_wide "select id, length(c1) as l1, length(c2) as l2, length(c3) as l3 from abs_wide_table order by 1, 2, 3, 4 limit 50" + order_qt_wide "select id, length(c1) as l1, length(c2) as l2, length(c3) as l3 from abs_wide_table order by 1, 2, 3, 4 limit 50" // Run query with adaptive disabled and collect result. set_adaptive(false) @@ -111,15 +111,13 @@ suite("adaptive_batch_size") { set_adaptive(true) def narrow_on = sql "select sum(c1), sum(c2), sum(c3) from abs_narrow_table" - qt_narrow "select sum(c1), sum(c2), sum(c3) from abs_narrow_table" + order_qt_narrow "select sum(c1), sum(c2), sum(c3) from abs_narrow_table" set_adaptive(false) def narrow_off = sql "select sum(c1), sum(c2), sum(c3) from abs_narrow_table" assertEquals(narrow_on.toString(), narrow_off.toString()) - sql "drop table abs_narrow_table" - // ── Test 3: AGG_KEYS table ──────────────────────────────────────────────── // Verifies that adaptive batch size does not break aggregation correctness @@ -149,15 +147,13 @@ suite("adaptive_batch_size") { set_adaptive(true) def agg_on = sql "select id, val from abs_agg_table order by 1, 2 limit 10" - qt_agg "select id, val from abs_agg_table order by 1, 2 limit 10" + order_qt_agg "select id, val from abs_agg_table order by 1, 2 limit 10" set_adaptive(false) def agg_off = sql "select id, val from abs_agg_table order by 1, 2 limit 10" assertEquals(agg_on.toString(), agg_off.toString()) - sql "drop table abs_agg_table" - // ── Test 4: UNIQUE_KEYS table ───────────────────────────────────────────── // Verifies that adaptive byte-stop in _unique_key_next_block does not @@ -180,15 +176,13 @@ suite("adaptive_batch_size") { set_adaptive(true) def uniq_on = sql "select count(*), sum(id) from abs_unique_table" - qt_unique "select count(*), sum(id) from abs_unique_table" + order_qt_unique "select count(*), sum(id) from abs_unique_table" set_adaptive(false) def uniq_off = sql "select count(*), sum(id) from abs_unique_table" assertEquals(uniq_on.toString(), uniq_off.toString()) - sql "drop table abs_unique_table" - // ── Test 5: verify setting preferred_block_size_bytes = 0 disables feature ─ @@ -205,15 +199,13 @@ suite("adaptive_batch_size") { sql "set preferred_block_size_bytes = 0" def flag_off = sql "select sum(v) from abs_flag_table" - qt_flag "select sum(v) from abs_flag_table" + order_qt_flag "select sum(v) from abs_flag_table" sql "set preferred_block_size_bytes = 8388608" def flag_on = sql "select sum(v) from abs_flag_table" assertEquals(flag_off.toString(), flag_on.toString()) - sql "drop table abs_flag_table" - // Reset session variables to defaults. sql "set preferred_block_size_bytes = 8388608" sql "set preferred_max_column_in_block_size_bytes = 1048576" --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
