apache / apache/hudi

[SUPPORT] How to implement incremental join

Open
#8,260 20 comments 0 reactions 0 assignees View on GitHub
priority:medium type:feature
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

- Problem
Now I want to do a poc on hudi with flinksql. That is I have table A and table B, and I join table A and table B to get a new table C. Now when I make changes to base table(table A or table B), how to make table C auto update?

- Example
1 create table hudi_patient
```sql
create table hudi_patient(
id bigint,
name String,
PRIMARY KEY (`id`) NOT ENFORCED
)
with(
'connector'='hudi',
'path'='/Users/gamblewin/hudi/hudi_patient',
'hoodie.datasource.write.recordkey.field'='id',
'hoodie.parquet.max.file.size'='268435456',
'hoodie.datasource.write.recordkey.field'='id',
'changelog.enabled'='true',
'table.type'='COPY_ON_WRITE'
);
```
2 insert data into hudi_patient
```sql
insert into hudi_patient(id, name) VALUES(1, 'otis');
```

3 create table hudi_disease
```sql
create table hudi_disease(
id bigint,
disease_name String,
PRIMARY KEY (`id`) NOT ENFORCED
)
with(
'connector'='hudi',
'path'='/Users/gamblewin/hudi/hudi_disease',
'hoodie.datasource.write.recordkey.field'='id',
'hoodie.parquet.max.file.size'='268435456',
'hoodie.datasource.write.recordkey.field'='id',

'table.type'='COPY_ON_WRITE'
);
```

4 insert data into hudi_disease
```sql
insert into hudi_disease(id, disease_name) VALUES (1, 'headache');
```

5 create a join table hudi_pat_disease
```sql
create table hudi_pat_disease(
id BIGINT,
name STRING,
disease_name STRING,
PRIMARY KEY(id) NOT ENFORCED
) WITH (
'connector'='hudi',
'path'='/Users/gamblewin/hudi/hudi_pat_diesease',
'hoodie.datasource.write.recordkey.field'='id',
'hoodie.parquet.max.file.size'='268435456',
'hoodie.datasource.write.recordkey.field'='id',
'changelog.enabled'='true',
'table.type'='COPY_ON_WRITE'
)
```

6 insert join data into hudi_pat_disease
```sql
insert into hudi_pat_disease
select
hudi_patient.id as id,
hudi_patient.name as name,
hudi_disease.disease_name as disease_name
from
hudi_patient
left join
hudi_disease
on
hudi_patient.id = hudi_disease.id;
```

7 now I can get data `1 otis headache` from `hudi_pat_disease`,
but when I `insert into hudi_patient(id, name) VALUES (2, 'gamblewin`),
I expect a new record `2, gambelwin, NULL` from table `hudi_pat_disease`,
but `hudi_pat_disease` will not automatically update this record based on changes of `hudi_patient`,
so how does hudi implement this auto incremental functionality.

Contributor guide

No contributing guide indexed for this repository

Research direction

Start by reproducing the supplied Flink SQL setup for hudi_patient, hudi_disease, and hudi_pat_disease, then inspect the Hudi incremental-processing behavior relevant to joined tables. Done would mean establishing whether changes to either source can automatically produce the expected update in the joined table, with the required configuration or documented limitation clearly identified.

Written by the indexing model from the issue text.

Assessment

Tech stack
sql
Domain
data-engineering, databases, stream-processing
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
20/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.