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 27cb451229c [improvement](file) Execute split residual predicates in
TableReader (#65925)
27cb451229c is described below
commit 27cb451229c7a42dd089c5248847828eb0536d02
Author: Gabriel <[email protected]>
AuthorDate: Thu Jul 23 16:52:33 2026 +0800
[improvement](file) Execute split residual predicates in TableReader
(#65925)
## What changed
- Add a split-local localization result so `TableReader` knows which
table predicates are enforced exactly by the current `FileReader`.
- Evaluate only unlocalized predicates at the end of
`TableReader::finalize_chunk`, after schema evolution, defaults,
partition values, and virtual columns are materialized.
- Make `FileScannerV2` skip row-level conjunct evaluation; it keeps
scheduling and accounting responsibilities only.
- Preserve predicate order by keeping unsafe/stateful predicates and
every later predicate on the table-materialization path.
- Apply the same table-level residual filtering to the Iceberg
position-delete system-table reader.
## Why
`FileScannerV2` previously reevaluated every original conjunct after
`FileReader` had already enforced localized predicates. This duplicated
expression work and made predicate execution ownership unclear.
Localization also depends on each split's physical schema, so ownership
cannot be cached at scanner or table-reader lifetime scope.
The new contract recomputes ownership for every split: localized
predicates belong to `FileReader`, while all remaining predicates belong
to `TableReader` after final materialization.
## Validation
- ASAN BE UT build: `cmake --build be/ut_build_ASAN --target
doris_be_test -j128`
- 287 focused TableReader/ColumnMapper/Iceberg/FileScannerV2 tests
passed.
- 41 JNI/Hudi/Paimon reader tests passed.
- `build-support/check-format.sh` passed with clang-format 16.0.6.
- clang-tidy was attempted but could not analyze the tree because of
existing toolchain/baseline errors (`stddef.h` not found and an
unmatched `NOLINTEND` in `core/types.h`).
---
be/src/exec/operator/scan_operator.cpp | 20 +-
be/src/exec/operator/scan_operator.h | 9 +-
be/src/exec/scan/file_scanner_v2.cpp | 145 +++++++--
be/src/exec/scan/file_scanner_v2.h | 33 +-
be/src/exec/scan/scanner.cpp | 16 +-
be/src/exec/scan/scanner.h | 8 +
be/src/format_v2/column_mapper.cpp | 69 ++++-
be/src/format_v2/column_mapper.h | 15 +-
be/src/format_v2/file_reader.h | 4 +-
be/src/format_v2/table/hudi_reader.cpp | 22 ++
be/src/format_v2/table/hudi_reader.h | 2 +
.../iceberg_position_delete_sys_table_reader.cpp | 31 +-
be/src/format_v2/table/paimon_reader.cpp | 22 ++
be/src/format_v2/table/paimon_reader.h | 2 +
be/src/format_v2/table_reader.cpp | 143 ++++++++-
be/src/format_v2/table_reader.h | 128 +++++---
.../segment/adaptive_block_size_predictor.cpp | 7 +-
.../segment/adaptive_block_size_predictor.h | 1 +
be/test/exec/scan/file_scanner_v2_test.cpp | 35 ++-
be/test/exec/scan/scanner_late_arrival_rf_test.cpp | 57 +++-
...eberg_position_delete_sys_table_reader_test.cpp | 73 +++++
be/test/format_v2/column_mapper_test.cpp | 68 ++++-
be/test/format_v2/table/hudi_reader_test.cpp | 147 +++++++++
be/test/format_v2/table/iceberg_reader_test.cpp | 11 -
be/test/format_v2/table/paimon_reader_test.cpp | 146 +++++++++
be/test/format_v2/table_reader_test.cpp | 340 ++++++++++++++++++++-
.../segment/adaptive_block_size_predictor_test.cpp | 12 +
27 files changed, 1424 insertions(+), 142 deletions(-)
diff --git a/be/src/exec/operator/scan_operator.cpp
b/be/src/exec/operator/scan_operator.cpp
index 3bb234d16b1..bc0c56dbfe7 100644
--- a/be/src/exec/operator/scan_operator.cpp
+++ b/be/src/exec/operator/scan_operator.cpp
@@ -71,13 +71,19 @@ bool ScanLocalState<Derived>::should_run_serial() const {
return _parent->cast<typename Derived::Parent>()._should_run_serial;
}
-Status ScanLocalStateBase::update_late_arrival_runtime_filter(RuntimeState*
state,
- int&
arrived_rf_num) {
+Status ScanLocalStateBase::update_late_arrival_runtime_filter(
+ RuntimeState* state, int applied_rf_num, int& arrived_rf_num,
+ VExprContextSPtrs& arrived_conjuncts) {
// Lock needed because _conjuncts can be accessed concurrently by multiple
scanner threads
LockGuard lock(_conjuncts_lock);
+ arrived_conjuncts.clear();
size_t conjuncts_before = _conjuncts.size();
RETURN_IF_ERROR(_helper.try_append_late_arrival_runtime_filter(state,
_parent->row_descriptor(),
arrived_rf_num, _conjuncts));
+ if (_conjuncts.size() > conjuncts_before) {
+ VExprContextSPtrs appended(_conjuncts.begin() + conjuncts_before,
_conjuncts.end());
+ _late_arrival_conjunct_batches.emplace_back(arrived_rf_num,
std::move(appended));
+ }
if (state->enable_adjust_conjunct_order_by_cost()) {
std::ranges::stable_sort(_conjuncts, [](const auto& a, const auto& b) {
return a->execute_cost() < b->execute_cost();
@@ -91,6 +97,16 @@ Status
ScanLocalStateBase::update_late_arrival_runtime_filter(RuntimeState* stat
if (_conjuncts.size() > conjuncts_before) {
RETURN_IF_ERROR(_on_runtime_filter_update());
}
+ for (const auto& [batch_arrived_rf_num, batch] :
_late_arrival_conjunct_batches) {
+ if (batch_arrived_rf_num <= applied_rf_num) {
+ continue;
+ }
+ for (const auto& conjunct : batch) {
+ VExprContextSPtr cloned;
+ RETURN_IF_ERROR(conjunct->clone(state, cloned));
+ arrived_conjuncts.push_back(std::move(cloned));
+ }
+ }
return Status::OK();
}
diff --git a/be/src/exec/operator/scan_operator.h
b/be/src/exec/operator/scan_operator.h
index ae93e155852..e955840f131 100644
--- a/be/src/exec/operator/scan_operator.h
+++ b/be/src/exec/operator/scan_operator.h
@@ -21,6 +21,8 @@
#include <optional>
#include <set>
#include <string>
+#include <utility>
+#include <vector>
#include "common/status.h"
#include "common/thread_safety_annotations.h"
@@ -96,7 +98,9 @@ public:
uint64_t get_condition_cache_digest() const { return
_condition_cache_digest; }
- Status update_late_arrival_runtime_filter(RuntimeState* state, int&
arrived_rf_num);
+ Status update_late_arrival_runtime_filter(RuntimeState* state, int
applied_rf_num,
+ int& arrived_rf_num,
+ VExprContextSPtrs&
arrived_conjuncts);
Status clone_conjunct_ctxs(VExprContextSPtrs& scanner_conjuncts);
@@ -142,6 +146,9 @@ protected:
AnnotatedMutex _conjuncts_lock;
RuntimeFilterConsumerHelper _helper;
+ // Preserve append identity independently of the cost-sorted operator
snapshot. Every scanner
+ // needs the exact RF contexts added since its own applied count.
+ std::vector<std::pair<int, VExprContextSPtrs>>
_late_arrival_conjunct_batches;
// magic number as seed to generate hash value for condition cache
uint64_t _condition_cache_digest = 0;
// condition cache filter stats
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index cdaa3d2f3b8..a20946cf095 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -24,6 +24,7 @@
#include <map>
#include <memory>
#include <optional>
+#include <sstream>
#include <string>
#include <utility>
@@ -226,7 +227,9 @@ Status rewrite_slot_refs_to_global_index(
#ifdef BE_TEST
FileScannerV2::FileScannerV2(RuntimeState* state, RuntimeProfile* profile,
std::unique_ptr<format::TableReader> table_reader)
- : Scanner(state, profile), _table_reader(std::move(table_reader)) {}
+ : Scanner(state, profile),
+ _table_reader(std::move(table_reader)),
+ _scanner_profile(profile) {}
Status FileScannerV2::TEST_validate_scan_range(const TFileScanRangeParams&
params,
const TFileRangeDesc& range) {
@@ -326,7 +329,9 @@ FileScannerV2::FileScannerV2(RuntimeState* state,
FileScanLocalState* local_stat
Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs&
conjuncts) {
RETURN_IF_ERROR(Scanner::init(state, conjuncts));
+ _initialize_scanner_residual_conjuncts();
auto* profile = _local_state->scanner_profile();
+ _scanner_profile = profile;
const auto hierarchy = file_scan_profile::ensure_hierarchy(profile);
_scanner_total_timer = hierarchy.scanner;
_io_timer = hierarchy.io;
@@ -360,6 +365,11 @@ Status FileScannerV2::init(RuntimeState* state, const
VExprContextSPtrs& conjunc
profile, "AdaptiveBatchActualBytes", TUnit::BYTES,
file_scan_profile::SCANNER, 1);
_adaptive_batch_probe_count_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "AdaptiveBatchProbeCount", TUnit::UNIT,
file_scan_profile::SCANNER, 1);
+ _scanner_residual_filter_timer = ADD_CHILD_TIMER_WITH_LEVEL(
+ profile, "ScannerResidualFilterTime", file_scan_profile::SCANNER,
1);
+ _scanner_residual_rows_filtered_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
+ profile, "ScannerResidualRowsFiltered", TUnit::UNIT,
file_scan_profile::SCANNER, 1);
+ _refresh_scanner_residual_profile();
SCOPED_TIMER(_scanner_total_timer);
SCOPED_TIMER(_init_timer);
_file_cache_statistics = std::make_unique<io::FileCacheStatistics>();
@@ -401,6 +411,7 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state,
Block* block, bool* e
SCOPED_TIMER(_get_block_timer);
while (true) {
RETURN_IF_CANCELLED(state);
+ RETURN_IF_ERROR(_sync_table_reader_conjuncts());
if (!_has_prepared_split) {
RETURN_IF_ERROR(_prepare_next_split(eof));
if (*eof) {
@@ -450,18 +461,33 @@ Status FileScannerV2::_get_block_impl(RuntimeState*
state, Block* block, bool* e
}
Status FileScannerV2::_filter_output_block(Block* block) {
- return
_contextualize_output_filter_status(Scanner::_filter_output_block(block),
- _get_current_format_type());
-}
-
-Status FileScannerV2::_contextualize_output_filter_status(Status status,
-
TFileFormatType::type format_type) {
- if (!status.ok() && format_type == TFileFormatType::FORMAT_ORC) {
- // Error-preserving expressions cannot be reordered into the ORC
reader and therefore run
- // at the scanner boundary; keep their error context identical to ORC
callback failures.
+ if (_scanner_residual_conjuncts.empty() || block->rows() == 0) {
+ return Status::OK();
+ }
+ SCOPED_TIMER(_scanner_residual_filter_timer);
+ const size_t rows_before_filter = block->rows();
+ auto status = VExprContext::filter_block(_scanner_residual_conjuncts,
block, block->columns());
+ if (!status.ok() && _params != nullptr &&
+ _get_current_format_type() == TFileFormatType::FORMAT_ORC) {
status.prepend("Orc row reader nextBatch failed. reason = ");
}
- return status;
+ RETURN_IF_ERROR(status);
+ const int64_t filtered_rows = cast_set<int64_t>(rows_before_filter -
block->rows());
+ _counter.num_rows_unselected += filtered_rows;
+ if (_scanner_residual_rows_filtered_counter != nullptr) {
+ COUNTER_UPDATE(_scanner_residual_rows_filtered_counter, filtered_rows);
+ }
+ return Status::OK();
+}
+
+size_t FileScannerV2::_last_block_rows_read(const Block& block) const {
+ const auto& stats = _table_reader->last_materialized_block_stats();
+ return stats.has_materialized_input ? stats.rows : block.rows();
+}
+
+size_t FileScannerV2::_last_block_bytes_read(const Block& block) const {
+ const auto& stats = _table_reader->last_materialized_block_stats();
+ return stats.has_materialized_input ? stats.allocated_bytes :
block.allocated_bytes();
}
Status FileScannerV2::_prepare_next_split(bool* eos) {
@@ -547,6 +573,7 @@ Status FileScannerV2::_init_table_reader(const
TFileRangeDesc& range) {
RETURN_IF_ERROR(_table_reader->init({
.projected_columns = _projected_columns,
.conjuncts = std::move(table_conjuncts),
+ .table_reader_owned_conjunct_count =
_table_reader_owned_conjunct_count,
.format = file_format,
.scan_params = const_cast<TFileScanRangeParams*>(_params),
.io_ctx = _io_ctx,
@@ -557,6 +584,9 @@ Status FileScannerV2::_init_table_reader(const
TFileRangeDesc& range) {
.push_down_count_columns = std::move(push_down_count_columns),
.condition_cache_digest =
_local_state->get_condition_cache_digest(),
}));
+ _table_reader_applied_rf_num = _applied_rf_num;
+ // RFs collected before TableReader initialization are already present in
the full snapshot.
+ _late_arrival_rf_conjuncts.clear();
return Status::OK();
}
@@ -601,15 +631,12 @@ 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 conjuncts;
- RETURN_IF_ERROR(_build_table_conjuncts(&conjuncts));
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),
- .conjuncts = std::move(conjuncts),
.partition_prune_conjuncts = std::move(partition_prune_conjuncts),
// A metadata COUNT split may span scheduler turns. Do not enter
that irreversible
// synthetic-row path while a runtime filter can still arrive
between batches.
@@ -799,10 +826,15 @@ format::ColumnDefinition
FileScannerV2::_build_table_column(const SlotDescriptor
}
Status FileScannerV2::_build_table_conjuncts(VExprContextSPtrs* conjuncts)
const {
+ return _build_table_conjuncts(_conjuncts, conjuncts);
+}
+
+Status FileScannerV2::_build_table_conjuncts(const VExprContextSPtrs& source,
+ VExprContextSPtrs* conjuncts)
const {
DORIS_CHECK(conjuncts != nullptr);
conjuncts->clear();
- conjuncts->reserve(_conjuncts.size());
- for (const auto& conjunct : _conjuncts) {
+ conjuncts->reserve(source.size());
+ for (const auto& conjunct : source) {
VExprSPtr root;
RETURN_IF_ERROR(format::clone_table_expr_tree(conjunct->root(),
&root));
RETURN_IF_ERROR(rewrite_slot_refs_to_global_index(&root,
_slot_id_to_global_index));
@@ -811,6 +843,68 @@ Status
FileScannerV2::_build_table_conjuncts(VExprContextSPtrs* conjuncts) const
return Status::OK();
}
+size_t FileScannerV2::_safe_conjunct_prefix_size(const VExprContextSPtrs&
conjuncts) {
+ for (size_t conjunct_index = 0; conjunct_index < conjuncts.size();
++conjunct_index) {
+ if
(!format::TableReader::is_safe_to_pre_execute(conjuncts[conjunct_index])) {
+ return conjunct_index;
+ }
+ }
+ return conjuncts.size();
+}
+
+void FileScannerV2::_initialize_scanner_residual_conjuncts() {
+ _table_reader_owned_conjunct_count =
_safe_conjunct_prefix_size(_conjuncts);
+ // Preserve the entire suffix, not only the unsafe expression. Otherwise a
later safe
+ // predicate could run below Scanner before a stateful/error-preserving
ordering barrier.
+ _scanner_residual_conjuncts.assign(
+ _conjuncts.begin() +
cast_set<ptrdiff_t>(_table_reader_owned_conjunct_count),
+ _conjuncts.end());
+ _refresh_scanner_residual_profile();
+}
+
+void FileScannerV2::_refresh_scanner_residual_profile() {
+ if (_scanner_profile == nullptr || _scanner_residual_conjuncts.empty()) {
+ return;
+ }
+ std::ostringstream predicates;
+ predicates << "[";
+ for (size_t conjunct_index = 0; conjunct_index <
_scanner_residual_conjuncts.size();
+ ++conjunct_index) {
+ if (conjunct_index > 0) {
+ predicates << ", ";
+ }
+ predicates <<
_scanner_residual_conjuncts[conjunct_index]->root()->debug_string();
+ }
+ predicates << "]";
+ _scanner_profile->add_info_string("ScannerResidualPredicates",
predicates.str());
+}
+
+Status FileScannerV2::_sync_table_reader_conjuncts() {
+ if (_table_reader == nullptr) {
+ return Status::OK();
+ }
+ if (_table_reader_applied_rf_num == _applied_rf_num) {
+ return Status::OK();
+ }
+ VExprContextSPtrs appended;
+ RETURN_IF_ERROR(_build_table_conjuncts(_late_arrival_rf_conjuncts,
&appended));
+ const size_t owned_count = _scanner_residual_conjuncts.empty()
+ ?
_safe_conjunct_prefix_size(_late_arrival_rf_conjuncts)
+ : 0;
+ // Preserve existing expression state and append the identity-tracked RF
delta. Cost sorting
+ // may move a late RF ahead of an old stateful predicate in the full
scanner snapshot.
+ RETURN_IF_ERROR(_table_reader->append_conjuncts_with_ownership(appended,
owned_count));
+ _table_reader_owned_conjunct_count += owned_count;
+ _scanner_residual_conjuncts.insert(
+ _scanner_residual_conjuncts.end(),
+ _late_arrival_rf_conjuncts.begin() +
cast_set<ptrdiff_t>(owned_count),
+ _late_arrival_rf_conjuncts.end());
+ _refresh_scanner_residual_profile();
+ _late_arrival_rf_conjuncts.clear();
+ _table_reader_applied_rf_num = _applied_rf_num;
+ return Status::OK();
+}
+
TFileFormatType::type FileScannerV2::_get_current_format_type() const {
return get_range_format_type(*_params, _current_range);
}
@@ -934,17 +1028,19 @@ void FileScannerV2::_update_adaptive_batch_size(const
Block& block) {
if (!_should_run_adaptive_batch_size()) {
return;
}
- COUNTER_SET(_adaptive_batch_actual_bytes_counter,
static_cast<int64_t>(block.bytes()));
- if (block.rows() == 0) {
+ const auto& stats = _table_reader->last_materialized_block_stats();
+ const size_t rows = stats.has_materialized_input ? stats.rows :
block.rows();
+ const size_t bytes = stats.has_materialized_input ? stats.bytes :
block.bytes();
+ COUNTER_SET(_adaptive_batch_actual_bytes_counter,
static_cast<int64_t>(bytes));
+ if (rows == 0) {
return;
}
- // The sample is taken after TableReader has finalized file-local columns
to table columns.
- // This matches the memory shape seen by upstream operators and catches
very wide nested
- // columns, such as map/string payloads, after the first probe batch.
+ // Residual predicates run after wide table columns are materialized.
Learn from that pre-filter
+ // shape so selective predicates cannot make the next reader batch
dangerously large.
if (!_block_size_predictor->has_history()) {
COUNTER_UPDATE(_adaptive_batch_probe_count_counter, 1);
}
- _block_size_predictor->update(block);
+ _block_size_predictor->update(rows, bytes);
}
Status FileScannerV2::close(RuntimeState* state) {
@@ -1145,9 +1241,8 @@ void
FileScannerV2::_report_file_reader_predicate_filtered_rows() {
const int64_t filtered_rows = _io_ctx != nullptr ?
_io_ctx->predicate_filtered_rows : 0;
const int64_t filtered_delta = filtered_rows -
_reported_predicate_filtered_rows;
if (filtered_delta > 0) {
- // File readers can evaluate localized conjuncts before a block
reaches Scanner. Count
- // those rows as scanner-level unselected rows so load statistics stay
identical no matter
- // whether a predicate is pushed down or evaluated by
Scanner::_filter_output_block().
+ // FileReader and TableReader both report their owned predicate rows
through the shared IO
+ // context. Preserve scanner-level load statistics without
re-evaluating either predicate.
_counter.num_rows_unselected += filtered_delta;
_reported_predicate_filtered_rows = filtered_rows;
}
diff --git a/be/src/exec/scan/file_scanner_v2.h
b/be/src/exec/scan/file_scanner_v2.h
index fc217506eea..c68cfb1db8b 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -88,15 +88,22 @@ public:
RuntimeProfile* profile, const io::FileCacheStatistics&
file_cache_statistics);
static bool TEST_should_skip_not_found(const Status& status, bool
ignore_not_found);
static bool TEST_should_skip_empty(const Status& status, bool stopped);
- static Status TEST_contextualize_output_filter_status(Status status,
-
TFileFormatType::type format_type) {
- return _contextualize_output_filter_status(std::move(status),
format_type);
- }
static bool TEST_should_run_adaptive_batch_size(bool predictor_initialized,
bool
current_split_uses_metadata_count) {
return _should_run_adaptive_batch_size(predictor_initialized,
current_split_uses_metadata_count);
}
+ void TEST_set_scanner_conjuncts(VExprContextSPtrs conjuncts) {
+ _conjuncts = std::move(conjuncts);
+ _initialize_scanner_residual_conjuncts();
+ }
+ Status TEST_filter_output_block(Block* block) { return
_filter_output_block(block); }
+ size_t TEST_table_reader_owned_conjunct_count() const {
+ return _table_reader_owned_conjunct_count;
+ }
+ size_t TEST_scanner_residual_conjunct_count() const {
+ return _scanner_residual_conjuncts.size();
+ }
#endif
FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t
limit,
@@ -115,6 +122,8 @@ public:
protected:
Status _get_block_impl(RuntimeState* state, Block* block, bool* eof)
override;
Status _filter_output_block(Block* block) override;
+ size_t _last_block_rows_read(const Block& block) const override;
+ size_t _last_block_bytes_read(const Block& block) const override;
void _collect_profile_before_close() override;
bool _should_update_load_counters() const override;
@@ -133,8 +142,6 @@ private:
std::map<std::string, Field>
partition_values);
static bool _should_skip_not_found(const Status& status, bool
ignore_not_found);
static bool _should_skip_empty(const Status& status, bool stopped);
- static Status _contextualize_output_filter_status(Status status,
- TFileFormatType::type
format_type);
bool _should_enable_file_meta_cache() const;
std::optional<format::GlobalRowIdContext> _create_global_rowid_context(
const TFileRangeDesc& range) const;
@@ -146,6 +153,12 @@ private:
Status _build_default_expr(const TFileScanSlotInfo& slot_info,
VExprContextSPtr* ctx) const;
static format::ColumnDefinition _build_table_column(const SlotDescriptor*
slot_desc);
Status _build_table_conjuncts(VExprContextSPtrs* conjuncts) const;
+ Status _build_table_conjuncts(const VExprContextSPtrs& source,
+ VExprContextSPtrs* conjuncts) const;
+ Status _sync_table_reader_conjuncts();
+ static size_t _safe_conjunct_prefix_size(const VExprContextSPtrs&
conjuncts);
+ void _initialize_scanner_residual_conjuncts();
+ void _refresh_scanner_residual_profile();
static Status _to_file_format(TFileFormatType::type format_type,
format::FileFormat* file_format);
void _reset_adaptive_batch_size_state();
@@ -181,6 +194,10 @@ private:
std::string _current_range_path;
std::unique_ptr<format::TableReader> _table_reader;
+ size_t _table_reader_owned_conjunct_count = 0;
+ // Scanner owns one persistent context vector for the first unsafe
conjunct and every later
+ // conjunct. Hybrid child readers may be recreated or switched, but this
state must not be.
+ VExprContextSPtrs _scanner_residual_conjuncts;
std::vector<format::ColumnDefinition> _projected_columns;
// File formats without embedded schema, such as CSV, still need the FE
slot descriptors in
// file-column order. This mirrors old FileScanner::_file_slot_descs and
is passed only to
@@ -214,6 +231,9 @@ private:
RuntimeProfile::Counter* _adaptive_batch_predicted_rows_counter = nullptr;
RuntimeProfile::Counter* _adaptive_batch_actual_bytes_counter = nullptr;
RuntimeProfile::Counter* _adaptive_batch_probe_count_counter = nullptr;
+ RuntimeProfile::Counter* _scanner_residual_filter_timer = nullptr;
+ RuntimeProfile::Counter* _scanner_residual_rows_filtered_counter = nullptr;
+ RuntimeProfile* _scanner_profile = nullptr;
std::unique_ptr<AdaptiveBlockSizePredictor> _block_size_predictor;
int64_t _reported_predicate_filtered_rows = 0;
int64_t _reported_condition_cache_hit_count = 0;
@@ -223,6 +243,7 @@ private:
int64_t _last_bytes_read_from_local = 0;
int64_t _last_bytes_read_from_remote = 0;
int64_t _reported_io_read_time = 0;
+ int _table_reader_applied_rf_num = 0;
};
} // namespace doris
diff --git a/be/src/exec/scan/scanner.cpp b/be/src/exec/scan/scanner.cpp
index 3eaf5ee77eb..138b3910d4f 100644
--- a/be/src/exec/scan/scanner.cpp
+++ b/be/src/exec/scan/scanner.cpp
@@ -19,6 +19,8 @@
#include <glog/logging.h>
+#include <iterator>
+
#include "common/config.h"
#include "common/status.h"
#include "core/block/column_with_type_and_name.h"
@@ -160,8 +162,11 @@ Status Scanner::get_block(RuntimeState* state, Block*
block, bool* eof) {
DCHECK(block->rows() == 0);
break;
}
- _num_rows_read += block->rows();
- _num_byte_read += block->allocated_bytes();
+ // Some scanners apply owned predicates before returning the
block. Account the
+ // materialized input, not only survivors, so the per-turn
progress bound remains
+ // effective for highly selective predicates.
+ _num_rows_read += _last_block_rows_read(*block);
+ _num_byte_read += _last_block_bytes_read(*block);
}
// 2. Filter the output block finally.
@@ -256,7 +261,9 @@ Status Scanner::try_append_late_arrival_runtime_filter() {
}
DCHECK(_applied_rf_num < _total_rf_num);
int arrived_rf_num = 0;
- RETURN_IF_ERROR(_local_state->update_late_arrival_runtime_filter(_state,
arrived_rf_num));
+ VExprContextSPtrs arrived_conjuncts;
+ RETURN_IF_ERROR(_local_state->update_late_arrival_runtime_filter(
+ _state, _applied_rf_num, arrived_rf_num, arrived_conjuncts));
if (arrived_rf_num == _applied_rf_num) {
// No newly arrived runtime filters, just return;
@@ -266,6 +273,9 @@ Status Scanner::try_append_late_arrival_runtime_filter() {
// avoid conjunct destroy in used by storage layer
_conjuncts.clear();
RETURN_IF_ERROR(_local_state->clone_conjunct_ctxs(_conjuncts));
+ _late_arrival_rf_conjuncts.insert(_late_arrival_rf_conjuncts.end(),
+
std::make_move_iterator(arrived_conjuncts.begin()),
+
std::make_move_iterator(arrived_conjuncts.end()));
_applied_rf_num = arrived_rf_num;
return Status::OK();
}
diff --git a/be/src/exec/scan/scanner.h b/be/src/exec/scan/scanner.h
index 1bf4cdb5bb2..8f9753cf112 100644
--- a/be/src/exec/scan/scanner.h
+++ b/be/src/exec/scan/scanner.h
@@ -231,6 +231,11 @@ public:
void update_block_avg_bytes(size_t block_avg_bytes) { _block_avg_bytes =
block_avg_bytes; }
protected:
+ virtual size_t _last_block_rows_read(const Block& block) const { return
block.rows(); }
+ virtual size_t _last_block_bytes_read(const Block& block) const {
+ return block.allocated_bytes();
+ }
+
RuntimeState* _state = nullptr;
ScanLocalStateBase* _local_state = nullptr;
@@ -258,6 +263,9 @@ protected:
// Cloned from _conjuncts of scan node.
// It includes predicate in SQL and runtime filters.
VExprContextSPtrs _conjuncts;
+ // Exact append-only RF delta for readers that preserve state across
multiple splits. It must
+ // not be reconstructed by position from the cost-sorted full conjunct
snapshot.
+ VExprContextSPtrs _late_arrival_rf_conjuncts;
VExprContextSPtrs _projections;
// Used in common subexpression elimination to compute intermediate
results.
std::vector<VExprContextSPtrs> _intermediate_projections;
diff --git a/be/src/format_v2/column_mapper.cpp
b/be/src/format_v2/column_mapper.cpp
index 7457098a76f..8ac3060bac4 100644
--- a/be/src/format_v2/column_mapper.cpp
+++ b/be/src/format_v2/column_mapper.cpp
@@ -399,6 +399,33 @@ std::string TableColumnMapperOptions::debug_string() const
{
return out.str();
}
+bool requires_char_or_varchar_truncation(const ColumnMapping& mapping) {
+ if (mapping.table_type == nullptr) {
+ return false;
+ }
+ const auto table_type = remove_nullable(mapping.table_type);
+ const auto primitive_type = table_type->get_primitive_type();
+ if (primitive_type != TYPE_VARCHAR && primitive_type != TYPE_CHAR) {
+ return false;
+ }
+ const auto target_len = assert_cast<const
DataTypeString*>(table_type.get())->len();
+ if (target_len <= 0) {
+ return false;
+ }
+ if (mapping.file_type == nullptr) {
+ return true;
+ }
+ const auto file_type = remove_nullable(mapping.file_type);
+ DORIS_CHECK(file_type != nullptr);
+ int file_len = -1;
+ if (file_type->get_primitive_type() == TYPE_VARCHAR ||
+ file_type->get_primitive_type() == TYPE_CHAR ||
+ file_type->get_primitive_type() == TYPE_STRING) {
+ file_len = assert_cast<const DataTypeString*>(file_type.get())->len();
+ }
+ return file_len < 0 || target_len < file_len;
+}
+
std::string ColumnDefinition::debug_string() const {
std::ostringstream out;
out << "ColumnDefinition{name=" << name << ", identifier=" <<
field_debug_string(identifier)
@@ -2156,7 +2183,7 @@ Status TableColumnMapper::_build_filter_entries(const
FileScanRequest& file_requ
Status TableColumnMapper::create_scan_request(
const std::vector<TableFilter>& table_filters,
const std::vector<ColumnDefinition>& projected_columns,
FileScanRequest* file_request,
- RuntimeState* runtime_state) {
+ RuntimeState* runtime_state, FilterLocalizationResult*
localization_result) {
// FileReader evaluates expressions against a file-local block. This
mapper owns the
// table-column to file-column conversion, so it also owns the file-local
block positions.
file_request->predicate_columns.clear();
@@ -2192,7 +2219,8 @@ Status TableColumnMapper::create_scan_request(
// 2. Build referenced predicate columns
// Hidden filter mappings must be built before localizing filters, so that
they can be localized together with visible mappings and referenced by
localized filter expressions.
RETURN_IF_ERROR(_build_hidden_filter_mappings(table_filters));
- RETURN_IF_ERROR(localize_filters(table_filters, file_request,
runtime_state));
+ RETURN_IF_ERROR(
+ localize_filters(table_filters, file_request, runtime_state,
localization_result));
for (const auto& mapping : _hidden_mappings) {
if (!mapping.file_local_id.has_value()) {
continue;
@@ -2206,9 +2234,9 @@ Status TableColumnMapper::create_scan_request(
if (is_visible_output) {
continue;
}
- // File-local filtering is an optimization; Scanner still evaluates
the original
- // table-level conjunct after TableReader returns. Only truly hidden
mappings are absent
- // from that scanner-visible block and may safely discard their
payload here.
+ // A localized predicate is enforced exactly before TableReader
materializes output. Only
+ // truly hidden mappings are absent from the final table block and may
discard their
+ // payload after that file-local evaluation.
if (std::ranges::any_of(file_request->predicate_columns,
[local_id](const LocalColumnIndex& projection)
{
return projection.column_id() == local_id;
@@ -2257,7 +2285,11 @@ ColumnMapping*
TableColumnMapper::_find_filter_mapping(GlobalIndex global_index)
Status TableColumnMapper::localize_filters(const std::vector<TableFilter>&
table_filters,
FileScanRequest* file_request,
- RuntimeState* runtime_state) {
+ RuntimeState* runtime_state,
+ FilterLocalizationResult*
localization_result) {
+ if (localization_result != nullptr) {
+ localization_result->localized_filters.assign(table_filters.size(),
false);
+ }
std::set<LocalColumnId> localized_predicate_columns;
FilterProjectionMap filter_projections;
auto filter_mappings = _filter_visible_mappings();
@@ -2295,17 +2327,28 @@ Status TableColumnMapper::localize_filters(const
std::vector<TableFilter>& table
// This keeps expression localization independent from filter iteration
order.
filter_mappings = _filter_visible_mappings();
const auto global_to_file_slot =
build_file_slot_rewrite_map(filter_mappings, _filter_entries);
- for (const auto& table_filter : table_filters) {
+ for (size_t filter_index = 0; filter_index < table_filters.size();
++filter_index) {
+ const auto& table_filter = table_filters[filter_index];
if (table_filter.conjunct != nullptr && table_filter.conjunct->root()
!= nullptr) {
const auto root = table_filter.conjunct->root();
const auto impl = root->get_impl();
const auto predicate = impl != nullptr ? impl : root;
- if (!predicate->is_deterministic() ||
+ if (!table_filter.can_localize || !predicate->is_deterministic() ||
!table_filter_has_only_local_entries(table_filter,
_filter_entries)) {
continue;
}
- // Scanner evaluates the original conjunct after final
materialization. Only predicates
- // whose result is stable across repeated execution may also run
as a file-local copy.
+ if (runtime_state != nullptr &&
+
runtime_state->query_options().truncate_char_or_varchar_columns &&
+ std::ranges::any_of(table_filter.global_indices,
[&](GlobalIndex global_index) {
+ const auto* mapping = _find_filter_mapping(global_index);
+ return mapping != nullptr &&
requires_char_or_varchar_truncation(*mapping);
+ })) {
+ // The table predicate observes the bounded value after
finalize; evaluating it on
+ // a wider file string would change equality and range
semantics.
+ continue;
+ }
+ // FileReader becomes the exact owner only for a stable predicate
whose complete
+ // expression can be rewritten against this split's physical
schema.
RewriteContext rewrite_context {.runtime_state = runtime_state};
VExprSPtr rewrite_root;
Status clone_status;
@@ -2316,8 +2359,7 @@ Status TableColumnMapper::localize_filters(const
std::vector<TableFilter>& table
// `element_at(MAP_VALUES(m)[1], 'age') > 30`. The current
file-local rewrite only
// understands top-level slots and struct-element paths rooted
at top-level slots;
// cloning such expressions can hit the generic TExpr
complex-type limitation.
- // Leave them above TableReader, where Scanner evaluates the
original table-level
- // conjunct after final materialization.
+ // Leave them for TableReader after final table-schema
materialization.
#ifndef NDEBUG
return Status::InternalError(
"Failed to clone table filter for file-local rewrite:
{}, expr={}",
@@ -2353,6 +2395,9 @@ Status TableColumnMapper::localize_filters(const
std::vector<TableFilter>& table
auto localized_conjunct =
VExprContext::create_shared(std::move(localized_root));
RETURN_IF_ERROR(rewrite_context.prepare_created_exprs(localized_conjunct.get()));
file_request->conjuncts.push_back(std::move(localized_conjunct));
+ if (localization_result != nullptr) {
+ localization_result->localized_filters[filter_index] = true;
+ }
for (const auto global_index : table_filter.global_indices) {
const auto* mapping = _find_filter_mapping(global_index);
if (mapping != nullptr && mapping->file_local_id.has_value() &&
diff --git a/be/src/format_v2/column_mapper.h b/be/src/format_v2/column_mapper.h
index 0b42ac39a5c..1ee306a8774 100644
--- a/be/src/format_v2/column_mapper.h
+++ b/be/src/format_v2/column_mapper.h
@@ -40,6 +40,13 @@ namespace doris::format {
struct ColumnDefinition;
struct TableFilter;
+// Reports which table filters were fully rewritten into exact file-local
predicates for the
+// current split. The result is aligned with the TableFilter input vector and
must not be reused for
+// another split because schema evolution can change localization
independently for every file.
+struct FilterLocalizationResult {
+ std::vector<bool> localized_filters;
+};
+
enum class TableColumnMappingMode {
// Match by ColumnDefinition::identifier TYPE_INT as field id.
BY_FIELD_ID,
@@ -164,6 +171,8 @@ struct TableColumnMapperOptions {
std::string debug_string() const;
};
+bool requires_char_or_varchar_truncation(const ColumnMapping& mapping);
+
Status clone_table_expr_tree(const VExprSPtr& expr, VExprSPtr* cloned_expr);
const Field* find_partition_value(const ColumnDefinition& table_column,
const std::map<std::string, Field>&
partition_values);
@@ -195,7 +204,8 @@ public:
virtual Status create_scan_request(const std::vector<TableFilter>&
table_filters,
const std::vector<ColumnDefinition>&
projected_columns,
FileScanRequest* file_request,
- RuntimeState* runtime_state = nullptr);
+ RuntimeState* runtime_state = nullptr,
+ FilterLocalizationResult*
localization_result = nullptr);
// Localize table-level filters to the file schema.
// Trivial mappings can copy structured predicates directly. Type changes
may be localized with
@@ -203,7 +213,8 @@ public:
// table-level finalize/filter fallback.
virtual Status localize_filters(const std::vector<TableFilter>&
table_filters,
FileScanRequest* file_request,
- RuntimeState* runtime_state = nullptr);
+ RuntimeState* runtime_state = nullptr,
+ FilterLocalizationResult*
localization_result = nullptr);
void clear() {
_mappings.clear();
_hidden_mappings.clear();
diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h
index 5f959c3e672..1f90957064b 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -76,7 +76,9 @@ struct FileScanRequest {
std::vector<LocalColumnId> predicate_only_columns;
// file-local column id -> file-local output block position.
std::map<LocalColumnId, LocalIndex> local_positions;
- // Row-level filters converted to file-local expressions from table-level
predicates.
+ // Row-level filters converted to file-local expressions from table-level
predicates. Readers
+ // must enforce these exactly on returned rows; metadata pruning alone
does not transfer
+ // predicate ownership away from TableReader.
VExprContextSPtrs conjuncts;
// Delete predicates converted to file-local expressions. A TRUE result
means that the row is
// deleted, so readers must invert each result when building their keep
filter.
diff --git a/be/src/format_v2/table/hudi_reader.cpp
b/be/src/format_v2/table/hudi_reader.cpp
index d05cfe4b03e..ef9c2772543 100644
--- a/be/src/format_v2/table/hudi_reader.cpp
+++ b/be/src/format_v2/table/hudi_reader.cpp
@@ -125,6 +125,27 @@ void HudiHybridReader::set_batch_size(size_t batch_size) {
}
}
+Status HudiHybridReader::append_conjuncts(const VExprContextSPtrs& conjuncts) {
+ // The wrapper snapshot initializes future children, while every existing
child needs the same
+ // late RF immediately so active and later reused splits keep identical
predicate ownership.
+ const size_t owned_count =
+
_appended_table_reader_owned_conjunct_count.value_or(conjuncts.size());
+ RETURN_IF_ERROR(format::TableReader::append_conjuncts(conjuncts));
+ if (_native_reader != nullptr) {
+
RETURN_IF_ERROR(_native_reader->append_conjuncts_with_ownership(conjuncts,
owned_count));
+ }
+ if (_jni_reader != nullptr) {
+
RETURN_IF_ERROR(_jni_reader->append_conjuncts_with_ownership(conjuncts,
owned_count));
+ }
+ return Status::OK();
+}
+
+const format::MaterializedBlockStats&
HudiHybridReader::last_materialized_block_stats() const {
+ // FileScannerV2 budgets cooperative work from the child that actually
materialized the block.
+ return _current_split_reader != nullptr ?
_current_split_reader->last_materialized_block_stats()
+ :
format::TableReader::last_materialized_block_stats();
+}
+
Status HudiHybridReader::_ensure_current_split_reader(const
format::SplitReadOptions& options) {
DORIS_CHECK(_scan_params != nullptr);
if (_is_jni_split(*_scan_params, options.current_range)) {
@@ -169,6 +190,7 @@ Status
HudiHybridReader::_init_child_reader(format::TableReader* reader,
RETURN_IF_ERROR(reader->init({
.projected_columns = _projected_columns,
.conjuncts = std::move(conjuncts),
+ .table_reader_owned_conjunct_count =
_table_reader_owned_conjunct_count,
.format = file_format,
.scan_params = _scan_params,
.io_ctx = _io_ctx,
diff --git a/be/src/format_v2/table/hudi_reader.h
b/be/src/format_v2/table/hudi_reader.h
index d81e8c80e93..42776422c4a 100644
--- a/be/src/format_v2/table/hudi_reader.h
+++ b/be/src/format_v2/table/hudi_reader.h
@@ -65,6 +65,8 @@ public:
Status abort_split() override;
Status close() override;
void set_batch_size(size_t batch_size) override;
+ Status append_conjuncts(const VExprContextSPtrs& conjuncts) override;
+ const format::MaterializedBlockStats& last_materialized_block_stats()
const override;
#ifdef BE_TEST
void TEST_install_batch_size_children() {
diff --git
a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
index b422243707e..16c7f829089 100644
--- a/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
+++ b/be/src/format_v2/table/iceberg_position_delete_sys_table_reader.cpp
@@ -162,6 +162,12 @@ Status
IcebergPositionDeleteSysTableV2Reader::prepare_split(
const format::SplitReadOptions& options) {
RETURN_IF_ERROR(close());
RETURN_IF_ERROR(format::TableReader::prepare_split(options));
+ if (current_split_pruned()) {
+ return Status::OK();
+ }
+ // This synthetic reader has no physical schema where a predicate can be
localized, so every
+ // split predicate must run after its system-table columns have been
materialized.
+ RETURN_IF_ERROR(_prepare_all_conjuncts_as_remaining());
// The inner delete-file reader has distinct counters, so the outer
preparation can safely
// contain its cache miss/open work without re-entering the same
RuntimeProfile timer.
SCOPED_TIMER(_profile.total_timer);
@@ -174,6 +180,7 @@ Status IcebergPositionDeleteSysTableV2Reader::prepare_split(
Status IcebergPositionDeleteSysTableV2Reader::get_block(Block* block, bool*
eos) {
SCOPED_TIMER(_profile.total_timer);
SCOPED_TIMER(_profile.exec_timer);
+ _reset_materialized_block_stats();
DORIS_CHECK(block != nullptr);
DORIS_CHECK(eos != nullptr);
DORIS_CHECK(block->columns() == _projected_columns.size());
@@ -191,9 +198,19 @@ Status
IcebergPositionDeleteSysTableV2Reader::get_block(Block* block, bool* eos)
return Status::OK();
}
- size_t read_rows = 0;
if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) {
- return _append_deletion_vector_block(block, &read_rows, eos);
+ size_t read_rows = 0;
+ RETURN_IF_ERROR(_append_deletion_vector_block(block, &read_rows, eos));
+ if (read_rows > 0) {
+ _record_materialized_block_stats(*block, read_rows);
+ RETURN_IF_ERROR(_filter_remaining_conjuncts(block, &read_rows));
+ }
+ if (read_rows == 0) {
+ // Yield after one deletion-vector batch so cancellation and
Scanner row budgets are
+ // observed even when residual predicates reject every synthesized
row.
+ block->clear_column_data(_projected_columns.size());
+ }
+ return Status::OK();
}
DORIS_CHECK(_position_reader != nullptr);
@@ -207,8 +224,18 @@ Status
IcebergPositionDeleteSysTableV2Reader::get_block(Block* block, bool* eos)
RETURN_IF_ERROR(_position_reader->get_block(&delete_block,
&position_reader_eof));
const size_t delete_rows = delete_block.rows();
if (delete_rows > 0) {
+ size_t read_rows = 0;
RETURN_IF_ERROR(
_append_position_delete_block(block, delete_block,
delete_rows, &read_rows));
+ _record_materialized_block_stats(*block, read_rows);
+ RETURN_IF_ERROR(_filter_remaining_conjuncts(block, &read_rows));
+ if (read_rows == 0) {
+ // A filtered materialized batch is still progress; return it
to Scanner instead of
+ // consuming an unbounded number of position-delete batches in
this call.
+ block->clear_column_data(_projected_columns.size());
+ *eos = false;
+ return Status::OK();
+ }
*eos = false;
return Status::OK();
}
diff --git a/be/src/format_v2/table/paimon_reader.cpp
b/be/src/format_v2/table/paimon_reader.cpp
index 942fbb23426..3e409fee9da 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -152,6 +152,27 @@ void PaimonHybridReader::set_batch_size(size_t batch_size)
{
}
}
+Status PaimonHybridReader::append_conjuncts(const VExprContextSPtrs&
conjuncts) {
+ // The wrapper snapshot initializes future children, while every existing
child needs the same
+ // late RF immediately so active and later reused splits keep identical
predicate ownership.
+ const size_t owned_count =
+
_appended_table_reader_owned_conjunct_count.value_or(conjuncts.size());
+ RETURN_IF_ERROR(format::TableReader::append_conjuncts(conjuncts));
+ if (_native_reader != nullptr) {
+
RETURN_IF_ERROR(_native_reader->append_conjuncts_with_ownership(conjuncts,
owned_count));
+ }
+ if (_jni_reader != nullptr) {
+
RETURN_IF_ERROR(_jni_reader->append_conjuncts_with_ownership(conjuncts,
owned_count));
+ }
+ return Status::OK();
+}
+
+const format::MaterializedBlockStats&
PaimonHybridReader::last_materialized_block_stats() const {
+ // FileScannerV2 budgets cooperative work from the child that actually
materialized the block.
+ return _current_split_reader != nullptr ?
_current_split_reader->last_materialized_block_stats()
+ :
format::TableReader::last_materialized_block_stats();
+}
+
Status PaimonHybridReader::_ensure_current_split_reader(const
format::SplitReadOptions& options) {
if (_is_jni_split(options.current_range)) {
DCHECK(options.current_split_format == format::FileFormat::JNI);
@@ -199,6 +220,7 @@ Status
PaimonHybridReader::_init_child_reader(format::TableReader* reader,
RETURN_IF_ERROR(reader->init({
.projected_columns = _projected_columns,
.conjuncts = std::move(conjuncts),
+ .table_reader_owned_conjunct_count =
_table_reader_owned_conjunct_count,
.format = file_format,
.scan_params = _scan_params,
.io_ctx = _io_ctx,
diff --git a/be/src/format_v2/table/paimon_reader.h
b/be/src/format_v2/table/paimon_reader.h
index ad29075fed1..6ee5e7d72f3 100644
--- a/be/src/format_v2/table/paimon_reader.h
+++ b/be/src/format_v2/table/paimon_reader.h
@@ -71,6 +71,8 @@ public:
Status abort_split() override;
Status close() override;
void set_batch_size(size_t batch_size) override;
+ Status append_conjuncts(const VExprContextSPtrs& conjuncts) override;
+ const format::MaterializedBlockStats& last_materialized_block_stats()
const override;
#ifdef BE_TEST
static bool TEST_is_jni_split(const TFileRangeDesc& range) { return
_is_jni_split(range); }
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index 4beaf8c9ff5..a073f01f237 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -678,6 +678,8 @@ Status TableReader::init(TableReadOptions&& options) {
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"PrepareSplitTime", table_profile, 1);
_profile.finalize_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"FinalizeBlockTime", table_profile, 1);
+ _profile.residual_filter_timer = ADD_CHILD_TIMER_WITH_LEVEL(
+ _scanner_profile, "ResidualFilterTime", table_profile, 1);
_profile.create_reader_timer =
ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile,
"CreateReaderTime", table_profile, 1);
_profile.pushdown_agg_timer =
@@ -721,6 +723,9 @@ Status TableReader::init(TableReadOptions&& options) {
_push_down_count_columns = options.push_down_count_columns;
_initial_condition_cache_digest = options.condition_cache_digest;
_condition_cache_digest = _initial_condition_cache_digest;
+ _table_reader_owned_conjunct_count =
+
options.table_reader_owned_conjunct_count.value_or(options.conjuncts.size());
+ DORIS_CHECK_LE(_table_reader_owned_conjunct_count,
options.conjuncts.size());
_projected_columns = std::move(options.projected_columns);
if (supports_iceberg_scan_semantics_v1(_scan_params)) {
for (auto& projected_column : _projected_columns) {
@@ -734,7 +739,64 @@ Status TableReader::init(TableReadOptions&& options) {
}
_system_properties = create_system_properties(_scan_params);
_mapper_options.mode = TableColumnMappingMode::BY_NAME;
- _conjuncts = std::move(options.conjuncts);
+ return _replace_conjuncts(options.conjuncts);
+}
+
+Status TableReader::_prepare_conjunct(const VExprContextSPtr& source,
VExprContextSPtr* prepared) {
+ DORIS_CHECK(source != nullptr);
+ DORIS_CHECK(source->root() != nullptr);
+ DORIS_CHECK(prepared != nullptr);
+ VExprSPtr root;
+ RETURN_IF_ERROR(clone_table_expr_tree(source->root(), &root));
+ auto conjunct = VExprContext::create_shared(std::move(root));
+ RETURN_IF_ERROR(conjunct->prepare(_runtime_state, RowDescriptor {}));
+ RETURN_IF_ERROR(conjunct->open(_runtime_state));
+ *prepared = std::move(conjunct);
+ return Status::OK();
+}
+
+Status TableReader::_replace_conjuncts(const VExprContextSPtrs& conjuncts) {
+ VExprContextSPtrs prepared;
+ prepared.reserve(conjuncts.size());
+ for (const auto& source : conjuncts) {
+ VExprContextSPtr conjunct;
+ RETURN_IF_ERROR(_prepare_conjunct(source, &conjunct));
+ prepared.push_back(std::move(conjunct));
+ }
+ _conjuncts = std::move(prepared);
+ return Status::OK();
+}
+
+Status TableReader::append_conjuncts_with_ownership(const VExprContextSPtrs&
conjuncts,
+ size_t
table_reader_owned_conjunct_count) {
+ DORIS_CHECK(!_appended_table_reader_owned_conjunct_count.has_value());
+ _appended_table_reader_owned_conjunct_count =
table_reader_owned_conjunct_count;
+ auto status = append_conjuncts(conjuncts);
+ _appended_table_reader_owned_conjunct_count.reset();
+ return status;
+}
+
+Status TableReader::append_conjuncts(const VExprContextSPtrs& conjuncts) {
+ const size_t owned_count =
+
_appended_table_reader_owned_conjunct_count.value_or(conjuncts.size());
+ DORIS_CHECK_LE(owned_count, conjuncts.size());
+ // Once Scanner owns a suffix, later predicates cannot be inserted into
the TableReader-owned
+ // prefix without reordering them ahead of that stateful/error-preserving
barrier.
+ DORIS_CHECK(owned_count == 0 || _table_reader_owned_conjunct_count ==
_conjuncts.size());
+ for (size_t conjunct_index = 0; conjunct_index < conjuncts.size();
++conjunct_index) {
+ const auto& source = conjuncts[conjunct_index];
+ VExprContextSPtr conjunct;
+ RETURN_IF_ERROR(_prepare_conjunct(source, &conjunct));
+ _conjuncts.push_back(conjunct);
+ if (conjunct_index < owned_count) {
+ ++_table_reader_owned_conjunct_count;
+ }
+ if (_current_task != nullptr && conjunct_index < owned_count) {
+ // The active reader has already fixed its localized predicate
set. Appended runtime
+ // filters must remain residual until the next split rebuilds its
FileScanRequest.
+ _remaining_conjuncts.push_back(std::move(conjunct));
+ }
+ }
return Status::OK();
}
@@ -742,20 +804,27 @@ Status TableReader::_build_table_filters_from_conjuncts()
{
_table_filters.clear();
_constant_pruning_safe_filter_count = 0;
bool in_safe_prefix = true;
- for (const auto& conjunct : _conjuncts) {
+ for (size_t conjunct_index = 0; conjunct_index < _conjuncts.size();
++conjunct_index) {
+ const auto& conjunct = _conjuncts[conjunct_index];
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
// `_table_filters` omits expressions without slot references, but
such an expression still
// occupies a position in the row-level conjunct order. Record how
many localized filters
// precede the first unsafe original conjunct so constant pruning
cannot jump over a
- // slotless non-deterministic/error-preserving barrier. Unsafe
predicates remain solely on
- // Scanner's original row-level path because localizing a clone would
execute their state
- // twice with independent state.
- if (in_safe_prefix && !_is_safe_to_pre_execute(conjunct)) {
+ // slotless non-deterministic/error-preserving barrier. An unsafe
predicate is either kept
+ // on TableReader's post-materialization path by a standalone caller
or carried only for
+ // analysis when FileScannerV2 owns the ordered suffix.
+ if (in_safe_prefix && !is_safe_to_pre_execute(conjunct)) {
in_safe_prefix = false;
}
+ const size_t filters_before = _table_filters.size();
RETURN_IF_ERROR(
build_table_filters_from_conjunct(conjunct, _runtime_state,
&_table_filters));
+ for (size_t filter_index = filters_before; filter_index <
_table_filters.size();
+ ++filter_index) {
+ _table_filters[filter_index].source_conjunct_index =
conjunct_index;
+ _table_filters[filter_index].can_localize = in_safe_prefix;
+ }
if (in_safe_prefix) {
_constant_pruning_safe_filter_count = _table_filters.size();
}
@@ -763,6 +832,59 @@ Status TableReader::_build_table_filters_from_conjuncts() {
return Status::OK();
}
+Status TableReader::_prepare_all_conjuncts_as_remaining() {
+ // Expression contexts carry mutable state (for example sequence/stateful
functions). Select
+ // from the TableReader-owned contexts instead of reopening clones for
every split.
+ _remaining_conjuncts.assign(
+ _conjuncts.begin(),
+ _conjuncts.begin() +
cast_set<ptrdiff_t>(_table_reader_owned_conjunct_count));
+ return Status::OK();
+}
+
+Status TableReader::_prepare_remaining_conjuncts(
+ const FilterLocalizationResult& localization_result) {
+ DORIS_CHECK(localization_result.localized_filters.size() ==
_table_filters.size());
+ std::vector<bool> localized_conjuncts(_conjuncts.size(), false);
+ for (size_t filter_index = 0; filter_index < _table_filters.size();
++filter_index) {
+ if (!localization_result.localized_filters[filter_index]) {
+ continue;
+ }
+ const size_t source_index =
_table_filters[filter_index].source_conjunct_index;
+ DORIS_CHECK(source_index < localized_conjuncts.size());
+ localized_conjuncts[source_index] = true;
+ }
+
+ _remaining_conjuncts.clear();
+ for (size_t conjunct_index = 0; conjunct_index <
_table_reader_owned_conjunct_count;
+ ++conjunct_index) {
+ if (localized_conjuncts[conjunct_index]) {
+ continue;
+ }
+ _remaining_conjuncts.push_back(_conjuncts[conjunct_index]);
+ }
+ return Status::OK();
+}
+
+Status TableReader::_filter_remaining_conjuncts(Block* block, size_t* rows) {
+ DORIS_CHECK(block != nullptr);
+ DORIS_CHECK(rows != nullptr);
+ if (*rows == 0 || _remaining_conjuncts.empty()) {
+ return Status::OK();
+ }
+ SCOPED_TIMER(_profile.residual_filter_timer);
+ const size_t rows_before_filter = *rows;
+ auto status = VExprContext::filter_block(_remaining_conjuncts, block,
block->columns());
+ if (!status.ok() && _format == FileFormat::ORC) {
+ status.prepend("Orc row reader nextBatch failed. reason = ");
+ }
+ RETURN_IF_ERROR(status);
+ *rows = block->columns() == 0 ? rows_before_filter : block->rows();
+ if (_io_ctx != nullptr) {
+ _io_ctx->predicate_filtered_rows += rows_before_filter - *rows;
+ }
+ return Status::OK();
+}
+
Status TableReader::_open_local_filter_exprs(const FileScanRequest&
file_request) {
RowDescriptor row_desc;
for (const auto& conjunct : file_request.conjuncts) {
@@ -1003,6 +1125,9 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
SCOPED_TIMER(_profile.total_timer);
SCOPED_TIMER(_profile.prepare_split_timer);
_current_split_pruned = false;
+ // Predicate localization belongs to the physical schema of one split.
Clear the previous
+ // ownership before any early return so a pruned or failed split cannot
leak it to the next one.
+ _remaining_conjuncts.clear();
_all_runtime_filters_applied_for_split =
options.all_runtime_filters_applied;
_condition_cache_digest_covers_current_split =
options.condition_cache_digest.has_value();
if (options.condition_cache_digest.has_value()) {
@@ -1016,7 +1141,7 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
_condition_cache_digest = _initial_condition_cache_digest;
}
if (options.conjuncts.has_value()) {
- _conjuncts = *options.conjuncts;
+ RETURN_IF_ERROR(_replace_conjuncts(*options.conjuncts));
}
// Update to current split format to handle ORC/PARQUET files in one table.
_format = options.current_split_format;
@@ -1086,7 +1211,7 @@ Status
TableReader::_evaluate_partition_prune_conjuncts(const VExprContextSPtrs&
// Keep only the safe prefix of the original conjunct order. If an
unsafe conjunct is
// skipped, a later predicate could prune the split before the unsafe
one reaches its
// normal row-level evaluation point.
- if (!_is_safe_to_pre_execute(conjunct)) {
+ if (!is_safe_to_pre_execute(conjunct)) {
break;
}
std::set<GlobalIndex> global_indices;
@@ -1122,7 +1247,7 @@ Status
TableReader::_evaluate_partition_prune_conjuncts(const VExprContextSPtrs&
can_filter_all);
}
-bool TableReader::_is_safe_to_pre_execute(const VExprContextSPtr& conjunct) {
+bool TableReader::is_safe_to_pre_execute(const VExprContextSPtr& conjunct) {
DORIS_CHECK(conjunct != nullptr);
DORIS_CHECK(conjunct->root() != nullptr);
const auto root = conjunct->root();
diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h
index 7adbb3b0a52..47d442f7fc1 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -78,10 +78,14 @@ namespace doris::format {
using DeleteRows = std::vector<int64_t>;
// Row-level predicates on table/global schema. They are rewritten to
file-local expressions when
-// possible, and remain the source of row-level filtering after localization.
+// possible; otherwise TableReader evaluates them after final table-schema
materialization.
struct TableFilter {
VExprContextSPtr conjunct;
std::vector<GlobalIndex> global_indices;
+ size_t source_conjunct_index = 0;
+ // False after the first unsafe source conjunct so file-local execution
cannot reorder a later
+ // predicate ahead of stateful or error-preserving table semantics.
+ bool can_localize = true;
};
struct ScanTask {
@@ -112,6 +116,7 @@ struct ReadProfile {
RuntimeProfile::Counter* exec_timer = nullptr;
RuntimeProfile::Counter* prepare_split_timer = nullptr;
RuntimeProfile::Counter* finalize_timer = nullptr;
+ RuntimeProfile::Counter* residual_filter_timer = nullptr;
RuntimeProfile::Counter* create_reader_timer = nullptr;
RuntimeProfile::Counter* pushdown_agg_timer = nullptr;
RuntimeProfile::Counter* open_reader_timer = nullptr;
@@ -128,12 +133,23 @@ struct ReadProfile {
RuntimeProfile::Counter* file_reader_close_timer = nullptr;
};
+struct MaterializedBlockStats {
+ bool has_materialized_input = false;
+ size_t rows = 0;
+ size_t bytes = 0;
+ size_t allocated_bytes = 0;
+};
+
struct TableReadOptions {
// Columns need to be read from file and output by table reader. They are
all in table/global
// schema semantics.
const std::vector<ColumnDefinition> projected_columns;
// All complex conjuncts from scan operator
const VExprContextSPtrs conjuncts;
+ // Number of leading conjuncts whose row-level execution is owned by
TableReader/FileReader.
+ // FileScannerV2 still passes the complete ordered list so mapping,
pruning guards, aggregate
+ // eligibility, and condition-cache analysis see the exact query
semantics. nullopt means all.
+ const std::optional<size_t> table_reader_owned_conjunct_count =
std::nullopt;
// File format of the underlying data files, needed for reader
initialization and reader-level
// filter pushdown.
const FileFormat format;
@@ -206,6 +222,10 @@ public:
#ifdef BE_TEST
size_t TEST_batch_size() const { return _batch_size; }
+ size_t TEST_conjunct_count() const { return _conjuncts.size(); }
+ size_t TEST_table_reader_owned_conjunct_count() const {
+ return _table_reader_owned_conjunct_count;
+ }
bool TEST_current_data_file_is_immutable() const {
DORIS_CHECK(_current_task != nullptr);
DORIS_CHECK(_current_task->data_file != nullptr);
@@ -226,6 +246,25 @@ public:
return _current_split_uses_metadata_count;
}
+ // Runtime filters that arrive after a split has opened cannot be pushed
into that file reader.
+ // Keep their expression contexts in TableReader and evaluate them as
residual predicates for
+ // the active reader; later splits can localize them normally.
+ virtual Status append_conjuncts(const VExprContextSPtrs& conjuncts);
+
+ // Append a full ordered snapshot delta while marking only its leading
prefix as owned by
+ // TableReader/FileReader. This non-virtual wrapper preserves the
long-standing virtual API and
+ // carries the ownership boundary through hybrid readers to their children.
+ Status append_conjuncts_with_ownership(const VExprContextSPtrs& conjuncts,
+ size_t
table_reader_owned_conjunct_count);
+
+ // Shared safety classification for deciding which ordered conjunct prefix
may execute below
+ // Scanner without changing stateful or error-preserving semantics.
+ static bool is_safe_to_pre_execute(const VExprContextSPtr& conjunct);
+
+ virtual const MaterializedBlockStats& last_materialized_block_stats()
const {
+ return _last_materialized_block_stats;
+ }
+
// 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.
@@ -245,6 +284,7 @@ public:
_remaining_file_level_count = -1;
_current_split_uses_metadata_count = false;
_current_split_pruned = false;
+ _remaining_conjuncts.clear();
return Status::OK();
}
@@ -254,6 +294,7 @@ public:
virtual Status get_block(Block* block, bool* eos) {
SCOPED_TIMER(_profile.total_timer);
SCOPED_TIMER(_profile.exec_timer);
+ _last_materialized_block_stats = {};
DORIS_CHECK(block->columns() == _projected_columns.size());
block->clear_column_data(_projected_columns.size());
@@ -326,7 +367,7 @@ public:
RETURN_IF_ERROR(_check_file_block_columns("after file reader
get_block", current_rows));
#endif
DORIS_CHECK(block->columns() ==
_data_reader.column_mapper->mappings().size());
- RETURN_IF_ERROR(finalize_chunk(block, current_rows));
+ RETURN_IF_ERROR(finalize_chunk(block, ¤t_rows));
#ifndef NDEBUG
RETURN_IF_ERROR(
_check_table_block_columns("after finalize_chunk", block,
current_rows));
@@ -335,6 +376,13 @@ public:
_current_reader_reached_eof = !stopped_during_read;
RETURN_IF_ERROR(close_current_reader());
}
+ if (current_rows == 0) {
+ // One materialized batch is one Scanner progress unit even
when residual
+ // predicates reject every row. Returning here preserves
row-budget and
+ // cancellation checks in Scanner::get_block().
+ block->clear_column_data(_projected_columns.size());
+ return Status::OK();
+ }
return Status::OK();
}
}
@@ -352,6 +400,7 @@ public:
_remaining_table_level_count = -1;
_remaining_file_level_count = -1;
_current_split_uses_metadata_count = false;
+ _remaining_conjuncts.clear();
return Status::OK();
}
@@ -437,14 +486,17 @@ protected:
// reader with the request. File scan request carries row-level
expression filters and
// file-level pruning hints. Only expression filters decide returned
rows.
auto file_request = std::make_shared<FileScanRequest>();
+ FilterLocalizationResult localization_result;
RETURN_IF_ERROR(_data_reader.column_mapper->create_scan_request(
- _table_filters, _projected_columns, file_request.get(),
_runtime_state));
+ _table_filters, _projected_columns, file_request.get(),
_runtime_state,
+ &localization_result));
bool constant_filter_pruned_split = false;
RETURN_IF_ERROR(_evaluate_constant_filters(&constant_filter_pruned_split));
if (constant_filter_pruned_split) {
RETURN_IF_ERROR(close_current_reader());
return Status::OK();
}
+ RETURN_IF_ERROR(_prepare_remaining_conjuncts(localization_result));
// COUNT(*) has no semantic column argument, but Nereids retains a
minimum-width scan slot
// so the scan node still has an output tuple. Record only the current
non-predicate file
// columns before table-format hooks add row-position or
equality-delete dependencies. This
@@ -520,9 +572,13 @@ protected:
}
Status _build_table_filters_from_conjuncts();
+ Status _replace_conjuncts(const VExprContextSPtrs& conjuncts);
+ Status _prepare_conjunct(const VExprContextSPtr& source, VExprContextSPtr*
prepared);
+ Status _prepare_remaining_conjuncts(const FilterLocalizationResult&
localization_result);
+ Status _prepare_all_conjuncts_as_remaining();
+ Status _filter_remaining_conjuncts(Block* block, size_t* rows);
Status _evaluate_partition_prune_conjuncts(const VExprContextSPtrs&
conjuncts,
bool* can_filter_all);
- static bool _is_safe_to_pre_execute(const VExprContextSPtr& conjunct);
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);
@@ -541,7 +597,7 @@ protected:
if (table_filter.conjunct == nullptr) {
continue;
}
- DORIS_CHECK(_is_safe_to_pre_execute(table_filter.conjunct));
+ DORIS_CHECK(is_safe_to_pre_execute(table_filter.conjunct));
// RuntimeFilterExpr does not implement execute_column_impl(); it
is evaluated by the
// row-level filter path through execute_filter(). Constant split
pruning uses
// VExprContext::execute() on a one-row synthetic block, so
runtime filters must not be
@@ -758,6 +814,7 @@ protected:
}
_table_filters.clear();
_constant_pruning_safe_filter_count = 0;
+ _remaining_conjuncts.clear();
_data_reader.file_schema.clear();
_data_reader.file_block_layout.clear();
_data_reader.block_template.clear();
@@ -773,15 +830,28 @@ protected:
}
}
+ void _reset_materialized_block_stats() { _last_materialized_block_stats =
{}; }
+
+ void _record_materialized_block_stats(const Block& block, size_t rows) {
+ _last_materialized_block_stats = {
+ .has_materialized_input = true,
+ .rows = rows,
+ .bytes = block.bytes(),
+ .allocated_bytes = block.allocated_bytes(),
+ };
+ }
+
// Finalize file-local block to table/global schema block.
- Status finalize_chunk(Block* block, const size_t rows) {
+ Status finalize_chunk(Block* block, size_t* rows) {
+ DORIS_CHECK(rows != nullptr);
SCOPED_TIMER(_profile.finalize_timer);
size_t idx = 0;
const auto& mappings = _data_reader.column_mapper->mappings();
for (const auto& mapping : mappings) {
ColumnPtr column;
- RETURN_IF_ERROR(_materialize_mapping_column(mapping,
&_data_reader.block_template, rows,
- &column, idx + 1 ==
mappings.size()));
+ RETURN_IF_ERROR(_materialize_mapping_column(mapping,
&_data_reader.block_template,
+ *rows, &column,
+ idx + 1 ==
mappings.size()));
block->replace_by_position(idx,
IColumn::mutate(std::move(column)));
idx++;
}
@@ -789,7 +859,12 @@ protected:
// Enforce CHAR/VARCHAR length declared by the table schema after all
file-to-table
// materialization has finished.
RETURN_IF_ERROR(_truncate_char_or_varchar_columns(block));
- return Status::OK();
+ // Preserve the cost of materialization before residual predicates
shrink the block. The
+ // scanner uses this snapshot for bounded progress and adaptive batch
sizing.
+ _record_materialized_block_stats(*block, *rows);
+ // Predicate ownership is split-local: only predicates not
acknowledged as exact by this
+ // split's FileScanRequest run here, after
virtual/default/schema-evolution values exist.
+ return _filter_remaining_conjuncts(block, rows);
}
// Materialize virtual columns in the table block, such as Iceberg _row_id
and
@@ -953,31 +1028,7 @@ protected:
// - table VARCHAR(10), file STRING: truncate to 10 because STRING has no
declared bound;
// - table STRING, any file type: no truncation because the target has no
bound.
static bool _should_truncate_char_or_varchar_column(const ColumnMapping&
mapping) {
- if (mapping.table_type == nullptr) {
- return false;
- }
- const auto table_type = remove_nullable(mapping.table_type);
- const auto primitive_type = table_type->get_primitive_type();
- if (primitive_type != TYPE_VARCHAR && primitive_type != TYPE_CHAR) {
- return false;
- }
- const auto target_len = assert_cast<const
DataTypeString*>(table_type.get())->len();
- if (target_len <= 0) {
- return false;
- }
- if (mapping.file_type == nullptr) {
- return true;
- }
- const auto file_type = remove_nullable(mapping.file_type);
- DORIS_CHECK(file_type != nullptr);
- int file_len = -1;
- if (file_type->get_primitive_type() == TYPE_VARCHAR ||
- file_type->get_primitive_type() == TYPE_CHAR ||
- file_type->get_primitive_type() == TYPE_STRING) {
- file_len = assert_cast<const
DataTypeString*>(file_type.get())->len();
- }
-
- return file_len < 0 || target_len < file_len;
+ return requires_char_or_varchar_truncation(mapping);
}
// Truncate a materialized CHAR/VARCHAR column in place by reusing the
vectorized substring
@@ -1074,9 +1125,8 @@ protected:
if (!_all_runtime_filters_applied_for_split) {
return false;
}
- // Scanner owns the original conjunct list and evaluates it after
TableReader finalizes
- // rows. Even a slotless conjunct that cannot become a TableFilter
must see every source
- // row before an aggregate reduces the stream to synthetic
COUNT/MINMAX rows.
+ // Even a slotless conjunct that cannot become a TableFilter must see
every source row
+ // before an aggregate reduces the stream to synthetic COUNT/MINMAX
rows.
if (!_conjuncts.empty()) {
return false;
}
@@ -1748,6 +1798,10 @@ protected:
// intentionally absent from that vector but must still act as ordering
barriers.
size_t _constant_pruning_safe_filter_count = 0;
VExprContextSPtrs _conjuncts;
+ size_t _table_reader_owned_conjunct_count = 0;
+ std::optional<size_t> _appended_table_reader_owned_conjunct_count;
+ VExprContextSPtrs _remaining_conjuncts;
+ MaterializedBlockStats _last_materialized_block_stats;
ReadProfile _profile;
// Parsed from row-position based delete files, including position delete
and deletion vector.
DeleteRows* _delete_rows = nullptr;
diff --git a/be/src/storage/segment/adaptive_block_size_predictor.cpp
b/be/src/storage/segment/adaptive_block_size_predictor.cpp
index d8cc700f579..7a5ad573a2e 100644
--- a/be/src/storage/segment/adaptive_block_size_predictor.cpp
+++ b/be/src/storage/segment/adaptive_block_size_predictor.cpp
@@ -32,11 +32,14 @@
AdaptiveBlockSizePredictor::AdaptiveBlockSizePredictor(size_t preferred_block_si
_metadata_hint_bytes_per_row(metadata_hint_bytes_per_row) {}
void AdaptiveBlockSizePredictor::update(const Block& block) {
- size_t rows = block.rows();
+ update(block.rows(), block.bytes());
+}
+
+void AdaptiveBlockSizePredictor::update(size_t rows, size_t bytes) {
if (rows == 0) {
return;
}
- double cur = static_cast<double>(block.bytes()) /
static_cast<double>(rows);
+ double cur = static_cast<double>(bytes) / static_cast<double>(rows);
if (!_has_history) {
_bytes_per_row = cur;
diff --git a/be/src/storage/segment/adaptive_block_size_predictor.h
b/be/src/storage/segment/adaptive_block_size_predictor.h
index e03f18c2a53..f327fec8517 100644
--- a/be/src/storage/segment/adaptive_block_size_predictor.h
+++ b/be/src/storage/segment/adaptive_block_size_predictor.h
@@ -60,6 +60,7 @@ public:
// Update EWMA estimates from a completed batch. Must be called only when
block.rows() > 0
// and the batch returned Status::OK().
void update(const Block& block);
+ void update(size_t rows, size_t bytes);
// Predict how many rows the next batch should read.
// Never exceeds |block_size_rows|; never returns less than 1.
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index 87d4c92a776..e1a2fd0cf88 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -749,18 +749,6 @@ TEST(FileScannerV2Test, EndOfFileIsSkippedAsEmptySplit) {
EXPECT_FALSE(FileScannerV2::TEST_should_skip_empty(Status::OK(), false));
}
-TEST(FileScannerV2Test, OrcScannerResidualFilterRetainsNextBatchContext) {
- auto status = FileScannerV2::TEST_contextualize_output_filter_status(
- Status::InvalidArgument("synthetic row filter failure"),
TFileFormatType::FORMAT_ORC);
- EXPECT_NE(status.to_string().find("nextBatch failed"), std::string::npos)
<< status;
- EXPECT_NE(status.to_string().find("synthetic row filter failure"),
std::string::npos) << status;
-
- status = FileScannerV2::TEST_contextualize_output_filter_status(
- Status::InvalidArgument("synthetic row filter failure"),
- TFileFormatType::FORMAT_PARQUET);
- EXPECT_EQ(status.to_string().find("nextBatch failed"), std::string::npos)
<< status;
-}
-
// 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.
@@ -900,4 +888,27 @@ TEST(FileScannerTest,
PartitionPruningStopsAtUnsafePredicate) {
EXPECT_EQ(partition_conjuncts[0], conjuncts[0]);
}
+TEST(FileScannerV2Test, ScannerOwnsUnsafeConjunctAndOrderedSuffixInProfile) {
+ const auto bool_type = std::make_shared<DataTypeUInt8>();
+ auto unsafe_predicate = std::make_shared<UnsafePartitionPredicate>();
+ unsafe_predicate->add_child(slot_ref(1, 0, bool_type, "part"));
+ VExprContextSPtrs conjuncts {
+ runtime_filter_context(slot_ref(1, 0, bool_type, "part"), 1),
+ runtime_filter_context(std::move(unsafe_predicate), 2),
+ runtime_filter_context(slot_ref(1, 0, bool_type, "part"), 3),
+ };
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("file_scanner_v2");
+ FileScannerV2 scanner(&state, &profile, nullptr);
+ scanner.TEST_set_scanner_conjuncts(std::move(conjuncts));
+
+ EXPECT_EQ(scanner.TEST_table_reader_owned_conjunct_count(), 1);
+ EXPECT_EQ(scanner.TEST_scanner_residual_conjunct_count(), 2);
+ const auto* residual_predicates =
profile.get_info_string("ScannerResidualPredicates");
+ ASSERT_NE(residual_predicates, nullptr);
+ EXPECT_FALSE(residual_predicates->empty());
+ EXPECT_NE(residual_predicates->find("SlotRef"), std::string::npos) <<
*residual_predicates;
+}
+
} // namespace doris
diff --git a/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
b/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
index 0d31b694951..b70f1cf9ea0 100644
--- a/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
+++ b/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
@@ -65,6 +65,23 @@ private:
std::list<Block> _blocks;
};
+class HighCostPredicate final : public VExpr {
+public:
+ HighCostPredicate() : VExpr(std::make_shared<DataTypeUInt8>(), false) {}
+
+ Status execute_column_impl(VExprContext*, const Block*, const Selector*,
size_t count,
+ ColumnPtr& result_column) const override {
+ result_column = ColumnUInt8::create(count, 1);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+ double execute_cost() const override { return 100.0; }
+
+private:
+ const std::string _expr_name = "high_cost_stateful_predicate";
+};
+
class ScannerLateArrivalRfTest : public RuntimeFilterTest {
public:
void SetUp() override {
@@ -83,6 +100,7 @@ public:
// the counter advances after RFs arrive and that the second call
short-circuits
// via the fast path at the top of the function.
TEST_F(ScannerLateArrivalRfTest, applied_rf_num_advances_after_late_arrival) {
+
_runtime_states[0]->_query_options.__set_enable_adjust_conjunct_order_by_cost(true);
std::vector<TRuntimeFilterDesc> rf_descs = {
TRuntimeFilterDescBuilder().add_planId_to_target_expr(0).build(),
TRuntimeFilterDescBuilder().add_planId_to_target_expr(0).build()};
@@ -104,26 +122,55 @@ TEST_F(ScannerLateArrivalRfTest,
applied_rf_num_advances_after_late_arrival) {
auto local_state =
std::make_shared<MockScanLocalState>(_runtime_states[0].get(), op.get());
+ auto initial_conjunct =
VExprContext::create_shared(std::make_shared<HighCostPredicate>());
+ ASSERT_TRUE(initial_conjunct->prepare(_runtime_states[0].get(),
row_desc).ok());
+ ASSERT_TRUE(initial_conjunct->open(_runtime_states[0].get()).ok());
+ local_state->_conjuncts.push_back(initial_conjunct);
+
std::vector<std::shared_ptr<Dependency>> rf_dependencies;
ASSERT_TRUE(local_state->_helper.init(_runtime_states[0].get(), true, 0,
0, rf_dependencies, "")
.ok());
+ std::shared_ptr<RuntimeFilterProducer> producer;
+ ASSERT_TRUE(RuntimeFilterProducer::create(_query_ctx.get(),
rf_descs.data(), &producer).ok());
+
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
+ local_state->_helper._consumers[0]->signal(producer.get());
+ ASSERT_TRUE(local_state->_helper
+ .acquire_runtime_filter(_runtime_states[0].get(),
local_state->_conjuncts,
+ row_desc)
+ .ok());
+ ASSERT_EQ(local_state->_conjuncts.size(), 2);
+
auto scanner = std::make_unique<TestScanner>(_runtime_states[0].get(),
local_state.get(),
-1 /*limit*/, &_profile);
- ASSERT_TRUE(scanner->init(_runtime_states[0].get(), {}).ok());
+ ASSERT_TRUE(scanner->init(_runtime_states[0].get(),
local_state->_conjuncts).ok());
+ auto second_scanner =
std::make_unique<TestScanner>(_runtime_states[0].get(), local_state.get(),
+ -1 /*limit*/,
&_profile);
+ ASSERT_TRUE(second_scanner->init(_runtime_states[0].get(),
local_state->_conjuncts).ok());
ASSERT_EQ(scanner->_total_rf_num, 2);
ASSERT_EQ(scanner->_applied_rf_num, 0);
- std::shared_ptr<RuntimeFilterProducer> producer;
- ASSERT_TRUE(RuntimeFilterProducer::create(_query_ctx.get(),
rf_descs.data(), &producer).ok());
-
producer->set_wrapper_state_and_ready_to_publish(RuntimeFilterWrapper::State::READY);
- local_state->_helper._consumers[0]->signal(producer.get());
local_state->_helper._consumers[1]->signal(producer.get());
// First call after both RFs arrived: counter must advance to total. Before
// the fix this stayed at 0 because the assignment was missing.
ASSERT_TRUE(scanner->try_append_late_arrival_runtime_filter().ok());
ASSERT_EQ(scanner->_applied_rf_num, 2);
+ ASSERT_EQ(scanner->_late_arrival_rf_conjuncts.size(), 1);
+ for (const auto& conjunct : scanner->_late_arrival_rf_conjuncts) {
+ EXPECT_NE(dynamic_cast<const
RuntimeFilterExpr*>(conjunct->root().get()), nullptr);
+ }
+ ASSERT_EQ(scanner->_conjuncts.size(), 3);
+ EXPECT_EQ(scanner->_conjuncts.back()->expr_name(),
"high_cost_stateful_predicate");
+
+ // The first scanner consumes the shared helper's expression, so another
scanner can only get
+ // the exact delta from the local state's append-only RF batch history.
+ ASSERT_TRUE(second_scanner->try_append_late_arrival_runtime_filter().ok());
+ ASSERT_EQ(second_scanner->_applied_rf_num, 2);
+ ASSERT_EQ(second_scanner->_late_arrival_rf_conjuncts.size(), 1);
+ EXPECT_NE(dynamic_cast<const RuntimeFilterExpr*>(
+
second_scanner->_late_arrival_rf_conjuncts[0]->root().get()),
+ nullptr);
// Second call: must hit the fast-path early return without re-cloning.
// We clear `_conjuncts` and verify the function does NOT repopulate them;
diff --git
a/be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp
b/be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp
index b453e7d32b1..e22060e4417 100644
---
a/be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp
+++
b/be/test/format/table/iceberg/iceberg_position_delete_sys_table_reader_test.cpp
@@ -35,10 +35,12 @@
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
+#include "exprs/vexpr.h"
#include "format/orc/orc_memory_stream_test.h"
#include "format/table/iceberg_scan_semantics.h"
#include "format/table/parquet_utils.h"
@@ -65,6 +67,31 @@ protected:
void _collect_profile_before_close() override { ++collect_calls; }
};
+class RejectAllRowsPredicate final : public VExpr {
+public:
+ RejectAllRowsPredicate() : VExpr(std::make_shared<DataTypeUInt8>(), false)
{}
+
+ Status execute_column_impl(VExprContext*, const Block*, const Selector*,
size_t count,
+ ColumnPtr& result_column) const override {
+ auto result = ColumnUInt8::create();
+ result->get_data().resize_fill(count, 0);
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _name; }
+ bool is_deterministic() const override { return false; }
+
+ Status clone_node(VExprSPtr* cloned_expr) const override {
+ DORIS_CHECK(cloned_expr != nullptr);
+ *cloned_expr = std::make_shared<RejectAllRowsPredicate>();
+ return Status::OK();
+ }
+
+private:
+ const std::string _name = "RejectAllRowsPredicate";
+};
+
SlotDescriptor* make_slot(ObjectPool* pool, int id, std::string name,
DataTypePtr type) {
TSlotDescriptor slot_desc;
slot_desc.__set_id(id);
@@ -1001,6 +1028,52 @@ TEST(IcebergPositionDeleteSysTableV2ReaderTest,
StopsBeforeExpandingDeletionVect
EXPECT_TRUE(eof);
}
+TEST(IcebergPositionDeleteSysTableV2ReaderTest,
+ AllFilteredDeletionVectorYieldsBeforeObservingCancellation) {
+ ObjectPool pool;
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("test_profile");
+ const auto nullable_int64 =
make_nullable(std::make_shared<DataTypeInt64>());
+ std::vector<SlotDescriptor*> file_slot_descs {
+ make_slot(&pool, 0, "pos", nullable_int64),
+ };
+
+ auto conjunct =
VExprContext::create_shared(std::make_shared<RejectAllRowsPredicate>());
+ RowDescriptor row_desc;
+ ASSERT_TRUE(conjunct->prepare(&state, row_desc).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+
+ format::iceberg::IcebergPositionDeleteSysTableV2Reader reader;
+ reader._runtime_state = &state;
+ reader._scanner_profile = &profile;
+ reader._io_ctx = std::make_shared<io::IOContext>();
+ reader._file_slot_descs = &file_slot_descs;
+ reader._projected_columns.resize(file_slot_descs.size());
+ reader._remaining_conjuncts = {conjunct};
+ reader._has_split = true;
+ reader._delete_file_kind =
+
format::iceberg::IcebergPositionDeleteSysTableV2Reader::DeleteFileKind::DELETION_VECTOR;
+ reader._batch_size = 1;
+ reader._dv_positions.add(uint64_t {7});
+ reader._dv_positions.add(uint64_t {9});
+ reader._dv_positions.add(uint64_t {11});
+ reader._next_dv_position.emplace(reader._dv_positions.begin());
+
+ Block block = make_output_block(file_slot_descs);
+ bool eof = false;
+ ASSERT_TRUE(reader.get_block(&block, &eof).ok());
+ EXPECT_FALSE(eof);
+ EXPECT_EQ(block.rows(), 0);
+ ASSERT_TRUE(reader._next_dv_position.has_value());
+ EXPECT_EQ(**reader._next_dv_position, 9);
+
+ reader._io_ctx->should_stop = true;
+ ASSERT_TRUE(reader.get_block(&block, &eof).ok());
+ EXPECT_TRUE(eof);
+ ASSERT_TRUE(reader._next_dv_position.has_value());
+ EXPECT_EQ(**reader._next_dv_position, 9);
+}
+
TEST(IcebergPositionDeleteSysTableV2ReaderTest,
ParquetRowUsesAnyFieldIdMapping) {
run_mixed_id_position_delete_test(format::FileFormat::PARQUET,
TFileFormatType::FORMAT_PARQUET,
"parquet");
diff --git a/be/test/format_v2/column_mapper_test.cpp
b/be/test/format_v2/column_mapper_test.cpp
index 4bb7b920231..6bfc06139e4 100644
--- a/be/test/format_v2/column_mapper_test.cpp
+++ b/be/test/format_v2/column_mapper_test.cpp
@@ -2315,6 +2315,39 @@ TEST(ColumnMapperLocalizeFiltersTest,
VisibleLocalFilterAddsPredicateColumnAndCo
EXPECT_TRUE(localized_slot->data_type()->equals(*int_type));
}
+TEST(ColumnMapperLocalizeFiltersTest, ReportsLocalizationForEachSplitMapping) {
+ const auto int_type = i32();
+ auto table_column = name_col("id", int_type);
+ const std::vector<ColumnDefinition> table_schema = {table_column};
+ TableFilter filter {
+ .conjunct = VExprContext::create_shared(int_gt(table_slot(0, 0,
int_type, "id"), 1)),
+ .global_indices = {GlobalIndex(0)}};
+
+ TableColumnMapper local_mapper({.mode = TableColumnMappingMode::BY_NAME});
+ ASSERT_TRUE(local_mapper.create_mapping(table_schema, {}, {name_col("id",
int_type, 7)}).ok());
+ FileScanRequest local_request;
+ FilterLocalizationResult local_result;
+ ASSERT_TRUE(local_mapper
+ .create_scan_request({filter}, table_schema,
&local_request, nullptr,
+ &local_result)
+ .ok());
+ ASSERT_EQ(local_result.localized_filters.size(), 1);
+ EXPECT_TRUE(local_result.localized_filters[0]);
+ ASSERT_EQ(local_request.conjuncts.size(), 1);
+
+ TableColumnMapper missing_mapper({.mode =
TableColumnMappingMode::BY_NAME});
+ ASSERT_TRUE(missing_mapper.create_mapping(table_schema, {}, {}).ok());
+ FileScanRequest missing_request;
+ FilterLocalizationResult missing_result;
+ ASSERT_TRUE(missing_mapper
+ .create_scan_request({filter}, table_schema,
&missing_request, nullptr,
+ &missing_result)
+ .ok());
+ ASSERT_EQ(missing_result.localized_filters.size(), 1);
+ EXPECT_FALSE(missing_result.localized_filters[0]);
+ EXPECT_TRUE(missing_request.conjuncts.empty());
+}
+
TEST(ColumnMapperLocalizeFiltersTest, VarbinaryFilterStaysAboveFileReader) {
const auto binary_type = varbinary();
const auto table_column = name_col("partition_key", binary_type);
@@ -2340,6 +2373,35 @@ TEST(ColumnMapperLocalizeFiltersTest,
VarbinaryFilterStaysAboveFileReader) {
EXPECT_TRUE(request.conjuncts.empty());
}
+TEST(ColumnMapperLocalizeFiltersTest,
VarcharWidthTruncationFilterStaysAboveFileReader) {
+ const auto table_type = std::make_shared<DataTypeString>(3, TYPE_VARCHAR);
+ const auto file_type = std::make_shared<DataTypeString>(10, TYPE_VARCHAR);
+ const auto table_column = name_col("value", table_type);
+ const auto file_column = name_col("value", file_type, 7);
+
+ TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_NAME});
+ ASSERT_TRUE(mapper.create_mapping({table_column}, {}, {file_column}).ok());
+
+ TableFilter filter {.conjunct =
VExprContext::create_shared(binary_predicate(
+ TExprOpcode::EQ, table_slot(0, 0, table_type,
"value"),
+ literal(table_type,
Field::create_field<TYPE_STRING>("abc")))),
+ .global_indices = {GlobalIndex(0)}};
+ TQueryOptions query_options;
+ query_options.__set_truncate_char_or_varchar_columns(true);
+ RuntimeState state {query_options, TQueryGlobals()};
+ FileScanRequest request;
+ FilterLocalizationResult localization_result;
+
+ ASSERT_TRUE(mapper.create_scan_request({filter}, {table_column}, &request,
&state,
+ &localization_result)
+ .ok());
+ ASSERT_EQ(localization_result.localized_filters.size(), 1);
+ EXPECT_FALSE(localization_result.localized_filters[0]);
+ EXPECT_TRUE(request.conjuncts.empty());
+ ASSERT_EQ(request.non_predicate_columns.size(), 1);
+ EXPECT_EQ(request.non_predicate_columns[0].column_id(), LocalColumnId(7));
+}
+
TEST(ColumnMapperLocalizeFiltersTest,
NestedVarbinaryFilterStaysAboveFileReader) {
const auto table_column = struct_name_col(
"payload", {name_col("id", i32()), name_col("binary_value",
varbinary())});
@@ -2490,7 +2552,7 @@ TEST(ColumnMapperScanRequestTest,
HiddenTopLevelFilterMappingUsesNameFallback) {
EXPECT_EQ(mapper.filter_entries().at(GlobalIndex(1)).local_index(),
LocalIndex(1));
}
-TEST(ColumnMapperScanRequestTest,
OrdinaryPredicateSlotRetainsPayloadForScannerBoundary) {
+TEST(ColumnMapperScanRequestTest, OrdinaryPredicateSlotRetainsOutputPayload) {
const auto int_type = i32();
auto quantity = name_col("ss_quantity", int_type);
auto tax = name_col("ss_ext_tax", int_type);
@@ -2515,8 +2577,8 @@ TEST(ColumnMapperScanRequestTest,
OrdinaryPredicateSlotRetainsPayloadForScannerB
EXPECT_EQ(request.predicate_columns[0].column_id(), LocalColumnId(0));
ASSERT_EQ(request.non_predicate_columns.size(), 1);
EXPECT_EQ(request.non_predicate_columns[0].column_id(), LocalColumnId(1));
- // The scanner evaluates its table-level conjuncts after TableReader
returns, so a visible
- // predicate slot cannot be replaced with a default-valued placeholder at
the file boundary.
+ // A visible predicate slot is still part of the table output and cannot
be replaced with a
+ // default-valued placeholder after file-local filtering.
EXPECT_TRUE(request.predicate_only_columns.empty());
}
diff --git a/be/test/format_v2/table/hudi_reader_test.cpp
b/be/test/format_v2/table/hudi_reader_test.cpp
index 6f9bea89ffe..b8dd97e57d5 100644
--- a/be/test/format_v2/table/hudi_reader_test.cpp
+++ b/be/test/format_v2/table/hudi_reader_test.cpp
@@ -37,6 +37,9 @@
#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/vexpr_context.h"
+#include "exprs/vliteral.h"
#include "format_v2/column_data.h"
#include "gen_cpp/ExternalTableSchema_types.h"
#include "gen_cpp/PlanNodes_types.h"
@@ -141,6 +144,54 @@ public:
}
};
+class AppendTrackingTableReader final : public TableReader {
+public:
+ Status append_conjuncts(const VExprContextSPtrs& conjuncts) override {
+ appended_conjuncts += conjuncts.size();
+ owned_conjuncts +=
_appended_table_reader_owned_conjunct_count.value_or(conjuncts.size());
+ return Status::OK();
+ }
+
+ size_t appended_conjuncts = 0;
+ size_t owned_conjuncts = 0;
+};
+
+class OneRowTableReader final : public TableReader {
+public:
+ Status prepare_split(const SplitReadOptions&) override { return
Status::OK(); }
+
+ Status get_block(Block* block, bool* eos) override {
+ auto column = ColumnInt32::create();
+ column->insert_value(1);
+ block->replace_by_position(0, std::move(column));
+ *eos = false;
+ return Status::OK();
+ }
+};
+
+class StatefulHybridPredicate final : public VExpr {
+public:
+ explicit StatefulHybridPredicate(std::vector<int>* observed_invocations)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _observed_invocations(observed_invocations) {}
+
+ Status execute_column_impl(VExprContext*, const Block*, const Selector*,
size_t count,
+ ColumnPtr& result_column) const override {
+ _observed_invocations->push_back(_invocation++);
+ result_column = ColumnUInt8::create(count, 1);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+ bool is_constant() const override { return false; }
+ bool is_deterministic() const override { return false; }
+
+private:
+ std::vector<int>* const _observed_invocations;
+ mutable int _invocation = 0;
+ const std::string _expr_name = "StatefulHybridPredicate";
+};
+
// Scenario: FileScannerV2 Hudi native reader uses the split schema id to
annotate the physical
// file schema before TableColumnMapper runs. This keeps schema-evolved Hudi
files on field-id
// mapping, including renamed nested children.
@@ -262,6 +313,102 @@ TEST(HudiHybridReaderTest,
AdaptiveBatchSizeReachesBothChildReaders) {
EXPECT_EQ(child_batch_sizes.second, 123);
}
+TEST(HudiHybridReaderTest, ReportsActiveChildMaterializedBlockStats) {
+ hudi::HudiHybridReader reader;
+ reader.TEST_install_batch_size_children();
+ reader._current_split_reader = reader._native_reader.get();
+ reader._native_reader->_last_materialized_block_stats = {
+ .has_materialized_input = true, .rows = 7, .bytes = 70,
.allocated_bytes = 96};
+
+ const auto& stats = reader.last_materialized_block_stats();
+ EXPECT_TRUE(stats.has_materialized_input);
+ EXPECT_EQ(stats.rows, 7);
+ EXPECT_EQ(stats.bytes, 70);
+ EXPECT_EQ(stats.allocated_bytes, 96);
+}
+
+TEST(HudiHybridReaderTest, LateConjunctReachesInitializedNativeAndJniChildren)
{
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ hudi::HudiHybridReader reader;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = {},
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+ auto native_reader = std::make_unique<AppendTrackingTableReader>();
+ auto jni_reader = std::make_unique<AppendTrackingTableReader>();
+ auto* native_reader_ptr = native_reader.get();
+ auto* jni_reader_ptr = jni_reader.get();
+ reader._native_reader = std::move(native_reader);
+ reader._jni_reader = std::move(jni_reader);
+
+ auto literal = VLiteral::create_shared(std::make_shared<DataTypeInt32>(),
+ Field::create_field<TYPE_INT>(1));
+ ASSERT_TRUE(reader.append_conjuncts_with_ownership(
+
{VExprContext::create_shared(std::move(literal))}, 0)
+ .ok());
+ EXPECT_EQ(native_reader_ptr->appended_conjuncts, 1);
+ EXPECT_EQ(jni_reader_ptr->appended_conjuncts, 1);
+ EXPECT_EQ(native_reader_ptr->owned_conjuncts, 0);
+ EXPECT_EQ(jni_reader_ptr->owned_conjuncts, 0);
+}
+
+TEST(HudiHybridReaderTest,
ScannerStatefulResidualSurvivesNativeJniNativeSwitch) {
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("hudi_scanner_stateful_residual");
+ TFileScanRangeParams scan_params;
+ scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET);
+ auto hybrid_reader = std::make_unique<hudi::HudiHybridReader>();
+ auto* hybrid_reader_ptr = hybrid_reader.get();
+ hybrid_reader_ptr->TEST_set_child_reader_factories(
+ [] { return std::make_unique<OneRowTableReader>(); },
+ [] { return std::make_unique<OneRowTableReader>(); });
+
+ std::vector<int> observed_invocations;
+ auto conjunct = VExprContext::create_shared(
+ std::make_shared<StatefulHybridPredicate>(&observed_invocations));
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor {}).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ FileScannerV2 scanner(&state, &profile, std::move(hybrid_reader));
+ scanner.TEST_set_scanner_conjuncts({std::move(conjunct)});
+
+ const std::vector<ColumnDefinition> projected_columns {
+ make_table_column(0, "id", std::make_shared<DataTypeInt32>()),
+ };
+ ASSERT_TRUE(hybrid_reader_ptr
+ ->init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ auto run_split = [&](FileFormat format, TFileFormatType::type
thrift_format) {
+ SplitReadOptions split;
+ split.current_split_format = format;
+ split.current_range.__set_format_type(thrift_format);
+ ASSERT_TRUE(hybrid_reader_ptr->prepare_split(split).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(hybrid_reader_ptr->get_block(&block, &eos).ok());
+ ASSERT_TRUE(scanner.TEST_filter_output_block(&block).ok());
+ };
+ run_split(FileFormat::PARQUET, TFileFormatType::FORMAT_PARQUET);
+ run_split(FileFormat::JNI, TFileFormatType::FORMAT_JNI);
+ run_split(FileFormat::PARQUET, TFileFormatType::FORMAT_PARQUET);
+
+ EXPECT_EQ(observed_invocations, std::vector<int>({0, 1, 2}));
+}
+
TEST(HudiHybridReaderTest,
NativeCountStarReportsMetadataRowsThroughHybridReader) {
const auto test_dir =
std::filesystem::temp_directory_path() /
"doris_hudi_hybrid_count_star_test";
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index 782463e335a..c42d68933f2 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -1146,11 +1146,6 @@ VExprContextSPtr prepared_conjunct(RuntimeState* state,
const VExprSPtr& expr) {
return ctx;
}
-void apply_final_conjuncts(Block* block, const VExprContextSPtrs& conjuncts) {
- const auto status = VExprContext::filter_block(conjuncts, block,
block->columns());
- ASSERT_TRUE(status.ok()) << status;
-}
-
TEST(IcebergV2ReaderTest, IcebergVirtualColumnsUseRowLineageMetadata) {
const auto test_dir =
std::filesystem::temp_directory_path() /
"doris_iceberg_virtual_columns_test";
@@ -1389,9 +1384,6 @@ TEST(IcebergV2ReaderTest,
IcebergRowIdPredicateFiltersAfterRowLineageMaterializa
bool eos = false;
ASSERT_TRUE(reader.get_block(&block, &eos).ok());
ASSERT_FALSE(eos);
- ASSERT_EQ(block.rows(), 3);
-
- apply_final_conjuncts(&block, conjuncts);
ASSERT_EQ(block.rows(), 1);
expect_nullable_int64_column_values(*block.get_by_position(0).column,
{1001});
expect_nullable_int64_column_values(*block.get_by_position(1).column,
{77});
@@ -1443,9 +1435,6 @@ TEST(IcebergV2ReaderTest,
IcebergLastUpdatedSequencePredicateFiltersAfterMateria
bool eos = false;
ASSERT_TRUE(reader.get_block(&block, &eos).ok());
ASSERT_FALSE(eos);
- ASSERT_EQ(block.rows(), 3);
-
- apply_final_conjuncts(&block, conjuncts);
ASSERT_EQ(block.rows(), 1);
expect_nullable_int64_column_values(*block.get_by_position(0).column,
{1001});
expect_nullable_int64_column_values(*block.get_by_position(1).column,
{77});
diff --git a/be/test/format_v2/table/paimon_reader_test.cpp
b/be/test/format_v2/table/paimon_reader_test.cpp
index 3c85eb4632f..2434d748dfa 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/be/test/format_v2/table/paimon_reader_test.cpp
@@ -45,6 +45,9 @@
#include "core/data_type/data_type_string.h"
#include "core/field.h"
#include "exec/common/endian.h"
+#include "exec/scan/file_scanner_v2.h"
+#include "exprs/vexpr_context.h"
+#include "exprs/vliteral.h"
#include "format/format_common.h"
#include "format/table/deletion_vector_reader.h"
#include "format/table/paimon_reader.h"
@@ -79,6 +82,54 @@ public:
}
};
+class AppendTrackingTableReader final : public TableReader {
+public:
+ Status append_conjuncts(const VExprContextSPtrs& conjuncts) override {
+ appended_conjuncts += conjuncts.size();
+ owned_conjuncts +=
_appended_table_reader_owned_conjunct_count.value_or(conjuncts.size());
+ return Status::OK();
+ }
+
+ size_t appended_conjuncts = 0;
+ size_t owned_conjuncts = 0;
+};
+
+class OneRowTableReader final : public TableReader {
+public:
+ Status prepare_split(const SplitReadOptions&) override { return
Status::OK(); }
+
+ Status get_block(Block* block, bool* eos) override {
+ auto column = ColumnInt32::create();
+ column->insert_value(1);
+ block->replace_by_position(0, std::move(column));
+ *eos = false;
+ return Status::OK();
+ }
+};
+
+class StatefulHybridPredicate final : public VExpr {
+public:
+ explicit StatefulHybridPredicate(std::vector<int>* observed_invocations)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _observed_invocations(observed_invocations) {}
+
+ Status execute_column_impl(VExprContext*, const Block*, const Selector*,
size_t count,
+ ColumnPtr& result_column) const override {
+ _observed_invocations->push_back(_invocation++);
+ result_column = ColumnUInt8::create(count, 1);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+ bool is_constant() const override { return false; }
+ bool is_deterministic() const override { return false; }
+
+private:
+ std::vector<int>* const _observed_invocations;
+ mutable int _invocation = 0;
+ const std::string _expr_name = "StatefulHybridPredicate";
+};
+
DataTypePtr table_type(const DataTypePtr& type) {
return type->is_nullable() ? type : make_nullable(type);
}
@@ -694,6 +745,101 @@ TEST(PaimonHybridReaderTest,
AdaptiveBatchSizeReachesBothChildReaders) {
EXPECT_EQ(child_batch_sizes.second, 321);
}
+TEST(PaimonHybridReaderTest, ReportsActiveChildMaterializedBlockStats) {
+ paimon::PaimonHybridReader reader;
+ reader.TEST_install_batch_size_children();
+ reader._current_split_reader = reader._native_reader.get();
+ reader._native_reader->_last_materialized_block_stats = {
+ .has_materialized_input = true, .rows = 7, .bytes = 70,
.allocated_bytes = 96};
+
+ const auto& stats = reader.last_materialized_block_stats();
+ EXPECT_TRUE(stats.has_materialized_input);
+ EXPECT_EQ(stats.rows, 7);
+ EXPECT_EQ(stats.bytes, 70);
+ EXPECT_EQ(stats.allocated_bytes, 96);
+}
+
+TEST(PaimonHybridReaderTest,
LateConjunctReachesInitializedNativeAndJniChildren) {
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ paimon::PaimonHybridReader reader;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = {},
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+ auto native_reader = std::make_unique<AppendTrackingTableReader>();
+ auto jni_reader = std::make_unique<AppendTrackingTableReader>();
+ auto* native_reader_ptr = native_reader.get();
+ auto* jni_reader_ptr = jni_reader.get();
+ reader._native_reader = std::move(native_reader);
+ reader._jni_reader = std::move(jni_reader);
+
+ auto literal = VLiteral::create_shared(std::make_shared<DataTypeInt32>(),
+ Field::create_field<TYPE_INT>(1));
+ ASSERT_TRUE(reader.append_conjuncts_with_ownership(
+
{VExprContext::create_shared(std::move(literal))}, 0)
+ .ok());
+ EXPECT_EQ(native_reader_ptr->appended_conjuncts, 1);
+ EXPECT_EQ(jni_reader_ptr->appended_conjuncts, 1);
+ EXPECT_EQ(native_reader_ptr->owned_conjuncts, 0);
+ EXPECT_EQ(jni_reader_ptr->owned_conjuncts, 0);
+}
+
+TEST(PaimonHybridReaderTest,
ScannerStatefulResidualSurvivesNativeJniNativeSwitch) {
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("paimon_scanner_stateful_residual");
+ auto scan_params = make_local_parquet_scan_params();
+ auto hybrid_reader = std::make_unique<paimon::PaimonHybridReader>();
+ auto* hybrid_reader_ptr = hybrid_reader.get();
+ hybrid_reader_ptr->TEST_set_child_reader_factories(
+ [] { return std::make_unique<OneRowTableReader>(); },
+ [] { return std::make_unique<OneRowTableReader>(); });
+
+ std::vector<int> observed_invocations;
+ auto conjunct = VExprContext::create_shared(
+ std::make_shared<StatefulHybridPredicate>(&observed_invocations));
+ ASSERT_TRUE(conjunct->prepare(&state, RowDescriptor {}).ok());
+ ASSERT_TRUE(conjunct->open(&state).ok());
+ FileScannerV2 scanner(&state, &profile, std::move(hybrid_reader));
+ scanner.TEST_set_scanner_conjuncts({std::move(conjunct)});
+
+ const std::vector<ColumnDefinition> projected_columns {
+ make_table_column(0, "id", std::make_shared<DataTypeInt32>()),
+ };
+ ASSERT_TRUE(hybrid_reader_ptr
+ ->init({
+ .projected_columns = projected_columns,
+ .conjuncts = {},
+ .format = FileFormat::PARQUET,
+ .scan_params = &scan_params,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ auto run_split = [&](FileFormat format, TFileRangeDesc range) {
+ SplitReadOptions split;
+ split.current_split_format = format;
+ split.current_range = std::move(range);
+ ASSERT_TRUE(hybrid_reader_ptr->prepare_split(split).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(hybrid_reader_ptr->get_block(&block, &eos).ok());
+ ASSERT_TRUE(scanner.TEST_filter_output_block(&block).ok());
+ };
+ run_split(FileFormat::PARQUET,
make_paimon_native_range(TFileFormatType::FORMAT_PARQUET));
+ run_split(FileFormat::JNI, make_paimon_jni_range());
+ run_split(FileFormat::PARQUET,
make_paimon_native_range(TFileFormatType::FORMAT_PARQUET));
+
+ EXPECT_EQ(observed_invocations, std::vector<int>({0, 1, 2}));
+}
+
TEST(PaimonHybridReaderTest,
NativeCountColumnReportsMetadataRowsThroughHybridReader) {
const auto test_dir =
std::filesystem::temp_directory_path() /
"doris_paimon_hybrid_count_column_test";
diff --git a/be/test/format_v2/table_reader_test.cpp
b/be/test/format_v2/table_reader_test.cpp
index 48b38d78e19..d92bbf2b0ac 100644
--- a/be/test/format_v2/table_reader_test.cpp
+++ b/be/test/format_v2/table_reader_test.cpp
@@ -297,6 +297,37 @@ private:
const std::string _expr_name = "NonDeterministicPartitionPredicate";
};
+class StatefulSequencePredicate final : public VExpr {
+public:
+ explicit StatefulSequencePredicate(std::vector<int>* observed_invocations)
+ : VExpr(std::make_shared<DataTypeUInt8>(), false),
+ _observed_invocations(observed_invocations) {}
+
+ Status execute_column_impl(VExprContext*, const Block*, const Selector*,
size_t count,
+ ColumnPtr& result_column) const override {
+ DORIS_CHECK(_observed_invocations != nullptr);
+ _observed_invocations->push_back(_invocation++);
+ auto result = ColumnUInt8::create();
+ result->get_data().resize_fill(count, 1);
+ result_column = std::move(result);
+ return Status::OK();
+ }
+
+ const std::string& expr_name() const override { return _expr_name; }
+ bool is_deterministic() const override { return false; }
+
+ Status clone_node(VExprSPtr* cloned_expr) const override {
+ DORIS_CHECK(cloned_expr != nullptr);
+ *cloned_expr =
std::make_shared<StatefulSequencePredicate>(_observed_invocations);
+ return Status::OK();
+ }
+
+private:
+ std::vector<int>* const _observed_invocations;
+ mutable int _invocation = 0;
+ const std::string _expr_name = "StatefulSequencePredicate";
+};
+
class NullableArrayBigintDefaultExpr final : public VExpr {
public:
explicit NullableArrayBigintDefaultExpr(DataTypePtr data_type)
@@ -534,6 +565,23 @@ void write_parquet_file(const std::string& file_path,
int32_t id, const std::str
builder.build()));
}
+void write_single_int_parquet_file(const std::string& file_path, const
std::string& column_name,
+ int32_t value) {
+ auto schema = arrow::schema({arrow::field(column_name, arrow::int32(),
false)});
+ auto table = arrow::Table::Make(schema, {build_int32_array({value})});
+
+ auto file_result = arrow::io::FileOutputStream::Open(file_path);
+ ASSERT_TRUE(file_result.ok()) << file_result.status();
+ std::shared_ptr<arrow::io::FileOutputStream> out = *file_result;
+
+ ::parquet::WriterProperties::Builder builder;
+ builder.version(::parquet::ParquetVersion::PARQUET_2_6);
+ builder.data_page_version(::parquet::ParquetDataPageVersion::V2);
+ builder.compression(::parquet::Compression::UNCOMPRESSED);
+ PARQUET_THROW_NOT_OK(::parquet::arrow::WriteTable(*table,
arrow::default_memory_pool(), out, 1,
+ builder.build()));
+}
+
void write_struct_parquet_file(const std::string& file_path, int32_t id) {
auto struct_type = arrow::struct_({arrow::field("id", arrow::int32(),
false)});
arrow::StructBuilder builder(
@@ -1055,6 +1103,8 @@ struct FakeFileReaderState {
bool stop_during_aggregate = false;
bool stop_during_read = false;
bool not_found_during_init = false;
+ int batch_count = 1;
+ int get_block_count = 0;
std::shared_ptr<FileScanRequest> last_request;
std::optional<FileAggregateRequest> last_aggregate_request;
std::shared_ptr<ConditionCacheContext> condition_cache_ctx;
@@ -1093,7 +1143,7 @@ public:
RETURN_IF_ERROR(FileReader::open(std::move(request)));
_state->last_request = _request;
++_state->open_count;
- _returned_batch = false;
+ _returned_batches = 0;
return Status::OK();
}
@@ -1102,7 +1152,8 @@ public:
DORIS_CHECK(rows != nullptr);
DORIS_CHECK(eof != nullptr);
DORIS_CHECK(_request != nullptr);
- if (_returned_batch) {
+ ++_state->get_block_count;
+ if (_returned_batches >= _state->batch_count) {
*rows = 0;
*eof = true;
return Status::OK();
@@ -1149,9 +1200,9 @@ public:
DORIS_CHECK(_state->io_ctx != nullptr);
_state->io_ctx->should_stop = true;
}
- _returned_batch = true;
+ ++_returned_batches;
*rows = 2;
- *eof = _state->eof_with_first_batch;
+ *eof = _state->eof_with_first_batch && _returned_batches >=
_state->batch_count;
if (_state->condition_cache_ctx != nullptr &&
!_state->condition_cache_ctx->is_hit &&
_state->condition_cache_ctx->filter_result != nullptr &&
!_state->condition_cache_ctx->filter_result->empty()) {
@@ -1207,7 +1258,7 @@ public:
private:
std::vector<ColumnDefinition> _schema;
std::shared_ptr<FakeFileReaderState> _state;
- bool _returned_batch = false;
+ int _returned_batches = 0;
};
class FakeTableReader final : public TableReader {
@@ -1381,13 +1432,54 @@ TEST(TableReaderTest,
ConstantPruningStopsAtUnsafePredicate) {
Block block = build_table_block(projected_columns);
bool eos = false;
ASSERT_TRUE(reader.get_block(&block, &eos).ok());
- EXPECT_FALSE(predicate_executed);
+ EXPECT_TRUE(predicate_executed);
EXPECT_FALSE(eos);
+ // The file was still opened, proving constant pruning did not jump over
the unsafe predicate;
+ // the predicate is evaluated only after the resulting table row is
materialized.
EXPECT_EQ(fake_state->open_count, 1);
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST(TableReaderTest, UnsafePredicateRunsAfterTableMaterialization) {
+ 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()};
+ bool predicate_executed = false;
+ auto unsafe_predicate =
+
std::make_shared<NonDeterministicPartitionPredicate>(&predicate_executed);
+ unsafe_predicate->add_child(table_int32_slot_ref(0, 0, "id"));
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {prepared_conjunct(&state,
unsafe_predicate)},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ SplitReadOptions split;
+ split.current_range.__set_path("fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_NE(fake_state->last_request, nullptr);
+ EXPECT_TRUE(fake_state->last_request->conjuncts.empty());
+ EXPECT_TRUE(predicate_executed);
ASSERT_TRUE(reader.close().ok());
}
-TEST(TableReaderTest, UnsafePredicateStaysOnScannerPath) {
+TEST(TableReaderTest,
ScannerOwnedUnsafePredicateIsPassedButNotExecutedByTableReader) {
std::vector<ColumnDefinition> file_schema;
file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
std::vector<ColumnDefinition> projected_columns;
@@ -1404,6 +1496,7 @@ TEST(TableReaderTest, UnsafePredicateStaysOnScannerPath) {
ASSERT_TRUE(reader.init({
.projected_columns = projected_columns,
.conjuncts = {prepared_conjunct(&state,
unsafe_predicate)},
+ .table_reader_owned_conjunct_count = 0,
.format = FileFormat::PARQUET,
.scan_params = nullptr,
.io_ctx = nullptr,
@@ -1418,9 +1511,178 @@ TEST(TableReaderTest,
UnsafePredicateStaysOnScannerPath) {
Block block = build_table_block(projected_columns);
bool eos = false;
ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+
ASSERT_NE(fake_state->last_request, nullptr);
EXPECT_TRUE(fake_state->last_request->conjuncts.empty());
+ EXPECT_EQ(reader.TEST_conjunct_count(), 1);
+ EXPECT_EQ(reader.TEST_table_reader_owned_conjunct_count(), 0);
EXPECT_FALSE(predicate_executed);
+ EXPECT_EQ(block.rows(), 2);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST(TableReaderTest, ResidualExpressionStateSurvivesAcrossSplits) {
+ 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()};
+ std::vector<int> observed_invocations;
+ auto stateful_predicate =
std::make_shared<StatefulSequencePredicate>(&observed_invocations);
+ stateful_predicate->add_child(table_int32_slot_ref(0, 0, "id"));
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {prepared_conjunct(&state,
stateful_predicate)},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ SplitReadOptions split;
+ split.current_range.__set_path("fake-table-reader-input");
+ for (int split_index = 0; split_index < 2; ++split_index) {
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_EQ(block.rows(), 2);
+ ASSERT_TRUE(reader.close().ok());
+ }
+
+ EXPECT_EQ(observed_invocations, std::vector<int>({0, 1}));
+}
+
+TEST(TableReaderTest, AllFilteredResidualReturnsAfterOneMaterializedBatch) {
+ 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()};
+ bool predicate_executed = false;
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ fake_state->batch_count = 2;
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {prepared_conjunct(
+ &state,
+
std::make_shared<NonDeterministicPartitionPredicate>(
+ &predicate_executed))},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ SplitReadOptions split;
+ split.current_range.__set_path("fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+
+ EXPECT_TRUE(predicate_executed);
+ EXPECT_EQ(block.rows(), 0);
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(fake_state->get_block_count, 1);
+ EXPECT_EQ(fake_state->close_count, 0);
+ EXPECT_TRUE(reader.last_materialized_block_stats().has_materialized_input);
+ EXPECT_EQ(reader.last_materialized_block_stats().rows, 2);
+ EXPECT_GT(reader.last_materialized_block_stats().bytes, 0);
+ EXPECT_GT(reader.last_materialized_block_stats().allocated_bytes, 0);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST(TableReaderTest, LateConjunctFiltersAlreadyOpenSplit) {
+ 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->batch_count = 2;
+ 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;
+ split.current_range.__set_path("fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_EQ(block.rows(), 2);
+
+ bool predicate_executed = false;
+ auto late_predicate =
std::make_shared<NonDeterministicPartitionPredicate>(&predicate_executed);
+ late_predicate->add_child(table_int32_slot_ref(0, 0, "id"));
+ ASSERT_TRUE(
+
reader.append_conjuncts({VExprContext::create_shared(std::move(late_predicate))}).ok());
+ block = build_table_block(projected_columns);
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(predicate_executed);
+ EXPECT_EQ(block.rows(), 0);
+ EXPECT_EQ(fake_state->get_block_count, 2);
+ ASSERT_TRUE(reader.close().ok());
+}
+
+TEST(TableReaderTest, ResidualFilteringHasDedicatedProfileTimer) {
+ 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()};
+ RuntimeProfile profile("scanner");
+ bool predicate_executed = false;
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ FakeTableReader reader(file_schema, fake_state);
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {prepared_conjunct(
+ &state,
+
std::make_shared<NonDeterministicPartitionPredicate>(
+ &predicate_executed))},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ })
+ .ok());
+
+ SplitReadOptions split;
+ split.current_range.__set_path("fake-table-reader-input");
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+
+ EXPECT_TRUE(predicate_executed);
+ ASSERT_NE(profile.get_counter("ResidualFilterTime"), nullptr);
+ EXPECT_GT(profile.get_counter("ResidualFilterTime")->value(), 0);
ASSERT_TRUE(reader.close().ok());
}
@@ -1471,9 +1733,11 @@ TEST(TableReaderTest,
ConstantPruningStopsAtUnsafeSlotlessPredicate) {
EXPECT_EQ(fake_state->open_count, 1);
ASSERT_NE(fake_state->last_request, nullptr);
// A slotless unsafe conjunct is an ordering barrier even though it has no
TableFilter entry.
- // The later predicate must stay on the scanner's row-level path instead
of running inside the
+ // The later predicate must stay on the post-materialization path instead
of running inside the
// file reader before the unsafe conjunct.
EXPECT_TRUE(fake_state->last_request->conjuncts.empty());
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
ASSERT_TRUE(reader.close().ok());
}
@@ -1756,8 +2020,13 @@ TEST(TableReaderTest,
SlotlessConjunctDisablesAggregatePushdown) {
// presence still prevents the fake aggregate count (3) from replacing the
two physical rows.
ASSERT_NE(fake_state->last_request, nullptr);
EXPECT_TRUE(fake_state->last_request->conjuncts.empty());
- EXPECT_EQ(block.rows(), 2);
+ // The two physical rows are then filtered at the table boundary, where
slotless predicates are
+ // evaluated exactly even though they cannot be localized to a file column.
+ EXPECT_EQ(block.rows(), 0);
+ EXPECT_FALSE(eos);
EXPECT_TRUE(predicate_executed);
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
ASSERT_TRUE(reader.close().ok());
}
@@ -4306,6 +4575,59 @@ TEST(TableReaderTest, VExprPredicateSurvivesReopenSplit)
{
std::filesystem::remove_all(test_dir);
}
+TEST(TableReaderTest, RecomputesPredicateExecutionLayerForEverySplit) {
+ const auto test_dir = std::filesystem::temp_directory_path() /
+ "doris_table_reader_split_local_predicate_test";
+ std::filesystem::remove_all(test_dir);
+ std::filesystem::create_directories(test_dir);
+
+ const auto local_file = (test_dir / "local.parquet").string();
+ const auto missing_file = (test_dir / "missing.parquet").string();
+ write_single_int_parquet_file(local_file, "id", 3);
+ write_single_int_parquet_file(missing_file, "other", 9);
+
+ 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()};
+ TableReader reader;
+ ASSERT_TRUE(reader.init({
+ .projected_columns = projected_columns,
+ .conjuncts = {prepared_conjunct(
+ &state,
table_int32_greater_than_expr(0, 0, 2))},
+ .format = FileFormat::PARQUET,
+ .scan_params = nullptr,
+ .io_ctx = nullptr,
+ .runtime_state = &state,
+ .scanner_profile = nullptr,
+ })
+ .ok());
+
+ ASSERT_TRUE(reader.prepare_split(build_split_options(local_file)).ok());
+ Block block = build_table_block(projected_columns);
+ bool eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_FALSE(eos);
+ expect_int32_column_values(*block.get_by_position(0).column, {3});
+ ASSERT_TRUE(reader.close().ok());
+
+ // The same predicate cannot be file-local when this split omits `id`. It
must be rebuilt as a
+ // table-level predicate over the materialized NULL instead of inheriting
the previous split's
+ // file-local ownership or escaping without exact evaluation.
+ ASSERT_TRUE(reader.prepare_split(build_split_options(missing_file)).ok());
+ block = build_table_block(projected_columns);
+ eos = false;
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_FALSE(eos);
+ EXPECT_EQ(block.rows(), 0);
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ EXPECT_TRUE(eos);
+
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(test_dir);
+}
+
TEST(TableReaderTest, CreateScanRequestDeduplicatesSharedPredicateColumns) {
const auto int_type = std::make_shared<DataTypeInt32>();
const std::vector<ColumnDefinition> projected_columns = {
diff --git a/be/test/storage/segment/adaptive_block_size_predictor_test.cpp
b/be/test/storage/segment/adaptive_block_size_predictor_test.cpp
index 60b6f37b8ce..64795dace19 100644
--- a/be/test/storage/segment/adaptive_block_size_predictor_test.cpp
+++ b/be/test/storage/segment/adaptive_block_size_predictor_test.cpp
@@ -89,6 +89,18 @@ TEST_F(AdaptiveBlockSizePredictorTest,
NoHistoryReturnsMaxRows) {
EXPECT_DOUBLE_EQ(pred.bytes_per_row_for_test(), expected_bpr);
}
+TEST_F(AdaptiveBlockSizePredictorTest,
ExplicitMaterializedSampleUsesPreFilterShape) {
+ AdaptiveBlockSizePredictor pred(kBlockBytes, 0.0);
+
+ // Callers that filter a block before returning it can still report the
rows and bytes that
+ // were actually materialized upstream.
+ pred.update(32, 32 * 4096);
+
+ EXPECT_TRUE(pred.has_history_for_test());
+ EXPECT_DOUBLE_EQ(pred.bytes_per_row_for_test(), 4096.0);
+ EXPECT_EQ(pred.predict_next_rows(), 2048);
+}
+
// ── Test 2: EWMA convergence
──────────────────────────────────────────────────
// When every update delivers the same sample, the EWMA stays exactly at that
// value (0.9*v + 0.1*v == v for any v).
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
