You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2026/09/16 16:49:29 UTC
(spark) branch master updated: [SPARK-59015][SQL] Rebind refreshed DSv2 relations to the current table schema
This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 66a28acd5b1c [SPARK-59015][SQL] Rebind refreshed DSv2 relations to the current table schema
66a28acd5b1c is described below
commit 66a28acd5b1cf34458de74f5ff6703a4851ad751
Author: yyanyy <yy...@gmail.com>
AuthorDate: Wed Sep 16 09:49:17 2026 -0700
[SPARK-59015][SQL] Rebind refreshed DSv2 relations to the current table schema
### What changes were proposed in this pull request?
When `V2TableRefreshUtil` refreshes a `DataSourceV2Relation` in `ALLOW_NEW_FIELDS` mode, the relation
is now replaced with a `Project` over a relation that exposes the current table schema: the relation
stays aligned with the physical scan, while the projection recreates the captured output — names,
expression IDs, types, nullability and metadata — for the already-analyzed parent plan. A new
`CapturedSchemaProjection` builds it, recursing through structs, arrays and maps, matching data columns
by name and metadata columns by their logical name. When the captured and current outputs already
match, the relation is returned unchanged, so an unchanged table keeps its plan and the rewrite is
idempotent under `transformDown`.
One shape cannot be recreated, and this PR makes it fail rather than answer wrongly: a `MAP` whose
**key** type gained a field. Keys that are distinct under the current key type can be equal under the
captured one, so narrowing the key back changes how many entries the map has. The rebuild goes through
Spark's map builder, which enforces key uniqueness, so the query raises `DUPLICATED_MAP_KEY`. Details
under user-facing changes.
### Why are the changes needed?
Spark refreshes versioned DSv2 tables between analysis and optimization. `V2TableRefreshUtil` does
`r.copy(table = currentTable)`, deliberately swapping in the current table while keeping the analyzed
output, which leaves:
```
relation.output = [id, salary] // captured at analysis
scan data schema = [id, salary, new_column] // current snapshot
```
`PushDownUtils.pruneColumns` then calls `toOutputAttrs(scan.readSchema(), relation)`, which builds its
name map only from `relation.output`, so any field the captured output does not know about throws
`java.util.NoSuchElementException: key not found: new_column`. The trigger is precisely that
**`Scan.readSchema()` field names are not a subset of `relation.output` field names** — not merely that
`readSchema` is wider than the query's project.
Reporting more than was requested is allowed: `SupportsPushDownRequiredColumns.pruneColumns` permits
partial pruning, its example being a source that only prunes top-level columns. That example alone
reproduces the failure:
```scala
sql("CREATE TABLE t (id INT, person STRUCT<name: STRING>) USING ...")
val df = spark.table("t") // captures person: STRUCT<name>
sql("ALTER TABLE t ADD COLUMN person.age INT FIRST")
df.collect()
```
```
[PLAN_VALIDATION_FAILED_RULE_IN_BATCH] ... The plan output schema has changed from
STRUCT<id: INT, person: STRUCT<name: STRING>> to
STRUCT<id: INT, person: STRUCT<age: INT, name: STRING>>
```
A one-shot query analyzes and optimizes back to back, so this window is normally invisible; it becomes
deterministic by forcing analysis on a `DataFrame` before changing the table, or via a temp view
created from a `Dataset`. This is not Connect-specific — the refresh phase lives in common
`QueryExecution`.
### Does this PR introduce _any_ user-facing change?
Yes. Refresh exists so an already-analyzed plan runs against the table's *current* schema, so where the
captured output can still be reproduced the query now succeeds, and where it cannot the query fails
instead of improvising an answer.
- **Succeeds where it used to fail.** Queries against a partially-pruning source, run after the table
gained a column, previously failed with `NoSuchElementException` or
`PLAN_VALIDATION_FAILED_RULE_IN_BATCH`. They now return the captured schema.
- **Fails where it used to succeed: a map key that gained a field.** The one compatibility break, and
it is intended. With a captured `MAP<STRUCT<name: STRING>, INT>`:
```sql
ALTER TABLE t ADD COLUMN m.key.id INT;
INSERT INTO t SELECT map(named_struct('name', 'a', 'id', 1), 10,
named_struct('name', 'a', 'id', 2), 20);
```
Both keys are distinct now, but both are `{name: 'a'}` in the captured schema, so the map cannot hold
two entries there. Previously the scan was asked to narrow the key itself and returned a map holding
duplicate keys, which Spark propagated unchecked: `size(m)` reported both entries while collecting
the row kept only the last. The projection now rebuilds the captured key through Spark's map builder,
so the query fails with `DUPLICATED_MAP_KEY` — or keeps the last entry under
`spark.sql.mapKeyDedupPolicy=LAST_WIN`. Narrowing a map key is a fold rather than a projection and is
not something a scan can be asked to perform; a captured key type that cannot represent the current
key set has no correct answer to return.
- **No change when the table has not changed** — the outputs match, rebinding is skipped, and the plan
is identical to before.
### How was this patch tested?
New `CapturedSchemaProjectionSuite` (27 tests) covers the projection at the expression level: nested
struct/array/map shapes with a field added at every level and a different ordinal shift per level; an
unchanged `VARIANT` leaf, which must be carried by reference rather than rebuilt; nulls; resolver-based
name matching; captured metadata columns; captured field metadata surviving alias cleanup; a map key
that gained a field under both `spark.sql.mapKeyDedupPolicy` values; and every rejection path.
`DataSourceV2DataFrameSuite` gains 9 end-to-end tests that run an already-analyzed plan across a table
change, mostly against a source that reports every data column from `readSchema()`: flat and nested
widening, a column added first, an external case-only rename, two metadata-column cases, a self-join,
and the map key above. Disabling the rebinding call in `V2TableRefreshUtil` fails 8 of the 9.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code 2.1.268
Closes #58298 from yyanyy/spark-dsv2-schema-rebind-20260821.
Authored-by: yyanyy <yy...@gmail.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
docs/sql-migration-guide.md | 2 +
.../spark/sql/connector/catalog/V2TableUtil.scala | 4 +-
.../org/apache/spark/sql/util/SchemaUtils.scala | 72 +-
.../sql/connector/catalog/InMemoryBaseTable.scala | 25 +-
.../datasources/v2/AnalyzedSchemaProjection.scala | 356 +++++++
.../datasources/v2/V2TableRefreshUtil.scala | 7 +-
.../sql/connector/DataSourceV2DataFrameSuite.scala | 565 ++++++++++-
.../v2/AnalyzedSchemaProjectionSuite.scala | 1025 ++++++++++++++++++++
8 files changed, 2034 insertions(+), 22 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 87d8c04c758b..dfe63fc83e4f 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -24,6 +24,8 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed after a compatible schema change, connectors can return more data columns from the current table schema in `Scan.readSchema()` than requested by `SupportsPushDownRequiredColumns.pruneColumns`. Previously, this partial pruning could fail planning because the scan reported columns absent from the analyzed relation output.
+- Since Spark 4.4, when an already-analyzed Data Source V2 query is refreshed, if rebinding an expanded struct used as a map key causes distinct current keys to collide under the analyzed schema, the query fails with `DUPLICATED_MAP_KEY` by default instead of returning duplicate keys. With `spark.sql.mapKeyDedupPolicy=LAST_WIN`, the last value is kept.
- Since Spark 4.4, the optimizer rule `SimplifyCaseConversionExpressions` no longer simplifies mixed case conversions `LOWER(UPPER(str))` to `LOWER(str)` or `UPPER(LOWER(str))` to `UPPER(str)`. This preserves standard Unicode case mapping semantics for non-ASCII characters such as Latin small letter dotless i (`ı` U+0131) and micro sign (`µ` U+00B5).
- Since Spark 4.4, the Oracle JDBC datasource maps Oracle DATE and TIMESTAMP to TimestampNTZType unconditionally, ignoring the JDBC read option `preferTimestampNTZ` (DATE only when the driver default `oracle.jdbc.mapDateToTimestamp=true` surfaces it as TIMESTAMP; otherwise DATE stays DateType). In 4.3 and earlier these were read as TimestampType, or as TimestampNTZType only when `preferTimestampNTZ=true`. This faithfully represents these zoneless Oracle types. Unaffected: Oracle DATE rea [...]
- Since Spark 4.4, a nondeterministic common expression that a conditional branch reads more than once is evaluated once, and every read sees that one value. `BETWEEN` and `NULLIF` read their input twice, so an expression such as `CASE WHEN c THEN monotonically_increasing_id() BETWEEN 3 AND 5 END` previously gave each read its own value and could return a different result. Queries whose results depended on the earlier behaviour change, and an optimized or physical plan may now show a `wi [...]
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V2TableUtil.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V2TableUtil.scala
index 2b8ac6404ea6..0da7550b0c26 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V2TableUtil.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/V2TableUtil.scala
@@ -17,8 +17,6 @@
package org.apache.spark.sql.connector.catalog
-import java.util.Locale
-
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.MetadataAttributeWithLogicalName
@@ -198,7 +196,7 @@ private[sql] object V2TableUtil extends SQLConfHelper {
}
private def normalize(name: String): String = {
- if (conf.caseSensitiveAnalysis) name else name.toLowerCase(Locale.ROOT)
+ SchemaUtils.foldName(name, conf.caseSensitiveAnalysis)
}
private def resolver: Resolver = conf.resolver
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala
index 61c6b1397b9d..5fad61badc8f 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/util/SchemaUtils.scala
@@ -443,12 +443,13 @@ private[spark] object SchemaUtils {
(dataType, otherDataType) match {
case (StructType(fields), StructType(otherFields)) =>
- val fieldsByName = index(fields, resolver)
- val otherFieldsByName = index(otherFields, resolver)
+ val otherCandidates = index(otherFields, resolver)
- fieldsByName.foreach { case (normalizedName, field) =>
- otherFieldsByName.get(normalizedName) match {
- case Some(otherField) =>
+ // Iterates the fields rather than the index so that every field is reported even when two
+ // of them fold to one key, and so that the errors come out in schema order.
+ fields.foreach { field =>
+ matchField(otherCandidates, field.name, resolver) match {
+ case Seq(otherField) =>
val nameParts = colPath :+ field.name
if (checkFieldIds) {
for (id <- field.id; otherId <- otherField.id if id != otherId) {
@@ -465,14 +466,18 @@ private[spark] object SchemaUtils {
mode,
checkFieldIds,
errors)
- case None =>
+ case Seq() =>
errors += s"${formatField(colPath, field)} has been removed"
+ case several =>
+ errors += s"${(colPath :+ field.name).fullyQuoted} matches more than one field: " +
+ several.map(_.name).mkString("[", ", ", "]")
}
}
if (mode == PROHIBIT_CHANGES || (mode == ALLOW_NEW_TOP_LEVEL_FIELDS && colPath.nonEmpty)) {
- otherFieldsByName.foreach { case (normalizedName, otherField) =>
- if (!fieldsByName.contains(normalizedName)) {
+ val candidates = index(fields, resolver)
+ otherFields.foreach { otherField =>
+ if (matchField(candidates, otherField.name, resolver).isEmpty) {
errors += s"${formatField(colPath, otherField)} has been added"
}
}
@@ -529,12 +534,51 @@ private[spark] object SchemaUtils {
if (field.nullable) s"$name $dataType" else s"$name $dataType NOT NULL"
}
- private def index(fields: Array[StructField], resolver: Resolver): Map[String, StructField] = {
- if (isCaseSensitiveAnalysis(resolver)) {
- fields.map(field => field.name -> field).toMap
- } else {
- fields.map(field => field.name.toLowerCase(Locale.ROOT) -> field).toMap
- }
+ /**
+ * Folds a name to the key that collects the fields a name may refer to.
+ *
+ * This is the first half of the identity rule name resolution is built on: `AttributeSeq` looks
+ * attributes up by this key and only then filters the candidates with the resolver. The fold
+ * alone is therefore not an identity: it can collect several fields a schema is allowed to keep
+ * apart, and the resolver alone can equate names the fold separates (`equalsIgnoreCase` equates
+ * U+017F LONG S with `s`, which this fold does not).
+ *
+ * In particular this key is NOT single-valued within a schema. `checkColumnNameDuplication` folds
+ * with the JVM default locale rather than with `Locale.ROOT`, so under a Turkish, Azeri or
+ * Lithuanian default locale it admits two names that fold to one key here. Callers must narrow a
+ * multi-candidate match with the resolver instead of assuming there is at most one.
+ */
+ def foldName(name: String, caseSensitiveAnalysis: Boolean): String = {
+ if (caseSensitiveAnalysis) name else name.toLowerCase(Locale.ROOT)
+ }
+
+ private def index(
+ fields: Array[StructField],
+ resolver: Resolver): Map[String, Seq[StructField]] = {
+ val caseSensitive = isCaseSensitiveAnalysis(resolver)
+ fields.groupBy(field => foldName(field.name, caseSensitive))
+ .map { case (folded, group) => folded -> group.toSeq }
+ }
+
+ /**
+ * Returns the fields `name` can refer to, the way column resolution decides it: fold to collect
+ * candidates, then narrow them with the resolver.
+ *
+ * Both halves are always applied. A fold collision is not ambiguous by itself, because the
+ * resolver can still tell the candidates apart, and picking one by position instead would pair a
+ * captured field with a field resolution would not have chosen. Equally, a lone candidate the
+ * fold collects is not a match until the resolver agrees: the fold equates U+0130 CAPITAL I WITH
+ * DOT with `i` plus a combining dot while `equalsIgnoreCase` does not, so accepting it here would
+ * call a schema compatible when no query can read that name out of it.
+ */
+ private def matchField(
+ candidatesByFoldedName: Map[String, Seq[StructField]],
+ name: String,
+ resolver: Resolver): Seq[StructField] = {
+ val caseSensitive = isCaseSensitiveAnalysis(resolver)
+ candidatesByFoldedName
+ .getOrElse(foldName(name, caseSensitive), Nil)
+ .filter(f => resolver(f.name, name))
}
/**
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
index 39634eb4c834..3edf4b1a8a22 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
@@ -194,6 +194,9 @@ abstract class InMemoryBaseTable(
private val allowUnsupportedTransforms =
properties.getOrDefault("allow-unsupported-transforms", "false").toBoolean
+ private val simulatePartialColumnPruning = properties
+ .getOrDefault(InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING, "false").toBoolean
+
private val acceptAnySchema = properties.getOrDefault("accept-any-schema", "false").toBoolean
private val autoSchemaEvolution = properties.getOrDefault("auto-schema-evolution", "true")
.toBoolean
@@ -572,10 +575,21 @@ abstract class InMemoryBaseTable(
// them by their logical (original) names, not their current names.
val schemaNames = tableSchema.map(_.name).toSet
val prunedFields = requiredSchema.filter {
- case MetadataStructFieldWithLogicalName(f, name) => metadataColumnNames.contains(name)
+ case MetadataStructFieldWithLogicalName(_, name) => metadataColumnNames.contains(name)
case f => schemaNames.contains(f.name)
}
- schema = StructType(prunedFields)
+ schema = if (simulatePartialColumnPruning) {
+ // Prune the metadata columns as required, but report every data column back from
+ // `Scan.readSchema()` regardless of what was required.
+ StructType(tableSchema ++ prunedFields.filter(isRequiredMetadataField))
+ } else {
+ StructType(prunedFields)
+ }
+ }
+
+ private def isRequiredMetadataField(field: StructField): Boolean = field match {
+ case MetadataStructFieldWithLogicalName(_, name) => metadataColumnNames.contains(name)
+ case _ => false
}
override def pushFilters(filters: Array[Filter]): Array[Filter] = {
@@ -1058,6 +1072,13 @@ object InMemoryBaseTable {
// SQL conf key that enables column ID assignment
val ASSIGN_COLUMN_IDS = "spark.sql.test.inMemoryTable.assignColumnIds"
+ // Test table property that simulates a connector which accepts a required schema but only
+ // partially applies it: every data column is reported from Scan.readSchema() while the required
+ // metadata columns are still pruned. `SupportsPushDownRequiredColumns` permits partial pruning of
+ // data columns; dropping the required metadata columns is not permitted, since
+ // `SupportsMetadataColumns` makes the read schema the only channel that delivers them.
+ val SIMULATE_PARTIAL_COLUMN_PRUNING = "simulate-partial-column-pruning"
+
/**
* Assigns fresh IDs to any top-level column or nested struct field that does not already
* have one. Recurses into struct fields within ArrayType and MapType so that every field
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AnalyzedSchemaProjection.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AnalyzedSchemaProjection.scala
new file mode 100644
index 000000000000..41770708234d
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AnalyzedSchemaProjection.scala
@@ -0,0 +1,356 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import org.apache.spark.{SparkException, SparkThrowable}
+import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.expressions.{Alias, ArrayTransform, AttributeReference, AttributeSeq, CreateNamedStruct, Expression, ExtractValue, GetStructField, If, IsNull, KnownNotNull, LambdaFunction, Literal, MetadataAttributeWithLogicalName, NamedLambdaVariable, TaggingExpression, TransformKeys, TransformValues, UnresolvedNamedLambdaVariable}
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.util.MetadataColumnHelper
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, Metadata, StructType}
+import org.apache.spark.sql.util.SchemaUtils
+
+/**
+ * Restores the output a plan was analyzed with on a relation whose table has since changed - the
+ * schema `V2TableUtil.validateCapturedColumns` calls the captured one.
+ *
+ * The relation exposes the table's current schema, so its output stays aligned with the physical
+ * scan, and a projection on top recreates the columns, types and expression IDs the parent plan was
+ * analyzed against.
+ */
+private[sql] object AnalyzedSchemaProjection extends SQLConfHelper {
+
+ /**
+ * Prevents [[CreateNamedStruct]] from inheriting metadata from a field value while leaving the
+ * value's type, nullability, evaluation, and code generation unchanged.
+ */
+ private case class MetadataPropagationBarrier(child: Expression) extends TaggingExpression {
+ override protected def withNewChildInternal(
+ newChild: Expression): MetadataPropagationBarrier = copy(child = newChild)
+ }
+
+ def rebindToAnalyzedSchema(relation: DataSourceV2Relation): LogicalPlan = {
+ // The relation still carries the output captured at analysis time; only its table has been
+ // swapped for the current one.
+ val capturedOutput = relation.output
+ val resolver = conf.resolver
+ val caseSensitive = conf.caseSensitiveAnalysis
+ val current = DataSourceV2Relation.create(
+ relation.table,
+ relation.catalog,
+ relation.identifier,
+ relation.options,
+ relation.timeTravelSpec)
+ val currentMetadataOutput = current.metadataOutput
+ val currentMetadata = capturedOutput.filter(_.isMetadataCol).map { captured =>
+ val logicalName = metadataLogicalName(captured)
+ matchFoldedName(currentMetadataOutput, logicalName, caseSensitive)(metadataLogicalName)
+ .map(pos => currentMetadataOutput(pos))
+ .getOrElse {
+ // The connector still reports this metadata column, so it can only be absent here
+ // because a data column has taken its name and the connector suppresses rather than
+ // renames the conflict (`canRenameConflictingMetadataColumns`). Validation owns
+ // rejecting that.
+ unexpectedSchemaChange(
+ s"captured metadata column $logicalName is missing from the current relation")
+ }
+ }
+
+ val currentOutput = current.output ++ currentMetadata
+
+ // Refresh may visit an already rebound relation. Preserve its attributes so the projection
+ // above it continues to reference valid expression IDs.
+ //
+ // A further schema change on such a relation adds a second projection instead of replacing
+ // the first. Only the cache stores a refreshed plan, so the effect is limited to that entry:
+ // it stops matching the single projection a query rebuilds from its own captured output, and
+ // is no longer reused. Results stay correct.
+ if (sameOutputShape(capturedOutput, currentOutput)) {
+ return relation
+ }
+
+ val capturedIndex = new AttributeIndex(capturedOutput, resolver, caseSensitive)
+ val reboundOutput = currentOutput.map { currentAttr =>
+ capturedIndex.get(currentAttr).filter(canReuse(_, currentAttr)).getOrElse(currentAttr)
+ }
+ val reboundRelation = relation.copy(output = reboundOutput)
+
+ val reboundIndex = new AttributeIndex(reboundOutput, resolver, caseSensitive)
+ val projectList = capturedOutput.map { capturedAttr =>
+ val currentAttr = reboundIndex.get(capturedAttr).getOrElse {
+ unexpectedSchemaChange(
+ s"captured column ${capturedAttr.name} is missing from current table ${relation.name}")
+ }
+ if (currentAttr.exprId == capturedAttr.exprId &&
+ sameAttributeShape(currentAttr, capturedAttr)) {
+ currentAttr
+ } else {
+ if (currentAttr.nullable != capturedAttr.nullable) {
+ unexpectedSchemaChange(
+ s"nullability changed for captured column ${capturedAttr.name} in ${relation.name}")
+ }
+ val projected = projectToType(
+ currentAttr, currentAttr.dataType, capturedAttr.dataType, resolver)
+ if (projected.dataType != capturedAttr.dataType ||
+ projected.nullable != capturedAttr.nullable) {
+ unexpectedSchemaChange(
+ s"failed to recreate captured column ${capturedAttr.name} in ${relation.name}")
+ }
+ Alias(projected, capturedAttr.name)(
+ exprId = capturedAttr.exprId,
+ qualifier = capturedAttr.qualifier,
+ explicitMetadata = Some(capturedAttr.metadata))
+ }
+ }
+
+ Project(projectList, reboundRelation)
+ }
+
+ private[v2] def projectToType(
+ input: Expression,
+ from: DataType,
+ to: DataType,
+ resolver: Resolver): Expression = {
+ if (from == to) {
+ return input
+ }
+
+ val projected = (from, to) match {
+ case (fromStruct: StructType, toStruct: StructType) =>
+ val structInput = if (input.nullable) KnownNotNull(input) else input
+ val fields = toStruct.fields.iterator.flatMap { targetField =>
+ val index = matchStructField(structInput, fromStruct, targetField.name, resolver)
+ val sourceField = fromStruct.fields(index)
+ val value = projectToType(
+ GetStructField(structInput, index, Some(sourceField.name)),
+ sourceField.dataType,
+ targetField.dataType,
+ resolver)
+ val namedValue = if (targetField.metadata == Metadata.empty) {
+ // An empty captured value is still an explicit instruction not to inherit metadata
+ // from the current GetStructField. CleanupAliases removes an empty-metadata Alias, so
+ // use a local passthrough barrier instead of changing shared alias cleanup behavior.
+ MetadataPropagationBarrier(value)
+ } else {
+ Alias(value, targetField.name)(explicitMetadata = Some(targetField.metadata))
+ }
+ Iterator(Literal(targetField.name), namedValue)
+ }.toSeq
+ val rebuilt = CreateNamedStruct(fields)
+ if (input.nullable) {
+ // The null literal takes the rebuilt type rather than `toStruct` so that the type check
+ // below still sees any mismatch: `If` merges its branch types and only requires them to
+ // match up to `sameType`, which ignores nullability and metadata.
+ If(IsNull(input), Literal.create(null, rebuilt.dataType), rebuilt)
+ } else {
+ rebuilt
+ }
+
+ case (ArrayType(fromElement, fromContainsNull), ArrayType(toElement, toContainsNull)) =>
+ if (fromContainsNull != toContainsNull) {
+ unexpectedSchemaChange(s"array element nullability changed from $from to $to")
+ }
+ val element = NamedLambdaVariable(
+ UnresolvedNamedLambdaVariable.freshVarName("element"),
+ fromElement,
+ fromContainsNull)
+ ArrayTransform(
+ input,
+ LambdaFunction(
+ projectToType(element, fromElement, toElement, resolver), Seq(element)))
+
+ case (
+ MapType(fromKey, fromValue, fromValueContainsNull),
+ MapType(toKey, toValue, toValueContainsNull)) =>
+ if (fromValueContainsNull != toValueContainsNull) {
+ unexpectedSchemaChange(s"map value nullability changed from $from to $to")
+ }
+
+ val withProjectedKeys = if (fromKey != toKey) {
+ val key = NamedLambdaVariable(
+ UnresolvedNamedLambdaVariable.freshVarName("key"),
+ fromKey,
+ nullable = false)
+ val value = NamedLambdaVariable(
+ UnresolvedNamedLambdaVariable.freshVarName("value"),
+ fromValue,
+ fromValueContainsNull)
+ // A field dropped from a key can collapse keys that are distinct under the current type
+ // into one captured key. Rebuilding the map through `TransformKeys` keeps the uniqueness
+ // invariant, so such a projection fails with DUPLICATED_MAP_KEY (or keeps the last entry
+ // under `spark.sql.mapKeyDedupPolicy=LAST_WIN`) rather than producing a map that holds
+ // duplicate keys.
+ TransformKeys(
+ input,
+ LambdaFunction(projectToType(key, fromKey, toKey, resolver), Seq(key, value)))
+ } else {
+ input
+ }
+
+ if (fromValue != toValue) {
+ val key = NamedLambdaVariable(
+ UnresolvedNamedLambdaVariable.freshVarName("key"),
+ toKey,
+ nullable = false)
+ val value = NamedLambdaVariable(
+ UnresolvedNamedLambdaVariable.freshVarName("value"),
+ fromValue,
+ fromValueContainsNull)
+ TransformValues(
+ withProjectedKeys,
+ LambdaFunction(
+ projectToType(value, fromValue, toValue, resolver), Seq(key, value)))
+ } else {
+ withProjectedKeys
+ }
+
+ case _ =>
+ unexpectedSchemaChange(s"cannot project incompatible data type $from to $to")
+ }
+
+ if (projected.dataType != to) {
+ unexpectedSchemaChange(
+ s"projected data type ${projected.dataType} does not match captured type $to")
+ }
+ projected
+ }
+
+ /**
+ * Returns the ordinal in `fromStruct` that the captured field name refers to, by asking the same
+ * resolution Spark uses for `st.field` in a query.
+ *
+ * Deciding this here instead would mean maintaining a second rule, and Spark's two rules are not
+ * the same: a top-level column is resolved by folding names with `toLowerCase(ROOT)` to collect
+ * candidates and only then filtering them with the resolver, while a struct field is resolved
+ * with the resolver alone. A name pair the fold separates but the resolver equates - `s` beside
+ * U+017F LONG S, say - is therefore a distinct column at the top level and ambiguous inside a
+ * struct. Delegating keeps a refreshed plan readable exactly where a fresh query is readable.
+ *
+ * An ambiguity is reported as the same user-facing error a fresh query reports. Refresh
+ * validation pairs a captured field with a current one the top-level way - candidates by folded
+ * name, narrowed with the resolver - so it can see one candidate where this sees two, and accepts
+ * a schema this rejects. A missing field, by contrast, cannot happen once validation has passed:
+ * the field it paired is always among the candidates the resolver alone collects here, so that
+ * branch stays an internal error.
+ */
+ private def matchStructField(
+ structInput: Expression,
+ fromStruct: StructType,
+ capturedName: String,
+ resolver: Resolver): Int = {
+ ExtractValue.extractValue(structInput, Literal(capturedName), resolver) match {
+ case Left(field: GetStructField) => field.ordinal
+ case Left(other) =>
+ unexpectedSchemaChange(
+ s"captured struct field $capturedName extracted a ${other.getClass.getSimpleName}")
+ case Right(ambiguous: SparkThrowable)
+ if ambiguous.getCondition == "AMBIGUOUS_REFERENCE_TO_FIELDS" =>
+ throw ambiguous.asInstanceOf[Throwable]
+ case Right(_) =>
+ unexpectedSchemaChange(
+ s"captured struct field $capturedName is missing from $fromStruct")
+ }
+ }
+
+ /**
+ * Returns the position of the first entry whose folded name matches `target`, if any.
+ *
+ * Only the metadata lookups use this. A metadata attribute matches on its logical name rather
+ * than on `name`, and `AttributeSeq` indexes by `name`, so that lookup cannot be delegated to
+ * Spark's resolution the way the data-column and struct-field lookups are. It keeps the fold and
+ * the first-match tie-break it has always had; the weaknesses of that rule - a fold collision can
+ * arise under a non-ROOT default locale, and the first match is then arbitrary - are unaddressed
+ * here on purpose, because metadata columns are out of this change's scope.
+ */
+ private def matchFoldedName[T](
+ candidates: Seq[T],
+ target: String,
+ caseSensitive: Boolean)(name: T => String): Option[Int] = {
+ val foldedTarget = SchemaUtils.foldName(target, caseSensitive)
+ val pos = candidates.indexWhere { candidate =>
+ SchemaUtils.foldName(name(candidate), caseSensitive) == foldedTarget
+ }
+ if (pos >= 0) Some(pos) else None
+ }
+
+ /**
+ * Indexes attributes by name for the rebinding lookups. Data and metadata attributes are indexed
+ * separately because a metadata attribute matches on its logical name, which a data column may
+ * also carry.
+ *
+ * Data attributes are looked up through [[AttributeSeq.resolve]], so a captured column binds to
+ * the attribute a fresh query would resolve its name to and an ambiguous name raises the same
+ * user-facing error rather than silently binding to one of the candidates. Metadata attributes
+ * keep the folded-name lookup they had, for the reason given on [[matchFoldedName]].
+ */
+ private class AttributeIndex(
+ attributes: Seq[AttributeReference],
+ resolver: Resolver,
+ caseSensitive: Boolean) {
+ private val dataAttrs = attributes.filterNot(_.isMetadataCol)
+ private val metadataAttrs = attributes.filter(_.isMetadataCol)
+ // Both indexes are built once, because `get` runs per column: `AttributeSeq` keeps its name
+ // index in per-instance lazy state, and recovering the attribute by expression ID would
+ // otherwise scan the attributes again on every lookup.
+ private val dataAttrSeq = AttributeSeq(dataAttrs)
+ private val dataAttrsById = dataAttrs.map(attr => attr.exprId -> attr).toMap
+
+ def get(target: AttributeReference): Option[AttributeReference] = {
+ if (target.isMetadataCol) {
+ matchFoldedName(metadataAttrs, metadataLogicalName(target), caseSensitive)(
+ metadataLogicalName).map(pos => metadataAttrs(pos))
+ } else {
+ dataAttrSeq.resolve(Seq(target.name), resolver).flatMap { resolved =>
+ // `resolve` renames the attribute it returns to the requested name; recover the attribute
+ // itself so the rebound output keeps its own name and expression ID.
+ dataAttrsById.get(resolved.references.head.exprId)
+ }
+ }
+ }
+ }
+
+ private def metadataLogicalName(attr: AttributeReference): String = attr match {
+ case MetadataAttributeWithLogicalName(_, logicalName) => logicalName
+ case _ =>
+ unexpectedSchemaChange(s"metadata attribute ${attr.name} has no logical name")
+ }
+
+ private def canReuse(captured: AttributeReference, current: AttributeReference): Boolean = {
+ captured.name == current.name && sameAttributeShape(captured, current)
+ }
+
+ private def sameOutputShape(
+ left: Seq[AttributeReference],
+ right: Seq[AttributeReference]): Boolean = {
+ left.length == right.length && left.zip(right).forall { case (l, r) =>
+ l.name == r.name && l.isMetadataCol == r.isMetadataCol && sameAttributeShape(l, r)
+ }
+ }
+
+ private def sameAttributeShape(left: AttributeReference, right: AttributeReference): Boolean = {
+ left.dataType == right.dataType &&
+ left.nullable == right.nullable &&
+ left.metadata == right.metadata
+ }
+
+ private def unexpectedSchemaChange(message: String): Nothing = {
+ throw SparkException.internalError(
+ s"Unexpected incompatible table schema after refresh validation: $message")
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
index da5f930795e7..e900aa3a6808 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
@@ -103,7 +103,12 @@ private[sql] object V2TableRefreshUtil extends SQLConfHelper with Logging {
validateTableIdentity(currentTable, r)
validateDataColumns(currentTable, r, schemaValidationMode)
validateMetadataColumns(currentTable, r, schemaValidationMode)
- r.copy(table = currentTable)
+ val refreshed = r.copy(table = currentTable)
+ if (schemaValidationMode == ALLOW_NEW_FIELDS) {
+ AnalyzedSchemaProjection.rebindToAnalyzedSchema(refreshed)
+ } else {
+ refreshed
+ }
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
index 028def6e18bc..1fd1f84c48e1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
@@ -23,9 +23,10 @@ import java.util.Collections
import scala.jdk.CollectionConverters._
import scala.reflect.ClassTag
-import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException}
import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode, SessionQueryTest, SparkSession}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
+import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
import org.apache.spark.sql.catalyst.plans.logical.{AppendData, CreateTableAsSelect, LogicalPlan, ReplaceTableAsSelect}
import org.apache.spark.sql.connector.catalog.{BasicInMemoryTableCatalog, CachingInMemoryTableCatalog, CatalogV2Util, Column, ColumnDefaultValue, DefaultValue, GenerationExpression, Identifier, InMemoryBaseTable, InMemoryTableCatalog, MixedColumnIdTableCatalog, NullColumnIdInMemoryTableCatalog, NullTableIdAndNullColumnIdInMemoryTableCatalog, NullTableIdInMemoryTableCatalog, SupportsV1OverwriteWithSaveAsTable, TableCatalog, TableInfo, TypeChangeResetsColIdTableCatalog}
@@ -37,7 +38,7 @@ import org.apache.spark.sql.connector.expressions.{ApplyTransform, Cast => V2Cas
import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, AlwaysTrue, Predicate => V2Predicate}
import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
import org.apache.spark.sql.execution.ExplainUtils.stripAQEPlan
-import org.apache.spark.sql.execution.datasources.v2.{AlterTableExec, CreateTableExec, DataSourceV2Relation, ReplaceTableExec}
+import org.apache.spark.sql.execution.datasources.v2.{AlterTableExec, CreateTableExec, DataSourceV2Relation, DataSourceV2ScanRelation, ReplaceTableExec}
import org.apache.spark.sql.functions.{lit, sum}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{BooleanType, CalendarIntervalType, DoubleType, IntegerType, LongType, StringType, StructType, TimestampType}
@@ -1584,6 +1585,12 @@ class DataSourceV2DataFrameSuite
stripAQEPlan(qe.executedPlan).asInstanceOf[T]
}
+ private def scanRelationOf(df: DataFrame): DataSourceV2ScanRelation = {
+ df.queryExecution.optimizedPlan.collectFirst {
+ case scan: DataSourceV2ScanRelation => scan
+ }.get
+ }
+
private def checkDefaultValues(
columns: Array[Column],
expectedDefaultValues: Array[ColumnDefaultValue],
@@ -1725,6 +1732,531 @@ class DataSourceV2DataFrameSuite
}
}
+ test("a captured nested field a fresh query cannot resolve fails the refreshed query too") {
+ // Spark resolves a top-level column by folding names with `toLowerCase(ROOT)` to collect
+ // candidates and only then filtering them with the resolver, but resolves a struct field with
+ // the resolver alone. Adding U+017F LONG S beside `s` therefore leaves a top-level `s`
+ // resolvable while making a nested `st.s` ambiguous. A refreshed plan has to report that
+ // ambiguity rather than keep reading one of the two, so that a stale plan and a fresh query
+ // agree on whether the name is readable.
+ //
+ // Spark's own DDL refuses the addition at either level (it looks for an existing field with the
+ // resolver and reports FIELD_ALREADY_EXISTS), so it is applied through the catalog the way
+ // another engine would. That is also why this only reaches tables mutated outside Spark.
+ val longS = new String(Character.toChars(0x17f))
+ val t = "testcat.ns1.ns2.tbl"
+
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, st STRUCT<s: INT>) USING foo")
+
+ // Analyze but do not execute: a Dataset memoizes its optimized plan, so collecting here would
+ // refresh once and never see the change below.
+ val stale = spark.table(t).selectExpr("st.s AS v")
+ assert(stale.queryExecution.analyzed.resolved)
+
+ catalog("testcat")
+ .alterTable(testIdent, TableChange.addColumn(Array("st", longS), IntegerType, true))
+
+ val fresh = intercept[AnalysisException](sql(s"SELECT st.s FROM $t").collect())
+ assert(fresh.getCondition == "AMBIGUOUS_REFERENCE_TO_FIELDS")
+
+ val refreshed = intercept[AnalysisException](stale.collect())
+ assert(
+ refreshed.getCondition == "AMBIGUOUS_REFERENCE_TO_FIELDS",
+ "the refreshed plan must report the ambiguity a fresh query reports")
+ }
+
+ // The same addition at the top level stays readable, for both a fresh and a refreshed plan.
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, s INT) USING foo")
+ val stale = spark.table(t).filter("id > 0")
+ assert(stale.queryExecution.analyzed.resolved)
+
+ val cat = catalog("testcat")
+ cat.alterTable(testIdent, TableChange.addColumn(Array(longS), IntegerType, true))
+ externalAppend(cat, testIdent, InternalRow(1, 10, 99))
+
+ checkAnswer(sql(s"SELECT s FROM $t"), Seq(Row(10)))
+ checkAnswer(stale, Seq(Row(1, 10)))
+ }
+ }
+
+ test("rebind a captured column renamed beside an addition the resolver cannot tell apart") {
+ // U+017F LONG S folds to itself under `toLowerCase`, so it is a distinct column name to the
+ // fold that name resolution and refresh validation key on, while `equalsIgnoreCase` equates it
+ // with `s`. Renaming `s` to `S` and adding U+017F is therefore a compatible change, and the
+ // captured `s` must keep reading the renamed column rather than becoming ambiguous.
+ val longS = new String(Character.toChars(0x17f))
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, s INT) USING foo")
+
+ // Analyze the plan but do not execute it: a Dataset memoizes its optimized plan, so a collect
+ // before the change would refresh once and never see the change below.
+ val df = spark.table(t).filter("id > 0")
+ assert(df.queryExecution.analyzed.resolved)
+
+ // Spark's own DDL refuses both changes because it checks for an existing field with the
+ // resolver, so apply them through the catalog the way another engine would. Data is written
+ // afterwards because this fixture migrates rows by exact field name and so drops the values
+ // of a renamed column.
+ val cat = catalog("testcat")
+ cat.alterTable(testIdent, TableChange.renameColumn(Array("s"), "S"))
+ cat.alterTable(testIdent, TableChange.addColumn(Array(longS), IntegerType, true))
+ externalAppend(cat, testIdent, InternalRow(2, 20, 99))
+
+ // Reading 99 instead of 20 would mean the captured name bound to the addition.
+ checkAnswer(df, Seq(Row(2, 20)))
+ assert(df.queryExecution.optimizedPlan.output.map(_.name) == Seq("id", "s"))
+ }
+ }
+
+ test("refresh keeps an exact-name binding under a Turkish locale fold collision") {
+ val capitalIDot = new String(Character.toChars(0x130))
+ val iCombiningDot = "i" + new String(Character.toChars(0x307))
+ val t = "testcat.ns1.ns2.tbl"
+
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+ withLocale("tr") {
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, `$iCombiningDot` INT) USING foo")
+
+ // Analyze but do not execute so that collection has to refresh this captured plan.
+ val stale = spark.table(t).selectExpr("id", s"`$iCombiningDot`")
+ assert(stale.queryExecution.analyzed.resolved)
+
+ // Under a Turkish default locale, duplicate validation admits U+0130 beside
+ // i+U+0307, although Spark's root-locale fold puts both names in one lookup bucket.
+ // Put the addition first so that choosing the bucket's first field would be observable.
+ val cat = catalog("testcat")
+ cat.alterTable(
+ testIdent,
+ TableChange.addColumn(
+ Array(capitalIDot),
+ IntegerType,
+ true,
+ null,
+ TableChange.ColumnPosition.first(),
+ null))
+ assert(
+ cat.loadTable(testIdent).columns().map(_.name).toSeq ==
+ Seq(capitalIDot, "id", iCombiningDot))
+ externalAppend(cat, testIdent, InternalRow(99, 1, 20))
+
+ // A fresh query establishes that the exact name reads 20. The captured plan must make
+ // the same choice rather than read 99 from the preceding addition.
+ checkAnswer(sql(s"SELECT id, `$iCombiningDot` FROM $t"), Seq(Row(1, 20)))
+ checkAnswer(stale, Seq(Row(1, 20)))
+ }
+ }
+ }
+ }
+
+ test("refresh reconciles a wider partially-pruned scan with stored temp view output") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ withView("v") {
+ sql(
+ s"""CREATE TABLE $t (id INT, salary INT) USING foo
+ |TBLPROPERTIES (
+ | '${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')
+ |""".stripMargin)
+ sql(s"INSERT INTO $t VALUES (1, 100)")
+
+ spark.table(t).filter("salary < 999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
+
+ val cat = catalog("testcat")
+ val addCol = TableChange.addColumn(Array("new_column"), IntegerType, true)
+ cat.alterTable(testIdent, addCol)
+ externalAppend(cat, testIdent, InternalRow(2, 200, -1))
+
+ val refreshedView = spark.table("v")
+ val scan = refreshedView.queryExecution.optimizedPlan.collectFirst {
+ case scan: DataSourceV2ScanRelation => scan
+ }.get
+ assert(scan.output.map(_.name) == Seq("id", "salary", "new_column"))
+ assert(refreshedView.queryExecution.optimizedPlan.output.map(_.name) == Seq("id", "salary"))
+ checkAnswer(refreshedView, Seq(Row(1, 100), Row(2, 200)))
+ }
+ }
+ }
+
+ test("refresh recreates a captured nested schema from a partially-pruned scan") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ sql(
+ s"""CREATE TABLE $t (id INT, person STRUCT<name: STRING>) USING foo
+ |TBLPROPERTIES (
+ | '${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')
+ |""".stripMargin)
+ sql(s"INSERT INTO $t VALUES (1, named_struct('name', 'Alice'))")
+
+ val df = spark.table(t)
+ assert(df.queryExecution.analyzed.resolved)
+
+ sql(s"ALTER TABLE $t ADD COLUMN person.age INT FIRST")
+ sql(s"INSERT INTO $t VALUES (2, named_struct('age', 25, 'name', 'Bob'))")
+ sql(s"INSERT INTO $t VALUES (3, NULL)")
+
+ checkAnswer(df, Seq(
+ Row(1, Row("Alice")),
+ Row(2, Row("Bob")),
+ Row(3, null)))
+ }
+ }
+
+ test("refresh recreates a captured nested schema without nested pruning or filter pushdown") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, person STRUCT<name: STRING, city: STRING>) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, named_struct('name', 'Alice', 'city', 'SF'))")
+
+ def nestedQuery(): DataFrame =
+ spark.table(t).where("person.name = 'Alice'").select("person.name")
+
+ def personFieldsRead(df: DataFrame): Seq[String] = {
+ val readSchema = scanRelationOf(df).scan.readSchema()
+ readSchema("person").dataType.asInstanceOf[StructType].fieldNames.toSeq
+ }
+
+ // Every filter this source reports is one Spark managed to translate and hand to
+ // `pushFilters`. The scan relation's own `pushedFilters` cannot be used here: it keeps only
+ // fully-pushed filters, and this source evaluates none on an unpartitioned table.
+ def filtersReachingSource(df: DataFrame): Seq[String] = {
+ scanRelationOf(df).scan match {
+ case scan: InMemoryBaseTable#InMemoryBatchScan =>
+ scan.pushedFilters.map(_.toString).toSeq
+ case other =>
+ fail(s"unexpected scan type ${other.getClass.getName}")
+ }
+ }
+
+ // capture the plan, then add a field inside the struct
+ val captured = nestedQuery()
+ assert(captured.queryExecution.analyzed.resolved)
+ sql(s"ALTER TABLE $t ADD COLUMN person.age INT FIRST")
+ sql(s"INSERT INTO $t VALUES (2, named_struct('age', 25, 'name', 'Bob', 'city', 'NY'))")
+
+ val fresh = nestedQuery()
+ checkAnswer(captured, Seq(Row("Alice")))
+ checkAnswer(fresh, Seq(Row("Alice")))
+
+ // a plan analyzed after the change prunes to the one field it reads and translates the
+ // predicate down to the source
+ assert(personFieldsRead(fresh) == Seq("name"))
+ assert(filtersReachingSource(fresh).nonEmpty)
+
+ // Rebinding rebuilds the struct as `If(IsNull(person), null, CreateNamedStruct(...))`, which
+ // the extraction, schema-pruning and pushdown rules cannot see through, so the captured plan
+ // reads the whole current struct and its predicate never reaches the source. Results stay
+ // correct, only wider than necessary. Tighten both assertions to match `fresh` once the
+ // projection uses an optimizer-friendly null-preserving form.
+ assert(personFieldsRead(captured) == Seq("age", "name", "city"))
+ assert(filtersReachingSource(captured).isEmpty)
+ }
+ }
+
+ test("refresh recreates a captured schema nested through a map and an array") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ // A struct is needed between the map and the array: CatalogV2Util.replace only walks into a
+ // collection whose element/value type is a struct, so ALTER TABLE cannot reach the inner
+ // field of MAP<STRING, ARRAY<STRUCT<...>>>. AnalyzedSchemaProjectionSuite covers that shape
+ // directly at the expression level.
+ sql(
+ s"""CREATE TABLE $t (
+ | id INT,
+ | data STRUCT<m: MAP<STRING, STRUCT<arr: ARRAY<STRUCT<v: STRING>>>>>)
+ |USING foo
+ |TBLPROPERTIES (
+ | '${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')
+ |""".stripMargin)
+ sql(
+ s"""INSERT INTO $t VALUES (
+ | 1,
+ | named_struct('m', map('k', named_struct('arr', array(named_struct('v', 'a'))))))
+ |""".stripMargin)
+
+ val df = spark.table(t)
+ assert(df.queryExecution.analyzed.resolved)
+
+ // Add a field at two depths, each FIRST, so the captured field that follows it moves to a
+ // different ordinal at each level.
+ sql(s"ALTER TABLE $t ADD COLUMN data.m.value.arr.element.extra INT FIRST")
+ sql(s"ALTER TABLE $t ADD COLUMN data.added INT FIRST")
+ sql(
+ s"""INSERT INTO $t VALUES (
+ | 2,
+ | named_struct(
+ | 'added', 9,
+ | 'm', map('k', named_struct('arr', array(named_struct('extra', 7, 'v', 'b'))))))
+ |""".stripMargin)
+
+ checkAnswer(df, Seq(
+ Row(1, Row(Map("k" -> Row(Seq(Row("a")))))),
+ Row(2, Row(Map("k" -> Row(Seq(Row("b"))))))))
+ }
+ }
+
+ // Narrowing a map key back to the captured type is the one case where rebinding cannot reproduce
+ // the captured output: dropping a key field can collapse keys that are distinct under the current
+ // type into one captured key, and the captured type then cannot say how many entries the map has.
+ // The projection rebuilds the map through Spark's map builder, so the map-key uniqueness
+ // invariant decides the outcome instead.
+ test("refresh fails when narrowing a map key collapses distinct keys") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ sql(
+ s"""CREATE TABLE $t (m MAP<STRUCT<name: STRING>, INT>) USING foo
+ |TBLPROPERTIES (
+ | '${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')
+ |""".stripMargin)
+
+ val df = spark.table(t).selectExpr("size(m) AS n", "map_keys(m) AS keys")
+ assert(df.queryExecution.analyzed.resolved)
+
+ sql(s"ALTER TABLE $t ADD COLUMN m.key.id INT")
+ sql(
+ s"""INSERT INTO $t SELECT map(
+ | named_struct('name', 'a', 'id', 1), 10,
+ | named_struct('name', 'a', 'id', 2), 20)
+ |""".stripMargin)
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ df.collect()
+ },
+ condition = "DUPLICATED_MAP_KEY",
+ parameters = Map(
+ "key" -> "[a]",
+ "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\""))
+ }
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ checkAnswer(df, Seq(Row(1, Seq(Row("a")))))
+ }
+ }
+ }
+
+ test("refresh does not evaluate an unselected sibling with a narrowed map key") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ sql(
+ s"""CREATE TABLE $t (
+ | payload STRUCT<x: INT, m: MAP<STRUCT<name: STRING>, INT>>)
+ |USING foo
+ |""".stripMargin)
+
+ val selectedMap = spark.table(t).select("payload.m")
+ val selectedSibling = spark.table(t).select("payload.x")
+ assert(selectedMap.queryExecution.analyzed.resolved)
+ assert(selectedSibling.queryExecution.analyzed.resolved)
+
+ sql(s"ALTER TABLE $t ADD COLUMN payload.m.key.id INT")
+ sql(
+ s"""INSERT INTO $t SELECT named_struct(
+ | 'x', 1,
+ | 'm', map(
+ | named_struct('name', 'a', 'id', 1), 10,
+ | named_struct('name', 'a', 'id', 2), 20))
+ |""".stripMargin)
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+ // Reading the map proves that this row's keys collapse under the captured key type.
+ val error = intercept[SparkRuntimeException] {
+ selectedMap.collect()
+ }
+ assert(error.getCondition == "DUPLICATED_MAP_KEY")
+
+ // Reading only the sibling must not evaluate the unused map reconstruction.
+ checkAnswer(selectedSibling, Seq(Row(1)))
+ }
+ }
+ }
+
+ test("refresh preserves a captured metadata column across a wider partially-pruned scan") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ withView("v") {
+ sql(
+ s"""CREATE TABLE $t (id INT, salary INT) USING foo
+ |TBLPROPERTIES (
+ | '${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')
+ |""".stripMargin)
+ sql(s"INSERT INTO $t VALUES (1, 100)")
+
+ spark.table(t).select("id", "salary", "index").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100, 0)))
+
+ val cat = catalog("testcat")
+ cat.alterTable(testIdent, TableChange.addColumn(Array("new_column"), IntegerType, true))
+ externalAppend(cat, testIdent, InternalRow(2, 200, -1))
+
+ val refreshedView = spark.table("v")
+ assert(refreshedView.schema.map(_.name) == Seq("id", "salary", "index"))
+ checkAnswer(refreshedView, Seq(Row(1, 100, 0), Row(2, 200, 0)))
+ }
+ }
+ }
+
+ // This test does not enable partial column pruning, so the scan reports only the captured
+ // columns and `toOutputAttrs` finds every name: the new data column never reaches the read
+ // schema, and so cannot collapse onto the captured metadata attribute. What is pinned here is
+ // that rebinding keeps the captured `index` bound to the metadata column, which this connector
+ // exposes under a renamed name, rather than to the new data column. Unlike the other tests in
+ // this group it therefore also passes without the rebinding call.
+ test("refresh keeps a captured metadata column when a new data column takes its name") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ withView("v") {
+ sql(s"CREATE TABLE $t (id INT, salary INT) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, 100)")
+
+ spark.table(t).select("id", "salary", "index").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100, 0)))
+
+ // The new data column conflicts with the `index` metadata column, so the current relation
+ // exposes the metadata column under a conflict-renamed name.
+ val cat = catalog("testcat")
+ cat.alterTable(testIdent, TableChange.addColumn(Array("index"), IntegerType, true))
+ externalAppend(cat, testIdent, InternalRow(2, 200, -1))
+
+ val refreshedView = spark.table("v")
+ assert(refreshedView.schema.map(_.name) == Seq("id", "salary", "index"))
+ // `index` still resolves to the metadata column (0), not to the new data column (-1).
+ checkAnswer(refreshedView, Seq(Row(1, 100, 0), Row(2, 200, 0)))
+ }
+ }
+ }
+
+ test("refresh restores the captured column order when a column is added first") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ withView("v") {
+ sql(
+ s"""CREATE TABLE $t (id INT, salary INT) USING foo
+ |TBLPROPERTIES (
+ | '${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')
+ |""".stripMargin)
+ sql(s"INSERT INTO $t VALUES (1, 100)")
+
+ spark.table(t).filter("salary < 999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
+
+ sql(s"ALTER TABLE $t ADD COLUMN new_column INT FIRST")
+ sql(s"INSERT INTO $t VALUES (-1, 2, 200)")
+
+ val refreshedView = spark.table("v")
+ val scan = refreshedView.queryExecution.optimizedPlan.collectFirst {
+ case scan: DataSourceV2ScanRelation => scan
+ }.get
+ assert(scan.output.map(_.name) == Seq("new_column", "id", "salary"))
+ // The projection restores the captured order, not the current table order.
+ assert(refreshedView.queryExecution.optimizedPlan.output.map(_.name) == Seq("id", "salary"))
+ checkAnswer(refreshedView, Seq(Row(1, 100), Row(2, 200)))
+ }
+ }
+ }
+
+ // A case-only rename is rejected by ALTER TABLE, so it can only reach the refresh path when an
+ // external writer performs it directly through the catalog. This test also does not enable
+ // partial column pruning, but it still fails without the rebinding call: the scan reports renamed
+ // `SALARY` while the captured output has `salary`, and `toOutputAttrs` looks read-schema names up
+ // in `relation.output`.
+ test("refresh restores the captured column name after an external case-only rename") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ withView("v") {
+ sql(s"CREATE TABLE $t (id INT, salary INT) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, 100)")
+
+ spark.table(t).filter("salary < 999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
+
+ val cat = catalog("testcat")
+ cat.alterTable(testIdent, TableChange.renameColumn(Array("salary"), "SALARY"))
+ assert(cat.loadTable(testIdent).columns().map(_.name).toSeq == Seq("id", "SALARY"))
+
+ // The scan reads the current name while the view keeps the captured one. Rows written
+ // before the rename are not asserted here: the in-memory connector resolves buffered rows
+ // by exact column name, so it cannot read them back under the new name.
+ val refreshedView = spark.table("v")
+ val scan = refreshedView.queryExecution.optimizedPlan.collectFirst {
+ case scan: DataSourceV2ScanRelation => scan
+ }.get
+ assert(scan.output.map(_.name) == Seq("id", "SALARY"))
+ assert(refreshedView.queryExecution.optimizedPlan.output.map(_.name) == Seq("id", "salary"))
+ assert(refreshedView.schema.map(_.name) == Seq("id", "salary"))
+ }
+ }
+ }
+
+ test("refresh keeps a captured column bound to its exact name past a folding sibling") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ withView("v") {
+ sql(s"CREATE TABLE $t (s INT) USING foo")
+ sql(s"INSERT INTO $t VALUES (1)")
+
+ spark.table(t).createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1)))
+
+ // U+017F is a distinct name to the duplicate-name check, which folds names with
+ // `toLowerCase`, and equal to `s` for the case-insensitive resolver. The column is added
+ // through the catalog, like the other external changes in this group.
+ val longS = new String(Character.toChars(0x17f))
+ val cat = catalog("testcat")
+ cat.alterTable(
+ testIdent,
+ TableChange.addColumn(
+ Array(longS), IntegerType, true, null, TableChange.ColumnPosition.first(), null))
+ assert(cat.loadTable(testIdent).columns().map(_.name).toSeq == Seq(longS, "s"))
+ externalAppend(cat, testIdent, InternalRow(7, 2))
+
+ // The captured `s` reads the current `s`, not the column that precedes it and matches only
+ // after folding.
+ checkAnswer(spark.table("v"), Seq(Row(1), Row(2)))
+ }
+ }
+ }
+
+ test("refresh rebinds every relation of a self-joined partially-pruned table") {
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ withView("v") {
+ sql(
+ s"""CREATE TABLE $t (id INT, salary INT) USING foo
+ |TBLPROPERTIES (
+ | '${InMemoryBaseTable.SIMULATE_PARTIAL_COLUMN_PRUNING}' = 'true')
+ |""".stripMargin)
+ sql(s"INSERT INTO $t VALUES (1, 100)")
+
+ sql(s"SELECT l.id, r.salary FROM $t l JOIN $t r ON l.id = r.id")
+ .createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
+
+ val cat = catalog("testcat")
+ cat.alterTable(testIdent, TableChange.addColumn(Array("new_column"), IntegerType, true))
+ externalAppend(cat, testIdent, InternalRow(2, 200, -1))
+
+ val refreshedView = spark.table("v")
+ val scans = refreshedView.queryExecution.optimizedPlan.collect {
+ case scan: DataSourceV2ScanRelation => scan
+ }
+ assert(scans.size == 2)
+ // Both sides see the current schema; column pruning then keeps only what each side needs.
+ assert(scans.forall(_.relation.output.map(_.name) == Seq("id", "salary", "new_column")))
+ assert(refreshedView.schema.map(_.name) == Seq("id", "salary"))
+ checkAnswer(refreshedView, Seq(Row(1, 100), Row(2, 200)))
+ }
+ }
+ }
+
test("SPARK-54157: detect multiple change types after DataFrame analysis") {
val t = "testcat.ns1.ns2.tbl"
withTable(t) {
@@ -3658,6 +4190,35 @@ class DataSourceV2DataFrameSuite
checkAnswer(
spark.table(t),
Seq(Row(1, 10, null), Row(2, 20, null), Row(3, 30, null), Row(4, 40, "40")))
+
+ // a second compatible change now refreshes a cached plan that was already rebound once
+ val secondChange = TableChange.addColumn(Array("extra"), StringType, true)
+ catalog("testcat").alterTable(ident, secondChange)
+
+ // refresh table is supposed to trigger recaching
+ spark.sql(s"REFRESH TABLE $t")
+
+ // recaching is expected to succeed
+ assert(spark.sharedState.cacheManager.numCachedEntries == 1)
+
+ // Rebinding an already rebound plan adds a second projection rather than replacing the first,
+ // so the retained entry no longer matches the single projection a derived query rebuilds from
+ // its own captured output, and stops being reused. Assert that directly: the entry surviving
+ // is not the same claim as the entry being usable. Flip this back to `assertCached` once
+ // refresh replaces the generated projection instead of nesting inside it.
+ assertNotCached(df.filter("id > 0"))
+
+ // Derived queries must still return the captured schema and the latest data.
+ checkAnswer(df.filter("id > 0"), Seq(Row(1, 10), Row(2, 20), Row(3, 30), Row(4, 40)))
+
+ // verify latest schema is propagated again
+ checkAnswer(
+ spark.table(t),
+ Seq(
+ Row(1, 10, null, null),
+ Row(2, 20, null, null),
+ Row(3, 30, null, null),
+ Row(4, 40, "40", null)))
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/AnalyzedSchemaProjectionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/AnalyzedSchemaProjectionSuite.scala
new file mode 100644
index 000000000000..ba62064ad6c6
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/AnalyzedSchemaProjectionSuite.scala
@@ -0,0 +1,1025 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.v2
+
+import java.util
+
+import org.apache.spark.{SparkException, SparkFunSuite, SparkRuntimeException, SparkThrowable}
+import org.apache.spark.sql.catalyst.analysis.{caseInsensitiveResolution, caseSensitiveResolution}
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSeq, CreateNamedStruct, Expression, ExpressionEvalHelper, ExtractValue, GetStructField, Literal, MetadataAttribute}
+import org.apache.spark.sql.catalyst.optimizer.FoldablePropagation
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, Project}
+import org.apache.spark.sql.catalyst.util.{GenericArrayData, MetadataColumnHelper}
+import org.apache.spark.sql.connector.catalog.{Column, MetadataColumn, SupportsMetadataColumns, Table, TableCapability}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DataType, IntegerType, MapType, Metadata, MetadataBuilder, StringType, StructField, StructType, VariantType}
+import org.apache.spark.sql.util.{CaseInsensitiveStringMap, SchemaUtils, SchemaValidationMode}
+import org.apache.spark.types.variant.VariantBuilder
+import org.apache.spark.unsafe.types.VariantVal
+
+class AnalyzedSchemaProjectionSuite extends SparkFunSuite with ExpressionEvalHelper {
+
+ test("rebind a root-level addition without inspecting unchanged column types") {
+ val capturedTable = new TestTable(Array(
+ Column.create("id", IntegerType),
+ Column.create("payload", VariantType)))
+ val currentTable = new TestTable(Array(
+ Column.create("added", StringType),
+ Column.create("id", IntegerType),
+ Column.create("payload", VariantType)))
+ val captured = DataSourceV2Relation.create(capturedTable, None, None)
+
+ val project = AnalyzedSchemaProjection
+ .rebindToAnalyzedSchema(captured.copy(table = currentTable))
+ .asInstanceOf[Project]
+
+ assert(project.child.output.map(_.name) == Seq("added", "id", "payload"))
+ val childByName = project.child.output.map(attr => attr.name -> attr).toMap
+ captured.output.foreach { capturedAttr =>
+ assert(childByName(capturedAttr.name).exprId == capturedAttr.exprId)
+ }
+ assert(project.output.map(_.name) == Seq("id", "payload"))
+ assert(project.output.map(_.dataType) == Seq(IntegerType, VariantType))
+ assert(project.output.map(_.exprId) == captured.output.map(_.exprId))
+ }
+
+ test("return the relation unchanged when the captured output still matches the table") {
+ val table = new TestTable(Array(
+ Column.create("id", IntegerType),
+ Column.create("payload", VariantType)))
+ val captured = DataSourceV2Relation.create(table, None, None)
+
+ // No projection is added, so an unchanged table keeps the plan it was analyzed with.
+ assert(AnalyzedSchemaProjection.rebindToAnalyzedSchema(captured) eq captured)
+ }
+
+ test("rebind an already rebound relation without stacking a second projection") {
+ val capturedTable = new TestTable(Array(Column.create("id", IntegerType)))
+ val currentTable = new TestTable(Array(
+ Column.create("added", StringType),
+ Column.create("id", IntegerType)))
+ val captured = DataSourceV2Relation.create(capturedTable, None, None)
+
+ val project = AnalyzedSchemaProjection
+ .rebindToAnalyzedSchema(captured.copy(table = currentTable))
+ .asInstanceOf[Project]
+
+ // `V2TableRefreshUtil` refreshes through `transformDown`, so the rule visits the relation it
+ // just rebound. Rebinding must be idempotent there: the second pass has to leave the relation
+ // and its attributes alone, or the projection above it would reference stale expression IDs.
+ val rebound = project.child.asInstanceOf[DataSourceV2Relation]
+ assert(AnalyzedSchemaProjection.rebindToAnalyzedSchema(rebound) eq rebound)
+ }
+
+ test("project a nested field addition around an unchanged variant leaf") {
+ val currentType = StructType(Seq(
+ StructField("added", IntegerType),
+ StructField("payload", VariantType, nullable = false)))
+ val capturedType =
+ StructType(Seq(StructField("payload", VariantType, nullable = false)))
+ val parsed = VariantBuilder.parseJson("""{"key":1}""", false)
+ val variant = new VariantVal(parsed.getValue, parsed.getMetadata)
+
+ val projected = project(
+ Literal(create_row(10, variant), currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, create_row(variant))
+ }
+
+ test("project structs by name while preserving null structs") {
+ val currentType = StructType(
+ Seq(StructField("age", IntegerType), StructField("name", StringType, nullable = false)))
+ val capturedType = StructType(Seq(StructField("name", StringType, nullable = false)))
+
+ val input = Literal(create_row(25, "Alice"), currentType)
+ val projected = project(input, currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, create_row("Alice"))
+ val projectedNull = project(Literal.create(null, currentType), currentType, capturedType)
+ assert(projectedNull.dataType == capturedType)
+ checkEvaluation(projectedNull, null)
+ }
+
+ test("project structs in arrays while preserving null arrays and elements") {
+ val currentElement = StructType(
+ Seq(StructField("extra", IntegerType), StructField("value", StringType, nullable = false)))
+ val capturedElement = StructType(Seq(StructField("value", StringType, nullable = false)))
+ val currentType = ArrayType(currentElement, containsNull = true)
+ val capturedType = ArrayType(capturedElement, containsNull = true)
+ val input = new GenericArrayData(Array(create_row(1, "a"), null))
+ val expected = new GenericArrayData(Array(create_row("a"), null))
+
+ val projected = project(Literal(input, currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, expected)
+ checkEvaluation(project(Literal.create(null, currentType), currentType, capturedType), null)
+ }
+
+ test("project structs nested in an array of arrays") {
+ val currentElement = StructType(
+ Seq(StructField("extra", IntegerType), StructField("value", StringType, nullable = false)))
+ val capturedElement = StructType(Seq(StructField("value", StringType, nullable = false)))
+ val currentType =
+ ArrayType(ArrayType(currentElement, containsNull = true), containsNull = true)
+ val capturedType =
+ ArrayType(ArrayType(capturedElement, containsNull = true), containsNull = true)
+ val input = new GenericArrayData(
+ Array[Any](new GenericArrayData(Array[Any](create_row(1, "a"), null)), null))
+ val expected = new GenericArrayData(
+ Array[Any](new GenericArrayData(Array[Any](create_row("a"), null)), null))
+
+ // The inner array is projected by a lambda nested inside the outer array's lambda.
+ val projected = project(Literal(input, currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, expected)
+ }
+
+ test("project structs in map keys and values") {
+ val currentKey = StructType(
+ Seq(StructField("extra", IntegerType), StructField("key", StringType, nullable = false)))
+ val capturedKey = StructType(Seq(StructField("key", StringType, nullable = false)))
+ val currentValue = StructType(
+ Seq(StructField("extra", IntegerType), StructField("value", IntegerType, nullable = false)))
+ val capturedValue = StructType(Seq(StructField("value", IntegerType, nullable = false)))
+ val currentType = MapType(currentKey, currentValue, valueContainsNull = true)
+ val capturedType = MapType(capturedKey, capturedValue, valueContainsNull = true)
+ val input =
+ create_map(Seq(create_row(1, "a"), create_row(2, "b")), Seq(create_row(3, 10), null))
+ val expected = create_map(Seq(create_row("a"), create_row("b")), Seq(create_row(10), null))
+
+ val projected = project(Literal(input, currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, expected)
+ checkEvaluation(project(Literal.create(null, currentType), currentType, capturedType), null)
+ }
+
+ test("project nested structs and preserve captured field metadata") {
+ val comment = new MetadataBuilder().putString("comment", "the name").build()
+ val currentInner = StructType(Seq(
+ StructField("added", IntegerType),
+ StructField("name", StringType, nullable = false, comment)))
+ val capturedInner =
+ StructType(Seq(StructField("name", StringType, nullable = false, comment)))
+ val currentType = StructType(Seq(
+ StructField("inner", currentInner),
+ StructField("added_outer", IntegerType)))
+ val capturedType = StructType(Seq(StructField("inner", capturedInner)))
+
+ val input = Literal(create_row(create_row(1, "Alice"), 2), currentType)
+ val projected = project(input, currentType, capturedType)
+ // StructField equality covers metadata, so this also asserts the comment survived.
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, create_row(create_row("Alice")))
+ checkEvaluation(
+ project(Literal(create_row(null, 2), currentType), currentType, capturedType),
+ create_row(null))
+ }
+
+ test("project a captured type through three levels of nested structs") {
+ // Pure struct nesting, three levels deep, with a field added at every level and a different
+ // ordinal shift at each one (l1 0 -> 1, l2 0 -> 2, leaf 0 -> 3), so reusing one level's
+ // ordinals at another level would pick the wrong field. `l1` is nullable while `l2` is not,
+ // which puts the KnownNotNull/If pair and the plain rebuild in one expression. The captured
+ // comment sits on `l2`, a middle level rather than a leaf, and differs from the current one:
+ // without `explicitMetadata` the alias would inherit the current comment from its
+ // GetStructField child and the type check below would fail.
+ val capturedComment = new MetadataBuilder().putString("comment", "captured").build()
+ val currentComment = new MetadataBuilder().putString("comment", "current").build()
+
+ val capturedL2 = StructType(Seq(StructField("leaf", StringType, nullable = false)))
+ val currentL2 = StructType(Seq(
+ StructField("added_c1", IntegerType),
+ StructField("added_c2", IntegerType),
+ StructField("added_c3", IntegerType),
+ StructField("leaf", StringType, nullable = false)))
+
+ val capturedL1 =
+ StructType(Seq(StructField("l2", capturedL2, nullable = false, capturedComment)))
+ val currentL1 = StructType(Seq(
+ StructField("added_b1", IntegerType),
+ StructField("added_b2", IntegerType),
+ StructField("l2", currentL2, nullable = false, currentComment)))
+
+ val capturedType = StructType(Seq(StructField("l1", capturedL1)))
+ val currentType = StructType(Seq(
+ StructField("added_a", IntegerType),
+ StructField("l1", currentL1)))
+
+ val input =
+ Literal(create_row(1, create_row(2, 3, create_row(4, 5, 6, "deep"))), currentType)
+ val projected = project(input, currentType, capturedType)
+ // StructField equality covers metadata, so this also asserts `l2` kept the captured comment.
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, create_row(create_row(create_row("deep"))))
+
+ // A null at the middle level, then a null at the head so two If(IsNull(...)) guards nest.
+ checkEvaluation(
+ project(Literal(create_row(1, null), currentType), currentType, capturedType),
+ create_row(null))
+ val projectedNull = project(Literal.create(null, currentType), currentType, capturedType)
+ assert(projectedNull.dataType == capturedType)
+ checkEvaluation(projectedNull, null)
+ }
+
+ test("project a captured type nested through a struct, a map, and an array") {
+ // A heterogeneous chain -- struct -> map value -> array -> struct -- with a field added at
+ // every level, each shifting the captured field to a different ordinal (m 0 -> 1, v 0 -> 1,
+ // w 0 -> 2). Reusing one level's ordinals at another level would pick the wrong field.
+ val capturedInMap = StructType(Seq(StructField("v", StringType, nullable = false)))
+ val currentInMap = StructType(Seq(
+ StructField("before_v", IntegerType),
+ StructField("v", StringType, nullable = false),
+ StructField("after_v", IntegerType)))
+ val capturedInArray = StructType(Seq(StructField("w", IntegerType, nullable = false)))
+ val currentInArray = StructType(Seq(
+ StructField("first_w", IntegerType),
+ StructField("second_w", IntegerType),
+ StructField("w", IntegerType, nullable = false)))
+
+ val capturedType = StructType(Seq(
+ StructField("m", MapType(
+ StringType, ArrayType(capturedInMap, containsNull = true), valueContainsNull = true)),
+ StructField("arr", ArrayType(capturedInArray, containsNull = true))))
+ val currentType = StructType(Seq(
+ StructField("added_first", IntegerType),
+ StructField("m", MapType(
+ StringType, ArrayType(currentInMap, containsNull = true), valueContainsNull = true)),
+ StructField("arr", ArrayType(currentInArray, containsNull = true)),
+ StructField("added_last", IntegerType)))
+
+ val currentMap = create_map(
+ Seq("k1", "k2"),
+ Seq(new GenericArrayData(Array[Any](create_row(1, "a", 2), null)), null))
+ val currentArray = new GenericArrayData(Array[Any](create_row(7, 8, 9), null))
+ val expectedMap = create_map(
+ Seq("k1", "k2"),
+ Seq(new GenericArrayData(Array[Any](create_row("a"), null)), null))
+ val expectedArray = new GenericArrayData(Array[Any](create_row(9), null))
+
+ val projected =
+ project(Literal(create_row(100, currentMap, currentArray, 200), currentType),
+ currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, create_row(expectedMap, expectedArray))
+
+ // A null map inside the struct: the map is only reached through GetStructField, so the whole
+ // TransformValues chain must stay null-safe.
+ checkEvaluation(
+ project(Literal(create_row(100, null, currentArray, 200), currentType),
+ currentType, capturedType),
+ create_row(null, expectedArray))
+
+ // A null struct at the head of the chain exercises the nullable branch, whose null literal
+ // must carry the fully rebuilt nested type.
+ val projectedNull = project(Literal.create(null, currentType), currentType, capturedType)
+ assert(projectedNull.dataType == capturedType)
+ checkEvaluation(projectedNull, null)
+ }
+
+ test("project structs in maps nested in arrays") {
+ val currentLeaf = StructType(
+ Seq(StructField("extra", IntegerType), StructField("value", StringType, nullable = false)))
+ val capturedLeaf = StructType(Seq(StructField("value", StringType, nullable = false)))
+ val currentMap = MapType(StringType, currentLeaf, valueContainsNull = true)
+ val capturedMap = MapType(StringType, capturedLeaf, valueContainsNull = true)
+ val currentType = ArrayType(currentMap, containsNull = true)
+ val capturedType = ArrayType(capturedMap, containsNull = true)
+ val input = new GenericArrayData(Array[Any](
+ create_map(Seq("k1", "k2"), Seq(create_row(1, "a"), null)),
+ null))
+ val expected = new GenericArrayData(Array[Any](
+ create_map(Seq("k1", "k2"), Seq(create_row("a"), null)),
+ null))
+
+ val projected = project(Literal(input, currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, expected)
+ }
+
+ test("project structs in maps nested in map values") {
+ val currentLeaf = StructType(
+ Seq(StructField("extra", IntegerType), StructField("value", StringType, nullable = false)))
+ val capturedLeaf = StructType(Seq(StructField("value", StringType, nullable = false)))
+ val currentInnerMap = MapType(StringType, currentLeaf, valueContainsNull = true)
+ val capturedInnerMap = MapType(StringType, capturedLeaf, valueContainsNull = true)
+ val currentType = MapType(StringType, currentInnerMap, valueContainsNull = true)
+ val capturedType = MapType(StringType, capturedInnerMap, valueContainsNull = true)
+ val input = create_map(
+ Seq("outer", "null"),
+ Seq(create_map(Seq("inner"), Seq(create_row(1, "a"))), null))
+ val expected = create_map(
+ Seq("outer", "null"),
+ Seq(create_map(Seq("inner"), Seq(create_row("a"))), null))
+
+ val projected = project(Literal(input, currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, expected)
+ }
+
+ test("project structs in arrays nested in map keys") {
+ val currentElement = StructType(
+ Seq(StructField("extra", IntegerType), StructField("key", StringType, nullable = false)))
+ val capturedElement = StructType(Seq(StructField("key", StringType, nullable = false)))
+ val currentKey = ArrayType(currentElement, containsNull = true)
+ val capturedKey = ArrayType(capturedElement, containsNull = true)
+ val currentType = MapType(currentKey, IntegerType, valueContainsNull = false)
+ val capturedType = MapType(capturedKey, IntegerType, valueContainsNull = false)
+ val inputKey = new GenericArrayData(Array[Any](create_row(1, "a"), null))
+ val expectedKey = new GenericArrayData(Array[Any](create_row("a"), null))
+
+ val projected = project(
+ Literal(create_map(Seq(inputKey), Seq(10)), currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, create_map(Seq(expectedKey), Seq(10)))
+ }
+
+ test("project only map values when the key type is unchanged") {
+ val currentValue = StructType(
+ Seq(StructField("extra", IntegerType), StructField("value", IntegerType, nullable = false)))
+ val capturedValue = StructType(Seq(StructField("value", IntegerType, nullable = false)))
+ val currentType = MapType(StringType, currentValue, valueContainsNull = true)
+ val capturedType = MapType(StringType, capturedValue, valueContainsNull = true)
+ val input = create_map(Seq("a"), Seq(create_row(1, 10)))
+ val expected = create_map(Seq("a"), Seq(create_row(10)))
+
+ val projected = project(Literal(input, currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, expected)
+ }
+
+ test("project only map keys when the value type is unchanged") {
+ val currentKey = StructType(
+ Seq(StructField("extra", IntegerType), StructField("key", StringType, nullable = false)))
+ val capturedKey = StructType(Seq(StructField("key", StringType, nullable = false)))
+ val currentType = MapType(currentKey, IntegerType, valueContainsNull = true)
+ val capturedType = MapType(capturedKey, IntegerType, valueContainsNull = true)
+ val input = create_map(Seq(create_row(1, "a")), Seq(10))
+ val expected = create_map(Seq(create_row("a")), Seq(10))
+
+ val projected = project(Literal(input, currentType), currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, expected)
+ }
+
+ test("dropping a map key field that made keys distinct fails at runtime") {
+ val currentKey = StructType(
+ Seq(StructField("extra", IntegerType), StructField("key", StringType, nullable = false)))
+ val capturedKey = StructType(Seq(StructField("key", StringType, nullable = false)))
+ val currentType = MapType(currentKey, IntegerType, valueContainsNull = true)
+ val capturedType = MapType(capturedKey, IntegerType, valueContainsNull = true)
+ // Both keys collapse to {key: "a"} once `extra` is dropped. Narrowing a key changes how many
+ // entries the map holds, so it is a fold rather than a projection, and the captured type has no
+ // correct answer to return. Rebuilding through `TransformKeys` keeps map-key uniqueness, so the
+ // outcome is whatever `mapKeyDedupPolicy` prescribes rather than a map with duplicate keys.
+ val input = create_map(Seq(create_row(1, "a"), create_row(2, "a")), Seq(10, 20))
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+ checkErrorInExpression[SparkRuntimeException](
+ project(Literal(input, currentType), currentType, capturedType),
+ condition = "DUPLICATED_MAP_KEY",
+ parameters = Map(
+ "key" -> "[a]",
+ "mapKeyDedupPolicy" -> "\"spark.sql.mapKeyDedupPolicy\""))
+ }
+
+ withSQLConf(
+ SQLConf.MAP_KEY_DEDUP_POLICY.key -> SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ checkEvaluation(
+ project(Literal(input, currentType), currentType, capturedType),
+ create_map(Seq(create_row("a")), Seq(20)))
+ }
+ }
+
+ test("match nested field names case-insensitively but not under case-sensitive analysis") {
+ val currentType = StructType(
+ Seq(StructField("added", IntegerType), StructField("NAME", StringType, nullable = false)))
+ val capturedType = StructType(Seq(StructField("name", StringType, nullable = false)))
+ val input = Literal(create_row(1, "Alice"), currentType)
+
+ // Case-insensitive analysis resolves the captured `name` to the current `NAME` and restores the
+ // captured field name.
+ val projected = project(input, currentType, capturedType)
+ assert(projected.dataType == capturedType)
+ checkEvaluation(projected, create_row("Alice"))
+
+ // Case-sensitive analysis treats them as different fields.
+ checkRejected(
+ input,
+ currentType,
+ capturedType,
+ "captured struct field name is missing from",
+ caseSensitive = true)
+ }
+
+ test("reject a captured column the fold separates from every current name") {
+ // U+0130 CAPITAL I WITH DOT separates the two halves of the identity rule: `equalsIgnoreCase`
+ // equates it with `i`, while folding with `toLowerCase` maps it to `i` plus a combining dot.
+ // Top-level resolution collects candidates by the fold before applying the resolver, so the
+ // fold alone already rules this pair out: it names two different columns, and rebinding must
+ // report the captured one missing rather than bind it to the other.
+ val capitalIWithDot = new String(Character.toChars(0x130))
+ val capturedNested = StructType(Seq(StructField("i", StringType, nullable = false)))
+ val currentNested =
+ StructType(Seq(StructField(capitalIWithDot, StringType, nullable = false)))
+ val currentTable = new TestTable(
+ Array(Column.create(capitalIWithDot, currentNested, false)))
+ val relation = DataSourceV2Relation(
+ table = currentTable,
+ output = Seq(AttributeReference("i", capturedNested, nullable = false)()),
+ catalog = None,
+ identifier = None,
+ options = CaseInsensitiveStringMap.empty())
+
+ // Refresh rejects this evolution before rebinding runs, so the internal error below reports a
+ // validation gap and is not reachable through a query.
+ val capturedSchema = StructType(Seq(StructField("i", capturedNested, nullable = false)))
+ val currentSchema =
+ StructType(Seq(StructField(capitalIWithDot, currentNested, nullable = false)))
+ Seq(false, true).foreach { caseSensitive =>
+ val resolver = if (caseSensitive) caseSensitiveResolution else caseInsensitiveResolution
+ assert(
+ SchemaUtils.validateSchemaCompatibility(
+ capturedSchema,
+ currentSchema,
+ resolver,
+ SchemaValidationMode.ALLOW_NEW_FIELDS,
+ checkFieldIds = false).nonEmpty,
+ s"validation must reject this pair (caseSensitive = $caseSensitive)")
+
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive.toString) {
+ val e = intercept[SparkException] {
+ AnalyzedSchemaProjection.rebindToAnalyzedSchema(relation)
+ }
+ assert(e.getCondition == "INTERNAL_ERROR")
+ assert(e.getMessage.contains("captured column i is missing from current table"))
+ }
+ }
+ }
+
+ test("bind a captured column to the one that folds alike, not a resolver-equal sibling") {
+ // A table can legally hold both `s` and U+017F LONG S: the duplicate-name check folds names
+ // with `toLowerCase`, which keeps them apart, while the resolver compares with
+ // `equalsIgnoreCase`, which does not. The captured name must read the column it folds to.
+ val longS = new String(Character.toChars(0x17f))
+ val currentTable = new TestTable(Array(
+ Column.create(longS, IntegerType),
+ Column.create("s", IntegerType)))
+ val relation = DataSourceV2Relation(
+ table = currentTable,
+ output = Seq(AttributeReference("s", IntegerType)()),
+ catalog = None,
+ identifier = None,
+ options = CaseInsensitiveStringMap.empty())
+
+ val rebound = AnalyzedSchemaProjection.rebindToAnalyzedSchema(relation).asInstanceOf[Project]
+ assert(rebound.child.output.map(_.name) == Seq(longS, "s"))
+ assert(rebound.output.map(_.name) == Seq("s"))
+ assert(rebound.output.map(_.exprId) == relation.output.map(_.exprId))
+ val read = rebound.projectList.head.references.toSeq
+ assert(read.map(_.name) == Seq("s"), "the captured column must not read the folding sibling")
+ assert(read.map(_.exprId) == Seq(rebound.child.output.last.exprId))
+ }
+
+ test("a resolver-equal sibling makes a captured struct field ambiguous") {
+ // Struct-field resolution compares with the resolver alone, so `S` and U+017F are both equal to
+ // `s` and a fresh `SELECT st.s` fails. Rebinding reports the same ambiguity rather than picking
+ // one, in either order. The same pair at the top level is NOT ambiguous, because top-level
+ // resolution folds first - see "bind a captured column to the one that folds alike ..." above.
+ val capturedType = StructType(Seq(StructField("s", IntegerType)))
+ Seq(
+ Seq(StructField(longS, IntegerType), StructField("s", IntegerType)),
+ Seq(StructField("S", IntegerType), StructField(longS, IntegerType))).foreach { fields =>
+ val currentType = StructType(fields)
+ val e = intercept[Throwable] {
+ project(Literal(create_row(1, 2), currentType), currentType, capturedType)
+ }
+ assert(condition(e) == "AMBIGUOUS_REFERENCE_TO_FIELDS", s"for ${fields.map(_.name)}")
+ }
+ }
+
+ test("duplicate struct field names report the ambiguity resolution reports") {
+ // The duplicate-name check rejects both of these schemas, so this path is unreachable through
+ // refresh. It still must not invent an answer: struct-field resolution calls both ambiguous, so
+ // reaching it without that check has to surface the same error, not silently take an ordinal.
+ val capturedType = StructType(Seq(StructField("dup", IntegerType)))
+ Seq(
+ Seq(StructField("dup", IntegerType), StructField("dup", IntegerType)),
+ Seq(StructField("dup", IntegerType), StructField("DUP", IntegerType))).foreach { fields =>
+ val currentType = StructType(fields)
+ val e = intercept[Throwable] {
+ project(Literal(create_row(1, 2), currentType), currentType, capturedType)
+ }
+ assert(condition(e) == "AMBIGUOUS_REFERENCE_TO_FIELDS", s"for ${fields.map(_.name)}")
+ }
+ }
+
+ test("reject a captured field that is missing from the current type") {
+ val currentType = StructType(Seq(StructField("name", StringType)))
+ val capturedType = StructType(Seq(StructField("missing", StringType)))
+
+ checkRejected(
+ Literal.create(null, currentType),
+ currentType,
+ capturedType,
+ "captured struct field missing is missing from")
+ }
+
+ test("reject an incompatible captured type") {
+ checkRejected(Literal(1), IntegerType, StringType, "cannot project incompatible data type")
+ }
+
+ test("reject a changed array element nullability") {
+ val current = ArrayType(IntegerType, containsNull = true)
+ val captured = ArrayType(IntegerType, containsNull = false)
+
+ checkRejected(
+ Literal.create(null, current), current, captured, "array element nullability changed")
+ }
+
+ test("reject a changed map value nullability") {
+ val current = MapType(StringType, IntegerType, valueContainsNull = true)
+ val captured = MapType(StringType, IntegerType, valueContainsNull = false)
+
+ checkRejected(
+ Literal.create(null, current), current, captured, "map value nullability changed")
+ }
+
+ test("reject a captured column whose nullability changed") {
+ // Validation is expected to reject a nullability change before rebinding runs, so getting here
+ // is an internal error. The differing nullability is also what stops the captured attribute
+ // from being reused, which is how the column reaches this check.
+ val relation = DataSourceV2Relation(
+ table = new TestTable(Array(Column.create("id", IntegerType, false))),
+ output = Seq(AttributeReference("id", IntegerType, nullable = true)()),
+ catalog = None,
+ identifier = None,
+ options = CaseInsensitiveStringMap.empty())
+
+ val e = intercept[SparkException] {
+ AnalyzedSchemaProjection.rebindToAnalyzedSchema(relation)
+ }
+ assert(e.getCondition == "INTERNAL_ERROR")
+ assert(e.getMessage.contains("Unexpected incompatible table schema after refresh validation"))
+ assert(e.getMessage.contains("nullability changed for captured column id"))
+ }
+
+ test("reject a captured metadata column that a data column hides") {
+ // The connector keeps reporting `index` as a metadata column, but a data column has taken that
+ // name, and this connector suppresses rather than renames the conflict. Validation is expected
+ // to reject this before rebinding runs, so it fails as an internal error here.
+ val relation = DataSourceV2Relation(
+ table = new TestMetadataTable(canRename = false),
+ output = Seq(
+ AttributeReference("id", IntegerType)(),
+ MetadataAttribute("index", IntegerType, nullable = false)),
+ catalog = None,
+ identifier = None,
+ options = CaseInsensitiveStringMap.empty())
+
+ val e = intercept[SparkException] {
+ AnalyzedSchemaProjection.rebindToAnalyzedSchema(relation)
+ }
+ assert(e.getCondition == "INTERNAL_ERROR")
+ assert(e.getMessage.contains("Unexpected incompatible table schema after refresh validation"))
+ assert(e.getMessage.contains("captured metadata column index is missing from"))
+ }
+
+ test("rebind a captured metadata column that the connector renames on conflict") {
+ // Same conflict, but this connector renames the metadata column, so it stays readable and
+ // rebinding succeeds instead of failing.
+ val relation = DataSourceV2Relation(
+ table = new TestMetadataTable(canRename = true),
+ output = Seq(
+ AttributeReference("id", IntegerType)(),
+ MetadataAttribute("index", IntegerType, nullable = false)),
+ catalog = None,
+ identifier = None,
+ options = CaseInsensitiveStringMap.empty())
+
+ val rebound = AnalyzedSchemaProjection.rebindToAnalyzedSchema(relation)
+ val project = rebound.asInstanceOf[Project]
+ assert(project.output.map(_.name) == Seq("id", "index"))
+ // Rebinding exists to keep the captured expression IDs valid for the parent plan.
+ assert(project.output.map(_.exprId) == relation.output.map(_.exprId))
+ // The relation exposes the new data column too, but only so the scan stays aligned with it --
+ // the projection must drop it rather than let it shadow the captured metadata column.
+ assert(project.child.output.map(_.name) == Seq("id", "index", "_index"))
+ // Checking names alone cannot tell the renamed metadata column apart from the new data column,
+ // since both are called `index` at some point. Assert what the captured `index` actually reads.
+ val indexChild =
+ project.projectList(1).asInstanceOf[Alias].child.asInstanceOf[AttributeReference]
+ assert(indexChild.isMetadataCol, "the captured index must read the metadata column")
+ assert(indexChild.name == "_index", "which the connector exposes under a renamed name")
+ // Metadata-column-ness survives here either way: the alias child is a NamedExpression, so an
+ // alias without explicit metadata would inherit its `__metadata_col` key. What the captured
+ // metadata pins is the exact value, not whatever the current relation happens to expose.
+ assert(project.output(1).isMetadataCol, "the projected index must stay a metadata column")
+ assert(project.output(1).metadata == relation.output(1).metadata)
+ }
+
+ test("optimizer cleanup distinguishes explicit empty from unspecified empty nested metadata") {
+ val currentComment = new MetadataBuilder().putString("comment", "current").build()
+ val currentType = StructType(Seq(
+ StructField("added", IntegerType),
+ StructField("name", StringType, nullable = false, currentComment)))
+ val capturedType =
+ StructType(Seq(StructField("name", StringType, nullable = false, Metadata.empty)))
+ val input = AttributeReference("person", currentType, nullable = false)()
+
+ val rebuilt = project(input, currentType, capturedType)
+ val cleanedRebuild = FoldablePropagation(
+ Project(Seq(Alias(rebuilt, "person")()), LocalRelation(input))).asInstanceOf[Project]
+ // The captured empty metadata is observable in the optimized output even though shared alias
+ // cleanup remains unchanged and may remove empty-metadata aliases.
+ assert(cleanedRebuild.output.head.dataType == capturedType)
+
+ val unspecifiedEmpty = CreateNamedStruct(Seq(
+ Literal("value"),
+ Alias(Literal(1), "value")()))
+ val cleanedUnspecified = FoldablePropagation(
+ Project(Seq(Alias(unspecifiedEmpty, "record")()), LocalRelation(input)))
+ .asInstanceOf[Project]
+ .projectList
+ .head
+ .asInstanceOf[Alias]
+ .child
+ .asInstanceOf[CreateNamedStruct]
+ assert(!cleanedUnspecified.children(1).isInstanceOf[Alias])
+ }
+
+ test("preserve captured field metadata on a rebuilt nested column") {
+ // The alias over a rebuilt struct cannot inherit metadata from its child, because that child is
+ // a CreateNamedStruct rather than a NamedExpression. Attribute reuse is inapplicable because
+ // this path rebuilds the expression, so only explicit captured metadata keeps the column
+ // comment on the rebound column.
+ val comment = new MetadataBuilder().putString("comment", "captured").build()
+ val capturedStruct = StructType(Seq(StructField("name", StringType, nullable = false)))
+ val currentStruct = StructType(Seq(
+ StructField("added", IntegerType),
+ StructField("name", StringType, nullable = false)))
+ val relation = DataSourceV2Relation(
+ table = new TestTable(Array(Column.create("person", currentStruct, true))),
+ output = Seq(AttributeReference("person", capturedStruct, nullable = true, comment)()),
+ catalog = None,
+ identifier = None,
+ options = CaseInsensitiveStringMap.empty())
+
+ val project = AnalyzedSchemaProjection.rebindToAnalyzedSchema(relation).asInstanceOf[Project]
+ assert(project.child.output.map(_.dataType) == Seq(currentStruct))
+ assert(project.output.map(_.dataType) == Seq(capturedStruct))
+ assert(project.output.head.metadata == comment, "the captured comment must survive the rebuild")
+ }
+
+ test("name matching matches Spark's own resolution (recorded outcomes)") {
+ nameCases.foreach { c =>
+ assert(
+ rebindTopLevel(c.captured, c.current) == c.topExpected,
+ s"top-level '${c.label}': captured ${escapeName(c.captured)} " +
+ s"against ${c.current.map(escapeName)}")
+ assert(
+ projectNested(c.captured, c.current) == c.nestedExpected,
+ s"nested '${c.label}': captured ${escapeName(c.captured)} " +
+ s"against ${c.current.map(escapeName)}")
+ }
+ }
+
+ test("name matching delegates to Spark's own resolution rather than reimplementing it") {
+ // Guards against a future rule of our own drifting from resolution: it compares against the
+ // live APIs, so it also covers name pairs the recorded table above does not list.
+ nameCases.foreach { c =>
+ assert(
+ rebindTopLevel(c.captured, c.current) == topAuthority(c.captured, c.current),
+ s"top-level '${c.label}' must answer what AttributeSeq.resolve answers")
+ assert(
+ projectNested(c.captured, c.current) == nestedAuthority(c.captured, c.current),
+ s"nested '${c.label}' must answer what ExtractValue.extractValue answers")
+ }
+ }
+
+ test("exact duplicate names resolve the way Spark resolves them") {
+ // Kept apart from `nameCases` because two columns with the identical name make `canReuse`
+ // give both the captured attribute, so the rebound output holds one attribute twice rather
+ // than two ambiguous ones. The duplicate-name check rejects this schema under every locale.
+ assert(topAuthority("dup", Seq("dup", "dup")) == Left("AMBIGUOUS_REFERENCE"))
+ assert(nestedAuthority("dup", Seq("dup", "dup")) == Left("AMBIGUOUS_REFERENCE_TO_FIELDS"))
+ assert(projectNested("dup", Seq("dup", "dup")) == Left("AMBIGUOUS_REFERENCE_TO_FIELDS"))
+ }
+
+ test("which locales admit a schema whose names collide under the fold") {
+ // Pins the upstream facts this rebinding must not assume away: the duplicate-name check folds
+ // with the JVM default locale (`SchemaUtils.checkColumnNameDuplication`) while every other fold
+ // here uses `Locale.ROOT`, so a Turkish or Lithuanian default locale admits a pair that an
+ // English one rejects. Rebinding must therefore not rely on the fold being single-valued.
+ def admits(names: Seq[String], locale: String): Boolean = {
+ val previous = util.Locale.getDefault
+ try {
+ util.Locale.setDefault(util.Locale.forLanguageTag(locale))
+ SchemaUtils.checkColumnNameDuplication(names, caseSensitiveAnalysis = false)
+ true
+ } catch {
+ case _: Throwable => false
+ } finally {
+ util.Locale.setDefault(previous)
+ }
+ }
+
+ assert(admits(Seq("S", longS), "en"), "the fold keeps S and U+017F apart under any locale")
+ assert(!admits(Seq(capitalIDot, iCombiningDot), "en"))
+ assert(admits(Seq(capitalIDot, iCombiningDot), "tr"))
+ assert(!admits(Seq(capitalIGrave, smallIGrave), "en"))
+ assert(admits(Seq(capitalIGrave, smallIGrave), "lt"))
+ assert(!admits(Seq("dup", "dup"), "en") && !admits(Seq("dup", "dup"), "tr"))
+ }
+
+ test("refresh validation does not blame field IDs for a fold collision") {
+ // With the addition after the captured column, `SchemaUtils.index` folds both current names to
+ // one key and keeps the later field, so the captured field is compared against the addition and
+ // its ID mismatch is reported - a legal change rejected for the wrong reason.
+ //
+ // Runs under a Turkish default locale because that is what makes the schema admissible at all:
+ // the duplicate-name check folds with the default locale and would otherwise reject the pair.
+ val captured = StructType(Seq(StructField(iCombiningDot, IntegerType).withId("1")))
+ val previousLocale = util.Locale.getDefault
+ try {
+ util.Locale.setDefault(util.Locale.forLanguageTag("tr"))
+ Seq(
+ "addition first" -> Seq(capitalIDot -> "2", iCombiningDot -> "1"),
+ "addition last" -> Seq(iCombiningDot -> "1", capitalIDot -> "2")).foreach {
+ case (label, fields) =>
+ val current = StructType(fields.map { case (n, id) =>
+ StructField(n, IntegerType).withId(id)
+ })
+ val errors = SchemaUtils.validateSchemaCompatibility(
+ captured,
+ current,
+ caseInsensitiveResolution,
+ SchemaValidationMode.ALLOW_NEW_FIELDS,
+ checkFieldIds = true)
+ assert(
+ !errors.exists(_.contains("field ID has changed")),
+ s"$label must not report a field ID change: $errors")
+ }
+ } finally {
+ util.Locale.setDefault(previousLocale)
+ }
+ }
+
+ test("array elements and map entries match names the same way as a bare struct") {
+ // `projectToType` recurses into containers through the same name matching, so a pair that is
+ // ambiguous in a bare struct must stay ambiguous inside an array, a map key and a map value.
+ val currentField = StructType(
+ Seq(StructField("S", IntegerType), StructField(longS, IntegerType)))
+ val capturedField = StructType(Seq(StructField("s", IntegerType)))
+
+ Seq[(String, DataType, DataType)](
+ ("array element",
+ ArrayType(currentField, containsNull = false),
+ ArrayType(capturedField, containsNull = false)),
+ ("map value",
+ MapType(IntegerType, currentField, valueContainsNull = false),
+ MapType(IntegerType, capturedField, valueContainsNull = false)),
+ ("map key",
+ MapType(currentField, IntegerType, valueContainsNull = false),
+ MapType(capturedField, IntegerType, valueContainsNull = false))).foreach {
+ case (label, from, to) =>
+ val e = intercept[Throwable](project(AttributeReference("c", from)(), from, to))
+ assert(
+ condition(e) == "AMBIGUOUS_REFERENCE_TO_FIELDS",
+ s"$label must report the ambiguity struct-field resolution reports, got ${condition(e)}")
+ }
+ }
+
+ test("refresh validation and rebinding agree on which current field a captured name refers to") {
+ // Refresh validation runs first and rebinding assumes it passed. Two things must therefore
+ // never happen: validation accepting a schema that rebinding then cannot map (which would
+ // surface as INTERNAL_ERROR), and rebinding mapping a name that validation would have
+ // rejected. The pairs below make the two halves of the identity rule disagree in each
+ // direction, so a rule that applied only one half fails here.
+ //
+ // The one divergence that is allowed is a struct field the resolver finds ambiguous: struct
+ // fields are resolved with the resolver alone, so rebinding reports the ambiguity a fresh
+ // query reports even though validation, which folds first, paired the field successfully.
+ val rows = Seq(
+ ("fold-equal and resolver-equal", "c", Seq("C")),
+ ("fold-equal but resolver-unequal", iCombiningDot, Seq(capitalIDot)),
+ ("fold-unequal but resolver-equal", "s", Seq(longS)),
+ ("fold-equal beside an ambiguous sibling", "s", Seq("S", longS)))
+
+ rows.foreach { case (label, captured, current) =>
+ val capturedType = StructType(Seq(StructField(captured, IntegerType)))
+ val currentType = StructType(current.map(n => StructField(n, IntegerType)))
+ val validationPasses = SchemaUtils.validateSchemaCompatibility(
+ capturedType,
+ currentType,
+ caseInsensitiveResolution,
+ SchemaValidationMode.ALLOW_NEW_FIELDS,
+ checkFieldIds = false).isEmpty
+
+ Seq(
+ ("top-level", rebindTopLevel(captured, current), topAuthority(captured, current)),
+ ("nested", projectNested(captured, current), nestedAuthority(captured, current))
+ ).foreach { case (level, ours, authority) =>
+ if (validationPasses) {
+ assert(
+ ours != Left("INTERNAL_ERROR"),
+ s"$level '$label': validation accepted this schema, so rebinding must map the name " +
+ s"or report what a fresh query reports, not an internal error")
+ assert(ours == authority, s"$level '$label': must answer what resolution answers")
+ }
+ }
+
+ // The other direction is guarded by order rather than by the rules agreeing: validation folds
+ // names, so it calls a captured field removed when only a name the fold separates but the
+ // resolver equates is left, while struct-field resolution would still read that field. What
+ // keeps the two from disagreeing in practice is that `V2TableRefreshUtil.refresh` throws
+ // `columnsChangedAfterAnalysis` before it rebinds, so this pair never reaches the projection.
+ if (!validationPasses) {
+ assert(
+ SchemaUtils.validateSchemaCompatibility(
+ capturedType,
+ currentType,
+ caseInsensitiveResolution,
+ SchemaValidationMode.ALLOW_NEW_FIELDS,
+ checkFieldIds = false).nonEmpty,
+ s"'$label' must be rejected by validation, which is what stops it reaching rebinding")
+ }
+ }
+ }
+
+ // U+017F LONG S folds to itself but `equalsIgnoreCase` equates it with `s`.
+ private val longS = new String(Character.toChars(0x17f))
+ // U+0130 CAPITAL I WITH DOT folds to `i` plus a combining dot, which `equalsIgnoreCase` does
+ // not equate with either. A Turkish default locale keeps the two apart for the duplicate check.
+ private val capitalIDot = new String(Character.toChars(0x130))
+ private val iCombiningDot = "i" + new String(Character.toChars(0x307))
+ // U+00CC/U+00EC differ only by case to both rules, so they are ambiguous rather than distinct.
+ private val capitalIGrave = new String(Character.toChars(0xcc))
+ private val smallIGrave = new String(Character.toChars(0xec))
+
+ /**
+ * One row of the name-matching contract.
+ *
+ * `topExpected` and `nestedExpected` are the recorded behaviour of Spark's own resolution:
+ * `AttributeSeq.resolve` for a top-level column and `ExtractValue.extractValue` for a struct
+ * field. They are written as literals rather than derived from those APIs at run time, so that a
+ * change on either side - ours or Catalyst's - turns this test red instead of silently following.
+ * `Left` is an expected error condition, `Right` an expected ordinal in `current`.
+ *
+ * The two levels differ because Spark's two rules differ: top-level resolution folds names with
+ * `toLowerCase(ROOT)` to collect candidates and only then filters them with the resolver, while
+ * struct-field resolution compares with the resolver alone. A pair the fold separates but the
+ * resolver equates is therefore a distinct column at the top level and ambiguous inside a struct.
+ */
+ private case class NameCase(
+ label: String,
+ captured: String,
+ current: Seq[String],
+ topExpected: Either[String, Int],
+ nestedExpected: Either[String, Int])
+
+ private val nameCases = Seq(
+ NameCase("ascii case rename", "c", Seq("C"), Right(0), Right(0)),
+ NameCase("longS after the captured name", "s", Seq("S", longS),
+ Right(0), Left("AMBIGUOUS_REFERENCE_TO_FIELDS")),
+ NameCase("longS before the captured name", "s", Seq(longS, "S"),
+ Right(1), Left("AMBIGUOUS_REFERENCE_TO_FIELDS")),
+ NameCase("fold-colliding addition first", iCombiningDot, Seq(capitalIDot, iCombiningDot),
+ Right(1), Right(1)),
+ NameCase("fold-colliding addition last", iCombiningDot, Seq(iCombiningDot, capitalIDot),
+ Right(0), Right(0)),
+ NameCase("resolver-equal pair first", smallIGrave, Seq(capitalIGrave, smallIGrave),
+ Left("AMBIGUOUS_REFERENCE"), Left("AMBIGUOUS_REFERENCE_TO_FIELDS")),
+ NameCase("resolver-equal pair last", smallIGrave, Seq(smallIGrave, capitalIGrave),
+ Left("AMBIGUOUS_REFERENCE"), Left("AMBIGUOUS_REFERENCE_TO_FIELDS")),
+ NameCase("case-folding duplicates", "dup", Seq("dup", "DUP"),
+ Left("AMBIGUOUS_REFERENCE"), Left("AMBIGUOUS_REFERENCE_TO_FIELDS")))
+
+ private def condition(t: Throwable): String = t match {
+ case s: SparkThrowable => s.getCondition
+ case other => other.getClass.getSimpleName
+ }
+
+ /** Rebinds a captured column against `current` and reports the ordinal it reads. */
+ private def rebindTopLevel(captured: String, current: Seq[String]): Either[String, Int] = {
+ val relation = DataSourceV2Relation(
+ table = new TestTable(current.map(n => Column.create(n, IntegerType)).toArray),
+ output = Seq(AttributeReference(captured, IntegerType)()),
+ catalog = None,
+ identifier = None,
+ options = CaseInsensitiveStringMap.empty())
+ try {
+ val rebound = AnalyzedSchemaProjection.rebindToAnalyzedSchema(relation).asInstanceOf[Project]
+ val read = rebound.projectList.head.references.head
+ Right(rebound.child.output.indexWhere(_.exprId == read.exprId))
+ } catch {
+ case e: Throwable => Left(condition(e))
+ }
+ }
+
+ /** Projects a captured struct field out of `current` and reports the ordinal it extracts. */
+ private def projectNested(captured: String, current: Seq[String]): Either[String, Int] = {
+ val currentType = StructType(current.map(n => StructField(n, IntegerType)))
+ val capturedType = StructType(Seq(StructField(captured, IntegerType)))
+ try {
+ val projected = project(AttributeReference("st", currentType)(), currentType, capturedType)
+ projected.collect { case g: GetStructField => g.ordinal } match {
+ case Seq(ordinal) => Right(ordinal)
+ case ordinals => Left(s"expected one extraction, got $ordinals")
+ }
+ } catch {
+ case e: Throwable => Left(condition(e))
+ }
+ }
+
+ private def topAuthority(captured: String, current: Seq[String]): Either[String, Int] = {
+ val attrs = current.map(n => AttributeReference(n, IntegerType)())
+ try {
+ AttributeSeq(attrs).resolve(Seq(captured), caseInsensitiveResolution) match {
+ case Some(resolved) =>
+ Right(attrs.indexWhere(_.exprId == resolved.references.head.exprId))
+ case None => Left("NO MATCH")
+ }
+ } catch {
+ case e: Throwable => Left(condition(e))
+ }
+ }
+
+ private def nestedAuthority(captured: String, current: Seq[String]): Either[String, Int] = {
+ val currentType = StructType(current.map(n => StructField(n, IntegerType)))
+ ExtractValue
+ .extractValue(
+ AttributeReference("st", currentType)(), Literal(captured), caseInsensitiveResolution)
+ .fold(
+ extracted => Right(extracted.asInstanceOf[GetStructField].ordinal),
+ throwable => Left(condition(throwable)))
+ }
+
+ private def escapeName(name: String): String =
+ if (name.forall(c => c > 0x20 && c < 0x7f)) s"'$name'"
+ else name.map(c => f"U+$c%04X").mkString("+")
+
+ private def project(
+ input: Expression,
+ from: DataType,
+ to: DataType,
+ caseSensitive: Boolean = false): Expression = {
+ val resolver = if (caseSensitive) caseSensitiveResolution else caseInsensitiveResolution
+ AnalyzedSchemaProjection.projectToType(input, from, to, resolver)
+ }
+
+ private def checkRejected(
+ input: Expression,
+ from: DataType,
+ to: DataType,
+ expectedMessage: String,
+ caseSensitive: Boolean = false): Unit = {
+ val e = intercept[SparkException](project(input, from, to, caseSensitive))
+ assert(e.getCondition == "INTERNAL_ERROR")
+ assert(e.getMessage.contains("Unexpected incompatible table schema after refresh validation"))
+ assert(e.getMessage.contains(expectedMessage))
+ }
+
+ /** A minimal table with no metadata columns. */
+ private class TestTable(cols: Array[Column]) extends Table {
+ override def name(): String = "test_table"
+
+ override def columns(): Array[Column] = cols
+
+ override def capabilities(): util.Set[TableCapability] = util.Set.of(TableCapability.BATCH_READ)
+ }
+
+ /** A table with an `index` metadata column whose name a data column has taken over. */
+ private class TestMetadataTable(canRename: Boolean) extends Table with SupportsMetadataColumns {
+ override def name(): String = "test_table"
+
+ override def columns(): Array[Column] = Array(
+ Column.create("id", IntegerType),
+ Column.create("index", IntegerType))
+
+ override def capabilities(): util.Set[TableCapability] = util.Set.of(TableCapability.BATCH_READ)
+
+ override def metadataColumns(): Array[MetadataColumn] = Array(new MetadataColumn {
+ override def name(): String = "index"
+ override def dataType(): DataType = IntegerType
+ override def isNullable: Boolean = false
+ })
+
+ override val canRenameConflictingMetadataColumns: Boolean = canRename
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org