This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 305a3a1bd [flink] Fix COUNT(column) aggregate pushdown to reject
nullable columns (#3271)
305a3a1bd is described below
commit 305a3a1bda93060145e8307a3aa4d4096b448f9c
Author: Junbo Wang <[email protected]>
AuthorDate: Mon May 18 20:40:51 2026 +0800
[flink] Fix COUNT(column) aggregate pushdown to reject nullable columns
(#3271)
---
.../fluss/flink/source/FlinkTableSource.java | 37 ++++++++----
.../flink/source/FlinkTableSourceBatchITCase.java | 68 ++++++++++++++++++++--
2 files changed, 87 insertions(+), 18 deletions(-)
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
index 859ef7cfa..b0bba81a7 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
@@ -78,7 +78,6 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.AggregateExpression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.functions.AsyncLookupFunction;
-import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.LookupFunction;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
@@ -843,19 +842,33 @@ public class FlinkTableSource
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 =
+
"org.apache.flink.table.planner.functions.aggfunctions.CountAggFunction"
+ .equals(functionName);
+ boolean isCount1Agg =
+
"org.apache.flink.table.planner.functions.aggfunctions.Count1AggFunction"
+ .equals(functionName);
+ if (!isCountAgg && !isCount1Agg) {
return false;
}
+
+ // For COUNT(column), reject if column is nullable (cannot handle NULL
filtering)
+ if (isCountAgg) {
+ List<org.apache.flink.table.expressions.Expression> args =
aggExpr.getChildren();
+ if (!args.isEmpty() && args.get(0) instanceof ResolvedExpression) {
+ ResolvedExpression arg = (ResolvedExpression) args.get(0);
+ if (arg.getOutputDataType().getLogicalType().isNullable()) {
+ return false;
+ }
+ }
+ }
+
selectRowCount = true;
this.producedDataType = dataType.getLogicalType();
return true;
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java
index 48ee39af1..51d05fa03 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceBatchITCase.java
@@ -324,9 +324,9 @@ abstract class FlinkTableSourceBatchITCase extends
FlinkTestBase {
List<String> expected =
Arrays.asList(
"+I[1, address1, name1]",
- "+I[2, address2, name2]",
+ "+I[2, null, name2]",
"+I[3, address3, name3]",
- "+I[4, address4, name4]",
+ "+I[4, null, name4]",
"+I[5, address5, name5]");
assertThat(collected).isSubsetOf(expected);
assertThat(collected).hasSize(2);
@@ -404,6 +404,33 @@ abstract class FlinkTableSourceBatchITCase extends
FlinkTestBase {
List<String> expected = Collections.singletonList("+I[5]");
assertThat(collected).isEqualTo(expected);
+ // test COUNT(column) pushdown on non-nullable column
+ query = String.format("SELECT COUNT(id) FROM %s", tableName);
+ assertThat(tEnv.explainSql(query))
+ .contains(
+ "aggregates=[grouping=[],
aggFunctions=[Count1AggFunction()]]]], fields=[count1$0]");
+ iterRows = tEnv.executeSql(query).collect();
+ collected = collectRowsWithTimeout(iterRows, 1);
+ assertThat(collected).isEqualTo(expected);
+
+ // test COUNT(column) on nullable column - should NOT push down
+ // For PK table, this will fail because it doesn't support full scan
in batch mode
+ assertThatThrownBy(
+ () ->
+ tEnv.explainSql(
+ String.format("SELECT COUNT(address)
FROM %s", tableName)))
+ .hasMessageContaining(
+ "Currently, Fluss only support queries on table with
datalake enabled or point queries on primary key when it's in batch execution
mode.");
+
+ assertThatThrownBy(
+ () ->
+ tEnv.explainSql(
+ String.format(
+ "SELECT COUNT(DISTINCT
address) FROM %s",
+ tableName)))
+ .hasMessageContaining(
+ "Currently, Fluss only support queries on table with
datalake enabled or point queries on primary key when it's in batch execution
mode.");
+
// test not push down grouping count.
assertThatThrownBy(
() ->
@@ -452,6 +479,32 @@ abstract class FlinkTableSourceBatchITCase extends
FlinkTestBase {
List<String> expected =
Collections.singletonList(String.format("+I[%s]", expectedRows));
assertThat(collected).isEqualTo(expected);
+ // test COUNT(column) pushdown
+ query = String.format("SELECT COUNT(id) FROM %s", tableName);
+ assertThat(tEnv.explainSql(query))
+ .contains(
+ "aggregates=[grouping=[],
aggFunctions=[Count1AggFunction()]]]], fields=[count1$0]");
+ iterRows = tEnv.executeSql(query).collect();
+ collected = collectRowsWithTimeout(iterRows, 1);
+ assertThat(collected).isEqualTo(expected);
+
+ // test COUNT(column) with NULL values - should NOT push down for
nullable columns
+ // This will fail because log table doesn't support full scan in batch
mode
+ assertThatThrownBy(
+ () ->
+ tEnv.explainSql(
+ String.format("SELECT COUNT(address)
FROM %s", tableName)))
+ .hasMessageContaining(
+ "Currently, Fluss only support queries on table with
datalake enabled or point queries on primary key when it's in batch execution
mode.");
+ assertThatThrownBy(
+ () ->
+ tEnv.explainSql(
+ String.format(
+ "SELECT COUNT(DISTINCT
address) FROM %s",
+ tableName)))
+ .hasMessageContaining(
+ "Currently, Fluss only support queries on table with
datalake enabled or point queries on primary key when it's in batch execution
mode.");
+
// test not push down grouping count.
assertThatThrownBy(
() ->
@@ -536,11 +589,11 @@ abstract class FlinkTableSourceBatchITCase extends
FlinkTestBase {
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};
appendWriter.append(row(values));
// make sure every bucket has records
appendWriter.flush();
@@ -571,12 +624,15 @@ abstract class FlinkTableSourceBatchITCase extends
FlinkTestBase {
waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), tablePath);
Collection<String> partitions = partitionNameById.values();
- // 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++) {
for (String partition : partitions) {
- Object[] values = new Object[] {i, "address" + i, "name" +
i, partition};
+ Object[] values =
+ new Object[] {
+ i, i % 2 == 0 ? null : "address" + i, "name" +
i, partition
+ };
appendWriter.append(row(values));
// make sure every bucket has records
appendWriter.flush();
