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 87bdb2d5d7 [spark] Support action predicate pruning for data evolution
self-merge (#9544)
87bdb2d5d7 is described below
commit 87bdb2d5d73ab288945752ec109b9ee52f99f2cd
Author: zhoulii <[email protected]>
AuthorDate: Wed Sep 2 17:52:54 2026 +0800
[spark] Support action predicate pruning for data evolution self-merge
(#9544)
---
.../MergeIntoPaimonDataEvolutionTable.scala | 104 ++++++++----
.../MergeIntoPaimonDataEvolutionTable.scala | 104 ++++++++----
.../paimon/spark/sql/RowTrackingTestBase.scala | 182 ++++++++++++++++++++-
3 files changed, 325 insertions(+), 65 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 1e59af4b57..d86165352d 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
@@ -28,6 +28,7 @@ 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.predicate.{Predicate, PredicateBuilder}
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.analysis.PaimonRelation
import org.apache.paimon.spark.catalyst.analysis.PaimonUpdateTable.toColumn
@@ -249,6 +250,32 @@ case class MergeIntoPaimonDataEvolutionTable(
private lazy val targetRelation: DataSourceV2Relation =
matchedUpdateScanTarget._2
+ private lazy val sourceToTargetAttributes: Map[ExprId, AttributeReference] =
{
+ val targetAttrs = targetRelation.output ++ targetRelation.metadataOutput
+ val sourceAttrs = sourceTable.output ++ sourceTable.metadataOutput
+ sourceAttrs.flatMap {
+ source =>
+ targetAttrs.find(target => resolver(target.name,
source.name)).map(source.exprId -> _)
+ }.toMap
+ }
+
+ private def rewriteSourceToTarget(expression: Expression): Expression = {
+ expression.transform {
+ case attr: AttributeReference if
sourceToTargetAttributes.contains(attr.exprId) =>
+ sourceToTargetAttributes(attr.exprId)
+ }
+ }
+
+ private def rewriteToTargetOnly(expression: Expression): Option[Expression]
= {
+ val rewritten = rewriteSourceToTarget(expression)
+ val targetExprIds = targetRelation.output.map(_.exprId).toSet
+ if (rewritten.references.forall(attr =>
targetExprIds.contains(attr.exprId))) {
+ Some(rewritten)
+ } else {
+ None
+ }
+ }
+
lazy val tableSchema: StructType = targetSparkTable.schema
override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -264,7 +291,7 @@ case class MergeIntoPaimonDataEvolutionTable(
if (readSnapshot != null) {
snapshotReader.withSnapshot(readSnapshot)
}
- pushDownMergePartitionFilter(snapshotReader)
+ pushDownMergeFilters(snapshotReader)
val plan = snapshotReader.read()
val tableSplits: Seq[DataSplit] = plan
.splits()
@@ -379,10 +406,17 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
- private def pushDownMergePartitionFilter(snapshotReader: SnapshotReader):
Unit = {
+ private def pushDownMergeFilters(snapshotReader: SnapshotReader): Unit = {
+ val predicates = Seq(mergePartitionPredicate,
selfMergeActionPredicate).flatten
+ if (predicates.nonEmpty) {
+ snapshotReader.withFilter(PredicateBuilder.and(predicates: _*))
+ }
+ }
+
+ private def mergePartitionPredicate: Option[Predicate] = {
val partitionRowType = table.schema().logicalPartitionType()
if (partitionRowType.getFieldCount == 0) {
- return
+ return None
}
// matchedCondition comes from MergeIntoTable.mergeCondition, which is the
MERGE ON condition.
@@ -392,12 +426,42 @@ case class MergeIntoPaimonDataEvolutionTable(
.getOrElse(Seq.empty)
if (partitionPredicates.nonEmpty) {
- val filter = convertConditionToPaimonPredicate(
+ convertConditionToPaimonPredicate(
partitionPredicates.reduce(And),
targetRelation.output,
rowType,
ignorePartialFailure = true)
- filter.foreach(snapshotReader.withFilter)
+ } else {
+ None
+ }
+ }
+
+ private def selfMergeActionPredicate: Option[Predicate] = {
+ if (
+ !useSelfMergeShortcut ||
+ !table.coreOptions().dataEvolutionMergeIntoFilePruning() ||
+ matchedActions.isEmpty ||
+ matchedActions.exists(_.condition.isEmpty)
+ ) {
+ return None
+ }
+
+ val actionCondition = matchedActions.flatMap(_.condition).reduce(Or)
+ if (!actionCondition.deterministic) {
+ return None
+ }
+
+ rewriteToTargetOnly(actionCondition).flatMap {
+ targetOnlyCondition =>
+ try {
+ convertConditionToPaimonPredicate(
+ targetOnlyCondition,
+ targetRelation.output,
+ rowType,
+ ignorePartialFailure = false)
+ } catch {
+ case NonFatal(_) => None
+ }
}
}
@@ -433,8 +497,8 @@ case class MergeIntoPaimonDataEvolutionTable(
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
persistSourceDss: Option[Dataset[Row]]): Seq[DataSplit] = {
// Self-Merge shortcut:
- // 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.
+ // Snapshot-level merge filters have already removed unrelated splits.
MergeRows preserves
+ // rows in the remaining splits which do not satisfy any matched action
condition.
if (useSelfMergeShortcut) {
return tableSplits
}
@@ -689,32 +753,14 @@ case class MergeIntoPaimonDataEvolutionTable(
targetAttrsDedup.filter(a => neededNames.exists(n => resolver(n,
a.name)))
val readPlan = touchedFileTargetRelation.copy(output =
allReadFieldsOnTarget)
- // Build mapping: source exprId -> target attr (matched by column name).
- val sourceToTarget = {
- val targetAttrs = targetRelation.output ++
targetRelation.metadataOutput
- val sourceAttrs = sourceTable.output ++ sourceTable.metadataOutput
- sourceAttrs.flatMap {
- s => targetAttrs.find(t => resolver(t.name, s.name)).map(t =>
s.exprId -> t)
- }.toMap
- }
-
- def rewriteSourceToTarget(
- expr: Expression,
- m: Map[ExprId, AttributeReference]): Expression = {
- expr.transform {
- case a: AttributeReference if m.contains(a.exprId) => m(a.exprId)
- }
- }
-
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))
- }
+ val newCond = action.condition.map(rewriteSourceToTarget)
+ val newAssignments =
+ action.assignments.map(a => Assignment(a.key,
rewriteSourceToTarget(a.value)))
action.copy(condition = newCond, assignments = newAssignments)
case DeleteAction(condition) =>
- DeleteAction(condition.map(c => rewriteSourceToTarget(c,
sourceToTarget)))
+ DeleteAction(condition.map(rewriteSourceToTarget))
case other =>
throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
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 9d24d1fcf8..b6a5650d1f 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
@@ -28,6 +28,7 @@ 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.predicate.{Predicate, PredicateBuilder}
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.analysis.PaimonRelation
import org.apache.paimon.spark.catalyst.analysis.PaimonUpdateTable.toColumn
@@ -249,6 +250,32 @@ case class MergeIntoPaimonDataEvolutionTable(
private lazy val targetRelation: DataSourceV2Relation =
matchedUpdateScanTarget._2
+ private lazy val sourceToTargetAttributes: Map[ExprId, AttributeReference] =
{
+ val targetAttrs = targetRelation.output ++ targetRelation.metadataOutput
+ val sourceAttrs = sourceTable.output ++ sourceTable.metadataOutput
+ sourceAttrs.flatMap {
+ source =>
+ targetAttrs.find(target => resolver(target.name,
source.name)).map(source.exprId -> _)
+ }.toMap
+ }
+
+ private def rewriteSourceToTarget(expression: Expression): Expression = {
+ expression.transform {
+ case attr: AttributeReference if
sourceToTargetAttributes.contains(attr.exprId) =>
+ sourceToTargetAttributes(attr.exprId)
+ }
+ }
+
+ private def rewriteToTargetOnly(expression: Expression): Option[Expression]
= {
+ val rewritten = rewriteSourceToTarget(expression)
+ val targetExprIds = targetRelation.output.map(_.exprId).toSet
+ if (rewritten.references.forall(attr =>
targetExprIds.contains(attr.exprId))) {
+ Some(rewritten)
+ } else {
+ None
+ }
+ }
+
lazy val tableSchema: StructType = targetSparkTable.schema
override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -264,7 +291,7 @@ case class MergeIntoPaimonDataEvolutionTable(
if (readSnapshot != null) {
snapshotReader.withSnapshot(readSnapshot)
}
- pushDownMergePartitionFilter(snapshotReader)
+ pushDownMergeFilters(snapshotReader)
val plan = snapshotReader.read()
val tableSplits: Seq[DataSplit] = plan
.splits()
@@ -379,10 +406,17 @@ case class MergeIntoPaimonDataEvolutionTable(
}
}
- private def pushDownMergePartitionFilter(snapshotReader: SnapshotReader):
Unit = {
+ private def pushDownMergeFilters(snapshotReader: SnapshotReader): Unit = {
+ val predicates = Seq(mergePartitionPredicate,
selfMergeActionPredicate).flatten
+ if (predicates.nonEmpty) {
+ snapshotReader.withFilter(PredicateBuilder.and(predicates: _*))
+ }
+ }
+
+ private def mergePartitionPredicate: Option[Predicate] = {
val partitionRowType = table.schema().logicalPartitionType()
if (partitionRowType.getFieldCount == 0) {
- return
+ return None
}
// matchedCondition comes from MergeIntoTable.mergeCondition, which is the
MERGE ON condition.
@@ -392,12 +426,42 @@ case class MergeIntoPaimonDataEvolutionTable(
.getOrElse(Seq.empty)
if (partitionPredicates.nonEmpty) {
- val filter = convertConditionToPaimonPredicate(
+ convertConditionToPaimonPredicate(
partitionPredicates.reduce(And),
targetRelation.output,
rowType,
ignorePartialFailure = true)
- filter.foreach(snapshotReader.withFilter)
+ } else {
+ None
+ }
+ }
+
+ private def selfMergeActionPredicate: Option[Predicate] = {
+ if (
+ !useSelfMergeShortcut ||
+ !table.coreOptions().dataEvolutionMergeIntoFilePruning() ||
+ matchedActions.isEmpty ||
+ matchedActions.exists(_.condition.isEmpty)
+ ) {
+ return None
+ }
+
+ val actionCondition = matchedActions.flatMap(_.condition).reduce(Or)
+ if (!actionCondition.deterministic) {
+ return None
+ }
+
+ rewriteToTargetOnly(actionCondition).flatMap {
+ targetOnlyCondition =>
+ try {
+ convertConditionToPaimonPredicate(
+ targetOnlyCondition,
+ targetRelation.output,
+ rowType,
+ ignorePartialFailure = false)
+ } catch {
+ case NonFatal(_) => None
+ }
}
}
@@ -433,8 +497,8 @@ case class MergeIntoPaimonDataEvolutionTable(
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
persistSourceDss: Option[Dataset[Row]]): Seq[DataSplit] = {
// Self-Merge shortcut:
- // 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.
+ // Snapshot-level merge filters have already removed unrelated splits.
MergeRows preserves
+ // rows in the remaining splits which do not satisfy any matched action
condition.
if (useSelfMergeShortcut) {
return tableSplits
}
@@ -688,32 +752,14 @@ case class MergeIntoPaimonDataEvolutionTable(
targetAttrsDedup.filter(a => neededNames.exists(n => resolver(n,
a.name)))
val readPlan = touchedFileTargetRelation.copy(output =
allReadFieldsOnTarget)
- // Build mapping: source exprId -> target attr (matched by column name).
- val sourceToTarget = {
- val targetAttrs = targetRelation.output ++
targetRelation.metadataOutput
- val sourceAttrs = sourceTable.output ++ sourceTable.metadataOutput
- sourceAttrs.flatMap {
- s => targetAttrs.find(t => resolver(t.name, s.name)).map(t =>
s.exprId -> t)
- }.toMap
- }
-
- def rewriteSourceToTarget(
- expr: Expression,
- m: Map[ExprId, AttributeReference]): Expression = {
- expr.transform {
- case a: AttributeReference if m.contains(a.exprId) => m(a.exprId)
- }
- }
-
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))
- }
+ val newCond = action.condition.map(rewriteSourceToTarget)
+ val newAssignments =
+ action.assignments.map(a => Assignment(a.key,
rewriteSourceToTarget(a.value)))
action.copy(condition = newCond, assignments = newAssignments)
case DeleteAction(condition) =>
- DeleteAction(condition.map(c => rewriteSourceToTarget(c,
sourceToTarget)))
+ DeleteAction(condition.map(rewriteSourceToTarget))
case other =>
throw new UnsupportedOperationException(s"Unsupported matched
action: $other.")
}
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 5c4e32f390..6eb128e168 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
@@ -1087,9 +1087,11 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
test("Data Evolution: merge into table with data-evolution for Self-Merge
with _ROW_ID shortcut") {
withTable("target") {
sql(
- "CREATE TABLE target (a INT, b INT, c STRING) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
+ "CREATE TABLE target (a INT, b INT, c STRING) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true',
'compaction.min.file-num' = '100')")
sql(
- "INSERT INTO target values (1, 10, 'c1'), (2, 20, 'c2'), (3, 30,
'c3'), (4, 40, 'c4'), (5, 50, 'c5')")
+ "INSERT INTO target SELECT /*+ REPARTITION(1) */ * FROM VALUES (1, 10,
'c1'), (2, 20, 'c2')")
+ sql(
+ "INSERT INTO target SELECT /*+ REPARTITION(1) */ * FROM VALUES (3, 30,
'c3'), (4, 40, 'c4'), (5, 50, 'c5')")
var updatePlan: LogicalPlan = null
val latch = new CountDownLatch(1)
@@ -1133,11 +1135,11 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
checkAnswer(
sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM target ORDER BY a"),
Seq(
- Row(1, 10, "c1", 0, 2),
- Row(2, 20, "c2", 1, 2),
- Row(3, 90, "c3c3", 2, 2),
- Row(4, 120, "c4c4", 3, 2),
- Row(5, 100, "c5", 4, 2))
+ Row(1, 10, "c1", 0, 1),
+ Row(2, 20, "c2", 1, 1),
+ Row(3, 90, "c3c3", 2, 3),
+ Row(4, 120, "c4c4", 3, 3),
+ Row(5, 100, "c5", 4, 3))
)
}
}
@@ -1193,6 +1195,172 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ Seq(false, true).foreach {
+ filePruning =>
+ test(s"Data Evolution: self-merge action predicate file pruning:
$filePruning") {
+ withSparkSQLConf(
+ "spark.paimon.data-evolution.merge-into.file-pruning" ->
filePruning.toString) {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'compaction.min.file-num' = '100')
+ |""".stripMargin)
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10
FROM range(1, 3)")
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10
FROM range(3, 5)")
+
+ val (_, resultedTableFiles) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID =
source._ROW_ID
+ |WHEN MATCHED AND source.id = 3
+ | THEN UPDATE SET b =
source.b + 100
+ |""".stripMargin)
+
+ val expectedFiles = if (filePruning) Seq(1L) else Seq(2L)
+ assert(
+ resultedTableFiles == expectedFiles,
+ s"Expected target scan metrics $expectedFiles, but got: " +
+ resultedTableFiles.mkString(", "))
+ checkAnswer(
+ sql("SELECT id, b FROM target ORDER BY id"),
+ Seq(Row(1, 10), Row(2, 20), Row(3, 130), Row(4, 40)))
+ }
+ }
+ }
+ }
+
+ test("Data Evolution: self-merge combines on and action predicates") {
+ withSparkSQLConf("spark.paimon.data-evolution.merge-into.file-pruning" ->
"true") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT, dt STRING) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'compaction.min.file-num' = '100')
+ |PARTITIONED BY (dt)
+ |""".stripMargin)
+ sql("INSERT INTO target VALUES (1, 10, 'p1'), (2, 20, 'p1')")
+ sql("INSERT INTO target VALUES (1, 30, 'p2'), (4, 40, 'p2')")
+
+ val (_, resultedTableFiles) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID = source._ROW_ID
+ | AND target.dt = 'p1'
+ |WHEN MATCHED AND source.id = 1
+ | THEN UPDATE SET b = source.b +
100
+ |""".stripMargin)
+
+ assert(
+ resultedTableFiles == Seq(1L),
+ s"Expected one target scan after combining predicates, but got: " +
+ resultedTableFiles.mkString(", "))
+ checkAnswer(
+ sql("SELECT id, b, dt FROM target ORDER BY dt, id"),
+ Seq(Row(1, 110, "p1"), Row(2, 20, "p1"), Row(1, 30, "p2"), Row(4,
40, "p2")))
+ }
+ }
+ }
+
+ test("Data Evolution: self-merge skips action pruning for unconditional
matched action") {
+ withSparkSQLConf("spark.paimon.data-evolution.merge-into.file-pruning" ->
"true") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'compaction.min.file-num' = '100')
+ |""".stripMargin)
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10 FROM
range(1, 3)")
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10 FROM
range(3, 5)")
+
+ val (_, resultedTableFiles) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID = source._ROW_ID
+ |WHEN MATCHED THEN UPDATE SET b =
source.b + 100
+ |""".stripMargin)
+
+ assert(
+ resultedTableFiles == Seq(2L),
+ s"Expected one unpruned target scan, but got:
${resultedTableFiles.mkString(", ")}")
+ checkAnswer(
+ sql("SELECT id, b FROM target ORDER BY id"),
+ Seq(Row(1, 110), Row(2, 120), Row(3, 130), Row(4, 140)))
+ }
+ }
+ }
+
+ test("Data Evolution: self-merge action predicate prunes files for delete") {
+ withSparkSQLConf("spark.paimon.data-evolution.merge-into.file-pruning" ->
"true") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'compaction.min.file-num' = '100')
+ |""".stripMargin)
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10 FROM
range(1, 3)")
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10 FROM
range(3, 5)")
+
+ val (_, resultedTableFiles) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID = source._ROW_ID
+ |WHEN MATCHED AND source.id = 3
THEN DELETE
+ |""".stripMargin)
+
+ assert(
+ resultedTableFiles == Seq(1L),
+ s"Expected one pruned target scan, but got:
${resultedTableFiles.mkString(", ")}")
+ checkAnswer(
+ sql("SELECT id, b FROM target ORDER BY id"),
+ Seq(Row(1, 10), Row(2, 20), Row(4, 40)))
+ }
+ }
+ }
+
+ test("Data Evolution: self-merge action predicate requires complete
conversion") {
+ withSparkSQLConf("spark.paimon.data-evolution.merge-into.file-pruning" ->
"true") {
+ withTable("target") {
+ sql("""
+ |CREATE TABLE target (id INT, b INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'compaction.min.file-num' = '100')
+ |""".stripMargin)
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10 FROM
range(1, 3)")
+ sql("INSERT INTO target SELECT /*+ REPARTITION(1) */ id, id * 10 FROM
range(3, 5)")
+
+ val (_, resultedTableFiles) =
+ executeMergeIntoAndCollectPlans("""
+ |MERGE INTO target
+ |USING target AS source
+ |ON target._ROW_ID = source._ROW_ID
+ |WHEN MATCHED AND target.id = 1
+ | THEN UPDATE SET b = source.b +
100
+ |WHEN MATCHED AND target.id + 1 = 4
+ | THEN UPDATE SET b = source.b +
200
+ |""".stripMargin)
+
+ assert(
+ resultedTableFiles == Seq(2L),
+ s"Expected action pruning to fall back completely, but got: " +
+ resultedTableFiles.mkString(", "))
+ checkAnswer(
+ sql("SELECT id, b FROM target ORDER BY id"),
+ Seq(Row(1, 110), Row(2, 20), Row(3, 230), Row(4, 40)))
+ }
+ }
+ }
+
test("Data Evolution: self-merge falls back for computed source projection")
{
withTable("target") {
sql("""
