andygrove commented on code in PR #5734:
URL: https://github.com/apache/datafusion-comet/pull/5734#discussion_r3959422667
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometInMemoryCacheBenchmark.scala:
##########
@@ -73,6 +73,13 @@ object CometInMemoryCacheBenchmark extends
CometBenchmarkBase {
"concat('str_c_', cast(id as string)) AS s3")
.createOrReplaceTempView(sourceTable)
+ val buildBenchmark =
+ new Benchmark("in-memory cache materialization", numRows, output =
output)
+ buildBenchmark.addCase("Comet cache writer") { _ =>
+ withCachedTable {}
+ }
+ buildBenchmark.run()
Review Comment:
Two things about this case. `withCachedTable`'s `finally` block runs inside
the timed lambda, so `uncacheTable` and `clearCache` are part of the
measurement along with row generation, Arrow conversion and LZ4. And the thing
this PR speeds up is a small share of that, which matches your own 1.926 s
against 1.935 s.
I measured `gatherColumnStats` on its own over an `OnHeapColumnVector` batch
of 10000 rows and got 505-522 ms down to 134-138 ms with 1505 MB down to 4.8 MB
allocated, so the win is real and bigger than you claimed, but nothing
committed here would catch it regressing. Would you consider a case that times
statistics collection alone, or at least moving the teardown out of the timed
region?
A one-line comment saying this case has no comparison arm on purpose would
help too, since every other case in the file has two and the explanation for
why a `DefaultCachedBatch` baseline is impossible lives down in
`withCachedTable`.
##########
spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala:
##########
@@ -351,6 +352,82 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
}
+ test("Comet in-memory cache statistics preserve typed bounds and null
counts") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val types = Seq(
+ "boolean",
+ "tinyint",
+ "smallint",
+ "int",
+ "bigint",
+ "float",
+ "double",
+ "decimal(10,2)",
+ "decimal(38,2)",
+ "string",
+ "date",
+ "timestamp",
+ "timestamp_ntz",
+ "binary")
+ val expressions = types.zipWithIndex.map { case (dt, i) =>
+ val value = dt match {
+ case "date" => "date_add(DATE '2000-01-01', cast(v AS INT))"
+ case "timestamp" | "timestamp_ntz" =>
+ s"cast(date_add(DATE '2000-01-01', cast(v AS INT)) AS $dt)"
+ case "string" | "binary" => s"cast(concat('字', cast(v AS STRING)) AS
$dt)"
+ case _ => s"cast(v AS $dt)"
+ }
+ s"$value AS c$i"
+ }
+ // Leading nulls, updates in both directions, duplicate values, all-null
and single-value
+ // columns exercise initialization as well as the primitive and
reference bounds loops.
+ Seq("(NULL), (2), (-3), (0), (1), (2), (NULL)", "(NULL), (NULL)",
"(NULL), (1)").foreach {
Review Comment:
These fixtures stay inside `[-3, 2]`, so the new `FloatType` and
`DoubleType` arms never see a value where `Float.compare` and `<` disagree. One
more fixture with `NaN`, `-0.0`, the infinities and the integral extremes would
cover them.
Worth knowing before you add it: the bounds stored here use `Float.compare`
total ordering, where `-0.0 < 0.0` and NaN is greatest, while Spark's `min` and
`max` treat `-0.0` and `0.0` as equal. A fixture holding both would fail the
`expected` comparison because `java.lang.Double.equals` is bit-wise. Those two
bounds are better asserted directly than against the aggregate.
##########
spark/src/test/scala/org/apache/comet/exec/CometInMemoryCacheSuite.scala:
##########
@@ -351,6 +352,82 @@ class CometInMemoryCacheSuite extends CometTestBase {
}
}
+ test("Comet in-memory cache statistics preserve typed bounds and null
counts") {
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "false",
Review Comment:
Setting `COMET_ENABLED=false` here means the cached plan is row-based, so
the batches reaching `gatherColumnStats` are Arrow-backed `CometVector`s from
`rowToArrowBatchIter`. The numbers you measured came from Spark's
`OnHeapColumnVector` going through `convertColumnarBatchToCachedBatch` instead,
and that is also the path where the string `.copy()` is load bearing, since
`getUTF8String` there returns a view into the vector's own `byteData` rather
than an owned copy.
The suite already has `withSparkColumnarCache` for exactly this shape. Would
you mind running the same assertions through it as well?
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala:
##########
@@ -92,20 +95,169 @@ class ArrowCachedBatchSerializer extends
SimpleMetricsCachedBatchSerializer {
val dt = attrs(c).dataType
val col = batch.column(c)
var r = 0
- while (r < numRows) {
- if (col.isNullAt(r)) {
- nulls(c) += 1
- } else if (tracksBounds(dt)) {
- val value = readValue(col, dt, r)
- if (lower(c) == null || compare(dt, value, lower(c)) < 0) {
- lower(c) = value
+ var nullCount = 0
+ // Dispatch once per column and box only the final bounds. r ==
nullCount identifies
+ // the first non-null value, including when a column starts with nulls.
+ dt match {
+ case BooleanType =>
+ var min = false
+ var max = false
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getBoolean(r)
+ if (r == nullCount || JBoolean.compare(value, min) < 0) min =
value
+ if (r == nullCount || JBoolean.compare(value, max) > 0) max =
value
+ }
+ r += 1
}
- if (upper(c) == null || compare(dt, value, upper(c)) > 0) {
- upper(c) = value
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ case ByteType =>
+ var min = 0.toByte
+ var max = 0.toByte
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getByte(r)
+ if (r == nullCount || JByte.compare(value, min) < 0) min = value
+ if (r == nullCount || JByte.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ case ShortType =>
+ var min = 0.toShort
+ var max = 0.toShort
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getShort(r)
+ if (r == nullCount || JShort.compare(value, min) < 0) min = value
+ if (r == nullCount || JShort.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ case IntegerType | DateType =>
+ var min = 0
+ var max = 0
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getInt(r)
+ if (r == nullCount || JInteger.compare(value, min) < 0) min =
value
+ if (r == nullCount || JInteger.compare(value, max) > 0) max =
value
+ }
+ r += 1
+ }
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ case LongType | TimestampType | TimestampNTZType =>
+ var min = 0L
+ var max = 0L
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getLong(r)
+ if (r == nullCount || JLong.compare(value, min) < 0) min = value
+ if (r == nullCount || JLong.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ case FloatType =>
+ var min = 0.0f
+ var max = 0.0f
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getFloat(r)
+ if (r == nullCount || JFloat.compare(value, min) < 0) min = value
+ if (r == nullCount || JFloat.compare(value, max) > 0) max = value
+ }
+ r += 1
+ }
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ case DoubleType =>
+ var min = 0.0d
+ var max = 0.0d
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ 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
+ }
+ r += 1
+ }
+ if (nullCount < numRows) {
+ lower(c) = min
+ upper(c) = max
+ }
+ case d: DecimalType =>
+ var min: Decimal = null
+ var max: Decimal = null
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getDecimal(r, d.precision, d.scale)
+ if (min == null || value.compare(min) < 0) min = value
+ if (max == null || value.compare(max) > 0) max = value
+ }
+ r += 1
+ }
+ lower(c) = min
+ upper(c) = max
+ case StringType =>
+ val ordering = TypeUtils.getInterpretedOrdering(dt)
+ var min: UTF8String = null
+ var max: UTF8String = null
+ while (r < numRows) {
+ if (col.isNullAt(r)) {
+ nullCount += 1
+ } else {
+ val value = col.getUTF8String(r)
+ // Compare the UTF-8 bytes directly, without allocating getBytes
arrays.
+ // Borrow the value for comparison, but retain owned copies of
the bounds.
+ if (min == null || ordering.compare(value, min) < 0) min =
value.copy()
+ if (max == null || ordering.compare(value, max) > 0) max =
value.copy()
+ }
+ r += 1
+ }
+ lower(c) = min
+ upper(c) = max
+ case _ =>
Review Comment:
`tracksBounds` decides what `buildFilter` pushes down, and the match above
decides what actually gets bounds. Nothing links the two lists, and they are
two hundred lines apart. If someone adds a type to `tracksBounds` without
adding an arm here, the pushed predicate compares against null bounds and
prunes every batch, which drops rows silently rather than failing.
Your new test pins that binary bounds stay unset, which is today's snapshot
rather than the implication. Would an `assert(!tracksBounds(other))` in this
arm, or a test that walks the types `tracksBounds` accepts and requires
non-null bounds for each, be worth adding?
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowCachedBatchSerializer.scala:
##########
@@ -92,20 +95,169 @@ class ArrowCachedBatchSerializer extends
SimpleMetricsCachedBatchSerializer {
val dt = attrs(c).dataType
val col = batch.column(c)
var r = 0
- while (r < numRows) {
- if (col.isNullAt(r)) {
- nulls(c) += 1
- } else if (tracksBounds(dt)) {
- val value = readValue(col, dt, r)
- if (lower(c) == null || compare(dt, value, lower(c)) < 0) {
- lower(c) = value
+ var nullCount = 0
+ // Dispatch once per column and box only the final bounds. r ==
nullCount identifies
+ // the first non-null value, including when a column starts with nulls.
+ dt match {
+ case BooleanType =>
Review Comment:
The repetition is the right call here, since pushing these arms back through
`Any` callbacks would reintroduce the boxing you are removing. A lighter shape
would be one small private method per type returning `(min, max, nullCount)`,
which leaves `gatherColumnStats` as a readable dispatch, drops the shared `var
r` and `var nullCount`, and removes the seven copies of the `if (nullCount <
numRows)` epilogue. That trades two boxes per row for three per column per
batch.
Three smaller things either way. The primitive arms use `nullCount <
numRows` to mean "no values" while the decimal and string arms use a null
sentinel, so it would read better to pick one. `Array.fill[Int](numCols)(0)`
above is now dead initialization, since every arm assigns `nulls(c)`
unconditionally at the end of the loop. And `// Compare the UTF-8 bytes
directly, without allocating getBytes arrays` in the string arm describes the
code this replaces rather than the code that is there; the borrow-and-copy line
under it is the part a future reader actually needs.
--
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]
