feat(schema): Migrate hudi-flink to use HoodieSchema instead of avro Schema #14355

rahil-c · 2025-11-25T23:00:02Z

Describe the issue this Pull Request addresses

This PR aims to change usages of avro'sSchema within the hudi-flink-datasource to use HoodieSchema instead.

Summary and Changelog

Change avro'sSchema within the hudi-flink-datasource to use HoodieSchema.

Impact

medium as we are changing apis to now take in HoodieSchema in this code path, we have CI as a sanity check to ensure functionality is preserved.

Risk Level

Medium

Documentation Update

None

Contributor's checklist

Read through contributor's guide
Enough context is provided in the sections above
Adequate tests were added if applicable

the-other-tim-brown · 2025-11-26T02:53:02Z

...ce/hudi-flink/src/test/java/org/apache/hudi/sink/utils/BucketStreamWriteFunctionWrapper.java

    this.gateway = new MockOperatorEventGateway();
    this.conf = conf;
-    this.rowType = (RowType) AvroSchemaConverter.convertToDataType(StreamerUtil.getSourceSchema(conf)).getLogicalType();
+    this.rowType = (RowType) AvroSchemaConverter.convertToDataType(StreamerUtil.getSourceSchema(conf).getAvroSchema()).getLogicalType();


How much work would be involved in adding a schema converter to convert from HoodieSchema to the RowType?

Will look into this

the-other-tim-brown · 2025-11-26T03:01:35Z

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestSchemaEvolution.java

-
-      writeClient.addColumn("salary", doubleType, null, "name", AFTER);
+      // Create nullable primitive types using HoodieSchema
+      HoodieSchema intType = HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.INT));


There is a helper method that allows you to just call HoodieSchema.createNullable(HoodieSchemaType.INT) for easy test setup

the-other-tim-brown · 2025-11-26T03:02:59Z

...asource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieFileGroupReaderOnFlink.java

