Re: [PR] [spark] Fix empty projection issue in Spark batch read to avoid Arrow metadata errors with COUNT(*)/COUNT(1) [fluss]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #2725:
URL: https://github.com/apache/fluss/pull/2725#discussion_r3165394329


##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala:
##########
@@ -44,14 +44,38 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
           Row(1000L, 25L, 605, "addr5") :: Nil
       )
 
+      // Verify empty projection handling for COUNT(*)
+      checkAnswer(
+        sql(s"SELECT count(*) FROM $DEFAULT_DATABASE.t"),
+        Row(5L) :: Nil
+      )

Review Comment:
   The PR title mentions `COUNT(1)` as well, but the added Spark coverage here only exercises `COUNT(*)` (and `COUNT(column)`). Consider adding at least one `SELECT COUNT(1) ...` assertion to ensure the empty-projection fix is covered for that variant too.
   ```suggestion
         )
         checkAnswer(
           sql(s"SELECT COUNT(1) FROM $DEFAULT_DATABASE.t"),
           Row(5L) :: Nil
         )
   ```



##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java:
##########
@@ -500,11 +553,11 @@ private String prepareLogTable() throws Exception {
 
         TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
 
-        // prepare table data
+        // prepare table data with NULL values in address column
         try (Table table = conn.getTable(tablePath)) {
             AppendWriter appendWriter = table.newAppend().createWriter();
             for (int i = 1; i <= 5; i++) {
-                Object[] values = new Object[] {i, "address" + i, "name" + i};
+                Object[] values = new Object[] {i, i % 2 == 0 ? null : "address" + i, "name" + i};

Review Comment:
   `prepareLogTable()` now inserts NULLs into the `address` column for even ids, but `testLimitLogTableScan()` (and any other tests using this helper) still expects `address2/address4` to be non-null. This will make the limit scan assertions fail. Consider keeping `prepareLogTable()` data non-null and creating a dedicated helper/extra insert for NULL-specific scenarios, or update the affected expected results accordingly.



##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java:
##########
@@ -602,19 +601,33 @@ public boolean applyAggregates(
             return false;
         }
 
-        FunctionDefinition functionDefinition = aggregateExpressions.get(0).getFunctionDefinition();
-        if (!(functionDefinition
-                        .getClass()
-                        .getCanonicalName()
-                        .equals(
-                                "org.apache.flink.table.planner.functions.aggfunctions.CountAggFunction")
-                || functionDefinition
-                        .getClass()
-                        .getCanonicalName()
-                        .equals(
-                                "org.apache.flink.table.planner.functions.aggfunctions.Count1AggFunction"))) {
+        AggregateExpression aggExpr = aggregateExpressions.get(0);
+        String functionName = aggExpr.getFunctionDefinition().getClass().getCanonicalName();
+
+        // Verify that the aggregate function is COUNT(*) or COUNT(1)
+        // CountAggFunction: COUNT(*) or COUNT(column)
+        // Count1AggFunction: COUNT(1) with constant argument
+        boolean isCountAgg =

Review Comment:
   The method-level comment says only `COUNT(*)/COUNT(1)` is supported (and mentions log table), but the updated logic now also allows `COUNT(nonNullableExpr)` (e.g., `COUNT(id)`) to be pushed down. Please update the comment to match the actual supported cases and scope to avoid confusing future maintainers.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org