viirya commented on code in PR #6853:
URL: https://github.com/apache/datafusion-comet/pull/6853#discussion_r4262942182
##########
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:
Nice change, and the comment on `arrowNullCount` explaining why Spark's
vectors are excluded is really helpful.
One question on the new no-bounds test. `nullCountsWithoutBounds` only reads
the stats row, and Spark's `DefaultCachedBatchSerializer` records the same
`(null, null, nullCount, ...)` layout for binary, nested and void columns. So
if these relations ever end up as `DefaultCachedBatch`, for example through the
serializer pinning that the comment at the top of this suite describes, or if
`NullType` is later delegated to the fallback, the row-path and native-scan
legs (here and at line 1203) would still pass without touching Comet's
collector. Could you assert `cachedBatchTypes(...)` is `CometCachedBatch` in
those two legs, the way `withSparkColumnarCache` does?
--
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]
