@@ -47,9 +47,11 @@ With Comet's serializer installed as `spark.sql.cache.serializer`:
   expects, so Spark can prune whole cached batches on a predicate before any of them is decoded.
 
 Relations whose schema Comet's Arrow writer cannot store — interval types, most notably — are
-delegated in full to Spark's default cache format, per relation. Nothing about the format depends
-on a runtime config, because `spark.sql.cache.serializer` is a static setting and a relation whose
-format could change mid-session could not be read back reliably. Turning
+delegated in full to Spark's default cache format, per relation. Which format a relation uses does
+not depend on a runtime config, because `spark.sql.cache.serializer` is a static setting and a
+relation whose format could change mid-session could not be read back reliably. The compression
+codec is a runtime config, but each batch records the codec it was written with, so data cached
+under one setting stays readable after the setting changes. Turning
 `spark.comet.exec.inMemoryCache.enabled` off at runtime only sends cached scans back to Spark's
 execution path; the cached data stays readable either way.
 
@@ -78,8 +80,8 @@ longer. A narrow projection pays far less, because it only inflates the columns
 
 | Codec  | Materialize | Footprint | Read 1 of 6 | Read 6 of 6 |
 | ------ | ----------: | --------: | ----------: | ----------: |
-| `zstd` |     1503 ms |    55 MiB |       47 ms |      296 ms |
-| `none` |     1091 ms |   315 MiB |       36 ms |       63 ms |
+| `zstd` |     1507 ms |    55 MiB |       45 ms |      295 ms |
+| `none` |     1081 ms |   315 MiB |       35 ms |       64 ms |
 
 `none` is the better setting for a relation that fits in memory uncompressed and is read at close
 to full width. The default is the other way round because a cache that does not fit costs more than
@@ -115,20 +117,20 @@ On a 5M-row relation of six flat columns:
 
 | Query shape                    | Spark cache scan + convert | `CometInMemoryTableScan` | Relative |
 | ------------------------------ | -------------------------: | -----------------------: | -------: |
-| Repeated scan (3 of 6 columns) |                     180 ms |                   147 ms |     1.2x |
-| Selective filter               |                      58 ms |                    50 ms |     1.2x |
-| Row count only (0 of 6)        |                      37 ms |                    34 ms |     1.1x |
-| Narrow projection (1 of 6)     |                      52 ms |                    48 ms |     1.1x |
-| Full projection (6 of 6)       |                     547 ms |                   291 ms |     1.9x |
+| Repeated scan (3 of 6 columns) |                     209 ms |                   172 ms |     1.2x |
+| Selective filter               |                      72 ms |                    61 ms |     1.2x |
+| Row count only (0 of 6)        |                      46 ms |                    38 ms |     1.2x |
+| Narrow projection (1 of 6)     |                      70 ms |                    58 ms |     1.2x |
+| Full projection (6 of 6)       |                     566 ms |                   324 ms |     1.7x |
 
 And on a 1M-row relation of six columns whose middle three are structs, one of them nested two
 levels deep:
 
 | Query shape                | Spark cache scan + convert | `CometInMemoryTableScan` | Relative |
 | -------------------------- | -------------------------: | -----------------------: | -------: |
-| Row count only (0 of 6)    |                      30 ms |                    29 ms |     1.0x |
-| Narrow projection (1 of 6) |                      98 ms |                    53 ms |     1.8x |
-| Full projection (6 of 6)   |                     250 ms |                   125 ms |     2.0x |
+| Row count only (0 of 6)    |                      38 ms |                    32 ms |     1.2x |
+| Narrow projection (1 of 6) |                     109 ms |                    58 ms |     1.9x |
+| Full projection (6 of 6)   |                     275 ms |                   130 ms |     2.1x |
 
 Both columns read the cache at the default codec, `zstd`. The codec table above shows what `none`
 changes, and it is the full projection that moves most: nothing has to be inflated, so it runs
