Implement basic Aggregate Merge Engine #2255

platinumhamburg · 2025-12-25T06:22:50Z

Purpose

Linked issue: close #2254

Brief change log

Tests

API and Format

Documentation

Copilot

Pull request overview

This PR implements the Aggregation Merge Engine for Fluss, which allows field-level aggregation of rows with the same primary key. The implementation includes 12 aggregate functions (sum, product, max, min, last_value, last_value_ignore_nulls, first_value, first_value_ignore_nulls, listagg, string_agg, bool_and, bool_or) with comprehensive schema evolution support.

Key changes:

Core aggregation engine with field-level aggregators and caching
Schema API enhancements to support aggregation function configuration
Comprehensive documentation with examples for all supported functions
Utility classes for string concatenation and value comparison

Reviewed changes

Copilot reviewed 56 out of 56 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
website/docs/table-design/table-types/pk-table/merge-engines/aggregation.md	Comprehensive documentation for the Aggregation Merge Engine with examples
fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java	Main row merger implementation with partial update support
fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/*.java	Aggregation context, caching, and field processing logic
fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/*.java	Field aggregator implementations for all functions
fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/factory/*.java	Factory classes for creating aggregators via SPI
fluss-common/src/main/java/org/apache/fluss/metadata/AggFunction.java	Enum defining all supported aggregation functions
fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java	Schema enhancements to support aggregation functions
fluss-common/src/main/java/org/apache/fluss/utils/*.java	Utility classes for string operations and comparisons
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	New configuration options for aggregation
Test files	Comprehensive test coverage for all components

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java

wuchong

Thanks @platinumhamburg , I didn't complete the review. I gave some feedbacks first after reading the documentation.

website/docs/table-design/table-types/pk-table/merge-engines/aggregation.md

…ggregation.md Co-authored-by: Jark Wu <imjark@gmail.com>

platinumhamburg · 2025-12-29T09:46:36Z

Thanks @platinumhamburg , I didn't complete the review. I gave some feedbacks first after reading the documentation.

@wuchong I've refactored both the code and documentation based on the feedback. Thanks for review again.

wuchong

Certainly! Here's a more polished, professional, and clear version of your message:

Thanks @platinumhamburg for the excellent work!

The code is well-designed, thoroughly tested, and clearly documented.

I’ve left some comments. One additional suggestion: could we add sanity checks for the parameters of AggregateFunctions when creating an aggregate merge table?

Currently, there’s no validation of the parameters being passed. Since these are persisted to ZooKeeper, malformed or incorrect inputs (e.g., typos in parameter keys) could lead to subtle runtime issues during data writes.

Adding upfront validation would improve robustness and user experience.

wuchong · 2025-12-29T12:29:29Z

fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctions.java

+     */
+    public static AggFunction LISTAGG(String delimiter) {
+        Map<String, String> params = new HashMap<>();
+        params.put("delimiter", delimiter);


Add delimiter as a public static constant to enable easy and consistent referencing across the codebase.

wuchong · 2025-12-29T12:34:02Z

fluss-common/src/main/java/org/apache/fluss/metadata/MergeEngineType.java

+     *
+     * @since 0.9
+     */
+    AGGREGATE;


AGGREGATION

wuchong · 2025-12-29T12:44:00Z

fluss-common/src/test/java/org/apache/fluss/utils/json/SchemaJsonSerdeTest.java

    }
+
+    @Test
+    void testSchemaWithAggregationFunctions() {


Since you have added SCHEMA_WITH_AGG into createObjects(), this json serialization and deserialization test has already been verified by parent org.apache.fluss.utils.json.JsonSerdeTestBase#testJsonSerde. So I think we can remove this test method.

wuchong · 2025-12-29T12:44:11Z

fluss-common/src/test/java/org/apache/fluss/utils/json/SchemaJsonSerdeTest.java

+    }
+
+    @Test
+    void testSchemaWithoutAggregationFunctions() {


ditto. we can remove it.

wuchong · 2025-12-29T12:48:20Z

fluss-common/src/test/java/org/apache/fluss/utils/json/SchemaJsonSerdeTest.java

+            Schema.newBuilder()
+                    .column("product_id", DataTypes.BIGINT().copy(false))
+                    .column("total_sales", DataTypes.BIGINT(), AggFunctions.SUM())
+                    .column("max_price", DataTypes.DECIMAL(10, 2), AggFunctions.MAX())


Could you add aggregate functions that has parameters, like LISTAGG, to verify the json format of such agg functions.

wuchong · 2025-12-29T17:12:46Z

...ver/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContextTest.java

+                        .primaryKey("id")
+                        .build();
+
+        TableConfig tableConfig = new TableConfig(new Configuration());


nit: can be removed.

wuchong · 2025-12-29T17:13:00Z

...ver/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContextTest.java

+                        .primaryKey("id1", "id2")
+                        .build();
+
+        TableConfig tableConfig = new TableConfig(new Configuration());


wuchong · 2025-12-29T17:14:06Z

...ver/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContextTest.java

+                        .build();
+
+        // No configuration at all
+        TableConfig tableConfig = new TableConfig(new Configuration());


wuchong · 2025-12-29T17:14:27Z

...ver/src/test/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContextTest.java

+                        .primaryKey("id")
+                        .build();
+
+        TableConfig tableConfig = new TableConfig(new Configuration());


wuchong · 2025-12-29T17:21:04Z

...st/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldAggregatorParameterizedTest.java

+                        .build();
+
+        Configuration conf = new Configuration();
+        conf.setString("table.merge-engine.aggregate.value.listagg-delimiter", "|");


replace to define delimiter in schema

platinumhamburg changed the title ~~basic aggregate merge engine support~~ Implement basic Aggregate Merge Engine Dec 25, 2025

platinumhamburg force-pushed the base-agg branch 21 times, most recently from 907559e to e7d0af7 Compare December 26, 2025 06:45

Implement basic Aggregate Merge Engine

d34b27a

platinumhamburg force-pushed the base-agg branch from e7d0af7 to d34b27a Compare December 26, 2025 07:06

improve listagg performance

34dd539

wuchong added the priority=critical label Dec 26, 2025

wuchong requested review from Copilot and wuchong December 26, 2025 13:44

Copilot started reviewing on behalf of wuchong December 26, 2025 13:44 View session

Copilot AI reviewed Dec 26, 2025

View reviewed changes

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java Show resolved Hide resolved

wuchong reviewed Dec 28, 2025

View reviewed changes

platinumhamburg and others added 4 commits December 29, 2025 10:34

Update website/docs/table-design/table-types/pk-table/merge-engines/a…

9ba7905

…ggregation.md Co-authored-by: Jark Wu <imjark@gmail.com>

Update website/docs/table-design/table-types/pk-table/merge-engines/a…

a7ae04e

…ggregation.md Co-authored-by: Jark Wu <imjark@gmail.com>

refine document

03fd91f

refine delete behaviour & schema api

400a6f1

platinumhamburg force-pushed the base-agg branch from 42ab4e0 to 400a6f1 Compare December 29, 2025 09:41

wuchong and others added 2 commits December 29, 2025 20:10

update docs and fix compile

fc3a605

fix unit test

ec3d44a

wuchong reviewed Dec 29, 2025

View reviewed changes

Implement basic Aggregate Merge Engine #2255

Are you sure you want to change the base?

Implement basic Aggregate Merge Engine #2255

Uh oh!

Conversation

platinumhamburg commented Dec 25, 2025

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

platinumhamburg commented Dec 29, 2025

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants