[Enhancement](parquet)update runtime filter when read next parquet row group.

hubgeter · hubgeter · commit b0d62d68ec8b · 2025-12-15T22:26:00.000+08:00
diff --git a/be/src/runtime_filter/runtime_filter_consumer_helper.h b/be/src/runtime_filter/runtime_filter_consumer_helper.h
@@ -52,6 +52,9 @@ class RuntimeFilterConsumerHelper {
     // parent_operator_profile is owned by LocalState so update it is safe at here.
     void collect_realtime_profile(RuntimeProfile* parent_operator_profile);
 
+    size_t rf_num() {
+        return _runtime_filter_descs.size();
+    }
 private:
     // Append late-arrival runtime filters to the vconjunct_ctx.
     Status _append_rf_into_conjuncts(RuntimeState* state,
diff --git a/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
@@ -318,7 +318,7 @@ Status RowGroupReader::next_batch(Block* block, size_t batch_size, size_t* read_
 
         RETURN_IF_ERROR(_fill_row_id_columns(block, *read_rows, modify_row_ids));
 
-        Status st = VExprContext::filter_block(_lazy_read_ctx.conjuncts, block, block->columns());
+        Status st = VExprContext::filter_block(*_lazy_read_ctx.conjuncts, block, block->columns());
         *read_rows = block->rows();
         return st;
     }
@@ -349,7 +349,7 @@ Status RowGroupReader::next_batch(Block* block, size_t batch_size, size_t* read_
             for (uint32_t i = 0; i < column_to_keep; ++i) {
                 columns_to_filter[i] = i;
             }
-            if (!_lazy_read_ctx.conjuncts.empty()) {
+            if (!_lazy_read_ctx.conjuncts->empty()) {
                 std::vector<IColumn::Filter*> filters;
                 if (_position_delete_ctx.has_filter) {
                     filters.push_back(_pos_delete_filter_ptr.get());
diff --git a/be/src/vec/exec/format/parquet/vparquet_group_reader.h b/be/src/vec/exec/format/parquet/vparquet_group_reader.h
@@ -79,7 +79,13 @@ class RowGroupReader : public ProfileCollector {
 
     // table name
     struct LazyReadContext {
-        VExprContextSPtrs conjuncts;
+        const VExprContextSPtrs* conjuncts;
+
+        //ParquetReader::set_fill_columns
+        const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>> *
+                fill_partition_columns;
+        const std::unordered_map<std::string, VExprContextSPtr>* fill_missing_columns;
+
         bool can_lazy_read = false;
         // block->rows() returns the number of rows of the first column,
         // so we should check and resize the first column
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -382,7 +382,7 @@ Status ParquetReader::init_reader(
         }
     }
     // build column predicates for column lazy read
-    _lazy_read_ctx.conjuncts = conjuncts;
+    _lazy_read_ctx.conjuncts = &conjuncts;
     return Status::OK();
 }
 
@@ -408,11 +408,18 @@ bool ParquetReader::_type_matches(const VSlotRef* slot_ref) const {
            !is_complex_type(table_col_type->get_primitive_type());
 }
 
-Status ParquetReader::set_fill_columns(
-        const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&
-                partition_columns,
-        const std::unordered_map<std::string, VExprContextSPtr>& missing_columns) {
-    SCOPED_RAW_TIMER(&_statistics.parse_meta_time);
+
+Status ParquetReader::_update_lazy_read_ctx() {
+    RowGroupReader::LazyReadContext new_lazy_read_ctx;
+    new_lazy_read_ctx.conjuncts = _lazy_read_ctx.conjuncts;
+    new_lazy_read_ctx.fill_partition_columns = _lazy_read_ctx.fill_partition_columns;
+    new_lazy_read_ctx.fill_missing_columns = _lazy_read_ctx.fill_missing_columns;
+    _lazy_read_ctx = std::move(new_lazy_read_ctx);
+
+    _top_runtime_vexprs.clear();
+    _push_down_predicates.clear();
+    _useless_predicates.clear();
+
     // std::unordered_map<column_name, std::pair<col_id, slot_id>>
     std::unordered_map<std::string, std::pair<uint32_t, int>> predicate_columns;
     // visit_slot for lazy mat.
@@ -432,7 +439,7 @@ Status ParquetReader::set_fill_columns(
         }
     };
 
-    for (const auto& conjunct : _lazy_read_ctx.conjuncts) {
+    for (const auto& conjunct : *_lazy_read_ctx.conjuncts) {
         auto expr = conjunct->root();
 
         if (expr->is_rf_wrapper()) {
@@ -454,8 +461,8 @@ Status ParquetReader::set_fill_columns(
 
                 int max_in_size =
                         _state->query_options().__isset.max_pushdown_conditions_per_column
-                                ? _state->query_options().max_pushdown_conditions_per_column
-                                : 1024;
+                        ? _state->query_options().max_pushdown_conditions_per_column
+                        : 1024;
                 if (direct_in_predicate->get_set_func()->size() == 0 ||
                     direct_in_predicate->get_set_func()->size() > max_in_size) {
                     continue;
@@ -519,7 +526,7 @@ Status ParquetReader::set_fill_columns(
         _lazy_read_ctx.all_predicate_col_ids.emplace_back(_row_id_column_iterator_pair.second);
     }
 
-    for (auto& kv : partition_columns) {
+    for (auto& kv : *_lazy_read_ctx.fill_partition_columns) {
         auto iter = predicate_columns.find(kv.first);
         if (iter == predicate_columns.end()) {
             _lazy_read_ctx.partition_columns.emplace(kv.first, kv.second);
@@ -529,7 +536,7 @@ Status ParquetReader::set_fill_columns(
         }
     }
 
-    for (auto& kv : missing_columns) {
+    for (auto& kv : *_lazy_read_ctx.fill_missing_columns) {
         auto iter = predicate_columns.find(kv.first);
         if (iter == predicate_columns.end()) {
             _lazy_read_ctx.missing_columns.emplace(kv.first, kv.second);
@@ -561,6 +568,19 @@ Status ParquetReader::set_fill_columns(
         }
     }
 
+    return Status::OK();
+}
+
+Status ParquetReader::set_fill_columns(
+        const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&
+                partition_columns,
+        const std::unordered_map<std::string, VExprContextSPtr>& missing_columns) {
+//    SCOPED_RAW_TIMER(&_statistics.parse_meta_time);
+
+    _lazy_read_ctx.fill_partition_columns = &partition_columns;
+    _lazy_read_ctx.fill_missing_columns = &missing_columns;
+    RETURN_IF_ERROR(_update_lazy_read_ctx());
+
     if (_filter_groups && (_total_groups == 0 || _t_metadata->num_rows == 0 || _range_size < 0)) {
         return Status::EndOfFile("No row group to read");
     }
@@ -698,6 +718,12 @@ Status ParquetReader::_next_row_group_reader() {
             continue;
         }
 
+
+        bool has_late_rf_cond = false;
+        RETURN_IF_ERROR(_call_late_rf_func(&has_late_rf_cond));
+        if (has_late_rf_cond) {
+            RETURN_IF_ERROR(_update_lazy_read_ctx());
+        }
         size_t before_predicate_size = _push_down_predicates.size();
         _push_down_predicates.reserve(before_predicate_size + _top_runtime_vexprs.size());
         for (const auto& vexpr : _top_runtime_vexprs) {
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.h b/be/src/vec/exec/format/parquet/vparquet_reader.h
@@ -165,6 +165,17 @@ class ParquetReader : public GenericReader, public ExprPushDownHelper {
 
     bool count_read_rows() override { return true; }
 
+    void set_update_late_rf_func(std::function<Status(bool*)>&& func) {
+        _call_late_rf_func = std::move(func);
+    }
+
+    std::function<Status(bool*)> _call_late_rf_func = [](bool* changed) {
+        *changed = false;
+        return Status::OK();
+    };
+
+    Status _update_lazy_read_ctx();
+
 protected:
     void _collect_profile_before_close() override;
 
@@ -345,6 +356,7 @@ class ParquetReader : public GenericReader, public ExprPushDownHelper {
     std::vector<std::unique_ptr<MutilColumnBlockPredicate>> _push_down_predicates;
     std::vector<std::shared_ptr<ColumnPredicate>> _useless_predicates;
     Arena _arena;
+
 };
 #include "common/compile_check_end.h"
 
diff --git a/be/src/vec/exec/scan/file_scanner.cpp b/be/src/vec/exec/scan/file_scanner.cpp
@@ -357,8 +357,10 @@ Status FileScanner::_process_conjuncts() {
     return Status::OK();
 }
 
-Status FileScanner::_process_late_arrival_conjuncts() {
+Status FileScanner::_process_late_arrival_conjuncts(bool* changed) {
+    *changed = false;
     if (_push_down_conjuncts.size() < _conjuncts.size()) {
+        *changed = true;
         _push_down_conjuncts.clear();
         _push_down_conjuncts.resize(_conjuncts.size());
         for (size_t i = 0; i != _conjuncts.size(); ++i) {
@@ -1052,8 +1054,17 @@ Status FileScanner::_get_next_reader() {
             // see IcebergTableReader::init_row_filters for example.
             parquet_reader->set_push_down_agg_type(_get_push_down_agg_type());
             if (push_down_predicates) {
-                RETURN_IF_ERROR(_process_late_arrival_conjuncts());
+                bool changed = false;
+                RETURN_IF_ERROR(_process_late_arrival_conjuncts(&changed));
             }
+            std::function<Status(bool*)> update_late_rf = [&](bool* changed) -> Status {
+                if (!_is_load) {
+                    RETURN_IF_ERROR(try_append_late_arrival_runtime_filter());
+                    RETURN_IF_ERROR(_process_late_arrival_conjuncts(changed));
+                }
+                return Status::OK();
+            };
+            parquet_reader->set_update_late_rf_func(std::move(update_late_rf));
             RETURN_IF_ERROR(_init_parquet_reader(std::move(parquet_reader), file_meta_cache_ptr));
 
             need_to_get_parsed_schema = true;
@@ -1074,7 +1085,8 @@ Status FileScanner::_get_next_reader() {
 
             orc_reader->set_push_down_agg_type(_get_push_down_agg_type());
             if (push_down_predicates) {
-                RETURN_IF_ERROR(_process_late_arrival_conjuncts());
+                bool changed = false;
+                RETURN_IF_ERROR(_process_late_arrival_conjuncts(&changed));
             }
             RETURN_IF_ERROR(_init_orc_reader(std::move(orc_reader), file_meta_cache_ptr));
 
diff --git a/be/src/vec/exec/scan/file_scanner.h b/be/src/vec/exec/scan/file_scanner.h
@@ -100,6 +100,9 @@ class FileScanner : public Scanner {
 
     void update_realtime_counters() override;
 
+
+
+
 protected:
     Status _get_block_impl(RuntimeState* state, Block* block, bool* eof) override;
 
@@ -251,7 +254,7 @@ class FileScanner : public Scanner {
     void _init_runtime_filter_partition_prune_block();
     Status _process_runtime_filters_partition_prune(bool& is_partition_pruned);
     Status _process_conjuncts();
-    Status _process_late_arrival_conjuncts();
+    Status _process_late_arrival_conjuncts(bool* changed);
     void _get_slot_ids(VExpr* expr, std::vector<int>* slot_ids);
     Status _generate_truncate_columns(bool need_to_get_parsed_schema);
     Status _set_fill_or_truncate_columns(bool need_to_get_parsed_schema);
diff --git a/be/src/vec/exec/scan/scanner.cpp b/be/src/vec/exec/scan/scanner.cpp
@@ -41,6 +41,7 @@ Scanner::Scanner(RuntimeState* state, pipeline::ScanLocalStateBase* local_state,
           _output_tuple_desc(_local_state->output_tuple_desc()),
           _output_row_descriptor(_local_state->_parent->output_row_descriptor()),
           _has_prepared(false) {
+    _total_rf_num = cast_set<int>(_local_state->_helper.rf_num());
     DorisMetrics::instance()->scanner_cnt->increment(1);
 }
 

Original file line number	Diff line number	Diff line change
`@@ -41,6 +41,7 @@ Scanner::Scanner(RuntimeState* state, pipeline::ScanLocalStateBase* local_state,`
`41`	`41`	`_output_tuple_desc(_local_state->output_tuple_desc()),`
`42`	`42`	`_output_row_descriptor(_local_state->_parent->output_row_descriptor()),`
`43`	`43`	`_has_prepared(false) {`
	`44`	`+ _total_rf_num = cast_set<int>(_local_state->_helper.rf_num());`
`44`	`45`	`DorisMetrics::instance()->scanner_cnt->increment(1);`
`45`	`46`	`}`
`46`	`47`