| // Licensed to the Apache Software Foundation (ASF) under one | |
| // or more contributor license agreements. See the NOTICE file | |
| // distributed with this work for additional information | |
| // regarding copyright ownership. The ASF licenses this file | |
| // to you under the Apache License, Version 2.0 (the | |
| // "License"); you may not use this file except in compliance | |
| // with the License. You may obtain a copy of the License at | |
| // | |
| // http://www.apache.org/licenses/LICENSE-2.0 | |
| // | |
| // Unless required by applicable law or agreed to in writing, | |
| // software distributed under the License is distributed on an | |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| // KIND, either express or implied. See the License for the | |
| // specific language governing permissions and limitations | |
| // under the License. | |
| #include "exec/rowid_fetcher.h" | |
| #include <fmt/format.h> | |
| #include <gen_cpp/data.pb.h> | |
| #include <gen_cpp/internal_service.pb.h> | |
| #include <gen_cpp/olap_file.pb.h> | |
| #include <gen_cpp/types.pb.h> | |
| #include <glog/logging.h> | |
| #include <stddef.h> | |
| #include <stdint.h> | |
| #include <algorithm> | |
| #include <cstdint> | |
| #include <memory> | |
| #include <ostream> | |
| #include <string> | |
| #include <unordered_map> | |
| #include <utility> | |
| #include <vector> | |
| #include "common/config.h" | |
| #include "common/exception.h" | |
| #include "common/signal_handler.h" | |
| #include "core/block/block.h" // Block | |
| #include "core/column/column.h" | |
| #include "core/data_type/data_type_struct.h" | |
| #include "core/data_type_serde/data_type_serde.h" | |
| #include "exec/operator/file_scan_operator.h" | |
| #include "exec/scan/file_scanner.h" | |
| #include "exec/scan/file_scanner_v2.h" | |
| #include "format/orc/vorc_reader.h" | |
| #include "format/parquet/vparquet_reader.h" | |
| #include "io/io_common.h" | |
| #include "runtime/descriptors.h" | |
| #include "runtime/exec_env.h" // ExecEnv | |
| #include "runtime/fragment_mgr.h" // FragmentMgr | |
| #include "runtime/runtime_state.h" // RuntimeState | |
| #include "runtime/workload_group/workload_group_manager.h" | |
| #include "semaphore" | |
| #include "storage/olap_common.h" | |
| #include "storage/rowset/beta_rowset.h" | |
| #include "storage/segment/column_reader.h" | |
| #include "storage/tablet/tablet_fwd.h" | |
| #include "storage/tablet/tablet_schema.h" | |
| #include "storage/utils.h" | |
| #include "util/jsonb/serialize.h" | |
| namespace doris { | |
| namespace { | |
| void set_topn_lazy_materialization_file_cache_stats( | |
| const io::FileCacheStatistics& stats, PTopNLazyMaterializationFileCacheStats* pstats) { | |
| pstats->set_local_io_count(stats.num_local_io_total); | |
| pstats->set_local_io_bytes(stats.bytes_read_from_local); | |
| pstats->set_remote_io_count(stats.num_remote_io_total); | |
| pstats->set_remote_io_bytes(stats.bytes_read_from_remote); | |
| pstats->set_skip_cache_io_count(stats.num_skip_cache_io_total); | |
| pstats->set_write_cache_bytes(stats.bytes_write_into_cache); | |
| pstats->set_local_io_time(stats.local_io_timer); | |
| pstats->set_remote_io_time(stats.remote_io_timer); | |
| pstats->set_write_cache_io_time(stats.write_cache_io_timer); | |
| } | |
| } // namespace | |
| struct IteratorKey { | |
| int64_t tablet_id; | |
| RowsetId rowset_id; | |
| uint64_t segment_id; | |
| int slot_id; | |
| // unordered map std::equal_to | |
| bool operator==(const IteratorKey& rhs) const { | |
| return tablet_id == rhs.tablet_id && rowset_id == rhs.rowset_id && | |
| segment_id == rhs.segment_id && slot_id == rhs.slot_id; | |
| } | |
| }; | |
| struct SegKey { | |
| int64_t tablet_id; | |
| RowsetId rowset_id; | |
| uint64_t segment_id; | |
| // unordered map std::equal_to | |
| bool operator==(const SegKey& rhs) const { | |
| return tablet_id == rhs.tablet_id && rowset_id == rhs.rowset_id && | |
| segment_id == rhs.segment_id; | |
| } | |
| }; | |
| struct HashOfSegKey { | |
| size_t operator()(const SegKey& key) const { | |
| size_t seed = 0; | |
| seed = HashUtil::hash64(&key.tablet_id, sizeof(key.tablet_id), seed); | |
| seed = HashUtil::hash64(&key.rowset_id.hi, sizeof(key.rowset_id.hi), seed); | |
| seed = HashUtil::hash64(&key.rowset_id.mi, sizeof(key.rowset_id.mi), seed); | |
| seed = HashUtil::hash64(&key.rowset_id.lo, sizeof(key.rowset_id.lo), seed); | |
| seed = HashUtil::hash64(&key.segment_id, sizeof(key.segment_id), seed); | |
| return seed; | |
| } | |
| }; | |
| struct HashOfIteratorKey { | |
| size_t operator()(const IteratorKey& key) const { | |
| size_t seed = 0; | |
| seed = HashUtil::hash64(&key.tablet_id, sizeof(key.tablet_id), seed); | |
| seed = HashUtil::hash64(&key.rowset_id.hi, sizeof(key.rowset_id.hi), seed); | |
| seed = HashUtil::hash64(&key.rowset_id.mi, sizeof(key.rowset_id.mi), seed); | |
| seed = HashUtil::hash64(&key.rowset_id.lo, sizeof(key.rowset_id.lo), seed); | |
| seed = HashUtil::hash64(&key.segment_id, sizeof(key.segment_id), seed); | |
| seed = HashUtil::hash64(&key.slot_id, sizeof(key.slot_id), seed); | |
| return seed; | |
| } | |
| }; | |
| struct IteratorItem { | |
| explicit IteratorItem(OlapReaderStatistics& stats) : storage_read_options(stats) {} | |
| std::unique_ptr<ColumnIterator> iterator; | |
| SegmentSharedPtr segment; | |
| // for holding the reference of storage read options to avoid use after release | |
| StorageReadOptions storage_read_options; | |
| }; | |
| // read_column is what slot resolves to, or its variant parent column for a subpath slot. | |
| static void set_slot_access_paths(const SlotDescriptor& slot, const TabletColumn& read_column, | |
| StorageReadOptions& storage_read_options) { | |
| const int32_t unique_id = | |
| read_column.unique_id() >= 0 ? read_column.unique_id() : read_column.parent_unique_id(); | |
| if (unique_id < 0) { | |
| return; | |
| } | |
| if (!slot.all_access_paths().empty()) { | |
| storage_read_options.all_access_paths[unique_id] = slot.all_access_paths(); | |
| } | |
| if (!slot.predicate_access_paths().empty()) { | |
| storage_read_options.predicate_access_paths[unique_id] = slot.predicate_access_paths(); | |
| } | |
| } | |
| struct SegItem { | |
| BaseTabletSPtr tablet; | |
| BetaRowsetSharedPtr rowset; | |
| // for holding the reference of segment to avoid use after release | |
| 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, | |
| const std::vector<Block>& scan_blocks, Block& result_block) { | |
| for (size_t column_id = 0; column_id < result_block.columns(); ++column_id) { | |
| auto dst_col_guard = result_block.mutate_column_scoped(column_id); | |
| MutableColumnPtr& dst_col = dst_col_guard.mutable_column(); | |
| std::vector<const IColumn*> scan_src_columns; | |
| scan_src_columns.reserve(row_id_block_idx.size()); | |
| std::vector<size_t> scan_positions; | |
| scan_positions.reserve(row_id_block_idx.size()); | |
| 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); | |
| scan_src_columns.emplace_back( | |
| scan_blocks[pos_block].get_by_position(column_id).column.get()); | |
| scan_positions.emplace_back(block_idx); | |
| } | |
| dst_col->insert_from_multi_column(scan_src_columns, scan_positions); | |
| } | |
| } | |
| Status RowIdStorageReader::read_by_rowids(const PMultiGetRequestV2& request, | |
| PMultiGetResponseV2* response) { | |
| if (request.request_block_descs_size()) { | |
| auto tquery_id = ((UniqueId)request.query_id()).to_thrift(); | |
| // todo: use mutableBlock instead of block | |
| std::vector<Block> result_blocks(request.request_block_descs_size()); | |
| OlapReaderStatistics stats; | |
| int64_t acquire_tablet_ms = 0; | |
| int64_t acquire_rowsets_ms = 0; | |
| int64_t acquire_segments_ms = 0; | |
| int64_t lookup_row_data_ms = 0; | |
| int64_t external_init_reader_avg_ms = 0; | |
| int64_t external_get_block_avg_ms = 0; | |
| size_t external_scan_range_cnt = 0; | |
| const auto file_cache_miss_policy = | |
| request.file_cache_remote_only_on_miss() | |
| ? io::FileCacheMissPolicy::REMOTE_ONLY_ON_MISS | |
| : io::FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK; | |
| // Add counters for different file mapping types | |
| std::unordered_map<FileMappingType, int64_t> file_type_counts; | |
| auto id_file_map = | |
| ExecEnv::GetInstance()->get_id_manager()->get_id_file_map(request.query_id()); | |
| // if id_file_map is null, means the BE not have scan range, just return ok | |
| if (!id_file_map) { | |
| // padding empty block to response | |
| LOG(INFO) << "id_file_map not found for query_id: " << print_id(request.query_id()); | |
| for (int i = 0; i < request.request_block_descs_size(); ++i) { | |
| response->add_blocks(); | |
| } | |
| return Status::OK(); | |
| } | |
| for (int i = 0; i < request.request_block_descs_size(); ++i) { | |
| const auto& request_block_desc = request.request_block_descs(i); | |
| PMultiGetBlockV2* pblock = response->add_blocks(); | |
| if (request_block_desc.row_id_size() >= 1) { | |
| // Since this block belongs to the same table, we only need to take the first type for judgment. | |
| auto first_file_id = request_block_desc.file_id(0); | |
| auto first_file_mapping = id_file_map->get_file_mapping(first_file_id); | |
| if (!first_file_mapping) { | |
| return Status::InternalError( | |
| "Backend:{} file_mapping not found, query_id: {}, file_id: {}", | |
| BackendOptions::get_localhost(), print_id(request.query_id()), | |
| first_file_id); | |
| } | |
| file_type_counts[first_file_mapping->type] += request_block_desc.row_id_size(); | |
| // prepare slots to build block | |
| std::vector<SlotDescriptor> slots; | |
| slots.reserve(request_block_desc.slots_size()); | |
| for (const auto& pslot : request_block_desc.slots()) { | |
| slots.push_back(SlotDescriptor(pslot)); | |
| } | |
| try { | |
| if (first_file_mapping->type == FileMappingType::INTERNAL) { | |
| RETURN_IF_ERROR(read_batch_doris_format_row( | |
| request_block_desc, id_file_map, slots, tquery_id, result_blocks[i], | |
| stats, &acquire_tablet_ms, &acquire_rowsets_ms, | |
| &acquire_segments_ms, &lookup_row_data_ms, file_cache_miss_policy)); | |
| } else { | |
| RETURN_IF_ERROR(read_batch_external_row( | |
| request.wg_id(), request_block_desc, id_file_map, slots, | |
| first_file_mapping, tquery_id, result_blocks[i], | |
| pblock->mutable_profile(), &external_init_reader_avg_ms, | |
| &external_get_block_avg_ms, &external_scan_range_cnt)); | |
| } | |
| } catch (const Exception& e) { | |
| return Status::Error<false>(e.code(), "Row id fetch failed because {}", | |
| e.what()); | |
| } | |
| } | |
| [[maybe_unused]] size_t compressed_size = 0; | |
| [[maybe_unused]] size_t uncompressed_size = 0; | |
| [[maybe_unused]] int64_t compress_time = 0; | |
| int be_exec_version = request.has_be_exec_version() ? request.be_exec_version() : 0; | |
| RETURN_IF_ERROR(result_blocks[i].serialize( | |
| be_exec_version, pblock->mutable_block(), &uncompressed_size, &compressed_size, | |
| &compress_time, segment_v2::CompressionTypePB::LZ4)); | |
| } | |
| // Build file type statistics string | |
| std::string file_type_stats; | |
| for (const auto& [type, count] : file_type_counts) { | |
| if (!file_type_stats.empty()) { | |
| file_type_stats += ", "; | |
| } | |
| file_type_stats += fmt::format("{}:{}", type, count); | |
| } | |
| LOG(INFO) << "Query stats: " | |
| << fmt::format( | |
| "query_id:{}, " | |
| "Internal table:" | |
| "hit_cached_pages:{}, total_pages_read:{}, compressed_bytes_read:{}, " | |
| "io_latency:{}ns, uncompressed_bytes_read:{}, bytes_read:{}, " | |
| "acquire_tablet_ms:{}, acquire_rowsets_ms:{}, acquire_segments_ms:{}, " | |
| "lookup_row_data_ms:{}, file_types:[{}]; " | |
| "External table : init_reader_ms:{}, get_block_ms:{}, " | |
| "external_scan_range_cnt:{}", | |
| print_id(request.query_id()), stats.cached_pages_num, | |
| stats.total_pages_num, stats.compressed_bytes_read, stats.io_ns, | |
| stats.uncompressed_bytes_read, stats.bytes_read, acquire_tablet_ms, | |
| acquire_rowsets_ms, acquire_segments_ms, lookup_row_data_ms, | |
| file_type_stats, external_init_reader_avg_ms, | |
| external_get_block_avg_ms, external_scan_range_cnt); | |
| set_topn_lazy_materialization_file_cache_stats( | |
| stats.file_cache_stats, | |
| response->mutable_topn_lazy_materialization_file_cache_stats()); | |
| } | |
| return Status::OK(); | |
| } | |
| Status RowIdStorageReader::read_batch_doris_format_row( | |
| const PRequestBlockDesc& request_block_desc, std::shared_ptr<IdFileMap> id_file_map, | |
| std::vector<SlotDescriptor>& slots, const TUniqueId& query_id, Block& result_block, | |
| OlapReaderStatistics& stats, int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, | |
| int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms, | |
| io::FileCacheMissPolicy file_cache_miss_policy) { | |
| if (result_block.is_empty_column()) [[likely]] { | |
| result_block = Block(slots, request_block_desc.row_id_size()); | |
| } | |
| std::vector<TabletColumn> fetch_columns; | |
| fetch_columns.reserve(request_block_desc.column_descs_size()); | |
| for (const ColumnPB& column_pb : request_block_desc.column_descs()) { | |
| fetch_columns.emplace_back(column_pb); | |
| } | |
| std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey> iterator_map; | |
| std::unordered_map<SegKey, SegItem, HashOfSegKey> seg_map; | |
| std::string row_store_buffer; | |
| RowStoreReadStruct row_store_read_struct(row_store_buffer); | |
| if (request_block_desc.fetch_row_store()) { | |
| for (int i = 0; i < request_block_desc.slots_size(); ++i) { | |
| row_store_read_struct.serdes.emplace_back(slots[i].get_data_type_ptr()->get_serde()); | |
| row_store_read_struct.col_uid_to_idx[slots[i].col_unique_id()] = i; | |
| row_store_read_struct.default_values.emplace_back(slots[i].col_default_value()); | |
| } | |
| } | |
| // 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); | |
| 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); | |
| } | |
| // 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(request_block_desc.row_id(j), | |
| 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, scan_batch.file_mapping, row_ids, slots, fetch_columns, | |
| 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(); | |
| } | |
| const std::string RowIdStorageReader::ScannersRunningTimeProfile = "ScannersRunningTime"; | |
| const std::string RowIdStorageReader::InitReaderAvgTimeProfile = "InitReaderAvgTime"; | |
| const std::string RowIdStorageReader::GetBlockAvgTimeProfile = "GetBlockAvgTime"; | |
| const std::string RowIdStorageReader::FileReadLinesProfile = "FileReadLines"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOCount = | |
| "TopNLazyMaterializationSecondPhaseLocalIOCount"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOBytes = | |
| "TopNLazyMaterializationSecondPhaseLocalIOBytes"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOCount = | |
| "TopNLazyMaterializationSecondPhaseRemoteIOCount"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOBytes = | |
| "TopNLazyMaterializationSecondPhaseRemoteIOBytes"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseSkipCacheIOCount = | |
| "TopNLazyMaterializationSecondPhaseSkipCacheIOCount"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheBytes = | |
| "TopNLazyMaterializationSecondPhaseWriteCacheBytes"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOTime = | |
| "TopNLazyMaterializationSecondPhaseLocalIOTime"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOTime = | |
| "TopNLazyMaterializationSecondPhaseRemoteIOTime"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheIOTime = | |
| "TopNLazyMaterializationSecondPhaseWriteCacheIOTime"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseRowsRead = | |
| "TopNLazyMaterializationSecondPhaseRowsRead"; | |
| const std::string RowIdStorageReader::TopNLazyMaterializationSecondPhaseSegmentsRead = | |
| "TopNLazyMaterializationSecondPhaseSegmentsRead"; | |
| bool RowIdStorageReader::should_use_file_scanner_v2(const TQueryOptions& query_options, | |
| const TFileScanRangeParams& scan_params, | |
| const TFileRangeDesc& range) { | |
| const auto format_type = | |
| range.__isset.format_type ? range.format_type : scan_params.format_type; | |
| // Phase two inherits the query options, including the Thrift presence bit. Reuse phase one's | |
| // policy so disabling V2 (or an older payload omitting the option) also keeps row fetches on V1. | |
| return FileScanLocalState::should_use_file_scanner_v2(query_options, false, scan_params) && | |
| (format_type == TFileFormatType::FORMAT_PARQUET || | |
| format_type == TFileFormatType::FORMAT_ORC) && | |
| FileScannerV2::is_supported(scan_params, range); | |
| } | |
| TFileRangeDesc RowIdStorageReader::build_external_fetch_range(const TFileRangeDesc& source_range) { | |
| // Rows were selected after delete filtering. Preserve the original path and row lineage | |
| // needed by virtual columns, and do not mutate the FileMapping shared by other fetches. | |
| auto range = source_range; | |
| range.table_format_params.iceberg_params.__set_delete_files({}); | |
| range.table_format_params.transactional_hive_params = TTransactionalHiveDesc {}; | |
| return range; | |
| } | |
| Status RowIdStorageReader::read_external_row_from_file_mapping( | |
| size_t idx, const std::multimap<segment_v2::rowid_t, size_t>& row_ids, | |
| const std::shared_ptr<FileMapping>& file_mapping, | |
| const std::vector<SlotDescriptor>& scan_slots, const TUniqueId& query_id, | |
| const std::shared_ptr<RuntimeState>& runtime_state, std::vector<Block>& scan_blocks, | |
| std::vector<std::pair<size_t, size_t>>& row_id_block_idx, | |
| std::vector<RowIdStorageReader::ExternalFetchStatistics>& fetch_statistics, | |
| const TFileScanRangeParams& rpc_scan_params, | |
| const std::unordered_map<std::string, int>& colname_to_slot_id, | |
| std::counting_semaphore<>& semaphore, TupleDescriptor& tuple_desc) { | |
| SCOPED_ATTACH_TASK(ExecEnv::GetInstance()->rowid_storage_reader_tracker()); | |
| signal::set_signal_task_id(query_id); | |
| // Release the concurrency permit on every exit path (including error returns | |
| // and exceptions). Completion accounting and status publishing are owned by | |
| // the caller, so the status is always published before the waiter is woken. | |
| Defer defer([&] { semaphore.release(); }); | |
| std::list<int64_t> read_ids; | |
| //Generate an ordered list with the help of the orderliness of the map. | |
| for (const auto& [row_id, result_block_idx] : row_ids) { | |
| if (read_ids.empty() || read_ids.back() != row_id) { | |
| read_ids.emplace_back(row_id); | |
| } | |
| row_id_block_idx[result_block_idx] = std::make_pair(idx, read_ids.size() - 1); | |
| } | |
| scan_blocks[idx] = Block(scan_slots, read_ids.size()); | |
| auto& external_info = file_mapping->get_external_file_info(); | |
| auto scan_range_desc = build_external_fetch_range(external_info.scan_range_desc); | |
| std::unique_ptr<RuntimeProfile> sub_runtime_profile = | |
| std::make_unique<RuntimeProfile>("ExternalRowIDFetcher"); | |
| { | |
| if (should_use_file_scanner_v2(runtime_state->query_options(), rpc_scan_params, | |
| scan_range_desc)) { | |
| auto file_scanner = FileScannerV2::create_unique( | |
| runtime_state.get(), sub_runtime_profile.get(), &rpc_scan_params, | |
| &colname_to_slot_id, &tuple_desc); | |
| RETURN_IF_ERROR(file_scanner->read_by_rows(scan_range_desc, read_ids, &scan_blocks[idx], | |
| &fetch_statistics[idx].init_reader_ms, | |
| &fetch_statistics[idx].get_block_ms)); | |
| } else { | |
| // Phase one can still use V1 for table-format variants unsupported by V2, so keep the | |
| // matching phase-two path instead of turning a previously valid query into an error. | |
| auto file_scanner = | |
| FileScanner::create_unique(runtime_state.get(), sub_runtime_profile.get(), | |
| &rpc_scan_params, &colname_to_slot_id, &tuple_desc); | |
| RETURN_IF_ERROR(file_scanner->prepare_for_read_lines(scan_range_desc)); | |
| RETURN_IF_ERROR(file_scanner->read_lines_from_range( | |
| scan_range_desc, read_ids, &scan_blocks[idx], external_info, | |
| &fetch_statistics[idx].init_reader_ms, &fetch_statistics[idx].get_block_ms)); | |
| } | |
| } | |
| if (scan_blocks[idx].rows() != read_ids.size()) { | |
| return Status::InternalError( | |
| "Row id fetch scan row count mismatch, " | |
| "query_id={}, path={}, expected_rows={}, actual_rows={}", | |
| print_id(query_id), scan_range_desc.path, read_ids.size(), scan_blocks[idx].rows()); | |
| } | |
| for (size_t column_id = 0; column_id < scan_blocks[idx].columns(); ++column_id) { | |
| const auto& column = scan_blocks[idx].get_by_position(column_id); | |
| if (column.column->size() != read_ids.size()) { | |
| return Status::InternalError( | |
| "Row id fetch scan column row count mismatch, " | |
| "query_id={}, path={}, column={}, expected_rows={}, actual_rows={}", | |
| print_id(query_id), scan_range_desc.path, column.name, read_ids.size(), | |
| column.column->size()); | |
| } | |
| } | |
| auto file_read_bytes_counter = | |
| sub_runtime_profile->get_counter(FileScannerV2::FileReadBytesProfile); | |
| if (file_read_bytes_counter != nullptr) { | |
| fetch_statistics[idx].file_read_bytes = PrettyPrinter::print( | |
| file_read_bytes_counter->value(), file_read_bytes_counter->type()); | |
| } | |
| auto file_read_times_counter = | |
| sub_runtime_profile->get_counter(FileScannerV2::FileReadTimeProfile); | |
| if (file_read_times_counter != nullptr) { | |
| fetch_statistics[idx].file_read_times = PrettyPrinter::print( | |
| file_read_times_counter->value(), file_read_times_counter->type()); | |
| } | |
| return Status::OK(); | |
| } | |
| std::string RowIdStorageReader::source_column_key(const SlotDescriptor& slot, uint32_t column_idx) { | |
| fmt::memory_buffer key; | |
| // Length-prefix each component so distinct sequences cannot alias, e.g. | |
| // paths ["a", "b"] -> "1:a1:b" while ["a:b"] -> "3:a:b". | |
| auto append = [&key](std::string_view component) { | |
| fmt::format_to(key, "{}:", component.size()); | |
| key.append(component.data(), component.data() + component.size()); | |
| }; | |
| append(slot.col_name()); | |
| append(std::to_string(column_idx)); | |
| append(std::to_string(slot.col_unique_id())); | |
| append(std::to_string(slot.column_paths().size())); | |
| for (const auto& path : slot.column_paths()) { | |
| append(path); | |
| } | |
| append(std::to_string(slot.all_access_paths().size())); | |
| // Encode each optional sub-path's presence bit separately from its element | |
| // count so an absent path ("0") never aliases a present-but-empty path | |
| // ("1" + size "0"). | |
| auto append_optional_path = [&append](bool is_set, const std::vector<std::string>& items) { | |
| append(is_set ? "1" : "0"); | |
| if (is_set) { | |
| append(std::to_string(items.size())); | |
| for (const auto& item : items) { | |
| append(item); | |
| } | |
| } | |
| }; | |
| for (const auto& path : slot.all_access_paths()) { | |
| append(fmt::format("{}", path.type)); | |
| append_optional_path(path.__isset.data_access_path, path.data_access_path.path); | |
| append_optional_path(path.__isset.meta_access_path, path.meta_access_path.path); | |
| } | |
| return fmt::to_string(key); | |
| } | |
| Status RowIdStorageReader::submit_external_scan_tasks( | |
| ScannerScheduler* scheduler, std::counting_semaphore<>& semaphore, size_t task_count, | |
| const std::function<std::string(size_t)>& make_task_id, | |
| const std::function<Status(size_t)>& run_task) { | |
| // `completed_count` is a plain counter guarded by `mtx`; the same mutex guards | |
| // the wait predicate below, so a worker can never notify between the waiter's | |
| // predicate check and its wait. | |
| AtomicStatus scan_status; | |
| std::condition_variable cv; | |
| std::mutex mtx; | |
| size_t completed_count = 0; | |
| // Only tasks the scheduler actually accepted are waited for. If a submission | |
| // fails we stop submitting, but still wait for the already-accepted tasks so | |
| // their workers cannot outlive the locals they capture by reference. | |
| size_t submitted_count = 0; | |
| for (size_t idx = 0; idx < task_count; ++idx) { | |
| semaphore.acquire(); | |
| auto run_one_task = [&, idx]() -> bool { | |
| Status task_status = Status::OK(); | |
| // Publish the status before the completion signal wakes the waiter, on every | |
| // path. A scanner that throws would otherwise leave scan_status OK while this | |
| // Defer still counts the task as finished, and the caller would report success | |
| // over a half-filled result block. | |
| Defer complete([&] { | |
| scan_status.update(task_status); | |
| std::lock_guard<std::mutex> lock(mtx); | |
| ++completed_count; | |
| cv.notify_one(); | |
| }); | |
| ASSIGN_STATUS_IF_CATCH_EXCEPTION(task_status = run_task(idx), task_status); | |
| return true; | |
| }; | |
| Status submit_st = scheduler->submit_scan_task( | |
| SimplifiedScanTask(run_one_task, nullptr, nullptr), make_task_id(idx)); | |
| if (!submit_st.ok()) { | |
| scan_status.update(submit_st); | |
| semaphore.release(); | |
| break; | |
| } | |
| ++submitted_count; | |
| } | |
| { | |
| std::unique_lock<std::mutex> lock(mtx); | |
| cv.wait(lock, [&] { return completed_count == submitted_count; }); | |
| } | |
| return scan_status.ok() ? Status::OK() : scan_status.status(); | |
| } | |
| TFileScanRangeParams RowIdStorageReader::build_external_scan_params( | |
| const TFileScanRangeParams& source_params, const TFileRangeDesc& range, | |
| const std::vector<SlotDescriptor>& scan_slots, | |
| const std::vector<uint32_t>& scan_column_idxs) { | |
| DORIS_CHECK(scan_slots.size() == scan_column_idxs.size()); | |
| auto params = source_params; | |
| params.required_slots.clear(); | |
| params.column_idxs.clear(); | |
| params.slot_name_to_schema_pos.clear(); | |
| const std::set partition_names(range.columns_from_path_keys.begin(), | |
| range.columns_from_path_keys.end()); | |
| for (size_t slot_idx = 0; slot_idx < scan_slots.size(); ++slot_idx) { | |
| const auto& slot = scan_slots[slot_idx]; | |
| const auto column_idx = scan_column_idxs[slot_idx]; | |
| TFileScanSlotInfo slot_info; | |
| slot_info.__set_slot_id(slot.id()); | |
| // Hive V2 checks the Thrift presence bit before trusting is_file_slot. Without it, | |
| // partition columns consume physical file indexes and invalidate the rebuilt projection. | |
| bool is_file_slot = !partition_names.contains(slot.col_name()); | |
| if (source_params.__isset.column_name_to_category) { | |
| // Lazy metadata slots may be absent from phase one's required_slots and have new | |
| // slot IDs here. The pinned schema's name map preserves their original categories. | |
| const auto it = source_params.column_name_to_category.find(slot.col_name()); | |
| const auto category = it != source_params.column_name_to_category.end() | |
| ? it->second | |
| : TColumnCategory::REGULAR; | |
| slot_info.__set_category(category); | |
| is_file_slot = | |
| category == TColumnCategory::REGULAR || category == TColumnCategory::GENERATED; | |
| } | |
| slot_info.__set_is_file_slot(is_file_slot); | |
| if (is_file_slot) { | |
| params.column_idxs.emplace_back(column_idx); | |
| } | |
| params.default_value_of_src_slot.emplace(slot.id(), TExpr {}); | |
| params.required_slots.emplace_back(slot_info); | |
| params.slot_name_to_schema_pos.emplace(slot.col_name(), column_idx); | |
| } | |
| return params; | |
| } | |
| Status RowIdStorageReader::read_batch_external_row( | |
| const uint64_t workload_group_id, const PRequestBlockDesc& request_block_desc, | |
| std::shared_ptr<IdFileMap> id_file_map, std::vector<SlotDescriptor>& slots, | |
| std::shared_ptr<FileMapping> first_file_mapping, const TUniqueId& query_id, | |
| Block& result_block, PRuntimeProfileTree* pprofile, int64_t* init_reader_avg_ms, | |
| int64_t* get_block_avg_ms, size_t* scan_range_cnt) { | |
| TFileScanRangeParams rpc_scan_params; | |
| TupleDescriptor tuple_desc(request_block_desc.desc(), false); | |
| std::unordered_map<std::string, int> colname_to_slot_id; | |
| std::shared_ptr<RuntimeState> runtime_state = nullptr; | |
| std::vector<SlotDescriptor> scan_slots; | |
| std::vector<size_t> result_column_to_scan_column; | |
| std::vector<uint32_t> scan_column_idxs; | |
| int max_file_scanners = 0; | |
| { | |
| if (result_block.is_empty_column()) [[likely]] { | |
| result_block = Block(slots, request_block_desc.row_id_size()); | |
| } | |
| if (request_block_desc.column_idxs_size() != slots.size()) { | |
| return Status::InternalError( | |
| "Row id fetch request has mismatched slots and column indexes, " | |
| "query_id={}, slots={}, column_idxs={}", | |
| print_id(query_id), slots.size(), request_block_desc.column_idxs_size()); | |
| } | |
| auto& external_info = first_file_mapping->get_external_file_info(); | |
| int plan_node_id = external_info.plan_node_id; | |
| const auto& first_scan_range_desc = external_info.scan_range_desc; | |
| DCHECK(id_file_map->get_external_scan_params().contains(plan_node_id)); | |
| const auto* old_scan_params = &(id_file_map->get_external_scan_params().at(plan_node_id)); | |
| std::unordered_map<std::string, size_t> source_column_to_scan_idx; | |
| result_column_to_scan_column.reserve(slots.size()); | |
| scan_slots.reserve(slots.size()); | |
| scan_column_idxs.reserve(slots.size()); | |
| for (auto slot_idx = 0; slot_idx < slots.size(); ++slot_idx) { | |
| const auto& slot = slots[slot_idx]; | |
| const auto column_idx = request_block_desc.column_idxs(slot_idx); | |
| const auto key = source_column_key(slot, column_idx); | |
| auto [it, inserted] = | |
| source_column_to_scan_idx.emplace(key, source_column_to_scan_idx.size()); | |
| result_column_to_scan_column.emplace_back(it->second); | |
| if (inserted) { | |
| scan_slots.emplace_back(slot); | |
| scan_column_idxs.emplace_back(column_idx); | |
| } | |
| } | |
| for (auto& slot : scan_slots) { | |
| tuple_desc.add_slot(&slot); | |
| colname_to_slot_id[slot.col_name()] = slot.id(); | |
| } | |
| rpc_scan_params = build_external_scan_params(*old_scan_params, first_scan_range_desc, | |
| scan_slots, scan_column_idxs); | |
| const auto& query_options = id_file_map->get_query_options(); | |
| const auto& query_globals = id_file_map->get_query_globals(); | |
| /* | |
| * The scan stage needs the information in query_options to generate different behaviors according to the specific variables: | |
| * query_options.hive_parquet_use_column_names, query_options.truncate_char_or_varchar_columns,query_globals.time_zone ... | |
| * | |
| * To ensure the same behavior as the scan stage, I get query_options query_globals from id_file_map, then create runtime_state | |
| * and pass it to vfile_scanner so that the runtime_state information is the same as the scan stage and the behavior is also consistent. | |
| */ | |
| runtime_state = RuntimeState::create_shared( | |
| query_id, -1, query_options, query_globals, ExecEnv::GetInstance(), | |
| ExecEnv::GetInstance()->rowid_storage_reader_tracker()); | |
| max_file_scanners = id_file_map->get_max_file_scanners(); | |
| } | |
| // Hash(TFileRangeDesc) => { all the rows that need to be read and their positions in the result block. } + file mapping | |
| // std::multimap<segment_v2::rowid_t, size_t> : The reason for using multimap is: may need the same row of data multiple times. | |
| std::map<std::string, | |
| std::pair<std::multimap<segment_v2::rowid_t, size_t>, std::shared_ptr<FileMapping>>> | |
| scan_rows; | |
| // Block corresponding to the order of `scan_rows` map. | |
| std::vector<Block> scan_blocks; | |
| // row_id (Indexing of vectors) => < In which block, which line in the block > | |
| std::vector<std::pair<size_t, size_t>> row_id_block_idx; | |
| // Count the time/bytes it takes to read each TFileRangeDesc. (for profile) | |
| std::vector<ExternalFetchStatistics> fetch_statistics; | |
| auto hash_file_range = [](const TFileRangeDesc& file_range_desc) { | |
| std::string value; | |
| value.resize(file_range_desc.path.size() + sizeof(file_range_desc.start_offset)); | |
| auto* ptr = value.data(); | |
| memcpy(ptr, &file_range_desc.start_offset, sizeof(file_range_desc.start_offset)); | |
| ptr += sizeof(file_range_desc.start_offset); | |
| memcpy(ptr, file_range_desc.path.data(), file_range_desc.path.size()); | |
| return value; | |
| }; | |
| for (int j = 0; j < request_block_desc.row_id_size(); ++j) { | |
| auto file_id = request_block_desc.file_id(j); | |
| 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); | |
| } | |
| const auto& external_info = file_mapping->get_external_file_info(); | |
| const auto& scan_range_desc = external_info.scan_range_desc; | |
| auto scan_range_hash = hash_file_range(scan_range_desc); | |
| if (scan_rows.contains(scan_range_hash)) { | |
| scan_rows.at(scan_range_hash).first.emplace(request_block_desc.row_id(j), j); | |
| } else { | |
| std::multimap<segment_v2::rowid_t, size_t> tmp {{request_block_desc.row_id(j), j}}; | |
| scan_rows.emplace(scan_range_hash, std::make_pair(tmp, file_mapping)); | |
| } | |
| } | |
| scan_blocks.resize(scan_rows.size()); | |
| row_id_block_idx.resize(request_block_desc.row_id_size()); | |
| fetch_statistics.resize(scan_rows.size()); | |
| // Get the workload group for subsequent scan task submission. | |
| std::vector<uint64_t> workload_group_ids; | |
| workload_group_ids.emplace_back(workload_group_id); | |
| auto wg = ExecEnv::GetInstance()->workload_group_mgr()->get_group(workload_group_ids); | |
| doris::TaskScheduler* exec_sched = nullptr; | |
| ScannerScheduler* scan_sched = nullptr; | |
| ScannerScheduler* remote_scan_sched = nullptr; | |
| wg->get_query_scheduler(&exec_sched, &scan_sched, &remote_scan_sched); | |
| DCHECK(remote_scan_sched); | |
| int64_t scan_running_time = 0; | |
| RETURN_IF_ERROR(scope_timer_run( | |
| [&]() -> Status { | |
| //semaphore: Limit the number of scan tasks submitted at one time | |
| std::counting_semaphore semaphore {max_file_scanners}; | |
| std::vector<std::pair<std::multimap<segment_v2::rowid_t, size_t>, | |
| std::shared_ptr<FileMapping>>> | |
| scan_info_list; | |
| scan_info_list.reserve(scan_rows.size()); | |
| for (const auto& [_, scan_info] : scan_rows) { | |
| scan_info_list.emplace_back(scan_info); | |
| } | |
| return submit_external_scan_tasks( | |
| remote_scan_sched, semaphore, scan_rows.size(), | |
| [&](size_t idx) { | |
| return fmt::format("{}-read_batch_external_row-{}", print_id(query_id), | |
| idx); | |
| }, | |
| [&](size_t idx) -> Status { | |
| const auto& [row_ids, file_mapping] = scan_info_list[idx]; | |
| return read_external_row_from_file_mapping( | |
| idx, row_ids, file_mapping, scan_slots, query_id, runtime_state, | |
| scan_blocks, row_id_block_idx, fetch_statistics, | |
| rpc_scan_params, colname_to_slot_id, semaphore, tuple_desc); | |
| }); | |
| }, | |
| &scan_running_time)); | |
| // Insert the read data into result_block. Use insert_indices_from() instead of | |
| // scatter_scan_blocks_to_result_block()/insert_from_multi_column(), because | |
| // scan_blocks may have fewer columns than result_block when duplicate physical columns | |
| // are deduplicated, and insert_from_multi_column() cannot handle ColumnString | |
| // cross-type (32/64) copies safely. | |
| const size_t result_column_count = result_block.columns(); | |
| for (size_t column_id = 0; column_id < result_column_count; column_id++) { | |
| auto dst_col_guard = result_block.mutate_column_scoped(column_id); | |
| MutableColumnPtr& dst_col = dst_col_guard.mutable_column(); | |
| bool dst_is_nullable = dst_col->is_nullable(); | |
| std::vector<ColumnPtr> nullable_src_columns(scan_blocks.size()); | |
| auto scan_column_id = result_column_to_scan_column[column_id]; | |
| for (const auto& [pos_block, block_idx] : row_id_block_idx) { | |
| DCHECK_GT(scan_blocks.size(), pos_block); | |
| DCHECK_GT(scan_blocks[pos_block].columns(), scan_column_id); | |
| const auto& src_column_ptr = | |
| scan_blocks[pos_block].get_by_position(scan_column_id).column; | |
| const auto* src_col = src_column_ptr.get(); | |
| if (dst_is_nullable && !src_col->is_nullable()) { | |
| if (!nullable_src_columns[pos_block]) { | |
| nullable_src_columns[pos_block] = make_nullable(src_column_ptr); | |
| } | |
| src_col = nullable_src_columns[pos_block].get(); | |
| } | |
| if (block_idx >= src_col->size()) { | |
| return Status::InternalError( | |
| "Row id fetch source index out of range, query_id={}, column={}, " | |
| "source_block={}, source_rows={}, row_index={}", | |
| print_id(query_id), result_block.get_by_position(column_id).name, pos_block, | |
| src_col->size(), block_idx); | |
| } | |
| uint32_t scan_position = cast_set<uint32_t>(block_idx); | |
| dst_col->insert_indices_from(*src_col, &scan_position, &scan_position + 1); | |
| } | |
| } | |
| // Statistical runtime profile information. | |
| std::unique_ptr<RuntimeProfile> runtime_profile = | |
| std::make_unique<RuntimeProfile>("ExternalRowIDFetcher"); | |
| { | |
| runtime_profile->add_info_string(ScannersRunningTimeProfile, | |
| std::to_string(scan_running_time) + "ms"); | |
| fmt::memory_buffer file_read_lines_buffer; | |
| format_to(file_read_lines_buffer, "["); | |
| fmt::memory_buffer file_read_bytes_buffer; | |
| format_to(file_read_bytes_buffer, "["); | |
| fmt::memory_buffer file_read_times_buffer; | |
| format_to(file_read_times_buffer, "["); | |
| size_t idx = 0; | |
| for (const auto& [_, scan_info] : scan_rows) { | |
| format_to(file_read_lines_buffer, "{}, ", scan_info.first.size()); | |
| *init_reader_avg_ms = fetch_statistics[idx].init_reader_ms; | |
| *get_block_avg_ms += fetch_statistics[idx].get_block_ms; | |
| format_to(file_read_bytes_buffer, "{}, ", fetch_statistics[idx].file_read_bytes); | |
| format_to(file_read_times_buffer, "{}, ", fetch_statistics[idx].file_read_times); | |
| idx++; | |
| } | |
| format_to(file_read_lines_buffer, "]"); | |
| format_to(file_read_bytes_buffer, "]"); | |
| format_to(file_read_times_buffer, "]"); | |
| *init_reader_avg_ms /= fetch_statistics.size(); | |
| *get_block_avg_ms /= fetch_statistics.size(); | |
| runtime_profile->add_info_string(InitReaderAvgTimeProfile, | |
| std::to_string(*init_reader_avg_ms) + "ms"); | |
| runtime_profile->add_info_string(GetBlockAvgTimeProfile, | |
| std::to_string(*init_reader_avg_ms) + "ms"); | |
| runtime_profile->add_info_string(FileReadLinesProfile, | |
| fmt::to_string(file_read_lines_buffer)); | |
| runtime_profile->add_info_string(FileScannerV2::FileReadBytesProfile, | |
| fmt::to_string(file_read_bytes_buffer)); | |
| runtime_profile->add_info_string(FileScannerV2::FileReadTimeProfile, | |
| fmt::to_string(file_read_times_buffer)); | |
| } | |
| runtime_profile->to_proto(pprofile, 2); | |
| *scan_range_cnt = scan_rows.size(); | |
| return Status::OK(); | |
| } | |
| Status RowIdStorageReader::read_doris_format_row( | |
| const std::shared_ptr<IdFileMap>& id_file_map, | |
| const std::shared_ptr<FileMapping>& file_mapping, const std::vector<uint32_t>& row_ids, | |
| std::vector<SlotDescriptor>& slots, const std::vector<TabletColumn>& fetch_columns, | |
| RowStoreReadStruct& row_store_read_struct, OlapReaderStatistics& stats, | |
| int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, int64_t* acquire_segments_ms, | |
| int64_t* lookup_row_data_ms, std::unordered_map<SegKey, SegItem, HashOfSegKey>& seg_map, | |
| std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>& iterator_map, | |
| io::FileCacheMissPolicy file_cache_miss_policy, Block& result_block) { | |
| 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}; | |
| BaseTabletSPtr tablet; | |
| BetaRowsetSharedPtr rowset; | |
| SegmentSharedPtr segment; | |
| if (seg_map.find(seg_key) == seg_map.end()) { | |
| tablet = scope_timer_run( | |
| [&]() { | |
| auto res = ExecEnv::get_tablet(tablet_id); | |
| return !res.has_value() ? nullptr | |
| : std::dynamic_pointer_cast<BaseTablet>(res.value()); | |
| }, | |
| acquire_tablet_ms); | |
| if (!tablet) { | |
| return Status::InternalError( | |
| "Backend:{} tablet not found, tablet_id: {}, rowset_id: {}, segment_id: {}, " | |
| "row_id: {}", | |
| BackendOptions::get_localhost(), tablet_id, rowset_id.to_string(), segment_id, | |
| row_ids[0]); | |
| } | |
| rowset = std::static_pointer_cast<BetaRowset>(scope_timer_run( | |
| [&]() { return id_file_map->get_temp_rowset(tablet_id, rowset_id); }, | |
| acquire_rowsets_ms)); | |
| if (!rowset) { | |
| return Status::InternalError( | |
| "Backend:{} rowset_id not found, tablet_id: {}, rowset_id: {}, segment_id: {}, " | |
| "row_id: {}", | |
| BackendOptions::get_localhost(), tablet_id, rowset_id.to_string(), segment_id, | |
| row_ids[0]); | |
| } | |
| SegmentCacheHandle segment_cache; | |
| RETURN_IF_ERROR(scope_timer_run( | |
| [&]() { | |
| return SegmentLoader::instance()->load_segments(rowset, &segment_cache, true); | |
| }, | |
| acquire_segments_ms)); | |
| auto it = std::find_if(segment_cache.get_segments().cbegin(), | |
| segment_cache.get_segments().cend(), | |
| [segment_id](const segment_v2::SegmentSharedPtr& seg) { | |
| return seg->id() == segment_id; | |
| }); | |
| if (it == segment_cache.get_segments().end()) { | |
| return Status::InternalError( | |
| "Backend:{} segment not found, tablet_id: {}, rowset_id: {}, segment_id: {}, " | |
| "row_id: {}", | |
| BackendOptions::get_localhost(), tablet_id, rowset_id.to_string(), segment_id, | |
| row_ids[0]); | |
| } | |
| segment = *it; | |
| seg_map[seg_key] = SegItem {.tablet = tablet, .rowset = rowset, .segment = segment}; | |
| } else { | |
| auto& seg_item = seg_map[seg_key]; | |
| tablet = seg_item.tablet; | |
| rowset = seg_item.rowset; | |
| segment = seg_item.segment; | |
| } | |
| // if row_store_read_struct not empty, means the line we should read from row_store | |
| if (!row_store_read_struct.default_values.empty()) { | |
| if (!tablet->tablet_schema()->has_row_store_for_all_columns()) { | |
| return Status::InternalError("Tablet {} does not have row store for all columns", | |
| tablet->tablet_id()); | |
| } | |
| auto result_columns_guard = result_block.mutate_columns_scoped(); | |
| MutableColumns& result_columns = result_columns_guard.mutable_columns(); | |
| io::IOContext io_ctx; | |
| io_ctx.reader_type = ReaderType::READER_QUERY; | |
| io_ctx.file_cache_stats = &stats.file_cache_stats; | |
| io_ctx.file_cache_miss_policy = file_cache_miss_policy; | |
| for (auto row_id : row_ids) { | |
| RowLocation loc(rowset_id, segment->id(), cast_set<uint32_t>(row_id)); | |
| row_store_read_struct.row_store_buffer.clear(); | |
| RETURN_IF_ERROR(scope_timer_run( | |
| [&]() { | |
| return tablet->lookup_row_data({}, loc, rowset, stats, | |
| row_store_read_struct.row_store_buffer, | |
| false, &io_ctx); | |
| }, | |
| lookup_row_data_ms)); | |
| RETURN_IF_ERROR(JsonbSerializeUtil::jsonb_to_columns( | |
| row_store_read_struct.serdes, row_store_read_struct.row_store_buffer.data(), | |
| row_store_read_struct.row_store_buffer.size(), | |
| row_store_read_struct.col_uid_to_idx, result_columns, | |
| row_store_read_struct.default_values, {})); | |
| } | |
| } else { | |
| for (int x = 0; x < slots.size(); ++x) { | |
| auto column_guard = result_block.mutate_column_scoped(x); | |
| MutableColumnPtr& column = column_guard.mutable_column(); | |
| IteratorKey iterator_key {.tablet_id = tablet_id, | |
| .rowset_id = rowset_id, | |
| .segment_id = segment_id, | |
| .slot_id = slots[x].id()}; | |
| IteratorItem& iterator_item = | |
| iterator_map.try_emplace(iterator_key, stats).first->second; | |
| if (iterator_item.segment == nullptr) { | |
| iterator_item.segment = segment; | |
| iterator_item.storage_read_options.io_ctx.reader_type = ReaderType::READER_QUERY; | |
| iterator_item.storage_read_options.io_ctx.file_cache_miss_policy = | |
| file_cache_miss_policy; | |
| // A rowid fetch bypasses TabletReader, so copy the rowset context it would have | |
| // supplied. For example, a rowset with commit_tso=100 must expose 100 through | |
| // __DORIS_COMMIT_TSO_COL__ instead of its on-disk placeholder 0. | |
| iterator_item.storage_read_options.version = rowset->version(); | |
| iterator_item.storage_read_options.commit_tso = rowset->commit_tso(); | |
| } | |
| if (x >= fetch_columns.size()) { | |
| return Status::InternalError( | |
| "fetch request carries {} column descs for {} slots, slot {} has none", | |
| fetch_columns.size(), slots.size(), slots[x].col_name()); | |
| } | |
| const TabletColumn& read_column = fetch_columns[x]; | |
| set_slot_access_paths(slots[x], read_column, iterator_item.storage_read_options); | |
| RETURN_IF_ERROR(segment->seek_and_read_by_rowid(read_column, &slots[x], row_ids, column, | |
| iterator_item.storage_read_options, | |
| iterator_item.iterator)); | |
| } | |
| } | |
| return Status::OK(); | |
| } | |
| } // namespace doris |