Fix: InterleaveExec fallback to UnionExec when children partitioning diverges

After optimizer rewrites (e.g. join_selection changing join modes, or
additional EnforceDistribution passes), InterleaveExec children may no
longer have consistent hash partitioning. Instead of panicking with an
assertion error, fall back to UnionExec. Correctness is preserved; only
the interleave optimization is lost.

This fixes materialization failures on complex UNION ALL queries (like
tickers_v2) where join_selection changes a child's partitioning after
InterleaveExec was created by an earlier EnforceDistribution pass.

Author: zhuqi-lucas

datafusion/physical-plan/src/union.rs

@@ -500,12 +500,16 @@ impl ExecutionPlan for InterleaveExec {
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        // New children are no longer interleavable, which might be a bug of optimization rewrite.
-        assert_or_internal_err!(
-            can_interleave(children.iter()),
-            "Can not create InterleaveExec: new children can not be interleaved"
-        );
-        Ok(Arc::new(InterleaveExec::try_new(children)?))
+        // After optimizer rewrites (e.g. join_selection changing join modes,
+        // or a second EnforceDistribution pass), children may no longer have
+        // consistent hash partitioning. Fall back to UnionExec instead of
+        // panicking — correctness is preserved, only the interleave
+        // optimization is lost.
+        if can_interleave(children.iter()) {
+            Ok(Arc::new(InterleaveExec::try_new(children)?))
+        } else {
+            Ok(Arc::new(UnionExec::try_new(children)?))
+        }
     }
 
     fn execute(