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 2b46053b5a [spark] Broadcast data evolution firstRowIdToPartitionMap
(#8051)
2b46053b5a is described below
commit 2b46053b5acc4bcd7cb845a6bb1a5e18429c52b5
Author: Liurnly <[email protected]>
AuthorDate: Sun May 31 21:46:15 2026 +0800
[spark] Broadcast data evolution firstRowIdToPartitionMap (#8051)
`firstRowIdToPartitionMap` is used by data evolution MERGE INTO writes
to locate the original partition and row count for each data file from
its `firstRowId`. This mapping is built from scanned `DataSplit`s and
then used by each Spark task while writing evolved rows.
For merge plans that affect many files across many partitions, the map
can become large. Capturing it directly in every task closure duplicates
the serialized map for each task and can make Spark task serialization
very large.
---
.../spark/commands/DataEvolutionPaimonWriter.scala | 42 ++++++++++++----------
.../spark/write/DataEvolutionTableDataWrite.scala | 6 ++--
2 files changed, 28 insertions(+), 20 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
index 2b78333223..c49368d1bf 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
@@ -19,7 +19,6 @@
package org.apache.paimon.spark.commands
import org.apache.paimon.CoreOptions
-import org.apache.paimon.data.BinaryRow
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.spark.write.{DataEvolutionTableDataWrite,
WriteHelper, WriteTaskResult}
import org.apache.paimon.table.FileStoreTable
@@ -28,6 +27,7 @@ import org.apache.paimon.table.source.DataSplit
import org.apache.paimon.types.DataType
import org.apache.paimon.types.DataTypeRoot.BLOB
import org.apache.paimon.types.VectorType.isVectorStoreFile
+import org.apache.paimon.utils.SerializationUtils
import org.apache.spark.sql._
@@ -39,21 +39,6 @@ import scala.collection.mutable
case class DataEvolutionPaimonWriter(paimonTable: FileStoreTable, dataSplits:
Seq[DataSplit])
extends WriteHelper {
- private lazy val firstRowIdToPartitionMap: mutable.HashMap[Long, (BinaryRow,
Long)] = {
- val firstRowIdToPartitionMap = new mutable.HashMap[Long, (BinaryRow, Long)]
- dataSplits.foreach(
- split =>
- split
- .dataFiles()
- .asScala
- .filter(file => !isBlobFile(file.fileName()) &&
!isVectorStoreFile(file.fileName()))
- .foreach(
- file =>
- firstRowIdToPartitionMap
- .put(file.firstRowId(), (split.partition(), file.rowCount()))))
- firstRowIdToPartitionMap
- }
-
// File rolling will never be performed
override val table: FileStoreTable =
paimonTable.copy(Collections.singletonMap(CoreOptions.TARGET_FILE_SIZE.key(),
"99999 G"))
@@ -77,14 +62,35 @@ case class DataEvolutionPaimonWriter(paimonTable:
FileStoreTable, dataSplits: Se
CoreOptions.BLOB_EXTERNAL_STORAGE_FIELD.key() + "') can be updated.")
}
+ val firstRowIdToPartitionMap = new mutable.HashMap[Long, (Array[Byte],
Long)]
+ dataSplits.foreach(
+ split =>
+ split
+ .dataFiles()
+ .asScala
+ .filter(file => !isBlobFile(file.fileName()) &&
!isVectorStoreFile(file.fileName()))
+ .foreach(
+ file =>
+ firstRowIdToPartitionMap
+ .put(
+ file.firstRowId(),
+ // BinaryRow stores data in transient memory segments and
relies on Java
+ // serialization hooks to restore them. Store bytes in Spark
closures and
+ // broadcasts so Kryo does not serialize BinaryRow internals
directly.
+ (SerializationUtils.serializeBinaryRow(split.partition()),
file.rowCount())
+ )))
+ val firstRowIdToPartitionMapBroadcast =
+ sparkSession.sparkContext.broadcast(firstRowIdToPartitionMap)
+ val writeBuilder = table.newBatchWriteBuilder()
+
val written =
data.mapPartitions {
iter =>
{
val write = DataEvolutionTableDataWrite(
- table.newBatchWriteBuilder(),
+ writeBuilder,
writeType,
- firstRowIdToPartitionMap,
+ firstRowIdToPartitionMapBroadcast.value,
catalogContextForBlobDescriptor)
try {
iter.foreach(row => write.write(row))
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
index c0e5190d5d..ba2e84ef8a 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/DataEvolutionTableDataWrite.scala
@@ -28,6 +28,7 @@ import org.apache.paimon.spark.util.SparkRowUtils
import org.apache.paimon.table.sink.{BatchWriteBuilder, CommitMessage,
CommitMessageImpl, TableWriteImpl}
import org.apache.paimon.types.RowType
import org.apache.paimon.utils.RecordWriter
+import org.apache.paimon.utils.SerializationUtils
import org.apache.spark.sql.Row
@@ -39,7 +40,7 @@ import scala.collection.mutable.ListBuffer
case class DataEvolutionTableDataWrite(
writeBuilder: BatchWriteBuilder,
writeType: RowType,
- firstRowIdToPartitionMap: mutable.HashMap[Long, (BinaryRow, Long)],
+ firstRowIdToPartitionMap: mutable.HashMap[Long, (Array[Byte], Long)],
catalogContext: CatalogContext)
extends InnerTableV1DataWrite {
@@ -73,7 +74,8 @@ case class DataEvolutionTableDataWrite(
s"Available first row IDs:
${firstRowIdToPartitionMap.keys.mkString(", ")}")
}
- val (partition, numRecords) = pair
+ val (partitionBytes, numRecords) = pair
+ val partition = SerializationUtils.deserializeBinaryRow(partitionBytes)
val writer = writeBuilder
.newWrite()
