This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 369ca5cb3596 refactor: Migrate
reconcileSchema/reconcileSchemaRequirements to HoodieSchema (#19154)
369ca5cb3596 is described below
commit 369ca5cb3596c1efe8fd8c2ea4ab922fda21bdf1
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).
---
.../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 14789a4c667d..932e11da2733 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 530734172ba6..8de3ff23f68e 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
@@ -279,7 +279,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)
}
