This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 2305552f3856 [SPARK-59975][SQL] Handle all-null partial and final
tuple/theta intersection aggregates
2305552f3856 is described below
commit 2305552f38563de120e44051af4ed648f631c82d
Author: bajka <[email protected]>
AuthorDate: Fri Oct 9 13:20:52 2026 -0700
[SPARK-59975][SQL] Handle all-null partial and final tuple/theta
intersection aggregates
### What changes were proposed in this pull request?
Previously `theta_intersection_agg`, `tuple_intersection_agg_double` and
`tuple_intersection_agg_integer` threw an error when all input sketches
were NULL,
even for group by, when a single all-NULL group caused the entire query to
fail, even if
other groups contained valid sketches. The same happens, even without NULL
sketches,
for a global aggregate with an empty input partition or zero input rows,
and for a query
with two or more DISTINCT aggregates (`RewriteDistinctAggregates` replaces
the regular
aggregates' inputs with NULL on the expanded rows).
The error could also occur during partial aggregation when one partition
contained
only NULL sketches for a group, even if another partition contained valid
sketches
for that same group.
This change serializes untouched intermediate states as NULL and skips them
during merging. Final evaluation returns an empty sketch when all inputs for
the group are NULL.
We changed so that:
- `serialize()`: return an empty byte array for an untouched intersection
state instead of calling `getResult()`, which throws
- `merge()`: skip untouched incoming states or untouched incoming states,
real empty sketches still participate in the intersection
### Why are the changes needed?
Intersection aggregates currently throw when a partial aggregation receives
only
null sketches, which can fail a query even when another partition has a
valid sketch
for the same group. Untouched partials should be passed through as null and
only
an all-null final result should become an empty sketch.
### Does this PR introduce _any_ user-facing change?
Yes. `theta_intersection_agg`, `tuple_intersection_agg_double` and
`tuple_intersection_agg_integer` now return an empty sketch for all-null
input instead of throwing. Null-only partials also no longer cause otherwise
valid distributed aggregations to fail.
### How was this patch tested?
Added unit tests for the updated serialization, deserialization, merge and
final-evaluation behavior and integration tests covering:
- all-NULL input across partitions returns an empty final sketch
- grouped input correctly handles both an all-NULL group and a
group with a NULL-only partition plus a valid-sketch partition
- a global aggregate with an empty input partition and with zero input rows
- a query with multiple DISTINCT aggregates
- a window aggregate whose frame contains only NULL sketches
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex
Closes #59252 from SavicStefan/fix_intersection.
Lead-authored-by: bajka <[email protected]>
Co-authored-by: Stefan Savić <[email protected]>
Signed-off-by: Daniel Tenedorio <[email protected]>
---
.../aggregate/thetasketchesAggregates.scala | 28 +++-
.../aggregate/tupleIntersectionAgg.scala | 17 +++
.../expressions/aggregate/tupleSketchState.scala | 20 ++-
.../aggregate/ThetasketchesAggSuite.scala | 125 +++++++++++++++-
.../aggregate/TupleSketchAggSuite.scala | 159 +++++++++++++++++++++
.../apache/spark/sql/DataFrameAggregateSuite.scala | 122 ++++++++++++++++
6 files changed, 464 insertions(+), 7 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/thetasketchesAggregates.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/thetasketchesAggregates.scala
index a5326ae76c9f..dec472709e15 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/thetasketchesAggregates.scala
+++
b/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 {
+ Array.emptyByteArray
+ }
+ }
+
+ override def eval(): Array[Byte] = {
+ if (intersection.hasResult()) {
+ intersection.getResult.toByteArrayCompressed
+ } else {
+ new UpdateSketchBuilder().build().compact().toByteArrayCompressed
+ }
+ }
}
case class FinalizedSketch(sketch: CompactSketch) extends ThetaSketchState {
override def serialize(): Array[Byte] = sketch.toByteArrayCompressed
@@ -522,6 +536,13 @@ case class ThetaUnionAgg(
Examples:
> SELECT theta_sketch_estimate(_FUNC_(sketch)) FROM (SELECT
theta_sketch_agg(col) as sketch FROM VALUES (1) tab(col) UNION ALL SELECT
theta_sketch_agg(col, 20) as sketch FROM VALUES (1) tab(col));
1
+ > SELECT theta_sketch_estimate(_FUNC_(sketch)) FROM VALUES (CAST(NULL AS
BINARY)) tab(sketch);
+ 0
+ """,
+ note = """
+ NULL input sketches are ignored. If a group has no non-NULL input sketch,
the result is
+ an empty sketch. The empty sketch is not neutral for a later intersection:
intersecting
+ it with any other sketch returns an empty sketch.
""",
group = "agg_funcs",
since = "4.1.0")
@@ -617,6 +638,9 @@ case class ThetaIntersectionAgg(
intersectionBuffer: ThetaSketchState,
input: ThetaSketchState): ThetaSketchState = {
(intersectionBuffer, input) match {
+ // Untouched input states do not contribute to the intersection.
+ case (_, IntersectionAggregationBuffer(intersection)) if
!intersection.hasResult() =>
+ intersectionBuffer
// If both arguments are intersection objects, merge them directly.
case (
IntersectionAggregationBuffer(intersectLeft),
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleIntersectionAgg.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleIntersectionAgg.scala
index 975544d919e5..734191ff834c 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleIntersectionAgg.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleIntersectionAgg.scala
@@ -67,6 +67,13 @@ import org.apache.spark.sql.types.{AbstractDataType,
BinaryType, DataType}
Examples:
> SELECT tuple_sketch_estimate_double(_FUNC_(sketch)) FROM (SELECT
tuple_sketch_agg_double(key, summary) as sketch FROM VALUES (1, 5.0D), (2,
10.0D), (3, 15.0D) tab(key, summary) UNION ALL SELECT
tuple_sketch_agg_double(key, summary) as sketch FROM VALUES (2, 3.0D), (3,
7.0D), (4, 12.0D) tab(key, summary));
2.0
+ > SELECT tuple_sketch_estimate_double(_FUNC_(sketch)) FROM VALUES
(CAST(NULL AS BINARY)) tab(sketch);
+ 0.0
+ """,
+ note = """
+ NULL input sketches are ignored. If a group has no non-NULL input sketch,
the result is
+ an empty sketch. The empty sketch is not neutral for a later intersection:
intersecting
+ it with any other sketch returns an empty sketch.
""",
group = "agg_funcs",
since = "4.2.0")
@@ -165,6 +172,13 @@ case class TupleIntersectionAggDouble(
Examples:
> SELECT tuple_sketch_estimate_integer(_FUNC_(sketch)) FROM (SELECT
tuple_sketch_agg_integer(key, summary) as sketch FROM VALUES (1, 1), (2, 2),
(3, 3) tab(key, summary) UNION ALL SELECT tuple_sketch_agg_integer(key,
summary) as sketch FROM VALUES (2, 2), (3, 3), (4, 4) tab(key, summary));
2.0
+ > SELECT tuple_sketch_estimate_integer(_FUNC_(sketch)) FROM VALUES
(CAST(NULL AS BINARY)) tab(sketch);
+ 0.0
+ """,
+ note = """
+ NULL input sketches are ignored. If a group has no non-NULL input sketch,
the result is
+ an empty sketch. The empty sketch is not neutral for a later intersection:
intersecting
+ it with any other sketch returns an empty sketch.
""",
group = "agg_funcs",
since = "4.2.0")
@@ -315,6 +329,9 @@ abstract class TupleIntersectionAggBase[S <: Summary]
input: TupleSketchState[S]): TupleSketchState[S] = {
(intersectionBuffer, input) match {
+ // Untouched input states do not contribute to the intersection.
+ case (_, IntersectionTupleAggregationBuffer(intersection)) if
!intersection.hasResult() =>
+ intersectionBuffer
// The input was serialized then deserialized.
case (
intersectionBuffer @
IntersectionTupleAggregationBuffer(intersection),
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleSketchState.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleSketchState.scala
index 4129194a5220..12e1f7399076 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleSketchState.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/tupleSketchState.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.catalyst.expressions.aggregate
-import org.apache.datasketches.tuple.{Intersection, Sketch, Summary, Union,
UpdatableSketch, UpdatableSummary}
+import org.apache.datasketches.tuple.{Intersection, Sketch, Sketches, Summary,
Union, UpdatableSketch, UpdatableSummary}
/**
* Sealed trait representing the internal state of tuple sketch aggregation
operations.
@@ -45,8 +45,22 @@ case class UnionTupleAggregationBuffer[S <: Summary](union:
Union[S])
case class IntersectionTupleAggregationBuffer[S <: Summary](intersection:
Intersection[S])
extends TupleSketchState[S] {
- override def serialize(): Array[Byte] = intersection.getResult.toByteArray
- override def eval(): Array[Byte] = intersection.getResult.toByteArray
+ override def serialize(): Array[Byte] = {
+ // An untouched intersection represents no contribution, not an empty
sketch.
+ if (intersection.hasResult()) {
+ intersection.getResult.toByteArray
+ } else {
+ Array.emptyByteArray
+ }
+ }
+
+ override def eval(): Array[Byte] = {
+ if (intersection.hasResult()) {
+ intersection.getResult.toByteArray
+ } else {
+ Sketches.createEmptySketch[S]().toByteArray
+ }
+ }
}
case class FinalizedTupleSketch[S <: Summary](sketch: Sketch[S]) extends
TupleSketchState[S] {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ThetasketchesAggSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ThetasketchesAggSuite.scala
index afb978b8204d..0ff528e5c0be 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ThetasketchesAggSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ThetasketchesAggSuite.scala
@@ -22,8 +22,8 @@ import scala.util.Random
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{BoundReference,
ThetaSketchEstimate}
-import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.catalyst.expressions.{BoundReference,
GenericInternalRow, ThetaSketchEstimate}
+import org.apache.spark.sql.catalyst.util.{ArrayData, ThetaSketchUtils}
import org.apache.spark.sql.types.{ArrayType, BinaryType, DataType,
DoubleType, FloatType, IntegerType, LongType, StringType}
import org.apache.spark.unsafe.types.UTF8String
@@ -73,6 +73,31 @@ class ThetasketchesAggSuite extends SparkFunSuite {
mergedBuf.getResult.getLowerBound(3).toLong to
mergedBuf.getResult.getUpperBound(3).toLong)
}
+ private def buildThetaSketch(keys: Seq[Int]): Array[Byte] = {
+ val agg = new ThetaSketchAgg(BoundReference(0, IntegerType, nullable =
false))
+ val buffer = keys.foldLeft(agg.createAggregationBuffer()) { (buffer, key)
=>
+ agg.update(buffer, InternalRow(key))
+ }
+ agg.eval(buffer).asInstanceOf[Array[Byte]]
+ }
+
+ private def createIntersectionBuffer(
+ agg: ThetaIntersectionAgg,
+ sketches: Seq[Array[Byte]]): ThetaSketchState = {
+ sketches.foldLeft(agg.createAggregationBuffer()) { (buffer, sketch) =>
+ agg.update(buffer, InternalRow(sketch))
+ }
+ }
+
+ private def checkIntersectionEstimate(
+ agg: ThetaIntersectionAgg,
+ buffer: ThetaSketchState,
+ expected: Double): Unit = {
+ val result = agg.eval(buffer).asInstanceOf[Array[Byte]]
+ assert(result != null && result.nonEmpty)
+ assert(ThetaSketchUtils.wrapCompactSketch(result,
agg.prettyName).getEstimate == expected)
+ }
+
test("SPARK-52407: Test min/max values of supported datatypes") {
val intRange = Integer.MIN_VALUE to Integer.MAX_VALUE by 10000000
val (intEstimate, intEstimateRange) = simulateUpdateMerge(IntegerType,
intRange)
@@ -172,4 +197,100 @@ class ThetasketchesAggSuite extends SparkFunSuite {
.eval(InternalRow(intersectionResult))
assert(estimate.asInstanceOf[Long] >= 95 && estimate.asInstanceOf[Long] <=
105)
}
+
+ gridTest(
+ "SPARK-59975: theta intersection keeps no-input partials empty 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).isEmpty)
+ assert(agg.serialize(agg.deserialize(Array.emptyByteArray)).isEmpty)
+ 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).isEmpty)
+ val updated = agg.update(buffer, InternalRow(buildThetaSketch(Seq(1, 2))))
+ checkIntersectionEstimate(agg, updated, 2.0)
+ }
+
+ test("SPARK-59975: theta intersection preserves no-input state 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).isEmpty)
+
+ val result = agg.merge(
+ agg.createAggregationBuffer(),
+ agg.deserialize(agg.serialize(merged)))
+ checkIntersectionEstimate(agg, result, 0.0)
+ }
+
+ gridTest("SPARK-59975: 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("SPARK-59975: 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("SPARK-59975: theta intersection does not skip real empty partials
" +
+ "(serialized, emptyFirst) =")(
+ Seq((false, false), (false, true), (true, false), (true, true))) {
+ case (serialized, emptyFirst) =>
+ val agg = new ThetaIntersectionAgg(BoundReference(0, BinaryType,
nullable = true))
+ val empty = createIntersectionBuffer(agg,
Seq(buildThetaSketch(Seq.empty)))
+ val populated = createIntersectionBuffer(agg,
Seq(buildThetaSketch(Seq(1, 2))))
+ assert(agg.serialize(empty).nonEmpty)
+ val partials = if (emptyFirst) Seq(empty, populated) else Seq(populated,
empty)
+ val result = partials.foldLeft(agg.createAggregationBuffer()) { (buffer,
partial) =>
+ val input = if (serialized) agg.deserialize(agg.serialize(partial))
else partial
+ agg.merge(buffer, input)
+ }
+ assert(agg.serialize(result).nonEmpty)
+ checkIntersectionEstimate(agg, result, 0.0)
+ }
+
+ test("SPARK-59975: theta intersection preserves an empty intersection of
disjoint sketches") {
+ val agg = new ThetaIntersectionAgg(BoundReference(0, BinaryType, nullable
= true))
+ val disjoint = createIntersectionBuffer(agg,
+ Seq(buildThetaSketch(Seq(1)), buildThetaSketch(Seq(2))))
+ val serialized = agg.serialize(disjoint)
+ assert(serialized.nonEmpty)
+ assert(ThetaSketchUtils.wrapCompactSketch(serialized,
agg.prettyName).getEstimate == 0.0)
+
+ val populated = createIntersectionBuffer(agg, Seq(buildThetaSketch(Seq(1,
2))))
+ val result = agg.merge(populated, agg.deserialize(serialized))
+ checkIntersectionEstimate(agg, result, 0.0)
+ }
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/TupleSketchAggSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/TupleSketchAggSuite.scala
new file mode 100644
index 000000000000..2bdfbb7fefe4
--- /dev/null
+++
b/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 {
+ private val sketchInput = BoundReference(0, BinaryType, nullable = true)
+ private val keyInput = BoundReference(0, IntegerType, nullable = false)
+
+ private def buildSketch[T](agg: TypedImperativeAggregate[T], keys:
Seq[Int]): Array[Byte] = {
+ val buffer = keys.foldLeft(agg.createAggregationBuffer()) { (buffer, key)
=>
+ agg.update(buffer, InternalRow(key))
+ }
+ agg.eval(buffer).asInstanceOf[Array[Byte]]
+ }
+
+ private def testIntersection[T <: AnyRef](
+ createAggregate: () => TypedImperativeAggregate[T],
+ createSketch: Seq[Int] => Array[Byte],
+ estimate: Array[Byte] => Double): Unit = {
+ val name = createAggregate().prettyName
+
+ def createBuffer(agg: TypedImperativeAggregate[T], sketches:
Seq[Array[Byte]]): T = {
+ sketches.foldLeft(agg.createAggregationBuffer()) { (buffer, sketch) =>
+ agg.update(buffer, InternalRow(sketch))
+ }
+ }
+
+ def checkEstimate(agg: TypedImperativeAggregate[T], buffer: T, expected:
Double): Unit = {
+ val result = agg.eval(buffer).asInstanceOf[Array[Byte]]
+ assert(result != null && result.nonEmpty)
+ assert(estimate(result) == expected)
+ }
+
+ gridTest(s"SPARK-59975: $name keeps no-input partials empty and returns an
empty final sketch")(
+ Seq(0, 2)) { numNulls =>
+ val agg = createAggregate()
+ val buffer = createBuffer(agg, Seq.fill[Array[Byte]](numNulls)(null))
+ assert(agg.serialize(buffer).isEmpty)
+ assert(agg.serialize(agg.deserialize(Array.emptyByteArray)).isEmpty)
+ checkEstimate(agg, buffer, 0.0)
+
assert(agg.eval(buffer).asInstanceOf[Array[Byte]].sameElements(createSketch(Seq.empty)))
+
+ // Final evaluation must not change the untouched intermediate state.
+ assert(agg.serialize(buffer).isEmpty)
+ val updated = agg.update(buffer, InternalRow(createSketch(Seq(1, 2))))
+ checkEstimate(agg, updated, 2.0)
+ }
+
+ test(s"SPARK-59975: $name preserves no-input state through partial merge
and final merge") {
+ val agg = createAggregate()
+ val partial = createBuffer(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).isEmpty)
+
+ val result = agg.merge(
+ agg.createAggregationBuffer(),
+ agg.deserialize(agg.serialize(merged)))
+ checkEstimate(agg, result, 0.0)
+ }
+
+ gridTest(s"SPARK-59975: $name skips no-input partials (serialized,
nullFirst) =")(
+ Seq((false, false), (false, true), (true, false), (true, true))) {
+ case (serialized, nullFirst) =>
+ val agg = createAggregate()
+ val untouched = createBuffer(agg, Seq(null, null))
+ val populated = createBuffer(agg, Seq(null, createSketch(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)
+ }
+ checkEstimate(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)))
+ checkEstimate(agg, result, 2.0)
+ }
+
+ test(s"SPARK-59975: $name skips untouched objects in mergeBuffersObjects")
{
+ val agg = createAggregate()
+ val destination = new GenericInternalRow(1)
+ val incoming = new GenericInternalRow(1)
+ agg.initialize(destination)
+ agg.initialize(incoming)
+ agg.update(destination, InternalRow(createSketch(Seq(1, 2))))
+ agg.update(incoming, InternalRow(null: Array[Byte]))
+
+ agg.mergeBuffersObjects(destination, incoming)
+ val result = agg.eval(destination).asInstanceOf[Array[Byte]]
+ assert(estimate(result) == 2.0)
+ }
+
+ gridTest(s"SPARK-59975: $name does not skip real empty partials
(serialized, emptyFirst) =")(
+ Seq((false, false), (false, true), (true, false), (true, true))) {
+ case (serialized, emptyFirst) =>
+ val agg = createAggregate()
+ val empty = createBuffer(agg, Seq(createSketch(Seq.empty)))
+ val populated = createBuffer(agg, Seq(createSketch(Seq(1, 2))))
+ assert(agg.serialize(empty).nonEmpty)
+ val partials = if (emptyFirst) Seq(empty, populated) else
Seq(populated, empty)
+ val result = partials.foldLeft(agg.createAggregationBuffer()) {
(buffer, partial) =>
+ val input = if (serialized) agg.deserialize(agg.serialize(partial))
else partial
+ agg.merge(buffer, input)
+ }
+ assert(agg.serialize(result).nonEmpty)
+ checkEstimate(agg, result, 0.0)
+ }
+
+ test(s"SPARK-59975: $name preserves an empty intersection of disjoint
sketches") {
+ val agg = createAggregate()
+ val disjoint = createBuffer(agg, Seq(createSketch(Seq(1)),
createSketch(Seq(2))))
+ val serialized = agg.serialize(disjoint)
+ assert(serialized.nonEmpty)
+ assert(estimate(serialized) == 0.0)
+
+ val populated = createBuffer(agg, Seq(createSketch(Seq(1, 2))))
+ val result = agg.merge(populated, agg.deserialize(serialized))
+ checkEstimate(agg, result, 0.0)
+ }
+ }
+
+ testIntersection(
+ () => new TupleIntersectionAggDouble(sketchInput),
+ keys => buildSketch(new TupleSketchAggDouble(keyInput, Literal(1.0)),
keys),
+ bytes =>
+ TupleSketchUtils.heapifyDoubleSketch(bytes,
"tuple_intersection_agg_double").getEstimate)
+
+ testIntersection(
+ () => new TupleIntersectionAggInteger(sketchInput),
+ keys => buildSketch(new TupleSketchAggInteger(keyInput, Literal(1)), keys),
+ bytes =>
+ TupleSketchUtils.heapifyIntegerSketch(bytes,
"tuple_intersection_agg_integer").getEstimate)
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
index 320045a7a46e..4d9d981d7fd0 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -5031,6 +5031,128 @@ class DataFrameAggregateSuite extends SharedSparkSession
assert(estimateResult == 2.0)
}
+ private val intersectionSketchFunctions = Map(
+ "theta_intersection_agg" ->
+ ("theta_sketch_agg(id)", "theta_intersection_agg",
"theta_sketch_estimate"),
+ "tuple_intersection_agg_double" ->
+ ("tuple_sketch_agg_double(id, 1.0D)",
+ "tuple_intersection_agg_double", "tuple_sketch_estimate_double"),
+ "tuple_intersection_agg_integer" ->
+ ("tuple_sketch_agg_integer(id, 1)",
+ "tuple_intersection_agg_integer", "tuple_sketch_estimate_integer")
+ )
+
+ namedGridTest(
+ "SPARK-59975: intersection returns an empty sketch for all-null input")(
+ intersectionSketchFunctions) {
+ case (_, intersection, estimate) =>
+ val rows = spark.sparkContext.parallelize(Seq.fill(4)(Row(null)), 2)
+ val schema = new StructType().add("sketch", BinaryType)
+ withTempView("intersection_null_inputs") {
+ spark.createDataFrame(rows,
schema).createOrReplaceTempView("intersection_null_inputs")
+ checkAnswer(
+ sql(s"""
+ |SELECT CAST($estimate($intersection(sketch)) AS DOUBLE)
+ |FROM intersection_null_inputs
+ |""".stripMargin),
+ Row(0.0))
+ }
+ }
+
+ namedGridTest(
+ "SPARK-59975: intersection skips null-only partials for mixed groups")(
+ intersectionSketchFunctions) {
+ case (buildSketch, intersection, estimate) =>
+ val sketch = sql(s"SELECT $buildSketch FROM range(1,
3)").head().getAs[Array[Byte]](0)
+ // Group 0 is all-null. Group 1 has a null-only partial and a populated
partial.
+ val partitions = Seq(
+ Seq(Row(0, null), Row(1, null)),
+ Seq(Row(0, null), Row(1, sketch)))
+ val rows = spark.sparkContext.parallelize(partitions,
partitions.size).flatMap(_.iterator)
+ val schema = new StructType().add("group_id", IntegerType).add("sketch",
BinaryType)
+ withTempView("intersection_mixed_inputs") {
+ spark.createDataFrame(rows,
schema).createOrReplaceTempView("intersection_mixed_inputs")
+ checkAnswer(
+ sql(s"""
+ |SELECT group_id, CAST($estimate($intersection(sketch)) AS DOUBLE)
+ |FROM intersection_mixed_inputs
+ |GROUP BY group_id
+ |""".stripMargin),
+ Seq(Row(0, 0.0), Row(1, 2.0)))
+ }
+ }
+
+ namedGridTest(
+ "SPARK-59975: intersection skips empty partitions in a global aggregate")(
+ intersectionSketchFunctions) {
+ case (buildSketch, intersection, estimate) =>
+ val sketch = sql(s"SELECT $buildSketch FROM range(1,
3)").head().getAs[Array[Byte]](0)
+ // The first partition is empty, but a global aggregate still emits and
serializes
+ // a default partial buffer for it.
+ val rows = spark.sparkContext.parallelize(Seq(Row(sketch)), 2)
+ val schema = new StructType().add("sketch", BinaryType)
+ withTempView("intersection_empty_partition") {
+ spark.createDataFrame(rows,
schema).createOrReplaceTempView("intersection_empty_partition")
+ checkAnswer(
+ sql(s"""
+ |SELECT CAST($estimate($intersection(sketch)) AS DOUBLE)
+ |FROM intersection_empty_partition
+ |""".stripMargin),
+ Row(2.0))
+ // Zero input rows return an empty sketch.
+ checkAnswer(
+ sql(s"""
+ |SELECT CAST($estimate($intersection(sketch)) AS DOUBLE)
+ |FROM intersection_empty_partition
+ |WHERE false
+ |""".stripMargin),
+ Row(0.0))
+ }
+ }
+
+ namedGridTest(
+ "SPARK-59975: intersection with multiple DISTINCT aggregates")(
+ intersectionSketchFunctions) {
+ case (buildSketch, intersection, estimate) =>
+ val sketch = sql(s"SELECT $buildSketch FROM range(1,
3)").head().getAs[Array[Byte]](0)
+ withTempView("intersection_distinct_inputs") {
+ Seq((1, 10, sketch), (2, 20, sketch)).toDF("a", "b", "sketch")
+ .createOrReplaceTempView("intersection_distinct_inputs")
+ // RewriteDistinctAggregates sets the intersection input to NULL on
the expanded rows
+ // of each DISTINCT group, so the intersection sees all-NULL groups.
+ val df = sql(s"""
+ |SELECT
+ | CAST($estimate($intersection(sketch)) AS DOUBLE),
+ | count(DISTINCT a),
+ | count(DISTINCT b)
+ |FROM intersection_distinct_inputs
+ |""".stripMargin)
+ assert(df.queryExecution.optimizedPlan.collectFirst { case e: Expand
=> e }.nonEmpty)
+ checkAnswer(df, Row(2.0, 2L, 2L))
+ }
+ }
+
+ namedGridTest(
+ "SPARK-59975: intersection window aggregate over a frame with only null
sketches")(
+ intersectionSketchFunctions) {
+ case (buildSketch, intersection, estimate) =>
+ val sketch = sql(s"SELECT $buildSketch FROM range(1,
3)").head().getAs[Array[Byte]](0)
+ withTempView("intersection_window_inputs") {
+ Seq[(Int, Array[Byte])]((1, null), (2, sketch)).toDF("id", "sketch")
+ .createOrReplaceTempView("intersection_window_inputs")
+ // A window aggregate evaluates the frame buffer directly, without
serializing it.
+ checkAnswer(
+ sql(s"""
+ |SELECT
+ | id,
+ | CAST($estimate($intersection(sketch) OVER (
+ | ORDER BY id ROWS BETWEEN CURRENT ROW AND CURRENT ROW)) AS
DOUBLE)
+ |FROM intersection_window_inputs
+ |""".stripMargin),
+ Seq(Row(1, 0.0), Row(2, 2.0)))
+ }
+ }
+
test("SPARK-54179: tuple_sketch_agg + operations + estimate comprehensive
test - double") {
val df1 = Seq((1, "a", 1.0), (1, "a", 2.0), (1, "b", 3.0), (1, "c", 4.0))
.toDF("id", "key", "summary")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
