AnishMahto commented on code in PR #58465:
URL: https://github.com/apache/spark/pull/58465#discussion_r3920447140
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala:
##########
@@ -227,59 +231,160 @@ object SchemaInferenceUtils {
* @param targetSchema The target schema that we want the table to have
* @return A sequence of TableChange objects representing the necessary
changes
*/
- def diffSchemas(currentSchema: StructType, targetSchema: StructType):
Seq[TableChange] = {
- val changes = scala.collection.mutable.ArrayBuffer.empty[TableChange]
+ def diffSchemas(currentSchema: StructType, targetSchema: StructType):
Seq[TableChange] =
+ diffStructs(
+ currentStruct = currentSchema,
+ targetStruct = targetSchema,
+ // Root call: path is empty because current and target are the top-level
schemas.
+ pathToStruct = Seq.empty
+ )
- // Helper function to get a map of field name to field
- def getFieldMap(schema: StructType): Map[String, StructField] = {
- schema.fields.map(field => field.name -> field).toMap
- }
+ /**
+ * Diffs two structs field-by-field, matching fields by exact name.
+ *
+ * @param currentStruct The struct as it exists in the current schema.
+ * @param targetStruct The struct as it should look in the target schema.
+ * @param pathToStruct Path segments from the top-level schema to this
+ * struct, if this is a nested struct. Empty for the
+ * root call.
+ */
+ private def diffStructs(
Review Comment:
Exact key matches (case-sensitive) is maintained behavior from the previous
implementation too, and the scaladoc section that discusses it for
incremental/full refresh cases is carried as-is.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala:
##########
@@ -227,59 +231,160 @@ object SchemaInferenceUtils {
* @param targetSchema The target schema that we want the table to have
* @return A sequence of TableChange objects representing the necessary
changes
*/
- def diffSchemas(currentSchema: StructType, targetSchema: StructType):
Seq[TableChange] = {
- val changes = scala.collection.mutable.ArrayBuffer.empty[TableChange]
+ def diffSchemas(currentSchema: StructType, targetSchema: StructType):
Seq[TableChange] =
+ diffStructs(
+ currentStruct = currentSchema,
+ targetStruct = targetSchema,
+ // Root call: path is empty because current and target are the top-level
schemas.
+ pathToStruct = Seq.empty
+ )
- // Helper function to get a map of field name to field
- def getFieldMap(schema: StructType): Map[String, StructField] = {
- schema.fields.map(field => field.name -> field).toMap
- }
+ /**
+ * Diffs two structs field-by-field, matching fields by exact name.
+ *
+ * @param currentStruct The struct as it exists in the current schema.
+ * @param targetStruct The struct as it should look in the target schema.
+ * @param pathToStruct Path segments from the top-level schema to this
+ * struct, if this is a nested struct. Empty for the
+ * root call.
+ */
+ private def diffStructs(
Review Comment:
As mentioned in the PR description, field order is intentionally not
considered when diffing two structs.
It's fair to argue that it should, but that would be out of scope for this
PR; I'm choosing to keep previous behavior, albeit now it applies recursively
to nested struct comparisons too.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala:
##########
@@ -227,59 +231,160 @@ object SchemaInferenceUtils {
* @param targetSchema The target schema that we want the table to have
* @return A sequence of TableChange objects representing the necessary
changes
*/
- def diffSchemas(currentSchema: StructType, targetSchema: StructType):
Seq[TableChange] = {
- val changes = scala.collection.mutable.ArrayBuffer.empty[TableChange]
+ def diffSchemas(currentSchema: StructType, targetSchema: StructType):
Seq[TableChange] =
+ diffStructs(
+ currentStruct = currentSchema,
+ targetStruct = targetSchema,
+ // Root call: path is empty because current and target are the top-level
schemas.
+ pathToStruct = Seq.empty
+ )
- // Helper function to get a map of field name to field
- def getFieldMap(schema: StructType): Map[String, StructField] = {
- schema.fields.map(field => field.name -> field).toMap
- }
+ /**
+ * Diffs two structs field-by-field, matching fields by exact name.
+ *
+ * @param currentStruct The struct as it exists in the current schema.
+ * @param targetStruct The struct as it should look in the target schema.
+ * @param pathToStruct Path segments from the top-level schema to this
+ * struct, if this is a nested struct. Empty for the
+ * root call.
+ */
+ private def diffStructs(
+ currentStruct: StructType,
+ targetStruct: StructType,
+ pathToStruct: Seq[String]): Seq[TableChange] = {
+ val topLevelFieldsInCurrent = currentStruct.fields.map(field => field.name
-> field).toMap
+ val topLevelFieldsInTarget = targetStruct.fields.map(field => field.name
-> field).toMap
+
+ // Fields present in target but not in current are columns that need to be
added.
+ val columnsAdded = topLevelFieldsInTarget.values.toSeq
+ .filterNot(fieldInTarget =>
+ topLevelFieldsInCurrent.contains(fieldInTarget.name)
+ )
+ .map { fieldInTarget =>
+ TableChange.addColumn(
+ (pathToStruct :+ fieldInTarget.name).toArray,
+ fieldInTarget.dataType,
+ fieldInTarget.nullable,
+ fieldInTarget.getComment().orNull
+ )
+ }
- val currentFields = getFieldMap(currentSchema)
- val targetFields = getFieldMap(targetSchema)
-
- // Find columns to add (in target but not in current)
- val columnsToAdd = targetFields.keySet.diff(currentFields.keySet)
- columnsToAdd.foreach { columnName =>
- val field = targetFields(columnName)
- changes += TableChange.addColumn(
- Array(columnName),
- field.dataType,
- field.nullable,
- field.getComment().orNull
+ // Fields present in current but not in target are columns that need to be
removed.
+ val columnsDeleted = topLevelFieldsInCurrent.values.toSeq
+ .filterNot(fieldInCurrent =>
+ topLevelFieldsInTarget.contains(fieldInCurrent.name)
)
+ .map(fieldInCurrent =>
+ TableChange
+ .deleteColumn(
+ (pathToStruct :+ fieldInCurrent.name).toArray,
+ false
+ )
+ )
+
+ // Fields in both current and target but vary in metadata or nested
sub-fields represent
+ // columns that need to be updated.
+ val columnsUpdated = topLevelFieldsInCurrent.values.toSeq.flatMap {
+ fieldInCurrent =>
+ topLevelFieldsInTarget.get(fieldInCurrent.name).toSeq.flatMap {
+ fieldInTarget =>
+ diffField(
+ currentField = fieldInCurrent,
+ targetField = fieldInTarget,
+ pathToField = pathToStruct :+ fieldInCurrent.name
+ )
+ }
}
- // Find columns to delete (in current but not in target)
- val columnsToDelete = currentFields.keySet.diff(targetFields.keySet)
- columnsToDelete.foreach { columnName =>
- changes += TableChange.deleteColumn(Array(columnName), false)
+ columnsAdded ++ columnsDeleted ++ columnsUpdated
+ }
+
+ /**
+ * Diffs the type, nullability, and comment of one field present in both
schemas. Other
+ * StructField.metadata entries (defaults, generated-column expressions,
connector-specific
+ * metadata) are not diffed: pipeline schema synchronization does not
support propagating
+ * them, and Spark's own ResolveSchemaEvolution likewise ignores them.
+ */
+ private def diffField(
+ currentField: StructField,
+ targetField: StructField,
+ pathToField: Seq[String]): Seq[TableChange] = {
+ warnOnFieldMetadataDrift(currentField, targetField, pathToField)
+ diffDataTypes(currentField.dataType, targetField.dataType, pathToField) ++
+ diffNullability(currentField.nullable, targetField.nullable,
pathToField) ++
+ diffComment(currentField.getComment(), targetField.getComment(),
pathToField)
+ }
+
+ /**
+ * Logs a warning when two fields' metadata bags differ beyond the "comment"
key
+ * (which is already handled by [[diffComment]]). Pipeline schema
synchronization does not
+ * support propagating other metadata entries (defaults, generated-column
expressions,
+ * connector-specific metadata), so these differences are left for the user
to reconcile.
+ */
+ private def warnOnFieldMetadataDrift(
+ currentField: StructField,
+ targetField: StructField,
+ pathToField: Seq[String]): Unit = {
+ val current = stripMetadataComment(currentField.metadata)
+ val target = stripMetadataComment(targetField.metadata)
+ if (current != target) {
+ logWarning(
+ s"Field ${pathToField.mkString(".")} has metadata changes that
pipeline schema " +
+ s"synchronization does not propagate and will be ignored. " +
+ s"Current: ${current.json}, Target: ${target.json}")
}
+ }
- // Find columns with type changes (in both but with different types)
- val commonColumns = currentFields.keySet.intersect(targetFields.keySet)
- commonColumns.foreach { columnName =>
- val currentField = currentFields(columnName)
- val targetField = targetFields(columnName)
+ private def stripMetadataComment(m: Metadata): Metadata =
+ new MetadataBuilder().withMetadata(m).remove("comment").build()
- // If data types are different, add a type update change
- if (currentField.dataType != targetField.dataType) {
- changes += TableChange.updateColumnType(Array(columnName),
targetField.dataType)
- }
+ private def diffNullability(
+ currentNullable: Boolean,
+ targetNullable: Boolean,
+ pathToField: Seq[String]
+ ): Option[TableChange] = {
+ Option.when(currentNullable != targetNullable)(
+ TableChange.updateColumnNullability(pathToField.toArray, targetNullable)
+ )
+ }
- // If nullability is different, add a nullability update change
- if (currentField.nullable != targetField.nullable) {
- changes += TableChange.updateColumnNullability(Array(columnName),
targetField.nullable)
- }
+ private def diffComment(
+ currentComment: Option[String],
+ targetComment: Option[String],
+ pathToField: Seq[String]
+ ): Option[TableChange] = {
+ Option.when(currentComment != targetComment)(
+ TableChange.updateColumnComment(pathToField.toArray,
targetComment.orNull)
+ )
+ }
- // If comments are different, add a comment update change
- val currentComment = currentField.getComment().orNull
- val targetComment = targetField.getComment().orNull
- if (currentComment != targetComment) {
- changes += TableChange.updateColumnComment(Array(columnName),
targetComment)
- }
- }
+ /** Diffs two data types at `path`, descending through matching complex
types. */
+ private def diffDataTypes(
+ currentType: DataType,
+ targetType: DataType,
+ pathToField: Seq[String]
+ ): Seq[TableChange] = (currentType, targetType) match {
+ case (currentStruct: StructType, targetStruct: StructType) =>
+ diffStructs(currentStruct, targetStruct, pathToField)
+
+ case (currentArray: ArrayType, targetArray: ArrayType) =>
Review Comment:
Btw turns out DSv2 doesn't support `TableChanges` to an array or map nested
into another array/map. But I believe this is a bug with DSv2 and not how we
construct the nested path here.
Filed a spark issue at https://issues.apache.org/jira/browse/SPARK-59188
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
