Repository navigation
[server] Add retract support for Aggregation Merge Engine - #2917
platinumhamburg wants to merge 2 commits into
Conversation
26cb205 to
d1478b5
Compare
|
It is a great work |
43d1e5c to
e7fa946
Compare
|
hello,is there any progress ? |
| } | ||
|
|
||
| if (isAggregationTable | ||
| && !schemaSupportsRetract |
There was a problem hiding this comment.
This rejects any update changelog (including UPDATE_AFTER-only upsert streams) for aggregation tables that have a non-retractable column, even with sink.ignore-delete=true. Previously these jobs worked because UB was stripped, so e.g. INSERT INTO agg_tbl SELECT ... FROM <pk table> with a max column will now fail to plan after upgrade. Upsert-only input is also safe for max/min/first_value/bool_and/bool_or, since re-applying a value doesn't change their result. Should we only reject when a retract would actually be required, or put this behind an opt-in retract option?
| // 1. writerSupportsRetract=true: retract-then-aggregate semantics | ||
| // 2. sink.ignore-delete=true on non-aggregation tables: serializer drops UPDATE_BEFORE | ||
| // and DELETE as IGNORE | ||
| if (schemaSupportsRetract || sinkIgnoreDelete) { |
There was a problem hiding this comment.
sinkIgnoreDelete now returns requestedMode (including UPDATE_BEFORE) for all tables, including plain PK and log tables. The serializer drops UB anyway when ignore-delete is set, so this only forces the planner to generate UB upstream. That's extra records, possibly extra stateful operators, and a plan change for existing jobs unrelated to aggregation. I think this branch should only apply to schemaSupportsRetract.
| return schemaSupportsRetract; | ||
| } | ||
|
|
||
| for (String column : columns(targetColumnIndexes)) { |
There was a problem hiding this comment.
Is this target-column subset path reachable for streaming updates? getChangelogMode runs before target columns are known and checks the full schema, so INSERT INTO t(id, sum_col) with update input on a table that also has a max column is already rejected in getChangelogMode. The docs say "all target aggregation fields", which suggests otherwise. Should we drop the subset logic, or document that the full schema must support retract?
| public RowDataSerializationSchema(boolean isAppendOnly, boolean ignoreDelete) { | ||
| this.isAppendOnly = isAppendOnly; | ||
| this.ignoreDelete = ignoreDelete; | ||
| public RowDataSerializationSchema(SinkOperationMode mode) { |
There was a problem hiding this comment.
Replacing the (boolean isAppendOnly, boolean ignoreDelete) constructor breaks DataStream API. Should we keep the old constructor and have it delegate to SinkOperationMode?
| enableUndoRecovery, | ||
| producerId); | ||
| producerId, | ||
| false); |
There was a problem hiding this comment.
schemaSupportsRetract is hardcoded to false, but OperationType.RETRACT is now visible to custom FlussSerializationSchemas, so a DataStream user returning RETRACT always gets an exception.
This is just a note, so I'm aware and keep track of it for DataStream API.
| /** | ||
| * The result of retracting a record ({@link UpsertWriter#retract(InternalRow)}). | ||
| * | ||
| * @since 0.10 |
There was a problem hiding this comment.
nit: Update the @since 0.10
| * <p>This is a per-record field inside the KvRecord body, independent of the batch/request-level | ||
| * {@link org.apache.fluss.rpc.protocol.MergeMode}. | ||
| * | ||
| * @since 0.10 |
There was a problem hiding this comment.
nit: @since 0.10 → the next release.
| } | ||
|
|
||
| @Test | ||
| void testGetChangelogModeExcludesUpdateBeforeForNonRetractAggregationTable() { |
There was a problem hiding this comment.
nit: the name says "Excludes" but the test asserts it throws.
| } | ||
|
|
||
| @Override | ||
| public boolean supportsRetract() { |
There was a problem hiding this comment.
nit: redundant override, the base class already returns false.
| schemaId); | ||
|
|
||
| batch.tryAppend(writeRecord, callback); | ||
| WriteBatch.AppendResult appendResult2 = batch.tryAppend(writeRecord, callback); |
There was a problem hiding this comment.
nit: appendResult2 → e.g. freshBatchResult.
|
Hi @platinumhamburg, thank you so much for this work 🚀 This is something really useful and I'm happy to help get it merged. If you're short on time, I'd gladly help with anything that's useful to you: rebasing, splitting the PR, adding tests, or picking up parts of the work. Just let me know what you'd prefer, and I', happy to follow your lead on direction. I've left some inline comments. Another thought I wanted to share here: Behaviour change for existing aggregation tables: For tables where every column supports retract (including tables with no explicit agg function, which default to Thanks again for driving this, really looking forward to seeing it land! 🙏 |
- Reject retract() when the idempotent writer is disabled: retract is not idempotent and relies on server-side batch dedup under retries. - Create V2 format batches only for RETRACT records; Sender hands V2 batches to the rpc layer only when the negotiated PUT_KV version >= 2. - Server rejects V2 batches sent with PUT_KV api version < 2. - API cleanup: add RetractResult, move apiVersion validation from KvTablet to ReplicaManager, unify KvRecordBatchBuilder append API.
6c53e63 to
329fa7f
Compare
Purpose
Linked issue: close #2916
Brief change log
Tests
API and Format
Documentation