[platinumhamburg]

Purpose

Closes #4997.

Support native non-null input counting in the aggregation merge engine without requiring writers to generate a constant 1 and configure SUM.

Brief change log

• Add AggFunctionType.COUNT, AggFunctions.COUNT(), and fields.<column>.agg = count support for BIGINT columns.
• Register the COUNT field aggregator and use the existing first-write aggregation hook. Every non-null input contributes one; NULL contributes zero, including on the first write.
• Respect target columns on the first partial write: untargeted columns remain NULL, so an untouched COUNT column does not prevent a subsequent partial delete from removing the row.
• Cover both KV formats, partial updates, partial deletes, non-target input values, first partial writes across schema versions with reordered columns, and BIGINT type validation. Also verify first partial writes for SUM-only tables.

Tests

Passed with Java 11:

./mvnw verify -pl fluss-server -am '-Dtest=TableDescriptorTest,TableSchemaTest,AggregateRowMergerTest,Field*AggTest,FieldAggregatorParameterizedTest' -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false
./mvnw verify -pl fluss-server -am -Dtest=FieldCountAggTest,AggregateRowMergerTest -Dsurefire.failIfNoSpecifiedTests=false -DfailIfNoTests=false

Focused verification passed for metadata validation and the existing aggregation suite. The COUNT and row-merger verification passed all 30 cases, including 14 COUNT cases. The partial-delete regression failed on both KV formats before the fix and passed afterward. Checkstyle, Spotless, and RAT checks passed. The full repository test suite was not run.

API and Format

Adds the public evolving AggFunctions.COUNT() factory and COUNT function type. Input and stored output use BIGINT. No RPC or storage format changes. COUNT follows the existing delete behavior and does not support retractions or COUNT DISTINCT.

Documentation

Adds COUNT semantics, SQL and Java usage, the distinction between counting inputs and summing precomputed counts, and the behavior of omitted versus explicitly null COUNT columns on the first partial write to the aggregation merge engine documentation.

[polyzos]

LGTM, thanks! 👍

One note: COUNT turns agg(null, null) into 0, unlike other aggregators. A partial update that doesn't target cnt stores 0, so a later partial delete keeps the row instead of removing it.

This is not a blocker and we can merge once the CI passes and just have it a a follow-up if you think it makes sense.

[platinumhamburg]

Thanks @polyzos, good catch! Addressed in f38fff0. First partial writes now respect the target columns, leaving an untouched COUNT column NULL so a later partial delete can remove the row. Explicitly targeting COUNT with a NULL input still stores 0.

Added regression coverage for both KV formats and verified that the partial-delete case fails before the fix and passes afterward. The targeted Java 11 verification passed: 24 metadata tests and 119 aggregation tests, including 10 COUNT cases.