Spark 3.4: Split update into delete and insert for position deltas #7646

aokolnychyi · 2023-05-18T15:45:54Z

This PR contains a subset of changes from #7637 and is required for #7633.

aokolnychyi · 2023-05-18T17:22:49Z

spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/SparkDistributionAndOrderingUtil.java

@@ -218,7 +218,7 @@ private static Distribution buildPositionDeleteUpdateDistribution(
  }

  public static SortOrder[] buildPositionDeltaOrdering(Table table, Command command) {
-    if (command == DELETE || command == UPDATE) {


This is purely to avoid test failures for now. I will rework distribution and ordering in a follow-up.

aokolnychyi · 2023-05-18T17:46:15Z

...k/v3.4/spark-extensions/src/jmh/java/org/apache/iceberg/spark/UpdateProjectionBenchmark.java

+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@BenchmarkMode(Mode.SingleShotTime)
+public class UpdateProjectionBenchmark {


The new approach is a bit slower (cause we have an iterator instead of a projection) but I don't think it would outweigh the benefits we can achieve by having a better clustering. Also, we can optimize the new approach further by providing codegen support for the new expression and getting rid of the position delete sort. I just wanted to confirm there is no severe degradation.

Benchmark Mode Cnt Score Error Units [OLD] UpdateProjectionBenchmark.copyOnWriteUpdate10Percent ss 5 15.721 ± 0.409 s/op [NEW] UpdateProjectionBenchmark.copyOnWriteUpdate10Percent ss 5 15.728 ± 0.162 s/op [OLD] UpdateProjectionBenchmark.copyOnWriteUpdate30Percent ss 5 15.165 ± 0.084 s/op [NEW] UpdateProjectionBenchmark.copyOnWriteUpdate30Percent ss 5 15.071 ± 0.104 s/op [OLD] UpdateProjectionBenchmark.copyOnWriteUpdate75Percent ss 5 15.581 ± 0.198 s/op [NEW] UpdateProjectionBenchmark.copyOnWriteUpdate75Percent ss 5 15.437 ± 0.118 s/op [OLD] UpdateProjectionBenchmark.mergeOnRead10Percent ss 5 4.682 ± 0.173 s/op [NEW] UpdateProjectionBenchmark.mergeOnRead10Percent ss 5 4.923 ± 0.082 s/op [OLD] UpdateProjectionBenchmark.mergeOnReadUpdate30Percent ss 5 9.475 ± 0.587 s/op [NEW] UpdateProjectionBenchmark.mergeOnReadUpdate30Percent ss 5 10.251 ± 0.968 s/op [OLD] UpdateProjectionBenchmark.mergeOnReadUpdate75Percent ss 5 23.025 ± 0.135 s/op [NEW] UpdateProjectionBenchmark.mergeOnReadUpdate75Percent ss 5 26.260 ± 0.733 s/op

Here are results for the existing benchmark for merging rows.

Benchmark Mode Cnt Score Error Units [OLD] MergeCardinalityCheckBenchmark.copyOnWriteMergeCardinalityCheck10PercentUpdates ss 5 11.287 ± 0.978 s/op [NEW] MergeCardinalityCheckBenchmark.copyOnWriteMergeCardinalityCheck10PercentUpdates ss 5 11.100 ± 0.465 s/op [OLD] MergeCardinalityCheckBenchmark.copyOnWriteMergeCardinalityCheck30PercentUpdates ss 5 11.344 ± 0.272 s/op [NEW] MergeCardinalityCheckBenchmark.copyOnWriteMergeCardinalityCheck30PercentUpdates ss 5 11.417 ± 1.082 s/op [OLD] MergeCardinalityCheckBenchmark.copyOnWriteMergeCardinalityCheck90PercentUpdates ss 5 11.835 ± 0.322 s/op [NEW] MergeCardinalityCheckBenchmark.copyOnWriteMergeCardinalityCheck90PercentUpdates ss 5 11.887 ± 3.269 s/op [OLD] MergeCardinalityCheckBenchmark.mergeOnReadMergeCardinalityCheck10PercentUpdates ss 5 7.817 ± 0.245 s/op [NEW] MergeCardinalityCheckBenchmark.mergeOnReadMergeCardinalityCheck10PercentUpdates ss 5 7.106 ± 0.240 s/op [OLD] MergeCardinalityCheckBenchmark.mergeOnReadMergeCardinalityCheck30PercentUpdates ss 5 12.440 ± 0.339 s/op [NEW] MergeCardinalityCheckBenchmark.mergeOnReadMergeCardinalityCheck30PercentUpdates ss 5 11.662 ± 0.258 s/op [OLD] MergeCardinalityCheckBenchmark.mergeOnReadMergeCardinalityCheck90PercentUpdates ss 5 26.052 ± 0.865 s/op [NEW] MergeCardinalityCheckBenchmark.mergeOnReadMergeCardinalityCheck90PercentUpdates ss 5 23.681 ± 1.110 s/op

The new approach performs a tad better for MoR MERGE. Could be related to the projection logic in the writer but it does not really matter as long as it is not worse.

aokolnychyi · 2023-05-18T19:54:15Z

...extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala

    }
  }

  private def buildMergeRowsOutput(
-      matchedOutputs: Seq[Seq[Expression]],
+      matchedOutputs: Seq[Seq[Seq[Expression]]],


Matched actions (the only type that can contain UPDATE) may produce a sequence of outputs per action now (delete + insert for deltas). Unmatched actions only produce one output.

aokolnychyi · 2023-05-18T19:55:12Z

...extensions/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala

      mergeRows: MergeRows,
      rowAttrs: Seq[Attribute],
      rowIdAttrs: Seq[Attribute],
      metadataAttrs: Seq[Attribute]): WriteDeltaProjections = {

-    val outputAttrs = mergeRows.output


Moved into parent to reuse.

aokolnychyi · 2023-05-18T19:55:31Z

...ns/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelIcebergCommand.scala

@@ -67,6 +72,97 @@ trait RewriteRowLevelIcebergCommand extends RewriteRowLevelCommand {
    ProjectingInternalRow(schema, projectedOrdinals)
  }

+  protected def buildDeltaProjections(


From the rule that rewrites MERGE.

aokolnychyi · 2023-05-18T19:56:15Z

...spark-extensions/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/UpdateRows.scala

+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.util.truncatedString
+
+case class UpdateRows(


Similar to MergeRows but simpler.

aokolnychyi · 2023-05-18T19:56:54Z

...-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/MergeRowsExec.scala

-  private def applyProjection(
-      actions: Seq[(BasePredicate, Option[UnsafeProjection])],
-      inputRow: InternalRow): InternalRow = {
+  // This method is responsible for processing a input row to emit the resultant row with an


This comment is copied from below.

aokolnychyi · 2023-05-18T19:59:16Z

...-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/MergeRowsExec.scala

-        null
+    val projectTargetCols = createProjection(targetOutput)
+
+    val cardinalityCheck = if (performCardinalityCheck) {


I considered using Option but I was a bit concerned how it would look like in bytecode. I'd need to use foreach on it, which has a nested if. I hope JVM would be smart enough to detect the empty method.

RussellSpitzer · 2023-05-18T20:46:42Z

...extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/UpdateRowsExec.scala

+    UnsafeProjection.create(exprs, child.output)
+  }
+
+  class UpdateAsDeleteAndInsertRowIterator(


I would put a doc somewhere in here

/** Splits an iterator of update merge rows into delete and update rows. Each input row becomes two output rows, first the delete, then the insert. **/

Or something like that

RussellSpitzer · 2023-05-18T21:01:48Z

...-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/MergeRowsExec.scala

-    } else {
-      processRow
+    private def applyMatchedActions(row: InternalRow): InternalRow = {
+      for (action <- matchedActions) {


Feel free to ignore this if it's too Scala ish but I would do something like

return matchedActions.find(row => action.cond.eval(row)).match{ case split: Split => cachedExtraRow = split.projectExtraRow(row) split.projectRow(row) case project: Project => project.apply(row) }.orElse(null)

For these find first situations, I would also probably just keep the option and return None but that's not required.

This is something that is invoked per record so I wanted it to produce as few objects as possible and as simple bytecode as possible, hoping JIT would then make smart choices.

I do like None and find but I am paranoid it would add more calls.

I think that's a good paranoia :) I just think it's a little less readable. I have no problem keeping it this way but you do already have benchmarks set up so you could test it... ;)

Let me make a note of this and test it in parallel to working on distribution and ordering.

RussellSpitzer

Looks good to me, Excited to read over the distribution work next

aokolnychyi · 2023-05-19T20:34:43Z

Thanks, @RussellSpitzer!

github-actions bot added the spark label May 18, 2023

aokolnychyi commented May 18, 2023

View reviewed changes

RussellSpitzer reviewed May 18, 2023

View reviewed changes

Spark 3.4: Split update into delete and insert for position deltas

461b6b6

aokolnychyi force-pushed the split-updates branch from 1fe47eb to 461b6b6 Compare May 19, 2023 18:59

RussellSpitzer approved these changes May 19, 2023

View reviewed changes

aokolnychyi merged commit 2f61a08 into apache:master May 19, 2023
31 checks passed

aokolnychyi mentioned this pull request May 23, 2023

Spark 3.4: Codegen support for UpdateRowsExec #7691

Closed

chenjunjiedada mentioned this pull request Aug 16, 2023

Spark: support use-table-distribution-and-ordering in session conf #8164

Open

chengxianglibra mentioned this pull request Aug 17, 2023

Data Deletion Issue with MERGE INTO ... WHEN MATCHED THEN DELETE statement in Iceberg 1.3 with Spark 3.4.1 #8126

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Spark 3.4: Split update into delete and insert for position deltas #7646

Spark 3.4: Split update into delete and insert for position deltas #7646

aokolnychyi commented May 18, 2023

aokolnychyi May 18, 2023

aokolnychyi May 18, 2023

aokolnychyi May 18, 2023

aokolnychyi May 18, 2023 •

edited

aokolnychyi May 18, 2023

aokolnychyi May 18, 2023

aokolnychyi May 18, 2023

aokolnychyi May 18, 2023

aokolnychyi May 18, 2023

RussellSpitzer May 18, 2023 •

edited

aokolnychyi May 18, 2023

RussellSpitzer May 18, 2023

aokolnychyi May 18, 2023 •

edited

RussellSpitzer May 19, 2023

aokolnychyi May 19, 2023

RussellSpitzer left a comment

aokolnychyi commented May 19, 2023

Spark 3.4: Split update into delete and insert for position deltas #7646

Spark 3.4: Split update into delete and insert for position deltas #7646

Conversation

aokolnychyi commented May 18, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi May 18, 2023 • edited

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

RussellSpitzer May 18, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi May 18, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

RussellSpitzer left a comment

Choose a reason for hiding this comment

aokolnychyi commented May 19, 2023

aokolnychyi May 18, 2023 •

edited

RussellSpitzer May 18, 2023 •

edited

aokolnychyi May 18, 2023 •

edited