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 2023/04/13 09:57:01 UTC
[spark] branch master updated: [SPARK-43088][SQL] Respect RequiresDistributionAndOrdering in CTAS/RTAS
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 aa1b16b1acd [SPARK-43088][SQL] Respect RequiresDistributionAndOrdering in CTAS/RTAS
aa1b16b1acd is described below
commit aa1b16b1acd67c22544f969eb8595b02bbc1b433
Author: aokolnychyi <ao...@apple.com>
AuthorDate: Thu Apr 13 17:56:44 2023 +0800
[SPARK-43088][SQL] Respect RequiresDistributionAndOrdering in CTAS/RTAS
### What changes were proposed in this pull request?
This PR makes sure Spark satisfies distribution and ordering requirements during CTAS/RTAS by carrying the analyzed plan to exec nodes and building `AppendData` for the new table.
### Why are the changes needed?
These changes are needed for multiple reasons:
- Some data sources may require a specific distribution/ordering/num partitions for **correctness**. All of that information is now being requested from Spark via `RequiresDistributionAndOrdering`, which is ignored by CTAS/RTAS commands. As a result, a data source may ask for a particular distribution or ordering and Spark may not respect it. This can cause correctness issues.
- Ignoring `RequiresDistributionAndOrdering` may severely degrade the write **performance** or even fail jobs as the underlying write may produce lots of small files.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests.
Closes #40734 from aokolnychyi/spark-43088.
Authored-by: aokolnychyi <ao...@apple.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/catalyst/plans/logical/v2Commands.scala | 112 ++++++++++-----------
.../datasources/v2/DataSourceV2Strategy.scala | 28 ++----
.../datasources/v2/WriteToDataSourceV2Exec.scala | 88 ++++++----------
3 files changed, 92 insertions(+), 136 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
index 9508b2fb993..6f5df8d48c4 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala
@@ -389,12 +389,47 @@ case class WriteDelta(
}
}
+trait V2CreateTableAsSelectPlan extends V2CreateTablePlan with AnalysisOnlyCommand {
+ def query: LogicalPlan
+
+ override lazy val resolved: Boolean = childrenResolved && {
+ // the table schema is created from the query schema, so the only resolution needed is to check
+ // that the columns referenced by the table's partitioning exist in the query schema
+ val references = partitioning.flatMap(_.references).toSet
+ references.map(_.fieldNames).forall(query.schema.findNestedField(_).isDefined)
+ }
+
+ override def childrenToAnalyze: Seq[LogicalPlan] = Seq(name, query)
+
+ override def tableSchema: StructType = query.schema
+
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[LogicalPlan]): V2CreateTableAsSelectPlan = {
+ assert(!isAnalyzed)
+ newChildren match {
+ case Seq(newName, newQuery) =>
+ withNameAndQuery(newName, newQuery)
+ case others =>
+ throw new IllegalArgumentException("Must be 2 children: " + others)
+ }
+ }
+
+ protected def withNameAndQuery(
+ newName: LogicalPlan,
+ newQuery: LogicalPlan): V2CreateTableAsSelectPlan
+}
+
/** A trait used for logical plan nodes that create or replace V2 table definitions. */
trait V2CreateTablePlan extends LogicalPlan {
- def tableName: Identifier
+ def name: LogicalPlan
def partitioning: Seq[Transform]
def tableSchema: StructType
+ def tableName: Identifier = {
+ assert(name.resolved)
+ name.asInstanceOf[ResolvedIdentifier].identifier
+ }
+
/**
* Creates a copy of this node with the new partitioning transforms. This method is used to
* rewrite the partition transforms normalized according to the table schema.
@@ -414,11 +449,6 @@ case class CreateTable(
override def child: LogicalPlan = name
- override def tableName: Identifier = {
- assert(child.resolved)
- child.asInstanceOf[ResolvedIdentifier].identifier
- }
-
override protected def withNewChildInternal(newChild: LogicalPlan): V2CreateTablePlan =
copy(name = newChild)
@@ -437,36 +467,20 @@ case class CreateTableAsSelect(
tableSpec: TableSpec,
writeOptions: Map[String, String],
ignoreIfExists: Boolean,
- analyzedQuery: Option[LogicalPlan] = None)
- extends BinaryCommand with V2CreateTablePlan with KeepAnalyzedQuery {
-
- override def tableSchema: StructType = query.schema
- override def left: LogicalPlan = name
- override def right: LogicalPlan = query
+ isAnalyzed: Boolean = false)
+ extends V2CreateTableAsSelectPlan {
- override def tableName: Identifier = {
- assert(left.resolved)
- left.asInstanceOf[ResolvedIdentifier].identifier
- }
-
- override lazy val resolved: Boolean = childrenResolved && {
- // the table schema is created from the query schema, so the only resolution needed is to check
- // that the columns referenced by the table's partitioning exist in the query schema
- val references = partitioning.flatMap(_.references).toSet
- references.map(_.fieldNames).forall(query.schema.findNestedField(_).isDefined)
- }
+ override def markAsAnalyzed(ac: AnalysisContext): LogicalPlan = copy(isAnalyzed = true)
override def withPartitioning(rewritten: Seq[Transform]): V2CreateTablePlan = {
this.copy(partitioning = rewritten)
}
- override def storeAnalyzedQuery(): Command = copy(analyzedQuery = Some(query))
-
- override protected def withNewChildrenInternal(
- newLeft: LogicalPlan,
- newRight: LogicalPlan
- ): CreateTableAsSelect =
- copy(name = newLeft, query = newRight)
+ override protected def withNameAndQuery(
+ newName: LogicalPlan,
+ newQuery: LogicalPlan): CreateTableAsSelect = {
+ copy(name = newName, query = newQuery)
+ }
}
/**
@@ -486,11 +500,6 @@ case class ReplaceTable(
override def child: LogicalPlan = name
- override def tableName: Identifier = {
- assert(child.resolved)
- child.asInstanceOf[ResolvedIdentifier].identifier
- }
-
override protected def withNewChildInternal(newChild: LogicalPlan): V2CreateTablePlan =
copy(name = newChild)
@@ -512,35 +521,20 @@ case class ReplaceTableAsSelect(
tableSpec: TableSpec,
writeOptions: Map[String, String],
orCreate: Boolean,
- analyzedQuery: Option[LogicalPlan] = None)
- extends BinaryCommand with V2CreateTablePlan with KeepAnalyzedQuery {
-
- override def tableSchema: StructType = query.schema
- override def left: LogicalPlan = name
- override def right: LogicalPlan = query
+ isAnalyzed: Boolean = false)
+ extends V2CreateTableAsSelectPlan {
- override lazy val resolved: Boolean = childrenResolved && {
- // the table schema is created from the query schema, so the only resolution needed is to check
- // that the columns referenced by the table's partitioning exist in the query schema
- val references = partitioning.flatMap(_.references).toSet
- references.map(_.fieldNames).forall(query.schema.findNestedField(_).isDefined)
- }
-
- override def tableName: Identifier = {
- assert(name.resolved)
- name.asInstanceOf[ResolvedIdentifier].identifier
- }
-
- override def storeAnalyzedQuery(): Command = copy(analyzedQuery = Some(query))
-
- override protected def withNewChildrenInternal(
- newLeft: LogicalPlan,
- newRight: LogicalPlan): LogicalPlan =
- copy(name = newLeft, query = newRight)
+ override def markAsAnalyzed(ac: AnalysisContext): LogicalPlan = copy(isAnalyzed = true)
override def withPartitioning(rewritten: Seq[Transform]): V2CreateTablePlan = {
this.copy(partitioning = rewritten)
}
+
+ override protected def withNameAndQuery(
+ newName: LogicalPlan,
+ newQuery: LogicalPlan): ReplaceTableAsSelect = {
+ copy(name = newName, query = newQuery)
+ }
}
/**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
index 55dc8b54e00..471a9393b7d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.execution.datasources.v2
-import scala.collection.JavaConverters._
import scala.collection.mutable
import org.apache.commons.lang3.StringUtils
@@ -48,7 +47,6 @@ import org.apache.spark.sql.execution.streaming.continuous.{WriteToContinuousDat
import org.apache.spark.sql.internal.StaticSQLConf.WAREHOUSE_PATH
import org.apache.spark.sql.sources.{BaseRelation, TableScan}
import org.apache.spark.sql.types.StructType
-import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.storage.StorageLevel
class DataSourceV2Strategy(session: SparkSession) extends Strategy with PredicateHelper {
@@ -187,16 +185,14 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
partitioning, qualifyLocInTableSpec(tableSpec), ifNotExists) :: Nil
case CreateTableAsSelect(ResolvedIdentifier(catalog, ident), parts, query, tableSpec,
- options, ifNotExists, analyzedQuery) =>
- assert(analyzedQuery.isDefined)
- val writeOptions = new CaseInsensitiveStringMap(options.asJava)
+ options, ifNotExists, true) =>
catalog match {
case staging: StagingTableCatalog =>
- AtomicCreateTableAsSelectExec(staging, ident, parts, analyzedQuery.get, planLater(query),
- qualifyLocInTableSpec(tableSpec), writeOptions, ifNotExists) :: Nil
+ AtomicCreateTableAsSelectExec(staging, ident, parts, query,
+ qualifyLocInTableSpec(tableSpec), options, ifNotExists) :: Nil
case _ =>
- CreateTableAsSelectExec(catalog.asTableCatalog, ident, parts, analyzedQuery.get,
- planLater(query), qualifyLocInTableSpec(tableSpec), writeOptions, ifNotExists) :: Nil
+ CreateTableAsSelectExec(catalog.asTableCatalog, ident, parts, query,
+ qualifyLocInTableSpec(tableSpec), options, ifNotExists) :: Nil
}
case RefreshTable(r: ResolvedTable) =>
@@ -221,19 +217,16 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
}
case ReplaceTableAsSelect(ResolvedIdentifier(catalog, ident),
- parts, query, tableSpec, options, orCreate, analyzedQuery) =>
- assert(analyzedQuery.isDefined)
- val writeOptions = new CaseInsensitiveStringMap(options.asJava)
+ parts, query, tableSpec, options, orCreate, true) =>
catalog match {
case staging: StagingTableCatalog =>
AtomicReplaceTableAsSelectExec(
staging,
ident,
parts,
- analyzedQuery.get,
- planLater(query),
+ query,
qualifyLocInTableSpec(tableSpec),
- writeOptions,
+ options,
orCreate = orCreate,
invalidateCache) :: Nil
case _ =>
@@ -241,10 +234,9 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
catalog.asTableCatalog,
ident,
parts,
- analyzedQuery.get,
- planLater(query),
+ query,
qualifyLocInTableSpec(tableSpec),
- writeOptions,
+ options,
orCreate = orCreate,
invalidateCache) :: Nil
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
index c53c603ffaa..8355ac8e703 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
@@ -17,8 +17,6 @@
package org.apache.spark.sql.execution.datasources.v2
-import java.util.UUID
-
import scala.collection.JavaConverters._
import org.apache.spark.{SparkEnv, SparkException, TaskContext}
@@ -27,17 +25,16 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, TableSpec, UnaryNode}
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, LogicalPlan, TableSpec, UnaryNode}
import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, WriteDeltaProjections}
import org.apache.spark.sql.catalyst.util.RowDeltaUtils.{DELETE_OPERATION, INSERT_OPERATION, UPDATE_OPERATION}
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, StagedTable, StagingTableCatalog, SupportsWrite, Table, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, StagedTable, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.metric.CustomMetric
-import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter, DataWriterFactory, DeltaWrite, DeltaWriter, LogicalWriteInfoImpl, PhysicalWriteInfoImpl, V1Write, Write, WriterCommitMessage}
+import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter, DataWriterFactory, DeltaWrite, DeltaWriter, PhysicalWriteInfoImpl, Write, WriterCommitMessage}
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
import org.apache.spark.sql.execution.metric.{CustomMetrics, SQLMetric, SQLMetrics}
-import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.{LongAccumulator, Utils}
/**
@@ -69,10 +66,9 @@ case class CreateTableAsSelectExec(
catalog: TableCatalog,
ident: Identifier,
partitioning: Seq[Transform],
- plan: LogicalPlan,
- query: SparkPlan,
+ query: LogicalPlan,
tableSpec: TableSpec,
- writeOptions: CaseInsensitiveStringMap,
+ writeOptions: Map[String, String],
ifNotExists: Boolean) extends TableWriteExecHelper {
val properties = CatalogV2Util.convertTableProperties(tableSpec)
@@ -90,11 +86,8 @@ case class CreateTableAsSelectExec(
CharVarcharUtils.getRawSchema(query.schema, conf).asNullable)
val table = catalog.createTable(ident, columns,
partitioning.toArray, properties.asJava)
- writeToTable(catalog, table, writeOptions, ident)
+ writeToTable(catalog, table, writeOptions, ident, query)
}
-
- override protected def withNewChildInternal(newChild: SparkPlan): CreateTableAsSelectExec =
- copy(query = newChild)
}
/**
@@ -110,10 +103,9 @@ case class AtomicCreateTableAsSelectExec(
catalog: StagingTableCatalog,
ident: Identifier,
partitioning: Seq[Transform],
- plan: LogicalPlan,
- query: SparkPlan,
+ query: LogicalPlan,
tableSpec: TableSpec,
- writeOptions: CaseInsensitiveStringMap,
+ writeOptions: Map[String, String],
ifNotExists: Boolean) extends TableWriteExecHelper {
val properties = CatalogV2Util.convertTableProperties(tableSpec)
@@ -130,11 +122,8 @@ case class AtomicCreateTableAsSelectExec(
CharVarcharUtils.getRawSchema(query.schema, conf).asNullable)
val stagedTable = catalog.stageCreate(
ident, columns, partitioning.toArray, properties.asJava)
- writeToTable(catalog, stagedTable, writeOptions, ident)
+ writeToTable(catalog, stagedTable, writeOptions, ident, query)
}
-
- override protected def withNewChildInternal(newChild: SparkPlan): AtomicCreateTableAsSelectExec =
- copy(query = newChild)
}
/**
@@ -151,10 +140,9 @@ case class ReplaceTableAsSelectExec(
catalog: TableCatalog,
ident: Identifier,
partitioning: Seq[Transform],
- plan: LogicalPlan,
- query: SparkPlan,
+ query: LogicalPlan,
tableSpec: TableSpec,
- writeOptions: CaseInsensitiveStringMap,
+ writeOptions: Map[String, String],
orCreate: Boolean,
invalidateCache: (TableCatalog, Table, Identifier) => Unit) extends TableWriteExecHelper {
@@ -180,11 +168,8 @@ case class ReplaceTableAsSelectExec(
CharVarcharUtils.getRawSchema(query.schema, conf).asNullable)
val table = catalog.createTable(
ident, columns, partitioning.toArray, properties.asJava)
- writeToTable(catalog, table, writeOptions, ident)
+ writeToTable(catalog, table, writeOptions, ident, query)
}
-
- override protected def withNewChildInternal(newChild: SparkPlan): ReplaceTableAsSelectExec =
- copy(query = newChild)
}
/**
@@ -203,10 +188,9 @@ case class AtomicReplaceTableAsSelectExec(
catalog: StagingTableCatalog,
ident: Identifier,
partitioning: Seq[Transform],
- plan: LogicalPlan,
- query: SparkPlan,
+ query: LogicalPlan,
tableSpec: TableSpec,
- writeOptions: CaseInsensitiveStringMap,
+ writeOptions: Map[String, String],
orCreate: Boolean,
invalidateCache: (TableCatalog, Table, Identifier) => Unit) extends TableWriteExecHelper {
@@ -233,11 +217,8 @@ case class AtomicReplaceTableAsSelectExec(
} else {
throw QueryCompilationErrors.cannotReplaceMissingTableError(ident)
}
- writeToTable(catalog, staged, writeOptions, ident)
+ writeToTable(catalog, staged, writeOptions, ident, query)
}
-
- override protected def withNewChildInternal(newChild: SparkPlan): AtomicReplaceTableAsSelectExec =
- copy(query = newChild)
}
/**
@@ -578,38 +559,27 @@ case class DeltaWithMetadataWritingSparkTask(
}
}
-private[v2] trait TableWriteExecHelper extends V2TableWriteExec with SupportsV1Write {
+private[v2] trait TableWriteExecHelper extends LeafV2CommandExec {
+ override def output: Seq[Attribute] = Nil
+
protected def writeToTable(
catalog: TableCatalog,
table: Table,
- writeOptions: CaseInsensitiveStringMap,
- ident: Identifier): Seq[InternalRow] = {
+ writeOptions: Map[String, String],
+ ident: Identifier,
+ query: LogicalPlan): Seq[InternalRow] = {
Utils.tryWithSafeFinallyAndFailureCallbacks({
- table match {
- case table: SupportsWrite =>
- val info = LogicalWriteInfoImpl(
- queryId = UUID.randomUUID().toString,
- query.schema,
- writeOptions)
- val writeBuilder = table.newWriteBuilder(info)
-
- val write = writeBuilder.build()
- val writtenRows = write match {
- case v1: V1Write => writeWithV1(v1.toInsertableRelation)
- case v2 => writeWithV2(v2.toBatch)
- }
-
- table match {
- case st: StagedTable => st.commitStagedChanges()
- case _ =>
- }
- writtenRows
+ val relation = DataSourceV2Relation.create(table, Some(catalog), Some(ident))
+ val append = AppendData.byPosition(relation, query, writeOptions)
+ val qe = session.sessionState.executePlan(append)
+ qe.assertCommandExecuted()
+ table match {
+ case st: StagedTable => st.commitStagedChanges()
case _ =>
- // Table does not support writes - staged changes are also rolled back below if table
- // is staging.
- throw QueryExecutionErrors.unsupportedTableWritesError(ident)
}
+
+ Nil
})(catchBlock = {
table match {
// Failure rolls back the staged writes and metadata changes.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org