From 8ff826b16d5d0d394331fdad2623b61ac5465a97 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Tue, 28 Jul 2026 10:35:53 -0700 Subject: [PATCH] [SPARK-58408][CORE] Register TimestampNanosVal in KryoSerializer Since SPARK-57735, cached-batch statistics rows carry TimestampNanosVal min/max bounds for nanosecond-timestamp columns (TimestampNanosColumnStats in the default cache serializer, and the Arrow cache serializer's vector-side statistics). Persisting such a cache with a serialized storage level under spark.serializer=KryoSerializer and spark.kryo.registrationRequired=true failed at cache materialization with "Class is not registered: org.apache.spark.unsafe.types.TimestampNanosVal", because the class was never added to the registration list. Register it next to UTF8String, which reached the list the same way (SPARK-51790). Co-authored-by: Claude Code --- .../org/apache/spark/serializer/KryoSerializer.scala | 5 ++++- .../org/apache/spark/sql/CacheTableInKryoSuite.scala | 8 ++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala index ea7af01121a1b..65f0ffcbe18ea 100644 --- a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala +++ b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala @@ -47,7 +47,7 @@ import org.apache.spark.internal.io.FileCommitProtocol._ import org.apache.spark.network.util.ByteUnit import org.apache.spark.scheduler.{CompressedMapStatus, HighlyCompressedMapStatus} import org.apache.spark.storage._ -import org.apache.spark.unsafe.types.UTF8String +import org.apache.spark.unsafe.types.{TimestampNanosVal, UTF8String} import org.apache.spark.util.{BoundedPriorityQueue, ByteBufferInputStream, NextIterator, SerializableConfiguration, SerializableJobConf, Utils} import org.apache.spark.util.collection.{BitSet, CompactBuffer} import org.apache.spark.util.io.ChunkedByteBuffer @@ -236,6 +236,9 @@ class KryoSerializer(conf: SparkConf) kryo.register(classOf[ArrayBuffer[Any]]) kryo.register(classOf[Array[Array[Byte]]]) kryo.register(classOf[UTF8String]) + // Nanosecond-timestamp min/max bounds inside cached-batch statistics rows + // (TimestampNanosColumnStats and the Arrow cache's vector-side stats). + kryo.register(classOf[TimestampNanosVal]) // We can't load those class directly in order to avoid unnecessary jar dependencies. // We load them safely, ignore it if the class not found. diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CacheTableInKryoSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CacheTableInKryoSuite.scala index 72a2da160547a..feee2bee298f3 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CacheTableInKryoSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/CacheTableInKryoSuite.scala @@ -57,6 +57,14 @@ class CacheTableInKryoSuite extends SharedSparkSession { } } + test("SPARK-58408: TimestampNanosVal stats should be registered in KryoSerializer") { + // Cached-batch statistics rows carry TimestampNanosVal min/max bounds for + // nanosecond-timestamp columns (TimestampNanosColumnStats), so persisting such a cache + // through Kryo with registrationRequired must find the class registered. + val df = sql("SELECT CAST('2025-01-06 12:30:45.123456789' AS TIMESTAMP_NTZ(9)) AS ts") + assert(df.persist(StorageLevel.DISK_ONLY).count() === 1) + } + test("SPARK-51813 DefaultCachedBatchKryoSerializer do not propagate nulls") { val ks = new KryoSerializer(this.sparkConf) val kryo = ks.newKryo()