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 7ca1cd3603 [spark] Retry V1 UPDATE data-evolution conflicts (#8428)
7ca1cd3603 is described below
commit 7ca1cd3603f4df873188eb0ea8496e64039d9448
Author: Kerwin Zhang <[email protected]>
AuthorDate: Thu Jul 2 21:58:40 2026 +0800
[spark] Retry V1 UPDATE data-evolution conflicts (#8428)
---
docs/docs/multimodal-table/data-evolution.mdx | 21 ++++---
docs/generated/spark_connector_configuration.html | 12 ++++
.../org/apache/paimon/errors/ErrorMessages.java | 29 ++++++++++
.../paimon/operation/commit/ConflictDetection.java | 11 +++-
.../apache/paimon/spark/SparkConnectorOptions.java | 17 ++++++
.../UpdatePaimonDataEvolutionTableCommand.scala | 65 +++++++++++++++++++++-
.../org/apache/paimon/spark/util/OptionUtils.scala | 8 +++
.../paimon/spark/sql/RowTrackingTestBase.scala | 36 +++++++++++-
8 files changed, 188 insertions(+), 11 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index abafa67240..ad8c03064c 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -117,16 +117,18 @@ commit.close()
## Partial Updates
-You can update selected columns with Spark `MERGE INTO`, the Flink
-`data_evolution_merge_into` procedure, or the PyPaimon table update API. Only
-the updated column files are written; untouched columns remain in their
original
-files.
+You can update selected columns with Spark SQL `UPDATE` or `MERGE INTO`, the
+Flink `data_evolution_merge_into` procedure, or the PyPaimon table update API.
+Only the updated column files are written; untouched columns remain in their
+original files.
<Tabs groupId="data-evolution-partial-update">
<TabItem value="spark-sql" label="Spark SQL">
```sql
+UPDATE target_table SET b = b + 10 WHERE id = 1;
+
CREATE TABLE source_table (id INT, b INT);
INSERT INTO source_table VALUES (1, 11), (2, 22);
@@ -201,9 +203,14 @@ commit.close()
Notes:
-- SQL `DELETE` and standalone SQL `UPDATE` statements are not supported for
- Data Evolution tables yet. Use Spark `MERGE INTO`, the Flink procedure, or
- PyPaimon APIs.
+- 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.
- The Flink `data_evolution_merge_into` procedure currently supports updating
diff --git a/docs/generated/spark_connector_configuration.html
b/docs/generated/spark_connector_configuration.html
index 0dbc8b830a..c9b5cab70f 100644
--- a/docs/generated/spark_connector_configuration.html
+++ b/docs/generated/spark_connector_configuration.html
@@ -80,6 +80,18 @@ under the License.
<td>Boolean</td>
<td>Whether to adjust the target split size based on pruned
(projected) columns. If enabled, split size estimation uses only the columns
actually being read.</td>
</tr>
+ <tr>
+
<td><h5>write.data-evolution.update-conflict-retry.max-attempts</h5></td>
+ <td style="word-wrap: break-word;">20</td>
+ <td>Integer</td>
+ <td>Maximum attempts for Spark V1 UPDATE on data-evolution tables
when concurrent partial-column updates conflict on the same row-id range and
update columns. Values less than 2 disable retry.</td>
+ </tr>
+ <tr>
+
<td><h5>write.data-evolution.update-conflict-retry.wait-ms</h5></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Long</td>
+ <td>Wait time in milliseconds between retry attempts for Spark V1
UPDATE on data-evolution tables after row-id range update conflicts.</td>
+ </tr>
<tr>
<td><h5>write.merge-schema</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git
a/paimon-api/src/main/java/org/apache/paimon/errors/ErrorMessages.java
b/paimon-api/src/main/java/org/apache/paimon/errors/ErrorMessages.java
new file mode 100644
index 0000000000..8fd9012b1b
--- /dev/null
+++ b/paimon-api/src/main/java/org/apache/paimon/errors/ErrorMessages.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.errors;
+
+/** Shared error messages. */
+public class ErrorMessages {
+
+ public static final String DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE =
+ "For Data Evolution table, multiple 'MERGE INTO' operations have
encountered conflicts,"
+ + " updating the same file, which can render some updates
ineffective.";
+
+ private ErrorMessages() {}
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
index 39bbca3338..fcf00c792b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ConflictDetection.java
@@ -22,6 +22,7 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.Snapshot.CommitKind;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.errors.ErrorMessages;
import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
@@ -541,10 +542,16 @@ public class ConflictDetection {
if (file.firstRowId() != null
&& file.nonNullRowIdRange().from < checkNextRowId
&& columnChecker.conflictsWith(file)) {
+ LOG.debug(
+ "Data evolution row id conflict detected for table
{}, commit user {}, "
+ + "snapshot {}, file {}.",
+ tableName,
+ commitUser,
+ snapshot.id(),
+ file);
return Optional.of(
new RuntimeException(
- "For Data Evolution table, multiple 'MERGE
INTO' operations have encountered conflicts,"
- + " updating the same file, which
can render some updates ineffective."));
+
ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE));
}
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
index 5217ea0513..4138ae295a 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
@@ -67,6 +67,23 @@ public class SparkConnectorOptions {
.withDescription(
"If true, v2 write will be used. Currently, only
HASH_FIXED and BUCKET_UNAWARE bucket modes are supported. Will fall back to v1
write for other bucket modes. Currently, Spark V2 write does not support
TableCapability.STREAMING_WRITE.");
+ public static final ConfigOption<Integer>
DATA_EVOLUTION_UPDATE_CONFLICT_RETRY_MAX_ATTEMPTS =
+ key("write.data-evolution.update-conflict-retry.max-attempts")
+ .intType()
+ .defaultValue(20)
+ .withDescription(
+ "Maximum attempts for Spark V1 UPDATE on
data-evolution tables when "
+ + "concurrent partial-column updates
conflict on the same row-id "
+ + "range and update columns. Values less
than 2 disable retry.");
+
+ public static final ConfigOption<Long>
DATA_EVOLUTION_UPDATE_CONFLICT_RETRY_WAIT_MS =
+ key("write.data-evolution.update-conflict-retry.wait-ms")
+ .longType()
+ .defaultValue(10L)
+ .withDescription(
+ "Wait time in milliseconds between retry attempts
for Spark V1 UPDATE "
+ + "on data-evolution tables after row-id
range update conflicts.");
+
public static final ConfigOption<Integer> MAX_FILES_PER_TRIGGER =
key("read.stream.maxFilesPerTrigger")
.intType()
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 0a614f8f12..daaef27a85 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
@@ -18,10 +18,13 @@
package org.apache.paimon.spark.commands
+import org.apache.paimon.errors.ErrorMessages
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
import org.apache.paimon.spark.schema.PaimonMetadataColumn.ROW_ID_COLUMN
+import org.apache.paimon.spark.util.OptionUtils
+import org.apache.spark.internal.Logging
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, EqualTo, Expression}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
@@ -36,9 +39,33 @@ case class UpdatePaimonDataEvolutionTableCommand(
condition: Expression,
alignedExpressions: Seq[(Expression, Attribute)])
extends PaimonLeafRunnableCommand
- with SupportsSubquery {
+ with SupportsSubquery
+ with Logging {
override def run(sparkSession: SparkSession): Seq[Row] = {
+ val maxAttempts = math.max(1,
OptionUtils.dataEvolutionUpdateConflictRetryMaxAttempts())
+ val retryWaitMs = math.max(0L,
OptionUtils.dataEvolutionUpdateConflictRetryWaitMs())
+ val canRetry = deterministicUpdate
+ var attempt = 1
+
+ while (attempt < maxAttempts) {
+ try {
+ return runOnce(sparkSession)
+ } catch {
+ case e: RuntimeException if canRetry &&
isDataEvolutionUpdateConflict(e) =>
+ val nextAttempt = attempt + 1
+ logInfo(
+ s"Retry Spark V1 UPDATE for data-evolution table after concurrent
update conflict " +
+ s"(next attempt $nextAttempt/$maxAttempts).")
+ sleepBeforeRetry(retryWaitMs)
+ attempt += 1
+ }
+ }
+
+ runOnce(sparkSession)
+ }
+
+ private def runOnce(sparkSession: SparkSession): Seq[Row] = {
val (updateTable, updateRelation) =
MergeIntoPaimonDataEvolutionTable.withMatchedUpdateScanOptions(v2Table,
relation)
val targetRowId = rowIdAttribute(updateRelation)
@@ -60,6 +87,42 @@ case class UpdatePaimonDataEvolutionTableCommand(
Nil).run(sparkSession)
}
+ private def deterministicUpdate: Boolean = {
+ condition.deterministic && alignedExpressions.forall {
+ case (expression, _) =>
+ expression.deterministic
+ }
+ }
+
+ private def sleepBeforeRetry(retryWaitMs: Long): Unit = {
+ if (retryWaitMs > 0) {
+ try {
+ Thread.sleep(retryWaitMs)
+ } catch {
+ case e: InterruptedException =>
+ Thread.currentThread().interrupt()
+ throw new RuntimeException(
+ "Interrupted while retrying data-evolution UPDATE conflict.",
+ e)
+ }
+ }
+ }
+
+ private def isDataEvolutionUpdateConflict(e: Throwable): Boolean = {
+ var current = e
+ while (current != null) {
+ val message = current.getMessage
+ if (
+ message != null &&
+ message.contains(ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE)
+ ) {
+ return true
+ }
+ current = current.getCause
+ }
+ false
+ }
+
private def updatedRowIdSource(
updateTable: SparkTable,
updateRelation: DataSourceV2Relation,
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
index b21b83de26..7aff2966e6 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
@@ -118,6 +118,14 @@ object OptionUtils extends SQLConfHelper with Logging {
getOptionString(SparkConnectorOptions.TYPE_WIDENING).toBoolean
}
+ def dataEvolutionUpdateConflictRetryMaxAttempts(): Int = {
+
getOptionString(SparkConnectorOptions.DATA_EVOLUTION_UPDATE_CONFLICT_RETRY_MAX_ATTEMPTS).toInt
+ }
+
+ def dataEvolutionUpdateConflictRetryWaitMs(): Long = {
+
getOptionString(SparkConnectorOptions.DATA_EVOLUTION_UPDATE_CONFLICT_RETRY_WAIT_MS).toLong
+ }
+
def v1FunctionEnabled(): Boolean = {
getOptionString(SparkCatalogOptions.V1FUNCTION_ENABLED).toBoolean
}
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 6c63e1919d..dc96e3ab1e 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
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.sql
import org.apache.paimon.Snapshot.CommitKind
+import org.apache.paimon.errors.ErrorMessages
import org.apache.paimon.spark.PaimonMetrics.RESULTED_TABLE_FILES
import org.apache.paimon.spark.PaimonSparkTestBase
import org.apache.paimon.spark.read.PaimonSplitScan
@@ -149,7 +150,7 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
|WHEN MATCHED THEN
|UPDATE SET t.id = s.id, t.b = s.b + t.b, t.c = s.c +
t.c
|""".stripMargin).collect(),
- "multiple 'MERGE INTO' operations have encountered conflicts"
+ ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE
)
}
@@ -1099,6 +1100,39 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: V1 update retries concurrent update conflicts") {
+ withSparkSQLConf(
+ "spark.paimon.write.use-v2-write" -> "false",
+ "spark.paimon.write.data-evolution.update-conflict-retry.max-attempts"
-> "50",
+ "spark.paimon.write.data-evolution.update-conflict-retry.wait-ms" -> "10"
+ ) {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, b INT, c INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
+ sql("INSERT INTO t VALUES (1, 0, 0)")
+
+ val ready = new CountDownLatch(4)
+ val start = new CountDownLatch(1)
+ val updates = (1 to 4).map {
+ _ =>
+ Future {
+ ready.countDown()
+ assert(start.await(30, TimeUnit.SECONDS))
+ for (_ <- 1 to 5) {
+ sql("UPDATE t SET b = b + 1 WHERE id = 1").collect()
+ }
+ }
+ }
+
+ assert(ready.await(30, TimeUnit.SECONDS))
+ start.countDown()
+ updates.foreach(Await.result(_, 120.seconds))
+
+ checkAnswer(sql("SELECT b FROM t"), Seq(Row(20)))
+ }
+ }
+ }
+
test("Data Evolution: V1 update partition column throws exception") {
withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
withTable("t") {
