datafusion_optimizer/

propagate_empty_relation.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! [`PropagateEmptyRelation`] eliminates nodes fed by `EmptyRelation`

use std::sync::Arc;

use datafusion_common::JoinType;
use datafusion_common::tree_node::Transformed;
use datafusion_common::{Column, DFSchemaRef, Result, ScalarValue, plan_err};
use datafusion_expr::logical_plan::LogicalPlan;
use datafusion_expr::{EmptyRelation, Expr, GroupingSet, Projection, Union, cast, lit};

use crate::optimizer::ApplyOrder;
use crate::{OptimizerConfig, OptimizerRule};

/// Optimization rule that bottom-up to eliminate plan by propagating empty_relation.
#[derive(Default, Debug)]
pub struct PropagateEmptyRelation;

impl PropagateEmptyRelation {
    #[expect(missing_docs)]
    pub fn new() -> Self {
        Self {}
    }
}

impl OptimizerRule for PropagateEmptyRelation {
    fn name(&self) -> &str {
        "propagate_empty_relation"
    }

    fn apply_order(&self) -> Option<ApplyOrder> {
        Some(ApplyOrder::BottomUp)
    }

    fn supports_rewrite(&self) -> bool {
        true
    }

    fn rewrite(
        &self,
        plan: LogicalPlan,
        _config: &dyn OptimizerConfig,
    ) -> Result<Transformed<LogicalPlan>> {
        match plan {
            LogicalPlan::EmptyRelation(_) => Ok(Transformed::no(plan)),
            LogicalPlan::Projection(_)
            | LogicalPlan::Filter(_)
            | LogicalPlan::Window(_)
            | LogicalPlan::Sort(_)
            | LogicalPlan::SubqueryAlias(_)
            | LogicalPlan::Repartition(_)
            | LogicalPlan::Limit(_) => {
                let empty = empty_child(&plan)?;
                if let Some(empty_plan) = empty {
                    return Ok(Transformed::yes(empty_plan));
                }
                Ok(Transformed::no(plan))
            }
            LogicalPlan::Join(ref join) => {
                let (left_empty, right_empty) = binary_plan_children_is_empty(&plan)?;
                let left_field_count = join.left.schema().fields().len();

                match join.join_type {
                    // For Full Join, only both sides are empty, the Join result is empty.
                    JoinType::Full if left_empty && right_empty => Ok(Transformed::yes(
                        LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        }),
                    )),
                    // For Full Join, if one side is empty, replace with a
                    // Projection that null-pads the empty side's columns.
                    JoinType::Full if right_empty => {
                        Ok(Transformed::yes(build_null_padded_projection(
                            Arc::clone(&join.left),
                            &join.schema,
                            left_field_count,
                            true,
                        )?))
                    }
                    JoinType::Full if left_empty => {
                        Ok(Transformed::yes(build_null_padded_projection(
                            Arc::clone(&join.right),
                            &join.schema,
                            left_field_count,
                            false,
                        )?))
                    }
                    JoinType::Inner if left_empty || right_empty => Ok(Transformed::yes(
                        LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        }),
                    )),
                    JoinType::Left if left_empty => Ok(Transformed::yes(
                        LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        }),
                    )),
                    // Left Join with empty right: all left rows survive
                    // with NULLs for right columns.
                    JoinType::Left if right_empty => {
                        Ok(Transformed::yes(build_null_padded_projection(
                            Arc::clone(&join.left),
                            &join.schema,
                            left_field_count,
                            true,
                        )?))
                    }
                    JoinType::Right if right_empty => Ok(Transformed::yes(
                        LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        }),
                    )),
                    // Right Join with empty left: all right rows survive
                    // with NULLs for left columns.
                    JoinType::Right if left_empty => {
                        Ok(Transformed::yes(build_null_padded_projection(
                            Arc::clone(&join.right),
                            &join.schema,
                            left_field_count,
                            false,
                        )?))
                    }
                    JoinType::LeftSemi if left_empty || right_empty => Ok(
                        Transformed::yes(LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        })),
                    ),
                    JoinType::RightSemi if left_empty || right_empty => Ok(
                        Transformed::yes(LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        })),
                    ),
                    JoinType::LeftAnti if left_empty => Ok(Transformed::yes(
                        LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        }),
                    )),
                    JoinType::LeftAnti if right_empty => {
                        Ok(Transformed::yes((*join.left).clone()))
                    }
                    JoinType::RightAnti if left_empty => {
                        Ok(Transformed::yes((*join.right).clone()))
                    }
                    JoinType::RightAnti if right_empty => Ok(Transformed::yes(
                        LogicalPlan::EmptyRelation(EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(&join.schema),
                        }),
                    )),
                    _ => Ok(Transformed::no(plan)),
                }
            }
            LogicalPlan::Aggregate(ref agg) => {
                // An aggregate over an empty input can be eliminated only when
                // there is no empty grouping set.  An empty grouping set `()`
                // (from `GROUPING SETS(())`, `ROLLUP(...)`, or `CUBE(...)`)
                // always produces exactly one row even on empty input, so it
                // must not be replaced by an empty relation.
                if !agg.group_expr.is_empty()
                    && !has_empty_grouping_set(&agg.group_expr)
                    && let Some(empty_plan) = empty_child(&plan)?
                {
                    return Ok(Transformed::yes(empty_plan));
                }
                Ok(Transformed::no(LogicalPlan::Aggregate(agg.clone())))
            }
            LogicalPlan::Union(ref union) => {
                let new_inputs = union
                    .inputs
                    .iter()
                    .filter(|input| match &***input {
                        LogicalPlan::EmptyRelation(empty) => empty.produce_one_row,
                        _ => true,
                    })
                    .cloned()
                    .collect::<Vec<_>>();

                if new_inputs.len() == union.inputs.len() {
                    Ok(Transformed::no(plan))
                } else if new_inputs.is_empty() {
                    Ok(Transformed::yes(LogicalPlan::EmptyRelation(
                        EmptyRelation {
                            produce_one_row: false,
                            schema: Arc::clone(plan.schema()),
                        },
                    )))
                } else if new_inputs.len() == 1 {
                    let mut new_inputs = new_inputs;
                    let input_plan = new_inputs.pop().unwrap(); // length checked
                    let child = Arc::unwrap_or_clone(input_plan);
                    if child.schema().eq(plan.schema()) {
                        Ok(Transformed::yes(child))
                    } else {
                        Ok(Transformed::yes(LogicalPlan::Projection(
                            Projection::new_from_schema(
                                Arc::new(child),
                                Arc::clone(plan.schema()),
                            ),
                        )))
                    }
                } else {
                    Ok(Transformed::yes(LogicalPlan::Union(Union {
                        inputs: new_inputs,
                        schema: Arc::clone(&union.schema),
                    })))
                }
            }

            _ => Ok(Transformed::no(plan)),
        }
    }
}

