fix: don't duplicate volatile expressions when pushing projection into file scan

A volatile expression (e.g. `random()`, `uuid()`) aliased once in a subquery
and referenced multiple times must be evaluated once and reused. The physical
projection-pushdown rule merged the outer projection into the file
`DataSourceExec`, inlining the aliased volatile expression at every reference
site and turning one call into N independent calls, so the references diverged.
This regressed in 52.0.0 and reproduces on file scans (parquet/csv) but not
in-memory tables.

Guard `FileScanConfig::try_swapping_with_projection`: decline the merge when a
volatile expression in the scan's existing projection is referenced by the
incoming projection, reusing the existing `is_volatile()` utility -- the same
gate the physical `ProjectionPushdown` and `FilterPushdown` rules already
apply. Deterministic expressions still merge freely.

Closes #23220.

Author: fordN

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;
@@ -649,6 +649,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,
@@ -945,6 +990,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();
@@ -3342,4 +3396,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));
+    }
 }

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