This is an automated email from the ASF dual-hosted git repository.
danny0405 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 ee093f0e148a fix(spark): make new schema evolution fields nullable
(#19665)
ee093f0e148a is described below
commit ee093f0e148a6063d6dc92560516d5e2fa848346
Author: Danny Chan <[email protected]>
AuthorDate: Fri Aug 21 12:14:01 2026 +0800
fix(spark): make new schema evolution fields nullable (#19665)
* fix(spark): restore null backfill for new columns
---
.../internal/utils/AvroSchemaEvolutionUtils.java | 33 ++++++++---
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 5 +-
.../org/apache/hudi/TestHoodieSchemaUtils.java | 66 +++++++++++++++++++++
.../apache/hudi/functional/TestCOWDataSource.scala | 69 ++++++++++++----------
4 files changed, 131 insertions(+), 42 deletions(-)
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 0310ccfb236e..2e7ade10dab5 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
@@ -302,17 +302,19 @@ public class AvroSchemaEvolutionUtils {
* {@code target} one. Source is considered to be new incoming schema, while
target could refer to prev table schema.
* For example,
* if colA in source is non-nullable, but is nullable in target, output
schema will have colA as nullable.
- * if "hoodie.datasource.write.new.columns.nullable" is set to true and if
colB is not present in source, but
- * is present in target, output schema will have colB as nullable.
+ * if colB is present in source, but not in target, output schema will have
colB as nullable. If colB is a complex
+ * type, its existing descendants retain their nullability constraints.
* if colC has different data type in source schema compared to target
schema and if its promotable, (say source is int,
* and target is long and since int can be promoted to long), colC will be
long data type in output schema.
*
*
* @param sourceSchema source schema that needs reconciliation
* @param targetSchema target schema that source schema will be reconciled
against
+ * @param shouldReorderColumns whether fields should be reordered to match
the target schema
* @return schema (based off {@code source} one) that has nullability
constraints and datatypes reconciled
*/
- public static HoodieSchema reconcileSchemaRequirements(HoodieSchema
sourceSchema, HoodieSchema targetSchema, boolean shouldReorderColumns) {
+ public static HoodieSchema reconcileSchemaRequirements(HoodieSchema
sourceSchema, HoodieSchema targetSchema,
+ boolean
shouldReorderColumns) {
if (targetSchema.isSchemaNull() || targetSchema.getFields().isEmpty()) {
return sourceSchema;
}
@@ -327,12 +329,30 @@ public class AvroSchemaEvolutionUtils {
List<String> colNamesSourceSchema =
sourceInternalSchema.getAllColsFullName();
List<String> colNamesTargetSchema =
targetInternalSchema.getAllColsFullName();
+ List<String> userColNamesSourceSchema = colNamesSourceSchema.stream()
+ .filter(field -> !META_FIELD_NAMES.contains(field))
+ .collect(Collectors.toList());
List<String> nullableUpdateColsInSource = new ArrayList<>();
List<String> typeUpdateColsInSource = new ArrayList<>();
- colNamesSourceSchema.forEach(field -> {
- // handle columns that needs to be made nullable
- if (colNamesTargetSchema.contains(field) &&
sourceInternalSchema.findField(field).isOptional() !=
targetInternalSchema.findField(field).isOptional()) {
+
+ // Only relax the topmost field in a wholly new subtree. Relaxing every
descendant would alter the
+ // element/field constraints supplied by the writer instead of only making
the evolved field backfillable.
+ Set<String> visitedNewColumns = new HashSet<>();
+ userColNamesSourceSchema.stream()
+ .filter(field -> !colNamesTargetSchema.contains(field))
+ .sorted()
+ .forEach(field -> {
+ String parent = TableChangesHelper.getParentName(field);
+ if (!visitedNewColumns.contains(parent)) {
+ nullableUpdateColsInSource.add(field);
+ }
+ visitedNewColumns.add(field);
+ });
+
+ userColNamesSourceSchema.forEach(field -> {
+ if (colNamesTargetSchema.contains(field)
+ && sourceInternalSchema.findField(field).isOptional() !=
targetInternalSchema.findField(field).isOptional()) {
nullableUpdateColsInSource.add(field);
}
// handle columns that needs type to be updated
@@ -364,4 +384,3 @@ public class AvroSchemaEvolutionUtils {
return
convert(SchemaChangeUtils.applyTableChanges2Schema(sourceInternalSchema,
schemaChange), sourceSchema.getFullName());
}
}
-
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 234ef5a556b1..f82f496ef027 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
@@ -137,7 +137,7 @@ object HoodieSchemaUtils {
val shouldReconcileSchema =
opts.getOrElse(DataSourceWriteOptions.RECONCILE_SCHEMA.key(),
DataSourceWriteOptions.RECONCILE_SCHEMA.defaultValue().toString).toBoolean
val canonicalizedSourceSchema = if (shouldCanonicalizeSchema) {
- canonicalizeSchema(sourceSchema, latestTableSchema, opts,
!shouldReconcileSchema)
+ canonicalizeSchema(sourceSchema, latestTableSchema,
!shouldReconcileSchema)
} else {
InternalSchemaConverter.fixNullOrdering(sourceSchema)
}
@@ -276,12 +276,11 @@ object HoodieSchemaUtils {
*
* TODO support casing reconciliation
*/
- private def canonicalizeSchema(sourceSchema: HoodieSchema,
latestTableSchema: HoodieSchema, opts : Map[String, String],
+ private def canonicalizeSchema(sourceSchema: HoodieSchema,
latestTableSchema: HoodieSchema,
shouldReorderColumns: Boolean): HoodieSchema
= {
reconcileSchemaRequirements(sourceSchema, latestTableSchema,
shouldReorderColumns)
}
-
private def reconcileSchemasLegacy(tableSchema: HoodieSchema, newSchema:
HoodieSchema): (HoodieSchema, Boolean) = {
// Legacy reconciliation implements following semantic
// - In case new-schema is a "compatible" projection of the existing
table's one (projection allowing
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 d4e1e3aeba7d..6175c385912d 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
@@ -21,8 +21,12 @@ package org.apache.hudi;
import org.apache.hudi.common.config.HoodieCommonConfig;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.schema.HoodieSchemaType;
+import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter;
+import org.apache.hudi.common.schema.internal.utils.AvroSchemaEvolutionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieNullSchemaTypeException;
@@ -31,6 +35,7 @@ import
org.apache.hudi.exception.SchemaBackwardsCompatibilityException;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.util.EnumMap;
@@ -340,6 +345,67 @@ public class TestHoodieSchemaUtils {
assertEquals(expected, deduceWriterSchema(incoming, table,
setNullForMissingColumns));
}
+ @ParameterizedTest
+ @CsvSource({
+ "false,false",
+ "true,false",
+ "true,true"
+ })
+ void testNewColumnsAreNullableAcrossReconciliationPaths(boolean
reconcileSchema, boolean useInternalSchema) {
+ HoodieSchema table = createRecord("newColumns",
+ createPrimitiveField("id", HoodieSchemaType.INT),
+ HoodieSchemaField.of("address", createRecord("address",
+ createPrimitiveField("city", HoodieSchemaType.STRING)), null,
null));
+ HoodieSchema incoming = createRecord("newColumns",
+ createPrimitiveField("id", HoodieSchemaType.INT),
+ HoodieSchemaField.of("address", createRecord("address",
+ createPrimitiveField("city", HoodieSchemaType.STRING),
+ createPrimitiveField("country", HoodieSchemaType.STRING)), null,
null),
+ createPrimitiveField("phone", HoodieSchemaType.STRING),
+ HoodieSchemaField.of("profile", createRecord("profile",
+ createPrimitiveField("name", HoodieSchemaType.STRING)), null,
null),
+ createArrayField("items", createRecord("item",
+ createPrimitiveField("sku", HoodieSchemaType.STRING))));
+ TypedProperties properties = new TypedProperties();
+ properties.setProperty(HoodieCommonConfig.RECONCILE_SCHEMA.key(),
Boolean.toString(reconcileSchema));
+
+ HoodieSchema actual = HoodieSchemaUtils.deduceWriterSchema(
+ incoming,
+ Option.of(table),
+ useInternalSchema ? Option.of(InternalSchemaConverter.convert(table))
: Option.empty(),
+ properties);
+
+
assertFalse(actual.getNestedField("address.city").get().getRight().isNullable());
+
assertNewFieldIsNullableWithNullDefault(actual.getNestedField("address.country").get().getRight());
+ assertNewFieldIsNullableWithNullDefault(actual.getField("phone").get());
+ HoodieSchemaField profile = actual.getField("profile").get();
+ assertNewFieldIsNullableWithNullDefault(profile);
+
assertFalse(profile.schema().getNonNullType().getField("name").get().isNullable());
+ HoodieSchemaField items = actual.getField("items").get();
+ assertNewFieldIsNullableWithNullDefault(items);
+ HoodieSchema itemElement =
items.schema().getNonNullType().getElementType();
+ assertFalse(itemElement.isNullable());
+ assertFalse(itemElement.getField("sku").get().isNullable());
+ }
+
+ @Test
+ void testMetadataFieldsAreExcludedFromNewColumnNullability() {
+ HoodieSchema table = createRecord("metadata", createPrimitiveField("id",
HoodieSchemaType.INT));
+ HoodieSchema incoming = createRecord("metadata",
+ createPrimitiveField("id", HoodieSchemaType.INT),
+ createPrimitiveField(HoodieRecord.COMMIT_TIME_METADATA_FIELD,
HoodieSchemaType.STRING));
+
+ HoodieSchema actual =
AvroSchemaEvolutionUtils.reconcileSchemaRequirements(incoming, table, false);
+
+
assertFalse(actual.getField(HoodieRecord.COMMIT_TIME_METADATA_FIELD).get().isNullable());
+ }
+
+ private static void
assertNewFieldIsNullableWithNullDefault(HoodieSchemaField field) {
+ assertTrue(field.isNullable());
+ assertTrue(field.hasDefaultValue());
+ assertEquals(HoodieSchema.NULL_VALUE, field.defaultVal().get());
+ }
+
private static HoodieSchema deduceWriterSchema(HoodieSchema incomingSchema,
HoodieSchema latestTableSchema) {
return deduceWriterSchema(incomingSchema, latestTableSchema, false);
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
index 9a3123ffde0c..38cdbb3af6c6 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
@@ -27,9 +27,9 @@ import org.apache.hudi.common.config.{HoodieCommonConfig,
HoodieMetadataConfig,
import
org.apache.hudi.common.config.TimestampKeyGeneratorConfig.{TIMESTAMP_INPUT_DATE_FORMAT,
TIMESTAMP_OUTPUT_DATE_FORMAT, TIMESTAMP_TIMEZONE_FORMAT, TIMESTAMP_TYPE_FIELD}
import org.apache.hudi.common.config.metrics.HoodieMetricsConfig
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.model.{HoodieRecord,
HoodieReplaceCommitMetadata, WriteOperationType}
+import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieRecord,
HoodieReplaceCommitMetadata, WriteOperationType}
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
-import
org.apache.hudi.common.schema.HoodieSchemaCompatibilityChecker.SchemaIncompatibilityType
+import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
HoodieTableVersion, TableSchemaResolver}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline,
TimelineUtils}
import org.apache.hudi.common.testutils.{HoodieTestDataGenerator,
HoodieTestUtils}
@@ -37,7 +37,7 @@ import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{deleteRecordsTo
import
org.apache.hudi.common.testutils.HoodieTestUtils.{INSTANT_FILE_NAME_GENERATOR,
INSTANT_GENERATOR}
import org.apache.hudi.common.util.{ClusteringUtils, Option}
import org.apache.hudi.config.HoodieWriteConfig
-import org.apache.hudi.exception.{HoodieException,
SchemaBackwardsCompatibilityException}
+import org.apache.hudi.exception.HoodieException
import org.apache.hudi.hive.HiveSyncConfigHolder
import org.apache.hudi.keygen.{ComplexKeyGenerator, CustomKeyGenerator,
GlobalDeleteKeyGenerator, NonpartitionedKeyGenerator, SimpleKeyGenerator,
TimestampBasedKeyGenerator}
import org.apache.hudi.keygen.constant.{KeyGeneratorOptions, KeyGeneratorType}
@@ -1966,11 +1966,12 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
assertEquals(0, result.filter(result("id") === 1).count())
}
- /** Test case to verify MAKE_NEW_COLUMNS_NULLABLE config parameter. */
- @Test
- def testSchemaEvolutionWithNewColumn(): Unit = {
- val df1 = spark.sql("select '1' as event_id, '2' as ts, '3' as version,
'foo' as event_date")
- var hudiOptions = Map[String, String](
+ @ParameterizedTest
+ @CsvSource(value = Array("false,false", "false,true", "true,false",
"true,true"))
+ def testSchemaEvolutionWithNewColumns(schemaOnRead: Boolean,
reconcileSchema: Boolean): Unit = {
+ val df1 = spark.sql(
+ "select '1' as event_id, '2' as ts, '3' as version, named_struct('city',
'Paris') as address")
+ val hudiOptions = Map[String, String](
HoodieWriteConfig.TBL_NAME.key() -> "test_hudi_merger",
KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key() -> "event_id",
KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key() -> "version",
@@ -1979,35 +1980,39 @@ class TestCOWDataSource extends
HoodieSparkClientTestBase with ScalaAssertionSup
HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key() ->
"org.apache.hudi.keygen.ComplexKeyGenerator",
KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.key() -> "true",
HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key() -> "false",
- HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key() ->
"org.apache.hudi.DefaultSparkRecordMerger"
+ HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key() ->
"org.apache.hudi.DefaultSparkRecordMerger",
+ HoodieCommonConfig.RECONCILE_SCHEMA.key() -> reconcileSchema.toString,
+ HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key() -> schemaOnRead.toString
)
df1.write.format("hudi").options(hudiOptions).mode(SaveMode.Append).save(basePath)
- // Try adding a string column. This operation is expected to throw 'schema
not compatible' exception since
- // 'MAKE_NEW_COLUMNS_NULLABLE' parameter is 'false' by default.
- val df2 = spark.sql("select '2' as event_id, '2' as ts, '3' as version,
'foo' as event_date, 'bar' as add_col")
- try {
-
(df2.write.format("hudi").options(hudiOptions).mode("append").save(basePath))
- fail("Option succeeded, but was expected to fail.")
- } catch {
- case ex: SchemaBackwardsCompatibilityException => {
-
assertTrue(ex.getMessage.contains(SchemaIncompatibilityType.READER_FIELD_MISSING_DEFAULT_VALUE.name()))
- }
- case ex: Exception => {
- fail(ex)
- }
- }
+ val df2 = spark.sql(
+ "select '2' as event_id, '2' as ts, '3' as version, "
+ + "named_struct('city', 'Berlin', 'country', 'DE') as address, '123'
as phone")
- // Try adding the string column again. This operation is expected to
succeed since 'MAKE_NEW_COLUMNS_NULLABLE'
- // parameter has been set to 'true'.
- hudiOptions = hudiOptions +
(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS.key() -> "true")
- try {
-
(df2.write.format("hudi").options(hudiOptions).mode("append").save(basePath))
- } catch {
- case ex: Exception => {
- fail(ex)
- }
+ df2.write.format("hudi").options(hudiOptions).mode("append").save(basePath)
+
+ val metaClient = createMetaClient(basePath)
+ val timeline =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants()
+ val commitMetadata =
timeline.readCommitMetadata(timeline.lastInstant().get())
+ val committedSchema =
HoodieSchema.parse(commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY))
+ val phoneField = committedSchema.getField("phone").get()
+ val countryField =
committedSchema.getNestedField("address.country").get().getRight
+ Seq(phoneField, countryField).foreach { field =>
+ assertTrue(field.isNullable)
+ assertTrue(field.hasDefaultValue)
+ assertEquals(HoodieSchema.NULL_VALUE, field.defaultVal().get())
}
+
+ val rows = spark.read.format("hudi").load(basePath)
+ .select("event_id", "phone", "address")
+ .orderBy("event_id")
+ .collect()
+ assertEquals(2, rows.length)
+ assertEquals(null, rows(0).getAs[String]("phone"))
+ assertEquals(null, rows(0).getAs[Row]("address").getAs[String]("country"))
+ assertEquals("123", rows(1).getAs[String]("phone"))
+ assertEquals("DE", rows(1).getAs[Row]("address").getAs[String]("country"))
}
def assertLastCommitIsUpsert(): Boolean = {