fn binary_plan_children_is_empty(plan: &LogicalPlan) -> Result<(bool, bool)> {
    match plan.inputs()[..] {
        [left, right] => {
            let left_empty = match left {
                LogicalPlan::EmptyRelation(empty) => !empty.produce_one_row,
                _ => false,
            };
            let right_empty = match right {
                LogicalPlan::EmptyRelation(empty) => !empty.produce_one_row,
                _ => false,
            };
            Ok((left_empty, right_empty))
        }
        _ => plan_err!("plan just can have two child"),
    }
}

fn empty_child(plan: &LogicalPlan) -> Result<Option<LogicalPlan>> {
    match plan.inputs()[..] {
        [child] => match child {
            LogicalPlan::EmptyRelation(empty) => {
                if !empty.produce_one_row {
                    Ok(Some(LogicalPlan::EmptyRelation(EmptyRelation {
                        produce_one_row: false,
                        schema: Arc::clone(plan.schema()),
                    })))
                } else {
                    Ok(None)
                }
            }
            _ => Ok(None),
        },
        _ => plan_err!("plan just can have one child"),
    }
}

/// Builds a Projection that replaces one side of an outer join with NULL literals.
///
/// When one side of an outer join is an `EmptyRelation`, the join can be eliminated
/// by projecting the surviving side's columns as-is and replacing the empty side's
/// columns with `CAST(NULL AS <type>)`.
///
/// The join schema is used as the projection's output schema to preserve nullability
/// guarantees (important for FULL JOIN where the surviving side's columns are marked
/// nullable in the join schema even if they aren't in the source schema).
///
/// # Example
///
/// For a `LEFT JOIN` where the right side is empty:
/// ```text
/// Left Join (orders.id = returns.order_id)        Projection(orders.id, orders.amount,
///   ├── TableScan: orders                   =>      CAST(NULL AS Int64) AS order_id,
///   └── EmptyRelation                               CAST(NULL AS Utf8) AS reason)
///                                                   └── TableScan: orders
/// ```
fn build_null_padded_projection(
    surviving_plan: Arc<LogicalPlan>,
    join_schema: &DFSchemaRef,
    left_field_count: usize,
    empty_side_is_right: bool,
) -> Result<LogicalPlan> {
    let exprs = join_schema
        .iter()
        .enumerate()
        .map(|(i, (qualifier, field))| {
            let on_empty_side = if empty_side_is_right {
                i >= left_field_count
            } else {
                i < left_field_count
            };

            if on_empty_side {
                cast(lit(ScalarValue::Null), field.data_type().clone())
                    .alias_qualified(qualifier.cloned(), field.name())
            } else {
                Expr::Column(Column::new(qualifier.cloned(), field.name()))
            }
        })
        .collect::<Vec<_>>();

    Ok(LogicalPlan::Projection(Projection::try_new_with_schema(
        exprs,
        surviving_plan,
        Arc::clone(join_schema),
    )?))
}

