[enhance](runtime filter) impl partition pruning in runtime filer

be/src/vec/exec/format/empty_block_reader.h

@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "vec/exec/format/generic_reader.h"
+
+namespace doris::vectorized {
+
+class Block;
+
+// A reader that always returns empty blocks.
+class EmptyBlockReader final : public GenericReader {
+    ENABLE_FACTORY_CREATOR(EmptyBlockReader);
+
+public:
+    EmptyBlockReader() = default;
+    ~EmptyBlockReader() final = default;
+
+    Status get_next_block(Block* block, size_t* read_rows, bool* eof) override {
+        *read_rows = 0;
+        *eof = true;
+        return Status::OK();
+    }
+};
+
+} // namespace doris::vectorized

be/src/vec/exec/scan/vfile_scanner.cpp

@@ -27,11 +27,13 @@
 #include <iterator>
 #include <map>
 #include <tuple>
+#include <unordered_map>
 #include <utility>
 
 #include "common/compiler_util.h" // IWYU pragma: keep
 #include "common/config.h"
 #include "common/logging.h"
+#include "common/status.h"
 #include "io/cache/block_file_cache_profile.h"
 #include "runtime/descriptors.h"
 #include "runtime/runtime_state.h"
@@ -52,6 +54,7 @@
 #include "vec/exec/format/arrow/arrow_stream_reader.h"
 #include "vec/exec/format/avro/avro_jni_reader.h"
 #include "vec/exec/format/csv/csv_reader.h"
+#include "vec/exec/format/empty_block_reader.h"
 #include "vec/exec/format/json/new_json_reader.h"
 #include "vec/exec/format/orc/vorc_reader.h"
 #include "vec/exec/format/parquet/vparquet_reader.h"
@@ -174,6 +177,78 @@ Status VFileScanner::prepare(RuntimeState* state, const VExprContextSPtrs& conju
     return Status::OK();
 }
 
