This is an automated email from the ASF dual-hosted git repository.
wenchen 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 3da5608 [SPARK-37283][SQL] Don't try to store a V1 table which
contains ANSI intervals in Hive compatible format
3da5608 is described below
commit 3da5608f26b8f743d595d618d168231d426fabf2
Author: Kousuke Saruta <saru...@oss.nttdata.com>
AuthorDate: Mon Nov 15 18:00:17 2021 +0800
[SPARK-37283][SQL] Don't try to store a V1 table which contains ANSI
intervals in Hive compatible format
### What changes were proposed in this pull request?
If, a table being created contains a column of ANSI interval types and the
underlying file format has a corresponding Hive SerDe (e.g. Parquet),
`HiveExternalcatalog` tries to store the table in Hive compatible format.
But, as ANSI interval types in Spark and interval type in Hive are not
compatible (Hive only supports interval_year_month and interval_day_time), the
following warning with stack trace will be logged.
```
spark-sql> CREATE TABLE tbl1(a INTERVAL YEAR TO MONTH) USING Parquet;
21/11/11 14:39:29 WARN SessionState: METASTORE_FILTER_HOOK will be ignored,
since hive.security.authorization.manager is set to instance of
HiveAuthorizerFactory.
21/11/11 14:39:29 WARN HiveExternalCatalog: Could not persist
`default`.`tbl1` in a Hive compatible way. Persisting it into Hive metastore in
Spark SQL specific format.
org.apache.hadoop.hive.ql.metadata.HiveException:
java.lang.IllegalArgumentException: Error: type expected at the position 0 of
'interval year to month' but 'interval year to month' is found.
at org.apache.hadoop.hive.ql.metadata.Hive.createTable(Hive.java:869)
at org.apache.hadoop.hive.ql.metadata.Hive.createTable(Hive.java:874)
at
org.apache.spark.sql.hive.client.HiveClientImpl.$anonfun$createTable$1(HiveClientImpl.scala:553)
at
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at
org.apache.spark.sql.hive.client.HiveClientImpl.$anonfun$withHiveState$1(HiveClientImpl.scala:303)
at
org.apache.spark.sql.hive.client.HiveClientImpl.liftedTree1$1(HiveClientImpl.scala:234)
at
org.apache.spark.sql.hive.client.HiveClientImpl.retryLocked(HiveClientImpl.scala:233)
at
org.apache.spark.sql.hive.client.HiveClientImpl.withHiveState(HiveClientImpl.scala:283)
at
org.apache.spark.sql.hive.client.HiveClientImpl.createTable(HiveClientImpl.scala:551)
at
org.apache.spark.sql.hive.HiveExternalCatalog.saveTableIntoHive(HiveExternalCatalog.scala:499)
at
org.apache.spark.sql.hive.HiveExternalCatalog.createDataSourceTable(HiveExternalCatalog.scala:397)
at
org.apache.spark.sql.hive.HiveExternalCatalog.$anonfun$createTable$1(HiveExternalCatalog.scala:274)
at
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at
org.apache.spark.sql.hive.HiveExternalCatalog.withClient(HiveExternalCatalog.scala:102)
at
org.apache.spark.sql.hive.HiveExternalCatalog.createTable(HiveExternalCatalog.scala:245)
at
org.apache.spark.sql.catalyst.catalog.ExternalCatalogWithListener.createTable(ExternalCatalogWithListener.scala:94)
at
org.apache.spark.sql.catalyst.catalog.SessionCatalog.createTable(SessionCatalog.scala:376)
at
org.apache.spark.sql.execution.command.CreateDataSourceTableCommand.run(createDataSourceTables.scala:120)
at
org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75)
at
org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73)
at
org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84)
at
org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:97)
at
org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:103)
at
org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:163)
at
org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:90)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at
org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)
at
org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:97)
at
org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:93)
at
org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:481)
at
org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:82)
at
org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:481)
at
org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)
at
org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
at
org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
at
org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
at
org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
at
org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:457)
at
org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:93)
at
org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:80)
at
org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:78)
at org.apache.spark.sql.Dataset.<init>(Dataset.scala:222)
at org.apache.spark.sql.Dataset$.$anonfun$ofRows$2(Dataset.scala:102)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:99)
at
org.apache.spark.sql.SparkSession.$anonfun$sql$1(SparkSession.scala:618)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:613)
at org.apache.spark.sql.SQLContext.sql(SQLContext.scala:651)
at
org.apache.spark.sql.hive.thriftserver.SparkSQLDriver.run(SparkSQLDriver.scala:67)
at
org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.processCmd(SparkSQLCLIDriver.scala:384)
at
org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.$anonfun$processLine$1(SparkSQLCLIDriver.scala:504)
at
org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.$anonfun$processLine$1$adapted(SparkSQLCLIDriver.scala:498)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at
org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.processLine(SparkSQLCLIDriver.scala:498)
at
org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver$.main(SparkSQLCLIDriver.scala:287)
at
org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver.main(SparkSQLCLIDriver.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
at
org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:955)
at
org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:180)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:203)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:90)
at
org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1043)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1052)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.IllegalArgumentException: Error: type expected at the
position 0 of 'interval year to month' but 'interval year to month' is found.
at
org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils$TypeInfoParser.expect(TypeInfoUtils.java:372)
at
org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils$TypeInfoParser.expect(TypeInfoUtils.java:355)
at
org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils$TypeInfoParser.parseType(TypeInfoUtils.java:416)
at
org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils$TypeInfoParser.parseTypeInfos(TypeInfoUtils.java:329)
at
org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils.getTypeInfosFromTypeString(TypeInfoUtils.java:814)
at
org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe.initialize(ParquetHiveSerDe.java:110)
at
org.apache.hadoop.hive.serde2.AbstractSerDe.initialize(AbstractSerDe.java:54)
at
org.apache.hadoop.hive.serde2.SerDeUtils.initializeSerDe(SerDeUtils.java:533)
at
org.apache.hadoop.hive.metastore.MetaStoreUtils.getDeserializer(MetaStoreUtils.java:453)
at
org.apache.hadoop.hive.metastore.MetaStoreUtils.getDeserializer(MetaStoreUtils.java:440)
at
org.apache.hadoop.hive.ql.metadata.Table.getDeserializerFromMetaStore(Table.java:281)
at
org.apache.hadoop.hive.ql.metadata.Table.checkValidity(Table.java:199)
at org.apache.hadoop.hive.ql.metadata.Hive.createTable(Hive.java:842)
... 73 more
21/11/11 14:39:29 WARN HiveConf: HiveConf of name
hive.internal.ss.authz.settings.applied.marker does not exist
21/11/11 14:39:29 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout
does not exist
21/11/11 14:39:29 WARN HiveConf: HiveConf of name hive.stats.retries.wait
does not exist
```
In such case, `HiveExternalCatalog` fallbacks to store the table in Spark
specific format but the stack trace is surprising and confusable.
So the solution this PR proposes is not to try to create such tables in
Hive compatible format because it should always fail, and logs a message which
says the table is stored in Spark specific format.
### Why are the changes needed?
To fix the confusable behavior.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New test.
Closes #34551 from sarutak/fix-ansi-interval-hive-type.
Authored-by: Kousuke Saruta <saru...@oss.nttdata.com>
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
---
.../spark/sql/hive/HiveExternalCatalog.scala | 19 +++++-
.../spark/sql/hive/MetastoreDataSourcesSuite.scala | 68 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 2 deletions(-)
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
index 568e814..179b424 100644
---
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
+++
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala
@@ -46,8 +46,7 @@ import
org.apache.spark.sql.execution.datasources.{PartitioningUtils, SourceOpti
import org.apache.spark.sql.hive.client.HiveClient
import org.apache.spark.sql.internal.HiveSerDe
import org.apache.spark.sql.internal.StaticSQLConf._
-import org.apache.spark.sql.types.{DataType, StructType}
-
+import org.apache.spark.sql.types.{AnsiIntervalType, ArrayType, DataType,
MapType, StructType}
/**
* A persistent implementation of the system catalog using Hive.
@@ -357,6 +356,8 @@ private[spark] class HiveExternalCatalog(conf: SparkConf,
hadoopConf: Configurat
val qualifiedTableName = table.identifier.quotedString
val maybeSerde = HiveSerDe.sourceToSerDe(provider)
+ val incompatibleTypes =
+ table.schema.filter(f =>
!isHiveCompatibleDataType(f.dataType)).map(_.dataType.simpleString)
val (hiveCompatibleTable, logMessage) = maybeSerde match {
case _ if options.skipHiveMetadata =>
@@ -365,6 +366,12 @@ private[spark] class HiveExternalCatalog(conf: SparkConf,
hadoopConf: Configurat
"Spark SQL specific format, which is NOT compatible with Hive."
(None, message)
+ case _ if incompatibleTypes.nonEmpty =>
+ val message =
+ s"Hive incompatible types found: ${incompatibleTypes.mkString(",
")}. " +
+ s"Persisting data source table $qualifiedTableName into Hive
metastore in " +
+ "Spark SQL specific format, which is NOT compatible with Hive."
+ (None, message)
// our bucketing is un-compatible with hive(different hash function)
case Some(serde) if table.bucketSpec.nonEmpty =>
val message =
@@ -1402,4 +1409,12 @@ object HiveExternalCatalog {
provider.isDefined && provider != Some(DDLUtils.HIVE_PROVIDER)
}
+ private[spark] def isHiveCompatibleDataType(dt: DataType): Boolean = dt
match {
+ case _: AnsiIntervalType => false
+ case s: StructType => s.forall(f => isHiveCompatibleDataType(f.dataType))
+ case a: ArrayType => isHiveCompatibleDataType(a.elementType)
+ case m: MapType =>
+ isHiveCompatibleDataType(m.keyType) &&
isHiveCompatibleDataType(m.valueType)
+ case _ => true
+ }
}
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
index ba44192..644d304 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
@@ -22,6 +22,7 @@ import java.io.File
import scala.collection.mutable.ArrayBuffer
import org.apache.hadoop.fs.Path
+import org.apache.log4j.Level
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
@@ -1403,6 +1404,73 @@ class MetastoreDataSourcesSuite extends QueryTest with
SQLTestUtils with TestHiv
}
}
+ test("SPARK-37283: Don't try to store a V1 table in Hive compatible format "
+
+ "if the table contains Hive incompatible types") {
+ import DayTimeIntervalType.{DAY, HOUR, MINUTE, SECOND}
+ import YearMonthIntervalType.{MONTH, YEAR}
+ withTable("t") {
+ val logAppender = new LogAppender(
+ s"Check whether a message is shown and it says that the V1 table
contains " +
+ "Hive incompatible types if the table contains Hive incompatible
types")
+ logAppender.setThreshold(Level.WARN)
+ withLogAppender(logAppender) {
+ sql(
+ """
+ |CREATE TABLE t(
+ | c1 INTERVAL DAY TO MINUTE,
+ | c2 STRING,
+ | c3 INTERVAL YEAR TO MONTH,
+ | c4 INT,
+ | c5 INTERVAL HOUR,
+ | c6 INTERVAL MONTH,
+ | c7 STRUCT<a: INT, b: STRING>,
+ | c8 STRUCT<a: INT, b: INTERVAL HOUR TO SECOND>,
+ | c9 ARRAY<INT>,
+ | c10 ARRAY<INTERVAL YEAR>,
+ | c11 MAP<INT, STRING>,
+ | c12 MAP<INT, INTERVAL DAY>,
+ | c13 MAP<INTERVAL MINUTE TO SECOND, STRING>
+ |) USING Parquet""".stripMargin)
+ }
+ val expectedMsg = "Hive incompatible types found: interval day to
minute, " +
+ "interval year to month, interval hour, interval month, " +
+ "struct<a:int,b:interval hour to second>, " +
+ "array<interval year>, map<int,interval day>, " +
+ "map<interval minute to second,string>. " +
+ "Persisting data source table `default`.`t` into Hive metastore in " +
+ "Spark SQL specific format, which is NOT compatible with Hive."
+ val actualMessages = logAppender.loggingEvents
+ .map(_.getRenderedMessage)
+ .filter(_.contains("incompatible"))
+ assert(actualMessages.contains(expectedMsg))
+ assert(hiveClient.getTable("default", "t").schema
+ .forall(_.dataType == ArrayType(StringType)))
+
+ val df = sql("SELECT * FROM t")
+ assert(df.schema ===
+ StructType(Seq(
+ StructField("c1", DayTimeIntervalType(DAY, MINUTE)),
+ StructField("c2", StringType),
+ StructField("c3", YearMonthIntervalType(YEAR, MONTH)),
+ StructField("c4", IntegerType),
+ StructField("c5", DayTimeIntervalType(HOUR)),
+ StructField("c6", YearMonthIntervalType(MONTH)),
+ StructField("c7",
+ StructType(Seq(
+ StructField("a", IntegerType),
+ StructField("b", StringType)))),
+ StructField("c8",
+ StructType(Seq(
+ StructField("a", IntegerType),
+ StructField("b", DayTimeIntervalType(HOUR, SECOND))))),
+ StructField("c9", ArrayType(IntegerType)),
+ StructField("c10", ArrayType(YearMonthIntervalType(YEAR))),
+ StructField("c11", MapType(IntegerType, StringType)),
+ StructField("c12", MapType(IntegerType, DayTimeIntervalType(DAY))),
+ StructField("c13", MapType(DayTimeIntervalType(MINUTE, SECOND),
StringType)))))
+ }
+ }
+
private def withDebugMode(f: => Unit): Unit = {
val previousValue = sparkSession.sparkContext.conf.get(DEBUG_MODE)
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
