Addressed comments

Author: xiedeyantu

datafusion/common/src/config.rs

@@ -1174,7 +1174,7 @@ config_namespace! {
         /// with a combined filter. This optimization is conservative and only applies when the
         /// branches share the same source and compatible wrapper nodes such as identical
         /// projections or aliases.
-        pub enable_unions_to_filter: bool, default = false
+        pub enable_unions_to_filter: bool, default = true
     }
 }
 

datafusion/optimizer/src/unions_to_filter.rs

@@ -28,6 +28,7 @@ use datafusion_expr::utils::disjunction;
 use datafusion_expr::{
     Distinct, Expr, Filter, LogicalPlan, Projection, SubqueryAlias, Union,
 };
+use log::debug;
 use std::collections::HashMap;
 use std::sync::Arc;
 
@@ -80,10 +81,15 @@ impl OptimizerRule for UnionsToFilter {
 
 fn try_rewrite_distinct_union(plan: LogicalPlan) -> Result<Option<LogicalPlan>> {
     let LogicalPlan::Union(Union { inputs, schema }) = plan else {
+        debug!("unions_to_filter skipped: input is not a UNION");
         return Ok(None);
     };
 
     if inputs.len() < 2 {
+        debug!(
+            "unions_to_filter skipped: UNION has {} input(s), need at least 2",
+            inputs.len()
+        );
         return Ok(None);
     }
 
@@ -110,6 +116,7 @@ fn try_rewrite_distinct_union(plan: LogicalPlan) -> Result<Option<LogicalPlan>>
     }
 
     if !transformed {
+        debug!("unions_to_filter skipped: no branch groups could be merged");
         return Ok(None);
     }
 
@@ -138,20 +145,23 @@ fn try_rewrite_distinct_union(plan: LogicalPlan) -> Result<Option<LogicalPlan>>
     Ok(Some(LogicalPlan::Distinct(Distinct::All(Arc::new(union)))))
 }
 
-struct Branch {
+struct UnionBranch {
     source: LogicalPlan,
     predicate: Expr,
     wrappers: Vec<Wrapper>,
 }
 
-fn extract_branch(plan: LogicalPlan) -> Result<Option<Branch>> {
+fn extract_branch(plan: LogicalPlan) -> Result<Option<UnionBranch>> {
     let (wrappers, plan) = peel_wrappers(plan);
 
     // Volatile or subquery expressions in the projection must not be merged:
     // they are evaluated once per branch in the original plan but would be
     // evaluated once per combined row after the rewrite, which can change the
     // output row set.
     if !wrapper_projections_are_safe(&wrappers) {
+        debug!(
+            "unions_to_filter skipped: projection wrapper contains volatile expression or subquery"
+        );
         return Ok(None);
     }
 
@@ -160,9 +170,12 @@ fn extract_branch(plan: LogicalPlan) -> Result<Option<Branch>> {
             predicate, input, ..
         }) => {
             if !is_mergeable_predicate(&predicate) {
+                debug!(
+                    "unions_to_filter skipped: branch predicate contains volatility or a subquery"
+                );
                 return Ok(None);
             }
-            Ok(Some(Branch {
+            Ok(Some(UnionBranch {
                 source: strip_passthrough_nodes(Arc::unwrap_or_clone(input)),
                 predicate,
                 wrappers,
@@ -172,8 +185,15 @@ fn extract_branch(plan: LogicalPlan) -> Result<Option<Branch>> {
         // Merging two such branches into one would silently drop the per-branch
         // row restriction (LIMIT) or rely on an order guarantee that UNION does
         // not preserve (ORDER BY).  Bail out to leave the UNION unchanged.
-        LogicalPlan::Limit(_) | LogicalPlan::Sort(_) => Ok(None),
-        other => Ok(Some(Branch {
+        LogicalPlan::Limit(_) => {
+            debug!("unions_to_filter skipped: branch contains LIMIT");
+            Ok(None)
+        }
+        LogicalPlan::Sort(_) => {
+            debug!("unions_to_filter skipped: branch contains ORDER BY / SORT");
+            Ok(None)
+        }
+        other => Ok(Some(UnionBranch {
             source: strip_passthrough_nodes(other.clone()),
             predicate: Expr::Literal(
                 datafusion_common::ScalarValue::Boolean(Some(true)),

datafusion/sqllogictest/test_files/union.slt

@@ -274,6 +274,10 @@ physical_plan
 05)----DataSourceExec: partitions=1, partition_sizes=[1]
 
 # unions_to_filter is disabled by default
+
+statement ok
+set datafusion.optimizer.enable_unions_to_filter = false;
+
 query TT
 EXPLAIN SELECT id, name FROM t1 WHERE id = 1 UNION SELECT id, name FROM t1 WHERE id = 2
 ----