This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 95b352564200 [SPARK-57773][SQL] Strip leaked internal metadata in
DataSourceV2Relation.create while preserving column IDs
95b352564200 is described below
commit 95b3525642001c3e62d5512d417a220913f1a9ab
Author: Wenchen Fan <[email protected]>
AuthorDate: Wed Jul 1 13:13:29 2026 +0800
[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 <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 316a3cc80120ceb917fc6245c6b6735dd1fb6261)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../apache/spark/sql/catalyst/util/package.scala | 8 +++--
.../datasources/v2/DataSourceV2Relation.scala | 10 ++++--
.../datasources/v2/DataSourceV2RelationSuite.scala | 41 +++++++++++++++++++++-
3 files changed, 54 insertions(+), 5 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala
index 5dc3962821a0..4fd85ef4923d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/package.scala
+++
b/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())
})
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 19371dcb94de..0b7c829939ff 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++
b/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)
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala
index a1f1c85fe4d5..46ed870cb322 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2RelationSuite.scala
+++
b/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))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
