Yaoxuan Wu created SPARK-60600:
----------------------------------
Summary: approx_top_k returns a count below the exact count when a
partial sketch was purged to zero items
Key: SPARK-60600
URL: https://issues.apache.org/jira/browse/SPARK-60600
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.1.1, 4.2.0
Reporter: Yaoxuan Wu
{code:python}
# two partitions: [0, 0, 0, 0, 0, 0, 0] and [0, 1, 2, 3, 4, 5, 6]; value 0
occurs 8 times
spark.range(0, 14, 1, 2).selectExpr("CASE WHEN id < 7 THEN 0 ELSE id - 7 END AS
v").createOrReplaceTempView("t")
spark.sql("SELECT approx_top_k(v, 1, 2) FROM t").collect()
# [Row(approx_top_k(v, 1, 2)=[Row(item=0, count=7)])] expected count >= 8
spark.sql("SELECT approx_top_k(v, 1, 2) FROM (SELECT /*+ COALESCE(1) */ * FROM
t)").collect()
# [Row(approx_top_k(v, 1, 2)=[Row(item=0, count=8)])]
{code}
The count Spark reports is {{ItemsSketch.getEstimate(item)}} for items from
{{getFrequentItems(NO_FALSE_POSITIVES)}} ({{ApproxTopKAggregates.scala}}). For
a tracked item, {{getEstimate}} is {{itemCount + offset}}, the same value as
{{getUpperBound(item)}}, which DataSketches documents as "a number which is
guaranteed to be no smaller than the real frequency". So a count of 7 for an
item that occurs 8 times is wrong: the second partition's contribution was
lost. The same through {{approx_top_k_accumulate}} / {{approx_top_k_combine}} /
{{approx_top_k_estimate}}.
Cause: the second partition's 7 distinct values (count 1 each) make its
{{ItemsSketch}} (maxItemsTracked 2, map capacity 6) purge every counter: it
keeps the subtracted count in its offset but has no active items. In
datasketches-java 6.2.0 (Spark's {{datasketches.version}}, also on master)
{{ItemsSketch.isEmpty()}} is {{getNumActiveItems() == 0}}, so {{merge(other)}}
returns early ({{if (other.isEmpty()) return this;}}) and {{toByteArray()}}
writes the empty image; the offset and stream weight of that partial sketch are
dropped in {{ApproxTopKAggregateBuffer.merge}} / {{serialize}}. This is fixed
upstream by apache/datasketches-java#770 ("Frequent items: empty means zero
stream weight, not zero active items", merged 2026-09-24, not yet released).
Spark needs that release, or a workaround in {{ApproxTopKAggregateBuffer}} that
merges / serializes sketches with a non-zero stream length explicitly.
Reproduces on 4.1.1 and 4.2.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
