This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 771ae4251be [feat](fe-be) Add DataSketches HLL downsample function
(#69257)
771ae4251be is described below
commit 771ae4251bed00e04e1c06111b8717ee58be35a0
Author: nooneuse <[email protected]>
AuthorDate: Wed Oct 28 18:36:59 2026 +0800
[feat](fe-be) Add DataSketches HLL downsample function (#69257)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
Apache DataSketches HLL sketches may occupy significant storage space
when a large `lgK` is used. For historical or cold partitions, retaining
the original precision may no longer be necessary, but Doris previously
had no scalar function that could read a serialized DataSketches HLL
sketch and output another serialized sketch with a lower `lgK`.
This PR introduces the following function:
```sql
datasketches_hll_downsample(sketch, target_lg_k [, target_type])
```
The function accepts a serialized Apache DataSketches HLL sketch and
produces another serialized sketch whose `lgK` is:
```text
min(source_lg_k, target_lg_k)
```
It can be used with `INSERT OVERWRITE` to rewrite sketch columns in
historical partitions, reducing their serialized size without rebuilding
the sketches from raw upstream data.
#### Arguments
- `sketch`
- A serialized Apache DataSketches HLL sketch.
- Supported Doris types: `STRING`, `VARCHAR`, and `VARBINARY`.
- The returned value uses the same type family as the input.
- `target_lg_k`
- A constant integer between `4` and `21`.
- The function never increases the source sketch's `lgK`.
- `target_type`
- An optional constant string.
- Supported values: `HLL_4`, `HLL_6`, and `HLL_8`.
- The default value is `HLL_4`.
When the source sketch already has an `lgK` less than or equal to
`target_lg_k` and already uses the requested target type, the function
returns the original serialized bytes unchanged.
If the source `lgK` is already lower than the requested value but the
target type is different, only the HLL representation type is converted;
the source `lgK` is preserved.
`NULL` input produces `NULL`. Invalid parameters or malformed serialized
sketches return an explicit error.
> Lowering `lgK` is irreversible and reduces the accuracy available to
subsequent HLL unions and estimates. The target value should therefore
be selected according to the acceptable accuracy for each data retention
period. Dense sketches generally receive the most significant storage
reduction, while sparse LIST or SET mode sketches may see little or no
reduction.
#### Usage examples
Downsample a sketch to at most `lgK = 10` using the default `HLL_4`
representation:
```sql
SELECT datasketches_hll_downsample(sketch_column, 10)
FROM user_sketches;
```
Use `HLL_6` as the target representation:
```sql
SELECT datasketches_hll_downsample(sketch_column, 10, 'HLL_6')
FROM user_sketches;
```
Rewrite the sketch column of a historical partition while preserving the
other columns:
```sql
INSERT OVERWRITE TABLE user_sketches PARTITION(p202401)
SELECT
event_date,
dimension_id,
datasketches_hll_downsample(sketch_column, 10, 'HLL_4'),
other_column
FROM user_sketches PARTITION(p202401);
```
Check whether a partition contains at least one sketch that still
requires conversion:
```sql
SELECT 1
FROM user_sketches PARTITION(p202401)
WHERE datasketches_hll_downsample(
sketch_column,
10,
'HLL_4'
) <> sketch_column
LIMIT 1;
```
An empty result means that every non-NULL sketch in the partition
already satisfies the requested `lgK` and target type.
The converted sketches remain compatible with the existing DataSketches
HLL aggregate function:
```sql
SELECT datasketches_hll_union_agg(
datasketches_hll_downsample(sketch_column, 10)
)
FROM user_sketches;
```
### Release note
Add the `DATASKETCHES_HLL_DOWNSAMPLE` scalar function for converting
serialized Apache DataSketches HLL sketches to a configurable lower
`lgK` and optional `HLL_4`, `HLL_6`, or `HLL_8` representation.
The function supports `STRING`, `VARCHAR`, and `VARBINARY` inputs,
propagates `NULL`, never increases the source `lgK`, and returns the
original bytes when no conversion is required.
Lowering `lgK` reduces serialized storage for dense sketches at the cost
of lower estimation accuracy. The operation is irreversible. All BE
nodes should be upgraded before the function is used.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [x] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/exprs/CMakeLists.txt | 3 +
...aggregate_function_datasketches_hll_union_agg.h | 64 +-
be/src/exprs/datasketches_hll_util.cpp | 417 +++++++++
be/src/exprs/datasketches_hll_util.h | 138 +++
.../exprs/function/function_datasketches_hll.cpp | 287 ++++++
be/src/exprs/function/simple_function_factory.h | 2 +
.../function/function_datasketches_hll_test.cpp | 958 +++++++++++++++++++++
.../doris/catalog/BuiltinScalarFunctions.java | 2 +
.../expression/rules/FoldConstantRuleOnBE.java | 4 +-
.../scalar/DataSketchesHllDownsample.java | 185 ++++
.../expressions/visitor/ScalarFunctionVisitor.java | 5 +
.../scalar/DataSketchesHllDownsampleTest.java | 107 +++
.../test_datasketches_hll_downsample.out | 147 ++++
.../test_datasketches_hll_downsample_auxiliary.out | 62 ++
.../test_datasketches_hll_downsample.groovy | 169 ++++
...st_datasketches_hll_downsample_auxiliary.groovy | 94 ++
16 files changed, 2596 insertions(+), 48 deletions(-)
diff --git a/be/src/exprs/CMakeLists.txt b/be/src/exprs/CMakeLists.txt
index 3965155e4a6..8d13a698d41 100644
--- a/be/src/exprs/CMakeLists.txt
+++ b/be/src/exprs/CMakeLists.txt
@@ -56,6 +56,9 @@ pch_reuse(Exprs)
# object must shadow a never-pulled archive member, but a unity batch is
# pulled in for its siblings and the linker sees a duplicate definition.
set(EXPRS_UNITY_SKIP
+ # Keep the DataSketches template-heavy implementation out of the
registration batches.
+ ${CMAKE_CURRENT_SOURCE_DIR}/datasketches_hll_util.cpp
+ ${CMAKE_CURRENT_SOURCE_DIR}/function/function_datasketches_hll.cpp
${CMAKE_CURRENT_SOURCE_DIR}/aggregate/aggregate_function_array_agg.cpp
${CMAKE_CURRENT_SOURCE_DIR}/aggregate/aggregate_function_avg.cpp
${CMAKE_CURRENT_SOURCE_DIR}/aggregate/aggregate_function_bit.cpp
diff --git
a/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h
b/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h
index c0575dadd4c..dc13e6d8d38 100644
--- a/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h
+++ b/be/src/exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h
@@ -41,6 +41,7 @@
#include "exec/common/hash_table/hash.h"
#include "exec/common/hash_table/phmap_fwd_decl.h"
#include "exprs/aggregate/aggregate_function.h"
+#include "exprs/datasketches_hll_util.h"
#include "util/var_int.h"
template <typename T>
struct HashCRC32;
@@ -56,9 +57,9 @@ struct AggregateFunctionHllSketchData {
static constexpr uint8_t EMPTY_STATE_LOG_K = 12;
static constexpr uint8_t MIN_UNION_LOG_K = 7;
static constexpr uint8_t DEFAULT_UNION_LOG_K =
datasketches::hll_constants::MAX_LOG_K;
- using Alloc = CustomStdAllocator<uint8_t>;
- using Sketch = datasketches::hll_sketch_alloc<Alloc>;
- using Union = datasketches::hll_union_alloc<Alloc>;
+ using Alloc = datasketches_hll::Alloc;
+ using Sketch = datasketches_hll::Sketch;
+ using Union = datasketches_hll::Union;
std::optional<Union> hll_union_data;
@@ -68,18 +69,23 @@ struct AggregateFunctionHllSketchData {
if (sketch_data.is_empty()) {
return;
}
- if (!hll_union_data.has_value()) {
- hll_union_data.emplace(lg_max_k, Alloc());
- }
try {
+ if (!hll_union_data.has_value()) {
+ hll_union_data.emplace(lg_max_k, Alloc());
+ }
hll_union_data->update(sketch_data);
} catch (const doris::Exception& e) {
+ // A failed library update cannot be reused; the wrapper reclaims
its allocations
+ // without dereferencing the possibly dangling library gadget.
+ hll_union_data.reset();
throw Exception(e.code(), "Internal error happened when update HLL
sketch: {}",
e.to_string());
} catch (const std::exception& e) {
+ hll_union_data.reset();
throw Exception(ErrorCode::INTERNAL_ERROR,
"Internal error happened when update HLL sketch:
{}", e.what());
} catch (...) {
+ hll_union_data.reset();
throw Exception(ErrorCode::INTERNAL_ERROR,
"Internal error happened when update HLL sketch:
unknown exception.");
}
@@ -96,12 +102,15 @@ struct AggregateFunctionHllSketchData {
hll_union_data.emplace(lg_max_k, Alloc());
hll_union_data->update(current);
} catch (const doris::Exception& e) {
+ hll_union_data.reset();
throw Exception(e.code(), "Internal error happened when update
HLL sketch: {}",
e.to_string());
} catch (const std::exception& e) {
+ hll_union_data.reset();
throw Exception(ErrorCode::INTERNAL_ERROR,
"Internal error happened when update HLL
sketch: {}", e.what());
} catch (...) {
+ hll_union_data.reset();
throw Exception(
ErrorCode::INTERNAL_ERROR,
"Internal error happened when update HLL sketch:
unknown exception.");
@@ -140,23 +149,7 @@ struct AggregateFunctionHllSketchData {
void read(BufferReadable& buf) {
StringRef d;
buf.read_binary(d);
-
- auto cache = [&]() -> Sketch {
- try {
- return Sketch::deserialize(d.data, d.size, Alloc());
- } catch (const doris::Exception& e) {
- throw Exception(e.code(), "Failed to deserialize HLL sketch
when read: {}",
- e.to_string());
- } catch (const std::exception& e) {
- throw Exception(ErrorCode::CORRUPTION, "HLL sketch data
corrupted when read: {}",
- e.what());
- } catch (...) {
- throw Exception(ErrorCode::CORRUPTION,
- "HLL sketch data corrupted when read: unknown
exception.");
- }
- }();
-
- merge(cache);
+ merge(datasketches_hll::deserialize(d, "read"));
}
double get_result() const {
if (hll_union_data.has_value()) {
@@ -249,30 +242,7 @@ private:
const auto& src_column = assert_cast<const typename
PrimitiveTypeTraits<T>::ColumnType&,
TypeCheckOnRelease::DISABLE>(column);
StringRef value =
src_column.get_data_at(static_cast<size_t>(row_num));
- if (value.empty()) {
- throw Exception(ErrorCode::CORRUPTION,
- "HLL sketch data corrupted when add: empty
input.");
- }
-
- using Sketch = typename Data::Sketch;
- using Alloc = typename Data::Alloc;
-
- auto sketch_data = [&]() -> Sketch {
- try {
- return Sketch::deserialize(value.begin(), value.size,
Alloc());
- } catch (const doris::Exception& e) {
- throw Exception(e.code(), "Failed to deserialize HLL
sketch when add: {}",
- e.to_string());
- } catch (const std::exception& e) {
- throw Exception(ErrorCode::CORRUPTION, "HLL sketch data
corrupted when add: {}",
- e.what());
- } catch (...) {
- throw Exception(ErrorCode::CORRUPTION,
- "HLL sketch data corrupted when add:
unknown exception.");
- }
- }();
-
- data.merge(sketch_data, lg_max_k);
+ data.merge(datasketches_hll::deserialize(value, "add"), lg_max_k);
}
}
};
diff --git a/be/src/exprs/datasketches_hll_util.cpp
b/be/src/exprs/datasketches_hll_util.cpp
new file mode 100644
index 00000000000..c443b6cdaba
--- /dev/null
+++ b/be/src/exprs/datasketches_hll_util.cpp
@@ -0,0 +1,417 @@
+// 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.
+
+#include "exprs/datasketches_hll_util.h"
+
+#include <bthread/bthread.h>
+
+#include <cstring>
+#include <exception>
+#include <mutex>
+#include <new>
+#include <stdexcept>
+
+#include "common/check.h"
+#include "common/exception.h"
+#include "common/status.h"
+#include "util/debug_points.h"
+
+namespace doris::datasketches_hll {
+
+struct alignas(std::max_align_t) AllocationHeader {
+ AllocationHeader* previous;
+ AllocationHeader* next;
+ detail::AllocationOwner* owner;
+ size_t count;
+};
+
+namespace {
+
+using namespace datasketches::hll_constants;
+
+thread_local detail::AllocationOwner* pthread_scope = nullptr;
+
+bthread_key_t scope_key() {
+ static bthread_key_t key;
+ static std::once_flag once;
+ std::call_once(once, [] { DORIS_CHECK(bthread_key_create(&key, nullptr) ==
0); });
+ return key;
+}
+
+detail::AllocationOwner* current_scope() {
+ return bthread_self() == 0
+ ? pthread_scope
+ :
static_cast<detail::AllocationOwner*>(bthread_getspecific(scope_key()));
+}
+
+void set_scope(detail::AllocationOwner* scope) {
+ if (bthread_self() == 0) {
+ pthread_scope = scope;
+ } else {
+ DORIS_CHECK(bthread_setspecific(scope_key(), scope) == 0);
+ }
+}
+
+class ActiveAllocationOwner {
+public:
+ explicit ActiveAllocationOwner(detail::AllocationOwner* owner) :
_previous(current_scope()) {
+ set_scope(owner);
+ }
+ ~ActiveAllocationOwner() { set_scope(_previous); }
+
+private:
+ detail::AllocationOwner* _previous;
+};
+
+// Reuse the library's protected static header readers without constructing a
sketch or
+// duplicating the mode/type bit decoding. This abstract adapter is never
instantiated.
+struct HeaderReader : datasketches::HllSketchImpl<Alloc> {
+ using datasketches::HllSketchImpl<Alloc>::extractCurMode;
+ using datasketches::HllSketchImpl<Alloc>::extractTgtHllType;
+};
+
+uint32_t read_uint32(const uint8_t* bytes) {
+ uint32_t value;
+ std::memcpy(&value, bytes, sizeof(value));
+ return value;
+}
+
+void check_set_size(const uint8_t* bytes, size_t size, uint8_t lg_k, bool
compact) {
+ datasketches::HllUtil<Alloc>::checkMemSize(HASH_SET_INT_ARR_START, size);
+ const auto count = read_uint32(bytes + HASH_SET_COUNT_INT);
+ // SET promotes to HLL above 3/4 of its largest table (K/8 entries). Bound
the count before
+ // computeLgArrInts(), whose power-of-two and load-factor arithmetic uses
uint32_t.
+ if (lg_k <= 7 || count > (RESIZE_NUMER * (uint32_t {1} << (lg_k - 3))) /
RESIZE_DENOM) {
+ throw std::invalid_argument("Invalid HLL SET coupon count or lgK");
+ }
+ size_t items = count;
+ if (!compact) {
+ // The library evaluates 1 << lgArr before checking the payload. For
example, lgArr=32
+ // is an undefined shift. Preserve its support for older images with
an unset lgArr.
+ const uint8_t lg_arr = bytes[LG_ARR_BYTE] < LG_INIT_SET_SIZE
+ ?
datasketches::HllUtil<Alloc>::computeLgArrInts(
+ datasketches::SET, count,
lg_k)
+ : bytes[LG_ARR_BYTE];
+ if (lg_arr > lg_k - 3) {
+ throw std::invalid_argument("Invalid HLL SET array exponent");
+ }
+ items = size_t {1} << lg_arr;
+ }
+ datasketches::HllUtil<Alloc>::checkMemSize(HASH_SET_INT_ARR_START + items
* sizeof(uint32_t),
+ size);
+ if (!compact) {
+ // newSet() copies the whole table but trusts the declared count. With
ten occupied
+ // slots and count=1, a later compact serializer allocates 16 bytes
and writes 52.
+ uint32_t occupied = 0;
+ for (size_t i = 0; i < items; ++i) {
+ occupied += read_uint32(bytes + HASH_SET_INT_ARR_START + i *
sizeof(uint32_t)) != EMPTY;
+ }
+ if (occupied != count) {
+ throw std::invalid_argument("HLL SET coupon count does not match
occupied slots");
+ }
+ }
+}
+
+void check_aux_map(const uint8_t* bytes, size_t size, size_t offset, uint8_t
lg_k, uint32_t count,
+ bool compact) {
+ const uint32_t registers = uint32_t {1} << lg_k;
+ // There can be at most one auxiliary entry per register. This also bounds
the uint32_t
+ // capacity arithmetic used by the library to rebuild a compact map.
+ if (count > registers) {
+ throw std::invalid_argument("HLL_4 auxiliary count exceeds register
count");
+ }
+ size_t items = count;
+ if (!compact) {
+ const auto lg_arr = bytes[LG_ARR_BYTE];
+ // An updatable map starts at LG_AUX_ARR_INTS[lgK] and grows at 3/4
load, so at most
+ // K entries need 2*K slots. Reject bad exponents before the library's
1 << lgArr.
+ if (lg_arr < LG_AUX_ARR_INTS[lg_k] || lg_arr > lg_k + 1) {
+ throw std::invalid_argument("Invalid HLL_4 auxiliary array
exponent");
+ }
+ items = size_t {1} << lg_arr;
+ }
+ datasketches::HllUtil<Alloc>::checkMemSize(offset + items *
sizeof(uint32_t), size);
+
+ // AuxHashMap::deserialize owns a raw map until it returns. mustAdd()
throws for duplicate
+ // slots before that ownership transfers, leaking the map. Check the same
masked slot keys
+ // before allocating it. This bitmap is needed only for sketches with an
auxiliary map.
+ DorisVector<uint8_t> seen(registers / 8, 0);
+ for (size_t i = 0; i < items; ++i) {
+ const auto pair = read_uint32(bytes + offset + i * sizeof(uint32_t));
+ if (pair == EMPTY) {
+ // Compact maps contain only real entries. Treating EMPTY as slot
0 would let
+ // an AUX_TOKEN at register 0 pass this check, but
mustFindValueFor(0) still fails.
+ if (compact) {
+ throw std::invalid_argument("HLL_4 compact auxiliary map
contains an EMPTY entry");
+ }
+ continue;
+ }
+ const auto slot = datasketches::HllUtil<Alloc>::getLow26(pair) &
(registers - 1);
+ const auto bit = static_cast<uint8_t>(1U << (slot % 8));
+ if (seen[slot / 8] & bit) {
+ throw std::invalid_argument("HLL_4 auxiliary map contains
duplicate slots");
+ }
+ seen[slot / 8] |= bit;
+ }
+ // A nonempty map is insufficient: every AUX_TOKEN must refer to a slot in
that map.
+ // For example, token at register 0 with only slot 3 in the map makes
mergeHll() throw
+ // after allocating its target array. Reject that image even on the
unchanged-byte path.
+ for (uint32_t slot = 0; slot < registers; ++slot) {
+ const auto byte = bytes[HLL_BYTE_ARR_START + slot / 2];
+ const auto value = static_cast<uint8_t>((byte >> ((slot % 2) * 4)) &
loNibbleMask);
+ if (value == AUX_TOKEN && !(seen[slot / 8] & (1U << (slot % 8)))) {
+ throw std::invalid_argument("HLL_4 AUX_TOKEN has no matching
auxiliary slot");
+ }
+ }
+}
+
+bool check_hll_size(const uint8_t* bytes, size_t size, uint8_t lg_k, bool
compact) {
+ datasketches::HllUtil<Alloc>::checkMemSize(HLL_BYTE_ARR_START, size);
+ const auto type = HeaderReader::extractTgtHllType(bytes[MODE_BYTE]);
+ const size_t offset =
+ HLL_BYTE_ARR_START +
datasketches::HllArray<Alloc>::hllArrBytes(type, lg_k);
+ datasketches::HllUtil<Alloc>::checkMemSize(offset, size);
+ const auto aux_count = read_uint32(bytes + AUX_COUNT_INT);
+ if (type != datasketches::HLL_4) {
+ // newHll() installs any nonempty auxiliary map through an unchecked
Hll4Array cast.
+ // HLL_6/HLL_8 have no map member, so accepting such an image writes
outside the object.
+ if (aux_count != 0) {
+ throw std::invalid_argument("HLL_6/HLL_8 cannot contain an
auxiliary map");
+ }
+ return false;
+ }
+ if (aux_count != 0) {
+ check_aux_map(bytes, size, offset, lg_k, aux_count, compact);
+ return true;
+ }
+ // HLL_4 uses nibble 0xf to reference its auxiliary map. The library
accepts auxCount=0
+ // with such a nibble, then dereferences a null map during
union/conversion, even on a no-op.
+ for (size_t i = HLL_BYTE_ARR_START; i < offset; ++i) {
+ if ((bytes[i] & loNibbleMask) == AUX_TOKEN ||
+ ((bytes[i] >> 4) & loNibbleMask) == AUX_TOKEN) {
+ throw std::invalid_argument("HLL_4 contains AUX_TOKEN without an
auxiliary map");
+ }
+ }
+ return false;
+}
+
+} // namespace
+
+AllocationScope::AllocationScope()
+ : _previous(current_scope()), _exceptions(std::uncaught_exceptions()) {
+ set_scope(&_owner);
+}
+
+AllocationScope::~AllocationScope() {
+ set_scope(_previous);
+ if (std::uncaught_exceptions() == _exceptions) {
+ _owner.release();
+ }
+}
+
+detail::AllocationOwner::~AllocationOwner() {
+ clear();
+}
+
+void detail::AllocationOwner::clear() noexcept {
+ while (_head != nullptr) {
+ auto* header = _head;
+ _head = header->next;
+ CustomStdAllocator<AllocationHeader>().deallocate(header,
header->count);
+ }
+}
+
+void detail::AllocationOwner::release() noexcept {
+ while (_head != nullptr) {
+ auto* header = _head;
+ _head = header->next;
+ // Returned sketches keep normal library ownership after a successful
call.
+ header->owner = nullptr;
+ }
+}
+
+void* allocate(size_t bytes) {
+ DBUG_EXECUTE_IF("datasketches_hll.before_allocate", { DBUG_RUN_CALLBACK();
});
+ auto* scope = current_scope();
+ const size_t count =
+ 1 + bytes / sizeof(AllocationHeader) + (bytes %
sizeof(AllocationHeader) != 0);
+ auto* header = CustomStdAllocator<AllocationHeader>().allocate(count);
+ *header = {.previous = nullptr,
+ .next = scope == nullptr ? nullptr : scope->_head,
+ .owner = scope,
+ .count = count};
+ if (scope != nullptr) {
+ if (header->next != nullptr) {
+ header->next->previous = header;
+ }
+ scope->_head = header;
+ }
+ return header + 1;
+}
+
+void deallocate(void* ptr) noexcept {
+ auto* header = static_cast<AllocationHeader*>(ptr) - 1;
+ if (header->owner != nullptr) {
+ if (header->previous != nullptr) {
+ header->previous->next = header->next;
+ } else {
+ header->owner->_head = header->next;
+ }
+ if (header->next != nullptr) {
+ header->next->previous = header->previous;
+ }
+ }
+ CustomStdAllocator<AllocationHeader>().deallocate(header, header->count);
+}
+
+struct Union::Impl {
+ using LibraryUnion = datasketches::hll_union_alloc<Alloc>;
+
+ detail::AllocationOwner allocations;
+ alignas(LibraryUnion) std::byte storage[sizeof(LibraryUnion)];
+ bool valid = false;
+
+ LibraryUnion* get() { return
std::launder(reinterpret_cast<LibraryUnion*>(storage)); }
+ const LibraryUnion* get() const {
+ return std::launder(reinterpret_cast<const LibraryUnion*>(storage));
+ }
+
+ ~Impl() {
+ if (valid) {
+ get()->~LibraryUnion();
+ }
+ // Failed library updates can leave a dangling gadget pointer.
AllocationOwner frees
+ // the remaining blocks directly, instead of dereferencing that
invalid gadget.
+ }
+};
+
+void Union::Deleter::operator()(Impl* ptr) const noexcept {
+ ptr->~Impl();
+ datasketches_hll::deallocate(ptr);
+}
+
+Union::Union(uint8_t lg_k, const Alloc& allocator)
+ : _impl(new(datasketches_hll::allocate(sizeof(Impl))) Impl()) {
+ ActiveAllocationOwner scope(&_impl->allocations);
+ new (_impl->storage) Impl::LibraryUnion(lg_k, allocator);
+ _impl->valid = true;
+}
+
+Union::~Union() = default;
+Union::Union(Union&&) noexcept = default;
+Union& Union::operator=(Union&&) noexcept = default;
+
+Union::Union(const Union& other) :
_impl(new(datasketches_hll::allocate(sizeof(Impl))) Impl()) {
+ DORIS_CHECK(other._impl->valid);
+ ActiveAllocationOwner scope(&_impl->allocations);
+ new (_impl->storage) Impl::LibraryUnion(*other._impl->get());
+ _impl->valid = true;
+}
+
+Union& Union::operator=(const Union& other) {
+ Union copy(other);
+ *this = std::move(copy);
+ return *this;
+}
+
+void Union::update(const Sketch& sketch) {
+ DORIS_CHECK(_impl->valid);
+ ActiveAllocationOwner scope(&_impl->allocations);
+ try {
+ _impl->get()->update(sketch);
+ } catch (...) {
+ // The library can free its gadget during promotion and then throw
before replacing
+ // the pointer. This state must be discarded; its allocation owner
remains intact.
+ _impl->valid = false;
+ throw;
+ }
+}
+
+Sketch Union::get_result(datasketches::target_hll_type type) const {
+ DORIS_CHECK(_impl->valid);
+ AllocationScope scope;
+ return _impl->get()->get_result(type);
+}
+
+double Union::get_estimate() const {
+ DORIS_CHECK(_impl->valid);
+ return _impl->get()->get_estimate();
+}
+
+double Union::get_lower_bound(uint8_t std_devs) const {
+ DORIS_CHECK(_impl->valid);
+ return _impl->get()->get_lower_bound(std_devs);
+}
+
+double Union::get_upper_bound(uint8_t std_devs) const {
+ DORIS_CHECK(_impl->valid);
+ return _impl->get()->get_upper_bound(std_devs);
+}
+
+uint8_t Union::get_lg_config_k() const {
+ DORIS_CHECK(_impl->valid);
+ return _impl->get()->get_lg_config_k();
+}
+
+double Union::get_rel_err(bool upper, bool unioned, uint8_t lg_k, uint8_t
std_devs) {
+ return Impl::LibraryUnion::get_rel_err(upper, unioned, lg_k, std_devs);
+}
+
+Sketch deserialize(const StringRef& value, std::string_view action) {
+ AllocationScope allocations;
+ // The library reads the preamble byte before it checks the length.
+ if (value.size == 0) {
+ throw Exception(ErrorCode::CORRUPTION, "HLL sketch data corrupted when
{}: empty input.",
+ action);
+ }
+ try {
+ // All modes have an eight-byte preamble. An invalid lgK would
otherwise reach shifts used
+ // to size the register/coupon arrays before the library validates it.
+ datasketches::HllUtil<Alloc>::checkMemSize(EMPTY_SKETCH_SIZE_BYTES,
value.size);
+ const auto* bytes = reinterpret_cast<const uint8_t*>(value.data);
+ const auto lg_k =
datasketches::HllUtil<Alloc>::checkLgK(bytes[LG_K_BYTE]);
+ const auto mode = HeaderReader::extractCurMode(bytes[MODE_BYTE]);
+ const bool compact = (bytes[FLAGS_BYTE] & COMPACT_FLAG_MASK) != 0;
+ bool has_aux_map = false;
+ if (mode == datasketches::SET) {
+ check_set_size(bytes, value.size, lg_k, compact);
+ } else if (mode == datasketches::HLL) {
+ has_aux_map = check_hll_size(bytes, value.size, lg_k, compact);
+ }
+ // AuxHashMap::deserialize reads uint32_t directly. STRING rows have
no alignment
+ // guarantee: a 233-byte HLL_6 before this HLL_4 leaves its auxiliary
entries unaligned.
+ // Only this library reader needs an aligned copy; the other payloads
use memcpy.
+ if (has_aux_map && reinterpret_cast<uintptr_t>(value.data) %
alignof(uint32_t) != 0) {
+ DorisVector<uint32_t> aligned(value.size / sizeof(uint32_t) +
+ (value.size % sizeof(uint32_t) !=
0));
+ std::memcpy(aligned.data(), value.data, value.size);
+ return Sketch::deserialize(aligned.data(), value.size, Alloc());
+ }
+ return Sketch::deserialize(value.data, value.size, Alloc());
+ } catch (const doris::Exception& e) {
+ throw Exception(e.code(), "Failed to deserialize HLL sketch when {}:
{}", action,
+ e.to_string());
+ } catch (const std::exception& e) {
+ throw Exception(ErrorCode::CORRUPTION, "HLL sketch data corrupted when
{}: {}", action,
+ e.what());
+ } catch (...) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "HLL sketch data corrupted when {}: unknown
exception.", action);
+ }
+}
+
+} // namespace doris::datasketches_hll
diff --git a/be/src/exprs/datasketches_hll_util.h
b/be/src/exprs/datasketches_hll_util.h
new file mode 100644
index 00000000000..f683e31d1a0
--- /dev/null
+++ b/be/src/exprs/datasketches_hll_util.h
@@ -0,0 +1,138 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <hll.hpp>
+#include <memory>
+#include <string_view>
+
+#include "core/custom_allocator.h"
+#include "core/string_ref.h"
+
+namespace doris::datasketches_hll {
+
+struct AllocationHeader;
+void* allocate(size_t bytes);
+void deallocate(void* ptr) noexcept;
+
+namespace detail {
+class AllocationOwner {
+public:
+ AllocationOwner() = default;
+ ~AllocationOwner();
+ AllocationOwner(const AllocationOwner&) = delete;
+ AllocationOwner& operator=(const AllocationOwner&) = delete;
+ void clear() noexcept;
+ void release() noexcept;
+
+private:
+ friend void* datasketches_hll::allocate(size_t bytes);
+ friend void datasketches_hll::deallocate(void* ptr) noexcept;
+ AllocationHeader* _head = nullptr;
+};
+} // namespace detail
+
+// DataSketches uses allocator.allocate() + placement new in its factories.
When a constructor
+// or a later merge throws, some allocations never reach an owning sketch.
Declare this scope
+// before local sketches, and destroy/reset any existing union before an
exception leaves it.
+// Normal destructors run first; the scope then frees only the allocations
they did not release.
+class AllocationScope {
+public:
+ AllocationScope();
+ ~AllocationScope();
+
+ AllocationScope(const AllocationScope&) = delete;
+ AllocationScope& operator=(const AllocationScope&) = delete;
+
+private:
+ detail::AllocationOwner _owner;
+ detail::AllocationOwner* _previous;
+ int _exceptions;
+};
+
+// Stateless across copies and rebinds. The current call's scope records the
allocation, so a
+// union can copy registers from a sketch created in an earlier scope or on
another thread.
+template <typename T>
+class SketchAllocator {
+public:
+ using value_type = T;
+ SketchAllocator() = default;
+ template <typename U>
+ SketchAllocator(const SketchAllocator<U>&) noexcept {}
+
+ T* allocate(size_t count) {
+ static_assert(alignof(T) <= alignof(std::max_align_t));
+ return static_cast<T*>(datasketches_hll::allocate(count * sizeof(T)));
+ }
+ void deallocate(T* ptr, size_t) noexcept {
datasketches_hll::deallocate(ptr); }
+
+ template <typename U>
+ bool operator==(const SketchAllocator<U>&) const noexcept {
+ return true;
+ }
+ template <typename U>
+ bool operator!=(const SketchAllocator<U>&) const noexcept {
+ return false;
+ }
+};
+
+// Shared by the DataSketches HLL aggregate and scalar functions so that both
track memory through
+// Doris allocators and report broken input the same way.
+using Alloc = SketchAllocator<uint8_t>;
+using Sketch = datasketches::hll_sketch_alloc<Alloc>;
+
+// The library can free its old gadget during LIST/SET promotion before
installing the new
+// pointer. If a subsequent allocation fails, that union is not safe to
destroy. Keep its
+// allocations under Doris ownership and discard the failed state without
invoking that
+// invalid library destructor. The public sketch format and estimates remain
unchanged.
+class Union {
+public:
+ explicit Union(uint8_t lg_k, const Alloc& allocator = Alloc());
+ ~Union();
+ Union(Union&&) noexcept;
+ Union& operator=(Union&&) noexcept;
+ Union(const Union&);
+ Union& operator=(const Union&);
+
+ void update(const Sketch& sketch);
+ Sketch get_result(datasketches::target_hll_type type =
datasketches::HLL_4) const;
+ double get_estimate() const;
+ double get_lower_bound(uint8_t std_devs) const;
+ double get_upper_bound(uint8_t std_devs) const;
+ uint8_t get_lg_config_k() const;
+ static double get_rel_err(bool upper, bool unioned, uint8_t lg_k, uint8_t
std_devs);
+
+private:
+ struct Impl;
+ struct Deleter {
+ void operator()(Impl* ptr) const noexcept;
+ };
+ std::unique_ptr<Impl, Deleter> _impl;
+};
+
+// Deserializes a serialized DataSketches HLL sketch. `action` names the
caller in the error
+// message, e.g. "add" or "read".
+//
+// Empty bytes are rejected before the library reads them. A doris::Exception
keeps its code, so
+// MEM_ALLOC_FAILED raised by the allocator is not reported as corrupted data;
any other failure of
+// the library means the bytes are not a valid sketch and becomes CORRUPTION.
+Sketch deserialize(const StringRef& value, std::string_view action);
+
+} // namespace doris::datasketches_hll
diff --git a/be/src/exprs/function/function_datasketches_hll.cpp
b/be/src/exprs/function/function_datasketches_hll.cpp
new file mode 100644
index 00000000000..f44dedd5e5d
--- /dev/null
+++ b/be/src/exprs/function/function_datasketches_hll.cpp
@@ -0,0 +1,287 @@
+// 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.
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <exception>
+#include <hll.hpp>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "common/check.h"
+#include "common/exception.h"
+#include "common/status.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/block/column_numbers.h"
+#include "core/block/column_with_type_and_name.h"
+#include "core/column/column.h"
+#include "core/column/column_const.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_varbinary.h"
+#include "core/string_ref.h"
+#include "exprs/datasketches_hll_util.h"
+#include "exprs/function/function.h"
+#include "exprs/function/simple_function_factory.h"
+#include "exprs/function_context.h"
+#include "util/string_util.h"
+
+namespace doris {
+
+// datasketches_hll_downsample(sketch, lg_k [, tgt_type]) rewrites a
serialized DataSketches HLL
+// sketch so that its lgK is min(lg_k, source lgK) and its HLL type is
tgt_type (HLL_4 by default).
+//
+// A sketch that already has lgK <= lg_k and type tgt_type is returned byte
for byte after it is
+// fully deserialized. Callers rely on this: `datasketches_hll_downsample(sk,
k, t) <> sk` is true
+// exactly when sk's lgK is above k or its type differs from t, which tells
whether a row still
+// needs to be rewritten. For example, with lg_k = 12 and tgt_type = HLL_4:
+// lgK 14 HLL_8 -> lgK 12 HLL_4 (new bytes)
+// lgK 10 HLL_8 -> lgK 10 HLL_4 (new bytes, only the type changes)
+// lgK 10 HLL_4 -> the input bytes
+class FunctionDataSketchesHllDownsample : public IFunction {
+public:
+ static constexpr auto name = "datasketches_hll_downsample";
+
+ static FunctionPtr create() { return
std::make_shared<FunctionDataSketchesHllDownsample>(); }
+
+ String get_name() const override { return name; }
+
+ size_t get_number_of_arguments() const override { return 0; }
+
+ bool is_variadic() const override { return true; }
+
+ DataTypePtr get_return_type_impl(const DataTypes& arguments) const
override {
+ // use_default_implementation_for_nulls() is disabled below, so,
unlike the usual
+ // get_return_type_impl contract, arguments here keep their original
nullability and the
+ // wrapping is not automatic: replicate it by hand from every
argument, as FE does for a
+ // PropagateNullable function.
+ const bool nullable = std::ranges::any_of(
+ arguments, [](const DataTypePtr& type) { return
type->is_nullable(); });
+ DataTypePtr base = remove_nullable(arguments[0]);
+ return nullable ? make_nullable(base) : base;
+ }
+
+ // lg_k and tgt_type are validated before the NULL check of the sketch, so
the generic
+ // nullable-argument shortcut must not return NULL for, e.g.,
downsample(NULL, 30) first.
+ bool use_default_implementation_for_nulls() const override { return false;
}
+
+ Status open(FunctionContext* context, FunctionContext::FunctionStateScope
scope) override {
+ if (scope == FunctionContext::FunctionStateScope::THREAD_LOCAL) {
+ return Status::OK();
+ }
+ auto state = std::make_shared<DownsampleState>();
+ // FE folds the arguments to literals when it can. A constant FE
cannot fold, e.g.
+ // `10 + crc32('') % 2`, may not be evaluated in open either, then it
is read from the
+ // first row in execute.
+ if (context->is_col_constant(1)) {
+
RETURN_IF_ERROR(parse_lg_k(*context->get_constant_col(1)->column_ptr,
state->lg_k));
+ }
+ if (context->get_num_args() == 2) {
+ state->tgt_type = datasketches::HLL_4;
+ } else if (context->is_col_constant(2)) {
+ RETURN_IF_ERROR(
+ parse_tgt_type(*context->get_constant_col(2)->column_ptr,
state->tgt_type));
+ }
+ context->set_function_state(scope, state);
+ return Status::OK();
+ }
+
+ Status execute_impl(FunctionContext* context, Block& block, const
ColumnNumbers& arguments,
+ uint32_t result, size_t input_rows_count) const
override {
+ const auto* state = reinterpret_cast<const DownsampleState*>(
+
context->get_function_state(FunctionContext::FunctionStateScope::FRAGMENT_LOCAL));
+ DORIS_CHECK(state != nullptr);
+ uint8_t lg_k = 0;
+ if (state->lg_k.has_value()) {
+ lg_k = *state->lg_k;
+ } else {
+
RETURN_IF_ERROR(read_lg_k(*block.get_by_position(arguments[1]).column, lg_k));
+ }
+ datasketches::target_hll_type tgt_type = datasketches::HLL_4;
+ if (state->tgt_type.has_value()) {
+ tgt_type = *state->tgt_type;
+ } else {
+
RETURN_IF_ERROR(read_tgt_type(*block.get_by_position(arguments[2]).column,
tgt_type));
+ }
+
+ const ColumnWithTypeAndName& sketch_arg =
block.get_by_position(arguments[0]);
+ if (sketch_arg.type->is_nullable() &&
sketch_arg.get_nullable_column_info().only_null) {
+ block.get_by_position(result).column =
+
block.get_by_position(result).type->create_column_const(input_rows_count,
+
Field());
+ return Status::OK();
+ }
+ const auto [sketch_column, sketch_const] =
unpack_if_const(sketch_arg.column);
+ const NullMap* null_map = nullptr;
+ ColumnPtr sketch_values = sketch_column;
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(sketch_column.get())) {
+ null_map = &nullable->get_null_map_data();
+ sketch_values = nullable->get_nested_column_ptr();
+ }
+
+ MutableColumnPtr res;
+ if (const auto* strings =
check_and_get_column<ColumnString>(sketch_values.get())) {
+ res = execute_column(*strings, null_map, sketch_const,
input_rows_count, lg_k,
+ tgt_type);
+ } else if (const auto* binaries =
+
check_and_get_column<ColumnVarbinary>(sketch_values.get())) {
+ res = execute_column(*binaries, null_map, sketch_const,
input_rows_count, lg_k,
+ tgt_type);
+ } else {
+ return Status::RuntimeError("Illegal column {} of argument of
function {}",
+ sketch_arg.column->get_name(),
get_name());
+ }
+
+ if (!block.get_by_position(result).type->is_nullable()) {
+ block.replace_by_position(result, std::move(res));
+ return Status::OK();
+ }
+ if (sketch_const) {
+ // wrap_in_nullable skips constant arguments; a constant sketch
here is not NULL, since
+ // an all-NULL sketch returned above.
+ block.replace_by_position(
+ result, ColumnNullable::create(std::move(res),
+
ColumnUInt8::create(input_rows_count, 0)));
+ return Status::OK();
+ }
+ block.replace_by_position(result,
wrap_in_nullable(ColumnPtr(std::move(res)), block,
+ arguments,
input_rows_count));
+ return Status::OK();
+ }
+
+private:
+ // Same bounds as the DataSketches library accepts for a sketch.
+ static constexpr int64_t MIN_LG_K = datasketches::hll_constants::MIN_LOG_K;
+ static constexpr int64_t MAX_LG_K = datasketches::hll_constants::MAX_LOG_K;
+
+ struct DownsampleState {
+ // Empty when the argument is not evaluated in open; it is then read
in execute.
+ std::optional<uint8_t> lg_k;
+ std::optional<datasketches::target_hll_type> tgt_type;
+ };
+
+ static Status parse_lg_k(const IColumn& column, std::optional<uint8_t>&
lg_k) {
+ uint8_t value = 0;
+ RETURN_IF_ERROR(read_lg_k(column, value));
+ lg_k = value;
+ return Status::OK();
+ }
+
+ static Status parse_tgt_type(const IColumn& column,
+ std::optional<datasketches::target_hll_type>&
tgt_type) {
+ datasketches::target_hll_type value = datasketches::HLL_4;
+ RETURN_IF_ERROR(read_tgt_type(column, value));
+ tgt_type = value;
+ return Status::OK();
+ }
+
+ // Reads lg_k from the first row. FE rejects a literal outside the range
or NULL, so these
+ // errors are only reachable through a constant evaluated by BE.
+ static Status read_lg_k(const IColumn& column, uint8_t& lg_k) {
+ if (column.is_null_at(0)) {
+ return Status::InvalidArgument("{}'s lg_k must be between {} and
{}, but was NULL",
+ name, MIN_LG_K, MAX_LG_K);
+ }
+ const int64_t value = remove_nullable(column.get_ptr())->get_int(0);
+ if (value < MIN_LG_K || value > MAX_LG_K) {
+ return Status::InvalidArgument("{}'s lg_k must be between {} and
{}, but was {}", name,
+ MIN_LG_K, MAX_LG_K, value);
+ }
+ lg_k = static_cast<uint8_t>(value);
+ return Status::OK();
+ }
+
+ static Status read_tgt_type(const IColumn& column,
datasketches::target_hll_type& tgt_type) {
+ if (column.is_null_at(0)) {
+ return Status::InvalidArgument(
+ "{}'s tgt_type must be one of HLL_4, HLL_6, HLL_8, but was
NULL", name);
+ }
+ const std::string value =
remove_nullable(column.get_ptr())->get_data_at(0).to_string();
+ const std::string lower = to_lower(value);
+ if (lower == "hll_4") {
+ tgt_type = datasketches::HLL_4;
+ } else if (lower == "hll_6") {
+ tgt_type = datasketches::HLL_6;
+ } else if (lower == "hll_8") {
+ tgt_type = datasketches::HLL_8;
+ } else {
+ return Status::InvalidArgument(
+ "{}'s tgt_type must be one of HLL_4, HLL_6, HLL_8, but was
'{}'", name, value);
+ }
+ return Status::OK();
+ }
+
+ template <typename ColumnType>
+ static MutableColumnPtr execute_column(const ColumnType& sketches, const
NullMap* null_map,
+ bool sketch_const, size_t
input_rows_count, uint8_t lg_k,
+ datasketches::target_hll_type
tgt_type) {
+ auto res = ColumnType::create();
+ for (size_t i = 0; i < input_rows_count; ++i) {
+ const size_t row = index_check_const(i, sketch_const);
+ if (null_map != nullptr && (*null_map)[row]) {
+ res->insert_default();
+ continue;
+ }
+ downsample(sketches.get_data_at(row), lg_k, tgt_type, *res);
+ }
+ return res;
+ }
+
+ template <typename ColumnType>
+ static void downsample(const StringRef& value, uint8_t lg_k,
+ datasketches::target_hll_type tgt_type, ColumnType&
res) {
+ const datasketches_hll::Sketch sketch =
datasketches_hll::deserialize(value, "downsample");
+ if (sketch.get_lg_config_k() <= lg_k && sketch.get_target_type() ==
tgt_type) {
+ res.insert_data(value.data, value.size);
+ return;
+ }
+ try {
+ // A union capped at min(lg_k, source lgK) folds the registers of
a dense sketch and
+ // replays the coupons of a LIST/SET sketch, without ever raising
the source lgK.
+ datasketches_hll::Union hll_union(std::min(lg_k,
sketch.get_lg_config_k()),
+ datasketches_hll::Alloc());
+ hll_union.update(sketch);
+ const auto bytes =
hll_union.get_result(tgt_type).serialize_compact();
+ res.insert_data(reinterpret_cast<const char*>(bytes.data()),
bytes.size());
+ } catch (const doris::Exception& e) {
+ throw Exception(e.code(), "Internal error happened when downsample
HLL sketch: {}",
+ e.to_string());
+ } catch (const std::exception& e) {
+ throw Exception(ErrorCode::INTERNAL_ERROR,
+ "Internal error happened when downsample HLL
sketch: {}", e.what());
+ } catch (...) {
+ throw Exception(
+ ErrorCode::INTERNAL_ERROR,
+ "Internal error happened when downsample HLL sketch:
unknown exception.");
+ }
+ }
+};
+
+void register_function_datasketches_hll(SimpleFunctionFactory& factory) {
+ factory.register_function<FunctionDataSketchesHllDownsample>();
+}
+
+} // namespace doris
diff --git a/be/src/exprs/function/simple_function_factory.h
b/be/src/exprs/function/simple_function_factory.h
index 60d5bb14827..075135eccc9 100644
--- a/be/src/exprs/function/simple_function_factory.h
+++ b/be/src/exprs/function/simple_function_factory.h
@@ -35,6 +35,7 @@ void register_function_size(SimpleFunctionFactory& factory);
void register_function_comparison(SimpleFunctionFactory& factory);
void register_function_comparison_eq_for_null(SimpleFunctionFactory& factory);
void register_function_hll(SimpleFunctionFactory& factory);
+void register_function_datasketches_hll(SimpleFunctionFactory& factory);
void register_function_logical(SimpleFunctionFactory& factory);
void register_function_cast(SimpleFunctionFactory& factory);
void register_function_encode_varchar(SimpleFunctionFactory& factory);
@@ -284,6 +285,7 @@ public:
register_function_quantile_state(instance);
register_function_bitmap_variadic(instance);
register_function_hll(instance);
+ register_function_datasketches_hll(instance);
register_function_comparison(instance);
register_function_encode_varchar(instance);
register_function_decode_as_varchar(instance);
diff --git a/be/test/exprs/function/function_datasketches_hll_test.cpp
b/be/test/exprs/function/function_datasketches_hll_test.cpp
new file mode 100644
index 00000000000..9e29a423ad5
--- /dev/null
+++ b/be/test/exprs/function/function_datasketches_hll_test.cpp
@@ -0,0 +1,958 @@
+// 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.
+
+#include <bthread/bthread.h>
+#include <gtest/gtest.h>
+
+#include <algorithm>
+#include <array>
+#include <cmath>
+#include <cstring>
+#include <functional>
+#include <hll.hpp>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "common/config.h"
+#include "common/exception.h"
+#include "common/status.h"
+#include "core/block/block.h"
+#include "core/column/column_const.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_varbinary.h"
+#include "core/string_buffer.hpp"
+#include "exprs/aggregate/aggregate_function_datasketches_hll_union_agg.h"
+#include "exprs/datasketches_hll_util.h"
+#include "exprs/function/function.h"
+#include "exprs/function/simple_function_factory.h"
+#include "exprs/function_context.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
+#include "testutil/function_utils.h"
+#include "util/debug_points.h"
+
+namespace doris {
+
+namespace {
+
+class ScopedSketchAllocationFailure {
+public:
+ explicit ScopedSketchAllocationFailure(size_t fail_at)
+ : _old_enabled(config::enable_debug_points) {
+ config::enable_debug_points = true;
+ DebugPoints::instance()->add_with_callback(
+ "datasketches_hll.before_allocate",
std::function<void()>([this, fail_at] {
+ ++calls;
+ if (calls == fail_at) {
+ throw Exception(ErrorCode::MEM_ALLOC_FAILED,
+ "Injected HLL allocation failure");
+ }
+ }));
+ }
+
+ ~ScopedSketchAllocationFailure() {
+ DebugPoints::instance()->remove("datasketches_hll.before_allocate");
+ config::enable_debug_points = _old_enabled;
+ }
+
+ size_t calls = 0;
+
+private:
+ bool _old_enabled;
+};
+
+} // namespace
+
+class FunctionDataSketchesHllDownsampleTest : public testing::Test {
+protected:
+ static void put_uint32(std::string& bytes, size_t offset, uint32_t value) {
+ std::memcpy(bytes.data() + offset, &value, sizeof(value));
+ }
+
+ static std::string aux_sketch_bytes(int lg_k, bool compact) {
+ // Force genuine HLL_4 exceptions without searching for extremely rare
hash values.
+ datasketches::Hll4Array<std::allocator<uint8_t>> sketch(lg_k, true,
{});
+ sketch.couponUpdate(datasketches::HllUtil<>::pair(1, 20));
+ sketch.couponUpdate(datasketches::HllUtil<>::pair(3, 21));
+ auto bytes = sketch.serialize(compact, 0);
+ return {reinterpret_cast<const char*>(bytes.data()), bytes.size()};
+ }
+
+ static std::string missing_aux_slot(bool compact) {
+ using namespace datasketches::hll_constants;
+ auto bytes = aux_sketch_bytes(8, compact);
+ const size_t offset = HLL_BYTE_ARR_START + (1 << 7);
+ if (compact) {
+ bytes.resize(offset + sizeof(uint32_t));
+ }
+ std::fill(bytes.begin() + offset, bytes.end(), '\0');
+ put_uint32(bytes, AUX_COUNT_INT, 1);
+ // Registers 1 and 3 are AUX_TOKEN, but only register 3 has an
auxiliary record.
+ put_uint32(bytes, offset, datasketches::HllUtil<>::pair(3, 21));
+ return bytes;
+ }
+
+ static std::string empty_compact_aux_entry() {
+ using namespace datasketches::hll_constants;
+ datasketches::Hll4Array<std::allocator<uint8_t>> sketch(8, true, {});
+ sketch.couponUpdate(datasketches::HllUtil<>::pair(0, 20));
+ auto serialized = sketch.serialize(true, 0);
+ std::string bytes(reinterpret_cast<const char*>(serialized.data()),
serialized.size());
+ put_uint32(bytes, HLL_BYTE_ARR_START + (1 << 7), EMPTY);
+ return bytes;
+ }
+
+ static void verify_unaligned_conversion(const std::string& prefix, const
std::string& bytes,
+ PrimitiveType primitive, int
target,
+ datasketches::target_hll_type type,
+ const std::string& name) {
+ auto block = input({prefix, bytes}, target, name, primitive);
+ const auto raw = block.get_by_position(0).column->get_data_at(1);
+ ASSERT_NE(reinterpret_cast<uintptr_t>(raw.data) % alignof(uint32_t),
0);
+ ASSERT_TRUE(execute(block).ok());
+ if (target >= 8 && type == datasketches::HLL_4) {
+ EXPECT_EQ(result(block, 1), bytes);
+ return;
+ }
+ const auto reference =
datasketches::hll_sketch::deserialize(bytes.data(), bytes.size());
+ datasketches::hll_union expected_union(std::min(target, 8));
+ expected_union.update(reference);
+ const auto expected =
expected_union.get_result(type).serialize_compact();
+ EXPECT_EQ(result(block, 1),
+ std::string(reinterpret_cast<const char*>(expected.data()),
expected.size()));
+ }
+
+ static void verify_set_count_corruption(const std::string& bytes) {
+ for (auto primitive : {TYPE_STRING, TYPE_VARCHAR, TYPE_VARBINARY}) {
+ for (int target : {4, 8, 21}) {
+ for (const auto& type : {"HLL_4", "HLL_6", "HLL_8"}) {
+ auto block = input({bytes}, target, type, primitive);
+ const auto status = execute(block);
+ EXPECT_EQ(status.code(), ErrorCode::CORRUPTION);
+ EXPECT_NE(status.to_string().find("coupon count does not
match occupied slots"),
+ std::string::npos);
+ }
+ }
+ }
+ }
+
+ static void verify_conversion_allocation_failure(const std::string& bytes,
+ const std::string& type,
size_t fail_at) {
+ SCOPED_TRACE(testing::Message() << type << '/' << fail_at);
+ ScopedSketchAllocationFailure failure(fail_at);
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllConversionAllocationFailure");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ {
+ auto block = input({bytes}, 8, type);
+ EXPECT_EQ(execute(block).code(), ErrorCode::MEM_ALLOC_FAILED);
+ }
+ EXPECT_EQ(failure.calls, fail_at);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+ }
+
+ static void expect_corruption(const std::string& bytes, const std::string&
message,
+ const std::string& type = "HLL_4") {
+ // The matching type at lgK=21 also covers the unchanged-byte path.
+ for (int target : {4, 21}) {
+ for (auto primitive : {TYPE_STRING, TYPE_VARBINARY}) {
+ auto block = input({bytes}, target, type, primitive);
+ auto status = execute(block);
+ EXPECT_EQ(status.code(), ErrorCode::CORRUPTION);
+ EXPECT_NE(status.to_string().find(message), std::string::npos);
+ }
+ }
+ }
+
+ static std::string sketch_bytes(int lg_k, datasketches::target_hll_type
type, uint64_t n,
+ bool compact = true, bool full = false) {
+ datasketches::hll_sketch sketch(lg_k, type, full);
+ for (uint64_t i = 0; i < n; ++i) {
+ sketch.update(i);
+ }
+ auto bytes = compact ? sketch.serialize_compact() :
sketch.serialize_updatable();
+ return {reinterpret_cast<const char*>(bytes.data()), bytes.size()};
+ }
+
+ static ColumnPtr integer(int value, size_t rows = 1) {
+ auto column = ColumnInt32::create();
+ column->insert_value(value);
+ return ColumnConst::create(std::move(column), rows);
+ }
+
+ static ColumnPtr text(const std::string& value, size_t rows = 1) {
+ auto column = ColumnString::create();
+ column->insert_data(value.data(), value.size());
+ return ColumnConst::create(std::move(column), rows);
+ }
+
+ static Status execute(Block& block, bool constants_in_open = true) {
+ const auto args = block.get_columns_with_type_and_name();
+ DataTypes types;
+ ColumnNumbers indexes;
+ std::vector<std::shared_ptr<ColumnPtrWrapper>> constants;
+ bool nullable = false;
+ for (size_t i = 0; i < args.size(); ++i) {
+ types.push_back(args[i].type);
+ indexes.push_back(i);
+ nullable |= args[i].type->is_nullable();
+ constants.push_back(constants_in_open &&
is_column_const(*args[i].column)
+ ?
std::make_shared<ColumnPtrWrapper>(args[i].column)
+ : nullptr);
+ }
+ auto result_type = nullable ? make_nullable(remove_nullable(types[0]))
: types[0];
+ auto function = SimpleFunctionFactory::instance().get_function(
+ "datasketches_hll_downsample", args, result_type);
+ EXPECT_NE(function, nullptr);
+ FunctionUtils utils(result_type, types, false);
+ auto* context = utils.get_fn_ctx();
+ context->set_constant_cols(constants);
+ RETURN_IF_ERROR(
+ function->open(context,
FunctionContext::FunctionStateScope::FRAGMENT_LOCAL));
+ RETURN_IF_ERROR(function->open(context,
FunctionContext::FunctionStateScope::THREAD_LOCAL));
+ auto rows = block.rows();
+ block.insert({nullptr, result_type, "result"});
+ auto status = function->execute(context, block, indexes, args.size(),
rows);
+ EXPECT_TRUE(
+ function->close(context,
FunctionContext::FunctionStateScope::THREAD_LOCAL).ok());
+ EXPECT_TRUE(
+ function->close(context,
FunctionContext::FunctionStateScope::FRAGMENT_LOCAL).ok());
+ return status;
+ }
+
+ static Block input(const std::vector<std::optional<std::string>>& values,
int lg_k,
+ const std::optional<std::string>& type = std::nullopt,
+ PrimitiveType primitive = TYPE_STRING, bool constant =
false,
+ bool nullable = false) {
+ DataTypePtr data_type = primitive == TYPE_VARBINARY
+ ?
DataTypePtr(std::make_shared<DataTypeVarbinary>())
+ : std::make_shared<DataTypeString>(-1,
primitive);
+ auto column = data_type->create_column();
+ auto null_map = ColumnUInt8::create();
+ for (const auto& value : values) {
+ if (value.has_value()) {
+ column->insert_data(value->data(), value->size());
+ } else {
+ column->insert_default();
+ }
+ null_map->insert_value(!value.has_value());
+ }
+ ColumnPtr data = std::move(column);
+ if (nullable) {
+ data = ColumnNullable::create(data, std::move(null_map));
+ data_type = make_nullable(data_type);
+ }
+ if (constant) {
+ data = ColumnConst::create(data, values.size());
+ }
+ Block block;
+ block.insert({data, data_type, "sk"});
+ block.insert({integer(lg_k, values.size()),
std::make_shared<DataTypeInt32>(), "lg_k"});
+ if (type.has_value()) {
+ block.insert({text(*type, values.size()),
std::make_shared<DataTypeString>(), "type"});
+ }
+ return block;
+ }
+
+ static std::string result(const Block& block, size_t row = 0) {
+ return block.get_by_position(block.columns() -
1).column->get_data_at(row).to_string();
+ }
+
+ static void verify_conversion(const std::string& src, int src_k,
+ datasketches::target_hll_type src_type,
uint64_t n, int target,
+ datasketches::target_hll_type target_type,
+ const std::string& name) {
+ SCOPED_TRACE(testing::Message()
+ << src_k << '/' << target << '/' << n << '/' << src_type
<< '/' << name);
+ Block block = input({src}, target, name);
+ ASSERT_TRUE(execute(block).ok());
+ auto out = result(block);
+ auto sketch = datasketches::hll_sketch::deserialize(out.data(),
out.size());
+ EXPECT_EQ(sketch.get_lg_config_k(), std::min(src_k, target));
+ EXPECT_EQ(sketch.get_target_type(), target_type);
+ EXPECT_EQ(sketch.is_empty(), n == 0);
+ EXPECT_EQ(out == src, src_k <= target && src_type == target_type);
+ auto again = input({out}, target, name);
+ ASSERT_TRUE(execute(again).ok());
+ EXPECT_EQ(result(again), out);
+ }
+
+ static void verify_nullable(PrimitiveType type, bool constant, const
std::string& src) {
+ auto block = input({src}, 10, "HLL_4", type, constant, true);
+ ASSERT_TRUE(execute(block).ok());
+ auto output = result(block);
+ auto mixed = input({std::nullopt, src, std::nullopt, src}, 10,
"HLL_4", type, false, true);
+ ASSERT_TRUE(execute(mixed).ok());
+ const auto& col = mixed.get_by_position(3).column;
+ EXPECT_TRUE(col->is_null_at(0));
+ EXPECT_TRUE(col->is_null_at(2));
+ EXPECT_EQ(result(mixed, 1), output);
+ EXPECT_EQ(result(mixed, 3), output);
+ auto all_null = input({std::nullopt}, 10, "HLL_4", type, constant,
true);
+ ASSERT_TRUE(execute(all_null).ok());
+ EXPECT_TRUE(all_null.get_by_position(3).column->is_null_at(0));
+ }
+};
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
ModesTypesPrecisionAndIdempotence) {
+ const std::vector<datasketches::target_hll_type> types = {
+ datasketches::HLL_4, datasketches::HLL_6, datasketches::HLL_8};
+ const std::vector<std::string> names = {"HLL_4", "hll_6", "HlL_8"};
+ for (auto src_type : types) {
+ for (int src_k : {4, 7, 10, 15, 21}) {
+ for (uint64_t n : {0, 5, 20, 4000}) {
+ for (bool compact : {false, true}) {
+ auto src = sketch_bytes(src_k, src_type, n, compact);
+ for (int target : {4, 10, 15, 21}) {
+ for (size_t t = 0; t < types.size(); ++t) {
+ verify_conversion(src, src_k, src_type, n, target,
types[t], names[t]);
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
DenseHighPrecisionAndDefaultType) {
+ for (const auto& pair : {std::pair {21, 15}, std::pair {15, 10}}) {
+ auto src = sketch_bytes(pair.first, datasketches::HLL_8, 100000, true,
true);
+ auto block = input({src}, pair.second);
+ ASSERT_TRUE(execute(block).ok());
+ auto out = result(block);
+ auto low = datasketches::hll_sketch::deserialize(out.data(),
out.size());
+ EXPECT_EQ(low.get_lg_config_k(), pair.second);
+ EXPECT_EQ(low.get_target_type(), datasketches::HLL_4);
+ EXPECT_LT(out.size(), src.size() / 10);
+ auto direct_bytes = sketch_bytes(pair.second, datasketches::HLL_4,
100000);
+ auto direct =
+ datasketches::hll_sketch::deserialize(direct_bytes.data(),
direct_bytes.size());
+ EXPECT_NEAR(low.get_composite_estimate(),
direct.get_composite_estimate(), 1e-8);
+ auto empty = input({sketch_bytes(21, datasketches::HLL_8, 0, true,
true)}, pair.second);
+ ASSERT_TRUE(execute(empty).ok());
+ EXPECT_EQ(result(empty).size(), 8);
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest, NullableConstAndBinaryColumns) {
+ for (auto type : {TYPE_STRING, TYPE_VARCHAR, TYPE_VARBINARY}) {
+ auto src = sketch_bytes(12, datasketches::HLL_8, 1000);
+ for (bool constant : {false, true}) {
+ verify_nullable(type, constant, src);
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
BeEvaluatedConstantsAndInvalidParameters) {
+ auto src = sketch_bytes(15, datasketches::HLL_8, 4000);
+ auto block = input({src, src}, 10, "HLL_6");
+ // Constant expressions can arrive as full columns and may be unavailable
in open().
+ block.get_by_position(1).column = integer(10,
2)->convert_to_full_column_if_const();
+ block.get_by_position(2).column = text("HLL_6",
2)->convert_to_full_column_if_const();
+ ASSERT_TRUE(execute(block, false).ok());
+ auto out = result(block);
+ EXPECT_EQ(result(block, 1), out);
+ EXPECT_EQ(datasketches::hll_sketch::deserialize(out.data(),
out.size()).get_lg_config_k(), 10);
+ for (bool in_open : {false, true}) {
+ for (int k : {3, 22, 268}) {
+ auto invalid = input({std::nullopt}, k, "HLL_4", TYPE_STRING,
true, true);
+ EXPECT_EQ(execute(invalid, in_open).code(),
ErrorCode::INVALID_ARGUMENT);
+ }
+ auto invalid = input({src}, 10, "bad");
+ EXPECT_EQ(execute(invalid, in_open).code(),
ErrorCode::INVALID_ARGUMENT);
+ for (size_t arg : {1, 2}) {
+ auto null_arg = input({src}, 10, "HLL_4");
+ auto& c = null_arg.get_by_position(arg);
+ auto data = remove_nullable(c.type)->create_column();
+ data->insert_default();
+ c.column = ColumnConst::create(
+ ColumnNullable::create(std::move(data),
ColumnUInt8::create(1, 1)), 1);
+ c.type = make_nullable(c.type);
+ EXPECT_EQ(execute(null_arg, in_open).code(),
ErrorCode::INVALID_ARGUMENT);
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest, InvalidSketchIncludingFastPath) {
+ auto dense = sketch_bytes(12, datasketches::HLL_4, 1000);
+ std::vector<std::string> invalid {"", "x", dense.substr(0, dense.size() /
2)};
+ for (size_t pos : {1, 2, 3}) {
+ auto bad = dense;
+ bad[pos] = static_cast<char>(255);
+ invalid.push_back(bad);
+ }
+ // Missing auxiliary map with an AUX_TOKEN is accepted by upstream
deserialize, but would
+ // dereference a null map in union/conversion. Reject it even on the
unchanged-byte path.
+ auto missing_aux = dense;
+ std::fill_n(missing_aux.begin() +
datasketches::hll_constants::AUX_COUNT_INT, 4, '\0');
+ missing_aux[datasketches::hll_constants::HLL_BYTE_ARR_START] |= 0xf;
+ invalid.push_back(missing_aux);
+ for (const auto& bytes : invalid) {
+ auto block = input({bytes}, 21, "HLL_4");
+ EXPECT_EQ(execute(block).code(), ErrorCode::CORRUPTION);
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest, AllocationFailureKeepsErrorCode)
{
+ auto bytes = sketch_bytes(12, datasketches::HLL_8, 1000);
+ const auto old = config::mem_alloc_fault_probability;
+ config::mem_alloc_fault_probability = 1;
+ ++enable_thread_catch_bad_alloc;
+ try {
+ static_cast<void>(datasketches_hll::deserialize(StringRef(bytes),
"downsample"));
+ ADD_FAILURE() << "Expected allocation failure";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::MEM_ALLOC_FAILED);
+ }
+ --enable_thread_catch_bad_alloc;
+ config::mem_alloc_fault_probability = old;
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest, RejectAuxMapsOnHll6AndHll8) {
+ using namespace datasketches::hll_constants;
+ for (auto type : {datasketches::HLL_6, datasketches::HLL_8}) {
+ for (bool compact : {false, true}) {
+ auto bytes = sketch_bytes(8, type, 1, compact, true);
+ const auto offset = bytes.size();
+ // Supply the full auxiliary payload: a short image would throw
before the unsafe
+ // Hll4Array cast, hiding the regression this test is meant to
exercise.
+ bytes.resize(offset + sizeof(uint32_t) * (compact ? 1 : 4), '\0');
+ put_uint32(bytes, AUX_COUNT_INT, 1);
+ bytes[LG_ARR_BYTE] = 2;
+ put_uint32(bytes, offset, datasketches::HllUtil<>::pair(3, 20));
+ expect_corruption(bytes, "cannot contain an auxiliary map",
+ type == datasketches::HLL_6 ? "HLL_6" : "HLL_8");
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
ArrayExponentsAndTruncatedPayloads) {
+ using namespace datasketches::hll_constants;
+ const auto set = sketch_bytes(8, datasketches::HLL_4, 10, false);
+ ASSERT_EQ(static_cast<uint8_t>(set[PREAMBLE_INTS_BYTE]), HASH_SET_PREINTS);
+ const auto aux = aux_sketch_bytes(8, false);
+ for (uint8_t exponent : {31, 32, 63, 64, 255}) {
+ SCOPED_TRACE(static_cast<int>(exponent));
+ auto bad_set = set;
+ bad_set[LG_ARR_BYTE] = static_cast<char>(exponent);
+ bad_set.resize(HASH_SET_INT_ARR_START);
+ expect_corruption(bad_set, "SET array exponent");
+ auto bad_aux = aux;
+ bad_aux[LG_ARR_BYTE] = static_cast<char>(exponent);
+ expect_corruption(bad_aux, "auxiliary array exponent");
+ }
+ auto small_aux = aux;
+ small_aux[LG_ARR_BYTE] = 0;
+ expect_corruption(small_aux, "auxiliary array exponent");
+ // Valid exponents still require a complete array before the library
allocates anything.
+ expect_corruption(set.substr(0, set.size() - 1), "corrupted");
+ expect_corruption(aux.substr(0, aux.size() - 1), "corrupted");
+ auto compact_aux = aux_sketch_bytes(8, true);
+ expect_corruption(compact_aux.substr(0, compact_aux.size() - 1),
"corrupted");
+ auto huge_count = set;
+ huge_count[LG_ARR_BYTE] = 0;
+ put_uint32(huge_count, HASH_SET_COUNT_INT, UINT32_MAX);
+ expect_corruption(huge_count, "SET coupon count");
+ put_uint32(compact_aux, AUX_COUNT_INT, UINT32_MAX);
+ expect_corruption(compact_aux, "auxiliary count exceeds register count");
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
AuxiliarySlotsAreUniqueAndDoNotLeak) {
+ using namespace datasketches::hll_constants;
+ for (bool compact : {false, true}) {
+ auto bytes = aux_sketch_bytes(8, compact);
+ const size_t offset = HLL_BYTE_ARR_START + (1 << 7);
+ std::fill(bytes.begin() + offset, bytes.end(), '\0');
+ put_uint32(bytes, offset, datasketches::HllUtil<>::pair(3, 20));
+ // Raw keys differ but deserialize masks both down to register 3.
+ put_uint32(bytes, offset + 4, datasketches::HllUtil<>::pair(3 + (1 <<
8), 21));
+ expect_corruption(bytes, "duplicate slots");
+
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllAuxiliaryMapError");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ for (int i = 0; i < 32; ++i) {
+ try {
+
static_cast<void>(datasketches_hll::deserialize(StringRef(bytes), "test"));
+ ADD_FAILURE() << "Expected duplicate auxiliary slots to fail";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::CORRUPTION);
+ }
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
ValidAuxiliaryMapsAndLegacyArrayExponents) {
+ using namespace datasketches::hll_constants;
+ for (int lg_k : {4, 8, 12}) {
+ for (bool compact : {false, true}) {
+ auto bytes = aux_sketch_bytes(lg_k, compact);
+ if (compact) {
+ // Compact auxiliary images historically omit lgArr; the
library ignores it.
+ bytes[LG_ARR_BYTE] = static_cast<char>(255);
+ }
+ for (const auto& [type, name] : {std::pair {datasketches::HLL_4,
"HLL_4"},
+ std::pair {datasketches::HLL_6,
"HLL_6"},
+ std::pair {datasketches::HLL_8,
"HLL_8"}}) {
+ verify_conversion(bytes, lg_k, datasketches::HLL_4, 2, 4,
type, name);
+ verify_conversion(bytes, lg_k, datasketches::HLL_4, 2, 21,
type, name);
+ }
+ }
+ }
+ for (bool compact : {false, true}) {
+ auto bytes = sketch_bytes(8, datasketches::HLL_4, 10, compact);
+ bytes[LG_ARR_BYTE] = compact ? static_cast<char>(255) : 0;
+ verify_conversion(bytes, 8, datasketches::HLL_4, 10, 4,
datasketches::HLL_4, "HLL_4");
+ verify_conversion(bytes, 8, datasketches::HLL_4, 10, 21,
datasketches::HLL_4, "HLL_4");
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
RejectMissingAuxiliarySlotsWithoutLeaks) {
+ for (bool compact : {false, true}) {
+ const auto bytes = missing_aux_slot(compact);
+ expect_corruption(bytes, "no matching auxiliary slot");
+ for (const auto& type : {"HLL_6", "HLL_8"}) {
+ expect_corruption(bytes, "no matching auxiliary slot", type);
+ }
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllMissingAuxiliarySlot");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ for (int i = 0; i < 32; ++i) {
+ try {
+
static_cast<void>(datasketches_hll::deserialize(StringRef(bytes), "test"));
+ ADD_FAILURE() << "Expected missing auxiliary slot to fail";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::CORRUPTION);
+ }
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
UnalignedAuxiliaryMapsKeepSketchContents) {
+ for (int lg_k : {4, 8, 12}) {
+ for (bool compact : {false, true}) {
+ const auto bytes = aux_sketch_bytes(lg_k, compact);
+ const auto reference =
+ datasketches::hll_sketch::deserialize(bytes.data(),
bytes.size());
+ const auto expected = reference.serialize_updatable();
+ for (size_t offset = 0; offset < alignof(uint32_t); ++offset) {
+ SCOPED_TRACE(testing::Message() << lg_k << '/' << compact <<
'/' << offset);
+ DorisVector<uint32_t> storage((bytes.size() + offset + 3) /
sizeof(uint32_t));
+ auto* start = reinterpret_cast<char*>(storage.data()) + offset;
+ std::memcpy(start, bytes.data(), bytes.size());
+ const auto sketch =
+ datasketches_hll::deserialize(StringRef(start,
bytes.size()), "test");
+ const auto actual = sketch.serialize_updatable();
+ EXPECT_EQ(std::string(reinterpret_cast<const
char*>(actual.data()), actual.size()),
+ std::string(reinterpret_cast<const
char*>(expected.data()),
+ expected.size()));
+ EXPECT_DOUBLE_EQ(sketch.get_composite_estimate(),
+ reference.get_composite_estimate());
+ }
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
MixedRowsKeepUnalignedConversionResults) {
+ const auto prefix = sketch_bytes(8, datasketches::HLL_6, 1000, true, true);
+ ASSERT_EQ(prefix.size(), 233);
+ for (bool compact : {false, true}) {
+ const auto bytes = aux_sketch_bytes(8, compact);
+ for (auto primitive : {TYPE_STRING, TYPE_VARCHAR, TYPE_VARBINARY}) {
+ for (int target : {4, 8, 21}) {
+ for (const auto& [type, name] : {std::pair
{datasketches::HLL_4, "HLL_4"},
+ std::pair
{datasketches::HLL_6, "HLL_6"},
+ std::pair
{datasketches::HLL_8, "HLL_8"}}) {
+ verify_unaligned_conversion(prefix, bytes, primitive,
target, type, name);
+ }
+ }
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
UnalignedAggregateStateKeepsEstimate) {
+ using Data = AggregateFunctionHllSketchData<TYPE_STRING>;
+ for (bool compact : {false, true}) {
+ const auto bytes = aux_sketch_bytes(8, compact);
+ const auto source = datasketches_hll::deserialize(StringRef(bytes),
"test");
+ Data expected;
+ expected.merge(source);
+ auto buffer = ColumnString::create();
+ BufferWritable writer(*buffer);
+ writer.write_binary(StringRef(bytes));
+ writer.commit();
+ BufferReadable inspect(buffer->get_data_at(0));
+ StringRef payload;
+ inspect.read_binary(payload);
+ ASSERT_NE(reinterpret_cast<uintptr_t>(payload.data) %
alignof(uint32_t), 0);
+ BufferReadable reader(buffer->get_data_at(0));
+ Data actual;
+ actual.read(reader);
+ EXPECT_DOUBLE_EQ(actual.get_result(), expected.get_result());
+ const auto actual_bytes =
actual.hll_union_data->get_result().serialize_compact();
+ const auto expected_bytes =
expected.hll_union_data->get_result().serialize_compact();
+ EXPECT_EQ(std::string(reinterpret_cast<const
char*>(actual_bytes.data()),
+ actual_bytes.size()),
+ std::string(reinterpret_cast<const
char*>(expected_bytes.data()),
+ expected_bytes.size()));
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
UnalignedDeserializationFailuresAreBalanced) {
+ const auto bytes = aux_sketch_bytes(8, true);
+ DorisVector<uint32_t> storage((bytes.size() + 4) / sizeof(uint32_t));
+ auto* start = reinterpret_cast<char*>(storage.data()) + 1;
+ std::memcpy(start, bytes.data(), bytes.size());
+ const StringRef input(start, bytes.size());
+ size_t allocations = 0;
+ {
+ ScopedSketchAllocationFailure count_only(0);
+ const auto sketch = datasketches_hll::deserialize(input, "test");
+ allocations = count_only.calls;
+ }
+ ASSERT_GT(allocations, 1);
+ for (size_t fail_at = 1; fail_at <= allocations; ++fail_at) {
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllUnalignedAllocationFailure");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ ScopedSketchAllocationFailure failure(fail_at);
+ try {
+ static_cast<void>(datasketches_hll::deserialize(input, "test"));
+ ADD_FAILURE() << "Expected allocation failure";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::MEM_ALLOC_FAILED);
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
RejectCompactEmptyAuxiliaryEntry) {
+ const auto bytes = empty_compact_aux_entry();
+ for (const auto& type : {"HLL_4", "HLL_6", "HLL_8"}) {
+ expect_corruption(bytes, "compact auxiliary map contains an EMPTY
entry", type);
+ }
+ using Data = AggregateFunctionHllSketchData<TYPE_STRING>;
+ auto buffer = ColumnString::create();
+ BufferWritable writer(*buffer);
+ writer.write_binary(StringRef(bytes));
+ writer.commit();
+ BufferReadable reader(buffer->get_data_at(0));
+ Data data;
+ try {
+ data.read(reader);
+ ADD_FAILURE() << "Expected compact EMPTY auxiliary entry to fail";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::CORRUPTION);
+ EXPECT_NE(e.to_string().find("compact auxiliary map contains an EMPTY
entry"),
+ std::string::npos);
+ }
+ EXPECT_FALSE(data.hll_union_data.has_value());
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest, RejectUpdatableSetCountMismatch)
{
+ using namespace datasketches::hll_constants;
+ for (auto type : {datasketches::HLL_4, datasketches::HLL_6,
datasketches::HLL_8}) {
+ const auto valid = sketch_bytes(8, type, 10, false);
+ ASSERT_EQ(static_cast<uint8_t>(valid[PREAMBLE_INTS_BYTE]),
HASH_SET_PREINTS);
+ ASSERT_EQ(valid.size(), HASH_SET_INT_ARR_START + 32 *
sizeof(uint32_t));
+ for (uint32_t count : {0, 1, 9, 11, 24}) {
+ auto bytes = valid;
+ put_uint32(bytes, HASH_SET_COUNT_INT, count);
+ verify_set_count_corruption(bytes);
+ }
+ // The legacy unset lgArr is reconstructed by the library from count=1
as 32 slots;
+ // the payload is still complete, so the occupied-slot check must
reject it too.
+ auto legacy = valid;
+ legacy[LG_ARR_BYTE] = 0;
+ put_uint32(legacy, HASH_SET_COUNT_INT, 1);
+ verify_set_count_corruption(legacy);
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
+ SetCountMismatchFailsBeforeAllocationAndAggregateRead) {
+ using namespace datasketches::hll_constants;
+ auto bytes = sketch_bytes(8, datasketches::HLL_4, 10, false);
+ put_uint32(bytes, HASH_SET_COUNT_INT, 1);
+ auto tracker =
+ MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
"HllSetCountMismatch");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ {
+ ScopedSketchAllocationFailure count_only(0);
+ for (int i = 0; i < 32; ++i) {
+ try {
+
static_cast<void>(datasketches_hll::deserialize(StringRef(bytes), "test"));
+ ADD_FAILURE() << "Expected SET count mismatch to fail";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::CORRUPTION);
+ }
+ }
+ EXPECT_EQ(count_only.calls, 0);
+ auto buffer = ColumnString::create();
+ BufferWritable writer(*buffer);
+ writer.write_binary(StringRef(bytes));
+ writer.commit();
+ BufferReadable reader(buffer->get_data_at(0));
+ AggregateFunctionHllSketchData<TYPE_STRING> data;
+ try {
+ data.read(reader);
+ ADD_FAILURE() << "Expected aggregate SET count mismatch to fail";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::CORRUPTION);
+ EXPECT_NE(e.to_string().find("coupon count does not match occupied
slots"),
+ std::string::npos);
+ }
+ EXPECT_FALSE(data.hll_union_data.has_value());
+ EXPECT_EQ(count_only.calls, 0);
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
ValidSetCountsAndLegacyExponentsKeepResults) {
+ using namespace datasketches::hll_constants;
+ for (auto type : {datasketches::HLL_4, datasketches::HLL_6,
datasketches::HLL_8}) {
+ for (uint64_t count : {10, 80, 300}) {
+ for (bool compact : {false, true}) {
+ auto bytes = sketch_bytes(12, type, count, compact);
+ ASSERT_EQ(static_cast<uint8_t>(bytes[PREAMBLE_INTS_BYTE]),
HASH_SET_PREINTS);
+ const auto reference =
+ datasketches::hll_sketch::deserialize(bytes.data(),
bytes.size());
+ // Preserve the library's reconstruction of an unset
serialized array exponent.
+ bytes[LG_ARR_BYTE] = 0;
+ const auto parsed =
datasketches_hll::deserialize(StringRef(bytes), "test");
+ const auto expected = reference.serialize_compact();
+ const auto actual = parsed.serialize_compact();
+ EXPECT_EQ(std::string(reinterpret_cast<const
char*>(actual.data()), actual.size()),
+ std::string(reinterpret_cast<const
char*>(expected.data()),
+ expected.size()));
+ EXPECT_DOUBLE_EQ(parsed.get_estimate(),
reference.get_estimate());
+ }
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
EveryDeserializationAllocationFailureIsBalanced) {
+ for (const auto& bytes : {sketch_bytes(8, datasketches::HLL_8, 1, true,
true),
+ sketch_bytes(8, datasketches::HLL_4, 10, false),
+ aux_sketch_bytes(8, true), aux_sketch_bytes(8,
false)}) {
+ size_t allocations = 0;
+ {
+ ScopedSketchAllocationFailure count_only(0);
+ auto sketch = datasketches_hll::deserialize(StringRef(bytes),
"test");
+ EXPECT_FALSE(sketch.is_empty());
+ allocations = count_only.calls;
+ }
+ ASSERT_GT(allocations, 1);
+ for (size_t fail_at = 1; fail_at <= allocations; ++fail_at) {
+ SCOPED_TRACE(fail_at);
+ ScopedSketchAllocationFailure failure(fail_at);
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllDeserializeAllocationFailure");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ try {
+
static_cast<void>(datasketches_hll::deserialize(StringRef(bytes), "test"));
+ ADD_FAILURE() << "Expected allocation failure";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::MEM_ALLOC_FAILED);
+ }
+ EXPECT_EQ(failure.calls, fail_at);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
EveryConversionAllocationFailureIsBalanced) {
+ for (const auto& bytes :
+ {aux_sketch_bytes(12, true), sketch_bytes(12, datasketches::HLL_4,
200, true)}) {
+ for (const auto& type : {"HLL_4", "HLL_6", "HLL_8"}) {
+ size_t allocations = 0;
+ {
+ auto block = input({bytes}, 8, type);
+ ScopedSketchAllocationFailure count_only(0);
+ ASSERT_TRUE(execute(block).ok());
+ allocations = count_only.calls;
+ }
+ ASSERT_GT(allocations, 1);
+ for (size_t fail_at = 1; fail_at <= allocations; ++fail_at) {
+ verify_conversion_allocation_failure(bytes, type, fail_at);
+ }
+ }
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
AggregateUpdateFailureResetsStateAndMemory) {
+ using Data = AggregateFunctionHllSketchData<TYPE_STRING>;
+ const auto large_bytes = sketch_bytes(12, datasketches::HLL_8, 1000, true,
true);
+ const auto small_bytes = sketch_bytes(8, datasketches::HLL_8, 1000, true,
true);
+ auto large = datasketches_hll::deserialize(StringRef(large_bytes), "test");
+ auto small = datasketches_hll::deserialize(StringRef(small_bytes), "test");
+ size_t allocations = 0;
+ {
+ Data data;
+ data.merge(large);
+ ScopedSketchAllocationFailure count_only(0);
+ data.merge(small);
+ allocations = count_only.calls;
+ }
+ ASSERT_GT(allocations, 1);
+ for (size_t fail_at = 1; fail_at <= allocations; ++fail_at) {
+ SCOPED_TRACE(fail_at);
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllAggregateAllocationFailure");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ {
+ Data data;
+ data.merge(large);
+ ScopedSketchAllocationFailure failure(fail_at);
+ try {
+ data.merge(small);
+ ADD_FAILURE() << "Expected aggregate allocation failure";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::MEM_ALLOC_FAILED);
+ }
+ EXPECT_FALSE(data.hll_union_data.has_value());
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+ }
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
SuccessfulAllocationScopeTransfersOwnership) {
+ const auto bytes = aux_sketch_bytes(8, true);
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllAllocationScopeOwnership");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ {
+ // A successful deserialize returns a sketch whose allocations outlive
its scope.
+ auto sketch = datasketches_hll::deserialize(StringRef(bytes), "test");
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_GT(tracker->consumption(), baseline);
+ EXPECT_EQ(sketch.get_target_type(), datasketches::HLL_4);
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
UnionCopyAndMoveKeepIndependentOwnership) {
+ const auto bytes = aux_sketch_bytes(8, true);
+ auto sketch = datasketches_hll::deserialize(StringRef(bytes), "test");
+ auto tracker =
+ MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
"HllUnionCopyMove");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ {
+ datasketches_hll::Union source(8);
+ source.update(sketch);
+ const auto estimate = source.get_estimate();
+ datasketches_hll::Union copy(source);
+ datasketches_hll::Union moved(std::move(copy));
+ EXPECT_DOUBLE_EQ(moved.get_estimate(), estimate);
+ moved = source;
+ EXPECT_DOUBLE_EQ(moved.get_estimate(), estimate);
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
NestedAllocationScopesReleaseOnlyTheirOwnMemory) {
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllNestedAllocationScope");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ {
+ datasketches_hll::AllocationScope outer;
+ auto* parent = datasketches_hll::Alloc().allocate(32);
+ try {
+ datasketches_hll::AllocationScope inner;
+ static_cast<void>(datasketches_hll::Alloc().allocate(64));
+ throw Exception(ErrorCode::MEM_ALLOC_FAILED, "Simulated orphan in
nested call");
+ } catch (const Exception&) {
+ }
+ parent[0] = 42;
+ EXPECT_EQ(parent[0], 42);
+ datasketches_hll::Alloc().deallocate(parent, 32);
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+}
+
+TEST_F(FunctionDataSketchesHllDownsampleTest,
AllocationScopesAreIsolatedAcrossBthreads) {
+ const auto work = +[](void*) -> void* {
+ ScopedInitThreadContext thread;
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"HllBthreadAllocationScope");
+ auto switch_tracker = SwitchThreadMemTrackerLimiter(tracker);
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ const auto baseline = tracker->consumption();
+ for (int i = 0; i < 8; ++i) {
+ try {
+ datasketches_hll::AllocationScope scope;
+ static_cast<void>(datasketches_hll::Alloc().allocate(64));
+ // Yield while the call owns allocations, so another bthread
can enter its
+ // own scope on the same worker. A pthread-local scope alone
is insufficient.
+ bthread_usleep(1000);
+ throw Exception(ErrorCode::MEM_ALLOC_FAILED, "Simulated
bthread orphan");
+ } catch (const Exception&) {
+ }
+ }
+ thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+ EXPECT_EQ(tracker->consumption(), baseline);
+ return nullptr;
+ };
+ std::array<bthread_t, 2> threads;
+ for (auto& thread : threads) {
+ ASSERT_EQ(bthread_start_background(&thread, nullptr, work, nullptr),
0);
+ }
+ for (auto thread : threads) {
+ EXPECT_EQ(bthread_join(thread, nullptr), 0);
+ }
+}
+
+} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinScalarFunctions.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinScalarFunctions.java
index 7f0140e8303..2d7aaecd38d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinScalarFunctions.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinScalarFunctions.java
@@ -163,6 +163,7 @@ import
org.apache.doris.nereids.trees.expressions.functions.scalar.CurrentUser;
import org.apache.doris.nereids.trees.expressions.functions.scalar.CutIpv6;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.CutToFirstSignificantSubdomain;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.DamerauLevenshteinDistance;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.DataSketchesHllDownsample;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Database;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Date;
import org.apache.doris.nereids.trees.expressions.functions.scalar.DateDiff;
@@ -773,6 +774,7 @@ public class BuiltinScalarFunctions implements
FunctionHelper {
scalar(CutIpv6.class, "cut_ipv6"),
scalar(CutToFirstSignificantSubdomain.class,
"cut_to_first_significant_subdomain"),
scalar(DamerauLevenshteinDistance.class,
"damerau_levenshtein_distance"),
+ scalar(DataSketchesHllDownsample.class,
"datasketches_hll_downsample"),
scalar(Database.class, "database", "schema", "current_database"),
scalar(Date.class, "date"),
scalar(DateDiff.class, "datediff"),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java
index 57a5a5fe1d9..1e885d9a2c8 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/FoldConstantRuleOnBE.java
@@ -42,6 +42,7 @@ import org.apache.doris.nereids.trees.expressions.Match;
import org.apache.doris.nereids.trees.expressions.functions.BoundFunction;
import org.apache.doris.nereids.trees.expressions.functions.ai.AIFunction;
import
org.apache.doris.nereids.trees.expressions.functions.generator.TableGeneratingFunction;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.DataSketchesHllDownsample;
import org.apache.doris.nereids.trees.expressions.functions.scalar.FromBase64;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.Ipv6StringToNumOrDefault;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.Ipv6StringToNumOrNull;
@@ -254,7 +255,8 @@ public class FoldConstantRuleOnBE implements
ExpressionPatternRuleFactory {
// Skip those function to avoid incorrect binary data processing
during constant folding
if (expr instanceof FromBase64 || expr instanceof Ipv6StringToNumOrNull
- || expr instanceof Ipv6StringToNumOrDefault || expr instanceof
AIFunction) {
+ || expr instanceof Ipv6StringToNumOrDefault || expr instanceof
AIFunction
+ || expr instanceof DataSketchesHllDownsample) {
return true;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DataSketchesHllDownsample.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DataSketchesHllDownsample.java
new file mode 100644
index 00000000000..c2a63c3056f
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DataSketchesHllDownsample.java
@@ -0,0 +1,185 @@
+// 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.
+
+package org.apache.doris.nereids.trees.expressions.functions.scalar;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import
org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
+import org.apache.doris.nereids.trees.expressions.functions.PropagateNullable;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLikeLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.VarBinaryType;
+import org.apache.doris.nereids.types.VarcharType;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * ScalarFunction 'datasketches_hll_downsample'.
+ *
+ * <p>datasketches_hll_downsample(sketch, lg_k [, tgt_type]) rewrites a
serialized Apache DataSketches HLL
+ * sketch so that its lgK is min(lg_k, source lgK) and its HLL type is
tgt_type (HLL_4 by default). A sketch
+ * that already satisfies both is returned byte for byte.
+ */
+public class DataSketchesHllDownsample extends ScalarFunction
+ implements ExplicitlyCastableSignature, PropagateNullable {
+
+ public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
+
FunctionSignature.ret(StringType.INSTANCE).args(StringType.INSTANCE,
IntegerType.INSTANCE),
+ FunctionSignature.ret(VarcharType.SYSTEM_DEFAULT)
+ .args(VarcharType.SYSTEM_DEFAULT, IntegerType.INSTANCE),
+
FunctionSignature.ret(VarBinaryType.INSTANCE).args(VarBinaryType.INSTANCE,
IntegerType.INSTANCE),
+ FunctionSignature.ret(StringType.INSTANCE)
+ .args(StringType.INSTANCE, IntegerType.INSTANCE,
VarcharType.SYSTEM_DEFAULT),
+ FunctionSignature.ret(VarcharType.SYSTEM_DEFAULT)
+ .args(VarcharType.SYSTEM_DEFAULT, IntegerType.INSTANCE,
VarcharType.SYSTEM_DEFAULT),
+ FunctionSignature.ret(VarBinaryType.INSTANCE)
+ .args(VarBinaryType.INSTANCE, IntegerType.INSTANCE,
VarcharType.SYSTEM_DEFAULT)
+ );
+
+ // the bounds the DataSketches library accepts for a sketch's lgK
+ private static final int MIN_LG_K = 4;
+ private static final int MAX_LG_K = 21;
+ private static final Set<String> TGT_TYPES = ImmutableSet.of("HLL_4",
"HLL_6", "HLL_8");
+
+ /**
+ * constructor with 2 arguments.
+ */
+ public DataSketchesHllDownsample(Expression arg0, Expression arg1) {
+ super("datasketches_hll_downsample", arg0, arg1);
+ }
+
+ /**
+ * constructor with 3 arguments.
+ */
+ public DataSketchesHllDownsample(Expression arg0, Expression arg1,
Expression arg2) {
+ super("datasketches_hll_downsample", arg0, arg1, arg2);
+ }
+
+ /** constructor for withChildren and reuse signature */
+ private DataSketchesHllDownsample(ScalarFunctionParams functionParams) {
+ super(functionParams);
+ }
+
+ @Override
+ public void checkLegalityBeforeTypeCoercion() {
+ DataType sketchType = getArgumentType(0);
+ if (!(sketchType.isStringLikeType() || sketchType.isVarBinaryType() ||
sketchType.isNullType())) {
+ throw new AnalysisException(getName()
+ + " function's first argument should be of
STRING/VARCHAR/VARBINARY type, but was "
+ + sketchType);
+ }
+ // validate the value FE can evaluate here, because constant folding,
e.g. of
+ // datasketches_hll_downsample(null, 30), may remove this function
before checkLegalityAfterRewrite;
+ // the types are checked before the type coercion casts the arguments
to the signature
+ checkLgK(ExpressionUtils.foldConstantArgument(getArgument(1)));
+ if (arity() == 3) {
+ checkTgtType(ExpressionUtils.foldConstantArgument(getArgument(2)));
+ }
+ }
+
+ @Override
+ public void checkLegalityAfterRewrite() {
+ checkLgK(getArgument(1));
+ if (arity() == 3) {
+ checkTgtType(getArgument(2));
+ }
+ }
+
+ private void checkLgK(Expression lgK) {
+ if (!lgK.isConstant() || !lgK.getDataType().isIntegralType()) {
+ throw new AnalysisException(getName() + " requires lg_k to be a
constant integer: " + toSql());
+ }
+ // the value of a constant FE cannot fold is validated by BE when it
is evaluated
+ if (!(lgK instanceof Literal)) {
+ return;
+ }
+ if (!(lgK instanceof IntegerLikeLiteral)) {
+ throw new AnalysisException(getName() + " requires lg_k to be a
constant integer: " + toSql());
+ }
+ // compare the full value: narrowing a wide literal to int would wrap
it, e.g. 4294967308 into 12
+ BigDecimal value = ((IntegerLikeLiteral) lgK).getBigDecimalValue();
+ if (value.compareTo(BigDecimal.valueOf(MIN_LG_K)) < 0 ||
value.compareTo(BigDecimal.valueOf(MAX_LG_K)) > 0) {
+ throw new AnalysisException(getName() + " requires lg_k to be
between " + MIN_LG_K + " and "
+ + MAX_LG_K + ", but was " + value);
+ }
+ }
+
+ private void checkTgtType(Expression tgtType) {
+ if (!tgtType.isConstant() ||
!tgtType.getDataType().isStringLikeType()) {
+ throw new AnalysisException(getName() + " requires tgt_type to be
a constant string: " + toSql());
+ }
+ if (!(tgtType instanceof Literal)) {
+ return;
+ }
+ if (!(tgtType instanceof StringLikeLiteral)) {
+ throw new AnalysisException(getName() + " requires tgt_type to be
a constant string: " + toSql());
+ }
+ String value = ((StringLikeLiteral) tgtType).getStringValue();
+ if (!TGT_TYPES.contains(value.toUpperCase(Locale.ROOT))) {
+ throw new AnalysisException(getName() + " requires tgt_type to be
one of HLL_4, HLL_6, HLL_8, but was '"
+ + value + "'");
+ }
+ }
+
+ /**
+ * withChildren.
+ */
+ @Override
+ public DataSketchesHllDownsample withChildren(List<Expression> children) {
+ Preconditions.checkArgument(children.size() == 2 || children.size() ==
3);
+ return new DataSketchesHllDownsample(getFunctionParams(children));
+ }
+
+ @Override
+ public List<FunctionSignature> getSignatures() {
+ // Catalog matching treats VARCHAR and STRING as interchangeable.
Select the input family
+ // first so the earlier STRING overload cannot change a VARCHAR result
into STRING.
+ DataType input = getArgumentType(0);
+ DataType output = input.isVarBinaryType() ? VarBinaryType.INSTANCE
+ : input.isVarcharType() ? VarcharType.SYSTEM_DEFAULT :
StringType.INSTANCE;
+ return SIGNATURES.stream().filter(signature ->
signature.returnType.equals(output))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public boolean foldable() {
+ // Keep parameter validation on BE even for a NULL sketch when, for
example, lg_k contains
+ // crc32() and cannot be evaluated on FE. Binary results also cannot
become FE strings.
+ return false;
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitDataSketchesHllDownsample(this, context);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
index 865669a0632..b5120939365 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
@@ -173,6 +173,7 @@ import
org.apache.doris.nereids.trees.expressions.functions.scalar.CurrentUser;
import org.apache.doris.nereids.trees.expressions.functions.scalar.CutIpv6;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.CutToFirstSignificantSubdomain;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.DamerauLevenshteinDistance;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.DataSketchesHllDownsample;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Database;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Date;
import org.apache.doris.nereids.trees.expressions.functions.scalar.DateDiff;
@@ -1185,6 +1186,10 @@ public interface ScalarFunctionVisitor<R, C> {
return visitScalarFunction(cutIpv6, context);
}
+ default R visitDataSketchesHllDownsample(DataSketchesHllDownsample
function, C context) {
+ return visitScalarFunction(function, context);
+ }
+
default R visitDatabase(Database database, C context) {
return visitScalarFunction(database, context);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DataSketchesHllDownsampleTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DataSketchesHllDownsampleTest.java
new file mode 100644
index 00000000000..988696464f3
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DataSketchesHllDownsampleTest.java
@@ -0,0 +1,107 @@
+// 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.
+
+package org.apache.doris.nereids.trees.expressions.functions.scalar;
+
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.rules.analysis.ExpressionAnalyzer;
+import org.apache.doris.nereids.rules.expression.rules.FoldConstantRuleOnFE;
+import org.apache.doris.nereids.trees.expressions.Add;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.LargeIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.VarBinaryType;
+import org.apache.doris.nereids.types.VarcharType;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigInteger;
+
+class DataSketchesHllDownsampleTest {
+ private static final Expression SKETCH = SlotReference.of("sk",
StringType.INSTANCE);
+
+ private DataSketchesHllDownsample analyze(Expression sketch, Expression k,
Expression type) {
+ return (DataSketchesHllDownsample)
ExpressionAnalyzer.analyzeFunction(null, null,
+ type == null ? new DataSketchesHllDownsample(sketch, k)
+ : new DataSketchesHllDownsample(sketch, k, type));
+ }
+
+ @Test
+ void testInputTypesAndArity() {
+ for (DataType type : ImmutableList.of(StringType.INSTANCE,
VarcharType.SYSTEM_DEFAULT,
+ VarBinaryType.INSTANCE)) {
+ Expression input = SlotReference.of("sk", type);
+ Assertions.assertEquals(type, analyze(input, new
IntegerLiteral(12), null).getDataType());
+ DataSketchesHllDownsample fn = analyze(input, new
IntegerLiteral(12), new StringLiteral("HLL_4"));
+ Assertions.assertEquals(type, fn.getDataType());
+ Assertions.assertEquals(fn.getArguments(),
fn.withChildren(fn.getArguments()).getArguments());
+ }
+ Assertions.assertThrows(AnalysisException.class,
+ () -> analyze(new IntegerLiteral(1), new IntegerLiteral(12),
null));
+ }
+
+ @Test
+ void testBoundsAndTypes() {
+ for (int k : new int[] {4, 7, 10, 15, 21}) {
+ for (String type : ImmutableList.of("HLL_4", "hll_6", "HlL_8")) {
+ Assertions.assertDoesNotThrow(() -> analyze(SKETCH, new
IntegerLiteral(k), new StringLiteral(type)));
+ }
+ }
+ for (Expression invalid : ImmutableList.of(new IntegerLiteral(3), new
IntegerLiteral(22),
+ new LargeIntLiteral(new BigInteger("18446744073709551628")),
+ new StringLiteral("12"), new NullLiteral(IntegerType.INSTANCE),
+ SlotReference.of("k", IntegerType.INSTANCE))) {
+ Assertions.assertThrows(AnalysisException.class, () ->
analyze(SKETCH, invalid, null));
+ // Invalid parameters must be rejected before NULL propagation can
remove the call.
+ Assertions.assertThrows(AnalysisException.class, () -> analyze(new
NullLiteral(), invalid, null));
+ }
+ for (Expression invalid : ImmutableList.of(new StringLiteral("HLL_5"),
new IntegerLiteral(4),
+ new NullLiteral(StringType.INSTANCE), SlotReference.of("type",
StringType.INSTANCE))) {
+ Assertions.assertThrows(AnalysisException.class,
+ () -> analyze(SKETCH, new IntegerLiteral(12), invalid));
+ }
+ }
+
+ @Test
+ void testConstantExpressionsAndRewriteValidation() {
+ DataSketchesHllDownsample fn = analyze(SKETCH,
+ new Add(new IntegerLiteral(10), new IntegerLiteral(2)),
+ new Concat(new StringLiteral("HLL_"), new StringLiteral("4")));
+ Expression folded = FoldConstantRuleOnFE.evaluateWithoutContext(fn);
+ Assertions.assertDoesNotThrow(folded::checkLegalityAfterRewrite);
+ Assertions.assertEquals(new IntegerLiteral(12), folded.child(1));
+ Expression onlyBe = new Add(new IntegerLiteral(10), new Crc32(new
StringLiteral("")));
+ Assertions.assertDoesNotThrow(() -> analyze(SKETCH, onlyBe,
null).checkLegalityAfterRewrite());
+ Expression nullCall = analyze(new NullLiteral(), onlyBe, null);
+ Assertions.assertInstanceOf(DataSketchesHllDownsample.class,
+ FoldConstantRuleOnFE.evaluateWithoutContext(nullCall));
+ DataSketchesHllDownsample invalid = fn.withChildren(
+ ImmutableList.of(SKETCH, new IntegerLiteral(30), new
StringLiteral("HLL_4")));
+ Assertions.assertThrows(AnalysisException.class,
invalid::checkLegalityAfterRewrite);
+ DataSketchesHllDownsample invalidType = fn.withChildren(
+ ImmutableList.of(SKETCH, new IntegerLiteral(12), new
StringLiteral("bad")));
+ Assertions.assertThrows(AnalysisException.class,
invalidType::checkLegalityAfterRewrite);
+ }
+}
diff --git
a/regression-test/data/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample.out
b/regression-test/data/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample.out
new file mode 100644
index 00000000000..8924538f0a9
--- /dev/null
+++
b/regression-test/data/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample.out
@@ -0,0 +1,147 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !matrix --
+1 48 04 02 true
+2 28 04 00 true
+3 8 04 00 true
+4 \N \N \N \N
+
+-- !matrix --
+1 53 04 06 true
+2 28 04 04 true
+3 8 04 04 true
+4 \N \N \N \N
+
+-- !matrix --
+1 56 04 0A true
+2 28 04 08 true
+3 8 04 08 true
+4 \N \N \N \N
+
+-- !matrix --
+1 104 07 02 true
+2 28 07 00 true
+3 8 07 00 true
+4 \N \N \N \N
+
+-- !matrix --
+1 137 07 06 true
+2 28 07 04 true
+3 8 07 04 true
+4 \N \N \N \N
+
+-- !matrix --
+1 168 07 0A true
+2 28 07 08 true
+3 8 07 08 true
+4 \N \N \N \N
+
+-- !matrix --
+1 552 0A 02 true
+2 28 0A 00 true
+3 8 0A 00 true
+4 \N \N \N \N
+
+-- !matrix --
+1 809 0A 06 true
+2 28 0A 04 true
+3 8 0A 04 true
+4 \N \N \N \N
+
+-- !matrix --
+1 1064 0A 0A true
+2 28 0A 08 true
+3 8 0A 08 true
+4 \N \N \N \N
+
+-- !matrix --
+1 2088 0C 02 true
+2 28 0F 00 true
+3 8 0C 00 true
+4 \N \N \N \N
+
+-- !matrix --
+1 3113 0C 06 true
+2 28 0F 04 true
+3 8 0C 04 true
+4 \N \N \N \N
+
+-- !matrix --
+1 4136 0C 0A true
+2 28 0F 08 true
+3 8 0C 08 true
+4 \N \N \N \N
+
+-- !matrix --
+1 2088 0C 02 true
+2 28 15 00 true
+3 8 0C 00 true
+4 \N \N \N \N
+
+-- !matrix --
+1 3113 0C 06 true
+2 28 15 04 true
+3 8 0C 04 true
+4 \N \N \N \N
+
+-- !matrix --
+1 4136 0C 0A true
+2 28 15 08 true
+3 8 0C 08 true
+4 \N \N \N \N
+
+-- !default --
+1 true true
+2 true true
+3 true true
+4 \N \N
+5 true true
+
+-- !varchar --
+1 true
+2 true
+3 true
+4 \N
+5 true
+
+-- !binary --
+1 true
+2 true
+3 true
+4 \N
+5 true
+
+-- !union --
+4944 4944
+
+-- !constant --
+552 552 true
+
+-- !constant_expr --
+1 true true
+2 true true
+3 true true
+4 \N \N
+5 true true
+
+-- !unfolded --
+1 true
+2 true
+3 true
+4 \N
+5 true
+
+-- !probe --
+1
+
+-- !overwrite --
+2024-01-01 1 11 552 0A 02
+2024-01-01 2 22 28 0A 00
+2024-01-01 3 33 8 0A 00
+2024-01-01 4 44 \N \N \N
+2024-02-01 5 55 4136 0C 0A
+
+-- !no_match --
+
+-- !no_pending --
+4 3 0
+
diff --git
a/regression-test/data/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample_auxiliary.out
b/regression-test/data/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample_auxiliary.out
new file mode 100644
index 00000000000..19dd5bc09f4
--- /dev/null
+++
b/regression-test/data/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample_auxiliary.out
@@ -0,0 +1,62 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !input --
+1 233
+2 176
+3 200
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !results --
+1 true true true
+2 true true true
+3 true true true
+
+-- !noop --
+2 true
+3 true
+
+-- !serialization --
+true
+
+-- !aggregate --
+2 true
+3 true
+
diff --git
a/regression-test/suites/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample.groovy
b/regression-test/suites/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample.groovy
new file mode 100644
index 00000000000..48b1c9575b9
--- /dev/null
+++
b/regression-test/suites/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample.groovy
@@ -0,0 +1,169 @@
+// 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.
+
+suite("test_datasketches_hll_downsample") {
+ // Fixtures produced by datasketches-cpp 46025e9: update(int32_t i), i in
[0,n), HLL_8.
+ // lgK=21/n=5 stays LIST; lgK=12/n=5000 is dense; lgK=12/n=0 is empty.
+ def sparse = 'AgEHFQMIBQjL18IEK/L7BoYv+Q11gWYHgbxdBg=='
+ def dense =
'CgEHDAAIAAp0NJgpTVCzQAAAAFDYUJ9AAAAAAAAAAADFBAAAAAAAAAQEAwEAAQABAAIDBQABAgADAAMDAAICAQAAAQMBAwMDAQIAAAABAgIAAgQCAgMFAAQAAgEDAAEAAAUAAgAEAwICAgYDCAMEAQIBAAACAggAAwMAAgIAAQIDAAIAAgEABAEBAwIAAwEABgQDAQICAAEBBQEDAQEDAgIABAADAAACBAIDAAMEAgADAgMAAwUAAQEEAwABAQIBAQAHAQECAAADAQECAwIDAgMBAAECAQAAAQYFAAEDBgIBAQABAQMBAQIAAwEBAwECAQEBBgAABwEAAAICAAMDBgACAgAABAMBAAACAAEDAAIDAQAAAAIBAQYCAgQAAwEFBAMCAQMBAQABBQMBAgACAwMDAgMCAAECAAECAQEBAgICAgACBAEBAQIBAwIBAAQAAQEAAgABAgIACQIDBAMGBgAABgE
[...]
+ def empty = 'AgEHDAMMAAg='
+ sql "DROP TABLE IF EXISTS test_ds_hll_downsample"
+ sql """CREATE TABLE test_ds_hll_downsample (
+ dt DATE, id INT, sk STRING, payload INT)
+ DUPLICATE KEY(dt, id)
+ PARTITION BY RANGE(dt) (
+ PARTITION p_old VALUES [('2024-01-01'), ('2024-02-01')),
+ PARTITION p_new VALUES [('2024-02-01'), ('2024-03-01')))
+ DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES ('replication_num'='1')"""
+ sql """INSERT INTO test_ds_hll_downsample VALUES
+ ('2024-01-01',1,from_base64('${dense}'),11),
+ ('2024-01-01',2,from_base64('${sparse}'),22),
+ ('2024-01-01',3,from_base64('${empty}'),33),
+ ('2024-01-01',4,NULL,44),
+ ('2024-02-01',5,from_base64('${dense}'),55)"""
+
+ for (int k : [4, 7, 10, 15, 21]) {
+ for (String t : ['HLL_4', 'hll_6', 'HlL_8']) {
+ // Header lgK and type bits are inspected through HEX, not through
UTF-8 SUBSTR(sk).
+ order_qt_matrix """SELECT id, length(out), substr(hex(out),7,2),
substr(hex(out),15,2),
+ datasketches_hll_downsample(out,${k},'${t}') = out AS
idempotent
+ FROM (SELECT id,datasketches_hll_downsample(sk,${k},'${t}') AS
out
+ FROM test_ds_hll_downsample PARTITION(p_old)) d ORDER BY
id"""
+ }
+ }
+ order_qt_default """SELECT id,
+ datasketches_hll_downsample(sk,10) =
datasketches_hll_downsample(sk,10,'HLL_4'),
+ datasketches_hll_downsample(sk,21,'HLL_8') = sk
+ FROM test_ds_hll_downsample ORDER BY id"""
+ order_qt_varchar """SELECT id,
+ datasketches_hll_downsample(CAST(sk AS VARCHAR(65533)),10) =
datasketches_hll_downsample(sk,10)
+ FROM test_ds_hll_downsample ORDER BY id"""
+ order_qt_binary """SELECT id,
+ hex(datasketches_hll_downsample(CAST(sk AS VARBINARY),10)) =
hex(datasketches_hll_downsample(sk,10))
+ FROM test_ds_hll_downsample ORDER BY id"""
+ order_qt_union """SELECT
round(datasketches_hll_union_agg(datasketches_hll_downsample(sk,10))),
+ round(datasketches_hll_union_agg(sk,10))
+ FROM test_ds_hll_downsample WHERE id=1"""
+ order_qt_constant """SELECT
length(datasketches_hll_downsample(from_base64('${dense}'),10)),
+ length(datasketches_hll_downsample(from_base64_binary('${dense}'),10)),
+ datasketches_hll_downsample(NULL,10) IS NULL"""
+ order_qt_constant_expr """SELECT id,
+ datasketches_hll_downsample(sk,8+2,concat('HLL_','4')) =
datasketches_hll_downsample(sk,10),
+ datasketches_hll_downsample(sk,10 + CAST(crc32('') AS INT),
+ IF(crc32('')=0,'HLL_4','HLL_6')) =
datasketches_hll_downsample(sk,10)
+ FROM test_ds_hll_downsample ORDER BY id"""
+ sql 'SET debug_skip_fold_constant=true'
+ order_qt_unfolded """SELECT id,
+ datasketches_hll_downsample(sk,8+2,concat('HLL_','4')) =
datasketches_hll_downsample(sk,10)
+ FROM test_ds_hll_downsample ORDER BY id"""
+ sql 'SET debug_skip_fold_constant=false'
+ order_qt_probe """SELECT id FROM test_ds_hll_downsample PARTITION(p_old)
+ WHERE datasketches_hll_downsample(sk,10) <> sk ORDER BY id LIMIT 1"""
+ sql """INSERT OVERWRITE TABLE test_ds_hll_downsample PARTITION(p_old)
+ SELECT dt,id,datasketches_hll_downsample(sk,10),payload
+ FROM test_ds_hll_downsample PARTITION(p_old)"""
+ order_qt_overwrite """SELECT
dt,id,payload,length(sk),substr(hex(sk),7,2),substr(hex(sk),15,2)
+ FROM test_ds_hll_downsample ORDER BY dt,id"""
+ order_qt_no_match """SELECT 1 FROM test_ds_hll_downsample PARTITION(p_old)
+ WHERE datasketches_hll_downsample(sk,10) <> sk LIMIT 1"""
+ order_qt_no_pending """SELECT count(*),count(sk),
+ coalesce(sum(datasketches_hll_downsample(sk,10) <> sk),0)
+ FROM test_ds_hll_downsample PARTITION(p_old)"""
+ for (String arg : ['3','22','4294967308','18446744073709551628']) {
+ test {
+ sql "SELECT datasketches_hll_downsample(sk,${arg}) FROM
test_ds_hll_downsample"
+ exception 'requires lg_k to be between 4 and 21'
+ }
+ }
+ for (String arg : ['NULL','10.5',"'10'",'id']) {
+ test {
+ sql "SELECT datasketches_hll_downsample(sk,${arg}) FROM
test_ds_hll_downsample"
+ exception 'requires lg_k to be a constant integer'
+ }
+ }
+ for (String arg : ['NULL','4','sk']) {
+ test {
+ sql "SELECT datasketches_hll_downsample(sk,10,${arg}) FROM
test_ds_hll_downsample"
+ exception 'requires tgt_type to be a constant string'
+ }
+ }
+ test {
+ sql "SELECT datasketches_hll_downsample(NULL,30)"
+ exception 'requires lg_k to be between 4 and 21'
+ }
+ test {
+ sql "SELECT datasketches_hll_downsample(sk,10,'HLL_5') FROM
test_ds_hll_downsample"
+ exception 'requires tgt_type to be one of HLL_4, HLL_6, HLL_8'
+ }
+ test {
+ sql "SELECT datasketches_hll_downsample(id,10) FROM
test_ds_hll_downsample"
+ exception 'first argument should be of STRING/VARCHAR/VARBINARY type'
+ }
+ test {
+ sql "SELECT datasketches_hll_downsample(NULL,22 + CAST(crc32('') AS
INT))"
+ exception "between 4 and 21"
+ }
+ test {
+ // The compact HLL_4 below has auxCount=0. Replace the first register
byte by AUX_TOKEN.
+ sql """SELECT datasketches_hll_downsample(
+ unhex(concat(substr(h,1,80),'0F',substr(h,83))),21)
+ FROM (SELECT
hex(datasketches_hll_downsample(from_base64('${dense}'),12)) AS h) t"""
+ exception "AUX_TOKEN without an auxiliary map"
+ }
+ // datasketches-cpp 46025e9 fixtures with complete malformed auxiliary
maps, or lgArr=32.
+ // Cover the shared deserialization entry through both the scalar and the
existing aggregate.
+ def malformedAux = [
+
['0a01070402280006000000000000f03f0000000000002f4000000000000000000f000000010000000000000000000000040000000003000050',
+ 'cannot contain an auxiliary map'],
+
['0a0107040228000a000000000000f03f0000000000002f4000000000000000000f000000010000000000000000000000000000010000000003000050',
+ 'cannot contain an auxiliary map'],
+ ['03010708200000010a000000', 'SET array exponent'],
+
['0a01070420200002565d6e7f888800400000003000002c4000000000000000000e00000002000000f0f000000000000000000000010000500000000003000054',
+ 'auxiliary array exponent'],
+
['0a01070402200002565d6e7f888800400000003000002c4000000000000000000e00000002000000f0f000000000000003000050130000540000000000000000',
+ 'duplicate slots'],
+
['0a01070402280002565d6e7f888800400000003000002c4000000000000000000e00000002000000f0f00000000000000300005013000054',
+ 'duplicate slots']
+ ]
+ for (def bad : malformedAux) {
+ test {
+ sql "SELECT datasketches_hll_downsample(unhex('${bad[0]}'),21)"
+ exception bad[1]
+ }
+ test {
+ sql "SELECT datasketches_hll_union_agg(unhex('${bad[0]}'))"
+ exception bad[1]
+ }
+ }
+ test {
+ // Keep only auxiliary slot 3, but leave the AUX_TOKEN at register 1
unchanged.
+ sql """SELECT datasketches_hll_downsample(unhex(
+
'0a01070402280002565d6e7f888800400000003000002c4000000000000000000e00000001000000f0f000000000000003000054'),21)"""
+ exception 'no matching auxiliary slot'
+ }
+ test {
+ sql """SELECT datasketches_hll_union_agg(unhex(
+
'0a01070402280002565d6e7f888800400000003000002c4000000000000000000e00000001000000f0f000000000000003000054'))"""
+ exception 'no matching auxiliary slot'
+ }
+ for (String bad : ["''", "from_base64('AA==')", "'not a sketch'"]) {
+ test {
+ sql "SELECT datasketches_hll_downsample(${bad},21)"
+ exception 'CORRUPTION'
+ }
+ }
+}
diff --git
a/regression-test/suites/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample_auxiliary.groovy
b/regression-test/suites/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample_auxiliary.groovy
new file mode 100644
index 00000000000..ae403798760
--- /dev/null
+++
b/regression-test/suites/query_p0/sql_functions/hll_functions/test_datasketches_hll_downsample_auxiliary.groovy
@@ -0,0 +1,94 @@
+// 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.
+
+suite("test_datasketches_hll_downsample_auxiliary") {
+ // datasketches-cpp 46025e9 Hll4Array(lgK=8), coupons pair(1,20),
pair(3,21).
+ // The complete register array is f0,f0 followed by 126 zero bytes.
+ def compact =
"concat('0a0107080328000286068707080800400000000300c06f400000000000000000fe00000002000000',
'f0f0', repeat('00',126), '0100005003000054')"
+ def updatable =
"concat('0a0107080320000286068707080800400000000300c06f400000000000000000fe00000002000000',
'f0f0', repeat('00',126), '00000000010000500000000003000054', repeat('00',16))"
+ sql 'DROP TABLE IF EXISTS test_ds_hll_downsample_auxiliary'
+ sql """CREATE TABLE test_ds_hll_downsample_auxiliary (id INT, sk STRING)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ('replication_num'='1')"""
+ sql """INSERT INTO test_ds_hll_downsample_auxiliary VALUES
+ (1,datasketches_hll_downsample(unhex(${compact}),8,'HLL_6')),
+ (2,unhex(${compact})), (3,unhex(${updatable}))"""
+ order_qt_input """SELECT id,length(sk) FROM
test_ds_hll_downsample_auxiliary ORDER BY id"""
+ for (int k : [4,8,21]) {
+ for (String type : ['HLL_4','HLL_6','HLL_8']) {
+ order_qt_results """SELECT id,
+ hex(datasketches_hll_downsample(sk,${k},'${type}')) =
+ hex(IF(id=3 AND ${k}>=8 AND '${type}'='HLL_4', sk,
+
datasketches_hll_downsample(unhex(${compact}),${k},'${type}'))),
+ hex(datasketches_hll_downsample(CAST(sk AS
VARBINARY),${k},'${type}')) =
+ hex(datasketches_hll_downsample(sk,${k},'${type}')),
+ hex(datasketches_hll_downsample(CAST(sk AS
VARCHAR(65533)),${k},'${type}')) =
+ hex(datasketches_hll_downsample(sk,${k},'${type}'))
+ FROM test_ds_hll_downsample_auxiliary ORDER BY id"""
+ }
+ }
+ order_qt_noop """SELECT id,datasketches_hll_downsample(sk,21,'HLL_4')=sk
+ FROM test_ds_hll_downsample_auxiliary WHERE id IN (2,3) ORDER BY id"""
+ order_qt_serialization """SELECT
+ hex(datasketches_hll_downsample(sk,4))=
+ hex(datasketches_hll_downsample(unhex(${compact}),4))
+ FROM test_ds_hll_downsample_auxiliary WHERE id=3 ORDER BY id"""
+ order_qt_aggregate """SELECT id,
+ datasketches_hll_union_agg(sk)=
+ datasketches_hll_union_agg(unhex(${compact}))
+ FROM test_ds_hll_downsample_auxiliary WHERE id IN (2,3) GROUP BY id
ORDER BY id"""
+
+ // Only register 0 is AUX_TOKEN; auxCount=1, but its compact entry was
replaced by EMPTY.
+ def bad =
"concat('0a01070803280002000000000000f03f0000000200e06f400000000000000000ff00000001000000',
'0f', repeat('00',131))"
+ for (int k : [4,21]) {
+ for (String type : ['HLL_4','HLL_6','HLL_8']) {
+ test {
+ sql "SELECT
datasketches_hll_downsample(unhex(${bad}),${k},'${type}')"
+ exception 'compact auxiliary map contains an EMPTY entry'
+ }
+ test {
+ sql "SELECT datasketches_hll_downsample(CAST(unhex(${bad}) AS
VARBINARY),${k},'${type}')"
+ exception 'compact auxiliary map contains an EMPTY entry'
+ }
+ }
+ }
+ test {
+ sql "SELECT datasketches_hll_union_agg(unhex(${bad}))"
+ exception 'compact auxiliary map contains an EMPTY entry'
+ }
+
+ // Complete updatable lgK=8 HLL_4 SET from update(int32_t i), i in [0,10).
+ // Only change the count; all 32 table slots remain present and ten are
occupied.
+ def setImage =
'03010708050000010a0000000000000081bc5d0600000000000000000000000000000000862ff90d00000000000000000000000000000000cbd7c20400000000000000000000000000000000c1e9170500000000d21673070000000034a2610e7581660700000000000000000000000000000000fc2d420a7b65e6082bf2fb06000000000000000000000000'
+ for (String count : ['01000000','0b000000']) {
+ def malformed =
"unhex(concat(substr('${setImage}',1,16),'${count}',substr('${setImage}',25)))"
+ for (int k : [4,8,21]) {
+ test {
+ sql "SELECT
datasketches_hll_downsample(${malformed},${k},'HLL_6')"
+ exception 'coupon count does not match occupied slots'
+ }
+ test {
+ sql "SELECT datasketches_hll_downsample(CAST(${malformed} AS
VARBINARY),${k},'HLL_4')"
+ exception 'coupon count does not match occupied slots'
+ }
+ }
+ test {
+ sql "SELECT datasketches_hll_union_agg(${malformed})"
+ exception 'coupon count does not match occupied slots'
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
