[Feature] Introduce Merge Engines for Primary-Key Table
- 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](https://github.com/alibaba/fluss/issues) and found nothing similar.
### Motivation
We propose the introduction of merge engines for primary-key tables, enhancing data management and query capabilities. This feature will encompass the following merge engines, each with distinct functionalities:
- Version-Based Merge: This merge engine will allow data consolidation based on version numbers. It is very similar to ClickHouse [ReplacingMergeTree](https://clickhouse.com/docs/en/guides/replacing-merge-tree) that keeps the row with the highest version number. This is also very similar to HBase [checkAndPut](https://hbase.apache.org/1.2/apidocs/org/apache/hadoop/hbase/client/Table.html#checkAndPut(byte[],%20byte[],%20byte[],%20byte[],%20org.apache.hadoop.hbase.client.Put)) that updates records only when the condition is true. In this way, the out-of-order data can be guaranteed to be ultimately consistent with the upstream.
- First-Row Based Merge: Designed to retain the first occurrence of a row with the same primary key, this engine boosts performance by minimizing writes to the kv store and generates only insert-only changelogs (some downstream jobs only accept insert-only changelogs).
- Aggregate Based Merge: This engine will facilitate the merging of data by aggregating records, which is particularly useful for summarizing information and analytics purposes. It will support operations like `sum`, `max`, `min`, `count`, `avg`, among others, providing robust analytic capabilities.
### Solution
Introduce table properties (configured via Flink DDL with options):
```properties
'table.merge-engine' = 'first_row|version|aggregate'
'table.merge-engine.version.column' = ''
'table.merge-engine.aggregate.' = 'sum | max | min | count | avg | first_value | last_value'
```
The `merge-engine` properties are table storage properties, and therefore can't be changed after table creation.
## How to Use
### Version-Based Merge
```sql
CREATE TABLE fluss_table (
id BIGINT,
ts BIGINT,
data STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'fluss',
'bootstrap.address' = '...',
'table.merge-engine' = 'version',
'table.merge-engine.version.column' = 'ts' -- updates when new ts >= old ts
);
```
### First-Row-Based Merge
```sql
CREATE TABLE fluss_table (
id BIGINT,
version BIGINT,
data STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'fluss',
'bootstrap.address' = '...',
'table.merge-engine' = 'first_row' -- only the first row of the primary key will be retained.
);
```
### Aggregate-Based Merge
```sql
CREATE TABLE fluss_table (
id BIGINT,
price DOUBLE,
sales BIGINT,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'fluss',
'bootstrap.address' = '...',
'table.merge-engine' = 'aggregate',
'table.merge-engine.aggregate.price' = 'max',
'table.merge-engine.aggregate.sales' = 'sum'
);
```
### Anything else?
Subtasks:
- [x] #133
- [x] #213
- [ ] #2133
### Willingness to contribute
- [ ] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
The issue describes merge-engine table properties configured through Flink DDL, with version, first-row, and aggregate examples, but names no source files, tests, or implementation entry points. Start by locating the primary-key table storage and table-property handling; done would require the proposed engines, configuration validation, and coverage for the documented behaviors.
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
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 20/100