@@ -44,7 +44,7 @@ use datafusion_physical_expr::projection::{ProjectionExprs, ProjectionMapping};
4444use datafusion_physical_expr:: utils:: reassign_expr_columns;
4545use datafusion_physical_expr:: { EquivalenceProperties , Partitioning , split_conjunction} ;
4646use datafusion_physical_expr_adapter:: PhysicalExprAdapterFactory ;
47- use datafusion_physical_expr_common:: physical_expr:: PhysicalExpr ;
47+ use datafusion_physical_expr_common:: physical_expr:: { PhysicalExpr , is_volatile } ;
4848use datafusion_physical_expr_common:: sort_expr:: { LexOrdering , PhysicalSortExpr } ;
4949use datafusion_physical_plan:: SortOrderPushdownResult ;
5050use datafusion_physical_plan:: coop:: cooperative;
@@ -649,6 +649,51 @@ fn project_output_partitioning(
649649 }
650650}
651651
652+ /// Returns `true` if merging `outer` into `inner` would duplicate a volatile
653+ /// expression; the caller should then decline the merge.
654+ ///
655+ /// `inner` is the scan's current projection and `outer` the projection being
656+ /// pushed into it; merging substitutes each `inner` expression into every
657+ /// `outer` reference to it. If a volatile `inner` expression (e.g. `random()`,
658+ /// `uuid()`) is referenced more than once, that single value gets inlined at
659+ /// each site and re-evaluated independently, so references meant to share a
660+ /// "locked-in" value diverge. This is the volatility guard the physical
661+ /// `ProjectionPushdown` and `FilterPushdown` rules already apply (see
662+ /// `datafusion_physical_expr_common::physical_expr::is_volatile`).
663+ ///
664+ /// References are counted with multiplicity by walking each `outer` expression
665+ /// (as `try_collapse_projection_chain` does), so a self-duplicating expression
666+ /// such as `r + r` counts as two references. A volatile expression referenced
667+ /// exactly once has nothing to duplicate and is left to merge.
668+ fn would_duplicate_volatile_exprs (
669+ inner : & ProjectionExprs ,
670+ outer : & ProjectionExprs ,
671+ ) -> bool {
672+ use datafusion_common:: tree_node:: { TreeNode , TreeNodeRecursion } ;
673+
674+ let inner_exprs = inner. as_ref ( ) ;
675+
676+ let mut ref_counts = vec ! [ 0usize ; inner_exprs. len( ) ] ;
677+ for proj_expr in outer. as_ref ( ) {
678+ proj_expr
679+ . expr
680+ . apply ( |e| {
681+ if let Some ( col) = e. as_ref ( ) . downcast_ref :: < Column > ( )
682+ && let Some ( count) = ref_counts. get_mut ( col. index ( ) )
683+ {
684+ * count += 1 ;
685+ }
686+ Ok ( TreeNodeRecursion :: Continue )
687+ } )
688+ . expect ( "infallible closure should not fail" ) ;
689+ }
690+
691+ ref_counts
692+ . iter ( )
693+ . enumerate ( )
694+ . any ( |( idx, & count) | count > 1 && is_volatile ( & inner_exprs[ idx] . expr ) )
695+ }
696+
652697impl DataSource for FileScanConfig {
653698 fn open (
654699 & self ,
@@ -945,6 +990,15 @@ impl DataSource for FileScanConfig {
945990 & self ,
946991 projection : & ProjectionExprs ,
947992 ) -> Result < Option < Arc < dyn DataSource > > > {
993+ // Don't merge a projection into the scan if it would inline a volatile
994+ // expression that the outer projection references, which would turn a
995+ // single "locked-in" value (e.g. `random()` aliased in a subquery) into
996+ // multiple independent evaluations. See #23220.
997+ if let Some ( inner) = self . file_source . projection ( )
998+ && would_duplicate_volatile_exprs ( inner, projection)
999+ {
1000+ return Ok ( None ) ;
1001+ }
9481002 match self . file_source . try_pushdown_projection ( projection) ? {
9491003 Some ( new_source) => {
9501004 let mut new_file_scan_config = self . clone ( ) ;
@@ -3342,4 +3396,179 @@ mod tests {
33423396 ) ;
33433397 Ok ( ( ) )
33443398 }
3399+
3400+ /// Helper: build a `ProjectionExprs` from `(expr, alias)` pairs.
3401+ fn make_projection ( pairs : Vec < ( Arc < dyn PhysicalExpr > , & str ) > ) -> ProjectionExprs {
3402+ ProjectionExprs :: new (
3403+ pairs
3404+ . into_iter ( )
3405+ . map ( |( expr, alias) | ProjectionExpr :: new ( expr, alias) ) ,
3406+ )
3407+ }
3408+
3409+ /// Helper: create a volatile (non-deterministic) function expression,
3410+ /// e.g. `random()`.
3411+ fn make_volatile_expr ( ) -> Arc < dyn PhysicalExpr > {
3412+ use datafusion_common:: config:: ConfigOptions ;
3413+ use datafusion_expr:: ScalarUDF ;
3414+ use datafusion_functions:: math:: random:: RandomFunc ;
3415+ use datafusion_physical_expr:: ScalarFunctionExpr ;
3416+
3417+ Arc :: new ( ScalarFunctionExpr :: new (
3418+ "random" ,
3419+ Arc :: new ( ScalarUDF :: from ( RandomFunc :: new ( ) ) ) ,
3420+ vec ! [ ] ,
3421+ Arc :: new ( Field :: new ( "random" , DataType :: Float64 , false ) ) ,
3422+ Arc :: new ( ConfigOptions :: default ( ) ) ,
3423+ ) )
3424+ }
3425+
3426+ /// Column-only inner projections always merge safely, even when
3427+ /// the outer projection references them multiple times.
3428+ #[ test]
3429+ fn test_would_duplicate_allows_column_only_inner ( ) {
3430+ let col_a: Arc < dyn PhysicalExpr > = Arc :: new ( Column :: new ( "a" , 0 ) ) ;
3431+ let col_b: Arc < dyn PhysicalExpr > = Arc :: new ( Column :: new ( "b" , 1 ) ) ;
3432+
3433+ let inner =
3434+ make_projection ( vec ! [ ( Arc :: clone( & col_a) , "a" ) , ( Arc :: clone( & col_b) , "b" ) ] ) ;
3435+
3436+ // Outer references col 0 twice
3437+ let outer = make_projection ( vec ! [
3438+ ( Arc :: new( Column :: new( "a" , 0 ) ) , "x" ) ,
3439+ ( Arc :: new( Column :: new( "a" , 0 ) ) , "y" ) ,
3440+ ] ) ;
3441+
3442+ assert ! ( !would_duplicate_volatile_exprs( & inner, & outer) ) ;
3443+ }
3444+
3445+ /// Deterministic computed expressions (arithmetic) referenced multiple
3446+ /// times are allowed to merge — only volatile expressions are protected.
3447+ #[ test]
3448+ fn test_would_duplicate_allows_deterministic_computed_multi_ref ( ) {
3449+ let col_a: Arc < dyn PhysicalExpr > = Arc :: new ( Column :: new ( "a" , 0 ) ) ;
3450+ let col_b: Arc < dyn PhysicalExpr > = Arc :: new ( Column :: new ( "b" , 1 ) ) ;
3451+ // Inner: [a + b, b] (index 0 is deterministic computed)
3452+ let inner = make_projection ( vec ! [
3453+ (
3454+ Arc :: new( BinaryExpr :: new(
3455+ Arc :: clone( & col_a) ,
3456+ Operator :: Plus ,
3457+ Arc :: clone( & col_b) ,
3458+ ) ) ,
3459+ "sum" ,
3460+ ) ,
3461+ ( Arc :: clone( & col_b) , "b" ) ,
3462+ ] ) ;
3463+
3464+ // Outer references index 0 twice
3465+ let outer = make_projection ( vec ! [
3466+ ( Arc :: new( Column :: new( "sum" , 0 ) ) , "x" ) ,
3467+ ( Arc :: new( Column :: new( "sum" , 0 ) ) , "y" ) ,
3468+ ] ) ;
3469+
3470+ // Deterministic arithmetic → allow merge even though duplicated
3471+ assert ! ( !would_duplicate_volatile_exprs( & inner, & outer) ) ;
3472+ }
3473+
3474+ /// A volatile expression the outer projection does not reference is
3475+ /// safe to merge (it is projected away, not duplicated).
3476+ #[ test]
3477+ fn test_would_duplicate_allows_unreferenced_volatile ( ) {
3478+ let col_a: Arc < dyn PhysicalExpr > = Arc :: new ( Column :: new ( "a" , 0 ) ) ;
3479+ // Inner: [random(), a]
3480+ let inner =
3481+ make_projection ( vec ! [ ( make_volatile_expr( ) , "r" ) , ( Arc :: clone( & col_a) , "a" ) ] ) ;
3482+
3483+ // Outer references only index 1 (the column), not the volatile expr
3484+ let outer = make_projection ( vec ! [ ( Arc :: new( Column :: new( "a" , 1 ) ) , "a" ) ] ) ;
3485+
3486+ assert ! ( !would_duplicate_volatile_exprs( & inner, & outer) ) ;
3487+ }
3488+
3489+ /// A volatile expression referenced multiple times must block merge:
3490+ /// this is the #23220 regression (`random()` aliased then referenced as
3491+ /// `x` and `y`).
3492+ #[ test]
3493+ fn test_would_duplicate_blocks_multi_ref_volatile ( ) {
3494+ // Inner: [random()]
3495+ let inner = make_projection ( vec ! [ ( make_volatile_expr( ) , "r" ) ] ) ;
3496+
3497+ // Outer references index 0 twice
3498+ let outer = make_projection ( vec ! [
3499+ ( Arc :: new( Column :: new( "r" , 0 ) ) , "x" ) ,
3500+ ( Arc :: new( Column :: new( "r" , 0 ) ) , "y" ) ,
3501+ ] ) ;
3502+
3503+ assert ! ( would_duplicate_volatile_exprs( & inner, & outer) ) ;
3504+ }
3505+
3506+ /// A volatile expression referenced exactly once has nothing to duplicate,
3507+ /// so the merge is allowed.
3508+ #[ test]
3509+ fn test_would_duplicate_allows_single_ref_volatile ( ) {
3510+ let col_a: Arc < dyn PhysicalExpr > = Arc :: new ( Column :: new ( "a" , 0 ) ) ;
3511+ // Inner: [random(), a]
3512+ let inner =
3513+ make_projection ( vec ! [ ( make_volatile_expr( ) , "r" ) , ( Arc :: clone( & col_a) , "a" ) ] ) ;
3514+
3515+ // Outer references the volatile expression exactly once
3516+ let outer = make_projection ( vec ! [
3517+ ( Arc :: new( Column :: new( "r" , 0 ) ) , "x" ) ,
3518+ ( Arc :: new( Column :: new( "a" , 1 ) ) , "a" ) ,
3519+ ] ) ;
3520+
3521+ assert ! ( !would_duplicate_volatile_exprs( & inner, & outer) ) ;
3522+ }
3523+
3524+ /// References are counted with multiplicity, so a single outer expression
3525+ /// that duplicates the value (e.g. `r + r`) still blocks the merge.
3526+ #[ test]
3527+ fn test_would_duplicate_blocks_single_expr_self_ref_volatile ( ) {
3528+ // Inner: [random()]
3529+ let inner = make_projection ( vec ! [ ( make_volatile_expr( ) , "r" ) ] ) ;
3530+
3531+ // Outer: [r + r] — one expression referencing `random()` twice
3532+ let outer = make_projection ( vec ! [ (
3533+ Arc :: new( BinaryExpr :: new(
3534+ Arc :: new( Column :: new( "r" , 0 ) ) ,
3535+ Operator :: Plus ,
3536+ Arc :: new( Column :: new( "r" , 0 ) ) ,
3537+ ) ) ,
3538+ "x" ,
3539+ ) ] ) ;
3540+
3541+ assert ! ( would_duplicate_volatile_exprs( & inner, & outer) ) ;
3542+ }
3543+
3544+ /// A volatile expression buried inside a larger expression (e.g.
3545+ /// `random() + 1`) is still detected and blocks merge.
3546+ #[ test]
3547+ fn test_would_duplicate_blocks_volatile_nested_in_arithmetic ( ) {
3548+ // Inner: [random() + 1]
3549+ let inner = make_projection ( vec ! [ (
3550+ Arc :: new( BinaryExpr :: new(
3551+ make_volatile_expr( ) ,
3552+ Operator :: Plus ,
3553+ Arc :: new( Literal :: new( ScalarValue :: Float64 ( Some ( 1.0 ) ) ) ) ,
3554+ ) ) ,
3555+ "expr" ,
3556+ ) ] ) ;
3557+
3558+ // Outer references index 0 twice
3559+ let outer = make_projection ( vec ! [
3560+ ( Arc :: new( Column :: new( "expr" , 0 ) ) , "x" ) ,
3561+ ( Arc :: new( Column :: new( "expr" , 0 ) ) , "y" ) ,
3562+ ] ) ;
3563+
3564+ assert ! ( would_duplicate_volatile_exprs( & inner, & outer) ) ;
3565+ }
3566+
3567+ /// Empty projections should not block merging.
3568+ #[ test]
3569+ fn test_would_duplicate_empty_projections ( ) {
3570+ let inner = make_projection ( vec ! [ ] ) ;
3571+ let outer = make_projection ( vec ! [ ] ) ;
3572+ assert ! ( !would_duplicate_volatile_exprs( & inner, & outer) ) ;
3573+ }
33453574}
0 commit comments