Skip to content

Commit 151a49f

Browse files
xuyangzhongwuchong
xuyangzhong
authored and
wuchong
committed
[doc] Add docs for delta join support with Flink 2.1 (#1875)
(cherry picked from commit aa4afe8)
1 parent 46e4fab commit 151a49f

File tree

4 files changed

+188
-2
lines changed

4 files changed

+188
-2
lines changed

‎website/docs/assets/delta_join.jpg‎

83.4 KB
Loading

‎website/docs/engine-flink/datastream.mdx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
---
22
title: "DataStream API"
3-
sidebar_position: 6
3+
sidebar_position: 7
44
---
55

66
# DataStream API

‎website/docs/engine-flink/delta-joins.md‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,186 @@
1+
---
2+
sidebar_label: Delta Joins
3+
title: Flink Delta Joins
4+
sidebar_position: 6
5+
---
6+
7+
# The Delta Join
8+
Beginning with **Apache Flink 2.1**, a new operator called [Delta Join](https://cwiki.apache.org/confluence/display/FLINK/FLIP-486%3A+Introduce+A+New+DeltaJoin) was introduced.
9+
Compared to traditional streaming joins, the delta join operator significantly reduces the amount of state that needs to be maintained during execution. This improvement helps mitigate several common issues associated with large state sizes, including:
10+
11+
- Excessive computing resource and storage consumption
12+
- Long checkpointing durations
13+
- Extended recovery times after failures or restarts
14+
- Heavy state bootstrap overhead after changing job logic
15+
16+
Starting with **Apache Fluss 0.8**, streaming join jobs running on **Flink 2.1 or later** will be automatically optimized into **delta joins** whenever applicable. This optimization happens transparently at query planning time, requiring no manual configuration.
17+
18+
## How Delta Join Works
19+
20+
Traditional streaming joins in Flink require maintaining both input sides entirely in state to match records across streams. Delta join, by contrast, uses a **index-key lookup mechanism** to transform the behavior of querying data from the state into querying data from the Fluss source table, thereby avoiding redundant storage of the same data in both the Fluss source table and the state. This drastically reduces state size and improves performance for many streaming analytics and enrichment workloads.
21+
22+
![](../assets/delta_join.jpg)
23+
24+
## Example: Delta Join in Flink 2.1
25+
26+
Below is an example demonstrating a delta join query supported by Flink 2.1.
27+
28+
#### Create Source and Sink Tables
29+
30+
```sql title="Flink SQL"
31+
USE CATALOG fluss_catalog;
32+
```
33+
34+
```sql title="Flink SQL"
35+
CREATE DATABASE my_db;
36+
```
37+
38+
```sql title="Flink SQL"
39+
USE my_db;
40+
```
41+
42+
#### Create Left Source Table
43+
```sql title="Flink SQL"
44+
CREATE TABLE `fluss_catalog`.`my_db`.`left_src` (
45+
`city_id` INT NOT NULL,
46+
`order_id` INT NOT NULL,
47+
`content` VARCHAR NOT NULL,
48+
PRIMARY KEY (city_id, order_id) NOT ENFORCED
49+
) WITH (
50+
-- prefix key
51+
'bucket.key' = 'city_id',
52+
-- in Flink 2.1, delta join only support append-only source
53+
'table.merge-engine' = 'first_row'
54+
);
55+
```
56+
57+
#### Create Right Source Table
58+
```sql title="Flink SQL"
59+
CREATE TABLE `fluss_catalog`.`my_db`.`right_src` (
60+
`city_id` INT NOT NULL,
61+
`city_name` VARCHAR NOT NULL,
62+
PRIMARY KEY (city_id) NOT ENFORCED
63+
) WITH (
64+
-- in Flink 2.1, delta join only support append-only source
65+
'table.merge-engine' = 'first_row'
66+
);
67+
```
68+
69+
#### Create Sink Table
70+
```sql title="Flink SQL"
71+
CREATE TABLE `fluss_catalog`.`my_db`.`snk` (
72+
`city_id` INT NOT NULL,
73+
`order_id` INT NOT NULL,
74+
`content` VARCHAR NOT NULL,
75+
`city_name` VARCHAR NOT NULL,
76+
PRIMARY KEY (city_id, order_id) NOT ENFORCED
77+
);
78+
```
79+
80+
#### Explain the Join Query
81+
```sql title="Flink SQL"
82+
EXPLAIN
83+
INSERT INTO `fluss_catalog`.`my_db`.`snk`
84+
SELECT T1.`city_id`, T1.`order_id`, T1.`content`, T2.`city_name`
85+
FROM `fluss_catalog`.`my_db`.`left_src` T1
86+
Join `fluss_catalog`.`my_db`.`right_src` T2
87+
ON T1.`city_id` = T2.`city_id`;
88+
```
89+
90+
If the printed plan includes `DeltaJoin` as shown below, it indicates that the optimizer has successfully transformed the traditional streaming join into a delta join.
91+
92+
```title="Flink Plan"
93+
== Abstract Syntax Tree ==
94+
LogicalSink(table=[fluss_catalog.my_db.snk], fields=[city_id, order_id, content, city_name])
95+
+- LogicalProject(city_id=[$0], order_id=[$1], content=[$2], city_name=[$4])
96+
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
97+
:- LogicalTableScan(table=[[fluss_catalog, my_db, left_src]])
98+
+- LogicalTableScan(table=[[fluss_catalog, my_db, right_src]])
99+
100+
== Optimized Physical Plan ==
101+
Sink(table=[fluss_catalog.my_db.snk], fields=[city_id, order_id, content, city_name])
102+
+- Calc(select=[city_id, order_id, content, city_name])
103+
+- DeltaJoin(joinType=[InnerJoin], where=[=(city_id, city_id0)], select=[city_id, order_id, content, city_id0, city_name])
104+
:- Exchange(distribution=[hash[city_id]])
105+
: +- TableSourceScan(table=[[fluss_catalog, my_db, left_src]], fields=[city_id, order_id, content])
106+
+- Exchange(distribution=[hash[city_id]])
107+
+- TableSourceScan(table=[[fluss_catalog, my_db, right_src]], fields=[city_id, city_name])
108+
109+
== Optimized Execution Plan ==
110+
Sink(table=[fluss_catalog.my_db.snk], fields=[city_id, order_id, content, city_name])
111+
+- Calc(select=[city_id, order_id, content, city_name])
112+
+- DeltaJoin(joinType=[InnerJoin], where=[(city_id = city_id0)], select=[city_id, order_id, content, city_id0, city_name])
113+
:- Exchange(distribution=[hash[city_id]])
114+
: +- TableSourceScan(table=[[fluss_catalog, my_db, left_src]], fields=[city_id, order_id, content])
115+
+- Exchange(distribution=[hash[city_id]])
116+
+- TableSourceScan(table=[[fluss_catalog, my_db, right_src]], fields=[city_id, city_name])
117+
```
118+
119+
## Understanding Index Keys
120+
121+
Delta Join relies on performing lookups by the join key (e.g., the `city_id` in the above example) on the Fluss source tables. This requires that the Fluss source tables are defined with appropriate index on the join key to enable efficient lookups.
122+
123+
Currently, Fluss only supports to defines a single index key per table, which is also referred to as the **prefix key**. The general secondary index which allows define multiple indexes with arbitrary fields is planned to be supported in the near future releases.
124+
125+
A prefix key defines the portion of a table’s primary key that can be used for efficient key-based lookups or index pruning.
126+
127+
In Fluss, the option `'bucket.key' = 'city_id'` specifies that data is organized (or bucketed) by `city_id`. When performing a delta join, this allows Flink to quickly locate and read only the subset of records corresponding to the specific prefix key value, rather than scanning or caching the entire table state.
128+
129+
For example:
130+
- Full primary key: `(city_id, order_id)`
131+
- Bucket key: `city_id`
132+
133+
This yields an **index** on the prefix key `city_id`, so that you can perform [Prefix Key Lookup](/docs/engine-flink/lookups/#prefix-lookup) by the `city_id`.
134+
135+
In this setup:
136+
* The delta join operator uses the prefix key (`city_id`) to retrieve only relevant right-side records matching each left-side event.
137+
* This eliminates the need to hold all records for every city in Flink state, significantly reducing state size.
138+
139+
Prefix keys thus form the foundation for efficient lookups in delta joins, enabling Flink to scale join workloads efficiently even under high throughput.
140+
141+
## Flink Version Support
142+
143+
The delta join feature is introduced since Flink 2.1 and still evolving, and its optimization capabilities vary across Flink versions.
144+
145+
Refer to the [Delta Join Issue](https://issues.apache.org/jira/browse/FLINK-37836) for the most up-to-date information.
146+
147+
148+
### Flink 2.1
149+
150+
#### Supported Features
151+
152+
- Support for optimizing a dual-stream join in the straightforward pattern of `insert only source -> join -> upsert sink` into a delta join.
153+
154+
#### Limitations
155+
156+
- The primary key or the prefix key of the tables must be included as part of the equivalence conditions in the join.
157+
- The join must be a INNER join.
158+
- The downstream nodes of the join can accept duplicate changes, such as a sink that provides UPSERT mode without `upsertMaterialize`.
159+
- All join inputs should be INSERT-ONLY streams.
160+
- This is why the option `'table.merge-engine' = 'first_row'` is added to the source table DDL.
161+
- All upstream nodes of the join should be `TableSourceScan` or `Exchange`.
162+
163+
### Flink 2.2 (upcoming)
164+
165+
#### Supported Features
166+
167+
- Support for optimizing a dual-stream join from CDC sources that do not include delete messages into a delta join.
168+
- Disable delete on the source table to guarantee there is no delete message in the table, by adding the option `'table.delete.behavior' = 'IGNORE'` or `'DISABLE'` on the table.
169+
- The source table is no more required to be a `first_row` merge engine table since this version.
170+
- Support `Project` and `Filter` between source and delta join.
171+
- Support cache in delta join.
172+
173+
#### Limitations
174+
175+
- The primary key or the prefix lookup key of the tables must be included as part of the equivalence conditions in the join.
176+
- The join must be a INNER join.
177+
- The downstream nodes of the join can accept duplicate changes, such as a sink that provides UPSERT mode.
178+
- When consuming a CDC stream, the join key used in the delta join must be part of the primary key.
179+
- All filters must be applied on the upsert key, and neither filters nor projections should contain non-deterministic functions.
180+
181+
## Future Plan
182+
183+
The Fluss and Flink community are actively working together on enhancing delta join. Planned improvements include:
184+
- Support for additional join types, such as LEFT JOIN and FULL OUTER JOIN.
185+
- Support for multi-way joins involving more than two streams. This also requires Fluss to support defining multiple secondary index keys on a single table.
186+
- Support for more complex query patterns to optimize a wider range of join scenarios.

‎website/docs/engine-flink/options.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
---
22
title: Connector Options
3-
sidebar_position: 7
3+
sidebar_position: 8
44
---
55

66
# Connector Options

0 commit comments

Comments
 (0)