This is an automated email from the ASF dual-hosted git repository.
nsivabalan 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 d8ebba4974e7 fix: relax existing column to nullable in reconcileSchema
when source made it nullable (#19337)
d8ebba4974e7 is described below
commit d8ebba4974e74d990af9f0c71ebd6073205e603e
Author: vamsikarnika <[email protected]>
AuthorDate: Wed Jul 22 07:59:16 2026 +0530
fix: relax existing column to nullable in reconcileSchema when source made
it nullable (#19337)
In reconcileSchema, detect existing columns whose incoming schema is
nullable but the table is required (nullabilityRelaxColumns), exclude them from
the early-return short-circuit, and relax them to nullable in the result via
updateColumnNullability(col, true). This only ever widens, never tightens (a
nullable→required incoming leaves the table nullable), consistent with
reconcileSchemaRequirements, and still null-fills genuinely missing columns.
---
.../internal/utils/AvroSchemaEvolutionUtils.java | 17 +++++-
.../utils/TestAvroSchemaEvolutionUtils.java | 60 ++++++++++++++++++++++
.../org/apache/hudi/TestHoodieSchemaUtils.java | 19 +++++++
3 files changed, 95 insertions(+), 1 deletion(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
index 4ce674bf853e..f6405fef0137 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/utils/AvroSchemaEvolutionUtils.java
@@ -78,7 +78,19 @@ public class AvroSchemaEvolutionUtils {
.stream()
.filter(f -> colNamesFromOldSchema.contains(f) &&
!inComingInternalSchema.findType(f).equals(oldTableSchema.findType(f)))
.collect(Collectors.toList());
- if (colNamesFromIncoming.size() == colNamesFromOldSchema.size() &&
diffFromOldSchema.size() == 0 && typeChangeColumns.isEmpty()) {
+ // check columns the incoming schema relaxed from required to nullable.
Since the result is built from
+ // oldTableSchema (to preserve column order/ids and to null-fill missing
columns), an existing column
+ // whose incoming counterpart became nullable would otherwise silently
keep the table's REQUIRED
+ // nullability, blocking a valid required -> nullable evolution. We only
ever relax (never tighten).
+ List<String> nullabilityRelaxColumns = colNamesFromIncoming
+ .stream()
+ .filter(f -> colNamesFromOldSchema.contains(f)
+ && !META_FIELD_NAMES.contains(f)
+ && inComingInternalSchema.findField(f).isOptional()
+ && !oldTableSchema.findField(f).isOptional())
+ .collect(Collectors.toList());
+ if (colNamesFromIncoming.size() == colNamesFromOldSchema.size() &&
diffFromOldSchema.size() == 0
+ && typeChangeColumns.isEmpty() && nullabilityRelaxColumns.isEmpty()) {
return oldTableSchema;
}
@@ -122,6 +134,9 @@ public class AvroSchemaEvolutionUtils {
typeChange.updateColumnType(col, inComingInternalSchema.findType(col));
});
+ // relax existing columns to nullable when the incoming schema made them
nullable (valid widening)
+ nullabilityRelaxColumns.forEach(col ->
typeChange.updateColumnNullability(col, true));
+
if (makeMissingFieldsNullable) {
// mark columns missing from incoming schema as nullable
Set<String> visited = new HashSet<>();
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
index 514048710034..d82dd84b87e1 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/utils/TestAvroSchemaEvolutionUtils.java
@@ -567,4 +567,64 @@ public class TestAvroSchemaEvolutionUtils {
// the evolved schema should be the old table schema, since there is no
type change at all.
Assertions.assertEquals(oldInternalSchema, evolvedSchema);
}
+
+ /**
+ * When the incoming schema relaxes an existing required column to nullable,
reconcileSchema must evolve
+ * that column to nullable in the result, even when
makeMissingFieldsNullable is true. Previously the
+ * result was rebuilt from the required table and the relaxation was
silently dropped, so records with
+ * null in that column failed the write / were quarantined.
+ */
+ @Test
+ public void testReconcileSchemaRelaxesExistingColumnToNullable() {
+ // table: id (required int), flag (required boolean) -- same column set as
the incoming schema
+ Types.RecordType oldRecord = Types.RecordType.get(
+ Types.Field.get(0, false, "id", Types.IntType.get()),
+ Types.Field.get(1, false, "flag", Types.BooleanType.get())
+ );
+ InternalSchema oldSchema = new InternalSchema(oldRecord);
+ // incoming: identical columns, but the source relaxed "flag" to nullable
+ Types.RecordType incomingRecord = Types.RecordType.get(
+ Types.Field.get(0, false, "id", Types.IntType.get()),
+ Types.Field.get(1, true, "flag", Types.BooleanType.get())
+ );
+ incomingRecord = (Types.RecordType)
InternalSchemaBuilder.getBuilder().refreshNewId(incomingRecord, new
AtomicInteger(0));
+ HoodieSchema incomingSchema =
InternalSchemaConverter.convert(incomingRecord, "test1");
+
+ InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldSchema, true);
+
+ Types.RecordType checkedRecord = Types.RecordType.get(
+ Types.Field.get(0, false, "id", Types.IntType.get()),
+ Types.Field.get(1, true, "flag", Types.BooleanType.get())
+ );
+ Assertions.assertEquals(checkedRecord, result.getRecord());
+ }
+
+ /**
+ * reconcileSchema must only ever relax (widen) an existing column's
nullability, never tighten it: if the
+ * incoming schema marks a column required but the table has it nullable,
the table stays nullable.
+ */
+ @Test
+ public void testReconcileSchemaDoesNotTightenNullableToRequired() {
+ // table: id (required int), flag (nullable boolean)
+ Types.RecordType oldRecord = Types.RecordType.get(
+ Types.Field.get(0, false, "id", Types.IntType.get()),
+ Types.Field.get(1, true, "flag", Types.BooleanType.get())
+ );
+ InternalSchema oldSchema = new InternalSchema(oldRecord);
+ // incoming: source tightened "flag" to required -- must NOT tighten the
table
+ Types.RecordType incomingRecord = Types.RecordType.get(
+ Types.Field.get(0, false, "id", Types.IntType.get()),
+ Types.Field.get(1, false, "flag", Types.BooleanType.get())
+ );
+ incomingRecord = (Types.RecordType)
InternalSchemaBuilder.getBuilder().refreshNewId(incomingRecord, new
AtomicInteger(0));
+ HoodieSchema incomingSchema =
InternalSchemaConverter.convert(incomingRecord, "test1");
+
+ InternalSchema result =
AvroSchemaEvolutionUtils.reconcileSchema(incomingSchema, oldSchema, true);
+
+ Types.RecordType checkedRecord = Types.RecordType.get(
+ Types.Field.get(0, false, "id", Types.IntType.get()),
+ Types.Field.get(1, true, "flag", Types.BooleanType.get())
+ );
+ Assertions.assertEquals(checkedRecord, result.getRecord());
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestHoodieSchemaUtils.java
b/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestHoodieSchemaUtils.java
index 18d488098c8f..d4e1e3aeba7d 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestHoodieSchemaUtils.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/java/org/apache/hudi/TestHoodieSchemaUtils.java
@@ -321,6 +321,25 @@ public class TestHoodieSchemaUtils {
assertEquals(expected, deduceWriterSchema(end, start, true));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testExistingColumnRelaxedToNullableEvolves(boolean
setNullForMissingColumns) {
+ // Table has field2 as a required boolean; the incoming (source) schema
relaxed it to nullable, same
+ // column set otherwise. The deduced writer schema must evolve field2 to
nullable regardless of the
+ // set.null.for.missing.columns flag -- with the flag on this used to
silently stay required, so records
+ // with null in field2 failed the write / were quarantined.
+ HoodieSchema table = createRecord("relaxRec",
+ createPrimitiveField("field1", HoodieSchemaType.INT),
+ createPrimitiveField("field2", HoodieSchemaType.BOOLEAN));
+ HoodieSchema incoming = createRecord("relaxRec",
+ createPrimitiveField("field1", HoodieSchemaType.INT),
+ createNullablePrimitiveField("field2", HoodieSchemaType.BOOLEAN));
+ HoodieSchema expected = createRecord("relaxRec",
+ createPrimitiveField("field1", HoodieSchemaType.INT),
+ createNullablePrimitiveField("field2", HoodieSchemaType.BOOLEAN));
+ assertEquals(expected, deduceWriterSchema(incoming, table,
setNullForMissingColumns));
+ }
+
private static HoodieSchema deduceWriterSchema(HoodieSchema incomingSchema,
HoodieSchema latestTableSchema) {
return deduceWriterSchema(incomingSchema, latestTableSchema, false);
}
