AnishMahto commented on code in PR #58465:
URL: https://github.com/apache/spark/pull/58465#discussion_r3938160422
##########
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,
Review Comment:
Alright so I dropped the nullability enforcement from SDP. A couple things:
- I tested locally with Iceberg, and the Iceberg catalog desirably throws
with an `Incompatible change: cannot add required column: <column added without
nullable>` error when the DSv2 updates are sent via `alterTable`.
- I decided to defer support for specifying/changing column default values
in `diffSchemas` for now, I realized its complex enough that it deserves it's
own PR. I filed two relevant bug tickets
(https://issues.apache.org/jira/browse/SPARK-59268,
https://issues.apache.org/jira/browse/SPARK-59269) instead. Note that both of
these are gaps that already existed prior to this change; we're not creating
any new regression.
--
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]
