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 26da929908 [spark] Fix data evolution matched updates with global
index (#8411)
26da929908 is described below
commit 26da929908649c46b8c24d4c35b9d636c7d4679a
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 1 17:59:57 2026 +0800
[spark] Fix data evolution matched updates with global index (#8411)
Fixes Data Evolution matched-update scans with global index by forcing
row-id discovery to use full global-index search on a pinned current
snapshot. This prevents rows inserted after global index creation from
being skipped during UPDATE or MERGE matched updates.
---
.../MergeIntoPaimonDataEvolutionTable.scala | 68 ++++++++++++++++++++--
.../MergeIntoPaimonDataEvolutionTable.scala | 68 ++++++++++++++++++++--
.../UpdatePaimonDataEvolutionTableCommand.scala | 17 ++++--
.../paimon/spark/sql/RowTrackingTestBase.scala | 56 ++++++++++++++++++
4 files changed, 193 insertions(+), 16 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 c14ee04648..e4e2d9b1e1 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
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.commands
+import org.apache.paimon.CoreOptions
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
import org.apache.paimon.Snapshot
import org.apache.paimon.data.BinaryRow
@@ -36,6 +37,7 @@ import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl}
import org.apache.paimon.table.source.DataSplit
import org.apache.paimon.table.source.snapshot.SnapshotReader
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil
import org.apache.paimon.types.DataTypeRoot.BLOB
import org.apache.paimon.types.RowType
import org.apache.paimon.types.VectorType.isVectorStoreFile
@@ -53,6 +55,8 @@ import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.paimon.shims.SparkShimLoader
import org.apache.spark.sql.types.{BooleanType, StructType}
+import java.util.{HashMap => JHashMap}
+
import scala.collection.{immutable, mutable}
import scala.collection.JavaConverters._
import scala.collection.Searching.{search, Found, InsertionPoint}
@@ -87,7 +91,19 @@ case class MergeIntoPaimonDataEvolutionTable(
import MergeIntoPaimonDataEvolutionTable._
- override val table: FileStoreTable =
v2Table.getTable.asInstanceOf[FileStoreTable]
+ private lazy val originalTargetRelation: DataSourceV2Relation =
+ PaimonRelation.getPaimonRelation(targetTable)
+
+ private lazy val matchedUpdateScanTarget: (SparkTable, DataSourceV2Relation)
=
+ if (matchedActions.nonEmpty) {
+ withMatchedUpdateScanOptions(v2Table, originalTargetRelation)
+ } else {
+ (v2Table, originalTargetRelation)
+ }
+
+ private lazy val targetSparkTable: SparkTable = matchedUpdateScanTarget._1
+
+ override lazy val table: FileStoreTable =
targetSparkTable.getTable.asInstanceOf[FileStoreTable]
private val updateColumns: Set[AttributeReference] = {
val columns = mutable.Set[AttributeReference]()
@@ -121,7 +137,9 @@ case class MergeIntoPaimonDataEvolutionTable(
private lazy val isSelfMergeOnRowId: Boolean = {
if (!isPaimonTable(sourceTable)) {
false
- } else if
(!targetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name))
{
+ } else if (
+
!originalTargetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name)
+ ) {
false
} else {
matchedCondition match {
@@ -139,10 +157,9 @@ case class MergeIntoPaimonDataEvolutionTable(
"NOT MATCHED BY SOURCE are not supported."
)
- private lazy val targetRelation: DataSourceV2Relation =
- PaimonRelation.getPaimonRelation(targetTable)
+ private lazy val targetRelation: DataSourceV2Relation =
matchedUpdateScanTarget._2
- lazy val tableSchema: StructType = v2Table.schema
+ lazy val tableSchema: StructType = targetSparkTable.schema
override def run(sparkSession: SparkSession): Seq[Row] = {
// Persist the schema that the analyzer evolved in memory (commit deferred
to execution).
@@ -821,6 +838,47 @@ object MergeIntoPaimonDataEvolutionTable {
final private val FIRST_ROW_ID_NAME = "_FIRST_ROW_ID";
final private val RAW_BLOB_PLACEHOLDER_MARKER_PREFIX =
"__paimon_raw_blob_placeholder_"
+ private[commands] def withMatchedUpdateScanOptions(
+ v2Table: SparkTable,
+ relation: DataSourceV2Relation): (SparkTable, DataSourceV2Relation) = {
+ val table = v2Table.getTable.asInstanceOf[FileStoreTable]
+ val fullSearchMode = CoreOptions.GlobalIndexSearchMode.FULL.toString
+ Option(TimeTravelUtil.tryTravelOrLatest(table)) match {
+ case None =>
+ (v2Table, relation)
+ case Some(snapshot) =>
+ val configuredSnapshotId =
+ Option(table.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key()))
+ val snapshotId = snapshot.id().toString
+ if (
+ configuredSnapshotId.contains(snapshotId) &&
+ fullSearchMode.equalsIgnoreCase(
+ table.options().get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key()))
+ ) {
+ (v2Table, relation)
+ } else {
+ val dynamicOptions = new JHashMap[String, String]()
+ timeTravelOptionKeys.foreach(dynamicOptions.put(_, null))
+ dynamicOptions.put(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
fullSearchMode)
+ dynamicOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId)
+
+ val scanTable = SparkTable.of(table.copy(dynamicOptions))
+ val scanRelation =
+ SparkShimLoader.shim.copyDataSourceV2Relation(relation, scanTable,
relation.output)
+ (scanTable, scanRelation)
+ }
+ }
+ }
+
+ private def timeTravelOptionKeys: Seq[String] = Seq(
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_VERSION.key()
+ )
+
private[commands] def isModifiedAssignment(assignment: Assignment): Boolean
= {
!sameAttributeReference(assignment.key, assignment.value)
}
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 7464176c87..15379a1147 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
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.commands
+import org.apache.paimon.CoreOptions
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
import org.apache.paimon.Snapshot
import org.apache.paimon.data.BinaryRow
@@ -36,6 +37,7 @@ import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl}
import org.apache.paimon.table.source.DataSplit
import org.apache.paimon.table.source.snapshot.SnapshotReader
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil
import org.apache.paimon.types.DataTypeRoot.BLOB
import org.apache.paimon.types.RowType
import org.apache.paimon.types.VectorType.isVectorStoreFile
@@ -53,6 +55,8 @@ import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.paimon.shims.SparkShimLoader
import org.apache.spark.sql.types.{BooleanType, StructType}
+import java.util.{HashMap => JHashMap}
+
import scala.collection.{immutable, mutable}
import scala.collection.JavaConverters._
import scala.collection.Searching.{search, Found, InsertionPoint}
@@ -87,7 +91,19 @@ case class MergeIntoPaimonDataEvolutionTable(
import MergeIntoPaimonDataEvolutionTable._
- override val table: FileStoreTable =
v2Table.getTable.asInstanceOf[FileStoreTable]
+ private lazy val originalTargetRelation: DataSourceV2Relation =
+ PaimonRelation.getPaimonRelation(targetTable)
+
+ private lazy val matchedUpdateScanTarget: (SparkTable, DataSourceV2Relation)
=
+ if (matchedActions.nonEmpty) {
+ withMatchedUpdateScanOptions(v2Table, originalTargetRelation)
+ } else {
+ (v2Table, originalTargetRelation)
+ }
+
+ private lazy val targetSparkTable: SparkTable = matchedUpdateScanTarget._1
+
+ override lazy val table: FileStoreTable =
targetSparkTable.getTable.asInstanceOf[FileStoreTable]
private val updateColumns: Set[AttributeReference] = {
val columns = mutable.Set[AttributeReference]()
@@ -121,7 +137,9 @@ case class MergeIntoPaimonDataEvolutionTable(
private lazy val isSelfMergeOnRowId: Boolean = {
if (!isPaimonTable(sourceTable)) {
false
- } else if
(!targetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name))
{
+ } else if (
+
!originalTargetRelation.name.equals(PaimonRelation.getPaimonRelation(sourceTable).name)
+ ) {
false
} else {
matchedCondition match {
@@ -139,10 +157,9 @@ case class MergeIntoPaimonDataEvolutionTable(
"NOT MATCHED BY SOURCE are not supported."
)
- private lazy val targetRelation: DataSourceV2Relation =
- PaimonRelation.getPaimonRelation(targetTable)
+ private lazy val targetRelation: DataSourceV2Relation =
matchedUpdateScanTarget._2
- lazy val tableSchema: StructType = v2Table.schema
+ lazy val tableSchema: StructType = targetSparkTable.schema
override def run(sparkSession: SparkSession): Seq[Row] = {
// Persist the schema that the analyzer evolved in memory (commit deferred
to execution).
@@ -820,6 +837,47 @@ object MergeIntoPaimonDataEvolutionTable {
final private val FIRST_ROW_ID_NAME = "_FIRST_ROW_ID";
final private val RAW_BLOB_PLACEHOLDER_MARKER_PREFIX =
"__paimon_raw_blob_placeholder_"
+ private[commands] def withMatchedUpdateScanOptions(
+ v2Table: SparkTable,
+ relation: DataSourceV2Relation): (SparkTable, DataSourceV2Relation) = {
+ val table = v2Table.getTable.asInstanceOf[FileStoreTable]
+ val fullSearchMode = CoreOptions.GlobalIndexSearchMode.FULL.toString
+ Option(TimeTravelUtil.tryTravelOrLatest(table)) match {
+ case None =>
+ (v2Table, relation)
+ case Some(snapshot) =>
+ val configuredSnapshotId =
+ Option(table.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key()))
+ val snapshotId = snapshot.id().toString
+ if (
+ configuredSnapshotId.contains(snapshotId) &&
+ fullSearchMode.equalsIgnoreCase(
+ table.options().get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key()))
+ ) {
+ (v2Table, relation)
+ } else {
+ val dynamicOptions = new JHashMap[String, String]()
+ timeTravelOptionKeys.foreach(dynamicOptions.put(_, null))
+ dynamicOptions.put(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
fullSearchMode)
+ dynamicOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId)
+
+ val scanTable = SparkTable.of(table.copy(dynamicOptions))
+ val scanRelation =
+ SparkShimLoader.shim.copyDataSourceV2Relation(relation, scanTable,
relation.output)
+ (scanTable, scanRelation)
+ }
+ }
+ }
+
+ private def timeTravelOptionKeys: Seq[String] = Seq(
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_VERSION.key()
+ )
+
private[commands] def isModifiedAssignment(assignment: Assignment): Boolean
= {
!sameAttributeReference(assignment.key, assignment.value)
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
index 9e2c35e3e4..0a614f8f12 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
@@ -39,8 +39,10 @@ case class UpdatePaimonDataEvolutionTableCommand(
with SupportsSubquery {
override def run(sparkSession: SparkSession): Seq[Row] = {
- val targetRowId = rowIdAttribute(relation)
- val sourceTable = updatedRowIdSource(targetRowId)
+ val (updateTable, updateRelation) =
+ MergeIntoPaimonDataEvolutionTable.withMatchedUpdateScanOptions(v2Table,
relation)
+ val targetRowId = rowIdAttribute(updateRelation)
+ val sourceTable = updatedRowIdSource(updateTable, updateRelation,
targetRowId)
val sourceRowId = sourceTable.output.head.asInstanceOf[AttributeReference]
val matchedCondition = EqualTo(targetRowId, sourceRowId)
@@ -49,8 +51,8 @@ case class UpdatePaimonDataEvolutionTableCommand(
alignedExpressions.map { case (expression, attribute) =>
Assignment(attribute, expression) })
MergeIntoPaimonDataEvolutionTable(
- v2Table,
- relation,
+ updateTable,
+ updateRelation,
sourceTable,
matchedCondition,
Seq(updateAction),
@@ -58,13 +60,16 @@ case class UpdatePaimonDataEvolutionTableCommand(
Nil).run(sparkSession)
}
- private def updatedRowIdSource(targetRowId: AttributeReference): Project = {
+ private def updatedRowIdSource(
+ updateTable: SparkTable,
+ updateRelation: DataSourceV2Relation,
+ targetRowId: AttributeReference): Project = {
val conditionReferences = condition.references.toSeq.collect {
case attr: AttributeReference => attr
}
val readOutput = deduplicateByExprId(conditionReferences :+ targetRowId)
val sourceScan =
- SparkShimLoader.shim.copyDataSourceV2Relation(relation, v2Table,
readOutput)
+ SparkShimLoader.shim.copyDataSourceV2Relation(updateRelation,
updateTable, readOutput)
// Keep the Filter visible for conditional UPDATEs. The data-evolution
MERGE command uses a
// self-merge shortcut for Project(PaimonRelation); if a WHERE update were
shaped that way, the
// shortcut would bypass the source join path and update every row.
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 c7f6ea86e3..f960b59d9b 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
@@ -1027,6 +1027,62 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: V1 update with global index updates unindexed rows") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, name STRING, b INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'btree-index.records-per-range' = '1000')
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 'old', 10)")
+ sql(
+ "CALL sys.create_global_index(table => 'test.t', index_column =>
'name', " +
+ "index_type => 'btree')")
+ sql("INSERT INTO t VALUES (2, 'new', 20)")
+
+ sql("UPDATE t SET b = 21 WHERE name = 'new'")
+
+ checkAnswer(
+ sql("SELECT id, name, b FROM t ORDER BY id"),
+ Seq(Row(1, "old", 10), Row(2, "new", 21))
+ )
+ }
+ }
+ }
+
+ test("Data Evolution: merge with global index updates unindexed rows") {
+ withTable("s", "t") {
+ sql("CREATE TABLE s (dummy INT, b INT)")
+ sql("INSERT INTO s VALUES (1, 21)")
+
+ sql("""
+ |CREATE TABLE t (id INT, name STRING, b INT) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'btree-index.records-per-range' = '1000')
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 'old', 10)")
+ sql(
+ "CALL sys.create_global_index(table => 'test.t', index_column =>
'name', " +
+ "index_type => 'btree')")
+ sql("INSERT INTO t VALUES (2, 'new', 20)")
+
+ sql("""
+ |MERGE INTO t
+ |USING s
+ |ON t.name = 'new' AND s.dummy = 1
+ |WHEN MATCHED THEN UPDATE SET t.b = s.b
+ |""".stripMargin)
+
+ checkAnswer(
+ sql("SELECT id, name, b FROM t ORDER BY id"),
+ Seq(Row(1, "old", 10), Row(2, "new", 21))
+ )
+ }
+ }
+
test("Data Evolution: V1 update table with data-evolution without
condition") {
withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
withTable("t") {
