From 4ca8c52d3e5100cc2506daacbd0366bf86274c66 Mon Sep 17 00:00:00 2001 From: HappenLee Date: Thu, 21 May 2026 12:00:36 +0800 Subject: [PATCH] [fix](exec) Backport batched rowid reads for lazy materialization to 4.2 Backport #63436 via the branch-4.1 adaptation in #69331. Group requests by segment, sort and deduplicate rowids before batch reads, and scatter results back to preserve request order and duplicate rows. Preserve branch-4.2 COW ownership, Variant V2 handling, rowid range checks, file cache policy, and external row-fetch fixes. Adapt the existing Variant segment-reader test to the batch interface. Validation: clang-format 16 and git diff --check passed. Build, unit tests, and SQL regression tests were not run for this backport. (cherry picked from commit 12f02ebf090f707b3e745ba98563957b792f561a) (cherry picked from commit 372719609707df77af19fab92179900638025abe) --- 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_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 0219ac0b7783f7..5fc288dd301545 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 file_mapping; + // (row_id, index_in_request) pairs for all rows in this segment. + std::vector> row_ids_with_positions; +}; + +static void scatter_scan_blocks_to_result_block( + const std::vector>& row_id_block_idx, + std::vector& 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(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(row_loc.ordinal_id()); + std::vector row_ids { + static_cast(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 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 scan_batches; + std::unordered_map batch_idx_by_seg; + // (batch_idx, position_in_batch) for each row in the original request. + std::vector> 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::max()) { return Status::InvalidArgument("internal row id exceeds uint32 range: {}", row_id); } - row_ids.emplace_back(static_cast(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::max()) { - return Status::InvalidArgument("internal row id exceeds uint32 range: {}", - next_row_id); - } - row_ids.emplace_back(static_cast(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(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 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 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 84f2e732050ae1..1b4e6fcb04ce9a 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(row_loc.row_id); + std::vector row_ids { + static_cast(row_loc.row_id)}; auto& column = result_columns[pos]; std::unique_ptr 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 6258e2110e59b1..ca7db7148b7eb2 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& row_ids, + MutableColumnPtr& result, StorageReadOptions& storage_read_options, std::unique_ptr& 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 single_row_loc {row_id}; const auto runtime_type = remove_nullable(slot->type()); const auto* variant_type = typeid_cast(runtime_type.get()); const auto* variant_v2_type = typeid_cast(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 41199a31379d25..5ec9f9b3c8a504 100644 --- a/be/src/storage/segment/segment.h +++ b/be/src/storage/segment/segment.h @@ -149,8 +149,9 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd 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& row_ids, MutableColumnPtr& result, StorageReadOptions& storage_read_options, std::unique_ptr& 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 aaf3794ddfaca9..bdaff4245fcd5f 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& 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) {