dtenedor commented on code in PR #59252:
URL: https://github.com/apache/spark/pull/59252#discussion_r4200070126
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala:
##########
@@ -5031,6 +5031,62 @@ class DataFrameAggregateSuite extends SharedSparkSession
assert(estimateResult == 2.0)
}
+ private val intersectionSketchFunctions = Seq(
Review Comment:
Two more cases that I believe hit the same exception before this change, and
that are probably the most common triggers in practice:
- A global aggregate where an input partition has no rows. For a global
aggregate, an empty partition still emits a default partial buffer
(`ObjectAggregationIterator.outputForEmptyGroupingKeyWithoutInput`), and that
buffer is serialized. So a selective `WHERE` filter, or a table with more
partitions than rows, fails even when no sketch is NULL. For example,
`spark.sparkContext.parallelize(Seq(Row(sketch)), 2)` leaves the first
partition empty and should return the sketch's estimate. Zero input rows (`...
WHERE false`) should return an empty sketch.
- The intersection aggregate together with two or more DISTINCT aggregates,
e.g. `SELECT theta_intersection_agg(sketch), count(DISTINCT a), count(DISTINCT
b) FROM t`. `RewriteDistinctAggregates` replaces the regular aggregates' inputs
with NULL on the expanded rows of each DISTINCT group, so the intersection sees
all-NULL groups even when no input sketch is NULL.
Optional: a window aggregate whose frame contains only NULL sketches (this
calls `eval` without serializing), and running the grouped test with
`spark.sql.objectHashAggregate.sortBased.fallbackThreshold=1` and with
`spark.sql.execution.useObjectHashAggregateExec=false`, since the sort-based
fallback and `SortAggregateExec` also serialize partial buffers.
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala:
##########
@@ -5031,6 +5031,62 @@ class DataFrameAggregateSuite extends SharedSparkSession
assert(estimateResult == 2.0)
}
+ private val intersectionSketchFunctions = Seq(
+ ("theta_sketch_agg(id)", "theta_intersection_agg",
"theta_sketch_estimate"),
+ ("tuple_sketch_agg_double(id, 1.0D)",
+ "tuple_intersection_agg_double", "tuple_sketch_estimate_double"),
+ ("tuple_sketch_agg_integer(id, 1)",
+ "tuple_intersection_agg_integer", "tuple_sketch_estimate_integer")
+ )
+
+ gridTest(
+ "intersection returns an empty sketch for all-null input")(
+ intersectionSketchFunctions) {
+ case (_, intersection, estimate) =>
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
Review Comment:
Nit: AQE and the shuffle partition count don't affect the partial
aggregation, whose partitions come from the input RDD, so these two confs can
be dropped (or add a comment saying why they're needed). Same in the next test.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/thetasketchesAggregates.scala:
##########
@@ -44,8 +44,22 @@ case class UnionAggregationBuffer(union: Union) extends
ThetaSketchState {
override def eval(): Array[Byte] = union.getResult.toByteArrayCompressed
}
case class IntersectionAggregationBuffer(intersection: Intersection) extends
ThetaSketchState {
- override def serialize(): Array[Byte] =
intersection.getResult.toByteArrayCompressed
- override def eval(): Array[Byte] =
intersection.getResult.toByteArrayCompressed
+ override def serialize(): Array[Byte] = {
Review Comment:
Optional simplification: `deserialize` already maps an empty byte array to a
fresh `createAggregationBuffer()`, and the new `!hasResult()` guard in `merge`
skips that fresh buffer. If `serialize` returned `Array.emptyByteArray` for an
untouched intersection, the `buffer == null` branch in `deserialize` and the
`case (_, null)` guard in `merge` wouldn't be needed (here and in
`tupleIntersectionAgg.scala`; the same change would apply to
`IntersectionTupleAggregationBuffer.serialize`). `deserialize` would also keep
returning a buffer object, as the `TypedImperativeAggregate` docs describe, and
this matches `HllUnionAgg`, which serializes its empty state as an empty array.
I traced the null path through `ObjectHashAggregateExec` (including the
sort-based fallback), `SortAggregateExec` and `AggregatingAccumulator` and
didn't find a problem, so this is about keeping the contract simple, not a bug.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/thetasketchesAggregates.scala:
##########
@@ -44,8 +44,22 @@ case class UnionAggregationBuffer(union: Union) extends
ThetaSketchState {
override def eval(): Array[Byte] = union.getResult.toByteArrayCompressed
}
case class IntersectionAggregationBuffer(intersection: Intersection) extends
ThetaSketchState {
- override def serialize(): Array[Byte] =
intersection.getResult.toByteArrayCompressed
- override def eval(): Array[Byte] =
intersection.getResult.toByteArrayCompressed
+ override def serialize(): Array[Byte] = {
+ // An untouched intersection represents no contribution, not an empty
sketch.
+ if (intersection.hasResult()) {
+ intersection.getResult.toByteArrayCompressed
+ } else {
+ null
+ }
+ }
+
+ override def eval(): Array[Byte] = {
+ if (intersection.hasResult()) {
+ intersection.getResult.toByteArrayCompressed
+ } else {
+ new UpdateSketchBuilder().build().compact().toByteArrayCompressed
Review Comment:
Since this is a user-facing change, could you document it in the
`ExpressionDescription` of `theta_intersection_agg`,
`tuple_intersection_agg_double` and `tuple_intersection_agg_integer`? For
example, a `note` such as "NULL input sketches are ignored. If a group has no
non-NULL input sketch, the result is an empty sketch." An example like `SELECT
theta_sketch_estimate(_FUNC_(sketch)) FROM VALUES (CAST(NULL AS BINARY))
tab(sketch);` returning `0` would also be run by `ExpressionInfoSuite`. It's
probably worth saying that the empty result is not neutral for a later
intersection: intersecting per-group results again gives an empty sketch if any
group had only NULL inputs.
In the PR description, a sentence on why the result is an empty sketch
rather than NULL would help reviewers: these functions are declared
non-nullable, and `theta_sketch_agg`, `theta_union_agg` and `tuple_union_agg_*`
already return an empty sketch for all-NULL input.
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala:
##########
@@ -5031,6 +5031,62 @@ class DataFrameAggregateSuite extends SharedSparkSession
assert(estimateResult == 2.0)
}
+ private val intersectionSketchFunctions = Seq(
+ ("theta_sketch_agg(id)", "theta_intersection_agg",
"theta_sketch_estimate"),
+ ("tuple_sketch_agg_double(id, 1.0D)",
+ "tuple_intersection_agg_double", "tuple_sketch_estimate_double"),
+ ("tuple_sketch_agg_integer(id, 1)",
+ "tuple_intersection_agg_integer", "tuple_sketch_estimate_integer")
+ )
+
+ gridTest(
+ "intersection returns an empty sketch for all-null input")(
Review Comment:
Nit: since the grid parameter is the whole 3-tuple, every test name contains
all three SQL snippets. Using just the intersection function name would read
better. Also, please prefix new test names with `SPARK-59975:` like the
neighboring tests (`SPARK-54179: ...`). The same goes for the new tests in
`ThetasketchesAggSuite` and `TuplesketchAggSuite`.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/TuplesketchAggSuite.scala:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{BoundReference,
GenericInternalRow, Literal}
+import org.apache.spark.sql.catalyst.util.TupleSketchUtils
+import org.apache.spark.sql.types.{BinaryType, IntegerType}
+
+class TuplesketchAggSuite extends SparkFunSuite {
Review Comment:
Nit: `TupleSketchAggSuite` would match `tupleSketchAgg.scala` and
`TupleSketchUtilsSuite`. The theta tests in `ThetasketchesAggSuite` are nearly
the same as the ones generated by `testIntersection` here, so sharing that
helper would remove the duplication.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ThetasketchesAggSuite.scala:
##########
@@ -172,4 +197,99 @@ class ThetasketchesAggSuite extends SparkFunSuite {
.eval(InternalRow(intersectionResult))
assert(estimate.asInstanceOf[Long] >= 95 && estimate.asInstanceOf[Long] <=
105)
}
+
+ gridTest(
+ "theta intersection keeps no-input partials null and returns an empty
final")(
+ Seq(0, 2)) { numNulls =>
+ val agg = new ThetaIntersectionAgg(BoundReference(0, BinaryType, nullable
= true))
+ val buffer = createIntersectionBuffer(agg,
Seq.fill[Array[Byte]](numNulls)(null))
+ assert(agg.serialize(buffer) == null)
+ assert(agg.deserialize(null) == null)
+ checkIntersectionEstimate(agg, buffer, 0.0)
+
assert(agg.eval(buffer).asInstanceOf[Array[Byte]].sameElements(buildThetaSketch(Seq.empty)))
+
+ // Final evaluation must not change the untouched intermediate state.
+ assert(agg.serialize(buffer) == null)
+ val updated = agg.update(buffer, InternalRow(buildThetaSketch(Seq(1, 2))))
+ checkIntersectionEstimate(agg, updated, 2.0)
+ }
+
+ test("theta intersection preserves null through partial merge and final
merge") {
+ val agg = new ThetaIntersectionAgg(BoundReference(0, BinaryType, nullable
= true))
+ val partial = createIntersectionBuffer(agg, Seq(null, null))
+ val partialMerge = agg.merge(
+ agg.createAggregationBuffer(),
+ agg.deserialize(agg.serialize(partial)))
+ val merged = agg.merge(partialMerge,
agg.deserialize(agg.serialize(partial)))
+ assert(agg.serialize(merged) == null)
+
+ val result = agg.merge(
+ agg.createAggregationBuffer(),
+ agg.deserialize(agg.serialize(merged)))
+ checkIntersectionEstimate(agg, result, 0.0)
+ }
+
+ gridTest("theta intersection skips no-input partials (serialized,
nullFirst)")(
+ Seq((false, false), (false, true), (true, false), (true, true))) {
+ case (serialized, nullFirst) =>
+ val agg = new ThetaIntersectionAgg(BoundReference(0, BinaryType,
nullable = true))
+ val untouched = createIntersectionBuffer(agg, Seq(null, null))
+ val populated = createIntersectionBuffer(agg, Seq(null,
buildThetaSketch(Seq(1, 2)), null))
+ val partials = if (nullFirst) Seq(untouched, populated) else
Seq(populated, untouched)
+ val merged = partials.foldLeft(agg.createAggregationBuffer()) { (buffer,
partial) =>
+ val input = if (serialized) agg.deserialize(agg.serialize(partial))
else partial
+ agg.merge(buffer, input)
+ }
+ checkIntersectionEstimate(agg, merged, 2.0)
+
+ // A further partial-merge serialization must preserve the populated
result.
+ val result = agg.merge(
+ agg.createAggregationBuffer(),
+ agg.deserialize(agg.serialize(merged)))
+ checkIntersectionEstimate(agg, result, 2.0)
+ }
+
+ test("theta intersection skips untouched objects in mergeBuffersObjects") {
+ val agg = new ThetaIntersectionAgg(BoundReference(0, BinaryType, nullable
= true))
+ val destination = new GenericInternalRow(1)
+ val incoming = new GenericInternalRow(1)
+ agg.initialize(destination)
+ agg.initialize(incoming)
+ agg.update(destination, InternalRow(buildThetaSketch(Seq(1, 2))))
+ agg.update(incoming, InternalRow(null: Array[Byte]))
+
+ agg.mergeBuffersObjects(destination, incoming)
+ val result = agg.eval(destination).asInstanceOf[Array[Byte]]
+ assert(ThetaSketchUtils.wrapCompactSketch(result,
agg.prettyName).getEstimate == 2.0)
+ }
+
+ gridTest("theta intersection does not skip real empty partials, " +
Review Comment:
Nit: the gridTest names are inconsistent: `(serialized, emptyFirst) =` here
vs `(serialized, nullFirst)` without `=` above (same in `TuplesketchAggSuite`).
--
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]
