[SPARK-57773][SQL] Strip leaked internal metadata in DataSourceV2Relation.create while preserving column IDs

### What changes were proposed in this pull request?

`DataSourceV2Relation.create` builds the relation output schema directly from `table.columns().asSchema` (after char/varchar replacement), without removing internal metadata. This PR makes `create` strip internal metadata (the keys in `INTERNAL_METADATA_KEYS`, via `removeInternalMetadata`) from the schema before it becomes the relation's output, **while preserving column IDs**.

`removeInternalMetadata` gains a `keepFieldIds: Boolean = false` parameter that skips `FIELD_ID_METADATA_KEY` in the same removal pass (the default preserves behavior for existing callers). `create` calls it with `keepFieldIds = true`:

```scala
val schema = removeInternalMetadata(
  CharVarcharUtils.replaceCharVarcharWithStringInSchema(table.columns.asSchema),
  keepFieldIds = true)
```

### Why are the changes needed?

`INTERNAL_METADATA_KEYS` exists so that internal-only metadata keys (e.g. `__metadata_col`, `__qualified_access_only`) do not surface to users, and `removeInternalMetadata` is already applied on other schema-producing paths. But `DataSourceV2Relation.create` never applied it, so any internal metadata key that a v2 source attaches to its columns leaks straight onto the relation output and `df.schema`. This hardens `create` so internal-only keys stay internal, consistent with the other paths.

Column IDs need special handling: `FIELD_ID_METADATA_KEY` is listed in `INTERNAL_METADATA_KEYS` so that other paths drop it, but it is also deliberately surfaced onto the relation's output (see SPARK-57544). Removing it in `create` would defeat that, so the new `keepFieldIds` flag preserves it in the same pass that strips the rest.

### Does this PR introduce _any_ user-facing change?

No. It removes internal-only metadata keys (which were never meant to be user-visible) from the v2 relation output, and preserves the column IDs that are intentionally surfaced.

### How was this patch tested?

New unit test in `DataSourceV2RelationSuite` that builds a table whose column carries both a column ID and a leaked internal metadata key, and asserts that `create` strips the internal key from the relation output while preserving the column ID. Verified the test fails without the source change.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.8)

Closes #56887 from cloud-fan/SPARK-57544-followup.

Authored-by: Wenchen Fan <wenchen@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala

@@ -251,11 +251,15 @@ package object util extends Logging {
     MetadataColumn.PRESERVE_ON_REINSERT
   )
 
-  def removeInternalMetadata(schema: StructType): StructType = {
+  def removeInternalMetadata(schema: StructType, keepFieldIds: Boolean = false): StructType = {
     StructType(schema.map { field =>
       var builder = new MetadataBuilder().withMetadata(field.metadata)
       INTERNAL_METADATA_KEYS.foreach { key =>
-        builder = builder.remove(key)
+        // Column IDs are listed as internal so most paths drop them, but some paths (e.g. the v2
+        // relation output) deliberately surface them, so allow callers to keep them.
+        if (!(keepFieldIds && key == FIELD_ID_METADATA_KEY)) {
+          builder = builder.remove(key)
+        }
       }
       field.copy(metadata = builder.build())
     })

sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala

@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, ExposesMetadataColumns, Histogram, HistogramBin, LeafNode, LogicalPlan, Statistics}
 import org.apache.spark.sql.catalyst.streaming.{StreamingSourceIdentifyingName, Unassigned}
 import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
-import org.apache.spark.sql.catalyst.util.{truncatedString, CharVarcharUtils}
+import org.apache.spark.sql.catalyst.util.{removeInternalMetadata, truncatedString, CharVarcharUtils}
 import org.apache.spark.sql.connector.catalog.{CatalogPlugin, FunctionCatalog, Identifier, SupportsMetadataColumns, Table, TableCapability, TableCatalog, V2TableUtil}
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.CatalogHelper
 import org.apache.spark.sql.connector.expressions.{FieldReference, NamedReference}
@@ -329,7 +329,13 @@ object DataSourceV2Relation {
     import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
     // The v2 source may return schema containing char/varchar type. We replace char/varchar
     // with "annotated" string type here as the query engine doesn't support char/varchar yet.
-    val schema = CharVarcharUtils.replaceCharVarcharWithStringInSchema(table.columns.asSchema)
+    // We also strip internal metadata that may have leaked onto the table columns, so it does
+    // not surface on the relation's output. Column IDs (FIELD_ID_METADATA_KEY) are an exception:
+    // although the key is listed in INTERNAL_METADATA_KEYS so that other paths drop it, the
+    // column-ID feature deliberately surfaces field IDs on the relation's output, so we keep them.
+    val schema = removeInternalMetadata(
+      CharVarcharUtils.replaceCharVarcharWithStringInSchema(table.columns.asSchema),
+      keepFieldIds = true)
     DataSourceV2Relation(table, toAttributes(schema), catalog, identifier, options, timeTravelSpec)
   }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala

@@ -17,11 +17,17 @@
 
 package org.apache.spark.sql.execution.datasources.v2
 
+import java.util
+
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.catalyst.catalog.{CatalogColumnStat, CatalogStatistics}
 import org.apache.spark.sql.catalyst.plans.logical.{Histogram, HistogramBin}
+import org.apache.spark.sql.catalyst.util.FieldMetadataUtils.FIELD_ID_METADATA_KEY
+import org.apache.spark.sql.catalyst.util.INTERNAL_METADATA_KEYS
+import org.apache.spark.sql.connector.catalog.{Column, Table, TableCapability}
 import org.apache.spark.sql.connector.expressions.FieldReference
-import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+import org.apache.spark.sql.types.{IntegerType, MetadataBuilder, StringType, StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
 class DataSourceV2RelationSuite extends SparkFunSuite {
 
@@ -107,4 +113,37 @@ class DataSourceV2RelationSuite extends SparkFunSuite {
     val idV2NoHist = colStats.get(FieldReference.column("id"))
     assert(!idV2NoHist.histogram().isPresent)
   }
+
+  test("create strips leaked internal metadata but preserves column IDs") {
+    // A column carrying both a column ID (surfaced on purpose) and every internal metadata key
+    // (listed in INTERNAL_METADATA_KEYS), simulating a v2 source that leaks internal metadata.
+    // The column ID key is excluded here since it is set via `.id(...)` below.
+    val leakedBuilder = new MetadataBuilder()
+    INTERNAL_METADATA_KEYS.filter(_ != FIELD_ID_METADATA_KEY).foreach { key =>
+      leakedBuilder.putString(key, "leaked")
+    }
+    val leakedMetadata = leakedBuilder.build()
+    val table = new Table {
+      override def name(): String = "t"
+      override def columns(): Array[Column] = Array(
+        Column.builderFor("id", IntegerType)
+          .id("1")
+          .metadata(leakedMetadata.json)
+          .build())
+      override def capabilities(): util.Set[TableCapability] =
+        util.Set.of[TableCapability]()
+    }
+
+    val relation =
+      DataSourceV2Relation.create(table, None, None, CaseInsensitiveStringMap.empty())
+    val field = relation.schema.head
+
+    // All leaked internal metadata keys are stripped from the relation output ...
+    INTERNAL_METADATA_KEYS.filter(_ != FIELD_ID_METADATA_KEY).foreach { key =>
+      assert(!field.metadata.contains(key), s"internal metadata key $key should be stripped")
+    }
+    // ... but the column ID is preserved.
+    assert(field.id.contains("1"))
+    assert(field.metadata.contains(FIELD_ID_METADATA_KEY))
+  }
 }