imonteroq commented on code in PR #59299:
URL: https://github.com/apache/spark/pull/59299#discussion_r4258495380
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala:
##########
@@ -1279,25 +1281,100 @@ object MergeIntoTable {
}
private def computePendingSchemaChanges(merge: MergeIntoTable):
Seq[TableChange] = {
+ val preserveColumnOrder = SQLConf.get.schemaEvolutionPreserveColumnOrder
+ val resolver = SQLConf.get.resolver
+
+ // Field paths already scheduled for addition: they anchor positioned adds
that
+ // follow a source sibling which is itself being added, and are skipped as
repeats.
+ val addedFieldPaths = mutable.ArrayBuffer.empty[Seq[String]]
+
+ def isAlreadyAdded(fieldPath: Seq[String]): Boolean = {
+ addedFieldPaths.exists { added =>
+ added.length == fieldPath.length &&
+ added.lazyZip(fieldPath).forall((a, b) => resolver(a, b))
+ }
+ }
+
+ // Positions the new field as in the source: after the nearest preceding
source
+ // sibling that exists in the target or is itself added earlier, or first
when
+ // none exists. Returns null when the source position cannot be determined.
+ def addedColumnPosition(fieldPath: Seq[String]):
TableChange.ColumnPosition = {
+ val parentPath = fieldPath.init
+
+ def siblingFields(schema: StructType): Seq[StructField] = {
+ if (parentPath.isEmpty) {
+ schema.fields.toImmutableArraySeq
+ } else {
+ schema
+ .findNestedField(parentPath, includeCollections = true, resolver =
resolver)
+ .map(_._2.dataType)
+ .collect { case struct: StructType =>
struct.fields.toImmutableArraySeq }
+ .getOrElse(Seq.empty)
+ }
+ }
+
+ val sourceSiblings = siblingFields(merge.sourceTable.schema)
+ val targetSiblings = siblingFields(merge.targetTable.schema)
+ val index = sourceSiblings.indexWhere(f => resolver(f.name,
fieldPath.last))
+ if (index < 0) {
+ null
+ } else {
+ sourceSiblings.take(index).reverseIterator
+ .map { sibling =>
+ targetSiblings.find(f => resolver(f.name, sibling.name))
+ .map(f => TableChange.ColumnPosition.after(f.name))
+ .orElse(Option.when(isAlreadyAdded(parentPath :+ sibling.name)) {
+ TableChange.ColumnPosition.after(sibling.name)
Review Comment:
Verified locally with your exact scenario. Fixed by moving positioning into
a shared post-pass (`assignAddPositions` ) that runs after dedup. Anchors now
carry the emitted field name of the surviving add, so `info.BONUS`/`info.bonus`
resolves to the same column instead of leaving a dangling AFTER bonus. Added a
case-variant test covering leaf + `INSERT *`.
--
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]
