Skip to content

[fix](be) Restore FileScannerV2 residual predicate ownership #66035

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Gabriel39 wants to merge 2 commits into from
Closed

[fix](be) Restore FileScannerV2 residual predicate ownership #66035

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3f899ba
Revert "[fix](file) Revert split residual predicate ownership (#65998)"
Gabriel39 Jul 25, 2026
3b25ddc
[fix](be) Restore FileScannerV2 residual predicate ownership
Gabriel39 Jul 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions be/src/exec/operator/scan_operator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,13 +71,19 @@ bool ScanLocalState<Derived>::should_run_serial() const {
return _parent->cast<typename Derived::Parent>()._should_run_serial;
}

Status ScanLocalStateBase::update_late_arrival_runtime_filter(RuntimeState* state,
int& arrived_rf_num) {
Status ScanLocalStateBase::update_late_arrival_runtime_filter(
RuntimeState* state, int applied_rf_num, int& arrived_rf_num,
VExprContextSPtrs& arrived_conjuncts) {
// Lock needed because _conjuncts can be accessed concurrently by multiple scanner threads
LockGuard lock(_conjuncts_lock);
arrived_conjuncts.clear();
size_t conjuncts_before = _conjuncts.size();
RETURN_IF_ERROR(_helper.try_append_late_arrival_runtime_filter(state, _parent->row_descriptor(),
arrived_rf_num, _conjuncts));
if (_conjuncts.size() > conjuncts_before) {
VExprContextSPtrs appended(_conjuncts.begin() + conjuncts_before, _conjuncts.end());
_late_arrival_conjunct_batches.emplace_back(arrived_rf_num, std::move(appended));
}
if (state->enable_adjust_conjunct_order_by_cost()) {
std::ranges::stable_sort(_conjuncts, [](const auto& a, const auto& b) {
return a->execute_cost() < b->execute_cost();
Expand All @@ -91,6 +97,16 @@ Status ScanLocalStateBase::update_late_arrival_runtime_filter(RuntimeState* stat
if (_conjuncts.size() > conjuncts_before) {
RETURN_IF_ERROR(_on_runtime_filter_update());
}
for (const auto& [batch_arrived_rf_num, batch] : _late_arrival_conjunct_batches) {
if (batch_arrived_rf_num <= applied_rf_num) {
continue;
}
for (const auto& conjunct : batch) {
VExprContextSPtr cloned;
RETURN_IF_ERROR(conjunct->clone(state, cloned));

@github-actions github-actions Bot Jul 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Avoid cloning RF deltas for scanners that cannot consume them

This runs under _conjuncts_lock and clones every unseen RF batch for every Scanner subclass. The base Scanner then clones the complete conjunct snapshot as well and stores these delta contexts in _late_arrival_rf_conjuncts, but only FileScannerV2 ever consumes or clears that vector; FileScanner V1, OlapScanner, JdbcScanner, and MetaScanner retain the opened clones until teardown. Please make delta delivery a FileScannerV2 capability/override (or otherwise optional) and avoid doing unused clone/open work under the shared lock.

arrived_conjuncts.push_back(std::move(cloned));
}
}
return Status::OK();
}

Expand Down
9 changes: 8 additions & 1 deletion be/src/exec/operator/scan_operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
#include <optional>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "common/status.h"
#include "common/thread_safety_annotations.h"
Expand Down Expand Up @@ -96,7 +98,9 @@ class ScanLocalStateBase : public PipelineXLocalState<> {

uint64_t get_condition_cache_digest() const { return _condition_cache_digest; }

Status update_late_arrival_runtime_filter(RuntimeState* state, int& arrived_rf_num);
Status update_late_arrival_runtime_filter(RuntimeState* state, int applied_rf_num,
int& arrived_rf_num,
VExprContextSPtrs& arrived_conjuncts);

Status clone_conjunct_ctxs(VExprContextSPtrs& scanner_conjuncts);

Expand Down Expand Up @@ -142,6 +146,9 @@ class ScanLocalStateBase : public PipelineXLocalState<> {

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
Expand Down
162 changes: 137 additions & 25 deletions be/src/exec/scan/file_scanner_v2.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <map>
#include <memory>
#include <optional>
#include <sstream>
#include <string>
#include <utility>

Expand Down Expand Up @@ -226,7 +227,9 @@ Status rewrite_slot_refs_to_global_index(
#ifdef BE_TEST
FileScannerV2::FileScannerV2(RuntimeState* state, RuntimeProfile* profile,
std::unique_ptr<format::TableReader> table_reader)
: Scanner(state, profile), _table_reader(std::move(table_reader)) {}
: Scanner(state, profile),
_table_reader(std::move(table_reader)),
_scanner_profile(profile) {}

Status FileScannerV2::TEST_validate_scan_range(const TFileScanRangeParams& params,
const TFileRangeDesc& range) {
Expand Down Expand Up @@ -326,7 +329,9 @@ FileScannerV2::FileScannerV2(RuntimeState* state, FileScanLocalState* local_stat

Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs& conjuncts) {
RETURN_IF_ERROR(Scanner::init(state, conjuncts));
_initialize_scanner_residual_conjuncts();
auto* profile = _local_state->scanner_profile();
_scanner_profile = profile;
const auto hierarchy = file_scan_profile::ensure_hierarchy(profile);
_scanner_total_timer = hierarchy.scanner;
_io_timer = hierarchy.io;
Expand Down Expand Up @@ -360,6 +365,11 @@ Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs& conjunc
profile, "AdaptiveBatchActualBytes", TUnit::BYTES, file_scan_profile::SCANNER, 1);
_adaptive_batch_probe_count_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "AdaptiveBatchProbeCount", TUnit::UNIT, file_scan_profile::SCANNER, 1);
_scanner_residual_filter_timer = ADD_CHILD_TIMER_WITH_LEVEL(
profile, "ScannerResidualFilterTime", file_scan_profile::SCANNER, 1);
_scanner_residual_rows_filtered_counter = ADD_CHILD_COUNTER_WITH_LEVEL(
profile, "ScannerResidualRowsFiltered", TUnit::UNIT, file_scan_profile::SCANNER, 1);
_refresh_scanner_residual_profile();
SCOPED_TIMER(_scanner_total_timer);
SCOPED_TIMER(_init_timer);
_file_cache_statistics = std::make_unique<io::FileCacheStatistics>();
Expand Down Expand Up @@ -401,6 +411,7 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state, Block* block, bool* e
SCOPED_TIMER(_get_block_timer);
while (true) {
RETURN_IF_CANCELLED(state);
RETURN_IF_ERROR(_sync_table_reader_conjuncts());
if (!_has_prepared_split) {
RETURN_IF_ERROR(_prepare_next_split(eof));
if (*eof) {
Expand Down Expand Up @@ -450,18 +461,33 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state, Block* block, bool* e
}

Status FileScannerV2::_filter_output_block(Block* block) {
return _contextualize_output_filter_status(Scanner::_filter_output_block(block),
_get_current_format_type());
}

Status FileScannerV2::_contextualize_output_filter_status(Status status,
TFileFormatType::type format_type) {
if (!status.ok() && format_type == TFileFormatType::FORMAT_ORC) {
// Error-preserving expressions cannot be reordered into the ORC reader and therefore run
// at the scanner boundary; keep their error context identical to ORC callback failures.
if (_scanner_residual_conjuncts.empty() || block->rows() == 0) {
return Status::OK();
}
SCOPED_TIMER(_scanner_residual_filter_timer);
const size_t rows_before_filter = block->rows();
auto status = VExprContext::filter_block(_scanner_residual_conjuncts, block, block->columns());
if (!status.ok() && _params != nullptr &&
_get_current_format_type() == TFileFormatType::FORMAT_ORC) {
status.prepend("Orc row reader nextBatch failed. reason = ");
}
return status;
RETURN_IF_ERROR(status);
const int64_t filtered_rows = cast_set<int64_t>(rows_before_filter - block->rows());
_counter.num_rows_unselected += filtered_rows;
if (_scanner_residual_rows_filtered_counter != nullptr) {
COUNTER_UPDATE(_scanner_residual_rows_filtered_counter, filtered_rows);
}
return Status::OK();
}

size_t FileScannerV2::_last_block_rows_read(const Block& block) const {

@github-actions github-actions Bot Jul 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve the Scanner progress bound for JNI batches

This fallback is useful only when the reader returns once per materialized batch. JniTableReader::get_block() neither records MaterializedBlockStats nor returns when a Java batch is fully filtered; it keeps reading until a survivor or EOF. Therefore direct JNI and Hudi/Paimon JNI splits can drain an entire reject-all split in one Scanner task, while adaptive sizing learns from survivors instead of the pre-filter Java block. Please reset/record JNI materialization stats before predicate filtering and return an empty progress batch after one rejected Java batch, with a JNI coverage case.

const auto& stats = _table_reader->last_materialized_block_stats();
return stats.has_materialized_input ? stats.rows : block.rows();
}

size_t FileScannerV2::_last_block_bytes_read(const Block& block) const {
const auto& stats = _table_reader->last_materialized_block_stats();
return stats.has_materialized_input ? stats.allocated_bytes : block.allocated_bytes();
}

Status FileScannerV2::_prepare_next_split(bool* eos) {
Expand Down Expand Up @@ -527,6 +553,21 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
RETURN_IF_ERROR(_to_file_format(format_type, &file_format));
DORIS_CHECK(_table_reader != nullptr);

if (!_late_arrival_rf_conjuncts.empty()) {
const size_t owned_count = _scanner_residual_conjuncts.empty()
? _safe_conjunct_prefix_size(_late_arrival_rf_conjuncts)
: 0;
_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());
_append_ordered_conjuncts.insert(_append_ordered_conjuncts.end(),
_late_arrival_rf_conjuncts.begin(),
_late_arrival_rf_conjuncts.end());
_late_arrival_rf_conjuncts.clear();
_refresh_scanner_residual_profile();
}
VExprContextSPtrs table_conjuncts;
RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts));
std::optional<std::vector<format::GlobalIndex>> push_down_count_columns;
Expand All @@ -547,6 +588,7 @@ Status FileScannerV2::_init_table_reader(const TFileRangeDesc& range) {
RETURN_IF_ERROR(_table_reader->init({
.projected_columns = _projected_columns,
.conjuncts = std::move(table_conjuncts),
.table_reader_owned_conjunct_count = _table_reader_owned_conjunct_count,
.format = file_format,
.scan_params = const_cast<TFileScanRangeParams*>(_params),
.io_ctx = _io_ctx,
Expand All @@ -557,6 +599,7 @@ 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;
return Status::OK();
}

Expand Down Expand Up @@ -601,15 +644,12 @@ Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range,
std::map<std::string, Field> partition_values) {
format::FileFormat current_split_format;
RETURN_IF_ERROR(_to_file_format(get_range_format_type(*_params, range), &current_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.
Expand Down Expand Up @@ -799,10 +839,15 @@ format::ColumnDefinition FileScannerV2::_build_table_column(const SlotDescriptor
}

Status FileScannerV2::_build_table_conjuncts(VExprContextSPtrs* conjuncts) const {
return _build_table_conjuncts(_append_ordered_conjuncts, conjuncts);
}

Status FileScannerV2::_build_table_conjuncts(const VExprContextSPtrs& source,
VExprContextSPtrs* conjuncts) const {
DORIS_CHECK(conjuncts != nullptr);
conjuncts->clear();
conjuncts->reserve(_conjuncts.size());
for (const auto& conjunct : _conjuncts) {
conjuncts->reserve(source.size());
for (const auto& conjunct : source) {
VExprSPtr root;
RETURN_IF_ERROR(format::clone_table_expr_tree(conjunct->root(), &root));
RETURN_IF_ERROR(rewrite_slot_refs_to_global_index(&root, _slot_id_to_global_index));
Expand All @@ -811,6 +856,72 @@ Status FileScannerV2::_build_table_conjuncts(VExprContextSPtrs* conjuncts) const
return Status::OK();
}

size_t FileScannerV2::_safe_conjunct_prefix_size(const VExprContextSPtrs& conjuncts) {
for (size_t conjunct_index = 0; conjunct_index < conjuncts.size(); ++conjunct_index) {
if (!format::TableReader::is_safe_to_pre_execute(conjuncts[conjunct_index])) {
return conjunct_index;
}
}
return conjuncts.size();
}

void FileScannerV2::_initialize_scanner_residual_conjuncts() {
_append_ordered_conjuncts = _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));
_append_ordered_conjuncts.insert(_append_ordered_conjuncts.end(),
_late_arrival_rf_conjuncts.begin(),
_late_arrival_rf_conjuncts.end());
_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);
}
Expand Down Expand Up @@ -934,17 +1045,19 @@ void FileScannerV2::_update_adaptive_batch_size(const Block& block) {
if (!_should_run_adaptive_batch_size()) {
return;
}
COUNTER_SET(_adaptive_batch_actual_bytes_counter, static_cast<int64_t>(block.bytes()));
if (block.rows() == 0) {
const auto& stats = _table_reader->last_materialized_block_stats();
const size_t rows = stats.has_materialized_input ? stats.rows : block.rows();
const size_t bytes = stats.has_materialized_input ? stats.bytes : block.bytes();
COUNTER_SET(_adaptive_batch_actual_bytes_counter, static_cast<int64_t>(bytes));
if (rows == 0) {
return;
}
// The sample is taken after TableReader has finalized file-local columns to table columns.
// This matches the memory shape seen by upstream operators and catches very wide nested
// columns, such as map/string payloads, after the first probe batch.
// Residual predicates run after wide table columns are materialized. Learn from that pre-filter
// shape so selective predicates cannot make the next reader batch dangerously large.
if (!_block_size_predictor->has_history()) {
COUNTER_UPDATE(_adaptive_batch_probe_count_counter, 1);
}
_block_size_predictor->update(block);
_block_size_predictor->update(rows, bytes);
}

Status FileScannerV2::close(RuntimeState* state) {
Expand Down Expand Up @@ -1145,9 +1258,8 @@ void FileScannerV2::_report_file_reader_predicate_filtered_rows() {
const int64_t filtered_rows = _io_ctx != nullptr ? _io_ctx->predicate_filtered_rows : 0;
const int64_t filtered_delta = filtered_rows - _reported_predicate_filtered_rows;
if (filtered_delta > 0) {
// File readers can evaluate localized conjuncts before a block reaches Scanner. Count
// those rows as scanner-level unselected rows so load statistics stay identical no matter
// whether a predicate is pushed down or evaluated by Scanner::_filter_output_block().
// FileReader and TableReader both report their owned predicate rows through the shared IO
// context. Preserve scanner-level load statistics without re-evaluating either predicate.
_counter.num_rows_unselected += filtered_delta;
_reported_predicate_filtered_rows = filtered_rows;
}
Expand Down
Loading
Loading