/// Returns `true` if any grouping set in the list of GROUP BY expressions is
/// the empty set `()`.
///
/// An empty grouping set acts as a "grand total" group: the aggregate must
/// always produce **exactly one row** for it, even when the input is empty.
/// This means an aggregate with an empty grouping set cannot be replaced by
/// an empty relation.
///
/// The three forms that can contain an empty grouping set:
/// - `GROUPING SETS (…, (), …)` — explicitly listed.
/// - `ROLLUP(exprs)` — always expands to include `()`.
/// - `CUBE(exprs)`  — always expands to include `()`.
fn has_empty_grouping_set(group_expr: &[Expr]) -> bool {
    match group_expr.first() {
        Some(Expr::GroupingSet(GroupingSet::GroupingSets(groups))) => {
            groups.iter().any(|g| g.is_empty())
        }
        // Both ROLLUP and CUBE always include the empty grouping set ().
        Some(Expr::GroupingSet(GroupingSet::Rollup(_)))
        | Some(Expr::GroupingSet(GroupingSet::Cube(_))) => true,
        _ => false,
    }
}

#[cfg(test)]
mod tests {

    use arrow::datatypes::{DataType, Field, Schema};

    use datafusion_common::{Column, DFSchema};
    use datafusion_expr::logical_plan::table_scan;
    use datafusion_expr::{
        Operator, binary_expr, col, lit, logical_plan::builder::LogicalPlanBuilder,
    };

    use crate::OptimizerContext;
    use crate::assert_optimized_plan_eq_snapshot;
    use crate::eliminate_filter::EliminateFilter;
    use crate::optimize_unions::OptimizeUnions;
    use crate::test::{
        assert_optimized_plan_with_rules, test_table_scan, test_table_scan_fields,
        test_table_scan_with_name,
    };

