This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new bda646ee62d [Enhancement](file scanner) Support row-id fetch in
FileScannerV2 (#67906)
bda646ee62d is described below
commit bda646ee62d71c50831e4ca3f538ebb3631e7660
Author: Gabriel <[email protected]>
AuthorDate: Mon Sep 14 09:51:57 2026 +0800
[Enhancement](file scanner) Support row-id fetch in FileScannerV2 (#67906)
### What problem does this PR solve?
Problem Summary:
FileScannerV2 did not support selective reads by absolute file row
position. As a result, TopN two-phase materialization still had to use
the legacy FileScanner for its second-phase Parquet/ORC fetch.
This change carries the selected row positions through TableReader and
FileScanRequest, builds selective Parquet row ranges, seeks requested
ORC rows, and routes supported second-phase fetches through
FileScannerV2. Table-format variants that V2 does not support retain the
existing V1 fallback. Phase two reuses the phase-one scanner selection
policy, including the option presence bit, so disabling or omitting
`enable_file_scanner_v2` retains V1. Explicit row-ID requests bypass
whole-chunk Parquet cache prefetch and range merging. Rebuilt fetch
projections preserve Thrift slot presence bits and the pinned full
schema's non-regular column categories, including slots pruned from
phase one. Hive partition columns and synthesized metadata columns do
not consume physical column indexes. Fetches retain original Iceberg
file paths and row lineage while removing delete files, without mutating
shared file mappings.
### Release note
Support row-id fetch for Parquet and ORC in FileScannerV2.
### Check List (For Author)
- Test
- [x] Unit Test
- `ParquetScanTest.ReadsOnlyRequestedAbsoluteFileRowsAcrossRowGroups`
- `NewOrcReaderTest.ReadsOnlyRequestedAbsoluteFileRows`
- `RowIdStorageReaderTest.ExternalScannerSelectionRespectsRolloutOption`
-
`RowIdStorageReaderTest.ExternalScannerSelectionKeepsUnsupportedFormatsOnV1`
- `ParquetScanTest.SparseRowIdsAvoidCachedRemoteChunkPrefetch`
-
`RowIdStorageReaderTest.ExternalFetchPartitionSlotsPreserveHivePositionMapping`
-
`RowIdStorageReaderTest.ExternalFetchPreservesPrunedMetadataCategories`
- `RowIdStorageReaderTest.ExternalFetchPreservesIcebergFileMetadata`
- `FileQueryScanNodeTest.testRowIdFetchRetainsCategoriesOfPrunedColumns`
- Added lazy metadata comparisons to
`test_iceberg_file_metadata_columns` for Parquet and ORC.
- Validation: 283 focused BE unit tests passed under ASAN using
`run-be-ut.sh`; 16 targeted FE tests passed after temporarily excluding
an unrelated pre-existing IVM test compilation error. Clang-format 16,
FE Checkstyle, and build hygiene passed. Clang-tidy was blocked by
pre-existing unmatched `NOLINTEND` diagnostics. Full SQL integration
tests were not run locally.
- Behavior changed:
- [x] Yes. TopN two-phase materialization uses FileScannerV2 for
supported Parquet and ORC scans.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exec/operator/file_scan_operator.cpp | 10 +-
be/src/exec/operator/file_scan_operator.h | 4 +-
be/src/exec/rowid_fetcher.cpp | 135 +++++++++-----
be/src/exec/rowid_fetcher.h | 11 ++
be/src/exec/scan/file_scanner_v2.cpp | 124 +++++++++++--
be/src/exec/scan/file_scanner_v2.h | 15 ++
be/src/format_v2/file_reader.cpp | 8 +-
be/src/format_v2/file_reader.h | 20 ++
be/src/format_v2/orc/orc_reader.cpp | 42 ++++-
be/src/format_v2/orc/orc_reader.h | 1 +
be/src/format_v2/parquet/parquet_reader.h | 1 +
be/src/format_v2/parquet/parquet_scan.cpp | 48 +++--
be/src/format_v2/parquet/parquet_scan.h | 1 +
be/src/format_v2/table_reader.cpp | 9 +-
be/src/format_v2/table_reader.h | 7 +-
be/test/exec/rowid_fetcher_test.cpp | 202 ++++++++++++++++++++-
be/test/format_v2/orc/orc_reader_test.cpp | 32 ++++
be/test/format_v2/parquet/parquet_scan_test.cpp | 171 +++++++++++++++++
.../doris/datasource/scan/FileQueryScanNode.java | 28 ++-
.../datasource/scan/PluginDrivenScanNode.java | 5 +-
.../datasource/scan/FileQueryScanNodeTest.java | 40 ++++
gensrc/thrift/PlanNodes.thrift | 3 +
.../test_iceberg_file_metadata_columns.groovy | 16 ++
23 files changed, 839 insertions(+), 94 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index 5290e3078fb..6e2fb81602e 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -135,13 +135,13 @@ ScannerScheduler*
FileScanLocalState::scan_scheduler(RuntimeState* state) const
bool FileScanLocalState::TEST_should_use_file_scanner_v2(const TQueryOptions&
query_options,
bool is_load,
const
TFileScanRangeParams& scan_params) {
- return _should_use_file_scanner_v2(query_options, is_load, scan_params);
+ return should_use_file_scanner_v2(query_options, is_load, scan_params);
}
#endif
-bool FileScanLocalState::_should_use_file_scanner_v2(const TQueryOptions&
query_options,
- bool is_load,
- const
TFileScanRangeParams& scan_params) {
+bool FileScanLocalState::should_use_file_scanner_v2(const TQueryOptions&
query_options,
+ bool is_load,
+ const
TFileScanRangeParams& scan_params) {
// ADBC only has a FileScannerV2 reader, and enable_file_scanner_v2 is a
session variable marked
// fuzzy=true, so the regression harness flips it to false at random.
Without letting adbc
// through unconditionally, those queries land in v1, which has no "adbc"
branch, and come back
@@ -190,7 +190,7 @@ Status
FileScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
state()->desc_tbl().get_tuple_descriptor(scan_params->src_tuple_id) != nullptr;
// TODO: Use scanner v2 for all queries.
const bool use_file_scanner_v2 =
- _should_use_file_scanner_v2(state()->query_options(), is_load,
*scan_params);
+ should_use_file_scanner_v2(state()->query_options(), is_load,
*scan_params);
_operator_profile->add_info_string("UseScannerV2", use_file_scanner_v2 ?
"true" : "false");
const auto* output_tuple_desc =
state()->desc_tbl().get_tuple_descriptor(_output_tuple_id);
DORIS_CHECK(output_tuple_desc != nullptr);
diff --git a/be/src/exec/operator/file_scan_operator.h
b/be/src/exec/operator/file_scan_operator.h
index 6128715ddb7..45901d82858 100644
--- a/be/src/exec/operator/file_scan_operator.h
+++ b/be/src/exec/operator/file_scan_operator.h
@@ -54,6 +54,8 @@ public:
int max_scanners_concurrency(RuntimeState* state) const override;
int min_scanners_concurrency(RuntimeState* state) const override;
ScannerScheduler* scan_scheduler(RuntimeState* state) const override;
+ static bool should_use_file_scanner_v2(const TQueryOptions& query_options,
bool is_load,
+ const TFileScanRangeParams&
scan_params);
#ifdef BE_TEST
static bool TEST_should_use_file_scanner_v2(const TQueryOptions&
query_options, bool is_load,
const TFileScanRangeParams&
scan_params);
@@ -69,8 +71,6 @@ private:
return PushDownType::PARTIAL_ACCEPTABLE;
}
bool _push_down_topn(const RuntimePredicate& predicate) override;
- static bool _should_use_file_scanner_v2(const TQueryOptions&
query_options, bool is_load,
- const TFileScanRangeParams&
scan_params);
PushDownType _should_push_down_is_null_predicate(VectorizedFnCall*
fn_call) const override {
return fn_call->fn().name.function_name == "is_null_pred" ||
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index 05a62e70e13..fca26050618 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -42,7 +42,9 @@
#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"
@@ -437,6 +439,28 @@ const std::string
RowIdStorageReader::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,
@@ -467,25 +491,30 @@ Status
RowIdStorageReader::read_external_row_from_file_mapping(
scan_blocks[idx] = Block(scan_slots, read_ids.size());
auto& external_info = file_mapping->get_external_file_info();
- auto& scan_range_desc = external_info.scan_range_desc;
-
- // Clear to avoid reading iceberg position delete file...
- scan_range_desc.table_format_params.iceberg_params = TIcebergFileDesc {};
-
- // Clear to avoid reading hive transactional delete delta file...
- scan_range_desc.table_format_params.transactional_hive_params =
TTransactionalHiveDesc {};
+ 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");
{
- std::unique_ptr<FileScanner> vfile_scanner_ptr =
- FileScanner::create_unique(runtime_state.get(),
sub_runtime_profile.get(),
- &rpc_scan_params,
&colname_to_slot_id, &tuple_desc);
-
-
RETURN_IF_ERROR(vfile_scanner_ptr->prepare_for_read_lines(scan_range_desc));
- RETURN_IF_ERROR(vfile_scanner_ptr->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 (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()) {
@@ -506,7 +535,7 @@ Status
RowIdStorageReader::read_external_row_from_file_mapping(
}
auto file_read_bytes_counter =
-
sub_runtime_profile->get_counter(FileScanner::FileReadBytesProfile);
+
sub_runtime_profile->get_counter(FileScannerV2::FileReadBytesProfile);
if (file_read_bytes_counter != nullptr) {
fetch_statistics[idx].file_read_bytes = PrettyPrinter::print(
@@ -514,7 +543,7 @@ Status
RowIdStorageReader::read_external_row_from_file_mapping(
}
auto file_read_times_counter =
- sub_runtime_profile->get_counter(FileScanner::FileReadTimeProfile);
+
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());
@@ -609,6 +638,47 @@ Status RowIdStorageReader::submit_external_scan_tasks(
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,
@@ -641,15 +711,6 @@ Status RowIdStorageReader::read_batch_external_row(
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));
- rpc_scan_params = *old_scan_params;
-
- rpc_scan_params.required_slots.clear();
- rpc_scan_params.column_idxs.clear();
- rpc_scan_params.slot_name_to_schema_pos.clear();
-
- std::set
partition_name_set(first_scan_range_desc.columns_from_path_keys.begin(),
-
first_scan_range_desc.columns_from_path_keys.end());
-
std::unordered_map<std::string, size_t> source_column_to_scan_idx;
result_column_to_scan_column.reserve(slots.size());
@@ -668,24 +729,12 @@ Status RowIdStorageReader::read_batch_external_row(
}
}
- for (auto slot_idx = 0; slot_idx < scan_slots.size(); ++slot_idx) {
- auto& slot = scan_slots[slot_idx];
+ for (auto& slot : scan_slots) {
tuple_desc.add_slot(&slot);
colname_to_slot_id[slot.col_name()] = slot.id();
- TFileScanSlotInfo slot_info;
- slot_info.slot_id = slot.id();
- auto column_idx = scan_column_idxs[slot_idx];
- if (partition_name_set.contains(slot.col_name())) {
- //This is partition column.
- slot_info.is_file_slot = false;
- } else {
- rpc_scan_params.column_idxs.emplace_back(column_idx);
- slot_info.is_file_slot = true;
- }
- rpc_scan_params.default_value_of_src_slot.emplace(slot.id(), TExpr
{});
- rpc_scan_params.required_slots.emplace_back(slot_info);
- rpc_scan_params.slot_name_to_schema_pos.emplace(slot.col_name(),
column_idx);
}
+ 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();
@@ -866,9 +915,9 @@ Status RowIdStorageReader::read_batch_external_row(
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(FileScanner::FileReadBytesProfile,
+ runtime_profile->add_info_string(FileScannerV2::FileReadBytesProfile,
fmt::to_string(file_read_bytes_buffer));
- runtime_profile->add_info_string(FileScanner::FileReadTimeProfile,
+ runtime_profile->add_info_string(FileScannerV2::FileReadTimeProfile,
fmt::to_string(file_read_times_buffer));
}
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index 7c2dd3d1378..9b98d7b9ab1 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -35,6 +35,7 @@
namespace doris {
class RuntimeState;
+class TQueryOptions;
class TupleDescriptor;
class ScannerScheduler;
namespace io {
@@ -81,6 +82,9 @@ public:
static Status read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response);
private:
+ static bool should_use_file_scanner_v2(const TQueryOptions& query_options,
+ const TFileScanRangeParams&
scan_params,
+ const TFileRangeDesc& range);
struct ExternalFetchStatistics;
static Status read_doris_format_row(
@@ -118,6 +122,13 @@ private:
const std::unordered_map<std::string, int>& colname_to_slot_id,
std::counting_semaphore<>& semaphore, TupleDescriptor& tuple_desc);
+ static TFileRangeDesc build_external_fetch_range(const TFileRangeDesc&
source_range);
+
+ static TFileScanRangeParams 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);
+
static std::string source_column_key(const SlotDescriptor& slot, uint32_t
column_idx);
friend class RowIdStorageReaderTest;
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index 96387b92f66..c64d64fbd1f 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -75,9 +75,13 @@
#include "runtime/runtime_state.h"
#include "service/backend_options.h"
#include "storage/id_manager.h"
+#include "util/stopwatch.hpp"
#include "util/string_util.h"
namespace doris {
+const std::string FileScannerV2::FileReadBytesProfile = "FileReadBytes";
+const std::string FileScannerV2::FileReadTimeProfile = "FileReadTime";
+
namespace {
constexpr int kIcebergPositionDeleteContent = 1;
@@ -397,12 +401,12 @@ Status FileScannerV2::init(RuntimeState* state, const
VExprContextSPtrs& conjunc
file_scan_profile::SCANNER, 1);
_file_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile, "FileNumber",
TUnit::UNIT,
file_scan_profile::SCANNER,
1);
- _file_read_bytes_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"FileReadBytes", TUnit::BYTES,
-
file_scan_profile::IO, 1);
+ _file_read_bytes_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
FileReadBytesProfile,
+ TUnit::BYTES,
file_scan_profile::IO, 1);
_file_read_calls_counter = ADD_CHILD_COUNTER_WITH_LEVEL(profile,
"FileReadCalls", TUnit::UNIT,
file_scan_profile::IO, 1);
_file_read_time_counter =
- ADD_CHILD_TIMER_WITH_LEVEL(profile, "FileReadTime",
file_scan_profile::IO, 1);
+ ADD_CHILD_TIMER_WITH_LEVEL(profile, FileReadTimeProfile,
file_scan_profile::IO, 1);
_adaptive_batch_predicted_rows_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "AdaptiveBatchPredictedRows", TUnit::UNIT,
file_scan_profile::SCANNER, 1);
_adaptive_batch_actual_bytes_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
@@ -593,18 +597,21 @@ Status FileScannerV2::_init_table_reader(const
TFileRangeDesc& range) {
VExprContextSPtrs table_conjuncts;
RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts));
std::optional<std::vector<format::GlobalIndex>> push_down_count_columns;
- const auto& push_down_count_slot_ids =
_local_state->get_push_down_count_slot_ids();
- if (push_down_count_slot_ids.has_value()) {
- push_down_count_columns.emplace();
- push_down_count_columns->reserve(push_down_count_slot_ids->size());
- for (const auto slot_id : *push_down_count_slot_ids) {
- const auto global_index_it =
_slot_id_to_global_index.find(slot_id);
- if (global_index_it == _slot_id_to_global_index.end()) {
- return Status::InternalError(
- "Pushed-down COUNT argument is not a projected file
scan slot, slot_id={}",
- slot_id);
+ if (_local_state != nullptr) {
+ const auto& push_down_count_slot_ids =
_local_state->get_push_down_count_slot_ids();
+ if (push_down_count_slot_ids.has_value()) {
+ push_down_count_columns.emplace();
+ push_down_count_columns->reserve(push_down_count_slot_ids->size());
+ for (const auto slot_id : *push_down_count_slot_ids) {
+ const auto global_index_it =
_slot_id_to_global_index.find(slot_id);
+ if (global_index_it == _slot_id_to_global_index.end()) {
+ return Status::InternalError(
+ "Pushed-down COUNT argument is not a projected
file scan slot, "
+ "slot_id={}",
+ slot_id);
+ }
+ push_down_count_columns->push_back(global_index_it->second);
}
- push_down_count_columns->push_back(global_index_it->second);
}
}
RETURN_IF_ERROR(_table_reader->init({
@@ -614,15 +621,97 @@ Status FileScannerV2::_init_table_reader(const
TFileRangeDesc& range) {
.scan_params = const_cast<TFileScanRangeParams*>(_params),
.io_ctx = _io_ctx,
.runtime_state = _state,
- .scanner_profile = _local_state->scanner_profile(),
+ .scanner_profile = _local_state != nullptr ?
_local_state->scanner_profile() : _profile,
.file_slot_descs = &_file_slot_descs,
- .push_down_agg_type = _local_state->get_push_down_agg_type(),
+ .push_down_agg_type = _local_state != nullptr ?
_local_state->get_push_down_agg_type()
+ :
TPushAggOp::type::NONE,
.push_down_count_columns = std::move(push_down_count_columns),
- .condition_cache_digest =
_local_state->get_condition_cache_digest(),
+ .condition_cache_digest =
+ _local_state != nullptr ?
_local_state->get_condition_cache_digest() : 0,
}));
return Status::OK();
}
+Status FileScannerV2::read_by_rows(const TFileRangeDesc& range, const
std::list<int64_t>& row_ids,
+ Block* result_block, int64_t*
init_reader_ms,
+ int64_t* get_block_ms) {
+ DORIS_CHECK(result_block != nullptr);
+ DORIS_CHECK(init_reader_ms != nullptr);
+ DORIS_CHECK(get_block_ms != nullptr);
+ _current_range = range;
+ RETURN_IF_ERROR(_validate_scan_range(*_params, range));
+ const auto format_type = get_range_format_type(*_params, range);
+ if (format_type != TFileFormatType::FORMAT_PARQUET &&
+ format_type != TFileFormatType::FORMAT_ORC) {
+ return Status::NotSupported(
+ "FileScannerV2 row-id fetch supports only Parquet and ORC,
file format={}",
+ to_string(format_type));
+ }
+
+ _file_cache_statistics = std::make_unique<io::FileCacheStatistics>();
+ _file_reader_stats = std::make_unique<io::FileReaderStats>();
+ _file_read_bytes_counter =
+ ADD_COUNTER_WITH_LEVEL(_profile, FileReadBytesProfile,
TUnit::BYTES, 1);
+ _file_read_time_counter = ADD_TIMER_WITH_LEVEL(_profile,
FileReadTimeProfile, 1);
+ RETURN_IF_ERROR(_init_io_ctx());
+ _io_ctx->file_cache_stats = _file_cache_statistics.get();
+ _io_ctx->file_reader_stats = _file_reader_stats.get();
+ _io_ctx->is_disposable = _state->query_options().disable_file_cache;
+
+ MonotonicStopWatch init_watch;
+ init_watch.start();
+ auto init_status = [&]() -> Status {
+ RETURN_IF_ERROR(_create_table_reader_for_format(range,
&_table_reader));
+ DORIS_CHECK(_table_reader != nullptr);
+ RETURN_IF_ERROR(_init_expr_ctxes());
+ RETURN_IF_ERROR(_init_table_reader(range));
+ std::map<std::string, Field> partition_values;
+ RETURN_IF_ERROR(_generate_partition_values(range, &partition_values));
+ format::FileFormat current_split_format;
+ RETURN_IF_ERROR(_to_file_format(format_type, ¤t_split_format));
+ std::vector<int64_t> requested_rows(row_ids.begin(), row_ids.end());
+ _table_reader->set_batch_size(std::max<size_t>(requested_rows.size(),
1));
+ RETURN_IF_ERROR(_table_reader->prepare_split({
+ .partition_values = std::move(partition_values),
+ .conjuncts = std::nullopt,
+ .partition_prune_conjuncts = {},
+ .all_runtime_filters_applied = true,
+ .condition_cache_digest = 0,
+ .cache = nullptr,
+ .current_range = range,
+ .current_split_format = current_split_format,
+ .global_rowid_context = std::nullopt,
+ .row_ids = std::move(requested_rows),
+ }));
+ return Status::OK();
+ }();
+ *init_reader_ms += init_watch.elapsed_time() / 1000 / 1000;
+ RETURN_IF_ERROR(init_status);
+
+ MonotonicStopWatch read_watch;
+ read_watch.start();
+ auto read_status = [&]() -> Status {
+ Block read_block = result_block->clone_empty();
+ ScopedMutableBlock mutable_result(result_block);
+ bool eof = false;
+ while (!eof) {
+ RETURN_IF_ERROR(_table_reader->get_block(&read_block, &eof));
+ if (read_block.rows() > 0) {
+
RETURN_IF_ERROR(mutable_result.mutable_block().merge(read_block));
+ }
+ }
+ return Status::OK();
+ }();
+ *get_block_ms += read_watch.elapsed_time() / 1000 / 1000;
+ RETURN_IF_ERROR(read_status);
+
+ RETURN_IF_ERROR(_table_reader->close());
+ _table_reader.reset();
+ COUNTER_UPDATE(_file_read_bytes_counter, _file_reader_stats->read_bytes);
+ COUNTER_UPDATE(_file_read_time_counter, _file_reader_stats->read_time_ns);
+ return Status::OK();
+}
+
Status FileScannerV2::_create_table_reader_for_format(
const TFileRangeDesc& range, std::unique_ptr<format::TableReader>*
reader) const {
DORIS_CHECK(reader != nullptr);
@@ -693,6 +782,7 @@ Status FileScannerV2::_prepare_table_reader_split(const
TFileRangeDesc& range,
.current_range = range,
.current_split_format = current_split_format,
.global_rowid_context = _create_global_rowid_context(range),
+ .row_ids = std::nullopt,
}));
return Status::OK();
}
diff --git a/be/src/exec/scan/file_scanner_v2.h
b/be/src/exec/scan/file_scanner_v2.h
index ed1dce98821..13a6d593a3c 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -17,6 +17,7 @@
#pragma once
+#include <list>
#include <map>
#include <memory>
#include <optional>
@@ -53,6 +54,8 @@ class FileScannerV2 final : public Scanner {
public:
static constexpr const char* NAME = "FileScannerV2";
static constexpr size_t ADAPTIVE_BATCH_INITIAL_PROBE_ROWS = 32;
+ static const std::string FileReadBytesProfile;
+ static const std::string FileReadTimeProfile;
struct RealtimeCounterDeltas {
int64_t scan_rows = 0;
@@ -108,6 +111,18 @@ public:
ShardedKVCache* kv_cache,
const std::unordered_map<std::string, int>*
colname_to_slot_id);
+ // Standalone scanner used by TopN two-phase materialization.
+ FileScannerV2(RuntimeState* state, RuntimeProfile* profile, const
TFileScanRangeParams* params,
+ const std::unordered_map<std::string, int>*
colname_to_slot_id,
+ TupleDescriptor* tuple_desc)
+ : Scanner(state, profile), _params(params) {
+ (void)colname_to_slot_id;
+ _output_tuple_desc = tuple_desc;
+ }
+
+ Status read_by_rows(const TFileRangeDesc& range, const std::list<int64_t>&
row_ids,
+ Block* result_block, int64_t* init_reader_ms, int64_t*
get_block_ms);
+
Status init(RuntimeState* state, const VExprContextSPtrs& conjuncts)
override;
Status _open_impl(RuntimeState* state) override;
Status close(RuntimeState* state) override;
diff --git a/be/src/format_v2/file_reader.cpp b/be/src/format_v2/file_reader.cpp
index 8a9df5d7953..2d94a23d673 100644
--- a/be/src/format_v2/file_reader.cpp
+++ b/be/src/format_v2/file_reader.cpp
@@ -73,7 +73,13 @@ std::string FileScanRequest::debug_string() const {
}
out << column_id << ":" << block_position;
}
- out << "}, conjunct_count=" << conjuncts.size()
+ out << "}, row_ids=";
+ if (row_ids.has_value()) {
+ out << join_debug_strings(*row_ids, [](int64_t row_id) { return
std::to_string(row_id); });
+ } else {
+ out << "nullopt";
+ }
+ out << ", conjunct_count=" << conjuncts.size()
<< ", metadata_pruning_safe_conjunct_count=" <<
metadata_pruning_safe_conjunct_count
<< ", constant_pruning_safe_table_filter_count=" <<
constant_pruning_safe_table_filter_count
<< ", delete_conjunct_count=" << delete_conjuncts.size() << ",
variant_schema_overrides="
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index 91a7873d84d..700ccf9439d 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -18,6 +18,7 @@
#include <algorithm>
#include <cstddef>
#include <cstdint>
+#include <functional>
#include <limits>
#include <map>
#include <memory>
@@ -102,6 +103,12 @@ struct FileScanRequest {
// predicate_columns, the value is semantically required and must still be
validated and read.
std::vector<LocalColumnId> count_star_placeholder_columns;
+ // Absolute zero-based file row positions selected by a row-id fetch. A
present but empty
+ // vector means that no rows should be read; nullopt keeps the normal
sequential scan path.
+ // Readers require strictly increasing positions so they can seek forward
without duplicating
+ // output rows.
+ std::optional<std::vector<int64_t>> row_ids = std::nullopt;
+
// Table formats may assign semantics that legacy physical files do not
encode. Each path here
// identifies an unannotated Parquet group that the physical reader must
validate and decode as
// Variant. Keeping this explicit prevents generic Parquet scans from
guessing based on names.
@@ -374,8 +381,21 @@ public:
virtual std::unique_ptr<TableColumnMapper> create_column_mapper(
TableColumnMapperOptions options) const;
+ virtual bool supports_rowid_fetch() const { return false; }
+
// Open the file reader with file-local scan request. The file reader
should initialize its internal state according to the request, but does not
need to interpret table/global schema semantics. For example, all schema
change, filter localization, default/generated/partition columns should be
handled in table reader layer. This method can only be called after init()
successfully.
virtual Status open(std::shared_ptr<FileScanRequest> request) {
+ if (request->row_ids.has_value()) {
+ if (!supports_rowid_fetch()) {
+ return Status::NotSupported("File reader does not support
row-id fetch");
+ }
+ const auto& row_ids = *request->row_ids;
+ if (std::ranges::any_of(row_ids, [](int64_t row_id) { return
row_id < 0; }) ||
+ std::ranges::adjacent_find(row_ids, std::greater_equal<>()) !=
row_ids.end()) {
+ return Status::InvalidArgument(
+ "Row-id fetch requires non-negative, strictly
increasing file row ids");
+ }
+ }
_request = std::move(request);
return Status::OK();
}
diff --git a/be/src/format_v2/orc/orc_reader.cpp
b/be/src/format_v2/orc/orc_reader.cpp
index b59eafbadfb..b7ffc1c9134 100644
--- a/be/src/format_v2/orc/orc_reader.cpp
+++ b/be/src/format_v2/orc/orc_reader.cpp
@@ -778,6 +778,7 @@ struct OrcReaderScanState {
std::vector<StripeRange> selected_stripe_ranges;
size_t current_stripe_range = 0;
bool stripe_pruning_applied = false;
+ size_t next_row_id = 0;
bool row_reader_created = false;
};
@@ -1239,6 +1240,7 @@ Status
OrcReader::open(std::shared_ptr<format::FileScanRequest> request) {
return Status::Uninitialized("OrcReader is not open");
}
RETURN_IF_ERROR(format::FileReader::open(std::move(request)));
+ _state->next_row_id = 0;
if (_request->local_positions.empty()) {
size_t next_position = 0;
@@ -1296,6 +1298,16 @@ Status
OrcReader::open(std::shared_ptr<format::FileScanRequest> request) {
_apply_current_stripe_range();
RETURN_IF_ERROR(_create_row_reader());
+ if (_request->row_ids.has_value()) {
+ for (const int64_t row_id : *_request->row_ids) {
+ if (static_cast<uint64_t>(row_id) <
_state->row_reader_range_first_row ||
+ static_cast<uint64_t>(row_id) >=
_state->row_reader_range_end_row) {
+ return Status::InvalidArgument(
+ "ORC row id {} is outside the current split row range
[{}, {})", row_id,
+ _state->row_reader_range_first_row,
_state->row_reader_range_end_row);
+ }
+ }
+ }
_eof = get_total_rows() == 0;
return Status::OK();
}
@@ -1650,7 +1662,12 @@ Status OrcReader::_create_row_reader() {
_state->orc_lazy_read_enabled ? _orc_filter.get() : nullptr);
_state->selected_type = &_state->row_reader->getSelectedType();
DORIS_CHECK(_state->selected_type->getKind() ==
::orc::TypeKind::STRUCT);
- _state->batch =
_state->row_reader->createRowBatch(DEFAULT_ORC_READ_BATCH_SIZE);
+ // Row-id fetch seeks before every read; a one-row batch preserves
exact selection instead
+ // of also returning the sequential rows that follow the requested
position.
+ const uint64_t batch_size = _request != nullptr &&
_request->row_ids.has_value()
+ ? 1
+ : DEFAULT_ORC_READ_BATCH_SIZE;
+ _state->batch = _state->row_reader->createRowBatch(batch_size);
_state->orc_lazy_selection_valid = false;
_state->orc_lazy_selected_rows.clear();
_state->orc_lazy_input_rows = 0;
@@ -1947,11 +1964,23 @@ Status OrcReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
}
bool has_next = false;
+ std::optional<uint64_t> fetched_row_id;
while (true) {
try {
+ if (_request->row_ids.has_value()) {
+ if (_state->next_row_id >= _request->row_ids->size()) {
+ _eof = true;
+ *eof = true;
+ return Status::OK();
+ }
+ fetched_row_id =
static_cast<uint64_t>((*_request->row_ids)[_state->next_row_id]);
+ _state->row_reader->seekToRow(*fetched_row_id);
+ }
// Condition-cache seeks can perform I/O, so keep them in the same
cancellation
// boundary as next().
- _skip_condition_cache_false_granules(rows, eof);
+ if (!_request->row_ids.has_value()) {
+ _skip_condition_cache_false_granules(rows, eof);
+ }
if (*eof) {
return Status::OK();
}
@@ -1959,6 +1988,9 @@ Status OrcReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
_state->orc_lazy_selected_rows.clear();
_state->orc_lazy_input_rows = 0;
has_next = _state->row_reader->next(*_state->batch);
+ if (_request->row_ids.has_value() && has_next) {
+ ++_state->next_row_id;
+ }
} catch (const std::exception& e) {
if (is_orc_stop(_io_ctx.get(), e)) {
file_block->clear_column_data(file_block->columns());
@@ -1979,6 +2011,10 @@ Status OrcReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
}
break;
}
+ if (_request->row_ids.has_value()) {
+ return Status::InternalError("ORC row id {} could not be read from
the current split",
+ *fetched_row_id);
+ }
bool advanced = false;
RETURN_IF_ERROR(_advance_to_next_stripe_range(&advanced));
if (!advanced) {
@@ -1989,7 +2025,7 @@ Status OrcReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
}
const auto batch_rows = static_cast<size_t>(_state->batch->numElements);
- const auto batch_first_row = _state->row_reader->getRowNumber();
+ const auto batch_first_row =
fetched_row_id.value_or(_state->row_reader->getRowNumber());
_state->current_batch_first_row = batch_first_row;
_state->condition_cache_next_row = _state->current_batch_first_row +
batch_rows;
auto* struct_batch =
dynamic_cast<::orc::StructVectorBatch*>(_state->batch.get());
diff --git a/be/src/format_v2/orc/orc_reader.h
b/be/src/format_v2/orc/orc_reader.h
index a67dc5ad7f5..a58af5451a7 100644
--- a/be/src/format_v2/orc/orc_reader.h
+++ b/be/src/format_v2/orc/orc_reader.h
@@ -58,6 +58,7 @@ public:
Status get_schema(std::vector<format::ColumnDefinition>* const
file_schema) const override;
std::unique_ptr<format::TableColumnMapper> create_column_mapper(
format::TableColumnMapperOptions options) const override;
+ bool supports_rowid_fetch() const override { return true; }
Status open(std::shared_ptr<format::FileScanRequest> request) override;
Status get_block(Block* file_block, size_t* rows, bool* eof) override;
Status get_aggregate_result(const format::FileAggregateRequest& request,
diff --git a/be/src/format_v2/parquet/parquet_reader.h
b/be/src/format_v2/parquet/parquet_reader.h
index 5a1ebd64614..61aad3362bb 100644
--- a/be/src/format_v2/parquet/parquet_reader.h
+++ b/be/src/format_v2/parquet/parquet_reader.h
@@ -68,6 +68,7 @@ public:
std::unique_ptr<format::TableColumnMapper> create_column_mapper(
format::TableColumnMapperOptions options) const override;
+ bool supports_rowid_fetch() const override { return true; }
Status open(std::shared_ptr<format::FileScanRequest> request) override;
diff --git a/be/src/format_v2/parquet/parquet_scan.cpp
b/be/src/format_v2/parquet/parquet_scan.cpp
index b4cd731f12b..dc2741f7d3e 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -556,7 +556,28 @@ Status build_native_row_group_read_plans(
row_group_plan.row_group_id = row_group_idx;
row_group_plan.first_file_row = row_group_first_rows[row_group_idx];
row_group_plan.row_group_rows = row_group.num_rows;
- row_group_plan.selected_ranges = {{.start = 0, .length =
row_group.num_rows}};
+ if (request.row_ids.has_value()) {
+ const auto& row_ids = *request.row_ids;
+ const int64_t row_group_end = row_group_plan.first_file_row +
row_group.num_rows;
+ auto row_id = std::ranges::lower_bound(row_ids,
row_group_plan.first_file_row);
+ const auto row_id_end = std::ranges::lower_bound(row_id,
row_ids.end(), row_group_end);
+ for (; row_id != row_id_end; ++row_id) {
+ const int64_t local_row = *row_id -
row_group_plan.first_file_row;
+ if (!row_group_plan.selected_ranges.empty() &&
+ row_group_plan.selected_ranges.back().start +
+
row_group_plan.selected_ranges.back().length ==
+ local_row) {
+ ++row_group_plan.selected_ranges.back().length;
+ } else {
+ row_group_plan.selected_ranges.push_back({.start =
local_row, .length = 1});
+ }
+ }
+ if (row_group_plan.selected_ranges.empty()) {
+ continue;
+ }
+ } else {
+ row_group_plan.selected_ranges = {{.start = 0, .length =
row_group.num_rows}};
+ }
row_group_plan.expensive_pruning_pending = true;
plan->row_groups.push_back(std::move(row_group_plan));
}
@@ -1214,14 +1235,14 @@ Status ParquetScanScheduler::open_next_row_group(
RETURN_IF_ERROR(detail::build_native_prefetch_ranges(
thrift_metadata, file_schema, request_scan_columns(request),
row_group_idx,
file_context.native_file->size(), compat.parquet_816_padding,
&native_ranges));
- if (request.non_predicate_positions.empty()) {
+ if (!request.row_ids.has_value() &&
request.non_predicate_positions.empty()) {
_current_merge_range_active =
file_context.set_native_random_access_ranges(
native_ranges,
detail::average_prefetch_range_size(native_ranges), _profile,
_merge_read_slice_size);
} else {
- // Independent predicate/output readers may revisit the same physical
leaf at different
- // cursors. MergeRangeFileReader has one consumptive cache per range,
so use the random
- // access reader for this layout instead of sharing one sequential
range cache.
+ // Row-ID reads must not merge whole chunks containing unselected
rows. Independent
+ // predicate/output readers also need random access: they can revisit
one physical leaf
+ // at different cursors, while MergeRangeFileReader has one
consumptive cache per range.
_current_merge_range_active =
file_context.set_native_random_access_ranges(
{}, 0, _profile, _merge_read_slice_size);
}
@@ -1263,8 +1284,9 @@ Status ParquetScanScheduler::open_next_row_group(
// changes row/column materialization order.
if (!_current_merge_range_active) {
const auto prefetch_columns =
adaptive_predicate_prefetch_columns(request);
- RETURN_IF_ERROR(prefetch_current_row_group_columns(
- file_context, file_schema, prefetch_columns,
&_current_predicate_prefetched));
+ RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context,
file_schema, request,
+ prefetch_columns,
+
&_current_predicate_prefetched));
}
for (const auto& col : request.non_predicate_columns) {
const auto local_id = col.column_id();
@@ -1302,7 +1324,7 @@ Status ParquetScanScheduler::open_next_row_group(
// With no row-level filters there is no lazy-read decision to wait
for, so start warming
// output chunks immediately after their readers are created. Filtered
scans still defer
// this until at least one row survives the predicate phase.
- RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context,
file_schema,
+ RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context,
file_schema, request,
physical_non_predicate_columns(request),
&_current_non_predicate_prefetched));
}
@@ -2519,10 +2541,14 @@ Status
ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
Status ParquetScanScheduler::prefetch_current_row_group_columns(
ParquetFileContext& file_context,
const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ const format::FileScanRequest& request,
const std::vector<format::LocalColumnIndex>& scan_columns, bool*
prefetched) {
DORIS_CHECK(prefetched != nullptr);
- if (_current_merge_range_active || *prefetched || scan_columns.empty() ||
- _current_row_group_id < 0 || file_context.native_metadata == nullptr) {
+ // Row-ID requests remain selective even without conjuncts. Whole-chunk
dry-run prefetch
+ // would download unselected bytes without query accounting; demand reads
retain IOContext stats.
+ if (request.row_ids.has_value() || _current_merge_range_active ||
*prefetched ||
+ scan_columns.empty() || _current_row_group_id < 0 ||
+ file_context.native_metadata == nullptr) {
return Status::OK();
}
*prefetched = true;
@@ -2640,7 +2666,7 @@ Status ParquetScanScheduler::read_current_row_group_batch(
// Do not prefetch lazy output columns until at least one row survives
filtering. This is
// the same decision point where the v2 reader switches from
predicate-only reads to
// materializing non-predicate columns, so fully filtered batches
avoid unnecessary IO.
- RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context,
file_schema,
+ RETURN_IF_ERROR(prefetch_current_row_group_columns(file_context,
file_schema, request,
physical_non_predicate_columns(request),
&_current_non_predicate_prefetched));
}
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index 0cccbc31cb2..25367c84595 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -242,6 +242,7 @@ private:
Status prefetch_current_row_group_columns(
ParquetFileContext& file_context,
const std::vector<std::unique_ptr<ParquetColumnSchema>>&
file_schema,
+ const format::FileScanRequest& request,
const std::vector<format::LocalColumnIndex>& scan_columns, bool*
prefetched);
Status read_current_row_group_batch(
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 76aec676893..3e654ab594a 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -1161,6 +1161,7 @@ Status TableReader::refresh_conjuncts(VExprContextSPtrs
conjuncts) {
_file_scan_request == nullptr ? nullptr :
&_file_scan_request->local_positions,
_file_scan_request == nullptr ? nullptr
:
&_file_scan_request->non_predicate_positions));
+ refreshed_request->row_ids = _row_ids;
// A refresh does not prove that every future runtime filter has arrived.
Keep carrier values
// available whenever the split started with pending filters.
if (_push_down_agg_type == TPushAggOp::type::COUNT &&
_push_down_count_columns.has_value() &&
@@ -1470,6 +1471,7 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
?
std::make_optional(options.current_range.load_id)
: std::nullopt;
_global_rowid_context = options.global_rowid_context;
+ _row_ids = options.row_ids;
_delete_rows = nullptr;
_deletion_vector = nullptr;
_aggregate_pushdown_tried = false;
@@ -1494,9 +1496,10 @@ Status TableReader::prepare_split(const
SplitReadOptions& options) {
// the NULL state of a COUNT argument. Require the new FE's explicit empty
argument list, which
// means COUNT(*)/COUNT(1). A non-empty list means COUNT(col), while
nullopt comes from an old FE
// whose COUNT semantics are unknown during a BE-first rolling upgrade.
- if (_push_down_agg_type == TPushAggOp::type::COUNT &&
_push_down_count_columns.has_value() &&
- _push_down_count_columns->empty() &&
options.all_runtime_filters_applied &&
- _conjuncts.empty() &&
options.current_range.__isset.table_format_params &&
+ if (!_row_ids.has_value() && _push_down_agg_type ==
TPushAggOp::type::COUNT &&
+ _push_down_count_columns.has_value() &&
_push_down_count_columns->empty() &&
+ options.all_runtime_filters_applied && _conjuncts.empty() &&
+ options.current_range.__isset.table_format_params &&
options.current_range.table_format_params.__isset.table_level_row_count) {
DORIS_CHECK(options.current_range.table_format_params.table_level_row_count >=
-1);
_remaining_table_level_count =
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 097cee93734..1f890a514fe 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -186,6 +186,9 @@ struct SplitReadOptions {
TFileRangeDesc current_range;
FileFormat current_split_format = FileFormat::PARQUET;
std::optional<GlobalRowIdContext> global_rowid_context;
+ // Optional absolute file-row selection used by TopN two-phase
materialization. TableReader
+ // carries it unchanged into the format-neutral FileScanRequest.
+ std::optional<std::vector<int64_t>> row_ids = std::nullopt;
};
// Base class for table-level readers.
@@ -456,6 +459,7 @@ protected:
auto file_request = std::make_shared<FileScanRequest>();
RETURN_IF_ERROR(_data_reader.column_mapper->create_scan_request(
_table_filters, _projected_columns, file_request.get(),
_runtime_state));
+ file_request->row_ids = _row_ids;
_constant_pruning_safe_filter_count =
std::min(_constant_pruning_safe_filter_count,
file_request->constant_pruning_safe_table_filter_count);
@@ -1048,7 +1052,7 @@ protected:
*pushed_down = false;
block->clear_column_data(_projected_columns.size());
_aggregate_pushdown_tried = true;
- if (!_supports_aggregate_pushdown(_push_down_agg_type)) {
+ if (_row_ids.has_value() ||
!_supports_aggregate_pushdown(_push_down_agg_type)) {
return Status::OK();
}
@@ -2251,6 +2255,7 @@ protected:
// irreversible aggregate rows, not only the table-level row-count
shortcut in prepare_split().
bool _all_runtime_filters_applied_for_split = true;
std::optional<GlobalRowIdContext> _global_rowid_context;
+ std::optional<std::vector<int64_t>> _row_ids;
bool _aggregate_pushdown_tried = false;
bool _current_split_pruned = false;
TableColumnMapperOptions _mapper_options;
diff --git a/be/test/exec/rowid_fetcher_test.cpp
b/be/test/exec/rowid_fetcher_test.cpp
index b2fcbbac236..8f14aa6654b 100644
--- a/be/test/exec/rowid_fetcher_test.cpp
+++ b/be/test/exec/rowid_fetcher_test.cpp
@@ -22,8 +22,13 @@
#include <string>
#include <vector>
+#include "exec/operator/file_scan_operator.h"
+#include "exec/scan/file_scanner_v2.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/table/hive_reader.h"
#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"
+#include "runtime/runtime_state.h"
namespace doris {
@@ -40,6 +45,8 @@ public:
protected:
struct SlotSpec {
std::string col_name = "c";
+ int32_t slot_id = 0;
+ PrimitiveType type = TYPE_INT;
int32_t col_unique_id = 1;
std::vector<std::string> column_paths = {};
TColumnAccessPaths access_paths = {};
@@ -47,11 +54,12 @@ protected:
static SlotDescriptor make_slot(const SlotSpec& spec) {
TSlotDescriptor tdesc = TSlotDescriptorBuilder()
- .type(TYPE_INT)
+ .type(spec.type)
.nullable(true)
.column_name(spec.col_name)
.column_pos(0)
.build();
+ tdesc.__set_id(spec.slot_id);
tdesc.__set_col_unique_id(spec.col_unique_id);
tdesc.__set_column_paths(spec.column_paths);
if (!spec.access_paths.empty()) {
@@ -76,6 +84,198 @@ protected:
}
};
+TEST_F(RowIdStorageReaderTest, ExternalScannerSelectionRespectsRolloutOption) {
+ for (auto format : {TFileFormatType::FORMAT_PARQUET,
TFileFormatType::FORMAT_ORC}) {
+ TFileScanRangeParams params;
+ params.__set_format_type(format);
+ TFileRangeDesc range;
+ for (const auto& table_format : {"hive", "iceberg", "tvf"}) {
+ TTableFormatFileDesc table;
+ table.__set_table_format_type(table_format);
+ params.__set_table_format_params(table);
+ range.__set_table_format_params(table);
+ for (int option = 0; option < 3; ++option) {
+ TQueryOptions options;
+ if (option != 0) {
+ options.__set_enable_file_scanner_v2(option == 2);
+ } else {
+ // An absent Thrift field must not enable V2 even if its
value defaults to true.
+ options.enable_file_scanner_v2 = true;
+ options.__isset.enable_file_scanner_v2 = false;
+ }
+
EXPECT_EQ(RowIdStorageReader::should_use_file_scanner_v2(options, params,
range),
+
FileScanLocalState::TEST_should_use_file_scanner_v2(options, false,
+
params));
+
EXPECT_EQ(RowIdStorageReader::should_use_file_scanner_v2(options, params,
range),
+ option == 2);
+ }
+ }
+ }
+}
+
+TEST_F(RowIdStorageReaderTest,
ExternalScannerSelectionKeepsUnsupportedFormatsOnV1) {
+ TQueryOptions options;
+ options.__set_enable_file_scanner_v2(true);
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ TFileRangeDesc range;
+ range.__set_format_type(TFileFormatType::FORMAT_JNI);
+ EXPECT_FALSE(RowIdStorageReader::should_use_file_scanner_v2(options,
params, range));
+ range.__set_format_type(TFileFormatType::FORMAT_ORC);
+ TTableFormatFileDesc table;
+ table.__set_table_format_type("transactional_hive");
+ params.__set_table_format_params(table);
+ range.__set_table_format_params(table);
+ EXPECT_FALSE(RowIdStorageReader::should_use_file_scanner_v2(options,
params, range));
+}
+
+TEST_F(RowIdStorageReaderTest, ExternalFetchPreservesIcebergFileMetadata) {
+ TFileRangeDesc range;
+ range.__set_path("normalized/data.parquet");
+ TIcebergFileDesc iceberg;
+ iceberg.__set_original_file_path("s3://bucket/data.parquet");
+ iceberg.__set_format_version(3);
+ iceberg.__set_first_row_id(128);
+ iceberg.__set_last_updated_sequence_number(7);
+ TIcebergDeleteFileDesc deletes;
+ deletes.__set_path("s3://bucket/deletes.parquet");
+ iceberg.__set_delete_files({deletes});
+ range.table_format_params.__set_iceberg_params(iceberg);
+
+ const auto fetch_range =
RowIdStorageReader::build_external_fetch_range(range);
+ const auto& fetch_iceberg = fetch_range.table_format_params.iceberg_params;
+ EXPECT_TRUE(fetch_iceberg.__isset.original_file_path);
+ EXPECT_EQ(fetch_iceberg.original_file_path, iceberg.original_file_path);
+ EXPECT_EQ(fetch_iceberg.format_version, 3);
+ EXPECT_EQ(fetch_iceberg.first_row_id, 128);
+ EXPECT_EQ(fetch_iceberg.last_updated_sequence_number, 7);
+ EXPECT_TRUE(fetch_iceberg.delete_files.empty());
+ EXPECT_EQ(range.table_format_params.iceberg_params, iceberg);
+}
+
+TEST_F(RowIdStorageReaderTest, ExternalFetchPreservesPrunedMetadataCategories)
{
+ TFileScanRangeParams source_params;
+ source_params.__set_column_name_to_category(
+ {{"_file", TColumnCategory::SYNTHESIZED},
+ {"_pos", TColumnCategory::SYNTHESIZED},
+ {"generated_col", TColumnCategory::GENERATED},
+ {"partition_col", TColumnCategory::PARTITION_KEY}});
+ // Phase one projects only the sort key; none of these fetch slots
survives in required_slots.
+ TFileScanSlotInfo sort_slot;
+ sort_slot.__set_slot_id(99);
+ sort_slot.__set_category(TColumnCategory::REGULAR);
+ source_params.__set_required_slots({sort_slot});
+ source_params.__set_column_idxs({0});
+ std::vector<SlotDescriptor> slots;
+ for (const auto* name : {"_file", "_pos", "generated_col",
"partition_col", "value"}) {
+ slots.emplace_back(
+ make_slot({.col_name = name,
+ .slot_id = static_cast<int32_t>(slots.size()),
+ .type = std::string_view(name) == "_file" ?
TYPE_STRING : TYPE_BIGINT}));
+ }
+ const auto params = RowIdStorageReader::build_external_scan_params(
+ source_params, TFileRangeDesc {}, slots, {3, 4, 1, 2, 0});
+ EXPECT_EQ(params.column_idxs, (std::vector<int32_t> {1, 0}));
+ const std::vector<TColumnCategory::type> categories {
+ TColumnCategory::SYNTHESIZED, TColumnCategory::SYNTHESIZED,
TColumnCategory::GENERATED,
+ TColumnCategory::PARTITION_KEY, TColumnCategory::REGULAR};
+ std::vector<format::ColumnDefinition> columns;
+ for (size_t i = 0; i < slots.size(); ++i) {
+ const auto& info = params.required_slots[i];
+ EXPECT_TRUE(info.__isset.category);
+ EXPECT_EQ(info.category, categories[i]);
+ EXPECT_EQ(info.is_file_slot, i == 2 || i == 4);
+ EXPECT_EQ(FileScannerV2::TEST_is_partition_slot(info,
slots[i].col_name()), i == 3);
+ auto column = FileScannerV2::_build_table_column(&slots[i]);
+ column.is_synthesized =
+ info.__isset.category && info.category ==
TColumnCategory::SYNTHESIZED;
+ columns.emplace_back(std::move(column));
+ }
+ format::TableColumnMapper mapper({.mode =
format::TableColumnMappingMode::BY_NAME,
+ .enable_iceberg_metadata_virtual_columns
= true});
+ ASSERT_TRUE(mapper.create_mapping({columns[0], columns[1]}, {}, {}).ok());
+ EXPECT_EQ(mapper.mappings()[0].virtual_column_type,
+ format::TableVirtualColumnType::ICEBERG_FILE_PATH);
+ EXPECT_EQ(mapper.mappings()[1].virtual_column_type,
+ format::TableVirtualColumnType::ICEBERG_ROW_POSITION);
+
+ // An authoritative empty map means ordinary physical columns, even for
metadata spellings.
+ source_params.__set_column_name_to_category({});
+ const auto physical_params =
RowIdStorageReader::build_external_scan_params(
+ source_params, TFileRangeDesc {}, slots, {3, 4, 1, 2, 0});
+ for (const auto& info : physical_params.required_slots) {
+ EXPECT_EQ(info.category, TColumnCategory::REGULAR);
+ EXPECT_TRUE(info.is_file_slot);
+ }
+}
+
+// Row-id fetch rebuilds the projection after TopN. Hive's positional mapper
must consume
+// indexes only for physical columns, including when partition columns precede
file columns.
+TEST_F(RowIdStorageReaderTest,
ExternalFetchPartitionSlotsPreserveHivePositionMapping) {
+ for (const auto format : {TFileFormatType::FORMAT_ORC,
TFileFormatType::FORMAT_PARQUET}) {
+ TQueryOptions options;
+ options.__set_hive_orc_use_column_names(false);
+ options.__set_hive_parquet_use_column_names(false);
+ RuntimeState state(options, TQueryGlobals {});
+ TFileScanRangeParams source_params;
+ source_params.__set_format_type(format);
+ source_params.__set_column_idxs({0, 1, 2});
+ TFileScanSlotInfo old_slot;
+ old_slot.__set_slot_id(99);
+ source_params.__set_required_slots({old_slot});
+ source_params.__set_slot_name_to_schema_pos({{"old_column", 0}});
+ TFileRangeDesc range;
+ range.__set_columns_from_path_keys({"partition_col"});
+
+ for (const auto& names : {std::vector<std::string> {"value",
"partition_col"},
+ std::vector<std::string> {"partition_col",
"value", "id"},
+ std::vector<std::string> {"partition_col"},
+ std::vector<std::string> {"value", "id"}}) {
+ SCOPED_TRACE(fmt::format("format={}, columns={}",
static_cast<int>(format),
+ fmt::join(names, ",")));
+ std::vector<SlotDescriptor> slots;
+ std::vector<uint32_t> indices;
+ std::vector<int32_t> file_indices;
+ for (const auto& name : names) {
+ slots.emplace_back(make_slot(
+ {.col_name = name, .slot_id =
static_cast<int32_t>(slots.size())}));
+ const uint32_t index = name == "partition_col" ? 3 : name ==
"value" ? 2 : 0;
+ indices.emplace_back(index);
+ if (name != "partition_col") {
+ file_indices.emplace_back(index);
+ }
+ }
+ const auto params =
RowIdStorageReader::build_external_scan_params(source_params, range,
+
slots, indices);
+ ASSERT_EQ(params.required_slots.size(), slots.size());
+ EXPECT_EQ(params.column_idxs, file_indices);
+
EXPECT_FALSE(params.slot_name_to_schema_pos.contains("old_column"));
+ format::ProjectedColumnBuildContext context {
+ .scan_params = ¶ms, .range = &range, .runtime_state =
&state};
+ format::hive::HiveReader reader;
+ for (size_t i = 0; i < slots.size(); ++i) {
+ const auto& slot_info = params.required_slots[i];
+ const auto& name = names[i];
+ const bool is_partition = name == "partition_col";
+ EXPECT_TRUE(slot_info.__isset.slot_id);
+ EXPECT_EQ(slot_info.slot_id, slots[i].id());
+ EXPECT_TRUE(slot_info.__isset.is_file_slot);
+ EXPECT_EQ(FileScannerV2::TEST_is_partition_slot(slot_info,
name), is_partition);
+ format::ColumnDefinition column;
+ column.name = name;
+ column.type = slots[i].get_data_type_ptr();
+ const auto status =
reader.annotate_projected_column(slot_info, &context, &column);
+ ASSERT_TRUE(status.ok()) << status;
+ if (!is_partition) {
+ EXPECT_EQ(column.get_identifier_position(), indices[i]);
+ }
+ }
+ EXPECT_EQ(context.next_file_column_idx, file_indices.size());
+ EXPECT_TRUE(reader.validate_projected_columns(context).ok());
+ }
+ }
+}
+
TEST_F(RowIdStorageReaderTest, SameSourceColumnSharesKey) {
// The bug case: one physical column projected twice must dedup onto one
scan column.
const SlotDescriptor first = make_slot({});
diff --git a/be/test/format_v2/orc/orc_reader_test.cpp
b/be/test/format_v2/orc/orc_reader_test.cpp
index 67b5d2a1676..d9be5a5e6d0 100644
--- a/be/test/format_v2/orc/orc_reader_test.cpp
+++ b/be/test/format_v2/orc/orc_reader_test.cpp
@@ -10212,6 +10212,38 @@ TEST_F(NewOrcReaderTest, CloseClearsFileLocalState) {
EXPECT_FALSE(reader->open(request).ok());
}
+TEST_F(NewOrcReaderTest, ReadsOnlyRequestedAbsoluteFileRows) {
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->non_predicate_columns = {field_projection(0)};
+ request->row_ids = {0, 2, 4};
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block({schema[0]});
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column = assert_cast<const ColumnInt32&>(
+ assert_cast<const
ColumnNullable&>(*block.get_by_position(0).column)
+ .get_nested_column());
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({1, 3, 5}));
+}
+
TEST_F(NewOrcReaderTest, ReadPrimitiveTypesWithNulls) {
const auto primitive_file_path = (_test_dir / "primitive.orc").string();
write_primitive_orc_file(primitive_file_path);
diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp
b/be/test/format_v2/parquet/parquet_scan_test.cpp
index 17d80897b7a..6d10bc2d427 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -25,7 +25,9 @@
#include <parquet/arrow/writer.h>
#include <parquet/encoding.h>
+#include <atomic>
#include <bit>
+#include <chrono>
#include <cstdlib>
#include <cstring>
#include <filesystem>
@@ -36,6 +38,7 @@
#include <optional>
#include <set>
#include <string>
+#include <thread>
#include <utility>
#include <vector>
@@ -72,6 +75,11 @@
#include "format_v2/parquet/reader/native_column_reader.h"
#include "gen_cpp/PlanNodes_types.h"
#include "gen_cpp/Types_types.h"
+#include "io/cache/block_file_cache.h"
+#include "io/cache/block_file_cache_factory.h"
+#include "io/cache/cached_remote_file_reader.h"
+#include "io/cache/fs_file_cache_storage.h"
+#include "io/fs/local_file_system.h"
#include "io/io_common.h"
#include "runtime/runtime_state.h"
#include "storage/index/zone_map/zonemap_eval_context.h"
@@ -79,6 +87,8 @@
#include "storage/utils.h"
#include "testutil/mock/mock_query_context.h"
#include "util/coding.h"
+#include "util/defer_op.h"
+#include "util/threadpool.h"
#include "util/thrift_util.h"
namespace doris {
@@ -2513,6 +2523,167 @@ TEST_F(ParquetScanTest,
GlobalRowIdUsesFileLocalPositionForScanRange) {
EXPECT_EQ(row_ids, std::vector<uint32_t>({2, 3}));
}
+TEST_F(ParquetScanTest, ReadsOnlyRequestedAbsoluteFileRowsAcrossRowGroups) {
+ write_int_pair_parquet_file(_file_path, 2);
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->non_predicate_columns = {field_projection(0)};
+ request->row_ids = {0, 3, 5};
+ ASSERT_TRUE(reader->open(request).ok());
+
+ std::vector<int32_t> ids;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block({schema[0]});
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ const auto& id_column =
int32_data_column(*block.get_by_position(0).column);
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ }
+ }
+
+ EXPECT_EQ(ids, std::vector<int32_t>({1, 4, 6}));
+}
+
+class CountingParquetRemoteReader final : public io::FileReader {
+public:
+ explicit CountingParquetRemoteReader(io::FileReaderSPtr reader) :
_reader(std::move(reader)) {}
+ Status close() override { return _reader->close(); }
+ const io::Path& path() const override { return _reader->path(); }
+ size_t size() const override { return _reader->size(); }
+ bool closed() const override { return _reader->closed(); }
+ int64_t mtime() const override { return _reader->mtime(); }
+ std::atomic<size_t> remote_bytes {0};
+ std::atomic<size_t> dryrun_bytes {0};
+
+protected:
+ Status read_at_impl(size_t offset, Slice result, size_t* bytes_read,
+ const io::IOContext* io_ctx) override {
+ RETURN_IF_ERROR(_reader->read_at(offset, result, bytes_read, io_ctx));
+ remote_bytes += *bytes_read;
+ if (io_ctx != nullptr && io_ctx->is_dryrun) {
+ dryrun_bytes += *bytes_read;
+ }
+ return Status::OK();
+ }
+
+private:
+ io::FileReaderSPtr _reader;
+};
+
+TEST_F(ParquetScanTest, SparseRowIdsAvoidCachedRemoteChunkPrefetch) {
+ using namespace format::parquet;
+ // Use multiple cache blocks so an eager chunk read cannot hide inside one
demand read.
+ std::vector<int32_t> ids(1024 * 1024);
+ std::iota(ids.begin(), ids.end(), 0);
+ auto table = arrow::Table::Make(arrow::schema({arrow::field("id",
arrow::int32(), false)}),
+ {build_int32_array(ids)});
+ write_table(_file_path, table, ids.size());
+
+ auto* env = ExecEnv::GetInstance();
+ auto* old_factory = env->file_cache_factory();
+ auto factory = std::make_unique<io::FileCacheFactory>();
+ const auto cache_path = (_test_dir / "cache").string();
+ io::FileCacheSettings settings;
+ settings.storage = "disk";
+ settings.capacity = 16 * 1024 * 1024;
+ settings.query_queue_size = settings.capacity;
+ settings.query_queue_elements = 1024;
+ settings.max_file_block_size = 64 * 1024;
+ const auto old_ttl_gc_interval =
config::file_cache_background_ttl_gc_interval_ms;
+ const auto old_ttl_info_interval =
config::file_cache_background_ttl_info_update_interval_ms;
+ // The TTL workers sleep between iterations, so bound the fixture's
shutdown latency.
+ config::file_cache_background_ttl_gc_interval_ms = 100;
+ config::file_cache_background_ttl_info_update_interval_ms = 100;
+ const auto old_column_buffer = config::parquet_column_max_buffer_mb;
+ config::parquet_column_max_buffer_mb = 1;
+ const auto old_block_size = config::file_cache_each_block_size;
+ config::file_cache_each_block_size = settings.max_file_block_size;
+ auto old_fd_cache = std::move(env->_file_cache_open_fd_cache);
+ auto old_pool = std::move(env->_segment_prefetch_thread_pool);
+ Defer restore([&] {
+ env->_segment_prefetch_thread_pool.reset();
+ factory.reset();
+ env->set_file_cache_factory(old_factory);
+ env->_file_cache_open_fd_cache = std::move(old_fd_cache);
+ env->_segment_prefetch_thread_pool = std::move(old_pool);
+ config::file_cache_each_block_size = old_block_size;
+ config::parquet_column_max_buffer_mb = old_column_buffer;
+ config::file_cache_background_ttl_gc_interval_ms = old_ttl_gc_interval;
+ config::file_cache_background_ttl_info_update_interval_ms =
old_ttl_info_interval;
+ });
+ env->set_file_cache_factory(factory.get());
+ env->_file_cache_open_fd_cache = std::make_unique<io::FDCache>();
+ ASSERT_TRUE(factory->create_file_cache(cache_path, settings).ok());
+ auto* cache = factory->get_by_path(cache_path);
+ ASSERT_NE(cache, nullptr);
+ for (int attempt = 0; attempt < 200 && !cache->get_async_open_success();
++attempt) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(5));
+ }
+ ASSERT_TRUE(cache->get_async_open_success());
+ ASSERT_TRUE(ThreadPoolBuilder("parquet_test_prefetch")
+ .set_min_threads(1)
+ .set_max_threads(1)
+
.build(&ExecEnv::GetInstance()->_segment_prefetch_thread_pool)
+ .ok());
+ io::FileReaderSPtr local;
+ ASSERT_TRUE(io::global_local_filesystem()->open_file(_file_path,
&local).ok());
+ auto remote = std::make_shared<CountingParquetRemoteReader>(local);
+ io::FileReaderOptions options;
+ options.cache_type = io::FileCachePolicy::FILE_BLOCK_CACHE;
+ options.cache_base_path = cache_path;
+ options.cache_write_mode = io::CacheWriteMode::SYNC_WRITE;
+ auto cached = std::make_shared<io::CachedRemoteFileReader>(remote,
options);
+ io::FileCacheStatistics cache_stats;
+ io::IOContext io_ctx;
+ io_ctx.file_cache_stats = &cache_stats;
+ io::FileDescription description;
+ description.path = _file_path;
+ description.file_size = local->size();
+ ParquetFileContext context;
+ ASSERT_TRUE(context.open(cached, &io_ctx, false, description).ok());
+ std::vector<std::unique_ptr<ParquetColumnSchema>> schema;
+ ASSERT_TRUE(build_parquet_column_schema(context.native_metadata->schema(),
&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->non_predicate_columns = {field_projection(0)};
+ request->local_positions.emplace(format::LocalColumnId(0),
format::LocalIndex(0));
+ request->row_ids = {0};
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RowGroupScanPlan plan;
+ ASSERT_TRUE(plan_parquet_row_groups(*context.native_metadata, schema,
*request, {}, false,
+ &plan, &state.timezone_obj(), &state,
&context)
+ .ok());
+ ParquetScanScheduler scheduler;
+ scheduler.set_plan(std::move(plan));
+ scheduler.set_scan_request(request);
+ scheduler.set_runtime_state(&state);
+ scheduler.set_timezone(&state.timezone_obj());
+ Block block;
+ block.insert({schema[0]->type->create_column(), schema[0]->type, "id"});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(scheduler.read_next_batch(context, schema, &block, &rows,
&eof).ok());
+ ExecEnv::GetInstance()->segment_prefetch_thread_pool()->wait();
+ ASSERT_EQ(rows, 1);
+
EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_element(0),
0);
+ EXPECT_FALSE(scheduler._current_merge_range_active);
+ EXPECT_FALSE(scheduler._current_non_predicate_prefetched);
+ EXPECT_LT(remote->remote_bytes.load(), local->size() / 2);
+ EXPECT_EQ(remote->dryrun_bytes.load(), 0);
+ // Source counters measure copied bytes; downloads can include cache-block
alignment padding.
+ EXPECT_GT(cache_stats.bytes_read_from_remote, 0);
+ EXPECT_LE(cache_stats.bytes_read_from_remote, remote->remote_bytes.load());
+}
+
TEST_F(ParquetScanTest, PredicateOnlyGlobalRowIdKeepsSignedFileLocalId) {
write_int_pair_parquet_file(_file_path, 6, false);
format::GlobalRowIdContext context {.version = 7, .backend_id = 123456789,
.file_id = 42};
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
index 775b811e5ee..63d45e48688 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
@@ -241,12 +241,16 @@ public abstract class FileQueryScanNode extends
FileScanNode {
setColumnPositionMapping();
}
+ protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String>
partitionKeys) {
+ return classifyColumn(slot.getColumn().getName(), partitionKeys);
+ }
+
/**
- * Classify a column's category for the BE reader.
+ * Classify projected and lazy columns with the same connector-specific
rules.
* Subclasses override this for format-specific classification.
*/
- protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String>
partitionKeys) {
- if (partitionKeys.contains(slot.getColumn().getName())) {
+ protected TColumnCategory classifyColumn(String columnName, List<String>
partitionKeys) {
+ if (partitionKeys.contains(columnName)) {
return TColumnCategory.PARTITION_KEY;
}
return TColumnCategory.REGULAR;
@@ -300,8 +304,24 @@ public abstract class FileQueryScanNode extends
FileScanNode {
// Pre-index columns into a Map for O(1) lookup
List<Column> columns = getPinnedFullSchema();
Map<String, Integer> columnNameMap = new HashMap<>(columns.size());
+ boolean needsRowIdFetch = desc.getSlots().stream()
+ .anyMatch(slot ->
slot.getColumn().getName().startsWith(Column.GLOBAL_ROWID_COL));
+ List<String> partitionKeys = needsRowIdFetch ? getPathPartitionKeys()
: Collections.emptyList();
+ Map<String, TColumnCategory> columnCategories = new HashMap<>();
for (int i = 0; i < columns.size(); i++) {
- columnNameMap.putIfAbsent(columns.get(i).getName(), i);
+ String columnName = columns.get(i).getName();
+ columnNameMap.putIfAbsent(columnName, i);
+ if (needsRowIdFetch) {
+ TColumnCategory category = classifyColumn(columnName,
partitionKeys);
+ if (category != TColumnCategory.REGULAR) {
+ columnCategories.put(columnName, category);
+ }
+ }
+ }
+ if (needsRowIdFetch) {
+ // Lazy slots are already absent from the scan tuple. Preserve
their categories from
+ // the pinned full schema so phase two cannot mistake metadata for
physical columns.
+ params.setColumnNameToCategory(columnCategories);
}
for (TFileScanSlotInfo slot : params.getRequiredSlots()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
index 3229689f017..9af04f0f833 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
@@ -782,8 +782,7 @@ public class PluginDrivenScanNode extends FileQueryScanNode
{
* Everything else falls through to {@code super} (partition key /
regular).
*/
@Override
- protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String>
partitionKeys) {
- String name = slot.getColumn().getName();
+ protected TColumnCategory classifyColumn(String name, List<String>
partitionKeys) {
if (name.startsWith(Column.GLOBAL_ROWID_COL)) {
return TColumnCategory.SYNTHESIZED;
}
@@ -794,7 +793,7 @@ public class PluginDrivenScanNode extends FileQueryScanNode
{
if (category == ConnectorColumnCategory.GENERATED) {
return TColumnCategory.GENERATED;
}
- return super.classifyColumn(slot, partitionKeys);
+ return super.classifyColumn(name, partitionKeys);
}
/**
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java
index 5f608d73497..1afdc77defe 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/FileQueryScanNodeTest.java
@@ -264,6 +264,46 @@ public class FileQueryScanNodeTest {
Assertions.assertFalse(node.getFileScanRangeParams().isSetHiveParquetTimeZone());
}
+ @Test
+ public void testRowIdFetchRetainsCategoriesOfPrunedColumns() throws
Exception {
+ TestFileQueryScanNode node = new TestFileQueryScanNode(new
SessionVariable()) {
+ @Override
+ protected TColumnCategory classifyColumn(String name, List<String>
partitionKeys) {
+ if (name.equals("_file") || name.equals("_pos")) {
+ return TColumnCategory.SYNTHESIZED;
+ }
+ if (name.equals("generated_col")) {
+ return TColumnCategory.GENERATED;
+ }
+ return super.classifyColumn(name, partitionKeys);
+ }
+ };
+ node.setTargetTable(table);
+ TupleDescriptor desc = node.getTupleDescriptor();
+ desc.setTable(table);
+ SlotDescriptor sortSlot = new SlotDescriptor(new SlotId(1),
desc.getId());
+ sortSlot.setColumn(new Column("id", Type.INT));
+ desc.addSlot(sortSlot);
+ SlotDescriptor rowIdSlot = new SlotDescriptor(new SlotId(2),
desc.getId());
+ rowIdSlot.setColumn(new Column(Column.GLOBAL_ROWID_COL, Type.STRING));
+ desc.addSlot(rowIdSlot);
+ List<Column> fullSchema = Arrays.asList(sortSlot.getColumn(), new
Column("_file", Type.STRING),
+ new Column("_pos", Type.BIGINT), new Column("generated_col",
Type.BIGINT));
+ Mockito.when(table.getBaseSchema(false)).thenReturn(fullSchema);
+ Mockito.when(table.getFullSchema()).thenReturn(fullSchema);
+
+ node.initSchemaParamsForTest();
+ UPDATE_REQUIRED_SLOTS_METHOD.invoke(node);
+
+ TFileScanRangeParams params = node.getFileScanRangeParams();
+ Assertions.assertEquals(2, params.getRequiredSlotsSize());
+ Assertions.assertEquals(Arrays.asList(0), params.getColumnIdxs());
+ Assertions.assertEquals(TColumnCategory.SYNTHESIZED,
params.getColumnNameToCategory().get("_file"));
+ Assertions.assertEquals(TColumnCategory.SYNTHESIZED,
params.getColumnNameToCategory().get("_pos"));
+ Assertions.assertEquals(TColumnCategory.GENERATED,
params.getColumnNameToCategory().get("generated_col"));
+
Assertions.assertFalse(params.getColumnNameToCategory().containsKey("id"));
+ }
+
@Test
public void testUpdateRequiredSlotsPreservesInlineDefaultValueExpr()
throws Exception {
SessionVariable sv = new SessionVariable();
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 1c30a652fe9..8e443ab0f03 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -597,6 +597,9 @@ struct TFileScanRangeParams {
// values unchanged. When present, only INT96 TIMESTAMP values are
converted with this zone.
36: optional string hive_parquet_time_zone
37: optional TLanceScanParams lance_scan_params
+ // Non-regular columns in the pinned full schema, including columns pruned
from phase one.
+ // When present, omitted names are REGULAR. Used to rebuild row-id fetch
projections.
+ 38: optional map<string, TColumnCategory> column_name_to_category
}
struct TFileRangeDesc {
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy
index d4dde31313e..6a19f385668 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_columns.groovy
@@ -163,5 +163,21 @@ suite("test_iceberg_file_metadata_columns",
"p0,external,iceberg,external_docker
"""
verifyFileMetadata(tableName, format)
+
+ // Compare against an eager scan: file paths depend on the writer, and
metadata slots
+ // pruned from phase one must still be synthesized during the row-id
fetch.
+ for (String projection : ["`_file`", "`_pos`", "`_file`, `_pos`",
+ "id, payload, `_file`, `_pos`"]) {
+ String query = "select ${projection} from ${tableName} order by id
limit 3"
+ sql "set topn_lazy_materialization_threshold = -1"
+ def eagerRows = sql query
+ sql "set topn_lazy_materialization_threshold = 10"
+ explain {
+ sql query
+ contains "VMaterializeNode"
+ }
+ assertEquals(eagerRows, sql(query))
+ }
+
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
