andygrove commented on code in PR #6853:
URL: https://github.com/apache/datafusion-comet/pull/6853#discussion_r4264988218
##########
spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala:
##########
@@ -1141,6 +1141,143 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
}
+ // Each column's null count summed over the cached batches, asserting along
the way that no
+ // batch records bounds for the column.
+ private def nullCountsWithoutBounds(view: String): Seq[Long] = {
+ val relation =
spark.sharedState.cacheManager.lookupCachedData(spark.table(view)).get
+ val stats = relation.cachedRepresentation.cacheBuilder.cachedColumnBuffers
+ .collect()
+ .map(_.asInstanceOf[SimpleMetricsCachedBatch].stats)
+ relation.cachedRepresentation.output.indices.map { i =>
+ stats.foreach(s => assert(s.isNullAt(i * 5) && s.isNullAt(i * 5 + 1),
s"column $i"))
+ stats.map(_.getInt(i * 5 + 2).toLong).sum
+ }
+ }
+
+ test("Comet in-memory cache statistics count nulls in columns that record no
bounds") {
+ // These columns record a null count and no bounds: binary and nested
types, and a bigint
+ // column with no values. Comet's Arrow vectors give the count without a
pass over the rows,
+ // while Spark's vectors are counted row by row, so it is checked from the
row write path,
+ // Comet's native scan and Spark's vectorized reader.
+ val df = spark
+ .range(1000)
+ .selectExpr("if(id % 7 = 0, NULL, id) AS v")
+ .selectExpr(
+ "cast(concat('b', v) AS binary) AS bin",
+ "if(v IS NULL, NULL, array(v, v + 1)) AS arr",
+ "if(v IS NULL, NULL, map(concat('k', v), v)) AS mp",
+ "if(v IS NULL, NULL, named_struct('a', v, 'b', cast(v AS string))) AS
st",
+ "cast(NULL AS bigint) AS no_values")
+ // 143 of the ids 0 to 999 are multiples of 7.
+ val expected = Seq(143L, 143L, 143L, 143L, 1000L)
+
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // Only the row write path can take a void column; Parquet cannot store
one.
+ val rows = df.withColumn("nt", lit(null))
+ rows.createOrReplaceTempView("no_bounds_rows")
+ rows.cache()
+ try {
+ rows.count()
+ assert(nullCountsWithoutBounds("no_bounds_rows") == expected :+ 1000L)
Review Comment:
You're right that the counts alone don't show which collector produced them.
In 1b3990d111 the check is in `nullCountsWithoutBounds`, so every leg asserts
the relation is a `CometCachedBatch` before it reads the statistics, and a
failure names the format it found.
While checking this I found that the `no_values` column already catches a
fallback by accident. Spark's `LongColumnStats` records `Long.MaxValue` and
`Long.MinValue` as the bounds of an all-null bigint column, so with `NullType`
delegated to Spark's format the row leg failed on column 4 rather than passing.
That depends on the test keeping a primitive column, so the explicit check is
still better. With the same mutant the test now fails with `no_bounds_rows was
cached as org.apache.spark.sql.execution.columnar.DefaultCachedBatch`.
--
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]
