[SPARK-28554][SQL] Adds a v1 fallback writer implementation for v2 data source codepaths #25348
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -26,16 +26,17 @@ import org.apache.spark.{SparkEnv, SparkException, TaskContext} | |
| import org.apache.spark.executor.CommitDeniedException | ||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.rdd.RDD | ||
| import org.apache.spark.sql.{Dataset, SaveMode} | ||
| import org.apache.spark.sql.catalog.v2.{Identifier, StagingTableCatalog, TableCatalog} | ||
| import org.apache.spark.sql.catalog.v2.expressions.Transform | ||
| import org.apache.spark.sql.catalyst.InternalRow | ||
| import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, NoSuchTableException, TableAlreadyExistsException} | ||
| import org.apache.spark.sql.catalyst.expressions.Attribute | ||
| import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan | ||
| import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode} | ||
| import org.apache.spark.sql.sources.{AlwaysTrue, CreatableRelationProvider, Filter} | ||
| import org.apache.spark.sql.sources.v2.{StagedTable, SupportsWrite} | ||
| import org.apache.spark.sql.sources.v2.writer.{BatchWrite, DataWriterFactory, SupportsDynamicOverwrite, SupportsOverwrite, SupportsTruncate, V1WriteBuilder, WriteBuilder, WriterCommitMessage} | ||
| import org.apache.spark.sql.util.CaseInsensitiveStringMap | ||
| import org.apache.spark.util.{LongAccumulator, Utils} | ||
|
|
||
|
|
@@ -63,10 +64,11 @@ case class CreateTableAsSelectExec( | |
| catalog: TableCatalog, | ||
| ident: Identifier, | ||
| partitioning: Seq[Transform], | ||
| plan: LogicalPlan, | ||
| query: SparkPlan, | ||
| properties: Map[String, String], | ||
| writeOptions: CaseInsensitiveStringMap, | ||
| ifNotExists: Boolean) extends SupportsV1Write { | ||
|
|
||
| import org.apache.spark.sql.catalog.v2.CatalogV2Implicits.IdentifierHelper | ||
|
|
||
|
|
@@ -83,12 +85,17 @@ case class CreateTableAsSelectExec( | |
| catalog.createTable( | ||
| ident, query.schema, partitioning.toArray, properties.asJava) match { | ||
| case table: SupportsWrite => | ||
| val writer = table.newWriteBuilder(writeOptions) | ||
| .withInputDataSchema(query.schema) | ||
| .withQueryId(UUID.randomUUID().toString) | ||
|
|
||
| writer match { | ||
| case v1: V1WriteBuilder => | ||
| val mode = if (ifNotExists) SaveMode.Ignore else SaveMode.ErrorIfExists | ||
|
Contributor
There was a problem hiding this comment. The table was just created above, so both of these modes are incorrect. This should be |
||
| writeWithV1(v1.buildForV1Write(), mode, writeOptions) | ||
| case v2 => | ||
| doWrite(v2.buildForBatch()) | ||
| } | ||
|
|
||
| case _ => | ||
| // table does not support writes | ||
|
|
@@ -114,6 +121,7 @@ case class AtomicCreateTableAsSelectExec( | |
| catalog: StagingTableCatalog, | ||
| ident: Identifier, | ||
| partitioning: Seq[Transform], | ||
| plan: LogicalPlan, | ||
| query: SparkPlan, | ||
| properties: Map[String, String], | ||
| writeOptions: CaseInsensitiveStringMap, | ||
|
|
@@ -129,7 +137,8 @@ case class AtomicCreateTableAsSelectExec( | |
| } | ||
| val stagedTable = catalog.stageCreate( | ||
| ident, query.schema, partitioning.toArray, properties.asJava) | ||
| val mode = if (ifNotExists) SaveMode.Ignore else SaveMode.ErrorIfExists | ||
|
Contributor
There was a problem hiding this comment. Same problem here. I think mode should always be Append because that's the only one with reliable behavior that matches v2. |
||
| writeToStagedTable(stagedTable, writeOptions, ident, mode) | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -147,10 +156,11 @@ case class ReplaceTableAsSelectExec( | |
| catalog: TableCatalog, | ||
| ident: Identifier, | ||
| partitioning: Seq[Transform], | ||
| plan: LogicalPlan, | ||
| query: SparkPlan, | ||
| properties: Map[String, String], | ||
| writeOptions: CaseInsensitiveStringMap, | ||
| orCreate: Boolean) extends SupportsV1Write { | ||
|
|
||
| import org.apache.spark.sql.catalog.v2.CatalogV2Implicits.IdentifierHelper | ||
|
|
||
|
|
@@ -173,12 +183,16 @@ case class ReplaceTableAsSelectExec( | |
| Utils.tryWithSafeFinallyAndFailureCallbacks({ | ||
| createdTable match { | ||
| case table: SupportsWrite => | ||
| val writer = table.newWriteBuilder(writeOptions) | ||
|
Contributor
There was a problem hiding this comment. This isn't a writer, it's a builder. |
||
| .withInputDataSchema(query.schema) | ||
| .withQueryId(UUID.randomUUID().toString) | ||
|
|
||
| writer match { | ||
| case v1: V1WriteBuilder => | ||
| writeWithV1(v1.buildForV1Write(), SaveMode.Overwrite, writeOptions) | ||
| case v2 => | ||
| doWrite(v2.buildForBatch()) | ||
| } | ||
|
|
||
| case _ => | ||
| // table does not support writes | ||
|
|
@@ -207,6 +221,7 @@ case class AtomicReplaceTableAsSelectExec( | |
| catalog: StagingTableCatalog, | ||
| ident: Identifier, | ||
| partitioning: Seq[Transform], | ||
| plan: LogicalPlan, | ||
| query: SparkPlan, | ||
| properties: Map[String, String], | ||
| writeOptions: CaseInsensitiveStringMap, | ||
|
|
@@ -227,7 +242,7 @@ case class AtomicReplaceTableAsSelectExec( | |
| } else { | ||
| throw new CannotReplaceMissingTableException(ident) | ||
| } | ||
| writeToStagedTable(staged, writeOptions, ident, SaveMode.Overwrite) | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -239,11 +254,16 @@ case class AtomicReplaceTableAsSelectExec( | |
| case class AppendDataExec( | ||
| table: SupportsWrite, | ||
| writeOptions: CaseInsensitiveStringMap, | ||
| plan: LogicalPlan, | ||
| query: SparkPlan) extends SupportsV1Write with BatchWriteHelper { | ||
|
|
||
| override protected def doExecute(): RDD[InternalRow] = { | ||
| newWriteBuilder() match { | ||
| case v1: V1WriteBuilder => | ||
| writeWithV1(v1.buildForV1Write(), SaveMode.Append, writeOptions) | ||
| case v2 => | ||
| doWrite(v2.buildForBatch()) | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -261,25 +281,26 @@ case class OverwriteByExpressionExec( | |
| table: SupportsWrite, | ||
| deleteWhere: Array[Filter], | ||
| writeOptions: CaseInsensitiveStringMap, | ||
| plan: LogicalPlan, | ||
| query: SparkPlan) extends SupportsV1Write with BatchWriteHelper { | ||
|
|
||
| private def isTruncate(filters: Array[Filter]): Boolean = { | ||
| filters.length == 1 && filters(0).isInstanceOf[AlwaysTrue] | ||
| } | ||
|
|
||
| override protected def doExecute(): RDD[InternalRow] = { | ||
| newWriteBuilder() match { | ||
| case v1: V1WriteBuilder if isTruncate(deleteWhere) => | ||
| writeWithV1(v1.buildForV1Write(), SaveMode.Overwrite, writeOptions) | ||
|
Contributor
There was a problem hiding this comment. Overwrite is ambiguous and doesn't specify whether the table data should be truncated, replaced dynamically by partition, etc. It isn't possible for v1 sources to guarantee the right behavior -- deleting data that matches
Member
There was a problem hiding this comment. Does it make sense to just do this but only if "deleteWhere") is empty?
Contributor
There was a problem hiding this comment. No. If you pass I was talking to @brkyvz directly and we think that we can use |
||
| case builder: SupportsTruncate if isTruncate(deleteWhere) => | ||
| doWrite(builder.truncate().buildForBatch()) | ||
|
|
||
| case builder: SupportsOverwrite => | ||
| doWrite(builder.overwrite(deleteWhere).buildForBatch()) | ||
|
|
||
| case _ => | ||
| throw new SparkException(s"Table does not support overwrite by expression: $table") | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -463,24 +484,30 @@ object DataWritingSparkTask extends Logging { | |
| } | ||
| } | ||
|
|
||
| private[v2] trait AtomicTableWriteExec extends SupportsV1Write { | ||
| import org.apache.spark.sql.catalog.v2.CatalogV2Implicits.IdentifierHelper | ||
|
|
||
| protected def writeToStagedTable( | ||
| stagedTable: StagedTable, | ||
| writeOptions: CaseInsensitiveStringMap, | ||
| ident: Identifier, | ||
| mode: SaveMode): RDD[InternalRow] = { | ||
| Utils.tryWithSafeFinallyAndFailureCallbacks({ | ||
| stagedTable match { | ||
| case table: SupportsWrite => | ||
| val writer = table.newWriteBuilder(writeOptions) | ||
| .withInputDataSchema(query.schema) | ||
| .withQueryId(UUID.randomUUID().toString) | ||
|
|
||
| val writtenRows = writer match { | ||
| case v1: V1WriteBuilder => | ||
| writeWithV1(v1.buildForV1Write(), SaveMode.Overwrite, writeOptions) | ||
| case v2 => | ||
| doWrite(v2.buildForBatch()) | ||
| } | ||
| stagedTable.commitStagedChanges() | ||
| writtenRows | ||
|
|
||
| case _ => | ||
| // Table does not support writes - staged changes are also rolled back below. | ||
| throw new SparkException( | ||
|
|
@@ -501,3 +528,19 @@ private[v2] case class DataWritingSparkTaskResult( | |
| * Sink progress information collected after commit. | ||
| */ | ||
| private[sql] case class StreamWriterCommitProgress(numOutputRows: Long) | ||
|
|
||
| /** | ||
| * A trait that allows Tables that use V1 Writer interfaces to write data. | ||
| */ | ||
| sealed trait SupportsV1Write extends V2TableWriteExec { | ||
| def plan: LogicalPlan | ||
|
|
||
| protected def writeWithV1( | ||
| relation: CreatableRelationProvider, | ||
| mode: SaveMode, | ||
| options: CaseInsensitiveStringMap): RDD[InternalRow] = { | ||
| relation.createRelation( | ||
| sqlContext, mode, options.asScala.toMap, Dataset.ofRows(sqlContext.sparkSession, plan)) | ||
|
Contributor
There was a problem hiding this comment. I think this should use the original options map that preserves the case that was passed in. |
||
| sparkContext.emptyRDD | ||
|
Member
There was a problem hiding this comment. I might be missing something here, but why does this return an RDD and why is it empty?
Member
There was a problem hiding this comment. Ok so I see this is used in Atomic Table Writes, that seems a bit wrong to me, shouldn't we just not support the atomic table writes with V1 Fallback? Seems like we are violating the contract by returning empty regardless of what happens.
Contributor
Author
There was a problem hiding this comment. What is the contract? All data write commands have returned an empty result since the beginning of time, e.g. look at
Member
There was a problem hiding this comment. If this is always empty why do we save it as writtenRows here? This is just to hold a reference to the empty result set?
Contributor
Author
There was a problem hiding this comment. yeah. It's pretty much dead code. I think if we decide to change what to return later, it's easier to change 1-2 places vs 'n' different operators. |
||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.sql.sources.v2.writer | ||
|
|
||
| import org.apache.spark.annotation.{Experimental, Unstable} | ||
| import org.apache.spark.sql.sources.CreatableRelationProvider | ||
|
|
||
| /** | ||
| * A trait that should be implemented by V1 DataSources that would like to leverage the DataSource | ||
| * V2 write code paths. | ||
| * | ||
| * @since 3.0.0 | ||
| */ | ||
| @Experimental | ||
| @Unstable | ||
| trait V1WriteBuilder extends WriteBuilder { | ||
|
Contributor
There was a problem hiding this comment. since it's a mixin trait, how about naming it
Contributor
Author
There was a problem hiding this comment. Hmm, all of the other mixins and their functions return a |
||
|
|
||
| /** | ||
| * Creates a [[CreatableRelationProvider]] that allows saving a DataFrame to a | ||
| * a destination (using data source-specific parameters). | ||
| * | ||
| * The relation will receive a string to string map of options that will be case sensitive, | ||
| * therefore the implementation of the data source should be able to handle case insensitive | ||
| * option checking. | ||
| * | ||
| * @since 3.0.0 | ||
| */ | ||
| def buildForV1Write(): CreatableRelationProvider = { | ||
|
Contributor
There was a problem hiding this comment. Why not add a path for
Contributor
Author
There was a problem hiding this comment. insert semantics are weird. It doesn't support the passing in of options as well. All sources that I checked support |
||
| throw new UnsupportedOperationException(getClass.getName + " does not support batch write") | ||
|
Contributor
There was a problem hiding this comment. Do we need to throw an exception by default? This is a mixin trait. If users mix it in, they must provide the implementation. |
||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why we reuse the v2
CreateTableAsSelectExecfor v1 fallback, but notAppendDataExec?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
because table creation happens after certain operations with side effects in CTAS and RTAS.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this is because CTAS and RTAS plans don't have a table instance. CTAS and RTAS create a table at runtime and then write to it.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This explains why we must reuse
CreateTableAsSelectExecfor v1 fallback, but why can't we reuseAppendDataExecfor v1 fallback as well?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's just cleaner to have a separation of which code path was used. (Shows up separately in the SQL tab, etc)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If end-users look at the SQL tab and see
AppendDataExecV1, they would expect to see v1 version of CTAS physical plan as well, and may report a bug if they don't see it.BTW I think there are other ways to implement this feature (users know if v1 fallback is triggered from SQL tab), e.g. we can use SQLMetrics to report it, which can be updated at runtime and support CTAS as well.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't think most (98%) of Spark users already don't know what the physical nodes stand for, etc. I was thinking of using metrics as well. We just wanted to keep the nodes separate, because the semantics are a bit different for things like
OverwriteByExpression, where the source's implementation of the Overwrite + Append (v2 behavior) may not be atomic, etcThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
good point on having different physical nodes for different semantics. It seems to me that the semantic of append is the same between v1 and v2. Shall we reuse
AppendDataExec? Then we can make a good story: If semantic is the same, reuse the v2 plan (CREATE TABLE, CTAS, APPEND). Otherwise, create a new physical node (OVERWRITE).There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I like the idea to use the same physical plan node when the semantics match, but I'm not sure that it is possible to make strong guarantees about not changing v1 if we do. The problem is that there are multiple v1 plan nodes for the same operation, which could have slightly different behavior.
We could take the time to inspect the v1 implementations and convert, but that adds risk and takes time. It also isn't needed to migrate to v2, and wouldn't speed up the migration, unless I'm missing something. So it probably doesn't provide enough value to make doing it worth while.