[server] Support sequence groups in default and agg merge engine - #4130
[server] Support sequence groups in default and agg merge engine#4130Kaixuan-Duan wants to merge 11 commits into
Conversation
Kaixuan-Duan
changed the title
The first-row fast paths used to return the incoming row without consulting the sequence groups, so a first record whose ordering values are all NULL initialized every protected column — while the partial update path already skipped the group, and a later equally sequence-less record was skipped as well. That left "first sequence-less value wins" and made equivalent full-row and partial writes disagree. All three merge paths (Default full row, full Aggregation, partial Aggregation) now run the arbitration with a nullable old row before taking the fast path: a group with a non-NULL ordering value is FORWARD, an all-NULL group is SKIP, and fields outside any group keep taking their incoming values. The fast path survives when every field is accepted; otherwise a skipped group encodes NULL while there is no stored row. This matches Paimon, which applies the empty-group check to the first record too.
An auto increment column is never written by a client, so a group containing one can never be updated: as an ordering column it is always NULL in an incoming row and the group is always skipped, while as a protected column it can never join the write target that the group requires to be covered entirely. Also reject an empty sequence or protected column list, which leaves a group permanently on SKIP, as a backup for the Builder checks so that every entry point into a Schema enforces the same invariants.
…n path Every KV merge used to create about a dozen short-lived arrays: one Comparable[] per group for the incoming sequence values, with INT and BIGINT boxed through Comparable<?>, a group decision array, a field-sized expansion of it, and, for partial writes, arbitration of groups the target set never touches. Sequence columns are now compared in place column by column through type-specific comparators built once per schema, so no sequence value is stored or boxed; the group decisions live in a buffer reused across records, safe under KvTablet's single-threaded write lock like the row encoders; callers query accepts(i) / decisionOf(i) per field instead of receiving the field-level expansion; and restrictTo drops the sequence columns of uncovered groups, so a partial write only evaluates the groups it references.
…anges nothing The processor recognizes a no-op by object identity, but the mergers re-encoded an unchanged row into a fresh BinaryValue when arbitration rejected the write, so a rejected record still produced a changelog event, consumed its record offsets, and rewrote the state. All three partial paths now check rejectsEveryTargetField and return the stored value itself: for the default engine SKIP and STALE alike reject, while an aggregating engine folds a stale record in through aggReversed, so there only SKIP does. The shortcut requires the stored row to be encoded with the target schema already, since returning it keeps that schema; a stored row on an older schema is upgraded through the merge as before.
The complete sequence-group description was duplicated across the Default and Aggregation merge-engine pages, hiding that its main role is strengthening Partial Update with independent ordering for columns written by multiple streams. The common documentation now lives in Primary Key Table > Partial Update, right after the partial-update example, and each merge-engine page keeps only its engine-specific semantics plus a link. The aggregation example now protects a first_value field, where an out-of-order record would actually change the result, instead of sum, whose result is the same whatever order the records arrive in.
There was a problem hiding this comment.
@Kaixuan-Duan Thanks for the contribution and the work on the review feedback. This PR looks like a substantial step toward aligning Fluss's sequence-group behavior with Paimon. In particular, sequence groups in the default merge engine provide an important capability for existing Paimon users moving to lake-stream tables. The implementation looks well structured, and I have left a few smaller engineering suggestions.
For the aggregate merge engine, the implementation also appears consistent with the Paimon behavior we have been targeting. Given my earlier request to align with Paimon, I think this is a reasonable implementation of that direction and helps preserve consistent behavior between Fluss and the lake table.
My remaining hesitation concerns the limitations inherited from Paimon itself. Paimon #3393 reports incorrect first_value results after compaction with aggregation and sequence.field, while the unmerged #3101 attempted to address related first/first-non-null/last-non-null correctness problems. Although these are not identical to this PR's sequence-group path, they highlight the same underlying issue: the sequence retained for an aggregated row may no longer identify the record that supplied a selected field value.
I do not have a clear picture of how widely Paimon users rely on aggregation functions within sequence groups. That leaves us with three difficult choices:
- Defer sequence-group support in the aggregate merge engine, while proceeding with support in the default merge engine. This avoids introducing the known aggregation limitations, but leaves a compatibility gap for users who depend on that combination.
- Accept the implementation in this PR and preserve compatibility with Paimon, including its current limitations. This minimizes behavior changes for existing users migrating to lake-stream tables, but we should document the affected cases and avoid promising correctness for arbitrary out-of-order inputs.
- Develop a correct implementation in Fluss and work with the Paimon community to bring the corresponding fix upstream. This addresses the underlying problem, but requires additional design and careful coordination around behavior, persisted state, and compatibility between Fluss and Paimon.
I see this primarily as a scope and compatibility decision rather than an implementation issue specific to this contribution, @wuchong what do you think?
| rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(source)); | ||
| } | ||
| } | ||
| return new BinaryValue(targetSchemaId, rowEncoder.finishRow()); |
There was a problem hiding this comment.
The earlier no-op feedback is now covered in PartialUpdater and both aggregation paths: they return oldValue itself when the write contributes nothing. The full-row Default path appears to be the remaining case.
For example, with ts protecting status, a stored row (1, PAID, 100) and an incoming row (1, CREATED, 99) should keep the stored row. This path does keep the values, but re-encodes them into a different BinaryValue. KvWriteProcessor then misses its newValue == oldValue check and writes an unchanged row to state and the changelog, consuming one log offset in WAL mode or two with before/after images.
Could we extend the same early return to this path, before encoding, with the existing-row and matching-schema guards used by the other paths? For Default, both SKIP and STALE mean no contribution; for Agg, only SKIP does. Any ungrouped non-key field must still prevent this decision-only shortcut.
| arbitrateSequenceGroups | ||
| ? partialUpdaterCache.getOrCreatePartialUpdater( | ||
| kvFormat, latestShemaId, latestSchema, targetColumns) | ||
| : new PartialUpdater( |
There was a problem hiding this comment.
Bypassing sequence-group arbitration in OVERWRITE mode is necessary for undo recovery: restoring a checkpointed value must not reject it because its sequence is older. While following that change, I noticed that partial OVERWRITE writes now bypass PartialUpdaterCache.
KvWriteProcessor configures the merger for each batch. When the targets cover only part of the schema, this branch therefore creates a fresh PartialUpdater, field-type array, field getters, and encoder each time. Previously this path used the cache, and the new allocation also applies to tables without sequence groups. UndoRecoveryManager reaches it through OVERWRITE plus partialUpdate(targetColumnIndexes).
Could we keep updater reuse for this path while retaining the explicit bypass of sequence arbitration? A cache configured for OVERWRITE mode, or an arbitration mode in the cache key, would both work. Simply routing this branch through the existing default cache factory would re-enable sequence-group arbitration and could reject the older value being restored.
| return this; | ||
| } | ||
|
|
||
| TargetColumns.checkSequenceGroupsAreFullyTargeted(latestSchema, targetColumns); |
There was a problem hiding this comment.
The new group-completeness check covers the split-group case raised earlier. One follow-up on where it runs: this call is before the partial-merger cache lookup, so every batch repeats the validation even when the same schema and target columns already have a cached merger. DefaultRowMerger does the same.
For a schema with sequence groups, each invocation rebuilds the target-name set, every group's field set, and the missing-field sets. A successful validation does not depend on the batch contents; it remains valid for that schema ID and target-column combination.
Could we validate when creating the cached partial merger/updater, so cache hits reuse the validated configuration? All creation paths should retain the check, including after eviction or a schema/target change. Missing-field collections could be built only when needed to report an invalid target set.
This is a per-batch allocation concern, particularly for small batches.
| RowEncoder encoder) { | ||
| // the groups are resolved against the target schema, which is the one being encoded | ||
| if (sequenceGroups != null) { | ||
| sequenceGroups.arbitrate(oldRow, newRow); |
There was a problem hiding this comment.
The reusable decision buffer addresses the allocation concern from the earlier review. There is one remaining duplication between the new no-op check and field aggregation: AggregateRowMerger arbitrates the rows to decide whether it can return oldValue, and this method arbitrates them again when the write needs to continue.
For an existing row in the target schema, any write that contributes a target field takes this two-pass path, in both full and partial aggregation. A first write that cannot return directly also repeats arbitration. The decisions from the first pass are still available in groupDecisions, so the second pass repeats the same sequence reads and comparisons.
Could we have one place perform arbitration and let both the early-return checks and the field processor consume those decisions? The different-schema path still needs its first arbitration, so removing this call alone would not cover all cases.
A smaller remaining point from the earlier partial-write feedback: restrictTo avoids reading sequence values for inactive groups, but arbitrate still visits their empty entries and writes SKIP. Resolving the active group IDs when the partial merger is created would avoid those visits as well. This should preserve the existing projection of non-target fields; it should not make returning newValue unconditional.
2 participants
Purpose
Linked issue: close #4129
Brief change log
Tests
API and Format
Documentation