diff --git a/be/benchmark/benchmark_bit_pack.hpp b/be/benchmark/benchmark_bit_pack.hpp index a4f269dd4710cd..78ba2ebcd978d6 100644 --- a/be/benchmark/benchmark_bit_pack.hpp +++ b/be/benchmark/benchmark_bit_pack.hpp @@ -50,8 +50,10 @@ void bit_pack(const T* input, uint8_t in_num, int bit_width, uint8_t* output) { } static void BM_BitPack(benchmark::State& state) { - int w = state.range(0); - int n = 255; + // Registrations cap these dimensions to the destination types; explicit casts keep the + // benchmark build independent of the compiler's implicit-conversion warning policy. + const auto w = static_cast(state.range(0)); + constexpr uint8_t n = 255; std::default_random_engine e; std::uniform_int_distribution u; @@ -75,8 +77,8 @@ static void BM_BitPack(benchmark::State& state) { } static void BM_BitPackOptimized(benchmark::State& state) { - int w = state.range(0); - int n = 255; + const auto w = static_cast(state.range(0)); + constexpr uint8_t n = 255; std::default_random_engine e; std::uniform_int_distribution u; diff --git a/be/benchmark/benchmark_column_array_view.hpp b/be/benchmark/benchmark_column_array_view.hpp index 09baf2bd435030..4ba5fc522a72ad 100644 --- a/be/benchmark/benchmark_column_array_view.hpp +++ b/be/benchmark/benchmark_column_array_view.hpp @@ -126,7 +126,7 @@ static ColumnPtr make_string_array_column() { // Wrap with outer Nullable (no rows are actually null, just the wrapper overhead). static ColumnPtr wrap_nullable(const ColumnPtr& col) { - return ColumnNullable::create(col->assume_mutable(), + return ColumnNullable::create(col->assert_mutable(), ColumnUInt8::create(col->size(), 0)); } diff --git a/be/benchmark/benchmark_column_array_view_distance.hpp b/be/benchmark/benchmark_column_array_view_distance.hpp index 34fd287f2030ff..567a1b84d5d70d 100644 --- a/be/benchmark/benchmark_column_array_view_distance.hpp +++ b/be/benchmark/benchmark_column_array_view_distance.hpp @@ -264,7 +264,7 @@ BENCHMARK(ArrayView_Distance_Const_Plain_Flat)->Unit(benchmark::kNanosecond); // ============================================================ static ColumnPtr wrap_nullable_for_dist(const ColumnPtr& col) { - return ColumnNullable::create(col->assume_mutable(), ColumnUInt8::create(col->size(), 0)); + return ColumnNullable::create(col->assert_mutable(), ColumnUInt8::create(col->size(), 0)); } static void Handwritten_Distance_Nullable_Plain(benchmark::State& state) { diff --git a/be/benchmark/benchmark_fastunion.hpp b/be/benchmark/benchmark_fastunion.hpp index ba469b75fa6ae3..ae574321aec348 100644 --- a/be/benchmark/benchmark_fastunion.hpp +++ b/be/benchmark/benchmark_fastunion.hpp @@ -19,7 +19,7 @@ #include -#include "util/bitmap_value.h" +#include "core/value/bitmap_value.h" using Roaring64Map = doris::detail::Roaring64Map; diff --git a/be/benchmark/benchmark_hll_merge.hpp b/be/benchmark/benchmark_hll_merge.hpp index d923d208fe4446..2fc6c47eaa68b8 100644 --- a/be/benchmark/benchmark_hll_merge.hpp +++ b/be/benchmark/benchmark_hll_merge.hpp @@ -17,7 +17,7 @@ #include -#include "olap/hll.h" +#include "core/value/hll.h" #include "util/hash_util.hpp" namespace doris { diff --git a/be/benchmark/benchmark_main.cpp b/be/benchmark/benchmark_main.cpp index 7c64fa2729cefb..775f5caffef5bd 100644 --- a/be/benchmark/benchmark_main.cpp +++ b/be/benchmark/benchmark_main.cpp @@ -24,19 +24,11 @@ #include "benchmark_arrow_validation.hpp" #include "benchmark_bit_pack.hpp" -#include "benchmark_bits.hpp" -#include "benchmark_block_bloom_filter.hpp" #include "benchmark_column_array_view.hpp" #include "benchmark_column_array_view_distance.hpp" -#include "benchmark_column_view.hpp" -#include "benchmark_damerau_levenshtein.hpp" #include "benchmark_fastunion.hpp" #include "benchmark_fmod.hpp" #include "benchmark_hll_merge.hpp" -#include "benchmark_hybrid_set.hpp" -#include "benchmark_pdep_unpack.hpp" -#include "benchmark_string.hpp" -#include "benchmark_string_replace.hpp" #include "benchmark_zone_map_index.hpp" #include "binary_cast_benchmark.hpp" #include "common/config.h" @@ -44,20 +36,16 @@ #include "core/column/column_string.h" #include "core/data_type/data_type.h" #include "core/data_type/data_type_string.h" +#include "parquet/benchmark_file_scanner_expr.hpp" #include "parquet/benchmark_parquet_decoder.hpp" #include "parquet/benchmark_parquet_kernels.hpp" #include "parquet/benchmark_parquet_reader.hpp" +#include "parquet/benchmark_parquet_selection.hpp" #include "runtime/exec_env.h" #include "runtime/memory/mem_tracker_limiter.h" #include "runtime/memory/thread_mem_tracker_mgr.h" #include "runtime/thread_context.h" -// benchmark_binary_plain_page_v2.hpp must be included LAST: it transitively pulls AWS SDK -// headers (via storage/cache/page_cache.h) whose symbols shadow types used by the benchmark -// headers above (notably binary_cast_benchmark.hpp). Keeping it last avoids the clash without -// disabling any benchmark. (Do not let clang-format reorder it above the others.) -#include "benchmark_binary_plain_page_v2.hpp" - namespace doris { // change if need static bool init_benchmark_config(const char* executable) { @@ -118,8 +106,6 @@ int main(int argc, char** argv) { if (!doris::init_benchmark_config(argv[0])) { return 1; } - doris::config::enable_bmi2_optimizations = true; - SCOPED_INIT_THREAD_CONTEXT(); doris::ExecEnv::GetInstance()->init_mem_tracker(); doris::thread_context()->thread_mem_tracker_mgr->init(); diff --git a/be/benchmark/binary_cast_benchmark.hpp b/be/benchmark/binary_cast_benchmark.hpp index 9949a783b05a1d..fb2650b40f5579 100644 --- a/be/benchmark/binary_cast_benchmark.hpp +++ b/be/benchmark/binary_cast_benchmark.hpp @@ -21,7 +21,7 @@ #include #include -#include "util/binary_cast.hpp" +#include "core/binary_cast.hpp" namespace doris { diff --git a/be/benchmark/parquet/AGENTS.md b/be/benchmark/parquet/AGENTS.md index 6c2fb8c85df1fc..4c1d3cf4d5e197 100644 --- a/be/benchmark/parquet/AGENTS.md +++ b/be/benchmark/parquet/AGENTS.md @@ -6,16 +6,20 @@ benchmark system described in the design document. ## What exists today -The benchmark binary registers three groups: +The benchmark binary registers five groups: - `ParquetDecoder`: native page decoder benchmarks using in-memory encoded pages. - `ParquetKernel`: isolated SIMD-sensitive decode and predicate kernels. +- `ParquetSelection`: isolated selection initialization and predicate compaction paths. - `ParquetReader`: local-file benchmarks that call the format V2 Parquet reader directly. +- `FileScannerExpr`: expression lifecycle benchmarks for split-local clone, prepare, and open. The relevant files are: - `benchmark_parquet_decoder.hpp`: deterministic page construction and decoder registration. +- `benchmark_parquet_selection.hpp`: selection initialization and compaction registration. - `benchmark_parquet_reader.hpp`: deterministic local Parquet fixtures and reader registration. +- `benchmark_file_scanner_expr.hpp`: runtime-filter expression lifecycle registration. - `parquet_benchmark_scenarios.h`: scenario definitions and the selected matrix. - `README.md`: short human-oriented build and invocation examples. - `be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp`: matrix invariants. @@ -23,7 +27,8 @@ The relevant files are: Do not describe this suite as end-to-end SQL, `FileScannerV2`, remote I/O, V1/V2 comparison, or a cross-engine benchmark. The reader benchmark starts at `format::parquet::ParquetReader` and does not include FE planning, scanner scheduling, `TableReader`, client latency, or Runtime Profile -collection. +collection. `FileScannerExpr` isolates expression lifecycle work and likewise does not execute a +scanner or read a file. ## Build and list cases @@ -37,7 +42,10 @@ List all Parquet cases and verify the expected registration counts: ```shell be/output/lib/benchmark_test --benchmark_list_tests \ - | grep -E '^Parquet(Decoder|Kernel|Reader)/' + | grep -E '^Parquet(Decoder|Kernel|Selection|Reader)/' + +be/output/lib/benchmark_test --benchmark_list_tests \ + | grep '^FileScannerExpr/' be/output/lib/benchmark_test --benchmark_list_tests \ | grep -c '^ParquetDecoder/' # currently 228 @@ -45,8 +53,14 @@ be/output/lib/benchmark_test --benchmark_list_tests \ be/output/lib/benchmark_test --benchmark_list_tests \ | grep -c '^ParquetKernel/' # currently 92 +be/output/lib/benchmark_test --benchmark_list_tests \ + | grep -c '^ParquetSelection/' # currently 25 + be/output/lib/benchmark_test --benchmark_list_tests \ | grep -c '^ParquetReader/' # currently 167 + +be/output/lib/benchmark_test --benchmark_list_tests \ + | grep -c '^FileScannerExpr/' # currently 8 ``` When running the binary directly from `be/build_RELEASE/bin`, make sure the JVM and third-party @@ -71,11 +85,23 @@ be/output/lib/benchmark_test \ --benchmark_out=parquet-kernel-smoke.json \ --benchmark_out_format=json +be/output/lib/benchmark_test \ + --benchmark_filter='^ParquetSelection/' \ + --benchmark_min_time=0.001s \ + --benchmark_out=parquet-selection-smoke.json \ + --benchmark_out_format=json + be/output/lib/benchmark_test \ --benchmark_filter='^ParquetReader/' \ --benchmark_min_time=0.001s \ --benchmark_out=parquet-reader-smoke.json \ --benchmark_out_format=json + +be/output/lib/benchmark_test \ + --benchmark_filter='^FileScannerExpr/' \ + --benchmark_min_time=0.001s \ + --benchmark_out=file-scanner-expr-smoke.json \ + --benchmark_out_format=json ``` Reject a smoke run if the process is non-zero, the expected number of JSON results is absent, or @@ -128,6 +154,18 @@ rates with both placement patterns, 0% through 100% raw-predicate selectivities, legacy and fused implementations in the same binary and validates both against an independent source-level oracle before timing. +`ParquetSelection` contains 25 cases that isolate the selection-vector work used by Parquet +predicate evaluation. It measures identity initialization, one raw-row filter, and two successive +filters. The filter matrix covers 0%, 1%, 10%, 50%, 90%, and 100% selectivity with clustered and +alternating matches. These cases include `SelectionVector::resize()` in the timed region because +initializing a new batch is part of the production predicate path. + +`FileScannerExpr` contains eight cases that clone, prepare, and open an already-prepared +`VDirectInPredicate` with 128, 1,024, 8,192, or 65,536 integer set values. Each cardinality registers +`impl_shared` and `impl_rematerialize` in the same binary. Set construction and the original +fragment-level materialization are outside the timed region. The cases model the repeated +split-local expression lifecycle only; they do not include scanner scheduling or file reads. + `ParquetReader` deliberately uses a single-variable matrix rather than a Cartesian product. After deduplication it contains 167 cases covering: @@ -302,10 +340,10 @@ be simulated by silently changing the local reader benchmark. ## Current validation record -The current expected registration counts are 228 decoder, 92 kernel, and 167 reader cases. A smoke -run is an execution record only, not a reviewed performance baseline, because repetitions, host -isolation, warmups, cache control, `perf` data, variance, and before/after comparison are not -collected. +The current expected registration counts are 228 decoder, 92 kernel, 25 selection, 167 reader, and +8 expression-lifecycle cases. A smoke run is an execution record only, not a reviewed performance +baseline, because repetitions, host isolation, warmups, cache control, `perf` data, variance, and +before/after comparison are not collected. ## Rules for extending the suite diff --git a/be/benchmark/parquet/README.md b/be/benchmark/parquet/README.md index fcbed4a548ba4c..156302b24238d1 100644 --- a/be/benchmark/parquet/README.md +++ b/be/benchmark/parquet/README.md @@ -21,6 +21,12 @@ List only the Parquet cases: be/output/lib/benchmark_test --benchmark_list_tests | grep '^Parquet' ``` +List the split-local runtime-filter expression lifecycle cases: + +```shell +be/output/lib/benchmark_test --benchmark_list_tests | grep '^FileScannerExpr/' +``` + ## Decoder cases `ParquetDecoder` measures the native decoder with data generation and encoder setup outside the @@ -75,6 +81,20 @@ taskset -c 8 be/output/lib/benchmark_test \ # Repeat fused as B2, then legacy as A2, changing only --benchmark_out. ``` +## Selection compaction cases + +`ParquetSelection` isolates the selection-vector paths used after raw and expression predicate +evaluation. It covers implicit identity initialization, a filter indexed by source row, and a +second compact filter applied after an earlier predicate has already made the selection sparse. + +```shell +be/output/lib/benchmark_test \ + --benchmark_filter='^ParquetSelection/(resize_identity|row_filter|cascade_filter)/' \ + --benchmark_min_time=1s \ + --benchmark_repetitions=10 \ + --benchmark_report_aggregates_only=true +``` + ## Local reader cases `ParquetReader` measures local open-to-first-block, full scan, predicate scan, complex residual @@ -113,3 +133,21 @@ be/output/lib/benchmark_test \ Every result reports throughput plus `raw_rows`, `selected_rows`, `fixture_bytes`, `ns/raw_row`, and (when at least one row survives) `ns/selected_row`. Keep CPU frequency, build type, compiler, machine placement, and benchmark filters fixed when comparing two commits. + +## Runtime-filter expression lifecycle cases + +`FileScannerExpr` measures only the repeated deep-clone, prepare, and open work for an already +prepared direct-IN runtime filter. Four cardinalities sweep 128 through 65,536 set values, with +shared-state and forced-rematerialization implementations registered in the same binary. Set +construction and the original fragment-level prepare/open are outside the timed region. + +```shell +be/output/lib/benchmark_test \ + --benchmark_filter='^FileScannerExpr/direct_in_clone_prepare_open/' \ + --benchmark_min_time=1s \ + --benchmark_repetitions=10 \ + --benchmark_report_aggregates_only=true +``` + +These cases do not execute `FileScannerV2`, schedule splits, or read Parquet files. They isolate the +expression lifecycle visible in scanner profiles so it can be compared without I/O noise. diff --git a/be/benchmark/parquet/benchmark_file_scanner_expr.hpp b/be/benchmark/parquet/benchmark_file_scanner_expr.hpp new file mode 100644 index 00000000000000..14654472f5ee64 --- /dev/null +++ b/be/benchmark/parquet/benchmark_file_scanner_expr.hpp @@ -0,0 +1,118 @@ +// 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 + +#include +#include +#include +#include +#include + +#include "core/data_type/data_type_number.h" +#include "exprs/create_predicate_function.h" +#include "exprs/vdirect_in_predicate.h" +#include "exprs/vexpr_context.h" +#include "exprs/vslot_ref.h" +#include "runtime/descriptors.h" +#include "runtime/runtime_state.h" + +namespace doris::parquet_benchmark::file_scanner_expr_detail { + +inline TExprNode make_direct_in_node() { + TExprNode node; + node.__set_type(create_type_desc(PrimitiveType::TYPE_BOOLEAN)); + node.__set_node_type(TExprNodeType::IN_PRED); + node.__set_opcode(TExprOpcode::FILTER_IN); + node.__set_num_children(1); + node.__set_is_nullable(false); + node.in_predicate.__set_is_not_in(false); + return node; +} + +inline void run_direct_in_clone_prepare_open(benchmark::State& state, size_t cardinality, + bool share_pruning_state) { + std::shared_ptr filter(create_set(PrimitiveType::TYPE_INT, cardinality, false)); + for (size_t index = 0; index < cardinality; ++index) { + const int32_t value = static_cast(index); + filter->insert(&value); + } + auto root = VDirectInPredicate::create_shared(make_direct_in_node(), std::move(filter), true); + root->add_child(VSlotRef::create_shared(0, 0, -1, std::make_shared(), + "runtime_filter_key")); + RuntimeState runtime_state {TQueryOptions(), TQueryGlobals()}; + RowDescriptor row_desc; + VExprContext original(root); + auto status = original.prepare(&runtime_state, row_desc); + if (status.ok()) { + status = original.open(&runtime_state); + } + if (!status.ok()) { + const auto error = status.to_string(); + state.SkipWithError(error.c_str()); + return; + } + + for (auto _ : state) { + VExprSPtr cloned_root; + if (share_pruning_state) { + status = root->deep_clone(&cloned_root); + } else { + auto rematerialized = VDirectInPredicate::create_shared(make_direct_in_node(), + root->get_set_func(), true); + rematerialized->add_child(VSlotRef::create_shared( + 0, 0, -1, std::make_shared(), "runtime_filter_key")); + cloned_root = std::move(rematerialized); + status = Status::OK(); + } + if (status.ok()) { + VExprContext cloned(cloned_root); + status = cloned.prepare(&runtime_state, row_desc); + if (status.ok()) { + status = cloned.open(&runtime_state); + } + benchmark::DoNotOptimize(cloned_root); + } + if (!status.ok()) { + const auto error = status.to_string(); + state.SkipWithError(error.c_str()); + return; + } + } + state.counters["set_values"] = static_cast(cardinality); +} + +inline bool register_file_scanner_expr_benchmarks() { + for (const size_t cardinality : std::array {128, 1024, 8192, 65536}) { + for (const bool share_pruning_state : {false, true}) { + const std::string name = "FileScannerExpr/direct_in_clone_prepare_open/values_" + + std::to_string(cardinality) + + (share_pruning_state ? "/impl_shared" : "/impl_rematerialize"); + benchmark::RegisterBenchmark(name.c_str(), [cardinality, share_pruning_state]( + benchmark::State& state) { + run_direct_in_clone_prepare_open(state, cardinality, share_pruning_state); + })->Unit(benchmark::kNanosecond); + } + } + return true; +} + +inline const bool FILE_SCANNER_EXPR_BENCHMARKS_REGISTERED = register_file_scanner_expr_benchmarks(); + +} // namespace doris::parquet_benchmark::file_scanner_expr_detail diff --git a/be/benchmark/parquet/benchmark_parquet_selection.hpp b/be/benchmark/parquet/benchmark_parquet_selection.hpp new file mode 100644 index 00000000000000..06c0a338defd11 --- /dev/null +++ b/be/benchmark/parquet/benchmark_parquet_selection.hpp @@ -0,0 +1,172 @@ +// 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 + +#include +#include +#include +#include +#include +#include + +#include "format_v2/parquet/selection_vector.h" +#include "parquet_benchmark_scenarios.h" + +namespace doris::parquet_benchmark::selection_detail { + +constexpr size_t SELECTION_ROWS = 1UL << 12; +constexpr int CASCADE_FIRST_SELECTIVITY = 90; + +inline std::vector make_filter(size_t rows, int selectivity_percent, Pattern pattern) { + std::vector filter(rows, 0); + const size_t selected_rows = + rows * static_cast(std::clamp(selectivity_percent, 0, 100)) / 100; + if (pattern == Pattern::CLUSTERED) { + std::fill_n(filter.begin(), selected_rows, uint8_t {1}); + } else if (selected_rows != 0) { + for (size_t selected = 0; selected < selected_rows; ++selected) { + filter[selected * rows / selected_rows] = 1; + } + } + return filter; +} + +// Keep the benchmark source buildable on revisions before the bulk compaction helpers. The +// fallback mirrors the former Parquet scan loops so one named matrix can compare both revisions. +template +size_t compact_with_row_filter(Selection* selection, const uint8_t* filter, size_t rows) { + if constexpr (requires { selection->compact_with_row_filter(filter, rows); }) { + return selection->compact_with_row_filter(filter, rows); + } else { + size_t output = 0; + for (size_t position = 0; position < rows; ++position) { + const auto row = selection->get_index(position); + if (filter[row] != 0) { + selection->set_index(output++, row); + } + } + return output; + } +} + +template +size_t compact_with_selection_filter(Selection* selection, const uint8_t* filter, size_t rows) { + if constexpr (requires { selection->compact_with_selection_filter(filter, rows); }) { + return selection->compact_with_selection_filter(filter, rows); + } else { + size_t output = 0; + for (size_t position = 0; position < rows; ++position) { + if (filter[position] != 0) { + selection->set_index(output++, selection->get_index(position)); + } + } + return output; + } +} + +inline std::vector expected_selection_rows( + const SelectionScenario& scenario, const std::vector& row_filter, + const std::vector& first_filter, const std::vector& selection_filter) { + std::vector rows(SELECTION_ROWS); + std::iota(rows.begin(), rows.end(), 0); + if (scenario.operation == SelectionOperation::RESIZE_IDENTITY) { + return rows; + } + if (scenario.operation == SelectionOperation::ROW_FILTER) { + std::erase_if(rows, [&](const auto row) { return row_filter[row] == 0; }); + return rows; + } + std::erase_if(rows, [&](const auto row) { return first_filter[row] == 0; }); + std::vector cascaded; + cascaded.reserve(rows.size()); + for (size_t position = 0; position < rows.size(); ++position) { + if (selection_filter[position] != 0) { + cascaded.push_back(rows[position]); + } + } + return cascaded; +} + +inline void run_selection(benchmark::State& state, const SelectionScenario& scenario) { + format::parquet::SelectionVector selection; + const auto row_filter = + make_filter(SELECTION_ROWS, scenario.selectivity_percent, scenario.pattern); + const auto first_filter = + make_filter(SELECTION_ROWS, CASCADE_FIRST_SELECTIVITY, Pattern::ALTERNATING); + const size_t first_selected = + static_cast(std::count(first_filter.begin(), first_filter.end(), uint8_t {1})); + const auto selection_filter = + make_filter(first_selected, scenario.selectivity_percent, scenario.pattern); + const auto expected_rows = + expected_selection_rows(scenario, row_filter, first_filter, selection_filter); + + size_t selected_rows = 0; + for (auto _ : state) { + selection.resize(SELECTION_ROWS); + switch (scenario.operation) { + case SelectionOperation::RESIZE_IDENTITY: + selected_rows = SELECTION_ROWS; + break; + case SelectionOperation::ROW_FILTER: + selected_rows = compact_with_row_filter(&selection, row_filter.data(), SELECTION_ROWS); + break; + case SelectionOperation::CASCADE_FILTER: + selected_rows = + compact_with_row_filter(&selection, first_filter.data(), SELECTION_ROWS); + selected_rows = compact_with_selection_filter(&selection, selection_filter.data(), + selected_rows); + break; + } + benchmark::DoNotOptimize(selected_rows); + benchmark::ClobberMemory(); + } + bool selection_matches = selected_rows == expected_rows.size(); + for (size_t position = 0; selection_matches && position < selected_rows; ++position) { + selection_matches = selection.get_index(position) == expected_rows[position]; + } + if (!selection_matches) { + state.SkipWithError("selection compaction produced unexpected row indices"); + return; + } + + state.SetItemsProcessed(static_cast(state.iterations() * SELECTION_ROWS)); + state.SetBytesProcessed(static_cast(state.iterations() * SELECTION_ROWS)); + state.counters["raw_rows"] = static_cast(SELECTION_ROWS); + state.counters["selected_rows"] = static_cast(selected_rows); + state.counters["ns/raw_row"] = benchmark::Counter( + static_cast(SELECTION_ROWS), + benchmark::Counter::kIsIterationInvariantRate | benchmark::Counter::kInvert); +} + +inline bool register_selection_benchmarks() { + for (const auto& scenario : selection_scenarios()) { + const std::string name = "ParquetSelection/" + to_string(scenario.operation) + "/sel_" + + std::to_string(scenario.selectivity_percent) + "/" + + to_string(scenario.pattern); + benchmark::RegisterBenchmark(name.c_str(), [=](benchmark::State& state) { + run_selection(state, scenario); + })->Unit(benchmark::kNanosecond); + } + return true; +} + +inline const bool SELECTION_BENCHMARKS_REGISTERED = register_selection_benchmarks(); + +} // namespace doris::parquet_benchmark::selection_detail diff --git a/be/benchmark/parquet/parquet_benchmark_scenarios.h b/be/benchmark/parquet/parquet_benchmark_scenarios.h index d2655b3380e9f0..a9c58c15d8cff1 100644 --- a/be/benchmark/parquet/parquet_benchmark_scenarios.h +++ b/be/benchmark/parquet/parquet_benchmark_scenarios.h @@ -37,6 +37,7 @@ enum class Encoding { enum class ValueType { INT32, INT64, FLOAT, DOUBLE, BYTE_ARRAY, FIXED_LEN_BYTE_ARRAY }; enum class Pattern { CLUSTERED, ALTERNATING }; enum class Projection { PREDICATE_ONLY, PREDICATE_PROJECTED }; +enum class SelectionOperation { RESIZE_IDENTITY, ROW_FILTER, CASCADE_FILTER }; enum class ReaderOperation { OPEN_TO_FIRST_BLOCK, FULL_SCAN, @@ -82,6 +83,12 @@ struct KernelScenario { NestedSelectionImplementation nested_implementation = NestedSelectionImplementation::FUSED; }; +struct SelectionScenario { + SelectionOperation operation; + int selectivity_percent; + Pattern pattern; +}; + struct SelectionRange { size_t first; size_t count; @@ -156,6 +163,20 @@ inline std::vector kernel_scenarios() { return scenarios; } +inline std::vector selection_scenarios() { + std::vector scenarios { + {SelectionOperation::RESIZE_IDENTITY, 100, Pattern::CLUSTERED}}; + for (const auto operation : + {SelectionOperation::ROW_FILTER, SelectionOperation::CASCADE_FILTER}) { + for (const int selectivity : {0, 1, 10, 50, 90, 100}) { + for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) { + scenarios.push_back({operation, selectivity, pattern}); + } + } + } + return scenarios; +} + inline std::vector reader_scenarios() { std::vector scenarios; std::setrefresh_conjuncts(std::move(refreshed_conjuncts))); + _table_reader_rf_num = _applied_rf_num; + } if (_should_run_adaptive_batch_size()) { _table_reader->set_batch_size(_predict_reader_batch_rows()); } @@ -549,6 +555,7 @@ Status FileScannerV2::_prepare_next_split(bool* eos) { } COUNTER_UPDATE(_file_counter, 1); _has_prepared_split = true; + _table_reader_rf_num = _applied_rf_num; *eos = false; return Status::OK(); } diff --git a/be/src/exec/scan/file_scanner_v2.h b/be/src/exec/scan/file_scanner_v2.h index 658ec44ad7f051..87a68e2ab6d176 100644 --- a/be/src/exec/scan/file_scanner_v2.h +++ b/be/src/exec/scan/file_scanner_v2.h @@ -178,6 +178,7 @@ class FileScannerV2 final : public Scanner { std::shared_ptr _split_source; bool _first_scan_range = false; bool _has_prepared_split = false; + int _table_reader_rf_num = 0; TFileRangeDesc _current_range; std::string _current_range_path; diff --git a/be/src/exprs/vdirect_in_predicate.h b/be/src/exprs/vdirect_in_predicate.h index 79d0f996d7c792..b272c816edb12a 100644 --- a/be/src/exprs/vdirect_in_predicate.h +++ b/be/src/exprs/vdirect_in_predicate.h @@ -17,6 +17,7 @@ #pragma once +#include #include #include @@ -37,6 +38,15 @@ namespace doris { class VDirectInPredicate final : public VExpr { ENABLE_FACTORY_CREATOR(VDirectInPredicate); + struct PruningState { + std::once_flag materialize_once; + Status materialization_status; + bool zonemap_materialized = false; + std::vector seg_filter_values; + Field seg_filter_min; + Field seg_filter_max; + }; + public: // `hybrid_set_values_match_child_type` tells whether values in `filter` can be interpreted with // the child expression type. Parquet/ORC dictionary-filter rewrites evaluate the original @@ -90,22 +100,24 @@ class VDirectInPredicate final : public VExpr { std::shared_ptr get_set_func() const override { return _filter; } ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override { - return expr_zonemap::eval_in_zonemap(ctx, get_child(0), false, _seg_filter_values, - _seg_filter_min, _seg_filter_max); + return expr_zonemap::eval_in_zonemap( + ctx, get_child(0), false, _pruning_state->seg_filter_values, + _pruning_state->seg_filter_min, _pruning_state->seg_filter_max); } bool can_evaluate_zonemap_filter() const override { - return _zonemap_materialized && + return _pruning_state->zonemap_materialized && std::dynamic_pointer_cast(get_child(0)) != nullptr; } ZoneMapFilterResult evaluate_dictionary_filter( const DictionaryEvalContext& ctx) const override { - return expr_zonemap::eval_in_dictionary(ctx, get_child(0), false, _seg_filter_values); + return expr_zonemap::eval_in_dictionary(ctx, get_child(0), false, + _pruning_state->seg_filter_values); } bool can_evaluate_dictionary_filter() const override { - return _zonemap_materialized && + return _pruning_state->zonemap_materialized && std::dynamic_pointer_cast(get_child(0)) != nullptr; } @@ -177,8 +189,12 @@ class VDirectInPredicate final : public VExpr { Status clone_node(VExprSPtr* cloned_expr) const override { DORIS_CHECK(cloned_expr != nullptr); - *cloned_expr = VDirectInPredicate::create_shared(clone_texpr_node(), _filter, - _hybrid_set_values_match_child_type); + auto cloned = VDirectInPredicate::create_shared(clone_texpr_node(), _filter, + _hybrid_set_values_match_child_type); + // Runtime-filter sets are immutable after publication, and file-local rewrites preserve + // the predicate's logical child type, so every split clone must reuse this materialization. + cloned->_pruning_state = _pruning_state; + *cloned_expr = std::move(cloned); return Status::OK(); } @@ -300,21 +316,27 @@ class VDirectInPredicate final : public VExpr { } Status _materialize_for_zonemap_filter() { - if (!_hybrid_set_values_match_child_type) { - _zonemap_materialized = false; - return Status::OK(); - } - DORIS_CHECK(_filter != nullptr); - auto& filter = *_filter; - const auto& data_type = remove_nullable(get_child(0)->data_type()); - expr_zonemap::InZonemapMaterializedSet materialized; - RETURN_IF_ERROR(expr_zonemap::materialize_hybrid_set_for_zonemap_filter(filter, data_type, - &materialized)); - _seg_filter_values = std::move(materialized.values); - _seg_filter_min = std::move(materialized.min_value); - _seg_filter_max = std::move(materialized.max_value); - _zonemap_materialized = true; - return Status::OK(); + const auto pruning_state = _pruning_state; + std::call_once(pruning_state->materialize_once, [&] { + if (!_hybrid_set_values_match_child_type) { + return; + } + DORIS_CHECK(_filter != nullptr); + auto& filter = *_filter; + const auto& data_type = remove_nullable(get_child(0)->data_type()); + expr_zonemap::InZonemapMaterializedSet materialized; + pruning_state->materialization_status = + expr_zonemap::materialize_hybrid_set_for_zonemap_filter(filter, data_type, + &materialized); + if (!pruning_state->materialization_status.ok()) { + return; + } + pruning_state->seg_filter_values = std::move(materialized.values); + pruning_state->seg_filter_min = std::move(materialized.min_value); + pruning_state->seg_filter_max = std::move(materialized.max_value); + pruning_state->zonemap_materialized = true; + }); + return pruning_state->materialization_status; } std::shared_ptr _filter; @@ -323,10 +345,7 @@ class VDirectInPredicate final : public VExpr { // literals for zonemap pruning or slot-IN rewrite. bool _hybrid_set_values_match_child_type = true; std::string _expr_name; - bool _zonemap_materialized = false; - std::vector _seg_filter_values; - Field _seg_filter_min; - Field _seg_filter_max; + std::shared_ptr _pruning_state = std::make_shared(); }; #include "common/compile_check_end.h" diff --git a/be/src/format_v2/column_mapper.cpp b/be/src/format_v2/column_mapper.cpp index c3dbe3fa9e7766..7e028370221c38 100644 --- a/be/src/format_v2/column_mapper.cpp +++ b/be/src/format_v2/column_mapper.cpp @@ -2161,13 +2161,20 @@ Status TableColumnMapper::_build_filter_entries(const FileScanRequest& file_requ Status TableColumnMapper::create_scan_request( const std::vector& table_filters, const std::vector& projected_columns, FileScanRequest* file_request, - RuntimeState* runtime_state) { + RuntimeState* runtime_state, + const std::map* fixed_local_positions) { // 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(); file_request->non_predicate_columns.clear(); file_request->predicate_only_columns.clear(); file_request->local_positions.clear(); + if (fixed_local_positions != nullptr) { + // A refreshed predicate may promote a lazy column, but the active split's block slots are + // immutable. Seed their positions before rebuilding expressions so every rewritten SlotRef + // continues to address the same physical column. + file_request->local_positions = *fixed_local_positions; + } file_request->conjuncts.clear(); file_request->delete_conjuncts.clear(); _filter_entries.clear(); diff --git a/be/src/format_v2/column_mapper.h b/be/src/format_v2/column_mapper.h index e03f836061ebf5..68d0ff357c0e52 100644 --- a/be/src/format_v2/column_mapper.h +++ b/be/src/format_v2/column_mapper.h @@ -193,10 +193,11 @@ class TableColumnMapper { // Convert a table-level scan request into a file-local scan request. table_filters preserve // row-level filtering semantics and are rewritten as file-local conjuncts. File-layer pruning // such as ZoneMap, dictionary, and bloom filter derives from those localized VExpr conjuncts. - virtual Status create_scan_request(const std::vector& table_filters, - const std::vector& projected_columns, - FileScanRequest* file_request, - RuntimeState* runtime_state = nullptr); + virtual Status create_scan_request( + const std::vector& table_filters, + const std::vector& projected_columns, FileScanRequest* file_request, + RuntimeState* runtime_state = nullptr, + const std::map* fixed_local_positions = nullptr); // Localize table-level filters to the file schema. // Trivial mappings can copy structured predicates directly. Type changes may be localized with diff --git a/be/src/format_v2/file_reader.h b/be/src/format_v2/file_reader.h index 3ff512975d2dcd..65a2d03417c605 100644 --- a/be/src/format_v2/file_reader.h +++ b/be/src/format_v2/file_reader.h @@ -305,6 +305,15 @@ class FileReader { return Status::OK(); } + // Readers opt in only when they can keep an immutable request for the active physical + // granule and switch a newer snapshot at a well-defined boundary. + virtual bool supports_scan_request_refresh() const { return false; } + + virtual Status queue_scan_request(std::shared_ptr request) { + (void)request; + return Status::NotSupported("FileReader does not support scan request refresh"); + } + virtual Status get_block(Block* file_block, size_t* rows, bool* eof) { if (rows != nullptr) { *rows = 0; diff --git a/be/src/format_v2/jni/jni_table_reader.cpp b/be/src/format_v2/jni/jni_table_reader.cpp index b58696fe4ad895..dfa00c1568ef2c 100644 --- a/be/src/format_v2/jni/jni_table_reader.cpp +++ b/be/src/format_v2/jni/jni_table_reader.cpp @@ -77,6 +77,23 @@ Status JniTableReader::prepare_split(const SplitReadOptions& options) { return _open_jni_scanner(); } +Status JniTableReader::refresh_conjuncts(VExprContextSPtrs conjuncts) { + if (_scanner_opened) { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.refresh_conjuncts_timer); + SCOPED_TIMER(_profile.file_reader_total_timer); + SCOPED_TIMER(_profile.file_reader_refresh_timer); + RowDescriptor row_desc; + for (const auto& conjunct : conjuncts) { + // JNI readers bypass TableReader::open_reader(), so a late predicate would otherwise + // replace the active snapshot without initializing its executable function state. + RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc)); + RETURN_IF_ERROR(conjunct->open(_runtime_state)); + } + } + return TableReader::refresh_conjuncts(std::move(conjuncts)); +} + Status JniTableReader::get_block(Block* output_block, bool* eos) { SCOPED_TIMER(_profile.total_timer); SCOPED_TIMER(_profile.exec_timer); diff --git a/be/src/format_v2/jni/jni_table_reader.h b/be/src/format_v2/jni/jni_table_reader.h index df7edc92f985fb..76e51c1c059644 100644 --- a/be/src/format_v2/jni/jni_table_reader.h +++ b/be/src/format_v2/jni/jni_table_reader.h @@ -48,6 +48,7 @@ class JniTableReader : public TableReader { Status init(TableReadOptions&& options) override; Status prepare_split(const SplitReadOptions& options) override; + Status refresh_conjuncts(VExprContextSPtrs conjuncts) override; Status get_block(Block* block, bool* eos) override; Status abort_split() override; Status close() override; diff --git a/be/src/format_v2/parquet/parquet_profile.cpp b/be/src/format_v2/parquet/parquet_profile.cpp index 02b64d7b63c867..e6717505cea898 100644 --- a/be/src/format_v2/parquet/parquet_profile.cpp +++ b/be/src/format_v2/parquet/parquet_profile.cpp @@ -31,6 +31,8 @@ void ParquetProfile::init(RuntimeProfile* profile) { static const char* parquet_profile = "ParquetReader"; total_time = ADD_CHILD_TIMER_WITH_LEVEL(profile, parquet_profile, file_scan_profile::FILE_READER, 1); + refresh_scan_request_time = + ADD_CHILD_TIMER_WITH_LEVEL(profile, "RefreshScanRequestTime", parquet_profile, 1); // Row-group counters are part of the long-standing ParquetReader profile contract. Keep them // below the format node so profile parsers and operators can attribute pruning to Parquet. @@ -256,6 +258,7 @@ void ParquetProfile::update_deferred_pruning_stats(const ParquetPruningStats& pr bool selected) const { const int64_t filtered = selected ? 0 : 1; COUNTER_UPDATE(filtered_row_groups, filtered); + COUNTER_UPDATE(filtered_row_groups_by_min_max, pruning_stats.filtered_row_groups_by_statistics); COUNTER_UPDATE(filtered_row_groups_by_dictionary, pruning_stats.filtered_row_groups_by_dictionary); COUNTER_UPDATE(filtered_row_groups_by_bloom_filter, diff --git a/be/src/format_v2/parquet/parquet_profile.h b/be/src/format_v2/parquet/parquet_profile.h index 2282a70db30540..57c03c79336f06 100644 --- a/be/src/format_v2/parquet/parquet_profile.h +++ b/be/src/format_v2/parquet/parquet_profile.h @@ -133,6 +133,7 @@ struct ParquetProfile { ParquetScanProfile scan_profile() const; RuntimeProfile::Counter* total_time = nullptr; + RuntimeProfile::Counter* refresh_scan_request_time = nullptr; RuntimeProfile::Counter* filtered_row_groups = nullptr; RuntimeProfile::Counter* filtered_row_groups_by_min_max = nullptr; diff --git a/be/src/format_v2/parquet/parquet_reader.cpp b/be/src/format_v2/parquet/parquet_reader.cpp index 4a80bc0403df8d..065adc63b66e1e 100644 --- a/be/src/format_v2/parquet/parquet_reader.cpp +++ b/be/src/format_v2/parquet/parquet_reader.cpp @@ -471,10 +471,24 @@ Status ParquetReader::open(std::shared_ptr request) { _state->scheduler.set_global_rowid_context(_global_rowid_context); _state->scheduler.set_scan_profile(_parquet_profile.scan_profile()); _state->scheduler.set_plan(std::move(row_group_plan)); + _state->scheduler.set_scan_request(request_snapshot); _eof = _state->scheduler.empty(); return Status::OK(); } +Status ParquetReader::queue_scan_request(std::shared_ptr request) { + SCOPED_TIMER(_parquet_profile.total_time); + SCOPED_TIMER(_parquet_profile.refresh_scan_request_time); + if (_state == nullptr || _state->file_context.native_metadata == nullptr) { + return Status::Uninitialized("ParquetReader is not open"); + } + DORIS_CHECK(request != nullptr); + RETURN_IF_ERROR(validate_requested_columns_supported(_state->file_schema, *request)); + _state->scheduler.queue_scan_request(request); + _request = std::move(request); + return Status::OK(); +} + Status ParquetReader::get_block(Block* file_block, size_t* rows, bool* eof) { SCOPED_TIMER(_parquet_profile.total_time); if (_state == nullptr || _state->file_context.native_metadata == nullptr) { @@ -489,15 +503,14 @@ Status ParquetReader::get_block(Block* file_block, size_t* rows, bool* eof) { *eof = true; return Status::OK(); } - auto request_snapshot = _request; - if (request_snapshot == nullptr) { + if (_request == nullptr) { return Status::Cancelled("ParquetReader is closed"); } const auto predicate_filtered_rows_before = _state->scheduler.predicate_filtered_rows(); const auto raw_rows_read_before = _state->scheduler.raw_rows_read(); Status st = _state->scheduler.read_next_batch(_state->file_context, _state->file_schema, - *request_snapshot, file_block, rows, eof); + file_block, rows, eof); if (!st.ok()) { if (_io_ctx != nullptr && _io_ctx->should_stop) { *rows = 0; diff --git a/be/src/format_v2/parquet/parquet_reader.h b/be/src/format_v2/parquet/parquet_reader.h index da6135b81ae838..fe95b93a9e0101 100644 --- a/be/src/format_v2/parquet/parquet_reader.h +++ b/be/src/format_v2/parquet/parquet_reader.h @@ -60,6 +60,10 @@ class ParquetReader : public format::FileReader { Status open(std::shared_ptr request) override; + bool supports_scan_request_refresh() const override { return true; } + + Status queue_scan_request(std::shared_ptr request) override; + Status get_block(Block* file_block, size_t* rows, bool* eof) override; Status get_aggregate_result(const format::FileAggregateRequest& request, diff --git a/be/src/format_v2/parquet/parquet_scan.cpp b/be/src/format_v2/parquet/parquet_scan.cpp index 33ef6a08fedb2c..18b42d90a96368 100644 --- a/be/src/format_v2/parquet/parquet_scan.cpp +++ b/be/src/format_v2/parquet/parquet_scan.cpp @@ -46,6 +46,7 @@ #include "format_v2/parquet/reader/native/column_chunk_reader.h" #include "format_v2/parquet/reader/native_column_reader.h" #include "format_v2/parquet/reader/row_position_column_reader.h" +#include "runtime/runtime_state.h" #include "util/defer_op.h" #include "util/time.h" @@ -587,14 +588,7 @@ void update_counter_if_not_null(RuntimeProfile::Counter* counter, int64_t value) uint16_t apply_filter_to_selection(const IColumn::Filter& filter, SelectionVector* selection, uint16_t selected_rows) { - uint16_t new_selected_rows = 0; - for (uint16_t selection_idx = 0; selection_idx < selected_rows; ++selection_idx) { - const auto row_idx = selection->get_index(selection_idx); - if (filter[row_idx] != 0) { - selection->set_index(new_selected_rows++, static_cast(row_idx)); - } - } - return new_selected_rows; + return cast_set(selection->compact_with_row_filter(filter.data(), selected_rows)); } Status execute_compact_filter_conjuncts(const VExprContextSPtrs& conjuncts, size_t rows, @@ -752,14 +746,8 @@ uint16_t apply_compact_filter_to_selection(const IColumn::Filter& filter, SelectionVector* selection, uint16_t selected_rows) { DORIS_CHECK(selection != nullptr); DORIS_CHECK(filter.size() == selected_rows); - uint16_t new_selected_rows = 0; - for (uint16_t selection_idx = 0; selection_idx < selected_rows; ++selection_idx) { - if (filter[selection_idx] != 0) { - selection->set_index(new_selected_rows++, static_cast( - selection->get_index(selection_idx))); - } - } - return new_selected_rows; + return cast_set( + selection->compact_with_selection_filter(filter.data(), selected_rows)); } IColumn::Filter selection_to_filter(const SelectionVector& selection, uint16_t selected_rows, @@ -855,6 +843,7 @@ void ParquetScanScheduler::set_plan(RowGroupScanPlan plan) { _row_group_plans = std::move(plan.row_groups); _condition_cache_filtered_rows = 0; _predicate_filtered_rows = 0; + _remaining_plans_need_replanning = false; reset(); } @@ -908,6 +897,40 @@ void ParquetScanScheduler::reset() { reset_current_row_group(); } +void ParquetScanScheduler::set_scan_request(std::shared_ptr request) { + DORIS_CHECK(request != nullptr); + _active_request = std::move(request); + _pending_request.reset(); + _predicate_schedule_request = nullptr; +} + +void ParquetScanScheduler::queue_scan_request(std::shared_ptr request) { + DORIS_CHECK(request != nullptr); + _pending_request = std::move(request); +} + +void ParquetScanScheduler::activate_pending_scan_request_at_row_group_boundary() { + if (_has_current_row_group || !_pending_predicate_selection.empty() || + _pending_request == nullptr) { + return; + } + // Column readers and predicate schedules retain request-derived state for one row group. Swap + // only after they are gone; the refreshed request may promote a lazy column to a predicate. + _active_request = std::move(_pending_request); + _predicate_schedule_request = nullptr; + // Footer plans and adaptive ordering describe the previous predicate snapshot. Reusing either + // after a late runtime filter would miss pruning or bias the new predicate order with stale data. + _remaining_plans_need_replanning = true; + _predicate_schedule = {}; + _predicate_positions_scratch.clear(); + _predicate_indices_by_position_scratch.clear(); + _materialized_predicate_positions_scratch.clear(); + _ordered_predicate_positions_scratch.clear(); + _predicate_runtime_stats.clear(); + _predicate_batch_sequence = 0; + _predicate_survival_ratio = -1; +} + void ParquetScanScheduler::reset_current_row_group() { // RuntimeProfile updates are amortized on the batch path, but a row-group transition destroys // the reader tree. Force the final delta out before clearing it so short row groups and early @@ -1053,6 +1076,27 @@ Status ParquetScanScheduler::open_next_row_group( file_context.reset_random_access_ranges(); _current_merge_range_active = false; ParquetPruningStats deferred_stats; + if (_remaining_plans_need_replanning) { + // A refreshed projection may require different dictionary, Bloom, or page-index + // metadata. Preserve already-safe selected ranges, but rebuild every request-shaped + // artifact before opening this row group. + candidate_plan.expensive_pruning_pending = true; + candidate_plan.page_skip_plans.clear(); + candidate_plan.offset_indexes.clear(); + const std::vector candidate {candidate_plan.row_group_id}; + std::vector footer_selected; + RETURN_IF_ERROR(select_row_groups_by_metadata( + file_context.native_metadata->to_thrift(), file_schema, request, &candidate, + &footer_selected, _enable_bloom_filter, &deferred_stats, _timezone, + _runtime_state, &file_context, _scan_profile.column_reader_profile, + ParquetMetadataProbeMode::FOOTER_ONLY)); + if (footer_selected.empty()) { + if (_parquet_profile != nullptr) { + _parquet_profile->update_deferred_pruning_stats(deferred_stats, false); + } + continue; + } + } bool selected = false; RETURN_IF_ERROR(finalize_native_row_group_read_plan( *file_context.native_metadata, file_schema, request, _enable_bloom_filter, @@ -2706,11 +2750,12 @@ void ParquetScanScheduler::mark_condition_cache_granules(const SelectionVector& Status ParquetScanScheduler::read_next_batch( ParquetFileContext& file_context, - const std::vector>& file_schema, - const format::FileScanRequest& request, Block* file_block, size_t* rows, bool* eof) { + const std::vector>& file_schema, Block* file_block, + size_t* rows, bool* eof) { + DORIS_CHECK(_active_request != nullptr); *rows = 0; if (!_pending_predicate_selection.empty()) { - RETURN_IF_ERROR(materialize_pending_predicate_batch(request, file_block, rows)); + RETURN_IF_ERROR(materialize_pending_predicate_batch(*_active_request, file_block, rows)); *eof = false; return Status::OK(); } @@ -2731,9 +2776,10 @@ Status ParquetScanScheduler::read_next_batch( }; while (true) { if (!_has_current_row_group) { + activate_pending_scan_request_at_row_group_boundary(); bool has_row_group = false; - RETURN_IF_ERROR( - open_next_row_group(file_context, file_schema, request, &has_row_group)); + RETURN_IF_ERROR(open_next_row_group(file_context, file_schema, *_active_request, + &has_row_group)); if (!has_row_group) { *eof = true; return Status::OK(); @@ -2769,8 +2815,9 @@ Status ParquetScanScheduler::read_next_batch( const int64_t physical_rows_read = batch_rows; const int64_t batch_first_file_row = _current_row_group_first_row + _current_row_group_rows_read; - RETURN_IF_ERROR(read_current_row_group_batch(file_context, file_schema, batch_rows, request, - batch_first_file_row, file_block, rows)); + RETURN_IF_ERROR(read_current_row_group_batch(file_context, file_schema, batch_rows, + *_active_request, batch_first_file_row, + file_block, rows)); _current_row_group_rows_read += physical_rows_read; _current_range_rows_read += physical_rows_read; if (_current_range_rows_read >= current_range.length) { diff --git a/be/src/format_v2/parquet/parquet_scan.h b/be/src/format_v2/parquet/parquet_scan.h index f0e202d99422a8..474963ed5d9aa3 100644 --- a/be/src/format_v2/parquet/parquet_scan.h +++ b/be/src/format_v2/parquet/parquet_scan.h @@ -188,6 +188,8 @@ class ParquetScanScheduler { _enable_strict_mode = enable_strict_mode; } void set_runtime_state(RuntimeState* runtime_state) { _runtime_state = runtime_state; } + void set_scan_request(std::shared_ptr request); + void queue_scan_request(std::shared_ptr request); // Release row-group readers before the owning RuntimeProfile is reported. Native readers // publish their accumulated page/decode statistics from their destructor. void close() { reset_current_row_group(); } @@ -204,13 +206,13 @@ class ParquetScanScheduler { Status read_next_batch(ParquetFileContext& file_context, const std::vector>& file_schema, - const format::FileScanRequest& request, Block* file_block, size_t* rows, - bool* eof); + Block* file_block, size_t* rows, bool* eof); private: static constexpr size_t PROFILE_FLUSH_BATCH_INTERVAL = 16; void reset_current_row_group(); + void activate_pending_scan_request_at_row_group_boundary(); void flush_current_reader_profiles(); bool finish_current_reader_batch_profiles(); const detail::PredicateConjunctSchedule& predicate_conjunct_schedule( @@ -310,6 +312,9 @@ class ParquetScanScheduler { SelectionVector _selection; std::vector _read_column_positions_scratch; const format::FileScanRequest* _predicate_schedule_request = nullptr; + std::shared_ptr _active_request; + std::shared_ptr _pending_request; + bool _remaining_plans_need_replanning = false; detail::PredicateConjunctSchedule _predicate_schedule; std::vector _predicate_positions_scratch; std::unordered_map _predicate_indices_by_position_scratch; diff --git a/be/src/format_v2/parquet/reader/native_column_reader.cpp b/be/src/format_v2/parquet/reader/native_column_reader.cpp index 60eb0dc3d92404..6b686597fa600e 100644 --- a/be/src/format_v2/parquet/reader/native_column_reader.cpp +++ b/be/src/format_v2/parquet/reader/native_column_reader.cpp @@ -738,7 +738,6 @@ Status NativeColumnReader::select_with_fixed_width_filter( DORIS_CHECK(used_filter != nullptr); DORIS_CHECK(execution_kind != nullptr); RETURN_IF_ERROR(validate_selected_span(batch_rows)); - RETURN_IF_ERROR(selection.verify(selected_rows, batch_rows)); const uint8_t* filter_data = nullptr; RETURN_IF_ERROR(selection.materialize_filter(selected_rows, batch_rows, &filter_data)); int64_t rows_read = 0; @@ -768,7 +767,6 @@ Status NativeColumnReader::select_with_runtime_filter( DORIS_CHECK(row_filter != nullptr); DORIS_CHECK(used_filter != nullptr); RETURN_IF_ERROR(validate_selected_span(batch_rows)); - RETURN_IF_ERROR(selection.verify(selected_rows, batch_rows)); row_filter->clear(); *used_filter = false; if (_nested || conjuncts.empty() || !std::ranges::all_of(conjuncts, [&](const auto& conjunct) { diff --git a/be/src/format_v2/parquet/selection_vector.h b/be/src/format_v2/parquet/selection_vector.h index 033478875fad95..99c854d8b4e896 100644 --- a/be/src/format_v2/parquet/selection_vector.h +++ b/be/src/format_v2/parquet/selection_vector.h @@ -68,17 +68,19 @@ class SelectionVector { _data = data; _size = count; _identity = data == nullptr; + _mutable_data_exposed = data != nullptr; ++_generation; } void resize(size_t count) { - _owned.resize(count); - _data = _owned.data(); + // Identity is the overwhelmingly common initial state. Keep it implicit until a caller + // actually changes an index, avoiding one write per source row for every scanner batch. + // Scanner batches repeatedly reuse this object. Retaining the initialized high-water mark + // avoids value-initializing the entire scratch vector before every sparse compaction. + _data = nullptr; _size = count; - for (size_t idx = 0; idx < count; ++idx) { - _data[idx] = static_cast(idx); - } _identity = true; + _mutable_data_exposed = false; ++_generation; } @@ -87,6 +89,7 @@ class SelectionVector { _data = nullptr; _size = 0; _identity = true; + _mutable_data_exposed = false; ++_generation; } @@ -95,9 +98,11 @@ class SelectionVector { bool is_set() const { return _data != nullptr; } Index* data() { + _materialize_identity(); // A mutable pointer can change indices without set_index(), so identity can no longer be // proven until resize() rebuilds it. This keeps the O(1) dense fast path conservative. _identity = false; + _mutable_data_exposed = true; ++_generation; return _data; } @@ -112,6 +117,7 @@ class SelectionVector { } void set_index(size_t idx, Index value) { + _materialize_identity(); _data[idx] = value; if (value != idx) { _identity = false; @@ -119,10 +125,18 @@ class SelectionVector { ++_generation; } + size_t compact_with_row_filter(const uint8_t* filter, size_t count) { + return _compact(filter, count, true); + } + + size_t compact_with_selection_filter(const uint8_t* filter, size_t count) { + return _compact(filter, count, false); + } + Status materialize_filter(size_t count, int64_t batch_rows, const uint8_t** filter) const { DORIS_CHECK(filter != nullptr); if (batch_rows >= 0 && std::cmp_equal(count, batch_rows) && _identity && - (_data == nullptr || count <= _size)) { + (_size == 0 || count <= _size)) { // A proven identity selection is equivalent to no FilterMap. Returning nullptr avoids // constructing and rescanning one dense byte per source row. *filter = nullptr; @@ -157,6 +171,17 @@ class SelectionVector { return Status::InvalidArgument("Parquet selection count {} exceeds vector size {}", count, _size); } + if (_data == nullptr && _size != 0 && count > _size) { + return Status::InvalidArgument("Parquet selection count {} exceeds vector size {}", + count, _size); + } + if (_identity) { + return Status::OK(); + } + if (!_mutable_data_exposed && _verified_generation == _generation && + _verified_count == count && _verified_batch_rows == batch_rows) { + return Status::OK(); + } size_t previous = 0; for (size_t idx = 0; idx < count; ++idx) { const size_t current = get_index(idx); @@ -173,19 +198,87 @@ class SelectionVector { } previous = current; } + if (!_mutable_data_exposed) { + _verified_generation = _generation; + _verified_count = count; + _verified_batch_rows = batch_rows; + } return Status::OK(); } private: + void _materialize_identity() { + if (_data != nullptr) { + return; + } + if (_owned.size() < _size) { + _owned.resize(_size); + } + _data = _owned.data(); + for (size_t idx = 0; idx < _size; ++idx) { + _data[idx] = static_cast(idx); + } + } + + size_t _compact(const uint8_t* filter, size_t count, bool filter_uses_row_index) { + DORIS_CHECK(filter != nullptr); + DORIS_CHECK(count <= _size); + Index* source = _data; + if (_data == nullptr) { + if (_owned.size() < _size) { + _owned.resize(_size); + } + _data = _owned.data(); + // An implicit identity maps both filter coordinate systems to the same position. + // Specialize this first compaction so split batches do not pay source/coordinate + // branches for every row after already avoiding identity materialization. + size_t output = 0; + while (output < count && filter[output] != 0) { + _data[output] = static_cast(output); + ++output; + } + bool remains_identity = true; + for (size_t position = output; position < count; ++position) { + if (filter[position] != 0) { + _data[output++] = static_cast(position); + remains_identity = false; + } + } + _identity = remains_identity; + ++_generation; + return output; + } + size_t output = 0; + bool remains_identity = true; + for (size_t position = 0; position < count; ++position) { + const Index row = source == nullptr ? static_cast(position) : source[position]; + const size_t filter_position = filter_uses_row_index ? row : position; + if (filter[filter_position] != 0) { + _data[output] = row; + remains_identity &= row == output; + ++output; + } + } + // Compaction is one logical mutation. Invalidating caches once is important when several + // predicates successively refine a wide batch. + _identity = remains_identity; + ++_generation; + return output; + } + std::vector _owned; Index* _data = nullptr; size_t _size = 0; bool _identity = true; + bool _mutable_data_exposed = false; uint64_t _generation = 0; mutable std::vector _filter; mutable uint64_t _filter_generation = std::numeric_limits::max(); mutable size_t _filter_count = 0; mutable int64_t _filter_batch_rows = -1; + mutable uint64_t _verified_generation = std::numeric_limits::max(); + mutable size_t _verified_count = 0; + mutable int64_t _verified_batch_rows = -1; }; inline void selection_to_ranges(const SelectionVector& selection, uint16_t selected_rows, diff --git a/be/src/format_v2/table/hudi_reader.cpp b/be/src/format_v2/table/hudi_reader.cpp index ee24d0f9ad7d02..838d98b5e52e14 100644 --- a/be/src/format_v2/table/hudi_reader.cpp +++ b/be/src/format_v2/table/hudi_reader.cpp @@ -80,6 +80,18 @@ Status HudiHybridReader::prepare_split(const format::SplitReadOptions& options) return _current_split_reader->prepare_split(options); } +Status HudiHybridReader::refresh_conjuncts(VExprContextSPtrs conjuncts) { + RETURN_IF_ERROR(format::TableReader::refresh_conjuncts(std::move(conjuncts))); + if (_current_split_reader == nullptr) { + return Status::OK(); + } + VExprContextSPtrs child_conjuncts; + RETURN_IF_ERROR(_clone_conjuncts(&child_conjuncts)); + // The hybrid wrapper owns no physical reader; forward a clone so the active child, rather than + // only the wrapper snapshot, observes late predicates for the remainder of this split. + return _current_split_reader->refresh_conjuncts(std::move(child_conjuncts)); +} + Status HudiHybridReader::get_block(Block* block, bool* eos) { DORIS_CHECK(_current_split_reader != nullptr); return _current_split_reader->get_block(block, eos); diff --git a/be/src/format_v2/table/hudi_reader.h b/be/src/format_v2/table/hudi_reader.h index dbb6f5e8231043..c06e1b238b62c4 100644 --- a/be/src/format_v2/table/hudi_reader.h +++ b/be/src/format_v2/table/hudi_reader.h @@ -59,6 +59,7 @@ class HudiHybridReader final : public format::TableReader { Status init(format::TableReadOptions&& options) override; Status prepare_split(const format::SplitReadOptions& options) override; + Status refresh_conjuncts(VExprContextSPtrs conjuncts) override; Status get_block(Block* block, bool* eos) override; bool current_split_pruned() const override; bool current_split_uses_metadata_count() const override; diff --git a/be/src/format_v2/table/paimon_reader.cpp b/be/src/format_v2/table/paimon_reader.cpp index a3f4092a470263..93183af178babb 100644 --- a/be/src/format_v2/table/paimon_reader.cpp +++ b/be/src/format_v2/table/paimon_reader.cpp @@ -115,6 +115,18 @@ Status PaimonHybridReader::prepare_split(const format::SplitReadOptions& options return _current_split_reader->prepare_split(options); } +Status PaimonHybridReader::refresh_conjuncts(VExprContextSPtrs conjuncts) { + RETURN_IF_ERROR(format::TableReader::refresh_conjuncts(std::move(conjuncts))); + if (_current_split_reader == nullptr) { + return Status::OK(); + } + VExprContextSPtrs child_conjuncts; + RETURN_IF_ERROR(_clone_conjuncts(&child_conjuncts)); + // The hybrid wrapper owns no physical reader; forward a clone so the active child, rather than + // only the wrapper snapshot, observes late predicates for the remainder of this split. + return _current_split_reader->refresh_conjuncts(std::move(child_conjuncts)); +} + Status PaimonHybridReader::get_block(Block* block, bool* eos) { DORIS_CHECK(_current_split_reader != nullptr); return _current_split_reader->get_block(block, eos); diff --git a/be/src/format_v2/table/paimon_reader.h b/be/src/format_v2/table/paimon_reader.h index 823fa6540d280c..8570f2efba624e 100644 --- a/be/src/format_v2/table/paimon_reader.h +++ b/be/src/format_v2/table/paimon_reader.h @@ -65,6 +65,7 @@ class PaimonHybridReader final : public format::TableReader { Status init(format::TableReadOptions&& options) override; Status prepare_split(const format::SplitReadOptions& options) override; + Status refresh_conjuncts(VExprContextSPtrs conjuncts) override; Status get_block(Block* block, bool* eos) override; bool current_split_pruned() const override; bool current_split_uses_metadata_count() const override; diff --git a/be/src/format_v2/table_reader.cpp b/be/src/format_v2/table_reader.cpp index 8a5da295fddd4a..be4fe0e406e217 100644 --- a/be/src/format_v2/table_reader.cpp +++ b/be/src/format_v2/table_reader.cpp @@ -686,6 +686,8 @@ Status TableReader::init(TableReadOptions&& options) { ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "PushDownAggTime", table_profile, 1); _profile.open_reader_timer = ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "OpenReaderTime", table_profile, 1); + _profile.refresh_conjuncts_timer = ADD_CHILD_TIMER_WITH_LEVEL( + _scanner_profile, "RefreshConjunctsTime", table_profile, 1); _profile.runtime_filter_partition_prune_timer = ADD_CHILD_TIMER_WITH_LEVEL( _scanner_profile, "FileScannerRuntimeFilterPartitionPruningTime", table_profile, 1); _profile.runtime_filter_partition_pruned_range_counter = ADD_CHILD_COUNTER_WITH_LEVEL( @@ -703,6 +705,8 @@ Status TableReader::init(TableReadOptions&& options) { _scanner_profile, "FileReaderCreateColumnMapperTime", file_reader_profile, 1); _profile.file_reader_open_timer = ADD_CHILD_TIMER_WITH_LEVEL( _scanner_profile, "FileReaderOpenTime", file_reader_profile, 1); + _profile.file_reader_refresh_timer = ADD_CHILD_TIMER_WITH_LEVEL( + _scanner_profile, "FileReaderRefreshScanRequestTime", file_reader_profile, 1); _profile.file_reader_get_block_timer = ADD_CHILD_TIMER_WITH_LEVEL( _scanner_profile, "FileReaderGetBlockTime", file_reader_profile, 1); _profile.file_reader_aggregate_timer = ADD_CHILD_TIMER_WITH_LEVEL( @@ -765,6 +769,112 @@ Status TableReader::_build_table_filters_from_conjuncts() { return Status::OK(); } +namespace { + +bool same_scan_projection(const LocalColumnIndex& lhs, const LocalColumnIndex& rhs) { + if (lhs.index != rhs.index || lhs.project_all_children != rhs.project_all_children || + lhs.children.size() != rhs.children.size()) { + return false; + } + for (size_t index = 0; index < lhs.children.size(); ++index) { + if (!same_scan_projection(lhs.children[index], rhs.children[index])) { + return false; + } + } + return true; +} + +const LocalColumnIndex* find_scan_projection(const FileScanRequest& request, + LocalColumnId column_id) { + const auto find_by_id = [column_id](const std::vector& projections) { + return std::ranges::find_if(projections, [column_id](const LocalColumnIndex& projection) { + return projection.column_id() == column_id; + }); + }; + auto it = find_by_id(request.predicate_columns); + if (it != request.predicate_columns.end()) { + return &*it; + } + it = find_by_id(request.non_predicate_columns); + return it == request.non_predicate_columns.end() ? nullptr : &*it; +} + +bool same_physical_scan_layout(const FileScanRequest& lhs, const FileScanRequest& rhs) { + if (lhs.local_positions != rhs.local_positions) { + return false; + } + for (const auto& [column_id, _] : lhs.local_positions) { + const auto* lhs_projection = find_scan_projection(lhs, column_id); + const auto* rhs_projection = find_scan_projection(rhs, column_id); + if (lhs_projection == nullptr || rhs_projection == nullptr || + !same_scan_projection(*lhs_projection, *rhs_projection)) { + return false; + } + } + return true; +} + +} // namespace + +Status TableReader::refresh_conjuncts(VExprContextSPtrs conjuncts) { + SCOPED_TIMER(_profile.total_timer); + SCOPED_TIMER(_profile.refresh_conjuncts_timer); + _conjuncts = std::move(conjuncts); + if (_data_reader.reader == nullptr) { + // The split is prepared but its physical reader has not opened yet. open_reader() will use + // this newest snapshot directly, so no pending request is needed. + return Status::OK(); + } + if (!_data_reader.reader->supports_scan_request_refresh()) { + return Status::OK(); + } + + RETURN_IF_ERROR(_build_table_filters_from_conjuncts()); + // create_scan_request() rebuilds mapping projections in place. Build late predicates with an + // isolated mapper so the active row group cannot observe an unprepared or incompatible mapper + // before its physical request reaches the reader's safe activation boundary. + auto refreshed_mapper = _data_reader.reader->create_column_mapper(_mapper_options); + DORIS_CHECK(refreshed_mapper != nullptr); + RETURN_IF_ERROR(refreshed_mapper->create_mapping(_projected_columns, _partition_values, + _data_reader.file_schema)); + auto refreshed_request = std::make_shared(); + RETURN_IF_ERROR(refreshed_mapper->create_scan_request( + _table_filters, _projected_columns, refreshed_request.get(), _runtime_state, + _file_scan_request == nullptr ? nullptr : &_file_scan_request->local_positions)); + // A refresh does not prove that every future runtime filter has arrived. Keep carrier values + // available whenever the split started with pending filters. + if (_push_down_agg_type == TPushAggOp::type::COUNT && _push_down_count_columns.has_value() && + _push_down_count_columns->empty() && _all_runtime_filters_applied_for_split) { + for (const auto& column : refreshed_request->non_predicate_columns) { + refreshed_request->count_star_placeholder_columns.push_back(column.column_id()); + } + } + RETURN_IF_ERROR(customize_file_scan_request(refreshed_request.get())); + if (_file_scan_request == nullptr || + !same_physical_scan_layout(*refreshed_request, *_file_scan_request)) { + // A reader cannot reinterpret columns already materialized with another block layout. + // Keep scanner-level filtering as the correctness fallback for hidden slots or nested + // projections instead of switching an incompatible physical shape mid-file. + return Status::OK(); + } + RETURN_IF_ERROR(_open_local_filter_exprs(*refreshed_request)); + + if (_condition_cache_ctx != nullptr && !_condition_cache_ctx->is_hit) { + // Rows before and after a late RF were evaluated by different predicate snapshots. Such a + // partial MISS bitmap must never be published under either snapshot's cache key. + _condition_cache = nullptr; + _condition_cache_ctx = nullptr; + _data_reader.reader->set_condition_cache_context(nullptr); + } + { + SCOPED_TIMER(_profile.file_reader_total_timer); + SCOPED_TIMER(_profile.file_reader_refresh_timer); + RETURN_IF_ERROR(_data_reader.reader->queue_scan_request(refreshed_request)); + } + _file_scan_request = std::move(refreshed_request); + return Status::OK(); +} + Status TableReader::_open_local_filter_exprs(const FileScanRequest& file_request) { RowDescriptor row_desc; for (const auto& conjunct : file_request.conjuncts) { diff --git a/be/src/format_v2/table_reader.h b/be/src/format_v2/table_reader.h index 6d679ec2d797cf..9ff2641f581b92 100644 --- a/be/src/format_v2/table_reader.h +++ b/be/src/format_v2/table_reader.h @@ -116,6 +116,7 @@ struct ReadProfile { RuntimeProfile::Counter* create_reader_timer = nullptr; RuntimeProfile::Counter* pushdown_agg_timer = nullptr; RuntimeProfile::Counter* open_reader_timer = nullptr; + RuntimeProfile::Counter* refresh_conjuncts_timer = nullptr; RuntimeProfile::Counter* runtime_filter_partition_prune_timer = nullptr; RuntimeProfile::Counter* runtime_filter_partition_pruned_range_counter = nullptr; RuntimeProfile::Counter* close_timer = nullptr; @@ -124,6 +125,7 @@ struct ReadProfile { RuntimeProfile::Counter* file_reader_schema_timer = nullptr; RuntimeProfile::Counter* file_reader_mapper_timer = nullptr; RuntimeProfile::Counter* file_reader_open_timer = nullptr; + RuntimeProfile::Counter* file_reader_refresh_timer = nullptr; RuntimeProfile::Counter* file_reader_get_block_timer = nullptr; RuntimeProfile::Counter* file_reader_aggregate_timer = nullptr; RuntimeProfile::Counter* file_reader_close_timer = nullptr; @@ -223,6 +225,10 @@ class TableReader { // 2. Parse delete predicates from split/task information, which will be used for later dynamic filtering and delete handling. virtual Status prepare_split(const SplitReadOptions& options); + // Refresh row-level predicates for an already prepared split. Physical readers that support + // this operation decide the safe boundary at which the new immutable request becomes active. + virtual Status refresh_conjuncts(VExprContextSPtrs conjuncts); + virtual bool current_split_pruned() const { return _current_split_pruned; } virtual bool current_split_uses_metadata_count() const { return _current_split_uses_metadata_count; @@ -453,8 +459,11 @@ class TableReader { // marker is independent of aggregate eligibility: with position deletes, for example, // metadata COUNT must fall back to reading rows, but an arbitrary unsupported TIME_MILLIS // placeholder still must not be validated or decoded merely to carry the surviving count. + // Pending runtime filters may later target this retained slot, so placeholder values are + // safe only after every filter for the split has arrived. if (_push_down_agg_type == TPushAggOp::type::COUNT && - _push_down_count_columns.has_value() && _push_down_count_columns->empty()) { + _push_down_count_columns.has_value() && _push_down_count_columns->empty() && + _all_runtime_filters_applied_for_split) { file_request->count_star_placeholder_columns.reserve( file_request->non_predicate_columns.size()); for (const auto& column : file_request->non_predicate_columns) { @@ -465,6 +474,7 @@ class TableReader { RETURN_IF_ERROR(_open_local_filter_exprs(*file_request)); _data_reader.file_block_layout.clear(); _data_reader.block_template.clear(); + _file_scan_request.reset(); _data_reader.file_block_layout.resize(file_request->local_positions.size()); // 4. Build file block layout from file schema and column mapping. The layout describes @@ -517,7 +527,8 @@ class TableReader { SCOPED_TIMER(_profile.file_reader_open_timer); RETURN_IF_ERROR(_data_reader.reader->open(file_request)); } - RETURN_IF_ERROR(_init_reader_condition_cache(*file_request)); + _file_scan_request = std::move(file_request); + RETURN_IF_ERROR(_init_reader_condition_cache(*_file_scan_request)); return Status::OK(); } @@ -763,6 +774,7 @@ class TableReader { _data_reader.file_schema.clear(); _data_reader.file_block_layout.clear(); _data_reader.block_template.clear(); + _file_scan_request.reset(); _current_task.reset(); _current_file_description.reset(); _current_reader_reached_eof = false; @@ -1877,6 +1889,9 @@ class TableReader { Block block_template; }; DataReader _data_reader; + // Latest immutable request queued to the physical reader. The file-block layout remains fixed + // for the split even while predicates are refreshed at a reader-defined granule boundary. + std::shared_ptr _file_scan_request; std::vector _projected_columns; std::unique_ptr _current_task; std::optional _current_file_description; diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h index 3ef113c1fd1097..3102cc8529613a 100644 --- a/be/src/runtime/query_context.h +++ b/be/src/runtime/query_context.h @@ -325,7 +325,6 @@ class QueryContext : public std::enable_shared_from_this { void _init_query_mem_tracker(); std::unordered_map _runtime_predicates; - std::unique_ptr _runtime_filter_mgr; const TQueryOptions _query_options; diff --git a/be/test/exprs/expr_zonemap_filter_test.cpp b/be/test/exprs/expr_zonemap_filter_test.cpp index 1d80f40531e663..668a528b6a1695 100644 --- a/be/test/exprs/expr_zonemap_filter_test.cpp +++ b/be/test/exprs/expr_zonemap_filter_test.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -809,10 +810,12 @@ TEST(ExprZonemapFilterTest, DirectInPredicateMaterializesStringSetForZonemap) { direct_in_expr.add_child(slot); ASSERT_TRUE(direct_in_expr._materialize_for_zonemap_filter().ok()); - EXPECT_TRUE(direct_in_expr._zonemap_materialized); - EXPECT_EQ(2, direct_in_expr._seg_filter_values.size()); - EXPECT_EQ(Field::create_field("aaa"), direct_in_expr._seg_filter_min); - EXPECT_EQ(Field::create_field("zzz"), direct_in_expr._seg_filter_max); + EXPECT_TRUE(direct_in_expr._pruning_state->zonemap_materialized); + EXPECT_EQ(2, direct_in_expr._pruning_state->seg_filter_values.size()); + EXPECT_EQ(Field::create_field("aaa"), + direct_in_expr._pruning_state->seg_filter_min); + EXPECT_EQ(Field::create_field("zzz"), + direct_in_expr._pruning_state->seg_filter_max); } TEST(ExprZonemapFilterTest, DirectInPredicateMaterializesZonemapValuesDuringPrepare) { @@ -841,11 +844,34 @@ TEST(ExprZonemapFilterTest, DirectInPredicateMaterializesZonemapValuesDuringPrep VExprContext context(direct_in_expr); ASSERT_TRUE(context.prepare(&runtime_state, row_desc).ok()); - EXPECT_TRUE(direct_in_expr->_zonemap_materialized); + EXPECT_TRUE(direct_in_expr->_pruning_state->zonemap_materialized); EXPECT_TRUE(direct_in_expr->can_evaluate_zonemap_filter()); - EXPECT_EQ(2, direct_in_expr->_seg_filter_values.size()); - EXPECT_EQ(int_field(1), direct_in_expr->_seg_filter_min); - EXPECT_EQ(int_field(30), direct_in_expr->_seg_filter_max); + EXPECT_EQ(2, direct_in_expr->_pruning_state->seg_filter_values.size()); + EXPECT_EQ(int_field(1), direct_in_expr->_pruning_state->seg_filter_min); + EXPECT_EQ(int_field(30), direct_in_expr->_pruning_state->seg_filter_max); +} + +TEST(ExprZonemapFilterTest, DirectInPredicateDeepCloneReusesMaterializedPruningState) { + auto type = int_type(); + std::shared_ptr filter(create_set(PrimitiveType::TYPE_INT, false)); + int32_t low_value = 1; + int32_t high_value = 30; + filter->insert(&low_value); + filter->insert(&high_value); + + auto direct_in_expr = + std::make_shared(make_in_predicate_node(false, 1), filter, true); + direct_in_expr->add_child(make_slot(0, type)); + ASSERT_TRUE(direct_in_expr->_materialize_for_zonemap_filter().ok()); + + VExprSPtr cloned_expr; + ASSERT_TRUE(direct_in_expr->deep_clone(&cloned_expr).ok()); + auto cloned_direct_in = std::dynamic_pointer_cast(cloned_expr); + ASSERT_NE(cloned_direct_in, nullptr); + EXPECT_EQ(direct_in_expr->_pruning_state, cloned_direct_in->_pruning_state); + EXPECT_TRUE(cloned_direct_in->can_evaluate_zonemap_filter()); + EXPECT_EQ(ZoneMapFilterResult::kNoMatch, cloned_direct_in->evaluate_zonemap_filter( + make_context(make_int_zonemap(10, 20), type))); } TEST(ExprZonemapFilterTest, DirectInPredicateRewritesStringSetToInPredicate) { @@ -874,7 +900,7 @@ TEST(ExprZonemapFilterTest, DirectInPredicateSkipsMaterializationWhenSetTypeDiff direct_in_expr.add_child(slot); ASSERT_TRUE(direct_in_expr._materialize_for_zonemap_filter().ok()); - EXPECT_FALSE(direct_in_expr._zonemap_materialized); + EXPECT_FALSE(direct_in_expr._pruning_state->zonemap_materialized); VExprSPtr in_expr; EXPECT_FALSE(direct_in_expr.get_slot_in_expr(in_expr)); } diff --git a/be/test/format_v2/jni/jni_table_reader_test.cpp b/be/test/format_v2/jni/jni_table_reader_test.cpp index 2eee5548ad7e31..a59a4bbdf560dc 100644 --- a/be/test/format_v2/jni/jni_table_reader_test.cpp +++ b/be/test/format_v2/jni/jni_table_reader_test.cpp @@ -27,8 +27,11 @@ #include #include +#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_context.h" +#include "exprs/vslot_ref.h" #include "format/jni/jni_data_bridge.h" #include "io/io_common.h" @@ -238,6 +241,29 @@ TEST(JniTableReaderTest, AdaptiveProbeSetBeforePrepareControlsFirstJniOpen) { EXPECT_TRUE(reader.TEST_scanner_opened()); } +TEST(JniTableReaderTest, RefreshedConjunctIsReadyBeforeFilteringOpenScanner) { + FakeJniTableReader reader; + ASSERT_TRUE(init_reader(&reader, nullptr).ok()); + ASSERT_TRUE(reader.prepare_split({ + .partition_values = {}, + .conjuncts = std::nullopt, + .partition_prune_conjuncts = {}, + .all_runtime_filters_applied = true, + .condition_cache_digest = std::nullopt, + .cache = nullptr, + .current_range = {}, + .current_split_format = FileFormat::JNI, + .global_rowid_context = std::nullopt, + }) + .ok()); + + auto refreshed = VExprContext::create_shared( + VSlotRef::create_shared(0, 0, 0, std::make_shared(), "filter_column")); + ASSERT_FALSE(refreshed->root()->ready_status().ok()); + ASSERT_TRUE(reader.refresh_conjuncts({refreshed}).ok()); + EXPECT_TRUE(refreshed->root()->ready_status().ok()); +} + TEST(JniTableReaderTest, CommonLifecycleTimersContainJniLifecycleWork) { constexpr auto delay = std::chrono::milliseconds(8); RuntimeProfile profile("JniLifecycleContainment"); diff --git a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp index 72ccbf4484484c..2145b6ab60da43 100644 --- a/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp +++ b/be/test/format_v2/parquet/parquet_benchmark_scenarios_test.cpp @@ -127,6 +127,26 @@ TEST(ParquetBenchmarkScenariosTest, NestedSelectionCoversSparseParentSurvivors) } } +TEST(ParquetBenchmarkScenariosTest, SelectionMatrixCoversIdentityAndSuccessiveCompaction) { + const auto scenarios = selection_scenarios(); + EXPECT_EQ(scenarios.size(), size_t {25}); + EXPECT_TRUE(std::ranges::any_of(scenarios, [](const SelectionScenario& scenario) { + return scenario.operation == SelectionOperation::RESIZE_IDENTITY; + })); + for (const auto operation : + {SelectionOperation::ROW_FILTER, SelectionOperation::CASCADE_FILTER}) { + for (const int selectivity : {0, 1, 10, 50, 90, 100}) { + for (const auto pattern : {Pattern::CLUSTERED, Pattern::ALTERNATING}) { + EXPECT_TRUE(std::ranges::any_of(scenarios, [&](const SelectionScenario& scenario) { + return scenario.operation == operation && + scenario.selectivity_percent == selectivity && + scenario.pattern == pattern; + })) << "missing selection compaction shape"; + } + } + } +} + TEST(ParquetBenchmarkScenariosTest, ReaderMatrixCoversNullableSparseAndProjectionAxes) { const auto scenarios = reader_scenarios(); // Keep the exact count aligned with the upstream complex-residual scenario retained by rebase. diff --git a/be/test/format_v2/parquet/parquet_reader_control_test.cpp b/be/test/format_v2/parquet/parquet_reader_control_test.cpp index 0c70ddaf01e9f8..e21439e885570d 100644 --- a/be/test/format_v2/parquet/parquet_reader_control_test.cpp +++ b/be/test/format_v2/parquet/parquet_reader_control_test.cpp @@ -180,6 +180,7 @@ TEST(SelectionVectorTest, MaterializedFilterIsReusedUntilSelectionChanges) { TEST(SelectionVectorTest, IdentitySelectionDoesNotMaterializeFilter) { SelectionVector selection(4); + EXPECT_FALSE(selection.is_set()); const uint8_t* filter = reinterpret_cast(1); ASSERT_TRUE(selection.materialize_filter(4, 4, &filter).ok()); EXPECT_EQ(filter, nullptr); @@ -259,6 +260,36 @@ TEST(NativeNestedSelectionTest, PreservesPriorLevelsAcrossPageContinuation) { EXPECT_EQ(definition_levels, (std::vector {3, 3, 2})); } +TEST(SelectionVectorTest, BulkCompactionSupportsBothFilterCoordinates) { + SelectionVector selection(6); + const uint8_t row_filter[] = {0, 1, 1, 0, 1, 0}; + ASSERT_EQ(selection.compact_with_row_filter(row_filter, 6), 3); + EXPECT_EQ(selection.get_index(0), 1); + EXPECT_EQ(selection.get_index(1), 2); + EXPECT_EQ(selection.get_index(2), 4); + + const uint8_t compact_filter[] = {1, 0, 1}; + ASSERT_EQ(selection.compact_with_selection_filter(compact_filter, 3), 2); + EXPECT_EQ(selection.get_index(0), 1); + EXPECT_EQ(selection.get_index(1), 4); + EXPECT_TRUE(selection.verify(2, 6).ok()); +} + +TEST(SelectionVectorTest, BatchResetRetainsMaterializedScratchHighWaterMark) { + SelectionVector selection(6); + ASSERT_NE(selection.data(), nullptr); + const uint8_t first_filter[] = {0, 1, 1, 0, 1, 0}; + ASSERT_EQ(selection.compact_with_row_filter(first_filter, 6), 3); + + selection.resize(6); + const uint8_t second_filter[] = {0, 0, 0, 0, 0, 1}; + ASSERT_EQ(selection.compact_with_row_filter(second_filter, 6), 1); + EXPECT_EQ(selection.get_index(0), 5); + // Positions beyond the logical result remain reusable scratch. Clearing and resizing the + // owned vector would value-initialize this slot on every scanner batch. + EXPECT_EQ(selection.get_index(5), 5); +} + TEST(ParquetColumnReaderControlTest, BaseSelectUsesSkipReadRanges) { CursorColumnReader reader; SelectionVector selection(3); @@ -318,6 +349,30 @@ TEST(ParquetColumnReaderControlTest, SchedulerOrsPageCrossingOncePerBatch) { EXPECT_EQ(lazy_ptr->page_crossing_checks(), 1); } +TEST(ParquetColumnReaderControlTest, PendingRequestActivatesOnlyAtRowGroupBoundary) { + ParquetScanScheduler scheduler; + auto initial = std::make_shared(); + auto refreshed = std::make_shared(); + refreshed->predicate_only_columns.push_back(format::LocalColumnId(7)); + + scheduler.set_scan_request(initial); + scheduler._has_current_row_group = true; + scheduler.queue_scan_request(refreshed); + scheduler.activate_pending_scan_request_at_row_group_boundary(); + EXPECT_EQ(scheduler._active_request, initial); + + scheduler._has_current_row_group = false; + scheduler._predicate_survival_ratio = 0.5; + scheduler._predicate_batch_sequence = 3; + scheduler._predicate_runtime_stats.emplace(1, detail::AdaptivePredicateStats {}); + scheduler.activate_pending_scan_request_at_row_group_boundary(); + EXPECT_EQ(scheduler._active_request, refreshed); + EXPECT_TRUE(scheduler._remaining_plans_need_replanning); + EXPECT_EQ(scheduler._predicate_survival_ratio, -1); + EXPECT_EQ(scheduler._predicate_batch_sequence, 0); + EXPECT_TRUE(scheduler._predicate_runtime_stats.empty()); +} + TEST(ParquetColumnReaderControlTest, PendingOutputDrainsBeforePageCrossingSample) { ParquetScanScheduler scheduler; scheduler._batch_size = 1; diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp index 75733164ed268f..8ed63a7fe2845e 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/be/test/format_v2/parquet/parquet_scan_test.cpp @@ -1398,6 +1398,22 @@ void write_dictionary_int_pair_parquet_file(const std::string& file_path) { write_table(file_path, table, 6, true, false, false); } +void write_dictionary_int_pair_parquet_file(const std::string& file_path, + const std::vector& dictionary_values) { + std::vector scores(dictionary_values.size()); + for (size_t index = 0; index < scores.size(); ++index) { + scores[index] = static_cast((index + 1) * 10); + } + auto schema = arrow::schema({ + arrow::field("id", arrow::int32(), false), + arrow::field("score", arrow::int32(), false), + }); + auto table = arrow::Table::Make( + schema, {build_int32_array(dictionary_values), build_int32_array(scores)}); + write_table(file_path, table, static_cast(dictionary_values.size()), true, false, + false); +} + void write_dictionary_bigint_pair_parquet_file(const std::string& file_path) { auto schema = arrow::schema({ arrow::field("id", arrow::int64(), false), @@ -2268,6 +2284,49 @@ TEST_F(ParquetScanTest, NoRequestedColumnsReturnsRowsOnlyAcrossRowGroups) { EXPECT_EQ(total_rows, 6); } +TEST_F(ParquetScanTest, LateRequestReplansUnopenedRowGroupsWithFooterStatistics) { + write_int_pair_parquet_file(_file_path, 2); + RuntimeProfile profile("profile"); + auto reader = create_reader(0, -1, &profile); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + reader->set_batch_size(2); + + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + auto initial = std::make_shared(); + format::FileScanRequestBuilder initial_builder(initial.get()); + ASSERT_TRUE(initial_builder.add_non_predicate_column(format::LocalColumnId(0)).ok()); + ASSERT_TRUE(initial_builder.add_non_predicate_column(format::LocalColumnId(1)).ok()); + ASSERT_TRUE(reader->open(initial).ok()); + + Block first_block = build_file_block(schema); + size_t first_rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&first_block, &first_rows, &eof).ok()); + ASSERT_EQ(first_rows, 2); + EXPECT_EQ(int32_data_column(*first_block.get_by_position(0).column).get_data(), + (ColumnInt32::Container {1, 2})); + + auto refreshed = std::make_shared(); + format::FileScanRequestBuilder refreshed_builder(refreshed.get()); + ASSERT_TRUE(refreshed_builder.add_predicate_column(format::LocalColumnId(0)).ok()); + ASSERT_TRUE(refreshed_builder.add_non_predicate_column(format::LocalColumnId(1)).ok()); + refreshed->conjuncts.push_back(create_int32_zonemap_conjunct(0, Int32ZoneMapExpr::Op::GT, 4)); + ASSERT_TRUE(reader->queue_scan_request(refreshed).ok()); + + Block refreshed_block = build_file_block(schema); + size_t refreshed_rows = 0; + ASSERT_TRUE(reader->get_block(&refreshed_block, &refreshed_rows, &eof).ok()); + ASSERT_EQ(refreshed_rows, 2); + EXPECT_EQ(int32_data_column(*refreshed_block.get_by_position(0).column).get_data(), + (ColumnInt32::Container {5, 6})); + // The middle row group is rejected from footer statistics before any data page is decoded. + EXPECT_EQ(counter_value(profile, "RawRowsRead"), 4); + EXPECT_EQ(counter_value(profile, "RowGroupsFilteredByMinMax"), 1); + EXPECT_NE(profile.get_counter("RefreshScanRequestTime"), nullptr); +} + TEST_F(ParquetScanTest, PredicateColumnsFilterRoundByRound) { write_int_pair_parquet_file(_file_path, 6, false); RuntimeProfile profile("profile"); @@ -3163,17 +3222,82 @@ TEST_F(ParquetScanTest, PredicateOnlyDictionaryRangeSkipsTypedValueMaterializati conjunct->close(); } +TEST_F(ParquetScanTest, DictionaryFiltersAreBuiltFromEachReaderSnapshot) { + struct ScanResult { + std::vector scores; + int64_t typed_compare_columns = 0; + }; + + auto scan = [&](int32_t lower_bound) { + RuntimeProfile profile("profile"); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto reader = create_reader(0, -1, &profile); + EXPECT_TRUE(reader->init(&state).ok()); + + std::vector schema; + EXPECT_TRUE(reader->get_schema(&schema).ok()); + auto request = std::make_shared(); + format::FileScanRequestBuilder request_builder(request.get()); + EXPECT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok()); + EXPECT_TRUE(request_builder.add_non_predicate_column(format::LocalColumnId(1)).ok()); + request->predicate_only_columns.push_back(format::LocalColumnId(0)); + auto conjunct = + create_int32_function_conjunct(0, "gt", TExprOpcode::GT, lower_bound, false); + EXPECT_TRUE(conjunct->prepare(&state, RowDescriptor()).ok()); + EXPECT_TRUE(conjunct->open(&state).ok()); + request->conjuncts.push_back(conjunct); + EXPECT_TRUE(reader->open(request).ok()); + + ScanResult result; + bool eof = false; + while (!eof) { + Block block = build_file_block(schema); + size_t rows = 0; + EXPECT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + const auto& score_column = int32_data_column(*block.get_by_position(1).column); + for (size_t row = 0; row < rows; ++row) { + result.scores.push_back(score_column.get_element(row)); + } + } + result.typed_compare_columns = counter_value(profile, "DictFilterTypedCompareColumns"); + conjunct->close(); + EXPECT_TRUE(reader->close().ok()); + return result; + }; + + write_dictionary_int_pair_parquet_file(_file_path); + const auto first = scan(2); + EXPECT_EQ(first.scores, std::vector({30, 40, 50, 60})); + EXPECT_EQ(first.typed_compare_columns, 1); + + const auto repeated = scan(2); + EXPECT_EQ(repeated.scores, first.scores); + EXPECT_EQ(repeated.typed_compare_columns, 1); + + const auto changed_predicate = scan(3); + EXPECT_EQ(changed_predicate.scores, std::vector({40, 50, 60})); + EXPECT_EQ(changed_predicate.typed_compare_columns, 1); + + write_dictionary_int_pair_parquet_file(_file_path, {7, 1, 8, 2, 9, 3}); + const auto changed_dictionary = scan(2); + EXPECT_EQ(changed_dictionary.scores, std::vector({10, 30, 50, 60})); + EXPECT_EQ(changed_dictionary.typed_compare_columns, 1); +} + TEST_F(ParquetScanTest, PredicateOnlyDictionaryTopNUsesDictionaryIds) { write_dictionary_int_pair_parquet_file(_file_path); RuntimeProfile profile("profile"); - auto reader = create_reader(0, -1, &profile); RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto prepared = + create_topn_conjunct(&state, 0, make_nullable(std::make_shared()), + Field::create_field(3)); + ASSERT_NE(state.get_query_ctx(), nullptr); + auto reader = create_reader(0, -1, &profile); ASSERT_TRUE(reader->init(&state).ok()); std::vector schema; ASSERT_TRUE(reader->get_schema(&schema).ok()); - auto prepared = - create_topn_conjunct(&state, 0, schema[0].type, Field::create_field(3)); + ASSERT_TRUE(schema[0].type->equals(*prepared.conjunct->root()->children()[0]->data_type())); auto request = std::make_shared(); format::FileScanRequestBuilder request_builder(request.get()); ASSERT_TRUE(request_builder.add_predicate_column(format::LocalColumnId(0)).ok()); diff --git a/be/test/format_v2/table/hudi_reader_test.cpp b/be/test/format_v2/table/hudi_reader_test.cpp index 96126281744f5a..e75eee47be39d6 100644 --- a/be/test/format_v2/table/hudi_reader_test.cpp +++ b/be/test/format_v2/table/hudi_reader_test.cpp @@ -141,6 +141,18 @@ class SlowInitTableReader final : public TableReader { } }; +class RefreshTrackingTableReader final : public TableReader { +public: + Status prepare_split(const SplitReadOptions&) override { return Status::OK(); } + + Status refresh_conjuncts(VExprContextSPtrs conjuncts) override { + ++refresh_count; + return TableReader::refresh_conjuncts(std::move(conjuncts)); + } + + int refresh_count = 0; +}; + // 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. @@ -269,6 +281,39 @@ TEST(HudiHybridReaderTest, AggregatesConditionCacheHitsFromBothChildren) { EXPECT_EQ(reader.condition_cache_hit_count(), 9); } +TEST(HudiHybridReaderTest, ForwardsLatePredicatesToActiveChild) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + TFileScanRangeParams scan_params; + scan_params.__set_format_type(TFileFormatType::FORMAT_PARQUET); + hudi::HudiHybridReader reader; + RefreshTrackingTableReader* child = nullptr; + reader.TEST_set_child_reader_factories( + [&] { + auto tracking = std::make_unique(); + child = tracking.get(); + return tracking; + }, + [] { return std::make_unique(); }); + ASSERT_TRUE(reader.init({ + .projected_columns = {}, + .conjuncts = {}, + .format = FileFormat::PARQUET, + .scan_params = &scan_params, + .io_ctx = nullptr, + .runtime_state = &state, + .scanner_profile = nullptr, + }) + .ok()); + + SplitReadOptions split; + split.current_split_format = FileFormat::PARQUET; + split.current_range.__set_format_type(TFileFormatType::FORMAT_PARQUET); + ASSERT_TRUE(reader.prepare_split(split).ok()); + ASSERT_NE(child, nullptr); + ASSERT_TRUE(reader.refresh_conjuncts({}).ok()); + EXPECT_EQ(child->refresh_count, 1); +} + TEST(HudiHybridReaderTest, NativeCountStarReportsMetadataRowsThroughHybridReader) { const auto test_dir = std::filesystem::temp_directory_path() / "doris_hudi_hybrid_count_star_test"; diff --git a/be/test/format_v2/table/paimon_reader_test.cpp b/be/test/format_v2/table/paimon_reader_test.cpp index 6ebec512f42fdf..32b82ab12acbe3 100644 --- a/be/test/format_v2/table/paimon_reader_test.cpp +++ b/be/test/format_v2/table/paimon_reader_test.cpp @@ -79,6 +79,18 @@ class SlowInitTableReader final : public TableReader { } }; +class RefreshTrackingTableReader final : public TableReader { +public: + Status prepare_split(const SplitReadOptions&) override { return Status::OK(); } + + Status refresh_conjuncts(VExprContextSPtrs conjuncts) override { + ++refresh_count; + return TableReader::refresh_conjuncts(std::move(conjuncts)); + } + + int refresh_count = 0; +}; + class SplitFormatTrackingTableReader final : public TableReader { public: Status prepare_split(const SplitReadOptions& options) override { @@ -754,6 +766,38 @@ TEST(PaimonHybridReaderTest, AggregatesConditionCacheHitsFromBothChildren) { EXPECT_EQ(reader.condition_cache_hit_count(), 8); } +TEST(PaimonHybridReaderTest, ForwardsLatePredicatesToActiveChild) { + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto scan_params = make_local_parquet_scan_params(); + paimon::PaimonHybridReader reader; + RefreshTrackingTableReader* child = nullptr; + reader.TEST_set_child_reader_factories( + [&] { + auto tracking = std::make_unique(); + child = tracking.get(); + return tracking; + }, + [] { return std::make_unique(); }); + ASSERT_TRUE(reader.init({ + .projected_columns = {}, + .conjuncts = {}, + .format = FileFormat::PARQUET, + .scan_params = &scan_params, + .io_ctx = nullptr, + .runtime_state = &state, + .scanner_profile = nullptr, + }) + .ok()); + + SplitReadOptions split; + split.current_split_format = FileFormat::PARQUET; + split.current_range = make_paimon_native_range(TFileFormatType::FORMAT_PARQUET); + ASSERT_TRUE(reader.prepare_split(split).ok()); + ASSERT_NE(child, nullptr); + ASSERT_TRUE(reader.refresh_conjuncts({}).ok()); + EXPECT_EQ(child->refresh_count, 1); +} + TEST(PaimonHybridReaderTest, NativeCountColumnReportsMetadataRowsThroughHybridReader) { const auto test_dir = std::filesystem::temp_directory_path() / "doris_paimon_hybrid_count_column_test"; diff --git a/be/test/format_v2/table_reader_test.cpp b/be/test/format_v2/table_reader_test.cpp index bb0db9329a2a1c..f4bc1911d17b36 100644 --- a/be/test/format_v2/table_reader_test.cpp +++ b/be/test/format_v2/table_reader_test.cpp @@ -1050,6 +1050,7 @@ struct FakeFileReaderState { int init_count = 0; int open_count = 0; int close_count = 0; + int refresh_count = 0; int64_t total_rows = 2; int64_t aggregate_count = -1; int64_t condition_cache_base_granule = 0; @@ -1060,6 +1061,7 @@ struct FakeFileReaderState { bool stop_during_read = false; bool not_found_during_init = false; std::shared_ptr last_request; + std::shared_ptr pending_request; std::optional last_aggregate_request; std::shared_ptr condition_cache_ctx; std::shared_ptr io_ctx; @@ -1101,6 +1103,14 @@ class FakeFileReader final : public FileReader { return Status::OK(); } + bool supports_scan_request_refresh() const override { return true; } + + Status queue_scan_request(std::shared_ptr request) override { + _state->pending_request = std::move(request); + ++_state->refresh_count; + return Status::OK(); + } + Status get_block(Block* file_block, size_t* rows, bool* eof) override { DORIS_CHECK(file_block != nullptr); DORIS_CHECK(rows != nullptr); @@ -1220,6 +1230,12 @@ class FakeTableReader final : public TableReader { std::shared_ptr state) : _file_schema(std::move(file_schema)), _state(std::move(state)) {} + VExprContextSPtr TEST_mapping_projection(size_t index) const { + DORIS_CHECK(_data_reader.column_mapper != nullptr); + DORIS_CHECK_LT(index, _data_reader.column_mapper->mappings().size()); + return _data_reader.column_mapper->mappings()[index].projection; + } + protected: Status create_file_reader(std::unique_ptr* reader) override { DORIS_CHECK(reader != nullptr); @@ -1610,6 +1626,101 @@ TEST(TableReaderTest, PrepareSplitReplacesInitialConjunctSnapshot) { ASSERT_TRUE(reader.close().ok()); } +TEST(TableReaderTest, ActiveReaderQueuesRefreshedRuntimeFilterRequest) { + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", std::make_shared())); + file_schema.push_back(make_file_column(1, "value", std::make_shared())); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + projected_columns.push_back(make_table_column(1, "value", std::make_shared())); + set_name_identifiers(&projected_columns); + + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + RuntimeProfile profile("scanner"); + auto fake_state = std::make_shared(); + fake_state->eof_with_first_batch = false; + 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 = &profile, + }) + .ok()); + + SplitReadOptions split_options; + split_options.current_range.__set_path("fake-table-reader-input"); + ASSERT_TRUE(reader.prepare_split(split_options).ok()); + Block block = build_table_block(projected_columns); + bool eos = false; + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + ASSERT_FALSE(eos); + + VExprContextSPtrs refreshed {VExprContext::create_shared( + runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0, 1)))}; + ASSERT_TRUE(reader.refresh_conjuncts(std::move(refreshed)).ok()); + ASSERT_EQ(fake_state->refresh_count, 1); + ASSERT_NE(fake_state->pending_request, nullptr); + EXPECT_EQ(fake_state->pending_request->local_positions, + fake_state->last_request->local_positions); + EXPECT_EQ(projection_ids(fake_state->pending_request->predicate_columns), + std::vector({0})); + EXPECT_EQ(projection_ids(fake_state->pending_request->non_predicate_columns), + std::vector({1})); + ASSERT_EQ(fake_state->pending_request->conjuncts.size(), 1); + EXPECT_TRUE(fake_state->pending_request->conjuncts.front()->root()->is_rf_wrapper()); + EXPECT_NE(profile.get_counter("RefreshConjunctsTime"), nullptr); + EXPECT_NE(profile.get_counter("FileReaderRefreshScanRequestTime"), nullptr); + ASSERT_TRUE(reader.close().ok()); +} + +TEST(TableReaderTest, RefreshKeepsActiveMappingProjectionSnapshot) { + std::vector file_schema; + file_schema.push_back(make_file_column(0, "id", std::make_shared())); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + set_name_identifiers(&projected_columns); + + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + auto fake_state = std::make_shared(); + fake_state->eof_with_first_batch = false; + 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_options; + split_options.current_range.__set_path("fake-table-reader-input"); + ASSERT_TRUE(reader.prepare_split(split_options).ok()); + Block block = build_table_block(projected_columns); + bool eos = false; + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + ASSERT_FALSE(eos); + + const auto active_projection = reader.TEST_mapping_projection(0); + ASSERT_NE(active_projection, nullptr); + ASSERT_TRUE(active_projection->root()->ready_status().ok()); + VExprContextSPtrs refreshed {VExprContext::create_shared( + runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0, 1)))}; + ASSERT_TRUE(reader.refresh_conjuncts(std::move(refreshed)).ok()); + + EXPECT_EQ(reader.TEST_mapping_projection(0), active_projection); + EXPECT_TRUE(reader.TEST_mapping_projection(0)->root()->ready_status().ok()); + ASSERT_TRUE(reader.close().ok()); +} + TEST(TableReaderTest, RefreshedConjunctDisablesTableLevelCount) { std::vector file_schema; file_schema.push_back(make_file_column(0, "id", std::make_shared())); @@ -1692,9 +1803,64 @@ TEST(TableReaderTest, PendingRuntimeFilterDisablesTableLevelCount) { EXPECT_EQ(fake_state->open_count, 1); EXPECT_EQ(block.rows(), 2); ASSERT_NE(fake_state->last_request, nullptr); - // Aggregate pushdown is disabled while a runtime filter is pending, but COUNT(*) semantics do - // not change. The retained output slot remains a value-less placeholder during row fallback. - EXPECT_TRUE(fake_state->last_request->is_count_star_placeholder(LocalColumnId(0))); + // A pending runtime filter may later target the retained output slot. The fallback reader must + // keep its real values until the refreshed physical request reaches a row-group boundary. + EXPECT_FALSE(fake_state->last_request->is_count_star_placeholder(LocalColumnId(0))); + ASSERT_TRUE(reader.close().ok()); +} + +TEST(TableReaderTest, CountStarFallbackKeepsLateRuntimeFilterCarrierValues) { + const auto test_dir = + std::filesystem::temp_directory_path() / "doris_table_reader_count_star_late_rf_test"; + std::filesystem::remove_all(test_dir); + std::filesystem::create_directories(test_dir); + const auto file_path = (test_dir / "split.parquet").string(); + write_int_pair_parquet_file(file_path, {1, 2, 3, 4, 5, 6}, {10, 20, 30, 40, 50, 60}, + {"one", "two", "three", "four", "five", "six"}, 2); + + std::vector projected_columns; + projected_columns.push_back(make_table_column(0, "id", std::make_shared())); + set_name_identifiers(&projected_columns); + + TQueryOptions query_options; + query_options.__set_batch_size(2); + RuntimeState state {query_options, TQueryGlobals()}; + TableReader reader; + ASSERT_TRUE(reader.init({ + .projected_columns = projected_columns, + .conjuncts = {}, + .format = FileFormat::PARQUET, + .scan_params = nullptr, + .io_ctx = nullptr, + .runtime_state = &state, + .scanner_profile = nullptr, + .push_down_agg_type = TPushAggOp::type::COUNT, + .push_down_count_columns = std::vector {}, + }) + .ok()); + auto split_options = build_split_options(file_path); + split_options.all_runtime_filters_applied = false; + ASSERT_TRUE(reader.prepare_split(split_options).ok()); + + Block first_block = build_table_block(projected_columns); + bool eos = false; + ASSERT_TRUE(reader.get_block(&first_block, &eos).ok()); + ASSERT_EQ(first_block.rows(), 2); + EXPECT_EQ(assert_cast(expect_not_null_table_column(first_block, 0)) + .get_data(), + (ColumnInt32::Container {1, 2})); + + VExprContextSPtrs refreshed {VExprContext::create_shared( + runtime_filter_wrapper_expr(table_int32_greater_than_expr(0, 0, 4)))}; + ASSERT_TRUE(reader.refresh_conjuncts(std::move(refreshed)).ok()); + std::vector remaining_ids; + while (!eos) { + Block block = build_table_block(projected_columns); + ASSERT_TRUE(reader.get_block(&block, &eos).ok()); + const auto& ids = assert_cast(expect_not_null_table_column(block, 0)); + remaining_ids.insert(remaining_ids.end(), ids.get_data().begin(), ids.get_data().end()); + } + EXPECT_EQ(remaining_ids, std::vector({5, 6})); ASSERT_TRUE(reader.close().ok()); }