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 1367e46bfe [spark] Support partition pruning for data evolution
self-merge (#9489)
1367e46bfe is described below
commit 1367e46bfe05931d1530fa312c70d9ed7375e4b1
Author: zhoulii <[email protected]>
AuthorDate: Mon Aug 31 18:11:29 2026 +0800
[spark] Support partition pruning for data evolution self-merge (#9489)
---
.../MergeIntoPaimonDataEvolutionTable.scala | 128 +++++++++--
.../MergeIntoPaimonDataEvolutionTable.scala | 128 +++++++++--
.../paimon/spark/sql/RowTrackingTestBase.scala | 243 +++++++++++++++++++--
3 files changed, 449 insertions(+), 50 deletions(-)
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 fd70124ab5..1e59af4b57 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
@@ -27,9 +27,9 @@ import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.index.GlobalIndexMeta
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
import org.apache.paimon.manifest.IndexManifestEntry
+import org.apache.paimon.options.Options
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.analysis.PaimonRelation
-import org.apache.paimon.spark.catalyst.analysis.PaimonRelation.isPaimonTable
import org.apache.paimon.spark.catalyst.analysis.PaimonUpdateTable.toColumn
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
@@ -45,6 +45,7 @@ import org.apache.paimon.types.VectorType.isVectorStoreFile
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.apache.spark.sql.PaimonUtils._
+import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer.resolver
import org.apache.spark.sql.catalyst.expressions.{Alias, And,
AttributeReference, EqualTo, Expression, ExprId, Literal, Or, PythonUDF,
SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral,
TrueLiteral}
@@ -61,6 +62,7 @@ import scala.collection.{immutable, mutable}
import scala.collection.JavaConverters._
import scala.collection.Searching.{search, Found, InsertionPoint}
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
+import scala.util.control.NonFatal
/** Command for Merge Into for Data Evolution paimon table. */
case class MergeIntoPaimonDataEvolutionTable(
@@ -150,25 +152,97 @@ case class MergeIntoPaimonDataEvolutionTable(
*
* without any extra shuffle, join, or sort.
*/
- private lazy val isSelfMergeOnRowId: Boolean = {
- if (!isPaimonTable(sourceTable)) {
- false
- } else if (
-
!originalTargetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name)
- ) {
- false
+ private case class SelfMergeSpec(residualCondition: Option[Expression])
+
+ private def passthroughSourceRelation(plan: LogicalPlan):
Option[DataSourceV2Relation] = {
+ EliminateSubqueryAliases(plan) match {
+ case relation: DataSourceV2Relation if
isPaimonRelationWithoutTimeTravel(relation) =>
+ Some(relation)
+ case Project(projectList, child) if isPassthroughProject(projectList,
child) =>
+ passthroughSourceRelation(child)
+ case _ =>
+ None
+ }
+ }
+
+ private def isPaimonRelationWithoutTimeTravel(relation:
DataSourceV2Relation): Boolean =
+ relation.table match {
+ case sparkTable: SparkTable =>
+
!TimeTravelUtil.hasTimeTravelOptions(Options.fromMap(sparkTable.getTable.options()))
+ case _ => false
+ }
+
+ private def isPassthroughProject(projectList: Seq[Expression], child:
LogicalPlan): Boolean = {
+ val childAttributes = child.output ++ child.metadataOutput
+
+ def isChildAttribute(attr: AttributeReference): Boolean =
+ childAttributes.exists(_.exprId == attr.exprId)
+
+ projectList.forall {
+ case attr: AttributeReference => isChildAttribute(attr)
+ case alias: Alias =>
+ alias.child match {
+ case attr: AttributeReference =>
+ resolver(alias.name, attr.name) && isChildAttribute(attr)
+ case _ => false
+ }
+ case _ => false
+ }
+ }
+
+ private lazy val sameSourceAndTargetTable: Boolean =
+ passthroughSourceRelation(sourceTable)
+ .exists(sourceRelation =>
originalTargetRelation.name.equals(sourceRelation.name))
+
+ private def isTargetRowId(attr: AttributeReference): Boolean = {
+ attr.name == ROW_ID_NAME && (originalTargetRelation.output ++
+ originalTargetRelation.metadataOutput).exists(_.exprId == attr.exprId)
+ }
+
+ private def isSourceRowId(attr: AttributeReference): Boolean = {
+ attr.name == ROW_ID_NAME && (sourceTable.output ++
sourceTable.metadataOutput)
+ .exists(_.exprId == attr.exprId)
+ }
+
+ private def isRowIdEquality(expression: Expression): Boolean = expression
match {
+ case EqualTo(left: AttributeReference, right: AttributeReference) =>
+ (isTargetRowId(left) && isSourceRowId(right)) ||
+ (isSourceRowId(left) && isTargetRowId(right))
+ case _ => false
+ }
+
+ private lazy val isExactSelfMergeOnRowId: Boolean =
+ sameSourceAndTargetTable && isRowIdEquality(matchedCondition)
+
+ private lazy val selfMergeSpec: Option[SelfMergeSpec] = {
+ if (!sameSourceAndTargetTable) {
+ None
} else {
- matchedCondition match {
- case EqualTo(left: AttributeReference, right: AttributeReference)
- if left.name == ROW_ID_NAME && right.name == ROW_ID_NAME =>
- true
- case _ => false
+ val conjuncts = splitConjunctivePredicates(matchedCondition)
+ val rowIdEqualities = conjuncts.filter(isRowIdEquality)
+ val residualConditions = conjuncts.filterNot(isRowIdEquality)
+ val partitionRowType = table.schema().logicalPartitionType()
+ val targetOnlyResidualConditions = residualConditions.filter {
+ condition => canEvaluate(condition, targetTable) &&
canEvaluateWithinJoin(condition)
+ }
+ val allResidualConditionsArePartitionPredicates =
+ extractMergePartitionFilters(targetOnlyResidualConditions,
partitionRowType).size ==
+ residualConditions.size &&
residualConditions.forall(canConvertToPaimonPredicate)
+
+ if (rowIdEqualities.size != 1 ||
!allResidualConditionsArePartitionPredicates) {
+ None
+ } else {
+ Some(SelfMergeSpec(residualConditions.reduceOption(And)))
}
}
}
+ private lazy val useSelfMergeShortcut: Boolean =
+ selfMergeSpec.isDefined && notMatchedActions.isEmpty &&
notMatchedBySourceActions.isEmpty
+
assert(
- !(isSelfMergeOnRowId && (notMatchedActions.nonEmpty ||
notMatchedBySourceActions.nonEmpty)),
+ !(isExactSelfMergeOnRowId &&
+ (notMatchedActions.nonEmpty || notMatchedBySourceActions.nonEmpty)),
"Self-Merge on _ROW_ID only supports WHEN MATCHED actions. WHEN NOT
MATCHED and " +
"WHEN NOT MATCHED BY SOURCE are not supported."
)
@@ -340,6 +414,18 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
+ private def canConvertToPaimonPredicate(expression: Expression): Boolean = {
+ try {
+ convertConditionToPaimonPredicate(
+ expression,
+ targetRelation.output,
+ rowType,
+ ignorePartialFailure = false).isDefined
+ } catch {
+ case NonFatal(_) => false
+ }
+ }
+
private def targetRelatedSplits(
sparkSession: SparkSession,
tableSplits: Seq[DataSplit],
@@ -347,8 +433,9 @@ case class MergeIntoPaimonDataEvolutionTable(
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
persistSourceDss: Option[Dataset[Row]]): Seq[DataSplit] = {
// Self-Merge shortcut:
- // In Self-Merge mode, every row in the table may be updated, so we scan
all splits.
- if (isSelfMergeOnRowId) {
+ // The snapshot-level partition filter has already removed unrelated
partitions. Every row in
+ // the remaining splits may be updated, so the shortcut scans all of them.
+ if (useSelfMergeShortcut) {
return tableSplits
}
@@ -583,7 +670,7 @@ case class MergeIntoPaimonDataEvolutionTable(
def matchedActionInstructions(actions: Seq[MergeAction]):
Seq[MergeRows.Instruction] =
actions.map(matchedActionInstruction) ++ keepCopyInstructions
- val targetActionOutput: Dataset[Row] = if (isSelfMergeOnRowId) {
+ val targetActionOutput: Dataset[Row] = if (useSelfMergeShortcut) {
// Self-Merge shortcut:
// - Scan the target table only (no source scan, no join), and read all
columns required by
// merge condition and update expressions.
@@ -632,8 +719,13 @@ case class MergeIntoPaimonDataEvolutionTable(
throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
+ // The shortcut removes the source scan and join. A target row has a
matching source only
+ // when the residual ON condition holds.
+ val sourceRowPresentCondition =
+ selfMergeSpec.flatMap(_.residualCondition).getOrElse(TrueLiteral)
+
val mergeRows = MergeRows(
- isSourceRowPresent = TrueLiteral,
+ isSourceRowPresent = sourceRowPresentCondition,
isTargetRowPresent = TrueLiteral,
matchedInstructions =
matchedActionInstructions(rewrittenMatchedActions),
notMatchedInstructions = Nil,
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 42ffd2ad3c..9d24d1fcf8 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
@@ -27,9 +27,9 @@ import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.index.GlobalIndexMeta
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
import org.apache.paimon.manifest.IndexManifestEntry
+import org.apache.paimon.options.Options
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.analysis.PaimonRelation
-import org.apache.paimon.spark.catalyst.analysis.PaimonRelation.isPaimonTable
import org.apache.paimon.spark.catalyst.analysis.PaimonUpdateTable.toColumn
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
@@ -45,6 +45,7 @@ import org.apache.paimon.types.VectorType.isVectorStoreFile
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.apache.spark.sql.PaimonUtils._
+import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.analysis.SimpleAnalyzer.resolver
import org.apache.spark.sql.catalyst.expressions.{Alias, And,
AttributeReference, EqualTo, Expression, ExprId, Literal, Or, PythonUDF,
SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral,
TrueLiteral}
@@ -61,6 +62,7 @@ import scala.collection.{immutable, mutable}
import scala.collection.JavaConverters._
import scala.collection.Searching.{search, Found, InsertionPoint}
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
+import scala.util.control.NonFatal
/** Command for Merge Into for Data Evolution paimon table. */
case class MergeIntoPaimonDataEvolutionTable(
@@ -150,25 +152,97 @@ case class MergeIntoPaimonDataEvolutionTable(
*
* without any extra shuffle, join, or sort.
*/
- private lazy val isSelfMergeOnRowId: Boolean = {
- if (!isPaimonTable(sourceTable)) {
- false
- } else if (
-
!originalTargetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name)
- ) {
- false
+ private case class SelfMergeSpec(residualCondition: Option[Expression])
+
+ private def passthroughSourceRelation(plan: LogicalPlan):
Option[DataSourceV2Relation] = {
+ EliminateSubqueryAliases(plan) match {
+ case relation: DataSourceV2Relation if
isPaimonRelationWithoutTimeTravel(relation) =>
+ Some(relation)
+ case Project(projectList, child) if isPassthroughProject(projectList,
child) =>
+ passthroughSourceRelation(child)
+ case _ =>
+ None
+ }
+ }
+
+ private def isPaimonRelationWithoutTimeTravel(relation:
DataSourceV2Relation): Boolean =
+ relation.table match {
+ case sparkTable: SparkTable =>
+
!TimeTravelUtil.hasTimeTravelOptions(Options.fromMap(sparkTable.getTable.options()))
+ case _ => false
+ }
+
+ private def isPassthroughProject(projectList: Seq[Expression], child:
LogicalPlan): Boolean = {
+ val childAttributes = child.output ++ child.metadataOutput
+
+ def isChildAttribute(attr: AttributeReference): Boolean =
+ childAttributes.exists(_.exprId == attr.exprId)
+
+ projectList.forall {
+ case attr: AttributeReference => isChildAttribute(attr)
+ case alias: Alias =>
+ alias.child match {
+ case attr: AttributeReference =>
+ resolver(alias.name, attr.name) && isChildAttribute(attr)
+ case _ => false
+ }
+ case _ => false
+ }
+ }
+
+ private lazy val sameSourceAndTargetTable: Boolean =
+ passthroughSourceRelation(sourceTable)
+ .exists(sourceRelation =>
originalTargetRelation.name.equals(sourceRelation.name))
+
+ private def isTargetRowId(attr: AttributeReference): Boolean = {
+ attr.name == ROW_ID_NAME && (originalTargetRelation.output ++
+ originalTargetRelation.metadataOutput).exists(_.exprId == attr.exprId)
+ }
+
+ private def isSourceRowId(attr: AttributeReference): Boolean = {
+ attr.name == ROW_ID_NAME && (sourceTable.output ++
sourceTable.metadataOutput)
+ .exists(_.exprId == attr.exprId)
+ }
+
+ private def isRowIdEquality(expression: Expression): Boolean = expression
match {
+ case EqualTo(left: AttributeReference, right: AttributeReference) =>
+ (isTargetRowId(left) && isSourceRowId(right)) ||
+ (isSourceRowId(left) && isTargetRowId(right))
+ case _ => false
+ }
+
+ private lazy val isExactSelfMergeOnRowId: Boolean =
+ sameSourceAndTargetTable && isRowIdEquality(matchedCondition)
+
+ private lazy val selfMergeSpec: Option[SelfMergeSpec] = {
+ if (!sameSourceAndTargetTable) {
+ None
} else {
- matchedCondition match {
- case EqualTo(left: AttributeReference, right: AttributeReference)
- if left.name == ROW_ID_NAME && right.name == ROW_ID_NAME =>
- true
- case _ => false
+ val conjuncts = splitConjunctivePredicates(matchedCondition)
+ val rowIdEqualities = conjuncts.filter(isRowIdEquality)
+ val residualConditions = conjuncts.filterNot(isRowIdEquality)
+ val partitionRowType = table.schema().logicalPartitionType()
+ val targetOnlyResidualConditions = residualConditions.filter {
+ condition => canEvaluate(condition, targetTable) &&
canEvaluateWithinJoin(condition)
+ }
+ val allResidualConditionsArePartitionPredicates =
+ extractMergePartitionFilters(targetOnlyResidualConditions,
partitionRowType).size ==
+ residualConditions.size &&
residualConditions.forall(canConvertToPaimonPredicate)
+
+ if (rowIdEqualities.size != 1 ||
!allResidualConditionsArePartitionPredicates) {
+ None
+ } else {
+ Some(SelfMergeSpec(residualConditions.reduceOption(And)))
}
}
}
+ private lazy val useSelfMergeShortcut: Boolean =
+ selfMergeSpec.isDefined && notMatchedActions.isEmpty &&
notMatchedBySourceActions.isEmpty
+
assert(
- !(isSelfMergeOnRowId && (notMatchedActions.nonEmpty ||
notMatchedBySourceActions.nonEmpty)),
+ !(isExactSelfMergeOnRowId &&
+ (notMatchedActions.nonEmpty || notMatchedBySourceActions.nonEmpty)),
"Self-Merge on _ROW_ID only supports WHEN MATCHED actions. WHEN NOT
MATCHED and " +
"WHEN NOT MATCHED BY SOURCE are not supported."
)
@@ -340,6 +414,18 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
+ private def canConvertToPaimonPredicate(expression: Expression): Boolean = {
+ try {
+ convertConditionToPaimonPredicate(
+ expression,
+ targetRelation.output,
+ rowType,
+ ignorePartialFailure = false).isDefined
+ } catch {
+ case NonFatal(_) => false
+ }
+ }
+
private def targetRelatedSplits(
sparkSession: SparkSession,
tableSplits: Seq[DataSplit],
@@ -347,8 +433,9 @@ case class MergeIntoPaimonDataEvolutionTable(
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
persistSourceDss: Option[Dataset[Row]]): Seq[DataSplit] = {
// Self-Merge shortcut:
- // In Self-Merge mode, every row in the table may be updated, so we scan
all splits.
- if (isSelfMergeOnRowId) {
+ // The snapshot-level partition filter has already removed unrelated
partitions. Every row in
+ // the remaining splits may be updated, so the shortcut scans all of them.
+ if (useSelfMergeShortcut) {
return tableSplits
}
@@ -582,7 +669,7 @@ case class MergeIntoPaimonDataEvolutionTable(
def matchedActionInstructions(actions: Seq[MergeAction]):
Seq[MergeRows.Instruction] =
actions.map(matchedActionInstruction) ++ keepCopyInstructions
- val targetActionOutput: Dataset[Row] = if (isSelfMergeOnRowId) {
+ val targetActionOutput: Dataset[Row] = if (useSelfMergeShortcut) {
// Self-Merge shortcut:
// - Scan the target table only (no source scan, no join), and read all
columns required by
// merge condition and update expressions.
@@ -631,8 +718,13 @@ case class MergeIntoPaimonDataEvolutionTable(
throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
+ // The shortcut removes the source scan and join. A target row has a
matching source only
+ // when the residual ON condition holds.
+ val sourceRowPresentCondition =
+ selfMergeSpec.flatMap(_.residualCondition).getOrElse(TrueLiteral)
+
val mergeRows = MergeRows(
- isSourceRowPresent = TrueLiteral,
+ isSourceRowPresent = sourceRowPresentCondition,
isTargetRowPresent = TrueLiteral,
matchedInstructions =
matchedActionInstructions(rewrittenMatchedActions),
notMatchedInstructions = Nil,
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
index 1560bbca92..6ae50630f8 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
@@ -940,43 +940,54 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
private def executeMergeIntoAndAssertPartitionPruned(mergeSql: String): Unit
= {
+ val (_, resultedTableFiles) = executeMergeIntoAndCollectPlans(mergeSql)
+ assert(
+ resultedTableFiles.nonEmpty,
+ "Expected target PaimonSplitScan in merge into executed plans.")
+ assert(
+ resultedTableFiles.contains(1),
+ s"Expected target scan to read only one partition file, but got resulted
table files: " +
+ resultedTableFiles.mkString(", ")
+ )
+ }
+
+ private def executeMergeIntoAndCollectPlans(mergeSql: String):
(Seq[LogicalPlan], Seq[Long]) = {
+ val mergeRowsPlans = new
java.util.concurrent.CopyOnWriteArrayList[LogicalPlan]()
val resultedTableFiles = new
java.util.concurrent.CopyOnWriteArrayList[Long]()
val listener = new QueryExecutionListener {
override def onSuccess(funcName: String, qe: QueryExecution, durationNs:
Long): Unit = {
- checkPlan(qe)
+ collectPlans(qe)
}
override def onFailure(funcName: String, qe: QueryExecution, exception:
Exception): Unit = {
- checkPlan(qe)
+ collectPlans(qe)
}
- private def checkPlan(qe: QueryExecution): Unit = {
+ private def collectPlans(qe: QueryExecution): Unit = {
+ if (qe.analyzed.collectFirst { case _: MergeRows => true }.nonEmpty) {
+ mergeRowsPlans.add(qe.analyzed)
+ }
collect(qe.executedPlan) {
case scanExec: BatchScanExec
if scanExec.scan.isInstanceOf[PaimonSplitScan] &&
scanExec.scan.description().startsWith("PaimonSplitScan:
[target]") =>
- val scan = scanExec.scan.asInstanceOf[PaimonSplitScan]
- metric(scan.reportDriverMetrics(), RESULTED_TABLE_FILES)
- }.foreach(resultedTableFile =>
resultedTableFiles.add(resultedTableFile))
+ metric(
+
scanExec.scan.asInstanceOf[PaimonSplitScan].reportDriverMetrics(),
+ RESULTED_TABLE_FILES)
+ }.foreach(resultedTableFiles.add)
}
}
spark.listenerManager.register(listener)
try {
- sql(mergeSql)
+ sql(mergeSql).collect()
Utils.waitUntilEventEmpty(spark)
} finally {
spark.listenerManager.unregister(listener)
}
- val metrics = resultedTableFiles.asScala
- assert(metrics.nonEmpty, "Expected target PaimonSplitScan in merge into
executed plans.")
- assert(
- metrics.contains(1),
- s"Expected target scan to read only one partition file, but got resulted
table files: " +
- metrics.mkString(", ")
- )
+ (mergeRowsPlans.asScala.toSeq,
resultedTableFiles.asScala.map(_.toLong).toSeq)
}
private def metric(metrics: Array[CustomTaskMetric], name: String): Long = {
@@ -1131,6 +1142,210 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: self-merge on _ROW_ID with partition pruning") {
+ withSparkSQLConf("spark.paimon.data-evolution.merge-into.file-pruning" ->
"false") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT, dt STRING)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |PARTITIONED BY (dt)
+ |""".stripMargin)
+ sql("""
+ |INSERT INTO target VALUES
+ | (1, 10, '2026-08-30'),
+ | (2, 20, '2026-08-30'),
+ | (3, 30, '2026-08-31'),
+ | (4, 40, '2026-09-01')
+ |""".stripMargin)
+
+ val (mergeRowsPlans, resultedTableFiles) =
executeMergeIntoAndCollectPlans(
+ """
+ |MERGE INTO target
+ |USING target AS source
+ |ON source._ROW_ID = target._ROW_ID AND target.dt = '2026-08-30'
+ |WHEN MATCHED AND target.id = 1 THEN UPDATE SET target.b =
source.b + 100
+ |""".stripMargin)
+
+ assert(mergeRowsPlans.nonEmpty, "Expected a MergeRows plan for
self-merge.")
+ assert(
+ mergeRowsPlans.forall(_.collectFirst {
+ case p: Join => p
+ case p: Sort => p
+ case p: RepartitionByExpression => p
+ }.isEmpty),
+ s"Found unexpected Join/Sort/Exchange in plans:
${mergeRowsPlans.mkString("\n")}"
+ )
+ assert(
+ resultedTableFiles.contains(1L),
+ s"Expected target scan to read one partition file, but got: " +
+ resultedTableFiles.mkString(", "))
+
+ checkAnswer(
+ sql("SELECT id, b, dt FROM target ORDER BY id"),
+ Seq(
+ Row(1, 110, "2026-08-30"),
+ Row(2, 20, "2026-08-30"),
+ Row(3, 30, "2026-08-31"),
+ Row(4, 40, "2026-09-01")))
+ }
+ }
+ }
+
+ test("Data Evolution: self-merge falls back for computed source projection")
{
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT, dt STRING)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |PARTITIONED BY (dt)
+ |""".stripMargin)
+ sql("INSERT INTO target VALUES (1, 10, 'p1'), (2, 20, 'p2')")
+
+ val (mergeRowsPlans, _) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING (
+ | SELECT _ROW_ID, b + 1 AS b FROM
target
+ |) source
+ |ON target._ROW_ID = source._ROW_ID
+ | AND target.dt = 'p1'
+ |WHEN MATCHED THEN UPDATE SET
target.b = source.b
+ |""".stripMargin)
+
+ assert(
+ mergeRowsPlans.exists(_.collectFirst { case _: Join => true
}.nonEmpty),
+ s"Expected general MERGE plan with Join, but got:
${mergeRowsPlans.mkString("\n")}"
+ )
+ checkAnswer(
+ sql("SELECT id, b, dt FROM target ORDER BY id"),
+ Seq(Row(1, 11, "p1"), Row(2, 20, "p2")))
+ }
+ }
+
+ test("Data Evolution: self-merge falls back for time-travel source") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT, dt STRING)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |PARTITIONED BY (dt)
+ |""".stripMargin)
+ sql("INSERT INTO target VALUES (1, 10, 'p1'), (2, 30, 'p2')")
+ val oldSnapshotId =
loadTable("target").snapshotManager().latestSnapshotId()
+ sql("UPDATE target SET b = 20 WHERE id = 1")
+
+ val (mergeRowsPlans, _) =
+ executeMergeIntoAndCollectPlans(s"""
+ |MERGE INTO target
+ |USING (
+ | SELECT _ROW_ID, b
+ | FROM target VERSION AS OF
$oldSnapshotId
+ |) source
+ |ON target._ROW_ID = source._ROW_ID
+ | AND target.dt = 'p1'
+ |WHEN MATCHED THEN UPDATE SET
target.b = source.b
+ |""".stripMargin)
+
+ assert(
+ mergeRowsPlans.exists(_.collectFirst { case _: Join => true
}.nonEmpty),
+ s"Expected general MERGE plan with Join, but got:
${mergeRowsPlans.mkString("\n")}"
+ )
+ checkAnswer(
+ sql("SELECT id, b, dt FROM target ORDER BY id"),
+ Seq(Row(1, 10, "p1"), Row(2, 30, "p2")))
+ }
+ }
+
+ test("Data Evolution: self-merge falls back for non-partition residual
condition") {
+ withTable("target") {
+ sql(
+ "CREATE TABLE target (id INT, b INT) TBLPROPERTIES " +
+ "('row-tracking.enabled' = 'true', 'data-evolution.enabled' =
'true')")
+ sql("INSERT INTO target VALUES (1, 10), (2, 20)")
+
+ val (mergeRowsPlans, _) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID = source._ROW_ID
AND target.id = 1
+ |WHEN MATCHED THEN UPDATE SET
target.b = source.b + 100
+ |""".stripMargin)
+
+ assert(
+ mergeRowsPlans.exists(_.collectFirst { case _: Join => true
}.nonEmpty),
+ s"Expected general MERGE plan with Join, but got:
${mergeRowsPlans.mkString("\n")}"
+ )
+ checkAnswer(sql("SELECT id, b FROM target ORDER BY id"), Seq(Row(1,
110), Row(2, 20)))
+ }
+ }
+
+ test("Data Evolution: self-merge falls back for dynamic partition
condition") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT, dt STRING)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |PARTITIONED BY (dt)
+ |""".stripMargin)
+ sql("INSERT INTO target VALUES (1, 10, 'p1'), (2, 20, 'p2')")
+
+ val (mergeRowsPlans, _) =
+ executeMergeIntoAndCollectPlans(
+ """
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID = source._ROW_ID AND target.dt = source.dt
+ |WHEN MATCHED THEN UPDATE SET target.b = source.b + 100
+ |""".stripMargin)
+
+ assert(
+ mergeRowsPlans.exists(_.collectFirst { case _: Join => true
}.nonEmpty),
+ s"Expected general MERGE plan with Join, but got:
${mergeRowsPlans.mkString("\n")}"
+ )
+ checkAnswer(
+ sql("SELECT id, b, dt FROM target ORDER BY id"),
+ Seq(Row(1, 110, "p1"), Row(2, 120, "p2")))
+ }
+ }
+
+ test("Data Evolution: self-merge with not matched action uses general MERGE
path") {
+ withSparkSQLConf("spark.paimon.data-evolution.merge-into.file-pruning" ->
"false") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT, dt STRING)
+ |TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |PARTITIONED BY (dt)
+ |""".stripMargin)
+ sql("INSERT INTO target VALUES (1, 10, 'p1'), (2, 20, 'p2')")
+
+ val (mergeRowsPlans, _) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID =
source._ROW_ID AND target.dt = 'p1'
+ |WHEN MATCHED THEN UPDATE SET
target.b = source.b + 100
+ |WHEN NOT MATCHED THEN INSERT (id,
b, dt)
+ | VALUES (source.id + 10,
source.b, source.dt)
+ |""".stripMargin)
+
+ assert(
+ mergeRowsPlans.exists(_.collectFirst { case _: Join => true
}.nonEmpty),
+ s"Expected general MERGE plan with Join, but got:
${mergeRowsPlans.mkString("\n")}"
+ )
+ checkAnswer(
+ sql("SELECT id, b, dt FROM target ORDER BY id"),
+ Seq(Row(1, 110, "p1"), Row(2, 20, "p2"), Row(12, 20, "p2")))
+ }
+ }
+ }
+
test("Data Evolution: V1 update table with data-evolution") {
withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
withTable("t") {
