diff --git a/delta-lake/common/src/main/scala/com/nvidia/spark/rapids/delta/GpuRapidsProcessDeltaMergeJoinExec.scala b/delta-lake/common/src/main/scala/com/nvidia/spark/rapids/delta/GpuRapidsProcessDeltaMergeJoinExec.scala index b9961523682..b814ced34b1 100644 --- a/delta-lake/common/src/main/scala/com/nvidia/spark/rapids/delta/GpuRapidsProcessDeltaMergeJoinExec.scala +++ b/delta-lake/common/src/main/scala/com/nvidia/spark/rapids/delta/GpuRapidsProcessDeltaMergeJoinExec.scala @@ -1,5 +1,5 @@ /* - * Copyright (c) 2023-2025, NVIDIA CORPORATION. + * Copyright (c) 2023-2026, NVIDIA CORPORATION. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,7 +18,7 @@ package com.nvidia.spark.rapids.delta import scala.collection.mutable.ArrayBuffer -import ai.rapids.cudf.{NvtxColor, Table} +import ai.rapids.cudf.{ColumnVector, NvtxColor, Scalar, Table} import com.nvidia.spark.rapids._ import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource} import com.nvidia.spark.rapids.AssertUtils.assertInTests @@ -49,7 +49,8 @@ object RapidsProcessDeltaMergeJoinStrategy extends SparkStrategy { notMatchedBySourceConditions = p.notMatchedBySourceConditions, notMatchedBySourceOutputs = p.notMatchedBySourceOutputs, noopCopyOutput = p.noopCopyOutput, - deleteRowOutput = p.deleteRowOutput)) + deleteRowOutput = p.deleteRowOutput, + rowDroppedColumnIndex = p.rowDroppedColumnIndex)) case _ => Nil } } @@ -66,7 +67,10 @@ case class RapidsProcessDeltaMergeJoin( notMatchedBySourceConditions: Seq[Expression], notMatchedBySourceOutputs: Seq[Seq[Seq[Expression]]], noopCopyOutput: Seq[Expression], - deleteRowOutput: Seq[Expression]) extends UnaryNode { + deleteRowOutput: Seq[Expression], + // Position of the row-dropped control column in every projected output row. When None, the + // column is located by its name in `output`, falling back to the position after `output`. + rowDroppedColumnIndex: Option[Int] = None) extends UnaryNode { @transient override lazy val references: AttributeSet = inputSet @@ -88,7 +92,8 @@ case class RapidsProcessDeltaMergeJoinExec( notMatchedConditions: Seq[Expression], notMatchedOutputs: Seq[Seq[Seq[Expression]]], noopCopyOutput: Seq[Expression], - deleteRowOutput: Seq[Expression]) extends UnaryExecNode { + deleteRowOutput: Seq[Expression], + rowDroppedColumnIndex: Option[Int] = None) extends UnaryExecNode { override protected def doExecute(): RDD[InternalRow] = { throw new IllegalStateException("Should have been replaced by a GpuRapidsProcessMergeJoinExec") @@ -122,7 +127,8 @@ class RapidsProcessDeltaMergeJoinMeta( notMatchedBySourceConditions = p.notMatchedBySourceConditions.map(convertExprToGpu), notMatchedBySourceOutputs = p.notMatchedBySourceOutputs.map(_.map(_.map(convertExprToGpu))), noopCopyOutput = p.noopCopyOutput.map(convertExprToGpu), - deleteRowOutput = p.deleteRowOutput.map(convertExprToGpu)) + deleteRowOutput = p.deleteRowOutput.map(convertExprToGpu), + rowDroppedColumnIndex = p.rowDroppedColumnIndex) } private def convertExprToGpu(e: Expression): Expression = { @@ -149,14 +155,11 @@ case class GpuRapidsProcessDeltaMergeJoinExec( notMatchedBySourceConditions: Seq[Expression], notMatchedBySourceOutputs: Seq[Seq[Seq[Expression]]], noopCopyOutput: Seq[Expression], - deleteRowOutput: Seq[Expression]) extends UnaryExecNode with GpuExec { + deleteRowOutput: Seq[Expression], + rowDroppedColumnIndex: Option[Int] = None) extends UnaryExecNode with GpuExec { require(matchedConditions.length == matchedOutputs.length) require(notMatchedConditions.length == notMatchedOutputs.length) - - // TODO add support for notMatchedBy* - // see https://github.com/NVIDIA/spark-rapids/issues/8415 - require(notMatchedBySourceConditions.isEmpty) - require(notMatchedBySourceOutputs.isEmpty) + require(notMatchedBySourceConditions.length == notMatchedBySourceOutputs.length) private lazy val inputTypes: Array[DataType] = GpuColumnVector.extractTypes(child.schema) private lazy val outputExprs: Seq[GpuBoundReference] = output.zipWithIndex.map { @@ -169,6 +172,10 @@ case class GpuRapidsProcessDeltaMergeJoinExec( private lazy val boundMatchedOutputs = matchedOutputs.map(_.map(_.map(bindForGpu))) private lazy val boundNotMatchedConditions = notMatchedConditions.map(bindForGpu) private lazy val boundNotMatchedOutputs = notMatchedOutputs.map(_.map(_.map(bindForGpu))) + private lazy val boundNotMatchedBySourceConditions = + notMatchedBySourceConditions.map(bindForGpu) + private lazy val boundNotMatchedBySourceOutputs = + notMatchedBySourceOutputs.map(_.map(_.map(bindForGpu))) private lazy val boundNoopCopyOutput = noopCopyOutput.map(bindForGpu) private lazy val boundDeleteRowOutput = deleteRowOutput.map(bindForGpu) @@ -195,8 +202,11 @@ case class GpuRapidsProcessDeltaMergeJoinExec( val localMatchedOutputs = boundMatchedOutputs val localNotMatchedConditions = boundNotMatchedConditions val localNotMatchedOutputs = boundNotMatchedOutputs + val localNotMatchedBySourceConditions = boundNotMatchedBySourceConditions + val localNotMatchedBySourceOutputs = boundNotMatchedBySourceOutputs val localNoopCopyOutput = boundNoopCopyOutput val localDeleteRowOutput = boundDeleteRowOutput + val localRowDroppedColumnIndex = rowDroppedColumnIndex child.executeColumnar().mapPartitions { iter => new GpuRapidsProcessDeltaMergeJoinIterator( iter = iter, @@ -209,9 +219,12 @@ case class GpuRapidsProcessDeltaMergeJoinExec( matchedOutputs = localMatchedOutputs, notMatchedConditions = localNotMatchedConditions, notMatchedOutputs = localNotMatchedOutputs, + notMatchedBySourceConditions = localNotMatchedBySourceConditions, + notMatchedBySourceOutputs = localNotMatchedBySourceOutputs, noopCopyOutput = localNoopCopyOutput, deleteRowOutput = localDeleteRowOutput, - allMetrics) + metrics = allMetrics, + rowDroppedColumnIndex = localRowDroppedColumnIndex) } } @@ -231,9 +244,12 @@ class GpuRapidsProcessDeltaMergeJoinIterator( matchedOutputs: Seq[Seq[Seq[GpuExpression]]], notMatchedConditions: Seq[GpuExpression], notMatchedOutputs: Seq[Seq[Seq[GpuExpression]]], + notMatchedBySourceConditions: Seq[GpuExpression], + notMatchedBySourceOutputs: Seq[Seq[Seq[GpuExpression]]], noopCopyOutput: Seq[GpuExpression], deleteRowOutput: Seq[GpuExpression], - metrics: Map[String, GpuMetric]) + metrics: Map[String, GpuMetric], + rowDroppedColumnIndex: Option[Int] = None) extends Iterator[ColumnarBatch] with AutoCloseable { private[this] val intermediateTypes: Array[DataType] = noopCopyOutput.map(_.dataType).toArray @@ -286,10 +302,13 @@ class GpuRapidsProcessDeltaMergeJoinIterator( private def processSingleBatch(input: ColumnarBatch): ColumnarBatch = { val (targetNoMatchBatch, targetMatchBatch) = splitBatchAndClose(input, inputTypes, targetRowHasNoMatch) - val noopCopyBatch = closeOnExcept(targetMatchBatch) { _ => - GpuProjectExec.projectAndClose(targetNoMatchBatch, noopCopyOutput, NoopMetric) + // Target rows without a source match are handled by the NOT MATCHED BY SOURCE clauses. + // A target row that satisfies none of the clause conditions is copied unchanged. + val targetNotMatchedBatches = closeOnExcept(targetMatchBatch) { _ => + processProjectionSeries(targetNoMatchBatch, + notMatchedBySourceConditions, notMatchedBySourceOutputs, noopCopyOutput) } - val bigTable = withResource(noopCopyBatch) { _ => + val bigTable = withResource(targetNotMatchedBatches) { _ => val (sourceNoMatchBatch, sourceMatchBatch) = splitBatchAndClose(targetMatchBatch, inputTypes, sourceRowHasNoMatch) val sourceNotMatchedBatches = closeOnExcept(sourceMatchBatch) { _ => @@ -300,9 +319,15 @@ class GpuRapidsProcessDeltaMergeJoinIterator( val sourceMatchedBatches = processProjectionSeries(sourceMatchBatch, matchedConditions, matchedOutputs, noopCopyOutput) withResource(sourceMatchedBatches) { _ => - val allBatches = (noopCopyBatch +: sourceNotMatchedBatches) ++ sourceMatchedBatches - // annoyingly Table.concatenate does not gracefully handle the degenerate case - if (allBatches.size == 1) { + val allBatches = targetNotMatchedBatches ++ sourceNotMatchedBatches ++ + sourceMatchedBatches + // annoyingly Table.concatenate does not gracefully handle the degenerate cases + if (allBatches.isEmpty) { + // every projection series skips empty inputs, so an empty input batch ends up here + withResource(GpuColumnVector.emptyBatchFromTypes(intermediateTypes)) { emptyBatch => + GpuColumnVector.from(emptyBatch) + } + } else if (allBatches.size == 1) { GpuColumnVector.from(allBatches.head) } else { withResource(allBatches.safeMap(GpuColumnVector.from)) { allTables => @@ -313,11 +338,13 @@ class GpuRapidsProcessDeltaMergeJoinIterator( } } val shouldNotDeleteBatch = withResource(bigTable) { _ => - // If ROW_DROPPED_COL is not in output schema - // then CDC must be disabled and it's the column after our output cols - val shouldDeleteColumnIndex = + // The command that built the plan knows where the control column sits. Without that, if + // ROW_DROPPED_COL is not in the output schema then CDC must be disabled and it's the column + // after our output cols. + val shouldDeleteColumnIndex = rowDroppedColumnIndex.getOrElse { output.zipWithIndex.find(_._1.name == GpuDeltaMergeConstants.ROW_DROPPED_COL).map(_._2) .getOrElse(output.size) + } val shouldDeleteColumn = bigTable.getColumn(shouldDeleteColumnIndex) withResource(shouldDeleteColumn.not()) { notDeleteColumn => withResource(bigTable.filter(notDeleteColumn)) { notDeleteTable => @@ -336,11 +363,13 @@ class GpuRapidsProcessDeltaMergeJoinIterator( closeOnExcept(new ArrayBuffer[ColumnarBatch]) { results => var leftOverBatch = input conditions.zip(outputs).foreach { case (condition, output) => - closeOnExcept(leftOverBatch) { _ => - if (leftOverBatch.numRows() > 0) { - val (matchBatch, notMatchBatch) = - splitBatchAndClose(leftOverBatch, inputTypes, condition) - leftOverBatch = notMatchBatch + if (leftOverBatch.numRows() > 0) { + // splitBatchAndClose closes the batch it is given, so only the not-matched remainder + // is still open if a projection below throws + val (matchBatch, notMatchBatch) = + splitBatchAndClose(leftOverBatch, inputTypes, condition) + leftOverBatch = notMatchBatch + closeOnExcept(notMatchBatch) { _ => withResource(matchBatch) { _ => output.foreach { exprs => results.append(GpuProjectExec.project(matchBatch, exprs)) @@ -364,15 +393,20 @@ class GpuRapidsProcessDeltaMergeJoinIterator( predicate: Expression): (ColumnarBatch, ColumnarBatch) = { withResource(input) { _ => withResource(GpuColumnVector.from(input)) { inTable => - val predCol = predicate.columnarEval(input) + // A clause condition that evaluates to NULL is false in SQL: the row moves on to the + // next clause or to the default output. cuDF's filter drops rows whose mask is NULL, and + // NOT NULL is NULL, so without this both halves would lose the row. + val predCol = withResource(predicate.columnarEval(input)) { evaluated => + nullsAsFalse(evaluated.getBase) + } val matchedBatch = closeOnExcept(predCol) { _ => - withResource(inTable.filter(predCol.getBase)) { matchedTable => + withResource(inTable.filter(predCol)) { matchedTable => GpuColumnVector.from(matchedTable, inputTypes) } } closeOnExcept(matchedBatch) { _ => val notPredCol = withResource(predCol) { _ => - predCol.getBase.not() + predCol.not() } val notMatchedBatch = withResource(notPredCol) { _ => withResource(inTable.filter(notPredCol)) { notMatchedTable => @@ -384,4 +418,14 @@ class GpuRapidsProcessDeltaMergeJoinIterator( } } } + + private def nullsAsFalse(mask: ColumnVector): ColumnVector = { + if (mask.hasNulls) { + withResource(Scalar.fromBool(false)) { falseScalar => + mask.replaceNulls(falseScalar) + } + } else { + mask.incRefCount() + } + } } diff --git a/delta-lake/delta-spark400db173/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuMergeIntoCommand.scala b/delta-lake/delta-spark400db173/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuMergeIntoCommand.scala index cf468603b41..7953bee7a00 100644 --- a/delta-lake/delta-spark400db173/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuMergeIntoCommand.scala +++ b/delta-lake/delta-spark400db173/src/main/scala/com/databricks/sql/transaction/tahoe/rapids/GpuMergeIntoCommand.scala @@ -28,7 +28,6 @@ import scala.collection.JavaConverters._ import scala.collection.mutable import com.databricks.sql.transaction.tahoe._ -import com.databricks.sql.transaction.tahoe.DeltaOperations.MergePredicate import com.databricks.sql.transaction.tahoe.actions.{AddCDCFile, AddFile, FileAction} import com.databricks.sql.transaction.tahoe.commands.DeltaCommand import com.databricks.sql.transaction.tahoe.commands.merge.MergeIntoMaterializeSource @@ -39,6 +38,7 @@ import com.databricks.sql.transaction.tahoe.util.{AnalysisHelper, SetAccumulator import com.fasterxml.jackson.databind.annotation.JsonDeserialize import com.nvidia.spark.rapids.{BaseExprMeta, GpuOverrides, RapidsConf} import com.nvidia.spark.rapids.delta._ +import com.nvidia.spark.rapids.delta.shims.UpdateCommandShims import org.apache.spark.SparkContext import org.apache.spark.sql._ @@ -46,18 +46,19 @@ import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute import org.apache.spark.sql.catalyst.catalog.CatalogTable import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder} -import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, BasePredicate, Expression, IsNull, Literal, NamedExpression, PredicateHelper, UnsafeProjection} +import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute, AttributeReference, BasePredicate, EqualNullSafe, Expression, If, IsNull, Literal, NamedExpression, Not, Or, PredicateHelper, UnsafeProjection} import org.apache.spark.sql.catalyst.expressions.codegen.GeneratePredicate -import org.apache.spark.sql.catalyst.plans.logical.{DeltaMergeIntoClause, DeltaMergeIntoMatchedClause, DeltaMergeIntoMatchedDeleteClause, DeltaMergeIntoMatchedUpdateClause, DeltaMergeIntoNotMatchedBySourceClause, DeltaMergeIntoNotMatchedClause, LogicalPlan, Project} +import org.apache.spark.sql.catalyst.plans.logical.{DeltaMergeIntoClause, DeltaMergeIntoMatchedClause, DeltaMergeIntoMatchedDeleteClause, DeltaMergeIntoMatchedUpdateClause, DeltaMergeIntoNotMatchedBySourceClause, DeltaMergeIntoNotMatchedBySourceDeleteClause, DeltaMergeIntoNotMatchedBySourceUpdateClause, DeltaMergeIntoNotMatchedClause, LogicalPlan, Project} import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap import org.apache.spark.sql.execution.SQLExecution import org.apache.spark.sql.execution.command.LeafRunnableCommand import org.apache.spark.sql.execution.datasources.LogicalRelation import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} +import org.apache.spark.sql.expressions.Window import org.apache.spark.sql.functions._ import org.apache.spark.sql.nvidia.DFUDFShims -import org.apache.spark.sql.types.{DataTypes, LongType, StringType, StructType} +import org.apache.spark.sql.types.{DataTypes, LongType, StringType, StructField, StructType} case class GpuMergeDataSizes( @JsonDeserialize(contentAs = classOf[java.lang.Long]) @@ -101,9 +102,10 @@ case class GpuMergeStats( insertExprs: Seq[String], deleteConditionExpr: String, - // Newer expressions used in MERGE with any number of MATCHED/NOT MATCHED + // Newer expressions used in MERGE with any number of MATCHED/NOT MATCHED/NOT MATCHED BY SOURCE matchedStats: Seq[GpuMergeClauseStats], notMatchedStats: Seq[GpuMergeClauseStats], + notMatchedBySourceStats: Seq[GpuMergeClauseStats], // Data sizes of source and target at different stages of processing source: GpuMergeDataSizes, @@ -129,8 +131,12 @@ case class GpuMergeStats( targetPartitionsAddedTo: Option[Long], targetRowsCopied: Long, targetRowsUpdated: Long, + targetRowsMatchedUpdated: Long, + targetRowsNotMatchedBySourceUpdated: Long, targetRowsInserted: Long, - targetRowsDeleted: Long + targetRowsDeleted: Long, + targetRowsMatchedDeleted: Long, + targetRowsNotMatchedBySourceDeleted: Long ) object GpuMergeStats { @@ -140,6 +146,7 @@ object GpuMergeStats { condition: Expression, matchedClauses: Seq[DeltaMergeIntoMatchedClause], notMatchedClauses: Seq[DeltaMergeIntoNotMatchedClause], + notMatchedBySourceClauses: Seq[DeltaMergeIntoNotMatchedBySourceClause], isPartitioned: Boolean): GpuMergeStats = { def metricValueIfPartitioned(metricName: String): Option[Long] = { @@ -150,9 +157,11 @@ object GpuMergeStats { // Merge condition expression conditionExpr = condition.sql, - // Newer expressions used in MERGE with any number of MATCHED/NOT MATCHED + // Newer expressions used in MERGE with any number of MATCHED/NOT MATCHED/ + // NOT MATCHED BY SOURCE matchedStats = matchedClauses.map(GpuMergeClauseStats(_)), notMatchedStats = notMatchedClauses.map(GpuMergeClauseStats(_)), + notMatchedBySourceStats = notMatchedBySourceClauses.map(GpuMergeClauseStats(_)), // Data sizes of source and target at different stages of processing source = GpuMergeDataSizes(rows = Some(metrics("numSourceRows").value)), @@ -179,8 +188,12 @@ object GpuMergeStats { targetPartitionsAddedTo = metricValueIfPartitioned("numTargetPartitionsAddedTo"), targetRowsCopied = metrics("numTargetRowsCopied").value, targetRowsUpdated = metrics("numTargetRowsUpdated").value, + targetRowsMatchedUpdated = metrics("numTargetRowsMatchedUpdated").value, + targetRowsNotMatchedBySourceUpdated = metrics("numTargetRowsNotMatchedBySourceUpdated").value, targetRowsInserted = metrics("numTargetRowsInserted").value, targetRowsDeleted = metrics("numTargetRowsDeleted").value, + targetRowsMatchedDeleted = metrics("numTargetRowsMatchedDeleted").value, + targetRowsNotMatchedBySourceDeleted = metrics("numTargetRowsNotMatchedBySourceDeleted").value, // Deprecated fields updateConditionExpr = null, @@ -276,14 +289,45 @@ case class GpuMergeIntoCommand( } /** Whether this merge statement has only a single insert (NOT MATCHED) clause. */ - private def isSingleInsertOnly: Boolean = matchedClauses.isEmpty && notMatchedClauses.length == 1 - /** Whether this merge statement has only MATCHED clauses. */ - private def isMatchedOnly: Boolean = notMatchedClauses.isEmpty && matchedClauses.nonEmpty + private def isSingleInsertOnly: Boolean = + matchedClauses.isEmpty && notMatchedBySourceClauses.isEmpty && notMatchedClauses.length == 1 + /** Whether this merge statement has no insert (NOT MATCHED) clause. */ + private def hasNoInserts: Boolean = notMatchedClauses.isEmpty // We over-count numTargetRowsDeleted when there are multiple matches; // this is the amount of the overcount, so we can subtract it to get a correct final metric. private var multipleMatchDeleteOnlyOvercount: Option[Long] = None + /** + * Whether a joined source/target row pair takes a WHEN MATCHED action. Databricks Runtime 16.0 + * and later report multiple matches only for source rows that satisfy the ON condition and at + * least one WHEN MATCHED clause condition (an undefined condition is implicitly true); source + * rows that match on ON alone take no action and do not make the target row ambiguous. + */ + private lazy val effectiveMatchPredicate: Expression = + if (matchedClauses.isEmpty) { + Literal.FalseLiteral + } else { + matchedClauses.map(_.condition.getOrElse(Literal.TrueLiteral)).reduce((a, b) => Or(a, b)) + } + + // Set by findTouchedFiles when some target rows match several source rows on the ON condition + // of which at most one takes a WHEN MATCHED action; writeAllChanges then keeps one pair per row. + private var hasNonEffectiveDuplicateMatches: Boolean = false + + /** + * A helper column name that none of `existing` resolves to under the session's resolver. + * `Dataset.withColumn` replaces an existing column of the same name, so a helper attached + * under a fixed name would replace a user column that the clause expressions still reference. + */ + private def uniqueColumnName(base: String, existing: Seq[String]): String = { + val resolver = conf.resolver + Iterator.from(0) + .map(i => if (i == 0) base else s"$base$i") + .find(candidate => !existing.exists(name => resolver(name, candidate))) + .get + } + private def checkIdentityColumnHighWaterMarks(deltaTxn: OptimisticTransaction): Unit = { // DBR implements this in MergeIntoCommandBase, but that method is protected and the trait // also requires the full CPU MERGE runMerge contract. Keep this local copy aligned with the @@ -316,7 +360,15 @@ case class GpuMergeIntoCommand( "numTargetRowsCopied" -> createMetric(sc, "number of target rows rewritten unmodified"), "numTargetRowsInserted" -> createMetric(sc, "number of inserted rows"), "numTargetRowsUpdated" -> createMetric(sc, "number of updated rows"), + "numTargetRowsMatchedUpdated" -> + createMetric(sc, "number of rows updated by a matched clause"), + "numTargetRowsNotMatchedBySourceUpdated" -> + createMetric(sc, "number of rows updated by a not matched by source clause"), "numTargetRowsDeleted" -> createMetric(sc, "number of deleted rows"), + "numTargetRowsMatchedDeleted" -> + createMetric(sc, "number of rows deleted by a matched clause"), + "numTargetRowsNotMatchedBySourceDeleted" -> + createMetric(sc, "number of rows deleted by a not matched by source clause"), "numTargetFilesBeforeSkipping" -> createMetric(sc, "number of target files before skipping"), "numTargetFilesAfterSkipping" -> createMetric(sc, "number of target files after skipping"), "numTargetFilesRemoved" -> createMetric(sc, "number of files removed to target"), @@ -414,15 +466,14 @@ case class GpuMergeIntoCommand( Option(condition), matchedClauses.map(DeltaOperations.MergePredicate(_)), notMatchedClauses.map(DeltaOperations.MergePredicate(_)), - // We do not support notMatchedBySourcePredicates yet and fall back to CPU - // See https://github.com/NVIDIA/spark-rapids/issues/8415 - notMatchedBySourcePredicates = Seq.empty[MergePredicate] + notMatchedBySourcePredicates = + notMatchedBySourceClauses.map(DeltaOperations.MergePredicate(_)) ), RowTracking.addPreservedRowTrackingTagIfNotSet(deltaTxn.snapshot)) // Record metrics val stats = GpuMergeStats.fromMergeSQLMetrics( - metrics, condition, matchedClauses, notMatchedClauses, + metrics, condition, matchedClauses, notMatchedClauses, notMatchedBySourceClauses, deltaTxn.metadata.partitionColumns.nonEmpty) recordDeltaEvent(targetDeltaLog, "delta.dml.merge.stats", data = stats) @@ -450,7 +501,8 @@ case class GpuMergeIntoCommand( /** * Find the target table files that contain the rows that satisfy the merge condition. This is * implemented as an inner-join between the source query/table and the target table using - * the merge condition. + * the merge condition. When there are NOT MATCHED BY SOURCE clauses a right outer join is used + * instead so that target rows without a source match are also collected. */ private def findTouchedFiles( spark: SparkSession, @@ -465,44 +517,68 @@ case class GpuMergeIntoCommand( val recordTouchedFileName = udf(new GpuDeltaRecordTouchedFileNameUDF(touchedFilesAccum)) .asNondeterministic() - // Skip data based on the merge condition - val targetOnlyPredicates = - splitConjunctivePredicates(condition).filter(_.references.subsetOf(target.outputSet)) - val dataSkippedFiles = deltaTxn.filterFiles(targetOnlyPredicates) + // Prune non-matching files if we don't need to collect them for NOT MATCHED BY SOURCE clauses. + val dataSkippedFiles = + if (notMatchedBySourceClauses.isEmpty) { + val targetOnlyPredicates = + splitConjunctivePredicates(condition).filter(_.references.subsetOf(target.outputSet)) + deltaTxn.filterFiles(targetOnlyPredicates) + } else { + deltaTxn.filterFiles(Seq(Literal.TrueLiteral)) + } // UDF to increment metrics val incrSourceRowCountCol = makeMetricUpdateUDF("numSourceRows") val sourceDF = getMergeSource.df .filter(incrSourceRowCountCol) - // Apply inner join to between source and target using the merge condition to find matches + // Join the source and target table using the merge condition to find touched files. An inner + // join collects all candidate files for MATCHED clauses, a right outer join also includes + // candidates for NOT MATCHED BY SOURCE clauses. // In addition, we attach two columns // - a monotonically increasing row id for target rows to later identify whether the same // target row is modified by multiple user or not // - the target file name the row is from to later identify the files touched by matched rows - val targetDF = Dataset.ofRows(spark, buildTargetPlanWithFiles(deltaTxn, dataSkippedFiles)) - .withColumn(ROW_ID_COL, monotonically_increasing_id()) - .withColumn(FILE_NAME_COL, input_file_name()) + val joinType = if (notMatchedBySourceClauses.isEmpty) "inner" else "right_outer" + val targetPlanDF = Dataset.ofRows(spark, buildTargetPlanWithFiles(deltaTxn, dataSkippedFiles)) + // The helper names are chosen to be absent from both sides (see uniqueColumnName), so + // attaching them cannot replace a user column and they resolve unambiguously after the join. + val userColumns = sourceDF.columns.toSeq ++ targetPlanDF.columns.toSeq + val rowIdCol = uniqueColumnName(ROW_ID_COL, userColumns) + val fileNameCol = uniqueColumnName(FILE_NAME_COL, userColumns :+ rowIdCol) + val targetDF = targetPlanDF + .withColumn(rowIdCol, monotonically_increasing_id()) + .withColumn(fileNameCol, input_file_name()) val joinToFindTouchedFiles = - sourceDF.join(targetDF, DFUDFShims.exprToColumn(condition), "inner") + sourceDF.join(targetDF, DFUDFShims.exprToColumn(condition), joinType) - // Process the matches from the inner join to record touched files and find multiple matches + // Process the matches from the join to record touched files and find multiple matches. A pair + // is an effective match when it takes a WHEN MATCHED action; only effective matches can make a + // target row ambiguous (Databricks Runtime 16.0+ semantics, see effectiveMatchPredicate). val collectTouchedFiles = joinToFindTouchedFiles - .select(col(ROW_ID_COL), recordTouchedFileName(col(FILE_NAME_COL)).as("one")) + .select(col(rowIdCol), recordTouchedFileName(col(fileNameCol)).as("one"), + when(DFUDFShims.exprToColumn(effectiveMatchPredicate), lit(1)).otherwise(lit(0)) + .as("effective")) - // Calculate frequency of matches per source row - val matchedRowCounts = collectTouchedFiles.groupBy(ROW_ID_COL).agg(sum("one").as("count")) + // Calculate frequency of matches per target row: all joined pairs and effective pairs + val matchedRowCounts = collectTouchedFiles.groupBy(rowIdCol) + .agg(sum("one").as("count"), sum("effective").as("effectiveCount")) // Get multiple matches and simultaneously collect (using touchedFilesAccum) the file names - // multipleMatchCount = # of target rows with more than 1 matching source row (duplicate match) - // multipleMatchSum = total # of duplicate matched rows + // multipleMatchCount = # of target rows with more than 1 effective matching source row + // multipleMatchSum = total # of effective matched rows of those target rows + // nonEffectiveDuplicates = # of target rows matched by several source rows of which at most + // one takes a WHEN MATCHED action; writeAllChanges keeps one pair for each of them val multipleMatchRow = matchedRowCounts - .filter("count > 1") - .select(coalesce(count("*"), lit(0)), coalesce(sum("count"), lit(0))) + .select( + coalesce(sum(when(col("effectiveCount") > 1, lit(1L))), lit(0L)), + coalesce(sum(when(col("effectiveCount") > 1, col("effectiveCount"))), lit(0L)), + coalesce(sum(when(col("count") > 1 && col("effectiveCount") <= 1, lit(1L))), lit(0L))) .collect() .head val multipleMatchCount = multipleMatchRow.getLong(0) val multipleMatchSum = multipleMatchRow.getLong(1) + hasNonEffectiveDuplicateMatches = multipleMatchRow.getLong(2) > 0 val hasMultipleMatches = multipleMatchCount > 0 @@ -671,7 +747,7 @@ case class GpuMergeIntoCommand( // Generate a new logical plan that has same output attributes exprIds as the target plan. // This allows us to apply the existing resolved update/insert expressions. val newTarget = buildTargetPlanWithFiles(deltaTxn, filesToRewrite) - val joinType = if (isMatchedOnly && + val joinType = if (hasNoInserts && spark.conf.get(DeltaSQLConf.MERGE_MATCHED_ONLY_ENABLED)) { "rightOuter" } else { @@ -691,12 +767,20 @@ case class GpuMergeIntoCommand( // allowed outside a very specific set of Catalyst nodes (Project, Filter, Window, Aggregate). val incrUpdatedCountExpr = metricUpdateExpr("numTargetRowsUpdated", deterministic = true) + val incrUpdatedMatchedCountExpr = + metricUpdateExpr("numTargetRowsMatchedUpdated", deterministic = true) + val incrUpdatedNotMatchedBySourceCountExpr = + metricUpdateExpr("numTargetRowsNotMatchedBySourceUpdated", deterministic = true) val incrInsertedCountExpr = metricUpdateExpr("numTargetRowsInserted", deterministic = true) val incrNoopCountExpr = metricUpdateExpr("numTargetRowsCopied", deterministic = true) val incrDeletedCountExpr = metricUpdateExpr("numTargetRowsDeleted", deterministic = true) + val incrDeletedMatchedCountExpr = + metricUpdateExpr("numTargetRowsMatchedDeleted", deterministic = true) + val incrDeletedNotMatchedBySourceCountExpr = + metricUpdateExpr("numTargetRowsNotMatchedBySourceDeleted", deterministic = true) // Apply an outer join to find both, matches and non-matches. We are adding two boolean fields // with value `true`, one to each side of the join. Whether this field is null or not after @@ -705,23 +789,104 @@ case class GpuMergeIntoCommand( // We add row IDs to the targetDF if we have a delete-when-matched clause with duplicate // matches and CDC is enabled, and additionally add row IDs to the source if we also have an // insert clause. See above at isDeleteWithDuplicateMatchesAndCdc definition for more details. - var sourceDF = getMergeSource.df - .withColumn(SOURCE_ROW_PRESENT_COL, makeMetricUpdateUDF("numSourceRowsInSecondScan")) - var targetDF = Dataset.ofRows(spark, newTarget) - .withColumn(TARGET_ROW_PRESENT_COL, lit(true)) + // Every helper column is attached under a name absent from both sides (see uniqueColumnName), + // so it cannot replace a user column that the clause expressions still reference and it + // resolves unambiguously on the joined plan. + val sourcePlanDF = getMergeSource.df + // Row tracking: the target's materialized row id and commit version columns ride along with + // the target output columns, so rewritten rows keep their ids as they do on the CPU. Copied + // rows keep both, updated rows keep the id and get a null commit version (the writer assigns + // the new one), inserted rows get both null. Without row tracking the helper adds nothing. + val (targetPlanDF, rowTrackingCols, rowTrackingUpdateExprs) = + UpdateCommandShims.preserveRowTrackingColumns( + Dataset.ofRows(spark, newTarget), deltaTxn.snapshot, Seq.empty, Seq.empty) + targetOutputCols = targetOutputCols ++ rowTrackingCols + // The processors rebuild their output attributes from this schema, and the Delta writer + // recognises the row tracking columns by the field metadata the helper put on them, so the + // metadata has to be carried; the columns are nullable because updates reset the version. + outputRowSchema = rowTrackingCols.foldLeft(outputRowSchema) { (schema, attr) => + schema.add(StructField(attr.name, attr.dataType, nullable = true, attr.metadata)) + } + val rowTrackingInsertExprs: Seq[Expression] = + rowTrackingCols.map(attr => Literal(null, attr.dataType)) + val userColumns = sourcePlanDF.columns.toSeq ++ targetPlanDF.columns.toSeq + val sourceRowPresentCol = uniqueColumnName(SOURCE_ROW_PRESENT_COL, userColumns) + val targetRowPresentCol = + uniqueColumnName(TARGET_ROW_PRESENT_COL, userColumns :+ sourceRowPresentCol) + val taken = userColumns ++ Seq(sourceRowPresentCol, targetRowPresentCol) + val dedupTargetRowIdCol = uniqueColumnName(TARGET_ROW_ID_COL, taken) + val dedupSourceRowIdCol = uniqueColumnName(SOURCE_ROW_ID_COL, taken :+ dedupTargetRowIdCol) + var sourceDF = sourcePlanDF + .withColumn(sourceRowPresentCol, makeMetricUpdateUDF("numSourceRowsInSecondScan")) + var targetDF = targetPlanDF.withColumn(targetRowPresentCol, lit(true)) if (isDeleteWithDuplicateMatchesAndCdc) { - targetDF = targetDF.withColumn(TARGET_ROW_ID_COL, monotonically_increasing_id()) + targetDF = targetDF.withColumn(dedupTargetRowIdCol, monotonically_increasing_id()) if (notMatchedClauses.nonEmpty) { // insert clause - sourceDF = sourceDF.withColumn(SOURCE_ROW_ID_COL, monotonically_increasing_id()) + sourceDF = sourceDF.withColumn(dedupSourceRowIdCol, monotonically_increasing_id()) } + } else if (hasNonEffectiveDuplicateMatches) { + // Row ids on both sides identify the joined pairs of one target row for de-duplication. + targetDF = targetDF.withColumn(dedupTargetRowIdCol, monotonically_increasing_id()) + sourceDF = sourceDF.withColumn(dedupSourceRowIdCol, monotonically_increasing_id()) + } + val rawJoinedDF = sourceDF.join(targetDF, DFUDFShims.exprToColumn(condition), joinType) + val joinedDF = if (hasNonEffectiveDuplicateMatches) { + // Some target rows matched several source rows on the ON condition, of which at most one + // takes a WHEN MATCHED action (findTouchedFiles rejected the ambiguous cases). Keep one + // joined pair per target row, preferring the pair that takes an action: the dropped pairs + // hold source rows that are matched (so they must not be inserted) and take no action, and + // when no pair takes an action the surviving one copies the target row unchanged. Source-only + // and target-only rows form partitions of their own and pass through. + val effective = + when(DFUDFShims.exprToColumn(effectiveMatchPredicate), lit(1)).otherwise(lit(0)) + val rankCol = uniqueColumnName(DUPLICATE_MATCH_RANK_COL, rawJoinedDF.columns.toSeq) + val onePairPerTargetRow = Window + .partitionBy(col(dedupTargetRowIdCol), + when(col(targetRowPresentCol).isNull, col(dedupSourceRowIdCol))) + .orderBy(effective.desc) + rawJoinedDF + .withColumn(rankCol, row_number().over(onePairPerTargetRow)) + .filter(col(rankCol) === lit(1)) + .drop(rankCol, dedupTargetRowIdCol, dedupSourceRowIdCol) + } else { + rawJoinedDF } - val joinedDF = sourceDF.join(targetDF, DFUDFShims.exprToColumn(condition), joinType) val joinedPlan = joinedDF.queryExecution.analyzed def resolveOnJoinedPlan(exprs: Seq[Expression]): Seq[Expression] = { tryResolveReferencesForExpressions(spark, exprs, joinedPlan) } + // Databricks 17.0 and later accept non-deterministic expressions in the values of update and + // insert actions (not in clause conditions, which the analysis rejects). Catalyst allows + // such expressions in a fixed set of operators and the processor node is not one of them, + // so each one is evaluated in a projection over the joined rows, under a generated name, and + // the clause outputs reference the projected column (the written row and its CDC post-image + // share it). The value is guarded by the rows the clause acts on, because the CPU command + // evaluates an action value only on those rows: a guarded division under ANSI mode must not + // fail on a joined pair that takes no action. The outputs bind against the projection's + // output; resolving them on the joined plan passes the projected attribute through. + val materializedValues = mutable.ArrayBuffer[NamedExpression]() + def materializeNonDeterministic( + exprs: Seq[Expression], + takesClause: Expression): Seq[Expression] = exprs.map { + case e if !e.deterministic => + val resolved = resolveOnJoinedPlan(Seq(e)).head + val existing = joinedPlan.output.map(_.name) ++ materializedValues.map(_.name) + val alias = Alias(If(takesClause, resolved, Literal(null, resolved.dataType)), + uniqueColumnName(NON_DETERMINISTIC_VALUE_COL, existing))() + materializedValues += alias + alias.toAttribute + case e => e + } + + // The rows a clause acts on: the row kind, none of the earlier clauses of that kind taken, + // and its own condition true. A NULL condition counts as false, as in the processor. + def clauseRouting(rowKind: Expression, conditions: Seq[Expression], index: Int): Expression = { + val taken = conditions.take(index).map(c => Not(EqualNullSafe(c, TrueLiteral))) + (rowKind +: taken :+ EqualNullSafe(conditions(index), TrueLiteral)).reduce(And) + } + // ==== Generate the expressions to process full-outer join output and generate target rows ==== // If there are N columns in the target table, there will be N + 3 columns after processing // - N columns for target table @@ -738,7 +903,7 @@ case class GpuMergeIntoCommand( // and rows for the CDC data which will be output to CDCReader.CDC_LOCATION. // See [[CDCReader]] for general details on how partitioning on the CDC type column works. - // In the following two functions `matchedClauseOutput` and `notMatchedClauseOutput`, we + // In the following functions `updateOutput`, `deleteOutput` and `insertOutput`, we // produce a Seq[Expression] for each intended output row. // Depending on the clause and whether CDC is enabled, we output between 0 and 3 rows, as a // Seq[Seq[Expression]] @@ -753,14 +918,22 @@ case class GpuMergeIntoCommand( // These ROW_ID_COL will always be dropped before the final write. if (isDeleteWithDuplicateMatchesAndCdc) { - targetOutputCols = targetOutputCols :+ UnresolvedAttribute(TARGET_ROW_ID_COL) - outputRowSchema = outputRowSchema.add(TARGET_ROW_ID_COL, DataTypes.LongType) + targetOutputCols = targetOutputCols :+ UnresolvedAttribute(dedupTargetRowIdCol) + outputRowSchema = outputRowSchema.add(dedupTargetRowIdCol, DataTypes.LongType) if (notMatchedClauses.nonEmpty) { // there is an insert clause, make SRC_ROW_ID_COL=null - targetOutputCols = targetOutputCols :+ Alias(Literal(null), SOURCE_ROW_ID_COL)() - outputRowSchema = outputRowSchema.add(SOURCE_ROW_ID_COL, DataTypes.LongType) + // Typed: the GPU processor concatenates this output with the insert output, which + // carries the source row id as a long; an untyped null literal makes the types differ. + targetOutputCols = targetOutputCols :+ Alias(Literal(null, LongType), dedupSourceRowIdCol)() + outputRowSchema = outputRowSchema.add(dedupSourceRowIdCol, DataTypes.LongType) } } + // ROW_DROPPED_COL is the first control column every clause output appends after the target + // output columns, whether or not it is part of outputRowSchema (it is only with CDC). The + // processors read it by this position rather than by name, so a user column that happens to + // carry the name cannot be mistaken for it. + val rowDroppedColumnIndex = targetOutputCols.size + if (cdcEnabled) { outputRowSchema = outputRowSchema .add(ROW_DROPPED_COL, DataTypes.BooleanType) @@ -768,51 +941,54 @@ case class GpuMergeIntoCommand( .add(CDC_TYPE_COLUMN_NAME, DataTypes.StringType) } - def matchedClauseOutput(clause: DeltaMergeIntoMatchedClause): Seq[Seq[Expression]] = { - val exprs = clause match { - case u: DeltaMergeIntoMatchedUpdateClause => - // Generate update expressions and set ROW_DELETED_COL = false and - // CDC_TYPE_COLUMN_NAME = CDC_TYPE_NOT_CDC - val mainDataOutput = u.resolvedActions.map(_.expr) :+ FalseLiteral :+ - incrUpdatedCountExpr :+ CDC_TYPE_NOT_CDC_LITERAL - if (cdcEnabled) { - // For update preimage, we have do a no-op copy with ROW_DELETED_COL = false and - // CDC_TYPE_COLUMN_NAME = CDC_TYPE_UPDATE_PREIMAGE and INCR_ROW_COUNT_COL as a no-op - // (because the metric will be incremented in `mainDataOutput`) - val preImageOutput = targetOutputCols :+ FalseLiteral :+ TrueLiteral :+ - Literal(CDC_TYPE_UPDATE_PREIMAGE) - // For update postimage, we have the same expressions as for mainDataOutput but with - // INCR_ROW_COUNT_COL as a no-op (because the metric will be incremented in - // `mainDataOutput`), and CDC_TYPE_COLUMN_NAME = CDC_TYPE_UPDATE_POSTIMAGE - val postImageOutput = mainDataOutput.dropRight(2) :+ TrueLiteral :+ - Literal(CDC_TYPE_UPDATE_POSTIMAGE) - Seq(mainDataOutput, preImageOutput, postImageOutput) - } else { - Seq(mainDataOutput) - } - case _: DeltaMergeIntoMatchedDeleteClause => - // Generate expressions to set the ROW_DELETED_COL = true and CDC_TYPE_COLUMN_NAME = - // CDC_TYPE_NOT_CDC - val mainDataOutput = targetOutputCols :+ TrueLiteral :+ incrDeletedCountExpr :+ - CDC_TYPE_NOT_CDC_LITERAL - if (cdcEnabled) { - // For delete we do a no-op copy with ROW_DELETED_COL = false, INCR_ROW_COUNT_COL as a - // no-op (because the metric will be incremented in `mainDataOutput`) and - // CDC_TYPE_COLUMN_NAME = CDC_TYPE_DELETE - val deleteCdcOutput = targetOutputCols :+ FalseLiteral :+ TrueLiteral :+ - Literal(CDC_TYPE_DELETE) - Seq(mainDataOutput, deleteCdcOutput) - } else { - Seq(mainDataOutput) - } + def updateOutput( + updateExprs: Seq[Expression], + incrMetricExpr: Expression): Seq[Seq[Expression]] = { + // Generate update expressions and set ROW_DELETED_COL = false and + // CDC_TYPE_COLUMN_NAME = CDC_TYPE_NOT_CDC + val mainDataOutput = updateExprs :+ FalseLiteral :+ incrMetricExpr :+ + CDC_TYPE_NOT_CDC_LITERAL + val exprs = if (cdcEnabled) { + // For update preimage, we have do a no-op copy with ROW_DELETED_COL = false and + // CDC_TYPE_COLUMN_NAME = CDC_TYPE_UPDATE_PREIMAGE and INCR_ROW_COUNT_COL as a no-op + // (because the metric will be incremented in `mainDataOutput`) + val preImageOutput = targetOutputCols :+ FalseLiteral :+ TrueLiteral :+ + Literal(CDC_TYPE_UPDATE_PREIMAGE) + // For update postimage, we have the same expressions as for mainDataOutput but with + // INCR_ROW_COUNT_COL as a no-op (because the metric will be incremented in + // `mainDataOutput`), and CDC_TYPE_COLUMN_NAME = CDC_TYPE_UPDATE_POSTIMAGE + val postImageOutput = mainDataOutput.dropRight(2) :+ TrueLiteral :+ + Literal(CDC_TYPE_UPDATE_POSTIMAGE) + Seq(mainDataOutput, preImageOutput, postImageOutput) + } else { + Seq(mainDataOutput) } exprs.map(resolveOnJoinedPlan) } - def notMatchedClauseOutput(clause: DeltaMergeIntoNotMatchedClause): Seq[Seq[Expression]] = { + def deleteOutput(incrMetricExpr: Expression): Seq[Seq[Expression]] = { + // Generate expressions to set the ROW_DELETED_COL = true and CDC_TYPE_COLUMN_NAME = + // CDC_TYPE_NOT_CDC + val mainDataOutput = targetOutputCols :+ TrueLiteral :+ incrMetricExpr :+ + CDC_TYPE_NOT_CDC_LITERAL + val exprs = if (cdcEnabled) { + // For delete we do a no-op copy with ROW_DELETED_COL = false, INCR_ROW_COUNT_COL as a + // no-op (because the metric will be incremented in `mainDataOutput`) and + // CDC_TYPE_COLUMN_NAME = CDC_TYPE_DELETE + val deleteCdcOutput = targetOutputCols :+ FalseLiteral :+ TrueLiteral :+ + Literal(CDC_TYPE_DELETE) + Seq(mainDataOutput, deleteCdcOutput) + } else { + Seq(mainDataOutput) + } + exprs.map(resolveOnJoinedPlan) + } + + def insertOutput( + insertExprs: Seq[Expression], + incrMetricExpr: Expression): Seq[Seq[Expression]] = { // Generate insert expressions and set ROW_DELETED_COL = false and // CDC_TYPE_COLUMN_NAME = CDC_TYPE_NOT_CDC - val insertExprs = clause.resolvedActions.map(_.expr) val mainDataOutput = resolveOnJoinedPlan( if (isDeleteWithDuplicateMatchesAndCdc) { // Must be delete-when-matched merge with duplicate matches + insert clause @@ -820,10 +996,11 @@ case class GpuMergeIntoCommand( // clause we know the target row-id will be null. See above at // isDeleteWithDuplicateMatchesAndCdc definition for more details. insertExprs :+ - Alias(Literal(null), TARGET_ROW_ID_COL)() :+ UnresolvedAttribute(SOURCE_ROW_ID_COL) :+ - FalseLiteral :+ incrInsertedCountExpr :+ CDC_TYPE_NOT_CDC_LITERAL + Alias(Literal(null, LongType), dedupTargetRowIdCol)() :+ + UnresolvedAttribute(dedupSourceRowIdCol) :+ + FalseLiteral :+ incrMetricExpr :+ CDC_TYPE_NOT_CDC_LITERAL } else { - insertExprs :+ FalseLiteral :+ incrInsertedCountExpr :+ CDC_TYPE_NOT_CDC_LITERAL + insertExprs :+ FalseLiteral :+ incrMetricExpr :+ CDC_TYPE_NOT_CDC_LITERAL } ) if (cdcEnabled) { @@ -837,6 +1014,28 @@ case class GpuMergeIntoCommand( } } + def clauseOutput(clause: DeltaMergeIntoClause, routing: Expression): Seq[Seq[Expression]] = + clause match { + case u: DeltaMergeIntoMatchedUpdateClause => + updateOutput(materializeNonDeterministic(u.resolvedActions.map(_.expr), routing) ++ + rowTrackingUpdateExprs, + And(incrUpdatedCountExpr, incrUpdatedMatchedCountExpr)) + case _: DeltaMergeIntoMatchedDeleteClause => + deleteOutput(And(incrDeletedCountExpr, incrDeletedMatchedCountExpr)) + case i: DeltaMergeIntoNotMatchedClause => + insertOutput(materializeNonDeterministic(i.resolvedActions.map(_.expr), routing) ++ + rowTrackingInsertExprs, + incrInsertedCountExpr) + case u: DeltaMergeIntoNotMatchedBySourceUpdateClause => + updateOutput(materializeNonDeterministic(u.resolvedActions.map(_.expr), routing) ++ + rowTrackingUpdateExprs, + And(incrUpdatedCountExpr, incrUpdatedNotMatchedBySourceCountExpr)) + case _: DeltaMergeIntoNotMatchedBySourceDeleteClause => + deleteOutput(And(incrDeletedCountExpr, incrDeletedNotMatchedBySourceCountExpr)) + case other => + throw new IllegalArgumentException(s"Unsupported merge clause: ${other.getClass.getName}") + } + def clauseCondition(clause: DeltaMergeIntoClause): Expression = { // if condition is None, then expression always evaluates to true val condExpr = clause.condition.getOrElse(TrueLiteral) @@ -844,24 +1043,35 @@ case class GpuMergeIntoCommand( } val targetRowHasNoMatch = - resolveOnJoinedPlan(Seq(IsNull(UnresolvedAttribute(SOURCE_ROW_PRESENT_COL)))).head + resolveOnJoinedPlan(Seq(IsNull(UnresolvedAttribute(sourceRowPresentCol)))).head val sourceRowHasNoMatch = - resolveOnJoinedPlan(Seq(IsNull(UnresolvedAttribute(TARGET_ROW_PRESENT_COL)))).head + resolveOnJoinedPlan(Seq(IsNull(UnresolvedAttribute(targetRowPresentCol)))).head + val matchedRow = And(Not(targetRowHasNoMatch), Not(sourceRowHasNoMatch)) val matchedConditions = matchedClauses.map(clauseCondition) - val matchedOutputs = matchedClauses.map(matchedClauseOutput) + val matchedOutputs = matchedClauses.zipWithIndex.map { case (clause, i) => + clauseOutput(clause, clauseRouting(matchedRow, matchedConditions, i)) + } val notMatchedConditions = notMatchedClauses.map(clauseCondition) - val notMatchedOutputs = notMatchedClauses.map(notMatchedClauseOutput) - // TODO support notMatchedBySourceClauses which is new in DBR 12.2 - // https://github.com/NVIDIA/spark-rapids/issues/8415 - val notMatchedBySourceConditions = Seq.empty - val notMatchedBySourceOutputs = Seq.empty + val notMatchedOutputs = notMatchedClauses.zipWithIndex.map { case (clause, i) => + clauseOutput(clause, clauseRouting(sourceRowHasNoMatch, notMatchedConditions, i)) + } + val notMatchedBySourceConditions = notMatchedBySourceClauses.map(clauseCondition) + val notMatchedBySourceOutputs = notMatchedBySourceClauses.zipWithIndex.map { + case (clause, i) => + clauseOutput(clause, clauseRouting(targetRowHasNoMatch, notMatchedBySourceConditions, i)) + } val noopCopyOutput = resolveOnJoinedPlan(targetOutputCols :+ FalseLiteral :+ incrNoopCountExpr :+ CDC_TYPE_NOT_CDC_LITERAL) val deleteRowOutput = resolveOnJoinedPlan(targetOutputCols :+ TrueLiteral :+ TrueLiteral :+ CDC_TYPE_NOT_CDC_LITERAL) - var outputDF = addMergeJoinProcessor(spark, joinedPlan, outputRowSchema, + val processorInputPlan = if (materializedValues.isEmpty) { + joinedPlan + } else { + Project(joinedPlan.output ++ materializedValues, joinedPlan) + } + var outputDF = addMergeJoinProcessor(spark, processorInputPlan, outputRowSchema, targetRowHasNoMatch = targetRowHasNoMatch, sourceRowHasNoMatch = sourceRowHasNoMatch, matchedConditions = matchedConditions, @@ -871,7 +1081,20 @@ case class GpuMergeIntoCommand( notMatchedBySourceConditions = notMatchedBySourceConditions, notMatchedBySourceOutputs = notMatchedBySourceOutputs, noopCopyOutput = noopCopyOutput, - deleteRowOutput = deleteRowOutput) + deleteRowOutput = deleteRowOutput, + rowDroppedColumnIndex = rowDroppedColumnIndex) + + // The two control columns are part of the output only with CDC. They are the last attributes + // carrying those names, so dropping them by attribute leaves a user column of the same name in + // place, where dropping by name would remove that column as well. + def dropControlColumns(df: DataFrame): DataFrame = if (!cdcEnabled) { + df + } else { + val output = df.queryExecution.analyzed.output + Seq(ROW_DROPPED_COL, INCR_ROW_COUNT_COL) + .flatMap(name => output.reverse.find(_.name == name)) + .foldLeft(df)((d, attr) => d.drop(DFUDFShims.exprToColumn(attr))) + } if (isDeleteWithDuplicateMatchesAndCdc) { // When we have a delete when matched clause with duplicate matches we have to remove @@ -884,15 +1107,14 @@ case class GpuMergeIntoCommand( // SOURCE_ROW_ID_COL and CDC_TYPE_COLUMN_NAME to avoid dropping valid duplicate inserted rows // and their corresponding CDC rows. val columnsToDedupeBy = if (notMatchedClauses.nonEmpty) { // insert clause - Seq(TARGET_ROW_ID_COL, SOURCE_ROW_ID_COL, CDC_TYPE_COLUMN_NAME) + Seq(dedupTargetRowIdCol, dedupSourceRowIdCol, CDC_TYPE_COLUMN_NAME) } else { - Seq(TARGET_ROW_ID_COL) + Seq(dedupTargetRowIdCol) } - outputDF = outputDF - .dropDuplicates(columnsToDedupeBy) - .drop(ROW_DROPPED_COL, INCR_ROW_COUNT_COL, TARGET_ROW_ID_COL, SOURCE_ROW_ID_COL) + outputDF = dropControlColumns(outputDF.dropDuplicates(columnsToDedupeBy)) + .drop(dedupTargetRowIdCol, dedupSourceRowIdCol) } else { - outputDF = outputDF.drop(ROW_DROPPED_COL, INCR_ROW_COUNT_COL) + outputDF = dropControlColumns(outputDF) } logDebug("writeAllChanges: join output plan:\n" + outputDF.queryExecution) @@ -914,6 +1136,10 @@ case class GpuMergeIntoCommand( metrics("numTargetRowsDeleted").value - multipleMatchDeleteOnlyOvercount.get assert(actualRowsDeleted >= 0) metrics("numTargetRowsDeleted").set(actualRowsDeleted) + val actualRowsMatchedDeleted = + metrics("numTargetRowsMatchedDeleted").value - multipleMatchDeleteOnlyOvercount.get + assert(actualRowsMatchedDeleted >= 0) + metrics("numTargetRowsMatchedDeleted").set(actualRowsMatchedDeleted) } newFiles @@ -932,7 +1158,8 @@ case class GpuMergeIntoCommand( notMatchedBySourceConditions: Seq[Expression], notMatchedBySourceOutputs: Seq[Seq[Seq[Expression]]], noopCopyOutput: Seq[Expression], - deleteRowOutput: Seq[Expression]): Dataset[Row] = { + deleteRowOutput: Seq[Expression], + rowDroppedColumnIndex: Int): Dataset[Row] = { def wrap(e: Expression): BaseExprMeta[Expression] = { GpuOverrides.wrapExpr(e, rapidsConf, None) } @@ -979,7 +1206,8 @@ case class GpuMergeIntoCommand( notMatchedBySourceConditions = notMatchedBySourceConditions, notMatchedBySourceOutputs = notMatchedBySourceOutputs, noopCopyOutput = noopCopyOutput, - deleteRowOutput = deleteRowOutput) + deleteRowOutput = deleteRowOutput, + rowDroppedColumnIndex = Some(rowDroppedColumnIndex)) Dataset.ofRows(spark, processedJoinPlan) } else { val joinedRowEncoder = ExpressionEncoder(RowEncoder.encoderFor(joinedPlan.schema)) @@ -993,11 +1221,14 @@ case class GpuMergeIntoCommand( matchedOutputs = matchedOutputs, notMatchedConditions = notMatchedConditions, notMatchedOutputs = notMatchedOutputs, + notMatchedBySourceConditions = notMatchedBySourceConditions, + notMatchedBySourceOutputs = notMatchedBySourceOutputs, noopCopyOutput = noopCopyOutput, deleteRowOutput = deleteRowOutput, joinedAttributes = joinedPlan.output, joinedRowEncoder = joinedRowEncoder, - outputRowEncoder = outputRowEncoder) + outputRowEncoder = outputRowEncoder, + rowDroppedColumnIndex = rowDroppedColumnIndex) Dataset.ofRows(spark, joinedPlan).mapPartitions(processor.processPartition)(outputRowEncoder) } @@ -1145,6 +1376,8 @@ object GpuMergeIntoCommand { val FILE_NAME_COL = "_file_name_" val SOURCE_ROW_PRESENT_COL = "_source_row_present_" val TARGET_ROW_PRESENT_COL = "_target_row_present_" + val DUPLICATE_MATCH_RANK_COL = "_duplicate_match_rank_" + val NON_DETERMINISTIC_VALUE_COL = "_non_deterministic_value_" val ROW_DROPPED_COL = GpuDeltaMergeConstants.ROW_DROPPED_COL val INCR_ROW_COUNT_COL = "_incr_row_count_" @@ -1166,12 +1399,18 @@ object GpuMergeIntoCommand { * @param notMatchedOutputs corresponding output for each not-matched clause. for each clause, * we have 1-2 output rows, each of which is a sequence of * expressions to apply to the joined row + * @param notMatchedBySourceConditions condition for each not-matched-by-source clause + * @param notMatchedBySourceOutputs corresponding output for each not-matched-by-source + * clause. for each clause, we have 1-3 output rows, each of + * which is a sequence of expressions to apply to the joined + * row * @param noopCopyOutput no-op expression to copy a target row to the output * @param deleteRowOutput expression to drop a row from the final output. this is used for * source rows that don't match any not-matched clauses * @param joinedAttributes schema of our outer-joined dataframe * @param joinedRowEncoder joinedDF row encoder * @param outputRowEncoder final output row encoder + * @param rowDroppedColumnIndex position of ROW_DROPPED_COL in every projected output row */ class JoinedRowProcessor( targetRowHasNoMatch: Expression, @@ -1180,11 +1419,14 @@ object GpuMergeIntoCommand { matchedOutputs: Seq[Seq[Seq[Expression]]], notMatchedConditions: Seq[Expression], notMatchedOutputs: Seq[Seq[Seq[Expression]]], + notMatchedBySourceConditions: Seq[Expression], + notMatchedBySourceOutputs: Seq[Seq[Seq[Expression]]], noopCopyOutput: Seq[Expression], deleteRowOutput: Seq[Expression], joinedAttributes: Seq[Attribute], joinedRowEncoder: ExpressionEncoder[Row], - outputRowEncoder: ExpressionEncoder[Row]) extends Serializable { + outputRowEncoder: ExpressionEncoder[Row], + rowDroppedColumnIndex: Int) extends Serializable { private def generateProjection(exprs: Seq[Expression]): UnsafeProjection = { UnsafeProjection.create(exprs, joinedAttributes) @@ -1202,43 +1444,38 @@ object GpuMergeIntoCommand { val matchedProjs = matchedOutputs.map(_.map(generateProjection)) val notMatchedPreds = notMatchedConditions.map(generatePredicate) val notMatchedProjs = notMatchedOutputs.map(_.map(generateProjection)) + val notMatchedBySourcePreds = notMatchedBySourceConditions.map(generatePredicate) + val notMatchedBySourceProjs = notMatchedBySourceOutputs.map(_.map(generateProjection)) val noopCopyProj = generateProjection(noopCopyOutput) val deleteRowProj = generateProjection(deleteRowOutput) val outputProj = UnsafeProjection.create(outputRowEncoder.schema) - // this is accessing ROW_DROPPED_COL. If ROW_DROPPED_COL is not in outputRowEncoder.schema - // then CDC must be disabled and it's the column after our output cols - def shouldDeleteRow(row: InternalRow): Boolean = { - row.getBoolean( - outputRowEncoder.schema.getFieldIndex(ROW_DROPPED_COL) - .getOrElse(outputRowEncoder.schema.fields.size) - ) - } + // ROW_DROPPED_COL sits right after the target output columns in every projected row, + // whether or not it is part of outputRowEncoder.schema (it is only with CDC). + def shouldDeleteRow(row: InternalRow): Boolean = row.getBoolean(rowDroppedColumnIndex) def processRow(inputRow: InternalRow): Iterator[InternalRow] = { - if (targetRowHasNoMatchPred.eval(inputRow)) { - // Target row did not match any source row, so just copy it to the output - Iterator(noopCopyProj.apply(inputRow)) + // Identify which set of clauses to execute: matched, not-matched or not-matched-by-source + val (predicates, projections, noopAction) = if (targetRowHasNoMatchPred.eval(inputRow)) { + // Target row did not match any source row, so update the target row. + (notMatchedBySourcePreds, notMatchedBySourceProjs, noopCopyProj) + } else if (sourceRowHasNoMatchPred.eval(inputRow)) { + // Source row did not match with any target row, so insert the new source row + (notMatchedPreds, notMatchedProjs, deleteRowProj) } else { - // identify which set of clauses to execute: matched or not-matched ones - val (predicates, projections, noopAction) = if (sourceRowHasNoMatchPred.eval(inputRow)) { - // Source row did not match with any target row, so insert the new source row - (notMatchedPreds, notMatchedProjs, deleteRowProj) - } else { - // Source row matched with target row, so update the target row - (matchedPreds, matchedProjs, noopCopyProj) - } + // Source row matched with target row, so update the target row + (matchedPreds, matchedProjs, noopCopyProj) + } - // find (predicate, projection) pair whose predicate satisfies inputRow - val pair = (predicates zip projections).find { - case (predicate, _) => predicate.eval(inputRow) - } + // find (predicate, projection) pair whose predicate satisfies inputRow + val pair = (predicates zip projections).find { + case (predicate, _) => predicate.eval(inputRow) + } - pair match { - case Some((_, projections)) => - projections.map(_.apply(inputRow)).iterator - case None => Iterator(noopAction.apply(inputRow)) - } + pair match { + case Some((_, projections)) => + projections.map(_.apply(inputRow)).iterator + case None => Iterator(noopAction.apply(inputRow)) } } diff --git a/delta-lake/delta-spark400db173/src/main/scala/com/nvidia/spark/rapids/delta/shims/MergeIntoCommandMetaShim.scala b/delta-lake/delta-spark400db173/src/main/scala/com/nvidia/spark/rapids/delta/shims/MergeIntoCommandMetaShim.scala index 0d53ef18f49..ba1fe821dd2 100644 --- a/delta-lake/delta-spark400db173/src/main/scala/com/nvidia/spark/rapids/delta/shims/MergeIntoCommandMetaShim.scala +++ b/delta-lake/delta-spark400db173/src/main/scala/com/nvidia/spark/rapids/delta/shims/MergeIntoCommandMetaShim.scala @@ -38,10 +38,6 @@ object MergeIntoCommandMetaShim { private def tagForGpuCommon( meta: RapidsMeta[_, _, _], mergeCmd: MergeIntoCommandBase): Unit = { - // see https://github.com/NVIDIA/spark-rapids/issues/8415 for more information - if (mergeCmd.notMatchedBySourceClauses.nonEmpty) { - meta.willNotWorkOnGpu("notMatchedBySourceClauses not supported on GPU") - } tagPersistentDeletionVectorFallback( meta, mergeCmd.targetFileIndex.deltaLog, diff --git a/integration_tests/src/main/python/delta_lake_merge_test.py b/integration_tests/src/main/python/delta_lake_merge_test.py index a27f5ae6b9b..7f4bf835995 100644 --- a/integration_tests/src/main/python/delta_lake_merge_test.py +++ b/integration_tests/src/main/python/delta_lake_merge_test.py @@ -191,6 +191,8 @@ def checker(data_path, do_merge): @pytest.mark.skipif((not is_databricks_runtime()) and is_before_spark_340(), reason="NOT MATCHED BY SOURCE added in Delta Lake 2.4") @pytest.mark.skipif(is_spark_41x(), reason="NOT MATCHED BY SOURCE is supported on the GPU with OSS Delta 4.1") +@pytest.mark.skipif(is_databricks173_or_later(), + reason="NOT MATCHED BY SOURCE is supported on the GPU with Databricks 17.3+") @pytest.mark.parametrize("enable_deletion_vectors", deletion_vector_values_with_xfail_reasons( enabled_xfail_reason='https://github.com/NVIDIA/spark-rapids/issues/12042'), ids=idfn) def test_delta_merge_not_matched_by_source_fallback(spark_tmp_path, spark_tmp_table_factory, enable_deletion_vectors): @@ -216,8 +218,9 @@ def checker(data_path, do_merge): @allow_non_gpu(*delta_meta_allow) @delta_lake @ignore_order -@pytest.mark.skipif(not is_spark_41x(), - reason="NOT MATCHED BY SOURCE is supported on the GPU with OSS Delta 4.1") +@pytest.mark.skipif(not (is_spark_41x() or is_databricks173_or_later()), + reason="NOT MATCHED BY SOURCE is supported on the GPU with OSS Delta 4.1 " + "and Databricks 17.3+") @pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) def test_delta_merge_not_matched_by_source(spark_tmp_path, spark_tmp_table_factory, use_cdf): def src_table_func(spark): @@ -250,6 +253,49 @@ def dest_table_func(spark): conf=delta_merge_enabled_conf) assert_equal(expected, actual) + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not (is_spark_41x() or is_databricks173_or_later()), + reason="NOT MATCHED BY SOURCE is supported on the GPU with OSS Delta 4.1 " + "and Databricks 17.3+") +def test_delta_merge_not_matched_by_source_null_safe_keys(spark_tmp_path, spark_tmp_table_factory): + # Composite null-safe merge key with an update-only NOT MATCHED BY SOURCE clause, the shape + # used by SCD type 2 pipelines to expire current rows that disappeared from the source. + def src_table_func(spark): + return spark.createDataFrame( + [(1, None, 100), (2, "x", 200), (5, None, 500)], "k1 INT, k2 STRING, v INT") + + def dest_table_func(spark): + return spark.createDataFrame( + [(1, None, 10, True), (2, "x", 20, True), (3, None, 30, True), (4, "y", 40, False)], + "k1 INT, k2 STRING, v INT, current BOOLEAN") + + merge_sql = "MERGE INTO {dest_table} AS dest " \ + "USING {src_table} AS src " \ + "ON dest.k1 <=> src.k1 AND dest.k2 <=> src.k2 " \ + "WHEN MATCHED AND dest.v <> src.v THEN UPDATE SET dest.v = src.v " \ + "WHEN NOT MATCHED THEN " \ + " INSERT (k1, k2, v, current) VALUES (src.k1, src.k2, src.v, true) " \ + "WHEN NOT MATCHED BY SOURCE AND dest.current THEN UPDATE SET dest.current = false" + + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=False, enable_deletion_vectors=False, + src_table_func=src_table_func, dest_table_func=dest_table_func, + merge_sql=merge_sql, compare_logs=False, conf=delta_merge_enabled_conf) + + expected = [(1, None, 100, True), (2, "x", 200, True), (3, None, 30, False), + (4, "y", 40, False), (5, None, 500, True)] + data_path = spark_tmp_path + "/DELTA_DATA" + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: [tuple(row) for row in + read_delta_path(spark, data_path + "/" + run).orderBy("k1").collect()], + conf=delta_merge_enabled_conf) + assert_equal(expected, actual) + @allow_non_gpu("BroadcastHashJoinExec,ColumnarToRowExec,BroadcastExchangeExec," "UnionExec,UnionWithLocalDataExec,RangeExec", delta_write_fallback_allow, *delta_meta_allow) @@ -471,15 +517,23 @@ def test_delta_merge_standard_upsert_db173_smoke(spark_tmp_path, spark_tmp_table conf=delta_merge_enabled_conf) -@allow_non_gpu("ExecutedCommandExec,BroadcastHashJoinExec,ColumnarToRowExec,BroadcastExchangeExec,DataWritingCommandExec", delta_write_fallback_allow, *delta_meta_allow) +@allow_non_gpu(*delta_meta_allow) @delta_lake @ignore_order @pytest.mark.skipif(not is_databricks173_or_later(), - reason="Issue-specific fallback coverage for Databricks 17.3+") -def test_delta_merge_not_matched_by_source_db173_fallback(spark_tmp_path, spark_tmp_table_factory): - def checker(data_path, do_merge): - assert_gpu_fallback_write(do_merge, read_delta_path, data_path, "ExecutedCommandExec", - conf=delta_merge_enabled_conf) + reason="NOT MATCHED BY SOURCE is supported on the GPU with Databricks 17.3+") +@pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) +def test_delta_merge_not_matched_by_source_db173(spark_tmp_path, spark_tmp_table_factory, use_cdf): + # Every row path is exercised: matched rows are updated, source-only rows are inserted, + # target-only rows with b > 0 are updated by the NOT MATCHED BY SOURCE clause and the + # remaining target-only rows are copied unchanged. The GPU merge processor must be present + # in the plan, which proves the command did not fall back to the CPU. + def src_table_func(spark): + return spark.createDataFrame([(a, a * 10) for a in range(0, 300, 3)], "a INT, b INT") + + def dest_table_func(spark): + return spark.createDataFrame( + [(a, -1 if a % 4 == 0 else a) for a in range(0, 200)], "a INT, b INT") merge_sql = "MERGE INTO {dest_table} " \ "USING {src_table} " \ @@ -490,14 +544,557 @@ def checker(data_path, do_merge): " INSERT (a, b) VALUES ({src_table}.a, {src_table}.b) " \ "WHEN NOT MATCHED BY SOURCE AND {dest_table}.b > 0 THEN " \ " UPDATE SET {dest_table}.b = 0" - delta_sql_merge_test(spark_tmp_path, spark_tmp_table_factory, - use_cdf=False, enable_deletion_vectors=False, - src_table_func=lambda spark: binary_op_df( - spark, SetValuesGen(IntegerType(), range(10))), - dest_table_func=lambda spark: binary_op_df( - spark, SetValuesGen(IntegerType(), range(20, 30))), - merge_sql=merge_sql, - check_func=checker) + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=use_cdf, enable_deletion_vectors=False, + src_table_func=src_table_func, dest_table_func=dest_table_func, + merge_sql=merge_sql, compare_logs=False, + assert_func=_assert_gpu_merge_processor, + conf=delta_merge_no_cpu_bridge_conf) + + def expected_row(a): + if a % 3 == 0: + return (a, a * 10) + if a % 4 == 0: + return (a, -1) + return (a, 0) + expected = [expected_row(a) for a in range(0, 200)] + \ + [(a, a * 10) for a in range(201, 300, 3)] + data_path = spark_tmp_path + "/DELTA_DATA" + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: [tuple(row) for row in + read_delta_path(spark, data_path + "/" + run).orderBy("a").collect()], + conf=delta_merge_enabled_conf) + assert_equal(expected, actual) + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="Per-clause merge metrics are reported by the Databricks 17.3 GPU merge command") +def test_delta_merge_delete_only_duplicate_source_metrics_db173(spark_tmp_path, spark_tmp_table_factory): + # An unconditional MATCHED DELETE is the only merge that allows several source rows to match + # the same target row. The delete counters are incremented once per joined pair and then + # compensated, so both numTargetRowsDeleted and numTargetRowsMatchedDeleted must equal the + # number of target rows actually deleted, on the CPU and on the GPU. + def src_table_func(spark): + return spark.createDataFrame( + [(a, b) for a in range(0, 50) for b in range(3)], "a INT, b INT") + + def dest_table_func(spark): + return spark.createDataFrame([(a, -1) for a in range(0, 100)], "a INT, b INT") + + merge_sql = "MERGE INTO {dest_table} USING {src_table} ON {dest_table}.a == {src_table}.a " \ + "WHEN MATCHED THEN DELETE" + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=False, enable_deletion_vectors=False, + src_table_func=src_table_func, dest_table_func=dest_table_func, + merge_sql=merge_sql, compare_logs=False, conf=delta_merge_enabled_conf) + + def merge_metrics(spark, path): + row = spark.sql(f"DESCRIBE HISTORY delta.`{path}`") \ + .where("operation = 'MERGE'").orderBy("version", ascending=False).first() + return {k: int(row["operationMetrics"][k]) + for k in ["numTargetRowsDeleted", "numTargetRowsMatchedDeleted"]} + data_path = spark_tmp_path + "/DELTA_DATA" + for run in ["CPU", "GPU"]: + actual = with_cpu_session(lambda spark: merge_metrics(spark, data_path + "/" + run), + conf=delta_merge_enabled_conf) + assert actual == {"numTargetRowsDeleted": 50, "numTargetRowsMatchedDeleted": 50}, \ + f"{run}: {actual}" + + +# Databricks Runtime 16.0+ reports a target row as ambiguously matched only when more than one +# source row satisfies the ON condition AND a WHEN MATCHED clause condition. Source rows that match +# on ON alone take no action: they are not inserted, they do not flag the target row for the +# NOT MATCHED BY SOURCE clauses, and they do not make the row ambiguous. The GPU command has to +# accept the same merges as the CPU, write the same table, and reject the same ambiguous ones. +_dup_match_target_rows = [(1, "a", True), (2, "b", True), (3, "c", True)] +_dup_match_sql_full = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.apply THEN UPDATE SET t.v = s.v " + "WHEN NOT MATCHED THEN INSERT (k, v, cur) VALUES (s.k, s.v, true) " + "WHEN NOT MATCHED BY SOURCE THEN UPDATE SET t.cur = false") +_dup_match_sql_plain = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.apply THEN UPDATE SET t.v = s.v " + "WHEN NOT MATCHED THEN INSERT (k, v, cur) VALUES (s.k, s.v, true)") +_dup_match_sql_no_matched_clause = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN NOT MATCHED THEN INSERT (k, v, cur) VALUES (s.k, s.v, true) " + "WHEN NOT MATCHED BY SOURCE THEN UPDATE SET t.cur = false") +_dup_match_sql_two_clauses = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.apply THEN UPDATE SET t.v = s.v " + "WHEN MATCHED AND NOT s.apply THEN DELETE " + "WHEN NOT MATCHED THEN INSERT (k, v, cur) VALUES (s.k, s.v, true)") +_dup_match_sql_target_condition = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND t.cur THEN UPDATE SET t.v = s.v " + "WHEN NOT MATCHED THEN INSERT (k, v, cur) VALUES (s.k, s.v, true)") +_dup_match_sql_conditional_delete = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.apply THEN DELETE") + +_dup_match_accepted_cases = [ + pytest.param([(1, "x", True), (1, "y", False), (4, "d", True)], _dup_match_sql_full, + id="one_effective_with_not_matched_by_source"), + pytest.param([(1, "x", False), (1, "y", False), (4, "d", True)], _dup_match_sql_full, + id="none_effective_with_not_matched_by_source"), + pytest.param([(1, "x", True), (1, "y", True), (4, "d", True)], _dup_match_sql_no_matched_clause, + id="no_matched_clause"), + pytest.param([(1, "x", True), (1, "y", False), (4, "d", True)], _dup_match_sql_plain, + id="one_effective_plain"), + pytest.param([(1, "x", True), (1, "y", False)], _dup_match_sql_conditional_delete, + id="conditional_delete_one_effective"), + pytest.param([(1, "x", False), (1, "y", True), (1, "z", False), (2, "q", False)], + _dup_match_sql_full, id="three_matches_one_effective"), +] +_dup_match_rejected_cases = [ + pytest.param([(1, "x", True), (1, "y", True), (4, "d", True)], _dup_match_sql_full, + id="both_effective"), + pytest.param([(1, "x", True), (1, "y", False), (4, "d", True)], _dup_match_sql_two_clauses, + id="each_row_takes_a_different_clause"), + pytest.param([(1, "x", True), (1, "y", True), (4, "d", True)], _dup_match_sql_target_condition, + id="target_only_condition"), +] + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="Databricks 16.0+ applies WHEN MATCHED conditions when detecting multiple matches") +@pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) +@pytest.mark.parametrize("src_rows,merge_sql", _dup_match_accepted_cases) +def test_delta_merge_duplicate_source_rows_matched_conditions_db173( + spark_tmp_path, spark_tmp_table_factory, src_rows, merge_sql, use_cdf): + def src_table_func(spark): + return spark.createDataFrame(src_rows, "k INT, v STRING, apply BOOLEAN") + + def dest_table_func(spark): + return spark.createDataFrame(_dup_match_target_rows, "k INT, v STRING, cur BOOLEAN") + + # With CDF on, the change rows are compared too: one pre and post image per applied target + # row and one insert row per source-only row, none for the pairs the window dropped. + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=use_cdf, enable_deletion_vectors=False, + src_table_func=src_table_func, dest_table_func=dest_table_func, + merge_sql=merge_sql, compare_logs=False, + assert_func=_assert_gpu_merge_processor, + conf=delta_merge_no_cpu_bridge_conf) + + # The row counters per clause must agree too: one action per target row, none for the + # source rows that matched on ON alone. + metric_keys = ["numTargetRowsUpdated", "numTargetRowsInserted", "numTargetRowsDeleted", + "numTargetRowsMatchedUpdated", "numTargetRowsNotMatchedBySourceUpdated"] + + def merge_metrics(spark, path): + row = spark.sql(f"DESCRIBE HISTORY delta.`{path}`") \ + .where("operation = 'MERGE'").orderBy("version", ascending=False).first() + return {k: int(row["operationMetrics"].get(k, 0)) for k in metric_keys} + data_path = spark_tmp_path + "/DELTA_DATA" + cpu_metrics, gpu_metrics = [ + with_cpu_session(lambda spark, run=run: merge_metrics(spark, data_path + "/" + run), + conf=delta_merge_enabled_conf) for run in ["CPU", "GPU"]] + assert cpu_metrics == gpu_metrics, f"CPU {cpu_metrics} vs GPU {gpu_metrics}" + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="Databricks 16.0+ applies WHEN MATCHED conditions when detecting multiple matches") +@pytest.mark.parametrize("src_rows,merge_sql", _dup_match_rejected_cases) +def test_delta_merge_duplicate_source_rows_ambiguous_error_db173( + spark_tmp_path, spark_tmp_table_factory, src_rows, merge_sql): + src_table = spark_tmp_table_factory.get() + + def do_merge(spark): + gpu_enabled = str(spark.conf.get("spark.rapids.sql.enabled", "false")).lower() == "true" + target_path = spark_tmp_path + ("/GPU" if gpu_enabled else "/CPU") + spark.createDataFrame(_dup_match_target_rows, "k INT, v STRING, cur BOOLEAN") \ + .write.format("delta") \ + .option("delta.enableDeletionVectors", "false") \ + .mode("overwrite") \ + .save(target_path) + spark.createDataFrame(src_rows, "k INT, v STRING, apply BOOLEAN") \ + .createOrReplaceTempView(src_table) + return spark.sql(merge_sql.format( + dest_table=f"delta.`{target_path}`", src_table=src_table)).collect() + + assert_gpu_and_cpu_error( + do_merge, + conf=delta_merge_no_cpu_bridge_conf, + error_message="DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE") + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="Databricks 16.0+ applies WHEN MATCHED conditions when detecting multiple matches") +@pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) +def test_delta_merge_duplicate_source_rows_helper_column_names_db173( + spark_tmp_path, spark_tmp_table_factory, use_cdf): + # De-duplicating the non-effective duplicate matches attaches helper columns to the joined + # rows. User columns that carry the helper names must survive, because the clause expressions + # still reference them: the helper names are generated to be absent from the join. + def src_table_func(spark): + return spark.createDataFrame( + [(1, True, "chosen", 10), (1, False, "ignored", 11), (4, True, "inserted", 40)], + "k INT, apply BOOLEAN, _duplicate_match_rank_ STRING, _source_row_id_ INT") + + def dest_table_func(spark): + return spark.createDataFrame([(1, "a", 100), (2, "b", 200), (3, "c", 300)], + "k INT, v STRING, _target_row_id_ INT") + + merge_sql = ("MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.apply THEN UPDATE SET t.v = s._duplicate_match_rank_, " + "t._target_row_id_ = s._source_row_id_ " + "WHEN NOT MATCHED THEN INSERT (k, v, _target_row_id_) " + "VALUES (s.k, s._duplicate_match_rank_, s._source_row_id_)") + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=use_cdf, enable_deletion_vectors=False, + src_table_func=src_table_func, dest_table_func=dest_table_func, + merge_sql=merge_sql, compare_logs=False, + assert_func=_assert_gpu_merge_processor, + conf=delta_merge_no_cpu_bridge_conf) + expected = [(1, "chosen", 10), (2, "b", 200), (3, "c", 300), (4, "inserted", 40)] + data_path = spark_tmp_path + "/DELTA_DATA" + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: sorted(tuple(row) for row in + read_delta_path(spark, data_path + "/" + run).collect()), + conf=delta_merge_enabled_conf) + assert expected == actual, f"{run}: expected {expected}, got {actual}" + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="NOT MATCHED BY SOURCE is supported on the GPU with Databricks 17.3+") +@pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) +def test_delta_merge_internal_column_names_db173(spark_tmp_path, spark_tmp_table_factory, use_cdf): + # The command attaches a row id and a file name to the target in findTouchedFiles. User + # columns carrying those names must survive untouched and stay readable by the clause + # expressions and conditions. (The presence flags of the join and the two control columns + # of the processors are generated the same way, but the Databricks CPU command rejects user + # columns with those names, so a parity test cannot use them; see the GPU-only test below.) + def src_table_func(spark): + return spark.createDataFrame([(1, True, 100, "s1"), (4, True, 400, "s4")], + "k INT, apply BOOLEAN, _row_id_ INT, _file_name_ STRING") + + def dest_table_func(spark): + return spark.createDataFrame([(1, "a", 10, "t1"), (2, "b", 20, "t2"), (3, "c", 30, "t3")], + "k INT, v STRING, _row_id_ INT, _file_name_ STRING") + + merge_sql = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.apply THEN UPDATE SET t.v = s._file_name_, t._row_id_ = s._row_id_ " + "WHEN NOT MATCHED THEN INSERT (k, v, _row_id_, _file_name_) " + "VALUES (s.k, s._file_name_, s._row_id_, s._file_name_) " + "WHEN NOT MATCHED BY SOURCE AND t._row_id_ > 25 THEN UPDATE SET t.v = concat(t.v, '-kept')") + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=use_cdf, enable_deletion_vectors=False, + src_table_func=src_table_func, dest_table_func=dest_table_func, + merge_sql=merge_sql, compare_logs=False, + assert_func=_assert_gpu_merge_processor, + conf=delta_merge_no_cpu_bridge_conf) + expected = [(1, "s1", 100, "t1"), (2, "b", 20, "t2"), (3, "c-kept", 30, "t3"), + (4, "s4", 400, "s4")] + data_path = spark_tmp_path + "/DELTA_DATA" + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: sorted(tuple(row) for row in + read_delta_path(spark, data_path + "/" + run).collect()), + conf=delta_merge_enabled_conf) + assert expected == actual, f"{run}: expected {expected}, got {actual}" + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="NOT MATCHED BY SOURCE is supported on the GPU with Databricks 17.3+") +@pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) +def test_delta_merge_control_column_names_gpu_db173(spark_tmp_path, spark_tmp_table_factory, use_cdf): + # The processors append the control columns _row_dropped_ and _incr_row_count_ to every + # output row and read the first one back by position. User columns with those names must + # neither be mistaken for the control columns nor dropped with them. The Databricks CPU + # command rejects such columns as ambiguous, so this runs on the GPU only, against + # spelled-out rows; the GPU processor must be in the plan. + def src_table_func(spark): + return spark.createDataFrame([(1, True, False, False), (4, True, True, True)], + "k INT, apply BOOLEAN, _row_dropped_ BOOLEAN, " + "_incr_row_count_ BOOLEAN") + + def dest_table_func(spark): + return spark.createDataFrame([(1, "a", True, True), (2, "b", True, False), + (3, "c", False, True)], + "k INT, v STRING, _row_dropped_ BOOLEAN, " + "_incr_row_count_ BOOLEAN") + + merge_sql = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.apply THEN UPDATE SET t.v = 'updated', " + "t._row_dropped_ = s._row_dropped_ " + "WHEN NOT MATCHED THEN INSERT (k, v, _row_dropped_, _incr_row_count_) " + "VALUES (s.k, 'inserted', s._row_dropped_, s._incr_row_count_) " + "WHEN NOT MATCHED BY SOURCE AND t._row_dropped_ THEN UPDATE SET t.v = concat(t.v, '-kept')") + expected = [(1, "updated", False, True), (2, "b-kept", True, False), (3, "c", False, True), + (4, "inserted", True, True)] + expected_changes = {"update_preimage": 2, "update_postimage": 2, "insert": 1} + + def check_func(data_path, do_merge): + gpu_path = data_path + "/GPU" + callback = spark_jvm().org.apache.spark.sql.rapids.ExecutionPlanCaptureCallback + callback.startCapture() + try: + with_gpu_session(lambda spark: do_merge(spark, gpu_path), + conf=delta_merge_no_cpu_bridge_conf) + captured_plans = callback.getResultsWithTimeout(10000) + finally: + callback.endCapture() + class_name = "GpuRapidsProcessDeltaMergeJoinExec" + assert any(callback.contains(plan, class_name) for plan in captured_plans), \ + f"{class_name} was not found in the captured MERGE plans" + actual = with_cpu_session( + lambda spark: sorted(tuple(row) for row in + read_delta_path(spark, gpu_path).collect()), + conf=delta_merge_enabled_conf) + assert expected == actual, f"expected {expected}, got {actual}" + if use_cdf: + # The change feed is read from version 0, which includes the setup's own inserts, + # so only the rows of the MERGE commit are counted. + def merge_changes(spark): + merge_version = spark.sql(f"DESCRIBE HISTORY delta.`{gpu_path}`") \ + .where("operation = 'MERGE'").orderBy("version", ascending=False) \ + .first()["version"] + return [row["_change_type"] for row in + read_delta_path_with_cdf(spark, gpu_path) + .where(f"_commit_version = {merge_version}").collect()] + changes = with_cpu_session(merge_changes, conf=delta_merge_enabled_conf) + actual_changes = {t: changes.count(t) for t in set(changes)} + assert expected_changes == actual_changes, \ + f"expected change rows {expected_changes}, got {actual_changes}" + + delta_sql_merge_test(spark_tmp_path, spark_tmp_table_factory, use_cdf, False, + src_table_func, dest_table_func, merge_sql, check_func) + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="NOT MATCHED BY SOURCE is supported on the GPU with Databricks 17.3+") +@pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) +def test_delta_merge_non_deterministic_action_values_db173(spark_tmp_path, spark_tmp_table_factory, + use_cdf): + # Databricks 17.0 and later accept non-deterministic expressions in the values of update and + # insert actions (not in conditions). Catalyst does not allow them in the GPU processor node, + # so the GPU command evaluates each one once per joined row in a projection above the join, + # guarded by the rows the clause acts on: the matched update divides by a source column that + # is zero on the second, non-effective source row of key 1, which under ANSI mode must not be + # evaluated (the CPU evaluates an action value only on the rows that take the clause), and + # that duplicate match also sends the merge through the de-duplication window. rand() is + # seeded so the CPU run is reproducible; its values still cannot equal the GPU's, so the + # tables are compared with each value replaced by the range it must fall in, and with CDF on + # the change rows of the MERGE commit must carry the same value as the table row. + def src_table_func(spark): + return spark.createDataFrame([(1, 10, 4, 2), (1, 10, 4, 0), (4, 40, 8, 4)], + "k INT, s INT, x INT, d INT") + + def dest_table_func(spark): + return spark.createDataFrame([(1, 0.5, 0.5), (2, 0.5, 0.5), (3, 0.5, 0.5)], + "k INT, v DOUBLE, u DOUBLE") + + merge_sql = ( + "MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED AND s.d <> 0 THEN UPDATE SET t.v = s.x / s.d + rand(7) " + "WHEN NOT MATCHED THEN INSERT (k, v, u) VALUES (s.k, rand(7), s.s + rand(7)) " + "WHEN NOT MATCHED BY SOURCE AND t.k = 2 THEN UPDATE SET t.u = rand(7) + 10") + # k=1 matched by its d=2 row (v = 2 + random), k=2 not matched by source (u random + 10), + # k=3 copied, k=4 inserted (v random, u random + 40) + expected = [(1, "random+2", "kept"), (2, "kept", "random+10"), (3, "kept", "kept"), + (4, "random", "random+40")] + ranges = ("CASE WHEN {c} = 0.5 THEN 'kept' WHEN {c} >= 0 AND {c} < 1 THEN 'random' " + "WHEN {c} >= 2 AND {c} < 3 THEN 'random+2' " + "WHEN {c} >= 10 AND {c} < 11 THEN 'random+10' " + "WHEN {c} >= 40 AND {c} < 41 THEN 'random+40' ELSE 'bad' END AS {c}") + conf = copy_and_update(delta_merge_no_cpu_bridge_conf, {"spark.sql.ansi.enabled": "true"}) + + def read_ranges(spark, path): + return read_delta_path(spark, path).selectExpr("k", ranges.format(c="v"), + ranges.format(c="u")) + + def check_func(data_path, do_merge): + # The merge's result rows are compared between the engines and the GPU processor is + # asserted in the plan; then the tables, with each value replaced by its range. + _assert_gpu_merge_processor(do_merge, data_path, conf) + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: sorted(tuple(row) for row in + read_ranges(spark, data_path + "/" + run).collect()), + conf=delta_merge_enabled_conf) + assert expected == actual, f"{run}: expected {expected}, got {actual}" + if use_cdf: + def changed_values(spark, path): + merge_version = spark.sql(f"DESCRIBE HISTORY delta.`{path}`") \ + .where("operation = 'MERGE'").orderBy("version", ascending=False) \ + .first()["version"] + changes = read_delta_path_with_cdf(spark, path) \ + .where(f"_commit_version = {merge_version} AND " + "_change_type IN ('insert', 'update_postimage')") + table = {r["k"]: (r["v"], r["u"]) for r in read_delta_path(spark, path).collect()} + return {r["k"]: (r["v"], r["u"]) for r in changes.collect()}, table + for run in ["CPU", "GPU"]: + changed, table = with_cpu_session( + lambda spark: changed_values(spark, data_path + "/" + run), + conf=delta_merge_enabled_conf) + assert sorted(changed.keys()) == [1, 2, 4], f"{run}: {changed}" + for k, values in changed.items(): + assert table[k] == values, \ + f"{run}: change row of k={k} carries {values}, the table row {table[k]}" + + delta_sql_merge_test(spark_tmp_path, spark_tmp_table_factory, use_cdf, False, + src_table_func, dest_table_func, merge_sql, check_func) + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="Per-clause merge metrics are reported by the Databricks 17.3 GPU merge command") +def test_delta_merge_delete_only_duplicate_cdc_internal_column_names_db173( + spark_tmp_path, spark_tmp_table_factory): + # An unconditional MATCHED DELETE with duplicate source matches and CDF on de-duplicates the + # change rows by row ids attached to both sides. User columns carrying those names must + # survive and stay readable by the insert action. + def src_table_func(spark): + return spark.createDataFrame([(1, 10), (1, 11), (4, 40), (4, 41)], + "k INT, _source_row_id_ INT") + + def dest_table_func(spark): + return spark.createDataFrame([(1, "a", 100), (2, "b", 200), (3, "c", 300)], + "k INT, v STRING, _target_row_id_ INT") + + merge_sql = ("MERGE INTO {dest_table} t USING {src_table} s ON t.k = s.k " + "WHEN MATCHED THEN DELETE " + "WHEN NOT MATCHED THEN INSERT (k, v, _target_row_id_) " + "VALUES (s.k, 'inserted', s._source_row_id_)") + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=True, enable_deletion_vectors=False, + src_table_func=src_table_func, dest_table_func=dest_table_func, + merge_sql=merge_sql, compare_logs=False, + assert_func=_assert_gpu_merge_processor, + conf=delta_merge_no_cpu_bridge_conf) + expected = [(2, "b", 200), (3, "c", 300), (4, "inserted", 40), (4, "inserted", 41)] + data_path = spark_tmp_path + "/DELTA_DATA" + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: sorted(tuple(row) for row in + read_delta_path(spark, data_path + "/" + run).collect()), + conf=delta_merge_enabled_conf) + assert expected == actual, f"{run}: expected {expected}, got {actual}" + + +# A clause condition that evaluates to NULL is false: the row moves on to the next clause or to +# the default action (copy a target row, skip a source row). The GPU merge processor splits each +# batch with the condition and its negation, and a NULL passes neither filter, so without +# nulls-as-false handling such rows silently disappeared. Target rows a = 0..19 have b NULL when +# a % 5 == 0; source rows are the even a = 0..28 with b = a * 10 and a flag that is NULL when +# a % 6 == 0, true when a % 4 == 0, false otherwise. +def _nullable_condition_src(spark): + def flag(a): + return None if a % 6 == 0 else a % 4 == 0 + return spark.createDataFrame([(a, a * 10, flag(a)) for a in range(0, 30, 2)], + "a INT, b INT, flag BOOLEAN") + + +def _nullable_condition_dest(spark): + return spark.createDataFrame([(a, None if a % 5 == 0 else a) for a in range(0, 20)], + "a INT, b INT") + + +_nullable_condition_clauses = ( + "WHEN MATCHED AND {src_table}.flag THEN UPDATE SET {dest_table}.b = {src_table}.b " + "WHEN MATCHED AND NOT {src_table}.flag THEN DELETE " + "WHEN NOT MATCHED AND {src_table}.flag THEN INSERT (a, b) VALUES ({src_table}.a, {src_table}.b) ") + + +def _nullable_condition_expected(with_not_matched_by_source): + rows = [] + for a in range(0, 20): + b = None if a % 5 == 0 else a + if a % 2 == 0: # matched + if a % 6 == 0: # flag NULL: neither matched clause applies, copied unchanged + rows.append((a, b)) + elif a % 4 == 0: # flag true: updated + rows.append((a, a * 10)) + # flag false: deleted + elif with_not_matched_by_source: # target only: condition b > 0 is NULL when b is NULL + rows.append((a, None if b is None else 0)) + else: + rows.append((a, b)) + # source only: a = 20..28, inserted only when the flag is true (NULL is not) + rows += [(a, a * 10) for a in range(20, 30, 2) if a % 6 != 0 and a % 4 == 0] + return rows + + +def _assert_nullable_condition_result(spark_tmp_path, with_not_matched_by_source): + expected = sorted(_nullable_condition_expected(with_not_matched_by_source)) + data_path = spark_tmp_path + "/DELTA_DATA" + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: sorted(tuple(row) for row in + read_delta_path(spark, data_path + "/" + run).collect()), + conf=delta_merge_enabled_conf) + assert expected == actual, f"{run}: expected {expected}, got {actual}" + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(is_before_spark_320(), reason="Delta Lake writes are not supported before Spark 3.2.x") +def test_delta_merge_nullable_matched_conditions(spark_tmp_path, spark_tmp_table_factory): + merge_sql = "MERGE INTO {dest_table} USING {src_table} ON {dest_table}.a == {src_table}.a " + \ + _nullable_condition_clauses + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=False, enable_deletion_vectors=False, + src_table_func=_nullable_condition_src, dest_table_func=_nullable_condition_dest, + merge_sql=merge_sql, compare_logs=False, conf=delta_merge_enabled_conf) + _assert_nullable_condition_result(spark_tmp_path, with_not_matched_by_source=False) + + +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not (is_spark_41x() or is_databricks173_or_later()), + reason="NOT MATCHED BY SOURCE is supported on the GPU with OSS Delta 4.1 " + "and Databricks 17.3+") +@pytest.mark.parametrize("use_cdf", [False, True], ids=idfn) +def test_delta_merge_nullable_not_matched_by_source_condition( + spark_tmp_path, spark_tmp_table_factory, use_cdf): + merge_sql = "MERGE INTO {dest_table} USING {src_table} ON {dest_table}.a == {src_table}.a " + \ + _nullable_condition_clauses + \ + "WHEN NOT MATCHED BY SOURCE AND {dest_table}.b > 0 THEN UPDATE SET {dest_table}.b = 0" + assert_delta_sql_merge_collect( + spark_tmp_path, spark_tmp_table_factory, + use_cdf=use_cdf, enable_deletion_vectors=False, + src_table_func=_nullable_condition_src, dest_table_func=_nullable_condition_dest, + merge_sql=merge_sql, compare_logs=False, + assert_func=_assert_gpu_merge_processor, + conf=delta_merge_no_cpu_bridge_conf) + _assert_nullable_condition_result(spark_tmp_path, with_not_matched_by_source=True) @allow_non_gpu("ExecutedCommandExec", *delta_meta_allow) @@ -578,6 +1175,107 @@ def read_func(spark, path): conf=delta_merge_enabled_conf) +@allow_non_gpu(*delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="NOT MATCHED BY SOURCE is supported on the GPU with Databricks 17.3+") +def test_delta_merge_not_matched_by_source_schema_evolution_db173(spark_tmp_path, spark_tmp_table_factory): + # Schema evolution adds a source column to the target. Target-only rows go through the + # NOT MATCHED BY SOURCE clause and get NULL in the new column, like the copied rows do. + data_path = spark_tmp_path + "/DELTA_DATA" + src_table = spark_tmp_table_factory.get() + + def src_table_func(spark): + return spark.createDataFrame([(1, "updated", "new-col"), (3, "inserted", "brand-new")], + "a INT, b STRING, c STRING") + + def dest_table_func(spark): + return spark.createDataFrame([(1, "old"), (2, "keep"), (4, "stale")], "a INT, b STRING") + + def setup_tables(spark): + setup_delta_dest_tables(spark, data_path, dest_table_func, + use_cdf=False, enable_deletion_vectors=False) + src_table_func(spark).createOrReplaceTempView(src_table) + + def do_merge(spark, path): + src_table_func(spark).createOrReplaceTempView(src_table) + return spark.sql( + "MERGE WITH SCHEMA EVOLUTION INTO delta.`{path}` AS dest USING {src_table} AS src " + "ON dest.a = src.a " + "WHEN MATCHED THEN UPDATE SET * " + "WHEN NOT MATCHED THEN INSERT * " + "WHEN NOT MATCHED BY SOURCE AND dest.a > 2 THEN UPDATE SET dest.b = concat(dest.b, '-gone')" + .format(path=path, src_table=src_table)).collect() + + with_cpu_session(setup_tables) + _assert_gpu_merge_processor(do_merge, data_path, delta_merge_no_cpu_bridge_conf) + expected = [(1, "updated", "new-col"), (2, "keep", None), (3, "inserted", "brand-new"), + (4, "stale-gone", None)] + for run in ["CPU", "GPU"]: + actual = with_cpu_session( + lambda spark: sorted(tuple(row) for row in + read_delta_path(spark, data_path + "/" + run) + .select("a", "b", "c").collect()), + conf=delta_merge_enabled_conf) + assert expected == actual, f"{run}: expected {expected}, got {actual}" + + +@allow_non_gpu("ColumnarToRowExec", *delta_meta_allow) +@delta_lake +@ignore_order +@pytest.mark.skipif(not is_databricks173_or_later(), + reason="DBR 17.3 row tracking regression coverage") +def test_delta_merge_preserves_row_tracking_db173(spark_tmp_path): + # Every clause type touches a row-tracked target. The row ids of the rows that existed before + # the merge must survive on both engines, whether the row is updated by a matched clause, by + # a not-matched-by-source clause, or copied, and the commit version moves only for the + # updated rows. The inserted row gets a fresh id that the file layout decides, and the + # join-based GPU merge lays out files differently from the CPU, so the ids are checked per + # row rather than by comparing the commit logs as the UPDATE and DELETE tests do. + conf = copy_and_update(delta_merge_enabled_conf, delta_row_tracking_dml_conf) + data_path = spark_tmp_path + "/DELTA_DATA" + with_cpu_session(lambda spark: setup_delta_row_tracking_dest_tables( + spark, data_path, row_tracking_dml_test_df), conf=conf) + merge_sql = ("MERGE INTO delta.`{path}` t " + "USING (SELECT * FROM VALUES (2, 'B', 'y'), (9, 'I', 'y') AS s(a, b, c)) s " + "ON t.a = s.a " + "WHEN MATCHED THEN UPDATE SET t.c = s.c " + "WHEN NOT MATCHED THEN INSERT * " + "WHEN NOT MATCHED BY SOURCE AND t.a = 4 THEN UPDATE SET t.c = 'z'") + + def tracked_rows(spark, path): + rows = spark.sql("SELECT a, b, c, _metadata.row_id AS row_id, " + "_metadata.row_commit_version AS row_commit_version " + "FROM delta.`{}`".format(path)).collect() + return {r["a"]: (r["b"], r["c"], r["row_id"], r["row_commit_version"]) for r in rows} + + data = {} + for run in ["CPU", "GPU"]: + path = data_path + "/" + run + before = with_cpu_session(lambda spark: tracked_rows(spark, path), conf=conf) + do_merge = lambda spark: spark.sql(merge_sql.format(path=path)).collect() + if run == "GPU": + assert_rapids_delta_write(do_merge, conf=conf) + else: + with_cpu_session(do_merge, conf=conf) + after = with_cpu_session(lambda spark: tracked_rows(spark, path), conf=conf) + assert sorted(after.keys()) == [1, 2, 3, 4, 9], "{}: {}".format(run, after) + for a in [1, 2, 3, 4]: + assert after[a][2] == before[a][2], \ + "{}: row id of a={} changed: {} -> {}".format(run, a, before[a], after[a]) + for a in [1, 3]: # copied unchanged + assert after[a][3] == before[a][3], \ + "{}: commit version of copied a={} changed: {} -> {}".format(run, a, before[a], after[a]) + for a in [2, 4]: # updated by the matched and the not-matched-by-source clause + assert after[a][3] > before[a][3], \ + "{}: commit version of updated a={} did not move: {} -> {}".format(run, a, before[a], after[a]) + assert after[9][2] > max(v[2] for v in before.values()), \ + "{}: inserted row id is not fresh: {}".format(run, after[9]) + data[run] = sorted((a,) + v[:2] for a, v in after.items()) + assert data["CPU"] == data["GPU"], "CPU {} vs GPU {}".format(data["CPU"], data["GPU"]) + + @allow_non_gpu(*delta_meta_allow) @delta_lake @ignore_order