This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 1e77af86a3 fix: don't duplicate volatile expressions when pushing
projection into file scan (#23395)
1e77af86a3 is described below
commit 1e77af86a3208f21c92078b9b0b2f6711c0ceeca
Author: Ford <[email protected]>
AuthorDate: Fri Jul 10 01:08:13 2026 -0700
fix: don't duplicate volatile expressions when pushing projection into file
scan (#23395)
## 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.
---
datafusion/datasource/src/file_scan_config/mod.rs | 231 ++++++++++++++++++++-
.../test_files/projection_pushdown.slt | 30 +++
2 files changed, 260 insertions(+), 1 deletion(-)
diff --git a/datafusion/datasource/src/file_scan_config/mod.rs
b/datafusion/datasource/src/file_scan_config/mod.rs
index 660d0cd7a5..1336ee69cd 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,
ProjectionMapping};
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;
@@ -625,6 +625,51 @@ fn project_output_partitioning(
}
}
+/// 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::<Column>()
+ && 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,
@@ -911,6 +956,15 @@ impl DataSource for FileScanConfig {
&self,
projection: &ProjectionExprs,
) -> Result<Option<Arc<dyn DataSource>>> {
+ // 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();
@@ -3314,4 +3368,179 @@ mod tests {
);
Ok(())
}
+
+ /// Helper: build a `ProjectionExprs` from `(expr, alias)` pairs.
+ fn make_projection(pairs: Vec<(Arc<dyn PhysicalExpr>, &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<dyn PhysicalExpr> {
+ 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<dyn PhysicalExpr> = Arc::new(Column::new("a", 0));
+ let col_b: Arc<dyn PhysicalExpr> = 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<dyn PhysicalExpr> = Arc::new(Column::new("a", 0));
+ let col_b: Arc<dyn PhysicalExpr> = 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<dyn PhysicalExpr> = 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<dyn PhysicalExpr> = 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 fcadd3be1f..4fedf297cb 100644
--- a/datafusion/sqllogictest/test_files/projection_pushdown.slt
+++ b/datafusion/sqllogictest/test_files/projection_pushdown.slt
@@ -2077,3 +2077,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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
