fix(unparser): apply empty projection fallback to Projection nodes

datafusion/core/tests/sql/unparser.rs

@@ -759,6 +759,26 @@ async fn optimized_filter_with_subquery_alias() -> Result<()> {
     Ok(())
 }
 
+/// `optimize_projections` prunes every column under `count(*)` when the
+/// filter column isn't needed above the filter, leaving a `Projection` with
+/// no expressions. Dialects without empty select list support must not
+/// receive `SELECT FROM`.
+#[tokio::test]
+async fn optimized_count_over_filter_empty_projection() -> Result<()> {
+    let ctx = SessionContext::new();
+    ctx.sql("create table t (id int)").await?.collect().await?;
+    let plan = ctx
+        .sql("select count(*) from t where id > 10")
+        .await?
+        .into_optimized_plan()?;
+    let sql = plan_to_sql(&plan)?.to_string();
+    assert_eq!(
+        sql,
+        r#"SELECT count(1) AS "count(*)" FROM (SELECT 1 FROM t WHERE (t.id > 10))"#
+    );
+    Ok(())
+}
+
 #[tokio::test]
 async fn optimized_filter_after_projection() -> Result<()> {
     let ctx = SessionContext::new();

datafusion/sql/src/unparser/plan.rs

@@ -311,7 +311,15 @@ impl Unparser<'_> {
         select: &mut SelectBuilder,
         fully_absorbed: bool,
     ) -> Result<bool> {
-        let mut exprs = p.expr.clone();
+        // A projection of no columns (e.g. under `count(*)` once
+        // `optimize_projections` has pruned every input column) gets the
+        // same fallback as an empty table scan projection, so dialects
+        // that reject `SELECT FROM` receive `SELECT 1 FROM`.
+        let mut exprs = if p.expr.is_empty() {
+            self.empty_projection_fallback()
+        } else {
+            p.expr.clone()
+        };
 
         // If an Unnest node is found within the select, find and unproject the unnest column
         let flatten_alias = select.current_flatten_alias();

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -1490,6 +1490,63 @@ fn test_table_scan_with_empty_projection_and_filter_default_dialect() {
     );
 }
 
+/// A `Projection` node with no expressions, as `optimize_projections`
+/// leaves under `count(*)` when the filter column isn't otherwise needed.
+///
+/// For `select count(*) from t where id > 10`, the optimized plan is:
+///
+/// ```text
+/// Projection: count(Int64(1)) AS count(*)
+///   Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
+///     Projection:                          <-- empty
+///       Filter: t.id > Int32(10)
+///         TableScan: t projection=[id]
+/// ```
+///
+/// See also `optimized_count_over_filter_empty_projection` in
+/// `datafusion/core/tests/sql/unparser.rs`.
+fn filter_with_empty_projection_helper() -> LogicalPlanBuilder {
+    table_scan(Some("table"), &test_schema(), None)
+        .unwrap()
+        .filter(col("id").gt(lit(10)))
+        .unwrap()
+        .project(Vec::<Expr>::new())
+        .unwrap()
+}
+
+#[test]
+fn test_empty_projection_node_default_dialect() {
+    let plan = filter_with_empty_projection_helper().build().unwrap();
+    let unparser = Unparser::new(&UnparserDefaultDialect {});
+    let sql = unparser.plan_to_sql(&plan).unwrap();
+    assert_snapshot!(
+        sql,
+        @r#"SELECT 1 FROM "table" WHERE ("table".id > 10)"#
+    );
+}
+
+#[test]
+fn test_empty_projection_node_postgres() {
+    let plan = filter_with_empty_projection_helper().build().unwrap();
+    let unparser = Unparser::new(&UnparserPostgreSqlDialect {});
+    let sql = unparser.plan_to_sql(&plan).unwrap();
+    assert_snapshot!(
+        sql,
+        @r#"SELECT FROM "table" WHERE ("table"."id" > 10)"#
+    );
+}
+
+#[test]
+fn test_empty_projection_node_bigquery() {
+    let plan = filter_with_empty_projection_helper().build().unwrap();
+    let unparser = Unparser::new(&BigQueryDialect {});
+    let sql = unparser.plan_to_sql(&plan).unwrap();
+    assert_snapshot!(
+        sql,
+        @"SELECT 1 FROM `table` WHERE (`table`.`id` > 10)"
+    );
+}
+
 fn table_scan_with_empty_projection_and_none_projection_helper(
     table_name: &str,
     table_schema: Schema,