[SUPPORT] How to implement incremental join
- 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