This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 85dd958b6b [spark] supports deletion actions in data evolution merge
into (#8446)
85dd958b6b is described below
commit 85dd958b6ba4d3b054dfb64d9d87c9a81318eb73
Author: Faiz <[email protected]>
AuthorDate: Fri Jul 3 21:47:40 2026 +0800
[spark] supports deletion actions in data evolution merge into (#8446)
---
docs/docs/multimodal-table/data-evolution.mdx | 54 ++-
.../MergeIntoPaimonDataEvolutionTable.scala | 416 +++++++++++++++------
.../MergeIntoPaimonDataEvolutionTable.scala | 407 ++++++++++++++------
.../spark/commands/PaimonRowLevelCommand.scala | 11 +
.../spark/sql/DataEvolutionDeletionTestBase.scala | 381 +++++++++++++++++++
5 files changed, 1042 insertions(+), 227 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index ad8c03064c..4ca866ff4a 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -40,6 +40,8 @@ Data Evolution mode offers the following advantages:
I/O cost of rewriting untouched columns.
- **Reduced file rewrites**: append new column data to dedicated files when
backfilling or evolving data.
+- **Delete support**: record row deletions with deletion vectors without
+ rewriting existing column files.
- **Optimized reads**: merge original and updated column files at read time.
To enable Data Evolution, create an append table with both
@@ -204,18 +206,62 @@ commit.close()
Notes:
- Spark SQL supports standalone `UPDATE` statements for Data Evolution tables.
- SQL `DELETE` statements are not supported for Data Evolution tables yet. Use
- Spark `UPDATE` or `MERGE INTO`, the Flink procedure, or PyPaimon APIs.
- Concurrent Spark SQL `UPDATE` statements that update the same data file and
columns may be retried automatically. Configure retry attempts with
`spark.paimon.write.data-evolution.update-conflict-retry.max-attempts` and
retry wait time with
`spark.paimon.write.data-evolution.update-conflict-retry.wait-ms`.
-- `MERGE INTO` for Data Evolution tables does not support the
- `WHEN NOT MATCHED BY SOURCE` clause.
+- In Spark SQL, `MERGE INTO` supports `WHEN NOT MATCHED BY SOURCE` for delete
+ actions on Data Evolution tables.
- The Flink `data_evolution_merge_into` procedure currently supports updating
or inserting columns, but not inserting new rows.
+## Deletes
+
+Data Evolution tables can use deletion vectors to record deleted rows without
+rewriting existing column files. To write deletes, enable deletion vectors
+together with row tracking and Data Evolution:
+
+:::warning
+
+Deleting rows with deletion vectors does not physically remove the original
+data immediately. If a table has many deleted rows, run compaction to
+materialize the deletions; otherwise, the deleted data still consumes storage
+space and may reduce read efficiency.
+
+:::
+
+```sql
+CREATE TABLE target_table (id INT, b INT, c INT) TBLPROPERTIES (
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true',
+ 'deletion-vectors.enabled' = 'true'
+);
+```
+
+Spark SQL supports `DELETE FROM` for Data Evolution tables:
+
+```sql
+DELETE FROM target_table WHERE id = 1;
+```
+
+Spark SQL also supports delete actions in `MERGE INTO`:
+
+```sql
+CREATE TABLE source_table (id INT, op STRING);
+INSERT INTO source_table VALUES (2, 'delete'), (3, 'update');
+
+MERGE INTO target_table AS t
+USING source_table AS s
+ON t.id = s.id
+WHEN MATCHED AND s.op = 'delete' THEN DELETE
+WHEN MATCHED AND s.op = 'update' THEN UPDATE SET t.b = t.b + 10
+WHEN NOT MATCHED BY SOURCE AND t.id > 10 THEN DELETE;
+```
+
+The `WHEN NOT MATCHED BY SOURCE` clause requires Spark 3.4 or later. Currently,
+only Spark supports writing deletes for Data Evolution tables.
+
## Self Updates
Self updates transform existing column values in place. In Flink SQL, create a
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index e4e2d9b1e1..5e28c056f2 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
import org.apache.paimon.Snapshot
import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.deletionvectors.DeletionVector
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.index.GlobalIndexMeta
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
@@ -71,23 +72,26 @@ case class MergeIntoPaimonDataEvolutionTable(
matchedActions: Seq[MergeAction],
notMatchedActions: Seq[MergeAction],
notMatchedBySourceActions: Seq[MergeAction])
- extends PaimonLeafRunnableCommand
- with WithFileStoreTable
- with ExpressionHelper
+ extends PaimonRowLevelCommand
with Logging {
- private lazy val writer = PaimonSparkWriter(table)
-
- assert(
- notMatchedBySourceActions.isEmpty,
- "notMatchedBySourceActions is not supported in
MergeIntoPaimonDataEvolutionTable.")
assert(
- matchedActions.forall(x => x.isInstanceOf[UpdateAction]),
- "Only SET clause is supported in MergeIntoPaimonDataEvolutionTable for
SQL: WHEN MATCHED.")
+ matchedActions.forall {
+ case _: UpdateAction | _: DeleteAction => true
+ case _ => false
+ },
+ "Only SET and DELETE clauses are supported in
MergeIntoPaimonDataEvolutionTable for SQL: " +
+ "WHEN MATCHED."
+ )
assert(
notMatchedActions.forall(x => x.isInstanceOf[InsertAction]),
"Only INSERT clause is supported in MergeIntoPaimonDataEvolutionTable for
SQL: WHEN NOT MATCHED."
)
+ assert(
+ notMatchedBySourceActions.forall(x => x.isInstanceOf[DeleteAction]),
+ "Only DELETE clause is supported in MergeIntoPaimonDataEvolutionTable for
SQL: WHEN NOT " +
+ "MATCHED BY SOURCE."
+ )
import MergeIntoPaimonDataEvolutionTable._
@@ -105,6 +109,18 @@ case class MergeIntoPaimonDataEvolutionTable(
override lazy val table: FileStoreTable =
targetSparkTable.getTable.asInstanceOf[FileStoreTable]
+ private lazy val hasTargetSideDeleteActions: Boolean =
+ matchedActions.exists(_.isInstanceOf[DeleteAction]) ||
+ notMatchedBySourceActions.exists(_.isInstanceOf[DeleteAction])
+
+ private lazy val hasMatchedDeleteActions: Boolean =
+ matchedActions.exists(_.isInstanceOf[DeleteAction])
+
+ assert(
+ !hasTargetSideDeleteActions ||
table.coreOptions().deletionVectorsEnabled(),
+ "MERGE INTO DELETE on DataEvolution table requires
deletion-vectors.enabled=true."
+ )
+
private val updateColumns: Set[AttributeReference] = {
val columns = mutable.Set[AttributeReference]()
for (action <- matchedActions) {
@@ -115,6 +131,7 @@ case class MergeIntoPaimonDataEvolutionTable(
columns += assignmentKeyAttribute(assignment)
}
}
+ case _ =>
}
}
columns.toSet
@@ -153,8 +170,8 @@ case class MergeIntoPaimonDataEvolutionTable(
assert(
!(isSelfMergeOnRowId && (notMatchedActions.nonEmpty ||
notMatchedBySourceActions.nonEmpty)),
- "Self-Merge on _ROW_ID only supports WHEN MATCHED THEN UPDATE. WHEN NOT
MATCHED and WHEN " +
- "NOT MATCHED BY SOURCE are not supported."
+ "Self-Merge on _ROW_ID only supports WHEN MATCHED actions. WHEN NOT
MATCHED and " +
+ "WHEN NOT MATCHED BY SOURCE are not supported."
)
private lazy val targetRelation: DataSourceV2Relation =
matchedUpdateScanTarget._2
@@ -169,7 +186,11 @@ case class MergeIntoPaimonDataEvolutionTable(
}
private def invokeMergeInto(sparkSession: SparkSession): Unit = {
+ val readSnapshot = table.snapshotManager().latestSnapshot()
val snapshotReader = table.newSnapshotReader()
+ if (readSnapshot != null) {
+ snapshotReader.withSnapshot(readSnapshot)
+ }
pushDownMergePartitionFilter(snapshotReader)
val plan = snapshotReader.read()
val tableSplits: Seq[DataSplit] = plan
@@ -212,7 +233,7 @@ case class MergeIntoPaimonDataEvolutionTable(
val persistSourceDss: Option[Dataset[Row]] =
if (
table.coreOptions().dataEvolutionMergeIntoSourcePersist()
- && (matchedActions.nonEmpty || notMatchedActions.nonEmpty)
+ && (matchedActions.nonEmpty || notMatchedActions.nonEmpty ||
notMatchedBySourceActions.nonEmpty)
) {
val dss = createDataset(sparkSession, sourceTable)
dss.persist()
@@ -221,29 +242,42 @@ case class MergeIntoPaimonDataEvolutionTable(
None
}
+ var targetActionCleanup = () => ()
try {
- // step 1: find the related data splits, make it target file plan
- val dataSplits: Seq[DataSplit] = targetRelatedSplits(
+ // step 1: find the related data splits for matched / insert actions.
+ lazy val matchedOrInsertDataSplits: Seq[DataSplit] = targetRelatedSplits(
sparkSession,
tableSplits,
firstRowIds,
firstRowIdToBlobFirstRowIds,
persistSourceDss)
- val touchedFileTargetRelation =
- createNewScanPlan(dataSplits, targetRelation)
+ lazy val touchedFileTargetRelation =
+ createNewScanPlan(matchedOrInsertDataSplits, targetRelation)
- // step 2: invoke update action
- val updateCommit =
+ // step 2: invoke target-side action
+ val matchedResult =
if (matchedActions.nonEmpty) {
- val updateResult =
- updateActionInvoke(
- dataSplits,
- sparkSession,
- touchedFileTargetRelation,
- firstRowIds,
- persistSourceDss)
- checkUpdateResult(updateResult)
- } else Nil
+ val result = targetActionInvoke(
+ matchedOrInsertDataSplits,
+ sparkSession,
+ touchedFileTargetRelation,
+ firstRowIds,
+ persistSourceDss)
+ targetActionCleanup = result.cleanup
+ result
+ } else {
+ TargetActionResult(Nil, None, () => ())
+ }
+ val notMatchedBySourceDvs =
+ if (notMatchedBySourceActions.nonEmpty) {
+ notMatchedBySourceDeletionVectors(sparkSession, readSnapshot,
persistSourceDss)
+ } else None
+
+ // Merge deletion vectors generated from NMBS and matched conditions
+ // These two statement branch could generate deletion vectors for same
files!
+ val deleteCommit = persistMergedDeletionVectors(
+ Seq(matchedResult.deletionVectors, notMatchedBySourceDvs).flatten,
+ readSnapshot)
// step 3: invoke insert action
val insertCommit =
@@ -251,11 +285,14 @@ case class MergeIntoPaimonDataEvolutionTable(
insertActionInvoke(sparkSession, touchedFileTargetRelation,
persistSourceDss)
else Nil
- if (plan.snapshotId() != null) {
- writer.rowIdCheckConflict(plan.snapshotId())
+ if (readSnapshot != null) {
+ writer.rowIdCheckConflict(readSnapshot.id())
}
- writer.commit(updateCommit ++ insertCommit, Snapshot.Operation.MERGE)
+ writer.commit(
+ matchedResult.commitMessages ++ deleteCommit ++ insertCommit,
+ Snapshot.Operation.MERGE)
} finally {
+ targetActionCleanup()
if (persistSourceDss.isDefined) {
persistSourceDss.get.unpersist(blocking = false)
}
@@ -349,12 +386,21 @@ case class MergeIntoPaimonDataEvolutionTable(
.map(_.get())
}
- private def updateActionInvoke(
+ private case class TargetActionResult(
+ commitMessages: Seq[CommitMessage],
+ deletionVectors: Option[Dataset[SparkDeletionVector]],
+ cleanup: () => Unit)
+
+ private def targetActionInvoke(
dataSplits: Seq[DataSplit],
sparkSession: SparkSession,
touchedFileTargetRelation: DataSourceV2Relation,
firstRowIds: immutable.IndexedSeq[Long],
- persistSourceDss: Option[Dataset[Row]]): Seq[CommitMessage] = {
+ persistSourceDss: Option[Dataset[Row]]): TargetActionResult = {
+ if (dataSplits.isEmpty) {
+ return TargetActionResult(Nil, None, () => ())
+ }
+
val conditionFields = extractFields(matchedCondition)
val allFields = mutable.SortedSet.empty[AttributeReference](
(o1, o2) => {
@@ -403,27 +449,37 @@ case class MergeIntoPaimonDataEvolutionTable(
val rawBlobMarkerAttributes = rawBlobUpdateColumns.map(
attr =>
AttributeReference(rawBlobMarkerNamesByColumn(attr.name), BooleanType,
nullable = false)())
- val mergeOutput = updateColumnsSorted ++ metadataColumns ++
rawBlobMarkerAttributes
-
- val realUpdateActions = matchedActions
- .map(s => s.asInstanceOf[UpdateAction])
- .map(
- update =>
- UpdateAction.apply(
- update.condition,
- update.assignments.filter(
- a => updateColumnsSorted.contains(assignmentKeyAttribute(a)))))
+ val deletedAttribute =
+ AttributeReference(MERGE_DELETED_NAME, BooleanType, nullable = false)()
+ // Row metadata, _FIRST_ROW_ID added by addFirstRowId, and the delete
marker.
+ val fixedMergeOutputColumnCount = metadataColumns.size + 2
+ val mergeOutput = (updateColumnsSorted ++ metadataColumns ++
rawBlobMarkerAttributes) :+
+ deletedAttribute
+
+ val targetMatchedActions = matchedActions.map {
+ case update: UpdateAction =>
+ UpdateAction.apply(
+ update.condition,
+ update.assignments.filter(a =>
updateColumnsSorted.contains(assignmentKeyAttribute(a))))
+ case delete: DeleteAction => delete
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported matched action:
$other.")
+ }
// All fields are composed by:
// 1. Match condition fields
- // 2. For each update action, the condition fields and the assignment
value fields
+ // 2. For each target-side action, the condition fields and the assignment
value fields
// 3. All updated fields exclude raw blob fields
- for (action <- realUpdateActions) {
+ for (action <- targetMatchedActions) {
action.condition.foreach(condition => allFields ++=
extractFields(condition))
- for (assignment <- action.assignments) {
- if (isModifiedAssignment(assignment)) {
- allFields ++= extractFields(assignment.value)
- }
+ action match {
+ case update: UpdateAction =>
+ for (assignment <- update.assignments) {
+ if (isModifiedAssignment(assignment)) {
+ allFields ++= extractFields(assignment.value)
+ }
+ }
+ case _ =>
}
}
allFields ++= updateColumnsSorted.filterNot(isRawBlobUpdateColumn)
@@ -468,11 +524,11 @@ case class MergeIntoPaimonDataEvolutionTable(
TrueLiteral
}
}
- updatedColumns ++ metadata ++ markers
+ updatedColumns ++ metadata ++ markers :+ FalseLiteral
}
// the output projection for target table copy
- def copyOutput: Seq[Expression] = {
+ def copyOutput(deleted: Boolean): Seq[Expression] = {
val copiedColumns = updateColumnsSorted.map {
attr =>
if (rawBlobUpdateColumns.exists(_.sameRef(attr))) {
@@ -481,22 +537,45 @@ case class MergeIntoPaimonDataEvolutionTable(
attr
}
}
- copiedColumns ++ metadataColumns ++ rawBlobUpdateColumns.map(_ =>
TrueLiteral)
+ val deletedLiteral = if (deleted) TrueLiteral else FalseLiteral
+ copiedColumns ++ metadataColumns ++ rawBlobUpdateColumns.map(_ =>
TrueLiteral) :+
+ deletedLiteral
}
def reorderPartialWriteColumns(dataset: Dataset[Row]): Dataset[Row] = {
- if (rawBlobMarkerAttributes.isEmpty) {
- dataset
- } else {
- val columns =
- updateColumnsSorted.map(attr => quotedColumn(attr.name)) ++
- Seq(quotedColumn(ROW_ID_NAME), quotedColumn(FIRST_ROW_ID_NAME)) ++
- rawBlobMarkerAttributes.map(attr => quotedColumn(attr.name))
- dataset.select(columns: _*)
+ val columns =
+ updateColumnsSorted.map(attr => quotedColumn(attr.name)) ++
+ Seq(quotedColumn(ROW_ID_NAME), quotedColumn(FIRST_ROW_ID_NAME)) ++
+ rawBlobMarkerAttributes.map(attr => quotedColumn(attr.name))
+ dataset.select(columns: _*)
+ }
+
+ val keepCopyInstructions = Seq(
+ SparkShimLoader.shim
+ .mergeRowsKeepCopy(TrueLiteral, copyOutput(deleted = false))
+ .asInstanceOf[MergeRows.Instruction])
+
+ def matchedActionInstruction(action: MergeAction): MergeRows.Instruction =
{
+ action match {
+ case update: UpdateAction =>
+ SparkShimLoader.shim
+ .mergeRowsKeepUpdate(
+ update.condition.getOrElse(TrueLiteral),
+ updateOutput(update, modifiedRawBlobNames(update)))
+ .asInstanceOf[MergeRows.Instruction]
+ case DeleteAction(condition) =>
+ SparkShimLoader.shim
+ .mergeRowsKeepUpdate(condition.getOrElse(TrueLiteral),
copyOutput(deleted = true))
+ .asInstanceOf[MergeRows.Instruction]
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
}
- val toWrite = if (isSelfMergeOnRowId) {
+ def matchedActionInstructions(actions: Seq[MergeAction]):
Seq[MergeRows.Instruction] =
+ actions.map(matchedActionInstruction) ++ keepCopyInstructions
+
+ val targetActionOutput: Dataset[Row] = if (isSelfMergeOnRowId) {
// Self-Merge shortcut:
// - Scan the target table only (no source scan, no join), and read all
columns required by
// merge condition and update expressions.
@@ -532,47 +611,34 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
- val rawBlobModifiedByAction = realUpdateActions.map(modifiedRawBlobNames)
-
- val rewrittenUpdateActions: Seq[UpdateAction] = realUpdateActions.map {
- ua =>
- val newCond = ua.condition.map(c => rewriteSourceToTarget(c,
sourceToTarget))
- val newAssignments = ua.assignments.map {
+ val rewrittenMatchedActions: Seq[MergeAction] = targetMatchedActions.map
{
+ case action: UpdateAction =>
+ val newCond = action.condition.map(c => rewriteSourceToTarget(c,
sourceToTarget))
+ val newAssignments = action.assignments.map {
a => Assignment(a.key, rewriteSourceToTarget(a.value,
sourceToTarget))
}
- ua.copy(condition = newCond, assignments = newAssignments)
+ action.copy(condition = newCond, assignments = newAssignments)
+ case DeleteAction(condition) =>
+ DeleteAction(condition.map(c => rewriteSourceToTarget(c,
sourceToTarget)))
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
val mergeRows = MergeRows(
isSourceRowPresent = TrueLiteral,
isTargetRowPresent = TrueLiteral,
- matchedInstructions = rewrittenUpdateActions
- .zip(rawBlobModifiedByAction)
- .map {
- case (action, rawBlobModified) =>
- SparkShimLoader.shim
- .mergeRowsKeepUpdate(
- action.condition.getOrElse(TrueLiteral),
- updateOutput(action, rawBlobModified))
- .asInstanceOf[MergeRows.Instruction]
- } ++ Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]),
+ matchedInstructions =
matchedActionInstructions(rewrittenMatchedActions),
notMatchedInstructions = Nil,
- notMatchedBySourceInstructions = Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]),
+ notMatchedBySourceInstructions = keepCopyInstructions,
checkCardinality = false,
output = mergeOutput,
child = readPlan
)
- val withFirstRowId = reorderPartialWriteColumns(
- addFirstRowId(sparkSession, mergeRows, firstRowIds))
- assert(
- withFirstRowId.schema.fields.length == updateColumnsSorted.size + 2 +
rawBlobUpdateColumns.size)
+ val withFirstRowId = addFirstRowId(sparkSession, mergeRows, firstRowIds)
+ val expectedOutputColumnCount =
+ updateColumnsSorted.size + fixedMergeOutputColumnCount +
rawBlobUpdateColumns.size
+ assert(withFirstRowId.schema.fields.length == expectedOutputColumnCount)
withFirstRowId
} else {
val allReadFieldsOnTarget = allFields.filter(
@@ -596,44 +662,163 @@ case class MergeIntoPaimonDataEvolutionTable(
Join(targetTableProj, sourceTableProj, LeftOuter,
Some(matchedCondition), JoinHint.NONE)
val rowFromSourceAttr = attribute(ROW_FROM_SOURCE, joinPlan)
val rowFromTargetAttr = attribute(ROW_FROM_TARGET, joinPlan)
+
val mergeRows = MergeRows(
isSourceRowPresent = rowFromSourceAttr,
isTargetRowPresent = rowFromTargetAttr,
- matchedInstructions = realUpdateActions
- .map(
- action => {
- SparkShimLoader.shim
- .mergeRowsKeepUpdate(
- action.condition.getOrElse(TrueLiteral),
- updateOutput(action, modifiedRawBlobNames(action)))
- .asInstanceOf[MergeRows.Instruction]
- }) ++ Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]),
+ matchedInstructions = matchedActionInstructions(targetMatchedActions),
notMatchedInstructions = Nil,
- notMatchedBySourceInstructions = Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]).toSeq,
+ notMatchedBySourceInstructions = keepCopyInstructions,
checkCardinality = false,
output = mergeOutput,
child = joinPlan
)
- val withFirstRowId = reorderPartialWriteColumns(
- addFirstRowId(sparkSession, mergeRows, firstRowIds))
- assert(
- withFirstRowId.schema.fields.length == updateColumnsSorted.size + 2 +
rawBlobUpdateColumns.size)
+ val withFirstRowId = addFirstRowId(sparkSession, mergeRows, firstRowIds)
+ val expectedOutputColumnCount =
+ updateColumnsSorted.size + fixedMergeOutputColumnCount +
rawBlobUpdateColumns.size
+ assert(withFirstRowId.schema.fields.length == expectedOutputColumnCount)
withFirstRowId
.repartition(col(FIRST_ROW_ID_NAME))
.sortWithinPartitions(FIRST_ROW_ID_NAME, ROW_ID_NAME)
}
- val writer = DataEvolutionPaimonWriter(table, dataSplits)
- writer.writePartialFields(
- toWrite,
- updateColumnsSorted.map(_.name),
- rawBlobUpdateColumns.map(attr => attr.name ->
rawBlobMarkerNamesByColumn(attr.name)).toMap)
+ val writePartialFields = updateColumnsSorted.nonEmpty
+ val shouldPersistOutput = writePartialFields && hasMatchedDeleteActions
+ val output: Dataset[Row] =
+ if (shouldPersistOutput) targetActionOutput.persist()
+ else targetActionOutput
+ var success = false
+ try {
+ val updateCommit =
+ if (writePartialFields) {
+ val dataEvolutionWriter = DataEvolutionPaimonWriter(table,
dataSplits)
+ val partialCommit = dataEvolutionWriter.writePartialFields(
+ reorderPartialWriteColumns(output),
+ updateColumnsSorted.map(_.name),
+ rawBlobUpdateColumns
+ .map(attr => attr.name -> rawBlobMarkerNamesByColumn(attr.name))
+ .toMap
+ )
+ checkUpdateResult(partialCommit)
+ } else {
+ Nil
+ }
+
+ val deletionVectors =
+ if (hasMatchedDeleteActions) {
+ val dataFilePathToMeta = candidateFileMap(dataSplits)
+ Some(
+ collectDataEvolutionDeletionVectors(
+ dataSplits,
+ dataFilePathToMeta,
+
output.filter(quotedColumn(MERGE_DELETED_NAME)).select(quotedColumn(ROW_ID_NAME)),
+ sparkSession))
+ } else {
+ None
+ }
+
+ success = true
+ TargetActionResult(
+ updateCommit,
+ deletionVectors,
+ if (shouldPersistOutput) () => output.unpersist(blocking = false) else
() => ())
+ } finally {
+ if (!success && shouldPersistOutput) {
+ output.unpersist(blocking = false)
+ }
+ }
+ }
+
+ private def notMatchedBySourceDeletionVectors(
+ sparkSession: SparkSession,
+ readSnapshot: Snapshot,
+ persistSourceDss: Option[Dataset[Row]]):
Option[Dataset[SparkDeletionVector]] = {
+ // Do not reuse tableSplits here because they may have been pruned by the
MERGE ON
+ // partition filter. NOT MATCHED BY SOURCE must scan target rows
independently.
+ val snapshotReader = table.newSnapshotReader()
+ if (readSnapshot != null) {
+ snapshotReader.withSnapshot(readSnapshot)
+ }
+ if (table.coreOptions().manifestDeleteFileDropStats()) {
+ snapshotReader.dropStats()
+ }
+ val dataSplits =
+ snapshotReader.read().splits().asScala.collect { case split: DataSplit
=> split }.toSeq
+ if (dataSplits.isEmpty) {
+ return None
+ }
+
+ val rowIdMetadataColumn = (targetRelation.output ++
targetRelation.metadataOutput)
+ .filter(attr => attr.name.equals(ROW_ID_NAME))
+ .groupBy(_.exprId)
+ .map { case (_, attrs) => attrs.head }
+ .head
+
+ // No need to project source/target, spark optimizer will take care of it
+ val deleteCondition = notMatchedBySourceDeleteCondition
+ val targetReadPlan = createNewScanPlan(dataSplits, targetRelation)
+ .copy(output = targetRelation.output :+ rowIdMetadataColumn)
+ val targetFilteredPlan =
+ if (deleteCondition == TrueLiteral) targetReadPlan
+ else Filter(deleteCondition, targetReadPlan)
+ val sourceChild =
persistSourceDss.map(_.queryExecution.logical).getOrElse(sourceTable)
+ val antiJoinPlan =
+ Join(targetFilteredPlan, sourceChild, LeftAnti, Some(matchedCondition),
JoinHint.NONE)
+ val deletePlan = Project(Seq(rowIdMetadataColumn), antiJoinPlan)
+
+ val dataFilePathToMeta = candidateFileMap(dataSplits)
+ val deletionVectors = collectDataEvolutionDeletionVectors(
+ dataSplits,
+ dataFilePathToMeta,
+ createDataset(sparkSession,
deletePlan).select(quotedColumn(ROW_ID_NAME)),
+ sparkSession)
+ Some(deletionVectors)
+ }
+
+ private def persistMergedDeletionVectors(
+ deletionVectorDatasets: Seq[Dataset[SparkDeletionVector]],
+ readSnapshot: Snapshot): Seq[CommitMessage] = {
+ if (deletionVectorDatasets.isEmpty) {
+ return Nil
+ }
+
+ val sparkSession = deletionVectorDatasets.head.sparkSession
+ import sparkSession.implicits._
+ val deletionVectors = deletionVectorDatasets
+ .reduce(_.union(_))
+ .groupByKey(_.dataFilePath)
+ .mapGroups {
+ (_, iter) =>
+ var first: SparkDeletionVector = null
+ var merged: DeletionVector = null
+ while (iter.hasNext) {
+ val current = iter.next()
+ val deletionVector =
DeletionVector.deserializeFromBytes(current.deletionVector)
+ if (first == null) {
+ first = current
+ merged = deletionVector
+ } else {
+ merged.merge(deletionVector)
+ }
+ }
+
+ SparkDeletionVector(
+ first.bucketPath,
+ first.partition,
+ first.bucket,
+ first.dataFilePath,
+ DeletionVector.serializeToBytes(merged))
+ }
+
+ writer.persistDeletionVectors(deletionVectors, readSnapshot)
+ }
+
+ private def notMatchedBySourceDeleteCondition: Expression = {
+ if (notMatchedBySourceActions.exists(_.condition.isEmpty)) {
+ TrueLiteral
+ } else {
+ notMatchedBySourceActions.flatMap(_.condition).reduce(Or)
+ }
}
private def insertActionInvoke(
@@ -670,6 +855,8 @@ case class MergeIntoPaimonDataEvolutionTable(
else Literal(null))
)
.asInstanceOf[MergeRows.Instruction]
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported not matched
action: $other.")
}.toSeq,
notMatchedBySourceInstructions = Nil,
checkCardinality = false,
@@ -836,6 +1023,7 @@ object MergeIntoPaimonDataEvolutionTable {
final private val ROW_FROM_TARGET = "__row_from_target"
final private val ROW_ID_NAME = "_ROW_ID"
final private val FIRST_ROW_ID_NAME = "_FIRST_ROW_ID";
+ final private val MERGE_DELETED_NAME = "_PAIMON_MERGE_INTO_DELETED"
final private val RAW_BLOB_PLACEHOLDER_MARKER_PREFIX =
"__paimon_raw_blob_placeholder_"
private[commands] def withMatchedUpdateScanOptions(
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 15379a1147..5e28c056f2 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
import org.apache.paimon.Snapshot
import org.apache.paimon.data.BinaryRow
+import org.apache.paimon.deletionvectors.DeletionVector
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.index.GlobalIndexMeta
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
@@ -71,23 +72,26 @@ case class MergeIntoPaimonDataEvolutionTable(
matchedActions: Seq[MergeAction],
notMatchedActions: Seq[MergeAction],
notMatchedBySourceActions: Seq[MergeAction])
- extends PaimonLeafRunnableCommand
- with WithFileStoreTable
- with ExpressionHelper
+ extends PaimonRowLevelCommand
with Logging {
- private lazy val writer = PaimonSparkWriter(table)
-
- assert(
- notMatchedBySourceActions.isEmpty,
- "notMatchedBySourceActions is not supported in
MergeIntoPaimonDataEvolutionTable.")
assert(
- matchedActions.forall(x => x.isInstanceOf[UpdateAction]),
- "Only SET clause is supported in MergeIntoPaimonDataEvolutionTable for
SQL: WHEN MATCHED.")
+ matchedActions.forall {
+ case _: UpdateAction | _: DeleteAction => true
+ case _ => false
+ },
+ "Only SET and DELETE clauses are supported in
MergeIntoPaimonDataEvolutionTable for SQL: " +
+ "WHEN MATCHED."
+ )
assert(
notMatchedActions.forall(x => x.isInstanceOf[InsertAction]),
"Only INSERT clause is supported in MergeIntoPaimonDataEvolutionTable for
SQL: WHEN NOT MATCHED."
)
+ assert(
+ notMatchedBySourceActions.forall(x => x.isInstanceOf[DeleteAction]),
+ "Only DELETE clause is supported in MergeIntoPaimonDataEvolutionTable for
SQL: WHEN NOT " +
+ "MATCHED BY SOURCE."
+ )
import MergeIntoPaimonDataEvolutionTable._
@@ -105,6 +109,18 @@ case class MergeIntoPaimonDataEvolutionTable(
override lazy val table: FileStoreTable =
targetSparkTable.getTable.asInstanceOf[FileStoreTable]
+ private lazy val hasTargetSideDeleteActions: Boolean =
+ matchedActions.exists(_.isInstanceOf[DeleteAction]) ||
+ notMatchedBySourceActions.exists(_.isInstanceOf[DeleteAction])
+
+ private lazy val hasMatchedDeleteActions: Boolean =
+ matchedActions.exists(_.isInstanceOf[DeleteAction])
+
+ assert(
+ !hasTargetSideDeleteActions ||
table.coreOptions().deletionVectorsEnabled(),
+ "MERGE INTO DELETE on DataEvolution table requires
deletion-vectors.enabled=true."
+ )
+
private val updateColumns: Set[AttributeReference] = {
val columns = mutable.Set[AttributeReference]()
for (action <- matchedActions) {
@@ -115,6 +131,7 @@ case class MergeIntoPaimonDataEvolutionTable(
columns += assignmentKeyAttribute(assignment)
}
}
+ case _ =>
}
}
columns.toSet
@@ -153,8 +170,8 @@ case class MergeIntoPaimonDataEvolutionTable(
assert(
!(isSelfMergeOnRowId && (notMatchedActions.nonEmpty ||
notMatchedBySourceActions.nonEmpty)),
- "Self-Merge on _ROW_ID only supports WHEN MATCHED THEN UPDATE. WHEN NOT
MATCHED and WHEN " +
- "NOT MATCHED BY SOURCE are not supported."
+ "Self-Merge on _ROW_ID only supports WHEN MATCHED actions. WHEN NOT
MATCHED and " +
+ "WHEN NOT MATCHED BY SOURCE are not supported."
)
private lazy val targetRelation: DataSourceV2Relation =
matchedUpdateScanTarget._2
@@ -169,7 +186,11 @@ case class MergeIntoPaimonDataEvolutionTable(
}
private def invokeMergeInto(sparkSession: SparkSession): Unit = {
+ val readSnapshot = table.snapshotManager().latestSnapshot()
val snapshotReader = table.newSnapshotReader()
+ if (readSnapshot != null) {
+ snapshotReader.withSnapshot(readSnapshot)
+ }
pushDownMergePartitionFilter(snapshotReader)
val plan = snapshotReader.read()
val tableSplits: Seq[DataSplit] = plan
@@ -212,7 +233,7 @@ case class MergeIntoPaimonDataEvolutionTable(
val persistSourceDss: Option[Dataset[Row]] =
if (
table.coreOptions().dataEvolutionMergeIntoSourcePersist()
- && (matchedActions.nonEmpty || notMatchedActions.nonEmpty)
+ && (matchedActions.nonEmpty || notMatchedActions.nonEmpty ||
notMatchedBySourceActions.nonEmpty)
) {
val dss = createDataset(sparkSession, sourceTable)
dss.persist()
@@ -221,28 +242,42 @@ case class MergeIntoPaimonDataEvolutionTable(
None
}
+ var targetActionCleanup = () => ()
try {
- // step 1: find the related data splits, make it target file plan
- val dataSplits: Seq[DataSplit] = targetRelatedSplits(
+ // step 1: find the related data splits for matched / insert actions.
+ lazy val matchedOrInsertDataSplits: Seq[DataSplit] = targetRelatedSplits(
sparkSession,
tableSplits,
firstRowIds,
firstRowIdToBlobFirstRowIds,
persistSourceDss)
- val touchedFileTargetRelation =
- createNewScanPlan(dataSplits, targetRelation)
+ lazy val touchedFileTargetRelation =
+ createNewScanPlan(matchedOrInsertDataSplits, targetRelation)
- // step 2: invoke update action
- val updateCommit =
+ // step 2: invoke target-side action
+ val matchedResult =
if (matchedActions.nonEmpty) {
- val updateResult = updateActionInvoke(
- dataSplits,
+ val result = targetActionInvoke(
+ matchedOrInsertDataSplits,
sparkSession,
touchedFileTargetRelation,
firstRowIds,
persistSourceDss)
- checkUpdateResult(updateResult)
- } else Nil
+ targetActionCleanup = result.cleanup
+ result
+ } else {
+ TargetActionResult(Nil, None, () => ())
+ }
+ val notMatchedBySourceDvs =
+ if (notMatchedBySourceActions.nonEmpty) {
+ notMatchedBySourceDeletionVectors(sparkSession, readSnapshot,
persistSourceDss)
+ } else None
+
+ // Merge deletion vectors generated from NMBS and matched conditions
+ // These two statement branch could generate deletion vectors for same
files!
+ val deleteCommit = persistMergedDeletionVectors(
+ Seq(matchedResult.deletionVectors, notMatchedBySourceDvs).flatten,
+ readSnapshot)
// step 3: invoke insert action
val insertCommit =
@@ -250,11 +285,14 @@ case class MergeIntoPaimonDataEvolutionTable(
insertActionInvoke(sparkSession, touchedFileTargetRelation,
persistSourceDss)
else Nil
- if (plan.snapshotId() != null) {
- writer.rowIdCheckConflict(plan.snapshotId())
+ if (readSnapshot != null) {
+ writer.rowIdCheckConflict(readSnapshot.id())
}
- writer.commit(updateCommit ++ insertCommit, Snapshot.Operation.MERGE)
+ writer.commit(
+ matchedResult.commitMessages ++ deleteCommit ++ insertCommit,
+ Snapshot.Operation.MERGE)
} finally {
+ targetActionCleanup()
if (persistSourceDss.isDefined) {
persistSourceDss.get.unpersist(blocking = false)
}
@@ -348,12 +386,21 @@ case class MergeIntoPaimonDataEvolutionTable(
.map(_.get())
}
- private def updateActionInvoke(
+ private case class TargetActionResult(
+ commitMessages: Seq[CommitMessage],
+ deletionVectors: Option[Dataset[SparkDeletionVector]],
+ cleanup: () => Unit)
+
+ private def targetActionInvoke(
dataSplits: Seq[DataSplit],
sparkSession: SparkSession,
touchedFileTargetRelation: DataSourceV2Relation,
firstRowIds: immutable.IndexedSeq[Long],
- persistSourceDss: Option[Dataset[Row]]): Seq[CommitMessage] = {
+ persistSourceDss: Option[Dataset[Row]]): TargetActionResult = {
+ if (dataSplits.isEmpty) {
+ return TargetActionResult(Nil, None, () => ())
+ }
+
val conditionFields = extractFields(matchedCondition)
val allFields = mutable.SortedSet.empty[AttributeReference](
(o1, o2) => {
@@ -402,27 +449,37 @@ case class MergeIntoPaimonDataEvolutionTable(
val rawBlobMarkerAttributes = rawBlobUpdateColumns.map(
attr =>
AttributeReference(rawBlobMarkerNamesByColumn(attr.name), BooleanType,
nullable = false)())
- val mergeOutput = updateColumnsSorted ++ metadataColumns ++
rawBlobMarkerAttributes
-
- val realUpdateActions = matchedActions
- .map(s => s.asInstanceOf[UpdateAction])
- .map(
- update =>
- UpdateAction.apply(
- update.condition,
- update.assignments.filter(
- a => updateColumnsSorted.contains(assignmentKeyAttribute(a)))))
+ val deletedAttribute =
+ AttributeReference(MERGE_DELETED_NAME, BooleanType, nullable = false)()
+ // Row metadata, _FIRST_ROW_ID added by addFirstRowId, and the delete
marker.
+ val fixedMergeOutputColumnCount = metadataColumns.size + 2
+ val mergeOutput = (updateColumnsSorted ++ metadataColumns ++
rawBlobMarkerAttributes) :+
+ deletedAttribute
+
+ val targetMatchedActions = matchedActions.map {
+ case update: UpdateAction =>
+ UpdateAction.apply(
+ update.condition,
+ update.assignments.filter(a =>
updateColumnsSorted.contains(assignmentKeyAttribute(a))))
+ case delete: DeleteAction => delete
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported matched action:
$other.")
+ }
// All fields are composed by:
// 1. Match condition fields
- // 2. For each update action, the condition fields and the assignment
value fields
+ // 2. For each target-side action, the condition fields and the assignment
value fields
// 3. All updated fields exclude raw blob fields
- for (action <- realUpdateActions) {
+ for (action <- targetMatchedActions) {
action.condition.foreach(condition => allFields ++=
extractFields(condition))
- for (assignment <- action.assignments) {
- if (isModifiedAssignment(assignment)) {
- allFields ++= extractFields(assignment.value)
- }
+ action match {
+ case update: UpdateAction =>
+ for (assignment <- update.assignments) {
+ if (isModifiedAssignment(assignment)) {
+ allFields ++= extractFields(assignment.value)
+ }
+ }
+ case _ =>
}
}
allFields ++= updateColumnsSorted.filterNot(isRawBlobUpdateColumn)
@@ -467,11 +524,11 @@ case class MergeIntoPaimonDataEvolutionTable(
TrueLiteral
}
}
- updatedColumns ++ metadata ++ markers
+ updatedColumns ++ metadata ++ markers :+ FalseLiteral
}
// the output projection for target table copy
- def copyOutput: Seq[Expression] = {
+ def copyOutput(deleted: Boolean): Seq[Expression] = {
val copiedColumns = updateColumnsSorted.map {
attr =>
if (rawBlobUpdateColumns.exists(_.sameRef(attr))) {
@@ -480,22 +537,45 @@ case class MergeIntoPaimonDataEvolutionTable(
attr
}
}
- copiedColumns ++ metadataColumns ++ rawBlobUpdateColumns.map(_ =>
TrueLiteral)
+ val deletedLiteral = if (deleted) TrueLiteral else FalseLiteral
+ copiedColumns ++ metadataColumns ++ rawBlobUpdateColumns.map(_ =>
TrueLiteral) :+
+ deletedLiteral
}
def reorderPartialWriteColumns(dataset: Dataset[Row]): Dataset[Row] = {
- if (rawBlobMarkerAttributes.isEmpty) {
- dataset
- } else {
- val columns =
- updateColumnsSorted.map(attr => quotedColumn(attr.name)) ++
- Seq(quotedColumn(ROW_ID_NAME), quotedColumn(FIRST_ROW_ID_NAME)) ++
- rawBlobMarkerAttributes.map(attr => quotedColumn(attr.name))
- dataset.select(columns: _*)
+ val columns =
+ updateColumnsSorted.map(attr => quotedColumn(attr.name)) ++
+ Seq(quotedColumn(ROW_ID_NAME), quotedColumn(FIRST_ROW_ID_NAME)) ++
+ rawBlobMarkerAttributes.map(attr => quotedColumn(attr.name))
+ dataset.select(columns: _*)
+ }
+
+ val keepCopyInstructions = Seq(
+ SparkShimLoader.shim
+ .mergeRowsKeepCopy(TrueLiteral, copyOutput(deleted = false))
+ .asInstanceOf[MergeRows.Instruction])
+
+ def matchedActionInstruction(action: MergeAction): MergeRows.Instruction =
{
+ action match {
+ case update: UpdateAction =>
+ SparkShimLoader.shim
+ .mergeRowsKeepUpdate(
+ update.condition.getOrElse(TrueLiteral),
+ updateOutput(update, modifiedRawBlobNames(update)))
+ .asInstanceOf[MergeRows.Instruction]
+ case DeleteAction(condition) =>
+ SparkShimLoader.shim
+ .mergeRowsKeepUpdate(condition.getOrElse(TrueLiteral),
copyOutput(deleted = true))
+ .asInstanceOf[MergeRows.Instruction]
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
}
- val toWrite = if (isSelfMergeOnRowId) {
+ def matchedActionInstructions(actions: Seq[MergeAction]):
Seq[MergeRows.Instruction] =
+ actions.map(matchedActionInstruction) ++ keepCopyInstructions
+
+ val targetActionOutput: Dataset[Row] = if (isSelfMergeOnRowId) {
// Self-Merge shortcut:
// - Scan the target table only (no source scan, no join), and read all
columns required by
// merge condition and update expressions.
@@ -531,47 +611,34 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
- val rawBlobModifiedByAction = realUpdateActions.map(modifiedRawBlobNames)
-
- val rewrittenUpdateActions: Seq[UpdateAction] = realUpdateActions.map {
- ua =>
- val newCond = ua.condition.map(c => rewriteSourceToTarget(c,
sourceToTarget))
- val newAssignments = ua.assignments.map {
+ val rewrittenMatchedActions: Seq[MergeAction] = targetMatchedActions.map
{
+ case action: UpdateAction =>
+ val newCond = action.condition.map(c => rewriteSourceToTarget(c,
sourceToTarget))
+ val newAssignments = action.assignments.map {
a => Assignment(a.key, rewriteSourceToTarget(a.value,
sourceToTarget))
}
- ua.copy(condition = newCond, assignments = newAssignments)
+ action.copy(condition = newCond, assignments = newAssignments)
+ case DeleteAction(condition) =>
+ DeleteAction(condition.map(c => rewriteSourceToTarget(c,
sourceToTarget)))
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
val mergeRows = MergeRows(
isSourceRowPresent = TrueLiteral,
isTargetRowPresent = TrueLiteral,
- matchedInstructions = rewrittenUpdateActions
- .zip(rawBlobModifiedByAction)
- .map {
- case (action, rawBlobModified) =>
- SparkShimLoader.shim
- .mergeRowsKeepUpdate(
- action.condition.getOrElse(TrueLiteral),
- updateOutput(action, rawBlobModified))
- .asInstanceOf[MergeRows.Instruction]
- } ++ Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]),
+ matchedInstructions =
matchedActionInstructions(rewrittenMatchedActions),
notMatchedInstructions = Nil,
- notMatchedBySourceInstructions = Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]),
+ notMatchedBySourceInstructions = keepCopyInstructions,
checkCardinality = false,
output = mergeOutput,
child = readPlan
)
- val withFirstRowId = reorderPartialWriteColumns(
- addFirstRowId(sparkSession, mergeRows, firstRowIds))
- assert(
- withFirstRowId.schema.fields.length == updateColumnsSorted.size + 2 +
rawBlobUpdateColumns.size)
+ val withFirstRowId = addFirstRowId(sparkSession, mergeRows, firstRowIds)
+ val expectedOutputColumnCount =
+ updateColumnsSorted.size + fixedMergeOutputColumnCount +
rawBlobUpdateColumns.size
+ assert(withFirstRowId.schema.fields.length == expectedOutputColumnCount)
withFirstRowId
} else {
val allReadFieldsOnTarget = allFields.filter(
@@ -595,44 +662,163 @@ case class MergeIntoPaimonDataEvolutionTable(
Join(targetTableProj, sourceTableProj, LeftOuter,
Some(matchedCondition), JoinHint.NONE)
val rowFromSourceAttr = attribute(ROW_FROM_SOURCE, joinPlan)
val rowFromTargetAttr = attribute(ROW_FROM_TARGET, joinPlan)
+
val mergeRows = MergeRows(
isSourceRowPresent = rowFromSourceAttr,
isTargetRowPresent = rowFromTargetAttr,
- matchedInstructions = realUpdateActions
- .map(
- action => {
- SparkShimLoader.shim
- .mergeRowsKeepUpdate(
- action.condition.getOrElse(TrueLiteral),
- updateOutput(action, modifiedRawBlobNames(action)))
- .asInstanceOf[MergeRows.Instruction]
- }) ++ Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]),
+ matchedInstructions = matchedActionInstructions(targetMatchedActions),
notMatchedInstructions = Nil,
- notMatchedBySourceInstructions = Seq(
- SparkShimLoader.shim
- .mergeRowsKeepCopy(TrueLiteral, copyOutput)
- .asInstanceOf[MergeRows.Instruction]).toSeq,
+ notMatchedBySourceInstructions = keepCopyInstructions,
checkCardinality = false,
output = mergeOutput,
child = joinPlan
)
- val withFirstRowId = reorderPartialWriteColumns(
- addFirstRowId(sparkSession, mergeRows, firstRowIds))
- assert(
- withFirstRowId.schema.fields.length == updateColumnsSorted.size + 2 +
rawBlobUpdateColumns.size)
+ val withFirstRowId = addFirstRowId(sparkSession, mergeRows, firstRowIds)
+ val expectedOutputColumnCount =
+ updateColumnsSorted.size + fixedMergeOutputColumnCount +
rawBlobUpdateColumns.size
+ assert(withFirstRowId.schema.fields.length == expectedOutputColumnCount)
withFirstRowId
.repartition(col(FIRST_ROW_ID_NAME))
.sortWithinPartitions(FIRST_ROW_ID_NAME, ROW_ID_NAME)
}
- val writer = DataEvolutionPaimonWriter(table, dataSplits)
- writer.writePartialFields(
- toWrite,
- updateColumnsSorted.map(_.name),
- rawBlobUpdateColumns.map(attr => attr.name ->
rawBlobMarkerNamesByColumn(attr.name)).toMap)
+ val writePartialFields = updateColumnsSorted.nonEmpty
+ val shouldPersistOutput = writePartialFields && hasMatchedDeleteActions
+ val output: Dataset[Row] =
+ if (shouldPersistOutput) targetActionOutput.persist()
+ else targetActionOutput
+ var success = false
+ try {
+ val updateCommit =
+ if (writePartialFields) {
+ val dataEvolutionWriter = DataEvolutionPaimonWriter(table,
dataSplits)
+ val partialCommit = dataEvolutionWriter.writePartialFields(
+ reorderPartialWriteColumns(output),
+ updateColumnsSorted.map(_.name),
+ rawBlobUpdateColumns
+ .map(attr => attr.name -> rawBlobMarkerNamesByColumn(attr.name))
+ .toMap
+ )
+ checkUpdateResult(partialCommit)
+ } else {
+ Nil
+ }
+
+ val deletionVectors =
+ if (hasMatchedDeleteActions) {
+ val dataFilePathToMeta = candidateFileMap(dataSplits)
+ Some(
+ collectDataEvolutionDeletionVectors(
+ dataSplits,
+ dataFilePathToMeta,
+
output.filter(quotedColumn(MERGE_DELETED_NAME)).select(quotedColumn(ROW_ID_NAME)),
+ sparkSession))
+ } else {
+ None
+ }
+
+ success = true
+ TargetActionResult(
+ updateCommit,
+ deletionVectors,
+ if (shouldPersistOutput) () => output.unpersist(blocking = false) else
() => ())
+ } finally {
+ if (!success && shouldPersistOutput) {
+ output.unpersist(blocking = false)
+ }
+ }
+ }
+
+ private def notMatchedBySourceDeletionVectors(
+ sparkSession: SparkSession,
+ readSnapshot: Snapshot,
+ persistSourceDss: Option[Dataset[Row]]):
Option[Dataset[SparkDeletionVector]] = {
+ // Do not reuse tableSplits here because they may have been pruned by the
MERGE ON
+ // partition filter. NOT MATCHED BY SOURCE must scan target rows
independently.
+ val snapshotReader = table.newSnapshotReader()
+ if (readSnapshot != null) {
+ snapshotReader.withSnapshot(readSnapshot)
+ }
+ if (table.coreOptions().manifestDeleteFileDropStats()) {
+ snapshotReader.dropStats()
+ }
+ val dataSplits =
+ snapshotReader.read().splits().asScala.collect { case split: DataSplit
=> split }.toSeq
+ if (dataSplits.isEmpty) {
+ return None
+ }
+
+ val rowIdMetadataColumn = (targetRelation.output ++
targetRelation.metadataOutput)
+ .filter(attr => attr.name.equals(ROW_ID_NAME))
+ .groupBy(_.exprId)
+ .map { case (_, attrs) => attrs.head }
+ .head
+
+ // No need to project source/target, spark optimizer will take care of it
+ val deleteCondition = notMatchedBySourceDeleteCondition
+ val targetReadPlan = createNewScanPlan(dataSplits, targetRelation)
+ .copy(output = targetRelation.output :+ rowIdMetadataColumn)
+ val targetFilteredPlan =
+ if (deleteCondition == TrueLiteral) targetReadPlan
+ else Filter(deleteCondition, targetReadPlan)
+ val sourceChild =
persistSourceDss.map(_.queryExecution.logical).getOrElse(sourceTable)
+ val antiJoinPlan =
+ Join(targetFilteredPlan, sourceChild, LeftAnti, Some(matchedCondition),
JoinHint.NONE)
+ val deletePlan = Project(Seq(rowIdMetadataColumn), antiJoinPlan)
+
+ val dataFilePathToMeta = candidateFileMap(dataSplits)
+ val deletionVectors = collectDataEvolutionDeletionVectors(
+ dataSplits,
+ dataFilePathToMeta,
+ createDataset(sparkSession,
deletePlan).select(quotedColumn(ROW_ID_NAME)),
+ sparkSession)
+ Some(deletionVectors)
+ }
+
+ private def persistMergedDeletionVectors(
+ deletionVectorDatasets: Seq[Dataset[SparkDeletionVector]],
+ readSnapshot: Snapshot): Seq[CommitMessage] = {
+ if (deletionVectorDatasets.isEmpty) {
+ return Nil
+ }
+
+ val sparkSession = deletionVectorDatasets.head.sparkSession
+ import sparkSession.implicits._
+ val deletionVectors = deletionVectorDatasets
+ .reduce(_.union(_))
+ .groupByKey(_.dataFilePath)
+ .mapGroups {
+ (_, iter) =>
+ var first: SparkDeletionVector = null
+ var merged: DeletionVector = null
+ while (iter.hasNext) {
+ val current = iter.next()
+ val deletionVector =
DeletionVector.deserializeFromBytes(current.deletionVector)
+ if (first == null) {
+ first = current
+ merged = deletionVector
+ } else {
+ merged.merge(deletionVector)
+ }
+ }
+
+ SparkDeletionVector(
+ first.bucketPath,
+ first.partition,
+ first.bucket,
+ first.dataFilePath,
+ DeletionVector.serializeToBytes(merged))
+ }
+
+ writer.persistDeletionVectors(deletionVectors, readSnapshot)
+ }
+
+ private def notMatchedBySourceDeleteCondition: Expression = {
+ if (notMatchedBySourceActions.exists(_.condition.isEmpty)) {
+ TrueLiteral
+ } else {
+ notMatchedBySourceActions.flatMap(_.condition).reduce(Or)
+ }
}
private def insertActionInvoke(
@@ -669,6 +855,8 @@ case class MergeIntoPaimonDataEvolutionTable(
else Literal(null))
)
.asInstanceOf[MergeRows.Instruction]
+ case other =>
+ throw new UnsupportedOperationException(s"Unsupported not matched
action: $other.")
}.toSeq,
notMatchedBySourceInstructions = Nil,
checkCardinality = false,
@@ -835,6 +1023,7 @@ object MergeIntoPaimonDataEvolutionTable {
final private val ROW_FROM_TARGET = "__row_from_target"
final private val ROW_ID_NAME = "_ROW_ID"
final private val FIRST_ROW_ID_NAME = "_FIRST_ROW_ID";
+ final private val MERGE_DELETED_NAME = "_PAIMON_MERGE_INTO_DELETED"
final private val RAW_BLOB_PLACEHOLDER_MARKER_PREFIX =
"__paimon_raw_blob_placeholder_"
private[commands] def withMatchedUpdateScanOptions(
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala
index 3fad7fb4df..a99c7b36ab 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonRowLevelCommand.scala
@@ -196,6 +196,17 @@ trait PaimonRowLevelCommand
sparkSession)
}
+ protected def collectDataEvolutionDeletionVectors(
+ candidateDataSplits: Seq[DataSplit],
+ dataFilePathToMeta: Map[String, SparkDataFileMeta],
+ dataset: Dataset[Row],
+ sparkSession: SparkSession): Dataset[SparkDeletionVector] = {
+ buildDeletionVectors(
+ dataFilePathToMeta,
+ dataEvolutionDeletionTargets(candidateDataSplits, dataset, sparkSession),
+ sparkSession)
+ }
+
private def pathAndIndexDeletionTargets(
dataset: Dataset[Row],
sparkSession: SparkSession):
Dataset[PaimonRowLevelCommand.DeletionTarget] = {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTestBase.scala
index 5d2ba1d697..24f44a7527 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataEvolutionDeletionTestBase.scala
@@ -199,4 +199,385 @@ abstract class DataEvolutionDeletionTestBase extends
PaimonSparkTestBase {
Seq(Row(5, 5, 5, 5L), Row(7, 107, 7, 7L), Row(8, 8, 8, 8L)))
}
}
+
+ test("Data Evolution deletion: self merge delete by row id") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+
+ sql("""
+ |MERGE INTO t
+ |USING t AS source
+ |ON t._ROW_ID = source._ROW_ID
+ |WHEN MATCHED AND source.id IN (2, 6, 9) THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, c, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, 0, 0L),
+ Row(1, 1, 1, 1L),
+ Row(3, 3, 3, 3L),
+ Row(4, 4, 4, 4L),
+ Row(5, 5, 5, 5L),
+ Row(7, 7, 7, 7L),
+ Row(8, 8, 8, 8L))
+ )
+ }
+ }
+
+ test("Data Evolution deletion: merge matched delete writes deletion vectors
only") {
+ withTable("s", "t") {
+ sql("CREATE TABLE s (id INT)")
+ sql("INSERT INTO s VALUES (1), (6), (8), (99)")
+
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+
+ val fileCountBefore = sql("SELECT COUNT(*) FROM `t$files` WHERE
file_path NOT LIKE '%.blob'")
+ .collect()
+ .head
+ .getLong(0)
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN DELETE
+ |""".stripMargin)
+
+ val fileCountAfter = sql("SELECT COUNT(*) FROM `t$files` WHERE file_path
NOT LIKE '%.blob'")
+ .collect()
+ .head
+ .getLong(0)
+ assert(fileCountAfter == fileCountBefore)
+
+ checkAnswer(
+ sql("SELECT id, b, c, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, 0, 0L),
+ Row(2, 2, 2, 2L),
+ Row(3, 3, 3, 3L),
+ Row(4, 4, 4, 4L),
+ Row(5, 5, 5, 5L),
+ Row(7, 7, 7, 7L),
+ Row(9, 9, 9, 9L))
+ )
+ }
+ }
+
+ test("Data Evolution deletion: merge matched update and delete keeps
first-match semantics") {
+ withTable("s", "t") {
+ sql("CREATE TABLE s (id INT, new_b INT, op STRING)")
+ sql("""
+ |INSERT INTO s VALUES
+ | (1, 100, 'update'),
+ | (2, 200, 'delete'),
+ | (6, 600, 'update'),
+ | (11, 1100, 'delete')
+ |""".stripMargin)
+
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED AND s.op = 'delete' THEN DELETE
+ |WHEN MATCHED AND s.id IN (1, 2, 6) THEN UPDATE SET t.b = s.new_b
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, c, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, 0, 0L),
+ Row(1, 100, 1, 1L),
+ Row(3, 3, 3, 3L),
+ Row(4, 4, 4, 4L),
+ Row(5, 5, 5, 5L),
+ Row(6, 600, 6, 6L),
+ Row(7, 7, 7, 7L),
+ Row(8, 8, 8, 8L),
+ Row(9, 9, 9, 9L))
+ )
+ }
+ }
+
+ test("Data Evolution deletion: merge not matched by source delete") {
+ assume(gteqSpark3_4)
+
+ withTable("s", "t") {
+ sql("CREATE TABLE s (id INT)")
+ sql("INSERT INTO s VALUES (1), (3)")
+
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 6)")
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN NOT MATCHED BY SOURCE THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, c, _ROW_ID FROM t ORDER BY id"),
+ Seq(Row(1, 1, 1, 1L), Row(3, 3, 3, 3L)))
+ }
+ }
+
+ test("Data Evolution deletion: merge update insert and not matched by source
delete") {
+ assume(gteqSpark3_4)
+
+ withTable("s", "t") {
+ sql("CREATE TABLE s (id INT, new_b INT)")
+ sql("INSERT INTO s VALUES (1, 100), (4, 400)")
+
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 4)")
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.b = s.new_b
+ |WHEN NOT MATCHED THEN INSERT (id, b, c) VALUES (s.id, s.new_b,
s.new_b)
+ |WHEN NOT MATCHED BY SOURCE THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(sql("SELECT id, b, c FROM t ORDER BY id"), Seq(Row(1, 100,
1), Row(4, 400, 400)))
+ }
+ }
+
+ test("Data Evolution deletion: repeated complex merge on table with deletion
vectors") {
+ assume(gteqSpark3_4)
+
+ withTempView("s") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(10, 12)")
+
+ sql("""
+ |CREATE OR REPLACE TEMP VIEW s AS
+ |SELECT * FROM VALUES
+ | (1, 100, 'update'),
+ | (2, 200, 'delete'),
+ | (12, 1200, 'insert')
+ |AS s(id, new_b, op)
+ |""".stripMargin)
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED AND s.op = 'delete' THEN DELETE
+ |WHEN MATCHED AND s.op = 'update' THEN UPDATE SET t.b = s.new_b
+ |WHEN NOT MATCHED THEN INSERT (id, b, c) VALUES (s.id, s.new_b,
s.new_b + 1000)
+ |WHEN NOT MATCHED BY SOURCE AND t.id IN (3, 8) THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, c FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, 0),
+ Row(1, 100, 1),
+ Row(4, 4, 4),
+ Row(5, 5, 5),
+ Row(6, 6, 6),
+ Row(7, 7, 7),
+ Row(9, 9, 9),
+ Row(10, 10, 10),
+ Row(11, 11, 11),
+ Row(12, 1200, 2200))
+ )
+
+ sql("""
+ |CREATE OR REPLACE TEMP VIEW s AS
+ |SELECT * FROM VALUES
+ | (1, 101, 'update'),
+ | (2, 202, 'update'),
+ | (4, 400, 'delete'),
+ | (13, 1300, 'insert')
+ |AS s(id, new_b, op)
+ |""".stripMargin)
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED AND s.op = 'delete' THEN DELETE
+ |WHEN MATCHED AND s.op = 'update' THEN UPDATE SET t.b = s.new_b
+ |WHEN NOT MATCHED THEN INSERT (id, b, c) VALUES (s.id, s.new_b,
s.new_b + 1000)
+ |WHEN NOT MATCHED BY SOURCE AND t.id IN (5, 10) THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, c FROM t ORDER BY id, b, c"),
+ Seq(
+ Row(0, 0, 0),
+ Row(1, 101, 1),
+ Row(2, 202, 1202),
+ Row(6, 6, 6),
+ Row(7, 7, 7),
+ Row(9, 9, 9),
+ Row(11, 11, 11),
+ Row(12, 1200, 2200),
+ Row(13, 1300, 2300))
+ )
+ }
+ }
+ }
+
+ test("Data Evolution deletion: not matched by source delete does not trigger
partial copy") {
+ assume(gteqSpark3_4)
+
+ withTable("s", "t") {
+ sql("CREATE TABLE s (id INT, new_b INT)")
+ sql("INSERT INTO s VALUES (1, 100)")
+
+ sql("""
+ |CREATE TABLE t (id INT, b INT, c INT)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(0, 5)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(5, 10)")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(10, 15)")
+
+ val thirdRangeFileCountBefore =
+ sql(
+ "SELECT COUNT(*) FROM `t$files` WHERE first_row_id = 10 AND
file_path NOT LIKE '%.blob'")
+ .collect()
+ .head
+ .getLong(0)
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.b = s.new_b
+ |WHEN NOT MATCHED BY SOURCE AND t.id >= 10 THEN DELETE
+ |""".stripMargin)
+
+ val thirdRangeFileCountAfter =
+ sql(
+ "SELECT COUNT(*) FROM `t$files` WHERE first_row_id = 10 AND
file_path NOT LIKE '%.blob'")
+ .collect()
+ .head
+ .getLong(0)
+ assert(thirdRangeFileCountAfter == thirdRangeFileCountBefore)
+
+ checkAnswer(
+ sql("SELECT id, b, c, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, 0, 0L),
+ Row(1, 100, 1, 1L),
+ Row(2, 2, 2, 2L),
+ Row(3, 3, 3, 3L),
+ Row(4, 4, 4, 4L),
+ Row(5, 5, 5, 5L),
+ Row(6, 6, 6, 6L),
+ Row(7, 7, 7, 7L),
+ Row(8, 8, 8, 8L),
+ Row(9, 9, 9, 9L))
+ )
+ }
+ }
+
+ test("Data Evolution deletion: merge blob update and delete") {
+ withTable("s", "t") {
+ sql("""
+ |CREATE TABLE t (id INT, b INT, picture BINARY)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'blob-field' = 'picture',
+ | 'blob.target-file-size' = '1 b')
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(1) */ id, b, picture FROM
VALUES
+ | (0, 0, X'00'), (1, 1, X'01'), (2, 2, X'02'), (3, 3, X'03'), (4,
4, X'04')
+ | AS v(id, b, picture)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(1) */ id, b, picture FROM
VALUES
+ | (5, 5, X'05'), (6, 6, X'06'), (7, 7, X'07'), (8, 8, X'08'), (9,
9, X'09')
+ | AS v(id, b, picture)
+ |""".stripMargin)
+
+ sql("CREATE TABLE s (id INT, picture BINARY, op STRING)")
+ sql("""
+ |INSERT INTO s VALUES
+ | (2, X'22', 'delete'),
+ | (6, X'66', 'update'),
+ | (7, X'4D', 'update'),
+ | (9, X'79', 'delete')
+ |""".stripMargin)
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |WHEN MATCHED AND s.op = 'delete' THEN DELETE
+ |WHEN MATCHED THEN UPDATE SET t.picture = s.picture
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, b, picture, _ROW_ID FROM t ORDER BY id"),
+ Seq(
+ Row(0, 0, Array[Byte](0), 0L),
+ Row(1, 1, Array[Byte](1), 1L),
+ Row(3, 3, Array[Byte](3), 3L),
+ Row(4, 4, Array[Byte](4), 4L),
+ Row(5, 5, Array[Byte](5), 5L),
+ Row(6, 6, Array[Byte](102), 6L),
+ Row(7, 7, Array[Byte](77), 7L),
+ Row(8, 8, Array[Byte](8), 8L)
+ )
+ )
+ }
+ }
}
