You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2019/04/16 06:27:25 UTC
[spark] branch master updated: [SPARK-27448][SQL] File source V2
table provider should be compatible with V1 provider
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 f9837d3 [SPARK-27448][SQL] File source V2 table provider should be compatible with V1 provider
f9837d3 is described below
commit f9837d3bf6c15600f926917a253bf1bdddb4d1b4
Author: Gengliang Wang <ge...@databricks.com>
AuthorDate: Tue Apr 16 14:26:38 2019 +0800
[SPARK-27448][SQL] File source V2 table provider should be compatible with V1 provider
## What changes were proposed in this pull request?
In the rule `PreprocessTableCreation`, if an existing table is appended with a different provider, the action will fail.
Currently, there are two implementations for file sources and creating a table with file source V2 will always fall back to V1 FileFormat. We should consider the following cases as valid:
1. Appending a table with file source V2 provider using the v1 file format
2. Appending a table with v1 file format provider using file source V2 format
## How was this patch tested?
Unit test
Closes #24356 from gengliangwang/fixTableProvider.
Authored-by: Gengliang Wang <ge...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/execution/datasources/rules.scala | 10 +++-
.../spark/sql/hive/MetastoreDataSourcesSuite.scala | 68 ++++++++++++++++++++++
2 files changed, 77 insertions(+), 1 deletion(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
index 4e7fc40..534e2fd 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Cast, Expres
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.command.DDLUtils
+import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.InsertableRelation
import org.apache.spark.sql.types.{AtomicType, StructType}
@@ -113,7 +114,9 @@ case class PreprocessTableCreation(sparkSession: SparkSession) extends Rule[Logi
val specifiedProvider = DataSource.lookupDataSource(tableDesc.provider.get, conf)
// TODO: Check that options from the resolved relation match the relation that we are
// inserting into (i.e. using the same compression).
- if (existingProvider != specifiedProvider) {
+ // If the one of the provider is [[FileDataSourceV2]] and the other one is its corresponding
+ // [[FileFormat]], the two providers are considered compatible.
+ if (fallBackV2ToV1(existingProvider) != fallBackV2ToV1(specifiedProvider)) {
throw new AnalysisException(s"The format of the existing table $tableName is " +
s"`${existingProvider.getSimpleName}`. It doesn't match the specified format " +
s"`${specifiedProvider.getSimpleName}`.")
@@ -235,6 +238,11 @@ case class PreprocessTableCreation(sparkSession: SparkSession) extends Rule[Logi
}
}
+ private def fallBackV2ToV1(cls: Class[_]): Class[_] = cls.newInstance match {
+ case f: FileDataSourceV2 => f.fallbackFileFormat
+ case _ => cls
+ }
+
private def normalizeCatalogTable(schema: StructType, table: CatalogTable): CatalogTable = {
SchemaUtils.checkSchemaColumnNameDuplication(
schema,
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 c9ff4fc..6a64599 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
@@ -968,6 +968,74 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv
}
}
+ test("append a table with file source V2 provider using the v1 file format") {
+ def createDF(from: Int, to: Int): DataFrame = {
+ (from to to).map(i => i -> s"str$i").toDF("c1", "c2")
+ }
+
+ withTable("appendCSV") {
+ createDF(0, 9)
+ .write
+ .mode(SaveMode.Append)
+ .format("org.apache.spark.sql.execution.datasources.v2.csv.CSVDataSourceV2")
+ .saveAsTable("appendCSV")
+ createDF(10, 19)
+ .write
+ .mode(SaveMode.Append)
+ .format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat")
+ .saveAsTable("appendCSV")
+ checkAnswer(
+ sql("SELECT p.c1, p.c2 FROM appendCSV p WHERE p.c1 > 5"),
+ (6 to 19).map(i => Row(i, s"str$i")))
+ }
+
+ withTable("appendCSV") {
+ createDF(0, 9).write.mode(SaveMode.Append).format("csv").saveAsTable("appendCSV")
+ createDF(10, 19)
+ .write
+ .mode(SaveMode.Append)
+ .format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat")
+ .saveAsTable("appendCSV")
+ checkAnswer(
+ sql("SELECT p.c1, p.c2 FROM appendCSV p WHERE p.c1 > 5"),
+ (6 to 19).map(i => Row(i, s"str$i")))
+ }
+ }
+
+ test("append a table with v1 file format provider using file source V2 format") {
+ def createDF(from: Int, to: Int): DataFrame = {
+ (from to to).map(i => i -> s"str$i").toDF("c1", "c2")
+ }
+
+ withTable("appendCSV") {
+ createDF(0, 9)
+ .write
+ .mode(SaveMode.Append)
+ .format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat")
+ .saveAsTable("appendCSV")
+ createDF(10, 19)
+ .write
+ .mode(SaveMode.Append)
+ .format("org.apache.spark.sql.execution.datasources.v2.csv.CSVDataSourceV2")
+ .saveAsTable("appendCSV")
+ checkAnswer(
+ sql("SELECT p.c1, p.c2 FROM appendCSV p WHERE p.c1 > 5"),
+ (6 to 19).map(i => Row(i, s"str$i")))
+ }
+
+ withTable("appendCSV") {
+ createDF(0, 9)
+ .write
+ .mode(SaveMode.Append)
+ .format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat")
+ .saveAsTable("appendCSV")
+ createDF(10, 19).write.mode(SaveMode.Append).format("csv").saveAsTable("appendCSV")
+ checkAnswer(
+ sql("SELECT p.c1, p.c2 FROM appendCSV p WHERE p.c1 > 5"),
+ (6 to 19).map(i => Row(i, s"str$i")))
+ }
+ }
+
test("SPARK-8156:create table to specific database by 'use dbname' ") {
val df = (1 to 3).map(i => (i, s"val_$i", i * 2)).toDF("a", "b", "c")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org