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()