From 4a902dbd7e2b66a7667fa0319eca2bb35baa3844 Mon Sep 17 00:00:00 2001 From: Ford Date: Fri, 10 Jul 2026 01:08:13 -0700 Subject: [PATCH] fix: don't duplicate volatile expressions when pushing projection into file scan (#23395) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Which issue does this PR close? - Closes #23220. ## Rationale for this change The reported bug showed that when volatile (non-deterministic) expressions are referenced multiple times in an outer function the query results can be incorrect. To produce correct results, a volatile expression (e.g. `random()`, `uuid()`) aliased once in a subquery and referenced multiple times must be evaluated once and reused. Since 52.0.0 this evaluation pattern for volatile functions has been broken; the physical projection-pushdown rule merges the outer projection into the file `DataSourceExec`, inlining the aliased volatile expression at each reference site. Instead of being evaluated once, the volatile expression is evaluated N times, so the references diverge: ```sql SELECT s.r AS x, s.r AS y FROM (SELECT random() AS r FROM t) AS s; -- x != y on >= 52.0.0 ``` This was correct in 51.0.0 and regressed in 52.0.0/53.0.0. It reproduces on file scans (Parquet/CSV) but not in-memory tables, and was surfaced downstream in Ibis. Worth noting, #10337 appears to report the same class of bug but a different cause (I'll try to look into that issue soon as well). ## What changes are included in this PR? `FileScanConfig::try_swapping_with_projection` now declines to merge a projection into the file source when the merge would inline a volatile expression that the incoming projection references. The check reuses the existing `is_volatile()` utility from `datafusion_physical_expr_common` — the same volatility gate the physical `ProjectionPushdown` and `FilterPushdown` rules already apply — so file-source projection merging is now consistent with them. Deterministic expressions still merge freely. The guard blocks when a volatile inner expression is referenced at least once (not only more than once), because a single outer expression can itself duplicate the value (e.g.`r + r`). ## Are these changes tested? Yes: - Unit tests in `file_scan_config.rs`: - volatile referenced ≥1× → blocked; - deterministic computed and column-only → allowed; - unreferenced volatile → allowed; - single-expression self-reference (`r + r`) → blocked; - volatile nested in arithmetic → blocked; - empty → allowed. - A regression test in `projection_pushdown.slt` asserting `x = y` for the aliased-`random()` pattern over a Parquet scan. - Full `datafusion-sqllogictest` suite passes. ## Are there any user-facing changes? No API changes! Query results for the affected pattern are corrected (a volatile expression aliased in a subquery is no longer duplicated by projection pushdown into a file scan). Physical plans for such queries now retain a `ProjectionExec` above the scan rather than inlining the volatile expression into the `DataSourceExec` projection. --- Cargo.lock | 1 + datafusion/datasource/Cargo.toml | 1 + .../datasource/src/file_scan_config/mod.rs | 231 +++++++++++++++++- .../test_files/projection_pushdown.slt | 30 +++ 4 files changed, 262 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index bdb5a98a6b713..19ba2afcfd7a8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1887,6 +1887,7 @@ dependencies = [ "datafusion-common-runtime", "datafusion-execution", "datafusion-expr", + "datafusion-functions", "datafusion-physical-expr", "datafusion-physical-expr-adapter", "datafusion-physical-expr-common", diff --git a/datafusion/datasource/Cargo.toml b/datafusion/datasource/Cargo.toml index 40e2271f45205..2ac42ed900095 100644 --- a/datafusion/datasource/Cargo.toml +++ b/datafusion/datasource/Cargo.toml @@ -74,6 +74,7 @@ zstd = { workspace = true, optional = true } [dev-dependencies] criterion = { workspace = true } +datafusion-functions = { workspace = true } insta = { workspace = true } tempfile = { workspace = true } diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 050cd54bb0adf..492471316d6bb 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -44,7 +44,7 @@ use datafusion_physical_expr::projection::ProjectionExprs; use datafusion_physical_expr::utils::reassign_expr_columns; use datafusion_physical_expr::{EquivalenceProperties, Partitioning, split_conjunction}; use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; -use datafusion_physical_expr_common::physical_expr::PhysicalExpr; +use datafusion_physical_expr_common::physical_expr::{PhysicalExpr, is_volatile}; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; use datafusion_physical_plan::SortOrderPushdownResult; use datafusion_physical_plan::coop::cooperative; @@ -577,6 +577,51 @@ impl From for FileScanConfigBuilder { } } +/// Returns `true` if merging `outer` into `inner` would duplicate a volatile +/// expression; the caller should then decline the merge. +/// +/// `inner` is the scan's current projection and `outer` the projection being +/// pushed into it; merging substitutes each `inner` expression into every +/// `outer` reference to it. If a volatile `inner` expression (e.g. `random()`, +/// `uuid()`) is referenced more than once, that single value gets inlined at +/// each site and re-evaluated independently, so references meant to share a +/// "locked-in" value diverge. This is the volatility guard the physical +/// `ProjectionPushdown` and `FilterPushdown` rules already apply (see +/// `datafusion_physical_expr_common::physical_expr::is_volatile`). +/// +/// References are counted with multiplicity by walking each `outer` expression +/// (as `try_collapse_projection_chain` does), so a self-duplicating expression +/// such as `r + r` counts as two references. A volatile expression referenced +/// exactly once has nothing to duplicate and is left to merge. +fn would_duplicate_volatile_exprs( + inner: &ProjectionExprs, + outer: &ProjectionExprs, +) -> bool { + use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; + + let inner_exprs = inner.as_ref(); + + let mut ref_counts = vec![0usize; inner_exprs.len()]; + for proj_expr in outer.as_ref() { + proj_expr + .expr + .apply(|e| { + if let Some(col) = e.as_ref().downcast_ref::() + && let Some(count) = ref_counts.get_mut(col.index()) + { + *count += 1; + } + Ok(TreeNodeRecursion::Continue) + }) + .expect("infallible closure should not fail"); + } + + ref_counts + .iter() + .enumerate() + .any(|(idx, &count)| count > 1 && is_volatile(&inner_exprs[idx].expr)) +} + impl DataSource for FileScanConfig { fn open( &self, @@ -859,6 +904,15 @@ impl DataSource for FileScanConfig { &self, projection: &ProjectionExprs, ) -> Result>> { + // Don't merge a projection into the scan if it would inline a volatile + // expression that the outer projection references, which would turn a + // single "locked-in" value (e.g. `random()` aliased in a subquery) into + // multiple independent evaluations. See #23220. + if let Some(inner) = self.file_source.projection() + && would_duplicate_volatile_exprs(inner, projection) + { + return Ok(None); + } match self.file_source.try_pushdown_projection(projection)? { Some(new_source) => { let mut new_file_scan_config = self.clone(); @@ -3191,4 +3245,179 @@ mod tests { ); Ok(()) } + + /// Helper: build a `ProjectionExprs` from `(expr, alias)` pairs. + fn make_projection(pairs: Vec<(Arc, &str)>) -> ProjectionExprs { + ProjectionExprs::new( + pairs + .into_iter() + .map(|(expr, alias)| ProjectionExpr::new(expr, alias)), + ) + } + + /// Helper: create a volatile (non-deterministic) function expression, + /// e.g. `random()`. + fn make_volatile_expr() -> Arc { + use datafusion_common::config::ConfigOptions; + use datafusion_expr::ScalarUDF; + use datafusion_functions::math::random::RandomFunc; + use datafusion_physical_expr::ScalarFunctionExpr; + + Arc::new(ScalarFunctionExpr::new( + "random", + Arc::new(ScalarUDF::from(RandomFunc::new())), + vec![], + Arc::new(Field::new("random", DataType::Float64, false)), + Arc::new(ConfigOptions::default()), + )) + } + + /// Column-only inner projections always merge safely, even when + /// the outer projection references them multiple times. + #[test] + fn test_would_duplicate_allows_column_only_inner() { + let col_a: Arc = Arc::new(Column::new("a", 0)); + let col_b: Arc = Arc::new(Column::new("b", 1)); + + let inner = + make_projection(vec![(Arc::clone(&col_a), "a"), (Arc::clone(&col_b), "b")]); + + // Outer references col 0 twice + let outer = make_projection(vec![ + (Arc::new(Column::new("a", 0)), "x"), + (Arc::new(Column::new("a", 0)), "y"), + ]); + + assert!(!would_duplicate_volatile_exprs(&inner, &outer)); + } + + /// Deterministic computed expressions (arithmetic) referenced multiple + /// times are allowed to merge — only volatile expressions are protected. + #[test] + fn test_would_duplicate_allows_deterministic_computed_multi_ref() { + let col_a: Arc = Arc::new(Column::new("a", 0)); + let col_b: Arc = Arc::new(Column::new("b", 1)); + // Inner: [a + b, b] (index 0 is deterministic computed) + let inner = make_projection(vec![ + ( + Arc::new(BinaryExpr::new( + Arc::clone(&col_a), + Operator::Plus, + Arc::clone(&col_b), + )), + "sum", + ), + (Arc::clone(&col_b), "b"), + ]); + + // Outer references index 0 twice + let outer = make_projection(vec![ + (Arc::new(Column::new("sum", 0)), "x"), + (Arc::new(Column::new("sum", 0)), "y"), + ]); + + // Deterministic arithmetic → allow merge even though duplicated + assert!(!would_duplicate_volatile_exprs(&inner, &outer)); + } + + /// A volatile expression the outer projection does not reference is + /// safe to merge (it is projected away, not duplicated). + #[test] + fn test_would_duplicate_allows_unreferenced_volatile() { + let col_a: Arc = Arc::new(Column::new("a", 0)); + // Inner: [random(), a] + let inner = + make_projection(vec![(make_volatile_expr(), "r"), (Arc::clone(&col_a), "a")]); + + // Outer references only index 1 (the column), not the volatile expr + let outer = make_projection(vec![(Arc::new(Column::new("a", 1)), "a")]); + + assert!(!would_duplicate_volatile_exprs(&inner, &outer)); + } + + /// A volatile expression referenced multiple times must block merge: + /// this is the #23220 regression (`random()` aliased then referenced as + /// `x` and `y`). + #[test] + fn test_would_duplicate_blocks_multi_ref_volatile() { + // Inner: [random()] + let inner = make_projection(vec![(make_volatile_expr(), "r")]); + + // Outer references index 0 twice + let outer = make_projection(vec![ + (Arc::new(Column::new("r", 0)), "x"), + (Arc::new(Column::new("r", 0)), "y"), + ]); + + assert!(would_duplicate_volatile_exprs(&inner, &outer)); + } + + /// A volatile expression referenced exactly once has nothing to duplicate, + /// so the merge is allowed. + #[test] + fn test_would_duplicate_allows_single_ref_volatile() { + let col_a: Arc = Arc::new(Column::new("a", 0)); + // Inner: [random(), a] + let inner = + make_projection(vec![(make_volatile_expr(), "r"), (Arc::clone(&col_a), "a")]); + + // Outer references the volatile expression exactly once + let outer = make_projection(vec![ + (Arc::new(Column::new("r", 0)), "x"), + (Arc::new(Column::new("a", 1)), "a"), + ]); + + assert!(!would_duplicate_volatile_exprs(&inner, &outer)); + } + + /// References are counted with multiplicity, so a single outer expression + /// that duplicates the value (e.g. `r + r`) still blocks the merge. + #[test] + fn test_would_duplicate_blocks_single_expr_self_ref_volatile() { + // Inner: [random()] + let inner = make_projection(vec![(make_volatile_expr(), "r")]); + + // Outer: [r + r] — one expression referencing `random()` twice + let outer = make_projection(vec![( + Arc::new(BinaryExpr::new( + Arc::new(Column::new("r", 0)), + Operator::Plus, + Arc::new(Column::new("r", 0)), + )), + "x", + )]); + + assert!(would_duplicate_volatile_exprs(&inner, &outer)); + } + + /// A volatile expression buried inside a larger expression (e.g. + /// `random() + 1`) is still detected and blocks merge. + #[test] + fn test_would_duplicate_blocks_volatile_nested_in_arithmetic() { + // Inner: [random() + 1] + let inner = make_projection(vec![( + Arc::new(BinaryExpr::new( + make_volatile_expr(), + Operator::Plus, + Arc::new(Literal::new(ScalarValue::Float64(Some(1.0)))), + )), + "expr", + )]); + + // Outer references index 0 twice + let outer = make_projection(vec![ + (Arc::new(Column::new("expr", 0)), "x"), + (Arc::new(Column::new("expr", 0)), "y"), + ]); + + assert!(would_duplicate_volatile_exprs(&inner, &outer)); + } + + /// Empty projections should not block merging. + #[test] + fn test_would_duplicate_empty_projections() { + let inner = make_projection(vec![]); + let outer = make_projection(vec![]); + assert!(!would_duplicate_volatile_exprs(&inner, &outer)); + } } diff --git a/datafusion/sqllogictest/test_files/projection_pushdown.slt b/datafusion/sqllogictest/test_files/projection_pushdown.slt index 344aef1f92cf9..6bc07ee9f1f9f 100644 --- a/datafusion/sqllogictest/test_files/projection_pushdown.slt +++ b/datafusion/sqllogictest/test_files/projection_pushdown.slt @@ -2074,3 +2074,33 @@ SELECT s, id FROM simple_struct WHERE s['value'] > 100 AND id < 4; # reset it explicitly. statement ok SET datafusion.execution.target_partitions = 4; + +##################### +# Section: volatile expressions are not duplicated by projection pushdown +# +# Regression test for #23220: a volatile expression (e.g. `random()`) aliased +# once in a subquery and referenced multiple times must be evaluated once and +# reused. Projection pushdown must not merge the outer projection into the file +# scan when doing so would inline and duplicate the volatile expression. +# Reproduces only against a file scan (not an in-memory table); if the volatile +# expression is duplicated, the two references diverge and `x = y` is false. +##################### + +statement ok +COPY (SELECT 1 AS id UNION ALL SELECT 2 UNION ALL SELECT 3) +TO 'test_files/scratch/projection_pushdown/volatile.parquet' +STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE volatile_scan STORED AS PARQUET +LOCATION 'test_files/scratch/projection_pushdown/volatile.parquet'; + +# The two references to the aliased `random()` value must be equal on every +# row: the expression is evaluated once and reused, not inlined twice. +query B rowsort +SELECT s.x = s.y +FROM (SELECT r AS x, r AS y FROM (SELECT random() AS r FROM volatile_scan) AS t) AS s; +---- +true +true +true