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 b759db26948 [fix](file) Revert split residual predicate ownership
(#65998)
b759db26948 is described below
commit b759db26948cebb22206ffa3a5e0a969616f2616
Author: Gabriel <[email protected]>
AuthorDate: Fri Jul 24 18:08:11 2026 +0800
[fix](file) Revert split residual predicate ownership (#65998)
---
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, 142 insertions(+), 1424 deletions(-)
diff --git a/be/src/exec/operator/scan_operator.cpp
b/be/src/exec/operator/scan_operator.cpp
index bc0c56dbfe7..3bb234d16b1 100644
--- a/be/src/exec/operator/scan_operator.cpp
+++ b/be/src/exec/operator/scan_operator.cpp
@@ -71,19 +71,13 @@ 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 applied_rf_num, int& arrived_rf_num,
- VExprContextSPtrs& arrived_conjuncts) {
+Status ScanLocalStateBase::update_late_arrival_runtime_filter(RuntimeState*
state,
+ int&
arrived_rf_num) {
// 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();
@@ -97,16 +91,6 @@ Status
ScanLocalStateBase::update_late_arrival_runtime_filter(
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 e955840f131..ae93e155852 100644
--- a/be/src/exec/operator/scan_operator.h
+++ b/be/src/exec/operator/scan_operator.h
@@ -21,8 +21,6 @@
#include <optional>
#include <set>
#include <string>
-#include <utility>
-#include <vector>
#include "common/status.h"
#include "common/thread_safety_annotations.h"
@@ -98,9 +96,7 @@ public:
uint64_t get_condition_cache_digest() const { return
_condition_cache_digest; }
- Status update_late_arrival_runtime_filter(RuntimeState* state, int
applied_rf_num,
- int& arrived_rf_num,
- VExprContextSPtrs&
arrived_conjuncts);
+ Status update_late_arrival_runtime_filter(RuntimeState* state, int&
arrived_rf_num);
Status clone_conjunct_ctxs(VExprContextSPtrs& scanner_conjuncts);
@@ -146,9 +142,6 @@ 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 a20946cf095..cdaa3d2f3b8 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -24,7 +24,6 @@
#include <map>
#include <memory>
#include <optional>
-#include <sstream>
#include <string>
#include <utility>
@@ -227,9 +226,7 @@ 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_profile(profile) {}
+ : Scanner(state, profile), _table_reader(std::move(table_reader)) {}
Status FileScannerV2::TEST_validate_scan_range(const TFileScanRangeParams&
params,
const TFileRangeDesc& range) {
@@ -329,9 +326,7 @@ 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;
@@ -365,11 +360,6 @@ 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>();
@@ -411,7 +401,6 @@ 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) {
@@ -461,33 +450,18 @@ Status FileScannerV2::_get_block_impl(RuntimeState*
state, Block* block, bool* e
}
Status FileScannerV2::_filter_output_block(Block* block) {
- 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_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();
+ return
_contextualize_output_filter_status(Scanner::_filter_output_block(block),
+ _get_current_format_type());
}
-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::_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.
+ status.prepend("Orc row reader nextBatch failed. reason = ");
+ }
+ return status;
}
Status FileScannerV2::_prepare_next_split(bool* eos) {
@@ -573,7 +547,6 @@ 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,
@@ -584,9 +557,6 @@ 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();
}
@@ -631,12 +601,15 @@ 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.
@@ -826,15 +799,10 @@ 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(source.size());
- for (const auto& conjunct : source) {
+ conjuncts->reserve(_conjuncts.size());
+ for (const auto& conjunct : _conjuncts) {
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));
@@ -843,68 +811,6 @@ Status FileScannerV2::_build_table_conjuncts(const
VExprContextSPtrs& source,
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);
}
@@ -1028,19 +934,17 @@ void FileScannerV2::_update_adaptive_batch_size(const
Block& block) {
if (!_should_run_adaptive_batch_size()) {
return;
}
- 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) {
+ COUNTER_SET(_adaptive_batch_actual_bytes_counter,
static_cast<int64_t>(block.bytes()));
+ if (block.rows() == 0) {
return;
}
- // 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.
+ // 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.
if (!_block_size_predictor->has_history()) {
COUNTER_UPDATE(_adaptive_batch_probe_count_counter, 1);
}
- _block_size_predictor->update(rows, bytes);
+ _block_size_predictor->update(block);
}
Status FileScannerV2::close(RuntimeState* state) {
@@ -1241,8 +1145,9 @@ 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) {
- // FileReader and TableReader both report their owned predicate rows
through the shared IO
- // context. Preserve scanner-level load statistics without
re-evaluating either predicate.
+ // 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().
_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 c68cfb1db8b..fc217506eea 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -88,22 +88,15 @@ 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,
@@ -122,8 +115,6 @@ 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;
@@ -142,6 +133,8 @@ 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;
@@ -153,12 +146,6 @@ 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();
@@ -194,10 +181,6 @@ 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
@@ -231,9 +214,6 @@ 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;
@@ -243,7 +223,6 @@ 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 138b3910d4f..3eaf5ee77eb 100644
--- a/be/src/exec/scan/scanner.cpp
+++ b/be/src/exec/scan/scanner.cpp
@@ -19,8 +19,6 @@
#include <glog/logging.h>
-#include <iterator>
-
#include "common/config.h"
#include "common/status.h"
#include "core/block/column_with_type_and_name.h"
@@ -162,11 +160,8 @@ Status Scanner::get_block(RuntimeState* state, Block*
block, bool* eof) {
DCHECK(block->rows() == 0);
break;
}
- // 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);
+ _num_rows_read += block->rows();
+ _num_byte_read += block->allocated_bytes();
}
// 2. Filter the output block finally.
@@ -261,9 +256,7 @@ Status Scanner::try_append_late_arrival_runtime_filter() {
}
DCHECK(_applied_rf_num < _total_rf_num);
int arrived_rf_num = 0;
- VExprContextSPtrs arrived_conjuncts;
- RETURN_IF_ERROR(_local_state->update_late_arrival_runtime_filter(
- _state, _applied_rf_num, arrived_rf_num, arrived_conjuncts));
+ RETURN_IF_ERROR(_local_state->update_late_arrival_runtime_filter(_state,
arrived_rf_num));
if (arrived_rf_num == _applied_rf_num) {
// No newly arrived runtime filters, just return;
@@ -273,9 +266,6 @@ 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 8f9753cf112..1bf4cdb5bb2 100644
--- a/be/src/exec/scan/scanner.h
+++ b/be/src/exec/scan/scanner.h
@@ -231,11 +231,6 @@ 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;
@@ -263,9 +258,6 @@ 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 ed0dcda9739..c3dbe3fa9e7 100644
--- a/be/src/format_v2/column_mapper.cpp
+++ b/be/src/format_v2/column_mapper.cpp
@@ -399,33 +399,6 @@ 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)
@@ -2188,7 +2161,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, FilterLocalizationResult*
localization_result) {
+ RuntimeState* runtime_state) {
// 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();
@@ -2224,8 +2197,7 @@ 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,
localization_result));
+ RETURN_IF_ERROR(localize_filters(table_filters, file_request,
runtime_state));
for (const auto& mapping : _hidden_mappings) {
if (!mapping.file_local_id.has_value()) {
continue;
@@ -2239,9 +2211,9 @@ Status TableColumnMapper::create_scan_request(
if (is_visible_output) {
continue;
}
- // 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.
+ // 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.
if (std::ranges::any_of(file_request->predicate_columns,
[local_id](const LocalColumnIndex& projection)
{
return projection.column_id() == local_id;
@@ -2290,11 +2262,7 @@ ColumnMapping*
TableColumnMapper::_find_filter_mapping(GlobalIndex global_index)
Status TableColumnMapper::localize_filters(const std::vector<TableFilter>&
table_filters,
FileScanRequest* file_request,
- RuntimeState* runtime_state,
- FilterLocalizationResult*
localization_result) {
- if (localization_result != nullptr) {
- localization_result->localized_filters.assign(table_filters.size(),
false);
- }
+ RuntimeState* runtime_state) {
std::set<LocalColumnId> localized_predicate_columns;
FilterProjectionMap filter_projections;
auto filter_mappings = _filter_visible_mappings();
@@ -2332,28 +2300,17 @@ 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 (size_t filter_index = 0; filter_index < table_filters.size();
++filter_index) {
- const auto& table_filter = table_filters[filter_index];
+ for (const auto& table_filter : table_filters) {
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 (!table_filter.can_localize || !predicate->is_deterministic() ||
+ if (!predicate->is_deterministic() ||
!table_filter_has_only_local_entries(table_filter,
_filter_entries)) {
continue;
}
- 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.
+ // 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.
RewriteContext rewrite_context {.runtime_state = runtime_state};
VExprSPtr rewrite_root;
Status clone_status;
@@ -2364,7 +2321,8 @@ 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 for TableReader after final table-schema
materialization.
+ // Leave them above TableReader, where Scanner evaluates the
original table-level
+ // conjunct after final materialization.
#ifndef NDEBUG
return Status::InternalError(
"Failed to clone table filter for file-local rewrite:
{}, expr={}",
@@ -2400,9 +2358,6 @@ 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 0860b687e4a..e03f836061e 100644
--- a/be/src/format_v2/column_mapper.h
+++ b/be/src/format_v2/column_mapper.h
@@ -40,13 +40,6 @@ 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,
@@ -172,8 +165,6 @@ 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);
@@ -205,8 +196,7 @@ 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,
- FilterLocalizationResult*
localization_result = nullptr);
+ RuntimeState* runtime_state = nullptr);
// Localize table-level filters to the file schema.
// Trivial mappings can copy structured predicates directly. Type changes
may be localized with
@@ -214,8 +204,7 @@ public:
// table-level finalize/filter fallback.
virtual Status localize_filters(const std::vector<TableFilter>&
table_filters,
FileScanRequest* file_request,
- RuntimeState* runtime_state = nullptr,
- FilterLocalizationResult*
localization_result = nullptr);
+ RuntimeState* runtime_state = 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 1f90957064b..5f959c3e672 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -76,9 +76,7 @@ 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. Readers
- // must enforce these exactly on returned rows; metadata pruning alone
does not transfer
- // predicate ownership away from TableReader.
+ // Row-level filters converted to file-local expressions from table-level
predicates.
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 271f14a7b48..ee24d0f9ad7 100644
--- a/be/src/format_v2/table/hudi_reader.cpp
+++ b/be/src/format_v2/table/hudi_reader.cpp
@@ -125,27 +125,6 @@ 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();
-}
-
int64_t HudiHybridReader::condition_cache_hit_count() const {
// Keep the wrapper count cumulative across native/JNI dispatch so
scanner-level delta
// accounting neither loses a child hit nor observes a counter reset on a
split switch.
@@ -197,7 +176,6 @@ 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 c13ac1215d7..dbb6f5e8231 100644
--- a/be/src/format_v2/table/hudi_reader.h
+++ b/be/src/format_v2/table/hudi_reader.h
@@ -65,8 +65,6 @@ 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;
int64_t condition_cache_hit_count() const override;
#ifdef BE_TEST
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 6ab67caa77d..76d39c72fb1 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
@@ -163,12 +163,6 @@ 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);
@@ -181,7 +175,6 @@ 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());
@@ -199,19 +192,9 @@ Status
IcebergPositionDeleteSysTableV2Reader::get_block(Block* block, bool* eos)
return Status::OK();
}
+ size_t read_rows = 0;
if (_delete_file_kind == DeleteFileKind::DELETION_VECTOR) {
- 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();
+ return _append_deletion_vector_block(block, &read_rows, eos);
}
DORIS_CHECK(_position_reader != nullptr);
@@ -225,18 +208,8 @@ 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 d815ef81c9b..5d8363848f3 100644
--- a/be/src/format_v2/table/paimon_reader.cpp
+++ b/be/src/format_v2/table/paimon_reader.cpp
@@ -152,27 +152,6 @@ 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();
-}
-
int64_t PaimonHybridReader::condition_cache_hit_count() const {
// Both children survive split switches, so the wrapper must publish their
cumulative totals;
// returning only the active child would make FileScannerV2's monotonic
delta go backwards.
@@ -227,7 +206,6 @@ 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 b4f076fc6e4..823fa6540d2 100644
--- a/be/src/format_v2/table/paimon_reader.h
+++ b/be/src/format_v2/table/paimon_reader.h
@@ -71,8 +71,6 @@ 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;
int64_t condition_cache_hit_count() const override;
#ifdef BE_TEST
diff --git a/be/src/format_v2/table_reader.cpp
b/be/src/format_v2/table_reader.cpp
index a073f01f237..4beaf8c9ff5 100644
--- a/be/src/format_v2/table_reader.cpp
+++ b/be/src/format_v2/table_reader.cpp
@@ -678,8 +678,6 @@ 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 =
@@ -723,9 +721,6 @@ 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) {
@@ -739,64 +734,7 @@ Status TableReader::init(TableReadOptions&& options) {
}
_system_properties = create_system_properties(_scan_params);
_mapper_options.mode = TableColumnMappingMode::BY_NAME;
- 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));
- }
- }
+ _conjuncts = std::move(options.conjuncts);
return Status::OK();
}
@@ -804,27 +742,20 @@ Status TableReader::_build_table_filters_from_conjuncts()
{
_table_filters.clear();
_constant_pruning_safe_filter_count = 0;
bool in_safe_prefix = true;
- for (size_t conjunct_index = 0; conjunct_index < _conjuncts.size();
++conjunct_index) {
- const auto& conjunct = _conjuncts[conjunct_index];
+ for (const auto& conjunct : _conjuncts) {
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. 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)) {
+ // 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)) {
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();
}
@@ -832,59 +763,6 @@ 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) {
@@ -1125,9 +1003,6 @@ 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()) {
@@ -1141,7 +1016,7 @@ Status TableReader::prepare_split(const SplitReadOptions&
options) {
_condition_cache_digest = _initial_condition_cache_digest;
}
if (options.conjuncts.has_value()) {
- RETURN_IF_ERROR(_replace_conjuncts(*options.conjuncts));
+ _conjuncts = *options.conjuncts;
}
// Update to current split format to handle ORC/PARQUET files in one table.
_format = options.current_split_format;
@@ -1211,7 +1086,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;
@@ -1247,7 +1122,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 48e3d996ce2..ea40280ee99 100644
--- a/be/src/format_v2/table_reader.h
+++ b/be/src/format_v2/table_reader.h
@@ -78,14 +78,10 @@ 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; otherwise TableReader evaluates them after final table-schema
materialization.
+// possible, and remain the source of row-level filtering after localization.
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 {
@@ -116,7 +112,6 @@ 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;
@@ -133,23 +128,12 @@ 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;
@@ -222,10 +206,6 @@ 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;
- }
void TEST_set_condition_cache_hit_count(int64_t hits) {
_condition_cache_hit_count = hits; }
bool TEST_current_data_file_is_immutable() const {
DORIS_CHECK(_current_task != nullptr);
@@ -247,25 +227,6 @@ 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.
@@ -285,7 +246,6 @@ public:
_remaining_file_level_count = -1;
_current_split_uses_metadata_count = false;
_current_split_pruned = false;
- _remaining_conjuncts.clear();
return Status::OK();
}
@@ -295,7 +255,6 @@ 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());
@@ -368,7 +327,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, ¤t_rows));
+ RETURN_IF_ERROR(finalize_chunk(block, current_rows));
#ifndef NDEBUG
RETURN_IF_ERROR(
_check_table_block_columns("after finalize_chunk", block,
current_rows));
@@ -377,13 +336,6 @@ 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();
}
}
@@ -401,7 +353,6 @@ public:
_remaining_table_level_count = -1;
_remaining_file_level_count = -1;
_current_split_uses_metadata_count = false;
- _remaining_conjuncts.clear();
return Status::OK();
}
@@ -487,17 +438,14 @@ 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,
- &localization_result));
+ _table_filters, _projected_columns, file_request.get(),
_runtime_state));
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
@@ -573,13 +521,9 @@ 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);
@@ -598,7 +542,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
@@ -815,7 +759,6 @@ 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();
@@ -831,28 +774,15 @@ 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, size_t* rows) {
- DORIS_CHECK(rows != nullptr);
+ Status finalize_chunk(Block* block, const size_t rows) {
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++;
}
@@ -860,12 +790,7 @@ 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));
- // 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);
+ return Status::OK();
}
// Materialize virtual columns in the table block, such as Iceberg _row_id
and
@@ -1029,7 +954,31 @@ 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) {
- return requires_char_or_varchar_truncation(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;
}
// Truncate a materialized CHAR/VARCHAR column in place by reusing the
vectorized substring
@@ -1126,8 +1075,9 @@ protected:
if (!_all_runtime_filters_applied_for_split) {
return false;
}
- // 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.
+ // 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.
if (!_conjuncts.empty()) {
return false;
}
@@ -1799,10 +1749,6 @@ 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 7a5ad573a2e..d8cc700f579 100644
--- a/be/src/storage/segment/adaptive_block_size_predictor.cpp
+++ b/be/src/storage/segment/adaptive_block_size_predictor.cpp
@@ -32,14 +32,11 @@
AdaptiveBlockSizePredictor::AdaptiveBlockSizePredictor(size_t preferred_block_si
_metadata_hint_bytes_per_row(metadata_hint_bytes_per_row) {}
void AdaptiveBlockSizePredictor::update(const Block& block) {
- update(block.rows(), block.bytes());
-}
-
-void AdaptiveBlockSizePredictor::update(size_t rows, size_t bytes) {
+ size_t rows = block.rows();
if (rows == 0) {
return;
}
- double cur = static_cast<double>(bytes) / static_cast<double>(rows);
+ double cur = static_cast<double>(block.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 f327fec8517..e03f18c2a53 100644
--- a/be/src/storage/segment/adaptive_block_size_predictor.h
+++ b/be/src/storage/segment/adaptive_block_size_predictor.h
@@ -60,7 +60,6 @@ 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 e1a2fd0cf88..87d4c92a776 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -749,6 +749,18 @@ 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.
@@ -888,27 +900,4 @@ 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 b70f1cf9ea0..0d31b694951 100644
--- a/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
+++ b/be/test/exec/scan/scanner_late_arrival_rf_test.cpp
@@ -65,23 +65,6 @@ 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 {
@@ -100,7 +83,6 @@ 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()};
@@ -122,55 +104,26 @@ 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(),
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_TRUE(scanner->init(_runtime_states[0].get(), {}).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 e22060e4417..b453e7d32b1 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,12 +35,10 @@
#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"
@@ -67,31 +65,6 @@ 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);
@@ -1028,52 +1001,6 @@ 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 d3428b6134a..29d4efbe17f 100644
--- a/be/test/format_v2/column_mapper_test.cpp
+++ b/be/test/format_v2/column_mapper_test.cpp
@@ -2337,39 +2337,6 @@ 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);
@@ -2395,35 +2362,6 @@ 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())});
@@ -2574,7 +2512,7 @@ TEST(ColumnMapperScanRequestTest,
HiddenTopLevelFilterMappingUsesNameFallback) {
EXPECT_EQ(mapper.filter_entries().at(GlobalIndex(1)).local_index(),
LocalIndex(1));
}
-TEST(ColumnMapperScanRequestTest, OrdinaryPredicateSlotRetainsOutputPayload) {
+TEST(ColumnMapperScanRequestTest,
OrdinaryPredicateSlotRetainsPayloadForScannerBoundary) {
const auto int_type = i32();
auto quantity = name_col("ss_quantity", int_type);
auto tax = name_col("ss_ext_tax", int_type);
@@ -2599,8 +2537,8 @@ TEST(ColumnMapperScanRequestTest,
OrdinaryPredicateSlotRetainsOutputPayload) {
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));
- // A visible predicate slot is still part of the table output and cannot
be replaced with a
- // default-valued placeholder after file-local filtering.
+ // 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.
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 f5bd70292e5..96126281744 100644
--- a/be/test/format_v2/table/hudi_reader_test.cpp
+++ b/be/test/format_v2/table/hudi_reader_test.cpp
@@ -37,9 +37,6 @@
#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"
@@ -144,54 +141,6 @@ 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.
@@ -313,102 +262,6 @@ 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, AggregatesConditionCacheHitsFromBothChildren) {
hudi::HudiHybridReader reader;
reader.TEST_install_batch_size_children();
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp
b/be/test/format_v2/table/iceberg_reader_test.cpp
index c42d68933f2..782463e335a 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -1146,6 +1146,11 @@ 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";
@@ -1384,6 +1389,9 @@ 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});
@@ -1435,6 +1443,9 @@ 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 77466439c6b..4186aa78f03 100644
--- a/be/test/format_v2/table/paimon_reader_test.cpp
+++ b/be/test/format_v2/table/paimon_reader_test.cpp
@@ -45,9 +45,6 @@
#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"
@@ -82,54 +79,6 @@ 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);
}
@@ -745,101 +694,6 @@ 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, AggregatesConditionCacheHitsFromBothChildren) {
paimon::PaimonHybridReader reader;
reader.TEST_install_batch_size_children();
diff --git a/be/test/format_v2/table_reader_test.cpp
b/be/test/format_v2/table_reader_test.cpp
index d92bbf2b0ac..48b38d78e19 100644
--- a/be/test/format_v2/table_reader_test.cpp
+++ b/be/test/format_v2/table_reader_test.cpp
@@ -297,37 +297,6 @@ 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)
@@ -565,23 +534,6 @@ 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(
@@ -1103,8 +1055,6 @@ 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;
@@ -1143,7 +1093,7 @@ public:
RETURN_IF_ERROR(FileReader::open(std::move(request)));
_state->last_request = _request;
++_state->open_count;
- _returned_batches = 0;
+ _returned_batch = false;
return Status::OK();
}
@@ -1152,8 +1102,7 @@ public:
DORIS_CHECK(rows != nullptr);
DORIS_CHECK(eof != nullptr);
DORIS_CHECK(_request != nullptr);
- ++_state->get_block_count;
- if (_returned_batches >= _state->batch_count) {
+ if (_returned_batch) {
*rows = 0;
*eof = true;
return Status::OK();
@@ -1200,9 +1149,9 @@ public:
DORIS_CHECK(_state->io_ctx != nullptr);
_state->io_ctx->should_stop = true;
}
- ++_returned_batches;
+ _returned_batch = true;
*rows = 2;
- *eof = _state->eof_with_first_batch && _returned_batches >=
_state->batch_count;
+ *eof = _state->eof_with_first_batch;
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()) {
@@ -1258,7 +1207,7 @@ public:
private:
std::vector<ColumnDefinition> _schema;
std::shared_ptr<FakeFileReaderState> _state;
- int _returned_batches = 0;
+ bool _returned_batch = false;
};
class FakeTableReader final : public TableReader {
@@ -1432,54 +1381,13 @@ TEST(TableReaderTest,
ConstantPruningStopsAtUnsafePredicate) {
Block block = build_table_block(projected_columns);
bool eos = false;
ASSERT_TRUE(reader.get_block(&block, &eos).ok());
- EXPECT_TRUE(predicate_executed);
+ EXPECT_FALSE(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,
ScannerOwnedUnsafePredicateIsPassedButNotExecutedByTableReader) {
+TEST(TableReaderTest, UnsafePredicateStaysOnScannerPath) {
std::vector<ColumnDefinition> file_schema;
file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
std::vector<ColumnDefinition> projected_columns;
@@ -1496,7 +1404,6 @@ TEST(TableReaderTest,
ScannerOwnedUnsafePredicateIsPassedButNotExecutedByTableRe
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,
@@ -1511,178 +1418,9 @@ TEST(TableReaderTest,
ScannerOwnedUnsafePredicateIsPassedButNotExecutedByTableRe
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());
}
@@ -1733,11 +1471,9 @@ 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 post-materialization path instead
of running inside the
+ // The later predicate must stay on the scanner's row-level 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());
}
@@ -2020,13 +1756,8 @@ 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());
- // 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_EQ(block.rows(), 2);
EXPECT_TRUE(predicate_executed);
- ASSERT_TRUE(reader.get_block(&block, &eos).ok());
- EXPECT_TRUE(eos);
ASSERT_TRUE(reader.close().ok());
}
@@ -4575,59 +4306,6 @@ 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 64795dace19..60b6f37b8ce 100644
--- a/be/test/storage/segment/adaptive_block_size_predictor_test.cpp
+++ b/be/test/storage/segment/adaptive_block_size_predictor_test.cpp
@@ -89,18 +89,6 @@ 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]
