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 721372af9b [spark] Skip driver metrics when SQL execution context is
missing (#10563)
721372af9b is described below
commit 721372af9b6fa7702a2f85a955f6218feb2e4f1c
Author: Joey <[email protected]>
AuthorDate: Fri Oct 16 08:51:09 2026 +0800
[spark] Skip driver metrics when SQL execution context is missing (#10563)
---
.../paimon/spark/catalyst/Compatibility.scala | 3 +-
.../paimon/spark/catalyst/Compatibility.scala | 3 +-
.../paimon/spark/catalyst/Compatibility.scala | 3 +-
.../paimon/spark/catalyst/Compatibility.scala | 3 +-
.../apache/paimon/spark/write/WriteHelper.scala | 8 ++
.../paimon/spark/write/PaimonBatchWriteTest.java | 38 +++++++
.../apache/paimon/spark/sql/WriteMetricsTest.scala | 122 +++++++++++++++++++++
7 files changed, 176 insertions(+), 4 deletions(-)
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
index b7c1e31f0f..80e8aab86b 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
@@ -66,6 +66,7 @@ object Compatibility {
}
def getExecutionMetrics(spark: SparkSession, executionId: Long):
Seq[SQLPlanMetric] = {
- spark.sharedState.statusStore.execution(executionId).get.metrics.toSeq
+ // The SQL status store is updated asynchronously and may not have this
execution yet.
+
spark.sharedState.statusStore.execution(executionId).map(_.metrics.toSeq).getOrElse(Seq.empty)
}
}
diff --git
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
index 4703e43663..1f3753c999 100644
---
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
+++
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
@@ -55,6 +55,7 @@ object Compatibility {
}
def getExecutionMetrics(spark: SparkSession, executionId: Long):
Seq[SQLPlanMetric] = {
- spark.sharedState.statusStore.execution(executionId).get.metrics.toSeq
+ // The SQL status store is updated asynchronously and may not have this
execution yet.
+
spark.sharedState.statusStore.execution(executionId).map(_.metrics.toSeq).getOrElse(Seq.empty)
}
}
diff --git
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
index 2c3dbbf06f..8f81a6668e 100644
---
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
+++
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
@@ -55,6 +55,7 @@ object Compatibility {
}
def getExecutionMetrics(spark: SparkSession, executionId: Long):
Seq[SQLPlanMetric] = {
- spark.sharedState.statusStore.execution(executionId).get.metrics.toSeq
+ // The SQL status store is updated asynchronously and may not have this
execution yet.
+
spark.sharedState.statusStore.execution(executionId).map(_.metrics.toSeq).getOrElse(Seq.empty)
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
index 7222e0a27c..89de804eec 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
@@ -61,6 +61,7 @@ object Compatibility {
}
def getExecutionMetrics(spark: SparkSession, executionId: Long):
Seq[SQLPlanMetric] = {
- spark.sharedState.statusStore.execution(executionId).get.metrics.toSeq
+ // The SQL status store is updated asynchronously and may not have this
execution yet.
+
spark.sharedState.statusStore.execution(executionId).map(_.metrics.toSeq).getOrElse(Seq.empty)
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
index 5393697a50..b96c36561f 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/WriteHelper.scala
@@ -51,7 +51,15 @@ trait WriteHelper extends Logging {
def postDriverMetrics(commitMetrics: Array[CustomTaskMetric]): Unit = {
val spark = PaimonSparkSession.active
val executionId =
spark.sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ if (executionId == null) {
+ logDebug("Skipping driver metrics because there is no SQL execution ID")
+ return
+ }
val executionMetrics = Compatibility.getExecutionMetrics(spark,
executionId.toLong).distinct
+ if (executionMetrics.isEmpty) {
+ logDebug(s"Skipping driver metrics because execution $executionId has no
SQL metrics")
+ return
+ }
val metricUpdates = executionMetrics.flatMap {
m =>
commitMetrics.find(
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/write/PaimonBatchWriteTest.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/write/PaimonBatchWriteTest.java
index 6db8176d63..c3442825a9 100644
---
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/write/PaimonBatchWriteTest.java
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/write/PaimonBatchWriteTest.java
@@ -26,8 +26,11 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.table.CatalogEnvironment;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.spark.sql.connector.metric.CustomTaskMetric;
+import org.apache.spark.sql.connector.write.WriterCommitMessage;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.Test;
@@ -35,13 +38,48 @@ import java.util.Collections;
import scala.Option;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Tests for {@link PaimonBatchWrite}. */
public class PaimonBatchWriteTest {
+ @Test
+ public void testCommitFailureStillPropagatesAndClosesCommit() throws
Exception {
+ FileStoreTable table = mock(FileStoreTable.class);
+
when(table.coreOptions()).thenReturn(CoreOptions.fromMap(Collections.emptyMap()));
+ BatchWriteBuilder builder = mock(BatchWriteBuilder.class);
+ when(table.newBatchWriteBuilder()).thenReturn(builder);
+ BatchTableCommit commit = mock(BatchTableCommit.class);
+ when(builder.newCommit()).thenReturn(commit);
+ IllegalStateException failure = new IllegalStateException("commit
failed");
+ doThrow(failure).when(commit).commit(anyList());
+
+ StructType schema = new StructType();
+ PaimonBatchWrite batchWrite =
+ new PaimonBatchWrite(
+ table, schema, schema, Option.empty(), Option.empty(),
Option.empty()) {
+ @Override
+ public void postDriverMetrics(CustomTaskMetric[] metrics) {
+ throw new AssertionError("Metrics must not be posted
after commit failure");
+ }
+ };
+
+ WriterCommitMessage[] messages = new WriterCommitMessage[0];
+ assertSame(
+ failure,
+ assertThrows(IllegalStateException.class, () ->
batchWrite.commit(messages)));
+ verify(commit).close();
+ batchWrite.abort(messages);
+ verify(commit, never()).abort(anyList());
+ }
+
@Test
public void testResolveBlobDescriptorSourceBeforeCreatingTaskWriters()
throws Exception {
Identifier sourceIdentifier = Identifier.fromString("db.source");
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/WriteMetricsTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/WriteMetricsTest.scala
new file mode 100644
index 0000000000..9ad3d4d563
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/WriteMetricsTest.scala
@@ -0,0 +1,122 @@
+/*
+ * 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.spark.sql
+
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.catalyst.Compatibility
+import org.apache.paimon.spark.write.PaimonBatchWrite
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import org.apache.spark.sql.execution.LocalTableScanExec
+import org.apache.spark.sql.execution.SQLExecution
+import org.apache.spark.sql.paimon.Utils
+
+class WriteMetricsTest extends PaimonSparkTestBase {
+
+ override protected def sparkConf: SparkConf =
+ super.sparkConf.set("spark.paimon.write.use-v2-write", "true")
+
+ test("missing execution metrics") {
+ assert(spark.sharedState.statusStore.execution(-1L).isEmpty)
+ assert(Compatibility.getExecutionMetrics(spark, -1L).isEmpty)
+ }
+
+ test("empty dataframe append with and without execution metrics") {
+ for (disableListener <- Seq(false, true)) {
+ withTable("source", "target") {
+ sql(
+ "CREATE TABLE source (id INT, v INT) TBLPROPERTIES ('primary-key' =
'id', 'bucket' = '1')")
+ sql(
+ "CREATE TABLE target (id INT, v INT) TBLPROPERTIES ('primary-key' =
'id', 'bucket' = '1')")
+ sql("INSERT INTO target VALUES (1, 10)")
+ val before = loadTable("target").snapshotManager().latestSnapshotId()
+ val source = spark.read.table("source").limit(0)
+ val count = source.count()
+ assert(count == 0)
+
assert(source.queryExecution.executedPlan.isInstanceOf[LocalTableScanExec])
+ def append(): Unit = {
+ source
+ .coalesce(math.max(1, (count / 10000).toInt))
+ .write
+ .format("paimon")
+ .mode("append")
+ .saveAsTable("target")
+ }
+ if (disableListener) {
+ withoutStatusListener(append())
+ } else {
+ append()
+ }
+ checkAnswer(spark.table("target"), Row(1, 10) :: Nil)
+ assert(loadTable("target").snapshotManager().latestSnapshotId() ==
before)
+ }
+ }
+ }
+
+ test("nonempty dataframe append without execution metrics") {
+ import testImplicits._
+ withTable("target") {
+ sql(
+ "CREATE TABLE target (id INT, v INT) TBLPROPERTIES ('primary-key' =
'id', 'bucket' = '1')")
+ withoutStatusListener {
+ Seq((1, 10), (2, 20))
+ .toDF("id", "v")
+ .write
+ .format("paimon")
+ .mode("append")
+ .saveAsTable("target")
+ }
+ assert(loadTable("target").latestSnapshot().isPresent)
+ checkAnswer(spark.table("target"), Seq(Row(1, 10), Row(2, 20)))
+ }
+ }
+
+ test("nonempty batch commit without execution id") {
+ withTable("target") {
+ sql(
+ "CREATE TABLE target (id INT, v INT) TBLPROPERTIES ('primary-key' =
'id', 'bucket' = '1')")
+ val table = loadTable("target")
+ val schema = spark.table("target").schema
+ val batchWrite = PaimonBatchWrite(table, schema, schema, None, None)
+ val writer = batchWrite.createBatchWriterFactory(null).createWriter(0,
0L)
+ try {
+ writer.write(new GenericInternalRow(Array[Any](1, 10)))
+ val message = writer.commit()
+
assert(spark.sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) ==
null)
+ batchWrite.commit(Array(message))
+ } finally {
+ writer.close()
+ }
+ assert(table.latestSnapshot().isPresent)
+ checkAnswer(spark.table("target"), Row(1, 10) :: Nil)
+ }
+ }
+
+ // Use the real SQL status store with no listener to deterministically cover
an absent
+ // execution record while Spark still supplies the write's actual execution
id.
+ private def withoutStatusListener(f: => Unit): Unit = {
+ Utils.waitUntilEventEmpty(spark)
+ val listener = spark.sharedState.statusStore.listener.get
+ spark.sparkContext.removeSparkListener(listener)
+ try f
+ finally spark.sparkContext.addSparkListener(listener)
+ }
+}