    use super::*;

    macro_rules! assert_optimized_plan_equal {
        (
            $plan:expr,
            @ $expected:literal $(,)?
        ) => {{
            let optimizer_ctx = OptimizerContext::new().with_max_passes(1);
            let rules: Vec<Arc<dyn crate::OptimizerRule + Send + Sync>> = vec![Arc::new(PropagateEmptyRelation::new())];
            assert_optimized_plan_eq_snapshot!(
                optimizer_ctx,
                rules,
                $plan,
                @ $expected,
            )
        }};
    }

    fn assert_together_optimized_plan(
        plan: LogicalPlan,
        expected: &str,
        eq: bool,
    ) -> Result<()> {
        assert_optimized_plan_with_rules(
            vec![
                Arc::new(EliminateFilter::new()),
                Arc::new(OptimizeUnions::new()),
                Arc::new(PropagateEmptyRelation::new()),
            ],
            plan,
            expected,
            eq,
        )
    }

    #[test]
    fn propagate_empty() -> Result<()> {
        let plan = LogicalPlanBuilder::empty(false)
            .filter(lit(true))?
            .limit(10, None)?
            .project(vec![binary_expr(lit(1), Operator::Plus, lit(1))])?
            .build()?;

        assert_optimized_plan_equal!(plan, @"EmptyRelation: rows=0")
    }

