HyukjinKwon commented on code in PR #59299:
URL: https://github.com/apache/spark/pull/59299#discussion_r4236575504
##########
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:
**Nit (P3):** `isAlreadyAdded` matches the sibling case-insensitively, but
the anchor emitted here is the source schema's spelling, not the name the
earlier `AddColumn` was created with. With the default case-insensitive
analysis and the flag on, `UPDATE SET info.BONUS = s.info.BONUS, info.extra =
s.info.extra` (source `info<salary, bonus, extra, status>`, target
`info<salary, status>`) produces `ADD info.BONUS AFTER salary` and then `ADD
info.extra AFTER bonus`. `CatalogV2Util.addField` looks up `bonus` by exact
name, finds nothing, and the MERGE fails during schema evolution. The same
thing happens when `info.BONUS` is added by a leaf assignment and `INSERT *`
contributes `info.extra`. ALTER TABLE ADD COLUMNS avoids this by normalizing
AFTER to the recorded column name (`Analyzer.scala` `resolvePosition`).
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** This lands between `coerceMergeNestedTypes` and
`coerceInsertNestedTypes`, splitting that pair, while
`SCHEMA_EVOLUTION_PRESERVE_COLUMN_ORDER` is declared after both of their
entries. Moving it below `coerceInsertNestedTypes` keeps the accessors in
declaration order.
**Verification:**
- **Inspection:** The accessor order mirrors the declaration order of the
three related entries.
##########
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:
**Nit (P3):** The grammar only accepts the clause right after the verb:
`INSERT [WITH SCHEMA EVOLUTION] INTO ...` and `MERGE [WITH SCHEMA EVOLUTION]
INTO ...` (SqlBaseParser.g4). Suggest `(MERGE WITH SCHEMA EVOLUTION INTO or
INSERT WITH SCHEMA EVOLUTION INTO)`.
**Verification:**
- **Inspection:** The quoted statement forms match the INSERT and MERGE WITH
SCHEMA EVOLUTION grammar alternatives.
##########
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:
**Nit (P3):** With `.internal()`, this key is hidden from `SET -v` and the
generated configuration docs, and per `ConfigEntry` it should not be exposed to
users. Yet `docs/sql-ref-syntax-dml-merge-into.md` and
`docs/sql-ref-syntax-dml-insert-table.md` now tell users to set it, and the PR
description lists it as a user-facing change. The peer schema-evolution knobs
next to it are internal and undocumented. Please either drop `.internal()` (if
this is meant to be a supported setting) or remove the SQL-reference sentences
while it stays experimental.
**Recommended change:** Remove .internal() from
SCHEMA_EVOLUTION_PRESERVE_COLUMN_ORDER so the setting documented in the SQL
reference is a public, listed configuration, matching the PR's stated
user-facing change.
**Why this works:** Making the entry public puts it in the generated config
docs and SET -v and gives it the stability expectations implied by the SQL
reference.
**Scope:** Declare the new schema-evolution ordering config public so its
documented use is supported.
**Compatibility:** Preserve behavior outside the stated repair boundary.
**Risks:** A public key becomes a compatibility commitment for its name and
semantics.
**Constraints:** Default stays false.
**Success:** spark.sql.schemaEvolution.preserveColumnOrder is listed as a
public SQL config, consistent with the SQL reference pages that mention it.
**Verification:**
- **Inspection:** The config entry is public and its key and default match
the SQL-reference text.
##########
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:
**Nit (P3):** `getOrElse(previous)` anchors a new field on its preceding new
sibling on the assumption that the sibling's `AddColumn` will be applied.
`filterSupportedChanges` runs afterwards, though, and `SupportsSchemaEvolution`
explicitly lets a connector skip individual changes. With target `(id, data)`,
source `(id, v, b, data)` and a connector that rejects the add for `v`,
alterTable receives `ADD b AFTER v` and fails on the missing anchor, instead of
the clearer error about `v` that the flag-off path produces.
`MergeIntoTable.pendingSchemaChanges` has the same ordering. Re-anchoring on
the surviving changes after filtering would keep the batch consistent.
See **Shared repair plan 1** in the review body.
--
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]
