add propagate_empty_relation optimizer rule

datafusion/expr/src/logical_plan/builder.rs

@@ -419,6 +419,18 @@ impl LogicalPlanBuilder {
         Ok(Self::from(union_with_alias(self.plan.clone(), plan, None)?))
     }
 
+    pub fn union_with_alias(
+        &self,
+        plan: LogicalPlan,
+        alias: Option<String>,
+    ) -> Result<Self> {
+        Ok(Self::from(union_with_alias(
+            self.plan.clone(),
+            plan,
+            alias,
+        )?))
+    }
+
     /// Apply a union, removing duplicate rows
     pub fn union_distinct(&self, plan: LogicalPlan) -> Result<Self> {
         // unwrap top-level Distincts, to avoid duplication

datafusion/expr/src/logical_plan/plan.rs

@@ -1145,6 +1145,26 @@ impl Projection {
         })
     }
 
+    /// Create a new Projection using the specified output schema
+    pub fn new_from_schema(
+        input: Arc<LogicalPlan>,
+        schema: DFSchemaRef,
+        alias: Option<String>,
+    ) -> Self {
+        let expr: Vec<Expr> = schema
+            .fields()
+            .iter()
+            .map(|field| field.qualified_column())
+            .map(Expr::Column)
+            .collect();
+        Self {
+            expr,
+            input,
+            schema,
+            alias,
+        }
+    }
+
     pub fn try_from_plan(plan: &LogicalPlan) -> datafusion_common::Result<&Projection> {
         match plan {
             LogicalPlan::Projection(it) => Ok(it),

datafusion/optimizer/src/lib.rs

@@ -26,6 +26,7 @@ pub mod inline_table_scan;
 pub mod limit_push_down;
 pub mod optimizer;
 pub mod projection_push_down;
+pub mod propagate_empty_relation;
 pub mod reduce_cross_join;
 pub mod reduce_outer_join;
 pub mod scalar_subquery_to_join;

datafusion/optimizer/src/optimizer.rs

@@ -27,6 +27,7 @@ use crate::filter_push_down::FilterPushDown;
 use crate::inline_table_scan::InlineTableScan;
 use crate::limit_push_down::LimitPushDown;
 use crate::projection_push_down::ProjectionPushDown;
+use crate::propagate_empty_relation::PropagateEmptyRelation;
 use crate::reduce_cross_join::ReduceCrossJoin;
 use crate::reduce_outer_join::ReduceOuterJoin;
 use crate::rewrite_disjunctive_predicate::RewriteDisjunctivePredicate;
@@ -165,6 +166,7 @@ impl Optimizer {
             Arc::new(ReduceCrossJoin::new()),
             Arc::new(CommonSubexprEliminate::new()),
             Arc::new(EliminateLimit::new()),
+            Arc::new(PropagateEmptyRelation::new()),
             Arc::new(RewriteDisjunctivePredicate::new()),
         ];
         if config.filter_null_keys {