andygrove opened a new issue, #6203: URL: https://github.com/apache/datafusion-comet/issues/6203
### Describe the bug `checkSparkAnswer` builds its expected answer by re-running the query's logical plan with `spark.comet.enabled=false` ([`CometTestBase.internalCheckSparkAnswer`](https://github.com/apache/datafusion-comet/blob/67803a7a422c44de07af1e5d25c1dbeae8df68d4/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala#L130-L149)). 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 67803a7a4, 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. ```diff --- 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: ```shell ./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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
