diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala index c7a2165480854..aae020735bfd0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala @@ -607,8 +607,8 @@ object VirtualColumn { } /** - * The internal representation of the MetadataAttribute, - * it sets `__metadata_col` to `true` in AttributeReference metadata + * The internal representation of the MetadataAttribute. It stores the metadata column's logical + * name under `__metadata_col` in AttributeReference metadata. * - apply() will create a metadata attribute reference * - unapply() will check if an attribute reference is the metadata attribute reference */ @@ -682,8 +682,9 @@ object MetadataStructFieldWithLogicalName { } /** - * The internal representation of the FileSourceMetadataAttribute, it sets `__metadata_col` - * and `__file_source_metadata_col` to `true` in AttributeReference's metadata. + * The internal representation of the FileSourceMetadataAttribute. It stores the metadata column's + * logical name under `__metadata_col` and sets `__file_source_metadata_col` to `true` in + * AttributeReference metadata. * This is a super type of [[FileSourceConstantMetadataAttribute]] and * [[FileSourceGeneratedMetadataAttribute]]. * 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 ec98d2c0fbe99..2b8ac6404ea6e 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 @@ -21,7 +21,8 @@ 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.util.{quoteIfNeeded, MetadataColumnHelper} +import org.apache.spark.sql.catalyst.expressions.MetadataAttributeWithLogicalName +import org.apache.spark.sql.catalyst.util.{quoteIdentifier, quoteIfNeeded} import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.IdentifierHelper import org.apache.spark.sql.errors.QueryCompilationErrors import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation @@ -100,7 +101,9 @@ private[sql] object V2TableUtil extends SQLConfHelper { * @return metadata columns captured by the relation */ def extractMetadataColumns(relation: DataSourceV2Relation): Seq[MetadataColumn] = { - val metaAttrNames = relation.output.filter(_.isMetadataCol).map(_.name) + val metaAttrNames = relation.output.collect { + case MetadataAttributeWithLogicalName(_, logicalName) => logicalName + } if (metaAttrNames.isEmpty) Nil else filter(metaAttrNames, metadataColumns(relation.table)) } @@ -111,6 +114,8 @@ private[sql] object V2TableUtil extends SQLConfHelper { * - Column ID changes (top-level and nested field IDs) * - Metadata column type or nullability changes * - Removed metadata columns (missing from current table) + * - Metadata columns hidden by a same-named data column added after capture (when the connector + * suppresses rather than renames the conflict) * * @param table the current table metadata * @param originMetaCols the originally captured metadata columns @@ -130,7 +135,42 @@ private[sql] object V2TableUtil extends SQLConfHelper { val originMetaSchema = CatalogV2Util.toStructType(originMetaCols) val metaCols = filter(originMetaColNames, metadataColumns(table)) val metaSchema = CatalogV2Util.toStructType(metaCols) - SchemaUtils.validateSchemaCompatibility(originMetaSchema, metaSchema, resolver, mode, checkIds) + val schemaErrors = SchemaUtils.validateSchemaCompatibility( + originMetaSchema, metaSchema, resolver, mode, checkIds) + schemaErrors ++ shadowedMetadataColumnErrors(table, metaCols) + } + + /** + * Reports captured metadata columns that a data column of the same name now hides. + * + * When a data column takes a metadata column's name, a fresh relation for a connector that does + * not rename the conflict (`canRenameConflictingMetadataColumns` is false) suppresses the + * metadata column via `metadataOutputWithOutConflicts`. A refreshed relation retains an already + * captured metadata attribute in its output. Later, `PushDownUtils.toOutputAttrs` reconciles the + * scan schema to that output by physical name, which can bind the same-named data field to the + * metadata attribute and return the data column's values. The `SupportsMetadataColumns` contract + * advises a non-renaming source to reject such a data-column name but does not enforce it, so + * this reports the conflict rather than leaving it silent. + */ + private def shadowedMetadataColumnErrors( + table: Table, + reportedMetaCols: Seq[MetadataColumn]): Seq[String] = { + if (reportedMetaCols.isEmpty || renamesConflictingMetadataColumns(table)) { + Nil + } else { + val dataColNames = table.columns.iterator.map(_.name).toSeq + reportedMetaCols + .filter(metaCol => dataColNames.exists(resolver(metaCol.name, _))) + .map { metaCol => + s"${quoteIdentifier(metaCol.name)} metadata column is hidden by a data column " + + "with the same name" + } + } + } + + private def renamesConflictingMetadataColumns(table: Table): Boolean = table match { + case hasMeta: SupportsMetadataColumns => hasMeta.canRenameConflictingMetadataColumns + case _ => false } private def filter(colNames: Seq[String], cols: Seq[MetadataColumn]): Seq[MetadataColumn] = { 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 1b5d7bb75fe66..9f7d954ea97b6 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 @@ -186,8 +186,10 @@ abstract class InMemoryBaseTable( private val metadataColumnNames = metadataColumns.map(_.name).toSet // Metadata column renaming is supported -- see [[InMemoryScanBuilder.pruneColumns]] and - // [[BatchScanBaseClass.createReaderFactory]] for implementation details. - override val canRenameConflictingMetadataColumns: Boolean = true + // [[BatchScanBaseClass.createReaderFactory]] for implementation details. Set the property to + // false to act like a connector that suppresses such conflicts instead of renaming them. + override val canRenameConflictingMetadataColumns: Boolean = + properties.getOrDefault("rename-conflicting-metadata-columns", "true").toBoolean private val allowUnsupportedTransforms = properties.getOrDefault("allow-unsupported-transforms", "false").toBoolean diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/V2TableUtilSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/V2TableUtilSuite.scala index 31930dfcb67e0..7e0283209d419 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/V2TableUtilSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/V2TableUtilSuite.scala @@ -22,16 +22,18 @@ import java.util import org.apache.spark.SparkFunSuite import org.apache.spark.sql.AnalysisException import org.apache.spark.sql.catalyst.expressions.{AttributeReference, MetadataAttribute} +import org.apache.spark.sql.catalyst.plans.SQLHelper import org.apache.spark.sql.connector.catalog.TableCapability.BATCH_READ import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.internal.connector.ColumnImpl import org.apache.spark.sql.types._ import org.apache.spark.sql.util.CaseInsensitiveStringMap -import org.apache.spark.sql.util.SchemaValidationMode.{ALLOW_NEW_TOP_LEVEL_FIELDS, PROHIBIT_CHANGES} +import org.apache.spark.sql.util.SchemaValidationMode.{ALLOW_NEW_FIELDS, ALLOW_NEW_TOP_LEVEL_FIELDS, PROHIBIT_CHANGES} import org.apache.spark.sql.util.SchemaValidationMode import org.apache.spark.util.ArrayImplicits.SparkArrayOps -class V2TableUtilSuite extends SparkFunSuite { +class V2TableUtilSuite extends SparkFunSuite with SQLHelper { test("validateCapturedColumns - no changes") { val cols = Array( @@ -447,6 +449,178 @@ class V2TableUtilSuite extends SparkFunSuite { assert(errors.isEmpty) } + test("validateCapturedMetadataColumns - renamed captured metadata column type changed") { + val dataCols = Array(col("index", LongType, nullable = true)) + val originMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + val originTable = TestTableWithRenamableMetadata("test", dataCols, originMetaCols) + val dataAttrs = dataCols.map(c => AttributeReference(c.name, c.dataType, c.nullable)()) + val relation = DataSourceV2Relation( + originTable, + dataAttrs.toImmutableArraySeq, + None, + None, + CaseInsensitiveStringMap.empty()).withMetadataColumns() + assert(relation.output.map(_.name) == Seq("index", "_index")) + val currentTable = TestTableWithRenamableMetadata( + "test", + dataCols, + Array(metaCol("index", StringType, nullable = false))) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + currentTable, + relation, + mode = PROHIBIT_CHANGES, + checkIds = true) + assert(errors == Seq("`index` type has changed from INT to STRING")) + } + + test("validateCapturedMetadataColumns - unchanged renamed captured metadata column") { + val dataCols = Array(col("index", LongType, nullable = true)) + val originMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + val originTable = TestTableWithRenamableMetadata("test", dataCols, originMetaCols) + val attrs = Seq( + AttributeReference("index", LongType, nullable = true)(), + MetadataAttribute("index", IntegerType, nullable = false).withName("_index")) + val relation = DataSourceV2Relation( + originTable, + attrs, + None, + None, + CaseInsensitiveStringMap.empty()) + val currentTable = TestTableWithRenamableMetadata("test", dataCols, originMetaCols) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + currentTable, + relation, + mode = PROHIBIT_CHANGES, + checkIds = true) + assert(errors.isEmpty, "renaming a captured metadata column must remain valid") + } + + test("validateCapturedMetadataColumns - metadata column hidden by a data column is rejected") { + val originMetaCols = Seq(metaCol("index", IntegerType, nullable = false)) + // The connector still reports the `index` metadata column, but a data column has taken its + // name. Because the connector suppresses rather than renames the conflict, the metadata column + // is no longer reachable, so a captured reference to it is broken. + val currentDataCols = Array(col("index", IntegerType, nullable = true)) + val currentMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + val table = TestTableWithMetadataSupport("test", currentDataCols, currentMetaCols) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + table, + originMetaCols, + mode = ALLOW_NEW_FIELDS, + checkIds = false) + assert(errors.size == 1) + assert(errors.head == "`index` metadata column is hidden by a data column with the same name") + } + + test("validateCapturedMetadataColumns - metadata column hidden by a data column is rejected " + + "under PROHIBIT_CHANGES") { + // The check is independent of the validation mode: it fires whenever a data column hides a + // still-reported metadata column, regardless of whether new fields are otherwise allowed. + val originMetaCols = Seq(metaCol("index", IntegerType, nullable = false)) + val currentDataCols = Array(col("index", IntegerType, nullable = true)) + val currentMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + val table = TestTableWithMetadataSupport("test", currentDataCols, currentMetaCols) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + table, + originMetaCols, + mode = PROHIBIT_CHANGES, + checkIds = true) + assert(errors.size == 1) + assert(errors.head == "`index` metadata column is hidden by a data column with the same name") + } + + test("validateCapturedMetadataColumns - hidden metadata column detection is case insensitive") { + val originMetaCols = Seq(metaCol("index", IntegerType, nullable = false)) + // The data column differs only in case from the metadata column, which still conflicts. + val currentDataCols = Array(col("INDEX", IntegerType, nullable = true)) + val currentMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + val table = TestTableWithMetadataSupport("test", currentDataCols, currentMetaCols) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + table, + originMetaCols, + mode = ALLOW_NEW_FIELDS, + checkIds = false) + assert(errors.size == 1) + assert(errors.head == "`index` metadata column is hidden by a data column with the same name") + } + + test("validateCapturedMetadataColumns - hidden metadata column detection uses SQL resolver " + + "for Unicode identifiers") { + val originMetaCols = Seq(metaCol("index", IntegerType, nullable = false)) + // These names are equal under the case-insensitive SQL resolver, but differ after root-locale + // lowercasing because the capital I with dot expands to an i followed by a combining dot. + val unicodeName = new String(Character.toChars(0x130)) + "ndex" + val currentDataCols = Array(col(unicodeName, IntegerType, nullable = true)) + val currentMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + val table = TestTableWithMetadataSupport("test", currentDataCols, currentMetaCols) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + table, + originMetaCols, + mode = ALLOW_NEW_FIELDS, + checkIds = false) + assert(errors.size == 1) + assert(errors.head == "`index` metadata column is hidden by a data column with the same name") + } + + test("validateCapturedMetadataColumns - hidden metadata column detection is case sensitive " + + "under case-sensitive analysis") { + val originMetaCols = Seq(metaCol("index", IntegerType, nullable = false)) + // Under case-sensitive analysis the data column does not take the metadata column's name, so + // the metadata column is still reachable and must not be reported as hidden. This has to match + // `metadataOutputWithOutConflicts`, which resolves the same names through `conf.resolver`. + val currentDataCols = Array(col("INDEX", IntegerType, nullable = true)) + val currentMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + val table = TestTableWithMetadataSupport("test", currentDataCols, currentMetaCols) + + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + val errors = V2TableUtil.validateCapturedMetadataColumns( + table, + originMetaCols, + mode = ALLOW_NEW_FIELDS, + checkIds = false) + assert(errors.isEmpty, "a differently cased data column does not hide the metadata column") + } + } + + test("validateCapturedMetadataColumns - data column matching a renamable metadata column is ok") { + val originMetaCols = Seq(metaCol("index", IntegerType, nullable = false)) + val currentDataCols = Array(col("index", IntegerType, nullable = true)) + val currentMetaCols = Array(metaCol("index", IntegerType, nullable = false)) + // The connector renames a conflicting metadata column, so it stays reachable (as `_index`) and + // the data column does not hide it. This must keep working. + val table = TestTableWithRenamableMetadata("test", currentDataCols, currentMetaCols) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + table, + originMetaCols, + mode = ALLOW_NEW_FIELDS, + checkIds = false) + assert(errors.isEmpty, "a renamable metadata column is not hidden by a same-named data column") + } + + test("validateCapturedMetadataColumns - dropped metadata column is reported as removed, " + + "not hidden") { + val originMetaCols = Seq(metaCol("index", IntegerType, nullable = false)) + // The connector no longer reports the metadata column and a data column now carries its name. + // The removal, not the shadowing, is the accurate diagnosis, so only one error is expected. + val currentDataCols = Array(col("index", IntegerType, nullable = true)) + val table = TestTableWithMetadataSupport("test", currentDataCols, Array.empty) + + val errors = V2TableUtil.validateCapturedMetadataColumns( + table, + originMetaCols, + mode = PROHIBIT_CHANGES, + checkIds = true) + assert(errors.size == 1) + assert(errors.head == "`index` INT NOT NULL has been removed") + } + test("extractMetadataColumns - doesn't access table metadata unless needed") { val dataCols = Array( col("id", LongType, nullable = true), @@ -848,6 +1022,16 @@ class V2TableUtilSuite extends SparkFunSuite { override def capabilities: util.Set[TableCapability] = util.Set.of(BATCH_READ) } + // table that renames metadata columns conflicting with data columns instead of suppressing them + private case class TestTableWithRenamableMetadata( + override val name: String, + override val columns: Array[Column], + override val metadataColumns: Array[MetadataColumn] = Array.empty) + extends Table with SupportsMetadataColumns { + override def capabilities: util.Set[TableCapability] = util.Set.of(BATCH_READ) + override def canRenameConflictingMetadataColumns: Boolean = true + } + // table that throws when metadataColumns is accessed private case class TestTableThatThrowsOnMetadataAccess( override val name: String, 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 c18b8ef09e969..028def6e18bca 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 @@ -1750,6 +1750,32 @@ class DataSourceV2DataFrameSuite } } + test("SPARK-59014: detect a data column hiding a captured metadata column after analysis") { + val t = "testcat.ns1.ns2.tbl" + withTable(t) { + // this table suppresses metadata/data name conflicts instead of renaming them, which is the + // default in SupportsMetadataColumns and the case where the metadata column becomes lost + sql(s"CREATE TABLE $t (id INT, data STRING) USING foo " + + "TBLPROPERTIES ('rename-conflicting-metadata-columns' = 'false')") + sql(s"INSERT INTO $t VALUES (1, 'a')") + + // create DataFrame projecting the metadata column and trigger analysis + val table = spark.table(t) + val df = table.select($"id", table.metadataColumn("index")) + + // a data column takes the metadata column's name after the plan was analyzed + sql(s"ALTER TABLE $t ADD COLUMN `index` INT") + + // execution should fail instead of silently reading the data column's values + checkError( + exception = intercept[AnalysisException] { df.collect() }, + condition = "INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.METADATA_COLUMNS_MISMATCH", + parameters = Map( + "tableName" -> "`testcat`.`ns1`.`ns2`.`tbl`", + "errors" -> "- `index` metadata column is hidden by a data column with the same name")) + } + } + test("SPARK-54157: cached temp view allows top-level column additions") { val t = "testcat.ns1.ns2.tbl" withTable(t) {