This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a91e4443bcfd9464fbfecf036ee01420e2a9a950 Author: voonhous <[email protected]> AuthorDate: Fri Jul 3 18:52:22 2026 +0800 refactor: Migrate reconcileSchema/reconcileSchemaRequirements to HoodieSchema (#19154) The reconcile adapters in AvroSchemaEvolutionUtils took Avro Schema in/out and immediately wrapped inputs via HoodieSchema.fromAvroSchema, while every caller already held a HoodieSchema and passed .toAvroSchema(). Migrate the three signatures to HoodieSchema: - reconcileSchema(Schema, InternalSchema) -> (HoodieSchema, InternalSchema) - reconcileSchema(Schema, Schema): Schema -> (HoodieSchema, HoodieSchema): HoodieSchema - reconcileSchemaRequirements(Schema, Schema): Schema -> HoodieSchema in/out Callers (BaseHoodieWriteClient, FileGroupReaderBasedMergeHandle, HoodieMergeHelper, HoodieSchemaUtils.scala x3, and the unit test) drop the .toAvroSchema()/.getAvroSchema() round-trips; the InternalSchema convert(...) boundary is unchanged. Drops the org.apache.avro.Schema import from AvroSchemaEvolutionUtils. Part of the Avro Schema -> HoodieSchema migration (#14263). (cherry picked from commit 369ca5cb3596c1efe8fd8c2ea4ab922fda21bdf1) --- .../apache/hudi/client/BaseHoodieWriteClient.java | 2 +- .../hudi/io/FileGroupReaderBasedMergeHandle.java | 2 +- .../table/action/commit/HoodieMergeHelper.java | 2 +- .../schema/utils/AvroSchemaEvolutionUtils.java | 26 ++++++++++------------ .../schema/utils/TestAvroSchemaEvolutionUtils.java | 6 ++--- .../scala/org/apache/hudi/HoodieSchemaUtils.scala | 8 +++---- 6 files changed, 21 insertions(+), 25 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java index 8284f61baa15..f5d050d88098 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java @@ -369,7 +369,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient internalSchema = InternalSchemaUtils.searchSchema(Long.parseLong(instantTime), SerDeHelper.parseSchemas(historySchemaStr)); } - InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(schema.toAvroSchema(), internalSchema, config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS)); + InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(schema, internalSchema, config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS)); if (evolvedSchema.equals(internalSchema)) { metadata.addMetadata(SerDeHelper.LATEST_SCHEMA, SerDeHelper.toJson(evolvedSchema)); //TODO save history schema by metaTable diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java index 2893913a0d13..ef0d7e08dc31 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java @@ -257,7 +257,7 @@ public class FileGroupReaderBasedMergeHandle<T, I, K, O> extends HoodieWriteMerg } boolean usePosition = config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS); Option<InternalSchema> internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema()) - .map(internalSchema -> AvroSchemaEvolutionUtils.reconcileSchema(writeSchemaWithMetaFields.toAvroSchema(), internalSchema, + .map(internalSchema -> AvroSchemaEvolutionUtils.reconcileSchema(writeSchemaWithMetaFields, internalSchema, config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS))); long maxMemoryPerCompaction = getMaxMemoryForMerge(); props.put(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(), String.valueOf(maxMemoryPerCompaction)); diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java index f887e4d5ac74..2f571e0d3e2b 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieMergeHelper.java @@ -170,7 +170,7 @@ public class HoodieMergeHelper<T> extends BaseMergeHelper { // TODO support bootstrap if (querySchemaOpt.isPresent() && !baseFile.getBootstrapBaseFile().isPresent()) { // check implicitly add columns, and position reorder(spark sql may change cols order) - InternalSchema querySchema = AvroSchemaEvolutionUtils.reconcileSchema(writerSchema.toAvroSchema(), + InternalSchema querySchema = AvroSchemaEvolutionUtils.reconcileSchema(writerSchema, querySchemaOpt.get(), writeConfig.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS)); long commitInstantTime = Long.parseLong(baseFile.getCommitTime()); InternalSchema fileSchema = InternalSchemaCache.getInternalSchemaByVersionId(commitInstantTime, metaClient); diff --git a/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java b/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java index 5ae6f203f0f3..c31b5c057181 100644 --- a/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java @@ -24,8 +24,6 @@ import org.apache.hudi.internal.schema.InternalSchema; import org.apache.hudi.internal.schema.action.TableChanges; import org.apache.hudi.internal.schema.action.TableChangesHelper; -import org.apache.avro.Schema; - import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -64,12 +62,12 @@ public class AvroSchemaEvolutionUtils { * nullable in the result. Otherwise, no updates will be made to those fields. * @return reconcile Schema */ - public static InternalSchema reconcileSchema(Schema incomingSchema, InternalSchema oldTableSchema, boolean makeMissingFieldsNullable) { + public static InternalSchema reconcileSchema(HoodieSchema incomingSchema, InternalSchema oldTableSchema, boolean makeMissingFieldsNullable) { /* If incoming schema is null, we fall back on table schema. */ - if (incomingSchema.getType() == Schema.Type.NULL) { + if (incomingSchema.isSchemaNull()) { return oldTableSchema; } - InternalSchema inComingInternalSchema = convert(HoodieSchema.fromAvroSchema(incomingSchema), oldTableSchema.getNameToPosition()); + InternalSchema inComingInternalSchema = convert(incomingSchema, oldTableSchema.getNameToPosition()); // check column add/missing List<String> colNamesFromIncoming = inComingInternalSchema.getAllColsFullName(); List<String> colNamesFromOldSchema = oldTableSchema.getAllColsFullName(); @@ -149,8 +147,8 @@ public class AvroSchemaEvolutionUtils { return evolvedSchema; } - public static Schema reconcileSchema(Schema incomingSchema, Schema oldTableSchema, boolean makeMissingFieldsNullable) { - return convert(reconcileSchema(incomingSchema, convert(HoodieSchema.fromAvroSchema(oldTableSchema)), makeMissingFieldsNullable), oldTableSchema.getFullName()).toAvroSchema(); + public static HoodieSchema reconcileSchema(HoodieSchema incomingSchema, HoodieSchema oldTableSchema, boolean makeMissingFieldsNullable) { + return convert(reconcileSchema(incomingSchema, convert(oldTableSchema), makeMissingFieldsNullable), oldTableSchema.getFullName()); } /** @@ -169,18 +167,18 @@ public class AvroSchemaEvolutionUtils { * @param targetSchema target schema that source schema will be reconciled against * @return schema (based off {@code source} one) that has nullability constraints and datatypes reconciled */ - public static Schema reconcileSchemaRequirements(Schema sourceSchema, Schema targetSchema, boolean shouldReorderColumns) { - if (targetSchema.getType() == Schema.Type.NULL || targetSchema.getFields().isEmpty()) { + public static HoodieSchema reconcileSchemaRequirements(HoodieSchema sourceSchema, HoodieSchema targetSchema, boolean shouldReorderColumns) { + if (targetSchema.isSchemaNull() || targetSchema.getFields().isEmpty()) { return sourceSchema; } - if (sourceSchema == null || sourceSchema.getType() == Schema.Type.NULL || sourceSchema.getFields().isEmpty()) { + if (sourceSchema == null || sourceSchema.isSchemaNull() || sourceSchema.getFields().isEmpty()) { return targetSchema; } - InternalSchema targetInternalSchema = convert(HoodieSchema.fromAvroSchema(targetSchema)); + InternalSchema targetInternalSchema = convert(targetSchema); // Use existing fieldIds for consistent field ordering between commits when shouldReorderColumns is true - InternalSchema sourceInternalSchema = convert(HoodieSchema.fromAvroSchema(sourceSchema), shouldReorderColumns ? targetInternalSchema.getNameToPosition() : Collections.emptyMap()); + InternalSchema sourceInternalSchema = convert(sourceSchema, shouldReorderColumns ? targetInternalSchema.getNameToPosition() : Collections.emptyMap()); List<String> colNamesSourceSchema = sourceInternalSchema.getAllColsFullName(); List<String> colNamesTargetSchema = targetInternalSchema.getAllColsFullName(); @@ -200,7 +198,7 @@ public class AvroSchemaEvolutionUtils { if (nullableUpdateColsInSource.isEmpty() && typeUpdateColsInSource.isEmpty()) { //standardize order of unions - return convert(sourceInternalSchema, sourceSchema.getFullName()).toAvroSchema(); + return convert(sourceInternalSchema, sourceSchema.getFullName()); } TableChanges.ColumnUpdateChange schemaChange = TableChanges.ColumnUpdateChange.get(sourceInternalSchema); @@ -218,7 +216,7 @@ public class AvroSchemaEvolutionUtils { } - return convert(SchemaChangeUtils.applyTableChanges2Schema(sourceInternalSchema, schemaChange), sourceSchema.getFullName()).toAvroSchema(); + return convert(SchemaChangeUtils.applyTableChanges2Schema(sourceInternalSchema, schemaChange), sourceSchema.getFullName()); } } diff --git a/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java b/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java index 662436de9d75..19d4205d3556 100644 --- a/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java +++ b/hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java @@ -486,7 +486,7 @@ public class TestAvroSchemaEvolutionUtils { ); evolvedRecord = (Types.RecordType)InternalSchemaBuilder.getBuilder().refreshNewId(evolvedRecord, new AtomicInteger(0)); HoodieSchema evolvedSchema = InternalSchemaConverter.convert(evolvedRecord, "test1"); - InternalSchema result = AvroSchemaEvolutionUtils.reconcileSchema(evolvedSchema.getAvroSchema(), oldSchema, false); + InternalSchema result = AvroSchemaEvolutionUtils.reconcileSchema(evolvedSchema, oldSchema, false); Types.RecordType checkedRecord = Types.RecordType.get( Types.Field.get(0, false, "id", Types.IntType.get()), Types.Field.get(1, true, "data", Types.StringType.get()), @@ -541,7 +541,7 @@ public class TestAvroSchemaEvolutionUtils { + "{\"name\":\"d2\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null}]}"); HoodieSchema simpleReconcileSchema = InternalSchemaConverter.convert(AvroSchemaEvolutionUtils - .reconcileSchema(incomingSchema.getAvroSchema(), InternalSchemaConverter.convert(schema), false), "schemaNameFallback"); + .reconcileSchema(incomingSchema, InternalSchemaConverter.convert(schema), false), "schemaNameFallback"); Assertions.assertEquals(simpleCheckSchema, simpleReconcileSchema); } @@ -563,7 +563,7 @@ public class TestAvroSchemaEvolutionUtils { InternalSchema oldInternalSchema = InternalSchemaConverter.convert(oldSchema); // set a non-default schema id for old table schema, e.g., 2. oldInternalSchema.setSchemaId(2); - InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema.getAvroSchema(), oldInternalSchema, false); + InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldInternalSchema, false); // the evolved schema should be the old table schema, since there is no type change at all. Assertions.assertEquals(oldInternalSchema, evolvedSchema); } diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala index 5a40f45c1241..6c353481595e 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieSchemaUtils.scala @@ -173,7 +173,7 @@ object HoodieSchemaUtils { if (!mergeIntoWrites && !shouldValidateSchemasCompatibility && !allowAutoEvolutionColumnDrop) { // Default behaviour val reconciledSchema = if (setNullForMissingColumns) { - HoodieSchema.fromAvroSchema(AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema.toAvroSchema(), latestTableSchema.toAvroSchema(), setNullForMissingColumns)) + AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, latestTableSchema, setNullForMissingColumns) } else { canonicalizedSourceSchema } @@ -205,7 +205,7 @@ object HoodieSchemaUtils { // Apply schema evolution, by auto-merging write schema and read schema val setNullForMissingColumns = opts.getOrElse(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key(), HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.defaultValue()).toBoolean - val mergedInternalSchema = AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema.toAvroSchema(), internalSchema, setNullForMissingColumns) + val mergedInternalSchema = AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, internalSchema, setNullForMissingColumns) val evolvedSchema = InternalSchemaConverter.convert(mergedInternalSchema, latestTableSchema.getFullName) val shouldRemoveMetaDataFromInternalSchema = sourceSchema.getFields.asScala.filter(f => f.name().equalsIgnoreCase(HoodieRecord.RECORD_KEY_METADATA_FIELD)).isEmpty if (shouldRemoveMetaDataFromInternalSchema) HoodieCommonSchemaUtils.removeMetadataFields(evolvedSchema) else evolvedSchema @@ -257,9 +257,7 @@ object HoodieSchemaUtils { */ private def canonicalizeSchema(sourceSchema: HoodieSchema, latestTableSchema: HoodieSchema, opts : Map[String, String], shouldReorderColumns: Boolean): HoodieSchema = { - HoodieSchema.fromAvroSchema( - reconcileSchemaRequirements(sourceSchema.toAvroSchema(), latestTableSchema.toAvroSchema(), shouldReorderColumns) - ) + reconcileSchemaRequirements(sourceSchema, latestTableSchema, shouldReorderColumns) }
