[fix](scan) Backport FileScannerV2 predicate ownership fixes

be/src/exec/operator/scan_operator.cpp

@@ -73,17 +73,34 @@ 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();
         });
     };
+    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();
 }
 

be/src/exec/operator/scan_operator.h

@@ -21,6 +21,8 @@
 #include <optional>
 #include <set>
 #include <string>
+#include <utility>
+#include <vector>
 
 #include "common/status.h"
 #include "common/thread_safety_annotations.h"
@@ -91,7 +93,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);
 
@@ -130,6 +134,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;
 

be/src/exec/scan/file_scanner_v2.cpp

@@ -359,6 +359,7 @@ FileScannerV2::FileScannerV2(RuntimeState* state, FileScanLocalState* local_stat
 
 Status FileScannerV2::init(RuntimeState* state, const VExprContextSPtrs& conjuncts) {
     RETURN_IF_ERROR(Scanner::init(state, conjuncts));
+    _transfer_conjuncts_to_table_reader();
     auto* profile = _local_state->scanner_profile();
     const auto hierarchy = file_scan_profile::ensure_hierarchy(profile);
     _scanner_total_timer = hierarchy.scanner;
@@ -434,6 +435,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) {
@@ -483,18 +485,22 @@ 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());
+    // TableReader is the single owner of predicates that cannot be localized, while FileReader is
+    // the exact owner of localized predicates. Re-evaluating either set here would duplicate
+    // mutable expression state and force predicate-only columns to be materialized for Scanner.
+    DORIS_CHECK(_conjuncts.empty());
+    (void)block;
+    return Status::OK();
 }
 
-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;
+size_t FileScannerV2::_last_block_rows_read(const Block& block) const {
+    const auto& stats = _table_reader->last_materialized_block_stats();
+    return stats.has_materialized_input ? stats.rows : block.rows();
+}
+
+size_t FileScannerV2::_last_block_bytes_read(const Block& block) const {
+    const auto& stats = _table_reader->last_materialized_block_stats();
+    return stats.has_materialized_input ? stats.allocated_bytes : block.allocated_bytes();
 }
 
 Status FileScannerV2::_prepare_next_split(bool* eos) {
@@ -560,6 +566,14 @@ 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()) {
+        _table_reader_owned_conjunct_count += _late_arrival_rf_conjuncts.size();
+        _append_ordered_conjuncts.insert(_append_ordered_conjuncts.end(),
+                                         _late_arrival_rf_conjuncts.begin(),
+                                         _late_arrival_rf_conjuncts.end());
+        _late_arrival_rf_conjuncts.clear();
+        _conjuncts.clear();
+    }
     VExprContextSPtrs table_conjuncts;
     RETURN_IF_ERROR(_build_table_conjuncts(&table_conjuncts));
     std::optional<std::vector<format::GlobalIndex>> push_down_count_columns;
@@ -580,6 +594,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,
@@ -590,6 +605,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();
 }
 
@@ -639,20 +655,17 @@ 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.
             .all_runtime_filters_applied = _applied_rf_num == _total_rf_num,
-            .condition_cache_digest = _current_condition_cache_digest(),
+            .condition_cache_digest = _current_table_condition_cache_digest(),
             .cache = _kv_cache,
             .current_range = range,
             .current_split_format = current_split_format,
@@ -776,6 +789,11 @@ Status FileScannerV2::_build_projected_columns(const format::TableReader& table_
                                          slot_info.slot_id);
         }
         auto column = _build_table_column(it->second);
+        // An old FE does not send post-filter liveness, so preserve the historical behavior and
+        // materialize the value. A new FE may mark it dead, but ColumnMapper still requires exact
+        // localization for the current split before making it predicate-only.
+        column.value_required_after_filter = !slot_info.__isset.value_required_after_filter ||
+                                             slot_info.value_required_after_filter;
         build_context.slot_desc = it->second;
         if (column.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
             _need_global_rowid_column = true;
@@ -834,10 +852,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));
@@ -847,6 +870,43 @@ Status FileScannerV2::_build_table_conjuncts(VExprContextSPtrs* conjuncts) const
     return Status::OK();
 }
 
+void FileScannerV2::_transfer_conjuncts_to_table_reader() {
+    _append_ordered_conjuncts = std::move(_conjuncts);
+    _conjuncts.clear();
+    _table_reader_owned_conjunct_count = _append_ordered_conjuncts.size();
+}
+
+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 = appended.size();
+    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;
+    _late_arrival_rf_conjuncts.clear();
+    _conjuncts.clear();
+    _table_reader_applied_rf_num = _applied_rf_num;
+    return Status::OK();
+}
+
+uint64_t FileScannerV2::_current_table_condition_cache_digest() const {
+    DORIS_CHECK(_state != nullptr);
+    DORIS_CHECK(_local_state != nullptr);
+    if (_local_state->get_condition_cache_digest() == 0) {
+        return 0;
+    }
+    return _build_condition_cache_digest(_state->query_options().condition_cache_digest,
+                                         _append_ordered_conjuncts);
+}
+
 TFileFormatType::type FileScannerV2::_get_current_format_type() const {
     return get_range_format_type(*_params, _current_range);
 }
@@ -974,17 +1034,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) {
@@ -1176,9 +1238,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;
     }

be/src/exec/scan/file_scanner_v2.h

@@ -89,15 +89,20 @@ class FileScannerV2 final : public Scanner {
             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);
+        _transfer_conjuncts_to_table_reader();
+    }
+    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 _conjuncts.size(); }
 #endif
 
     FileScannerV2(RuntimeState* state, FileScanLocalState* parent, int64_t limit,
@@ -116,6 +121,9 @@ class FileScannerV2 final : public Scanner {
 protected:
     Status _get_block_impl(RuntimeState* state, Block* block, bool* eof) override;
     Status _filter_output_block(Block* block) override;
+    bool _retains_output_conjuncts() const override { return false; }
+    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;
 
@@ -134,8 +142,6 @@ class FileScannerV2 final : public Scanner {
                                        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;
@@ -147,6 +153,11 @@ class FileScannerV2 final : public Scanner {
     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();
+    void _transfer_conjuncts_to_table_reader();
+    uint64_t _current_table_condition_cache_digest() const;
     static Status _to_file_format(TFileFormatType::type format_type,
                                   format::FileFormat* file_format);
     void _reset_adaptive_batch_size_state();
@@ -182,6 +193,10 @@ class FileScannerV2 final : public Scanner {
     std::string _current_range_path;
 
     std::unique_ptr<format::TableReader> _table_reader;
+    size_t _table_reader_owned_conjunct_count = 0;
+    // Preserve append order for partition pruning and split-local condition-cache digests even
+    // though Scanner no longer owns an executable predicate list.
+    VExprContextSPtrs _append_ordered_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
@@ -224,6 +239,7 @@ class FileScannerV2 final : public Scanner {
     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