Support schema on write for flink
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
### Feature Description
**What the feature achieves:**
Support out-of-box schema on write for flink.
**Why this feature is needed:**
Currently, spark writers support oob schema on write: https://hudi.apache.org/docs/schema_evolution/#schema-evolution-in-action, which allows user writing records to hudi table with backwards-compatible schema evolution scenarios, such as adding a null field or promoting field type. But flink writer don't have full support for schema on write, we should supplement this feature for flink writer.
E.g., the following writing to COW table with schema promoting is not supported.
```java
// writer with schema1
DataStream dataStream = execEnv.fromData(
createRowData("id1", "Alice", 25, Timestamp.from(Instant.now()), "par1"),
createRowData("id2", "Lily", 21, Timestamp.from(Instant.now()), "par1"),
createRowData("id3", "Julia", 15, Timestamp.from(Instant.now()), "par1")
);
HoodiePipeline.Builder builder = HoodiePipeline.builder("test_sink")
.column("uuid string not null")
.column("name string")
.column("age int")
.column("`ts` timestamp(3)")
.column("`partition` string")
.pk("uuid")
.partition("partition")
.options(options);
builder.sink(dataStream, false);
execute(execEnv, false, "Api_Sink_Test");
// writer with schema2, promoting `age` from INT to DOUBLE
dataStream = execEnv.fromData(
createRowData("id1", "Alice", 25.1, Timestamp.from(Instant.now().plusMillis(1000)), "par1"),
createRowData("id2", "Lily", 21.2, Timestamp.from(Instant.now().plusMillis(1000)), "par1"),
createRowData("id3", "Julia", 15.3, Timestamp.from(Instant.now().plusMillis(100)), "par1")
);
builder = HoodiePipeline.builder("test_sink")
.column("uuid string not null")
.column("name string")
.column("age double")
.column("`ts` timestamp(3)")
.column("`partition` string")
.pk("uuid")
.partition("partition")
.options(options);
builder.sink(dataStream, false);
execute(execEnv, false, "Api_Sink_Test_1");
```
### User Experience
**How users will use this feature:**
- Configuration changes needed
- API changes
- Usage examples
### Hudi RFC Requirements
**RFC PR link:** (if applicable)
**Why RFC is/isn't needed:**
- Does this change public interfaces/APIs? (Yes/No)
- Does this change storage format? (Yes/No)
- Justification:
Contributor guide
No contributing guide indexed for this repository
Research direction
Review open pull request #14248 first, then compare the Flink writer behavior with the Spark schema-on-write example linked in the issue. Use the HoodiePipeline.Builder example and its execute calls as the entry point; done means the two shown schemas, including INT-to-DOUBLE promotion, work for the Flink COW table.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, stream-processing
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 25/100