This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 298905410dd [fix](be) Backport batched rowid reads for lazy
materialization to 4.1 (#69331)
298905410dd is described below
commit 298905410dd5d20819dbe7fe14ced507a3da929f
Author: Mryange <[email protected]>
AuthorDate: Thu Oct 22 11:37:11 2026 +0800
[fix](be) Backport batched rowid reads for lazy materialization to 4.1
(#69331)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #63436
Problem Summary:
Backport #63436 to branch-4.1. TopN lazy materialization can hang when a
join produces duplicate or out-of-order physical rowids. Root cause:
reusing a column iterator for these requests can underflow the RLE
decoder's unsigned skip count. Group rowids by segment, sort and
deduplicate them before batch reads, then scatter the results back to
preserve request order and duplicate rows.
The backport preserves branch-4.1 COW ownership, Variant V2 handling,
rowid range checks, file cache policy, and existing external row-fetch
fixes. The existing Variant segment-reader test is adapted to the batch
interface. No additional prerequisite PRs are included.
### Release note
Fix TopN lazy materialization hanging on duplicate or out-of-order
rowids from joins.
### Check List (For Author)
- Test:
- [ ] Regression test
- [ ] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- BE ASAN clean build passed: `./build.sh --be --clean -j48`.
- clang-format 16 and `git diff --check` passed.
- Unit tests and SQL regression tests have not been run.
- Behavior changed:
- [ ] No.
- [x] Yes. Duplicate and out-of-order rowids are normalized before
storage reads; output order and duplicates are preserved.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
Co-authored-by: HappenLee <[email protected]>
---
be/src/exec/rowid_fetcher.cpp | 115 +++++++++++++++------
be/src/service/point_query_executor.cpp | 5 +-
be/src/storage/segment/segment.cpp | 17 ++-
be/src/storage/segment/segment.h | 5 +-
.../variant/variant_column_writer_reader_test.cpp | 7 +-
5 files changed, 105 insertions(+), 44 deletions(-)
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index 0219ac0b778..5fc288dd301 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -378,6 +378,31 @@ struct SegItem {
SegmentSharedPtr segment;
};
+// Groups all row_ids belonging to the same segment for batched reading.
+// Position index tracks where each row_id originated in the original request,
+// so results can be scattered back to the correct output positions.
+struct DorisFormatReadBatch {
+ std::shared_ptr<FileMapping> file_mapping;
+ // (row_id, index_in_request) pairs for all rows in this segment.
+ std::vector<std::pair<segment_v2::rowid_t, size_t>> row_ids_with_positions;
+};
+
+static void scatter_scan_blocks_to_result_block(
+ const std::vector<std::pair<size_t, size_t>>& row_id_block_idx,
+ std::vector<Block>& scan_blocks, Block& result_block) {
+ for (size_t column_id = 0; column_id < result_block.columns();
++column_id) {
+ auto column_guard = result_block.mutate_column_scoped(column_id);
+ MutableColumnPtr& dst_col = column_guard.mutable_column();
+ for (const auto& [pos_block, block_idx] : row_id_block_idx) {
+ DCHECK(scan_blocks.size() > pos_block);
+ DCHECK(scan_blocks[pos_block].columns() > column_id);
+ const auto& src_col =
*scan_blocks[pos_block].get_by_position(column_id).column;
+ uint32_t scan_position = cast_set<uint32_t>(block_idx);
+ dst_col->insert_indices_from(src_col, &scan_position,
&scan_position + 1);
+ }
+ }
+}
+
Status RowIdStorageReader::read_by_rowids(const PMultiGetRequest& request,
PMultiGetResponse* response) {
// read from storage engine row id by row id
@@ -481,7 +506,8 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequest& request,
row_location.row_location.segment_id,
row_location.row_location.row_id);
for (int x = 0; x < slots.size(); ++x) {
- auto row_id =
static_cast<segment_v2::rowid_t>(row_loc.ordinal_id());
+ std::vector<segment_v2::rowid_t> row_ids {
+ static_cast<segment_v2::rowid_t>(row_loc.ordinal_id())};
// The scoped mutation republishes a potentially replaced V2
column when the reader
// transfers ownership of its first assembled batch.
auto column_guard = result_block.mutate_column_scoped(x);
@@ -500,8 +526,8 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequest& request,
segment = iterator_item.segment;
set_slot_access_paths(slots[x], full_read_schema,
iterator_item.storage_read_options);
RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
- full_read_schema, &slots[x], row_id, column,
iterator_item.storage_read_options,
- iterator_item.iterator));
+ full_read_schema, &slots[x], row_ids, column,
+ iterator_item.storage_read_options,
iterator_item.iterator));
}
}
// serialize block if not empty
@@ -705,45 +731,76 @@ Status RowIdStorageReader::read_batch_doris_format_row(
}
}
- std::vector<uint32_t> row_ids;
- int k = 1;
- auto max_k = 0;
- for (int j = 0; j < request_block_desc.row_id_size();) {
+ // Phase 1: Group all row_ids by their (tablet_id, rowset_id, segment_id)
key.
+ // Unlike the old code which only batched adjacent rows with the same
file_id,
+ // this merges non-contiguous same-segment requests into a single batch,
+ // maximizing the number of rows read per seek_and_read_by_rowid call.
+ std::vector<DorisFormatReadBatch> scan_batches;
+ std::unordered_map<SegKey, size_t, HashOfSegKey> batch_idx_by_seg;
+ // (batch_idx, position_in_batch) for each row in the original request.
+ std::vector<std::pair<size_t, size_t>>
row_id_block_idx(request_block_desc.row_id_size());
+ for (int j = 0; j < request_block_desc.row_id_size(); ++j) {
auto file_id = request_block_desc.file_id(j);
const auto row_id = request_block_desc.row_id(j);
if (row_id > std::numeric_limits<uint32_t>::max()) {
return Status::InvalidArgument("internal row id exceeds uint32
range: {}", row_id);
}
- row_ids.emplace_back(static_cast<uint32_t>(row_id));
auto file_mapping = id_file_map->get_file_mapping(file_id);
if (!file_mapping) {
return Status::InternalError(
"Backend:{} file_mapping not found, query_id: {}, file_id:
{}",
BackendOptions::get_localhost(), print_id(query_id),
file_id);
}
- for (k = 1; j + k < request_block_desc.row_id_size(); ++k) {
- if (request_block_desc.file_id(j + k) == file_id) {
- const auto next_row_id = request_block_desc.row_id(j + k);
- if (next_row_id > std::numeric_limits<uint32_t>::max()) {
- return Status::InvalidArgument("internal row id exceeds
uint32 range: {}",
- next_row_id);
- }
- row_ids.emplace_back(static_cast<uint32_t>(next_row_id));
- } else {
- break;
+
+ // Derive segment key and group by it — rows from the same segment are
batched together
+ // even if they are interleaved with rows from other segments in the
request.
+ auto [tablet_id, rowset_id, segment_id] =
file_mapping->get_doris_format_info();
+ SegKey seg_key {.tablet_id = tablet_id, .rowset_id = rowset_id,
.segment_id = segment_id};
+ auto [it, inserted] = batch_idx_by_seg.emplace(seg_key,
scan_batches.size());
+ if (inserted) {
+ // First time seeing this segment, create a new batch for it.
+ scan_batches.emplace_back();
+ scan_batches.back().file_mapping = file_mapping;
+ }
+ // Record (row_id, original_request_index) for later sorting and
scattering.
+
scan_batches[it->second].row_ids_with_positions.emplace_back(static_cast<uint32_t>(row_id),
+ j);
+ }
+
+ // Phase 2: For each segment, sort row_ids ascending (required by
ColumnIterator),
+ // deduplicate, then read all rows in a single batch call.
+ std::vector<Block> scan_blocks(scan_batches.size());
+ for (size_t batch_idx = 0; batch_idx < scan_batches.size(); ++batch_idx) {
+ auto& scan_batch = scan_batches[batch_idx];
+ auto& row_ids_with_positions = scan_batch.row_ids_with_positions;
+ std::sort(row_ids_with_positions.begin(), row_ids_with_positions.end(),
+ [](const auto& lhs, const auto& rhs) { return lhs.first <
rhs.first; });
+
+ // Column iterators read rowids monotonically. Deduplicate consecutive
identical row_ids
+ // (different file_ids may map to the same row), then scatter rows
back to their original
+ // request positions.
+ std::vector<uint32_t> row_ids;
+ row_ids.reserve(row_ids_with_positions.size());
+
+ // Also builds the scatter map: row_id_block_idx[original_request_idx]
->
+ // (batch_idx, deduplicated_position_in_batch).
+ for (const auto& [row_id, result_idx] : row_ids_with_positions) {
+ if (row_ids.empty() || row_ids.back() != row_id) {
+ row_ids.emplace_back(row_id);
}
+ row_id_block_idx[result_idx] = std::make_pair(batch_idx,
row_ids.size() - 1);
}
+ scan_blocks[batch_idx] = Block(slots, row_ids.size());
RETURN_IF_ERROR(read_doris_format_row(
- id_file_map, file_mapping, row_ids, slots, full_read_schema,
row_store_read_struct,
- stats, acquire_tablet_ms, acquire_rowsets_ms,
acquire_segments_ms,
- lookup_row_data_ms, seg_map, iterator_map,
file_cache_miss_policy, result_block));
-
- j += k;
- max_k = std::max(max_k, k);
- row_ids.clear();
+ id_file_map, scan_batch.file_mapping, row_ids, slots,
full_read_schema,
+ row_store_read_struct, stats, acquire_tablet_ms,
acquire_rowsets_ms,
+ acquire_segments_ms, lookup_row_data_ms, seg_map, iterator_map,
+ file_cache_miss_policy, scan_blocks[batch_idx]));
}
+ scatter_scan_blocks_to_result_block(row_id_block_idx, scan_blocks,
result_block);
+
return Status::OK();
}
@@ -1509,11 +1566,9 @@ Status RowIdStorageReader::read_doris_format_row(
file_cache_miss_policy;
}
set_slot_access_paths(slots[x], full_read_schema,
iterator_item.storage_read_options);
- for (auto row_id : row_ids) {
- RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
- full_read_schema, &slots[x], row_id, column,
- iterator_item.storage_read_options,
iterator_item.iterator));
- }
+ RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
+ full_read_schema, &slots[x], row_ids, column,
+ iterator_item.storage_read_options,
iterator_item.iterator));
}
}
return Status::OK();
diff --git a/be/src/service/point_query_executor.cpp
b/be/src/service/point_query_executor.cpp
index 3e9088525c7..95419a1ed38 100644
--- a/be/src/service/point_query_executor.cpp
+++ b/be/src/service/point_query_executor.cpp
@@ -590,7 +590,8 @@ Status PointQueryExecutor::_lookup_row_data() {
const auto& segment = *it;
for (int cid : _reusable->missing_col_uids()) {
int pos = _reusable->get_col_uid_to_idx().at(cid);
- auto row_id =
static_cast<segment_v2::rowid_t>(row_loc.row_id);
+ std::vector<segment_v2::rowid_t> row_ids {
+ static_cast<segment_v2::rowid_t>(row_loc.row_id)};
auto& column = result_columns[pos];
std::unique_ptr<ColumnIterator> iter;
SlotDescriptor* slot =
_reusable->tuple_desc()->slots()[pos];
@@ -598,7 +599,7 @@ Status PointQueryExecutor::_lookup_row_data() {
storage_read_options.stats = &_read_stats;
storage_read_options.io_ctx = io_ctx;
RETURN_IF_ERROR(segment->seek_and_read_by_rowid(*_tablet->tablet_schema(), slot,
- row_id,
column,
+ row_ids,
column,
storage_read_options, iter));
}
}
diff --git a/be/src/storage/segment/segment.cpp
b/be/src/storage/segment/segment.cpp
index c293434abee..7de9dfe55ff 100644
--- a/be/src/storage/segment/segment.cpp
+++ b/be/src/storage/segment/segment.cpp
@@ -1180,9 +1180,17 @@ Status Segment::read_key_by_rowid(uint32_t row_id,
std::string* key) {
}
Status Segment::seek_and_read_by_rowid(const TabletSchema& schema,
SlotDescriptor* slot,
- uint32_t row_id, MutableColumnPtr&
result,
+ const std::vector<uint32_t>& row_ids,
+ MutableColumnPtr& result,
StorageReadOptions&
storage_read_options,
std::unique_ptr<ColumnIterator>&
iterator_hint) {
+ if (row_ids.empty()) {
+ return Status::OK();
+ }
+ DORIS_CHECK(std::is_sorted(row_ids.begin(), row_ids.end()));
+ DORIS_CHECK(std::adjacent_find(row_ids.begin(), row_ids.end()) ==
row_ids.end());
+ // ColumnIterator::seek_and_read expects monotonically increasing row_ids
without
+ // duplicates for correct ordinal scanning. Enforce this contract at the
entry point.
auto io_ctx = storage_read_options.io_ctx;
io_ctx.reader_type = ReaderType::READER_QUERY;
io_ctx.file_cache_stats = &storage_read_options.stats->file_cache_stats;
@@ -1193,7 +1201,6 @@ Status Segment::seek_and_read_by_rowid(const
TabletSchema& schema, SlotDescripto
.io_ctx = io_ctx,
};
- std::vector<segment_v2::rowid_t> single_row_loc {row_id};
const auto runtime_type = remove_nullable(slot->type());
const auto* variant_type = typeid_cast<const
DataTypeVariant*>(runtime_type.get());
const auto* variant_v2_type = typeid_cast<const
DataTypeVariantV2*>(runtime_type.get());
@@ -1220,7 +1227,7 @@ Status Segment::seek_and_read_by_rowid(const
TabletSchema& schema, SlotDescripto
RETURN_IF_ERROR(iterator_hint->init(opt));
}
RETURN_IF_ERROR(
- iterator_hint->read_by_rowids(single_row_loc.data(), 1,
file_storage_column));
+ iterator_hint->read_by_rowids(row_ids.data(), row_ids.size(),
file_storage_column));
ColumnPtr source_ptr;
// storage may have different type with schema, so we need to cast the
column
RETURN_IF_ERROR(variant_util::cast_column(
@@ -1233,7 +1240,7 @@ Status Segment::seek_and_read_by_rowid(const
TabletSchema& schema, SlotDescripto
result = IColumn::mutate(std::move(source_ptr));
return Status::OK();
}
- RETURN_IF_CATCH_EXCEPTION(result->insert_range_from(*source_ptr, 0,
1));
+ RETURN_IF_CATCH_EXCEPTION(result->insert_range_from(*source_ptr, 0,
row_ids.size()));
} else {
int index = (slot->col_unique_id() >= 0) ?
schema.field_index(slot->col_unique_id())
:
schema.field_index(slot->col_name());
@@ -1252,7 +1259,7 @@ Status Segment::seek_and_read_by_rowid(const
TabletSchema& schema, SlotDescripto
RETURN_IF_ERROR(new_column_iterator(column, &iterator_hint,
&storage_read_options));
RETURN_IF_ERROR(iterator_hint->init(opt));
}
- RETURN_IF_ERROR(iterator_hint->read_by_rowids(single_row_loc.data(),
1, result));
+ RETURN_IF_ERROR(iterator_hint->read_by_rowids(row_ids.data(),
row_ids.size(), result));
}
return Status::OK();
}
diff --git a/be/src/storage/segment/segment.h b/be/src/storage/segment/segment.h
index e236e22da9e..3149fd867d6 100644
--- a/be/src/storage/segment/segment.h
+++ b/be/src/storage/segment/segment.h
@@ -147,8 +147,9 @@ public:
Status read_key_by_rowid(uint32_t row_id, std::string* key);
- Status seek_and_read_by_rowid(const TabletSchema& schema, SlotDescriptor*
slot, uint32_t row_id,
- MutableColumnPtr& result,
+ // row_ids must be strictly increasing.
+ Status seek_and_read_by_rowid(const TabletSchema& schema, SlotDescriptor*
slot,
+ const std::vector<uint32_t>& row_ids,
MutableColumnPtr& result,
StorageReadOptions& storage_read_options,
std::unique_ptr<ColumnIterator>&
iterator_hint);
diff --git a/be/test/storage/variant/variant_column_writer_reader_test.cpp
b/be/test/storage/variant/variant_column_writer_reader_test.cpp
index fd9cf208648..e46e58c217c 100644
--- a/be/test/storage/variant/variant_column_writer_reader_test.cpp
+++ b/be/test/storage/variant/variant_column_writer_reader_test.cpp
@@ -3533,11 +3533,8 @@ TEST_F(VariantColumnWriterReaderTest,
test_segment_rowid_read_by_reader_version)
[](const segment_v2::SegmentSharedPtr& segment, const
TabletSchema& schema,
SlotDescriptor* slot, const std::vector<uint32_t>& row_ids,
MutableColumnPtr& result,
StorageReadOptions& read_options, ColumnIteratorUPtr& iterator)
-> Status {
- for (const uint32_t row_id : row_ids) {
- RETURN_IF_ERROR(segment->seek_and_read_by_rowid(schema, slot,
row_id, result,
- read_options,
iterator));
- }
- return Status::OK();
+ return segment->seek_and_read_by_rowid(schema, slot, row_ids, result,
read_options,
+ iterator);
};
ASSERT_FALSE(_tablet_schema->column(0).variant_is_v2());
for (size_t mode = 0; mode < 2; ++mode) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