@@ -147,10 +149,9 @@ Read what this compares carefully. Comet execution is on in both columns, so the
 on Comet either way and only the cache-scan boundary moves: on the left, Spark's
 `InMemoryTableScanExec` feeds those same Comet operators through a `CometSparkColumnarToColumnar`
 bridge; on the right, `CometInMemoryTableScan` feeds them directly. Both columns read the same
-Comet-written `CometCachedBatch` — `spark.sql.cache.serializer` is static, so one session cannot
-also materialize Spark's format to compare against. These numbers are therefore "keep the cached
-scan native" against "fall back to a Spark cache scan and convert", not Comet against Spark
-execution, and not a comparison with Spark's own cache format.
+Comet-written `CometCachedBatch`. These numbers are therefore "keep the cached scan native" against
+"fall back to a Spark cache scan and convert", not Comet against Spark execution, and not a
+comparison with Spark's own cache format. That comparison is under [Limitations](#limitations).
 
 ## Kryo
 
@@ -174,10 +175,19 @@ registrator.
 
 ## Limitations
 
-Reads that feed **Spark** operators rather than Comet ones are still slower than Spark's own cache
-format, by roughly 1.7x to 2.5x depending on how wide the projection is. Those reads pay a row
-conversion that Spark's format avoids with generated code over its own layout. This is why the
-feature is off by default.
+Reads that feed **Spark** operators rather than Comet ones are slower than Spark's own cache
+format, and the narrower the read, the wider the gap. Measured by the same benchmark over the same
+5M-row relation, with Comet off so that Spark operators consume the cached data:
+
+| Read shape              | Spark's cache format | Comet's cache format | Slowdown |
+| ----------------------- | -------------------: | -------------------: | -------: |
+| Row count only (0 of 6) |                35 ms |               183 ms |     5.2x |
+| 1 of 6 columns          |                54 ms |               257 ms |     4.8x |
+| 3 of 6 columns          |                98 ms |               331 ms |     3.4x |
+| 6 of 6 columns          |               410 ms |               623 ms |     1.5x |
+
+This is why the feature is off by default. The cause is not yet established;
+[#5485](https://github.com/apache/datafusion-comet/issues/5485) tracks it.
 
 Comet's serializer exists because Spark's own Arrow cache format
 ([SPARK-57268](https://issues.apache.org/jira/browse/SPARK-57268)) is only available from Spark
 
@@ -26,7 +26,7 @@ import scala.collection.JavaConverters._
 import org.apache.spark.TaskContext
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, GenericInternalRow, IsNotNull, IsNull, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, GenericInternalRow, IsNotNull, IsNull, StartsWith, UnsafeProjection}
 import org.apache.spark.sql.catalyst.util.TypeUtils
 import org.apache.spark.sql.columnar.{CachedBatch, SimpleMetricsCachedBatch, SimpleMetricsCachedBatchSerializer}
 import org.apache.spark.sql.comet.util.Utils
@@ -456,6 +456,8 @@ class ArrowCachedBatchSerializer extends SimpleMetricsCachedBatchSerializer {
   // makes a comparison against them evaluate to null and therefore prune the batch. That would
   // silently drop rows, so predicates over columns without bounds are not pushed down at all.
   // Null counts and row counts are recorded for every column, so IsNull and IsNotNull stay safe.
+  // A column that does have bounds can still be unsafe to prune on through StartsWith; see
+  // prunesOnCollatedPrefix.
   override def buildFilter(
       predicates: Seq[Expression],
       cachedAttributes: Seq[Attribute]): (Int, Iterator[CachedBatch]) => Iterator[CachedBatch] = {
@@ -465,12 +467,25 @@ class ArrowCachedBatchSerializer extends SimpleMetricsCachedBatchSerializer {
 
     val prunablePredicates = predicates.filter {
       case _: IsNull | _: IsNotNull => true
-      case p => p.references.forall(a => prunable.contains(a.exprId))
+      case p =>
+        p.references.forall(a => prunable.contains(a.exprId)) && !prunesOnCollatedPrefix(p)
     }
 
     super.buildFilter(prunablePredicates, cachedAttributes)
   }
 
+  // Spark prunes StartsWith by cutting each bound to the prefix's length in characters, which
+  // assumes a matching value begins with as many characters as the prefix has. That holds under
+  // binary comparison but not under every collation: under UTF8_LCASE, U+0130 is one character
+  // that lowercases to two, so a batch of values beginning with it is pruned against a
+  // two-character prefix they all match. Spark's own cache prunes the same way. A predicate that
+  // holds such a StartsWith anywhere is left out whole, since leaving out a conjunct only prunes
+  // less. Equality and range predicates compare with the collation itself, so they still prune.
+  private def prunesOnCollatedPrefix(predicate: Expression): Boolean = predicate.exists {
+    case StartsWith(left, _) => left.dataType != StringType
+    case _ => false
+  }
+
   // Comet's Arrow writer only handles the types listed in supportsSchema. Reporting false here
   // sends the relation down the row path, where it is delegated to Spark's default serializer,
   // instead of failing at cache materialization inside Utils.serializeBatches.
 
@@ -317,48 +317,71 @@ private[comet] object CachedBatchIpc {
         nodes.add(new ArrowFieldNode(node.length(), node.nullCount()))
       }
 
+      // Under a codec, a buffer that compressing would not have shrunk is stored as it was, behind
+      // a -1 length prefix, and decompressing hands it back as a slice of whatever holds it rather
+      // than as an allocation of its own. Those windows are therefore copied into an allocation of
+      // their own: sharing one with the compressed windows, the smallest of them would keep every
+      // compressed byte of the batch alive for as long as the decoded vectors.
+      val compressedBody = compression.getCodec != NoCompressionCodec.COMPRESSION_TYPE
       val offsets = new Array[Long](bufferIndices.length)
       val lengths = new Array[Long](bufferIndices.length)
-      var total = 0L
+      val storedRaw = new Array[Boolean](bufferIndices.length)
+      var packedTotal = 0L
+      var rawTotal = 0L
       var k = 0
       while (k < bufferIndices.length) {
         val buffer = batch.buffers(bufferIndices(k))
         offsets(k) = buffer.offset()
         lengths(k) = buffer.length()
-        total += DataSizeRoundingUtil.roundUpTo8Multiple(lengths(k))
+        storedRaw(k) = compressedBody &&
+          lengths(k) >= CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH &&
+          chunks.readLong(bodyStart + offsets(k)) == CompressionUtil.NO_COMPRESSION_LENGTH
+        val size = DataSizeRoundingUtil.roundUpTo8Multiple(lengths(k))
+        if (storedRaw(k)) rawTotal += size else packedTotal += size
         k += 1
       }
 
-      // allocator.buffer(0) is legal but yields a buffer no window can be sliced from, and an
-      // all-empty projection (every selected column a NullVector, say) would ask for exactly that.
-      val body = allocator.buffer(math.max(total, 1L))
+      var packed: ArrowBuf = null
+      var raw: ArrowBuf = null
       val compressedBatch =
         try {
+          if (packedTotal > 0) packed = allocator.buffer(packedTotal)
+          if (rawTotal > 0) raw = allocator.buffer(rawTotal)
+
+          def window(into: ArrowBuf, at: Long, i: Int): ArrowBuf = {
+            chunks.copyTo(into, at, bodyStart + offsets(i), lengths(i))
+            val slice = into.slice(at, lengths(i))
+            slice.writerIndex(lengths(i))
+            slice
+          }
+
           val buffers = new java.util.ArrayList[ArrowBuf](bufferIndices.length)
-          var position = 0L
+          var packedPosition = 0L
+          var rawPosition = 0L
           var i = 0
           while (i < bufferIndices.length) {
-            val length = lengths(i)
-            if (length > 0) {
-              chunks.copyTo(body, position, bodyStart + offsets(i), length)
+            val size = DataSizeRoundingUtil.roundUpTo8Multiple(lengths(i))
+            if (lengths(i) == 0) {
+              // Empty, so there is nothing to copy and no allocation for it to keep alive.
+              buffers.add(allocator.getEmpty)
+            } else if (storedRaw(i)) {
+              buffers.add(window(raw, rawPosition, i))
+              rawPosition += size
+            } else {
+              buffers.add(window(packed, packedPosition, i))
+              packedPosition += size
             }
-            val window = body.slice(position, length)
-            window.writerIndex(length)
-            buffers.add(window)
-            position += DataSizeRoundingUtil.roundUpTo8Multiple(length)
             i += 1
           }
           new ArrowRecordBatch(batch.length().toInt, nodes, buffers, compression, false)
-        } catch {
-          case NonFatal(e) =>
-            body.close()
-            throw e
+        } finally {
+          // The constructor retained each window; slice() alone does not. Dropping these
+          // references leaves the batch as sole owner of both allocations, so closing the batch
+          // is what frees them -- and if anything above threw, this is what frees them instead.
+          if (packed != null) packed.close()
+          if (raw != null) raw.close()
         }
 
-      // The constructor retained each window; slice() alone does not. Dropping `body`'s own
-      // reference leaves the batch as sole owner of the one allocation, so closing the batch is
-      // what frees it -- and closing `body` again would drive its reference count negative.
-      body.close()
       val plainBatch =
         try decompressed(compressedBatch, allocator)
         finally compressedBatch.close()
@@ -394,18 +417,8 @@ private[comet] object CachedBatchIpc {
 
     /** Copy `length` bytes starting at `from` in the payload into `dst` at `dstIndex`. */
     def copyTo(dst: ArrowBuf, dstIndex: Long, from: Long, length: Long): Unit = {
-      // The flatbuffer accessors behind `from` and `length` are unchecked, so a window reaching
-      // past the payload is a corrupt or mismatched batch, and is reported as one rather than as
-      // an index error from inside the copy.
-      if (from < 0 || length < 0 || from + length > starts.last) {
-        throw new SparkException(
-          s"Comet cached batch records a buffer at [$from, ${from + length}) outside its " +
-            s"${starts.last}-byte payload")
-      }
-      // The last chunk whose start is at or before `from`. Empty chunks share a start with the
-      // chunk after them and are skipped by the loop below.
-      val found = java.util.Arrays.binarySearch(starts, 0, chunks.length, from)
-      var c = if (found >= 0) found else -found - 2
+      checkRange(from, length)
+      var c = chunkAt(from)
       var copied = 0L
       while (copied < length) {
         val chunk = chunks(c)
@@ -421,6 +434,40 @@ private[comet] object CachedBatchIpc {
         c += 1
       }
     }
+
+    /** The little-endian long at `from` in the payload, which may straddle two chunks. */
+    def readLong(from: Long): Long = {
+      checkRange(from, java.lang.Long.BYTES)
+      var value = 0L
+      var i = java.lang.Long.BYTES - 1
+      while (i >= 0) {
+        val at = from + i
+        var c = chunkAt(at)
+        // Skip the empty chunks that share a start with the one holding `at`.
+        while (starts(c + 1) <= at) c += 1
+        val chunk = chunks(c)
+        value = (value << 8) | (chunk.get(chunk.position() + (at - starts(c)).toInt) & 0xffL)
+        i -= 1
+      }
+      value
+    }
+
+    // The flatbuffer accessors behind a window's position and length are unchecked, so a window
+    // reaching past the payload is a corrupt or mismatched batch, and is reported as one rather
+    // than as an index error from inside a read.
+    private def checkRange(from: Long, length: Long): Unit =
+      if (from < 0 || length < 0 || from + length > starts.last) {
+        throw new SparkException(
+          s"Comet cached batch records a buffer at [$from, ${from + length}) outside its " +
+            s"${starts.last}-byte payload")
+      }
+
+    // The last chunk whose start is at or before `from`. Empty chunks share a start with the
+    // chunk after them; callers step past those.
+    private def chunkAt(from: Long): Int = {
+      val found = java.util.Arrays.binarySearch(starts, 0, chunks.length, from)
+      if (found >= 0) found else -found - 2
+    }
   }
 
   /**