This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 397ab8fc830 branch-4.1: [Enhancement](file scanner) Support row-id
fetch in FileScannerV2 #67906 (#67922)
397ab8fc830 is described below
commit 397ab8fc830eada75328c08ac10d0da19dc49f04
Author: Gabriel <[email protected]>
AuthorDate: Tue Sep 15 09:24:06 2026 +0800
branch-4.1: [Enhancement](file scanner) Support row-id fetch in
FileScannerV2 #67906 (#67922)
### What problem does this PR solve?
Backport #67906 to branch-4.1.
TopN two-phase materialization can fetch selected Parquet and ORC file
rows through FileScannerV2. The second phase follows the scanner rollout
policy, avoids whole-chunk Parquet prefetch for sparse row selections,
and preserves partition/generated/synthesized column categories and
Iceberg file metadata. Exact-row fetches reject short results before
reordering, fill output batches across sparse ranges, and use
demand-page reads for Parquet projections.
Compatibility adjustments for branch-4.1:
- Preserve the existing Lance dataset-level uint64 row-ID fetch path,
physical split scheduling, condition-cache state, and Variant
projections.
- Apply name-based column classification to the existing IcebergScanNode
and keep relation-snapshot schema categories separate from physical
column positions.
- Use the existing Iceberg v3 row-lineage regression suite for
eager/lazy comparisons; the master-only `_file`/`_pos` feature and
connector framework are not prerequisites for this backport.
### Release note
Support row-id fetch for Parquet and ORC in FileScannerV2.
### Check List (For Author)
- FE: `FileQueryScanNodeTest` passed (14 tests, zero failures/errors);
Maven validate passed with zero Checkstyle violations.
- BE: 274 related ASAN unit tests passed across RowIdStorageReader,
FileScanner, FileScannerV2, Parquet, and ORC. clang-format 16 passed for
all 20 affected C++ files.
- Regression evidence: all five focused cases failed before the fixes
and passed afterward. A 1,024-row sparse selection now needs 8 output
batches at a 128-row cap; a one-row fetch over 20 flat columns retains
about 1.53 MiB of stream buffers instead of 160 MiB with default
settings.
- Validation limitation: the external Iceberg regression suite was not
run locally.
- Behavior changed: Yes. Supported TopN second-phase fetches use
FileScannerV2 when enabled.
- Does this need documentation: No.
---
be/src/exec/operator/file_scan_operator.cpp | 12 +-
be/src/exec/operator/file_scan_operator.h | 4 +-
be/src/exec/rowid_fetcher.cpp | 138 +++++---
be/src/exec/rowid_fetcher.h | 11 +
be/src/exec/scan/file_scanner_v2.cpp | 133 ++++++-
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 | 78 ++++-
be/src/format_v2/parquet/parquet_scan.h | 1 +
.../parquet/reader/native_column_reader.cpp | 18 +-
.../parquet/reader/native_column_reader.h | 5 +-
be/src/format_v2/table_reader.cpp | 9 +-
be/src/format_v2/table_reader.h | 7 +-
be/test/exec/rowid_fetcher_test.cpp | 243 +++++++++++++
be/test/format_v2/orc/orc_reader_test.cpp | 32 ++
be/test/format_v2/parquet/parquet_scan_test.cpp | 389 +++++++++++++++++++++
.../apache/doris/datasource/FileQueryScanNode.java | 29 +-
.../datasource/iceberg/source/IcebergScanNode.java | 10 +-
.../doris/datasource/FileQueryScanNodeTest.java | 40 +++
gensrc/thrift/PlanNodes.thrift | 3 +
...test_iceberg_v3_row_lineage_query_insert.groovy | 23 ++
25 files changed, 1162 insertions(+), 110 deletions(-)
diff --git a/be/src/exec/operator/file_scan_operator.cpp
b/be/src/exec/operator/file_scan_operator.cpp
index c33fe2f8bd8..f8a72456e84 100644
--- a/be/src/exec/operator/file_scan_operator.cpp
+++ b/be/src/exec/operator/file_scan_operator.cpp
@@ -137,13 +137,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) {
const bool is_transactional_hive =
scan_params.__isset.table_format_params &&
scan_params.table_format_params.table_format_type ==
"transactional_hive";
@@ -156,7 +156,7 @@ bool
FileScanLocalState::_can_generate_physical_splits(const TQueryOptions& quer
bool is_load,
const
TFileScanRangeParams& scan_params,
const TFileRangeDesc&
range) {
- if (!_should_use_file_scanner_v2(query_options, is_load, scan_params)) {
+ if (!should_use_file_scanner_v2(query_options, is_load, scan_params)) {
return false;
}
const auto format = range.__isset.format_type ? range.format_type :
scan_params.format_type;
@@ -210,7 +210,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 862ff4d438d..80c16e1177d 100644
--- a/be/src/exec/operator/file_scan_operator.h
+++ b/be/src/exec/operator/file_scan_operator.h
@@ -56,6 +56,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);
@@ -80,8 +82,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);
static bool _can_generate_physical_splits(const TQueryOptions&
query_options, bool is_load,
const TFileScanRangeParams&
scan_params,
const TFileRangeDesc& range);
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index 61c5a529d73..8c3b0023831 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -56,7 +56,9 @@
#include "core/data_type_serde/data_type_serde.h"
#include "core/string_ref.h"
#include "core/types.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 "format_v2/table/lance_reader.h"
@@ -776,6 +778,69 @@ 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;
+}
+
+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_lance_rows_by_row_ids(
const TFileRangeDesc& scan_range_desc, const std::vector<uint64_t>&
row_ids,
const std::vector<SlotDescriptor>& slots, RuntimeState* runtime_state,
@@ -851,13 +916,7 @@ Status
RowIdStorageReader::read_external_row_from_file_mapping(
scan_blocks[idx] = Block(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");
@@ -868,9 +927,7 @@ Status
RowIdStorageReader::read_external_row_from_file_mapping(
scan_range_desc, read_ids, slots, runtime_state.get(),
sub_runtime_profile.get(),
rpc_scan_params, &scan_blocks[idx], &fetch_statistics[idx]));
} else {
- // Parquet/ORC row IDs are consumed as row ordinals within the exact
physical file range
- // recorded by phase one. Keep using FileScanner so the format reader
can resolve those
- // ordinals against that range; unlike Lance, ranges cannot be merged
at dataset level.
+ // Parquet/ORC IDs remain signed physical file positions; Lance keeps
native uint64 IDs.
std::list<int64_t> legacy_read_ids;
for (const auto row_id : read_ids) {
if (row_id >
static_cast<uint64_t>(std::numeric_limits<int64_t>::max())) {
@@ -879,18 +936,29 @@ Status
RowIdStorageReader::read_external_row_from_file_mapping(
}
legacy_read_ids.emplace_back(static_cast<int64_t>(row_id));
}
- 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, legacy_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, legacy_read_ids, &scan_blocks[idx],
+ &fetch_statistics[idx].init_reader_ms,
&fetch_statistics[idx].get_block_ms));
+ } else {
+ // Keep phase two on V1 whenever the rollout policy or table
format requires it.
+ std::unique_ptr<FileScanner> 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, legacy_read_ids, &scan_blocks[idx],
external_info,
+ &fetch_statistics[idx].init_reader_ms,
&fetch_statistics[idx].get_block_ms));
+ }
}
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(
@@ -898,7 +966,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());
@@ -947,30 +1015,16 @@ Status RowIdStorageReader::read_batch_external_row(
}
}
- 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());
- for (auto slot_idx = 0; slot_idx < slots.size(); ++slot_idx) {
+ std::vector<uint32_t> scan_column_idxs;
+ scan_column_idxs.reserve(slots.size());
+ for (int slot_idx = 0; slot_idx < slots.size(); ++slot_idx) {
auto& slot = slots[slot_idx];
tuple_desc.add_slot(&slot);
colname_to_slot_id.emplace(slot.col_name(), slot.id());
- TFileScanSlotInfo slot_info;
- slot_info.slot_id = slot.id();
- auto column_idx = request_block_desc.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);
+
scan_column_idxs.emplace_back(request_block_desc.column_idxs(slot_idx));
}
+ rpc_scan_params = build_external_scan_params(rpc_scan_params,
first_scan_range_desc, slots,
+ scan_column_idxs);
const auto& query_options = id_file_map->get_query_options();
const auto& query_globals = id_file_map->get_query_globals();
@@ -1174,9 +1228,9 @@ Status RowIdStorageReader::read_batch_external_row(
std::to_string(*get_block_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));
for (const auto& [time_name, time_value] : lance_fetch_times_ns) {
runtime_profile->add_info_string(time_name,
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index ceb7a51c1a0..ab8019e8f32 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -40,6 +40,7 @@ namespace doris {
class DorisNodesInfo;
class RuntimeProfile;
class RuntimeState;
+class TQueryOptions;
class TupleDescriptor;
namespace io {
enum class FileCacheMissPolicy : uint8_t;
@@ -122,7 +123,17 @@ public:
static Status read_by_rowids(const PMultiGetRequest& request,
PMultiGetResponse* response);
static Status read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response);
+ static bool should_use_file_scanner_v2(const TQueryOptions& query_options,
+ const TFileScanRangeParams&
scan_params,
+ const TFileRangeDesc& range);
+
private:
+ friend class RowIdStorageReaderTest;
+ 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);
struct ExternalFetchStatistics;
static Status read_doris_format_row(
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index ac775fd94bf..c0cdfaf7b78 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -75,8 +75,12 @@
#include "runtime/runtime_state.h"
#include "service/backend_options.h"
#include "storage/id_manager.h"
+#include "util/stopwatch.hpp"
namespace doris {
+const std::string FileScannerV2::FileReadBytesProfile = "FileReadBytes";
+const std::string FileScannerV2::FileReadTimeProfile = "FileReadTime";
+
namespace {
constexpr int kIcebergPositionDeleteContent = 1;
@@ -432,12 +436,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(
@@ -715,18 +719,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({
@@ -736,15 +743,106 @@ 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,
+ .file_context = nullptr,
+ .condition_cache_source_range = std::nullopt,
+ .condition_cache_split_context = nullptr,
+ .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);
+ // Reordering uses a dense position for every requested ID. A replaced or
truncated file
+ // must fail here instead of exposing a short source column to unchecked
indexed inserts.
+ if (result_block->rows() != row_ids.size()) {
+ return Status::Corruption("FileScannerV2 row-ID fetch returned {}
rows, expected {}",
+ result_block->rows(), row_ids.size());
+ }
+ return Status::OK();
+}
+
Status FileScannerV2::_create_table_reader_for_format(
const TFileRangeDesc& range, std::unique_ptr<format::TableReader>*
reader) const {
DORIS_CHECK(reader != nullptr);
@@ -825,6 +923,7 @@ Status FileScannerV2::_prepare_table_reader_split(const
TFileRangeDesc& range,
.format_split_id_end = _current_split.format_split_id_end,
.global_rowid_context =
_create_global_rowid_context(_current_split.source_identity_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 5f6869a8312..dd10d2952b1 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;
@@ -139,6 +142,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 34fd4e5b273..894cd49fa6b 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() << ",
residual_predicate_columns="
+ 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() << ",
residual_predicate_columns="
<< join_debug_strings(
residual_predicate_columns,
[](LocalColumnId column_id) { return
std::to_string(column_id.value()); })
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index 586f9dd4ff5..8d0cc487676 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>
@@ -117,6 +118,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.
@@ -409,8 +416,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 868f451963f..85754b9997a 100644
--- a/be/src/format_v2/orc/orc_reader.cpp
+++ b/be/src/format_v2/orc/orc_reader.cpp
@@ -755,6 +755,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;
};
@@ -1285,6 +1286,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;
@@ -1342,6 +1344,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();
}
@@ -1757,7 +1769,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;
@@ -2054,11 +2071,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();
}
@@ -2066,6 +2095,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());
@@ -2086,6 +2118,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) {
@@ -2096,7 +2132,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 0c42b32f617..5994ad6993c 100644
--- a/be/src/format_v2/orc/orc_reader.h
+++ b/be/src/format_v2/orc/orc_reader.h
@@ -77,6 +77,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 b912af0a165..1b162b48bd6 100644
--- a/be/src/format_v2/parquet/parquet_reader.h
+++ b/be/src/format_v2/parquet/parquet_reader.h
@@ -62,6 +62,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 2652e3e5f75..254e43bbc9b 100644
--- a/be/src/format_v2/parquet/parquet_scan.cpp
+++ b/be/src/format_v2/parquet/parquet_scan.cpp
@@ -842,7 +842,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;
prepare_row_group_physical_projection(row_group, file_schema, request,
&row_group_plan);
plan->row_groups.push_back(std::move(row_group_plan));
@@ -1580,14 +1601,14 @@ Status ParquetScanScheduler::open_next_row_group(
RETURN_IF_ERROR(detail::build_native_prefetch_ranges(
thrift_metadata, file_schema,
request_scan_columns(row_group_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);
}
@@ -1621,7 +1642,7 @@ Status ParquetScanScheduler::open_next_row_group(
file_context.native_io_ctx, _runtime_state,
file_context.native_page_cache_enabled,
file_context.native_page_cache_file_key,
_current_dictionary_filters.contains(local_id),
_scan_profile.column_reader_profile,
- &column_reader));
+ &column_reader, !request.row_ids.has_value()));
_current_predicate_columns[local_id] = std::move(column_reader);
}
// Start warming filter-column chunks as soon as their row group is
selected. The native
@@ -1630,8 +1651,9 @@ Status ParquetScanScheduler::open_next_row_group(
if (!_current_merge_range_active) {
const auto prefetch_columns =
adaptive_predicate_prefetch_columns(request,
row_group_request.predicate_columns);
- 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 : row_group_request.non_predicate_columns) {
const auto local_id = col.column_id();
@@ -1660,7 +1682,7 @@ Status ParquetScanScheduler::open_next_row_group(
row_group_idx, _current_selected_ranges,
_current_offset_indexes, _timezone,
file_context.native_io_ctx, _runtime_state,
file_context.native_page_cache_enabled,
file_context.native_page_cache_file_key, false,
_scan_profile.column_reader_profile,
- &column_reader));
+ &column_reader, !request.row_ids.has_value()));
_current_non_predicate_columns[local_id] = std::move(column_reader);
}
if (!_current_merge_range_active &&
@@ -1670,7 +1692,8 @@ Status ParquetScanScheduler::open_next_row_group(
// 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,
physical_non_predicate_columns(row_group_request),
+ file_context, file_schema, request,
+ physical_non_predicate_columns(row_group_request),
&_current_non_predicate_prefetched));
}
if (_parquet_profile != nullptr) {
@@ -2206,7 +2229,7 @@ Status
ParquetScanScheduler::prepare_current_dictionary_filters(
row_group_idx, _current_selected_ranges,
_current_offset_indexes, _timezone,
file_context.native_io_ctx, _runtime_state,
file_context.native_page_cache_enabled,
file_context.native_page_cache_file_key, true,
_scan_profile.column_reader_profile,
- &column_reader));
+ &column_reader, !request.row_ids.has_value()));
MutableColumnPtr dictionary_values;
{
SCOPED_TIMER(_scan_profile.dict_filter_read_dict_time);
@@ -2898,10 +2921,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;
@@ -3022,9 +3049,10 @@ Status
ParquetScanScheduler::read_current_row_group_batch(
// materializing non-predicate columns, so fully filtered batches
avoid unnecessary IO.
const auto& physical_request =
_current_row_group_request != nullptr ?
*_current_row_group_request : request;
- RETURN_IF_ERROR(prefetch_current_row_group_columns(
- file_context, file_schema,
physical_non_predicate_columns(physical_request),
- &_current_non_predicate_prefetched));
+ RETURN_IF_ERROR(
+ prefetch_current_row_group_columns(file_context, file_schema,
request,
+
physical_non_predicate_columns(physical_request),
+
&_current_non_predicate_prefetched));
}
if (selected_rows > _batch_size) {
@@ -3185,6 +3213,13 @@ Status ParquetScanScheduler::read_next_batch(
*eof = false;
return Status::OK();
}
+ // Phase-two IDs have already survived filtering. Append sparse ranges
directly to one
+ // output block so TableReader finalization and result merging happen once
per caller batch.
+ // Filtered requests and pending projection changes retain their
single-batch coordinates.
+ const bool append_row_id_ranges =
+ _active_request->row_ids.has_value() &&
_active_request->predicate_columns.empty() &&
+ _active_request->conjuncts.empty() &&
_active_request->delete_conjuncts.empty() &&
+ _active_request->count_star_placeholder_columns.empty() &&
_pending_request == nullptr;
int64_t predicate_batch_rows = std::max(_batch_size,
_empty_predicate_batch_rows);
const int64_t max_predicate_batch_rows = std::min<int64_t>(
std::numeric_limits<uint16_t>::max(),
@@ -3237,13 +3272,17 @@ Status ParquetScanScheduler::read_next_batch(
continue;
}
- const int64_t batch_rows = std::min<int64_t>(predicate_batch_rows,
remaining_rows);
+ const int64_t row_cap = append_row_id_ranges ? _batch_size -
static_cast<int64_t>(*rows)
+ : predicate_batch_rows;
+ const int64_t batch_rows = std::min<int64_t>(row_cap, remaining_rows);
const int64_t physical_rows_read = batch_rows;
const int64_t batch_first_file_row =
_current_row_group_first_row + _current_row_group_rows_read;
+ size_t batch_output_rows = 0;
RETURN_IF_ERROR(read_current_row_group_batch(file_context,
file_schema, batch_rows,
*_active_request,
batch_first_file_row,
- file_block, rows));
+ file_block,
&batch_output_rows));
+ *rows += batch_output_rows;
_current_row_group_rows_read += physical_rows_read;
_current_range_rows_read += physical_rows_read;
if (_current_range_rows_read >= current_range.length) {
@@ -3259,6 +3298,9 @@ Status ParquetScanScheduler::read_next_batch(
_publish_adaptive_state(*_active_request);
continue;
}
+ if (append_row_id_ranges && *rows < static_cast<size_t>(_batch_size)) {
+ continue;
+ }
*eof = false;
return Status::OK();
}
diff --git a/be/src/format_v2/parquet/parquet_scan.h
b/be/src/format_v2/parquet/parquet_scan.h
index dff822fe21c..7046d342304 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -313,6 +313,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/parquet/reader/native_column_reader.cpp
b/be/src/format_v2/parquet/reader/native_column_reader.cpp
index 273964e5d1f..e0fab04951b 100644
--- a/be/src/format_v2/parquet/reader/native_column_reader.cpp
+++ b/be/src/format_v2/parquet/reader/native_column_reader.cpp
@@ -266,7 +266,7 @@ Status NativeColumnReader::create(
const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState*
runtime_state,
bool enable_page_cache, const std::string& page_cache_file_key,
bool enable_dictionary_filter, ParquetColumnReaderProfile profile,
- std::unique_ptr<ParquetColumnReader>* reader) {
+ std::unique_ptr<ParquetColumnReader>* reader, bool enable_read_ahead) {
if (reader == nullptr) {
return Status::InvalidArgument("Native parquet reader result is null");
}
@@ -307,10 +307,11 @@ Status NativeColumnReader::create(
auto native_reader = std::unique_ptr<NativeColumnReader>(
new NativeColumnReader(column_schema, std::move(logical_type),
std::move(native_type),
std::move(variant_plan), profile));
- RETURN_IF_ERROR(native_reader->init(
- std::move(file), metadata, row_group_id, field,
std::move(schema_node),
- std::move(projected_ids), selected_ranges, offset_indexes,
timezone, io_ctx,
- runtime_state, enable_page_cache, page_cache_file_key,
enable_dictionary_filter));
+ RETURN_IF_ERROR(native_reader->init(std::move(file), metadata,
row_group_id, field,
+ std::move(schema_node),
std::move(projected_ids),
+ selected_ranges, offset_indexes,
timezone, io_ctx,
+ runtime_state, enable_page_cache,
page_cache_file_key,
+ enable_dictionary_filter,
enable_read_ahead));
*reader = std::move(native_reader);
return Status::OK();
}
@@ -322,7 +323,7 @@ Status NativeColumnReader::init(
const std::unordered_map<int, tparquet::OffsetIndex>& offset_indexes,
const cctz::time_zone* timezone, io::IOContext* io_ctx, RuntimeState*
runtime_state,
bool enable_page_cache, const std::string& page_cache_file_key,
- bool enable_dictionary_filter) {
+ bool enable_dictionary_filter, bool enable_read_ahead) {
DORIS_CHECK(file != nullptr);
DORIS_CHECK(metadata != nullptr);
DORIS_CHECK(field != nullptr);
@@ -347,7 +348,10 @@ Status NativeColumnReader::init(
const size_t max_group_buffer = config::parquet_rowgroup_max_buffer_mb <<
20;
const size_t max_column_buffer = config::parquet_column_max_buffer_mb <<
20;
- const size_t max_buffer_size = std::min(max_group_buffer,
max_column_buffer);
+ // Sparse exact-row fetches need demand pages, not one read-ahead buffer
per physical leaf.
+ // Passing zero through the native tree also prevents wide nested
projections multiplying it.
+ const size_t max_buffer_size =
+ enable_read_ahead ? std::min(max_group_buffer, max_column_buffer)
: 0;
RuntimeState* native_runtime_state = runtime_state;
const bool runtime_page_cache_enabled =
runtime_state == nullptr ||
diff --git a/be/src/format_v2/parquet/reader/native_column_reader.h
b/be/src/format_v2/parquet/reader/native_column_reader.h
index 150ab4352d0..69588484696 100644
--- a/be/src/format_v2/parquet/reader/native_column_reader.h
+++ b/be/src/format_v2/parquet/reader/native_column_reader.h
@@ -76,7 +76,8 @@ public:
RuntimeState* runtime_state, bool enable_page_cache,
const std::string& page_cache_file_key, bool
enable_dictionary_filter,
ParquetColumnReaderProfile profile,
- std::unique_ptr<ParquetColumnReader>* reader);
+ std::unique_ptr<ParquetColumnReader>* reader,
+ bool enable_read_ahead = true);
~NativeColumnReader() override;
@@ -115,7 +116,7 @@ private:
const std::unordered_map<int, tparquet::OffsetIndex>&
offset_indexes,
const cctz::time_zone* timezone, io::IOContext* io_ctx,
RuntimeState* runtime_state,
bool enable_page_cache, const std::string& page_cache_file_key,
- bool enable_dictionary_filter);
+ bool enable_dictionary_filter, bool enable_read_ahead);
Status read_with_filter(int64_t rows, const uint8_t* filter_data, bool
filter_all,
MutableColumnPtr& column, const DataTypePtr&
output_type,
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 8e03d3d2282..2c7ffe933c7 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -1343,6 +1343,7 @@ Status TableReader::refresh_conjuncts(VExprContextSPtrs
conjuncts,
_table_filters, _projected_columns, refreshed_request.get(),
_runtime_state,
_file_scan_request == nullptr ? nullptr :
&_file_scan_request->local_positions));
refreshed_request->predicate_snapshot_digest = _predicate_snapshot_digest;
+ 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() &&
@@ -1751,6 +1752,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;
@@ -1779,9 +1781,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 797bba2a0a1..273ef982021 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -194,6 +194,9 @@ struct SplitReadOptions {
int64_t format_split_id = -1;
int64_t format_split_id_end = -1;
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.
@@ -478,6 +481,7 @@ protected:
RETURN_IF_ERROR(_data_reader.column_mapper->create_scan_request(
_table_filters, _projected_columns, file_request.get(),
_runtime_state));
file_request->predicate_snapshot_digest = _predicate_snapshot_digest;
+ 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);
@@ -1072,7 +1076,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();
}
@@ -2060,6 +2064,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;
std::optional<FileAggregateResult> _metadata_aggregate_result;
bool _current_split_pruned = false;
diff --git a/be/test/exec/rowid_fetcher_test.cpp
b/be/test/exec/rowid_fetcher_test.cpp
new file mode 100644
index 00000000000..a1ba94cf702
--- /dev/null
+++ b/be/test/exec/rowid_fetcher_test.cpp
@@ -0,0 +1,243 @@
+// 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 <gtest/gtest.h>
+
+#include <string>
+#include <vector>
+
+#include "common/consts.h"
+#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 {
+
+class RowIdStorageReaderTest : public testing::Test {
+protected:
+ struct SlotSpec {
+ std::string col_name = "c";
+ int32_t slot_id = 0;
+ PrimitiveType type = TYPE_INT;
+ int32_t col_unique_id = 1;
+ };
+
+ static SlotDescriptor make_slot(const SlotSpec& spec) {
+ TSlotDescriptor tdesc = TSlotDescriptorBuilder()
+ .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);
+ return SlotDescriptor(tdesc);
+ }
+};
+
+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_LANCE);
+ 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;
+ // Branch-4.1 recognizes these built-in virtual names; arbitrary
synthesized names are
+ // not virtual columns and fall through to the legacy partition-slot
classification.
+ source_params.__set_column_name_to_category(
+ {{BeConsts::GLOBAL_ROWID_COL, TColumnCategory::SYNTHESIZED},
+ {BeConsts::ICEBERG_ROWID_COL, 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 :
+ {BeConsts::GLOBAL_ROWID_COL, BeConsts::ICEBERG_ROWID_COL,
std::string("generated_col"),
+ std::string("partition_col"), std::string("value")}) {
+ slots.emplace_back(make_slot(
+ {.col_name = name,
+ .slot_id = static_cast<int32_t>(slots.size()),
+ .type = name == BeConsts::GLOBAL_ROWID_COL ? 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};
+ 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);
+ }
+ // 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());
+ }
+ }
+}
+
+} // namespace doris
diff --git a/be/test/format_v2/orc/orc_reader_test.cpp
b/be/test/format_v2/orc/orc_reader_test.cpp
index 1884afaf73c..10f3a524aeb 100644
--- a/be/test/format_v2/orc/orc_reader_test.cpp
+++ b/be/test/format_v2/orc/orc_reader_test.cpp
@@ -10079,6 +10079,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 1292b90cb49..c045683e424 100644
--- a/be/test/format_v2/parquet/parquet_scan_test.cpp
+++ b/be/test/format_v2/parquet/parquet_scan_test.cpp
@@ -25,7 +25,10 @@
#include <parquet/arrow/writer.h>
#include <parquet/encoding.h>
+#include <atomic>
#include <bit>
+#include <chrono>
+#include <cstdlib>
#include <cstring>
#include <filesystem>
#include <fstream>
@@ -35,6 +38,7 @@
#include <optional>
#include <set>
#include <string>
+#include <thread>
#include <utility>
#include <vector>
@@ -50,6 +54,7 @@
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
#include "core/field.h"
+#include "exec/scan/file_scanner_v2.h"
#include "exprs/bloom_filter_func.h"
#include "exprs/create_predicate_function.h"
#include "exprs/runtime_filter_expr.h"
@@ -69,16 +74,25 @@
#include "format_v2/parquet/parquet_column_schema.h"
#include "format_v2/parquet/parquet_reader.h"
#include "format_v2/parquet/reader/native/block_split_bloom_filter.h"
+#include "format_v2/parquet/reader/native/column_reader.h"
#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/descriptor_helper.h"
#include "runtime/runtime_state.h"
#include "storage/index/zone_map/zonemap_eval_context.h"
#include "storage/index/zone_map/zonemap_filter_result.h"
#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 {
@@ -2596,6 +2610,381 @@ 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}));
+}
+
+TEST_F(ParquetScanTest, StandaloneRowIdFetchRejectsMissingRows) {
+ write_int_pair_parquet_file(_file_path, 2);
+ TDescriptorTableBuilder descriptors_builder;
+ TTupleDescriptorBuilder tuple_builder;
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(TYPE_INT)
+ .nullable(false)
+ .column_name("id")
+ .column_pos(0)
+ .build());
+ tuple_builder.build(&descriptors_builder);
+ ObjectPool pool;
+ DescriptorTbl* descriptors = nullptr;
+ ASSERT_TRUE(DescriptorTbl::create(&pool, descriptors_builder.desc_tbl(),
&descriptors).ok());
+ auto* tuple = descriptors->get_tuple_descriptor(0);
+ TFileScanRangeParams params;
+ params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ params.__set_file_type(TFileType::FILE_LOCAL);
+ params.__set_src_tuple_id(0);
+ params.__set_dest_tuple_id(0);
+ TFileScanSlotInfo slot;
+ slot.__set_slot_id(0);
+ slot.__set_is_file_slot(true);
+ slot.__set_category(TColumnCategory::REGULAR);
+ params.__set_required_slots({slot});
+ params.__set_column_idxs({0});
+ params.__set_slot_name_to_schema_pos({{"id", 0}});
+ TFileRangeDesc range;
+ range.__set_path(_file_path);
+ range.__set_start_offset(0);
+ range.__set_size(std::filesystem::file_size(_file_path));
+ range.__set_file_size(range.size);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ state.set_desc_tbl(descriptors);
+ // EOF is valid for a scan, but exact fetches must not expose missing rows
to reordering.
+ for (const auto& requested : {std::list<int64_t> {}, {0, 3, 5}, {0, 3, 6},
{6}}) {
+ RuntimeProfile profile("row_id_fetch");
+ FileScannerV2 scanner(&state, &profile, ¶ms, nullptr, tuple);
+ Block block;
+ block.insert({ColumnInt32::create(),
std::make_shared<DataTypeInt32>(), "id"});
+ int64_t init_ms = 0;
+ int64_t read_ms = 0;
+ auto status = scanner.read_by_rows(range, requested, &block, &init_ms,
&read_ms);
+ if (requested.empty() || requested.back() < 6) {
+ ASSERT_TRUE(status.ok()) << status.to_string();
+ ASSERT_EQ(block.rows(), requested.size());
+ const auto& values =
int32_data_column(*block.get_by_position(0).column);
+ size_t row = 0;
+ for (auto id : requested) {
+ EXPECT_EQ(values.get_element(row++), id + 1);
+ }
+ } else {
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("row-ID fetch"),
std::string::npos);
+ }
+ }
+}
+
+TEST_F(ParquetScanTest, SparseRowIdsFillBatchesAcrossRangesAndRowGroups) {
+ std::vector<int32_t> ids(4096);
+ 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, 500);
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ reader->set_batch_size(128);
+ 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),
+
field_projection(format::ROW_POSITION_COLUMN_ID)};
+ request->local_positions.emplace(format::LocalColumnId(0),
format::LocalIndex(0));
+
request->local_positions.emplace(format::LocalColumnId(format::ROW_POSITION_COLUMN_ID),
+ format::LocalIndex(1));
+ request->row_ids.emplace();
+ for (int64_t id = 0; id < 4096; id += 4) {
+ request->row_ids->push_back(id);
+ }
+ ASSERT_TRUE(reader->open(request).ok());
+ size_t fetched = 0;
+ size_t batches = 0;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block({schema[0],
format::row_position_column_definition()});
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ if (rows == 0) {
+ continue;
+ }
+ EXPECT_LE(rows, 128);
+ const auto& column =
int32_data_column(*block.get_by_position(0).column);
+ const auto& positions =
int64_data_column(*block.get_by_position(1).column);
+ for (size_t row = 0; row < rows; ++row) {
+ ASSERT_LT(fetched, request->row_ids->size());
+ EXPECT_EQ(positions.get_element(row),
(*request->row_ids)[fetched]);
+ EXPECT_EQ(column.get_element(row), (*request->row_ids)[fetched++]);
+ }
+ ++batches;
+ }
+ RecordProperty("nonempty_batches", std::to_string(batches));
+ EXPECT_EQ(fetched, 1024);
+ EXPECT_EQ(batches, 8);
+}
+
+TEST_F(ParquetScanTest, SparseRowIdsAppendNestedColumnsAcrossRanges) {
+ write_int_list_parquet_file(_file_path);
+ auto reader = create_reader();
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ reader->set_batch_size(3);
+ 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),
field_projection(1)};
+ request->row_ids = {0, 2};
+ ASSERT_TRUE(reader->open(request).ok());
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ ASSERT_EQ(rows, 2);
+
EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_element(0),
1);
+
EXPECT_EQ(int32_data_column(*block.get_by_position(0).column).get_element(1),
3);
+ const IColumn* list_column = block.get_by_position(1).column.get();
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(*list_column)) {
+ list_column = &nullable->get_nested_column();
+ }
+ const auto& lists = assert_cast<const ColumnArray&>(*list_column);
+ ASSERT_EQ(lists.get_offsets().size(), 2);
+ EXPECT_EQ(lists.get_offsets()[0], 2);
+ EXPECT_EQ(lists.get_offsets()[1], 2);
+ const auto& elements = int32_data_column(lists.get_data());
+ ASSERT_EQ(elements.size(), 2);
+ EXPECT_EQ(elements.get_element(0), 1);
+ EXPECT_EQ(elements.get_element(1), 2);
+}
+
+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, SparseRowIdsBoundWideProjectionReadAhead) {
+ using namespace format::parquet;
+ // Twenty leaves exceed the default aggregate budget if each retains 8 MiB
of read-ahead.
+ constexpr int leaf_count = 20;
+ std::vector<int32_t> ids(3 * 1024 * 1024);
+ std::iota(ids.begin(), ids.end(), 0);
+ auto values = build_int32_array(ids);
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ std::vector<std::shared_ptr<arrow::Array>> arrays;
+ for (int leaf = 0; leaf < leaf_count; ++leaf) {
+ fields.push_back(arrow::field("c" + std::to_string(leaf),
arrow::int32(), false));
+ arrays.push_back(values);
+ }
+ auto table = arrow::Table::Make(arrow::schema(fields), arrays);
+ auto output = arrow::io::FileOutputStream::Open(_file_path).ValueOrDie();
+ ::parquet::WriterProperties::Builder writer_properties;
+ writer_properties.disable_dictionary();
+ writer_properties.compression(::parquet::Compression::UNCOMPRESSED);
+ writer_properties.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ // Arrow otherwise caps each row group at 1M rows, making every chunk
smaller than 8 MiB.
+ writer_properties.max_row_group_length(ids.size());
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), output,
+ ids.size(),
writer_properties.build()));
+ ASSERT_TRUE(output->Close().ok());
+ io::FileReaderSPtr local;
+ ASSERT_TRUE(io::global_local_filesystem()->open_file(_file_path,
&local).ok());
+ auto remote = std::make_shared<CountingParquetRemoteReader>(local);
+ io::IOContext io_ctx;
+ io::FileDescription description;
+ description.path = _file_path;
+ description.file_size = local->size();
+ ParquetFileContext context;
+ ASSERT_TRUE(context.open(remote, &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>();
+ Block block;
+ for (int leaf = 0; leaf < leaf_count; ++leaf) {
+ request->non_predicate_columns.push_back(field_projection(leaf));
+ request->local_positions.emplace(format::LocalColumnId(leaf),
format::LocalIndex(leaf));
+ block.insert({schema[leaf]->type->create_column(), schema[leaf]->type,
schema[leaf]->name});
+ }
+ request->row_ids = {0};
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto plan = std::make_shared<RowGroupScanPlan>();
+ ASSERT_TRUE(plan_parquet_row_groups(*context.native_metadata, schema,
*request, {}, false,
+ plan.get(), &state.timezone_obj(),
&state, &context)
+ .ok());
+ ParquetScanScheduler scheduler;
+ scheduler.set_plan(plan);
+ scheduler.set_batch_size(1);
+ scheduler.set_scan_request(request);
+ scheduler.set_runtime_state(&state);
+ scheduler.set_timezone(&state.timezone_obj());
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(scheduler.read_next_batch(context, schema, &block, &rows,
&eof).ok());
+ ASSERT_EQ(rows, 1);
+ for (int leaf = 0; leaf < leaf_count; ++leaf) {
+
EXPECT_EQ(int32_data_column(*block.get_by_position(leaf).column).get_element(0),
0);
+ }
+ size_t retained_buffer_bytes = 0;
+ ASSERT_EQ(scheduler._current_non_predicate_columns.size(), leaf_count);
+ for (const auto& [id, column_reader] :
scheduler._current_non_predicate_columns) {
+ const auto* adapter = dynamic_cast<const
NativeColumnReader*>(column_reader.get());
+ ASSERT_NE(adapter, nullptr);
+ const auto* scalar = dynamic_cast<const
native::ScalarColumnReader<false, false>*>(
+ adapter->_native_reader.get());
+ ASSERT_NE(scalar, nullptr);
+ retained_buffer_bytes += scalar->_stream_reader->_buf_size;
+ }
+ RecordProperty("retained_buffer_bytes",
std::to_string(retained_buffer_bytes));
+ RecordProperty("remote_bytes",
std::to_string(remote->remote_bytes.load()));
+ EXPECT_LT(retained_buffer_bytes,
size_t(config::parquet_rowgroup_max_buffer_mb) << 20);
+ EXPECT_LT(remote->remote_bytes.load(),
size_t(config::parquet_rowgroup_max_buffer_mb) << 20);
+ EXPECT_EQ(remote->dryrun_bytes.load(), 0);
+}
+
+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_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::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::make_shared<RowGroupScanPlan>(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/FileQueryScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
index f57d84ca273..e95a9388483 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
@@ -276,12 +276,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;
@@ -339,6 +343,25 @@ public abstract class FileQueryScanNode extends
FileScanNode {
columnNameMap.putIfAbsent(columnNames.get(i), i);
}
+ boolean needsRowIdFetch = desc.getSlots().stream()
+ .anyMatch(slot ->
slot.getColumn().getName().startsWith(Column.GLOBAL_ROWID_COL));
+ if (needsRowIdFetch) {
+ // Lazy slots are absent from the scan tuple. Use the relation's
full schema so
+ // metadata categories survive pruning without changing physical
file positions.
+ List<Column> columns = desc.getTable() instanceof ExternalTable
+ ? ((ExternalTable)
desc.getTable()).getFullSchema(getRelationSnapshot())
+ : desc.getTable().getFullSchema();
+ List<String> partitionKeys = getPathPartitionKeys();
+ Map<String, TColumnCategory> columnCategories = new HashMap<>();
+ for (Column column : columns) {
+ TColumnCategory category = classifyColumn(column.getName(),
partitionKeys);
+ if (category != TColumnCategory.REGULAR) {
+ columnCategories.put(column.getName(), category);
+ }
+ }
+ params.setColumnNameToCategory(columnCategories);
+ }
+
for (TFileScanSlotInfo slot : params.getRequiredSlots()) {
if (!slot.isIsFileSlot()) {
continue;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
index 2a45a1b749f..ebae13b7d4c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
@@ -2570,17 +2570,17 @@ public class IcebergScanNode extends FileQueryScanNode {
}
@Override
- protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String>
partitionKeys) {
- if
(Column.ICEBERG_ROWID_COL.equalsIgnoreCase(slot.getColumn().getName())) {
+ protected TColumnCategory classifyColumn(String columnName, List<String>
partitionKeys) {
+ if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(columnName)) {
return TColumnCategory.SYNTHESIZED;
}
- if (slot.getColumn().getName().startsWith(Column.GLOBAL_ROWID_COL)) {
+ if (columnName.startsWith(Column.GLOBAL_ROWID_COL)) {
return TColumnCategory.SYNTHESIZED;
}
- if (IcebergUtils.isIcebergRowLineageColumn(slot.getColumn())) {
+ if (IcebergUtils.isIcebergRowLineageColumn(columnName)) {
return TColumnCategory.GENERATED;
}
- return super.classifyColumn(slot, partitionKeys);
+ return super.classifyColumn(columnName, partitionKeys);
}
private List<Split> doGetSplits(int numBackends) throws UserException {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java
index a29471aff20..bb3a25c5c97 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/FileQueryScanNodeTest.java
@@ -253,6 +253,46 @@ public class FileQueryScanNodeTest {
Assert.assertEquals(64 * MB, sv.getFileSplitSizeOnBe());
}
+ @Test
+ public void testRowIdFetchRetainsCategoriesOfPrunedColumns() throws
Exception {
+ TestFileQueryScanNode node = new TestFileQueryScanNode(new
SessionVariable()) {
+ @Override
+ protected TColumnCategory classifyColumn(String name, List<String>
partitionKeys) {
+ if (name.equals("metadata_path") ||
name.equals("metadata_position")) {
+ 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);
+ sortSlot.setColumn(new Column("id", Type.INT));
+ desc.addSlot(sortSlot);
+ SlotDescriptor rowIdSlot = new SlotDescriptor(new SlotId(2), desc);
+ rowIdSlot.setColumn(new Column(Column.GLOBAL_ROWID_COL, Type.STRING));
+ desc.addSlot(rowIdSlot);
+ List<Column> fullSchema = Arrays.asList(sortSlot.getColumn(), new
Column("metadata_path", Type.STRING),
+ new Column("metadata_position", Type.BIGINT), new
Column("generated_col", Type.BIGINT));
+ Mockito.when(table.getBaseSchema(false)).thenReturn(fullSchema);
+ Mockito.when(table.getFullSchema()).thenReturn(fullSchema);
+
+ node.params = new TFileScanRangeParams();
+ UPDATE_REQUIRED_SLOTS_METHOD.invoke(node);
+
+ TFileScanRangeParams params = node.getFileScanRangeParams();
+ Assert.assertEquals(2, params.getRequiredSlotsSize());
+ Assert.assertEquals(Arrays.asList(0), params.getColumnIdxs());
+ Assert.assertEquals(TColumnCategory.SYNTHESIZED,
params.getColumnNameToCategory().get("metadata_path"));
+ Assert.assertEquals(TColumnCategory.SYNTHESIZED,
params.getColumnNameToCategory().get("metadata_position"));
+ Assert.assertEquals(TColumnCategory.GENERATED,
params.getColumnNameToCategory().get("generated_col"));
+ Assert.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 35a04f3f99f..45b65a0d63d 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -693,6 +693,9 @@ struct TFileScanRangeParams {
35: optional string serialized_table_cache_key
// 31-33 and 36 are used in master; do not allocate them in branch-4.1.
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_v3_row_lineage_query_insert.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy
index 28e28bdd887..a4579535a9e 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy
@@ -118,6 +118,29 @@ suite("test_iceberg_v3_row_lineage_query_insert",
"p0,external,iceberg,external_
"_last_updated_sequence_number should be non-null for
${tableName}, row=${rowLineageRows[i]}")
}
+ def originalThreshold = sql("select
@@topn_lazy_materialization_threshold")[0][0]
+ def originalScannerV2 = sql("select @@enable_file_scanner_v2")[0][0]
+ try {
+ sql "set enable_file_scanner_v2 = true"
+ // Phase two must retain generated-column categories and
file-level lineage metadata.
+ for (String projection : ["_row_id",
"_last_updated_sequence_number",
+ "id, _row_id,
_last_updated_sequence_number"]) {
+ String query = "select ${projection} from ${tableName} order
by id limit 2"
+ 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))
+ }
+ } finally {
+ // UNSET requires the experimental prefix in 4.1; preserve the
caller's values even on failure.
+ sql "set topn_lazy_materialization_threshold =
${originalThreshold}"
+ sql "set enable_file_scanner_v2 = ${originalScannerV2}"
+ }
+
long firstRowId = rowLineageRows[0][1].toString().toLong()
long secondRowId = rowLineageRows[1][1].toString().toLong()
assertTrue(firstRowId < secondRowId,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
