This is an automated email from the ASF dual-hosted git repository.
maxgekk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 89397b5eecf [SPARK-40521][SQL] Return only exists partitions in
`PartitionsAlreadyExistException` from Hive's create partition
89397b5eecf is described below
commit 89397b5eecf9bdb54e740c50dada5bd183e8d9fc
Author: Max Gekk <max.g...@gmail.com>
AuthorDate: Fri Oct 7 15:40:36 2022 +0300
[SPARK-40521][SQL] Return only exists partitions in
`PartitionsAlreadyExistException` from Hive's create partition
### What changes were proposed in this pull request?
In the PR, I propose to modify Hive catalog implementation of create
partition in the case when Hive reports that some of partitions exist already.
In that case, Spark will check existence of all input partitions one-by-one and
return **only existing** partitions in `PartitionAlreadyExistsException`.
### Why are the changes needed?
1. To do not confuse Spark SQL users about existing partitions.
2. To be consistent with other catalogs V2 and V1 In-Memory.
### Does this PR introduce _any_ user-facing change?
Yes, it changes user-facing error message.
Before:
```
spark-sql> CREATE TABLE t (id bigint, data string) USING HIVE PARTITIONED
BY (id);
spark-sql> ALTER TABLE t ADD PARTITION (id=2) LOCATION 'loc1';
spark-sql> ALTER TABLE t ADD PARTITION (id=1) LOCATION 'loc' PARTITION
(id=2) LOCATION 'loc1';
The following partitions already exists in table 't' database 'default':
Map(id -> 1)
===
Map(id -> 2)
```
The error shows an existent and a non-existent partitions.
After:
```
...
spark-sql> ALTER TABLE t ADD PARTITION (id=1) LOCATION 'loc' PARTITION
(id=2) LOCATION 'loc1';
The following partitions already exists in table 't' database 'default':
Map(id -> 2)
```
The error contains only existent partitions that cause issues.
### How was this patch tested?
By running the modified test suites:
```
$ build/sbt -Phive-2.3 -Phive-thriftserver "test:testOnly
*.AlterTableAddPartitionSuite"
```
Closes #38134 from MaxGekk/fix-hive-partition-exists.
Authored-by: Max Gekk <max.g...@gmail.com>
Signed-off-by: Max Gekk <max.g...@gmail.com>
---
.../command/AlterTableAddPartitionSuiteBase.scala | 18 ------------------
.../command/v1/AlterTableAddPartitionSuite.scala | 21 +++++++++++++++++++++
.../command/v2/AlterTableAddPartitionSuite.scala | 19 +++++++++++++++++++
.../spark/sql/hive/client/HiveClientImpl.scala | 6 +++++-
4 files changed, 45 insertions(+), 19 deletions(-)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableAddPartitionSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableAddPartitionSuiteBase.scala
index dee14953c09..8472a581fa7 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableAddPartitionSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableAddPartitionSuiteBase.scala
@@ -20,7 +20,6 @@ package org.apache.spark.sql.execution.command
import java.time.{Duration, Period}
import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
-import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException
import org.apache.spark.sql.internal.SQLConf
/**
@@ -167,23 +166,6 @@ trait AlterTableAddPartitionSuiteBase extends QueryTest
with DDLCommandTestUtils
}
}
- test("partition already exists") {
- withNamespaceAndTable("ns", "tbl") { t =>
- sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED
BY (id)")
- sql(s"ALTER TABLE $t ADD PARTITION (id=2) LOCATION 'loc1'")
-
- val errMsg = intercept[PartitionsAlreadyExistException] {
- sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'" +
- " PARTITION (id=2) LOCATION 'loc1'")
- }.getMessage
- assert(errMsg.contains("The following partitions already exists"))
-
- sql(s"ALTER TABLE $t ADD IF NOT EXISTS PARTITION (id=1) LOCATION 'loc'" +
- " PARTITION (id=2) LOCATION 'loc1'")
- checkPartitions(t, Map("id" -> "1"), Map("id" -> "2"))
- }
- }
-
test("SPARK-33474: Support typed literals as partition spec values") {
withNamespaceAndTable("ns", "tbl") { t =>
sql(s"CREATE TABLE $t(name STRING, part DATE) USING PARQUET PARTITIONED
BY (part)")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala
index b2e626be1b1..6b2308766f6 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v1/AlterTableAddPartitionSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.command.v1
import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException
import org.apache.spark.sql.execution.command
import org.apache.spark.sql.internal.SQLConf
@@ -135,6 +136,26 @@ trait AlterTableAddPartitionSuiteBase extends
command.AlterTableAddPartitionSuit
}
}
}
+
+ // TODO: Move this test to the common trait as soon as it is migrated on
checkError()
+ test("partition already exists") {
+ withNamespaceAndTable("ns", "tbl") { t =>
+ sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED
BY (id)")
+ sql(s"ALTER TABLE $t ADD PARTITION (id=2) LOCATION 'loc1'")
+
+ val errMsg = intercept[PartitionsAlreadyExistException] {
+ sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'" +
+ " PARTITION (id=2) LOCATION 'loc1'")
+ }.getMessage
+ assert(errMsg ===
+ """The following partitions already exists in table 'tbl' database 'ns':
+ |Map(id -> 2)""".stripMargin)
+
+ sql(s"ALTER TABLE $t ADD IF NOT EXISTS PARTITION (id=1) LOCATION 'loc'" +
+ " PARTITION (id=2) LOCATION 'loc1'")
+ checkPartitions(t, Map("id" -> "1"), Map("id" -> "2"))
+ }
+ }
}
/**
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableAddPartitionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableAddPartitionSuite.scala
index fabe399c340..2784f1e4bdd 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableAddPartitionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/AlterTableAddPartitionSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.command.v2
import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.analysis.PartitionsAlreadyExistException
import org.apache.spark.sql.execution.command
/**
@@ -99,4 +100,22 @@ class AlterTableAddPartitionSuite
}
}
}
+
+ // TODO: Move this test to the common trait as soon as it is migrated on
checkError()
+ test("partition already exists") {
+ withNamespaceAndTable("ns", "tbl") { t =>
+ sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing PARTITIONED
BY (id)")
+ sql(s"ALTER TABLE $t ADD PARTITION (id=2) LOCATION 'loc1'")
+
+ val errMsg = intercept[PartitionsAlreadyExistException] {
+ sql(s"ALTER TABLE $t ADD PARTITION (id=1) LOCATION 'loc'" +
+ " PARTITION (id=2) LOCATION 'loc1'")
+ }.getMessage
+ assert(errMsg === s"The following partitions already exists in table
$t:2 -> id")
+
+ sql(s"ALTER TABLE $t ADD IF NOT EXISTS PARTITION (id=1) LOCATION 'loc'" +
+ " PARTITION (id=2) LOCATION 'loc1'")
+ checkPartitions(t, Map("id" -> "1"), Map("id" -> "2"))
+ }
+ }
}
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 61951cde8d0..bef320174ec 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -635,7 +635,11 @@ private[hive] class HiveClientImpl(
ignoreIfExists: Boolean): Unit = withHiveState {
def replaceExistException(e: Throwable): Unit = e match {
case _: HiveException if e.getCause.isInstanceOf[AlreadyExistsException]
=>
- throw new PartitionsAlreadyExistException(db, table, parts.map(_.spec))
+ val t = shim.getTable(client, db, table)
+ val exists = parts.filter { part =>
+ shim.getPartition(client, t, part.spec.asJava, forceCreate = false)
!= null
+ }
+ throw new PartitionsAlreadyExistException(db, table,
exists.map(_.spec))
case _ => throw e
}
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
