You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/06/18 13:44:52 UTC
(datafusion) branch main updated: refactor: Simplify `approx_distinct` (-200 LoC) (#22921)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new ad8e7b7f2b refactor: Simplify `approx_distinct` (-200 LoC) (#22921)
ad8e7b7f2b is described below
commit ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9
Author: Yongting You <20...@gmail.com>
AuthorDate: Thu Jun 18 21:29:59 2026 +0800
refactor: Simplify `approx_distinct` (-200 LoC) (#22921)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
Attempt to simplify the `approx_distinct` implementation, the existing
complexity is due to there is no generic API to calculate hash for a
`array.elem(i)`, so we have to implement specialization for many
different types like primitive/string/stringview, and bloated the code
size.
This PR used a existing `create_hashes` for batched hashing that is
applicable to all array types, and it reduced 261 lines of code in
`approx_distinct.rs`
### Performance
<details>
<summary> Cargo bench result </summary>
```sh
cargo bench -p datafusion-functions-aggregate \
--bench approx_distinct \
-- --baseline main
```
```sh
nuplot not found, using plotters backend
Benchmarking approx_distinct i64 80% distinct: Collecting 100 samples in estimated 5.0135 s (884k ite
approx_distinct i64 80% distinct
time: [5.6406 µs 5.6477 µs 5.6550 µs]
change: [−0.9680% −0.7111% −0.4639%] (p = 0.00 < 0.05)
Change within noise threshold.
Found 4 outliers among 100 measurements (4.00%)
2 (2.00%) low mild
2 (2.00%) high mild
Benchmarking approx_distinct utf8 short 80% distinct: Collecting 100 samples in estimated 5.0360 s (3
approx_distinct utf8 short 80% distinct
time: [12.970 µs 12.977 µs 12.985 µs]
change: [+15.898% +16.116% +16.339%] (p = 0.00 < 0.05)
Performance has regressed.
Found 7 outliers among 100 measurements (7.00%)
2 (2.00%) low severe
1 (1.00%) low mild
4 (4.00%) high severe
Benchmarking approx_distinct utf8view short 80% distinct: Collecting 100 samples in estimated 5.0171
approx_distinct utf8view short 80% distinct
time: [8.7402 µs 8.7455 µs 8.7511 µs]
change: [+22.516% +22.703% +22.893%] (p = 0.00 < 0.05)
Performance has regressed.
Found 9 outliers among 100 measurements (9.00%)
1 (1.00%) low severe
5 (5.00%) high mild
3 (3.00%) high severe
Benchmarking approx_distinct utf8 long 80% distinct: Collecting 100 samples in estimated 5.0120 s (26
approx_distinct utf8 long 80% distinct
time: [19.060 µs 19.085 µs 19.108 µs]
change: [+9.9923% +10.224% +10.429%] (p = 0.00 < 0.05)
Performance has regressed.
Found 10 outliers among 100 measurements (10.00%)
1 (1.00%) low severe
4 (4.00%) low mild
4 (4.00%) high mild
1 (1.00%) high severe
Benchmarking approx_distinct utf8view long 80% distinct: Collecting 100 samples in estimated 5.0800 s
approx_distinct utf8view long 80% distinct
time: [21.281 µs 21.306 µs 21.335 µs]
change: [+1.8930% +2.0965% +2.3087%] (p = 0.00 < 0.05)
Performance has regressed.
Benchmarking approx_distinct i64 99% distinct: Collecting 100 samples in estimated 5.0037 s (884k ite
approx_distinct i64 99% distinct
time: [5.6507 µs 5.6645 µs 5.6805 µs]
change: [+0.1104% +0.3620% +0.6143%] (p = 0.00 < 0.05)
Change within noise threshold.
Found 4 outliers among 100 measurements (4.00%)
2 (2.00%) low mild
1 (1.00%) high mild
1 (1.00%) high severe
Benchmarking approx_distinct utf8 short 99% distinct: Collecting 100 samples in estimated 5.0298 s (3
approx_distinct utf8 short 99% distinct
time: [12.956 µs 12.968 µs 12.979 µs]
change: [+15.776% +16.044% +16.296%] (p = 0.00 < 0.05)
Performance has regressed.
Found 6 outliers among 100 measurements (6.00%)
4 (4.00%) low mild
2 (2.00%) high severe
Benchmarking approx_distinct utf8view short 99% distinct: Collecting 100 samples in estimated 5.0295
approx_distinct utf8view short 99% distinct
time: [8.8005 µs 8.8054 µs 8.8106 µs]
change: [+22.620% +22.909% +23.202%] (p = 0.00 < 0.05)
Performance has regressed.
Found 7 outliers among 100 measurements (7.00%)
2 (2.00%) low severe
2 (2.00%) low mild
3 (3.00%) high severe
Benchmarking approx_distinct utf8 long 99% distinct: Collecting 100 samples in estimated 5.0467 s (26
approx_distinct utf8 long 99% distinct
time: [19.134 µs 19.197 µs 19.309 µs]
change: [+9.0293% +9.4204% +9.8109%] (p = 0.00 < 0.05)
Performance has regressed.
Found 4 outliers among 100 measurements (4.00%)
2 (2.00%) low severe
1 (1.00%) low mild
1 (1.00%) high severe
Benchmarking approx_distinct utf8view long 99% distinct: Collecting 100 samples in estimated 5.0953 s
approx_distinct utf8view long 99% distinct
time: [21.295 µs 21.332 µs 21.384 µs]
change: [+1.9350% +2.2141% +2.4823%] (p = 0.00 < 0.05)
Performance has regressed.
Found 1 outliers among 100 measurements (1.00%)
1 (1.00%) high mild
Benchmarking approx_distinct u8 bitmap: Collecting 100 samples in estimated 5.0022 s (4.2M iterations
approx_distinct u8 bitmap
time: [1.1961 µs 1.1976 µs 1.1994 µs]
change: [−3.8254% −3.1554% −2.5920%] (p = 0.00 < 0.05)
Performance has improved.
Found 5 outliers among 100 measurements (5.00%)
2 (2.00%) low mild
3 (3.00%) high mild
Benchmarking approx_distinct i8 bitmap: Collecting 100 samples in estimated 5.0058 s (4.2M iterations
approx_distinct i8 bitmap
time: [1.2043 µs 1.2058 µs 1.2075 µs]
change: [−0.6865% −0.3102% −0.0076%] (p = 0.07 > 0.05)
No change in performance detected.
Found 9 outliers among 100 measurements (9.00%)
1 (1.00%) low severe
3 (3.00%) low mild
4 (4.00%) high mild
1 (1.00%) high severe
Benchmarking approx_distinct u16 bitmap: Collecting 100 samples in estimated 5.0052 s (1.1M iteration
approx_distinct u16 bitmap
time: [4.3272 µs 4.3392 µs 4.3521 µs]
change: [−1.5999% −1.1667% −0.7366%] (p = 0.00 < 0.05)
Change within noise threshold.
Found 6 outliers among 100 measurements (6.00%)
5 (5.00%) low mild
1 (1.00%) high mild
Benchmarking approx_distinct i16 bitmap: Collecting 100 samples in estimated 5.0171 s (1.1M iteration
approx_distinct i16 bitmap
time: [4.4383 µs 4.4431 µs 4.4479 µs]
change: [+2.4499% +2.8213% +3.1689%] (p = 0.00 < 0.05)
Performance has regressed.
Found 5 outliers among 100 measurements (5.00%)
4 (4.00%) low mild
1 (1.00%) high mild
Benchmarking approx_distinct_grouped/Int64 50000 groups: Collecting 10 samples in estimated 5.4914 s
approx_distinct_grouped/Int64 50000 groups
time: [9.8851 ms 9.9005 ms 9.9293 ms]
change: [−3.5210% −2.9384% −2.4270%] (p = 0.00 < 0.05)
Performance has improved.
Found 1 outliers among 10 measurements (10.00%)
1 (10.00%) high mild
Benchmarking approx_distinct_grouped/Utf8 50000 groups: Collecting 10 samples in estimated 5.0165 s (
approx_distinct_grouped/Utf8 50000 groups
time: [10.116 ms 10.148 ms 10.186 ms]
change: [−4.9237% −4.6468% −4.3582%] (p = 0.00 < 0.05)
Performance has improved.
Benchmarking approx_distinct_grouped/Utf8View 50000 groups: Collecting 10 samples in estimated 5.4867
approx_distinct_grouped/Utf8View 50000 groups
time: [9.9450 ms 9.9498 ms 9.9556 ms]
change: [−3.4418% −3.0161% −2.5685%] (p = 0.00 < 0.05)
Performance has improved.
Found 1 outliers among 10 measurements (10.00%)
1 (10.00%) high severe
```
</details>
It shows some get 5% faster due to batched hashing, some utf cases get
slower (the worst one 22% slower)
I think it's still a good idea to ignore the regression and simplify the
code due to:
#### Amdahl's Law
If we make function X 20% faster, but function X only takes 1% of query
time, then the complexity to win the performance might not be worthy:
specifically the microbench only measured `update_batch()` function,
this piece of code is highly vectorizable, and it can very unlikely to
be significant on real queries.
I tried to construct a query that is very heavy on `update_batch`, still
can't observe end-to-end difference:
```sql id="8d3m2p"
> select approx_distinct(v1)
from (
select arrow_cast(v1, 'Utf8View')
from generate_series(100000000)
as t1(v1)) as t_string(v1);
+------------------------------+
| approx_distinct(t_string.v1) |
+------------------------------+
| 99201889 |
+------------------------------+
1 row(s) fetched.
Elapsed 0.139 seconds.
-- Runtime almost the same on PR v.s. main
```
#### LLVM Optimization
For the slowest microbench, I think the root cause is that LLVM can
optimize the manually simplified code more easily.
The existing implementation has the following fast path:
https://github.com/apache/datafusion/blob/b8998c762bb864a9f3607a518384b03dcf40eb61/datafusion/functions-aggregate/src/approx_distinct.rs#L254-L261
The same optimization also exists in the common, simpler API
`create_hashes`:
https://github.com/apache/datafusion/blob/b8998c762bb864a9f3607a518384b03dcf40eb61/datafusion/common/src/hash_utils.rs#L352
The existing implementation is still faster likely because the code is
manually specialized, while `create_hashes` is more branchy. This makes
LLVM easier to figure out how to optimize and bring 20% speedup.
However, this kind of optimization can be applied endlessly and would
introduce complexity everywhere, so I do not think it is worth
preserving here.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
1. Extend `create_hashes` with a hash state that is optimized for
statistical quality
2. Simplify `approx_distinct` with create_hashes
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
3. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
---
Cargo.lock | 1 -
datafusion/common/src/hash_utils.rs | 148 +++++--
datafusion/functions-aggregate/Cargo.toml | 1 -
.../functions-aggregate/src/approx_distinct.rs | 489 ++++++---------------
datafusion/functions-aggregate/src/hyperloglog.rs | 14 +-
5 files changed, 241 insertions(+), 412 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 1cb3fe0ecb..c2b729677d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2241,7 +2241,6 @@ dependencies = [
"datafusion-macros",
"datafusion-physical-expr",
"datafusion-physical-expr-common",
- "foldhash 0.2.0",
"half",
"log",
"num-traits",
diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs
index 02db75498a..e9c4c26e37 100644
--- a/datafusion/common/src/hash_utils.rs
+++ b/datafusion/common/src/hash_utils.rs
@@ -31,8 +31,54 @@ use itertools::Itertools;
use std::collections::HashMap;
use std::hash::{BuildHasher, Hash, Hasher};
-/// The hash random state used throughout DataFusion for hashing.
+/// [`RandomState`] is optimized for speed and suitable for hash tables and
+/// bloom filters. [`QualityRandomState`] is optimized for statistical quality
+/// and suitable for algorithms such as HyperLogLog. The tradeoff is that the
+/// fast variant gives up some statistical quality, while the quality variant
+/// is slightly slower.
+///
+/// See: <https://docs.rs/foldhash/0.2.0/src/foldhash/lib.rs.html#17-21>
pub type RandomState = FixedState;
+pub type QualityRandomState = foldhash::quality::FixedState;
+
+/// Fixed quality hash state used by HyperLogLog sketches.
+///
+/// The seed is part of the HLL wire/storage semantics: serialized sketches only
+/// remain mergeable if every producer uses the same hash state.
+pub const HLL_RANDOM_STATE: QualityRandomState = QualityRandomState::with_seed(0);
+
+/// Hash state used by [`create_hashes`].
+///
+/// Multi-column hashing folds the previous column hash into a fresh hasher
+/// before hashing the next column. This trait keeps that seeded hasher in the
+/// same foldhash tier as the top-level hash state.
+pub trait HashState: BuildHasher {
+ type SeededState: BuildHasher;
+
+ fn seeded_state(&self, seed: u64) -> Self::SeededState;
+}
+
+impl HashState for FixedState {
+ type SeededState = foldhash::fast::SeedableRandomState;
+
+ fn seeded_state(&self, seed: u64) -> Self::SeededState {
+ foldhash::fast::SeedableRandomState::with_seed(
+ seed,
+ foldhash::SharedSeed::global_fixed(),
+ )
+ }
+}
+
+impl HashState for foldhash::quality::FixedState {
+ type SeededState = foldhash::quality::SeedableRandomState;
+
+ fn seeded_state(&self, seed: u64) -> Self::SeededState {
+ foldhash::quality::SeedableRandomState::with_seed(
+ seed,
+ foldhash::SharedSeed::global_fixed(),
+ )
+ }
+}
#[cfg(not(feature = "force_hash_collisions"))]
use crate::cast::{
@@ -99,7 +145,7 @@ thread_local! {
/// ```
pub fn with_hashes<I, T, F, R>(
arrays: I,
- random_state: &RandomState,
+ random_state: &impl HashState,
callback: F,
) -> Result<R>
where
@@ -141,7 +187,11 @@ where
}
#[cfg(not(feature = "force_hash_collisions"))]
-fn hash_null(random_state: &RandomState, hashes_buffer: &'_ mut [u64], mul_col: bool) {
+fn hash_null<S: HashState>(
+ random_state: &S,
+ hashes_buffer: &'_ mut [u64],
+ mul_col: bool,
+) {
if mul_col {
hashes_buffer.iter_mut().for_each(|hash| {
// stable hash for null value
@@ -155,13 +205,13 @@ fn hash_null(random_state: &RandomState, hashes_buffer: &'_ mut [u64], mul_col:
}
pub trait HashValue {
- fn hash_one(&self, state: &RandomState) -> u64;
+ fn hash_one<S: BuildHasher>(&self, state: &S) -> u64;
/// Write this value into an existing hasher (same data as `hash_one`).
fn hash_write(&self, hasher: &mut impl Hasher);
}
impl<T: HashValue + ?Sized> HashValue for &T {
- fn hash_one(&self, state: &RandomState) -> u64 {
+ fn hash_one<S: BuildHasher>(&self, state: &S) -> u64 {
T::hash_one(self, state)
}
fn hash_write(&self, hasher: &mut impl Hasher) {
@@ -172,7 +222,7 @@ impl<T: HashValue + ?Sized> HashValue for &T {
macro_rules! hash_value {
($($t:ty),+) => {
$(impl HashValue for $t {
- fn hash_one(&self, state: &RandomState) -> u64 {
+ fn hash_one<S: BuildHasher>(&self, state: &S) -> u64 {
state.hash_one(self)
}
fn hash_write(&self, hasher: &mut impl Hasher) {
@@ -187,7 +237,7 @@ hash_value!(bool, str, [u8], IntervalDayTime, IntervalMonthDayNano);
macro_rules! hash_float_value {
($(($t:ty, $i:ty)),+) => {
$(impl HashValue for $t {
- fn hash_one(&self, state: &RandomState) -> u64 {
+ fn hash_one<S: BuildHasher>(&self, state: &S) -> u64 {
// +0.0 and -0.0 differ only in the sign bit but compare equal
// under IEEE 754; normalize -0.0 → +0.0 so Hash agrees with Eq.
let bits = <$i>::from_ne_bytes(self.to_ne_bytes());
@@ -204,25 +254,13 @@ macro_rules! hash_float_value {
}
hash_float_value!((half::f16, u16), (f32, u32), (f64, u64));
-/// Create a `SeedableRandomState` whose per-hasher seed incorporates `seed`.
-/// This folds the previous hash into the hasher's initial state so only the
-/// new value needs to pass through the hash function — same cost as `hash_one`.
-#[cfg(not(feature = "force_hash_collisions"))]
-#[inline]
-fn seeded_state(seed: u64) -> foldhash::fast::SeedableRandomState {
- foldhash::fast::SeedableRandomState::with_seed(
- seed,
- foldhash::SharedSeed::global_fixed(),
- )
-}
-
/// Builds hash values of PrimitiveArray and writes them into `hashes_buffer`
/// If `rehash==true` this folds the existing hash into the hasher state
/// and hashes only the new value (avoiding a separate combine step).
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_array_primitive<T>(
array: &PrimitiveArray<T>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
rehash: bool,
) where
@@ -237,7 +275,7 @@ fn hash_array_primitive<T>(
if array.null_count() == 0 {
if rehash {
for (hash, &value) in hashes_buffer.iter_mut().zip(array.values().iter()) {
- let mut hasher = seeded_state(*hash).build_hasher();
+ let mut hasher = random_state.seeded_state(*hash).build_hasher();
value.hash_write(&mut hasher);
*hash = hasher.finish();
}
@@ -249,7 +287,7 @@ fn hash_array_primitive<T>(
} else if rehash {
for i in array.nulls().unwrap().valid_indices() {
let value = unsafe { array.value_unchecked(i) };
- let mut hasher = seeded_state(hashes_buffer[i]).build_hasher();
+ let mut hasher = random_state.seeded_state(hashes_buffer[i]).build_hasher();
value.hash_write(&mut hasher);
hashes_buffer[i] = hasher.finish();
}
@@ -267,7 +305,7 @@ fn hash_array_primitive<T>(
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_array<T>(
array: &T,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
rehash: bool,
) where
@@ -322,7 +360,7 @@ fn hash_string_view_array_inner<
const REHASH: bool,
>(
array: &GenericByteViewArray<T>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) {
assert_eq!(
@@ -351,7 +389,7 @@ fn hash_string_view_array_inner<
// all views are inlined, no need to access external buffers
if !HAS_BUFFERS || view_len <= 12 {
if REHASH {
- let mut hasher = seeded_state(*hash).build_hasher();
+ let mut hasher = random_state.seeded_state(*hash).build_hasher();
v.hash_write(&mut hasher);
*hash = hasher.finish();
} else {
@@ -362,7 +400,7 @@ fn hash_string_view_array_inner<
// view is not inlined, so we need to hash the bytes as well
let value = view_bytes(view_len, v);
if REHASH {
- let mut hasher = seeded_state(*hash).build_hasher();
+ let mut hasher = random_state.seeded_state(*hash).build_hasher();
value.hash_write(&mut hasher);
*hash = hasher.finish();
} else {
@@ -377,7 +415,7 @@ fn hash_string_view_array_inner<
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_generic_byte_view_array<T: ByteViewType>(
array: &GenericByteViewArray<T>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
rehash: bool,
) {
@@ -396,7 +434,7 @@ fn hash_generic_byte_view_array<T: ByteViewType>(
}
(false, false, true) => {
for (hash, &view) in hashes_buffer.iter_mut().zip(array.views().iter()) {
- let mut hasher = seeded_state(*hash).build_hasher();
+ let mut hasher = random_state.seeded_state(*hash).build_hasher();
view.hash_write(&mut hasher);
*hash = hasher.finish();
}
@@ -449,7 +487,7 @@ fn hash_dictionary_inner<
const MULTI_COL: bool,
>(
array: &DictionaryArray<K>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
// Hash each dictionary value once, and then use that computed
@@ -491,7 +529,7 @@ fn hash_dictionary_inner<
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_dictionary<K: ArrowDictionaryKeyType>(
array: &DictionaryArray<K>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
multi_col: bool,
) -> Result<()> {
@@ -547,7 +585,7 @@ fn hash_dictionary<K: ArrowDictionaryKeyType>(
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_struct_array(
array: &StructArray,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
let nulls = array.nulls();
@@ -577,7 +615,7 @@ fn hash_struct_array(
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_map_array(
array: &MapArray,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
let nulls = array.nulls();
@@ -628,7 +666,7 @@ fn hash_map_array(
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_list_array<OffsetSize>(
array: &GenericListArray<OffsetSize>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()>
where
@@ -679,7 +717,7 @@ where
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_list_view_array<OffsetSize>(
array: &GenericListViewArray<OffsetSize>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()>
where
@@ -718,7 +756,7 @@ where
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_union_array(
array: &UnionArray,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
let DataType::Union(union_fields, _mode) = array.data_type() else {
@@ -750,7 +788,7 @@ fn hash_union_array(
fn hash_union_array_default(
array: &UnionArray,
union_fields: &UnionFields,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
let mut child_hashes: HashMap<i8, Vec<u64>> =
@@ -791,7 +829,7 @@ fn hash_union_array_default(
fn hash_sparse_union_array(
array: &UnionArray,
union_fields: &UnionFields,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
use std::collections::HashMap;
@@ -846,7 +884,7 @@ fn hash_sparse_union_array(
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_fixed_list_array(
array: &FixedSizeListArray,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
let values = array.values();
@@ -885,7 +923,7 @@ fn hash_run_array_inner<
const REHASH: bool,
>(
array: &RunArray<R>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
) -> Result<()> {
// We find the relevant runs that cover potentially sliced arrays, so we can only hash those
@@ -952,7 +990,7 @@ fn hash_run_array_inner<
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_run_array<R: RunEndIndexType>(
array: &RunArray<R>,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
rehash: bool,
) -> Result<()> {
@@ -979,7 +1017,7 @@ fn hash_run_array<R: RunEndIndexType>(
#[cfg(not(feature = "force_hash_collisions"))]
fn hash_single_array(
array: &dyn Array,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &mut [u64],
rehash: bool,
) -> Result<()> {
@@ -1052,7 +1090,7 @@ fn hash_single_array(
#[cfg(feature = "force_hash_collisions")]
fn hash_single_array(
_array: &dyn Array,
- _random_state: &RandomState,
+ _random_state: &impl HashState,
hashes_buffer: &mut [u64],
_rehash: bool,
) -> Result<()> {
@@ -1105,7 +1143,7 @@ impl AsDynArray for &ArrayRef {
/// `hashes_buffer` should be pre-sized appropriately.
pub fn create_hashes<'a, I, T>(
arrays: I,
- random_state: &RandomState,
+ random_state: &impl HashState,
hashes_buffer: &'a mut [u64],
) -> Result<&'a mut [u64]>
where
@@ -1814,6 +1852,30 @@ mod tests {
assert_eq!(hashes1, hashes2);
}
+ #[test]
+ fn test_create_hashes_with_quality_hash_state() {
+ let int_array: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4]));
+ let str_array: ArrayRef = Arc::new(StringArray::from(vec!["a", "b", "c", "d"]));
+ let quality_state = foldhash::quality::FixedState::with_seed(0);
+
+ let mut one_col_hashes = vec![0; int_array.len()];
+ create_hashes([&int_array], &quality_state, &mut one_col_hashes).unwrap();
+ let expected_hashes: Vec<_> = [1i32, 2, 3, 4]
+ .iter()
+ .map(|value| quality_state.hash_one(value))
+ .collect();
+ assert_eq!(one_col_hashes, expected_hashes);
+
+ let mut two_col_hashes = vec![0; int_array.len()];
+ create_hashes(
+ [&int_array, &str_array],
+ &quality_state,
+ &mut two_col_hashes,
+ )
+ .unwrap();
+ assert_ne!(two_col_hashes, one_col_hashes);
+ }
+
#[test]
fn test_with_hashes() {
let array: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4]));
diff --git a/datafusion/functions-aggregate/Cargo.toml b/datafusion/functions-aggregate/Cargo.toml
index 778e6a24bf..ff89808f0b 100644
--- a/datafusion/functions-aggregate/Cargo.toml
+++ b/datafusion/functions-aggregate/Cargo.toml
@@ -50,7 +50,6 @@ datafusion-functions-aggregate-common = { workspace = true }
datafusion-macros = { workspace = true }
datafusion-physical-expr = { workspace = true }
datafusion-physical-expr-common = { workspace = true }
-foldhash = "0.2"
half = { workspace = true }
log = { workspace = true }
num-traits = { workspace = true }
diff --git a/datafusion/functions-aggregate/src/approx_distinct.rs b/datafusion/functions-aggregate/src/approx_distinct.rs
index 0c5a438454..04ae5c1b35 100644
--- a/datafusion/functions-aggregate/src/approx_distinct.rs
+++ b/datafusion/functions-aggregate/src/approx_distinct.rs
@@ -18,20 +18,19 @@
//! Defines physical expressions that can evaluated at runtime during query execution
use crate::hyperloglog::{HLL_HASH_STATE, HyperLogLog, NUM_REGISTERS, count_from_hashes};
-use arrow::array::{Array, BinaryArray, StringViewArray};
use arrow::array::{
- AsArray, BinaryBuilder, BooleanArray, GenericBinaryArray, GenericStringArray,
- OffsetSizeTrait, PrimitiveArray, UInt64Array,
+ Array, ArrayRef, BinaryArray, BinaryBuilder, BooleanArray, PrimitiveArray,
+ UInt64Array,
};
use arrow::buffer::NullBuffer;
use arrow::datatypes::{
- ArrowPrimitiveType, Date32Type, Date64Type, FieldRef, Int32Type, Int64Type,
- Time32MillisecondType, Time32SecondType, Time64MicrosecondType, Time64NanosecondType,
- TimeUnit, TimestampMicrosecondType, TimestampMillisecondType,
+ ArrowPrimitiveType, DataType, Date32Type, Date64Type, Field, FieldRef, Int32Type,
+ Int64Type, Time32MillisecondType, Time32SecondType, Time64MicrosecondType,
+ Time64NanosecondType, TimeUnit, TimestampMicrosecondType, TimestampMillisecondType,
TimestampNanosecondType, TimestampSecondType, UInt32Type, UInt64Type,
};
-use arrow::{array::ArrayRef, datatypes::DataType, datatypes::Field};
use datafusion_common::ScalarValue;
+use datafusion_common::hash_utils::create_hashes;
use datafusion_common::{
DataFusionError, Result, downcast_value, internal_datafusion_err, internal_err,
not_impl_err,
@@ -51,8 +50,8 @@ use datafusion_functions_aggregate_common::aggregate::groups_accumulator::nulls:
use datafusion_functions_aggregate_common::noop_accumulator::NoopAccumulator;
use datafusion_macros::user_doc;
use std::fmt::{Debug, Formatter};
-use std::hash::{BuildHasher, Hash};
-use std::marker::PhantomData;
+use std::hash::Hash;
+use std::mem::{size_of, size_of_val};
use std::sync::Arc;
make_udaf_expr_and_func!(
@@ -124,192 +123,129 @@ impl<A: Accumulator> Accumulator for ApproxDistinctBitmapWrapper<A> {
}
#[derive(Debug)]
-struct NumericHLLAccumulator<T>
-where
- T: ArrowPrimitiveType,
- T::Native: Hash,
-{
- hll: HyperLogLog<T::Native>,
+struct HLLAccumulator {
+ hll: HyperLogLog<u8>,
+ hashes: Vec<u64>,
}
-impl<T> NumericHLLAccumulator<T>
-where
- T: ArrowPrimitiveType,
- T::Native: Hash,
-{
+impl HLLAccumulator {
pub fn new() -> Self {
Self {
hll: HyperLogLog::new(),
+ hashes: Vec::new(),
}
}
}
-#[derive(Debug)]
-struct StringHLLAccumulator<T>
-where
- T: OffsetSizeTrait,
-{
- hll: HyperLogLog<str>,
- phantom_data: PhantomData<T>,
-}
+impl Accumulator for HLLAccumulator {
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ let array = values[0].as_ref();
+ self.hashes.clear();
+ self.hashes.resize(array.len(), 0);
+ create_hashes([array], &HLL_HASH_STATE, &mut self.hashes)?;
-impl<T> StringHLLAccumulator<T>
-where
- T: OffsetSizeTrait,
-{
- pub fn new() -> Self {
- Self {
- hll: HyperLogLog::new(),
- phantom_data: PhantomData,
+ match array.logical_nulls() {
+ None => {
+ for &hash in &self.hashes {
+ self.hll.add_hashed(hash);
+ }
+ }
+ Some(nulls) => {
+ for row in 0..array.len() {
+ if nulls.is_valid(row) {
+ self.hll.add_hashed(self.hashes[row]);
+ }
+ }
+ }
}
+ Ok(())
}
-}
-
-#[derive(Debug)]
-struct StringViewHLLAccumulator {
- hll: HyperLogLog<str>,
-}
-impl StringViewHLLAccumulator {
- pub fn new() -> Self {
- Self {
- hll: HyperLogLog::new(),
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ assert_eq!(1, states.len(), "expect only 1 element in the states");
+ let binary_array = downcast_value!(states[0], BinaryArray);
+ for v in binary_array.iter() {
+ let v = v.ok_or_else(|| {
+ internal_datafusion_err!("Impossibly got empty binary array from states")
+ })?;
+ let other = v.try_into()?;
+ self.hll.merge(&other);
}
+ Ok(())
+ }
+
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ let value = ScalarValue::from(&self.hll);
+ Ok(vec![value])
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ Ok(ScalarValue::UInt64(Some(self.hll.count() as u64)))
+ }
+
+ fn size(&self) -> usize {
+ size_of_val(self) + self.hashes.capacity() * size_of::<u64>()
}
}
+/// Specialize the numeric case for extra performance.
#[derive(Debug)]
-struct BinaryHLLAccumulator<T>
+struct NumericHLLAccumulator<T>
where
- T: OffsetSizeTrait,
+ T: ArrowPrimitiveType,
+ T::Native: Hash,
{
- hll: HyperLogLog<[u8]>,
- phantom_data: PhantomData<T>,
+ hll: HyperLogLog<T::Native>,
}
-impl<T> BinaryHLLAccumulator<T>
+impl<T> NumericHLLAccumulator<T>
where
- T: OffsetSizeTrait,
+ T: ArrowPrimitiveType,
+ T::Native: Hash,
{
pub fn new() -> Self {
Self {
hll: HyperLogLog::new(),
- phantom_data: PhantomData,
}
}
}
-macro_rules! default_accumulator_impl {
- () => {
- fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
- assert_eq!(1, states.len(), "expect only 1 element in the states");
- let binary_array = downcast_value!(states[0], BinaryArray);
- for v in binary_array.iter() {
- let v = v.ok_or_else(|| {
- internal_datafusion_err!(
- "Impossibly got empty binary array from states"
- )
- })?;
- let other = v.try_into()?;
- self.hll.merge(&other);
- }
- Ok(())
- }
-
- fn state(&mut self) -> Result<Vec<ScalarValue>> {
- let value = ScalarValue::from(&self.hll);
- Ok(vec![value])
- }
-
- fn evaluate(&mut self) -> Result<ScalarValue> {
- Ok(ScalarValue::UInt64(Some(self.hll.count() as u64)))
- }
-
- fn size(&self) -> usize {
- // HLL has static size
- std::mem::size_of_val(self)
- }
- };
-}
-
-impl<T> Accumulator for BinaryHLLAccumulator<T>
+impl<T> Accumulator for NumericHLLAccumulator<T>
where
- T: OffsetSizeTrait,
+ T: ArrowPrimitiveType + Debug,
+ T::Native: Hash,
{
fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let array: &GenericBinaryArray<T> =
- downcast_value!(values[0], GenericBinaryArray, T);
- // flatten because we would skip nulls
+ let array: &PrimitiveArray<T> = downcast_value!(values[0], PrimitiveArray, T);
self.hll.extend(array.into_iter().flatten());
Ok(())
}
- default_accumulator_impl!();
-}
-
-impl Accumulator for StringViewHLLAccumulator {
- fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let array: &StringViewArray = downcast_value!(values[0], StringViewArray);
-
- if array.data_buffers().is_empty() {
- // Fast path: with no data buffers every value is inline, so they all
- // take the u128 path — no need to check the length per row.
- for (i, &view) in array.views().iter().enumerate() {
- if !array.is_null(i) {
- self.hll.add_hashed(HLL_HASH_STATE.hash_one(view));
- }
- }
- } else {
- // Mixed batch: decide per row by length. Short strings still use the
- // u128 path so they match how they'd be hashed in an all-inline
- // batch; only the genuinely out-of-line strings materialize a &str.
- for (i, &view) in array.views().iter().enumerate() {
- if array.is_null(i) {
- continue;
- }
- // The low 32 bits of the u128 view encode the string length.
- if (view as u32) <= 12 {
- self.hll.add_hashed(HLL_HASH_STATE.hash_one(view));
- } else {
- self.hll.add(array.value(i));
- }
- }
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ assert_eq!(1, states.len(), "expect only 1 element in the states");
+ let binary_array = downcast_value!(states[0], BinaryArray);
+ for v in binary_array.iter() {
+ let v = v.ok_or_else(|| {
+ internal_datafusion_err!("Impossibly got empty binary array from states")
+ })?;
+ let other = v.try_into()?;
+ self.hll.merge(&other);
}
-
Ok(())
}
- default_accumulator_impl!();
-}
-
-impl<T> Accumulator for StringHLLAccumulator<T>
-where
- T: OffsetSizeTrait,
-{
- fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let array: &GenericStringArray<T> =
- downcast_value!(values[0], GenericStringArray, T);
- // flatten because we would skip nulls
- self.hll.extend(array.into_iter().flatten());
- Ok(())
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ let value = ScalarValue::from(&self.hll);
+ Ok(vec![value])
}
- default_accumulator_impl!();
-}
-
-impl<T> Accumulator for NumericHLLAccumulator<T>
-where
- T: ArrowPrimitiveType + Debug,
- T::Native: Hash,
-{
- fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
- let array: &PrimitiveArray<T> = downcast_value!(values[0], PrimitiveArray, T);
- // flatten because we would skip nulls
- self.hll.extend(array.into_iter().flatten());
- Ok(())
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ Ok(ScalarValue::UInt64(Some(self.hll.count() as u64)))
}
- default_accumulator_impl!();
+ fn size(&self) -> usize {
+ size_of_val(self)
+ }
}
/// Maximum number of distinct hashes kept in the sparse representation of a
@@ -496,125 +432,6 @@ impl GroupHll {
}
}
-/// Computes HyperLogLog hashes for the rows of an input array, type by type.
-///
-/// The hashing matches the per-group [`Accumulator`] implementations exactly so
-/// that the grouped and ungrouped paths produce identical estimates.
-trait HllValueHasher: Send + Sync + 'static {
- /// Invoke `f(row_index, hash)` for every row that is valid according to
- /// `nulls`. `nulls = None` means every row is valid (caller has
- /// pre-combined value-nulls and filter into a single buffer).
- fn for_each_hash(
- array: &dyn Array,
- nulls: Option<&NullBuffer>,
- f: impl FnMut(usize, u64),
- );
-}
-
-struct NumericHasher<T>(PhantomData<T>);
-
-impl<T> HllValueHasher for NumericHasher<T>
-where
- T: ArrowPrimitiveType + Send + Sync + 'static,
- T::Native: Hash,
-{
- #[inline]
- fn for_each_hash(
- array: &dyn Array,
- nulls: Option<&NullBuffer>,
- mut f: impl FnMut(usize, u64),
- ) {
- let array: &PrimitiveArray<T> = array.as_primitive::<T>();
- match nulls {
- None => {
- for (i, v) in array.values().iter().enumerate() {
- f(i, HLL_HASH_STATE.hash_one(v));
- }
- }
- Some(nulls) => {
- for i in 0..array.len() {
- if nulls.is_valid(i) {
- f(i, HLL_HASH_STATE.hash_one(array.value(i)));
- }
- }
- }
- }
- }
-}
-
-struct Utf8Hasher<O>(PhantomData<O>);
-
-impl<O: OffsetSizeTrait> HllValueHasher for Utf8Hasher<O> {
- #[inline]
- fn for_each_hash(
- array: &dyn Array,
- nulls: Option<&NullBuffer>,
- mut f: impl FnMut(usize, u64),
- ) {
- let array: &GenericStringArray<O> = array.as_string::<O>();
- for i in 0..array.len() {
- if nulls.is_none_or(|n| n.is_valid(i)) {
- f(i, HLL_HASH_STATE.hash_one(array.value(i)));
- }
- }
- }
-}
-
-struct Utf8ViewHasher;
-
-impl HllValueHasher for Utf8ViewHasher {
- #[inline]
- fn for_each_hash(
- array: &dyn Array,
- nulls: Option<&NullBuffer>,
- mut f: impl FnMut(usize, u64),
- ) {
- let array: &StringViewArray = array.as_string_view();
- // Mirror `StringViewHLLAccumulator`: hash the raw inline view when all
- // strings are stored inline (≤ 12 bytes), avoiding `&str` materialization.
- if array.data_buffers().is_empty() {
- let views = array.views();
- for i in 0..array.len() {
- if nulls.is_none_or(|n| n.is_valid(i)) {
- f(i, HLL_HASH_STATE.hash_one(views[i]));
- }
- }
- } else {
- // Mixed batch: short strings (≤ 12 bytes) are still inline and must
- // be hashed as the raw u128 view to match the all-inline fast path.
- let views = array.views();
- for i in 0..array.len() {
- if nulls.is_none_or(|n| n.is_valid(i)) {
- let view = views[i];
- if (view as u32) <= 12 {
- f(i, HLL_HASH_STATE.hash_one(view));
- } else {
- f(i, HLL_HASH_STATE.hash_one(array.value(i)));
- }
- }
- }
- }
- }
-}
-
-struct BinaryHasher<O>(PhantomData<O>);
-
-impl<O: OffsetSizeTrait> HllValueHasher for BinaryHasher<O> {
- #[inline]
- fn for_each_hash(
- array: &dyn Array,
- nulls: Option<&NullBuffer>,
- mut f: impl FnMut(usize, u64),
- ) {
- let array: &GenericBinaryArray<O> = array.as_binary::<O>();
- for i in 0..array.len() {
- if nulls.is_none_or(|n| n.is_valid(i)) {
- f(i, HLL_HASH_STATE.hash_one(array.value(i)));
- }
- }
- }
-}
-
/// A [`GroupsAccumulator`] for `approx_distinct` that keeps one adaptive
/// (sparse → dense) HyperLogLog sketch per group.
///
@@ -646,20 +463,21 @@ impl<O: OffsetSizeTrait> HllValueHasher for BinaryHasher<O> {
/// replayed into a dense sketch. New values for `b` update the dense registers
/// directly, and serialized state is the raw [`NUM_REGISTERS`]-byte register
/// array.
-struct HllGroupsAccumulator<H: HllValueHasher> {
+struct HllGroupsAccumulator {
/// Per-group sketches, indexed by `group_index`.
groups: Vec<GroupHll>,
/// Incrementally maintained estimate of heap bytes used by `groups`.
allocated_bytes: usize,
- phantom: PhantomData<H>,
+ /// Reused workspace for vectorized value hashing.
+ hashes: Vec<u64>,
}
-impl<H: HllValueHasher> HllGroupsAccumulator<H> {
+impl HllGroupsAccumulator {
fn new() -> Self {
Self {
groups: Vec::new(),
allocated_bytes: 0,
- phantom: PhantomData,
+ hashes: Vec::new(),
}
}
@@ -677,7 +495,7 @@ impl<H: HllValueHasher> HllGroupsAccumulator<H> {
}
}
-impl<H: HllValueHasher> GroupsAccumulator for HllGroupsAccumulator<H> {
+impl GroupsAccumulator for HllGroupsAccumulator {
fn update_batch(
&mut self,
values: &[ArrayRef],
@@ -686,17 +504,30 @@ impl<H: HllValueHasher> GroupsAccumulator for HllGroupsAccumulator<H> {
total_num_groups: usize,
) -> Result<()> {
self.ensure_groups(total_num_groups);
- let groups = &mut self.groups;
+ let array = values[0].as_ref();
+ self.hashes.clear();
+ self.hashes.resize(array.len(), 0);
+ create_hashes([array], &HLL_HASH_STATE, &mut self.hashes)?;
+
let mut delta: isize = 0;
- // Pre-combine value-nulls and filter into one mask so the callback
- // needs no per-row branching.
+ // Pre-combine value-nulls and filter into one mask so the update loop
+ // only visits rows that should affect the sketch.
let filter_nulls = opt_filter.map(filter_to_nulls);
- let value_nulls = values[0].logical_nulls();
+ let value_nulls = array.logical_nulls();
let combined_nulls =
NullBuffer::union(filter_nulls.as_ref(), value_nulls.as_ref());
- H::for_each_hash(values[0].as_ref(), combined_nulls.as_ref(), |row, hash| {
- delta += groups[group_indices[row]].add_hash(hash);
- });
+ match combined_nulls {
+ None => {
+ for (row, &hash) in self.hashes.iter().enumerate() {
+ delta += self.groups[group_indices[row]].add_hash(hash);
+ }
+ }
+ Some(nulls) => {
+ for row in nulls.valid_indices() {
+ delta += self.groups[group_indices[row]].add_hash(self.hashes[row]);
+ }
+ }
+ }
self.apply_delta(delta);
Ok(())
}
@@ -750,7 +581,9 @@ impl<H: HllValueHasher> GroupsAccumulator for HllGroupsAccumulator<H> {
}
fn size(&self) -> usize {
- self.groups.capacity() * size_of::<GroupHll>() + self.allocated_bytes
+ self.groups.capacity() * size_of::<GroupHll>()
+ + self.allocated_bytes
+ + self.hashes.capacity() * size_of::<u64>()
}
}
@@ -886,6 +719,7 @@ impl AggregateUDFImpl for ApproxDistinct {
fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
let data_type = acc_args.expr_fields[0].data_type();
+ // For primitive types, use specialized accumulators for better performance.
let accumulator: Box<dyn Accumulator> = match data_type {
DataType::Boolean
| DataType::UInt8
@@ -924,11 +758,11 @@ impl AggregateUDFImpl for ApproxDistinct {
DataType::Timestamp(TimeUnit::Nanosecond, _) => {
Box::new(NumericHLLAccumulator::<TimestampNanosecondType>::new())
}
- DataType::Utf8 => Box::new(StringHLLAccumulator::<i32>::new()),
- DataType::LargeUtf8 => Box::new(StringHLLAccumulator::<i64>::new()),
- DataType::Utf8View => Box::new(StringViewHLLAccumulator::new()),
- DataType::Binary => Box::new(BinaryHLLAccumulator::<i32>::new()),
- DataType::LargeBinary => Box::new(BinaryHLLAccumulator::<i64>::new()),
+ DataType::Utf8
+ | DataType::LargeUtf8
+ | DataType::Utf8View
+ | DataType::Binary
+ | DataType::LargeBinary => Box::new(HLLAccumulator::new()),
DataType::Null => {
Box::new(NoopAccumulator::new(ScalarValue::UInt64(Some(0))))
}
@@ -950,71 +784,13 @@ impl AggregateUDFImpl for ApproxDistinct {
args: AccumulatorArgs,
) -> Result<Box<dyn GroupsAccumulator>> {
let data_type = args.expr_fields[0].data_type();
- let accumulator: Box<dyn GroupsAccumulator> = match data_type {
- DataType::UInt32 => {
- Box::new(HllGroupsAccumulator::<NumericHasher<UInt32Type>>::new())
- }
- DataType::UInt64 => {
- Box::new(HllGroupsAccumulator::<NumericHasher<UInt64Type>>::new())
- }
- DataType::Int32 => {
- Box::new(HllGroupsAccumulator::<NumericHasher<Int32Type>>::new())
- }
- DataType::Int64 => {
- Box::new(HllGroupsAccumulator::<NumericHasher<Int64Type>>::new())
- }
- DataType::Date32 => {
- Box::new(HllGroupsAccumulator::<NumericHasher<Date32Type>>::new())
- }
- DataType::Date64 => {
- Box::new(HllGroupsAccumulator::<NumericHasher<Date64Type>>::new())
- }
- DataType::Time32(TimeUnit::Second) => {
- Box::new(HllGroupsAccumulator::<NumericHasher<Time32SecondType>>::new())
- }
- DataType::Time32(TimeUnit::Millisecond) => Box::new(HllGroupsAccumulator::<
- NumericHasher<Time32MillisecondType>,
- >::new()),
- DataType::Time64(TimeUnit::Microsecond) => Box::new(HllGroupsAccumulator::<
- NumericHasher<Time64MicrosecondType>,
- >::new()),
- DataType::Time64(TimeUnit::Nanosecond) => Box::new(HllGroupsAccumulator::<
- NumericHasher<Time64NanosecondType>,
- >::new()),
- DataType::Timestamp(TimeUnit::Second, _) => Box::new(HllGroupsAccumulator::<
- NumericHasher<TimestampSecondType>,
- >::new()),
- DataType::Timestamp(TimeUnit::Millisecond, _) => {
- Box::new(HllGroupsAccumulator::<
- NumericHasher<TimestampMillisecondType>,
- >::new())
- }
- DataType::Timestamp(TimeUnit::Microsecond, _) => {
- Box::new(HllGroupsAccumulator::<
- NumericHasher<TimestampMicrosecondType>,
- >::new())
- }
- DataType::Timestamp(TimeUnit::Nanosecond, _) => Box::new(
- HllGroupsAccumulator::<NumericHasher<TimestampNanosecondType>>::new(),
- ),
- DataType::Utf8 => Box::new(HllGroupsAccumulator::<Utf8Hasher<i32>>::new()),
- DataType::LargeUtf8 => {
- Box::new(HllGroupsAccumulator::<Utf8Hasher<i64>>::new())
- }
- DataType::Utf8View => Box::new(HllGroupsAccumulator::<Utf8ViewHasher>::new()),
- DataType::Binary => {
- Box::new(HllGroupsAccumulator::<BinaryHasher<i32>>::new())
- }
- DataType::LargeBinary => {
- Box::new(HllGroupsAccumulator::<BinaryHasher<i64>>::new())
- }
- other => {
- return not_impl_err!(
- "GroupsAccumulator for 'approx_distinct' is not implemented for data type {other}"
- );
- }
- };
- Ok(accumulator)
+ if is_hll_groups_type(data_type) {
+ Ok(Box::new(HllGroupsAccumulator::new()))
+ } else {
+ not_impl_err!(
+ "GroupsAccumulator for 'approx_distinct' is not implemented for data type {data_type}"
+ )
+ }
}
fn documentation(&self) -> Option<&Documentation> {
@@ -1054,6 +830,7 @@ fn is_hll_groups_type(data_type: &DataType) -> bool {
mod tests {
use super::*;
use arrow::array::{AsArray, Int64Array, StringViewArray};
+ use std::hash::BuildHasher;
use std::sync::Arc;
// A string longer than the 12-byte inline limit
@@ -1079,7 +856,7 @@ mod tests {
buf
}
- fn distinct_count(acc: &mut StringViewHLLAccumulator) -> u64 {
+ fn distinct_count(acc: &mut HLLAccumulator) -> u64 {
match acc.evaluate().unwrap() {
ScalarValue::UInt64(Some(v)) => v,
other => panic!("unexpected evaluate result: {other:?}"),
@@ -1214,7 +991,7 @@ mod tests {
let filter =
BooleanArray::from(vec![Some(true), None, Some(false), None, Some(true)]);
- let mut acc = HllGroupsAccumulator::<NumericHasher<Int64Type>>::new();
+ let mut acc = HllGroupsAccumulator::new();
// put all rows in group 0
let group_indices = vec![0usize; 5];
acc.update_batch(&[values], &group_indices, Some(&filter), 1)
@@ -1243,7 +1020,7 @@ mod tests {
assert!(!batch2.as_string_view().data_buffers().is_empty());
let group_indices = vec![0usize, 0];
- let mut acc = HllGroupsAccumulator::<Utf8ViewHasher>::new();
+ let mut acc = HllGroupsAccumulator::new();
acc.update_batch(&[batch1], &group_indices, None, 1)
.unwrap();
acc.update_batch(&[batch2], &group_indices, None, 1)
@@ -1262,7 +1039,7 @@ mod tests {
// Multiset: {"aaa" x2, "bbb", LONG}, so 3 distinct values.
let mixed: ArrayRef =
Arc::new(StringViewArray::from(vec!["aaa", "bbb", LONG, "aaa"]));
- let mut acc_single = StringViewHLLAccumulator::new();
+ let mut acc_single = HLLAccumulator::new();
acc_single.update_batch(&[mixed]).unwrap();
// Same multiset, but split so "aaa" lands in both an all-inline batch
@@ -1272,7 +1049,7 @@ mod tests {
assert!(inline_only.as_string_view().data_buffers().is_empty());
assert!(!with_buffer.as_string_view().data_buffers().is_empty());
- let mut acc_split = StringViewHLLAccumulator::new();
+ let mut acc_split = HLLAccumulator::new();
acc_split.update_batch(&[inline_only]).unwrap();
acc_split.update_batch(&[with_buffer]).unwrap();
diff --git a/datafusion/functions-aggregate/src/hyperloglog.rs b/datafusion/functions-aggregate/src/hyperloglog.rs
index 182fe15cf0..9968e5a981 100644
--- a/datafusion/functions-aggregate/src/hyperloglog.rs
+++ b/datafusion/functions-aggregate/src/hyperloglog.rs
@@ -55,14 +55,7 @@ where
phantom: PhantomData<T>,
}
-/// Fixed seed for the hashing so that values are consistent across runs
-///
-/// Note that when we later move on to have serialized HLL register binaries
-/// shared across cluster, this HLL_HASH_STATE will have to be consistent across all
-/// parties otherwise we might have corruption. So ideally for later this seed
-/// shall be part of the serialized form (or stay unchanged across versions).
-pub(crate) const HLL_HASH_STATE: foldhash::quality::FixedState =
- foldhash::quality::FixedState::with_seed(0);
+pub(crate) use datafusion_common::hash_utils::HLL_RANDOM_STATE as HLL_HASH_STATE;
impl<T> Default for HyperLogLog<T>
where
@@ -93,9 +86,8 @@ where
}
}
- /// choice of hash function: foldhash is already an dependency
- /// and it fits the requirements of being a 64bit hash with
- /// reasonable performance.
+ /// The HLL hash state is shared through `datafusion_common::hash_utils`
+ /// so sketches remain compatible across accumulators.
#[inline]
fn hash_value(&self, obj: &T) -> u64 {
HLL_HASH_STATE.hash_one(obj)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org