You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/06/22 05:32:56 UTC
(datafusion) 01/01: feat(parquet): intra-file early stopping via statistics + dynamic filters (#22450)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-22450-ba67bb467a30c3dbe7f78089451ea4c4d1a191d5
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 9e68a86ddcd1b9968331842406cc2f65263d784c
Author: Qi Zhu <zh...@gmail.com>
AuthorDate: Mon Jun 22 13:32:35 2026 +0800
feat(parquet): intra-file early stopping via statistics + dynamic filters (#22450)
## Which issue does this PR close?
Closes #22407.
## Rationale for this change
DataFusion already prunes parquet at three granularities — **file**
(`EarlyStoppingStream` + `FilePruner`), **row group at scan-startup**
(`PruningPredicate` → `RowGroupAccessPlanFilter`), and **row inside an
open RG** (`RowFilter`).
There's a gap in the middle: once row-group pruning runs at file open,
that
decision is **frozen** because any dynamic filter is still `lit(true)`
at
that point. As `TopK` tightens its threshold at runtime, subsequent RGs
in
the already-opened file keep getting decoded even when their stats
already
prove they cannot beat the threshold. This is the dominant cost for
`ORDER BY ... LIMIT` queries on multi-RG files where file-level pruning
can't help (single large file, or scrambled-RG multi-file).
See the issue for a full architectural diagram and a concrete trace
showing where the wasted I/O / decompression / decode lives.
## What changes are included in this PR?
A single decoder paused at row-group boundaries, with a pruner consulted
between row groups and the decoder rebuilt via `into_builder()` to skip
the row groups the pruner just rejected. Three coordinated pieces:
1. **`RowGroupPruner`**
(`datafusion/datasource-parquet/src/push_decoder.rs`)
mirrors `FilePruner` at row-group granularity. It uses the
`DynamicFilterTracker` API from #22460 to subscribe once to every
not-yet-complete dynamic filter in the predicate; `tracker.changed()`
is a single atomic load — no tree traversal per check. The cached
`PruningPredicate` is rebuilt only when a watched filter has actually
moved, then evaluated against the next pending row group's statistics
via the existing `RowGroupPruningStatistics` adapter. Predicate
construction errors and predicate evaluation errors are counted into
two separate metrics so a flaky predicate path can never silently
drop data.
2. **Single-decoder iteration model**
(`PushDecoderStreamState::transition`). The opener builds **one**
`ParquetPushDecoder` from the prepared access plan, and the stream
uses arrow-rs 59's `ParquetRecordBatchReader` iterator to pause at
row-group boundaries. At each boundary the pruner is consulted
against the head of `rg_plan` (the remaining row-group indices). If
the pruner proves the head RG unwinnable, that index is dropped from
the plan and the decoder is **rebuilt via**
`decoder.into_builder().with_row_groups(remaining).build()` so the
skipped RGs are bypassed entirely — no decode, no row-filter eval.
Already-fetched buffered bytes for downstream RGs carry across the
rebuild.
3. **Gate: build the pruner only when the predicate actually moves.**
The opener creates a `RowGroupPruner` only when
`DynamicFilterTracking::classify(&predicate)` reports `Watching` (at
least one not-yet-complete dynamic filter) **and** more than one row
group remains in the access plan. Static or already-complete
predicates were fully consumed by `prune_by_statistics` at file open,
so re-evaluating them per RG boundary would be wasted work.
The earlier multi-decoder design (`PendingDecoderRun`,
`ParquetAccessPlan::split_runs`, `force_per_row_group`) is removed —
arrow-rs 59's `into_builder` + `with_row_groups` makes a single decoder
strictly more capable.
### Observability
- New `Count` metric `row_groups_pruned_dynamic_filter` on
`ParquetFileMetrics` surfaces the runtime saving.
- New `dynamic_rg_pruning=eligible` marker on `ParquetSource`'s
`EXPLAIN` (`fmt_extra` Default + Verbose) signals plan-time
eligibility, emitted whenever the predicate has a still-watching
dynamic portion. **Eligible** rather than **true** because the static
plan can't predict the runtime outcome.
### Benchmarks (`benchmarks/sort_pushdown_inexact`, 5 iterations)
| Query | main | this PR | Δ |
|---|---|---|---|
| Q1 `ORDER BY l_orderkey DESC LIMIT 100` | 6.99 ms | 3.80 ms | **−46%**
|
| Q2 `ORDER BY l_orderkey DESC LIMIT 1000` | 3.29 ms | 1.33 ms |
**−60%** |
| Q3 `SELECT * ... DESC LIMIT 100` | 11.17 ms | 9.91 ms | −11% |
| Q4 `SELECT * ... DESC LIMIT 1000` | 9.28 ms | 7.95 ms | −14% |
Narrow-projection queries gain the most — their per-RG cost is dominated
by metadata + sort-column read, which this PR eliminates for unwinnable
RGs. Wide-projection queries gain less because the *kept* RG's
all-column decode dominates total time, but still see meaningful
savings.
## Are these changes tested?
Three layers:
- **6 unit tests**:
- 3 in `push_decoder.rs::tests`: `RowGroupPruner` basic pruning,
tracker-driven dynamic-filter updates, fallback when the predicate
has no analyzable bounds.
- 3 in `source.rs::tests`: `dynamic_rg_pruning=eligible` marker
present on dynamic predicate, absent on static predicate, absent
when there is no predicate at all.
- **3 integration tests** in
`datafusion/core/tests/parquet/dynamic_row_group_pruning.rs`:
asserts `row_groups_pruned_dynamic_filter >= 1` end-to-end on a 5-RG
`ORDER BY DESC LIMIT 5` scan; a regression test for the
`prepare_access_plan` reorder bug that uses `ORDER BY ASC` against a
file written in descending value order so the sort-pushdown reorder
is exercised; and a quiet-without-TopK test that asserts the metric
stays at 0 (no spurious firing).
- **New SLT**
`datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt`:
asserts both `EXPLAIN` surfaces — plain `EXPLAIN` shows
`dynamic_rg_pruning=eligible`, and `EXPLAIN ANALYZE` pins
`row_groups_pruned_dynamic_filter=4` (five RGs, four pruned at
runtime).
`cargo clippy --all-targets --all-features -- -D warnings` clean.
## Are there any user-facing changes?
Two visible additions, both opt-in via existing dynamic-filter
infrastructure:
- New `row_groups_pruned_dynamic_filter` counter visible in
`EXPLAIN ANALYZE` for queries whose plan carries a
`DynamicFilterPhysicalExpr` (today: only TopK with
`enable_topk_dynamic_filter_pushdown=true`, which is the default).
- New `dynamic_rg_pruning=eligible` marker visible in `EXPLAIN`
output for the same queries.
No config changes, no API breakage, no behavior change for queries
without a dynamic predicate.
---
.../tests/parquet/dynamic_row_group_pruning.rs | 435 ++++++++++++++++++
datafusion/core/tests/parquet/mod.rs | 56 ++-
datafusion/datasource-parquet/src/access_plan.rs | 72 ---
.../datasource-parquet/src/decoder_projection.rs | 12 +-
datafusion/datasource-parquet/src/metrics.rs | 14 +
datafusion/datasource-parquet/src/opener/mod.rs | 195 +++-----
datafusion/datasource-parquet/src/push_decoder.rs | 502 ++++++++++++++++++---
datafusion/datasource-parquet/src/row_filter.rs | 13 +-
.../datasource-parquet/src/row_group_filter.rs | 16 +-
datafusion/datasource-parquet/src/source.rs | 111 ++++-
datafusion/sqllogictest/test_files/clickbench.slt | 8 +-
.../test_files/dynamic_filter_pushdown_config.slt | 34 +-
.../test_files/dynamic_row_group_pruning.slt | 112 +++++
.../sqllogictest/test_files/explain_analyze.slt | 28 +-
datafusion/sqllogictest/test_files/limit.slt | 2 +-
.../sqllogictest/test_files/limit_pruning.slt | 2 +-
.../test_files/preserve_file_partitioning.slt | 6 +-
.../test_files/projection_pushdown.slt | 50 +-
.../test_files/push_down_filter_parquet.slt | 40 +-
.../test_files/push_down_filter_regression.slt | 20 +-
.../test_files/repartition_subset_satisfaction.slt | 6 +-
.../sqllogictest/test_files/sort_pushdown.slt | 46 +-
.../test_files/statistics_registry.slt | 6 +-
datafusion/sqllogictest/test_files/topk.slt | 18 +-
24 files changed, 1385 insertions(+), 419 deletions(-)
diff --git a/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs b/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs
new file mode 100644
index 0000000000..b72c56ace5
--- /dev/null
+++ b/datafusion/core/tests/parquet/dynamic_row_group_pruning.rs
@@ -0,0 +1,435 @@
+// 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.
+
+//! End-to-end test for **runtime row-group pruning** driven by a TopK
+//! `SortExec`'s `DynamicFilterPhysicalExpr`.
+//!
+//! A 5-row-group parquet file is constructed with disjoint statistics on
+//! the sort column (`v`): row group `i` contains values
+//! `[i*100, (i+1)*100)`. The query `ORDER BY v DESC LIMIT 5` fills the
+//! TopK heap from the row group with the largest values; the threshold
+//! then proves the remaining row groups cannot contribute. The runtime
+//! `RowGroupPruner` in the parquet scan must observe the tightened
+//! threshold and increment `row_groups_pruned_dynamic_filter`.
+//!
+//! We assert a property (`pruned >= 1`) rather than an exact count
+//! because batch-arrival timing affects how soon the TopK heap fills,
+//! and we don't want this test to become flaky.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, Int64Array, RecordBatch};
+use arrow_schema::{DataType, Field, Schema};
+
+use crate::parquet::Unit::RowGroup;
+use crate::parquet::{ContextWithParquet, Scenario};
+
+/// Build five `RecordBatch`es whose `v` column ranges are disjoint:
+/// batch `i` carries `v` values `[i*100, (i+1)*100)`. When written with
+/// `max_row_group_row_count = 100` each batch lands in its own row group.
+fn build_five_disjoint_batches(schema: &Arc<Schema>) -> Vec<RecordBatch> {
+ (0..5i64)
+ .map(|rg| {
+ let base = rg * 100;
+ let values: Vec<i64> = (base..base + 100).collect();
+ let col: ArrayRef = Arc::new(Int64Array::from(values));
+ RecordBatch::try_new(Arc::clone(schema), vec![col]).unwrap()
+ })
+ .collect()
+}
+
+/// Build five `RecordBatch`es in *descending* value order: batch 0 holds
+/// `v ∈ [400, 500)`, batch 4 holds `v ∈ [0, 100)`. The physical row-group
+/// order on disk therefore does **not** match the order a `ORDER BY v ASC`
+/// query wants — sort-pushdown's `reorder_by_statistics` must rearrange
+/// the access plan so the scan reads RG 4 first, then RG 3, etc.
+fn build_five_disjoint_batches_desc(schema: &Arc<Schema>) -> Vec<RecordBatch> {
+ (0..5i64)
+ .map(|rg| {
+ let base = (4 - rg) * 100;
+ let values: Vec<i64> = (base..base + 100).collect();
+ let col: ArrayRef = Arc::new(Int64Array::from(values));
+ RecordBatch::try_new(Arc::clone(schema), vec![col]).unwrap()
+ })
+ .collect()
+}
+
+/// `ORDER BY v DESC LIMIT 5` against a 5-RG file with disjoint per-RG
+/// stats must trigger runtime RG pruning: the first RG read fills the
+/// heap, and the tightened threshold proves every other RG unreachable.
+#[tokio::test]
+async fn dynamic_rg_pruning_metric_fires_for_topk_descending_limit() {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ let batches = build_five_disjoint_batches(&schema);
+
+ // `with_custom_data` honors the custom schema + batches and ignores
+ // `Scenario`. `Unit::RowGroup(100)` enables `pushdown_filters`, which
+ // is required for the TopK dynamic filter to reach the parquet scan.
+ let mut ctx = ContextWithParquet::with_custom_data(
+ Scenario::Int,
+ RowGroup(100),
+ Arc::clone(&schema),
+ batches,
+ )
+ .await;
+
+ let output = ctx.query("SELECT v FROM t ORDER BY v DESC LIMIT 5").await;
+
+ assert_eq!(output.result_rows, 5, "query must return LIMIT rows",);
+
+ let pruned = output
+ .row_groups_pruned_dynamic_filter()
+ .expect("`row_groups_pruned_dynamic_filter` metric must be registered");
+ assert!(
+ pruned >= 1,
+ "dynamic RG pruner must skip at least one row group; \
+ pruned={pruned}\n{}",
+ output.description(),
+ );
+}
+
+/// Regression for the rg_plan / `reorder_by_statistics` ordering bug.
+///
+/// When `sort_order_for_reorder` is set on the parquet scan,
+/// `prepare_access_plan` calls
+/// [`PreparedAccessPlan::reorder_by_statistics`], which rearranges
+/// `row_group_indexes` so the decoder reads row groups in stats-optimal
+/// order (smallest-min first for ASC, etc.). The stream's per-RG plan
+/// (`rg_plan`) — which the runtime pruner walks one entry at a time —
+/// **must use this reordered list**, not the access plan's natural
+/// (index-ascending) order. Otherwise the pruner would consult the
+/// metadata of RG K while the decoder is actually about to yield RG K',
+/// silently producing wrong results.
+///
+/// This test makes the failure visible:
+///
+/// - File is written with RGs in *descending* `v` order (RG 0 has the
+/// largest values, RG 4 has the smallest).
+/// - Query is `ORDER BY v ASC LIMIT 5`, so sort-pushdown reorders the
+/// access plan to read RG 4 first, then RG 3, etc.
+/// - The smallest five values (which form the entire correct LIMIT
+/// answer) live in RG 4 alone. After they are emitted, the TopK
+/// threshold tightens enough that the per-RG pruner skips every other
+/// RG.
+///
+/// Without the fix, `rg_plan` would be `[0, 1, 2, 3, 4]` while the
+/// decoder reads `[4, 3, 2, 1, 0]`. The first yielded reader (for RG 4
+/// in the decoder) would be tracked as if it were RG 0, the pruner
+/// would check RG 1's stats (id range 300..400) against a threshold
+/// already tightened to `v < 5`, prune RG 1 (because nothing in
+/// 300..400 can satisfy `v < 5`), and then the rebuild via
+/// `into_builder` would scan a row group whose data does not match its
+/// expected metadata. The query would return fewer than five rows or
+/// the wrong rows.
+#[tokio::test]
+async fn dynamic_rg_pruning_handles_sort_pushdown_reorder() {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ let batches = build_five_disjoint_batches_desc(&schema);
+
+ let mut ctx = ContextWithParquet::with_custom_data(
+ Scenario::Int,
+ RowGroup(100),
+ Arc::clone(&schema),
+ batches,
+ )
+ .await;
+
+ let output = ctx.query("SELECT v FROM t ORDER BY v ASC LIMIT 5").await;
+
+ // Correctness — the five smallest values in the file are 0..=4.
+ // If `rg_plan` is misaligned with the decoder's read order, the
+ // pruner consults the wrong RG's stats and the result row count or
+ // values would drift.
+ assert_eq!(output.result_rows, 5, "query must return LIMIT rows");
+ let formatted = output.pretty_results();
+ for v in 0..=4i64 {
+ assert!(
+ formatted.contains(&format!("| {v} ")),
+ "output must contain the smallest value {v}; got:\n{formatted}",
+ );
+ }
+
+ // Behavior — the per-RG pruner must engage. We don't pin the exact
+ // count (batch-arrival timing affects how soon the heap fills); we
+ // only require that at least one row group is skipped at runtime.
+ let pruned = output
+ .row_groups_pruned_dynamic_filter()
+ .expect("`row_groups_pruned_dynamic_filter` metric must be registered");
+ assert!(
+ pruned >= 1,
+ "with `sort_order_for_reorder` active and a tight TopK, the \
+ runtime pruner must skip at least one row group; pruned={pruned}\n{}",
+ output.description(),
+ );
+}
+
+/// A query without ORDER BY does not produce a TopK and therefore no
+/// `DynamicFilterPhysicalExpr` reaches the scan. The runtime pruner must
+/// stay quiet — the metric should be 0.
+#[tokio::test]
+async fn dynamic_rg_pruning_metric_quiet_without_topk() {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ let batches = build_five_disjoint_batches(&schema);
+
+ let mut ctx = ContextWithParquet::with_custom_data(
+ Scenario::Int,
+ RowGroup(100),
+ Arc::clone(&schema),
+ batches,
+ )
+ .await;
+
+ // Plain `SELECT *` — no sort, no limit, no dynamic filter.
+ let output = ctx.query("SELECT v FROM t").await;
+ assert_eq!(output.result_rows, 500);
+
+ let pruned = output.row_groups_pruned_dynamic_filter().unwrap_or(0);
+ assert_eq!(
+ pruned,
+ 0,
+ "without TopK there is no dynamic filter, so the runtime pruner \
+ must not fire; pruned={pruned}\n{}",
+ output.description(),
+ );
+}
+
+/// Regression for "into_builder called mid-row-group" — surfaced by
+/// ClickBench Q24 / Q26 (`SELECT … WHERE x <> '' ORDER BY ts LIMIT 10`).
+///
+/// The push-decoder state machine re-enters Step 2 on every iteration of
+/// the `transition` loop, including iterations where Step 3 returned
+/// `NeedsData` and pushed byte ranges but has not yet produced a reader
+/// for the upcoming row group. At those moments the decoder is in
+/// `ReadingRowGroup` state but `is_at_row_group_boundary()` is `false`,
+/// and the runtime row-group pruner's `into_builder()` rebuild path
+/// errored out with:
+///
+/// ```text
+/// Parquet error: into_builder called mid-row-group;
+/// check is_at_row_group_boundary() first
+/// ```
+///
+/// The fix in `push_decoder.rs::Step 2` gates the prune-and-rebuild on
+/// `is_at_row_group_boundary()`. This test reproduces the trigger: a
+/// many-RG file (so the pruner has work to do) plus an `ORDER BY` query
+/// whose TopK threshold tightens enough to make the pruner want to
+/// rebuild more than once during the scan. Before the fix the query
+/// returned an `Execution` / `Parquet` error; after the fix it returns
+/// the expected ten rows and the pruner fires.
+#[tokio::test]
+async fn dynamic_rg_pruner_does_not_call_into_builder_mid_row_group() {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ // 20 disjoint row groups of 50 values each. With 20 RGs the pruner
+ // gets multiple boundaries to attempt rebuilds, so any path that
+ // calls `into_builder` outside a boundary is hit reliably.
+ let batches: Vec<RecordBatch> = (0..20i64)
+ .map(|rg| {
+ let base = rg * 50;
+ let values: Vec<i64> = (base..base + 50).collect();
+ let col: ArrayRef = Arc::new(Int64Array::from(values));
+ RecordBatch::try_new(Arc::clone(&schema), vec![col]).unwrap()
+ })
+ .collect();
+
+ let mut ctx = ContextWithParquet::with_custom_data(
+ Scenario::Int,
+ RowGroup(50),
+ Arc::clone(&schema),
+ batches,
+ )
+ .await;
+
+ let output = ctx.query("SELECT v FROM t ORDER BY v ASC LIMIT 10").await;
+
+ // Correctness: smallest ten values are 0..=9.
+ assert_eq!(output.result_rows, 10, "query must return LIMIT rows");
+ let formatted = output.pretty_results();
+ for v in 0..=9i64 {
+ assert!(
+ formatted.contains(&format!("| {v} ")),
+ "output must contain smallest value {v}; got:\n{formatted}",
+ );
+ }
+
+ // Behavior: with 20 disjoint RGs and a tight TopK, the dynamic
+ // pruner must skip a meaningful share of them. We don't pin the
+ // exact count — what matters is that the scan *completed* without
+ // the mid-row-group rebuild error.
+ let pruned = output
+ .row_groups_pruned_dynamic_filter()
+ .expect("`row_groups_pruned_dynamic_filter` metric must be registered");
+ assert!(
+ pruned >= 1,
+ "dynamic RG pruner must skip at least one row group; \
+ pruned={pruned}\n{}",
+ output.description(),
+ );
+}
+
+/// Build five sorted `RecordBatch`es with 1000 values each so that, when
+/// the writer is configured with `row_per_group=1000` and
+/// `data_page_row_count_limit=100`, every row group ends up with **ten
+/// data pages** of 100 rows each. RG `i` covers `[i*1000, (i+1)*1000)`,
+/// monotonically ascending — page index will then have tight per-page
+/// `min`/`max` and can prune at sub-RG granularity.
+fn build_five_thousand_row_rgs(schema: &Arc<Schema>) -> Vec<RecordBatch> {
+ (0..5i64)
+ .map(|rg| {
+ let base = rg * 1000;
+ let values: Vec<i64> = (base..base + 1000).collect();
+ let col: ArrayRef = Arc::new(Int64Array::from(values));
+ RecordBatch::try_new(Arc::clone(schema), vec![col]).unwrap()
+ })
+ .collect()
+}
+
+/// Co-existence test for **page-index `RowSelection`** + dynamic RG
+/// pruning. Tests that the `into_builder` rebuild preserves the
+/// `RowSelection` derived from page-index pruning across RG drops.
+///
+/// Layout: 5 RGs × 1000 rows, with `data_page_row_count_limit=100` so
+/// each RG has 10 pages of 100 rows.
+///
+/// Query: `SELECT v FROM t WHERE v >= 500 ORDER BY v DESC LIMIT 5`.
+/// - `v >= 500` engages the page index: in RG 0 (values 0..1000) the
+/// first 5 pages (values 0..500) are pruned, the last 5 (500..1000)
+/// are scanned. RGs 1..4 keep all their pages (every page has
+/// `max >= 500`). The decoder receives a `RowSelection` that masks
+/// out those first 5 pages of RG 0.
+/// - `ORDER BY v DESC LIMIT 5` fills the TopK heap from RG 4
+/// (`max=4999`); the tightened threshold (≥ 4995) then proves RGs
+/// 0..3 unreachable and the runtime pruner drops them in one
+/// `into_builder` rebuild.
+///
+/// If `into_builder` did **not** preserve the row selection (or
+/// truncated / shifted it incorrectly), either the result rows would
+/// drift or the count of pruned pages would drop to zero.
+#[tokio::test]
+async fn dynamic_rg_pruning_coexists_with_page_index_row_selection() {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ let batches = build_five_thousand_row_rgs(&schema);
+
+ // `RowGroupAndPage(1000, 100)` enables both `pushdown_filters` and
+ // page-index pruning, and writes a parquet file with 1000-row RGs
+ // partitioned into 100-row pages.
+ let mut ctx = ContextWithParquet::with_custom_data(
+ Scenario::Int,
+ crate::parquet::Unit::RowGroupAndPage(1000, 100),
+ Arc::clone(&schema),
+ batches,
+ )
+ .await;
+
+ let output = ctx
+ .query("SELECT v FROM t WHERE v >= 500 ORDER BY v DESC LIMIT 5")
+ .await;
+
+ // Correctness — top-5 values descending are 4995..=4999 (all in RG 4).
+ assert_eq!(output.result_rows, 5, "query must return LIMIT rows");
+ let formatted = output.pretty_results();
+ for v in 4995..=4999i64 {
+ assert!(
+ formatted.contains(&format!("| {v} ")),
+ "output must contain top-5 descending value {v}; got:\n{formatted}",
+ );
+ }
+
+ // Page-index pruning must have engaged: RG 0's first 5 pages are
+ // entirely < 500. If `into_builder` dropped the row-selection state,
+ // this metric would still report the original count (it is captured
+ // at file open). Combined with the dynamic-pruner assertion below it
+ // proves both mechanisms were active and that the rebuild left the
+ // selection coherent — otherwise the result rows above would drift.
+ let pages_pruned = output.metric_value("page_index_pages_pruned").unwrap_or(0);
+ assert!(
+ pages_pruned >= 5,
+ "page index must prune at least 5 pages (RG 0 pages 0..5 for v < 500); \
+ pruned={pages_pruned}\n{}",
+ output.description(),
+ );
+
+ let pruned = output
+ .row_groups_pruned_dynamic_filter()
+ .expect("`row_groups_pruned_dynamic_filter` metric must be registered");
+ assert!(
+ pruned >= 1,
+ "with TopK + tight threshold the runtime pruner must skip at least \
+ one row group; pruned={pruned}\n{}",
+ output.description(),
+ );
+}
+
+/// Co-existence test: a `WHERE` clause that gets pushed into the parquet
+/// `RowFilter` plus a `TopK` that drives the dynamic RG pruner.
+///
+/// `v % 2 = 0` cannot be statically pruned and is not page-index-amenable
+/// either, so it must run per-row inside the parquet decoder as a
+/// `RowFilter`. `ORDER BY v DESC LIMIT 3` then fills the TopK heap and
+/// tightens the threshold, triggering runtime RG pruning. The decoder
+/// rebuild that happens via
+/// `into_builder().with_row_groups(remaining).build()` must preserve the
+/// installed `RowFilter` (and any `RowSelection` derived from page-index
+/// pruning) across the rebuild — if it didn't, either:
+///
+/// - The post-prune RGs would silently drop their per-row filtering and
+/// the result would contain odd values, OR
+/// - The rebuilt decoder would re-emit rows the original was about to
+/// yield, double-counting against the limit.
+///
+/// This test catches both regressions: it pins both the exact result rows
+/// (top three even values descending: 498, 496, 494) and asserts the
+/// dynamic pruner fired at least once.
+#[tokio::test]
+async fn dynamic_rg_pruning_coexists_with_row_filter() {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ let batches = build_five_disjoint_batches(&schema);
+
+ let mut ctx = ContextWithParquet::with_custom_data(
+ Scenario::Int,
+ RowGroup(100),
+ Arc::clone(&schema),
+ batches,
+ )
+ .await;
+
+ // `v % 2 = 0` survives stats pruning (every RG straddles even / odd),
+ // so the predicate is pushed into the decoder as a `RowFilter` and
+ // evaluated per row. The TopK on top still tightens the threshold and
+ // engages the runtime RG pruner.
+ let output = ctx
+ .query("SELECT v FROM t WHERE v % 2 = 0 ORDER BY v DESC LIMIT 3")
+ .await;
+
+ assert_eq!(output.result_rows, 3, "query must return LIMIT rows");
+ let formatted = output.pretty_results();
+ for v in [498i64, 496, 494] {
+ assert!(
+ formatted.contains(&format!("| {v} ")),
+ "output must contain top-3 even descending value {v}; got:\n{formatted}",
+ );
+ }
+
+ let pruned = output
+ .row_groups_pruned_dynamic_filter()
+ .expect("`row_groups_pruned_dynamic_filter` metric must be registered");
+ assert!(
+ pruned >= 1,
+ "with WHERE v % 2 = 0 + TopK the runtime pruner must still skip at \
+ least one row group; pruned={pruned}\n{}",
+ output.description(),
+ );
+}
diff --git a/datafusion/core/tests/parquet/mod.rs b/datafusion/core/tests/parquet/mod.rs
index 12296f8498..1cc4bb32d9 100644
--- a/datafusion/core/tests/parquet/mod.rs
+++ b/datafusion/core/tests/parquet/mod.rs
@@ -46,6 +46,7 @@ use tempfile::NamedTempFile;
mod content_defined_chunking;
mod custom_reader;
+mod dynamic_row_group_pruning;
#[cfg(feature = "parquet_encryption")]
mod encryption;
mod expr_adapter;
@@ -99,6 +100,10 @@ enum Unit {
RowGroup(usize),
// pass max row per page in parquet writer
Page(usize),
+ // pass max row per row_group AND max row per page. Use when a test
+ // needs both multi-RG layout AND multiple pages within each RG so the
+ // page index can prune at sub-RG granularity.
+ RowGroupAndPage(usize, usize),
}
/// Test fixture that has an execution context that has an external
@@ -147,6 +152,12 @@ struct TestOutput {
}
impl TestOutput {
+ /// Pretty-printed result batches, useful for asserting concrete row
+ /// values in regression tests.
+ fn pretty_results(&self) -> &str {
+ &self.pretty_results
+ }
+
/// retrieve the value of the named metric, if any
fn metric_value(&self, metric_name: &str) -> Option<usize> {
if let Some(pm) = self.pruning_metric(metric_name) {
@@ -259,6 +270,13 @@ impl TestOutput {
.map(|pm| pm.total_pruned())
}
+ /// The number of row groups pruned at runtime by the dynamic
+ /// row-group pruner (e.g. driven by a TopK `SortExec` threshold
+ /// pushed down via `DynamicFilterPhysicalExpr`).
+ fn row_groups_pruned_dynamic_filter(&self) -> Option<usize> {
+ self.metric_value("row_groups_pruned_dynamic_filter")
+ }
+
fn description(&self) -> String {
format!(
"Input:\n{}\nQuery:\n{}\nOutput:\n{}\nMetrics:\n{}",
@@ -305,13 +323,32 @@ impl ContextWithParquet {
Unit::RowGroup(row_per_group) => {
config = config.with_parquet_bloom_filter_pruning(true);
config.options_mut().execution.parquet.pushdown_filters = true;
- make_test_file_rg(scenario, row_per_group, custom_schema, custom_batches)
- .await
+ make_test_file_rg(
+ scenario,
+ row_per_group,
+ None,
+ custom_schema,
+ custom_batches,
+ )
+ .await
}
Unit::Page(row_per_page) => {
config = config.with_parquet_page_index_pruning(true);
make_test_file_page(scenario, row_per_page).await
}
+ Unit::RowGroupAndPage(row_per_group, row_per_page) => {
+ config = config.with_parquet_bloom_filter_pruning(true);
+ config = config.with_parquet_page_index_pruning(true);
+ config.options_mut().execution.parquet.pushdown_filters = true;
+ make_test_file_rg(
+ scenario,
+ row_per_group,
+ Some(row_per_page),
+ custom_schema,
+ custom_batches,
+ )
+ .await
+ }
};
let parquet_path = file.path().to_string_lossy();
@@ -1139,6 +1176,7 @@ fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
async fn make_test_file_rg(
scenario: Scenario,
row_per_group: usize,
+ row_per_page: Option<usize>,
custom_schema: Option<SchemaRef>,
custom_batches: Option<Vec<RecordBatch>>,
) -> NamedTempFile {
@@ -1148,11 +1186,19 @@ async fn make_test_file_rg(
.tempfile()
.expect("tempfile creation");
- let props = WriterProperties::builder()
+ let mut props_builder = WriterProperties::builder()
.set_max_row_group_row_count(Some(row_per_group))
.set_bloom_filter_enabled(true)
- .set_statistics_enabled(EnabledStatistics::Page)
- .build();
+ .set_statistics_enabled(EnabledStatistics::Page);
+ if let Some(rpp) = row_per_page {
+ // Bound rows per page so the page index can prune at sub-RG
+ // granularity. `write_batch_size` must also be set so the writer
+ // does not buffer the whole RG into one page.
+ props_builder = props_builder
+ .set_data_page_row_count_limit(rpp)
+ .set_write_batch_size(rpp);
+ }
+ let props = props_builder.build();
let (batches, schema) =
if let (Some(schema), Some(batches)) = (custom_schema, custom_batches) {
diff --git a/datafusion/datasource-parquet/src/access_plan.rs b/datafusion/datasource-parquet/src/access_plan.rs
index 593ed365cd..8189c2378c 100644
--- a/datafusion/datasource-parquet/src/access_plan.rs
+++ b/datafusion/datasource-parquet/src/access_plan.rs
@@ -150,24 +150,6 @@ pub enum RowGroupAccess {
Selection(RowSelection),
}
-/// A consecutive set of row groups that share the same row filter requirement.
-#[derive(Debug, Clone, PartialEq)]
-pub(crate) struct RowGroupRun {
- /// True if this run needs row filter evaluation.
- pub(crate) needs_filter: bool,
- /// The access plan for this run.
- pub(crate) access_plan: ParquetAccessPlan,
-}
-
-impl RowGroupRun {
- fn new(needs_filter: bool, access_plan: ParquetAccessPlan) -> Self {
- Self {
- needs_filter,
- access_plan,
- }
- }
-}
-
impl RowGroupAccess {
/// Return true if this row group should be scanned
pub fn should_scan(&self) -> bool {
@@ -398,12 +380,6 @@ impl ParquetAccessPlan {
&self.fully_matched
}
- /// Return true if any scanned row group is fully matched.
- fn has_fully_matched(&self) -> bool {
- self.row_group_index_iter()
- .any(|idx| self.is_fully_matched(idx))
- }
-
/// Set to scan only the [`RowSelection`] in the specified row group.
///
/// Behavior is different depending on the existing access
@@ -589,54 +565,6 @@ impl ParquetAccessPlan {
self.row_groups
}
- /// Split this plan into consecutive row group runs that share the same row
- /// filter requirement.
- pub(crate) fn split_runs(self, needs_filter: bool) -> Vec<RowGroupRun> {
- if !needs_filter || !self.has_fully_matched() {
- return vec![RowGroupRun::new(needs_filter, self)];
- }
-
- let num_row_groups = self.row_groups.len();
- let row_groups = self.row_groups;
- let fully_matched = self.fully_matched;
- let mut runs: Vec<RowGroupRun> = Vec::new();
-
- for (idx, (access, fully_matched)) in
- row_groups.into_iter().zip(fully_matched).enumerate()
- {
- if !access.should_scan() {
- continue;
- }
-
- let row_group_needs_filter = !fully_matched;
- if let Some(run) = runs
- .last_mut()
- .filter(|run| run.needs_filter == row_group_needs_filter)
- {
- run.access_plan.set(idx, access);
- if fully_matched {
- run.access_plan.mark_fully_matched(idx);
- }
- } else {
- let mut run_plan = ParquetAccessPlan::new_none(num_row_groups);
- run_plan.set(idx, access);
- if fully_matched {
- run_plan.mark_fully_matched(idx);
- }
- runs.push(RowGroupRun::new(row_group_needs_filter, run_plan));
- }
- }
-
- if runs.is_empty() {
- vec![RowGroupRun::new(
- needs_filter,
- ParquetAccessPlan::new_none(num_row_groups),
- )]
- } else {
- runs
- }
- }
-
/// Prepare this plan and resolve to the final `PreparedAccessPlan`
pub(crate) fn prepare(
self,
diff --git a/datafusion/datasource-parquet/src/decoder_projection.rs b/datafusion/datasource-parquet/src/decoder_projection.rs
index 27a84f2f50..192600ce7a 100644
--- a/datafusion/datasource-parquet/src/decoder_projection.rs
+++ b/datafusion/datasource-parquet/src/decoder_projection.rs
@@ -20,7 +20,8 @@
//! [`DecoderProjection`] owns the two halves of "project a decoded parquet
//! batch onto the scan's output schema":
//!
-//! * the [`ProjectionMask`] installed on every parquet decoder run, and
+//! * the [`ProjectionMask`] installed on the parquet decoder (and on any
+//! rebuild performed via `into_builder` at a row-group boundary), and
//! * the per-batch transform ([`DecoderProjection::map`]) that applies the
//! projector and, when needed, rebuilds the batch with the user's
//! `output_schema` to recover metadata / nullability the file schema does
@@ -46,13 +47,14 @@ use parquet::schema::types::SchemaDescriptor;
use crate::opener::{VirtualColumnsState, append_fields};
use crate::row_filter::build_projection_read_plan;
-/// Per-file decoder projection: the [`ProjectionMask`] installed on every
-/// parquet decoder run, plus the per-batch transform that maps the decoder's
+/// Per-file decoder projection: the [`ProjectionMask`] installed on the
+/// parquet decoder, plus the per-batch transform that maps the decoder's
/// output onto the scan's `output_schema`.
///
/// Built once per file by the opener via [`Self::try_new`]; the
-/// push-decoder stream installs [`Self::projection_mask`] on each decoder
-/// and calls [`Self::map`] on every decoded batch.
+/// push-decoder stream installs [`Self::projection_mask`] on the decoder
+/// (and on any rebuild performed via `into_builder` at a row-group
+/// boundary) and calls [`Self::map`] on every decoded batch.
pub(crate) struct DecoderProjection {
projection_mask: ProjectionMask,
projector: Projector,
diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs
index 4865975ec7..cbdcb73196 100644
--- a/datafusion/datasource-parquet/src/metrics.rs
+++ b/datafusion/datasource-parquet/src/metrics.rs
@@ -53,6 +53,14 @@ pub struct ParquetFileMetrics {
pub limit_pruned_row_groups: PruningMetrics,
/// Number of row groups pruned by statistics
pub row_groups_pruned_statistics: PruningMetrics,
+ /// Number of row groups pruned at runtime by a dynamic predicate
+ /// (e.g. the threshold expression a TopK `SortExec` pushes down).
+ ///
+ /// Unlike [`Self::row_groups_pruned_statistics`], which is decided once
+ /// at access-plan time, this counter reflects row groups that survived
+ /// the initial pruning but were proved unreachable mid-scan after the
+ /// dynamic filter tightened.
+ pub row_groups_pruned_dynamic_filter: Count,
/// Total number of bytes scanned
pub bytes_scanned: Count,
/// Total rows filtered out by predicates pushed into parquet scan
@@ -198,6 +206,11 @@ impl ParquetFileMetrics {
.with_category(MetricCategory::Rows)
.gauge("predicate_cache_records", partition);
+ let row_groups_pruned_dynamic_filter = MetricBuilder::new(metrics)
+ .with_new_label("filename", filename.to_string())
+ .with_type(MetricType::Summary)
+ .counter("row_groups_pruned_dynamic_filter", partition);
+
Self {
files_ranges_pruned_statistics,
predicate_evaluation_errors,
@@ -217,6 +230,7 @@ impl ParquetFileMetrics {
scan_efficiency_ratio,
predicate_cache_inner_records,
predicate_cache_records,
+ row_groups_pruned_dynamic_filter,
}
}
diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs
index 67abae69e7..4eba21bf02 100644
--- a/datafusion/datasource-parquet/src/opener/mod.rs
+++ b/datafusion/datasource-parquet/src/opener/mod.rs
@@ -26,7 +26,9 @@ use self::encryption::EncryptionContext;
use crate::access_plan::PreparedAccessPlan;
use crate::decoder_projection::DecoderProjection;
use crate::page_filter::PagePruningAccessPlanFilter;
-use crate::push_decoder::{DecoderBuilderConfig, PushDecoderStreamState};
+use crate::push_decoder::{
+ DecoderBuilderConfig, PushDecoderStreamState, RgPlanEntry, RowGroupPruner,
+};
use crate::row_filter::RowFilterGenerator;
use crate::row_group_filter::{BloomFilterStatistics, RowGroupAccessPlanFilter};
use crate::{
@@ -53,7 +55,7 @@ use datafusion_common::{
ColumnStatistics, HashSet, Result, ScalarValue, Statistics, exec_err, internal_err,
};
use datafusion_datasource::{PartitionedFile, TableSchema};
-use datafusion_physical_expr::expressions::Column;
+use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking};
use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;
use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
@@ -1379,7 +1381,7 @@ impl RowGroupsPrunedParquetOpen {
prepared.virtual_state.as_deref(),
)?;
- let (decoder, pending_decoders, remaining_limit) = {
+ let (decoder, rg_plan) = {
let pushdown_predicate = prepared
.pushdown_filters
.then_some(prepared.predicate.as_ref())
@@ -1392,50 +1394,41 @@ impl RowGroupsPrunedParquetOpen {
&prepared.file_metrics,
);
- // Split into consecutive runs of row groups that share the same filter
- // requirement. Fully matched row groups skip the RowFilter; others need it.
- // Reverse the run order for reverse scans so the combined decoder stream
- // preserves the requested global row group order.
- let mut runs = access_plan.split_runs(row_filter_generator.has_row_filter());
- if prepared.reverse_row_groups {
- runs.reverse();
- }
- let run_count = runs.len();
- let decoder_limit = prepared.limit.filter(|_| run_count == 1);
- let remaining_limit = prepared.limit.filter(|_| run_count > 1);
-
+ // Build the prepared access plan first — `prepare_access_plan` may
+ // call `reorder_by_statistics` (for `sort_order_for_reorder`) and
+ // `reverse` (for `reverse_row_groups`), both of which mutate
+ // `row_group_indexes` to the physical scan order the decoder will
+ // actually read. We MUST build our `rg_plan` from this reordered
+ // list, otherwise our per-RG pruner check would consult the
+ // metadata of a different RG than the decoder is about to yield.
let decoder_config = DecoderBuilderConfig {
projection_mask: decoder_projection.projection_mask(),
batch_size: prepared.batch_size,
arrow_reader_metrics: &arrow_reader_metrics,
force_filter_selections: prepared.force_filter_selections,
- decoder_limit,
+ decoder_limit: prepared.limit,
};
- // Build a decoder per run.
- let mut decoders = VecDeque::with_capacity(runs.len());
- for run in runs {
- let prepared_access_plan = prepare_access_plan(run.access_plan)?;
- let mut builder =
- decoder_config.build(prepared_access_plan, reader_metadata.clone());
- if run.needs_filter {
- if let Some(row_filter) = row_filter_generator.next_filter() {
- builder = builder.with_row_filter(row_filter);
- }
- if let Some(max_predicate_cache_size) =
- prepared.max_predicate_cache_size
- {
- builder = builder
- .with_max_predicate_cache_size(max_predicate_cache_size);
- }
+ let prepared_access_plan = prepare_access_plan(access_plan)?;
+ let rg_plan: VecDeque<RgPlanEntry> = prepared_access_plan
+ .row_group_indexes
+ .iter()
+ .copied()
+ .map(|rg_index| RgPlanEntry { rg_index })
+ .collect();
+
+ let mut builder =
+ decoder_config.build(prepared_access_plan, reader_metadata.clone());
+ if let Some(row_filter) = row_filter_generator.next_filter() {
+ builder = builder.with_row_filter(row_filter);
+ if let Some(max_predicate_cache_size) = prepared.max_predicate_cache_size
+ {
+ builder =
+ builder.with_max_predicate_cache_size(max_predicate_cache_size);
}
- decoders.push_back(builder.build()?);
}
- let decoder = decoders
- .pop_front()
- .expect("at least one decoder must be created");
- (decoder, decoders, remaining_limit)
+ (builder.build()?, rg_plan)
};
let predicate_cache_inner_records =
@@ -1445,16 +1438,52 @@ impl RowGroupsPrunedParquetOpen {
let files_ranges_pruned_statistics =
prepared.file_metrics.files_ranges_pruned_statistics.clone();
+
+ // Build a dynamic row-group pruner only when all three conditions hold:
+ // 1) the scan has a predicate (so there is something to evaluate),
+ // 2) the predicate has at least one not-yet-complete dynamic filter
+ // (`DynamicFilterTracking::Watching`) — static or already-complete
+ // predicates were fully consumed by `prune_by_statistics` at file
+ // open, so re-evaluating them per RG boundary would be wasted work,
+ // 3) there is at least one pending RG that could be skipped.
+ // The pruner subscribes once to every still-incomplete dynamic filter
+ // via the `DynamicFilterTracker` watch channel (#22460), so detecting
+ // a threshold change is a single atomic load — not a tree walk per
+ // RG check.
+ let row_group_pruner = match (&prepared.predicate, rg_plan.len() > 1) {
+ (Some(predicate), true)
+ if matches!(
+ DynamicFilterTracking::classify(predicate),
+ DynamicFilterTracking::Watching(_)
+ ) =>
+ {
+ Some(RowGroupPruner::new(
+ Arc::clone(predicate),
+ Arc::clone(&prepared.physical_file_schema),
+ Arc::clone(reader_metadata.metadata()),
+ prepared.predicate_creation_errors.clone(),
+ prepared.file_metrics.predicate_evaluation_errors.clone(),
+ ))
+ }
+ _ => None,
+ };
+ let row_groups_pruned_dynamic = prepared
+ .file_metrics
+ .row_groups_pruned_dynamic_filter
+ .clone();
+
let stream = PushDecoderStreamState {
- decoder,
- pending_decoders,
- remaining_limit,
+ decoder: Some(decoder),
+ active_reader: None,
+ rg_plan,
reader: prepared.async_file_reader,
decoder_projection,
arrow_reader_metrics,
predicate_cache_inner_records,
predicate_cache_records,
baseline_metrics: prepared.baseline_metrics,
+ row_group_pruner,
+ row_groups_pruned_dynamic,
}
.into_stream();
@@ -3247,92 +3276,6 @@ mod test {
assert_eq!(values, vec![7, 4, 5, 6, 3]);
}
- #[test]
- fn test_split_decoder_runs_no_fully_matched() {
- // All row groups need filtering: single run.
- let plan = ParquetAccessPlan::new(vec![
- RowGroupAccess::Scan,
- RowGroupAccess::Scan,
- RowGroupAccess::Scan,
- ]);
- let runs = plan.split_runs(true);
- assert_eq!(runs.len(), 1);
- assert!(runs[0].needs_filter);
- assert_eq!(runs[0].access_plan.row_group_indexes(), vec![0, 1, 2]);
- }
-
- #[test]
- fn test_split_decoder_runs_all_fully_matched() {
- // All row groups are fully matched: single run, no filter.
- let mut plan = ParquetAccessPlan::new(vec![
- RowGroupAccess::Scan,
- RowGroupAccess::Scan,
- RowGroupAccess::Scan,
- ]);
- plan.mark_fully_matched(0);
- plan.mark_fully_matched(1);
- plan.mark_fully_matched(2);
-
- let runs = plan.split_runs(true);
- assert_eq!(runs.len(), 1);
- assert!(!runs[0].needs_filter);
- assert_eq!(runs[0].access_plan.row_group_indexes(), vec![0, 1, 2]);
- }
-
- #[test]
- fn test_split_decoder_runs_mixed() {
- // [F, M, M, F, M] creates 4 runs preserving order.
- let mut plan = ParquetAccessPlan::new(vec![
- RowGroupAccess::Scan, // 0: filtered
- RowGroupAccess::Scan, // 1: matched
- RowGroupAccess::Scan, // 2: matched
- RowGroupAccess::Scan, // 3: filtered
- RowGroupAccess::Scan, // 4: matched
- ]);
- plan.mark_fully_matched(1);
- plan.mark_fully_matched(2);
- plan.mark_fully_matched(4);
-
- let runs = plan.split_runs(true);
- assert_eq!(runs.len(), 4);
-
- assert!(runs[0].needs_filter);
- assert_eq!(runs[0].access_plan.row_group_indexes(), vec![0]);
-
- assert!(!runs[1].needs_filter);
- assert_eq!(runs[1].access_plan.row_group_indexes(), vec![1, 2]);
-
- assert!(runs[2].needs_filter);
- assert_eq!(runs[2].access_plan.row_group_indexes(), vec![3]);
-
- assert!(!runs[3].needs_filter);
- assert_eq!(runs[3].access_plan.row_group_indexes(), vec![4]);
- }
-
- #[test]
- fn test_split_decoder_runs_with_skipped_groups() {
- // Skipped row groups are excluded from all runs.
- let mut plan = ParquetAccessPlan::new(vec![
- RowGroupAccess::Scan, // 0: filtered
- RowGroupAccess::Skip, // 1: pruned
- RowGroupAccess::Scan, // 2: matched
- RowGroupAccess::Scan, // 3: filtered
- ]);
- plan.mark_fully_matched(2);
-
- let runs = plan.split_runs(true);
- assert_eq!(runs.len(), 3);
-
- assert!(runs[0].needs_filter);
- assert_eq!(runs[0].access_plan.row_group_indexes(), vec![0]);
-
- assert!(!runs[1].needs_filter);
- assert_eq!(runs[1].access_plan.row_group_indexes(), vec![2]);
-
- assert!(runs[2].needs_filter);
- assert_eq!(runs[2].access_plan.row_group_indexes(), vec![3]);
- }
-
/// Helpers for tests that exercise parquet virtual columns
/// (e.g. `row_number`) plumbed through `TableSchema`/`ParquetOpener`.
mod virtual_columns {
diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs
index 3156b9e35f..31bd365a46 100644
--- a/datafusion/datasource-parquet/src/push_decoder.rs
+++ b/datafusion/datasource-parquet/src/push_decoder.rs
@@ -19,42 +19,54 @@
//!
//! This module owns the push-decoder lifecycle:
//!
-//! - [`DecoderBuilderConfig`] holds the shared options applied to every
-//! [`ParquetPushDecoderBuilder`] in a file scan, exposing a single `build`
-//! entry point per decoder run.
-//! - [`PushDecoderStreamState`] is the per-file stream driver that polls one
-//! or more decoders to completion, yielding projected [`RecordBatch`]es.
-//! A scan can produce multiple decoders (for example, when fully matched
-//! row groups split it into runs with different filter requirements); the
-//! state machine drains them in order so the output is contiguous.
+//! - [`DecoderBuilderConfig`] holds the shared options applied to the
+//! [`ParquetPushDecoderBuilder`] for a file scan, exposing a single `build`
+//! entry point.
+//! - [`PushDecoderStreamState`] is the per-file stream driver. It owns a
+//! **single** [`ParquetPushDecoder`] plus an [`RgPlanEntry`] queue
+//! (`rg_plan`) and uses arrow-rs's [`ParquetRecordBatchReader`] iterator
+//! to pause at row-group boundaries. At each boundary the optional
+//! [`RowGroupPruner`] is consulted; row groups it proves unwinnable are
+//! dropped from the head of `rg_plan` and the decoder is rebuilt via
+//! [`ParquetPushDecoder::into_builder`] +
+//! [`ParquetPushDecoderBuilder::with_row_groups`] so the skipped RGs are
+//! bypassed entirely — no decode, no row-filter eval.
//!
//! The opener constructs both halves and hands the state off to
//! [`PushDecoderStreamState::into_stream`] for consumption.
use std::collections::VecDeque;
+use std::sync::Arc;
use arrow::array::RecordBatch;
+use arrow::datatypes::SchemaRef;
use futures::StreamExt;
use futures::stream::BoxStream;
+use log::debug;
use parquet::DecodeResult;
use parquet::arrow::ProjectionMask;
use parquet::arrow::arrow_reader::metrics::ArrowReaderMetrics;
-use parquet::arrow::arrow_reader::{ArrowReaderMetadata, RowSelectionPolicy};
+use parquet::arrow::arrow_reader::{
+ ArrowReaderMetadata, ParquetRecordBatchReader, RowSelectionPolicy,
+};
use parquet::arrow::async_reader::AsyncFileReader;
use parquet::arrow::push_decoder::{ParquetPushDecoder, ParquetPushDecoderBuilder};
+use parquet::file::metadata::ParquetMetaData;
use datafusion_common::{DataFusionError, Result};
-use datafusion_physical_plan::metrics::{BaselineMetrics, Gauge};
+use datafusion_physical_expr::expressions::DynamicFilterTracking;
+use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+use datafusion_physical_plan::metrics::{BaselineMetrics, Count, Gauge};
+use datafusion_pruning::{PruningPredicate, build_pruning_predicate};
use crate::access_plan::PreparedAccessPlan;
use crate::decoder_projection::DecoderProjection;
+use crate::row_group_filter::RowGroupPruningStatistics;
-/// Shared options applied to every [`ParquetPushDecoderBuilder`] in a file scan.
-///
-/// A single scan may produce multiple decoders (for example, when fully matched
-/// row groups split the scan into consecutive runs with different filter
-/// requirements). All decoders in that scan share the same projection, batch
-/// size, metrics sink, and selection policy.
+/// Shared options applied to the [`ParquetPushDecoderBuilder`] for a file
+/// scan, and to any later rebuilds performed via
+/// [`ParquetPushDecoder::into_builder`] at row-group boundaries (e.g. when
+/// the [`RowGroupPruner`] drops subsequent row groups).
pub(crate) struct DecoderBuilderConfig<'a> {
/// Projection mask installed on every decoder in the scan. Sourced from
/// the file's [`DecoderProjection`].
@@ -66,9 +78,9 @@ pub(crate) struct DecoderBuilderConfig<'a> {
}
impl DecoderBuilderConfig<'_> {
- /// Build a [`ParquetPushDecoderBuilder`] for a single decoder run.
+ /// Build a [`ParquetPushDecoderBuilder`] from a prepared access plan.
///
- /// The caller is expected to attach the run-specific
+ /// The caller is expected to attach the
/// [`RowFilter`](parquet::arrow::arrow_reader::RowFilter) and predicate
/// cache size on the returned builder.
pub(crate) fn build(
@@ -94,6 +106,133 @@ impl DecoderBuilderConfig<'_> {
}
}
+#[derive(Debug, Clone)]
+pub(crate) struct RgPlanEntry {
+ pub(crate) rg_index: usize,
+}
+
+/// Runtime row-group pruner driven by a dynamic predicate (e.g. the
+/// threshold expression a `TopK` operator pushes down).
+///
+/// Mirrors the [`FilePruner`](datafusion_pruning::FilePruner) pattern at
+/// the row-group level: subscribes once to every still-incomplete dynamic
+/// filter inside the predicate via
+/// [`DynamicFilterTracker`](datafusion_physical_expr::expressions::DynamicFilterTracker)
+/// and only rebuilds the [`PruningPredicate`] when one of those
+/// subscriptions reports an update, then evaluates the cached predicate
+/// against the statistics of the requested row groups.
+pub(crate) struct RowGroupPruner {
+ predicate: Arc<dyn PhysicalExpr>,
+ arrow_schema: SchemaRef,
+ parquet_metadata: Arc<ParquetMetaData>,
+ /// Classifies the predicate's dynamic-filter content. The `Watching`
+ /// variant carries a tracker that subscribes to every not-yet-complete
+ /// dynamic filter; for `Static` / `AllComplete` the predicate cannot
+ /// change so a single up-front `pruning_predicate` build suffices.
+ tracking: DynamicFilterTracking,
+ /// First-call sentinel: forces an initial `pruning_predicate` build
+ /// even when `tracking` is `Static` / `AllComplete`.
+ needs_initial_build: bool,
+ /// Cached pruning predicate. `None` means we couldn't build one for the
+ /// current generation (e.g. the predicate has no analyzable bounds);
+ /// in that case we conservatively don't prune.
+ pruning_predicate: Option<Arc<PruningPredicate>>,
+ /// Metric for `build_pruning_predicate` failures (predicate creation).
+ predicate_creation_errors: Count,
+ /// Metric for `PruningPredicate::prune` failures (evaluating an
+ /// already-built predicate against row-group statistics).
+ predicate_evaluation_errors: Count,
+}
+
+impl RowGroupPruner {
+ pub(crate) fn new(
+ predicate: Arc<dyn PhysicalExpr>,
+ arrow_schema: SchemaRef,
+ parquet_metadata: Arc<ParquetMetaData>,
+ predicate_creation_errors: Count,
+ predicate_evaluation_errors: Count,
+ ) -> Self {
+ let tracking = DynamicFilterTracking::classify(&predicate);
+ Self {
+ predicate,
+ arrow_schema,
+ parquet_metadata,
+ tracking,
+ needs_initial_build: true,
+ pruning_predicate: None,
+ predicate_creation_errors,
+ predicate_evaluation_errors,
+ }
+ }
+
+ /// Returns `true` when the statistics for `row_group_indices` prove that
+ /// every requested row group can be skipped under the current value of
+ /// the dynamic predicate.
+ ///
+ /// On any error (predicate construction, statistics evaluation) the
+ /// pruner conservatively returns `false` and logs the failure, so a
+ /// flaky pruning path never silently drops data.
+ pub(crate) fn should_prune(&mut self, row_group_indices: &[usize]) -> bool {
+ if row_group_indices.is_empty() {
+ return false;
+ }
+
+ // Refresh the cached `PruningPredicate` on the first call and
+ // whenever a watched dynamic filter has advanced since we last
+ // looked. `changed()` is a single atomic load per still-incomplete
+ // filter — no tree walk on every check.
+ let dynamic_changed = self
+ .tracking
+ .watcher()
+ .is_some_and(|tracker| tracker.changed());
+ if self.needs_initial_build || dynamic_changed {
+ self.pruning_predicate = build_pruning_predicate(
+ Arc::clone(&self.predicate),
+ &self.arrow_schema,
+ &self.predicate_creation_errors,
+ );
+ self.needs_initial_build = false;
+ }
+
+ let Some(pp) = self.pruning_predicate.as_ref() else {
+ return false;
+ };
+
+ let row_group_metadatas = row_group_indices
+ .iter()
+ .map(|&i| self.parquet_metadata.row_group(i))
+ .collect::<Vec<_>>();
+ let stats = RowGroupPruningStatistics {
+ parquet_schema: self.parquet_metadata.file_metadata().schema_descr(),
+ row_group_metadatas,
+ arrow_schema: self.arrow_schema.as_ref(),
+ // Match the existing static row-group pruning behavior: when a
+ // statistic's null count is missing, treat it as zero. This is
+ // sound for runtime pruning because the predicate only needs to
+ // prove a row group *cannot* contain matching rows.
+ missing_null_counts_as_zero: true,
+ };
+
+ match pp.prune(&stats) {
+ // `prune` returns `false` per container that the predicate proves
+ // cannot contain matching rows. We can skip the run only when
+ // every requested row group is in that state.
+ Ok(values) => values.iter().all(|&keep| !keep),
+ Err(e) => {
+ // The predicate was already built successfully (we hold `pp`);
+ // this failure is in *evaluating* it against the row-group
+ // stats, so it belongs in the evaluation-errors counter, not
+ // creation-errors.
+ debug!(
+ "Ignoring error evaluating runtime row-group pruning predicate: {e}"
+ );
+ self.predicate_evaluation_errors.add(1);
+ false
+ }
+ }
+ }
+}
+
/// State for a stream that decodes a single Parquet file using a push-based decoder.
///
/// The [`transition`](Self::transition) method drives the decoder in a loop: it requests
@@ -101,17 +240,9 @@ impl DecoderBuilderConfig<'_> {
/// [`ParquetPushDecoder`], and yields projected [`RecordBatch`]es until the file is
/// fully consumed.
pub(crate) struct PushDecoderStreamState {
- pub(crate) decoder: ParquetPushDecoder,
- /// Additional decoders to process after the current one finishes.
- /// Used when fully matched row groups split the scan into consecutive
- /// runs with different filter configurations, maintaining original order.
- pub(crate) pending_decoders: VecDeque<ParquetPushDecoder>,
- /// Global remaining row limit across all decoder runs.
- ///
- /// Decoder-local limits are only safe for single-run scans. When the scan
- /// is split across multiple decoders, the combined stream limit is enforced
- /// here instead.
- pub(crate) remaining_limit: Option<usize>,
+ pub(crate) decoder: Option<ParquetPushDecoder>,
+ pub(crate) active_reader: Option<ParquetRecordBatchReader>,
+ pub(crate) rg_plan: VecDeque<RgPlanEntry>,
pub(crate) reader: Box<dyn AsyncFileReader>,
/// Per-file projection: the mask installed on every decoder and the
/// per-batch transform applied by [`Self::project_batch`].
@@ -120,6 +251,20 @@ pub(crate) struct PushDecoderStreamState {
pub(crate) predicate_cache_inner_records: Gauge,
pub(crate) predicate_cache_records: Gauge,
pub(crate) baseline_metrics: BaselineMetrics,
+ /// Dynamic row-group pruner consulted at every row-group boundary.
+ ///
+ /// When the file scan was opened with a still-watching dynamic predicate
+ /// (typically the threshold expression a `TopK` `SortExec` pushed down),
+ /// we re-evaluate that predicate against the next pending RG's
+ /// statistics and drop RGs the current threshold proves cannot
+ /// contribute. The decoder is rebuilt via
+ /// [`ParquetPushDecoder::into_builder`] +
+ /// [`ParquetPushDecoderBuilder::with_row_groups`] so the skipped RGs are
+ /// bypassed entirely. `None` when the scan has no watching dynamic
+ /// predicate or only one row group remains.
+ pub(crate) row_group_pruner: Option<RowGroupPruner>,
+ /// Count of row groups skipped at runtime by [`Self::row_group_pruner`].
+ pub(crate) row_groups_pruned_dynamic: Count,
}
impl PushDecoderStreamState {
@@ -148,10 +293,85 @@ impl PushDecoderStreamState {
/// with `unfold`'s ownership across yield points.
async fn transition(mut self) -> Option<(Result<RecordBatch>, Self)> {
loop {
- if self.remaining_limit == Some(0) {
- return None;
+ // Step 1: drain a batch from the active reader if any.
+ if let Some(reader) = self.active_reader.as_mut() {
+ match reader.next() {
+ Some(Ok(batch)) => {
+ let mut timer = self.baseline_metrics.elapsed_compute().timer();
+ self.copy_arrow_reader_metrics();
+ let result = self.project_batch(&batch);
+ timer.stop();
+ drop(timer);
+ return Some((result, self));
+ }
+ Some(Err(e)) => {
+ return Some((Err(DataFusionError::from(e)), self));
+ }
+ None => {
+ // Reader exhausted: drop and fall through to per-RG
+ // boundary handling, then try_next_reader.
+ self.active_reader = None;
+ }
+ }
}
- match self.decoder.try_decode() {
+
+ // Step 2: when the decoder is sitting on a row-group boundary,
+ // scan the entire `rg_plan` and drop every RG the pruner proves
+ // cannot contribute — head, interior, and tail alike. Evaluating
+ // per-RG stats against the cached `PruningPredicate` is cheap;
+ // the expensive part is the `into_builder` rebuild, so we do at
+ // most one rebuild per boundary regardless of how many RGs were
+ // dropped. Buffered bytes for already-fetched RGs carry across
+ // the rebuild.
+ //
+ // `into_builder` errors out mid-row-group, so we gate the prune
+ // pass on `is_at_row_group_boundary()`. When the decoder is
+ // mid-RG (e.g. byte ranges have been pushed but no reader has
+ // been handed back yet), step 3 drives it forward and we get
+ // another chance at the next boundary — the pruner is stateful
+ // and idempotent, so deferring loses nothing.
+ let at_boundary = self
+ .decoder
+ .as_ref()
+ .expect("decoder present")
+ .is_at_row_group_boundary();
+ if at_boundary && !self.rg_plan.is_empty() {
+ let mut pruned_count = 0usize;
+ if let Some(pruner) = self.row_group_pruner.as_mut() {
+ let mut kept = VecDeque::with_capacity(self.rg_plan.len());
+ while let Some(entry) = self.rg_plan.pop_front() {
+ if pruner.should_prune(&[entry.rg_index]) {
+ pruned_count += 1;
+ self.row_groups_pruned_dynamic.add(1);
+ } else {
+ kept.push_back(entry);
+ }
+ }
+ self.rg_plan = kept;
+ }
+ if pruned_count > 0 {
+ if self.rg_plan.is_empty() {
+ return None;
+ }
+ let decoder = self.decoder.take().expect("decoder present");
+ let new_indices: Vec<usize> =
+ self.rg_plan.iter().map(|e| e.rg_index).collect();
+ let rebuilt = match decoder.into_builder() {
+ Ok(b) => b.with_row_groups(new_indices).build(),
+ Err(e) => Err(e),
+ };
+ match rebuilt {
+ Ok(d) => self.decoder = Some(d),
+ Err(e) => {
+ return Some((Err(DataFusionError::from(e)), self));
+ }
+ }
+ }
+ }
+
+ // Step 3: drive the decoder.
+ let decoder = self.decoder.as_mut().expect("decoder present");
+ match decoder.try_next_reader() {
Ok(DecodeResult::NeedsData(ranges)) => {
let data = self
.reader
@@ -160,43 +380,26 @@ impl PushDecoderStreamState {
.map_err(DataFusionError::from);
match data {
Ok(data) => {
- if let Err(e) = self.decoder.push_ranges(ranges, data) {
+ if let Err(e) = self
+ .decoder
+ .as_mut()
+ .expect("decoder present")
+ .push_ranges(ranges, data)
+ {
return Some((Err(DataFusionError::from(e)), self));
}
}
Err(e) => return Some((Err(e), self)),
}
}
- Ok(DecodeResult::Data(batch)) => {
- let batch = if let Some(remaining_limit) = self.remaining_limit {
- if batch.num_rows() > remaining_limit {
- self.remaining_limit = Some(0);
- batch.slice(0, remaining_limit)
- } else {
- self.remaining_limit =
- Some(remaining_limit - batch.num_rows());
- batch
- }
- } else {
- batch
- };
- let mut timer = self.baseline_metrics.elapsed_compute().timer();
- self.copy_arrow_reader_metrics();
- let result = self.project_batch(&batch);
- timer.stop();
- // Release the borrow on baseline_metrics before moving self
- drop(timer);
- return Some((result, self));
- }
- Ok(DecodeResult::Finished) => {
- // If there are pending decoders (e.g. for consecutive runs
- // with different filter configurations), switch to the next.
- if let Some(next) = self.pending_decoders.pop_front() {
- self.decoder = next;
- continue;
- }
- return None;
+ Ok(DecodeResult::Data(reader)) => {
+ // Pop the RG this reader is for (we already filtered
+ // pruned ones in step 2, so `rg_plan.front()` is the RG
+ // the decoder is about to read).
+ self.rg_plan.pop_front();
+ self.active_reader = Some(reader);
}
+ Ok(DecodeResult::Finished) => return None,
Err(e) => {
return Some((Err(DataFusionError::from(e)), self));
}
@@ -219,3 +422,178 @@ impl PushDecoderStreamState {
self.decoder_projection.map(batch)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ use arrow::array::{Int64Array, RecordBatch};
+ use arrow::datatypes::{DataType, Field, Schema};
+ use bytes::Bytes;
+ use datafusion_common::ScalarValue;
+ use datafusion_expr::Operator;
+ use datafusion_physical_expr::expressions::{
+ BinaryExpr, Column, DynamicFilterPhysicalExpr, lit,
+ };
+ use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricBuilder};
+ use parquet::arrow::ArrowWriter;
+ use parquet::file::metadata::ParquetMetaDataPushDecoder;
+ use parquet::file::properties::WriterProperties;
+
+ /// Build a tiny in-memory Parquet file with three row groups whose `v`
+ /// column statistics are disjoint: RG0 → 0..1000, RG1 → 1000..2000,
+ /// RG2 → 2000..3000. Returns (metadata, schema).
+ fn build_three_rg_file() -> (Arc<ParquetMetaData>, SchemaRef) {
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ let mut buf = Vec::new();
+ let props = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(1000))
+ .build();
+ let mut writer =
+ ArrowWriter::try_new(&mut buf, Arc::clone(&schema), Some(props)).unwrap();
+ for rg in 0..3i64 {
+ let base = rg * 1000;
+ let vals: Vec<i64> = (base..base + 1000).collect();
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int64Array::from(vals))],
+ )
+ .unwrap();
+ writer.write(&batch).unwrap();
+ writer.flush().unwrap();
+ }
+ writer.close().unwrap();
+
+ let file = Bytes::from(buf);
+ let len = file.len() as u64;
+ let mut md = ParquetMetaDataPushDecoder::try_new(len).unwrap();
+ // One range covering the whole file. Using `expect` rather than
+ // `allow` per this crate's `clippy::allow-attributes` lint.
+ #[expect(
+ clippy::single_range_in_vec_init,
+ reason = "we want a single range covering the whole file"
+ )]
+ let ranges = vec![0..len];
+ md.push_ranges(ranges, vec![file]).unwrap();
+ let DecodeResult::Data(meta) = md.try_decode().unwrap() else {
+ panic!("decoding metadata");
+ };
+ assert_eq!(meta.num_row_groups(), 3, "test fixture must have 3 RGs");
+ (Arc::new(meta), schema)
+ }
+
+ /// Create a fresh `(creation_errors, evaluation_errors)` counter pair
+ /// for tests. The names mirror the two metrics
+ /// [`RowGroupPruner::new`] consumes — predicate construction is
+ /// accounted separately from per-row-group evaluation.
+ fn pruner_error_counters() -> (Count, Count) {
+ let metrics = ExecutionPlanMetricsSet::new();
+ let creation =
+ MetricBuilder::new(&metrics).counter("num_predicate_creation_errors", 0);
+ let evaluation =
+ MetricBuilder::new(&metrics).counter("predicate_evaluation_errors", 0);
+ (creation, evaluation)
+ }
+
+ /// `v > literal` predicate on a single-column schema.
+ fn gt_predicate(threshold: i64) -> Arc<dyn PhysicalExpr> {
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("v", 0)),
+ Operator::Gt,
+ lit(ScalarValue::Int64(Some(threshold))),
+ ))
+ }
+
+ #[test]
+ fn row_group_pruner_skips_only_disqualified_row_groups() {
+ let (meta, schema) = build_three_rg_file();
+ let (creation, evaluation) = pruner_error_counters();
+ let mut pruner = RowGroupPruner::new(
+ gt_predicate(1500),
+ Arc::clone(&schema),
+ Arc::clone(&meta),
+ creation,
+ evaluation,
+ );
+
+ // RG0 (0..1000) is entirely below threshold → fully prunable.
+ assert!(pruner.should_prune(&[0]), "RG0 should be pruned");
+ // RG1 (1000..2000) straddles the threshold → not safe to prune.
+ assert!(!pruner.should_prune(&[1]), "RG1 must NOT be pruned");
+ // RG2 (2000..3000) is entirely above threshold → keep.
+ assert!(!pruner.should_prune(&[2]), "RG2 must NOT be pruned");
+ // Run covering both RG0 and RG1 cannot be skipped — RG1 is alive.
+ assert!(
+ !pruner.should_prune(&[0, 1]),
+ "mixed run with a live RG must NOT be pruned"
+ );
+ // Empty input is a no-op (defensive guard).
+ assert!(!pruner.should_prune(&[]));
+ }
+
+ #[test]
+ fn row_group_pruner_tracks_dynamic_filter_updates() {
+ let (meta, schema) = build_three_rg_file();
+ let dynamic = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::new(Column::new("v", 0))],
+ gt_predicate(500),
+ ));
+ let (creation, evaluation) = pruner_error_counters();
+ let mut pruner = RowGroupPruner::new(
+ Arc::clone(&dynamic) as Arc<dyn PhysicalExpr>,
+ Arc::clone(&schema),
+ Arc::clone(&meta),
+ creation,
+ evaluation,
+ );
+
+ // Initial threshold 500 → only the lower half of RG0 fails, so RG0
+ // (0..1000) straddles the threshold and stays alive.
+ assert!(!pruner.should_prune(&[0]));
+ assert!(!pruner.should_prune(&[1]));
+
+ // Tighten the threshold via the dynamic filter — TopK fills its
+ // heap and updates the threshold to 2500.
+ dynamic
+ .update(gt_predicate(2500))
+ .expect("update threshold");
+
+ // After the update the pruner must rebuild its `PruningPredicate`
+ // (driven by the `DynamicFilterTracker`'s change notification) and
+ // re-evaluate. RG0 and RG1 are both entirely below 2500 now.
+ assert!(
+ pruner.should_prune(&[0]),
+ "RG0 must be pruned after threshold tightens to 2500"
+ );
+ assert!(
+ pruner.should_prune(&[1]),
+ "RG1 must be pruned after threshold tightens to 2500"
+ );
+ assert!(
+ !pruner.should_prune(&[2]),
+ "RG2 (2000..3000) still straddles 2500"
+ );
+ }
+
+ #[test]
+ fn row_group_pruner_falls_back_to_conservative_when_predicate_has_no_bounds() {
+ // A predicate the pruning analyzer can't decompose (e.g. a bare
+ // column reference of bool type would normally be valid, but a
+ // non-binary expression on a non-bool column doesn't yield bounds).
+ // We use `lit(true)` which produces no column references, so
+ // `build_pruning_predicate` will return None.
+ let (meta, schema) = build_three_rg_file();
+ let (creation, evaluation) = pruner_error_counters();
+ let mut pruner = RowGroupPruner::new(
+ lit(true) as Arc<dyn PhysicalExpr>,
+ Arc::clone(&schema),
+ Arc::clone(&meta),
+ creation,
+ evaluation,
+ );
+ // No pruning predicate could be built → conservatively keep RGs.
+ assert!(!pruner.should_prune(&[0]));
+ assert!(!pruner.should_prune(&[1]));
+ assert!(!pruner.should_prune(&[2]));
+ }
+}
diff --git a/datafusion/datasource-parquet/src/row_filter.rs b/datafusion/datasource-parquet/src/row_filter.rs
index 8ce359942c..3ec3bdff76 100644
--- a/datafusion/datasource-parquet/src/row_filter.rs
+++ b/datafusion/datasource-parquet/src/row_filter.rs
@@ -1094,12 +1094,11 @@ pub fn build_row_filter(
.map(|filters| Some(RowFilter::new(filters)))
}
-/// Builds row filters for decoder runs.
+/// Builds row filters for a parquet decoder.
///
-/// A [`RowFilter`] must be owned by a decoder, so scans split across multiple
-/// decoder runs need a fresh filter for each run that evaluates row predicates.
-/// The first filter is built eagerly during construction so callers can cheaply
-/// query [`has_row_filter`](Self::has_row_filter) before splitting the scan.
+/// A [`RowFilter`] is owned by a decoder. The first filter is built eagerly
+/// during construction so the caller can attach it to the decoder via
+/// [`next_filter`](Self::next_filter) without a redundant build call.
pub(crate) struct RowFilterGenerator<'a> {
predicate: Option<&'a Arc<dyn PhysicalExpr>>,
physical_file_schema: &'a SchemaRef,
@@ -1129,10 +1128,6 @@ impl<'a> RowFilterGenerator<'a> {
generator
}
- pub(crate) fn has_row_filter(&self) -> bool {
- self.first_row_filter.is_some()
- }
-
pub(crate) fn next_filter(&mut self) -> Option<RowFilter> {
self.first_row_filter.take().or_else(|| self.build())
}
diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs
index 8893c86a5c..bbf6cd3876 100644
--- a/datafusion/datasource-parquet/src/row_group_filter.rs
+++ b/datafusion/datasource-parquet/src/row_group_filter.rs
@@ -457,12 +457,16 @@ impl RowGroupAccessPlanFilter {
}
}
-/// Wraps a slice of [`RowGroupMetaData`] in a way that implements [`PruningStatistics`]
-struct RowGroupPruningStatistics<'a> {
- parquet_schema: &'a SchemaDescriptor,
- row_group_metadatas: Vec<&'a RowGroupMetaData>,
- arrow_schema: &'a Schema,
- missing_null_counts_as_zero: bool,
+/// Wraps a slice of [`RowGroupMetaData`] in a way that implements [`PruningStatistics`].
+///
+/// Visible to sibling modules so runtime row-group pruners (e.g. the dynamic
+/// TopK pruner in `push_decoder.rs`) can reuse this adapter without
+/// duplicating the statistics-to-`PruningStatistics` plumbing.
+pub(crate) struct RowGroupPruningStatistics<'a> {
+ pub(crate) parquet_schema: &'a SchemaDescriptor,
+ pub(crate) row_group_metadatas: Vec<&'a RowGroupMetaData>,
+ pub(crate) arrow_schema: &'a Schema,
+ pub(crate) missing_null_counts_as_zero: bool,
}
impl<'a> RowGroupPruningStatistics<'a> {
diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs
index 4c1e1b386d..a2503c1071 100644
--- a/datafusion/datasource-parquet/src/source.rs
+++ b/datafusion/datasource-parquet/src/source.rs
@@ -44,7 +44,7 @@ use datafusion_datasource::TableSchema;
use datafusion_datasource::file::FileSource;
use datafusion_datasource::file_scan_config::FileScanConfig;
use datafusion_functions::core::file_row_index::FileRowIndexFunc;
-use datafusion_physical_expr::expressions::Column;
+use datafusion_physical_expr::expressions::{Column, DynamicFilterTracking};
use datafusion_physical_expr::projection::ProjectionExprs;
use datafusion_physical_expr::{EquivalenceProperties, conjunction};
use datafusion_physical_expr_adapter::expr_references_scalar_udf;
@@ -740,6 +740,33 @@ impl FileSource for ParquetSource {
write!(f, ", reverse_row_groups=true")?;
}
+ // Plan-time marker for dynamic RG-level pruning: if the
+ // predicate is dynamic (e.g. a TopK threshold expression),
+ // the parquet opener will pause the single decoder at row
+ // group boundaries and consult `RowGroupPruner` to drop
+ // RGs the current threshold proves unwinnable, rebuilding
+ // the decoder via `into_builder().with_row_groups(...)` to
+ // skip them. The actual pruning count appears as
+ // `row_groups_pruned_dynamic_filter` in EXPLAIN ANALYZE.
+ // We use `contains_dynamic_filter()` (matches both `Watching`
+ // and `AllComplete`) rather than the stricter `Watching(_)`
+ // check the opener uses to construct the pruner. Reason: the
+ // opener gate is evaluated at file-open time, when a TopK
+ // threshold has not yet been pushed — at that moment a still-
+ // useful pruner needs `Watching`. `fmt_extra`, on the other
+ // hand, is called *also* by `EXPLAIN ANALYZE` after execution
+ // completes, at which point TopK has marked its dynamic
+ // filter complete and `classify` returns `AllComplete`. The
+ // marker is plan-time metadata ("this scan was eligible for
+ // runtime RG pruning"), so it should still show in that
+ // post-run rendering.
+ if let Some(predicate) = self.filter()
+ && DynamicFilterTracking::classify(&predicate)
+ .contains_dynamic_filter()
+ {
+ write!(f, ", dynamic_rg_pruning=eligible")?;
+ }
+
// Try to build the pruning predicates.
// These are only generated here because it's useful to have *some*
// idea of what pushdown is happening when viewing plans.
@@ -1184,6 +1211,88 @@ mod tests {
assert!(source.filter().is_some());
}
+ /// Render a `ParquetSource`'s `fmt_extra` output as a `String` for
+ /// inspection in tests.
+ fn render_fmt_extra(source: &ParquetSource, t: DisplayFormatType) -> String {
+ use std::fmt::Display;
+
+ struct Wrap<'a> {
+ source: &'a ParquetSource,
+ t: DisplayFormatType,
+ }
+ impl Display for Wrap<'_> {
+ fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+ self.source.fmt_extra(self.t, f)
+ }
+ }
+ Wrap { source, t }.to_string()
+ }
+
+ /// EXPLAIN must surface a `dynamic_rg_pruning=eligible` marker when the
+ /// predicate carries a `DynamicFilterPhysicalExpr`. This is the
+ /// plan-time signal that the runtime row-group pruner will fire at
+ /// every RG boundary.
+ #[test]
+ fn fmt_extra_marks_dynamic_predicate_as_pruning_eligible() {
+ use arrow::datatypes::{DataType, Field, Schema};
+ use datafusion_physical_expr::expressions::{Column, DynamicFilterPhysicalExpr};
+
+ let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int64, false)]));
+ let dynamic = Arc::new(DynamicFilterPhysicalExpr::new(
+ vec![Arc::new(Column::new("v", 0))],
+ lit(true),
+ )) as Arc<dyn PhysicalExpr>;
+
+ let source =
+ ParquetSource::new(Arc::clone(&schema)).with_predicate(Arc::clone(&dynamic));
+
+ let rendered = render_fmt_extra(&source, DisplayFormatType::Default);
+ assert!(
+ rendered.contains("dynamic_rg_pruning=eligible"),
+ "expected marker in Default fmt_extra, got: {rendered}"
+ );
+
+ let rendered_verbose = render_fmt_extra(&source, DisplayFormatType::Verbose);
+ assert!(
+ rendered_verbose.contains("dynamic_rg_pruning=eligible"),
+ "expected marker in Verbose fmt_extra, got: {rendered_verbose}"
+ );
+ }
+
+ /// EXPLAIN must NOT show the dynamic-RG-pruning marker when the
+ /// predicate is purely static — the optimization will not fire, so
+ /// surfacing it would mislead the reader.
+ #[test]
+ fn fmt_extra_omits_marker_for_static_predicate() {
+ use arrow::datatypes::Schema;
+
+ let schema = Arc::new(Schema::empty());
+ let predicate = lit(true);
+ let source = ParquetSource::new(schema).with_predicate(predicate);
+
+ let rendered = render_fmt_extra(&source, DisplayFormatType::Default);
+ assert!(
+ !rendered.contains("dynamic_rg_pruning"),
+ "did not expect marker for static predicate, got: {rendered}"
+ );
+ }
+
+ /// EXPLAIN must NOT show the marker when there is no predicate at all
+ /// (e.g. unfiltered table scan).
+ #[test]
+ fn fmt_extra_omits_marker_when_no_predicate() {
+ use arrow::datatypes::Schema;
+
+ let schema = Arc::new(Schema::empty());
+ let source = ParquetSource::new(schema);
+
+ let rendered = render_fmt_extra(&source, DisplayFormatType::Default);
+ assert!(
+ !rendered.contains("dynamic_rg_pruning"),
+ "did not expect marker for predicate-less scan, got: {rendered}"
+ );
+ }
+
/// Helpers for the `try_pushdown_sort` regression tests below.
mod pushdown_sort_helpers {
use super::*;
diff --git a/datafusion/sqllogictest/test_files/clickbench.slt b/datafusion/sqllogictest/test_files/clickbench.slt
index 60f7aadb8c..7c2a8bcaa1 100644
--- a/datafusion/sqllogictest/test_files/clickbench.slt
+++ b/datafusion/sqllogictest/test_files/clickbench.slt
@@ -652,7 +652,7 @@ physical_plan
03)----ProjectionExec: expr=[WatchID@0 as WatchID, JavaEnable@1 as JavaEnable, Title@2 as Title, GoodEvent@3 as GoodEvent, EventTime@4 as EventTime, CounterID@6 as CounterID, ClientIP@7 as ClientIP, RegionID@8 as RegionID, UserID@9 as UserID, CounterClass@10 as CounterClass, OS@11 as OS, UserAgent@12 as UserAgent, URL@13 as URL, Referer@14 as Referer, IsRefresh@15 as IsRefresh, RefererCategoryID@16 as RefererCategoryID, RefererRegionID@17 as RefererRegionID, URLCategoryID@18 as URLCatego [...]
04)------FilterExec: URL@13 LIKE %google%
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[WatchID, JavaEnable, Title, GoodEvent, EventTime, EventDate, CounterID, ClientIP, RegionID, UserID, CounterClass, OS, UserAgent, URL, Referer, IsRefresh, RefererCategoryID, RefererRegionID, URLCategoryID, URLRegionID, ResolutionWidth, ResolutionHeight, ResolutionDepth, FlashMajor, FlashMinor, FlashMinor2, NetMajor, NetMinor, UserAgentMajor, UserAgentMi [...]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[WatchID, JavaEnable, Title, GoodEvent, EventTime, EventDate, CounterID, ClientIP, RegionID, UserID, CounterClass, OS, UserAgent, URL, Referer, IsRefresh, RefererCategoryID, RefererRegionID, URLCategoryID, URLRegionID, ResolutionWidth, ResolutionHeight, ResolutionDepth, FlashMajor, FlashMinor, FlashMinor2, NetMajor, NetMinor, UserAgentMajor, UserAgentMi [...]
query IITIIIIIIIIITTIIIIIIIIIITIIITIIIITTIIITIIIIIIIIIITIIIIITIIIIIITIIIIIIIIIITTTTIIIIIIIITITTITTTTTTTTTTIIIID
SELECT * FROM hits WHERE "URL" LIKE '%google%' ORDER BY "EventTime" LIMIT 10;
@@ -675,7 +675,7 @@ physical_plan
03)----SortExec: TopK(fetch=10), expr=[EventTime@1 ASC NULLS LAST], preserve_partitioning=[true]
04)------FilterExec: SearchPhrase@1 != , projection=[SearchPhrase@1, EventTime@0]
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[EventTime, SearchPhrase], file_type=parquet, predicate=SearchPhrase@39 != AND DynamicFilter [ empty ], pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 != OR != SearchPhrase_max@1), required_guarantees=[SearchPhrase not in ()]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[EventTime, SearchPhrase], file_type=parquet, predicate=SearchPhrase@39 != AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 != OR != SearchPhrase_max@1), required_guarantees=[SearchPhrase not in ()]
query T
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "EventTime" LIMIT 10;
@@ -695,7 +695,7 @@ physical_plan
02)--SortExec: TopK(fetch=10), expr=[SearchPhrase@0 ASC NULLS LAST], preserve_partitioning=[true]
03)----FilterExec: SearchPhrase@0 !=
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[SearchPhrase], file_type=parquet, predicate=SearchPhrase@39 != AND DynamicFilter [ empty ], pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 != OR != SearchPhrase_max@1), required_guarantees=[SearchPhrase not in ()]
+05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[SearchPhrase], file_type=parquet, predicate=SearchPhrase@39 != AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 != OR != SearchPhrase_max@1), required_guarantees=[SearchPhrase not in ()]
query T
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "SearchPhrase" LIMIT 10;
@@ -718,7 +718,7 @@ physical_plan
03)----SortExec: TopK(fetch=10), expr=[EventTime@1 ASC NULLS LAST, SearchPhrase@0 ASC NULLS LAST], preserve_partitioning=[true]
04)------FilterExec: SearchPhrase@1 != , projection=[SearchPhrase@1, EventTime@0]
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[EventTime, SearchPhrase], file_type=parquet, predicate=SearchPhrase@39 != AND DynamicFilter [ empty ], pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 != OR != SearchPhrase_max@1), required_guarantees=[SearchPhrase not in ()]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/clickbench_hits_10.parquet]]}, projection=[EventTime, SearchPhrase], file_type=parquet, predicate=SearchPhrase@39 != AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=SearchPhrase_null_count@2 != row_count@3 AND (SearchPhrase_min@0 != OR != SearchPhrase_max@1), required_guarantees=[SearchPhrase not in ()]
query T
SELECT "SearchPhrase" FROM hits WHERE "SearchPhrase" <> '' ORDER BY "EventTime", "SearchPhrase" LIMIT 10;
diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
index 7ddebc235a..c58047c4ab 100644
--- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
+++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
@@ -90,7 +90,7 @@ logical_plan
02)--TableScan: test_parquet projection=[id, value, name]
physical_plan
01)SortExec: TopK(fetch=3), expr=[value@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[value@1 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[value@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
statement ok
set datafusion.explain.analyze_level = summary;
@@ -104,7 +104,7 @@ Plan with Metrics
03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, <slt:ignore>]
04)------FilterExec: value@1 > 3, metrics=[output_rows=10, <slt:ignore>, selectivity=100% (10/10)]
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, <slt:ignore>]
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=1 [...]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarante [...]
statement ok
set datafusion.explain.analyze_level = dev;
@@ -157,7 +157,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Disable Join dynamic filter pushdown
statement ok
@@ -235,7 +235,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# RIGHT JOIN correctness: all right rows appear, unmatched left rows produce NULLs
query ITT
@@ -284,7 +284,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(id@0, id@0)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# LEFT SEMI JOIN (physical LeftSemi): reverse table roles so optimizer keeps LeftSemi
# (right_parquet has 3 rows < left_parquet has 5 rows, so no swap occurs).
@@ -304,7 +304,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(id@0, id@0)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# LEFT SEMI (physical LeftSemi) correctness: only right rows with matching left ids
query IT rowsort
@@ -338,7 +338,7 @@ physical_plan
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet
03)--SortExec: expr=[data@1 DESC], preserve_partitioning=[false]
04)----FilterExec: DynamicFilter [ empty ]
-05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[data@1 DESC], reverse_row_groups=true
+05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[data@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
statement count 0
SET datafusion.execution.parquet.pushdown_filters = true;
@@ -361,7 +361,7 @@ physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(id@0, id@0)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet
03)--SortExec: expr=[data@1 DESC], preserve_partitioning=[false]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[data@1 DESC], reverse_row_groups=true
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[data@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
statement count 0
RESET datafusion.execution.parquet.pushdown_filters;
@@ -383,7 +383,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(id@0, id@0)], null_aware
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# LEFT MARK JOIN: the OR prevents decorrelation to LeftSemi, so the optimizer
# uses LeftMark. Self-generated dynamic filter pushes to the probe side.
@@ -407,7 +407,7 @@ physical_plan
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)]
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ]
+05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# LEFT MARK correctness: all right rows match EXISTS, so all 3 appear
query IT rowsort
@@ -444,8 +444,8 @@ logical_plan
physical_plan
01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(id@0, id@0)]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ]
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Correctness check
query IT
@@ -478,8 +478,8 @@ logical_plan
physical_plan
01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
02)--HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(id@0, id@0)], null_aware
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ]
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Correctness check
query IT
@@ -516,7 +516,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Enable TopK, disable Join
statement ok
@@ -588,7 +588,7 @@ physical_plan
02)--CoalescePartitionsExec
03)----AggregateExec: mode=Partial, gby=[], aggr=[max(agg_parquet.score)]
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/agg_data.parquet]]}, projection=[score], file_type=parquet, predicate=category@0 = alpha AND DynamicFilter [ empty ], pruning_predicate=category_null_count@2 != row_count@3 AND category_min@0 <= alpha AND alpha <= category_max@1, required_guarantees=[category in (alpha)]
+05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/agg_data.parquet]]}, projection=[score], file_type=parquet, predicate=category@0 = alpha AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=category_null_count@2 != row_count@3 AND category_min@0 <= alpha AND alpha <= category_max@1, required_guarantees=[category in (alpha)]
# Test 4b: COUNT + MAX — DynamicFilter should NOT appear here in mixed aggregates
@@ -736,7 +736,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Test 6: Regression test for issue #20213 - dynamic filter applied to wrong table
# when subquery join has same column names on both sides.
diff --git a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt
new file mode 100644
index 0000000000..2149cacfc0
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt
@@ -0,0 +1,112 @@
+# 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.
+
+# End-to-end SLT for **dynamic row-group pruning** driven by a TopK
+# `SortExec`'s `DynamicFilterPhysicalExpr`.
+#
+# Builds a 5-row-group parquet file with disjoint per-RG ranges of `v`:
+# RG 0: 0..3, RG 1: 3..6, RG 2: 6..9, RG 3: 9..12, RG 4: 12..15
+# `ORDER BY v DESC LIMIT 3` fills the TopK heap from the row group with
+# the largest values; the tightened threshold then proves every other
+# row group unreachable. At each row-group boundary the runtime
+# `RowGroupPruner` evaluates the current threshold against the next RGs'
+# statistics, drops the ones it proves unwinnable, and rebuilds the
+# decoder via `into_builder().with_row_groups(...)` to skip them. Each
+# drop bumps `row_groups_pruned_dynamic_filter`.
+
+statement ok
+set datafusion.execution.parquet.pushdown_filters = true;
+
+statement ok
+set datafusion.explain.analyze_level = summary;
+
+statement ok
+CREATE TABLE source_data AS VALUES
+-- RG 0
+ (0), (1), (2),
+-- RG 1
+ (3), (4), (5),
+-- RG 2
+ (6), (7), (8),
+-- RG 3
+ (9), (10), (11),
+-- RG 4
+ (12), (13), (14);
+
+statement ok
+COPY (SELECT column1 as v FROM source_data)
+TO 'test_files/scratch/dynamic_row_group_pruning/data.parquet'
+STORED AS PARQUET
+OPTIONS (
+ 'format.max_row_group_size' '3'
+);
+
+statement ok
+drop table source_data;
+
+statement ok
+CREATE EXTERNAL TABLE t
+STORED AS PARQUET
+LOCATION 'test_files/scratch/dynamic_row_group_pruning/data.parquet';
+
+# Sanity: query returns the right rows.
+query I
+SELECT v FROM t ORDER BY v DESC LIMIT 3;
+----
+14
+13
+12
+
+# Plain `EXPLAIN` must surface the plan-time eligibility marker
+# `dynamic_rg_pruning=eligible` on the `DataSourceExec` line: the
+# predicate is dynamic, so the runtime row-group pruner will be
+# consulted at each decoder-run boundary. This is the only knob users
+# have for spotting the optimization without running the query.
+query TT
+explain select v from t order by v desc limit 3;
+----
+logical_plan
+01)Sort: t.v DESC NULLS FIRST, fetch=3
+02)--TableScan: t projection=[v]
+physical_plan
+01)SortExec: TopK(fetch=3), expr=[v@0 DESC], preserve_partitioning=[false]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/data.parquet]]}, projection=[v], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[v@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
+
+# `EXPLAIN ANALYZE` must surface the runtime metric
+# `row_groups_pruned_dynamic_filter` with a non-zero value. Five
+# disjoint row groups, `LIMIT 3` fits inside the highest RG, so the
+# pruner skips the other four. Note the exact `=4`: the data is small
+# enough that the TopK heap fills in a single batch, and execution is
+# single-threaded, so the count is deterministic. Time- and size-keyed
+# fields are masked with `<slt:ignore>`.
+query TT
+explain analyze select v from t order by v desc limit 3;
+----
+Plan with Metrics
+01)SortExec: TopK(fetch=3), expr=[v@0 DESC], preserve_partitioning=[false], filter=[v@0 IS NULL OR v@0 > 12], metrics=[output_rows=3, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/data.parquet]]}, projection=[v], file_type=parquet, predicate=DynamicFilter [ v@0 IS NULL OR v@0 > 12 ], sort_order_for_reorder=[v@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=v_null_count@0 > 0 OR v_null_count@0 != row_count@2 AND v_max@1 > 12, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=<slt:ignore>, out [...]
+
+statement ok
+drop table t;
+
+# Config reset — without these the SLT runner flags the file for
+# leaking session state into subsequent files.
+statement ok
+RESET datafusion.execution.parquet.pushdown_filters;
+
+statement ok
+RESET datafusion.explain.analyze_level;
diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt
index b1856e0add..623580fce9 100644
--- a/datafusion/sqllogictest/test_files/explain_analyze.slt
+++ b/datafusion/sqllogictest/test_files/explain_analyze.slt
@@ -231,7 +231,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -247,7 +247,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -262,7 +262,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -277,7 +277,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -292,7 +292,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[elapsed_compute=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -319,7 +319,7 @@ EXPLAIN (ANALYZE, METRICS 'none', LEVEL summary) select * from cat_tracking wher
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- (METRICS 'rows', LEVEL summary) — row-count metrics only ----
@@ -328,7 +328,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', LEVEL summary) select * from cat_tracking wher
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- Quoted-string METRICS with multiple categories ----
@@ -337,7 +337,7 @@ EXPLAIN (ANALYZE, METRICS 'rows,bytes', LEVEL summary) select * from cat_trackin
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- (METRICS 'timing', LEVEL summary) — timing metrics only ----
@@ -346,7 +346,7 @@ EXPLAIN (ANALYZE, METRICS 'timing', LEVEL summary) select * from cat_tracking wh
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[elapsed_compute=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- TIMING sugar: `METRICS 'rows,bytes', TIMING off` ↔ rows+bytes only ----
# Equivalent to METRICS 'rows,bytes' since the sugar removes the timing
@@ -357,7 +357,7 @@ EXPLAIN (ANALYZE, METRICS 'rows,bytes', TIMING off, LEVEL summary) select * from
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- TIMING sugar: `METRICS 'rows', TIMING on` ↔ rows + timing ----
@@ -366,7 +366,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', TIMING on, LEVEL summary) select * from cat_tr
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, elapsed_compute=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- SUMMARY sugar: `SUMMARY on` ↔ `LEVEL summary` ----
# Equivalent to METRICS 'rows', LEVEL summary above.
@@ -376,7 +376,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', SUMMARY on) select * from cat_tracking where s
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- Statement option overrides session config ----
# Session says 'timing' but statement-level `METRICS 'rows'` wins.
@@ -389,7 +389,7 @@ EXPLAIN (ANALYZE, METRICS 'rows', LEVEL summary) select * from cat_tracking wher
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
# ---- pgjson format: structural golden with no metrics ----
@@ -451,7 +451,7 @@ EXPLAIN (ANALYZE, METRICS rows, LEVEL summary) select * from cat_tracking where
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND spe [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_c [...]
statement ok
reset datafusion.sql_parser.dialect;
diff --git a/datafusion/sqllogictest/test_files/limit.slt b/datafusion/sqllogictest/test_files/limit.slt
index ca2b36727d..b984705908 100644
--- a/datafusion/sqllogictest/test_files/limit.slt
+++ b/datafusion/sqllogictest/test_files/limit.slt
@@ -868,7 +868,7 @@ physical_plan
01)ProjectionExec: expr=[1 as foo]
02)--SortPreservingMergeExec: [part_key@0 ASC NULLS LAST], fetch=1
03)----SortExec: TopK(fetch=1), expr=[part_key@0 ASC NULLS LAST], preserve_partitioning=[true]
-04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/test_limit_with_partitions/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/test_limit_with_partitions/part-1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/test_limit_with_partitions/part-2.parquet]]}, projection=[part_key], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[part_key@0 AS [...]
+04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/test_limit_with_partitions/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/test_limit_with_partitions/part-1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/test_limit_with_partitions/part-2.parquet]]}, projection=[part_key], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[part_key@0 AS [...]
query I
with selection as (
diff --git a/datafusion/sqllogictest/test_files/limit_pruning.slt b/datafusion/sqllogictest/test_files/limit_pruning.slt
index 3c3f0222f3..4ef0b5c74f 100644
--- a/datafusion/sqllogictest/test_files/limit_pruning.slt
+++ b/datafusion/sqllogictest/test_files/limit_pruning.slt
@@ -120,7 +120,7 @@ explain analyze select * from tracking_data where species > 'M' AND s >= 50 orde
----
Plan with Metrics
01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, elapsed_compute=<slt:ignore>, output_bytes=<slt:ignore>]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND speci [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_cou [...]
statement ok
drop table tracking_data;
diff --git a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt
index 175d7d90cd..412d606df9 100644
--- a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt
+++ b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt
@@ -367,7 +367,7 @@ physical_plan
08)--------------FilterExec: service@2 = log
09)----------------RepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=1
10)------------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension/data.parquet]]}, projection=[d_dkey, env, service], file_type=parquet, predicate=service@2 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)]
-11)------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], file_type=parquet, [...]
+11)------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], file_type=parquet, [...]
# Verify results without optimization
query TTTIR rowsort
@@ -418,7 +418,7 @@ physical_plan
06)----------FilterExec: service@2 = log
07)------------RepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=1
08)--------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension/data.parquet]]}, projection=[d_dkey, env, service], file_type=parquet, predicate=service@2 = log, pruning_predicate=service_null_count@2 != row_count@3 AND service_min@0 <= log AND log <= service_max@1, required_guarantees=[service in (log)]
-09)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], file_type=parquet, pred [...]
+09)--------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], file_type=parquet, pred [...]
query TTTIR rowsort
SELECT f.f_dkey, MAX(d.env), MAX(d.service), count(*), sum(f.value)
@@ -643,7 +643,7 @@ physical_plan
05)--------RepartitionExec: partitioning=Hash([d_dkey@1], 3), input_partitions=3
06)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=C/data.parquet]]}, projection=[env, d_dkey], file_type=parquet
07)--------RepartitionExec: partitioning=Hash([f_dkey@1], 3), input_partitions=3
-08)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], file_type=parquet, predicate=DynamicFilter [ empty ]
+08)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg [...]
query TTR rowsort
SELECT f.f_dkey, d.env, sum(f.value)
diff --git a/datafusion/sqllogictest/test_files/projection_pushdown.slt b/datafusion/sqllogictest/test_files/projection_pushdown.slt
index 344aef1f92..c1cb8ed561 100644
--- a/datafusion/sqllogictest/test_files/projection_pushdown.slt
+++ b/datafusion/sqllogictest/test_files/projection_pushdown.slt
@@ -444,7 +444,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as simple_struct.s[value]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as simple_struct.s[value]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query II
@@ -467,7 +467,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) + 1 as simple_struct.s[value] + Int64(1)], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) + 1 as simple_struct.s[value] + Int64(1)], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query II
@@ -490,7 +490,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as simple_struct.s[value], get_field(s@1, label) as simple_struct.s[label]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as simple_struct.s[value], get_field(s@1, label) as simple_struct.s[label]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query IIT
@@ -513,7 +513,7 @@ logical_plan
03)----TableScan: nested_struct projection=[id, nested]
physical_plan
01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/nested.parquet]]}, projection=[id, get_field(nested@1, outer, inner) as nested_struct.nested[outer][inner]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/nested.parquet]]}, projection=[id, get_field(nested@1, outer, inner) as nested_struct.nested[outer][inner]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query II
@@ -535,7 +535,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, label) || _suffix as simple_struct.s[label] || Utf8("_suffix")], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, label) || _suffix as simple_struct.s[label] || Utf8("_suffix")], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query IT
@@ -595,7 +595,7 @@ physical_plan
01)SortExec: TopK(fetch=2), expr=[simple_struct.s[value]@1 ASC NULLS LAST], preserve_partitioning=[false]
02)--ProjectionExec: expr=[id@1 as id, __datafusion_extracted_1@0 as simple_struct.s[value]]
03)----FilterExec: id@1 > 1
-04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet, predicate=id@0 > 1 AND DynamicFilter [ empty ], pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 > 1, required_guarantees=[]
+04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet, predicate=id@0 > 1 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 > 1, required_guarantees=[]
# Verify correctness
query II
@@ -621,7 +621,7 @@ physical_plan
01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
02)--ProjectionExec: expr=[id@1 as id, __datafusion_extracted_1@0 + 1 as simple_struct.s[value] + Int64(1)]
03)----FilterExec: id@1 > 1
-04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet, predicate=id@0 > 1 AND DynamicFilter [ empty ], pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 > 1, required_guarantees=[]
+04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet, predicate=id@0 > 1 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 > 1, required_guarantees=[]
# Verify correctness
query II
@@ -713,7 +713,7 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [id@0 ASC NULLS LAST], fetch=3
02)--SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part2.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part3.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part4.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_ [...]
+03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part2.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part3.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part4.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_ [...]
# Verify correctness
query II
@@ -737,7 +737,7 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [id@0 ASC NULLS LAST], fetch=3
02)--SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part2.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part3.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part4.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_ [...]
+03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part2.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part3.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/multi/part4.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_ [...]
# Verify correctness
query II
@@ -874,7 +874,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as simple_struct.s[value], get_field(s@1, value) + 10 as simple_struct.s[value] + Int64(10), get_field(s@1, label) as simple_struct.s[label]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as simple_struct.s[value], get_field(s@1, value) + 10 as simple_struct.s[value] + Int64(10), get_field(s@1, label) as simple_struct.s[label]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query IIIT
@@ -897,7 +897,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, 42 as constant], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, 42 as constant], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query II
@@ -919,7 +919,7 @@ logical_plan
02)--TableScan: simple_struct projection=[id]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query I
@@ -947,7 +947,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, id@0 + 100 as computed], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, id@0 + 100 as computed], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query II
@@ -1039,7 +1039,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) + id@0 as combined], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) + id@0 as combined], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query II
@@ -1095,7 +1095,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, 42 as answer, get_field(s@1, label) as simple_struct.s[label]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, 42 as answer, get_field(s@1, label) as simple_struct.s[label]], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query IIT
@@ -1118,7 +1118,7 @@ logical_plan
03)----TableScan: simple_struct projection=[id, s]
physical_plan
01)SortExec: TopK(fetch=2), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) + 100 as simple_struct.s[value] + Int64(100), get_field(s@1, label) || _test as simple_struct.s[label] || Utf8("_test")], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) + 100 as simple_struct.s[value] + Int64(100), get_field(s@1, label) || _test as simple_struct.s[label] || Utf8("_test")], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Verify correctness
query IIT
@@ -1317,7 +1317,7 @@ logical_plan
physical_plan
01)ProjectionExec: expr=[id@0 as id]
02)--SortExec: TopK(fetch=2), expr=[__datafusion_extracted_1@1 ASC NULLS LAST], preserve_partitioning=[false]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as __datafusion_extracted_1], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id, get_field(s@1, value) as __datafusion_extracted_1], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness
query I
@@ -1424,7 +1424,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(__datafusion_extracted_1@0, __datafusion_extracted_2 * Int64(10)@2)], projection=[id@1, id@3]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_2, id, get_field(s@1, level) * 10 as __datafusion_extracted_2 * Int64(10)], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_2, id, get_field(s@1, level) * 10 as __datafusion_extracted_2 * Int64(10)], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness - value = level * 10
# simple_struct: (1,100), (2,200), (3,150), (4,300), (5,250)
@@ -1460,7 +1460,7 @@ physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)]
02)--FilterExec: __datafusion_extracted_1@0 > 150, projection=[id@1]
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet, predicate=get_field(s@1, value) > 150
-04)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ]
+04)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness - id matches and value > 150
query II
@@ -1500,7 +1500,7 @@ physical_plan
02)--FilterExec: __datafusion_extracted_1@0 > 100, projection=[id@1]
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet, predicate=get_field(s@1, value) > 100
04)--FilterExec: __datafusion_extracted_2@0 > 3, projection=[id@1]
-05)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_2, id], file_type=parquet, predicate=get_field(s@1, level) > 3 AND DynamicFilter [ empty ]
+05)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_2, id], file_type=parquet, predicate=get_field(s@1, level) > 3 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness - id matches, value > 100, and level > 3
# Matching ids where value > 100: 2(200), 3(150), 4(300), 5(250)
@@ -1536,7 +1536,7 @@ physical_plan
01)ProjectionExec: expr=[id@0 as id, __datafusion_extracted_1@1 as simple_struct.s[label], __datafusion_extracted_2@2 as join_right.s[role]]
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@1, id@1)], projection=[id@1, __datafusion_extracted_1@0, __datafusion_extracted_2@2]
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, label) as __datafusion_extracted_1, id], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, role) as __datafusion_extracted_2, id], file_type=parquet, predicate=DynamicFilter [ empty ]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, role) as __datafusion_extracted_2, id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness
query ITT
@@ -1568,7 +1568,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[id], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness
query II
@@ -1607,7 +1607,7 @@ physical_plan
02)--HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@1, id@0)], projection=[id@1, __datafusion_extracted_2@0, __datafusion_extracted_3@3]
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_2, id], file_type=parquet
04)----FilterExec: __datafusion_extracted_1@0 > 5, projection=[id@1, __datafusion_extracted_3@2]
-05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_1, id, get_field(s@1, level) as __datafusion_extracted_3], file_type=parquet, predicate=get_field(s@1, level) > 5 AND DynamicFilter [ empty ]
+05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_1, id, get_field(s@1, level) as __datafusion_extracted_3], file_type=parquet, predicate=get_field(s@1, level) > 5 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness - left join with level > 5 condition
# Only join_right rows with level > 5 are matched: id=1 (level=10), id=4 (level=8)
@@ -1899,7 +1899,7 @@ physical_plan
01)ProjectionExec: expr=[__datafusion_extracted_3@0 as s.s[value], __datafusion_extracted_4@1 as j.s[role]]
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@2, id@2)], filter=__datafusion_extracted_1@1 > __datafusion_extracted_2@0, projection=[__datafusion_extracted_3@4, __datafusion_extracted_4@1]
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_2, get_field(s@1, role) as __datafusion_extracted_4, id], file_type=parquet
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, get_field(s@1, value) as __datafusion_extracted_3, id], file_type=parquet, predicate=DynamicFilter [ empty ]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, get_field(s@1, value) as __datafusion_extracted_3, id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness - only admin roles match (ids 1 and 4)
query II
@@ -1935,7 +1935,7 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@1, id@1)], filter=__datafusion_extracted_1@0 > __datafusion_extracted_2@1, projection=[id@1, id@3]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/simple.parquet]]}, projection=[get_field(s@1, value) as __datafusion_extracted_1, id], file_type=parquet
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_2, id], file_type=parquet, predicate=DynamicFilter [ empty ]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/projection_pushdown/join_right.parquet]]}, projection=[get_field(s@1, level) as __datafusion_extracted_2, id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify correctness - all rows match since value >> level for all ids
# simple_struct: (1,100), (2,200), (3,150), (4,300), (5,250)
diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
index b5a06bc7cb..6cb92025ca 100644
--- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
+++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
@@ -158,7 +158,7 @@ physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/small_table.parquet]]}, projection=[k], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet
03)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/large_table.parquet]]}, projection=[k, v], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet, predicate=v@1 >= 50 AND DynamicFilter [ empty ], pruning_predicate=v_null_count@1 != row_count@2 AND v_max@0 >= 50, required_guarantees=[]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/large_table.parquet]]}, projection=[k, v], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet, predicate=v@1 >= 50 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=v_null_count@1 != row_count@2 AND v_max@0 >= 50, required_guarantees=[]
statement ok
drop table small_table;
@@ -206,7 +206,7 @@ EXPLAIN ANALYZE SELECT t FROM topk_pushdown ORDER BY t * t LIMIT 10;
----
Plan with Metrics
01)SortExec: TopK(fetch=10), expr=[t@0 * t@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[t@0 * t@0 < 1884329474306198481], metrics=[output_rows=10, output_batches=1, row_replacements=10]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_pushdown.parquet]]}, projection=[t], output_ordering=[t@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ t@0 * t@0 < 1884329474306198481 ], metrics=[output_rows=128, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=782 total → 782 matched, row_groups_pruned_bloom_filter=782 total → 782 matched [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_pushdown.parquet]]}, projection=[t], output_ordering=[t@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ t@0 * t@0 < 1884329474306198481 ], dynamic_rg_pruning=eligible, metrics=[output_rows=128, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=782 total → 782 matched, row_groups_pruned_bloom_f [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -257,7 +257,7 @@ EXPLAIN SELECT * FROM topk_single_col ORDER BY b DESC LIMIT 1;
----
physical_plan
01)SortExec: TopK(fetch=1), expr=[b@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_single_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[b@1 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_single_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[b@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
statement ok
set datafusion.explain.analyze_categories = 'rows';
@@ -268,7 +268,7 @@ EXPLAIN ANALYZE SELECT * FROM topk_single_col ORDER BY b DESC LIMIT 1;
----
Plan with Metrics
01)SortExec: TopK(fetch=1), expr=[b@1 DESC], preserve_partitioning=[false], filter=[b@1 IS NULL OR b@1 > bd], metrics=[output_rows=1, output_batches=1, row_replacements=1]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_single_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 IS NULL OR b@1 > bd ], sort_order_for_reorder=[b@1 DESC], reverse_row_groups=true, pruning_predicate=b_null_count@0 > 0 OR b_null_count@0 != row_count@2 AND b_max@1 > bd, required_guarantees=[], metrics=[output_rows=4, output_batches=1, files_ranges_pruned_statisti [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_single_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 IS NULL OR b@1 > bd ], sort_order_for_reorder=[b@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@0 > 0 OR b_null_count@0 != row_count@2 AND b_max@1 > bd, required_guarantees=[], metrics=[output_rows=4, output_batches=1, [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -319,7 +319,7 @@ EXPLAIN ANALYZE SELECT * FROM topk_multi_col ORDER BY b ASC NULLS LAST, a DESC L
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[b@1 ASC NULLS LAST, a@0 DESC], preserve_partitioning=[false], filter=[b@1 < bb OR b@1 = bb AND (a@0 IS NULL OR a@0 > ac)], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_multi_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 < bb OR b@1 = bb AND (a@0 IS NULL OR a@0 > ac) ], sort_order_for_reorder=[b@1 ASC NULLS LAST, a@0 DESC], pruning_predicate=b_null_count@1 != row_count@2 AND b_min@0 < bb OR b_null_count@1 != row_count@2 AND b_min@0 <= bb AND bb <= b_max@3 AND (a_null_count@4 > 0 OR [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_multi_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 < bb OR b@1 = bb AND (a@0 IS NULL OR a@0 > ac) ], sort_order_for_reorder=[b@1 ASC NULLS LAST, a@0 DESC], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_min@0 < bb OR b_null_count@1 != row_count@2 AND b_min@0 <= bb AND bb <= b_max@ [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -389,7 +389,7 @@ FROM join_probe p INNER JOIN join_build AS build
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_ro [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_nul [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -475,8 +475,8 @@ Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c@3, d@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, b@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t1.parquet]]}, projection=[a, x], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t2.parquet]]}, projection=[b, c, y], file_type=parquet, predicate=DynamicFilter [ b@0 >= aa AND b@0 <= ab AND b@0 IN (SET) ([aa, ab]) ], pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= aa AND b_null_count@1 != row_count@2 AND b_min@3 <= ab AND (b_null_count@1 != row_count@2 AND b_min@3 <= aa AND aa <= b_max@0 OR b_null_count@1 != row_coun [...]
-05)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t3.parquet]]}, projection=[d, z], file_type=parquet, predicate=DynamicFilter [ d@0 >= ca AND d@0 <= cb AND hash_lookup ], pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= ca AND d_null_count@1 != row_count@2 AND d_min@3 <= cb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched [...]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t2.parquet]]}, projection=[b, c, y], file_type=parquet, predicate=DynamicFilter [ b@0 >= aa AND b@0 <= ab AND b@0 IN (SET) ([aa, ab]) ], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= aa AND b_null_count@1 != row_count@2 AND b_min@3 <= ab AND (b_null_count@1 != row_count@2 AND b_min@3 <= aa AND aa <= b_max@0 [...]
+05)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t3.parquet]]}, projection=[d, z], file_type=parquet, predicate=DynamicFilter [ d@0 >= ca AND d@0 <= cb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= ca AND d_null_count@1 != row_count@2 AND d_min@3 <= cb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_s [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -541,7 +541,7 @@ physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, d@0)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/parent_build.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=a@0 = aa, pruning_predicate=a_null_count@2 != row_count@3 AND a_min@0 <= aa AND aa <= a_max@1, required_guarantees=[a in (aa)]
03)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/parent_probe.parquet]]}, projection=[d, e, f], file_type=parquet, predicate=e@1 = ba AND d@0 = aa AND DynamicFilter [ empty ], pruning_predicate=e_null_count@2 != row_count@3 AND e_min@0 <= ba AND ba <= e_max@1 AND d_null_count@6 != row_count@3 AND d_min@4 <= aa AND aa <= d_max@5, required_guarantees=[d in (aa), e in (ba)]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/parent_probe.parquet]]}, projection=[d, e, f], file_type=parquet, predicate=e@1 = ba AND d@0 = aa AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=e_null_count@2 != row_count@3 AND e_min@0 <= ba AND ba <= e_max@1 AND d_null_count@6 != row_count@3 AND d_min@4 <= aa AND aa <= d_max@5, required_guarantees=[d in (aa), e in (ba)]
statement ok
drop table parent_build;
@@ -606,7 +606,7 @@ Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[e@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[e@0 < bb], metrics=[output_rows=2, output_batches=1, row_replacements=2]
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, d@0)], projection=[e@2], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_build.parquet]]}, projection=[a], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_r [...]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_probe.parquet]]}, projection=[d, e], file_type=parquet, predicate=DynamicFilter [ d@0 >= aa AND d@0 <= ab AND d@0 IN (SET) ([aa, ab]) ] AND DynamicFilter [ e@1 < bb ], pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= aa AND d_null_count@1 != row_count@2 AND d_min@3 <= ab AND (d_null_count@1 != row_count@2 AND d_min@3 <= aa AND aa <= d_m [...]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_probe.parquet]]}, projection=[d, e], file_type=parquet, predicate=DynamicFilter [ d@0 >= aa AND d@0 <= ab AND d@0 IN (SET) ([aa, ab]) ] AND DynamicFilter [ e@1 < bb ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= aa AND d_null_count@1 != row_count@2 AND d_min@3 <= ab AND (d_null_count@1 != row_count@2 AN [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -655,7 +655,7 @@ EXPLAIN ANALYZE SELECT b, a FROM topk_proj ORDER BY a LIMIT 2;
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[a@1 ASC NULLS LAST], preserve_partitioning=[false], filter=[a@1 < 2], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[b, a], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@1 ASC NULLS LAST], pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[b, a], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@1 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups [...]
# Case 2: prune — `SELECT a` — filter stays as `a < 2` on the scan.
query TT
@@ -663,7 +663,7 @@ EXPLAIN ANALYZE SELECT a FROM topk_proj ORDER BY a LIMIT 2;
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[a@0 < 2], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 m [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pr [...]
# Case 3: expression — `SELECT a+1 AS a_plus_1` — the TopK filter is on
# `a_plus_1`, the scan predicate must read `a@0 + 1`.
@@ -672,7 +672,7 @@ EXPLAIN ANALYZE SELECT a + 1 AS a_plus_1, b FROM topk_proj ORDER BY a_plus_1 LIM
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[a_plus_1@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[a_plus_1@0 < 3], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a_plus_1, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pr [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a_plus_1, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], dynamic_rg_pruning=eligible, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 [...]
# Case 4: alias shadowing — `SELECT a+1 AS a` — the projection renames
# `a+1` to `a`, so the TopK's `a < 3` must still be rewritten to
@@ -682,7 +682,7 @@ EXPLAIN ANALYZE SELECT a + 1 AS a, b FROM topk_proj ORDER BY a LIMIT 2;
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[a@0 < 3], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_ [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -744,7 +744,7 @@ Plan with Metrics
04)----AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[min(join_agg_probe.value)], metrics=[output_rows=2, output_batches=2, spill_count=0, spilled_rows=0]
05)------RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=1, metrics=[output_rows=2, output_batches=2, spill_count=0, spilled_rows=0]
06)--------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[min(join_agg_probe.value)], metrics=[output_rows=2, output_batches=1, spill_count=0, spilled_rows=0, skipped_aggregation_rows=0, reduction_factor=100% (2/2)]
-07)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_agg_probe.parquet]]}, projection=[a, value], file_type=parquet, predicate=DynamicFilter [ a@0 >= h1 AND a@0 <= h2 AND a@0 IN (SET) ([h1, h2]) ], pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= h1 AND a_null_count@1 != row_count@2 AND a_min@3 <= h2 AND (a_null_count@1 != row_count@2 AND a_min@3 <= h1 AND h1 <= a_max@0 OR a_null_count@1 [...]
+07)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_agg_probe.parquet]]}, projection=[a, value], file_type=parquet, predicate=DynamicFilter [ a@0 >= h1 AND a@0 <= h2 AND a@0 IN (SET) ([h1, h2]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= h1 AND a_null_count@1 != row_count@2 AND a_min@3 <= h2 AND (a_null_count@1 != row_count@2 AND a_min@3 <= h1 AND h1 [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -807,7 +807,7 @@ ON nulls_build.a = nulls_probe.a AND nulls_build.b = nulls_probe.b;
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=1, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=3, input_batches=1, input_rows=1, avg_fanout=100% (1/1), probe_hit_rate=100% (1/1)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_build.parquet]]}, projection=[a, b], file_type=parquet, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_probe.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= 1 AND b@1 <= 2 AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:1}, {c0:,c1:2}, {c0:ab,c1:}]) ], pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_coun [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_probe.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= 1 AND b@1 <= 2 AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:1}, {c0:,c1:2}, {c0:ab,c1:}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab A [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -873,7 +873,7 @@ ON lj_build.a = lj_probe.a AND lj_build.b = lj_probe.b;
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_ [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_ [...]
# LEFT SEMI JOIN: only matching build rows are returned; probe scan still
# receives the dynamic filter.
@@ -889,7 +889,7 @@ WHERE EXISTS (
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=4, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_cou [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -959,7 +959,7 @@ FROM hl_probe p INNER JOIN hl_build AS build
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND hash_lookup ], pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_null_count@5 != row_count@2 A [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_ [...]
statement ok
drop table hl_build;
@@ -1008,7 +1008,7 @@ FROM int_build b INNER JOIN int_probe p
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id1@0, id1@0), (id2@1, id2@1)], projection=[id1@0, id2@1, value@2, data@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_build.parquet]]}, projection=[id1, id2, value], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pr [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_probe.parquet]]}, projection=[id1, id2, data], file_type=parquet, predicate=DynamicFilter [ id1@0 >= 1 AND id1@0 <= 2 AND id2@1 >= 10 AND id2@1 <= 20 AND hash_lookup ], pruning_predicate=id1_null_count@1 != row_count@2 AND id1_max@0 >= 1 AND id1_null_count@1 != row_count@2 AND id1_min@3 <= 2 AND id2_null_count@5 != row_count@2 AND id2_max@4 >= 10 AND id2_nul [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_probe.parquet]]}, projection=[id1, id2, data], file_type=parquet, predicate=DynamicFilter [ id1@0 >= 1 AND id1@0 <= 2 AND id2@1 >= 10 AND id2@1 <= 20 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=id1_null_count@1 != row_count@2 AND id1_max@0 >= 1 AND id1_null_count@1 != row_count@2 AND id1_min@3 <= 2 AND id2_null_count@5 != row_count@2 AN [...]
statement ok
reset datafusion.explain.analyze_categories;
diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt
index b86bd2c51d..f28a314764 100644
--- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt
+++ b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt
@@ -146,7 +146,7 @@ physical_plan
01)AggregateExec: mode=Final, gby=[], aggr=[max(agg_dyn_test.id)]
02)--CoalescePartitionsExec
03)----AggregateExec: mode=Partial, gby=[], aggr=[max(agg_dyn_test.id)]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-01/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-02/j5fUeSDQo22oPyPU.parquet], [WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-03/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-04/j5fUeSDQo2 [...]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-01/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-02/j5fUeSDQo22oPyPU.parquet], [WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-03/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-04/j5fUeSDQo2 [...]
query I
select max(id) from agg_dyn_test where id > 1;
@@ -161,7 +161,7 @@ physical_plan
01)AggregateExec: mode=Final, gby=[], aggr=[max(agg_dyn_test.id)]
02)--CoalescePartitionsExec
03)----AggregateExec: mode=Partial, gby=[], aggr=[max(agg_dyn_test.id)]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-01/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-02/j5fUeSDQo22oPyPU.parquet], [WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-03/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-04/j5fUeSDQo2 [...]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-01/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-02/j5fUeSDQo22oPyPU.parquet], [WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-03/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-04/j5fUeSDQo2 [...]
# Expect dynamic filter available inside data source
query TT
@@ -171,7 +171,7 @@ physical_plan
01)AggregateExec: mode=Final, gby=[], aggr=[max(agg_dyn_test.id), min(agg_dyn_test.id)]
02)--CoalescePartitionsExec
03)----AggregateExec: mode=Partial, gby=[], aggr=[max(agg_dyn_test.id), min(agg_dyn_test.id)]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-01/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-02/j5fUeSDQo22oPyPU.parquet], [WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-03/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-04/j5fUeSDQo2 [...]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-01/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-02/j5fUeSDQo22oPyPU.parquet], [WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-03/j5fUeSDQo22oPyPU.parquet, WORKSPACE_ROOT/datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-04/j5fUeSDQo2 [...]
# Dynamic filter should not be available for grouping sets
query TT
@@ -236,7 +236,7 @@ Plan with Metrics
01)AggregateExec: mode=Final, gby=[], aggr=[max(agg_dyn_e2e.column1)], metrics=[]
02)--CoalescePartitionsExec, metrics=[]
03)----AggregateExec: mode=Partial, gby=[], aggr=[max(agg_dyn_e2e.column1)], metrics=[]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_0.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_2.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_3.parquet]] [...]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_0.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_2.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn/file_3.parquet]] [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -323,7 +323,7 @@ Plan with Metrics
01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_single.a)], metrics=[]
02)--CoalescePartitionsExec, metrics=[]
03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_single.a)], metrics=[]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 ], pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1, required_guarantees=[], metrics=[]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1, required_guarantees=[], metrics=[]
# MAX(a) -> DynamicFilter [ a > 8 ]
query TT
@@ -333,7 +333,7 @@ Plan with Metrics
01)AggregateExec: mode=Final, gby=[], aggr=[max(agg_dyn_single.a)], metrics=[]
02)--CoalescePartitionsExec, metrics=[]
03)----AggregateExec: mode=Partial, gby=[], aggr=[max(agg_dyn_single.a)], metrics=[]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 > 8 ], pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 > 8, required_guarantees=[], metrics=[]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 > 8 ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 > 8, required_guarantees=[], metrics=[]
# MIN(a), MAX(a) -> DynamicFilter [ a < 1 OR a > 8 ]
query TT
@@ -343,7 +343,7 @@ Plan with Metrics
01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_single.a), max(agg_dyn_single.a)], metrics=[]
02)--CoalescePartitionsExec, metrics=[]
03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_single.a), max(agg_dyn_single.a)], metrics=[]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR a@0 > 8 ], pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR a_null_count@1 != row_count@2 AND a_max@3 > 8, requir [...]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_single/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR a@0 > 8 ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR a_null_count@1 != row_co [...]
# MIN(a+1) -> no dynamic filter (expression input is not a plain column)
query TT
@@ -387,7 +387,7 @@ Plan with Metrics
01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_two_col.a), max(agg_dyn_two_col.b)], metrics=[]
02)--CoalescePartitionsExec, metrics=[]
03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_two_col.a), max(agg_dyn_two_col.b)], metrics=[]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_two_col/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_two_col/file_1.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR b@1 > 9 ], pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR b_null_count@4 != row_count@2 AND b_max@3 > 9, r [...]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_two_col/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_two_col/file_1.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR b@1 > 9 ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR b_null_count@4 != r [...]
statement ok
drop table agg_dyn_two_col;
@@ -423,7 +423,7 @@ Plan with Metrics
01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_mixed.a), max(agg_dyn_mixed.a), max(agg_dyn_mixed.b), min(agg_dyn_mixed.c + Int64(1))], metrics=[]
02)--CoalescePartitionsExec, metrics=[]
03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_mixed.a), max(agg_dyn_mixed.a), max(agg_dyn_mixed.b), min(agg_dyn_mixed.c + Int64(1))], metrics=[]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_1.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR a@0 > 8 OR b@1 > 12 ], pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR a_null_count@1 != row_count@2 AND a_m [...]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_1.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 < 1 OR a@0 > 8 OR b@1 > 12 ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 1 OR a_null_c [...]
statement ok
drop table agg_dyn_mixed;
@@ -455,7 +455,7 @@ Plan with Metrics
01)AggregateExec: mode=Final, gby=[], aggr=[min(agg_dyn_nulls.a)], metrics=[]
02)--CoalescePartitionsExec, metrics=[]
03)----AggregateExec: mode=Partial, gby=[], aggr=[min(agg_dyn_nulls.a)], metrics=[]
-04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_nulls/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_nulls/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ true ], metrics=[]
+04)------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_nulls/file_0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_regression/agg_dyn_nulls/file_1.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ true ], dynamic_rg_pruning=eligible, metrics=[]
statement ok
reset datafusion.explain.analyze_categories;
diff --git a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt
index dbf31dec5e..043a62314c 100644
--- a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt
+++ b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt
@@ -380,7 +380,7 @@ physical_plan
12)----------------------CoalescePartitionsExec
13)------------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2]
14)--------------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files [...]
-15)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 [...]
+15)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 [...]
# Verify results without subset satisfaction
query TPR rowsort
@@ -475,7 +475,7 @@ physical_plan
10)------------------CoalescePartitionsExec
11)--------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2]
12)----------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=A/data.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=D/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/dimension/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scr [...]
-13)------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC [...]
+13)------------------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC [...]
# Verify results match with subset satisfaction
query TPR rowsort
@@ -517,7 +517,7 @@ prod 2023-01-01T09:12:30 197.7
# Config reset
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
# reset it explicitly.
statement ok
set datafusion.execution.target_partitions = 4;
diff --git a/datafusion/sqllogictest/test_files/sort_pushdown.slt b/datafusion/sqllogictest/test_files/sort_pushdown.slt
index 36fb38f5b4..f2442762f3 100644
--- a/datafusion/sqllogictest/test_files/sort_pushdown.slt
+++ b/datafusion/sqllogictest/test_files/sort_pushdown.slt
@@ -43,7 +43,7 @@ logical_plan
02)--TableScan: sorted_parquet projection=[id, value, name]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_data.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_data.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
# Test 1.2: Verify results are correct
query IIT
@@ -74,7 +74,7 @@ logical_plan
02)--TableScan: sorted_parquet projection=[id, value, name]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_data.parquet]]}, projection=[id, value, name], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_data.parquet]]}, projection=[id, value, name], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Re-enable
statement ok
@@ -91,7 +91,7 @@ logical_plan
physical_plan
01)GlobalLimitExec: skip=2, fetch=3
02)--SortExec: TopK(fetch=5), expr=[id@0 DESC], preserve_partitioning=[false]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_data.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_data.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
query IIT
SELECT * FROM sorted_parquet ORDER BY id DESC LIMIT 3 OFFSET 2;
@@ -155,7 +155,7 @@ logical_plan
03)----TableScan: multi_rg_sorted projection=[id, category, value], partial_filters=[multi_rg_sorted.category = Utf8View("alpha") OR multi_rg_sorted.category = Utf8View("gamma")]
physical_plan
01)SortExec: TopK(fetch=5), expr=[id@0 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/multi_rg_sorted.parquet]]}, projection=[id, category, value], file_type=parquet, predicate=(category@1 = alpha OR category@1 = gamma) AND DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, pruning_predicate=category_null_count@2 != row_count@3 AND category_min@0 <= alpha AND alpha <= category_max@1 OR category_null_count@2 != row_count@3 [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/multi_rg_sorted.parquet]]}, projection=[id, category, value], file_type=parquet, predicate=(category@1 = alpha OR category@1 = gamma) AND DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=category_null_count@2 != row_count@3 AND category_min@0 <= alpha AND alpha <= category_max@1 OR category [...]
# Verify the results are correct despite reverse scanning with row selection
# Expected: gamma values (6, 5) then alpha values (2, 1), in DESC order by id
@@ -272,7 +272,7 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [id@0 DESC], fetch=3
02)--SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[true]
-03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_multi/part1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_multi/part2.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_multi/part3.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true
+03)----DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_multi/part1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_multi/part2.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/sorted_multi/part3.parquet]]}, projection=[id, value, name], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_group [...]
# Verify correctness with repartitioning and multiple files
query IIT
@@ -381,7 +381,7 @@ logical_plan
03)----TableScan: timeseries_parquet projection=[timeframe, period_end, value], partial_filters=[timeseries_parquet.timeframe = Utf8View("quarterly")]
physical_plan
01)SortExec: TopK(fetch=2), expr=[period_end@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], file_type=parquet, predicate=timeframe@0 = quarterly AND DynamicFilter [ empty ], sort_order_for_reorder=[period_end@1 DESC], reverse_row_groups=true, pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= quarterly AND quarterly <= timeframe_max@1, required_guarantees=[timefra [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], file_type=parquet, predicate=timeframe@0 = quarterly AND DynamicFilter [ empty ], sort_order_for_reorder=[period_end@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= quarterly AND quarterly <= timeframe_max@1, [...]
# Test 2.2: Verify the results are correct
query TIR
@@ -440,7 +440,7 @@ logical_plan
02)--TableScan: timeseries_parquet projection=[timeframe, period_end, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[timeframe@0 ASC NULLS LAST, period_end@1 DESC], preserve_partitioning=[false], sort_prefix=[timeframe@0 ASC NULLS LAST]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], output_ordering=[timeframe@0 ASC NULLS LAST, period_end@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], output_ordering=[timeframe@0 ASC NULLS LAST, period_end@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Test 2.7: Disable sort pushdown and verify filter still works
statement ok
@@ -458,7 +458,7 @@ logical_plan
03)----TableScan: timeseries_parquet projection=[timeframe, period_end, value], partial_filters=[timeseries_parquet.timeframe = Utf8View("quarterly")]
physical_plan
01)SortExec: TopK(fetch=2), expr=[period_end@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], output_ordering=[timeframe@0 ASC NULLS LAST, period_end@1 ASC NULLS LAST], file_type=parquet, predicate=timeframe@0 = quarterly AND DynamicFilter [ empty ], pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= quarterly AND quarterly <= timeframe_max@1, required_guarantees=[t [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], output_ordering=[timeframe@0 ASC NULLS LAST, period_end@1 ASC NULLS LAST], file_type=parquet, predicate=timeframe@0 = quarterly AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= quarterly AND quarterly <= timeframe_ [...]
# Results should still be correct
query TIR
@@ -491,7 +491,7 @@ logical_plan
03)----TableScan: timeseries_parquet projection=[timeframe, period_end, value], partial_filters=[timeseries_parquet.timeframe = Utf8View("daily") OR timeseries_parquet.timeframe = Utf8View("weekly")]
physical_plan
01)SortExec: TopK(fetch=3), expr=[period_end@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], file_type=parquet, predicate=(timeframe@0 = daily OR timeframe@0 = weekly) AND DynamicFilter [ empty ], sort_order_for_reorder=[period_end@1 DESC], reverse_row_groups=true, pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= daily AND daily <= timeframe_max@1 OR timeframe_nu [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], file_type=parquet, predicate=(timeframe@0 = daily OR timeframe@0 = weekly) AND DynamicFilter [ empty ], sort_order_for_reorder=[period_end@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= daily AND daily <= ti [...]
# Test 2.9: Complex case - literal constant in sort expression itself
# The literal 'constant' is ignored in sort analysis
@@ -511,7 +511,7 @@ logical_plan
03)----TableScan: timeseries_parquet projection=[timeframe, period_end, value], partial_filters=[timeseries_parquet.timeframe = Utf8View("monthly")]
physical_plan
01)SortExec: TopK(fetch=2), expr=[period_end@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], file_type=parquet, predicate=timeframe@0 = monthly AND DynamicFilter [ empty ], sort_order_for_reorder=[period_end@1 DESC], reverse_row_groups=true, pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= monthly AND monthly <= timeframe_max@1, required_guarantees=[timeframe in [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timeseries_sorted.parquet]]}, projection=[timeframe, period_end, value], file_type=parquet, predicate=timeframe@0 = monthly AND DynamicFilter [ empty ], sort_order_for_reorder=[period_end@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=timeframe_null_count@2 != row_count@3 AND timeframe_min@0 <= monthly AND monthly <= timeframe_max@1, requi [...]
# Verify results
query TIR
@@ -600,7 +600,7 @@ logical_plan
02)--TableScan: timestamp_parquet projection=[id, ts, volume, price]
physical_plan
01)SortExec: TopK(fetch=3), expr=[ts@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timestamp_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[ts@1 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timestamp_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[ts@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
# Verify results
query IPIR
@@ -626,7 +626,7 @@ logical_plan
02)--TableScan: timestamp_parquet projection=[id, ts, volume, price]
physical_plan
01)SortExec: TopK(fetch=3), expr=[date_trunc(day, ts@1) DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timestamp_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[date_trunc(day, ts@1) DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/timestamp_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[date_trunc(day, ts@1) DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
# Verify results (descending day)
query IPIR
@@ -686,7 +686,7 @@ logical_plan
02)--TableScan: multi_month_parquet projection=[id, ts, volume, price]
physical_plan
01)SortExec: TopK(fetch=2), expr=[ts@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/multi_month_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[ts@1 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/multi_month_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[ts@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
query IPIR
SELECT * FROM multi_month_parquet
@@ -712,7 +712,7 @@ logical_plan
02)--TableScan: multi_month_parquet projection=[id, ts, volume, price]
physical_plan
01)SortExec: TopK(fetch=2), expr=[date_trunc(month, ts@1) DESC, ts@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/multi_month_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[date_trunc(month, ts@1) DESC, ts@1 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/multi_month_sorted.parquet]]}, projection=[id, ts, volume, price], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[date_trunc(month, ts@1) DESC, ts@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
query IPIR
SELECT * FROM multi_month_parquet
@@ -754,7 +754,7 @@ logical_plan
02)--TableScan: int_parquet projection=[id, small_val, big_val]
physical_plan
01)SortExec: TopK(fetch=2), expr=[CAST(small_val@1 AS Int64) DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/int_sorted.parquet]]}, projection=[id, small_val, big_val], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[CAST(small_val@1 AS Int64) DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/int_sorted.parquet]]}, projection=[id, small_val, big_val], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[CAST(small_val@1 AS Int64) DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
query III
SELECT * FROM int_parquet
@@ -796,7 +796,7 @@ logical_plan
02)--TableScan: float_parquet projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[ceil(value@1) DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/float_sorted.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[ceil(value@1) DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/float_sorted.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[ceil(value@1) DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
query IR
SELECT * FROM float_parquet
@@ -839,7 +839,7 @@ logical_plan
02)--TableScan: signed_parquet projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[abs(value@1) DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/signed_sorted.parquet]]}, projection=[id, value], output_ordering=[value@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/signed_sorted.parquet]]}, projection=[id, value], output_ordering=[value@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Results should still be correct (no optimization applied)
query IR
@@ -1988,7 +1988,7 @@ logical_plan
02)--TableScan: tb_overlap projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=5), expr=[id@0 DESC, value@1 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tb_overlap/file_z.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tb_overlap/file_y.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tb_overlap/file_x.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC, value@1 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tb_overlap/file_z.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tb_overlap/file_y.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tb_overlap/file_x.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC, value@1 DESC], reverse_row_groups= [...]
query II
SELECT * FROM tb_overlap ORDER BY id DESC, value DESC LIMIT 5;
@@ -2073,7 +2073,7 @@ logical_plan
02)--TableScan: tc_limit projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tc_limit/file_c.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tc_limit/file_b.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tc_limit/file_a.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tc_limit/file_c.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tc_limit/file_b.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tc_limit/file_a.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pru [...]
query II
SELECT * FROM tc_limit ORDER BY id DESC LIMIT 3;
@@ -2544,7 +2544,7 @@ logical_plan
02)--TableScan: th_reorder projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/th_reorder/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/th_reorder/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
# Results must be correct regardless of RG reorder.
query II
@@ -2563,7 +2563,7 @@ logical_plan
02)--TableScan: th_reorder projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/th_reorder/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/th_reorder/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
query II
SELECT * FROM th_reorder ORDER BY id DESC LIMIT 3;
@@ -2649,7 +2649,7 @@ logical_plan
02)--TableScan: tj_scrambled projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tj_scrambled/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tj_scrambled/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
# Test J.2: Results must be correct
query II
@@ -2839,7 +2839,7 @@ logical_plan
02)--TableScan: tl_sorted projection=[id, value]
physical_plan
01)SortExec: TopK(fetch=3), expr=[id@0 DESC, value@1 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tl_multikey/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC, value@1 ASC NULLS LAST], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/tl_multikey/data.parquet]]}, projection=[id, value], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[id@0 DESC, value@1 ASC NULLS LAST], reverse_row_groups=true, dynamic_rg_pruning=eligible
query II
SELECT id, value FROM tl_sorted ORDER BY id DESC, value ASC LIMIT 3;
diff --git a/datafusion/sqllogictest/test_files/statistics_registry.slt b/datafusion/sqllogictest/test_files/statistics_registry.slt
index c856e779a0..89258bec29 100644
--- a/datafusion/sqllogictest/test_files/statistics_registry.slt
+++ b/datafusion/sqllogictest/test_files/statistics_registry.slt
@@ -104,9 +104,9 @@ physical_plan
02)--RepartitionExec: partitioning=Hash([small_id@2], 4), input_partitions=1, maintains_sort_order=true
03)----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@0, customer_id@1)], projection=[region_id@1, order_id@2, small_id@4]
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/customers.parquet]]}, projection=[customer_id, region_id], output_ordering=[region_id@1 ASC NULLS LAST], file_type=parquet
-05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
06)--RepartitionExec: partitioning=Hash([small_id@0], 4), input_partitions=1, maintains_sort_order=true
-07)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/dim_small.parquet]]}, projection=[small_id, label], output_ordering=[small_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+07)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/dim_small.parquet]]}, projection=[small_id, label], output_ordering=[small_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# -- With registry -----------------------------------------------------------
# Conservative estimate 100 > 50: dim_small correctly swapped to build side
@@ -127,7 +127,7 @@ physical_plan
04)--RepartitionExec: partitioning=Hash([small_id@2], 4), input_partitions=1, maintains_sort_order=true
05)----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@0, customer_id@1)], projection=[region_id@1, order_id@2, small_id@4]
06)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/customers.parquet]]}, projection=[customer_id, region_id], output_ordering=[region_id@1 ASC NULLS LAST], file_type=parquet
-07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ]
+07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# -- Verify results are identical regardless of join order --------------------
diff --git a/datafusion/sqllogictest/test_files/topk.slt b/datafusion/sqllogictest/test_files/topk.slt
index 8cab67dac0..d669e845ac 100644
--- a/datafusion/sqllogictest/test_files/topk.slt
+++ b/datafusion/sqllogictest/test_files/topk.slt
@@ -316,7 +316,7 @@ explain select number, letter, age from partial_sorted order by number desc, let
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[number@0 DESC, letter@1 ASC NULLS LAST, age@2 DESC], preserve_partitioning=[false], sort_prefix=[number@0 DESC, letter@1 ASC NULLS LAST]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Explain variations of the above query with different orderings, and different sort prefixes.
@@ -326,28 +326,28 @@ explain select number, letter, age from partial_sorted order by age desc limit 3
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[age@2 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[age@2 DESC], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[age@2 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible
query TT
explain select number, letter, age from partial_sorted order by number desc, letter desc limit 3;
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[number@0 DESC, letter@1 DESC], preserve_partitioning=[false], sort_prefix=[number@0 DESC]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
query TT
explain select number, letter, age from partial_sorted order by number asc limit 3;
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[number@0 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[number@0 ASC NULLS LAST]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[number@0 ASC NULLS LAST], dynamic_rg_pruning=eligible
query TT
explain select number, letter, age from partial_sorted order by letter asc, number desc limit 3;
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[letter@1 ASC NULLS LAST, number@0 DESC], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[letter@1 ASC NULLS LAST, number@0 DESC]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[letter@1 ASC NULLS LAST, number@0 DESC], dynamic_rg_pruning=eligible
# Explicit NULLS ordering cases (reversing the order of the NULLS on the number and letter orderings)
query TT
@@ -355,14 +355,14 @@ explain select number, letter, age from partial_sorted order by number desc, let
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[number@0 DESC, letter@1 ASC], preserve_partitioning=[false], sort_prefix=[number@0 DESC]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
query TT
explain select number, letter, age from partial_sorted order by number desc NULLS LAST, letter asc limit 3;
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[number@0 DESC NULLS LAST, letter@1 ASC NULLS LAST], preserve_partitioning=[false]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[number@0 DESC NULLS LAST, letter@1 ASC NULLS LAST], reverse_row_groups=true
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], file_type=parquet, predicate=DynamicFilter [ empty ], sort_order_for_reorder=[number@0 DESC NULLS LAST, letter@1 ASC NULLS LAST], reverse_row_groups=true, dynamic_rg_pruning=eligible
# Verify that the sort prefix is correctly computed on the normalized ordering (removing redundant aliased columns)
@@ -371,7 +371,7 @@ explain select number, letter, age, number as column4, letter as column5 from pa
----
physical_plan
01)SortExec: TopK(fetch=3), expr=[number@0 DESC, letter@1 ASC NULLS LAST, age@2 DESC], preserve_partitioning=[false], sort_prefix=[number@0 DESC, letter@1 ASC NULLS LAST]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age, number@0 as column4, letter@1 as column5], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age, number@0 as column4, letter@1 as column5], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Verify that the sort prefix is correctly computed over normalized, order-maintaining projections (number + 1, number, number + 1, age)
query TT
@@ -383,7 +383,7 @@ physical_plan
03)----ProjectionExec: expr=[__common_expr_1@0 as number_plus, number@1 as number, __common_expr_1@0 as other_number_plus, age@2 as age]
04)------ProjectionExec: expr=[CAST(number@0 AS Int64) + 1 as __common_expr_1, number@0 as number, age@1 as age]
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
-06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, age], output_ordering=[number@0 DESC], file_type=parquet, predicate=DynamicFilter [ empty ]
+06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, age], output_ordering=[number@0 DESC], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
# Cleanup
statement ok
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org