You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2020/12/11 18:04:13 UTC
[spark] branch branch-2.4 updated: [SPARK-33742][SQL][2.4] Throw
PartitionsAlreadyExistException from HiveExternalCatalog.createPartitions()
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-2.4 by this push:
new 3e64795 [SPARK-33742][SQL][2.4] Throw PartitionsAlreadyExistException from HiveExternalCatalog.createPartitions()
3e64795 is described below
commit 3e64795cbf472a15c1b77ec932cb501653dadcb9
Author: Max Gekk <ma...@gmail.com>
AuthorDate: Fri Dec 11 09:56:46 2020 -0800
[SPARK-33742][SQL][2.4] Throw PartitionsAlreadyExistException from HiveExternalCatalog.createPartitions()
### What changes were proposed in this pull request?
Throw `PartitionsAlreadyExistException` from `createPartitions()` in Hive external catalog when a partition exists. Currently, `HiveExternalCatalog.createPartitions()` throws `AlreadyExistsException` wrapped by `AnalysisException`.
In the PR, I propose to catch `AlreadyExistsException` in `HiveClientImpl` and replace it by `PartitionsAlreadyExistException`.
### Why are the changes needed?
The behaviour of Hive external catalog deviates from V1/V2 in-memory catalogs that throw `PartitionsAlreadyExistException`. To improve user experience with Spark SQL, it would be better to throw the same exception.
### Does this PR introduce _any_ user-facing change?
Yes
### How was this patch tested?
By running existing test suites:
```
$ build/sbt -Phive -Phive-thriftserver "hive/test:testOnly org.apache.spark.sql.hive.client.VersionsSuite"
$ build/sbt -Phive -Phive-thriftserver "hive/test:testOnly org.apache.spark.sql.hive.execution.HiveDDLSuite"
```
Authored-by: Max Gekk <max.gekkgmail.com>
Signed-off-by: Dongjoon Hyun <dongjoonapache.org>
(cherry picked from commit fab2995972761503563fa2aa547c67047c51bd33)
Signed-off-by: Max Gekk <max.gekkgmail.com>
Closes #30732 from MaxGekk/hive-partition-exception-2.4.
Authored-by: Max Gekk <ma...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../spark/sql/hive/client/HiveClientImpl.scala | 19 ++++++++++++++----
.../spark/sql/hive/client/VersionsSuite.scala | 23 +++++++++++++++++++++-
.../spark/sql/hive/execution/HiveDDLSuite.scala | 19 +++++++++++++++++-
3 files changed, 55 insertions(+), 6 deletions(-)
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 701f108..c983fb3 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
@@ -19,6 +19,7 @@ package org.apache.spark.sql.hive.client
import java.io.{File, PrintStream}
import java.lang.{Iterable => JIterable}
+import java.lang.reflect.InvocationTargetException
import java.util.{Locale, Map => JMap}
import scala.collection.JavaConverters._
@@ -30,10 +31,10 @@ import org.apache.hadoop.hive.common.StatsSetupConst
import org.apache.hadoop.hive.conf.HiveConf
import org.apache.hadoop.hive.conf.HiveConf.ConfVars
import org.apache.hadoop.hive.metastore.{TableType => HiveTableType}
-import org.apache.hadoop.hive.metastore.api.{Database => HiveDatabase, FieldSchema, Order}
+import org.apache.hadoop.hive.metastore.api.{AlreadyExistsException, Database => HiveDatabase, FieldSchema, Order}
import org.apache.hadoop.hive.metastore.api.{SerDeInfo, StorageDescriptor}
import org.apache.hadoop.hive.ql.Driver
-import org.apache.hadoop.hive.ql.metadata.{Hive, Partition => HivePartition, Table => HiveTable}
+import org.apache.hadoop.hive.ql.metadata.{Hive, HiveException, Partition => HivePartition, Table => HiveTable}
import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer.HIVE_COLUMN_ORDER_ASC
import org.apache.hadoop.hive.ql.processors._
import org.apache.hadoop.hive.ql.session.SessionState
@@ -43,7 +44,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.metrics.source.HiveCatalogMetrics
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchPartitionException}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchPartitionException, PartitionsAlreadyExistException}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.Expression
@@ -541,7 +542,17 @@ private[hive] class HiveClientImpl(
table: String,
parts: Seq[CatalogTablePartition],
ignoreIfExists: Boolean): Unit = withHiveState {
- shim.createPartitions(client, db, table, parts, ignoreIfExists)
+ def replaceExistException(e: Throwable): Unit = e match {
+ case _: HiveException if e.getCause.isInstanceOf[AlreadyExistsException] =>
+ throw new PartitionsAlreadyExistException(db, table, parts.map(_.spec))
+ case _ => throw e
+ }
+ try {
+ shim.createPartitions(client, db, table, parts, ignoreIfExists)
+ } catch {
+ case e: InvocationTargetException => replaceExistException(e.getCause)
+ case e: Throwable => replaceExistException(e)
+ }
}
override def dropPartitions(
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala
index 7c66ff6..2a4ef11 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala
@@ -30,7 +30,7 @@ import org.apache.spark.SparkFunSuite
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
-import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchPermanentFunctionException}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchPermanentFunctionException, PartitionsAlreadyExistException}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, Literal}
import org.apache.spark.sql.catalyst.util.quietly
@@ -458,6 +458,27 @@ class VersionsSuite extends SparkFunSuite with Logging {
assert(client.getPartitionOption("default", "src_part", spec).isEmpty)
}
+ test(s"$version: createPartitions if already exists") {
+ val partitions = Seq(CatalogTablePartition(
+ Map("key1" -> "101", "key2" -> "102"),
+ storageFormat))
+ try {
+ client.createPartitions("default", "src_part", partitions, ignoreIfExists = false)
+ val errMsg = intercept[PartitionsAlreadyExistException] {
+ client.createPartitions("default", "src_part", partitions, ignoreIfExists = false)
+ }.getMessage
+ assert(errMsg.contains("partitions already exists"))
+ } finally {
+ client.dropPartitions(
+ "default",
+ "src_part",
+ partitions.map(_.spec),
+ ignoreIfNotExists = true,
+ purge = false,
+ retainData = false)
+ }
+ }
+
///////////////////////////////////////////////////////////////////////////
// Function related API
///////////////////////////////////////////////////////////////////////////
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index 90915e0..35e83ae 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -31,7 +31,7 @@ import org.scalatest.BeforeAndAfterEach
import org.apache.spark.SparkException
import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, PartitionsAlreadyExistException, TableAlreadyExistsException}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.execution.command.{DDLSuite, DDLUtils}
import org.apache.spark.sql.functions._
@@ -2351,4 +2351,21 @@ class HiveDDLSuite
}
}
}
+
+ test("SPARK-33742: partition already exists") {
+ withTable("t") {
+ sql(s"CREATE TABLE t (data string) PARTITIONED BY (id bigint)")
+ 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'")
+ checkAnswer(sql("SHOW PARTITIONS t"), Seq(Row("id=1"), Row("id=2")))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org