mbutrovich commented on code in PR #4819:
URL: https://github.com/apache/datafusion-comet/pull/4819#discussion_r3615516093
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -823,6 +823,57 @@ object CometCollectSet extends
CometAggregateExpressionSerde[CollectSet] {
}
}
+object CometApproxCountDistinct extends
CometAggregateExpressionSerde[HyperLogLogPlusPlus] {
+
+ override def supportsMixedPartialFinal: Boolean = false
+
+ // Types that Comet's native `xxhash64` hashes identically to Spark's
`XxHash64Function`.
+ private def hashableType(dt: DataType): Boolean = dt match {
Review Comment:
`hashableType` returns true for every `DecimalType`, but the native xxhash64
kernel only matches Spark for precision <= 18. Verified against Spark:
`XxHash64Function` uses the interpreted `HashExpression.hash` path, which for
`Decimal` does (`sql/catalyst/.../expressions/hash.scala:625-631`) `if
(precision <= Decimal.MAX_LONG_DIGITS) hashLong(d.toUnscaledLong, seed) else
hashUnsafeBytes(d.toJavaBigDecimal.unscaledValue().toByteArray, ...)`, where
`Decimal.MAX_LONG_DIGITS == 18`. So precision > 18 hashes the `BigInteger`
two's-complement big-endian bytes (variable length), whereas Comet's
large-decimal path hashes the 16-byte little-endian `i128`
(`native/spark-expr/src/hash_funcs/utils.rs:712`, `hash_array_decimal!` ->
`array.value(i).to_le_bytes()`). Different length, endianness, and encoding.
The shared `HashUtils` predicate used by `CometXxHash64` already carves this
out (`spark/src/main/scala/org/apache/comet/serde/hash.scala:135`, "DecimalType
with precision > 18 is not suppo
rted"). Restrict the arm to `case d: DecimalType if d.precision <= 18 => true`
so wider decimals fall back. Note that fully delegating to
`HashUtils.supportLevelForChildren` is not appropriate here, since it permits
struct/array/map, and `normalize_floats` only canonicalizes top-level float
columns, not nested ones.
Required test for this gap (add to `approx_count_distinct.sql`). First a
query that exposes the current behavior: on the unpatched PR it runs natively
and the count can diverge from Spark, so this comparison should fail today and
is what proves the gap.
```sql
statement
CREATE TABLE acd_wide_dec(d decimal(38,4)) USING parquet
statement
INSERT INTO acd_wide_dec
SELECT CAST(id AS decimal(38,4)) + 0.0001 FROM range(20000)
-- decimal precision > 18: Spark hashes via BigDecimal, Comet's native path
does not match
query
SELECT approx_count_distinct(d) FROM acd_wide_dec
```
After the `precision <= 18` fix, this should instead assert the fallback so
Comet never executes the mismatched native hash:
```sql
query expect_fallback(Unsupported input data type)
SELECT approx_count_distinct(d) FROM acd_wide_dec
```
Keep a supported-decimal case too (for example `decimal(18,4)`) to confirm
the boundary still runs natively and matches.
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ self.signature.hash(state);
+ self.p.hash(state);
+ }
+}
+
+impl HllPlusPlus {
+ pub fn new(p: i32) -> Self {
+ assert!(p >= 4, "HLL++ requires at least 4 bits of precision");
Review Comment:
Keeping this as a hard `assert!(p >= 4, ...)` is good: it enforces the HLL++
minimum-precision invariant explicitly and is self-documenting. Consider naming
the source of truth in the message so the invariant is traceable, e.g. "Spark's
HyperLogLogPlusPlusHelper requires p >= 4 (relativeSD <= 39%)".
##########
native/spark-expr/src/agg_funcs/hll_plus_plus_const.rs:
##########
@@ -0,0 +1,3993 @@
+// Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
I checked DataFusion and arrow-rs directly. The vendoring is necessary for a
bit-identical port, with two caveats worth raising.
- DataFusion's HLL cannot be reused.
`datafusion/functions-aggregate/src/hyperloglog.rs` implements the 2017 Ertl
variant: a closed-form `hll_sigma`/`hll_tau` estimator with no lookup tables
(`hyperloglog.rs:196-233`), fixed precision `HLL_P = 14` (`:42`), one `u8` per
register (`:54`), and foldhash seed 0 (`common/src/hash_utils.rs`,
`HLL_RANDOM_STATE = with_seed(0)`). Spark uses the 2013 Google HLL++ variant:
xxhash64 seed 42, `p` derived from `relativeSD` (default 9), 6-bit registers
packed 10 per long, and empirical bias-correction plus linear-counting tables.
Every compatibility-critical axis differs (hash, precision, leading-vs-trailing
zeros, register storage, estimator), so DataFusion's HLL cannot be made
bit-identical. arrow-rs has no HLL at all.
- The tables are Spark's, verbatim, and exist nowhere reusable.
`RAW_ESTIMATE_DATA`, `BIAS_DATA`, and `THRESHOLDS` are transcribed from
`HyperLogLogPlusPlusHelper.scala:310-383` (about 5733 doubles). They are the
2013 paper's appendix data and are not present in DataFusion or arrow-rs.
Bit-identical output requires Spark's exact values (the transcription even
preserves Spark's float-formatting artifacts), so this is the correct choice.
Two requests on this:
1. The 3993 lines is roughly 2x formatting bloat. `BIAS_DATA` is emitted one
value per line while `RAW_ESTIMATE_DATA` is per-row. Add `#[rustfmt::skip]` to
`BIAS_DATA` and use the compact per-row layout, which roughly halves the file
with no loss of side-by-side auditability against Spark. Do not compress the
data itself (delta-encoding or a binary blob), since verbatim comparability
against Spark is the point.
2. Altitude, for later not this PR: a Spark-compatible
`approx_count_distinct` is a natural fit for the `datafusion-spark` crate
(`datafusion/datafusion/spark/src/function/aggregate/`, which already carries
`avg`, `try_sum`, `collect`), so the algorithm and tables could eventually be
shared rather than Comet-only. Landing in Comet first is the normal path, so
this is not a blocker. Worth a tracking issue for the upstreaming, with the
note that the one coupling to unwind is the xxhash64 dependency (the PR reuses
Comet's `create_xxhash64_hashes`).
Minor reuse note inside `hll_plus_plus.rs`: `emit_words` hand-rolls
`EmitTo::All`/`First(n)` slicing, where DataFusion exposes
`EmitTo::take_needed` (used by `approx_distinct.rs`). It is not a drop-in given
the flat `Vec<i64>` chunk layout, so this is optional.
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
Review Comment:
The manual `impl std::hash::Hash for HllPlusPlus` just hashes `name`,
`signature`, and `p`, which is exactly what `#[derive(Hash)]` produces
(`Signature`, `String`, and `usize` are all `Hash`). This is pure structure
with no ordering/float sensitivity, so the idiomatic form is to add `Hash` to
the derive list and delete the impl.
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ self.signature.hash(state);
+ self.p.hash(state);
+ }
+}
+
+impl HllPlusPlus {
+ pub fn new(p: i32) -> Self {
+ assert!(p >= 4, "HLL++ requires at least 4 bits of precision");
+ Self {
+ name: "approx_count_distinct".to_string(),
+ signature: Signature::any(1, Volatility::Immutable),
+ p: p as usize,
+ }
+ }
+}
+
+impl AggregateUDFImpl for HllPlusPlus {
+ fn name(&self) -> &str {
+ &self.name
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Int64)
+ }
+
+ fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn
Accumulator>> {
+ Ok(Box::new(HllPlusPlusAccumulator::new(self.p)))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // The partial state is the packed register buffer: one `Long` per
word, exactly matching
+ // the `MS[i]` buffer attributes of Spark's `HyperLogLogPlusPlus`.
+ Ok((0..num_words(self.p))
+ .map(|i| Arc::new(Field::new(format!("MS[{i}]"), DataType::Int64,
false)) as FieldRef)
+ .collect())
+ }
+
+ fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+ true
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ _args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(HllPlusPlusGroupsAccumulator::new(self.p)))
+ }
+}
+
+/// Normalize a float/double column the way Spark's `NormalizeNaNAndZero` does
before hashing:
+/// every NaN becomes the canonical NaN and `-0.0` becomes `0.0`. Returns the
input unchanged for
+/// non-floating-point types.
+fn normalize_floats(array: &ArrayRef) -> ArrayRef {
+ match array.data_type() {
+ DataType::Float32 => {
+ let normalized: Float32Array =
array.as_primitive::<Float32Type>().unary(|v| {
+ if v.is_nan() {
+ f32::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ DataType::Float64 => {
+ let normalized: Float64Array =
array.as_primitive::<Float64Type>().unary(|v| {
+ if v.is_nan() {
+ f64::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ _ => Arc::clone(array),
+ }
+}
+
+/// Hash a value column with Spark's `XxHash64` (seed 42), normalizing floats
first.
+fn hash_values(array: &ArrayRef) -> Result<Vec<u64>> {
+ let normalized = normalize_floats(array);
+ let mut hashes = vec![HASH_SEED; normalized.len()];
+ create_xxhash64_hashes(&[normalized], &mut hashes)?;
+ Ok(hashes)
+}
+
+/// Update the packed register buffer from a hashed value. Mirrors
`HyperLogLogPlusPlusHelper.update`.
+#[inline]
+fn update_word(words: &mut [i64], p: usize, hash: u64) {
+ let idx = (hash >> (64 - p)) as usize;
+ let w_padding = 1u64 << (p - 1);
+ let pw = (((hash << p) | w_padding).leading_zeros() + 1) as u64;
+
+ let word_offset = idx / REGISTERS_PER_WORD;
+ let shift = REGISTER_SIZE * (idx - word_offset * REGISTERS_PER_WORD) as
u64;
+ let word = words[word_offset] as u64;
+ let mask = REGISTER_WORD_MASK << shift;
+ let m_idx = (word & mask) >> shift;
+ if pw > m_idx {
+ words[word_offset] = ((word & !mask) | (pw << shift)) as i64;
+ }
+}
+
+/// Merge `src` packed registers into `dst` by taking the per-register
maximum. Mirrors
+/// `HyperLogLogPlusPlusHelper.merge`.
+#[inline]
+fn merge_words(dst: &mut [i64], src: &[i64], m: usize) {
+ let mut idx = 0;
+ for (d, s) in dst.iter_mut().zip(src.iter()) {
+ let word1 = *d as u64;
+ let word2 = *s as u64;
+ let mut word = 0u64;
+ let mut i = 0;
+ let mut mask = REGISTER_WORD_MASK;
+ while idx < m && i < REGISTERS_PER_WORD {
+ word |= (word1 & mask).max(word2 & mask);
+ mask <<= REGISTER_SIZE;
+ i += 1;
+ idx += 1;
+ }
+ *d = word as i64;
+ }
+}
+
+/// The precomputed `alpha * m * m` constant, matching
`HyperLogLogPlusPlusHelper.alphaM2`.
+fn alpha_m2(p: usize, m: f64) -> f64 {
+ match p {
+ 4 => 0.673 * m * m,
+ 5 => 0.697 * m * m,
+ 6 => 0.709 * m * m,
+ _ => (0.7213 / (1.0 + 1.079 / m)) * m * m,
+ }
+}
+
+/// Estimate the bias using KNN interpolation over Spark's appendix tables.
+fn estimate_bias(p: usize, e: f64) -> f64 {
+ let estimates = RAW_ESTIMATE_DATA[p - 4];
+ let biases = BIAS_DATA[p - 4];
+ let num_estimates = estimates.len();
+
+ // Index of the interpolation estimate closest to `e` (matches Java's
Arrays.binarySearch).
+ let nearest = match estimates.binary_search_by(|v|
v.partial_cmp(&e).unwrap()) {
+ Ok(ix) => ix,
+ Err(ix) => ix,
+ };
+
+ let distance = |i: usize| {
+ let diff = e - estimates[i];
+ diff * diff
+ };
+
+ let mut low = nearest.saturating_sub(K - 1);
+ let mut high = (low + K).min(num_estimates);
+ while high < num_estimates && distance(high) < distance(low) {
+ low += 1;
+ high += 1;
+ }
+
+ let bias_sum: f64 = biases[low..high].iter().sum();
+ bias_sum / (high - low) as f64
+}
+
+/// Compute the HyperLogLog++ cardinality estimate from a packed register
buffer. Faithful port of
+/// `HyperLogLogPlusPlusHelper.query`.
+fn query(p: usize, words: &[i64]) -> i64 {
+ let m_usize = 1usize << p;
+ let m = m_usize as f64;
+
+ let mut z_inverse = 0.0;
+ let mut v = 0.0;
+ let mut idx = 0;
+ for &word in words {
+ let word = word as u64;
+ let mut i = 0;
+ let mut shift = 0;
+ while idx < m_usize && i < REGISTERS_PER_WORD {
Review Comment:
The explicit `while idx < m && i < REGISTERS_PER_WORD` loop here is correct
to keep literal rather than refactoring into iterator chains: `z_inverse += 1.0
/ (1 << m_idx)` is a floating-point summation whose order must match Spark for
the result to stay bit-identical. Worth a one-line comment saying exactly that
("register iteration order matches Spark so the float summation is
bit-identical"), so a future reader does not "idiomatize" it and silently
change the last ULP.
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -823,6 +823,57 @@ object CometCollectSet extends
CometAggregateExpressionSerde[CollectSet] {
}
}
+object CometApproxCountDistinct extends
CometAggregateExpressionSerde[HyperLogLogPlusPlus] {
+
+ override def supportsMixedPartialFinal: Boolean = false
Review Comment:
`false` is safe, but it may contradict the PR's own claim that the register
buffer is bit-identical to Spark's `aggBufferSchema`. If that claim holds, the
buffer is byte-compatible and `true` would be justified, the same reasoning
`CometBloomFilterAggregate` uses to set it `true`. Either set it `true` (and
gain mixed-engine plans) or soften the "partial state matches Spark's
aggBufferSchema" wording in the audit doc. Which is intended?
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ self.signature.hash(state);
+ self.p.hash(state);
+ }
+}
+
+impl HllPlusPlus {
+ pub fn new(p: i32) -> Self {
+ assert!(p >= 4, "HLL++ requires at least 4 bits of precision");
+ Self {
+ name: "approx_count_distinct".to_string(),
+ signature: Signature::any(1, Volatility::Immutable),
+ p: p as usize,
+ }
+ }
+}
+
+impl AggregateUDFImpl for HllPlusPlus {
+ fn name(&self) -> &str {
+ &self.name
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Int64)
+ }
+
+ fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn
Accumulator>> {
+ Ok(Box::new(HllPlusPlusAccumulator::new(self.p)))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // The partial state is the packed register buffer: one `Long` per
word, exactly matching
+ // the `MS[i]` buffer attributes of Spark's `HyperLogLogPlusPlus`.
+ Ok((0..num_words(self.p))
+ .map(|i| Arc::new(Field::new(format!("MS[{i}]"), DataType::Int64,
false)) as FieldRef)
+ .collect())
+ }
+
+ fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+ true
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ _args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(HllPlusPlusGroupsAccumulator::new(self.p)))
+ }
+}
+
+/// Normalize a float/double column the way Spark's `NormalizeNaNAndZero` does
before hashing:
+/// every NaN becomes the canonical NaN and `-0.0` becomes `0.0`. Returns the
input unchanged for
+/// non-floating-point types.
+fn normalize_floats(array: &ArrayRef) -> ArrayRef {
+ match array.data_type() {
+ DataType::Float32 => {
+ let normalized: Float32Array =
array.as_primitive::<Float32Type>().unary(|v| {
+ if v.is_nan() {
+ f32::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ DataType::Float64 => {
+ let normalized: Float64Array =
array.as_primitive::<Float64Type>().unary(|v| {
+ if v.is_nan() {
+ f64::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ _ => Arc::clone(array),
+ }
+}
+
+/// Hash a value column with Spark's `XxHash64` (seed 42), normalizing floats
first.
+fn hash_values(array: &ArrayRef) -> Result<Vec<u64>> {
+ let normalized = normalize_floats(array);
+ let mut hashes = vec![HASH_SEED; normalized.len()];
+ create_xxhash64_hashes(&[normalized], &mut hashes)?;
+ Ok(hashes)
+}
+
+/// Update the packed register buffer from a hashed value. Mirrors
`HyperLogLogPlusPlusHelper.update`.
+#[inline]
+fn update_word(words: &mut [i64], p: usize, hash: u64) {
+ let idx = (hash >> (64 - p)) as usize;
+ let w_padding = 1u64 << (p - 1);
+ let pw = (((hash << p) | w_padding).leading_zeros() + 1) as u64;
+
+ let word_offset = idx / REGISTERS_PER_WORD;
+ let shift = REGISTER_SIZE * (idx - word_offset * REGISTERS_PER_WORD) as
u64;
+ let word = words[word_offset] as u64;
+ let mask = REGISTER_WORD_MASK << shift;
+ let m_idx = (word & mask) >> shift;
+ if pw > m_idx {
+ words[word_offset] = ((word & !mask) | (pw << shift)) as i64;
+ }
+}
+
+/// Merge `src` packed registers into `dst` by taking the per-register
maximum. Mirrors
+/// `HyperLogLogPlusPlusHelper.merge`.
+#[inline]
+fn merge_words(dst: &mut [i64], src: &[i64], m: usize) {
+ let mut idx = 0;
+ for (d, s) in dst.iter_mut().zip(src.iter()) {
+ let word1 = *d as u64;
+ let word2 = *s as u64;
+ let mut word = 0u64;
+ let mut i = 0;
+ let mut mask = REGISTER_WORD_MASK;
+ while idx < m && i < REGISTERS_PER_WORD {
+ word |= (word1 & mask).max(word2 & mask);
+ mask <<= REGISTER_SIZE;
+ i += 1;
+ idx += 1;
+ }
+ *d = word as i64;
+ }
+}
+
+/// The precomputed `alpha * m * m` constant, matching
`HyperLogLogPlusPlusHelper.alphaM2`.
+fn alpha_m2(p: usize, m: f64) -> f64 {
+ match p {
+ 4 => 0.673 * m * m,
+ 5 => 0.697 * m * m,
+ 6 => 0.709 * m * m,
+ _ => (0.7213 / (1.0 + 1.079 / m)) * m * m,
+ }
+}
+
+/// Estimate the bias using KNN interpolation over Spark's appendix tables.
+fn estimate_bias(p: usize, e: f64) -> f64 {
Review Comment:
`estimate_bias` indexes `RAW_ESTIMATE_DATA[p - 4]` / `BIAS_DATA[p - 4]`,
which is only in bounds for `4 <= p <= 18`. That holds today because the
constructor enforces `p >= 4` and every caller guards with `p < 19`, but
nothing states it here. To make the invariant explicit and self-documenting,
add an assertion at the top of this function such as
`debug_assert!((4..19).contains(&p));` (or `assert!`), so the table-bounds
contract lives at the indexing site.
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ self.signature.hash(state);
+ self.p.hash(state);
+ }
+}
+
+impl HllPlusPlus {
+ pub fn new(p: i32) -> Self {
+ assert!(p >= 4, "HLL++ requires at least 4 bits of precision");
+ Self {
+ name: "approx_count_distinct".to_string(),
+ signature: Signature::any(1, Volatility::Immutable),
+ p: p as usize,
+ }
+ }
+}
+
+impl AggregateUDFImpl for HllPlusPlus {
+ fn name(&self) -> &str {
+ &self.name
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Int64)
+ }
+
+ fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn
Accumulator>> {
+ Ok(Box::new(HllPlusPlusAccumulator::new(self.p)))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // The partial state is the packed register buffer: one `Long` per
word, exactly matching
+ // the `MS[i]` buffer attributes of Spark's `HyperLogLogPlusPlus`.
+ Ok((0..num_words(self.p))
+ .map(|i| Arc::new(Field::new(format!("MS[{i}]"), DataType::Int64,
false)) as FieldRef)
+ .collect())
+ }
+
+ fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+ true
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ _args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(HllPlusPlusGroupsAccumulator::new(self.p)))
+ }
+}
+
+/// Normalize a float/double column the way Spark's `NormalizeNaNAndZero` does
before hashing:
+/// every NaN becomes the canonical NaN and `-0.0` becomes `0.0`. Returns the
input unchanged for
+/// non-floating-point types.
+fn normalize_floats(array: &ArrayRef) -> ArrayRef {
Review Comment:
`normalize_floats` reimplements the existing generic `normalize_float<T:
num::Float>` in
`native/spark-expr/src/math_funcs/internal/normalize_nan.rs:110`, which already
canonicalizes NaN and maps `-0.0 -> 0.0` and is the helper
`NormalizeNaNAndZero` uses. Make that fn `pub(crate)` and call it from both
float arms here, which collapses the copy-paste and keeps one source of truth
for Spark's `NormalizeNaNAndZero` semantics. This is pure per-element mapping
with no float-order dependence, so it is behavior-preserving.
`num::Float::nan()` is the same canonical `0x7fc00000` / `0x7ff8...` pattern as
`f32::NAN` / `f64::NAN`, so bit-identity holds.
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ self.signature.hash(state);
+ self.p.hash(state);
+ }
+}
+
+impl HllPlusPlus {
+ pub fn new(p: i32) -> Self {
+ assert!(p >= 4, "HLL++ requires at least 4 bits of precision");
+ Self {
+ name: "approx_count_distinct".to_string(),
+ signature: Signature::any(1, Volatility::Immutable),
+ p: p as usize,
+ }
+ }
+}
+
+impl AggregateUDFImpl for HllPlusPlus {
+ fn name(&self) -> &str {
+ &self.name
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Int64)
+ }
+
+ fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn
Accumulator>> {
+ Ok(Box::new(HllPlusPlusAccumulator::new(self.p)))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // The partial state is the packed register buffer: one `Long` per
word, exactly matching
+ // the `MS[i]` buffer attributes of Spark's `HyperLogLogPlusPlus`.
+ Ok((0..num_words(self.p))
+ .map(|i| Arc::new(Field::new(format!("MS[{i}]"), DataType::Int64,
false)) as FieldRef)
+ .collect())
+ }
+
+ fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+ true
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ _args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(HllPlusPlusGroupsAccumulator::new(self.p)))
+ }
+}
+
+/// Normalize a float/double column the way Spark's `NormalizeNaNAndZero` does
before hashing:
+/// every NaN becomes the canonical NaN and `-0.0` becomes `0.0`. Returns the
input unchanged for
+/// non-floating-point types.
+fn normalize_floats(array: &ArrayRef) -> ArrayRef {
+ match array.data_type() {
+ DataType::Float32 => {
+ let normalized: Float32Array =
array.as_primitive::<Float32Type>().unary(|v| {
+ if v.is_nan() {
+ f32::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ DataType::Float64 => {
+ let normalized: Float64Array =
array.as_primitive::<Float64Type>().unary(|v| {
+ if v.is_nan() {
+ f64::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ _ => Arc::clone(array),
+ }
+}
+
+/// Hash a value column with Spark's `XxHash64` (seed 42), normalizing floats
first.
+fn hash_values(array: &ArrayRef) -> Result<Vec<u64>> {
Review Comment:
`hash_values` allocates a fresh `Vec<u64>` every batch, then fills,
consumes, and drops it. `create_xxhash64_hashes` takes a caller-owned `&mut
[u64]` and does not allocate, so this can be a reusable scratch buffer on each
accumulator: per batch do `buf.clear(); buf.resize(len, HASH_SEED)` (reuses
capacity) and pass `&mut buf`. Turns a per-batch malloc/free into an
amortized-zero allocation. The buffer must be re-seeded to `HASH_SEED` each
batch, not just zeroed, since the hash folds into the existing values. This is
the one item aligned with the "no wasted allocations" goal, though it does not
change the parity benchmark since hashing dominates.
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -823,6 +823,57 @@ object CometCollectSet extends
CometAggregateExpressionSerde[CollectSet] {
}
}
+object CometApproxCountDistinct extends
CometAggregateExpressionSerde[HyperLogLogPlusPlus] {
Review Comment:
`CometApproxCountDistinct` does not override
`getUnsupportedReasons()`/`getIncompatibleReasons()`, so they default to empty.
Runtime fallback still works because `getSupportLevel` returns
`Unsupported(Some(reason))`, but the generated Compatibility Guide will show no
caveats for this expression, unlike siblings such as `CometAverage` and
`CometXxHash64` (via `HashUtils.unsupportedReasons`). When the decimal fix
lands, add `getUnsupportedReasons()` entries (decimal precision > 18, collated
strings, unsupported input types) so the docs reflect the real limits.
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ self.signature.hash(state);
+ self.p.hash(state);
+ }
+}
+
+impl HllPlusPlus {
+ pub fn new(p: i32) -> Self {
+ assert!(p >= 4, "HLL++ requires at least 4 bits of precision");
+ Self {
+ name: "approx_count_distinct".to_string(),
+ signature: Signature::any(1, Volatility::Immutable),
+ p: p as usize,
+ }
+ }
+}
+
+impl AggregateUDFImpl for HllPlusPlus {
+ fn name(&self) -> &str {
+ &self.name
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Int64)
+ }
+
+ fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn
Accumulator>> {
+ Ok(Box::new(HllPlusPlusAccumulator::new(self.p)))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // The partial state is the packed register buffer: one `Long` per
word, exactly matching
+ // the `MS[i]` buffer attributes of Spark's `HyperLogLogPlusPlus`.
+ Ok((0..num_words(self.p))
+ .map(|i| Arc::new(Field::new(format!("MS[{i}]"), DataType::Int64,
false)) as FieldRef)
+ .collect())
+ }
+
+ fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+ true
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ _args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(HllPlusPlusGroupsAccumulator::new(self.p)))
+ }
+}
+
+/// Normalize a float/double column the way Spark's `NormalizeNaNAndZero` does
before hashing:
+/// every NaN becomes the canonical NaN and `-0.0` becomes `0.0`. Returns the
input unchanged for
+/// non-floating-point types.
+fn normalize_floats(array: &ArrayRef) -> ArrayRef {
+ match array.data_type() {
+ DataType::Float32 => {
+ let normalized: Float32Array =
array.as_primitive::<Float32Type>().unary(|v| {
+ if v.is_nan() {
+ f32::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ DataType::Float64 => {
+ let normalized: Float64Array =
array.as_primitive::<Float64Type>().unary(|v| {
+ if v.is_nan() {
+ f64::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ _ => Arc::clone(array),
+ }
+}
+
+/// Hash a value column with Spark's `XxHash64` (seed 42), normalizing floats
first.
+fn hash_values(array: &ArrayRef) -> Result<Vec<u64>> {
+ let normalized = normalize_floats(array);
+ let mut hashes = vec![HASH_SEED; normalized.len()];
+ create_xxhash64_hashes(&[normalized], &mut hashes)?;
+ Ok(hashes)
+}
+
+/// Update the packed register buffer from a hashed value. Mirrors
`HyperLogLogPlusPlusHelper.update`.
+#[inline]
+fn update_word(words: &mut [i64], p: usize, hash: u64) {
+ let idx = (hash >> (64 - p)) as usize;
+ let w_padding = 1u64 << (p - 1);
+ let pw = (((hash << p) | w_padding).leading_zeros() + 1) as u64;
+
+ let word_offset = idx / REGISTERS_PER_WORD;
+ let shift = REGISTER_SIZE * (idx - word_offset * REGISTERS_PER_WORD) as
u64;
+ let word = words[word_offset] as u64;
+ let mask = REGISTER_WORD_MASK << shift;
+ let m_idx = (word & mask) >> shift;
+ if pw > m_idx {
+ words[word_offset] = ((word & !mask) | (pw << shift)) as i64;
+ }
+}
+
+/// Merge `src` packed registers into `dst` by taking the per-register
maximum. Mirrors
+/// `HyperLogLogPlusPlusHelper.merge`.
+#[inline]
+fn merge_words(dst: &mut [i64], src: &[i64], m: usize) {
+ let mut idx = 0;
+ for (d, s) in dst.iter_mut().zip(src.iter()) {
+ let word1 = *d as u64;
+ let word2 = *s as u64;
+ let mut word = 0u64;
+ let mut i = 0;
+ let mut mask = REGISTER_WORD_MASK;
+ while idx < m && i < REGISTERS_PER_WORD {
+ word |= (word1 & mask).max(word2 & mask);
+ mask <<= REGISTER_SIZE;
+ i += 1;
+ idx += 1;
+ }
+ *d = word as i64;
+ }
+}
+
+/// The precomputed `alpha * m * m` constant, matching
`HyperLogLogPlusPlusHelper.alphaM2`.
+fn alpha_m2(p: usize, m: f64) -> f64 {
+ match p {
+ 4 => 0.673 * m * m,
+ 5 => 0.697 * m * m,
+ 6 => 0.709 * m * m,
+ _ => (0.7213 / (1.0 + 1.079 / m)) * m * m,
+ }
+}
+
+/// Estimate the bias using KNN interpolation over Spark's appendix tables.
+fn estimate_bias(p: usize, e: f64) -> f64 {
+ let estimates = RAW_ESTIMATE_DATA[p - 4];
+ let biases = BIAS_DATA[p - 4];
+ let num_estimates = estimates.len();
+
+ // Index of the interpolation estimate closest to `e` (matches Java's
Arrays.binarySearch).
+ let nearest = match estimates.binary_search_by(|v|
v.partial_cmp(&e).unwrap()) {
+ Ok(ix) => ix,
+ Err(ix) => ix,
+ };
+
+ let distance = |i: usize| {
+ let diff = e - estimates[i];
+ diff * diff
+ };
+
+ let mut low = nearest.saturating_sub(K - 1);
+ let mut high = (low + K).min(num_estimates);
+ while high < num_estimates && distance(high) < distance(low) {
+ low += 1;
+ high += 1;
+ }
+
+ let bias_sum: f64 = biases[low..high].iter().sum();
+ bias_sum / (high - low) as f64
+}
+
+/// Compute the HyperLogLog++ cardinality estimate from a packed register
buffer. Faithful port of
+/// `HyperLogLogPlusPlusHelper.query`.
+fn query(p: usize, words: &[i64]) -> i64 {
+ let m_usize = 1usize << p;
+ let m = m_usize as f64;
+
+ let mut z_inverse = 0.0;
+ let mut v = 0.0;
+ let mut idx = 0;
+ for &word in words {
+ let word = word as u64;
+ let mut i = 0;
+ let mut shift = 0;
+ while idx < m_usize && i < REGISTERS_PER_WORD {
+ let m_idx = (word >> shift) & REGISTER_WORD_MASK;
+ z_inverse += 1.0 / ((1u64 << m_idx) as f64);
+ if m_idx == 0 {
+ v += 1.0;
+ }
+ shift += REGISTER_SIZE;
+ i += 1;
+ idx += 1;
+ }
+ }
+
+ let e = alpha_m2(p, m) / z_inverse;
+ let e_bias_corrected = if p < 19 && e < 5.0 * m {
Review Comment:
`e_bias_corrected` is computed eagerly, so `estimate_bias` (a binary search
plus KNN interpolation over the appendix tables) runs even when the
linear-counting branch is taken and its result is discarded. `query` runs once
per group, so small-cardinality groups all pay for dead work. Make it lazy:
compute `e - estimate_bias(p, e)` only inside the two branches that return it.
This returns the same value in every case, so it does not affect bit-identity,
it only skips the interpolation when linear counting wins.
##########
native/spark-expr/src/agg_funcs/hll_plus_plus.rs:
##########
@@ -0,0 +1,649 @@
+// 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.
+
+//! Spark-compatible `approx_count_distinct`, a faithful port of Spark's
+//! `HyperLogLogPlusPlus` / `HyperLogLogPlusPlusHelper`. Values are hashed
with Spark's
+//! `XxHash64` (seed 42, floats normalized first) and the registers are stored
using the exact
+//! same packed-`Long` buffer layout Spark uses (10 six-bit registers per
64-bit word). Keeping
+//! the wire format identical means the partial-aggregation state matches
Spark's
+//! `aggBufferSchema`, and the cardinality estimate uses the same
bias-correction tables, so
+//! results are bit-identical to Spark.
+
+use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA,
THRESHOLDS};
+use crate::hash_funcs::create_xxhash64_hashes;
+use arrow::array::{
+ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array,
Int64Array,
+};
+use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type};
+use datafusion::common::{Result, ScalarValue};
+use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion::logical_expr::{
+ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
Volatility,
+};
+use std::sync::Arc;
+
+/// Number of interpolation points used when estimating the bias (Spark's `K`).
+const K: usize = 6;
+/// The seed Spark's `HyperLogLogPlusPlus` uses for `XxHash64`.
+const HASH_SEED: u64 = 42;
+/// Bits per register (Spark's `REGISTER_SIZE`). A 64-bit hash yields at most
64 leading zeros.
+const REGISTER_SIZE: u64 = 6;
+/// Registers packed per 64-bit word (Spark's `REGISTERS_PER_WORD`); only 60
of 64 bits are used.
+const REGISTERS_PER_WORD: usize = 10;
+/// Mask for a single register within a word (Spark's `REGISTER_WORD_MASK`).
+const REGISTER_WORD_MASK: u64 = (1 << REGISTER_SIZE) - 1;
+
+/// Number of 64-bit words needed to store all `m = 1 << p` registers,
matching Spark's `numWords`.
+fn num_words(p: usize) -> usize {
+ (1usize << p) / REGISTERS_PER_WORD + 1
+}
+
+/// Compute the precision `p` for a given `relativeSD`, matching Spark's
+/// `HyperLogLogPlusPlusHelper`. Kept here so the native tests can exercise
it, but the
+/// serde computes the same value and passes `p` through the protobuf.
+pub fn hllpp_precision(relative_sd: f64) -> i32 {
+ (2.0 * (1.106 / relative_sd).ln() / 2.0_f64.ln()).ceil() as i32
+}
+
+/// Spark-compatible `approx_count_distinct` aggregate.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct HllPlusPlus {
+ name: String,
+ signature: Signature,
+ /// Number of addressing bits (precision). Determines the register count
`m = 1 << p`.
+ p: usize,
+}
+
+impl std::hash::Hash for HllPlusPlus {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.name.hash(state);
+ self.signature.hash(state);
+ self.p.hash(state);
+ }
+}
+
+impl HllPlusPlus {
+ pub fn new(p: i32) -> Self {
+ assert!(p >= 4, "HLL++ requires at least 4 bits of precision");
+ Self {
+ name: "approx_count_distinct".to_string(),
+ signature: Signature::any(1, Volatility::Immutable),
+ p: p as usize,
+ }
+ }
+}
+
+impl AggregateUDFImpl for HllPlusPlus {
+ fn name(&self) -> &str {
+ &self.name
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::Int64)
+ }
+
+ fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn
Accumulator>> {
+ Ok(Box::new(HllPlusPlusAccumulator::new(self.p)))
+ }
+
+ fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
+ // The partial state is the packed register buffer: one `Long` per
word, exactly matching
+ // the `MS[i]` buffer attributes of Spark's `HyperLogLogPlusPlus`.
+ Ok((0..num_words(self.p))
+ .map(|i| Arc::new(Field::new(format!("MS[{i}]"), DataType::Int64,
false)) as FieldRef)
+ .collect())
+ }
+
+ fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+ true
+ }
+
+ fn create_groups_accumulator(
+ &self,
+ _args: AccumulatorArgs,
+ ) -> Result<Box<dyn GroupsAccumulator>> {
+ Ok(Box::new(HllPlusPlusGroupsAccumulator::new(self.p)))
+ }
+}
+
+/// Normalize a float/double column the way Spark's `NormalizeNaNAndZero` does
before hashing:
+/// every NaN becomes the canonical NaN and `-0.0` becomes `0.0`. Returns the
input unchanged for
+/// non-floating-point types.
+fn normalize_floats(array: &ArrayRef) -> ArrayRef {
+ match array.data_type() {
+ DataType::Float32 => {
+ let normalized: Float32Array =
array.as_primitive::<Float32Type>().unary(|v| {
+ if v.is_nan() {
+ f32::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ DataType::Float64 => {
+ let normalized: Float64Array =
array.as_primitive::<Float64Type>().unary(|v| {
+ if v.is_nan() {
+ f64::NAN
+ } else {
+ v + 0.0
+ }
+ });
+ Arc::new(normalized)
+ }
+ _ => Arc::clone(array),
+ }
+}
+
+/// Hash a value column with Spark's `XxHash64` (seed 42), normalizing floats
first.
+fn hash_values(array: &ArrayRef) -> Result<Vec<u64>> {
+ let normalized = normalize_floats(array);
+ let mut hashes = vec![HASH_SEED; normalized.len()];
+ create_xxhash64_hashes(&[normalized], &mut hashes)?;
+ Ok(hashes)
+}
+
+/// Update the packed register buffer from a hashed value. Mirrors
`HyperLogLogPlusPlusHelper.update`.
+#[inline]
+fn update_word(words: &mut [i64], p: usize, hash: u64) {
+ let idx = (hash >> (64 - p)) as usize;
+ let w_padding = 1u64 << (p - 1);
+ let pw = (((hash << p) | w_padding).leading_zeros() + 1) as u64;
+
+ let word_offset = idx / REGISTERS_PER_WORD;
+ let shift = REGISTER_SIZE * (idx - word_offset * REGISTERS_PER_WORD) as
u64;
+ let word = words[word_offset] as u64;
+ let mask = REGISTER_WORD_MASK << shift;
+ let m_idx = (word & mask) >> shift;
+ if pw > m_idx {
+ words[word_offset] = ((word & !mask) | (pw << shift)) as i64;
+ }
+}
+
+/// Merge `src` packed registers into `dst` by taking the per-register
maximum. Mirrors
+/// `HyperLogLogPlusPlusHelper.merge`.
+#[inline]
+fn merge_words(dst: &mut [i64], src: &[i64], m: usize) {
+ let mut idx = 0;
+ for (d, s) in dst.iter_mut().zip(src.iter()) {
+ let word1 = *d as u64;
+ let word2 = *s as u64;
+ let mut word = 0u64;
+ let mut i = 0;
+ let mut mask = REGISTER_WORD_MASK;
+ while idx < m && i < REGISTERS_PER_WORD {
+ word |= (word1 & mask).max(word2 & mask);
+ mask <<= REGISTER_SIZE;
+ i += 1;
+ idx += 1;
+ }
+ *d = word as i64;
+ }
+}
+
+/// The precomputed `alpha * m * m` constant, matching
`HyperLogLogPlusPlusHelper.alphaM2`.
+fn alpha_m2(p: usize, m: f64) -> f64 {
+ match p {
+ 4 => 0.673 * m * m,
+ 5 => 0.697 * m * m,
+ 6 => 0.709 * m * m,
+ _ => (0.7213 / (1.0 + 1.079 / m)) * m * m,
+ }
+}
+
+/// Estimate the bias using KNN interpolation over Spark's appendix tables.
+fn estimate_bias(p: usize, e: f64) -> f64 {
+ let estimates = RAW_ESTIMATE_DATA[p - 4];
+ let biases = BIAS_DATA[p - 4];
+ let num_estimates = estimates.len();
+
+ // Index of the interpolation estimate closest to `e` (matches Java's
Arrays.binarySearch).
+ let nearest = match estimates.binary_search_by(|v|
v.partial_cmp(&e).unwrap()) {
+ Ok(ix) => ix,
+ Err(ix) => ix,
+ };
+
+ let distance = |i: usize| {
+ let diff = e - estimates[i];
+ diff * diff
+ };
+
+ let mut low = nearest.saturating_sub(K - 1);
+ let mut high = (low + K).min(num_estimates);
+ while high < num_estimates && distance(high) < distance(low) {
+ low += 1;
+ high += 1;
+ }
+
+ let bias_sum: f64 = biases[low..high].iter().sum();
+ bias_sum / (high - low) as f64
+}
+
+/// Compute the HyperLogLog++ cardinality estimate from a packed register
buffer. Faithful port of
+/// `HyperLogLogPlusPlusHelper.query`.
+fn query(p: usize, words: &[i64]) -> i64 {
+ let m_usize = 1usize << p;
+ let m = m_usize as f64;
+
+ let mut z_inverse = 0.0;
+ let mut v = 0.0;
+ let mut idx = 0;
+ for &word in words {
+ let word = word as u64;
+ let mut i = 0;
+ let mut shift = 0;
+ while idx < m_usize && i < REGISTERS_PER_WORD {
+ let m_idx = (word >> shift) & REGISTER_WORD_MASK;
+ z_inverse += 1.0 / ((1u64 << m_idx) as f64);
+ if m_idx == 0 {
+ v += 1.0;
+ }
+ shift += REGISTER_SIZE;
+ i += 1;
+ idx += 1;
+ }
+ }
+
+ let e = alpha_m2(p, m) / z_inverse;
+ let e_bias_corrected = if p < 19 && e < 5.0 * m {
+ e - estimate_bias(p, e)
+ } else {
+ e
+ };
+
+ let estimate = if v > 0.0 {
+ // Linear counting for small cardinalities.
+ let h = m * (m / v).ln();
+ if (p < 19 && h <= THRESHOLDS[p - 4]) || e <= 2.5 * m {
+ h
+ } else {
+ e_bias_corrected
+ }
+ } else {
+ e_bias_corrected
+ };
+
+ // Match Java's Math.round: floor(x + 0.5).
+ (estimate + 0.5).floor() as i64
+}
+
+/// Per-group accumulator (also used for the ungrouped / merge path).
+#[derive(Debug)]
+struct HllPlusPlusAccumulator {
+ p: usize,
+ words: Vec<i64>,
+}
+
+impl HllPlusPlusAccumulator {
+ fn new(p: usize) -> Self {
+ Self {
+ p,
+ words: vec![0i64; num_words(p)],
+ }
+ }
+}
+
+impl Accumulator for HllPlusPlusAccumulator {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ let array = &values[0];
+ let hashes = hash_values(array)?;
+ for (i, &hash) in hashes.iter().enumerate() {
+ if array.is_null(i) {
+ continue;
+ }
+ update_word(&mut self.words, self.p, hash);
+ }
+ Ok(())
+ }
+
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ let m = 1usize << self.p;
+ let num_rows = states[0].len();
+ let cols: Vec<&Int64Array> = states
+ .iter()
+ .map(|s| s.as_primitive::<arrow::datatypes::Int64Type>())
+ .collect();
+ let mut partial = vec![0i64; self.words.len()];
+ for row in 0..num_rows {
+ for (w, col) in cols.iter().enumerate() {
+ partial[w] = col.value(row);
+ }
+ merge_words(&mut self.words, &partial, m);
+ }
+ Ok(())
+ }
+
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ Ok(self
+ .words
+ .iter()
+ .map(|&w| ScalarValue::Int64(Some(w)))
+ .collect())
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ Ok(ScalarValue::Int64(Some(query(self.p, &self.words))))
+ }
+
+ fn size(&self) -> usize {
+ std::mem::size_of_val(self) + self.words.capacity() *
std::mem::size_of::<i64>()
+ }
+}
+
+/// Vectorized grouped accumulator. The `numWords` words for all groups live
in a single flat
+/// buffer, with group `g` occupying `words[g * num_words .. (g + 1) *
num_words]`.
+#[derive(Debug)]
+struct HllPlusPlusGroupsAccumulator {
+ p: usize,
+ m: usize,
+ num_words: usize,
+ words: Vec<i64>,
+}
+
+impl HllPlusPlusGroupsAccumulator {
+ fn new(p: usize) -> Self {
+ Self {
+ p,
+ m: 1usize << p,
Review Comment:
The groups accumulator stores `m: 1usize << p` but the scalar accumulator
recomputes `1usize << self.p` inline where needed. `m` is a pure function of
`p`, which never changes after construction, so the stored copy is redundant
derivable state. Drop the field and compute `1usize << self.p` in `merge_batch`
to match the scalar path. Keep the `num_words` field, since that one is used on
hot paths (`resize`, `group_slice`, `emit_words`) and caching it is a
reasonable trade.
##########
spark/src/test/resources/sql-tests/expressions/aggregate/approx_count_distinct.sql:
##########
@@ -0,0 +1,131 @@
+-- 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.
+
+-- approx_count_distinct estimates the number of distinct values using
HyperLogLog++.
+-- Comet ports Spark's HyperLogLogPlusPlus exactly, so results are compared
for equality.
+
+-- ============================================================
+-- A table with a range of cardinalities across many magnitudes
Review Comment:
Two coverage suggestions:
- Add a `TimestampType` case under a non-UTC `spark.sql.session.timeZone` to
confirm timezone-independence (the hashed value is UTC micros, so it should
match, but the datetime checklist calls for it).
- Add a non-`UTF8_BINARY` collated string column to confirm it falls back.
Verified this is currently safe: `case StringType` is a stable-identifier
pattern matching `StringType.equals`, which is true only for the default
UTF8_BINARY collation, so a collated `StringType(collationId)` falls through to
`case _ => false`. Spark hashes collated strings via the collation sort key
(`isCollationAware = true` in `HyperLogLogPlusPlusHelper.scala:103`), which the
plain-bytes native hash would not reproduce, so the fallback is correct. A test
locks it in:
```sql
statement
CREATE TABLE acd_coll(s string COLLATE UTF8_LCASE) USING parquet
statement
INSERT INTO acd_coll VALUES ('a'), ('A'), ('b')
query expect_fallback(Unsupported input data type)
SELECT approx_count_distinct(s) FROM acd_coll
```
--
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]
