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 3403989f38 [spark] Ignore source row ids outside the table in the data
evolution MERGE row-id shortcut (#10055)
3403989f38 is described below
commit 3403989f386efce21be358eee69e38f02eb823d4
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Tue Sep 22 10:45:18 2026 +0800
[spark] Ignore source row ids outside the table in the data evolution MERGE
row-id shortcut (#10055)
---
.../MergeIntoPaimonDataEvolutionTable.scala | 42 +++++++++++++--
.../MergeIntoPaimonDataEvolutionTable.scala | 42 +++++++++++++--
.../paimon/spark/sql/RowTrackingTestBase.scala | 63 ++++++++++++++++++++++
3 files changed, 139 insertions(+), 8 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 5cf8f812d0..15a8f89429 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
@@ -25,7 +25,7 @@ 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}
+import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
import org.apache.paimon.manifest.IndexManifestEntry
import org.apache.paimon.options.Options
import org.apache.paimon.predicate.{Predicate, PredicateBuilder}
@@ -327,7 +327,7 @@ case class MergeIntoPaimonDataEvolutionTable(
.map(_.asInstanceOf[DataSplit])
.toSeq
- val firstRowIds: immutable.IndexedSeq[Long] = tableSplits
+ val normalDataFiles: Seq[DataFileMeta] = tableSplits
.flatMap(_.dataFiles().asScala)
.filter {
file =>
@@ -335,10 +335,19 @@ case class MergeIntoPaimonDataEvolutionTable(
!isBlobFile(file.fileName()) &&
!isVectorStoreFile(file.fileName())
}
+ val firstRowIds: immutable.IndexedSeq[Long] = normalDataFiles
.map(file => file.firstRowId().asInstanceOf[Long])
.distinct
.sorted
.toIndexedSeq
+ // Exclusive end of the row-id range starting at each entry of
`firstRowIds`. Files sharing a
+ // first row id are column groups of one range and end at the same row id.
+ val rowIdRangeEnds: immutable.IndexedSeq[Long] = {
+ val ends = normalDataFiles
+ .groupBy(_.firstRowId().longValue())
+ .map { case (first, files) => first -> files.map(file => first +
file.rowCount()).max }
+ firstRowIds.map(ends)
+ }
val firstRowIdToBlobFirstRowIds: Map[Long, List[Long]] = {
val map = new mutable.HashMap[Long, List[Long]]()
@@ -377,6 +386,7 @@ case class MergeIntoPaimonDataEvolutionTable(
sparkSession,
tableSplits,
firstRowIds,
+ rowIdRangeEnds,
firstRowIdToBlobFirstRowIds,
persistSourceDss)
lazy val touchedFileTargetRelation =
@@ -522,6 +532,7 @@ case class MergeIntoPaimonDataEvolutionTable(
sparkSession: SparkSession,
tableSplits: Seq[DataSplit],
firstRowIds: immutable.IndexedSeq[Long],
+ rowIdRangeEnds: immutable.IndexedSeq[Long],
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
persistSourceDss: Option[Dataset[Row]]): Seq[DataSplit] = {
// Self-Merge shortcut:
@@ -547,6 +558,7 @@ case class MergeIntoPaimonDataEvolutionTable(
sourceDss,
sparkSession,
firstRowIds,
+ rowIdRangeEnds,
firstRowIdToBlobFirstRowIds,
sourceRowIdAttr.name).toSet
@@ -557,6 +569,7 @@ case class MergeIntoPaimonDataEvolutionTable(
targetDss.alias("_left").join(sourceDss, toColumn(matchedCondition),
"inner"),
sparkSession,
firstRowIds,
+ rowIdRangeEnds,
firstRowIdToBlobFirstRowIds,
"_left." + ROW_ID_NAME
).toSet
@@ -1261,12 +1274,18 @@ case class MergeIntoPaimonDataEvolutionTable(
dataset: Dataset[Row],
sparkSession: SparkSession,
firstRowIds: immutable.IndexedSeq[Long],
+ rowIdRangeEnds: immutable.IndexedSeq[Long],
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
identifier: String): Array[Long] = {
import sparkSession.implicits._
- val firstRowIdUdf = udf((rowId: Long) => floorBinarySearch(firstRowIds,
rowId))
+ // The column may come straight from the source table, so it can hold row
ids that no longer
+ // exist (rows overwritten or dropped since the source was built) or
arbitrary values. Those
+ // match no target row: they map to no file at all rather than to the
nearest one.
+ val firstRowIdUdf =
+ udf((rowId: Long) => rowIdRangeStart(firstRowIds, rowIdRangeEnds, rowId))
dataset
- .select(firstRowIdUdf(col(identifier)))
+ .select(firstRowIdUdf(col(identifier)).as(FIRST_ROW_ID_NAME))
+ .filter(col(FIRST_ROW_ID_NAME).isNotNull)
.distinct()
.as[Long]
.flatMap(
@@ -1533,6 +1552,21 @@ object MergeIntoPaimonDataEvolutionTable {
}
}
+ /**
+ * The start of the row-id range `[firstRowIds(i), rowIdRangeEnds(i))`
holding `rowId`, or `None`
+ * when no range does.
+ */
+ private def rowIdRangeStart(
+ firstRowIds: immutable.IndexedSeq[Long],
+ rowIdRangeEnds: immutable.IndexedSeq[Long],
+ rowId: Long): Option[Long] = {
+ val index = firstRowIds.search(rowId) match {
+ case Found(foundIndex) => foundIndex
+ case InsertionPoint(insertionIndex) => insertionIndex - 1
+ }
+ if (index >= 0 && rowId < rowIdRangeEnds(index)) Some(firstRowIds(index))
else None
+ }
+
private def floorBinarySearch(indexed: immutable.IndexedSeq[Long], value:
Long): Long = {
if (indexed.isEmpty) {
throw new IllegalArgumentException("The input sorted sequence is empty.")
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 5cf8f812d0..15a8f89429 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
@@ -25,7 +25,7 @@ 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}
+import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
import org.apache.paimon.manifest.IndexManifestEntry
import org.apache.paimon.options.Options
import org.apache.paimon.predicate.{Predicate, PredicateBuilder}
@@ -327,7 +327,7 @@ case class MergeIntoPaimonDataEvolutionTable(
.map(_.asInstanceOf[DataSplit])
.toSeq
- val firstRowIds: immutable.IndexedSeq[Long] = tableSplits
+ val normalDataFiles: Seq[DataFileMeta] = tableSplits
.flatMap(_.dataFiles().asScala)
.filter {
file =>
@@ -335,10 +335,19 @@ case class MergeIntoPaimonDataEvolutionTable(
!isBlobFile(file.fileName()) &&
!isVectorStoreFile(file.fileName())
}
+ val firstRowIds: immutable.IndexedSeq[Long] = normalDataFiles
.map(file => file.firstRowId().asInstanceOf[Long])
.distinct
.sorted
.toIndexedSeq
+ // Exclusive end of the row-id range starting at each entry of
`firstRowIds`. Files sharing a
+ // first row id are column groups of one range and end at the same row id.
+ val rowIdRangeEnds: immutable.IndexedSeq[Long] = {
+ val ends = normalDataFiles
+ .groupBy(_.firstRowId().longValue())
+ .map { case (first, files) => first -> files.map(file => first +
file.rowCount()).max }
+ firstRowIds.map(ends)
+ }
val firstRowIdToBlobFirstRowIds: Map[Long, List[Long]] = {
val map = new mutable.HashMap[Long, List[Long]]()
@@ -377,6 +386,7 @@ case class MergeIntoPaimonDataEvolutionTable(
sparkSession,
tableSplits,
firstRowIds,
+ rowIdRangeEnds,
firstRowIdToBlobFirstRowIds,
persistSourceDss)
lazy val touchedFileTargetRelation =
@@ -522,6 +532,7 @@ case class MergeIntoPaimonDataEvolutionTable(
sparkSession: SparkSession,
tableSplits: Seq[DataSplit],
firstRowIds: immutable.IndexedSeq[Long],
+ rowIdRangeEnds: immutable.IndexedSeq[Long],
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
persistSourceDss: Option[Dataset[Row]]): Seq[DataSplit] = {
// Self-Merge shortcut:
@@ -547,6 +558,7 @@ case class MergeIntoPaimonDataEvolutionTable(
sourceDss,
sparkSession,
firstRowIds,
+ rowIdRangeEnds,
firstRowIdToBlobFirstRowIds,
sourceRowIdAttr.name).toSet
@@ -557,6 +569,7 @@ case class MergeIntoPaimonDataEvolutionTable(
targetDss.alias("_left").join(sourceDss, toColumn(matchedCondition),
"inner"),
sparkSession,
firstRowIds,
+ rowIdRangeEnds,
firstRowIdToBlobFirstRowIds,
"_left." + ROW_ID_NAME
).toSet
@@ -1261,12 +1274,18 @@ case class MergeIntoPaimonDataEvolutionTable(
dataset: Dataset[Row],
sparkSession: SparkSession,
firstRowIds: immutable.IndexedSeq[Long],
+ rowIdRangeEnds: immutable.IndexedSeq[Long],
firstRowIdToBlobFirstRowIds: Map[Long, List[Long]],
identifier: String): Array[Long] = {
import sparkSession.implicits._
- val firstRowIdUdf = udf((rowId: Long) => floorBinarySearch(firstRowIds,
rowId))
+ // The column may come straight from the source table, so it can hold row
ids that no longer
+ // exist (rows overwritten or dropped since the source was built) or
arbitrary values. Those
+ // match no target row: they map to no file at all rather than to the
nearest one.
+ val firstRowIdUdf =
+ udf((rowId: Long) => rowIdRangeStart(firstRowIds, rowIdRangeEnds, rowId))
dataset
- .select(firstRowIdUdf(col(identifier)))
+ .select(firstRowIdUdf(col(identifier)).as(FIRST_ROW_ID_NAME))
+ .filter(col(FIRST_ROW_ID_NAME).isNotNull)
.distinct()
.as[Long]
.flatMap(
@@ -1533,6 +1552,21 @@ object MergeIntoPaimonDataEvolutionTable {
}
}
+ /**
+ * The start of the row-id range `[firstRowIds(i), rowIdRangeEnds(i))`
holding `rowId`, or `None`
+ * when no range does.
+ */
+ private def rowIdRangeStart(
+ firstRowIds: immutable.IndexedSeq[Long],
+ rowIdRangeEnds: immutable.IndexedSeq[Long],
+ rowId: Long): Option[Long] = {
+ val index = firstRowIds.search(rowId) match {
+ case Found(foundIndex) => foundIndex
+ case InsertionPoint(insertionIndex) => insertionIndex - 1
+ }
+ if (index >= 0 && rowId < rowIdRangeEnds(index)) Some(firstRowIds(index))
else None
+ }
+
private def floorBinarySearch(indexed: immutable.IndexedSeq[Long], value:
Long): Long = {
if (indexed.isEmpty) {
throw new IllegalArgumentException("The input sorted sequence is empty.")
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 f3ba6808be..fef629d1cf 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
@@ -1252,6 +1252,69 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: merge into with _ROW_ID shortcut ignores source row
ids outside the table") {
+ withTable("source", "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, 'p1')")
+ sql("INSERT INTO target VALUES (3, 30, 'p2'), (4, 40, 'p2')")
+
+ // The source is built from this snapshot and keeps the row ids 0..3,
then partition p1 is
+ // overwritten: its rows get fresh row ids and 0 and 1 no longer exist
anywhere. Add a value
+ // below and above every live range as well.
+ sql("CREATE TABLE source (rid BIGINT, b INT)")
+ sql("INSERT INTO source SELECT _ROW_ID, b + 100 FROM target")
+ sql("INSERT INTO source VALUES (-1L, 999), (1000L, 999)")
+ sql("INSERT OVERWRITE target PARTITION (dt = 'p1') VALUES (5, 50), (6,
60)")
+ checkAnswer(
+ sql("SELECT id, _ROW_ID FROM target ORDER BY id"),
+ Seq(Row(3, 2), Row(4, 3), Row(5, 4), Row(6, 5)))
+
+ var findSplitsPlan: LogicalPlan = null
+ val listener = new QueryExecutionListener {
+ override def onSuccess(funcName: String, qe: QueryExecution,
durationNs: Long): Unit = {
+ if (qe.analyzed.collectFirst { case _: Deduplicate => true
}.nonEmpty) {
+ findSplitsPlan = qe.analyzed
+ }
+ }
+ override def onFailure(funcName: String, qe: QueryExecution,
exception: Exception): Unit =
+ onSuccess(funcName, qe, 0L)
+ }
+ spark.listenerManager.register(listener)
+ try {
+ sql("""
+ |MERGE INTO target
+ |USING source
+ |ON target._ROW_ID = source.rid
+ |WHEN MATCHED THEN UPDATE SET b = source.b
+ |WHEN NOT MATCHED THEN INSERT (id, b, dt) VALUES
(CAST(source.rid AS INT), source.b, 'p3')
+ |""".stripMargin)
+ Utils.waitUntilEventEmpty(spark)
+ } finally {
+ spark.listenerManager.unregister(listener)
+ }
+ // The row-id shortcut was taken: touched files were found without a
join.
+ assert(findSplitsPlan != null && findSplitsPlan.collect { case plan:
Join => plan }.isEmpty)
+
+ checkAnswer(
+ sql("SELECT id, b, dt FROM target ORDER BY dt, id"),
+ Seq(
+ Row(5, 50, "p1"),
+ Row(6, 60, "p1"),
+ Row(3, 130, "p2"),
+ Row(4, 140, "p2"),
+ Row(-1, 999, "p3"),
+ Row(0, 110, "p3"),
+ Row(1, 120, "p3"),
+ Row(1000, 999, "p3"))
+ )
+ }
+ }
+
test("Data Evolution: merge into table with data-evolution for Self-Merge
with _ROW_ID shortcut") {
withTable("target") {
sql(
