apache / apache/fluss

[Feature] Introduce Merge Engines for Primary-Key Table

Open
#212 6 comments 2 reactions 0 assignees View on GitHub
component=kv
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.