+Status VFileScanner::_process_runtime_filters_partition_pruning(bool& can_filter_all) {
+    VExprContextSPtrs ctxs;
+
+    for (auto& conjunct : _push_down_conjuncts) {
+        auto impl = conjunct->root()->get_impl();
+        if (impl) {
+            // If impl is not null, which means this a conjuncts from runtime filter.
+            auto* runtime_filter = typeid_cast<VRuntimeFilterWrapper*>(impl.get());
+            ctxs.emplace_back(std::make_shared<VExprContext>(runtime_filter->get_impl()));
+        }
+    }
+
+    size_t partition_value_column_size = 0;
+    // 1. Get partition key values to string columns.
+    std::unordered_map<SlotId, MutableColumnPtr> parititon_slot_id_to_column;
+    for (auto const& partition_col_desc : _partition_col_descs) {
+        auto partiton_data = std::get<0>(partition_col_desc.second);
+        const auto* partiton_slot_desc = std::get<1>(partition_col_desc.second);
+        auto partition_value_column = ColumnString::create();
+        partition_value_column->insert_data(partiton_data.c_str(), partiton_data.size());
+        parititon_slot_id_to_column[partiton_slot_desc->id()] = std::move(partition_value_column);
+        partition_value_column_size = partition_value_column->size();
+    }
+
+    // 2. Build a temp block from the partition column, then execute conjuncts and filter block.
+    // 2.1 Build a temp block from the partition column to match the conjuncts executing.
+    Block temp_block;
+    int index = 0;
+    bool first_cloumn_filled = false;
+    for (auto const* slot_desc : _real_tuple_desc->slots()) {
+        if (!slot_desc->need_materialize()) {
+            // should be ignored from reading
+            continue;
+        }
+        if (parititon_slot_id_to_column.find(slot_desc->id()) !=
+            parititon_slot_id_to_column.end()) {
+            auto data_type = slot_desc->get_data_type_ptr();
+            auto partition_value_column = std::move(parititon_slot_id_to_column[slot_desc->id()]);
+            if (data_type->is_nullable()) {
+                temp_block.insert(
+                        {ColumnNullable::create(
+                                 std::move(partition_value_column),
+                                 ColumnUInt8::create(partition_value_column_size, 0)),
+                         std::make_shared<DataTypeNullable>(std::make_shared<DataTypeString>()),
+                         ""});
+            } else {
+                temp_block.insert({std::move(partition_value_column),
+                                   std::make_shared<DataTypeString>(), ""});
+            }
+            if (index == 0) {
+                first_cloumn_filled = true;
+            }
+        } else {
+            temp_block.insert(ColumnWithTypeAndName(slot_desc->get_empty_mutable_column(),
+                                                    slot_desc->get_data_type_ptr(),
+                                                    slot_desc->col_name()));
+        }
+        index++;
+    }
+
+    // 2.2 Execute conjuncts.
+    if (!first_cloumn_filled) {
+        // VExprContext.execute has an optimization, the filtering is executed when block->rows() > 0
+        // The following process may be tricky and time-consuming, but we have no other way.
+        temp_block.get_by_position(0).column->assume_mutable()->resize(partition_value_column_size);
+    }
+    IColumn::Filter result_filter(temp_block.rows(), 1);
+    RETURN_IF_ERROR(VExprContext::execute_conjuncts(ctxs, nullptr, &temp_block, &result_filter,
+                                                    &can_filter_all));
+    return Status::OK();
+}
+
 Status VFileScanner::_process_conjuncts_for_dict_filter() {
     _slot_id_to_filter_conjuncts.clear();
     _not_single_slot_filter_conjuncts.clear();
@@ -752,6 +827,16 @@ Status VFileScanner::_get_next_reader() {
         const TFileRangeDesc& range = _current_range;
         _current_range_path = range.path;
 
+        // runtime filter partition pruning
+        // so we need get partition columns first
+        RETURN_IF_ERROR(_generate_parititon_columns());
+        bool can_filter_all = false;
+        RETURN_IF_ERROR(_process_runtime_filters_partition_pruning(can_filter_all));
+        if (can_filter_all) {
+            _cur_reader = EmptyBlockReader::create_unique();
+            return Status::OK();
+        }
+
         // create reader for specific format
         Status init_status;
         // for compatibility, if format_type is not set in range, use the format type of params
@@ -858,7 +943,7 @@ Status VFileScanner::_get_next_reader() {
                         &_slot_id_to_filter_conjuncts);
                 std::unique_ptr<PaimonParquetReader> paimon_reader =
                         PaimonParquetReader::create_unique(std::move(parquet_reader), _profile,
-                                                           *_params);
+                                                           _state, *_params, range);
                 RETURN_IF_ERROR(paimon_reader->init_row_filters(range, _io_ctx.get()));
                 _cur_reader = std::move(paimon_reader);
             } else {
@@ -920,8 +1005,8 @@ Status VFileScanner::_get_next_reader() {
                         &_file_col_names, _colname_to_value_range, _push_down_conjuncts, false,
                         _real_tuple_desc, _default_val_row_desc.get(),
                         &_not_single_slot_filter_conjuncts, &_slot_id_to_filter_conjuncts);
-                std::unique_ptr<PaimonOrcReader> paimon_reader =
-                        PaimonOrcReader::create_unique(std::move(orc_reader), _profile, *_params);
+                std::unique_ptr<PaimonOrcReader> paimon_reader = PaimonOrcReader::create_unique(
+                        std::move(orc_reader), _profile, _state, *_params, range);
                 RETURN_IF_ERROR(paimon_reader->init_row_filters(range, _io_ctx.get()));
                 _cur_reader = std::move(paimon_reader);
             } else {
@@ -1012,7 +1097,8 @@ Status VFileScanner::_get_next_reader() {
         _missing_cols.clear();
         RETURN_IF_ERROR(_cur_reader->get_columns(&_name_to_col_type, &_missing_cols));
         _cur_reader->set_push_down_agg_type(_get_push_down_agg_type());
-        RETURN_IF_ERROR(_generate_fill_columns());
+        RETURN_IF_ERROR(_generate_missing_columns());
+        RETURN_IF_ERROR(_cur_reader->set_fill_columns(_partition_col_descs, _missing_col_descs));
         if (VLOG_NOTICE_IS_ON && !_missing_cols.empty() && _is_load) {
             fmt::memory_buffer col_buf;
             for (auto& col : _missing_cols) {
@@ -1042,10 +1128,8 @@ Status VFileScanner::_get_next_reader() {
     return Status::OK();
 }
 
-Status VFileScanner::_generate_fill_columns() {
+Status VFileScanner::_generate_parititon_columns() {
     _partition_col_descs.clear();
-    _missing_col_descs.clear();
-
     const TFileRangeDesc& range = _current_range;
     if (range.__isset.columns_from_path && !_partition_slot_descs.empty()) {
         for (const auto& slot_desc : _partition_slot_descs) {
@@ -1066,7 +1150,11 @@ Status VFileScanner::_generate_fill_columns() {
             }
         }
     }
+    return Status::OK();
+}
 
+Status VFileScanner::_generate_missing_columns() {
+    _missing_col_descs.clear();
     if (!_missing_cols.empty()) {
         for (auto slot_desc : _real_tuple_desc->slots()) {
             if (!slot_desc->is_materialized()) {
@@ -1084,8 +1172,7 @@ Status VFileScanner::_generate_fill_columns() {
             _missing_col_descs.emplace(slot_desc->col_name(), it->second);
         }
     }
-
-    return _cur_reader->set_fill_columns(_partition_col_descs, _missing_col_descs);
+    return Status::OK();
 }
 
 Status VFileScanner::_init_expr_ctxes() {

be/src/vec/exec/scan/vfile_scanner.h

@@ -213,7 +213,9 @@ class VFileScanner : public VScanner {
     Status _convert_to_output_block(Block* block);
     Status _truncate_char_or_varchar_columns(Block* block);
     void _truncate_char_or_varchar_column(Block* block, int idx, int len);
-    Status _generate_fill_columns();
+    Status _generate_parititon_columns();
+    Status _generate_missing_columns();
+    Status _process_runtime_filters_partition_pruning(bool& is_partition_pruning);
     Status _process_conjuncts_for_dict_filter();
     Status _process_late_arrival_conjuncts();
     void _get_slot_ids(VExpr* expr, std::vector<int>* slot_ids);