[SPARK-33808][SQL] DataSource V2: Build logical writes in the optimizer #30806

aokolnychyi · 2020-12-16T15:31:47Z

What changes were proposed in this pull request?

This PR adds logic to build logical writes introduced in SPARK-33779.

Note: This PR contains a subset of changes discussed in PR #29066.

Why are the changes needed?

These changes are the next step as discussed in the design doc for SPARK-23889.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests.

aokolnychyi · 2020-12-16T15:33:02Z

I added @rdblue as a co-author of the change as he addressed the initial comments to my PR and I included his changes.

aokolnychyi · 2020-12-16T15:34:38Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala

@@ -94,7 +96,8 @@ case class OverwriteByExpression(
    deleteExpr: Expression,
    query: LogicalPlan,
    writeOptions: Map[String, String],
-    isByName: Boolean) extends V2WriteCommand {
+    isByName: Boolean,
+    write: Option[Write] = None) extends V2WriteCommand {


Making this optional allows us to reuse the same plan before we construct a write and after. Having None here means the logical write hasn't been constructed yet. This allows us to have idempotent rules in the optimizer.

aokolnychyi · 2020-12-16T15:37:25Z

...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

-          AppendDataExecV1(v1, writeOptions.asOptions, query, refreshCache(r)) :: Nil
+          AppendDataExecV1(
+            v1, writeOptions.asOptions, query,
+            refreshCache(r), write.map(_.asInstanceOf[V1Write])) :: Nil


I think there is one open point we need to discuss: do we want to always apply the new logic or should we expose a feature flag and construct logical writes only if the flag is enabled? I'd vote for always constructing writes using the new logic as it feels quite same and it does not have the burden of maintaining one more config. In addition, this will allow us to simply this PR a bit and get rid of optional writes in exec nodes.

I'd be curious about what everybody thinks here. I will be okay either way.

aokolnychyi · 2020-12-16T15:38:17Z

cc @dongjoon-hyun @dbtsai @sunchao @rdblue @viirya @cloud-fan @HyukjinKwon @HeartSaVioR @holdenk

aokolnychyi · 2020-12-16T15:40:25Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala


-  override protected def run(): Seq[InternalRow] = {
-    writeWithV1(newWriteBuilder().buildForV1Write(), refreshCache = refreshCache)
+  override protected def buildAndRun(): Seq[InternalRow] = {


We can get rid of all buildAndRun methods if we are OK to apply the new logic all the time.

I'm interested to hear what @dongjoon-hyun thinks about this.

I think we should have a different physical node for each write so that the explain plan shows what is happening. Otherwise, the approach to support building the batch write here or building it in the optimizer was mainly to be able to turn this on and off in our environment. I doubt that is needed in other situations.

I think I would be for removing all of the buildAndRun methods and always building the write in the optimizer.

Does it cause many code change on top of this? If it is not intrusive, it sounds reasonable in that context. I'd give +1 for the direction, @rdblue .

Getting rid of buildAndRun would also ensure we don't have to maintain the same logic in two places.

I'm also +1 on getting rid of buildAndRun.

It sounds like an intermediate consensus, I'll update the PR but we can revisit this once we have more input from others.

aokolnychyi · 2020-12-16T15:41:05Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala

@@ -89,6 +90,21 @@ sealed trait V1FallbackWriters extends V2CommandExec with SupportsV1Write {

  def table: SupportsWrite
  def writeOptions: CaseInsensitiveStringMap
+  def refreshCache: () => Unit
+  def write: Option[V1Write] = None


Same here: Option[V1Write] can become just V1Write if we are OK to apply the new logic all the time.

aokolnychyi · 2020-12-16T15:41:27Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala

    val session = sqlContext.sparkSession
    // The `plan` is already optimized, we should not analyze and optimize it again.
    relation.insert(AlreadyOptimized.dataFrame(session, plan), overwrite = false)
-    refreshCache()


Refresh moved to run.

cc @sunchao

SparkQA · 2020-12-16T16:15:11Z

Test build #132896 has finished for PR 30806 at commit ce334cc.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait V2ExistingTableWriteExec extends V2TableWriteExec

aokolnychyi · 2020-12-16T16:17:10Z

Hm, the failure is a bit weird and does not seem related.

Setting status of ce334cc48c8ea615a2465a9111fd0d9f6f606eb2 to FAILURE with url https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/132896/ and message: 'Build finished. '
FileNotFoundException means that the credentials Jenkins is using is probably wrong. Or the user account does not have write access to the repo.
org.kohsuke.github.GHFileNotFoundException: https://api.github.com/repos/apache/spark/statuses/ce334cc48c8ea615a2465a9111fd0d9f6f606eb2 {"message":"Not Found","documentation_url":"https://docs.github.com/rest/reference/repos#create-a-commit-status"}

rdblue · 2020-12-16T17:02:01Z

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala

@@ -39,6 +39,9 @@ class SparkOptimizer(
    // TODO: move SchemaPruning into catalyst
    SchemaPruning :: V2ScanRelationPushDown :: PruneFileSourcePartitions :: Nil

+  override def dataSourceRewriteRules: Seq[Rule[LogicalPlan]] =
+    V2Writes :: Nil


I would probably put this in the early pushdown batch, even though the name doesn't match. The rewrite batch needs to run before this so that writes created by it run V2Writes afterward. That's the same reason why early pushdown runs after plan rewrites.

Ok, it makes sense after thinking more about it.

SparkQA · 2020-12-16T17:04:28Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37498/

SparkQA · 2020-12-16T17:43:17Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37498/

SparkQA · 2020-12-16T21:08:48Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37502/

SparkQA · 2020-12-16T21:15:59Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37502/

SparkQA · 2020-12-17T00:18:29Z

Test build #132900 has finished for PR 30806 at commit db6abdb.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

jzhuge

Looks good overall. Minor suggestions and nits.

jzhuge · 2020-12-17T05:59:14Z

...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

@@ -188,15 +189,20 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
            orCreate = orCreate) :: Nil
      }

-    case AppendData(r: DataSourceV2Relation, query, writeOptions, _) =>
+    case AppendData(r: DataSourceV2Relation, query, writeOptions, _, write) =>


Is write guaranteed not be None?

How about rewriting this case as follows?

case AppendData(r @ DataSourceV2Relation(v1: SupportsWrite, _, _, _, _), query, writeOptions, _, Some(v1Write: V1Write)) if v1.supports(TableCapability.V1_BATCH_WRITE) => AppendDataExecV1(v1, writeOptions.asOptions, query, refreshCache(r), v1Write) :: Nil case AppendData(r @ DataSourceV2Relation(v2: SupportsWrite, _, _, _, _), query, writeOptions, _, Some(write)) => AppendDataExec(v2, writeOptions.asOptions, planLater(query), refreshCache(r), write) :: Nil

It is not exactly the same as the existing code. Some unmatched cases (not sure how many or if any) will fall through. Exception will be thrown later, instead of right here upon instance cast or Option.get.

+1 to this idea. It's guaranteed that the write will be Some not None at the planner, so matching Some(write) is better.

It's possible that the implementation declares V1_BATCH_WRITE but doesn't return V1Write. We should give clear error message if it happens:

case AppendData(r @ DataSourceV2Relation(v1: SupportsWrite, _, _, _, _), query, writeOptions, _, Some(write)) if v1.supports(TableCapability.V1_BATCH_WRITE) => if (!write.isInstanceOf[V1Write]) throw ... ...

Good idea to add more meaningful exception here.

I've updated this place. Could you take a look, @jzhuge and @cloud-fan?

Looks great! Thanks for taking care of Overwrite* cases as well.

jzhuge · 2020-12-17T05:59:21Z

...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

-          AppendDataExecV1(v1, writeOptions.asOptions, query, refreshCache(r)) :: Nil
+          AppendDataExecV1(
+            v1, writeOptions.asOptions, query,
+            refreshCache(r), write.get.asInstanceOf[V1Write]) :: Nil


Possible to avoid instance cast? See my suggestion above.

jzhuge · 2020-12-17T05:59:24Z

...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

-          AppendDataExec(v2, writeOptions.asOptions, planLater(query), refreshCache(r)) :: Nil
+          AppendDataExec(
+            v2, writeOptions.asOptions, planLater(query),
+            refreshCache(r), write.get) :: Nil


Possible to avoid Option.get? See my suggestion above.

Got rid of it.

jzhuge · 2020-12-17T05:59:38Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala

-  protected def writeWithV1(
-      relation: InsertableRelation,
-      refreshCache: () => Unit = () => ()): Seq[InternalRow] = {
+  protected def writeWithV1(relation: InsertableRelation): Seq[InternalRow] = {


Nicely simplified

jzhuge · 2020-12-17T05:59:50Z

...e/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala

-            case v1: V1WriteBuilder => writeWithV1(v1.buildForV1Write())
-            case v2 => writeWithV2(v2.buildForBatch())
+          val write = writeBuilder.build()
+          val writtenRows = write match {


Nit: merge line 451-454 into:

val writtenRows = table.newWriteBuilder(info).build() match {

I don't feel strongly about this place and can update it. However, I do prefer to split different logical parts into different variables. Here, I've separated building a logical write from actually writing the records. Let me know what are your thoughts, @jzhuge.

Fine with me

cloud-fan · 2020-12-17T12:28:41Z

sql/core/src/main/java/org/apache/spark/sql/connector/write/V1WriteBuilder.java

@@ -33,6 +33,11 @@
 */
 @Unstable
 public interface V1WriteBuilder extends WriteBuilder {


This is an unstable API, can we just remove it and only use V1Write?

I think that should simplify the V1 fallback API. I'll update it, @cloud-fan.

Got rid of V1WriteBuilder and tried to update docs too. Let me know if I missed places, @cloud-fan.

aokolnychyi · 2020-12-17T20:10:49Z

...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

      }

-    case OverwriteByExpression(r: DataSourceV2Relation, deleteExpr, query, writeOptions, _) =>
-      // fail if any filter cannot be converted. correctness depends on removing all matching data.


I removed the filter conversion as it is done earlier now.

aokolnychyi · 2020-12-17T20:12:37Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala

+
+    case o @ OverwriteByExpression(r: DataSourceV2Relation, deleteExpr, query, options, _, None) =>
+      // fail if any filter cannot be converted. correctness depends on removing all matching data.
+      val filters = splitConjunctivePredicates(deleteExpr).flatMap { pred =>


I kept the old logic but I am not sure whether we should also normalize filters. Thoughts, @cloud-fan @rdblue?

I think we should, to follow DS v1 and file source.

I've created SPARK-33868 as a follow-up item. I will keep the old behavior in this PR.

SparkQA · 2020-12-17T20:27:28Z

Test build #132965 has finished for PR 30806 at commit e6335c0.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-12-17T21:04:39Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37568/

SparkQA · 2020-12-17T21:36:35Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37568/

aokolnychyi · 2020-12-21T11:31:08Z

Retest this, please.

SparkQA · 2020-12-21T11:47:23Z

Test build #133150 has finished for PR 30806 at commit e6335c0.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-12-21T12:39:31Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37750/

SparkQA · 2020-12-21T12:44:52Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37750/

Lead-authored-by: Anton Okolnychyi <aokolnychyi@apple.com> Co-authored-by: Ryan Blue <blue@apache.org>

aokolnychyi · 2020-12-21T13:19:15Z

I think certain checks are expected to fail:

[error] spark-sql: Failed binary compatibility check against org.apache.spark:spark-sql_2.12:3.0.0! Found 1 potential problems (filtered 914)
[error]  * interface org.apache.spark.sql.connector.write.V1WriteBuilder does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("org.apache.spark.sql.connector.write.V1WriteBuilder")

Per discussion here.

SparkQA · 2020-12-21T13:44:46Z

Test build #133160 has finished for PR 30806 at commit 84241e0.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-12-21T14:20:35Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37759/

cloud-fan · 2020-12-21T14:25:51Z

...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

+        case v2Write =>
+          throw new AnalysisException(
+            s"Table ${v1.name} declares ${TableCapability.V1_BATCH_WRITE} capability but " +
+            s"${v2Write.getClass} is not an instance of ${classOf[V1Write]}")


classOf[V1Write].getName

cloud-fan · 2020-12-21T14:31:18Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala

@@ -65,7 +66,8 @@ case class AppendData(
    table: NamedRelation,
    query: LogicalPlan,
    writeOptions: Map[String, String],
-    isByName: Boolean) extends V2WriteCommand {
+    isByName: Boolean,
+    write: Option[Write] = None) extends V2WriteCommand {
  override def withNewQuery(newQuery: LogicalPlan): AppendData = copy(query = newQuery)
  override def withNewTable(newTable: NamedRelation): AppendData = copy(table = newTable)


Shall we add override lazy val resolved = ... && write.isDefined in V2WriteCommand? It's safer to make sure that the analyzer creates the Write object.

Sounds like a good idea but we actually construct the Write object in the optimizer after the operator optimization is done to ensure we operate on optimal expressions.

ah I see, let's leave it then.

cloud-fan

LGTM

cloud-fan · 2020-12-21T14:33:15Z

We need to update project/MimaExcludes.scala to fix the mima errors.

SparkQA · 2020-12-21T14:59:43Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37759/

cloud-fan · 2020-12-21T15:42:17Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala

@@ -132,7 +135,8 @@ case class OverwritePartitionsDynamic(
    table: NamedRelation,
    query: LogicalPlan,
    writeOptions: Map[String, String],
-    isByName: Boolean) extends V2WriteCommand {
+    isByName: Boolean,
+    write: Option[Write] = None) extends V2WriteCommand {


Not related to this PR. I'm thinking that if we should have an optional Scan object in DataSourceV2Relation, instead of having a new logical plan DataSourceV2ScanRelation. It's simpler and consistent with the write logical plans. cc @rdblue

I think that's a good idea.

Quick question: DataSourceV2Relation is also used inside write nodes like AppendData. If we add an optional scan, will that mean we will leak a read-specific concept into write plans?

cc @rdblue @cloud-fan

For AppendData, we intentionally do not treat the table as a child, which means the pushdown rule won't apply for it and the Scan object will always be None in AppendData.

SparkQA · 2020-12-22T00:00:42Z

Test build #133168 has finished for PR 30806 at commit 882e321.

This patch fails from timeout after a configured wait of 500m.
This patch merges cleanly.
This patch adds no public classes.

rdblue · 2020-12-22T00:03:00Z

Retest this please.

SparkQA · 2020-12-22T01:17:43Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37784/

SparkQA · 2020-12-22T01:47:46Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37784/

AmplabJenkins · 2020-12-22T01:47:49Z

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/37784/

SparkQA · 2020-12-22T02:56:07Z

Test build #133186 has finished for PR 30806 at commit 882e321.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2020-12-22T03:30:40Z

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/133186/

cloud-fan · 2020-12-22T08:23:52Z

thanks, merging to master!

aokolnychyi · 2020-12-22T09:19:14Z

Thanks @cloud-fan @jzhuge @rdblue @sunchao @dongjoon-hyun!

aokolnychyi commented Dec 16, 2020

View reviewed changes

rdblue reviewed Dec 16, 2020

View reviewed changes

github-actions bot added the SQL label Dec 16, 2020

jzhuge approved these changes Dec 17, 2020

View reviewed changes

cloud-fan reviewed Dec 17, 2020

View reviewed changes

aokolnychyi commented Dec 17, 2020

View reviewed changes

[SPARK-33808][SQL] DataSource V2: Build logical writes in the optimizer

84241e0

Lead-authored-by: Anton Okolnychyi <aokolnychyi@apple.com> Co-authored-by: Ryan Blue <blue@apache.org>

aokolnychyi force-pushed the spark-33808 branch from e6335c0 to 84241e0 Compare December 21, 2020 13:18

cloud-fan reviewed Dec 21, 2020

View reviewed changes

cloud-fan approved these changes Dec 21, 2020

View reviewed changes

Review comments

882e321

cloud-fan reviewed Dec 21, 2020

View reviewed changes

github-actions bot added the BUILD label Dec 21, 2020

rdblue approved these changes Dec 21, 2020

View reviewed changes

cloud-fan closed this in 2562183 Dec 22, 2020

ulysses-you mentioned this pull request Nov 30, 2021

[SPARK-37287][SQL] Pull out dynamic partition and bucket sort from FileFormatWriter #34568

Closed

[SPARK-33808][SQL] DataSource V2: Build logical writes in the optimizer #30806

[SPARK-33808][SQL] DataSource V2: Build logical writes in the optimizer #30806

Conversation

aokolnychyi commented Dec 16, 2020

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

aokolnychyi commented Dec 16, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi commented Dec 16, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 16, 2020

aokolnychyi commented Dec 16, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 16, 2020

SparkQA commented Dec 16, 2020

SparkQA commented Dec 16, 2020

SparkQA commented Dec 16, 2020

SparkQA commented Dec 17, 2020

jzhuge left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Dec 17, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Dec 17, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 17, 2020

SparkQA commented Dec 17, 2020

SparkQA commented Dec 17, 2020

aokolnychyi commented Dec 21, 2020

SparkQA commented Dec 21, 2020

SparkQA commented Dec 21, 2020

SparkQA commented Dec 21, 2020

aokolnychyi commented Dec 21, 2020

SparkQA commented Dec 21, 2020

SparkQA commented Dec 21, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan left a comment

Choose a reason for hiding this comment

cloud-fan commented Dec 21, 2020

SparkQA commented Dec 21, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Dec 22, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi commented Dec 16, 2020 •

edited

Loading

aokolnychyi Dec 17, 2020 •

edited

Loading

aokolnychyi Dec 17, 2020 •

edited

Loading

aokolnychyi Dec 22, 2020 •

edited

Loading