Skip to content

[SPARK-53598][SQL] Check the existence of numParts before reading large table property #52355

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pan3793 wants to merge 2 commits into from
Closed

[SPARK-53598][SQL] Check the existence of numParts before reading large table property #52355

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f2e9c68
[SPARK-53598][SQL] Check numParts field for Hive tables before readin…
pan3793 Sep 16, 2025
0434dd0
simplify
pan3793 Sep 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -767,7 +767,7 @@ object CatalogTable {
props.get(key).orElse {
if (props.exists { case (mapKey, _) => mapKey.startsWith(key) }) {
props.get(s"$key.numParts") match {
case None => throw QueryCompilationErrors.insufficientTablePropertyError(key)
case None => None
case Some(numParts) =>
val parts = (0 until numParts.toInt).map { index =>
val keyPart = s"$key.part.$index"
Expand Down
118 changes: 67 additions & 51 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1378,62 +1378,78 @@ class MetastoreDataSourcesSuite extends QueryTest
}

test("read table with corrupted schema") {
try {
val schema = StructType(StructField("int", IntegerType) :: Nil)
val hiveTableWithoutNumPartsProp = CatalogTable(
identifier = TableIdentifier("t", Some("default")),
tableType = CatalogTableType.MANAGED,
schema = HiveExternalCatalog.EMPTY_DATA_SCHEMA,
provider = Some("json"),
storage = CatalogStorageFormat.empty,
properties = Map(
DATASOURCE_PROVIDER -> "json",
DATASOURCE_SCHEMA_PART_PREFIX + 0 -> schema.json))

hiveClient.createTable(hiveTableWithoutNumPartsProp, ignoreIfExists = false)

checkError(
exception = intercept[AnalysisException] {
sharedState.externalCatalog.getTable("default", "t")
},
condition = "INSUFFICIENT_TABLE_PROPERTY.MISSING_KEY",
parameters = Map("key" -> toSQLConf("spark.sql.sources.schema"))
)

val hiveTableWithNumPartsProp = CatalogTable(
identifier = TableIdentifier("t2", Some("default")),
tableType = CatalogTableType.MANAGED,
schema = HiveExternalCatalog.EMPTY_DATA_SCHEMA,
provider = Some("json"),
storage = CatalogStorageFormat.empty,
properties = Map(
DATASOURCE_PROVIDER -> "json",
DATASOURCE_SCHEMA_PREFIX + "numParts" -> "3",
DATASOURCE_SCHEMA_PART_PREFIX + 0 -> schema.json))

hiveClient.createTable(hiveTableWithNumPartsProp, ignoreIfExists = false)
Seq(true, false).foreach { isHiveTable =>
try {
val schema = StructType(StructField("int", IntegerType) :: Nil)
val hiveTableWithoutNumPartsProp = CatalogTable(
Copy link
Member

@vrozov vrozov Sep 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can construction of CatalogTable be refactored to a separate method that takes 3 parameters (table name, table type and num of part) to avoid code duplication here and on line 1405?

Copy link
Member Author

@pan3793 pan3793 Sep 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this requires parameterizing more things, I prefer to keep it as-is unless you strongly disagree

identifier = TableIdentifier("t", Some("default")),
tableType = CatalogTableType.MANAGED,
schema = HiveExternalCatalog.EMPTY_DATA_SCHEMA,
provider = if (isHiveTable) None else Some("json"),
storage = CatalogStorageFormat.empty,
properties = Map(
DATASOURCE_SCHEMA_PART_PREFIX + 0 -> schema.json) ++ {
if (isHiveTable) {
Map.empty
} else {
Map(DATASOURCE_PROVIDER -> "json")
}
})

checkError(
exception = intercept[AnalysisException] {
sharedState.externalCatalog.getTable("default", "t2")
},
condition = "INSUFFICIENT_TABLE_PROPERTY.MISSING_KEY_PART",
parameters = Map(
"key" -> toSQLConf("spark.sql.sources.schema.part.1"),
"totalAmountOfParts" -> "3")
)
hiveClient.createTable(hiveTableWithoutNumPartsProp, ignoreIfExists = false)

withDebugMode {
val tableMeta = sharedState.externalCatalog.getTable("default", "t")
assert(tableMeta.identifier == TableIdentifier("t", Some("default")))
assert(tableMeta.properties(DATASOURCE_PROVIDER) == "json")
val tableMeta2 = sharedState.externalCatalog.getTable("default", "t2")
assert(tableMeta2.identifier == TableIdentifier("t2", Some("default")))
assert(tableMeta2.properties(DATASOURCE_PROVIDER) == "json")
assert(!tableMeta.properties.contains(DATASOURCE_PROVIDER))

val hiveTableWithNumPartsProp = CatalogTable(
identifier = TableIdentifier("t2", Some("default")),
tableType = CatalogTableType.MANAGED,
schema = HiveExternalCatalog.EMPTY_DATA_SCHEMA,
provider = if (isHiveTable) None else Some("json"),
storage = CatalogStorageFormat.empty,
properties = Map(
DATASOURCE_SCHEMA_PREFIX + "numParts" -> "3",
DATASOURCE_SCHEMA_PART_PREFIX + 0 -> schema.json) ++ {
if (isHiveTable) {
Map.empty
} else {
Map(DATASOURCE_PROVIDER -> "json")
}
})

hiveClient.createTable(hiveTableWithNumPartsProp, ignoreIfExists = false)

checkError(
exception = intercept[AnalysisException] {
sharedState.externalCatalog.getTable("default", "t2")
},
condition = "INSUFFICIENT_TABLE_PROPERTY.MISSING_KEY_PART",
parameters = Map(
"key" -> toSQLConf("spark.sql.sources.schema.part.1"),
"totalAmountOfParts" -> "3")
)

withDebugMode {
val tableMeta = sharedState.externalCatalog.getTable("default", "t")
assert(tableMeta.identifier == TableIdentifier("t", Some("default")))
if (isHiveTable) {
assert(!tableMeta.properties.contains(DATASOURCE_PROVIDER))
} else {
assert(tableMeta.properties(DATASOURCE_PROVIDER) == "json")
}
val tableMeta2 = sharedState.externalCatalog.getTable("default", "t2")
assert(tableMeta2.identifier == TableIdentifier("t2", Some("default")))
if (isHiveTable) {
assert(!tableMeta2.properties.contains(DATASOURCE_PROVIDER))
} else {
assert(tableMeta2.properties(DATASOURCE_PROVIDER) == "json")
}
}
} finally {
hiveClient.dropTable("default", "t", ignoreIfNotExists = true, purge = true)
hiveClient.dropTable("default", "t2", ignoreIfNotExists = true, purge = true)
}
} finally {
hiveClient.dropTable("default", "t", ignoreIfNotExists = true, purge = true)
hiveClient.dropTable("default", "t2", ignoreIfNotExists = true, purge = true)
}
}

Expand Down