fix(optimizer): optimize_projections rule bug

datafusion/optimizer/src/optimize_projections/mod.rs

@@ -469,6 +469,26 @@ fn merge_consecutive_projections(proj: Projection) -> Result<Transformed<Project
         return Projection::try_new_with_schema(expr, input, schema).map(Transformed::no);
     };
 
+    // In order to pass the detection of `assert_schema_is_the_same` function, avoid changing
+    // the plan schema here. e.g.
+    // Projection: public.test.a -- proj
+    //   Projection: public.test.a AS public.test.a -- prev_projection
+    //     TableScan: public.test
+    if prev_projection.expr.iter().any(|expr| match expr {
+        Expr::Alias(Alias {
+            expr,
+            name: alias_name,
+            ..
+        }) => match expr.as_ref() {
+            Expr::Column(column) => &column.flat_name() == alias_name,
+            _ => false,
+        },
+        _ => false,
+    }) {
+        // no change
+        return Projection::try_new_with_schema(expr, input, schema).map(Transformed::no);
+    }
+
     // Count usages (referrals) of each projection expression in its input fields:
     let mut column_referral_map = HashMap::<&Column, usize>::new();
     expr.iter()
@@ -1019,7 +1039,44 @@ mod tests {
 
         let expected = "Projection: test.a AS alias\
         \n  TableScan: test projection=[a]";
-        assert_optimized_plan_equal(plan, expected)
+        assert_optimized_plan_equal(plan, expected)?;
+
+        // SELECT "test.a" FROM (SELECT a AS "test.a" FROM test)
+        let table_scan = test_table_scan_with_name("test")?;
+        let plan = LogicalPlanBuilder::from(table_scan)
+            .project(vec![col("a").alias("test.a")])?
+            .project(vec![Expr::Column(Column::new_unqualified("test.a"))])?
+            .build()?;
+
+        let expected = "Projection: test.a AS test.a\
+        \n  TableScan: test projection=[a]";
+        assert_optimized_plan_equal(plan, expected)?;
+
+        // SELECT "public.test.a" FROM (SELECT a AS "public.test.a" FROM public.test)
+        let table_scan = test_table_scan_with_name("public.test")?;
+        let plan = LogicalPlanBuilder::from(table_scan)
+            .project(vec![col("a").alias("public.test.a")])?
+            .project(vec![Expr::Column(Column::new_unqualified("public.test.a"))])?
+            .build()?;
+
+        let expected = "Projection: public.test.a AS public.test.a\
+        \n  TableScan: public.test projection=[a]";
+        assert_optimized_plan_equal(plan, expected)?;
+
+        // SELECT "datafusion.public.test.a" FROM (SELECT a AS "datafusion.public.test.a" FROM datafusion.public.test)
+        let table_scan = test_table_scan_with_name("datafusion.public.test")?;
+        let plan = LogicalPlanBuilder::from(table_scan)
+            .project(vec![col("a").alias("datafusion.public.test.a")])?
+            .project(vec![Expr::Column(Column::new_unqualified(
+                "datafusion.public.test.a",
+            ))])?
+            .build()?;
+
+        let expected = "Projection: datafusion.public.test.a AS datafusion.public.test.a\
+        \n  TableScan: datafusion.public.test projection=[a]";
+        assert_optimized_plan_equal(plan, expected)?;
+
+        Ok(())
     }
 
     #[test]