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 bd5e9f9a03 [spark] Fix timestamp precision in key dynamic writes
(#9322)
bd5e9f9a03 is described below
commit bd5e9f9a03dfa0514af2db0bccbb707ede85f76a
Author: Zouxxyy <[email protected]>
AuthorDate: Fri Aug 21 08:28:53 2026 +0800
[spark] Fix timestamp precision in key dynamic writes (#9322)
---
.../paimon/spark/commands/BucketProcessor.scala | 26 +++++++++-------------
.../paimon/spark/commands/PaimonSparkWriter.scala | 11 +++------
.../paimon/spark/sql/DynamicBucketTableTest.scala | 23 +++++++++++++++++++
3 files changed, 36 insertions(+), 24 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/BucketProcessor.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/BucketProcessor.scala
index 4a147deee9..82ab11cf43 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/BucketProcessor.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/BucketProcessor.scala
@@ -22,7 +22,7 @@ import org.apache.paimon.CoreOptions
import org.apache.paimon.bucket.BucketFunction
import org.apache.paimon.codegen.CodeGenUtils
import org.apache.paimon.crosspartition.{GlobalIndexAssigner, KeyPartOrRow}
-import org.apache.paimon.data.{BinaryRow, GenericRow, InternalRow =>
PaimonInternalRow, JoinedRow}
+import org.apache.paimon.data.{BinaryRow, InternalRow => PaimonInternalRow}
import org.apache.paimon.disk.IOManager
import org.apache.paimon.index.HashBucketAssigner
import org.apache.paimon.schema.TableSchema
@@ -31,7 +31,6 @@ import org.apache.paimon.spark.SparkUtils.createIOManager
import org.apache.paimon.spark.util.EncoderUtils
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.RowPartitionKeyExtractor
-import org.apache.paimon.types.RowType
import org.apache.paimon.utils.{CloseableIterator, SerializationUtils}
import org.apache.spark.TaskContext
@@ -39,6 +38,7 @@ import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.{Deserializer,
Serializer}
+import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow,
JoinedRow => SparkJoinedRow}
import org.apache.spark.sql.types.StructType
import java.util.UUID
@@ -173,26 +173,19 @@ case class DynamicBucketProcessor(
case class GlobalDynamicBucketProcessor(
fileStoreTable: FileStoreTable,
- rowType: RowType,
numAssigners: Integer,
encoderGroup: EncoderSerDeGroup)
extends BucketProcessor[(KeyPartOrRow, Array[Byte])] {
override def processPartition(
rowIterator: Iterator[(KeyPartOrRow, Array[Byte])]): Iterator[Row] = {
- new GlobalIndexAssignerIterator(
- rowIterator,
- fileStoreTable,
- rowType,
- numAssigners,
- encoderGroup)
+ new GlobalIndexAssignerIterator(rowIterator, fileStoreTable, numAssigners,
encoderGroup)
}
}
class GlobalIndexAssignerIterator(
rowIterator: Iterator[(KeyPartOrRow, Array[Byte])],
fileStoreTable: FileStoreTable,
- rowType: RowType,
numAssigners: Integer,
encoderGroup: EncoderSerDeGroup)
extends Iterator[Row]
@@ -200,6 +193,8 @@ class GlobalIndexAssignerIterator(
private val queue = mutable.Queue[Row]()
+ private val dataRowType = fileStoreTable.rowType()
+
val ioManager: IOManager = createIOManager
var currentResult: Row = _
@@ -215,12 +210,11 @@ class GlobalIndexAssignerIterator(
numAssigners,
TaskContext.getPartitionId(),
(row, bucket) => {
- val extraRow: GenericRow = new GenericRow(2)
- extraRow.setField(0, row.getRowKind.toByteValue)
- extraRow.setField(1, bucket)
- queue.enqueue(
- encoderGroup.internalToRow(
- DataConverter.fromPaimon(new JoinedRow(row, extraRow), rowType)))
+ val dataRow = DataConverter.fromPaimon(row, dataRowType)
+ val extraRow = new GenericInternalRow(2)
+ extraRow.setByte(0, row.getRowKind.toByteValue)
+ extraRow.setInt(1, bucket)
+ queue.enqueue(encoderGroup.internalToRow(new SparkJoinedRow(dataRow,
extraRow)))
}
)
rowIterator.foreach {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 35d7098529..2bbd53b8ab 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -30,7 +30,7 @@ import org.apache.paimon.fs.Path
import org.apache.paimon.index.{BucketAssigner, SimpleHashBucketAssigner}
import org.apache.paimon.io.{CompactIncrement, DataIncrement}
import org.apache.paimon.manifest.FileKind
-import org.apache.paimon.spark.{SparkPostponeStagedCommitter, SparkRow,
SparkTypeUtils}
+import org.apache.paimon.spark.{SparkPostponeStagedCommitter, SparkRow}
import org.apache.paimon.spark.catalog.functions.BucketFunction
import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL,
ROW_KIND_COL}
import org.apache.paimon.spark.sort.TableSorter
@@ -40,7 +40,7 @@ import org.apache.paimon.spark.write.{PaimonDataWrite,
WriteHelper, WriteTaskRes
import org.apache.paimon.table.{FileStoreTable, SpecialFields}
import org.apache.paimon.table.BucketMode._
import org.apache.paimon.table.sink._
-import org.apache.paimon.types.{RowKind, RowType}
+import org.apache.paimon.types.RowKind
import org.apache.paimon.utils.{SerializationUtils, UriReaderFactory}
import org.apache.spark.{Partitioner, TaskContext}
@@ -247,7 +247,6 @@ case class PaimonSparkWriter(
val written: Dataset[_ <: WriteTaskResult] = bucketMode match {
case KEY_DYNAMIC =>
// Topology: input -> bootstrap -> shuffle by key hash ->
bucket-assigner -> shuffle by partition & bucket
- val rowType =
SparkTypeUtils.toPaimonType(withInitBucketCol.schema).asInstanceOf[RowType]
val assignerParallelism =
Option(coreOptions.dynamicBucketAssignerParallelism)
.map(_.toInt)
.getOrElse(sparkParallelism)
@@ -259,11 +258,7 @@ case class PaimonSparkWriter(
uriReaderFactory)
val globalDynamicBucketProcessor =
- GlobalDynamicBucketProcessor(
- table,
- rowType,
- assignerParallelism,
- encoderGroupWithBucketCol)
+ GlobalDynamicBucketProcessor(table, assignerParallelism,
encoderGroupWithBucketCol)
val repartitioned = repartitionByPartitionsAndBucket(
sparkSession.createDataFrame(
bootstrapped.mapPartitions(globalDynamicBucketProcessor.processPartition),
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DynamicBucketTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DynamicBucketTableTest.scala
index fee7973aca..47c91ef460 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DynamicBucketTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DynamicBucketTableTest.scala
@@ -18,7 +18,10 @@
package org.apache.paimon.spark.sql
+import org.apache.paimon.catalog.Identifier
+import org.apache.paimon.schema.Schema
import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.types.DataTypes
import org.apache.spark.sql.Row
@@ -221,4 +224,24 @@ class DynamicBucketTableTest extends PaimonSparkTestBase {
)
}
}
+
+ test("Paimon cross partition table: write timestamp with precision 3") {
+ withTable("t") {
+ val schema = Schema.newBuilder
+ .column("etl_dt", DataTypes.DATE.notNull)
+ .column("order_id", DataTypes.BIGINT.notNull)
+ .column("despatch_end", DataTypes.TIMESTAMP(3))
+ .partitionKeys("etl_dt")
+ .primaryKey("order_id")
+ .option("bucket", "-1")
+ .option("deletion-vectors.enabled", "true")
+ .option("dynamic-bucket.assigner-parallelism", "1")
+ .build
+ paimonCatalog.createTable(Identifier.create(dbName0, "t"), schema, false)
+
+ sql("INSERT INTO t VALUES (DATE '2026-08-14', 1L, TIMESTAMP '2026-08-14
10:00:00')")
+
+ checkAnswer(sql("SELECT order_id FROM t"), Row(1L))
+ }
+ }
}