    #[test]
    fn cooperate_with_eliminate_filter() -> Result<()> {
        let table_scan = test_table_scan()?;
        let left = LogicalPlanBuilder::from(table_scan).build()?;
        let right_table_scan = test_table_scan_with_name("test2")?;
        let right = LogicalPlanBuilder::from(right_table_scan)
            .project(vec![col("a")])?
            .filter(lit(false))?
            .build()?;

        let plan = LogicalPlanBuilder::from(left)
            .join_using(
                right,
                JoinType::Inner,
                vec![Column::from_name("a".to_string())],
            )?
            .filter(col("a").lt_eq(lit(1i64)))?
            .build()?;

        let expected = "EmptyRelation: rows=0";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn propagate_union_empty() -> Result<()> {
        let left = LogicalPlanBuilder::from(test_table_scan()?).build()?;
        let right = LogicalPlanBuilder::from(test_table_scan_with_name("test2")?)
            .filter(lit(false))?
            .build()?;

        let plan = LogicalPlanBuilder::from(left).union(right)?.build()?;

        let expected = "Projection: a, b, c\n  TableScan: test";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn propagate_union_multi_empty() -> Result<()> {
        let one =
            LogicalPlanBuilder::from(test_table_scan_with_name("test1")?).build()?;
        let two = LogicalPlanBuilder::from(test_table_scan_with_name("test2")?)
            .filter(lit(false))?
            .build()?;
        let three = LogicalPlanBuilder::from(test_table_scan_with_name("test3")?)
            .filter(lit(false))?
            .build()?;
        let four =
            LogicalPlanBuilder::from(test_table_scan_with_name("test4")?).build()?;

        let plan = LogicalPlanBuilder::from(one)
            .union(two)?
            .union(three)?
            .union(four)?
            .build()?;

        let expected = "Union\
            \n  TableScan: test1\
            \n  TableScan: test4";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn propagate_union_all_empty() -> Result<()> {
        let one = LogicalPlanBuilder::from(test_table_scan_with_name("test1")?)
            .filter(lit(false))?
            .build()?;
        let two = LogicalPlanBuilder::from(test_table_scan_with_name("test2")?)
            .filter(lit(false))?
            .build()?;
        let three = LogicalPlanBuilder::from(test_table_scan_with_name("test3")?)
            .filter(lit(false))?
            .build()?;
        let four = LogicalPlanBuilder::from(test_table_scan_with_name("test4")?)
            .filter(lit(false))?
            .build()?;

        let plan = LogicalPlanBuilder::from(one)
            .union(two)?
            .union(three)?
            .union(four)?
            .build()?;

        let expected = "EmptyRelation: rows=0";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn propagate_union_children_different_schema() -> Result<()> {
        let one_schema = Schema::new(vec![Field::new("t1a", DataType::UInt32, false)]);
        let t1_scan = table_scan(Some("test1"), &one_schema, None)?.build()?;
        let one = LogicalPlanBuilder::from(t1_scan)
            .filter(lit(false))?
            .build()?;

        let two_schema = Schema::new(vec![Field::new("t2a", DataType::UInt32, false)]);
        let t2_scan = table_scan(Some("test2"), &two_schema, None)?.build()?;
        let two = LogicalPlanBuilder::from(t2_scan).build()?;

        let three_schema = Schema::new(vec![Field::new("t3a", DataType::UInt32, false)]);
        let t3_scan = table_scan(Some("test3"), &three_schema, None)?.build()?;
        let three = LogicalPlanBuilder::from(t3_scan).build()?;

        let plan = LogicalPlanBuilder::from(one)
            .union(two)?
            .union(three)?
            .build()?;

        let expected = "Union\
            \n  TableScan: test2\
            \n  TableScan: test3";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn propagate_union_alias() -> Result<()> {
        let left = LogicalPlanBuilder::from(test_table_scan()?).build()?;
        let right = LogicalPlanBuilder::from(test_table_scan_with_name("test2")?)
            .filter(lit(false))?
            .build()?;

        let plan = LogicalPlanBuilder::from(left).union(right)?.build()?;

        let expected = "Projection: a, b, c\n  TableScan: test";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn cross_join_empty() -> Result<()> {
        let table_scan = test_table_scan()?;
        let left = LogicalPlanBuilder::from(table_scan).build()?;
        let right = LogicalPlanBuilder::empty(false).build()?;

        let plan = LogicalPlanBuilder::from(left)
            .cross_join(right)?
            .filter(col("a").lt_eq(lit(1i64)))?
            .build()?;

        let expected = "EmptyRelation: rows=0";
        assert_together_optimized_plan(plan, expected, true)
    }

    fn assert_empty_left_empty_right_lp(
        left_empty: bool,
        right_empty: bool,
        join_type: JoinType,
        eq: bool,
    ) -> Result<()> {
        let left_lp = if left_empty {
            let left_table_scan = test_table_scan()?;

            LogicalPlanBuilder::from(left_table_scan)
                .filter(lit(false))?
                .build()
        } else {
            let scan = test_table_scan_with_name("left").unwrap();
            LogicalPlanBuilder::from(scan).build()
        }?;

        let right_lp = if right_empty {
            let right_table_scan = test_table_scan_with_name("right")?;

            LogicalPlanBuilder::from(right_table_scan)
                .filter(lit(false))?
                .build()
        } else {
            let scan = test_table_scan_with_name("right").unwrap();
            LogicalPlanBuilder::from(scan).build()
        }?;

        let plan = LogicalPlanBuilder::from(left_lp)
            .join_using(
                right_lp,
                join_type,
                vec![Column::from_name("a".to_string())],
            )?
            .build()?;

        let expected = "EmptyRelation: rows=0";
        assert_together_optimized_plan(plan, expected, eq)
    }

    // TODO: fix this long name
    fn assert_anti_join_empty_join_table_is_base_table(
        anti_left_join: bool,
    ) -> Result<()> {
        // if we have an anti join with an empty join table, then the result is the base_table
        let (left, right, join_type, expected) = if anti_left_join {
            let left = test_table_scan()?;
            let right = LogicalPlanBuilder::from(test_table_scan()?)
                .filter(lit(false))?
                .build()?;
            let expected = left.display_indent().to_string();
            (left, right, JoinType::LeftAnti, expected)
        } else {
            let right = test_table_scan()?;
            let left = LogicalPlanBuilder::from(test_table_scan()?)
                .filter(lit(false))?
                .build()?;
            let expected = right.display_indent().to_string();
            (left, right, JoinType::RightAnti, expected)
        };

        let plan = LogicalPlanBuilder::from(left)
            .join_using(right, join_type, vec![Column::from_name("a".to_string())])?
            .build()?;

        assert_together_optimized_plan(plan, &expected, true)
    }

    #[test]
    fn test_join_empty_propagation_rules() -> Result<()> {
        // test full join with empty left and empty right
        assert_empty_left_empty_right_lp(true, true, JoinType::Full, true)?;

        // test left join with empty left
        assert_empty_left_empty_right_lp(true, false, JoinType::Left, true)?;

        // test right join with empty right
        assert_empty_left_empty_right_lp(false, true, JoinType::Right, true)?;

        // test left semi join with empty left
        assert_empty_left_empty_right_lp(true, false, JoinType::LeftSemi, true)?;

        // test left semi join with empty right
        assert_empty_left_empty_right_lp(false, true, JoinType::LeftSemi, true)?;

        // test right semi join with empty left
        assert_empty_left_empty_right_lp(true, false, JoinType::RightSemi, true)?;

        // test right semi join with empty right
        assert_empty_left_empty_right_lp(false, true, JoinType::RightSemi, true)?;

        // test left anti join empty left
        assert_empty_left_empty_right_lp(true, false, JoinType::LeftAnti, true)?;

        // test right anti join empty right
        assert_empty_left_empty_right_lp(false, true, JoinType::RightAnti, true)?;

        // test left anti join empty right
        assert_anti_join_empty_join_table_is_base_table(true)?;

        // test right anti join empty left
        assert_anti_join_empty_join_table_is_base_table(false)
    }

    #[test]
    fn test_join_empty_propagation_rules_noop() -> Result<()> {
        // these cases should not result in an empty relation

        // test left join with empty right
        assert_empty_left_empty_right_lp(false, true, JoinType::Left, false)?;

        // test right join with empty left
        assert_empty_left_empty_right_lp(true, false, JoinType::Right, false)?;

        // test left semi with non-empty left and right
        assert_empty_left_empty_right_lp(false, false, JoinType::LeftSemi, false)?;

        // test right semi with non-empty left and right
        assert_empty_left_empty_right_lp(false, false, JoinType::RightSemi, false)?;

        // test left anti join with non-empty left and right
        assert_empty_left_empty_right_lp(false, false, JoinType::LeftAnti, false)?;

        // test left anti with non-empty left and empty right
        assert_empty_left_empty_right_lp(false, true, JoinType::LeftAnti, false)?;

        // test right anti join with non-empty left and right
        assert_empty_left_empty_right_lp(false, false, JoinType::RightAnti, false)?;

        // test right anti with empty left and non-empty right
        assert_empty_left_empty_right_lp(true, false, JoinType::RightAnti, false)
    }

    #[test]
    fn test_left_join_right_empty_null_pad() -> Result<()> {
        let left =
            LogicalPlanBuilder::from(test_table_scan_with_name("left")?).build()?;
        let right_empty = LogicalPlanBuilder::from(test_table_scan_with_name("right")?)
            .filter(lit(false))?
            .build()?;

        let plan = LogicalPlanBuilder::from(left)
            .join_using(
                right_empty,
                JoinType::Left,
                vec![Column::from_name("a".to_string())],
            )?
            .build()?;

        let expected = "Projection: left.a, left.b, left.c, CAST(NULL AS UInt32) AS a, CAST(NULL AS UInt32) AS b, CAST(NULL AS UInt32) AS c\n  TableScan: left";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn test_right_join_left_empty_null_pad() -> Result<()> {
        let left_empty = LogicalPlanBuilder::from(test_table_scan_with_name("left")?)
            .filter(lit(false))?
            .build()?;
        let right =
            LogicalPlanBuilder::from(test_table_scan_with_name("right")?).build()?;

        let plan = LogicalPlanBuilder::from(left_empty)
            .join_using(
                right,
                JoinType::Right,
                vec![Column::from_name("a".to_string())],
            )?
            .build()?;

        let expected = "Projection: CAST(NULL AS UInt32) AS a, CAST(NULL AS UInt32) AS b, CAST(NULL AS UInt32) AS c, right.a, right.b, right.c\n  TableScan: right";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn test_full_join_right_empty_null_pad() -> Result<()> {
        let left =
            LogicalPlanBuilder::from(test_table_scan_with_name("left")?).build()?;
        let right_empty = LogicalPlanBuilder::from(test_table_scan_with_name("right")?)
            .filter(lit(false))?
            .build()?;

        let plan = LogicalPlanBuilder::from(left)
            .join_using(
                right_empty,
                JoinType::Full,
                vec![Column::from_name("a".to_string())],
            )?
            .build()?;

        let expected = "Projection: left.a, left.b, left.c, CAST(NULL AS UInt32) AS a, CAST(NULL AS UInt32) AS b, CAST(NULL AS UInt32) AS c\n  TableScan: left";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn test_full_join_left_empty_null_pad() -> Result<()> {
        let left_empty = LogicalPlanBuilder::from(test_table_scan_with_name("left")?)
            .filter(lit(false))?
            .build()?;
        let right =
            LogicalPlanBuilder::from(test_table_scan_with_name("right")?).build()?;

        let plan = LogicalPlanBuilder::from(left_empty)
            .join_using(
                right,
                JoinType::Full,
                vec![Column::from_name("a".to_string())],
            )?
            .build()?;

        let expected = "Projection: CAST(NULL AS UInt32) AS a, CAST(NULL AS UInt32) AS b, CAST(NULL AS UInt32) AS c, right.a, right.b, right.c\n  TableScan: right";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn test_left_join_complex_on_right_empty_null_pad() -> Result<()> {
        let left =
            LogicalPlanBuilder::from(test_table_scan_with_name("left")?).build()?;
        let right_empty = LogicalPlanBuilder::from(test_table_scan_with_name("right")?)
            .filter(lit(false))?
            .build()?;

        // Complex ON condition: left.a = right.a AND left.b > right.b
        let plan = LogicalPlanBuilder::from(left)
            .join(
                right_empty,
                JoinType::Left,
                (
                    vec![Column::from_name("a".to_string())],
                    vec![Column::from_name("a".to_string())],
                ),
                Some(col("left.b").gt(col("right.b"))),
            )?
            .build()?;

        let expected = "Projection: left.a, left.b, left.c, CAST(NULL AS UInt32) AS a, CAST(NULL AS UInt32) AS b, CAST(NULL AS UInt32) AS c\n  TableScan: left";
        assert_together_optimized_plan(plan, expected, true)
    }

    #[test]
    fn test_empty_with_non_empty() -> Result<()> {
        let table_scan = test_table_scan()?;

        let fields = test_table_scan_fields();

        let empty = LogicalPlan::EmptyRelation(EmptyRelation {
            produce_one_row: false,
            schema: Arc::new(DFSchema::from_unqualified_fields(
                fields.into(),
                Default::default(),
            )?),
        });

        let one = LogicalPlanBuilder::from(empty.clone()).build()?;
        let two = LogicalPlanBuilder::from(table_scan).build()?;
        let three = LogicalPlanBuilder::from(empty).build()?;

        // Union
        //  EmptyRelation
        //  TableScan: test
        //  EmptyRelation
        let plan = LogicalPlanBuilder::from(one)
            .union(two)?
            .union(three)?
            .build()?;

        let expected = "Projection: a, b, c\
        \n  TableScan: test";

        assert_together_optimized_plan(plan, expected, true)
    }
}