This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 72df240a6d4 [fix](exec) Backport batched rowid reads for lazy
materialization to 4.2 (#69429)
72df240a6d4 is described below
commit 72df240a6d44a45ef35827ab92a462043f3c9564
Author: Mryange <[email protected]>
AuthorDate: Fri Oct 23 14:05:04 2026 +0800
[fix](exec) Backport batched rowid reads for lazy materialization to 4.2
(#69429)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #63436, #69331
Problem Summary:
Backport #63436 to branch-4.2 using the branch-4.1 adaptation from
#69331. 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 requests by segment, sort and deduplicate rowids
before batch reads, then scatter results back to preserve request order
and duplicate rows.
The backport preserves branch-4.2 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)
- Static checks only: clang-format 16 and `git diff --check` passed.
- Build, unit tests, and SQL regression tests have not been run for this
4.2 backport.
- 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 84f2e732050..1b4e6fcb04c 100644
--- a/be/src/service/point_query_executor.cpp
+++ b/be/src/service/point_query_executor.cpp
@@ -584,7 +584,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];
@@ -609,7 +610,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 6258e2110e5..ca7db7148b7 100644
--- a/be/src/storage/segment/segment.cpp
+++ b/be/src/storage/segment/segment.cpp
@@ -1207,9 +1207,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;
@@ -1220,7 +1228,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());
@@ -1247,7 +1254,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(
@@ -1260,7 +1267,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());
@@ -1279,7 +1286,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 41199a31379..5ec9f9b3c8a 100644
--- a/be/src/storage/segment/segment.h
+++ b/be/src/storage/segment/segment.h
@@ -149,8 +149,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 aaf3794ddfa..bdaff4245fc 100644
--- a/be/test/storage/variant/variant_column_writer_reader_test.cpp
+++ b/be/test/storage/variant/variant_column_writer_reader_test.cpp
@@ -3863,11 +3863,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]
