apache / apache/fluss

[server] Support sequence groups in default and agg merge engine

Open
#4,129 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
2.1k
Forks
625
Avg merge
3d 14h
Merged PRs (30d)
97

Description

## Search before asking

- [x] I searched in the issues and found nothing similar.

## What

A **sequence group** puts one or more columns under the order of a *sequence column*, so that those
columns only take an incoming value when the sequence column is not older than the stored one. Every
group is arbitrated on its own, so within a single write one group may move forward while another
does not.

It is declared with `fields..sequence-group`, whose value lists the columns it
protects, matching Apache Paimon's DDL:

```sql
CREATE TABLE orders (
order_id BIGINT,
pay_status STRING,
pay_time BIGINT,
ship_status STRING,
ship_time BIGINT,
PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'fields.pay_time.sequence-group' = 'pay_status',
'fields.ship_time.sequence-group' = 'ship_status'
);

INSERT INTO orders VALUES (1, 'paid', 100, 'shipped', 100);
-- pay_time moves forward while ship_time falls behind,
-- so only the payment columns take the incoming values
INSERT INTO orders VALUES (1, 'refunded', 200, 'lost', 99);
-- result: 1, refunded, 200, shipped, 100
```

## Why

As soon as more than one writer updates the same row, Fluss has no way to reject a write that is
older than what is stored.

- **The default merge engine keeps the last written row**, whether or not it is actually the newest.
A retried or delayed record silently overwrites newer values, and the only way to avoid it today is
to order the writes outside Fluss.
- **The versioned merge engine arbitrates the whole row with a single version column.** Two streams
updating disjoint columns cannot each carry their own version, so a write that is newer for its own
columns is rejected because of another stream's version.
- **In the aggregation merge engine the order-dependent functions follow the arrival order.**
`first_value` and `listagg` have no notion of which record came first in business time, so an
out-of-order record changes the result.

A sequence group addresses all three by giving each group of columns its own order, which is exactly
what Paimon does for its `partial-update` engine. Keeping the same DDL lets tables move between the
two systems unchanged.

## How

One arbitration, read differently by the two engines.

The groups declared on a schema are resolved once into field positions. For a given stored row and
incoming row, each group then yields one of three decisions:

| Decision | Condition |
| -------- | ---------------------------------- |
| SKIP | the group's sequence columns are all NULL in the incoming row |
| FORWARD | the incoming sequence is not older than the stored one |
| STALE | the incoming sequence is older than the stored one |

What a merger does with them is where the two engines differ:

| Decision | Default merge engine | Aggregation merge engine |
| -------- | ------------------------ | -------------------------------------------- |
| SKIP | keeps the stored value | contributes nothing at all |
| FORWARD | takes the incoming value | aggregates, and the sequence moves forward |
| STALE | keeps the stored value | still aggregates, but the sequence stays put |

Without aggregate functions a group is a **version filter**: whatever is older is dropped, so SKIP
and STALE behave alike. With them it is an **ordering key** instead: an older record is a fact that
still belongs in the total, so it is aggregated as one that happened earlier, and only the sequence
refuses to move backwards.

Naming several sequence columns declares a composite key, compared in the declared order until one
differs. Sequence groups apply to full-row and partial-column writes alike. A table declaring no
sequence group keeps its current merge path unchanged.

## Scope

Supported on the default and the aggregation merge engine. A sequence group on a log table, or with
the `first_row` or `versioned` merge engine, is rejected at table creation, since none of them
consults it while merging.

`DELETE` is not arbitrated: a Fluss delete record carries the primary key alone, so it holds no
sequence values for any group to compare against the stored row, and it keeps removing the whole row
as it does today. Arbitrating a delete would first need the delete record to carry the sequence
columns, which is a change to the write protocol and is better proposed separately. Paimon's
`partial-update.remove-record-on-delete` and `partial-update.remove-record-on-sequence-group`
therefore have no counterpart here.

## Anything else?

Paimon's sequence group documentation:
https://paimon.apache.org/docs/master/primary-key-table/merge-engine/partial-update/#sequence-group

## Willingness to contribute

- [x] I'm willing to submit a PR!

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by tracing schema handling and the default and aggregation merge-engine entry points described in the issue, then compare how full-row and partial-column writes reach them. Done means sequence groups arbitrate each group as specified, unsupported engines and log tables are rejected at creation, and tables without groups retain their current behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
databases
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.