-  private static Schema getRecordAvroSchema(String schemaStr) {
-    Schema recordSchema = new Schema.Parser().parse(schemaStr);
-    return AvroSchemaConverter.convertToSchema(RowDataAvroQueryContexts.fromAvroSchema(recordSchema).getRowType().getLogicalType());
+  private static HoodieSchema getRecordAvroSchema(String schemaStr) {


Let's rename this to getRecordSchema?

the-other-tim-brown · 2025-11-26T03:03:52Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java


  @VisibleForTesting
-  public Schema getTableAvroSchema() {
+  public HoodieSchema getTableHoodieSchema() {


Can we just call this getTableSchema?

the-other-tim-brown · 2025-11-26T03:05:03Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java

+      Schema avroSchema = schemaResolver.getTableAvroSchema();
+      return HoodieSchema.fromAvroSchema(avroSchema);


I think we'll have a few PRs all doing this same operation. What do you think about just pushing this down to the TableSchemaResolver so we don't have to modify as much code later?

@the-other-tim-brown Wondering if you can help me better understand the ask, is it just for schemaResolver to have some helper called getTableSchema which basically just delegates to getTableAvroSchema but then wraps it around the HoodieSchema.fromAvroSchema?

Based on offline discussion with @the-other-tim-brown my understanding is that we will raise this as part of a seperate pr

@the-other-tim-brown let me know if this pr:#17456 this aligns with what you said above.

Lets update this to call the new method in the table schema resolver

the-other-tim-brown · 2025-11-26T03:05:27Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java

+  private HoodieSchema inferSchemaFromDdl() {
+    Schema avroSchema = AvroSchemaConverter.convertToSchema(this.tableRowType);
+    HoodieSchema schema = HoodieSchema.fromAvroSchema(avroSchema);
+    return HoodieSchemaUtils.addMetadataFields(schema,conf.get(FlinkOptions.CHANGELOG_ENABLED));


nitpick: add space after comma

the-other-tim-brown · 2025-11-26T03:08:05Z

...link-datasource/hudi-flink/src/main/java/org/apache/hudi/schema/FilebasedSchemaProvider.java

    FileSystem fs = HadoopFSUtils.getFs(sourceSchemaFile, HadoopConfigurations.getHadoopConf(new Configuration()));
    try {
-      this.sourceSchema = new Schema.Parser().parse(fs.open(new Path(sourceSchemaFile)));
+      this.sourceSchema = new HoodieSchema.Parser().parse(fs.open(new Path(sourceSchemaFile)));


I know you did not set this up this way but currently this code is failing to close the input stream provided to the parsers. Let's make sure we close these while we're updating this code.

the-other-tim-brown · 2025-11-26T03:10:10Z

...ink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java

      // get full schema iterator.
-      final Schema tableSchema = AvroSchemaCache.intern(new Schema.Parser().parse(tableState.getAvroSchema()));
+      final HoodieSchema schema = HoodieSchemaCache.intern(
+          new HoodieSchema.Parser().parse(tableState.getAvroSchema()));


Just an FYI, there is HoodieSchema.parse for a less verbose way of doing this

the-other-tim-brown · 2025-12-02T02:58:48Z

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java

+    if (hoodieSchema instanceof HoodieSchema.Decimal) {
+      return convertDecimal(hoodieSchema);
+    } else if (hoodieSchema instanceof HoodieSchema.Timestamp) {
+      return convertTimestamp(hoodieSchema);
+    } else if (hoodieSchema instanceof HoodieSchema.Time) {
+      return convertTime(hoodieSchema);
+    }


Instead of using instanceOf just use the switch on the type below

the-other-tim-brown · 2025-12-02T03:03:09Z

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java

+  private static HoodieSchema nullableSchema(HoodieSchema schema) {
+    return schema.isNullable()
+        ? schema
+        : HoodieSchema.createNullable(schema);
+  }


HoodieSchema.createNullable will already check if the existing schema is nullable for you.

the-other-tim-brown · 2025-12-02T03:05:09Z

hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java

+    // Time
+    HoodieSchema timeSchema = HoodieSchemaConverter.convertToSchema(
+        DataTypes.TIME(3).notNull().getLogicalType());
+    assertEquals(HoodieSchemaType.TIME, timeSchema.getType());


Check the time precision here as well and add a test for Time-Micros

Will make this change.

Just one thing to callout here in case someone from hudi flink review this as well. I noticed that the AvroSchemaConverter in the hudi flink module says it does not support micros for Time.https://github.com/apache/hudi/blob/master/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroSchemaConverter.java#L291

However when checking the avro spec I think time technically time is supported with micros
https://avro.apache.org/docs/1.8.0/spec.html#Time+%28microsecond+precision%29, so will add this functionaltiy in the HoodieSchemaConverter class

the-other-tim-brown · 2025-12-02T03:05:57Z

hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java

+  }
+
+  @Test
+  public void testArrayType() {


Let's test arrays and maps with nullable elements as well as non-null

the-other-tim-brown · 2025-12-02T03:08:37Z

hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java

+        DataTypes.FIELD("decimal_col", DataTypes.DECIMAL(10, 2).notNull())
+    ).notNull().getLogicalType();
+
+    HoodieSchema hoodieSchema = HoodieSchemaConverter.convertToSchema(flinkRowType);


Let's add a conversion for a decimal backed by a fixed size byte array

the-other-tim-brown · 2025-12-02T03:09:02Z

hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/util/TestHoodieSchemaConverter.java

+        ).notNull())
+    ).notNull().getLogicalType();
+
+    HoodieSchema hoodieSchema = HoodieSchemaConverter.convertToSchema(flinkRowType);


Could we start by generating a HoodieSchema directly for some of these tests?

…ata type

rahil-c · 2025-12-03T00:21:58Z

@the-other-tim-brown addressed feedback, let me know if something does not look ideal or if we the pr is close to landing

hudi-bot · 2025-12-03T02:29:12Z

CI report:

bccc169 UNKNOWN
5395b5a UNKNOWN
76ae7a7 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

the-other-tim-brown · 2025-12-03T03:15:09Z

...nk-datasource/hudi-flink/src/main/java/org/apache/hudi/util/JsonDeserializationFunction.java

    // Read from file source
    RowType rowType =
-        (RowType) AvroSchemaConverter.convertToDataType(StreamerUtil.getSourceSchema(conf))
+        (RowType) AvroSchemaConverter.convertToDataType(StreamerUtil.getSourceSchema(conf).getAvroSchema())


Can the new converter be used here?

the-other-tim-brown · 2025-12-03T03:15:38Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java

+      Schema avroSchema = schemaResolver.getTableAvroSchema();
+      return HoodieSchema.fromAvroSchema(avroSchema);


Lets update this to call the new method in the table schema resolver

the-other-tim-brown · 2025-12-03T03:15:55Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java

-    final Schema tableAvroSchema = getTableAvroSchema();
-    final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
+    final HoodieSchema tableSchema = getTableSchema();
+    //TODO make a convertor


Remove this TODO?

the-other-tim-brown · 2025-12-03T03:16:13Z

...source/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataParquetReader.java

  public HoodieSchema getSchema() {
    if (fileSchema == null) {
-      fileSchema = AvroSchemaConverter.convertToSchema(getRowType().notNull().getLogicalType());
+      //TODO to create a converter for HoodieSchema


Remove this TODO?

the-other-tim-brown · 2025-12-03T03:16:37Z

...source/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataParquetReader.java


  @Override
  public ClosableIterator<HoodieRecord<RowData>> getRecordIterator(HoodieSchema readerSchema, HoodieSchema requestedSchema) throws IOException {
    //TODO boundary to follow up in later pr


Can this TODO be removed now?

the-other-tim-brown · 2025-12-03T03:17:21Z

...source/hudi-flink/src/main/java/org/apache/hudi/table/format/HoodieRowDataParquetReader.java

  @Override
  public ClosableIterator<String> getRecordKeyIterator() throws IOException {
-    Schema schema = HoodieAvroUtils.getRecordKeySchema();
+    //TODO add a util for this in HoodieSchemaUtils


Can we address this now? Should be a fairly small change

the-other-tim-brown · 2025-12-03T03:18:56Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java

    assert (requiredFields.size() == requiredPos.length);
    Iterator<Integer> positionIterator = Arrays.stream(requiredPos).iterator();
-    requiredFields.forEach(f -> recordBuilder.set(f, getVal(record, positionIterator.next())));
+    requiredFields.forEach(f -> recordBuilder.set(f.name(), getVal(record, positionIterator.next())));


Can we use the position here instead of the name? Setting with the name requires an extra lookup.

the-other-tim-brown · 2025-12-03T03:19:41Z

...ink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cdc/CdcInputFormat.java


    private int[] getRequiredPos(String tableSchema, Schema required) {
-      Schema dataSchema = HoodieAvroUtils.removeMetadataFields(new Schema.Parser().parse(tableSchema));
+      Schema dataSchema = HoodieAvroUtils.removeMetadataFields(HoodieSchema.parse(tableSchema).getAvroSchema());


Can we make this work on HoodieSchema?

the-other-tim-brown · 2025-12-03T03:21:44Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java

+  public static HoodieSchema getTableAvroSchema(HoodieTableMetaClient metaClient, boolean includeMetadataFields) throws Exception {
    TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
-    return schemaUtil.getTableAvroSchema(includeMetadataFields);
+    return HoodieSchema.fromAvroSchema(schemaUtil.getTableAvroSchema(includeMetadataFields));


let's call the new method here as well to directly return the HoodieSchema

the-other-tim-brown · 2025-12-03T03:22:42Z

...flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadOperator.java

+    final HoodieSchema tableSchema;
    try {
-      tableAvroSchema = schemaResolver.getTableAvroSchema();
+      tableSchema = HoodieSchema.fromAvroSchema(schemaResolver.getTableAvroSchema());


Let's call the new method here as well?

rahil-c · 2025-12-03T14:33:20Z

@danny0405 Can you take a look at this PR, specifically the main area would be the HoodieSchemaConverter class which converts HoodieSchema to Flink Type and vice versa. Note this was based off of the original AvroSchemaConverter class in the hudi flink module https://github.com/apache/hudi/blob/master/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/AvroSchemaConverter.java.

danny0405 · 2025-12-04T03:43:06Z

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java

+  }
+
+  private static DataType convertTime(HoodieSchema schema) {
+    if (!(schema instanceof HoodieSchema.Time)) {


we can check the HoodieSchemaType instead of instanceof ?

danny0405 · 2025-12-04T03:43:35Z

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/util/HoodieSchemaConverter.java

+  }
+
+  private static DataType convertTimestamp(HoodieSchema schema) {
+    if (!(schema instanceof HoodieSchema.Timestamp)) {


we can check the HoodieSchemaType instead of instanceof ?

danny0405 · 2025-12-04T03:44:50Z

...link-datasource/hudi-flink/src/main/java/org/apache/hudi/schema/FilebasedSchemaProvider.java

    try {
-      this.sourceSchema = new Schema.Parser().parse(fs.open(new Path(sourceSchemaFile)));
+      try (InputStream stream = fs.open(new Path(sourceSchemaFile))) {
+        this.sourceSchema = new HoodieSchema.Parser().parse(stream);


we have HoodieSchema.parse(

Intial flink schema changes

2e872e4

github-actions bot added the size:L PR with lines of changes in (300, 1000] label Nov 25, 2025

rahil-c added 2 commits November 25, 2025 15:13

checkstyle fixes

bccc169

fix compilation issues

cf94d6c

rahil-c marked this pull request as ready for review November 26, 2025 02:28

rahil-c mentioned this pull request Nov 26, 2025

Phase 8: Flink Engine Migration #14273

Open

the-other-tim-brown reviewed Nov 26, 2025

View reviewed changes

add converter

96b5fcf

rahil-c changed the title ~~Migrate hudi-flink to use HoodieSchema instead of avro Schema~~ refactor: Migrate hudi-flink to use HoodieSchema instead of avro Schema Nov 30, 2025

address tim comments, add converter for hoodie schema to flink types

96b1c26

github-actions bot added size:XL PR with lines of changes > 1000 and removed size:L PR with lines of changes in (300, 1000] labels Dec 1, 2025

fix clustering IT

83ca6a7

the-other-tim-brown reviewed Dec 2, 2025

View reviewed changes

rahil-c mentioned this pull request Dec 2, 2025

feat(schema): Add helper to get HoodieSchema in TableSchemaResolver #17456

Merged

3 tasks

rahil-c changed the title ~~refactor: Migrate hudi-flink to use HoodieSchema instead of avro Schema~~ feat(schema) Migrate hudi-flink to use HoodieSchema instead of avro Schema Dec 2, 2025

rahil-c changed the title ~~feat(schema) Migrate hudi-flink to use HoodieSchema instead of avro Schema~~ feat(schema): Migrate hudi-flink to use HoodieSchema instead of avro Schema Dec 2, 2025

rahil-c added 4 commits December 2, 2025 14:37

address comments

6ddca22

start with hoodie schema for certain tests when converting to flink d…

5395b5a

…ata type

Merge branch 'master' into rahil/flink-migration

12b3889

use helper from TableSchemaResolver

76ae7a7

the-other-tim-brown reviewed Dec 3, 2025

View reviewed changes

voonhous mentioned this pull request Dec 3, 2025

feat(schema): phase 12 - Perform Data Source Helpers Migration #14382

Open

3 tasks

danny0405 reviewed Dec 4, 2025

View reviewed changes

		Schema avroSchema = schemaResolver.getTableAvroSchema();
		return HoodieSchema.fromAvroSchema(avroSchema);

feat(schema): Migrate hudi-flink to use HoodieSchema instead of avro Schema #14355

Are you sure you want to change the base?

feat(schema): Migrate hudi-flink to use HoodieSchema instead of avro Schema #14355

Conversation

rahil-c commented Nov 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Describe the issue this Pull Request addresses

Summary and Changelog

Impact

Risk Level

Documentation Update

Contributor's checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rahil-c Dec 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rahil-c commented Dec 3, 2025

Uh oh!

hudi-bot commented Dec 3, 2025

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rahil-c commented Nov 25, 2025 •

edited

Loading

rahil-c Dec 2, 2025 •

edited

Loading