feat: add Comet CachedBatchSerializer for native in-memory cache

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -226,6 +226,19 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
+  val COMET_CACHE_SERIALIZER_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.cache.serializer.enabled")
+      .category(CATEGORY_EXEC)
+      .doc(
+        "When enabled, Comet installs a CachedBatchSerializer that stores Spark's in-memory " +
+          "table cache as compressed Arrow IPC. Repeated scans of cached data are then read " +
+          "natively without a per-read conversion. Schemas Comet cannot handle transparently " +
+          "fall back to Spark's default cache serializer. This sets " +
+          "spark.sql.cache.serializer for the session unless that property is already set " +
+          "to a non-default value. Disabled by default.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_EXEC_ENABLED: ConfigEntry[Boolean] = conf(s"$COMET_EXEC_CONFIG_PREFIX.enabled")
     .category(CATEGORY_EXEC)
     .doc(

spark/src/main/scala/org/apache/spark/Plugins.scala

@@ -28,7 +28,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.{EXECUTOR_MEMORY, EXECUTOR_MEMORY_OVERHEAD, EXECUTOR_MEMORY_OVERHEAD_FACTOR}
 import org.apache.spark.sql.internal.StaticSQLConf
 
-import org.apache.comet.CometConf.{COMET_METRICS_ENABLED, COMET_ONHEAP_ENABLED}
+import org.apache.comet.CometConf.{COMET_CACHE_SERIALIZER_ENABLED, COMET_METRICS_ENABLED, COMET_ONHEAP_ENABLED}
 import org.apache.comet.CometSparkSessionExtensions
 
 /**
@@ -57,6 +57,9 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl
     // register CometSparkSessionExtensions if it isn't already registered
     CometDriverPlugin.registerCometSessionExtension(sc.conf)
 
+    // Install the Comet cache serializer if requested
+    CometDriverPlugin.setCacheSerializerIfEnabled(sc.conf)
+
     // Register Comet metrics
     CometDriverPlugin.registerCometMetrics(sc)
 
@@ -104,6 +107,26 @@ class CometDriverPlugin extends DriverPlugin with Logging with ShimCometDriverPl
 }
 
 object CometDriverPlugin extends Logging {
+  private[spark] val COMET_CACHE_SERIALIZER =
+    "org.apache.spark.sql.comet.CometCachedBatchSerializer"
+
+  /**
+   * If the Comet cache serializer is enabled, install it as Spark's cache serializer. This is a
+   * static SQL config, so it must be set on the SparkConf before the session is created. A
+   * user-provided non-default serializer is respected and not overridden.
+   */
+  private[spark] def setCacheSerializerIfEnabled(conf: SparkConf): Unit = {
+    if (conf.getBoolean(COMET_CACHE_SERIALIZER_ENABLED.key, defaultValue = false)) {
+      val key = StaticSQLConf.SPARK_CACHE_SERIALIZER.key
+      val default = StaticSQLConf.SPARK_CACHE_SERIALIZER.defaultValueString
+      val existing = conf.get(key, default)
+      if (existing == default) {
+        logInfo(s"Setting $key=$COMET_CACHE_SERIALIZER")
+        conf.set(key, COMET_CACHE_SERIALIZER)
+      }
+    }
+  }
+
   def registerCometMetrics(sc: SparkContext): Unit = {
     if (sc.getConf.getBoolean(
         COMET_METRICS_ENABLED.key,