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 71b815d6ff [spark] Stabilize row tracking concurrent tests (#8307)
71b815d6ff is described below
commit 71b815d6ff833def99b62818ed9790df1ed5cfdb
Author: QuakeWang <[email protected]>
AuthorDate: Mon Jun 22 18:07:59 2026 +0800
[spark] Stabilize row tracking concurrent tests (#8307)
Row tracking concurrent tests could fail on transient commit conflicts
and snapshot reads while merge and compaction run in parallel. The
previous retry logic was duplicated and, in one path, retried without a
bound.
This PR centralizes the retry helper for these concurrent tests, adds a
bounded retry with a small backoff, preserves the final exception on
failure, and narrows snapshot missing detection to require `Snapshot
file` and `does not exist` in the same throwable message.
---
.../paimon/spark/sql/RowTrackingTestBase.scala | 137 ++++++++++++---------
1 file changed, 76 insertions(+), 61 deletions(-)
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 5efcaf79b8..f53089470d 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
@@ -44,6 +44,43 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
import testImplicits._
+ private val MaxRetryAttempts = 20
+ private val RetryIntervalMillis = 10L
+
+ private def doWithRetry(doAction: () => Unit, retryableMessages: String*):
Unit = {
+ var attempts = 0
+ while (true) {
+ try {
+ doAction.apply()
+ return
+ } catch {
+ case e: Exception =>
+ attempts += 1
+ if (!isRetryable(e, retryableMessages) || attempts >=
MaxRetryAttempts) {
+ throw e
+ }
+ Thread.sleep(RetryIntervalMillis)
+ }
+ }
+ }
+
+ private def isRetryable(e: Throwable, retryableMessages: Seq[String]):
Boolean = {
+ hasMessage(e, "Snapshot file", "does not exist") ||
+ retryableMessages.exists(message => hasMessage(e, message))
+ }
+
+ private def hasMessage(e: Throwable, fragments: String*): Boolean = {
+ var current = e
+ while (current != null) {
+ val message = current.getMessage
+ if (message != null && fragments.forall(message.contains)) {
+ return true
+ }
+ current = current.getCause
+ }
+ false
+ }
+
test("Data Evolution: concurrent merge and compact") {
withTable("s", "t") {
sql(s"""
@@ -104,27 +141,16 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
Seq((1, 1, 1)).toDF("id", "b", "c").createOrReplaceTempView("s")
def doMerge(): Unit = {
- var success = false
- while (!success) {
- try {
- sql(s"""
- |MERGE INTO t
- |USING s
- |ON t.id = s.id
- |WHEN MATCHED THEN
- |UPDATE SET t.id = s.id, t.b = s.b + t.b, t.c = s.c + t.c
- |""".stripMargin).collect()
- success = true
- } catch {
- case e: Exception =>
- if (
- !e.getMessage.contains(
- "multiple 'MERGE INTO' operations have encountered
conflicts")
- ) {
- throw e
- }
- }
- }
+ doWithRetry(
+ () => sql(s"""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |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"
+ )
}
val mergeInto1 = Future {
@@ -159,25 +185,25 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
val mergeB = Future {
for (_ <- 1 to 10) {
- sql(s"""
- |MERGE INTO t
- |USING sb
- |ON t.id = sb.id
- |WHEN MATCHED THEN
- |UPDATE SET t.b = sb.b + t.b
- |""".stripMargin).collect()
+ doWithRetry(() => sql(s"""
+ |MERGE INTO t
+ |USING sb
+ |ON t.id = sb.id
+ |WHEN MATCHED THEN
+ |UPDATE SET t.b = sb.b + t.b
+ |""".stripMargin).collect())
}
}
val mergeC = Future {
for (_ <- 1 to 10) {
- sql(s"""
- |MERGE INTO t
- |USING sc
- |ON t.id = sc.id
- |WHEN MATCHED THEN
- |UPDATE SET t.c = sc.c + t.c
- |""".stripMargin).collect()
+ doWithRetry(() => sql(s"""
+ |MERGE INTO t
+ |USING sc
+ |ON t.id = sc.id
+ |WHEN MATCHED THEN
+ |UPDATE SET t.c = sc.c + t.c
+ |""".stripMargin).collect())
}
}
@@ -199,33 +225,19 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
sql("INSERT INTO t VALUES (1, 0, 0)")
Seq((1, 1, 1)).toDF("id", "b", "c").createOrReplaceTempView("s")
- def doWithRetry(doAction: () => Unit): Unit = {
- var success = false
- while (!success) {
- try {
- doAction.apply()
- success = true
- } catch {
- case e: Exception =>
- if (
- !e.getMessage.contains("multiple 'MERGE INTO' and 'COMPACT'
operations")
- && !e.getMessage.contains("Row ID existence conflict")
- ) {
- throw e
- }
- }
- }
- }
-
val mergeInto = Future {
for (i <- 1 to 10) {
- doWithRetry(() => sql(s"""
- |MERGE INTO t
- |USING s
- |ON t.id = s.id
- |WHEN MATCHED THEN
- |UPDATE SET t.id = s.id, t.b = s.b + t.b,
t.c = s.c + t.c
- |""".stripMargin).collect())
+ doWithRetry(
+ () => sql(s"""
+ |MERGE INTO t
+ |USING s
+ |ON t.id = s.id
+ |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' and 'COMPACT' operations",
+ "Row ID existence conflict"
+ )
if (i > 1) {
sql(s"INSERT INTO t VALUES ($i, $i, $i)")
}
@@ -244,7 +256,10 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
while (!canBeCompacted) {
Thread.sleep(1)
}
- doWithRetry(() => sql("CALL sys.compact(table => 't')"))
+ doWithRetry(
+ () => sql("CALL sys.compact(table => 't')"),
+ "multiple 'MERGE INTO' and 'COMPACT' operations",
+ "Row ID existence conflict")
}
}
