In-memory cache tests that use checkSparkAnswer compare the cache with itself

[andygrove]

Describe the bug

checkSparkAnswer builds its expected answer by re-running the query's logical plan with spark.comet.enabled=false (CometTestBase.internalCheckSparkAnswer). When the query reads a cached view, the cache manager still substitutes the InMemoryRelation, and spark.sql.cache.serializer is static, so that run reads the same CometCachedBatch bytes through the same ArrowCachedBatchSerializer and prunes with the same buildFilter. A wrong value in the stored data or a wrong bound in the statistics shows up on both sides, and the assertion passes.

23 of the 53 test definitions in CometInMemoryCacheSuite and CometInMemoryCacheKryoSuite check a cached query with checkSparkAnswer. For several of them it is the only check on the property the test is named for: Comet in-memory cache pruning handles NaN floating-point values, cache a Spark columnar plan whose vectors are not Arrow-backed, cache a non-Arrow-backed Spark columnar plan with complex types, Comet in-memory cache supports DISK_ONLY storage level, and the Kryo round trips. The NaN test has a second problem: isnan has no case in SimpleMetricsCachedBatchSerializer.buildFilter, so of its three predicates only d = 0.0D OR f = CAST(0.0 AS FLOAT) can prune anything.

Steps to reproduce

On main at 67803a7, default Spark 4.1 profile, apply two deliberate bugs to ArrowCachedBatchSerializer: every cached double is stored as value + 1.0, and NaN is left out of the double bounds.

--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala
@@ -281,8 +281,8 @@
         nullCount += 1
       } else {
         val value = col.getDouble(r)
-        if (r == nullCount || JDouble.compare(value, min) < 0) min = value
-        if (r == nullCount || JDouble.compare(value, max) > 0) max = value
+        if (!value.isNaN && (r == nullCount || JDouble.compare(value, min) < 0)) min = value
+        if (!value.isNaN && (r == nullCount || JDouble.compare(value, max) > 0)) max = value
       }
       r += 1
     }
@@ -412,11 +412,24 @@
       val writeDirectly =
         Utils.isArrowBacked(batch) && CachedBatchIpc.matchesReaderLayout(batch, readerFields)

+      def mutate(b: ColumnarBatch): Unit = (0 until b.numCols()).foreach { c =>
+        b.column(c) match {
+          case v: org.apache.comet.vector.CometVector =>
+            v.getValueVector match {
+              case f: org.apache.arrow.vector.Float8Vector =>
+                (0 until f.getValueCount).foreach(i => if (!f.isNull(i)) f.set(i, f.get(i) + 1.0))
+              case _ =>
+            }
+          case _ =>
+        }
+      }
       val (bytes, columnSizes) = if (writeDirectly) {
+        mutate(batch)
         CachedBatchIpc.serialize(batch, codec, CometArrowAllocator, settings.chunkSize)
       } else {
         val arrowBatch =
           CometArrowConverters.columnarBatchToArrowBatch(batch, arrowSchema, CometArrowAllocator)
+        mutate(arrowBatch)
         try CachedBatchIpc.serialize(arrowBatch, codec, CometArrowAllocator, settings.chunkSize)
         finally arrowBatch.close()
       }

Then run the two suites:

./mvnw test -Dtest=none \
  -Dsuites="org.apache.comet.exec.CometInMemoryCacheSuite,org.apache.comet.exec.CometInMemoryCacheKryoSuite"

54 of the 57 tests pass. The three that fail are the ones that compute their expected values without the cache: statistics preserve numeric extremes and floating-point ordering, round-trips all supported types and round-trips under every compression codec. cache a Spark columnar plan whose vectors are not Arrow-backed passes even though it selects and sums a double column.

Expected behavior

A cache test fails when the cache stores a wrong value or prunes a batch it should keep. That needs an expected answer the cache cannot reach: collect it before cacheTable, or evaluate the defining query with Comet off and the cache cleared. The collation tests and round-trips all supported types already work this way.

Additional context

A differential check catches both injected bugs. For each write path (Comet's native scan written directly, Spark's vectorized reader converted, and row input), write a fixture as many small Parquet files so the cache holds many batches with different bounds, cache it, and compare about 60 prunable predicates with ground truth computed from the DataFrame that generated the fixture. The predicates cover NaN, ±0, ±Inf, nulls, multibyte strings, decimals, timestamps, booleans, IN, OR and startswith. On unmodified main every one of them matches on all three paths. I have this as a scratch suite and can turn it into a real one.

Don't use an uncached Parquet read as the ground truth for signed-zero predicates. With Comet off, Spark's own reader skips a row group that holds only -0.0 for f = 0.0.

This bears on #5634, which uses a green CI run as evidence that the cache format is ready to be on by default.

Found while auditing the in-memory cache.

[LinSimon-901101]

take