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/07/10 08:08:36 UTC
(datafusion) branch gh-readonly-queue/main/pr-23395-f4a700d79de4deb36ff21700c2b85f91d5507c07 created (now 1e77af86a3)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23395-f4a700d79de4deb36ff21700c2b85f91d5507c07
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 1e77af86a3 fix: don't duplicate volatile expressions when pushing projection into file scan (#23395)
This branch includes the following new commits:
new 1e77af86a3 fix: don't duplicate volatile expressions when pushing projection into file scan (#23395)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: don't duplicate volatile expressions when pushing projection into file scan (#23395)
Posted by gi...@apache.org.
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-23395-f4a700d79de4deb36ff21700c2b85f91d5507c07
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 1e77af86a3208f21c92078b9b0b2f6711c0ceeca
Author: Ford <fo...@edgeandnode.com>
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org