This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 4f48ccfaa7d [improvement](be) Improve FileScannerV2 profiling and
pruning (#65449)
4f48ccfaa7d is described below
commit 4f48ccfaa7d95c94f31bedde4bdf982ead940488
Author: Gabriel <[email protected]>
AuthorDate: Fri Jul 10 19:57:01 2026 +0800
[improvement](be) Improve FileScannerV2 profiling and pruning (#65449)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
FileScannerV2 collected file-cache statistics but did not publish the
FileCache profile subtree or account for cache-write bytes in the query
resource context. It also lacked V1-compatible split pruning for
late-arriving runtime filters on partition columns and always propagated
missing-file errors, even when `ignore_not_found_file_in_external_table`
was enabled.
This PR:
- publishes the shared FileCache profile counters from FileScannerV2 and
records cache-write bytes;
- passes the latest scanner conjunct snapshot into
`TableReader::prepare_split`, where partition-only runtime filters are
evaluated before a concrete reader is opened;
- reports `FileScannerRuntimeFilterPartitionPruningTime` and
`RuntimeFilterPartitionPrunedRangeNum`;
- skips and counts `NOT_FOUND` splits when the external-table
missing-file option is enabled;
- resets native, JNI, Hudi hybrid, and Paimon hybrid split state before
continuing;
- adds focused unit coverage for profile reporting, partition pruning,
and split cleanup after `NOT_FOUND`.
### Release note
FileScannerV2 profiles now expose FileCache statistics, runtime-filter
partition range pruning statistics, and ignored missing-file counts.
FileScannerV2 can prune partition ranges before opening a reader and can
continue after missing external files when the existing ignore option is
enabled.
### Check List (For Author)
- Test: Manual test / Unit Test attempted
- clang-format 16 dry-run passed for all changed C++ files
- `git diff --check` passed
- targeted BE unit tests were started, but CMake configuration was
blocked because `thirdparty/installed` is missing Protobuf; tests did
not reach compilation
- Behavior changed: Yes. FileScannerV2 publishes additional profile
counters, prunes partition ranges using runtime filters, and skips
configured ignorable `NOT_FOUND` splits.
- Does this need documentation: Yes. The existing FileScanner V1/V2
profile document is updated in place.
---
be/src/exec/scan/file_scanner_v2.cpp | 97 ++++++++++++++++++++++++------
be/src/exec/scan/file_scanner_v2.h | 10 ++-
be/src/format_v2/jni/jni_table_reader.cpp | 8 +++
be/src/format_v2/jni/jni_table_reader.h | 1 +
be/src/format_v2/table/hudi_reader.cpp | 10 +++
be/src/format_v2/table/hudi_reader.h | 2 +
be/src/format_v2/table/iceberg_reader.cpp | 3 +
be/src/format_v2/table/paimon_reader.cpp | 10 +++
be/src/format_v2/table/paimon_reader.h | 2 +
be/src/format_v2/table_reader.cpp | 83 ++++++++++++++++++++++++-
be/src/format_v2/table_reader.h | 27 +++++++++
be/test/exec/scan/file_scanner_v2_test.cpp | 35 +++++++++++
be/test/format_v2/table_reader_test.cpp | 86 ++++++++++++++++++++++++++
13 files changed, 351 insertions(+), 23 deletions(-)
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index 59afe25f3c4..10327f37374 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -58,6 +58,7 @@
#include "format_v2/table/paimon_reader.h"
#include "format_v2/table/remote_doris_reader.h"
#include "format_v2/table_reader.h"
+#include "io/cache/block_file_cache_profile.h"
#include "io/fs/file_meta_cache.h"
#include "io/io_common.h"
#include "runtime/descriptors.h"
@@ -243,6 +244,15 @@ FileScannerV2::RealtimeCounterDeltas
FileScannerV2::TEST_collect_realtime_counte
last_read_rows,
last_bytes_read_from_local,
last_bytes_read_from_remote);
}
+
+void FileScannerV2::TEST_report_file_cache_profile(
+ RuntimeProfile* profile, const io::FileCacheStatistics&
file_cache_statistics) {
+ _report_file_cache_profile(profile, file_cache_statistics);
+}
+
+bool FileScannerV2::TEST_should_skip_not_found(const Status& status, bool
ignore_not_found) {
+ return _should_skip_not_found(status, ignore_not_found);
+}
#endif
bool FileScannerV2::is_supported(const TFileScanRangeParams& params, const
TFileRangeDesc& range) {
@@ -283,6 +293,8 @@ Status FileScannerV2::init(RuntimeState* state, const
VExprContextSPtrs& conjunc
RETURN_IF_ERROR(Scanner::init(state, conjuncts));
_get_block_timer =
ADD_TIMER_WITH_LEVEL(_local_state->scanner_profile(),
"FileScannerV2GetBlockTime", 1);
+ _not_found_file_counter =
ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),
+ "NotFoundFileNum",
TUnit::UNIT, 1);
_file_counter =
ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),
"FileNumber", TUnit::UNIT, 1);
_file_read_bytes_counter =
ADD_COUNTER_WITH_LEVEL(_local_state->scanner_profile(),
@@ -334,7 +346,17 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state,
Block* block, bool* e
if (_should_run_adaptive_batch_size()) {
_table_reader->set_batch_size(_predict_reader_batch_rows());
}
- RETURN_IF_ERROR(_table_reader->get_block(block, eof));
+ const auto status = _table_reader->get_block(block, eof);
+ if (_should_skip_not_found(status,
config::ignore_not_found_file_in_external_table)) {
+ RETURN_IF_ERROR(_table_reader->abort_split());
+ COUNTER_UPDATE(_not_found_file_counter, 1);
+ _state->update_num_finished_scan_range(1);
+ _has_prepared_split = false;
+
block->clear_column_data(cast_set<int64_t>(_projected_columns.size()));
+ *eof = false;
+ continue;
+ }
+ RETURN_IF_ERROR(status);
}
if (*eof) {
_state->update_num_finished_scan_range(1);
@@ -348,23 +370,40 @@ Status FileScannerV2::_get_block_impl(RuntimeState*
state, Block* block, bool* e
}
Status FileScannerV2::_prepare_next_split(bool* eos) {
- bool has_next = _first_scan_range;
- if (!_first_scan_range) {
- RETURN_IF_ERROR(_split_source->get_next(&has_next, &_current_range));
- }
- _first_scan_range = false;
- if (!has_next || _should_stop) {
- *eos = true;
+ while (true) {
+ bool has_next = _first_scan_range;
+ if (!_first_scan_range) {
+ RETURN_IF_ERROR(_split_source->get_next(&has_next,
&_current_range));
+ }
+ _first_scan_range = false;
+ if (!has_next || _should_stop) {
+ *eos = true;
+ return Status::OK();
+ }
+ DORIS_CHECK(_table_reader != nullptr);
+ _current_range_path = _current_range.path;
+
+ std::map<std::string, Field> partition_values;
+ RETURN_IF_ERROR(_generate_partition_values(_current_range,
&partition_values));
+ const auto status =
+ _prepare_table_reader_split(_current_range,
std::move(partition_values));
+ if (_should_skip_not_found(status,
config::ignore_not_found_file_in_external_table)) {
+ RETURN_IF_ERROR(_table_reader->abort_split());
+ COUNTER_UPDATE(_not_found_file_counter, 1);
+ _state->update_num_finished_scan_range(1);
+ continue;
+ }
+ RETURN_IF_ERROR(status);
+ if (_table_reader->current_split_pruned()) {
+ _state->update_num_finished_scan_range(1);
+ continue;
+ }
+ _init_adaptive_batch_size_state(get_range_format_type(*_params,
_current_range));
+ COUNTER_UPDATE(_file_counter, 1);
+ _has_prepared_split = true;
+ *eos = false;
return Status::OK();
}
- DORIS_CHECK(_table_reader != nullptr);
- _current_range_path = _current_range.path;
- _init_adaptive_batch_size_state(get_range_format_type(*_params,
_current_range));
- RETURN_IF_ERROR(_prepare_table_reader_split(_current_range));
- COUNTER_UPDATE(_file_counter, 1);
- _has_prepared_split = true;
- *eos = false;
- return Status::OK();
}
Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
@@ -425,13 +464,17 @@ Status FileScannerV2::_create_table_reader_for_format(
return Status::OK();
}
-Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range)
{
- std::map<std::string, Field> partition_values;
- RETURN_IF_ERROR(_generate_partition_values(range, &partition_values));
+Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range,
+ std::map<std::string, Field>
partition_values) {
format::FileFormat current_split_format;
RETURN_IF_ERROR(_to_file_format(get_range_format_type(*_params, range),
¤t_split_format));
+ VExprContextSPtrs partition_prune_conjuncts;
+ if (_state->query_options().enable_runtime_filter_partition_prune) {
+ RETURN_IF_ERROR(_build_table_conjuncts(&partition_prune_conjuncts));
+ }
RETURN_IF_ERROR(_table_reader->prepare_split({
.partition_values = std::move(partition_values),
+ .partition_prune_conjuncts = std::move(partition_prune_conjuncts),
.cache = _kv_cache,
.current_range = range,
.current_split_format = current_split_format,
@@ -440,6 +483,10 @@ Status FileScannerV2::_prepare_table_reader_split(const
TFileRangeDesc& range) {
return Status::OK();
}
+bool FileScannerV2::_should_skip_not_found(const Status& status, bool
ignore_not_found) {
+ return ignore_not_found && status.is<ErrorCode::NOT_FOUND>();
+}
+
bool FileScannerV2::_should_enable_file_meta_cache() const {
return ExecEnv::GetInstance()->file_meta_cache()->enabled() &&
_split_source->num_scan_ranges() <
config::max_external_file_meta_cache_num / 3;
@@ -874,6 +921,12 @@ FileScannerV2::UncachedReaderBytesStorage
FileScannerV2::_uncached_reader_bytes_
void FileScannerV2::_collect_profile_before_close() {
_report_file_reader_predicate_filtered_rows();
Scanner::_collect_profile_before_close();
+ if (config::enable_file_cache && _state->query_options().enable_file_cache
&&
+ _profile != nullptr) {
+ _report_file_cache_profile(_profile, *_file_cache_statistics);
+
_state->get_query_ctx()->resource_ctx()->io_context()->update_bytes_write_into_cache(
+ _file_cache_statistics->bytes_write_into_cache);
+ }
if (_file_reader_stats != nullptr) {
COUNTER_SET(_file_read_bytes_counter,
cast_set<int64_t>(_file_reader_stats->read_bytes));
COUNTER_SET(_file_read_calls_counter,
cast_set<int64_t>(_file_reader_stats->read_calls));
@@ -884,6 +937,12 @@ void FileScannerV2::_collect_profile_before_close() {
_report_condition_cache_profile();
}
+void FileScannerV2::_report_file_cache_profile(
+ RuntimeProfile* profile, const io::FileCacheStatistics&
file_cache_statistics) {
+ io::FileCacheProfileReporter cache_profile(profile);
+ cache_profile.update(&file_cache_statistics);
+}
+
bool FileScannerV2::_should_update_load_counters() const {
if (_is_load) {
return true;
diff --git a/be/src/exec/scan/file_scanner_v2.h
b/be/src/exec/scan/file_scanner_v2.h
index a911d42c3ad..07e772b4188 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -80,6 +80,9 @@ public:
UncachedReaderBytesStorage uncached_reader_bytes_storage, int64_t*
last_read_bytes,
int64_t* last_read_rows, int64_t* last_bytes_read_from_local,
int64_t* last_bytes_read_from_remote);
+ static void TEST_report_file_cache_profile(
+ RuntimeProfile* profile, const io::FileCacheStatistics&
file_cache_statistics);
+ static bool TEST_should_skip_not_found(const Status& status, bool
ignore_not_found);
#endif
FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t
limit,
@@ -108,7 +111,9 @@ private:
Status _init_table_reader(const TFileRangeDesc& range);
Status _create_table_reader_for_format(const TFileRangeDesc& range,
std::unique_ptr<format::TableReader>* reader) const;
- Status _prepare_table_reader_split(const TFileRangeDesc& range);
+ Status _prepare_table_reader_split(const TFileRangeDesc& range,
+ std::map<std::string, Field>
partition_values);
+ static bool _should_skip_not_found(const Status& status, bool
ignore_not_found);
bool _should_enable_file_meta_cache() const;
std::optional<format::GlobalRowIdContext> _create_global_rowid_context(
const TFileRangeDesc& range) const;
@@ -135,6 +140,8 @@ private:
int64_t* last_read_rows, int64_t* last_bytes_read_from_local,
int64_t* last_bytes_read_from_remote);
static UncachedReaderBytesStorage
_uncached_reader_bytes_storage(TFileType::type file_type);
+ static void _report_file_cache_profile(RuntimeProfile* profile,
+ const io::FileCacheStatistics&
file_cache_statistics);
void _report_file_reader_predicate_filtered_rows();
void _report_condition_cache_profile();
@@ -167,6 +174,7 @@ private:
ShardedKVCache* _kv_cache = nullptr;
RuntimeProfile::Counter* _get_block_timer = nullptr;
+ RuntimeProfile::Counter* _not_found_file_counter = nullptr;
RuntimeProfile::Counter* _file_counter = nullptr;
RuntimeProfile::Counter* _file_read_bytes_counter = nullptr;
RuntimeProfile::Counter* _file_read_calls_counter = nullptr;
diff --git a/be/src/format_v2/jni/jni_table_reader.cpp
b/be/src/format_v2/jni/jni_table_reader.cpp
index deefe26837e..7cec22a7c59 100644
--- a/be/src/format_v2/jni/jni_table_reader.cpp
+++ b/be/src/format_v2/jni/jni_table_reader.cpp
@@ -47,6 +47,9 @@ Status JniTableReader::prepare_split(const SplitReadOptions&
options) {
_current_range = options.current_range;
RETURN_IF_ERROR(validate_scan_range(options.current_range));
RETURN_IF_ERROR(TableReader::prepare_split(options));
+ if (current_split_pruned()) {
+ return Status::OK();
+ }
DORIS_CHECK(!_closed);
DORIS_CHECK(!_scanner_opened);
if (_is_table_level_count_active()) {
@@ -95,6 +98,11 @@ Status JniTableReader::get_block(Block* output_block, bool*
eos) {
}
}
+Status JniTableReader::abort_split() {
+ RETURN_IF_ERROR(_close_jni_scanner());
+ return TableReader::abort_split();
+}
+
Status JniTableReader::_get_next_jni_block(size_t* rows, bool* eof) {
DORIS_CHECK(rows != nullptr);
DORIS_CHECK(eof != nullptr);
diff --git a/be/src/format_v2/jni/jni_table_reader.h
b/be/src/format_v2/jni/jni_table_reader.h
index cfe1eb32c69..f9c854e1971 100644
--- a/be/src/format_v2/jni/jni_table_reader.h
+++ b/be/src/format_v2/jni/jni_table_reader.h
@@ -49,6 +49,7 @@ public:
Status init(TableReadOptions&& options) override;
Status prepare_split(const SplitReadOptions& options) override;
Status get_block(Block* block, bool* eos) override;
+ Status abort_split() override;
Status close() override;
protected:
diff --git a/be/src/format_v2/table/hudi_reader.cpp
b/be/src/format_v2/table/hudi_reader.cpp
index d0b50cbf8aa..3600b0ff706 100644
--- a/be/src/format_v2/table/hudi_reader.cpp
+++ b/be/src/format_v2/table/hudi_reader.cpp
@@ -66,6 +66,16 @@ Status HudiHybridReader::get_block(Block* block, bool* eos) {
return _current_split_reader->get_block(block, eos);
}
+bool HudiHybridReader::current_split_pruned() const {
+ DORIS_CHECK(_current_split_reader != nullptr);
+ return _current_split_reader->current_split_pruned();
+}
+
+Status HudiHybridReader::abort_split() {
+ DORIS_CHECK(_current_split_reader != nullptr);
+ return _current_split_reader->abort_split();
+}
+
Status HudiHybridReader::close() {
Status close_status = Status::OK();
if (_native_reader != nullptr) {
diff --git a/be/src/format_v2/table/hudi_reader.h
b/be/src/format_v2/table/hudi_reader.h
index aeaaedf6ab6..58f222c67ab 100644
--- a/be/src/format_v2/table/hudi_reader.h
+++ b/be/src/format_v2/table/hudi_reader.h
@@ -58,6 +58,8 @@ public:
Status init(format::TableReadOptions&& options) override;
Status prepare_split(const format::SplitReadOptions& options) override;
Status get_block(Block* block, bool* eos) override;
+ bool current_split_pruned() const override;
+ Status abort_split() override;
Status close() override;
private:
diff --git a/be/src/format_v2/table/iceberg_reader.cpp
b/be/src/format_v2/table/iceberg_reader.cpp
index 1be6b1e50e3..5c0693cbba6 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -193,6 +193,9 @@ Status IcebergTableReader::prepare_split(const
format::SplitReadOptions& options
}
}
RETURN_IF_ERROR(TableReader::prepare_split(options));
+ if (current_split_pruned()) {
+ return Status::OK();
+ }
if (_is_table_level_count_active()) {
return Status::OK();
}
diff --git a/be/src/format_v2/table/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index c1a0f441950..c2114968946 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -91,6 +91,16 @@ Status PaimonHybridReader::get_block(Block* block, bool*
eos) {
return _current_split_reader->get_block(block, eos);
}
+bool PaimonHybridReader::current_split_pruned() const {
+ DORIS_CHECK(_current_split_reader != nullptr);
+ return _current_split_reader->current_split_pruned();
+}
+
+Status PaimonHybridReader::abort_split() {
+ DORIS_CHECK(_current_split_reader != nullptr);
+ return _current_split_reader->abort_split();
+}
+
Status PaimonHybridReader::close() {
Status close_status = Status::OK();
if (_native_reader != nullptr) {
diff --git a/be/src/format_v2/table/paimon_reader.h
b/be/src/format_v2/table/paimon_reader.h
index a7aa227ddb4..ae179e6135a 100644
--- a/be/src/format_v2/table/paimon_reader.h
+++ b/be/src/format_v2/table/paimon_reader.h
@@ -63,6 +63,8 @@ public:
Status init(format::TableReadOptions&& options) override;
Status prepare_split(const format::SplitReadOptions& options) override;
Status get_block(Block* block, bool* eos) override;
+ bool current_split_pruned() const override;
+ Status abort_split() override;
Status close() override;
#ifdef BE_TEST
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 37b931b0a9e..17938066420 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -502,6 +502,10 @@ Status TableReader::init(TableReadOptions&& options) {
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"PushDownAggTime", table_profile, 1);
_profile.open_reader_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "OpenReaderTime",
table_profile, 1);
+ _profile.runtime_filter_partition_prune_timer = ADD_TIMER_WITH_LEVEL(
+ _scanner_profile,
"FileScannerRuntimeFilterPartitionPruningTime", 1);
+ _profile.runtime_filter_partition_pruned_range_counter =
ADD_COUNTER_WITH_LEVEL(
+ _scanner_profile, "RuntimeFilterPartitionPrunedRangeNum",
TUnit::UNIT, 1);
}
return Status::OK();
}
@@ -733,12 +737,12 @@ std::unique_ptr<io::FileDescription>
create_file_description(const TFileRangeDes
Status TableReader::prepare_split(const SplitReadOptions& options) {
SCOPED_TIMER(_profile.prepare_split_timer);
+ _current_split_pruned = false;
// Update to current split format to handle ORC/PARQUET files in one table.
_format = options.current_split_format;
_partition_values = std::move(options.partition_values);
- _current_task = std::make_unique<ScanTask>();
- _current_task->data_file = create_file_description(options.current_range);
- _current_file_description = *_current_task->data_file;
+ _current_task.reset();
+ _current_file_description.reset();
_current_file_range_desc = options.current_range;
_current_range_compress_type = options.current_range.__isset.compress_type
?
options.current_range.compress_type
@@ -751,6 +755,15 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
_aggregate_pushdown_tried = false;
_remaining_table_level_count = -1;
_current_reader_reached_eof = false;
+
RETURN_IF_ERROR(_evaluate_partition_prune_conjuncts(options.partition_prune_conjuncts,
+
&_current_split_pruned));
+ if (_current_split_pruned) {
+ COUNTER_UPDATE(_profile.runtime_filter_partition_pruned_range_counter,
1);
+ return Status::OK();
+ }
+ _current_task = std::make_unique<ScanTask>();
+ _current_task->data_file = create_file_description(options.current_range);
+ _current_file_description = *_current_task->data_file;
if (_push_down_agg_type == TPushAggOp::type::COUNT &&
options.current_range.__isset.table_format_params &&
options.current_range.table_format_params.__isset.table_level_row_count) {
@@ -764,6 +777,70 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
return _parse_delete_predicates(options);
}
+Status TableReader::_evaluate_partition_prune_conjuncts(const
VExprContextSPtrs& conjuncts,
+ bool* can_filter_all) {
+ DORIS_CHECK(can_filter_all != nullptr);
+ SCOPED_TIMER(_profile.runtime_filter_partition_prune_timer);
+ *can_filter_all = false;
+ if (conjuncts.empty() || _partition_values.empty()) {
+ return Status::OK();
+ }
+
+ VExprContextSPtrs partition_conjuncts;
+ for (const auto& conjunct : conjuncts) {
+ DORIS_CHECK(conjunct != nullptr);
+ DORIS_CHECK(conjunct->root() != nullptr);
+ std::set<GlobalIndex> global_indices;
+ collect_global_indices(conjunct->root(), &global_indices);
+ if (global_indices.empty()) {
+ continue;
+ }
+ const bool partition_only = std::ranges::all_of(global_indices,
[&](GlobalIndex index) {
+ if (index.value() >= _projected_columns.size()) {
+ return false;
+ }
+ const auto& column = _projected_columns[index.value()];
+ return column.is_partition_key &&
+ find_partition_value(column, _partition_values) != nullptr;
+ });
+ if (partition_only) {
+ partition_conjuncts.push_back(conjunct);
+ }
+ }
+ if (partition_conjuncts.empty()) {
+ return Status::OK();
+ }
+
+ Block block;
+ RETURN_IF_ERROR(_build_partition_prune_block(&block));
+ RowDescriptor row_desc;
+ for (const auto& conjunct : partition_conjuncts) {
+ RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc));
+ RETURN_IF_ERROR(conjunct->open(_runtime_state));
+ }
+ IColumn::Filter result_filter(block.rows(), 1);
+ return VExprContext::execute_conjuncts(partition_conjuncts, nullptr,
&block, &result_filter,
+ can_filter_all);
+}
+
+Status TableReader::_build_partition_prune_block(Block* block) const {
+ DORIS_CHECK(block != nullptr);
+ DORIS_CHECK(!_projected_columns.empty());
+ block->clear();
+ for (const auto& column : _projected_columns) {
+ DORIS_CHECK(column.type != nullptr);
+ ColumnPtr value_column =
column.type->create_column_const_with_default_value(1);
+ if (column.is_partition_key) {
+ const auto* partition_value = find_partition_value(column,
_partition_values);
+ if (partition_value != nullptr) {
+ value_column = column.type->create_column_const(1,
*partition_value);
+ }
+ }
+ block->insert({std::move(value_column), column.type, column.name});
+ }
+ return Status::OK();
+}
+
Status TableReader::_parse_delete_predicates(const SplitReadOptions& options) {
DeleteFileDesc desc {.fs_name = options.current_range.fs_name};
bool has_delete_file = false;
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index c4d8cc232f8..2a8a95817e8 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -107,6 +107,8 @@ struct ReadProfile {
RuntimeProfile::Counter* create_reader_timer = nullptr;
RuntimeProfile::Counter* pushdown_agg_timer = nullptr;
RuntimeProfile::Counter* open_reader_timer = nullptr;
+ RuntimeProfile::Counter* runtime_filter_partition_prune_timer = nullptr;
+ RuntimeProfile::Counter* runtime_filter_partition_pruned_range_counter =
nullptr;
};
struct TableReadOptions {
@@ -134,6 +136,9 @@ struct TableReadOptions {
struct SplitReadOptions {
// Split-level information for reader initialization, which may include
file path, partition values, delete file info, etc. The content is table format
specific and opaque to table reader base class; it's the responsibility of the
concrete table reader implementation to parse necessary information for reader
initialization and filter pushdown.
std::map<std::string, Field> partition_values;
+ // Latest scanner conjuncts rewritten to table/global column indices.
Runtime filters can
+ // arrive after TableReader::init(), so split preparation must receive a
fresh snapshot.
+ VExprContextSPtrs partition_prune_conjuncts;
ShardedKVCache* cache;
TFileRangeDesc current_range;
FileFormat current_split_format = FileFormat::PARQUET;
@@ -168,6 +173,24 @@ public:
// 2. Parse delete predicates from split/task information, which will be
used for later dynamic filtering and delete handling.
virtual Status prepare_split(const SplitReadOptions& options);
+ virtual bool current_split_pruned() const { return _current_split_pruned; }
+
+ // Discard the active split after the caller decides an error is
ignorable, for example a
+ // stale external-table file listing that returns NOT_FOUND. The next
prepare_split() must start
+ // with no concrete reader or split-local state left from the failed split.
+ virtual Status abort_split() {
+ if (_data_reader.reader != nullptr) {
+ RETURN_IF_ERROR(close_current_reader());
+ } else {
+ _current_task.reset();
+ _current_file_description.reset();
+ }
+ _delete_rows = nullptr;
+ _remaining_table_level_count = -1;
+ _current_split_pruned = false;
+ return Status::OK();
+ }
+
// Public entry point for reading a table-schema block. The base class
opens the current reader,
// advances across EOF, and closes exhausted readers. Subclasses provide
protected hooks for
// table-format-specific behavior.
@@ -382,6 +405,9 @@ protected:
}
Status _build_table_filters_from_conjuncts();
+ Status _evaluate_partition_prune_conjuncts(const VExprContextSPtrs&
conjuncts,
+ bool* can_filter_all);
+ Status _build_partition_prune_block(Block* block) const;
Status _open_local_filter_exprs(const FileScanRequest& file_request);
Status _init_reader_condition_cache(const FileScanRequest& file_request);
void _finalize_reader_condition_cache();
@@ -1496,6 +1522,7 @@ protected:
int64_t _remaining_table_level_count = -1;
std::optional<GlobalRowIdContext> _global_rowid_context;
bool _aggregate_pushdown_tried = false;
+ bool _current_split_pruned = false;
TableColumnMapperOptions _mapper_options;
private:
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index f85a5da4c70..6ce0f4ce4cd 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -339,6 +339,41 @@ TEST(FileScannerV2Test,
RealtimeCounterDeltasDoNotChargeLocalFileFallbackAsRemot
EXPECT_EQ(0, deltas.scan_bytes_from_remote_storage);
}
+TEST(FileScannerV2Test, FileCacheStatisticsArePublishedToScannerProfile) {
+ RuntimeProfile profile("file_scanner_v2");
+ io::FileCacheStatistics file_cache_statistics;
+ file_cache_statistics.num_local_io_total = 3;
+ file_cache_statistics.num_remote_io_total = 5;
+ file_cache_statistics.num_peer_io_total = 7;
+ file_cache_statistics.bytes_read_from_local = 11;
+ file_cache_statistics.bytes_read_from_remote = 13;
+ file_cache_statistics.bytes_read_from_peer = 17;
+ file_cache_statistics.bytes_write_into_cache = 19;
+ file_cache_statistics.peer_hosts = {"peer-a", "peer-b"};
+
+ FileScannerV2::TEST_report_file_cache_profile(&profile,
file_cache_statistics);
+
+ ASSERT_NE(profile.get_counter("FileCache"), nullptr);
+ EXPECT_EQ(profile.get_counter("NumLocalIOTotal")->value(), 3);
+ EXPECT_EQ(profile.get_counter("NumRemoteIOTotal")->value(), 5);
+ EXPECT_EQ(profile.get_counter("NumPeerIOTotal")->value(), 7);
+ EXPECT_EQ(profile.get_counter("BytesScannedFromCache")->value(), 11);
+ EXPECT_EQ(profile.get_counter("BytesScannedFromRemote")->value(), 13);
+ EXPECT_EQ(profile.get_counter("BytesScannedFromPeer")->value(), 17);
+ EXPECT_EQ(profile.get_counter("BytesWriteIntoCache")->value(), 19);
+ ASSERT_NE(profile.get_info_string("PeerCacheNodes"), nullptr);
+ EXPECT_EQ(*profile.get_info_string("PeerCacheNodes"), "peer-a, peer-b");
+}
+
+TEST(FileScannerV2Test, NotFoundIsSkippedOnlyWhenConfigured) {
+ const auto not_found = Status::NotFound("missing external file");
+ EXPECT_TRUE(FileScannerV2::TEST_should_skip_not_found(not_found, true));
+ EXPECT_FALSE(FileScannerV2::TEST_should_skip_not_found(not_found, false));
+ EXPECT_FALSE(
+
FileScannerV2::TEST_should_skip_not_found(Status::InternalError("read failed"),
true));
+ EXPECT_FALSE(FileScannerV2::TEST_should_skip_not_found(Status::OK(),
true));
+}
+
// Scenario: partition slots are identified from the explicit FE category when
present, otherwise
// from the legacy is_file_slot flag. Scanner-generated rowid columns must
never be treated as
// partition columns even if FE marks them as non-file slots.
diff --git a/be/test/format_v2/table_reader_test.cpp
b/be/test/format_v2/table_reader_test.cpp
index 8d5b2309d9c..252bfa5c04d 100644
--- a/be/test/format_v2/table_reader_test.cpp
+++ b/be/test/format_v2/table_reader_test.cpp
@@ -978,6 +978,7 @@ struct FakeFileReaderState {
bool eof_with_first_batch = true;
bool inject_delete_conjunct = false;
bool stop_during_aggregate = false;
+ bool not_found_during_init = false;
std::shared_ptr<FileScanRequest> last_request;
std::shared_ptr<ConditionCacheContext> condition_cache_ctx;
std::shared_ptr<io::IOContext> io_ctx;
@@ -995,6 +996,9 @@ public:
Status init(RuntimeState* state) override {
(void)state;
++_state->init_count;
+ if (_state->not_found_during_init) {
+ return Status::NotFound("fake table reader input is missing");
+ }
_eof = false;
return Status::OK();
}
@@ -1171,6 +1175,46 @@ private:
std::unique_ptr<segment_v2::ConditionCache> _cache;
};
+TEST(TableReaderTest, PrepareSplitPrunesPartitionRuntimeFilter) {
+ std::vector<ColumnDefinition> projected_columns;
+ auto partition_column = make_table_column(0, "part",
std::make_shared<DataTypeInt32>());
+ partition_column.is_partition_key = true;
+ projected_columns.push_back(std::move(partition_column));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("scanner");
+ TableReader reader;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ SplitReadOptions pruned_split;
+ pruned_split.current_range.__set_path("unused-pruned-file");
+ pruned_split.partition_values.emplace("part",
Field::create_field<TYPE_INT>(7));
+
pruned_split.partition_prune_conjuncts.push_back(VExprContext::create_shared(
+ runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0,
10))));
+ ASSERT_TRUE(reader.prepare_split(pruned_split).ok());
+ EXPECT_TRUE(reader.current_split_pruned());
+ ASSERT_NE(profile.get_counter("RuntimeFilterPartitionPrunedRangeNum"),
nullptr);
+
EXPECT_EQ(profile.get_counter("RuntimeFilterPartitionPrunedRangeNum")->value(),
1);
+
+ SplitReadOptions retained_split;
+ retained_split.current_range.__set_path("unused-retained-file");
+ retained_split.partition_values.emplace("part",
Field::create_field<TYPE_INT>(11));
+
retained_split.partition_prune_conjuncts.push_back(VExprContext::create_shared(
+ runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0,
10))));
+ ASSERT_TRUE(reader.prepare_split(retained_split).ok());
+ EXPECT_FALSE(reader.current_split_pruned());
+}
+
TEST(TableReaderTest, CanUseInjectedFileReaderForStandaloneUnitTest) {
std::vector<ColumnDefinition> file_schema;
file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
@@ -1229,6 +1273,48 @@ TEST(TableReaderTest,
CanUseInjectedFileReaderForStandaloneUnitTest) {
EXPECT_TRUE(eos);
}
+TEST(TableReaderTest, AbortSplitClearsReaderAfterIgnorableNotFound) {
+ std::vector<ColumnDefinition> file_schema;
+ file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ fake_state->not_found_during_init = true;
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ SplitReadOptions split_options;
+ split_options.current_range.__set_path("missing-fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ const auto status = reader.get_block(&block, &eos);
+ ASSERT_TRUE(status.is<ErrorCode::NOT_FOUND>()) << status;
+ ASSERT_TRUE(reader.abort_split().ok());
+ EXPECT_EQ(fake_state->init_count, 1);
+ EXPECT_EQ(fake_state->close_count, 1);
+
+ fake_state->not_found_during_init = false;
+ split_options.current_range.__set_path("existing-fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split_options).ok());
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_EQ(fake_state->init_count, 2);
+ EXPECT_EQ(fake_state->close_count, 2);
+ ASSERT_TRUE(reader.close().ok());
+}
+
TEST(TableReaderTest, PushDownCountRecordsReaderRowsBeforeClosingReader) {
std::vector<ColumnDefinition> file_schema;
file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
