imonteroq commented on code in PR #59299:
URL: https://github.com/apache/spark/pull/59299#discussion_r4258517959
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala:
##########
@@ -239,20 +253,53 @@ object ResolveSchemaEvolution extends Rule[LogicalPlan] {
newFieldMap(f.name).dataType,
fieldPath :+ f.name,
isByName = true,
- throwError)
+ throwError,
+ preserveColumnOrder)
}
// Collect newly added fields
- val adds = newFields
- .filterNot(f => currentFieldMap.contains(f.name))
- .map { f =>
- // Make the type nullable, since existing rows in the table will have
NULLs for this column.
- TableChange.addColumn((fieldPath :+ f.name).toArray,
f.dataType.asNullable)
+ val adds = newFields.zipWithIndex
+ .collect {
+ case (f, index) if !currentFieldMap.contains(f.name) =>
+ // Make the type nullable, since existing rows in the table will
have NULLs
+ // for this column.
+ TableChange.addColumn(
+ (fieldPath :+ f.name).toArray,
+ f.dataType.asNullable,
+ /* isNullable */ true,
+ /* comment */ null,
+ if (preserveColumnOrder) {
+ addedFieldPosition(currentFieldMap, newFields, index)
+ } else {
+ null
+ },
+ /* defaultValue */ null)
}
updates ++ adds
}
+ /**
+ * Returns the position of a field added at the given index among the source
schema's
+ * fields: [[ColumnPosition.first]] when the field is the first of its
siblings, or
+ * [[ColumnPosition.after]] the preceding source sibling. A preceding
sibling that
+ * already exists in the target schema is anchored by its target field name,
while a
+ * preceding sibling that is itself a new field is anchored by its source
name, since
+ * additions are emitted in source order.
+ */
+ private def addedFieldPosition(
+ currentFieldMap: Map[String, StructField],
+ newFields: Seq[StructField],
+ index: Int): TableChange.ColumnPosition = {
+ if (index == 0) {
+ TableChange.ColumnPosition.first()
+ } else {
+ val previous = newFields(index - 1).name
+ TableChange.ColumnPosition.after(
+ currentFieldMap.get(previous).map(_.name).getOrElse(previous))
Review Comment:
Same root cause, adopted the repair plan. `assignAddPositions` now runs
after `filterSupportedChanges`, anchoring on the nearest surviving accepted
add, so a rejected `v` is skipped as an anchor and `b` lands on `id`. Both the
`INSERT` and `MERGE` paths go through it. Added a `reject-added-columns`
property to the in-memory catalog plus tests on both paths.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -8751,6 +8751,18 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val SCHEMA_EVOLUTION_PRESERVE_COLUMN_ORDER =
+ buildConf("spark.sql.schemaEvolution.preserveColumnOrder")
+ .internal()
Review Comment:
Agreed, dropped `.internal()`, the flag is meant to be user-facing.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -8751,6 +8751,18 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val SCHEMA_EVOLUTION_PRESERVE_COLUMN_ORDER =
+ buildConf("spark.sql.schemaEvolution.preserveColumnOrder")
+ .internal()
+ .doc("When true, fields added to a table by schema evolution (MERGE INTO
or " +
+ "INSERT INTO ... WITH SCHEMA EVOLUTION) keep the position they occupy
in the " +
Review Comment:
Fixed, the doc now says `MERGE`/`INSERT WITH SCHEMA EVOLUTION INTO`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -10247,6 +10259,9 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def coerceMergeNestedTypes: Boolean =
getConf(SQLConf.MERGE_INTO_NESTED_TYPE_COERCION_ENABLED)
+ def schemaEvolutionPreserveColumnOrder: Boolean =
Review Comment:
Moved below `coerceInsertNestedTypes`, accessors now mirror declaration
order.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSchemaEvolution.scala:
##########
@@ -113,7 +119,12 @@ object ResolveSchemaEvolution extends Rule[LogicalPlan] {
targetTable.schema,
sourceSchema,
isByName)
- filterSupportedChanges(targetTable, candidateChanges)
+ val supportedChanges = filterSupportedChanges(targetTable,
candidateChanges)
+ if (isByName && conf.schemaEvolutionPreserveColumnOrder) {
Review Comment:
Added `preserve order - by position appends extra columns`: target `(id,
data)`, by-position source `(a, b, c)` with the flag on asserts `(id, data,
c)`. Fails as `(c, id, data)` if the guard is dropped, exactly the regression
described.
--
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]
