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 92f9fb099523 [SPARK-58849][SQL] Make DESCRIBE TABLE resilient to
corrupt partition metadata
92f9fb099523 is described below
commit 92f9fb099523791793a4461a8dbbc50089134b19
Author: Szehon Ho <[email protected]>
AuthorDate: Fri Aug 21 12:52:26 2026 +0800
[SPARK-58849][SQL] Make DESCRIBE TABLE resilient to corrupt partition
metadata
### What changes were proposed in this pull request?
Make `DESCRIBE TABLE` tolerate inconsistent partition metadata. When a
table's declared partition
columns do not match the last fields of the schema being described, the
text output replaces the
usual `# Partition Information` section with an `# Invalid Partition
Information` section that
lists both the declared partition columns and the last columns of the table
schema, so neither set
is presented as the valid one.
The condition is checked explicitly instead of catching the
`AssertionError` raised by
`CatalogTable.partitionSchema`. The check runs against the schema that is
actually described (for a
legacy table with an empty stored schema, that is the inferred one) and
compares names with
`conf.resolver` rather than exact string equality.
`DESCRIBE TABLE ... AS JSON` is deliberately left unchanged. Its
`partition_columns` field comes
from `CatalogTable.toJsonLinkedHashMap`, which reads `partitionColumnNames`
directly and never
evaluates the assertion, so that path already worked for corrupt metadata
and keeps reporting the
declared partition columns.
The following paths still use the strict `partitionSchema` assertion and
are intentionally out of
scope:
- `DESC ... PARTITION`, `SHOW TABLE EXTENDED ... PARTITION`, `SHOW
PARTITIONS ... PARTITION`, and
`SHOW CREATE TABLE ... AS SERDE`. Unfiltered `SHOW PARTITIONS` and plain
`SHOW CREATE TABLE` do
not call `partitionSchema` for this mismatch.
- `DataSource.resolveRelation`, which calls `partitionSchema` while
inferring the schema of a
legacy table whose stored schema is empty. `DESCRIBE TABLE` on such a
partitioned table therefore
still fails, inside schema inference rather than in the partition section.
Replacing that assertion with a structured Spark error can be handled
separately.
### Why are the changes needed?
`DescribeTableCommand.describePartitionInfo` called
`CatalogTable.partitionSchema`, which asserts
that the schema's last fields match `partitionColumnNames`. Corrupt catalog
metadata therefore made
`DESCRIBE TABLE` itself fail with an `AssertionError`, preventing users
from inspecting the
remaining metadata that would help them diagnose or repair the table.
### Does this PR introduce _any_ user-facing change?
Yes. Previously `DESCRIBE TABLE` failed with an `AssertionError` when the
declared partition columns
did not match the last columns of the schema. It now returns the rest of
the table information and
reports both column lists under an `# Invalid Partition Information`
section. `DESCRIBE TABLE ... AS
JSON` output is unchanged.
### How was this patch tested?
Added `DESCRIBE TABLE is resilient to corrupt partition metadata` to
`org.apache.spark.sql.execution.command.v1.DescribeTableSuite`, covering
plain `DESCRIBE TABLE`,
`DESCRIBE TABLE EXTENDED`, and asserting that `DESCRIBE TABLE EXTENDED ...
AS JSON` still reports
`partition_columns` unchanged.
Run locally, with no failures:
- `build/sbt 'sql/testOnly *DescribeTableSuite'` -- 106 tests, covering the
V1 and V2 suites with
each test run for both the V1 and V2 command versions.
- `build/sbt 'hive/testOnly *command.DescribeTableSuite'` -- 68 tests.
The fixture stays in the V1 in-memory suite rather than moving up to
`DescribeTableSuiteBase`,
which the Hive suite also extends. Under the Hive catalog the deliberately
malformed `CatalogTable`
cannot be read back at all: `HiveExternalCatalog.restoreHiveSerdeTable`
calls `reorderSchema`, which
resolves each declared partition column by name and throws
`AnalysisException`
`_LEGACY_ERROR_TEMP_3088` ("The metadata is corrupted. Unable to find the
partition column names
from the schema") when one is absent from the schema. The table therefore
fails before
`DESCRIBE TABLE` can inspect it.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5).
Closes #58092 from szehon-ho/codex/es-1878719-hardening.
Authored-by: Szehon Ho <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 7e87aa545a431efd4290c7cdb8b9592ddfd82954)
Signed-off-by: Wenchen Fan <[email protected]>
---
docs/sql-ref-syntax-aux-describe-table.md | 10 ++++++
.../spark/sql/execution/command/tables.scala | 36 +++++++++++++++-----
.../execution/command/v1/DescribeTableSuite.scala | 38 +++++++++++++++++++++-
3 files changed, 75 insertions(+), 9 deletions(-)
diff --git a/docs/sql-ref-syntax-aux-describe-table.md
b/docs/sql-ref-syntax-aux-describe-table.md
index 5c417689dde2..e2db0ab3d88c 100644
--- a/docs/sql-ref-syntax-aux-describe-table.md
+++ b/docs/sql-ref-syntax-aux-describe-table.md
@@ -164,6 +164,16 @@ to return the metadata pertaining to a partition or column
respectively.
| MapType | `{ "name" : "map", "key_type": <type_json>,
"value_type": <type_json>, "value_nullable": <boolean> }`
|
| StructType | `{ "name" : "struct", "fields": [ {"name" :
"field1", "type" : <type_json>, “nullable”: <boolean>, "comment": “<comment>”,
"default": “<default_val>”}, ... ] }` |
+**Note**
+- If a V1 (session catalog) table's declared partition columns do not match
the last columns
+ of its schema, the catalog metadata is considered inconsistent. `DESCRIBE
TABLE` still
+ returns the rest of the metadata and reports both column lists under an
+ `# Invalid Partition Information` section in place of `# Partition
Information`. Repair
+ the table metadata so the declared partition columns match the last columns
of the
+ schema. Until then, commands that inspect a specific partition may still
fail:
+ `DESCRIBE TABLE ... PARTITION`, `SHOW TABLE EXTENDED ... PARTITION`, and
+ `SHOW PARTITIONS ... PARTITION`.
+
### Examples
```sql
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index df050742ec66..4d513b4d3b06 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -654,15 +654,16 @@ case class DescribeTableCommand(
}
describeSchema(metadata.schema, result, header = false)
} else {
- if (metadata.schema.isEmpty) {
+ val schema = if (metadata.schema.isEmpty) {
// In older version(prior to 2.1) of Spark, the table schema can be
empty and should be
// inferred at runtime. We should still support it.
- describeSchema(sparkSession.table(metadata.identifier).schema, result,
header = false)
+ sparkSession.table(metadata.identifier).schema
} else {
- describeSchema(metadata.schema, result, header = false)
+ metadata.schema
}
+ describeSchema(schema, result, header = false)
- describePartitionInfo(metadata, result)
+ describePartitionInfo(metadata, schema, result)
describeClusteringInfo(metadata, result)
if (partitionSpec.nonEmpty) {
@@ -682,10 +683,29 @@ case class DescribeTableCommand(
result.toSeq
}
- private def describePartitionInfo(table: CatalogTable, buffer:
ArrayBuffer[Row]): Unit = {
- if (table.partitionColumnNames.nonEmpty) {
- append(buffer, "# Partition Information", "", "")
- describeSchema(table.partitionSchema, buffer, header = true)
+ private def describePartitionInfo(
+ table: CatalogTable,
+ schema: StructType,
+ buffer: ArrayBuffer[Row]): Unit = {
+ val partitionColumnNames = table.partitionColumnNames
+ if (partitionColumnNames.nonEmpty) {
+ // Same positional convention as `CatalogTable.partitionSchema`, but
reported instead of
+ // asserted so that a table with inconsistent metadata stays describable.
+ val partitionFields = schema.takeRight(partitionColumnNames.length)
+ val consistent = partitionFields.length == partitionColumnNames.length &&
+ partitionFields.map(_.name).zip(partitionColumnNames).forall {
+ case (schemaColumn, partitionColumn) => conf.resolver(schemaColumn,
partitionColumn)
+ }
+ if (consistent) {
+ append(buffer, "# Partition Information", "", "")
+ describeSchema(StructType(partitionFields), buffer, header = true)
+ } else {
+ append(buffer, "# Invalid Partition Information", "", "")
+ append(buffer, "Declared Partition Columns",
+ partitionColumnNames.mkString("[", ", ", "]"), "")
+ append(buffer, "Last Columns in Table Schema",
+ partitionFields.map(_.name).mkString("[", ", ", "]"), "")
+ }
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DescribeTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DescribeTableSuite.scala
index 97fa08fef3a6..4e2c4a4c6975 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DescribeTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/DescribeTableSuite.scala
@@ -24,11 +24,13 @@ import org.json4s.jackson.JsonMethods.parse
import org.apache.spark.SPARK_VERSION
import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable, CatalogTableType}
import
org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
import org.apache.spark.sql.execution.command
import org.apache.spark.sql.execution.command.{DescribeTableJson, Field,
SqlPathEntry, TableColumn, Type}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType}
/**
* This base suite contains unified tests for the `DESCRIBE TABLE` command
that checks V1
@@ -874,6 +876,40 @@ class DescribeTableSuite extends DescribeTableSuiteBase
with CommandSuiteBase {
}
}
+ test("DESCRIBE TABLE is resilient to corrupt partition metadata") {
+ withNamespaceAndTable("ns", "table") { tbl =>
+ // Simulate corrupt metadata where the declared partition column does
not match the
+ // last field in the table schema.
+ val table = CatalogTable(
+ identifier = TableIdentifier("table", Some("ns")),
+ tableType = CatalogTableType.MANAGED,
+ storage = CatalogStorageFormat.empty,
+ schema = new StructType()
+ .add("id", IntegerType)
+ .add("actual_part", StringType),
+ provider = Some(getProvider()),
+ partitionColumnNames = Seq("declared_part"))
+ spark.sessionState.catalog.createTable(table, ignoreIfExists = false)
+
+ val expectedInvalidInfo = Seq(
+ Row("# Invalid Partition Information", "", ""),
+ Row("Declared Partition Columns", "[declared_part]", ""),
+ Row("Last Columns in Table Schema", "[actual_part]", ""))
+
+ Seq("DESCRIBE TABLE", "DESCRIBE TABLE EXTENDED").foreach { command =>
+ val description = spark.sql(s"$command $tbl").collect().toSeq
+ assert(description.contains(Row("id", "int", null)))
+ assert(description.contains(Row("actual_part", "string", null)))
+ assert(description.containsSlice(expectedInvalidInfo))
+ assert(!description.exists(_.getString(0) == "# Partition
Information"))
+ }
+
+ val jsonValue = spark.sql(s"DESCRIBE TABLE EXTENDED $tbl AS
JSON").head().getString(0)
+ val parsedOutput = parse(jsonValue).extract[DescribeTableJson]
+ assert(parsedOutput.partition_columns === Some(List("declared_part")))
+ }
+ }
+
test("DESCRIBE TABLE EXTENDED of a table with a default column value") {
withTable("t") {
spark.sql(s"CREATE TABLE t (id bigint default 42) $defaultUsing")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
