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 2021/09/08 04:17:09 UTC
[spark] branch master updated: [SPARK-35803][SQL] Support
DataSource V2 CreateTempViewUsing
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 feba05f [SPARK-35803][SQL] Support DataSource V2 CreateTempViewUsing
feba05f is described below
commit feba05f181f594b53cf9c832cd44b00be2646938
Author: Pablo Langa <so...@gmail.com>
AuthorDate: Wed Sep 8 12:16:29 2021 +0800
[SPARK-35803][SQL] Support DataSource V2 CreateTempViewUsing
### What changes were proposed in this pull request?
Currently only DataSources V1 are supported in the CreateTempViewUsing command. This PR refactor DataframeReader to reuse the code for the creation of a DataFrame from a DataSource V2
### Why are the changes needed?
Improve the support of DataSourve V2 in this command
### Does this PR introduce _any_ user-facing change?
It does not change the current behavior, it only adds a new functionality
### How was this patch tested?
Unit testing
Closes #33922 from planga82/feature/spark35803_crateview_datasourceV2.
Authored-by: Pablo Langa <so...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../org/apache/spark/sql/DataFrameReader.scala | 55 ++-----------------
.../spark/sql/execution/datasources/ddl.scala | 24 +++++---
.../datasources/v2/DataSourceV2Utils.scala | 64 +++++++++++++++++++++-
.../spark/sql/connector/DataSourceV2Suite.scala | 12 ++++
4 files changed, 95 insertions(+), 60 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
index 8a0e449..cab0ea2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
@@ -21,8 +21,6 @@ import java.util.{Locale, Properties}
import scala.collection.JavaConverters._
-import com.fasterxml.jackson.databind.ObjectMapper
-
import org.apache.spark.Partition
import org.apache.spark.annotation.Stable
import org.apache.spark.api.java.JavaRDD
@@ -33,15 +31,13 @@ import org.apache.spark.sql.catalyst.csv.{CSVHeaderChecker, CSVOptions, Univocit
import org.apache.spark.sql.catalyst.expressions.ExprUtils
import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser, JSONOptions}
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils, FailureSafeParser}
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, SupportsCatalogOptions, SupportsRead}
-import org.apache.spark.sql.connector.catalog.TableCapability._
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.command.DDLUtils
import org.apache.spark.sql.execution.datasources.DataSource
import org.apache.spark.sql.execution.datasources.csv._
import org.apache.spark.sql.execution.datasources.jdbc._
import org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource
-import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2Utils}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.unsafe.types.UTF8String
@@ -207,55 +203,12 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
throw QueryCompilationErrors.pathOptionNotSetCorrectlyWhenReadingError()
}
- DataSource.lookupDataSourceV2(source, sparkSession.sessionState.conf).map { provider =>
- val catalogManager = sparkSession.sessionState.catalogManager
- val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
- source = provider, conf = sparkSession.sessionState.conf)
-
- val optionsWithPath = getOptionsWithPaths(paths: _*)
-
- val finalOptions = sessionOptions.filterKeys(!optionsWithPath.contains(_)).toMap ++
- optionsWithPath.originalMap
- val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
- val (table, catalog, ident) = provider match {
- case _: SupportsCatalogOptions if userSpecifiedSchema.nonEmpty =>
- throw new IllegalArgumentException(
- s"$source does not support user specified schema. Please don't specify the schema.")
- case hasCatalog: SupportsCatalogOptions =>
- val ident = hasCatalog.extractIdentifier(dsOptions)
- val catalog = CatalogV2Util.getTableProviderCatalog(
- hasCatalog,
- catalogManager,
- dsOptions)
- (catalog.loadTable(ident), Some(catalog), Some(ident))
- case _ =>
- // TODO: Non-catalog paths for DSV2 are currently not well defined.
- val tbl = DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
- (tbl, None, None)
- }
- import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
- table match {
- case _: SupportsRead if table.supports(BATCH_READ) =>
- Dataset.ofRows(
- sparkSession,
- DataSourceV2Relation.create(table, catalog, ident, dsOptions))
-
- case _ => loadV1Source(paths: _*)
- }
+ DataSource.lookupDataSourceV2(source, sparkSession.sessionState.conf).flatMap { provider =>
+ DataSourceV2Utils.loadV2Source(sparkSession, provider, userSpecifiedSchema, extraOptions,
+ source, paths: _*)
}.getOrElse(loadV1Source(paths: _*))
}
- private def getOptionsWithPaths(paths: String*): CaseInsensitiveMap[String] = {
- if (paths.isEmpty) {
- extraOptions
- } else if (paths.length == 1) {
- extraOptions + ("path" -> paths.head)
- } else {
- val objectMapper = new ObjectMapper()
- extraOptions + ("paths" -> objectMapper.writeValueAsString(paths.toArray))
- }
- }
-
private def loadV1Source(paths: String*) = {
val legacyPathOptionBehavior = sparkSession.sessionState.conf.legacyPathOptionBehavior
val (finalPaths, finalOptions) = if (!legacyPathOptionBehavior && paths.length == 1) {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala
index 22b98da..90463c0 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala
@@ -24,9 +24,11 @@ import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.command.{DDLUtils, LeafRunnableCommand}
import org.apache.spark.sql.execution.command.ViewHelper.createTemporaryViewRelation
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils
import org.apache.spark.sql.internal.StaticSQLConf
import org.apache.spark.sql.types._
@@ -87,15 +89,21 @@ case class CreateTempViewUsing(
throw QueryCompilationErrors.cannotCreateTempViewUsingHiveDataSourceError()
}
- val dataSource = DataSource(
- sparkSession,
- userSpecifiedSchema = userSpecifiedSchema,
- className = provider,
- options = options)
-
val catalog = sparkSession.sessionState.catalog
- val analyzedPlan = Dataset.ofRows(
- sparkSession, LogicalRelation(dataSource.resolveRelation())).logicalPlan
+ val analyzedPlan = DataSource.lookupDataSourceV2(provider, sparkSession.sessionState.conf)
+ .flatMap { tblProvider =>
+ DataSourceV2Utils.loadV2Source(sparkSession, tblProvider, userSpecifiedSchema,
+ CaseInsensitiveMap(options), provider)
+ }.getOrElse {
+ val dataSource = DataSource(
+ sparkSession,
+ userSpecifiedSchema = userSpecifiedSchema,
+ className = provider,
+ options = options)
+
+ Dataset.ofRows(
+ sparkSession, LogicalRelation(dataSource.resolveRelation()))
+ }.logicalPlan
if (global) {
val db = sparkSession.sessionState.conf.getConf(StaticSQLConf.GLOBAL_TEMP_DATABASE)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala
index 79f8e57..a0874fc 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala
@@ -19,8 +19,15 @@ package org.apache.spark.sql.execution.datasources.v2
import java.util.regex.Pattern
+import scala.collection.JavaConverters._
+
+import com.fasterxml.jackson.databind.ObjectMapper
+
import org.apache.spark.internal.Logging
-import org.apache.spark.sql.connector.catalog.{SessionConfigSupport, Table, TableProvider}
+import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
+import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, SessionConfigSupport, SupportsCatalogOptions, SupportsRead, Table, TableProvider}
+import org.apache.spark.sql.connector.catalog.TableCapability.BATCH_READ
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
@@ -83,4 +90,59 @@ private[sql] object DataSourceV2Utils extends Logging {
options.asCaseSensitiveMap())
}
}
+
+ def loadV2Source(
+ sparkSession: SparkSession,
+ provider: TableProvider,
+ userSpecifiedSchema: Option[StructType],
+ extraOptions: CaseInsensitiveMap[String],
+ source: String,
+ paths: String*): Option[DataFrame] = {
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
+ source = provider, conf = sparkSession.sessionState.conf)
+
+ val optionsWithPath = getOptionsWithPaths(extraOptions, paths: _*)
+
+ val finalOptions = sessionOptions.filterKeys(!optionsWithPath.contains(_)).toMap ++
+ optionsWithPath.originalMap
+ val dsOptions = new CaseInsensitiveStringMap(finalOptions.asJava)
+ val (table, catalog, ident) = provider match {
+ case _: SupportsCatalogOptions if userSpecifiedSchema.nonEmpty =>
+ throw new IllegalArgumentException(
+ s"$source does not support user specified schema. Please don't specify the schema.")
+ case hasCatalog: SupportsCatalogOptions =>
+ val ident = hasCatalog.extractIdentifier(dsOptions)
+ val catalog = CatalogV2Util.getTableProviderCatalog(
+ hasCatalog,
+ catalogManager,
+ dsOptions)
+ (catalog.loadTable(ident), Some(catalog), Some(ident))
+ case _ =>
+ // TODO: Non-catalog paths for DSV2 are currently not well defined.
+ val tbl = DataSourceV2Utils.getTableFromProvider(provider, dsOptions, userSpecifiedSchema)
+ (tbl, None, None)
+ }
+ import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+ table match {
+ case _: SupportsRead if table.supports(BATCH_READ) =>
+ Option(Dataset.ofRows(
+ sparkSession,
+ DataSourceV2Relation.create(table, catalog, ident, dsOptions)))
+ case _ => None
+ }
+ }
+
+ private def getOptionsWithPaths(
+ extraOptions: CaseInsensitiveMap[String],
+ paths: String*): CaseInsensitiveMap[String] = {
+ if (paths.isEmpty) {
+ extraOptions
+ } else if (paths.length == 1) {
+ extraOptions + ("path" -> paths.head)
+ } else {
+ val objectMapper = new ObjectMapper()
+ extraOptions + ("paths" -> objectMapper.writeValueAsString(paths.toArray))
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index 6eeaa55..7909dd7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -435,6 +435,18 @@ class DataSourceV2Suite extends QueryTest with SharedSparkSession with AdaptiveS
}
}
}
+
+ test("SPARK-35803: Support datasorce V2 in CREATE VIEW USING") {
+ Seq(classOf[SimpleDataSourceV2], classOf[JavaSimpleDataSourceV2]).foreach { cls =>
+ withClue(cls.getName) {
+ sql(s"CREATE or REPLACE TEMPORARY VIEW s1 USING ${cls.getName}")
+ checkAnswer(sql("select * from s1"), (0 until 10).map(i => Row(i, -i)))
+ checkAnswer(sql("select j from s1"), (0 until 10).map(i => Row(-i)))
+ checkAnswer(sql("select * from s1 where i > 5"),
+ (6 until 10).map(i => Row(i, -i)))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org