From cd666d25d8253ee99e84b36ef44e9760cba9a4ba Mon Sep 17 00:00:00 2001 From: comphead Date: Fri, 30 Oct 2026 10:37:45 -0700 Subject: [PATCH 1/3] perf: merge hll_sketch_agg partial states into one union per group merge_batch merged every partial sketch through a new SparkHllUnion, built by SparkHllSketch::merge_sketch. Once a group's sketch holds the dense register array, datasketches 0.3.0 copies that array into the empty union and rescans all 2^lgConfigK registers, so every merged row cost as much as the whole array. HllSketchAccumulator is now an enum of a sketch and a union. Rows are hashed into the sketch, and the first merge_batch folds it into a union that merges every partial sketch from then on, as HllUnionAccumulator does. Rows that arrive after a merge reach the union through a sketch of their own. Adds a Final-mode case to the hll_agg bench, with groups that turn dense at lgConfigK 18. Part of #7341. --- native/spark-expr/benches/hll_agg.rs | 86 +++++++++- native/spark-expr/src/agg_funcs/hll_sketch.rs | 28 +-- .../src/agg_funcs/hll_sketch_agg.rs | 160 ++++++++++++++---- 3 files changed, 207 insertions(+), 67 deletions(-) diff --git a/native/spark-expr/benches/hll_agg.rs b/native/spark-expr/benches/hll_agg.rs index c933533f7c6..6405c564a80 100644 --- a/native/spark-expr/benches/hll_agg.rs +++ b/native/spark-expr/benches/hll_agg.rs @@ -20,6 +20,8 @@ //! Each runs as a grouped `AggregateMode::Partial` over 8192 rows x 10 batches, mirroring //! the other aggregate benches. The union input is a column of real sketch bytes minted //! up front with `SparkHllSketch`, since the kernel deserializes and merges each one. +//! `hll_sketch_agg_final_dense` runs `hll_sketch_agg` as an `AggregateMode::Final` over +//! partial states instead, with groups that turn dense at lgConfigK 18. use arrow::array::{ArrayRef, BinaryBuilder, Int64Builder, RecordBatch, StringBuilder}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; @@ -43,19 +45,32 @@ use tokio::runtime::Runtime; const LG_CONFIG_K: u8 = 12; /// Number of distinct sketches minted for the union input; rows cycle through them. const SKETCH_POOL: usize = 256; +/// `lgConfigK` for the Final-mode case, where a dense group holds 2^18 one-byte registers. +const FINAL_LG_CONFIG_K: u8 = 18; +/// Runs the aggregate in `mode` over `partitions`. `input_schema` is the schema of the rows the +/// Partial stage reads, which in Final mode is not the schema of `partitions`. async fn agg_test( partitions: &[Vec], + mode: AggregateMode, + input_schema: &SchemaRef, group_col: Arc, input_exprs: Vec>, aggregate_udf: Arc, alias: &str, ) { - let schema = &partitions[0][0].schema(); let scan: Arc = Arc::new(DataSourceExec::new(Arc::new( - MemorySourceConfig::try_new(partitions, Arc::clone(schema), None).unwrap(), + MemorySourceConfig::try_new(partitions, partitions[0][0].schema(), None).unwrap(), ))); - let aggregate = create_aggregate(scan, group_col, input_exprs, schema, aggregate_udf, alias); + let aggregate = create_aggregate( + scan, + mode, + group_col, + input_exprs, + input_schema, + aggregate_udf, + alias, + ); let mut stream = aggregate .execute(0, Arc::new(TaskContext::default())) .unwrap(); @@ -66,14 +81,15 @@ async fn agg_test( fn create_aggregate( scan: Arc, + mode: AggregateMode, group_col: Arc, input_exprs: Vec>, - schema: &SchemaRef, + input_schema: &SchemaRef, aggregate_udf: Arc, alias: &str, ) -> Arc { let aggr_expr = AggregateExprBuilder::new(aggregate_udf, input_exprs) - .schema(schema.clone()) + .schema(input_schema.clone()) .alias(alias) .with_ignore_nulls(false) .with_distinct(false) @@ -82,12 +98,12 @@ fn create_aggregate( Arc::new( AggregateExec::try_new( - AggregateMode::Partial, + mode, PhysicalGroupBy::new_single(vec![(group_col, "c0".to_string())]), vec![aggr_expr.into()], vec![None], // no filter expressions scan, - Arc::clone(schema), + Arc::clone(input_schema), ) .unwrap(), ) @@ -134,10 +150,38 @@ fn create_record_batch(num_rows: usize, pool: &[Vec]) -> RecordBatch { RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap() } +/// Partial `hll_sketch_agg` states for the Final-mode case: 20 groups, each merging 400 partial +/// sketches of 200 distinct values. After 123 of them a group has more than 24,576 distinct +/// values, the most a lgConfigK 18 sketch holds before it moves to the register array. `c0` is +/// the Int64 group key and `c1` the state, one partial per group per round, the way a reducer +/// reads map outputs. +fn partial_states() -> RecordBatch { + let (groups, partials, values) = (20i64, 400i64, 200i64); + let mut keys = Int64Builder::with_capacity((groups * partials) as usize); + let mut states = BinaryBuilder::new(); + for round in 0..partials { + for group in 0..groups { + let mut sketch = SparkHllSketch::new(FINAL_LG_CONFIG_K); + let start = (round * groups + group) * values; + for v in start..start + values { + sketch.update_i64(v); + } + keys.append_value(group); + states.append_value(sketch.to_sketch_bytes()); + } + } + RecordBatch::try_from_iter([ + ("c0", Arc::new(keys.finish()) as ArrayRef), + ("c1", Arc::new(states.finish()) as ArrayRef), + ]) + .unwrap() +} + fn criterion_benchmark(c: &mut Criterion) { let num_rows = 8192; let pool = sketch_pool(); let batch = create_record_batch(num_rows, &pool); + let schema = batch.schema(); let batches = vec![batch; 10]; let partitions = &[batches]; @@ -155,6 +199,8 @@ fn criterion_benchmark(c: &mut Criterion) { b.to_async(&rt).iter(|| { black_box(agg_test( partitions, + AggregateMode::Partial, + &schema, c0.clone(), vec![c1.clone()], sketch_udf.clone(), @@ -169,6 +215,8 @@ fn criterion_benchmark(c: &mut Criterion) { b.to_async(&rt).iter(|| { black_box(agg_test( partitions, + AggregateMode::Partial, + &schema, c0.clone(), vec![c2.clone()], union_udf.clone(), @@ -177,6 +225,30 @@ fn criterion_benchmark(c: &mut Criterion) { }) }); + // The Final stage reads the states, but its aggregate is declared against the Int64 values + // the Partial stage hashed. + let final_partitions = &[vec![partial_states()]]; + let final_input_schema: SchemaRef = Arc::new(Schema::new(vec![ + Field::new("c0", DataType::Int64, false), + Field::new("c1", DataType::Int64, false), + ])); + let final_udf: Arc = Arc::new(AggregateUDF::new_from_impl(HllSketchAgg::new( + FINAL_LG_CONFIG_K as i32, + ))); + group.bench_function("hll_sketch_agg_final_dense", |b| { + b.to_async(&rt).iter(|| { + black_box(agg_test( + final_partitions, + AggregateMode::Final, + &final_input_schema, + c0.clone(), + vec![c1.clone()], + final_udf.clone(), + "hll_sketch_agg", + )) + }) + }); + group.finish(); } diff --git a/native/spark-expr/src/agg_funcs/hll_sketch.rs b/native/spark-expr/src/agg_funcs/hll_sketch.rs index ebb52156cd3..2cebe5a846d 100644 --- a/native/spark-expr/src/agg_funcs/hll_sketch.rs +++ b/native/spark-expr/src/agg_funcs/hll_sketch.rs @@ -269,14 +269,6 @@ impl SparkHllSketch { self.inner.estimate() } - /// Merge another sketch into this one via a union, keeping HLL_8 output. - pub fn merge_sketch(&mut self, other: &SparkHllSketch) { - let mut u = SparkHllUnion::new(self.lg_config_k()); - u.merge(self); - u.merge(other); - *self = u.to_sketch(); - } - fn is_dense(&self) -> bool { layout::is_dense(self.lg_config_k(), self.dense, self.inner.estimate()) } @@ -305,20 +297,17 @@ impl SparkHllUnion { } } + /// The `lgMaxK` the union was created with. + pub fn lg_max_k(&self) -> u8 { + self.inner.lg_max_k() + } + /// Merge a sketch into the union. pub fn merge(&mut self, sketch: &SparkHllSketch) { self.dense = self.is_dense() || sketch.is_dense(); self.inner.update(&sketch.inner); } - /// The union result as an HLL_8 sketch. - fn to_sketch(&self) -> SparkHllSketch { - SparkHllSketch { - inner: self.inner.to_sketch(HllType::Hll8), - dense: self.is_dense(), - } - } - /// The union result as an HLL_8 sketch's serialized bytes. pub fn to_sketch_bytes(&self) -> Vec { self.inner.to_sketch(HllType::Hll8).serialize() @@ -579,12 +568,5 @@ mod tests { assert!(union.inner.estimate() < 100.0); assert_eq!(allocated(&union.to_sketch_bytes()), 1 << 16); assert_eq!(union.heap_size(), 1 << 16); - - // `merge_sketch` goes through a union too, and keeps the flag. - let mut small = sketch(0..5); - small.merge_sketch(&barely_used); - assert!(small.estimate() < 100.0); - assert_eq!(allocated(&small.to_sketch_bytes()), 1 << 16); - assert_eq!(small.heap_size(), 1 << 16); } } diff --git a/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs b/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs index 3bcb5d44e1c..0d854e37955 100644 --- a/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs +++ b/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use crate::agg_funcs::hll_sketch::SparkHllSketch; +use crate::agg_funcs::hll_sketch::{SparkHllSketch, SparkHllUnion}; use arrow::array::Array; use arrow::array::ArrayRef; use arrow::array::BinaryArray; @@ -78,45 +78,76 @@ impl AggregateUDFImpl for HllSketchAgg { } } +/// Input rows are hashed into a sketch. The first `merge_batch` turns that sketch into a union, +/// which merges every partial sketch from then on, as `HllUnionAccumulator` does. Spark's +/// `HllSketchAgg.merge` builds a new union for each merged row instead. With `datasketches` 0.3.0 +/// that copies a dense accumulator's register array into the new union and rescans all +/// `2^lgConfigK` registers, so every merged row would cost as much as the whole array. #[derive(Debug)] -pub struct HllSketchAccumulator { - sketch: SparkHllSketch, +pub enum HllSketchAccumulator { + Sketch(SparkHllSketch), + Union(SparkHllUnion), } impl HllSketchAccumulator { pub fn new(lg_config_k: u8) -> Self { - Self { - sketch: SparkHllSketch::new(lg_config_k), + Self::Sketch(SparkHllSketch::new(lg_config_k)) + } + + fn update_sketch(sketch: &mut SparkHllSketch, arr: &ArrayRef) -> Result<()> { + // Downcast once per batch rather than going through `ScalarValue::try_from_array` per + // row: for the string and binary cases that copies every value onto the heap only to + // hash it and drop it again. + match arr.data_type() { + DataType::Int8 => Self::update_ints(sketch, as_primitive_array::(arr)), + DataType::Int16 => Self::update_ints(sketch, as_primitive_array::(arr)), + DataType::Int32 => Self::update_ints(sketch, as_primitive_array::(arr)), + DataType::Int64 => Self::update_ints(sketch, as_primitive_array::(arr)), + DataType::Utf8 => Self::update_byte_slices(sketch, downcast_value!(arr, StringArray)), + DataType::Binary => Self::update_byte_slices(sketch, downcast_value!(arr, BinaryArray)), + other => { + return Err(DataFusionError::Internal(format!( + "hll_sketch_agg received an unsupported input type: {other:?}" + ))) + } } + Ok(()) } /// Spark widens every accepted integral to `long` before hashing, so all four widths /// funnel through the same `i64` update. Nulls are ignored, matching `HllSketchAgg`. - fn update_ints(&mut self, arr: &PrimitiveArray) + fn update_ints(sketch: &mut SparkHllSketch, arr: &PrimitiveArray) where T: ArrowPrimitiveType, T::Native: Into, { for i in 0..arr.len() { if !arr.is_null(i) { - self.sketch.update_i64(arr.value(i).into()); + sketch.update_i64(arr.value(i).into()); } } } /// StringType hashes its UTF-8 bytes and BinaryType its bytes directly, so both share /// this loop. - fn update_byte_slices(&mut self, arr: &GenericByteArray) + fn update_byte_slices(sketch: &mut SparkHllSketch, arr: &GenericByteArray) where T: ByteArrayType, for<'a> &'a T::Native: AsRef<[u8]>, { for i in 0..arr.len() { if !arr.is_null(i) { - self.sketch.update_bytes(arr.value(i).as_ref()); + sketch.update_bytes(arr.value(i).as_ref()); } } } + + fn to_sketch_bytes(&self) -> Vec { + match self { + Self::Sketch(sketch) => sketch.to_sketch_bytes(), + Self::Union(union) => union.to_sketch_bytes(), + } + } } impl Accumulator for HllSketchAccumulator { @@ -124,53 +155,54 @@ impl Accumulator for HllSketchAccumulator { if values.is_empty() { return Ok(()); } - let arr = &values[0]; - // Downcast once per batch rather than going through `ScalarValue::try_from_array` per - // row: for the string and binary cases that copies every value onto the heap only to - // hash it and drop it again. - match arr.data_type() { - DataType::Int8 => self.update_ints(as_primitive_array::(arr)), - DataType::Int16 => self.update_ints(as_primitive_array::(arr)), - DataType::Int32 => self.update_ints(as_primitive_array::(arr)), - DataType::Int64 => self.update_ints(as_primitive_array::(arr)), - DataType::Utf8 => self.update_byte_slices(downcast_value!(arr, StringArray)), - DataType::Binary => self.update_byte_slices(downcast_value!(arr, BinaryArray)), - other => { - return Err(DataFusionError::Internal(format!( - "hll_sketch_agg received an unsupported input type: {other:?}" - ))) + match self { + Self::Sketch(sketch) => Self::update_sketch(sketch, &values[0]), + // DataFusion gives an accumulator either rows or partial sketches. Rows that follow + // a merge anyway reach the union through a sketch of their own. + Self::Union(union) => { + let mut sketch = SparkHllSketch::new(union.lg_max_k()); + Self::update_sketch(&mut sketch, &values[0])?; + union.merge(&sketch); + Ok(()) } } - Ok(()) } fn evaluate(&mut self) -> Result { // Spark's HllSketchAgg is declared non-nullable: an empty/all-null group // still returns a serialized empty sketch (which estimates to 0), never NULL. - Ok(ScalarValue::Binary(Some(self.sketch.to_sketch_bytes()))) + Ok(ScalarValue::Binary(Some(self.to_sketch_bytes()))) } fn size(&self) -> usize { - // What the sketch holds now, not the 1 << lgConfigK dense maximum: a group stays in a - // few-dozen-byte coupon list until it has seen enough distinct values. - std::mem::size_of_val(self) + self.sketch.heap_size() + // What the sketch or union holds now, not the 1 << lgConfigK dense maximum: a group stays + // in a few-dozen-byte coupon list until it has seen enough distinct values. + let heap = match self { + Self::Sketch(sketch) => sketch.heap_size(), + Self::Union(union) => union.heap_size(), + }; + std::mem::size_of_val(self) + heap } fn state(&mut self) -> Result> { - Ok(vec![ScalarValue::Binary(Some( - self.sketch.to_sketch_bytes(), - ))]) + Ok(vec![ScalarValue::Binary(Some(self.to_sketch_bytes()))]) } fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { let arr = downcast_value!(states[0], BinaryArray); + if let Self::Sketch(sketch) = self { + // Rows hashed before the first merge carry over into the union. + let mut union = SparkHllUnion::new(sketch.lg_config_k()); + union.merge(sketch); + *self = Self::Union(union); + } + let Self::Union(union) = self else { + unreachable!("the sketch was replaced by a union above") + }; for i in 0..arr.len() { - if arr.is_null(i) { - continue; + if !arr.is_null(i) { + union.merge(&SparkHllSketch::from_bytes(arr.value(i))?); } - let peer = SparkHllSketch::from_bytes(arr.value(i))?; - // Merge peer into self by unioning; reuse SparkHllUnion via sketch merge. - self.sketch.merge_sketch(&peer); } Ok(()) } @@ -288,6 +320,60 @@ mod tests { ); } + /// A Final aggregation hands a group its partial sketches over many `merge_batch` calls, and + /// they all have to go into one union. After every call the accumulator holds exactly what a + /// single `SparkHllUnion` fed the same sketches holds, in SET mode and once dense. + #[test] + fn merge_batch_keeps_one_union_across_calls() { + let partials: Vec> = (0..40i64) + .map(|p| { + let mut sketch = SparkHllSketch::new(12); + for v in p * 50..(p + 1) * 50 { + sketch.update_i64(v); + } + sketch.to_sketch_bytes() + }) + .collect(); + let mut acc = HllSketchAccumulator::new(12); + let mut union = SparkHllUnion::new(12); + for chunk in partials.chunks(7) { + acc.merge_batch(&[Arc::new(BinaryArray::from_iter_values(chunk))]) + .unwrap(); + for bytes in chunk { + union.merge(&SparkHllSketch::from_bytes(bytes).unwrap()); + } + assert_eq!(acc.to_sketch_bytes(), union.to_sketch_bytes()); + } + // 2,000 distinct values are well past the 384 coupons a lgConfigK 12 sketch holds before + // it moves to the register array. + assert!(acc.size() >= 1 << 12, "size {}", acc.size()); + let est = crate::agg_funcs::estimate_from_bytes(&sketch_bytes(&mut acc)).unwrap(); + assert!((est - 2000).abs() <= 60, "estimate {est}"); + } + + /// DataFusion gives an accumulator either rows or partial sketches, but one that gets both + /// still has to count all of them: the rows before its first merge, the merged sketch, and + /// the rows after. + #[test] + fn rows_before_and_after_a_merge_all_count() { + let ints = |values: std::ops::Range| -> ArrayRef { + Arc::new(Int64Array::from_iter_values(values)) + }; + let mut peer = SparkHllSketch::new(12); + for v in 1000..2000 { + peer.update_i64(v); + } + let mut acc = HllSketchAccumulator::new(12); + acc.update_batch(&[ints(0..1000)]).unwrap(); + acc.merge_batch(&[Arc::new(BinaryArray::from_iter_values([ + peer.to_sketch_bytes() + ]))]) + .unwrap(); + acc.update_batch(&[ints(2000..3000)]).unwrap(); + let est = crate::agg_funcs::estimate_from_bytes(&sketch_bytes(&mut acc)).unwrap(); + assert!((est - 3000).abs() <= 90, "estimate {est}"); + } + /// 64 singleton groups at lgConfigK=21, grouped through a real Partial/Final plan in a 16 MiB /// pool. Each group's sketch holds one coupon in an 8-slot LIST, 32 bytes, and Spark runs the /// same query without trouble. Charging every group the 2 MiB dense register array instead From 0d514bb0cde1a6a6303617a62f3ac325320aa570 Mon Sep 17 00:00:00 2001 From: comphead Date: Fri, 30 Oct 2026 13:39:07 -0700 Subject: [PATCH 2/3] address review: get the union through union_mut, compare with Spark in SQL merge_batch now takes its union from union_mut, which turns a Sketch into a Union and returns it. It calls itself after the swap, so the only way out is the Union arm and the unreachable! is gone. Adds expressions/aggregate/hll_sketch_agg.sql, which compares hll_sketch_agg with Spark: every input type with NULLs and bounds, and four files that the Final aggregation merges as one partial per group. Coupon-set groups and groups merging partials that already hold the register array match exactly. Where coupon-set partials take the union to the register array, the estimate depends on merge order in Spark too, so that query compares within 1%. --- .../src/agg_funcs/hll_sketch_agg.rs | 24 ++++-- .../expressions/aggregate/hll_sketch_agg.sql | 86 +++++++++++++++++++ 2 files changed, 101 insertions(+), 9 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql diff --git a/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs b/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs index 0d854e37955..1b2627b778c 100644 --- a/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs +++ b/native/spark-expr/src/agg_funcs/hll_sketch_agg.rs @@ -148,6 +148,20 @@ impl HllSketchAccumulator { Self::Union(union) => union.to_sketch_bytes(), } } + + /// The union that partial sketches merge into. The first call turns the sketch into one, so + /// rows hashed before the first merge carry over. + fn union_mut(&mut self) -> &mut SparkHllUnion { + match self { + Self::Union(union) => union, + Self::Sketch(sketch) => { + let mut union = SparkHllUnion::new(sketch.lg_config_k()); + union.merge(sketch); + *self = Self::Union(union); + self.union_mut() + } + } + } } impl Accumulator for HllSketchAccumulator { @@ -190,15 +204,7 @@ impl Accumulator for HllSketchAccumulator { fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> { let arr = downcast_value!(states[0], BinaryArray); - if let Self::Sketch(sketch) = self { - // Rows hashed before the first merge carry over into the union. - let mut union = SparkHllUnion::new(sketch.lg_config_k()); - union.merge(sketch); - *self = Self::Union(union); - } - let Self::Union(union) = self else { - unreachable!("the sketch was replaced by a union above") - }; + let union = self.union_mut(); for i in 0..arr.len() { if !arr.is_null(i) { union.merge(&SparkHllSketch::from_bytes(arr.value(i))?); diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql new file mode 100644 index 00000000000..130016409bf --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql @@ -0,0 +1,86 @@ +-- 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. + +-- MinSparkVersion: 4.0 +-- Config: spark.comet.expression.HllSketchAgg.allowIncompatible=true +-- Config: spark.comet.expression.HllSketchEstimate.allowIncompatible=true +-- Config: spark.sql.files.openCostInBytes=134217728 + +-- Until a group's sketch holds the register array, its estimate depends only on the coupons it +-- collected. Merging two or more partials that already hold it leaves the estimate to the +-- registers alone. Comet matches Spark exactly in both cases. + +statement +CREATE TABLE hll_types(g int, i int, l bigint, s string, b binary) USING parquet + +statement +INSERT INTO hll_types VALUES + (0, 1, 1, 'a', X'01'), + (0, 1, 1, 'a', X'01'), + (0, -2147483648, -9223372036854775808, '', X''), + (0, 2147483647, 9223372036854775807, 'héllo', X'00FF'), + (0, NULL, NULL, NULL, NULL), + (1, 0, 0, 'b', X'02'), + (1, NULL, NULL, NULL, NULL), + (2, NULL, NULL, NULL, NULL) + +-- NULLs, empty strings and empty binaries are skipped, so the all-NULL group estimates to 0. +query +SELECT g, + hll_sketch_estimate(hll_sketch_agg(i)), + hll_sketch_estimate(hll_sketch_agg(l)), + hll_sketch_estimate(hll_sketch_agg(s)), + hll_sketch_estimate(hll_sketch_agg(b)) +FROM hll_types GROUP BY g ORDER BY g + +-- Each INSERT coalesces to one task, so it writes one file, and an open cost as large as a split +-- keeps every file in its own partition. Each file then becomes one partial sketch per group, of +-- 10,000 distinct values, and the Final aggregation merges four of them. +statement +CREATE TABLE hll_partials(g int, v bigint) USING parquet + +statement +INSERT INTO hll_partials SELECT /*+ COALESCE(1) */ CAST(id % 2 AS INT), id FROM range(0, 20000) + +statement +INSERT INTO hll_partials SELECT /*+ COALESCE(1) */ CAST(id % 2 AS INT), id FROM range(20000, 40000) + +statement +INSERT INTO hll_partials SELECT /*+ COALESCE(1) */ CAST(id % 2 AS INT), id FROM range(40000, 60000) + +statement +INSERT INTO hll_partials SELECT /*+ COALESCE(1) */ CAST(id % 2 AS INT), id FROM range(60000, 80000) + +-- At lgConfigK 21 a sketch holds 196,608 coupons before the register array, so the partials and +-- the unions all stay coupon sets. +query +SELECT g, hll_sketch_estimate(hll_sketch_agg(v, 21)) FROM hll_partials GROUP BY g ORDER BY g + +-- At lgConfigK 12 every partial already holds the register array. +query +SELECT g, hll_sketch_estimate(hll_sketch_agg(v, 12)) FROM hll_partials GROUP BY g ORDER BY g + +query +SELECT hll_sketch_estimate(hll_sketch_agg(v, 12)) FROM hll_partials + +-- At lgConfigK 18 the partials are coupon sets, and a group's union moves to the register array +-- during the third merge, past 24,576 coupons. From there the estimate depends on the order the +-- remaining coupons arrive in, in Spark too, which gives different estimates for different orders +-- of the four partials. So the estimates are compared within 1% of the 40,000 distinct values. +query tolerance=400 +SELECT g, CAST(hll_sketch_estimate(hll_sketch_agg(v, 18)) AS DOUBLE) +FROM hll_partials GROUP BY g ORDER BY g From 4202bb42b6515e4178b2ecce84418924ab7f5a5e Mon Sep 17 00:00:00 2001 From: comphead Date: Fri, 30 Oct 2026 16:44:24 -0700 Subject: [PATCH 3/3] test: check the order-dependent hll_sketch_agg shape against the true count query tolerance only accepts an absolute tolerance up to 1e-6, so the lgConfigK 18 query failed with "absTol 400.0 is out of range". Where coupon-set partials take the union to the register array, Spark's own estimate changes with merge order, so check that each engine's estimate is within 1% of the 40,000 distinct values instead of comparing the two. --- .../sql-tests/expressions/aggregate/hll_sketch_agg.sql | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql b/spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql index 130016409bf..173f96af8f6 100644 --- a/spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql +++ b/spark/src/test/resources/sql-tests/expressions/aggregate/hll_sketch_agg.sql @@ -80,7 +80,8 @@ SELECT hll_sketch_estimate(hll_sketch_agg(v, 12)) FROM hll_partials -- At lgConfigK 18 the partials are coupon sets, and a group's union moves to the register array -- during the third merge, past 24,576 coupons. From there the estimate depends on the order the -- remaining coupons arrive in, in Spark too, which gives different estimates for different orders --- of the four partials. So the estimates are compared within 1% of the 40,000 distinct values. -query tolerance=400 -SELECT g, CAST(hll_sketch_estimate(hll_sketch_agg(v, 18)) AS DOUBLE) +-- of the four partials. So each engine's estimate is checked against the 40,000 distinct values, +-- within 1%, rather than against the other engine's. +query +SELECT g, abs(hll_sketch_estimate(hll_sketch_agg(v, 18)) - 40000) <= 400 FROM hll_partials GROUP BY g ORDER BY g