[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 <50296686+SavicStefan@users.noreply.github.com>
Co-authored-by: Stefan Savić <stefan.savic@databricks.com>
Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>
(cherry picked from commit 2305552)
Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

Author: 2 people

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),

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),

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] {

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)
+  }
 }