[SPARK-59014][SQL] Reject a data column that hides a captured metadata column during DSv2 refresh validation

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]].
  *

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,41 @@ 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 connector that does not rename the
+   * conflict (`canRenameConflictingMetadataColumns` is false) suppresses the metadata column via
+   * `metadataOutputWithOutConflicts`. A suppressed metadata column can no longer be resolved, so a
+   * captured reference to it is broken, and on a partially pruned scan it silently reads the data
+   * column's values instead. 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(c => normalize(c.name)).toSet
+      reportedMetaCols
+        .filter(metaCol => dataColNames.contains(normalize(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] = {

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

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,159 @@ 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 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 +1003,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,

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) {