This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new a9641f1770f [enhancement](scan) Optimize Parquet V2 direct predicate
filtering (#66360)
a9641f1770f is described below
commit a9641f1770fc079f3b422ea36326fe10ca713ef7
Author: Gabriel <[email protected]>
AuthorDate: Mon Aug 3 20:56:05 2026 +0800
[enhancement](scan) Optimize Parquet V2 direct predicate filtering (#66360)
## Proposed changes
Backport the selected Parquet V2 direct-predicate filtering changes from
#66261 to `branch-4.1`, then address the review and performance findings
on this branch.
- keep identity selection-vector state implicit and compact selected
rows in bulk
- retain the selection scratch high-water mark across scanner batches
and specialize first compaction from implicit identity
- refresh late runtime-filter requests at safe row-group boundaries
- re-run footer-statistics pruning and reset adaptive predicate state
for unopened row groups after a refresh
- preserve real COUNT(*) carrier values while runtime filters are
pending
- initialize refreshed JNI predicates and attribute refresh work to
TableReader/FileReader/Parquet profiles
- preserve Hudi/Paimon child-reader predicate state
- remove query-scoped dictionary-filter cache state
- share immutable `VDirectInPredicate` pruning materialization across
split-local expression clones
- add correctness-checked selection and direct-IN lifecycle
microbenchmarks
## Test
- `./run-be-ut.sh --run
--filter='FileScannerV2Test.*:*Parquet*:*TableReaderTest.*:Hudi*ReaderTest.*:Paimon*ReaderTest.*:SelectionVectorTest.*:DictionaryFilterCostTest.*'
-j48`
- 639 tests from 47 test suites passed under ASAN
- targeted late-RF, COUNT(*), dictionary-snapshot, shared-IN-state, and
SelectionVector tests
- 19 tests from 5 test suites passed under ASAN
- Release benchmark build and smoke run
- expected registrations: 228 decoder, 92 kernel, 25 selection, 167
reader, and 8 expression-lifecycle cases
- all 25 selection and 8 expression-lifecycle cases executed with zero
benchmark errors
- `git diff --check`
## Selection-vector microbenchmark
The final benchmark source validates every surviving original row ID
after the timed region. Base, pre-fix PR, and final binaries use the
same benchmark source and Clang `-O3 -DNDEBUG -mavx2` on the same host.
Each comparison uses one pinned CPU, three warmups, eight adjacent
A-B-B-A quartets, and at least 0.3 seconds per invocation. The table
reports median paired CPU-time ratios; negative values are improvements.
| Operation | Final selectivity | Final vs pre-fix PR | Final vs branch
base |
|---|---:|---:|---:|
| Identity initialization | 100% | -15.23% | -99.12% |
| Row filter | 1% | -24.23% | -23.76% |
| Row filter | 50% | -16.10% | -34.50% |
| Row filter | 90% | -45.91% | -45.95% |
| Row filter | 100% | -31.72% | -17.32% |
| Successive filters | 1% | -33.25% | -35.79% |
| Successive filters | 50% | -29.80% | -35.10% |
| Successive filters | 90% | -25.27% | -25.16% |
| Successive filters | 100% | -24.93% | -23.72% |
All final-vs-base paired-ratio CVs are at most 5.85%. The previous
16.43%/59.94% dense row-filter regressions and 27.92%-61.05%
successive-filter regressions are no longer present. Retaining `_owned`
avoids repeated value initialization; the implicit-identity
specialization removes the remaining source/coordinate branches from the
first compaction.
## Direct-IN expression lifecycle microbenchmark
`FileScannerExpr/direct_in_clone_prepare_open` isolates deep-clone,
prepare, and open for an already prepared direct-IN runtime filter. Set
construction and the original fragment prepare/open are outside the
timed region. Shared and forced-rematerialization implementations run in
the same Release binary on one pinned CPU, with 10 repetitions and at
least 0.5 seconds per repetition.
| IN values | Rematerialize median | Shared median | Speedup |
|---:|---:|---:|---:|
| 128 | 207.470 us | 1.634 us | 126.9x |
| 1,024 | 1.674 ms | 1.642 us | 1,019.5x |
| 8,192 | 13.514 ms | 1.672 us | 8,082.2x |
| 65,536 | 108.337 ms | 1.650 us | 65,663.1x |
The shared path remains approximately constant because split clones
reuse the immutable, fragment-originated pruning state; the
rematerialization path scales linearly with runtime-filter cardinality.
## Reader-level microbenchmark
The earlier reader-level comparison used an identical Parquet-only
Release benchmark binary on the same host, with one pinned CPU, warm
cache, three warmups, A-B-B-A order, 10 repetitions, and at least one
second per repetition. It covers nullable INT32 predicate scans with a
lazy payload for PLAIN and dictionary encoding.
| Encoding | Selectivity | CPU time vs base | Wall time vs base |
|---|---:|---:|---:|
| Dictionary | 1% | +0.44% | +0.46% |
| Dictionary | 10% | +1.13% | +1.22% |
| Dictionary | 50% | +1.41% | +1.51% |
| Dictionary | 90% | +0.41% | +0.53% |
| PLAIN | 1% | -0.65% | -0.66% |
| PLAIN | 10% | +0.32% | +0.38% |
| PLAIN | 50% | -1.21% | -1.18% |
| PLAIN | 90% | -1.34% | -1.29% |
The reader-level point estimates span -1.34% to +1.41% CPU time with
mixed signs, so this benchmark did not detect a material aggregate
regression. It starts at `format::parquet::ParquetReader`; it does not
cover scanner scheduling or end-to-end SQL execution.
---
be/benchmark/benchmark_bit_pack.hpp | 10 +-
be/benchmark/benchmark_column_array_view.hpp | 2 +-
.../benchmark_column_array_view_distance.hpp | 2 +-
be/benchmark/benchmark_fastunion.hpp | 2 +-
be/benchmark/benchmark_hll_merge.hpp | 2 +-
be/benchmark/benchmark_main.cpp | 18 +--
be/benchmark/binary_cast_benchmark.hpp | 2 +-
be/benchmark/parquet/AGENTS.md | 52 ++++++-
be/benchmark/parquet/README.md | 38 +++++
.../parquet/benchmark_file_scanner_expr.hpp | 118 ++++++++++++++
.../parquet/benchmark_parquet_selection.hpp | 172 +++++++++++++++++++++
be/benchmark/parquet/parquet_benchmark_scenarios.h | 33 ++++
be/src/exec/scan/file_scanner_v2.cpp | 7 +
be/src/exec/scan/file_scanner_v2.h | 1 +
be/src/exprs/vdirect_in_predicate.h | 71 +++++----
be/src/format_v2/column_mapper.cpp | 9 +-
be/src/format_v2/column_mapper.h | 9 +-
be/src/format_v2/file_reader.h | 9 ++
be/src/format_v2/jni/jni_table_reader.cpp | 17 ++
be/src/format_v2/jni/jni_table_reader.h | 1 +
be/src/format_v2/parquet/parquet_profile.cpp | 3 +
be/src/format_v2/parquet/parquet_profile.h | 1 +
be/src/format_v2/parquet/parquet_reader.cpp | 19 ++-
be/src/format_v2/parquet/parquet_reader.h | 4 +
be/src/format_v2/parquet/parquet_scan.cpp | 93 ++++++++---
be/src/format_v2/parquet/parquet_scan.h | 9 +-
.../parquet/reader/native_column_reader.cpp | 2 -
be/src/format_v2/parquet/selection_vector.h | 105 ++++++++++++-
be/src/format_v2/table/hudi_reader.cpp | 12 ++
be/src/format_v2/table/hudi_reader.h | 1 +
be/src/format_v2/table/paimon_reader.cpp | 12 ++
be/src/format_v2/table/paimon_reader.h | 1 +
be/src/format_v2/table_reader.cpp | 110 +++++++++++++
be/src/format_v2/table_reader.h | 19 ++-
be/src/runtime/query_context.h | 1 -
be/test/exprs/expr_zonemap_filter_test.cpp | 44 ++++--
be/test/format_v2/jni/jni_table_reader_test.cpp | 26 ++++
.../parquet/parquet_benchmark_scenarios_test.cpp | 20 +++
.../parquet/parquet_reader_control_test.cpp | 55 +++++++
be/test/format_v2/parquet/parquet_scan_test.cpp | 130 +++++++++++++++-
be/test/format_v2/table/hudi_reader_test.cpp | 45 ++++++
be/test/format_v2/table/paimon_reader_test.cpp | 44 ++++++
be/test/format_v2/table_reader_test.cpp | 172 ++++++++++++++++++++-
43 files changed, 1386 insertions(+), 117 deletions(-)
diff --git a/be/benchmark/benchmark_bit_pack.hpp
b/be/benchmark/benchmark_bit_pack.hpp
index a4f269dd471..78ba2ebcd97 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<int>(state.range(0));
+ constexpr uint8_t n = 255;
std::default_random_engine e;
std::uniform_int_distribution<int64_t> 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<int>(state.range(0));
+ constexpr uint8_t n = 255;
std::default_random_engine e;
std::uniform_int_distribution<int64_t> u;
diff --git a/be/benchmark/benchmark_column_array_view.hpp
b/be/benchmark/benchmark_column_array_view.hpp
index 09baf2bd435..4ba5fc522a7 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 34fd287f203..567a1b84d5d 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 ba469b75fa6..ae574321aec 100644
--- a/be/benchmark/benchmark_fastunion.hpp
+++ b/be/benchmark/benchmark_fastunion.hpp
@@ -19,7 +19,7 @@
#include <string>
-#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 d923d208fe4..2fc6c47eaa6 100644
--- a/be/benchmark/benchmark_hll_merge.hpp
+++ b/be/benchmark/benchmark_hll_merge.hpp
@@ -17,7 +17,7 @@
#include <benchmark/benchmark.h>
-#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 7c64fa2729c..775f5caffef 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 9949a783b05..fb2650b40f5 100644
--- a/be/benchmark/binary_cast_benchmark.hpp
+++ b/be/benchmark/binary_cast_benchmark.hpp
@@ -21,7 +21,7 @@
#include <random>
#include <vector>
-#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 6c2fb8c85df..4c1d3cf4d5e 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 fcbed4a548b..156302b2423 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 00000000000..14654472f5e
--- /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 <benchmark/benchmark.h>
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#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<HybridSetBase> filter(create_set(PrimitiveType::TYPE_INT,
cardinality, false));
+ for (size_t index = 0; index < cardinality; ++index) {
+ const int32_t value = static_cast<int32_t>(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<DataTypeInt32>(),
+ "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<DataTypeInt32>(),
"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<double>(cardinality);
+}
+
+inline bool register_file_scanner_expr_benchmarks() {
+ for (const size_t cardinality : std::array<size_t, 4> {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 00000000000..06c0a338def
--- /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 <benchmark/benchmark.h>
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <numeric>
+#include <string>
+#include <vector>
+
+#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<uint8_t> make_filter(size_t rows, int selectivity_percent,
Pattern pattern) {
+ std::vector<uint8_t> filter(rows, 0);
+ const size_t selected_rows =
+ rows * static_cast<size_t>(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 <typename Selection>
+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 <typename Selection>
+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<format::parquet::SelectionVector::Index>
expected_selection_rows(
+ const SelectionScenario& scenario, const std::vector<uint8_t>&
row_filter,
+ const std::vector<uint8_t>& first_filter, const std::vector<uint8_t>&
selection_filter) {
+ std::vector<format::parquet::SelectionVector::Index> 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<format::parquet::SelectionVector::Index> 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<size_t>(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<int64_t>(state.iterations() *
SELECTION_ROWS));
+ state.SetBytesProcessed(static_cast<int64_t>(state.iterations() *
SELECTION_ROWS));
+ state.counters["raw_rows"] = static_cast<double>(SELECTION_ROWS);
+ state.counters["selected_rows"] = static_cast<double>(selected_rows);
+ state.counters["ns/raw_row"] = benchmark::Counter(
+ static_cast<double>(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 d2655b3380e..a9c58c15d8c 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<KernelScenario> kernel_scenarios() {
return scenarios;
}
+inline std::vector<SelectionScenario> selection_scenarios() {
+ std::vector<SelectionScenario> 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<ReaderScenario> reader_scenarios() {
std::vector<ReaderScenario> scenarios;
std::set<std::tuple<ReaderOperation, Encoding, int, Pattern, int,
Projection, int, int,
@@ -347,6 +368,18 @@ inline std::string to_string(Projection value) {
return value == Projection::PREDICATE_ONLY ? "predicate_only" :
"predicate_projected";
}
+inline std::string to_string(SelectionOperation value) {
+ switch (value) {
+ case SelectionOperation::RESIZE_IDENTITY:
+ return "resize_identity";
+ case SelectionOperation::ROW_FILTER:
+ return "row_filter";
+ case SelectionOperation::CASCADE_FILTER:
+ return "cascade_filter";
+ }
+ return "unknown";
+}
+
inline std::string to_string(ReaderOperation value) {
switch (value) {
case ReaderOperation::OPEN_TO_FIRST_BLOCK:
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index 6a7bf74571e..5343b034ea2 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -442,6 +442,12 @@ Status FileScannerV2::_get_block_impl(RuntimeState* state,
Block* block, bool* e
}
{
+ if (_table_reader_rf_num != _applied_rf_num) {
+ VExprContextSPtrs refreshed_conjuncts;
+ RETURN_IF_ERROR(_build_table_conjuncts(&refreshed_conjuncts));
+
RETURN_IF_ERROR(_table_reader->refresh_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 658ec44ad7f..87a68e2ab6d 100644
--- a/be/src/exec/scan/file_scanner_v2.h
+++ b/be/src/exec/scan/file_scanner_v2.h
@@ -178,6 +178,7 @@ private:
std::shared_ptr<SplitSourceConnector> _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 79d0f996d7c..b272c816edb 100644
--- a/be/src/exprs/vdirect_in_predicate.h
+++ b/be/src/exprs/vdirect_in_predicate.h
@@ -17,6 +17,7 @@
#pragma once
+#include <mutex>
#include <utility>
#include <vector>
@@ -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<Field> 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 @@ public:
std::shared_ptr<HybridSetBase> 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<VSlotRef>(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<VSlotRef>(get_child(0)) != nullptr;
}
@@ -177,8 +189,12 @@ public:
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 @@ private:
}
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<HybridSetBase> _filter;
@@ -323,10 +345,7 @@ private:
// 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<Field> _seg_filter_values;
- Field _seg_filter_min;
- Field _seg_filter_max;
+ std::shared_ptr<PruningState> _pruning_state =
std::make_shared<PruningState>();
};
#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 c3dbe3fa9e7..7e028370221 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<TableFilter>& table_filters,
const std::vector<ColumnDefinition>& projected_columns,
FileScanRequest* file_request,
- RuntimeState* runtime_state) {
+ RuntimeState* runtime_state,
+ const std::map<LocalColumnId, LocalIndex>* 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 e03f836061e..68d0ff357c0 100644
--- a/be/src/format_v2/column_mapper.h
+++ b/be/src/format_v2/column_mapper.h
@@ -193,10 +193,11 @@ public:
// 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<TableFilter>&
table_filters,
- const std::vector<ColumnDefinition>&
projected_columns,
- FileScanRequest* file_request,
- RuntimeState* runtime_state = nullptr);
+ virtual Status create_scan_request(
+ const std::vector<TableFilter>& table_filters,
+ const std::vector<ColumnDefinition>& projected_columns,
FileScanRequest* file_request,
+ RuntimeState* runtime_state = nullptr,
+ const std::map<LocalColumnId, LocalIndex>* 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 3ff512975d2..65a2d03417c 100644
--- a/be/src/format_v2/file_reader.h
+++ b/be/src/format_v2/file_reader.h
@@ -305,6 +305,15 @@ public:
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<FileScanRequest>
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 b58696fe4ad..dfa00c1568e 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 df7edc92f98..76e51c1c059 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 @@ public:
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 02b64d7b63c..e6717505cea 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 2282a70db30..57c03c79336 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 4a80bc0403d..065adc63b66 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<format::FileScanRequest> 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<format::FileScanRequest>
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 da6135b81ae..fe95b93a9e0 100644
--- a/be/src/format_v2/parquet/parquet_reader.h
+++ b/be/src/format_v2/parquet/parquet_reader.h
@@ -60,6 +60,10 @@ public:
Status open(std::shared_ptr<format::FileScanRequest> request) override;
+ bool supports_scan_request_refresh() const override { return true; }
+
+ Status queue_scan_request(std::shared_ptr<format::FileScanRequest>
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 33ef6a08fed..18b42d90a96 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<SelectionVector::Index>(row_idx));
- }
- }
- return new_selected_rows;
+ return
cast_set<uint16_t>(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<SelectionVector::Index>(
-
selection->get_index(selection_idx)));
- }
- }
- return new_selected_rows;
+ return cast_set<uint16_t>(
+ 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<format::FileScanRequest>
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<format::FileScanRequest>
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<int> candidate {candidate_plan.row_group_id};
+ std::vector<int> 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<std::unique_ptr<ParquetColumnSchema>>& file_schema,
- const format::FileScanRequest& request, Block* file_block, size_t*
rows, bool* eof) {
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& 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 f0e202d9942..474963ed5d9 100644
--- a/be/src/format_v2/parquet/parquet_scan.h
+++ b/be/src/format_v2/parquet/parquet_scan.h
@@ -188,6 +188,8 @@ public:
_enable_strict_mode = enable_strict_mode;
}
void set_runtime_state(RuntimeState* runtime_state) { _runtime_state =
runtime_state; }
+ void set_scan_request(std::shared_ptr<format::FileScanRequest> request);
+ void queue_scan_request(std::shared_ptr<format::FileScanRequest> 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 @@ public:
Status read_next_batch(ParquetFileContext& file_context,
const
std::vector<std::unique_ptr<ParquetColumnSchema>>& 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 @@ private:
SelectionVector _selection;
std::vector<uint32_t> _read_column_positions_scratch;
const format::FileScanRequest* _predicate_schedule_request = nullptr;
+ std::shared_ptr<format::FileScanRequest> _active_request;
+ std::shared_ptr<format::FileScanRequest> _pending_request;
+ bool _remaining_plans_need_replanning = false;
detail::PredicateConjunctSchedule _predicate_schedule;
std::vector<size_t> _predicate_positions_scratch;
std::unordered_map<size_t, size_t> _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 60eb0dc3d92..6b686597fa6 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 033478875fa..99c854d8b4e 100644
--- a/be/src/format_v2/parquet/selection_vector.h
+++ b/be/src/format_v2/parquet/selection_vector.h
@@ -68,17 +68,19 @@ public:
_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<Index>(idx);
- }
_identity = true;
+ _mutable_data_exposed = false;
++_generation;
}
@@ -87,6 +89,7 @@ public:
_data = nullptr;
_size = 0;
_identity = true;
+ _mutable_data_exposed = false;
++_generation;
}
@@ -95,9 +98,11 @@ public:
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 @@ public:
}
void set_index(size_t idx, Index value) {
+ _materialize_identity();
_data[idx] = value;
if (value != idx) {
_identity = false;
@@ -119,10 +125,18 @@ public:
++_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 @@ public:
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 @@ public:
}
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<Index>(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<Index>(output);
+ ++output;
+ }
+ bool remains_identity = true;
+ for (size_t position = output; position < count; ++position) {
+ if (filter[position] != 0) {
+ _data[output++] = static_cast<Index>(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<Index>(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<Index> _owned;
Index* _data = nullptr;
size_t _size = 0;
bool _identity = true;
+ bool _mutable_data_exposed = false;
uint64_t _generation = 0;
mutable std::vector<uint8_t> _filter;
mutable uint64_t _filter_generation = std::numeric_limits<uint64_t>::max();
mutable size_t _filter_count = 0;
mutable int64_t _filter_batch_rows = -1;
+ mutable uint64_t _verified_generation =
std::numeric_limits<uint64_t>::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 ee24d0f9ad7..838d98b5e52 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 dbb6f5e8231..c06e1b238b6 100644
--- a/be/src/format_v2/table/hudi_reader.h
+++ b/be/src/format_v2/table/hudi_reader.h
@@ -59,6 +59,7 @@ public:
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 a3f4092a470..93183af178b 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 823fa6540d2..8570f2efba6 100644
--- a/be/src/format_v2/table/paimon_reader.h
+++ b/be/src/format_v2/table/paimon_reader.h
@@ -65,6 +65,7 @@ public:
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 8a5da295fdd..be4fe0e406e 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<LocalColumnIndex>&
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<FileScanRequest>();
+ 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 6d679ec2d79..9ff2641f581 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 @@ public:
// 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 @@ protected:
// 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 @@ protected:
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 @@ protected:
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 @@ protected:
_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 @@ protected:
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<FileScanRequest> _file_scan_request;
std::vector<ColumnDefinition> _projected_columns;
std::unique_ptr<ScanTask> _current_task;
std::optional<io::FileDescription> _current_file_description;
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
index 3ef113c1fd1..3102cc85296 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -325,7 +325,6 @@ private:
void _init_query_mem_tracker();
std::unordered_map<int, RuntimePredicate> _runtime_predicates;
-
std::unique_ptr<RuntimeFilterMgr> _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 1d80f40531e..668a528b6a1 100644
--- a/be/test/exprs/expr_zonemap_filter_test.cpp
+++ b/be/test/exprs/expr_zonemap_filter_test.cpp
@@ -22,6 +22,7 @@
#include <array>
#include <map>
#include <memory>
+#include <mutex>
#include <set>
#include <string>
#include <vector>
@@ -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<TYPE_STRING>("aaa"),
direct_in_expr._seg_filter_min);
- EXPECT_EQ(Field::create_field<TYPE_STRING>("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<TYPE_STRING>("aaa"),
+ direct_in_expr._pruning_state->seg_filter_min);
+ EXPECT_EQ(Field::create_field<TYPE_STRING>("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<HybridSetBase> 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<VDirectInPredicate>(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<VDirectInPredicate>(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 2eee5548ad7..a59a4bbdf56 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 <thread>
#include <vector>
+#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<DataTypeUInt8>(), "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 72ccbf44844..2145b6ab60d 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 0c70ddaf01e..e21439e8855 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<const uint8_t*>(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<level_t> {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<format::FileScanRequest>();
+ auto refreshed = std::make_shared<format::FileScanRequest>();
+ 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 75733164ed2..8ed63a7fe28 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<int32_t>&
dictionary_values) {
+ std::vector<int32_t> scores(dictionary_values.size());
+ for (size_t index = 0; index < scores.size(); ++index) {
+ scores[index] = static_cast<int32_t>((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<int64_t>(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<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto initial = std::make_shared<format::FileScanRequest>();
+ 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::FileScanRequest>();
+ 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<int32_t> 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<format::ColumnDefinition> schema;
+ EXPECT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ 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<int32_t>({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<int32_t>({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<int32_t>({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<DataTypeInt32>()),
+ Field::create_field<TYPE_INT>(3));
+ ASSERT_NE(state.get_query_ctx(), nullptr);
+ auto reader = create_reader(0, -1, &profile);
ASSERT_TRUE(reader->init(&state).ok());
std::vector<format::ColumnDefinition> schema;
ASSERT_TRUE(reader->get_schema(&schema).ok());
- auto prepared =
- create_topn_conjunct(&state, 0, schema[0].type,
Field::create_field<TYPE_INT>(3));
+
ASSERT_TRUE(schema[0].type->equals(*prepared.conjunct->root()->children()[0]->data_type()));
auto request = std::make_shared<format::FileScanRequest>();
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 96126281744..e75eee47be3 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 @@ public:
}
};
+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<RefreshTrackingTableReader>();
+ child = tracking.get();
+ return tracking;
+ },
+ [] { return std::make_unique<TableReader>(); });
+ 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 6ebec512f42..32b82ab12ac 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 @@ public:
}
};
+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<RefreshTrackingTableReader>();
+ child = tracking.get();
+ return tracking;
+ },
+ [] { return std::make_unique<TableReader>(); });
+ 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 bb0db9329a2..f4bc1911d17 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<FileScanRequest> last_request;
+ std::shared_ptr<FileScanRequest> pending_request;
std::optional<FileAggregateRequest> last_aggregate_request;
std::shared_ptr<ConditionCacheContext> condition_cache_ctx;
std::shared_ptr<io::IOContext> io_ctx;
@@ -1101,6 +1103,14 @@ public:
return Status::OK();
}
+ bool supports_scan_request_refresh() const override { return true; }
+
+ Status queue_scan_request(std::shared_ptr<FileScanRequest> 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 @@ public:
std::shared_ptr<FakeFileReaderState> 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<FileReader>* reader) override {
DORIS_CHECK(reader != nullptr);
@@ -1610,6 +1626,101 @@ TEST(TableReaderTest,
PrepareSplitReplacesInitialConjunctSnapshot) {
ASSERT_TRUE(reader.close().ok());
}
+TEST(TableReaderTest, ActiveReaderQueuesRefreshedRuntimeFilterRequest) {
+ std::vector<ColumnDefinition> file_schema;
+ file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ file_schema.push_back(make_file_column(1, "value",
std::make_shared<DataTypeString>()));
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ projected_columns.push_back(make_table_column(1, "value",
std::make_shared<DataTypeString>()));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RuntimeProfile profile("scanner");
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ 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<int32_t>({0}));
+
EXPECT_EQ(projection_ids(fake_state->pending_request->non_predicate_columns),
+ std::vector<int32_t>({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<ColumnDefinition> file_schema;
+ file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
+
+ std::vector<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ set_name_identifiers(&projected_columns);
+
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ auto fake_state = std::make_shared<FakeFileReaderState>();
+ 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<ColumnDefinition> file_schema;
file_schema.push_back(make_file_column(0, "id",
std::make_shared<DataTypeInt32>()));
@@ -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<ColumnDefinition> projected_columns;
+ projected_columns.push_back(make_table_column(0, "id",
std::make_shared<DataTypeInt32>()));
+ 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<GlobalIndex> {},
+ })
+ .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<const
ColumnInt32&>(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<int32_t> remaining_ids;
+ while (!eos) {
+ Block block = build_table_block(projected_columns);
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ const auto& ids = assert_cast<const
ColumnInt32&>(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<int32_t>({5, 6}));
ASSERT_TRUE(reader.close().ok());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
