You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "platinumhamburg (via GitHub)" <gi...@apache.org> on 2026/01/27 06:57:10 UTC
[PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
platinumhamburg opened a new pull request, #2492:
URL: https://github.com/apache/fluss/pull/2492
This commit introduces AggMode (Aggregation Mode) to control how the server handles data aggregation when writing to tables with aggregation merge engine.
Key changes:
1. New AggMode enum with three modes:
- AGGREGATE (default): Data is aggregated through server-side merge engine
- OVERWRITE: Bypass merge engine, directly replace values (for undo recovery)
- LOCAL_AGGREGATE: Reserved for future client-side pre-aggregation
2. Client-side changes:
- Upsert interface: Added aggregationMode(AggMode) method for fluent API
- UpsertWriterImpl: Propagates aggMode through WriteRecord
- KvWriteBatch: Validates aggMode consistency within batch
- ClientRpcMessageUtils: Validates aggMode consistency across batches
- WriteRecord: Added aggMode field for upsert/delete operations
3. Server-side changes:
- KvTablet: Pre-creates overwriteRowMerger for OVERWRITE mode
- putAsLeader(): Selects appropriate RowMerger based on aggMode
- Replica/ReplicaManager: Propagates aggMode through call chain
- TabletService: Extracts aggMode from PutKvRequest
4. Protocol changes:
- FlussApi.proto: Added optional agg_mode field to PutKvRequest
5. Test coverage:
- KvTabletAggModeTest: 9 tests covering OVERWRITE mode scenarios
- KvWriteBatchTest: 3 tests for aggMode consistency validation
- ClientRpcMessageUtilsTest: 4 tests for multi-batch aggMode validation
This feature enables Flink connector to perform undo recovery by restoring exact historical values during checkpoint failover, bypassing the merge engine.
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #2491 2491
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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
Re: [PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #2492:
URL: https://github.com/apache/fluss/pull/2492#issuecomment-3821483355
Thanks @wuchong , I've addressed all comments above, please take another look when you have time.
--
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
Re: [PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2492:
URL: https://github.com/apache/fluss/pull/2492#issuecomment-3822286534
@platinumhamburg LGTM. Could you rebase and resolve the conflicts?
--
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
Re: [PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #2492:
URL: https://github.com/apache/fluss/pull/2492#discussion_r2744320317
##########
fluss-common/src/main/java/org/apache/fluss/rpc/protocol/AggMode.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.rpc.protocol;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * Aggregation mode for write operations.
+ *
+ * <p>This enum controls how the server handles data aggregation when writing to tables with
+ * aggregation merge engine.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public enum AggMode {
Review Comment:
I think this new design makes more sense, and I'll revise it according to this design.
--
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
Re: [PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2492:
URL: https://github.com/apache/fluss/pull/2492
--
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
Re: [PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2492:
URL: https://github.com/apache/fluss/pull/2492#discussion_r2742565353
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TableUpsert.java:
##########
@@ -91,9 +94,15 @@ public Upsert partialUpdate(String... targetColumnNames) {
return partialUpdate(targetColumns);
}
+ @Override
+ public Upsert aggregationMode(AggMode mode) {
+ checkNotNull(mode, "aggregation mode");
+ return new TableUpsert(tablePath, tableInfo, writerClient, this.targetColumns, mode);
+ }
+
@Override
public UpsertWriter createWriter() {
- return new UpsertWriterImpl(tablePath, tableInfo, targetColumns, writerClient);
+ return new UpsertWriterImpl(tablePath, tableInfo, targetColumns, writerClient, aggMode);
Review Comment:
We should add check that if `aggMode` is `OVERWRITE` or `LOCAL_AGGREGATE`, the table should be aggregation merge engine.
##########
fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java:
##########
@@ -120,6 +121,15 @@ public abstract boolean tryAppend(WriteRecord writeRecord, WriteCallback callbac
public abstract void abortRecordAppends();
+ /**
+ * Get the aggregation mode for this batch (only applicable to KV batches).
+ *
+ * @return the aggregation mode, defaults to AGGREGATE for log batches
+ */
+ public AggMode getAggMode() {
Review Comment:
It's very confusing that `ArrowLogWriteBatch` has a default `AggMode#AGGREGATE`. I checked the code, it seems that providing this method only in `KvWriteBatch` can work.
##########
fluss-common/src/main/java/org/apache/fluss/rpc/protocol/AggMode.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.rpc.protocol;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * Aggregation mode for write operations.
+ *
+ * <p>This enum controls how the server handles data aggregation when writing to tables with
+ * aggregation merge engine.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public enum AggMode {
Review Comment:
I’ve reconsidered the enum API design, and it’s indeed confusing to have an `aggMode` flag on every `PutRequest`, especially since it implies aggregation-specific semantics while can being used across different merge engines. A more general and consistent approach would be to introduce a **`MergeMode`** that applies uniformly to all merge engines.
Here’s the proposed refinement:
1. **Rename `AggMode#AGGREGATE` to `MergeMode#DEFAULT`**
This represents the default merge behavior for each engine:
- For the *aggregate* merge engine: perform aggregation.
- For the *first-row* merge engine: retain the first observed row.
- For the *versioned* merge engine: keep the row with the highest version.
2. **Rename `AggMode#OVERWRITE` to `MergeMode#OVERWRITE`**
This mode bypasses all merge logic and directly replaces existing values with new ones—similar to an `OVERWRITE INTO` statement. It’s useful for scenarios like undo/recovery across all merge engines.
3. **Remove `AggMode#LOCAL_AGGREGATE`**
This is purely a client-side optimization that performs pre-aggregation before sending data. It actually doesn't change the protocol between client and server. The client still send the result using `MergeMode#DEFAULT`, as the on-wire schema remains unchanged.
This change simplifies the API, improves consistency across merge engines, and clarifies the intent of each mode.
If you agree on the new API, we can rename the enum type in this PR, and postpone the general support MergeMode for all merge eninges in next PR. But if you decide to support it in this PR, please remember to add tests.
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TableUpsert.java:
##########
Review Comment:
We should add check if the `aggMode` is not the default value, we should throw exception, because the `TypedUpsertWriterImpl` doesn't support the `aggMode` yet.
--
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
Re: [PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #2492:
URL: https://github.com/apache/fluss/pull/2492#issuecomment-3821355849
> Thanks @platinumhamburg , the code looks in a high quality. I left some comments below.
>
> Besides, I noticed that there is no any (agg mode or overwrite mode) IT case from client side? Is it planned in next PRs?
Yes, in the next PR.
--
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
Re: [PR] [FLUSS] Add AggMode support for undo recovery in aggregation tables [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #2492:
URL: https://github.com/apache/fluss/pull/2492#issuecomment-3822574326
> @platinumhamburg LGTM. Could you rebase and resolve the conflicts?
@wuchong Yes, I've rebased this PR and resolved all conflicts.
--
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