You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/03/07 05:23:45 UTC
(hudi) branch master updated: [HUDI-6947] Refactored HoodieSchemaUtils.deduceWriterSchema with many flags (#10810)
This is an automated email from the ASF dual-hosted git repository.
yihua 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 2e39bfb6940 [HUDI-6947] Refactored HoodieSchemaUtils.deduceWriterSchema with many flags (#10810)
2e39bfb6940 is described below
commit 2e39bfb694099293b77eec9977e5e46af97af18b
Author: Geser Dugarov <ge...@gmail.com>
AuthorDate: Thu Mar 7 12:23:38 2024 +0700
[HUDI-6947] Refactored HoodieSchemaUtils.deduceWriterSchema with many flags (#10810)
---
.../scala/org/apache/hudi/HoodieSchemaUtils.scala | 176 +++++++++++----------
1 file changed, 93 insertions(+), 83 deletions(-)
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 cfc43453e9c..9aeff64f237 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
@@ -76,107 +76,117 @@ object HoodieSchemaUtils {
latestTableSchemaOpt: Option[Schema],
internalSchemaOpt: Option[InternalSchema],
opts: Map[String, String]): Schema = {
- val setNullForMissingColumns = opts.getOrDefault(DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.key(),
- DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.defaultValue).toBoolean
- val shouldReconcileSchema = opts.getOrDefault(DataSourceWriteOptions.RECONCILE_SCHEMA.key(),
- DataSourceWriteOptions.RECONCILE_SCHEMA.defaultValue().toString).toBoolean
- val shouldValidateSchemasCompatibility = opts.getOrDefault(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key,
- HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.defaultValue).toBoolean
-
latestTableSchemaOpt match {
- // In case table schema is empty we're just going to use the source schema as a
- // writer's schema.
+ // If table schema is empty, then we use the source schema as a writer's schema.
case None => AvroInternalSchemaConverter.fixNullOrdering(sourceSchema)
// Otherwise, we need to make sure we reconcile incoming and latest table schemas
case Some(latestTableSchemaWithMetaFields) =>
- // NOTE: Meta-fields will be unconditionally injected by Hudi writing handles, for the sake of
- // deducing proper writer schema we're stripping them to make sure we can perform proper
- // analysis
- //add call to fix null ordering to ensure backwards compatibility
+ // NOTE: Meta-fields will be unconditionally injected by Hudi writing handles, for the sake of deducing proper writer schema
+ // we're stripping them to make sure we can perform proper analysis
+ // add call to fix null ordering to ensure backwards compatibility
val latestTableSchema = AvroInternalSchemaConverter.fixNullOrdering(removeMetadataFields(latestTableSchemaWithMetaFields))
+
// Before validating whether schemas are compatible, we need to "canonicalize" source's schema
// relative to the table's one, by doing a (minor) reconciliation of the nullability constraints:
// for ex, if in incoming schema column A is designated as non-null, but it's designated as nullable
// in the table's one we want to proceed aligning nullability constraints w/ the table's schema
// Also, we promote types to the latest table schema if possible.
- val shouldCanonicalizeSchema = opts.getOrDefault(CANONICALIZE_SCHEMA.key,
- CANONICALIZE_SCHEMA.defaultValue.toString).toBoolean
- val mergeIntoWrites = opts.getOrDefault(SQL_MERGE_INTO_WRITES.key(),
- SQL_MERGE_INTO_WRITES.defaultValue.toString).toBoolean
-
+ val shouldCanonicalizeSchema = opts.getOrDefault(CANONICALIZE_SCHEMA.key, CANONICALIZE_SCHEMA.defaultValue.toString).toBoolean
val canonicalizedSourceSchema = if (shouldCanonicalizeSchema) {
canonicalizeSchema(sourceSchema, latestTableSchema, opts)
} else {
AvroInternalSchemaConverter.fixNullOrdering(sourceSchema)
}
- val allowAutoEvolutionColumnDrop = opts.getOrDefault(HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.key,
- HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.defaultValue).toBoolean
-
+ val shouldReconcileSchema = opts.getOrDefault(DataSourceWriteOptions.RECONCILE_SCHEMA.key(),
+ DataSourceWriteOptions.RECONCILE_SCHEMA.defaultValue().toString).toBoolean
if (shouldReconcileSchema) {
- internalSchemaOpt match {
- case Some(internalSchema) =>
- // Apply schema evolution, by auto-merging write schema and read schema
- val mergedInternalSchema = AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, internalSchema)
- val evolvedSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema, latestTableSchema.getFullName)
- val shouldRemoveMetaDataFromInternalSchema = sourceSchema.getFields().filter(f => f.name().equalsIgnoreCase(HoodieRecord.RECORD_KEY_METADATA_FIELD)).isEmpty
- if (shouldRemoveMetaDataFromInternalSchema) HoodieAvroUtils.removeMetadataFields(evolvedSchema) else evolvedSchema
- case None =>
- // In case schema reconciliation is enabled we will employ (legacy) reconciliation
- // strategy to produce target writer's schema (see definition below)
- val (reconciledSchema, isCompatible) =
- reconcileSchemasLegacy(latestTableSchema, canonicalizedSourceSchema)
-
- // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
- // w/ the table's one and allow schemas to diverge. This is required in cases where
- // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
- // only incoming dataset's projection has to match the table's schema, and not the whole one
- if (!shouldValidateSchemasCompatibility || isCompatible) {
- reconciledSchema
- } else {
- log.error(
- s"""Failed to reconcile incoming batch schema with the table's one.
- |Incoming schema ${sourceSchema.toString(true)}
- |Incoming schema (canonicalized) ${canonicalizedSourceSchema.toString(true)}
- |Table's schema ${latestTableSchema.toString(true)}
- |""".stripMargin)
- throw new SchemaCompatibilityException("Failed to reconcile incoming schema with the table's one")
- }
- }
+ deduceWriterSchemaWithReconcile(sourceSchema, canonicalizedSourceSchema, latestTableSchema, internalSchemaOpt, opts)
+ } else {
+ deduceWriterSchemaWithoutReconcile(sourceSchema, canonicalizedSourceSchema, latestTableSchema, opts)
+ }
+ }
+ }
+
+ /**
+ * Deducing with disabled reconciliation.
+ * We have to validate that the source's schema is compatible w/ the table's latest schema,
+ * such that we're able to read existing table's records using [[sourceSchema]].
+ */
+ private def deduceWriterSchemaWithoutReconcile(sourceSchema: Schema,
+ canonicalizedSourceSchema: Schema,
+ latestTableSchema: Schema,
+ opts: Map[String, String]): Schema = {
+ // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
+ // w/ the table's one and allow schemas to diverge. This is required in cases where
+ // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
+ // only incoming dataset's projection has to match the table's schema, and not the whole one
+ val mergeIntoWrites = opts.getOrDefault(SQL_MERGE_INTO_WRITES.key(), SQL_MERGE_INTO_WRITES.defaultValue.toString).toBoolean
+ val shouldValidateSchemasCompatibility = opts.getOrDefault(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key,
+ HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.defaultValue).toBoolean
+ val allowAutoEvolutionColumnDrop = opts.getOrDefault(HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.key,
+ HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.defaultValue).toBoolean
+ val setNullForMissingColumns = opts.getOrDefault(DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.key(),
+ DataSourceWriteOptions.SET_NULL_FOR_MISSING_COLUMNS.defaultValue).toBoolean
+
+ if (!mergeIntoWrites && !shouldValidateSchemasCompatibility && !allowAutoEvolutionColumnDrop) {
+ // Default behaviour
+ val reconciledSchema = if (setNullForMissingColumns) {
+ AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, latestTableSchema)
+ } else {
+ canonicalizedSourceSchema
+ }
+ checkValidEvolution(reconciledSchema, latestTableSchema)
+ reconciledSchema
+ } else {
+ // If it's merge into writes, we don't check for projection nor schema compatibility. Writers down the line will take care of it.
+ // Or it's not merge into writes, and we don't validate schema, but we allow to drop columns automatically.
+ // Or it's not merge into writes, we validate schema, and schema is compatible.
+ if (shouldValidateSchemasCompatibility) {
+ checkSchemaCompatible(latestTableSchema, canonicalizedSourceSchema, true,
+ allowAutoEvolutionColumnDrop, java.util.Collections.emptySet())
+ }
+ canonicalizedSourceSchema
+ }
+ }
+
+ /**
+ * Deducing with enabled reconciliation.
+ * Marked as Deprecated.
+ */
+ private def deduceWriterSchemaWithReconcile(sourceSchema: Schema,
+ canonicalizedSourceSchema: Schema,
+ latestTableSchema: Schema,
+ internalSchemaOpt: Option[InternalSchema],
+ opts: Map[String, String]): Schema = {
+ internalSchemaOpt match {
+ case Some(internalSchema) =>
+ // Apply schema evolution, by auto-merging write schema and read schema
+ val mergedInternalSchema = AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, internalSchema)
+ val evolvedSchema = AvroInternalSchemaConverter.convert(mergedInternalSchema, latestTableSchema.getFullName)
+ val shouldRemoveMetaDataFromInternalSchema = sourceSchema.getFields().filter(f => f.name().equalsIgnoreCase(HoodieRecord.RECORD_KEY_METADATA_FIELD)).isEmpty
+ if (shouldRemoveMetaDataFromInternalSchema) HoodieAvroUtils.removeMetadataFields(evolvedSchema) else evolvedSchema
+ case None =>
+ // In case schema reconciliation is enabled we will employ (legacy) reconciliation
+ // strategy to produce target writer's schema (see definition below)
+ val (reconciledSchema, isCompatible) =
+ reconcileSchemasLegacy(latestTableSchema, canonicalizedSourceSchema)
+
+ // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
+ // w/ the table's one and allow schemas to diverge. This is required in cases where
+ // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
+ // only incoming dataset's projection has to match the table's schema, and not the whole one
+ val shouldValidateSchemasCompatibility = opts.getOrDefault(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key, HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.defaultValue).toBoolean
+ if (!shouldValidateSchemasCompatibility || isCompatible) {
+ reconciledSchema
} else {
- // In case reconciliation is disabled, we have to validate that the source's schema
- // is compatible w/ the table's latest schema, such that we're able to read existing table's
- // records using [[sourceSchema]].
- //
- // NOTE: In some cases we need to relax constraint of incoming dataset's schema to be compatible
- // w/ the table's one and allow schemas to diverge. This is required in cases where
- // partial updates will be performed (for ex, `MERGE INTO` Spark SQL statement) and as such
- // only incoming dataset's projection has to match the table's schema, and not the whole one
-
- if (mergeIntoWrites) {
- // if its merge into writes, do not check for projection nor schema compatibility. Writers down the line will
- // take care of it.
- canonicalizedSourceSchema
- } else {
- if (!shouldValidateSchemasCompatibility) {
- // if no validation is enabled, check for col drop
- if (allowAutoEvolutionColumnDrop) {
- canonicalizedSourceSchema
- } else {
- val reconciledSchema = if (setNullForMissingColumns) {
- AvroSchemaEvolutionUtils.reconcileSchema(canonicalizedSourceSchema, latestTableSchema)
- } else {
- canonicalizedSourceSchema
- }
- checkValidEvolution(reconciledSchema, latestTableSchema)
- reconciledSchema
- }
- } else {
- checkSchemaCompatible(latestTableSchema, canonicalizedSourceSchema, true,
- allowAutoEvolutionColumnDrop, java.util.Collections.emptySet())
- canonicalizedSourceSchema
- }
- }
+ log.error(
+ s"""Failed to reconcile incoming batch schema with the table's one.
+ |Incoming schema ${sourceSchema.toString(true)}
+ |Incoming schema (canonicalized) ${canonicalizedSourceSchema.toString(true)}
+ |Table's schema ${latestTableSchema.toString(true)}
+ |""".stripMargin)
+ throw new SchemaCompatibilityException("Failed to reconcile incoming schema with the table's one")
}
}
}