This is an automated email from the ASF dual-hosted git repository.
PointKernel pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datasketches-cuda.git
The following commit(s) were added to refs/heads/main by this push:
new 803bd4f Migrate Theta sketches to CUDA (#12)
803bd4f is described below
commit 803bd4f97c67f4a478c4ce4f4f1d8ed491b7b776
Author: Yunsong Wang <[email protected]>
AuthorDate: Fri Oct 23 09:29:37 2026 -0700
Migrate Theta sketches to CUDA (#12)
This PR adds a CUDA Theta sketch for numeric keys, with batch updates,
union (`merge`), intersection, A-not-B, estimates and confidence bounds,
reset, and CPU-compatible compact serialization.
The sketch keeps at most `k = 2^lg_k` unique hashes in a sorted device
array. Only nonzero hashes below a threshold, theta, are retained. When
the combined retained and incoming hashes exceed k, the sketch keeps the
smallest k and lowers theta to the first discarded hash. Theta records
the sampling fraction used to estimate the total distinct count.
Each update chunk follows three steps:
1. A fused kernel hashes the keys, rejects hashes outside theta, filters
nearby duplicates, and writes the survivors to a temporary buffer.
2. CUB sorts the survivors, removes remaining duplicates, and merges
them with the retained hashes.
3. The sketch trims the result to k entries and updates theta.
Early duplicate filtering reduces sorting work and temporary storage. It
uses warp matching and a small shared-memory table, discarding only
exact hash matches. Table collisions can let extra duplicates through;
the later sort/unique removes them, so filtering does not change the
result.
Large batches are processed in chunks so theta can tighten before the
whole batch is read. Chunk sizes adapt to theta and the observed
duplicate rate. The survivor buffer is capped at `max(2^22, 16*k)`
hashes: **32 MiB at `lg_k=12`**. This bounds update scratch
independently of batch size for a fixed k; sorting, merging, and CUB
temporary storage add to the peak. If the buffer would overflow, the
chunk makes no changes and is retried at a smaller size. Successful
chunks commit independently, so they remain installed if a later chunk
fails.
The sorted representation also supports set operations and
serialization. Union merges the arrays; intersection and A-not-B use
binary searches in the other array. Set operations respect both
sketches' sampling thresholds. Hash normalization matches the CPU
implementation and is shared with HLL. Serialization uses the CPU's
ordered, uncompressed compact-v3 format.
Device operations take an explicit stream and finish the work they
submit before returning. Counts are read back to size subsequent
allocations, so there are no async variants. Construction and
deserialization also take a memory resource, and their stream must
outlive the sketch.
The implementation supports `lg_k` from 5 to 26, custom seeds, sampling
probabilities from `2^-63` to 1, and compute capability 7.0 or newer.
Strings/byte spans and other serialization formats remain unsupported.
Deserialization requires exactly one supported image, with no trailing
bytes, whose entries fit the caller's destination k.
---
.github/workflows/ci.yml | 2 +-
.gitignore | 4 +-
CMakeLists.txt | 25 +-
README.md | 56 ++-
benchmarks/CMakeLists.txt | 31 ++
benchmarks/theta/update_bench.cu | 190 +++++++++
cmake/thirdparty/get_nvbench.cmake | 43 ++
.../detail/{hll => common}/normalizing_hasher.cuh | 24 +-
.../cuda/detail/hll/composite_finalizer.cuh | 34 +-
.../detail/hll/composite_interpolation_table.cuh | 20 +-
include/datasketches/cuda/detail/hll/policy.cuh | 6 +-
include/datasketches/cuda/detail/theta/policy.cuh | 78 ++++
.../datasketches/cuda/detail/theta/preamble.hpp | 179 +++++++++
include/datasketches/cuda/detail/theta/screen.cuh | 124 ++++++
.../datasketches/cuda/detail/theta/sketch_impl.hpp | 433 +++++++++++++++++++++
include/datasketches/cuda/detail/theta/theta.inl | 168 ++++++++
include/datasketches/cuda/detail/theta/tuning.cuh | 49 +++
include/datasketches/cuda/theta.hpp | 277 +++++++++++++
test/hll/CMakeLists.txt | 11 -
test/hll/composite_finalizer_device_test.cu | 12 +-
test/hll/normalizing_hasher_test.cpp | 4 +-
test/hll/policy_compile_test.cu | 2 +-
test/theta/CMakeLists.txt | 32 ++
test/theta/basic_sketch_test.cu | 426 ++++++++++++++++++++
test/theta/cpu_reference.cpp | 92 +++++
test/theta/cpu_reference.hpp | 44 +++
test/theta/parity_typed_test.cu | 95 +++++
test/theta/set_operations_test.cu | 137 +++++++
28 files changed, 2522 insertions(+), 76 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index da1f731..03e1b01 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -50,7 +50,7 @@ jobs:
uses: actions/cache@v4
with:
path: /root/.cache/CPM
- key: cpm-${{ hashFiles('cmake/thirdparty/*.cmake',
'cmake/CPM.cmake', 'test/hll/CMakeLists.txt') }}
+ key: cpm-${{ hashFiles('CMakeLists.txt', 'cmake/thirdparty/*.cmake',
'cmake/get_cpm.cmake', 'test/**/CMakeLists.txt') }}
restore-keys: |
cpm-
diff --git a/.gitignore b/.gitignore
index 31e117a..7271574 100644
--- a/.gitignore
+++ b/.gitignore
@@ -52,4 +52,6 @@ build/
# Sketch binary test files
-*.sk
\ No newline at end of file
+*.sk
+# clangd compile database symlinked into the source tree by the build
+compile_commands.json
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 4b6b2d2..507ba81 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -36,6 +36,9 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_LIST_DIR}")
set(_default ON)
endif()
option(BUILD_TESTS "Build unit tests" ${_default})
+# Benchmarks default OFF even at top level: they pull in nvbench, which is not
+# needed for a plain build-and-test cycle.
+option(BUILD_BENCHMARKS "Build benchmarks" OFF)
find_package(CUDAToolkit REQUIRED)
@@ -44,9 +47,10 @@ include(cmake/thirdparty/get_cccl.cmake)
# datasketches-cpp is a *public* dependency: include/datasketches/cuda/hll.hpp,
# detail/hll/preamble.hpp, detail/hll/sketch_impl.hpp, and
# detail/hll/reduction_state.hpp use upstream symbols (target_hll_type,
-# HllUtil, and INVERSE_POWERS_OF_2).
-# Configure it at top level so consumers via add_subdirectory/CPM inherit
-# the include dirs and the `datasketches` target automatically.
+# HllUtil, and INVERSE_POWERS_OF_2); Theta uses upstream MurmurHash3, seed
+# hashing, and binomial bounds. Configure it at top level so consumers via
+# add_subdirectory/CPM inherit the include dirs and the `datasketches` target
+# automatically.
include(cmake/thirdparty/get_datasketches_cpp.cmake)
# When the host C++ compiler is from a non-system prefix (e.g. a bundled GCC),
@@ -75,7 +79,22 @@ target_compile_features(datasketches_cuda INTERFACE
cxx_std_17 cuda_std_17)
if(BUILD_TESTS)
enable_testing()
+ CPMAddPackage(
+ NAME Catch2
+ GITHUB_REPOSITORY catchorg/Catch2
+ VERSION 3.5.3
+ )
+ if(Catch2_SOURCE_DIR)
+ list(APPEND CMAKE_MODULE_PATH ${Catch2_SOURCE_DIR}/extras)
+ endif()
+ include(Catch)
+
add_subdirectory(test/hll)
+ add_subdirectory(test/theta)
+endif()
+
+if(BUILD_BENCHMARKS)
+ add_subdirectory(benchmarks)
endif()
# Install rules — always present, same as datasketches-cpp's convention.
diff --git a/README.md b/README.md
index c7d2197..43ef140 100644
--- a/README.md
+++ b/README.md
@@ -38,12 +38,13 @@ If you are interested in making contributions to this site,
please see our [Comm
## Scope
-This is a header-only INTERFACE library. The current release implements
-HyperLogLog with the `HLL_8` target type, byte-compatible with
-`datasketches::hll_sketch` for round-trip serialization. Other sketch families
-and HLL variants are on the roadmap (see [Known Issues](#known-issues)).
+This is a header-only INTERFACE library. It implements HyperLogLog with the
+`HLL_8` target type and an initial GPU Theta sketch with union, intersection,
+and A-not-B. HLL images and ordered, uncompressed compact Theta v3 images are
+compatible with `datasketches-cpp`. Other sketch families and variants remain
+on the roadmap (see [Known Issues](#known-issues)).
-Public header:
+HLL example:
```cpp
#include <cuda/devices>
@@ -70,6 +71,33 @@ argument; construction and deserialization also require an
explicit device
memory resource. Streams used with `update_async` or `merge_async` must be
synchronized or otherwise ordered before the sketch is destroyed.
+Theta example:
+
+```cpp
+#include <datasketches/cuda/theta.hpp>
+
+datasketches::cuda::theta_sketch<std::uint64_t> a(stream, mr, /*lg_k=*/12);
+datasketches::cuda::theta_sketch<std::uint64_t> b(stream, mr, /*lg_k=*/12);
+a.update(stream, dev_a.begin(), dev_a.end());
+b.update(stream, dev_b.begin(), dev_b.end());
+
+a.intersect(stream, b);
+double estimate = a.get_estimate();
+
+auto bytes = a.serialize_compact(stream);
+auto cpu = datasketches::compact_theta_sketch::deserialize(
+ bytes.data(), bytes.size());
+```
+
+`theta_sketch` keeps ordered hashes in device memory. Each update chunk runs a
+kernel that hashes, screens against theta, drops nearby duplicates, and
+compacts the survivors, then folds them in with CUB radix sort, unique, and
+merge. Updates are split so theta tightens partway through a large batch and
+the survivors of one pass fit a budget of `max(2^22, 16*k)` hashes, bounding
+scratch memory independently of the batch size. Batch and set operations
+synchronize because result counts size later allocations, so prefer fewer,
+larger updates over many small ones.
+
## Build & Runtime Dependencies
Required:
@@ -77,13 +105,16 @@ Required:
- CMake >= 3.30
- A C++17-capable host compiler (GCC 13.2+ verified; older GCC may work if it
accepts C++17 and is supported by the CUDA toolkit)
- CUDA Toolkit >= 12.0 (12.4 verified)
-- An NVIDIA GPU with compute capability supported by the active CUDA Toolkit
(configured via `CMAKE_CUDA_ARCHITECTURES`; defaults to `native`)
+- An NVIDIA GPU supported by the active CUDA Toolkit (configured via
+ `CMAKE_CUDA_ARCHITECTURES`; defaults to `native`). Theta requires compute
+ capability 7.0 or newer because its screen kernel uses warp match intrinsics.
Fetched automatically via CPM at configure time (no manual install required):
- [NVIDIA/cccl](https://github.com/NVIDIA/cccl) — pinned to commit
`cba1df5786a2ffabc85887a9bfb1b7febee6232d` as synthetic version `3.5.2` while
this library develops against unreleased cudax HLL APIs. This should move to a
real CCCL release once the required APIs are tagged.
- [apache/datasketches-cpp](https://github.com/apache/datasketches-cpp)
`5.2.0` (fall-back if `find_package(DataSketches 5.0.0 CONFIG)` does not locate
a system install)
- [Catch2](https://github.com/catchorg/Catch2) `3.5.3` (test-only)
+- [NVIDIA/nvbench](https://github.com/NVIDIA/nvbench) (benchmark-only, fetched
when `BUILD_BENCHMARKS=ON`)
## Compilation and Test
@@ -98,6 +129,7 @@ ctest --test-dir build --output-on-failure
Common options:
- `-DBUILD_TESTS=OFF` to skip building tests (defaults to `ON` at top level,
`OFF` when consumed via `add_subdirectory`/CPM)
+- `-DBUILD_BENCHMARKS=ON` to build the nvbench benchmarks (defaults to `OFF`;
binaries land in `build/benchmarks`)
- `-DCMAKE_CUDA_ARCHITECTURES=<arch>` to target a specific GPU (e.g. `80` for
A100; defaults to `native`)
- `-DCPM_CCCL_SOURCE=/path/to/local/cccl` to point CPM at a local CCCL
checkout instead of fetching
@@ -137,7 +169,17 @@ additional dependency setup.
## Known Issues
- **HLL_8 only.** `HLL_4` and `HLL_6` packing are not yet implemented;
constructing with those throws `std::invalid_argument`. `AuxHashMap` (the HLL_4
exception table) is also pending.
+- **Initial Theta surface.** Theta currently accepts primitive numeric keys and
+ ordered, uncompressed compact-v3 images. Strings and byte spans, unordered or
+ update-sketch images, compressed v4, and legacy wire versions are pending.
+ Imported images must fit the destination's `2^lg_k` entries: trim a CPU
update
+ sketch before compacting it, or use a larger destination `lg_k`. Sampling
+ probabilities below `2^-63` are rejected because their hash threshold rounds
+ to zero.
- **No LIST / SET deserialization.** The wire format's small-cardinality modes
are rejected at parse. Sketches must already be in HLL mode.
- **Round-trip diverges on `FLAGS` (oooFlag) and `hipAccum`.** GPU output
always sets `oooFlag=1` (pins CPU side to the Composite estimator) and
`hipAccum=0` (no HIP tracking on parallel atomic update). All other bytes
round-trip exactly.
- **CCCL uses a synthetic development version.** Until upstream tags a CCCL
release containing the required cudax HLL policy and explicit stream /
memory-resource APIs, `cmake/thirdparty/get_cccl.cmake` uses `CPMFindPackage`
with synthetic version `3.5.2` and a pinned CCCL main commit. This prevents
automatically accepting older CCCL installs from disk while keeping an explicit
`CPM_CCCL_SOURCE` override available for development.
-- **No driver on some dev hosts.** CI gates the runtime parity test
(`parity_test.cu`); host-only tests (preamble, reduction state, normalizing
hasher, composite finalizer, policy compile) pass without a GPU.
+- **GPU runtime validation.** CI compiles the test binaries across the
supported
+ build matrix and runs them on a GPU using the latest configuration. Run
+ `ctest --test-dir build --output-on-failure` on a host with a supported GPU
+ to validate runtime behavior and CPU/GPU parity locally.
diff --git a/benchmarks/CMakeLists.txt b/benchmarks/CMakeLists.txt
new file mode 100644
index 0000000..39d2b31
--- /dev/null
+++ b/benchmarks/CMakeLists.txt
@@ -0,0 +1,31 @@
+# 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(${CMAKE_CURRENT_LIST_DIR}/../cmake/thirdparty/get_nvbench.cmake)
+
+function(ConfigureBench BENCH_NAME)
+ add_executable(${BENCH_NAME} ${ARGN})
+ target_link_libraries(${BENCH_NAME} PRIVATE
+ nvbench::main
+ datasketches::cuda
+ datasketches
+ CUDA::cudart)
+ set_target_properties(${BENCH_NAME} PROPERTIES
+ RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/benchmarks")
+endfunction()
+
+ConfigureBench(THETA_UPDATE_BENCH theta/update_bench.cu)
diff --git a/benchmarks/theta/update_bench.cu b/benchmarks/theta/update_bench.cu
new file mode 100644
index 0000000..4bea405
--- /dev/null
+++ b/benchmarks/theta/update_bench.cu
@@ -0,0 +1,190 @@
+/*
+ * 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 <cstdint>
+#include <cuda/devices>
+#include <cuda/iterator>
+#include <cuda/memory_pool>
+#include <cuda/std/cstdint>
+#include <cuda/stream>
+
+#include <thrust/device_vector.h>
+#include <thrust/execution_policy.h>
+#include <thrust/transform.h>
+
+#include <datasketches/cuda/theta.hpp>
+
+#include <nvbench/nvbench.cuh>
+
+namespace {
+
+using key_type = ::cuda::std::uint64_t;
+
+//! @brief Generates keys with a controlled distinct count and arrival
locality.
+//!
+//! Every key appears `count / distinct` times, with one extra occurrence for
+//! the first `count % distinct` keys. Scattered and grouped layouts contain
+//! exactly the same multiset: scattered repeats are `distinct` positions
apart,
+//! while grouped repeats are adjacent.
+//!
+//! Both axes matter and they are not interchangeable. Duplicate-heavy input
+//! keeps theta high for longer regardless of ordering, which exercises the
+//! chunking path. Locality separately decides whether repeats are visible to
any
+//! kernel that only sees a bounded window of consecutive keys, so a benchmark
+//! that only ever generates scattered repeats cannot distinguish an
optimization that
+//! exploits locality from one that does nothing.
+struct generate_key {
+ ::cuda::std::uint64_t distinct;
+ ::cuda::std::uint64_t repeats;
+ ::cuda::std::uint64_t extra;
+ ::cuda::std::uint64_t grouped_prefix;
+ bool grouped;
+
+ __host__ __device__ key_type operator()(::cuda::std::uint64_t index) const
noexcept
+ {
+ const auto key = !grouped
+ ? index % distinct
+ : (index < grouped_prefix ? index /
(repeats + 1)
+ : extra + (index
- grouped_prefix) / repeats);
+ ::cuda::std::uint64_t value = key + 0x9e3779b97f4a7c15ULL;
+ value ^= value >> 30;
+ value *= 0xbf58476d1ce4e5b9ULL;
+ value ^= value >> 27;
+ value *= 0x94d049bb133111ebULL;
+ value ^= value >> 31;
+ return value;
+ }
+};
+
+thrust::device_vector<key_type> make_keys(std::size_t count,
+ ::cuda::std::uint64_t distinct,
+ bool grouped = false)
+{
+ const auto repeats = count / distinct;
+ const auto extra = count % distinct;
+ thrust::device_vector<key_type> keys(count);
+ thrust::transform(thrust::device,
+ ::cuda::counting_iterator<::cuda::std::uint64_t>{0},
+ ::cuda::counting_iterator<::cuda::std::uint64_t>{count},
+ keys.begin(),
+ generate_key{distinct, repeats, extra, extra * (repeats +
1), grouped});
+ return keys;
+}
+
+//! @brief Cost of filling an empty sketch.
+//!
+//! The sketch starts at maximum theta and enters estimation mode only if the
+//! batch contains more than k distinct keys.
+void theta_update_cold(nvbench::state& state)
+{
+ const auto num_keys = static_cast<std::size_t>(state.get_int64("Keys"));
+ const auto distinct_pct = state.get_int64("DistinctPct");
+ const auto lg_k = static_cast<std::uint8_t>(state.get_int64("LgK"));
+ const auto grouped = state.get_string("Layout") == "grouped";
+ const auto distinct = std::max<std::uint64_t>(1, num_keys * distinct_pct
/ 100);
+
+ if (grouped && distinct == num_keys) {
+ state.skip("Distinct keys have the same scattered and grouped layout");
+ return;
+ }
+ const auto keys = make_keys(num_keys, distinct, grouped);
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+
+ state.add_element_count(num_keys, "Keys");
+ state.add_global_memory_reads<key_type>(num_keys);
+
+ state.exec(nvbench::exec_tag::sync, [&](nvbench::launch& launch) {
+ ::cuda::stream_ref stream{launch.get_stream()};
+ datasketches::cuda::theta_sketch<key_type> sketch(stream, mr, lg_k);
+ sketch.update(stream, keys.begin(), keys.end());
+ });
+}
+
+//! @brief Cost of feeding a fixed key count through many small update() calls.
+//!
+//! Each call carries a fixed cost in launches and synchronizations that does
not
+//! shrink with the batch, so this separates per-call overhead from streaming
+//! throughput. Sweeping Batch at a fixed Keys shows where the two cross over.
+void theta_update_batched(nvbench::state& state)
+{
+ const auto num_keys = static_cast<std::size_t>(state.get_int64("Keys"));
+ const auto batch = static_cast<std::size_t>(state.get_int64("Batch"));
+ const auto lg_k = static_cast<std::uint8_t>(state.get_int64("LgK"));
+
+ const auto keys = make_keys(num_keys, num_keys);
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+
+ state.add_element_count(num_keys, "Keys");
+ state.add_global_memory_reads<key_type>(num_keys);
+
+ state.exec(nvbench::exec_tag::sync, [&](nvbench::launch& launch) {
+ ::cuda::stream_ref stream{launch.get_stream()};
+ datasketches::cuda::theta_sketch<key_type> sketch(stream, mr, lg_k);
+ for (std::size_t offset = 0; offset < num_keys; offset += batch) {
+ const auto end = std::min(offset + batch, num_keys);
+ sketch.update(stream, keys.begin() + offset, keys.begin() + end);
+ }
+ });
+}
+
+//! @brief Cost of repeatedly merging two identical saturated sketches.
+void theta_merge(nvbench::state& state)
+{
+ const auto num_keys = static_cast<std::size_t>(state.get_int64("Keys"));
+ const auto lg_k = static_cast<std::uint8_t>(state.get_int64("LgK"));
+
+ const auto keys = make_keys(num_keys, num_keys);
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+
+ // Both sketches contain the same hashes, so this measures full-overlap
union.
+ // Repeated merges preserve their state and perform the same work each time.
+ ::cuda::stream setup_stream{::cuda::devices[0]};
+ datasketches::cuda::theta_sketch<key_type> other(setup_stream, mr, lg_k);
+ other.update(setup_stream, keys.begin(), keys.end());
+ datasketches::cuda::theta_sketch<key_type> sketch(setup_stream, mr, lg_k);
+ sketch.update(setup_stream, keys.begin(), keys.end());
+
+ state.add_element_count(std::size_t{1} << lg_k, "Retained");
+
+ state.exec(nvbench::exec_tag::sync, [&](nvbench::launch& launch) {
+ ::cuda::stream_ref stream{launch.get_stream()};
+ sketch.merge(stream, other);
+ });
+}
+
+} // namespace
+
+NVBENCH_BENCH(theta_update_cold)
+ .set_name("theta_update_cold")
+ .add_int64_axis("Keys", {1 << 20, 1 << 24, 100'000'000})
+ .add_int64_axis("DistinctPct", {100, 1})
+ .add_string_axis("Layout", {"scattered", "grouped"})
+ .add_int64_axis("LgK", {12, 20});
+
+NVBENCH_BENCH(theta_update_batched)
+ .set_name("theta_update_batched")
+ .add_int64_axis("Keys", {1 << 24})
+ .add_int64_axis("Batch", {1 << 16, 1 << 20, 1 << 24})
+ .add_int64_axis("LgK", {12});
+
+NVBENCH_BENCH(theta_merge)
+ .set_name("theta_merge_full_overlap")
+ .add_int64_axis("Keys", {1 << 22})
+ .add_int64_axis("LgK", {12, 16});
diff --git a/cmake/thirdparty/get_nvbench.cmake
b/cmake/thirdparty/get_nvbench.cmake
new file mode 100644
index 0000000..6fbf15a
--- /dev/null
+++ b/cmake/thirdparty/get_nvbench.cmake
@@ -0,0 +1,43 @@
+# 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.
+
+# NVIDIA/nvbench: benchmark harness, used only by BUILD_BENCHMARKS targets and
+# never by the installed interface library.
+#
+# nvbench brings its own CCCL. NVBench_ENABLE_CUPTI is left at its default and
+# NVBench_ENABLE_NVML is disabled so the benchmarks build on machines without
+# the NVML development package.
+#
+# Developer override: -DCPM_nvbench_SOURCE=/path/to/local/checkout
(CPM-native).
+if(NOT COMMAND CPMAddPackage)
+ include(${CMAKE_CURRENT_LIST_DIR}/../get_cpm.cmake)
+endif()
+
+function(find_and_configure_nvbench)
+ CPMAddPackage(
+ NAME nvbench
+ GITHUB_REPOSITORY NVIDIA/nvbench
+ # Keep the harness stable across builds and branch comparisons.
+ GIT_TAG a7fe2f3d47980a48a21dac5d407dd8eff007dfc4
+ OPTIONS
+ "NVBench_ENABLE_NVML OFF"
+ "NVBench_ENABLE_EXAMPLES OFF"
+ "NVBench_ENABLE_TESTING OFF"
+ )
+endfunction()
+
+find_and_configure_nvbench()
diff --git a/include/datasketches/cuda/detail/hll/normalizing_hasher.cuh
b/include/datasketches/cuda/detail/common/normalizing_hasher.cuh
similarity index 81%
rename from include/datasketches/cuda/detail/hll/normalizing_hasher.cuh
rename to include/datasketches/cuda/detail/common/normalizing_hasher.cuh
index 2417d6a..3f7db19 100644
--- a/include/datasketches/cuda/detail/hll/normalizing_hasher.cuh
+++ b/include/datasketches/cuda/detail/common/normalizing_hasher.cuh
@@ -20,7 +20,7 @@
#pragma once
// normalizing_hasher<Key>: per-key normalization that mirrors
-// datasketches-cpp's hll_sketch_alloc::update(...) overloads before hashing.
+// datasketches-cpp's HLL and Theta update(...) overloads before hashing.
//
// datasketches-cpp routes all primitive types through MurmurHash3_x64_128
// operating on a canonical 8-byte representation (HllSketch-internal.hpp):
@@ -46,7 +46,7 @@
#include <cuda/experimental/__cuco/hash_functions.cuh>
-namespace datasketches::cuda::detail::hll {
+namespace datasketches::cuda::detail {
// Inner hasher: MurmurHash3_x64_128 operating on uint64_t (the canonical
// 8-byte representation after normalization).
@@ -54,7 +54,7 @@ using _murmur_u64 =
::cuda::experimental::cuco::hash<::cuda::std::uint64_t,
::cuda::experimental::cuco::hash_algorithm::murmurhash3_x64_128>;
-//! @brief Default datasketches HLL hash seed. Single source of truth for the
+//! @brief Default datasketches hash seed. Single source of truth for the
//! seed across this project; mirrors datasketches-cpp `DEFAULT_SEED`
//! (`common_defs.hpp:34`). Kept as a self-contained `constexpr` so the device
//! headers need not include the heavy host-only `common_defs.hpp`. The value
is
@@ -102,23 +102,13 @@ class normalizing_hasher {
return ::cuda::std::bit_cast<::cuda::std::uint64_t>(
static_cast<::cuda::std::int64_t>(static_cast<_Signed>(__k)));
}
- // float: convert to double, then apply the same canonicalization as
double.
- // Inline rather than calling _canonicalize(double) because _canonicalize
is
- // a member template on _Key — calling it with a double argument while
- // _Key=float would bind to `const float&` via implicit narrowing
conversion,
- // reconstructing a float and looping forever.
- else if constexpr (::cuda::std::is_same_v<_Key, float>) {
+ // Promote float to double; canonicalize signed zero and NaN for both
types.
+ else if constexpr (::cuda::std::is_same_v<_Key, float> ||
+ ::cuda::std::is_same_v<_Key, double>) {
double __d = static_cast<double>(__k);
if (::cuda::std::isnan(__d)) { return
::cuda::std::uint64_t{0x7ff8000000000000ULL}; }
if (__d == 0.0) __d = 0.0; // -0.0 → +0.0
return ::cuda::std::bit_cast<::cuda::std::uint64_t>(__d);
- }
- // double: canonicalize -0.0 → +0.0 and NaN → Java quiet-NaN canonical.
- // Mirrors update(double) in HllSketch-internal.hpp:179-190.
- else if constexpr (::cuda::std::is_same_v<_Key, double>) {
- if (::cuda::std::isnan(__k)) { return
::cuda::std::uint64_t{0x7ff8000000000000ULL}; }
- double __d = (__k == 0.0) ? 0.0 : __k; // -0.0 → +0.0
- return ::cuda::std::bit_cast<::cuda::std::uint64_t>(__d);
} else {
// Type-dependent false to prevent eager instantiation in non-C++20 mode.
static_assert(sizeof(_Key) == 0,
@@ -129,4 +119,4 @@ class normalizing_hasher {
}
};
-} // namespace datasketches::cuda::detail::hll
+} // namespace datasketches::cuda::detail
diff --git a/include/datasketches/cuda/detail/hll/composite_finalizer.cuh
b/include/datasketches/cuda/detail/hll/composite_finalizer.cuh
index 15ce7e7..d0820f3 100644
--- a/include/datasketches/cuda/detail/hll/composite_finalizer.cuh
+++ b/include/datasketches/cuda/detail/hll/composite_finalizer.cuh
@@ -19,8 +19,8 @@
#pragma once
-#include <cstdint>
#include <cuda/std/cmath>
+#include <cuda/std/cstdint>
#include <cuda_runtime.h>
@@ -29,9 +29,10 @@
namespace datasketches::cuda::detail::hll {
//! @brief Returns the HLL raw harmonic-mean estimate.
-[[nodiscard]] __host__ __device__ inline double raw_estimate(double z,
std::uint8_t lg_k) noexcept
+[[nodiscard]] __host__ __device__ inline double raw_estimate(double z,
+
::cuda::std::uint8_t lg_k) noexcept
{
- const std::uint32_t config_k = 1u << lg_k;
+ const ::cuda::std::uint32_t config_k = 1u << lg_k;
double correction_factor;
if (lg_k == 4) {
correction_factor = 0.673;
@@ -45,7 +46,8 @@ namespace datasketches::cuda::detail::hll {
return (correction_factor * config_k * config_k) / z;
}
-[[nodiscard]] __host__ __device__ inline double harmonic_number(std::uint32_t
value) noexcept
+[[nodiscard]] __host__ __device__ inline double harmonic_number(
+ ::cuda::std::uint32_t value) noexcept
{
switch (value) {
case 0: return 0.0;
@@ -92,10 +94,10 @@ namespace datasketches::cuda::detail::hll {
}
//! @brief Returns the low-cardinality bitmap estimate.
-[[nodiscard]] __host__ __device__ inline double bitmap_estimate(std::uint32_t
num_zeroes,
- std::uint8_t
lg_k) noexcept
+[[nodiscard]] __host__ __device__ inline double
bitmap_estimate(::cuda::std::uint32_t num_zeroes,
+
::cuda::std::uint8_t lg_k) noexcept
{
- const std::uint32_t config_k = 1u << lg_k;
+ const ::cuda::std::uint32_t config_k = 1u << lg_k;
if (num_zeroes == 0) { return config_k * ::cuda::std::log(config_k / 0.5); }
return config_k * (harmonic_number(config_k) - harmonic_number(num_zeroes));
}
@@ -144,9 +146,8 @@ namespace datasketches::cuda::detail::hll {
return term0 + term1 + term2 + term3;
}
-[[nodiscard]] __host__ __device__ inline double interpolate_composite(const
double* x_values,
-
std::uint32_t y_stride,
- double
x) noexcept
+[[nodiscard]] __host__ __device__ inline double interpolate_composite(
+ const double* x_values, ::cuda::std::uint32_t y_stride, double x) noexcept
{
constexpr int length =
static_cast<int>(composite_interpolation::num_x_values);
if (x == x_values[length - 1]) { return static_cast<double>(y_stride) *
(length - 1); }
@@ -167,17 +168,16 @@ namespace datasketches::cuda::detail::hll {
//! @brief DataSketches Composite cardinality estimate for an HLL_8 register
//! array represented by its harmonic sum and zero-register count.
-[[nodiscard]] __host__ __device__ inline double composite_estimate(double z,
-
std::uint32_t num_zeroes,
-
std::uint8_t lg_k) noexcept
+[[nodiscard]] __host__ __device__ inline double composite_estimate(
+ double z, ::cuda::std::uint32_t num_zeroes, ::cuda::std::uint8_t lg_k)
noexcept
{
- const double raw_est = raw_estimate(z, lg_k);
- const double* x_values = composite_interpolation::x_values_for(lg_k);
- const std::uint32_t y_stride = composite_interpolation::y_stride_for(lg_k);
+ const double raw_est = raw_estimate(z, lg_k);
+ const double* x_values =
composite_interpolation::x_values_for(lg_k);
+ const ::cuda::std::uint32_t y_stride =
composite_interpolation::y_stride_for(lg_k);
if (raw_est < x_values[0]) { return 0.0; }
- constexpr std::uint32_t last = composite_interpolation::last_x_offset;
+ constexpr ::cuda::std::uint32_t last =
composite_interpolation::last_x_offset;
if (raw_est > x_values[last]) {
const double final_y = static_cast<double>(y_stride) * last;
return raw_est * (final_y / x_values[last]);
diff --git
a/include/datasketches/cuda/detail/hll/composite_interpolation_table.cuh
b/include/datasketches/cuda/detail/hll/composite_interpolation_table.cuh
index 62853d5..7ab192e 100644
--- a/include/datasketches/cuda/detail/hll/composite_interpolation_table.cuh
+++ b/include/datasketches/cuda/detail/hll/composite_interpolation_table.cuh
@@ -19,16 +19,16 @@
#pragma once
-#include <cstdint>
+#include <cuda/std/cstdint>
#include <cuda_runtime.h>
namespace datasketches::cuda::detail::hll::composite_interpolation {
-inline constexpr std::uint8_t min_lg_k = 4;
-inline constexpr std::uint8_t max_lg_k = 18;
-inline constexpr std::uint32_t num_x_values = 257;
-inline constexpr std::uint32_t last_x_offset = num_x_values - 1;
+inline constexpr ::cuda::std::uint8_t min_lg_k = 4;
+inline constexpr ::cuda::std::uint8_t max_lg_k = 18;
+inline constexpr ::cuda::std::uint32_t num_x_values = 257;
+inline constexpr ::cuda::std::uint32_t last_x_offset = num_x_values - 1;
// DataSketches C++ 5.2.0, commit de8553ba372e618382c2e7b44b0ffc9422b9458c:
// hll/include/CompositeInterpolationXTable-internal.hpp.
@@ -42,7 +42,7 @@ inline constexpr std::uint32_t last_x_offset = num_x_values -
1;
#define DATASKETCHES_CUDA_HLL_TABLE_DECL inline constexpr
#endif
-DATASKETCHES_CUDA_HLL_TABLE_DECL std::uint32_t y_strides[] = {
+DATASKETCHES_CUDA_HLL_TABLE_DECL ::cuda::std::uint32_t y_strides[] = {
1, 2, 3, 5, 10, 20, 40, 80, 160, 320, 640, 1280, 2560, 5120, 10240};
DATASKETCHES_CUDA_HLL_TABLE_DECL double x_values[15][num_x_values] = {
@@ -51,17 +51,19 @@ DATASKETCHES_CUDA_HLL_TABLE_DECL double
x_values[15][num_x_values] = {
#undef DATASKETCHES_CUDA_HLL_TABLE_DECL
-[[nodiscard]] __host__ __device__ constexpr bool valid_lg_k(std::uint8_t lg_k)
noexcept
+[[nodiscard]] __host__ __device__ constexpr bool
valid_lg_k(::cuda::std::uint8_t lg_k) noexcept
{
return lg_k >= min_lg_k && lg_k <= max_lg_k;
}
-[[nodiscard]] __host__ __device__ inline const double*
x_values_for(std::uint8_t lg_k) noexcept
+[[nodiscard]] __host__ __device__ inline const double* x_values_for(
+ ::cuda::std::uint8_t lg_k) noexcept
{
return x_values[lg_k - min_lg_k];
}
-[[nodiscard]] __host__ __device__ inline std::uint32_t
y_stride_for(std::uint8_t lg_k) noexcept
+[[nodiscard]] __host__ __device__ inline ::cuda::std::uint32_t y_stride_for(
+ ::cuda::std::uint8_t lg_k) noexcept
{
return y_strides[lg_k - min_lg_k];
}
diff --git a/include/datasketches/cuda/detail/hll/policy.cuh
b/include/datasketches/cuda/detail/hll/policy.cuh
index 506e6bd..4a50291 100644
--- a/include/datasketches/cuda/detail/hll/policy.cuh
+++ b/include/datasketches/cuda/detail/hll/policy.cuh
@@ -29,8 +29,8 @@
#include <cuda/experimental/__cuco/hash_functions.cuh>
+#include <datasketches/cuda/detail/common/normalizing_hasher.cuh>
#include <datasketches/cuda/detail/hll/composite_finalizer.cuh>
-#include <datasketches/cuda/detail/hll/normalizing_hasher.cuh>
namespace datasketches::cuda::detail::hll {
@@ -61,13 +61,13 @@ struct policy {
// MurmurHash3_x64_128, so GPU coupons are byte-compatible with the CPU
sketch
// for all supported types. Unsupported types produce a compile error in
// normalizing_hasher::_canonicalize.
- using hasher = normalizing_hasher<_Key>;
+ using hasher = detail::normalizing_hasher<_Key>;
using hash_result_type = __uint128_t;
using register_type = ::cuda::std::int32_t;
//! @brief Default datasketches HLL hash seed (`common_defs.hpp:34`).
//! Sourced from the single project-wide constant in
`normalizing_hasher.cuh`.
- static constexpr ::cuda::std::uint64_t default_seed =
detail::hll::default_seed;
+ static constexpr ::cuda::std::uint64_t default_seed = detail::default_seed;
hasher hasher_{default_seed};
diff --git a/include/datasketches/cuda/detail/theta/policy.cuh
b/include/datasketches/cuda/detail/theta/policy.cuh
new file mode 100644
index 0000000..486bda9
--- /dev/null
+++ b/include/datasketches/cuda/detail/theta/policy.cuh
@@ -0,0 +1,78 @@
+/*
+ * 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 <cuda/std/algorithm>
+#include <cuda/std/cstddef>
+#include <cuda/std/cstdint>
+
+#include <cuda_runtime.h>
+
+#include <datasketches/cuda/detail/common/normalizing_hasher.cuh>
+
+namespace datasketches::cuda::detail::theta {
+
+inline constexpr ::cuda::std::uint64_t max_theta = 0x7fffffffffffffffULL;
+inline constexpr ::cuda::std::uint8_t min_lg_k = 5;
+inline constexpr ::cuda::std::uint8_t max_lg_k = 26;
+inline constexpr ::cuda::std::uint8_t default_lg_k = 12;
+
+template <class Key>
+struct theta_hash {
+ detail::normalizing_hasher<Key> hasher;
+
+ __host__ __device__ explicit constexpr theta_hash(::cuda::std::uint64_t
seed) noexcept
+ : hasher(seed)
+ {
+ }
+
+ [[nodiscard]] __host__ __device__ constexpr ::cuda::std::uint64_t operator()(
+ const Key& key) const noexcept
+ {
+ // DataSketches Theta uses the low MurmurHash3 word and an unsigned shift
+ // by one, reserving zero as an empty-table sentinel.
+ return static_cast<::cuda::std::uint64_t>(hasher(key)) >> 1;
+ }
+};
+
+struct screen_hash {
+ ::cuda::std::uint64_t theta;
+
+ [[nodiscard]] __host__ __device__ constexpr bool operator()(
+ ::cuda::std::uint64_t hash) const noexcept
+ {
+ return hash != 0 && hash < theta;
+ }
+};
+
+struct membership_filter {
+ const ::cuda::std::uint64_t* other;
+ ::cuda::std::size_t other_size;
+ ::cuda::std::uint64_t theta;
+ bool keep_matches;
+
+ [[nodiscard]] __host__ __device__ bool operator()(::cuda::std::uint64_t
hash) const noexcept
+ {
+ if (!screen_hash{theta}(hash)) return false;
+ return ::cuda::std::binary_search(other, other + other_size, hash) ==
keep_matches;
+ }
+};
+
+} // namespace datasketches::cuda::detail::theta
diff --git a/include/datasketches/cuda/detail/theta/preamble.hpp
b/include/datasketches/cuda/detail/theta/preamble.hpp
new file mode 100644
index 0000000..cb552e6
--- /dev/null
+++ b/include/datasketches/cuda/detail/theta/preamble.hpp
@@ -0,0 +1,179 @@
+/*
+ * 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 <cstring>
+#include <cuda/std/cstddef>
+#include <cuda/std/cstdint>
+#include <cuda/std/span>
+#include <stdexcept>
+#include <vector>
+
+#include <datasketches/cuda/detail/theta/policy.cuh>
+
+#include <MurmurHash3.h>
+
+namespace datasketches::cuda::detail::theta {
+
+inline constexpr ::cuda::std::uint8_t uncompressed_serial_version = 3;
+inline constexpr ::cuda::std::uint8_t sketch_type = 3;
+
+inline constexpr ::cuda::std::uint8_t flag_big_endian = 1U << 0;
+inline constexpr ::cuda::std::uint8_t flag_read_only = 1U << 1;
+inline constexpr ::cuda::std::uint8_t flag_empty = 1U << 2;
+inline constexpr ::cuda::std::uint8_t flag_compact = 1U << 3;
+inline constexpr ::cuda::std::uint8_t flag_ordered = 1U << 4;
+
+struct compact_image {
+ bool empty;
+ ::cuda::std::uint64_t theta;
+ std::vector<::cuda::std::uint64_t> entries;
+};
+
+template <class T>
+void write_at(std::vector<::cuda::std::uint8_t>& bytes, ::cuda::std::size_t
offset, const T& value)
+{
+ std::memcpy(bytes.data() + offset, &value, sizeof(T));
+}
+
+template <class T>
+T read_at(::cuda::std::span<const ::cuda::std::uint8_t> bytes,
::cuda::std::size_t offset)
+{
+ T value{};
+ std::memcpy(&value, bytes.data() + offset, sizeof(T));
+ return value;
+}
+
+inline std::vector<::cuda::std::uint8_t> serialize_compact_v3(
+ bool empty,
+ ::cuda::std::uint16_t seed_hash,
+ ::cuda::std::uint64_t theta,
+ const std::vector<::cuda::std::uint64_t>& entries)
+{
+ const bool estimation = !empty && theta < max_theta;
+ const ::cuda::std::uint8_t preamble_longs =
+ estimation ? 3 : (empty || entries.size() == 1 ? 1 : 2);
+ const ::cuda::std::size_t preamble_bytes =
+ ::cuda::std::size_t{preamble_longs} * sizeof(::cuda::std::uint64_t);
+
+ std::vector<::cuda::std::uint8_t> bytes(
+ preamble_bytes + entries.size() * sizeof(::cuda::std::uint64_t), 0);
+ bytes[0] = preamble_longs;
+ bytes[1] = uncompressed_serial_version;
+ bytes[2] = sketch_type;
+ bytes[5] = flag_read_only | flag_compact | flag_ordered | (empty ?
flag_empty : 0);
+ write_at(bytes, 6, seed_hash);
+ if (preamble_longs > 1) {
+ write_at(bytes, 8, static_cast<::cuda::std::uint32_t>(entries.size()));
+ }
+ if (estimation) write_at(bytes, 16, theta);
+ if (!entries.empty()) {
+ std::memcpy(bytes.data() + preamble_bytes,
+ entries.data(),
+ entries.size() * sizeof(::cuda::std::uint64_t));
+ }
+ return bytes;
+}
+
+inline compact_image parse_compact_v3(::cuda::std::span<const
::cuda::std::uint8_t> bytes,
+ ::cuda::std::uint64_t seed)
+{
+ if (bytes.size() < sizeof(::cuda::std::uint64_t)) {
+ throw std::invalid_argument("theta_sketch::deserialize: image shorter than
one preamble long");
+ }
+
+ const auto preamble_longs = bytes[0];
+ if (preamble_longs < 1 || preamble_longs > 3) {
+ throw std::invalid_argument("theta_sketch::deserialize: invalid preamble
length");
+ }
+ const ::cuda::std::size_t preamble_bytes =
+ ::cuda::std::size_t{preamble_longs} * sizeof(::cuda::std::uint64_t);
+ if (bytes.size() < preamble_bytes) {
+ throw std::invalid_argument("theta_sketch::deserialize: truncated
preamble");
+ }
+
+ if (bytes[1] != uncompressed_serial_version) {
+ throw std::invalid_argument(
+ "theta_sketch::deserialize supports only uncompressed serialization
version 3");
+ }
+ if (bytes[2] != sketch_type) {
+ throw std::invalid_argument("theta_sketch::deserialize: unexpected sketch
type");
+ }
+
+ const auto flags = bytes[5];
+ if ((flags & flag_big_endian) != 0) {
+ throw std::invalid_argument("theta_sketch::deserialize: big-endian images
are unsupported");
+ }
+ if ((flags & flag_compact) == 0 || (flags & flag_ordered) == 0) {
+ throw std::invalid_argument(
+ "theta_sketch::deserialize requires an ordered compact Theta image");
+ }
+
+ const bool empty = (flags & flag_empty) != 0;
+ const ::cuda::std::uint16_t seed_hash =
read_at<::cuda::std::uint16_t>(bytes, 6);
+ if (!empty && seed_hash != ::compute_seed_hash(seed)) {
+ throw std::invalid_argument("theta_sketch::deserialize: seed hash
mismatch");
+ }
+
+ ::cuda::std::uint32_t count = 0;
+ ::cuda::std::uint64_t theta = max_theta;
+ if (empty) {
+ if (preamble_longs != 1) {
+ throw std::invalid_argument("theta_sketch::deserialize: empty image has
invalid preamble");
+ }
+ } else if (preamble_longs == 1) {
+ count = 1;
+ } else {
+ count = read_at<::cuda::std::uint32_t>(bytes, 8);
+ if (preamble_longs == 3) theta = read_at<::cuda::std::uint64_t>(bytes, 16);
+ }
+
+ if (theta == 0 || theta > max_theta) {
+ throw std::invalid_argument("theta_sketch::deserialize: theta is out of
range");
+ }
+ if (preamble_longs == 3 && theta == max_theta) {
+ throw std::invalid_argument("theta_sketch::deserialize: estimation image
has maximum theta");
+ }
+
+ const ::cuda::std::size_t expected =
+ preamble_bytes + ::cuda::std::size_t{count} *
sizeof(::cuda::std::uint64_t);
+ if (bytes.size() != expected) {
+ throw std::invalid_argument("theta_sketch::deserialize: image size
mismatch");
+ }
+
+ compact_image image{empty, theta, std::vector<::cuda::std::uint64_t>(count)};
+ if (count != 0) {
+ std::memcpy(
+ image.entries.data(), bytes.data() + preamble_bytes, count *
sizeof(::cuda::std::uint64_t));
+ }
+ for (::cuda::std::size_t i = 0; i < image.entries.size(); ++i) {
+ const auto hash = image.entries[i];
+ if (hash == 0 || hash >= theta) {
+ throw std::invalid_argument("theta_sketch::deserialize: retained hash is
out of range");
+ }
+ if (i != 0 && image.entries[i - 1] >= hash) {
+ throw std::invalid_argument(
+ "theta_sketch::deserialize: retained hashes are not strictly ordered");
+ }
+ }
+ return image;
+}
+
+} // namespace datasketches::cuda::detail::theta
diff --git a/include/datasketches/cuda/detail/theta/screen.cuh
b/include/datasketches/cuda/detail/theta/screen.cuh
new file mode 100644
index 0000000..73bdf43
--- /dev/null
+++ b/include/datasketches/cuda/detail/theta/screen.cuh
@@ -0,0 +1,124 @@
+/*
+ * 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 <cuda/atomic>
+#include <cuda/std/cstddef>
+#include <cuda/std/cstdint>
+#include <cuda/utility>
+
+#include <cuda_runtime.h>
+
+#include <cub/warp/warp_scan.cuh>
+
+#include <datasketches/cuda/detail/theta/policy.cuh>
+#include <datasketches/cuda/detail/theta/tuning.cuh>
+
+namespace datasketches::cuda::detail::theta {
+
+// Hash, screen, and compact without materializing one hash per input key.
+// Warp and block duplicate filters are best-effort: only exact matches are
+// dropped; missed duplicates are removed by the subsequent sort/unique.
+// Count all survivors but guard writes by capacity so the caller can retry.
+// Each warp reserves one output run for all its items, amortizing atomics
+// without a block-wide synchronization at each item.
+template <class KeyIt, class Hasher>
+__global__ void screen_kernel(KeyIt keys,
+ ::cuda::std::size_t num_keys,
+ Hasher hasher,
+ ::cuda::std::uint64_t theta,
+ ::cuda::std::uint64_t* __restrict__ out,
+ ::cuda::std::size_t capacity,
+ ::cuda::std::uint64_t* __restrict__ out_count)
+{
+ constexpr ::cuda::std::uint32_t warp_width = 32;
+ constexpr ::cuda::std::uint32_t full_mask = 0xffffffffu;
+ const ::cuda::std::uint32_t lane = threadIdx.x % warp_width;
+
+ using warp_scan = cub::WarpScan<::cuda::std::int32_t>;
+ __shared__ typename warp_scan::TempStorage
scan_storage[tuning::block_threads / warp_width];
+
+ // A slot contains the entire hash; it publishes no other data. Relaxed
+ // atomics suffice: stale observations can only emit another duplicate.
+ using filter_cell = ::cuda::atomic_ref<::cuda::std::uint64_t,
::cuda::thread_scope_block>;
+ __shared__ ::cuda::std::uint64_t filter[tuning::filter_slots];
+ for (auto i = static_cast<::cuda::std::size_t>(threadIdx.x); i <
tuning::filter_slots;
+ i += tuning::block_threads) {
+ filter[i] = 0;
+ }
+ __syncthreads();
+
+ // A grid from tuning::grid_size covers the input in one step; the stride
loop
+ // keeps the kernel correct for any smaller grid a caller might pass.
+ const auto stride = static_cast<::cuda::std::size_t>(gridDim.x) *
tuning::tile_keys;
+
+ for (auto base = static_cast<::cuda::std::size_t>(blockIdx.x) *
tuning::tile_keys;
+ base < num_keys;
+ base += stride) {
+ ::cuda::std::uint64_t hashes[tuning::items_per_thread];
+ ::cuda::std::int32_t mine = 0;
+
+ ::cuda::static_for<tuning::items_per_thread>([&](auto i) {
+ // strided within the tile so every load step stays fully coalesced
+ const auto idx = base + i * tuning::block_threads + threadIdx.x;
+ hashes[i] = 0;
+ const ::cuda::std::uint64_t hash =
+ idx < num_keys ? hasher(keys[idx]) : ::cuda::std::uint64_t{0};
+ const bool survives = screen_hash{theta}(hash);
+
+ // Stage one: one lane per distinct hash in this warp continues.
+ const auto peers = __match_any_sync(full_mask, survives ? hash :
::cuda::std::uint64_t{0});
+ const bool leader =
+ survives &&
static_cast<::cuda::std::uint32_t>(__ffs(static_cast<int>(peers)) - 1) == lane;
+
+ // Stage two: drop it if this block emitted the same hash recently.
+ if (leader) {
+ filter_cell cell{filter[hash % tuning::filter_slots]};
+ if (cell.load(::cuda::memory_order_relaxed) != hash) {
+ cell.store(hash, ::cuda::memory_order_relaxed);
+ hashes[i] = hash;
+ ++mine;
+ }
+ }
+ });
+
+ // One atomic per warp: the warp scan yields each lane's offset and the
total.
+ ::cuda::std::int32_t scan = 0;
+ ::cuda::std::int32_t warp_total = 0;
+ warp_scan(scan_storage[threadIdx.x / warp_width]).ExclusiveSum(mine, scan,
warp_total);
+ if (warp_total == 0) { continue; }
+
+ ::cuda::std::uint64_t warp_base = 0;
+ if (lane == 0) {
+ warp_base =
+ ::cuda::atomic_ref<::cuda::std::uint64_t,
::cuda::thread_scope_device>{*out_count}
+ .fetch_add(static_cast<::cuda::std::uint64_t>(warp_total),
::cuda::memory_order_relaxed);
+ }
+ auto cursor = __shfl_sync(full_mask, warp_base, 0) +
static_cast<::cuda::std::uint64_t>(scan);
+ ::cuda::static_for<tuning::items_per_thread>([&](auto i) {
+ if (hashes[i] != 0) {
+ if (cursor < capacity) { out[cursor] = hashes[i]; }
+ ++cursor;
+ }
+ });
+ }
+}
+
+} // namespace datasketches::cuda::detail::theta
diff --git a/include/datasketches/cuda/detail/theta/sketch_impl.hpp
b/include/datasketches/cuda/detail/theta/sketch_impl.hpp
new file mode 100644
index 0000000..bbf6263
--- /dev/null
+++ b/include/datasketches/cuda/detail/theta/sketch_impl.hpp
@@ -0,0 +1,433 @@
+/*
+ * 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 <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <cuda/memory_pool>
+#include <cuda/std/bit>
+#include <cuda/std/functional>
+#include <cuda/std/span>
+#include <cuda/stream>
+#include <stdexcept>
+#include <utility>
+#include <vector>
+
+#include <cuda_runtime.h>
+
+#include <cub/device/device_merge.cuh>
+#include <cub/device/device_radix_sort.cuh>
+#include <cub/device/device_select.cuh>
+
+#include <cuda/experimental/container.cuh>
+#include <cuda/experimental/execution.cuh>
+#include <cuda/experimental/memory_resource.cuh>
+
+#include <datasketches/cuda/detail/common/error.hpp>
+#include <datasketches/cuda/detail/theta/policy.cuh>
+#include <datasketches/cuda/detail/theta/preamble.hpp>
+#include <datasketches/cuda/detail/theta/screen.cuh>
+
+namespace datasketches::cuda::detail::theta {
+
+template <class Key, class MR = ::cuda::device_memory_pool_ref>
+struct sketch_impl {
+ using hash_result_type = std::uint64_t;
+ using count_type = std::uint64_t;
+ using hash_buffer_type = ::cuda::device_buffer<hash_result_type>;
+ using count_buffer_type = ::cuda::device_buffer<count_type>;
+ using env_type =
::cuda::experimental::env_t<::cuda::mr::device_accessible>;
+
+ struct buffer_result {
+ hash_buffer_type data;
+ std::size_t size;
+ };
+
+ std::uint8_t lg_k_;
+ std::uint64_t seed_;
+ std::uint64_t initial_theta_;
+ std::uint64_t theta_;
+ bool is_empty_;
+ std::size_t survivor_budget_; // Screen output capacity in hashes; tests
lower it.
+ ::cuda::stream_ref allocation_stream_;
+ MR mr_;
+ hash_buffer_type hashes_;
+
+ sketch_impl(::cuda::stream_ref stream, MR mr, std::uint8_t lg_k,
std::uint64_t seed, float p)
+ : lg_k_(check_lg_k(lg_k)),
+ seed_(seed),
+ initial_theta_(checked_theta(p)),
+ theta_(initial_theta_),
+ is_empty_(true),
+ survivor_budget_(std::max(tuning::min_survivor_budget,
tuning::chunk_target_multiple * k())),
+ allocation_stream_(stream),
+ mr_(std::move(mr)),
+ hashes_(make_hash_buffer(stream, 0))
+ {
+ }
+
+ sketch_impl(const sketch_impl&) = delete;
+ sketch_impl& operator=(const sketch_impl&) = delete;
+ sketch_impl(sketch_impl&&) = default;
+ sketch_impl& operator=(sketch_impl&&) = default;
+
+ static std::uint8_t check_lg_k(std::uint8_t lg_k)
+ {
+ if (lg_k < min_lg_k || lg_k > max_lg_k) {
+ throw std::invalid_argument("theta_sketch lg_k must be in [5, 26]");
+ }
+ return lg_k;
+ }
+
+ static std::uint64_t checked_theta(float p)
+ {
+ const auto theta = p > 0.0F && p < 1.0F
+ ?
static_cast<std::uint64_t>(static_cast<double>(max_theta) * p)
+ : max_theta;
+ if (!(p > 0.0F && p <= 1.0F) || theta == 0) {
+ throw std::invalid_argument(
+ "theta_sketch sampling probability must be in (0, 1] and yield a
nonzero theta");
+ }
+ return theta;
+ }
+
+ [[nodiscard]] std::size_t k() const noexcept { return std::size_t{1} <<
lg_k_; }
+
+ [[nodiscard]] std::uint64_t effective_theta() const noexcept
+ {
+ return is_empty_ ? max_theta : theta_;
+ }
+
+ [[nodiscard]] double theta_rate() const noexcept
+ {
+ return static_cast<double>(theta_) / static_cast<double>(max_theta);
+ }
+
+ [[nodiscard]] env_type env(::cuda::stream_ref stream) const { return
env_type{mr_, stream}; }
+
+ [[nodiscard]] hash_buffer_type make_hash_buffer(::cuda::stream_ref stream,
+ std::size_t count) const
+ {
+ return ::cuda::make_buffer<hash_result_type,
::cuda::mr::device_accessible>(
+ stream, mr_, count, ::cuda::no_init);
+ }
+
+ [[nodiscard]] count_buffer_type make_count_buffer(::cuda::stream_ref stream)
const
+ {
+ return ::cuda::make_buffer<count_type, ::cuda::mr::device_accessible>(
+ stream, mr_, 1, count_type{0});
+ }
+
+ [[nodiscard]] static std::size_t read_count(::cuda::stream_ref stream,
+ const count_buffer_type& count)
+ {
+ count_type host_count{};
+ DATASKETCHES_CUDA_TRY(cudaMemcpyAsync(
+ &host_count, count.data(), sizeof(host_count), cudaMemcpyDeviceToHost,
stream.get()));
+ stream.sync();
+ return static_cast<std::size_t>(host_count);
+ }
+
+ template <class Predicate>
+ [[nodiscard]] buffer_result select(::cuda::stream_ref stream,
+ const hash_result_type* input,
+ std::size_t count,
+ Predicate predicate) const
+ {
+ auto output = make_hash_buffer(stream, count);
+ if (count == 0) return {std::move(output), 0};
+ auto selected = make_count_buffer(stream);
+ DATASKETCHES_CUDA_TRY(cub::DeviceSelect::If(input,
+ output.data(),
+ selected.data(),
+
static_cast<std::int64_t>(count),
+ predicate,
+ env(stream)));
+ return {std::move(output), read_count(stream, selected)};
+ }
+
+ [[nodiscard]] buffer_result unique_sorted(::cuda::stream_ref stream,
+ const hash_result_type* sorted,
+ std::size_t count) const
+ {
+ auto output = make_hash_buffer(stream, count);
+ auto selected = make_count_buffer(stream);
+ DATASKETCHES_CUDA_TRY(cub::DeviceSelect::Unique(
+ sorted, output.data(), selected.data(),
static_cast<std::int64_t>(count), env(stream)));
+ return {std::move(output), read_count(stream, selected)};
+ }
+
+ [[nodiscard]] buffer_result sort_unique(::cuda::stream_ref stream,
+ hash_buffer_type input,
+ std::size_t count,
+ std::uint64_t bound) const
+ {
+ auto alternate = make_hash_buffer(stream, count);
+ cub::DoubleBuffer<hash_result_type> keys(input.data(), alternate.data());
+ DATASKETCHES_CUDA_TRY(cub::DeviceRadixSort::SortKeys(
+ keys, static_cast<std::int64_t>(count), 0,
::cuda::std::bit_width(bound), env(stream)));
+ return unique_sorted(stream, keys.Current(), count);
+ }
+
+ // Count every survivor but bound writes; an oversized result must be
retried.
+ template <class RandomAccessIt>
+ [[nodiscard]] buffer_result screen(::cuda::stream_ref stream,
+ RandomAccessIt first,
+ std::size_t count) const
+ {
+ const std::size_t capacity = std::min(count, survivor_budget_);
+ auto output = make_hash_buffer(stream, capacity);
+ auto selected = make_count_buffer(stream);
+ screen_kernel<<<tuning::grid_size(count), tuning::block_threads, 0,
stream.get()>>>(
+ first, count, theta_hash<Key>{seed_}, theta_, output.data(), capacity,
selected.data());
+ DATASKETCHES_CUDA_TRY(cudaGetLastError());
+ return {std::move(output), read_count(stream, selected)};
+ }
+
+ //! @brief Merge sorted, individually unique ranges; only cross-range
duplicates remain.
+ [[nodiscard]] buffer_result merge_unique(::cuda::stream_ref stream,
+ const hash_result_type* first,
+ std::size_t first_size,
+ const hash_result_type* second,
+ std::size_t second_size) const
+ {
+ const std::size_t merged_size = first_size + second_size;
+ auto merged = make_hash_buffer(stream, merged_size);
+ if (merged_size == 0) return {std::move(merged), 0};
+
+ if (first_size == 0 || second_size == 0) {
+ DATASKETCHES_CUDA_TRY(cudaMemcpyAsync(merged.data(),
+ first_size == 0 ? second : first,
+ merged_size *
sizeof(hash_result_type),
+ cudaMemcpyDeviceToDevice,
+ stream.get()));
+ return {std::move(merged), merged_size};
+ }
+ DATASKETCHES_CUDA_TRY(cub::DeviceMerge::MergeKeys(first,
+
static_cast<std::int64_t>(first_size),
+ second,
+
static_cast<std::int64_t>(second_size),
+ merged.data(),
+ ::cuda::std::less<>{},
+ env(stream)));
+ return unique_sorted(stream, merged.data(), merged_size);
+ }
+
+ void install(::cuda::stream_ref stream, buffer_result input, std::uint64_t
theta, bool empty)
+ {
+ const auto retained = std::min(input.size, k());
+ const auto* data = input.data.data();
+ const bool reuse = input.data.size() == retained;
+ auto next = reuse ? std::move(input.data) :
make_hash_buffer(stream, retained);
+ if (!reuse && retained != 0) {
+ DATASKETCHES_CUDA_TRY(cudaMemcpyAsync(next.data(),
+ data,
+ retained *
sizeof(hash_result_type),
+ cudaMemcpyDeviceToDevice,
+ stream.get()));
+ }
+ if (input.size > k()) {
+ DATASKETCHES_CUDA_TRY(
+ cudaMemcpyAsync(&theta, data + k(), sizeof(theta),
cudaMemcpyDeviceToHost, stream.get()));
+ }
+ stream.sync();
+ next.set_stream(allocation_stream_);
+ hashes_ = std::move(next);
+ theta_ = empty ? initial_theta_ : theta;
+ is_empty_ = empty;
+ }
+
+ // Predict survivors from theta and the last chunk's duplicate rate. Hedge
+ // that rate and cap growth because the distribution can change mid-update.
+ // At maximum theta, double chunks so distinct inputs tighten theta early.
+ // Guarded writes and halving on overflow guarantee progress for any input.
+ [[nodiscard]] std::size_t next_chunk(std::size_t remaining,
+ std::size_t previous,
+ double duplicate_factor) const noexcept
+ {
+ const double hedged = std::min(1.0, tuning::duplicate_hedge *
duplicate_factor);
+ const double rate = theta_rate() * hedged;
+ const double target = rate > 0.0
+ ? static_cast<double>(survivor_budget_) /
(tuning::budget_margin * rate)
+ : static_cast<double>(remaining);
+ std::size_t sized =
+ target >= static_cast<double>(remaining) ? remaining :
static_cast<std::size_t>(target);
+ if (hedged < 1.0 && previous != 0 && sized / tuning::max_chunk_growth >
previous) {
+ sized = previous * tuning::max_chunk_growth;
+ }
+ sized = std::max<std::size_t>(sized, 1);
+ if (theta_ == max_theta) {
+ const auto doubling =
+ std::max({tuning::chunk_target_multiple * k(), tuning::min_chunk_keys,
previous * 2});
+ return std::min({doubling, sized, remaining});
+ }
+ return std::min(std::max(sized, tuning::min_chunk_keys), remaining);
+ }
+
+ template <class RandomAccessIt>
+ [[nodiscard]] std::size_t update_chunk(::cuda::stream_ref stream,
+ RandomAccessIt first,
+ std::size_t count)
+ {
+ auto screened = screen(stream, first, count);
+ if (screened.size > screened.data.size() || screened.size == 0) return
screened.size;
+
+ auto incoming = sort_unique(stream, std::move(screened.data),
screened.size, theta_);
+ auto combined =
+ hashes_.empty()
+ ? std::move(incoming)
+ : merge_unique(stream, hashes_.data(), hashes_.size(),
incoming.data.data(), incoming.size);
+ install(stream, std::move(combined), theta_, false);
+ return screened.size;
+ }
+
+ template <class RandomAccessIt>
+ void update(::cuda::stream_ref stream, RandomAccessIt first, RandomAccessIt
last)
+ {
+ const auto distance = last - first;
+ if (distance < 0) {
+ throw std::invalid_argument("theta_sketch::update requires a
non-negative range");
+ }
+ const auto count = static_cast<std::size_t>(distance);
+
+ std::size_t offset = 0;
+ std::size_t previous = 0;
+ double duplicate_factor = 1.0;
+ while (offset < count) {
+ auto chunk = next_chunk(count - offset, previous, duplicate_factor);
+ // Theta is fixed until something is installed, so a pass's survivors per
+ // key over the theta fraction it was screened at is its duplicate
factor.
+ const double rate = theta_rate();
+ auto outcome = update_chunk(stream, first + offset, chunk);
+ while (outcome > survivor_budget_) {
+ duplicate_factor = static_cast<double>(outcome) /
static_cast<double>(chunk) / rate;
+ // The exact survivor count of the failed pass usually lands the retry
+ // within the budget at once; halving guarantees progress regardless,
+ // since a chunk of at most the budget cannot overflow.
+ chunk = std::min(chunk / 2, next_chunk(count - offset, previous,
duplicate_factor));
+ outcome = update_chunk(stream, first + offset, chunk);
+ }
+ // Commit the empty flag only after a chunk succeeds. A chunk whose keys
+ // are all rejected still counts as seen, but a failed first allocation
+ // must leave an empty sketch unchanged.
+ is_empty_ = false;
+ duplicate_factor = static_cast<double>(outcome) /
static_cast<double>(chunk) / rate;
+ offset += chunk;
+ previous = chunk;
+ }
+ }
+
+ template <class OtherMR>
+ void merge(::cuda::stream_ref stream, const sketch_impl<Key, OtherMR>& other)
+ {
+ if (other.is_empty_) return;
+ if (::compute_seed_hash(seed_) != ::compute_seed_hash(other.seed_)) {
+ throw std::invalid_argument("theta_sketch::merge: seed hash mismatch");
+ }
+
+ const std::uint64_t theta = std::min(theta_, other.effective_theta());
+ auto combined = merge_unique(
+ stream, hashes_.data(), hashes_.size(), other.hashes_.data(),
other.hashes_.size());
+ auto screened = select(stream, combined.data.data(), combined.size,
screen_hash{theta});
+ install(stream, std::move(screened), theta, false);
+ }
+
+ template <class OtherMR>
+ void intersect(::cuda::stream_ref stream, const sketch_impl<Key, OtherMR>&
other)
+ {
+ if (is_empty_) return;
+ if (other.is_empty_) {
+ reset(stream);
+ return;
+ }
+ retain_matching(stream, other, true);
+ }
+
+ template <class OtherMR>
+ void a_not_b(::cuda::stream_ref stream, const sketch_impl<Key, OtherMR>&
other)
+ {
+ if (is_empty_ || (!hashes_.empty() && other.is_empty_)) return;
+ retain_matching(stream, other, false);
+ }
+
+ template <class OtherMR>
+ void retain_matching(::cuda::stream_ref stream,
+ const sketch_impl<Key, OtherMR>& other,
+ bool keep_matches)
+ {
+ if (::compute_seed_hash(seed_) != ::compute_seed_hash(other.seed_)) {
+ throw std::invalid_argument(keep_matches ? "theta_sketch::intersect:
seed hash mismatch"
+ : "theta_sketch::a_not_b: seed
hash mismatch");
+ }
+ const auto theta = std::min(effective_theta(), other.effective_theta());
+ auto result =
+ select(stream,
+ hashes_.data(),
+ hashes_.size(),
+ membership_filter{other.hashes_.data(), other.hashes_.size(),
theta, keep_matches});
+ const bool empty = result.size == 0 && theta == max_theta;
+ install(stream, std::move(result), theta, empty);
+ }
+
+ void reset(::cuda::stream_ref stream)
+ {
+ install(stream, {make_hash_buffer(stream, 0), 0}, max_theta, true);
+ }
+
+ [[nodiscard]] std::vector<hash_result_type>
get_retained_hashes(::cuda::stream_ref stream) const
+ {
+ std::vector<hash_result_type> entries(hashes_.size());
+ if (!entries.empty()) {
+ DATASKETCHES_CUDA_TRY(cudaMemcpyAsync(entries.data(),
+ hashes_.data(),
+ entries.size() *
sizeof(hash_result_type),
+ cudaMemcpyDeviceToHost,
+ stream.get()));
+ }
+ stream.sync();
+ return entries;
+ }
+
+ [[nodiscard]] std::vector<std::uint8_t> serialize_compact(::cuda::stream_ref
stream) const
+ {
+ return serialize_compact_v3(
+ is_empty_, ::compute_seed_hash(seed_), effective_theta(),
get_retained_hashes(stream));
+ }
+
+ void load_compact(::cuda::stream_ref stream, const compact_image& image)
+ {
+ if (image.entries.size() > k()) {
+ throw std::invalid_argument(
+ "theta_sketch::deserialize: retained entries exceed configured nominal
k");
+ }
+ auto next = make_hash_buffer(stream, image.entries.size());
+ if (!image.entries.empty()) {
+ DATASKETCHES_CUDA_TRY(cudaMemcpyAsync(next.data(),
+ image.entries.data(),
+ image.entries.size() *
sizeof(hash_result_type),
+ cudaMemcpyHostToDevice,
+ stream.get()));
+ }
+ install(stream, {std::move(next), image.entries.size()}, image.theta,
image.empty);
+ }
+};
+
+} // namespace datasketches::cuda::detail::theta
diff --git a/include/datasketches/cuda/detail/theta/theta.inl
b/include/datasketches/cuda/detail/theta/theta.inl
new file mode 100644
index 0000000..a842a81
--- /dev/null
+++ b/include/datasketches/cuda/detail/theta/theta.inl
@@ -0,0 +1,168 @@
+/*
+ * 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
+
+namespace datasketches::cuda {
+
+template <class Key, class MR>
+theta_sketch<Key, MR>::theta_sketch(
+ ::cuda::stream_ref stream, MR mr, std::uint8_t lg_k, std::uint64_t seed,
float p)
+ : impl_(stream, std::move(mr), lg_k, seed, p)
+{
+}
+
+template <class Key, class MR>
+template <class RandomAccessIt>
+void theta_sketch<Key, MR>::update(::cuda::stream_ref stream,
+ RandomAccessIt first,
+ RandomAccessIt last)
+{
+ impl_.update(stream, first, last);
+}
+
+template <class Key, class MR>
+template <class OtherMR>
+void theta_sketch<Key, MR>::merge(::cuda::stream_ref stream,
+ const theta_sketch<Key, OtherMR>& other)
+{
+ impl_.merge(stream, other.impl_);
+}
+
+template <class Key, class MR>
+template <class OtherMR>
+void theta_sketch<Key, MR>::intersect(::cuda::stream_ref stream,
+ const theta_sketch<Key, OtherMR>& other)
+{
+ impl_.intersect(stream, other.impl_);
+}
+
+template <class Key, class MR>
+template <class OtherMR>
+void theta_sketch<Key, MR>::a_not_b(::cuda::stream_ref stream,
+ const theta_sketch<Key, OtherMR>& other)
+{
+ impl_.a_not_b(stream, other.impl_);
+}
+
+template <class Key, class MR>
+void theta_sketch<Key, MR>::reset(::cuda::stream_ref stream)
+{
+ impl_.reset(stream);
+}
+
+template <class Key, class MR>
+bool theta_sketch<Key, MR>::is_empty() const noexcept
+{
+ return impl_.is_empty_;
+}
+
+template <class Key, class MR>
+bool theta_sketch<Key, MR>::is_estimation_mode() const noexcept
+{
+ return !impl_.is_empty_ && impl_.theta_ < detail::theta::max_theta;
+}
+
+template <class Key, class MR>
+bool theta_sketch<Key, MR>::is_ordered() const noexcept
+{
+ return true;
+}
+
+template <class Key, class MR>
+std::uint8_t theta_sketch<Key, MR>::get_lg_k() const noexcept
+{
+ return impl_.lg_k_;
+}
+
+template <class Key, class MR>
+std::uint64_t theta_sketch<Key, MR>::get_theta64() const noexcept
+{
+ return impl_.effective_theta();
+}
+
+template <class Key, class MR>
+double theta_sketch<Key, MR>::get_theta() const noexcept
+{
+ return static_cast<double>(get_theta64()) /
static_cast<double>(detail::theta::max_theta);
+}
+
+template <class Key, class MR>
+std::uint16_t theta_sketch<Key, MR>::get_seed_hash() const noexcept
+{
+ return ::compute_seed_hash(impl_.seed_);
+}
+
+template <class Key, class MR>
+std::size_t theta_sketch<Key, MR>::get_num_retained() const noexcept
+{
+ return impl_.hashes_.size();
+}
+
+template <class Key, class MR>
+double theta_sketch<Key, MR>::get_estimate() const noexcept
+{
+ return static_cast<double>(get_num_retained()) / get_theta();
+}
+
+template <class Key, class MR>
+double theta_sketch<Key, MR>::get_lower_bound(std::uint8_t num_std_devs) const
+{
+ if (!is_estimation_mode()) return static_cast<double>(get_num_retained());
+ return ::datasketches::binomial_bounds::get_lower_bound(
+ get_num_retained(), get_theta(), num_std_devs);
+}
+
+template <class Key, class MR>
+double theta_sketch<Key, MR>::get_upper_bound(std::uint8_t num_std_devs) const
+{
+ if (!is_estimation_mode()) return static_cast<double>(get_num_retained());
+ return ::datasketches::binomial_bounds::get_upper_bound(
+ get_num_retained(), get_theta(), num_std_devs);
+}
+
+template <class Key, class MR>
+std::vector<typename theta_sketch<Key, MR>::hash_type> theta_sketch<Key,
MR>::get_retained_hashes(
+ ::cuda::stream_ref stream) const
+{
+ return impl_.get_retained_hashes(stream);
+}
+
+template <class Key, class MR>
+std::vector<std::uint8_t> theta_sketch<Key,
MR>::serialize_compact(::cuda::stream_ref stream) const
+{
+ return impl_.serialize_compact(stream);
+}
+
+template <class Key, class MR>
+theta_sketch<Key, MR> theta_sketch<Key, MR>::deserialize(
+ ::cuda::stream_ref stream,
+ ::cuda::std::span<const std::uint8_t> bytes,
+ MR mr,
+ std::uint8_t lg_k,
+ std::uint64_t seed)
+{
+ stream.sync();
+ auto image = detail::theta::parse_compact_v3(bytes, seed);
+ theta_sketch sketch(stream, std::move(mr), lg_k, seed);
+ sketch.impl_.load_compact(stream, image);
+ return sketch;
+}
+
+} // namespace datasketches::cuda
diff --git a/include/datasketches/cuda/detail/theta/tuning.cuh
b/include/datasketches/cuda/detail/theta/tuning.cuh
new file mode 100644
index 0000000..f152e95
--- /dev/null
+++ b/include/datasketches/cuda/detail/theta/tuning.cuh
@@ -0,0 +1,49 @@
+/*
+ * 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 <cuda/std/cstddef>
+#include <cuda/std/cstdint>
+
+namespace datasketches::cuda::detail::theta {
+
+struct tuning {
+ static constexpr ::cuda::std::int32_t block_threads = 256;
+ static constexpr ::cuda::std::int32_t items_per_thread = 8;
+ static constexpr ::cuda::std::size_t filter_slots = 1024;
+ static constexpr ::cuda::std::size_t tile_keys = block_threads *
items_per_thread;
+
+ // Amortize launches while bounding sort and screen scratch independently of
input size.
+ static constexpr ::cuda::std::size_t chunk_target_multiple = 16;
+ static constexpr ::cuda::std::size_t min_chunk_keys =
::cuda::std::size_t{1} << 20;
+ static constexpr ::cuda::std::size_t min_survivor_budget =
::cuda::std::size_t{1} << 22;
+ static constexpr ::cuda::std::size_t max_chunk_growth = 64;
+ static constexpr double budget_margin = 1.1;
+ static constexpr double duplicate_hedge = 2.0;
+
+ static ::cuda::std::uint32_t grid_size(::cuda::std::size_t count) noexcept
+ {
+ constexpr auto max_blocks = ::cuda::std::size_t{0x7fffffff};
+ const auto blocks = (count + tile_keys - 1) / tile_keys;
+ return static_cast<::cuda::std::uint32_t>(blocks < max_blocks ? blocks :
max_blocks);
+ }
+};
+
+} // namespace datasketches::cuda::detail::theta
diff --git a/include/datasketches/cuda/theta.hpp
b/include/datasketches/cuda/theta.hpp
new file mode 100644
index 0000000..f305370
--- /dev/null
+++ b/include/datasketches/cuda/theta.hpp
@@ -0,0 +1,277 @@
+/*
+ * 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 <cuda/memory_pool>
+#include <cuda/std/span>
+#include <cuda/stream>
+#include <utility>
+#include <vector>
+
+#include <datasketches/cuda/detail/theta/policy.cuh>
+#include <datasketches/cuda/detail/theta/sketch_impl.hpp>
+
+#include <binomial_bounds.hpp>
+
+namespace datasketches::cuda {
+
+//! @brief GPU Theta sketch with ordered compact serialization compatible with
+//! the datasketches::compact_theta_sketch serialization version 3.
+//!
+//! Updates are batch-oriented: a kernel hashes, screens against theta, and
+//! compacts the survivors, which CUB sort, unique, and merge fold into the
+//! ordered set of the k = 2^lg_k smallest retained hashes. Large updates are
+//! split so theta tightens partway through and scratch memory stays bounded.
+//! Set operations work directly on the ordered device-resident hashes.
+//!
+//! Supports primitive device keys, ordered uncompressed compact-v3
+//! serialization, custom seeds, and p-sampling.
+//!
+//! Every device operation takes an explicit `cuda::stream_ref` and completes
+//! before returning, because the retained count sizes the next allocation.
+//! Accessors therefore take no stream. No-op updates and set operations may
+//! return without synchronizing unrelated work queued on the stream, and
+//! set operations can take empty-sketch shortcuts without checking seeds.
+//!
+//! **Stream lifetime.** The caller MUST keep the stream supplied at
construction
+//! or deserialization alive until the sketch is destroyed. Retained buffers
are
+//! rebound to that stream for stream-ordered deallocation.
+//!
+//! @tparam Key Primitive input key type.
+//! @tparam MR Device-accessible memory resource type. Defaults to
+//! `::cuda::device_memory_pool_ref`.
+template <class Key, class MR = ::cuda::device_memory_pool_ref>
+class theta_sketch {
+ public:
+ using key_type = Key;
+ using hash_type = std::uint64_t;
+
+ static constexpr std::uint8_t default_lg_k = detail::theta::default_lg_k;
+ static constexpr std::uint64_t default_seed = detail::default_seed;
+
+ //! @brief Construct an empty sketch on a caller-provided stream.
+ //!
+ //! @param[in] stream CUDA stream used for stream-ordered initialization.
+ //! @param[in] mr Memory resource for device allocations.
+ //! @param[in] lg_k Base 2 logarithm of the nominal number of retained
entries.
+ //! @param[in] seed Hash seed; sketches built with different seeds cannot be
+ //! combined.
+ //! @param[in] p Sampling probability, in [2^-63, 1]. Smaller probabilities
+ //! round the hash threshold to zero and are rejected.
+ //! @throws std::invalid_argument if `lg_k` is outside [5, 26] or `p` is
+ //! outside [2^-63, 1].
+ theta_sketch(::cuda::stream_ref stream,
+ MR mr,
+ std::uint8_t lg_k = default_lg_k,
+ std::uint64_t seed = default_seed,
+ float p = 1.0F);
+
+ theta_sketch(const theta_sketch&) = delete;
+ theta_sketch& operator=(const theta_sketch&) = delete;
+ theta_sketch(theta_sketch&&) = default;
+ theta_sketch& operator=(theta_sketch&&) = default;
+ ~theta_sketch() = default;
+
+ //! @brief Bulk update on a caller-provided stream.
+ //!
+ //! Hashes each key, screens it against theta, and folds the surviving hashes
+ //! into the retained set. Large batches are split internally so theta
tightens
+ //! during the call. Completes the update before returning; an empty range is
+ //! a no-op.
+ //! Chunks commit independently. If a later allocation fails, earlier chunks
+ //! remain installed; a failure before the first successful chunk leaves the
+ //! sketch unchanged.
+ //!
+ //! @tparam RandomAccessIt Random-access iterator type over device-accessible
+ //! keys.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ //! @param[in] first Iterator to the first element to update.
+ //! @param[in] last Iterator past the last element to update.
+ //! @throws std::invalid_argument if `last` precedes `first`.
+ template <class RandomAccessIt>
+ void update(::cuda::stream_ref stream, RandomAccessIt first, RandomAccessIt
last);
+
+ //! @brief Replace this sketch with the union of this and `other`.
+ //!
+ //! Completes any device work it submits before returning.
+ //!
+ //! @tparam OtherMR Memory resource type of `other`.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ //! @param[in] other The other sketch to union into `*this`.
+ //! @throws std::invalid_argument if the seed hashes disagree and no
+ //! empty-sketch shortcut applies.
+ template <class OtherMR>
+ void merge(::cuda::stream_ref stream, const theta_sketch<Key, OtherMR>&
other);
+
+ //! @brief Replace this sketch with the intersection of this and `other`.
+ //!
+ //! Completes any device work it submits before returning.
+ //!
+ //! @tparam OtherMR Memory resource type of `other`.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ //! @param[in] other The other sketch to intersect with `*this`.
+ //! @throws std::invalid_argument if the seed hashes disagree and no
+ //! empty-sketch shortcut applies.
+ template <class OtherMR>
+ void intersect(::cuda::stream_ref stream, const theta_sketch<Key, OtherMR>&
other);
+
+ //! @brief Replace this sketch with the set difference this-minus-`other`.
+ //!
+ //! Completes any device work it submits before returning.
+ //!
+ //! @tparam OtherMR Memory resource type of `other`.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ //! @param[in] other The sketch to subtract from `*this`.
+ //! @throws std::invalid_argument if the seed hashes disagree and no
+ //! empty-sketch shortcut applies.
+ template <class OtherMR>
+ void a_not_b(::cuda::stream_ref stream, const theta_sketch<Key, OtherMR>&
other);
+
+ //! @brief Restore the initial empty state.
+ //!
+ //! Synchronizes `stream` before returning.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ void reset(::cuda::stream_ref stream);
+
+ //! @brief True iff the represented set is known to be empty.
+ //!
+ //! A sampled sketch with no retained hashes may still represent a nonempty
+ //! set. Exact set operations can establish emptiness after keys were seen.
+ //! @return True iff the represented set is known to be empty.
+ [[nodiscard]] bool is_empty() const noexcept;
+
+ //! @brief True iff theta has fallen below its maximum, so the retained count
+ //! no longer equals the exact distinct count.
+ //!
+ //! @return True iff the sketch is in estimation mode.
+ [[nodiscard]] bool is_estimation_mode() const noexcept;
+
+ //! @brief True iff the retained hashes are in ascending order.
+ //!
+ //! Always true: retained hashes are kept ordered so the compact image can be
+ //! written without a sort.
+ //!
+ //! @return True.
+ [[nodiscard]] bool is_ordered() const noexcept;
+
+ //! @brief Base 2 logarithm of the nominal number of entries.
+ //!
+ //! @return The `lg_k` the sketch was constructed with.
+ [[nodiscard]] std::uint8_t get_lg_k() const noexcept;
+
+ //! @brief Current theta as a raw 64-bit hash threshold.
+ //!
+ //! @return Theta in `[1, 2^63)`.
+ [[nodiscard]] std::uint64_t get_theta64() const noexcept;
+
+ //! @brief Current theta as a fraction of the hash space.
+ //!
+ //! @return Theta in `(0, 1]`.
+ [[nodiscard]] double get_theta() const noexcept;
+
+ //! @brief Hash of the seed, used to reject incompatible set operations.
+ //!
+ //! @return The 16-bit seed hash.
+ [[nodiscard]] std::uint16_t get_seed_hash() const noexcept;
+
+ //! @brief Number of hashes the sketch currently retains, at most `2^lg_k`.
+ //!
+ //! @return The retained entry count.
+ [[nodiscard]] std::size_t get_num_retained() const noexcept;
+
+ //! @brief Cardinality estimate.
+ //!
+ //! @return The cardinality estimate.
+ [[nodiscard]] double get_estimate() const noexcept;
+
+ //! @brief Lower bound on the estimate.
+ //!
+ //! @param[in] num_std_devs Confidence level: 1, 2, or 3.
+ //! @return The lower bound, or the exact count when not in estimation mode.
+ //! @throws std::invalid_argument if in estimation mode and `num_std_devs`
+ //! is outside [1, 3].
+ [[nodiscard]] double get_lower_bound(std::uint8_t num_std_devs) const;
+
+ //! @brief Upper bound on the estimate.
+ //!
+ //! @param[in] num_std_devs Confidence level: 1, 2, or 3.
+ //! @return The upper bound, or the exact count when not in estimation mode.
+ //! @throws std::invalid_argument if in estimation mode and `num_std_devs`
+ //! is outside [1, 3].
+ [[nodiscard]] double get_upper_bound(std::uint8_t num_std_devs) const;
+
+ //! @brief Copy the ordered retained hashes to host memory.
+ //!
+ //! Synchronizes `stream` before returning.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ //! @return The retained hashes, in ascending order.
+ [[nodiscard]] std::vector<hash_type> get_retained_hashes(::cuda::stream_ref
stream) const;
+
+ //! @brief Serialize as an ordered, uncompressed compact Theta v3 image.
+ //!
+ //! Synchronizes `stream` before returning.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ //! @return The serialized sketch.
+ [[nodiscard]] std::vector<std::uint8_t> serialize_compact(::cuda::stream_ref
stream) const;
+
+ //! @brief Deserialize an ordered, uncompressed compact Theta v3 image.
+ //!
+ //! Compact Theta images do not encode nominal k, so `lg_k` must be supplied
+ //! by the caller when it differs from the default.
+ //! CPU compact sketches can retain more than their nominal k before an
+ //! explicit trim. Trim those sketches before serialization, or choose an
+ //! `lg_k` large enough to hold every retained entry in the image.
+ //! The byte span must contain exactly one image, without trailing bytes.
+ //! Synchronizes `stream` before reading the host-accessible byte span.
+ //!
+ //! @param[in] stream CUDA stream this operation is executed in.
+ //! @param[in] bytes Host-accessible wire-format compact Theta v3 image.
+ //! @param[in] mr Memory resource for device allocations.
+ //! @param[in] lg_k Base 2 logarithm of the nominal number of retained
entries.
+ //! @param[in] seed Hash seed the image was produced with.
+ //! @throws std::invalid_argument if `bytes` is malformed, is not an ordered
+ //! uncompressed v3 image, disagrees with `seed`, or holds more entries
than
+ //! `lg_k` allows.
+ //! @return The deserialized sketch.
+ static theta_sketch deserialize(::cuda::stream_ref stream,
+ ::cuda::std::span<const std::uint8_t> bytes,
+ MR mr,
+ std::uint8_t lg_k = default_lg_k,
+ std::uint64_t seed = default_seed);
+
+ private:
+ template <class, class>
+ friend class theta_sketch; // Allow set operations across memory-resource
specializations.
+
+ detail::theta::sketch_impl<Key, MR> impl_; // Implementation details.
+};
+
+} // namespace datasketches::cuda
+
+#include <datasketches/cuda/detail/theta/theta.inl>
diff --git a/test/hll/CMakeLists.txt b/test/hll/CMakeLists.txt
index b255b3e..a9e2f28 100644
--- a/test/hll/CMakeLists.txt
+++ b/test/hll/CMakeLists.txt
@@ -15,17 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-CPMAddPackage(
- NAME Catch2
- GITHUB_REPOSITORY catchorg/Catch2
- VERSION 3.5.3
-)
-
-if(Catch2_ADDED)
- list(APPEND CMAKE_MODULE_PATH ${Catch2_SOURCE_DIR}/extras)
- include(Catch)
-endif()
-
# Test targets get datasketches::cuda (our INTERFACE lib, brings CCCL + CUDA),
# datasketches (apache/datasketches-cpp INTERFACE lib for parity & helpers),
# and Catch2WithMain. CUDA::cudart is added explicitly for runtime calls in
diff --git a/test/hll/composite_finalizer_device_test.cu
b/test/hll/composite_finalizer_device_test.cu
index 4adfbd2..df41d20 100644
--- a/test/hll/composite_finalizer_device_test.cu
+++ b/test/hll/composite_finalizer_device_test.cu
@@ -20,6 +20,8 @@
#include <algorithm>
#include <cstddef>
#include <cstdint>
+#include <cuda/std/cstddef>
+#include <cuda/std/cstdint>
#include <vector>
#include <cuda_runtime.h>
@@ -37,13 +39,15 @@ namespace {
struct finalizer_case {
double z;
- std::uint32_t zeroes;
- std::uint8_t lg_k;
+ ::cuda::std::uint32_t zeroes;
+ ::cuda::std::uint8_t lg_k;
};
-__global__ void finalize_kernel(const finalizer_case* cases, double* results,
std::size_t size)
+__global__ void finalize_kernel(const finalizer_case* cases,
+ double* results,
+ ::cuda::std::size_t size)
{
- const std::size_t index = blockIdx.x * blockDim.x + threadIdx.x;
+ const ::cuda::std::size_t index = blockIdx.x * blockDim.x + threadIdx.x;
if (index < size) {
const auto c = cases[index];
results[index] = datasketches::cuda::detail::hll::composite_estimate(c.z,
c.zeroes, c.lg_k);
diff --git a/test/hll/normalizing_hasher_test.cpp
b/test/hll/normalizing_hasher_test.cpp
index 409cc57..6a31b22 100644
--- a/test/hll/normalizing_hasher_test.cpp
+++ b/test/hll/normalizing_hasher_test.cpp
@@ -31,9 +31,9 @@
#include <catch2/catch_test_macros.hpp>
-#include <datasketches/cuda/detail/hll/normalizing_hasher.cuh>
+#include <datasketches/cuda/detail/common/normalizing_hasher.cuh>
-using datasketches::cuda::detail::hll::normalizing_hasher;
+using datasketches::cuda::detail::normalizing_hasher;
// Helper: invoke normalizing_hasher with seed 9001 (same as all tests).
template <class T>
diff --git a/test/hll/policy_compile_test.cu b/test/hll/policy_compile_test.cu
index ea240aa..7475f41 100644
--- a/test/hll/policy_compile_test.cu
+++ b/test/hll/policy_compile_test.cu
@@ -57,7 +57,7 @@ static_assert(
static_assert(policy_t::default_seed == datasketches::DEFAULT_SEED,
"policy::default_seed must stay pinned to datasketches-cpp
DEFAULT_SEED "
"(common_defs.hpp:34)");
-static_assert(datasketches::cuda::detail::hll::default_seed ==
datasketches::DEFAULT_SEED,
+static_assert(datasketches::cuda::detail::default_seed ==
datasketches::DEFAULT_SEED,
"project-wide default_seed must stay pinned to datasketches-cpp
DEFAULT_SEED");
TEST_CASE("policy satisfies cudax concept", "[policy][compile]")
diff --git a/test/theta/CMakeLists.txt b/test/theta/CMakeLists.txt
new file mode 100644
index 0000000..035f6bc
--- /dev/null
+++ b/test/theta/CMakeLists.txt
@@ -0,0 +1,32 @@
+# 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.
+
+function(ConfigureThetaTest TEST_NAME)
+ add_executable(${TEST_NAME} ${ARGN})
+ target_link_libraries(${TEST_NAME} PRIVATE
+ Catch2::Catch2WithMain
+ datasketches::cuda
+ datasketches
+ CUDA::cudart)
+ set_target_properties(${TEST_NAME} PROPERTIES
+ RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/tests")
+ catch_discover_tests(${TEST_NAME} EXTRA_ARGS --allow-running-no-tests)
+endfunction()
+
+ConfigureThetaTest(THETA_BASIC_TEST basic_sketch_test.cu cpu_reference.cpp)
+ConfigureThetaTest(THETA_SET_OPERATIONS_TEST set_operations_test.cu
cpu_reference.cpp)
+ConfigureThetaTest(THETA_TYPED_PARITY_TEST parity_typed_test.cu
cpu_reference.cpp)
diff --git a/test/theta/basic_sketch_test.cu b/test/theta/basic_sketch_test.cu
new file mode 100644
index 0000000..70b1138
--- /dev/null
+++ b/test/theta/basic_sketch_test.cu
@@ -0,0 +1,426 @@
+/*
+ * 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 <cmath>
+#include <cstdint>
+#include <cuda/devices>
+#include <cuda/memory_pool>
+#include <cuda/std/span>
+#include <cuda/stream>
+#include <limits>
+#include <new>
+#include <numeric>
+#include <vector>
+
+#include <thrust/copy.h>
+#include <thrust/device_vector.h>
+#include <thrust/execution_policy.h>
+#include <thrust/random.h>
+#include <thrust/sequence.h>
+#include <thrust/shuffle.h>
+
+#include <catch2/catch_test_macros.hpp>
+
+#include <datasketches/cuda/theta.hpp>
+
+#include "cpu_reference.hpp"
+
+namespace {
+
+struct allocation_failure {
+ std::size_t allocations = 0;
+ std::size_t fail_at = std::numeric_limits<std::size_t>::max();
+
+ void check()
+ {
+ if (allocations++ == fail_at) throw std::bad_alloc{};
+ }
+};
+
+struct failing_resource {
+ ::cuda::device_memory_pool_ref upstream;
+ allocation_failure* failure;
+
+ void* allocate_sync(std::size_t bytes, std::size_t alignment)
+ {
+ failure->check();
+ return upstream.allocate_sync(bytes, alignment);
+ }
+
+ void* allocate(::cuda::stream_ref stream, std::size_t bytes, std::size_t
alignment)
+ {
+ failure->check();
+ return upstream.allocate(stream, bytes, alignment);
+ }
+
+ void deallocate_sync(void* ptr, std::size_t bytes, std::size_t alignment)
noexcept
+ {
+ upstream.deallocate_sync(ptr, bytes, alignment);
+ }
+
+ void deallocate(::cuda::stream_ref stream,
+ void* ptr,
+ std::size_t bytes,
+ std::size_t alignment) noexcept
+ {
+ upstream.deallocate(stream, ptr, bytes, alignment);
+ }
+
+ bool operator==(const failing_resource& other) const noexcept
+ {
+ return upstream == other.upstream && failure == other.failure;
+ }
+
+ bool operator!=(const failing_resource& other) const noexcept { return
!(*this == other); }
+
+ friend void get_property(const failing_resource&,
::cuda::mr::device_accessible) noexcept {}
+};
+
+thrust::device_vector<std::uint64_t> make_keys(std::size_t count,
std::uint64_t first)
+{
+ thrust::device_vector<std::uint64_t> keys(count);
+ thrust::sequence(thrust::device, keys.begin(), keys.end(), first);
+ return keys;
+}
+
+using theta_test::cpu_image;
+
+std::vector<std::uint8_t> cpu_image(const
thrust::device_vector<std::uint64_t>& keys,
+ std::uint8_t lg_k,
+ std::uint64_t seed,
+ float p = 1.0F)
+{
+ std::vector<std::uint64_t> host(keys.size());
+ thrust::copy(keys.begin(), keys.end(), host.begin());
+ return theta_test::cpu_image(host, lg_k, seed, p);
+}
+
+} // namespace
+
+TEST_CASE("Theta starts empty and exact", "[theta][basic]")
+{
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ datasketches::cuda::theta_sketch<std::uint64_t> sketch(stream, mr, 12);
+
+ REQUIRE(sketch.is_empty());
+ REQUIRE(sketch.is_ordered());
+ REQUIRE_FALSE(sketch.is_estimation_mode());
+ REQUIRE(sketch.get_lg_k() == 12);
+ REQUIRE(sketch.get_num_retained() == 0);
+ REQUIRE(sketch.get_estimate() == 0.0);
+ REQUIRE(sketch.get_theta() == 1.0);
+
+ REQUIRE(sketch.serialize_compact(stream) ==
cpu_image(std::vector<std::uint64_t>{}, 12, 9001));
+}
+
+TEST_CASE("Theta failed first chunk preserves the empty sketch",
"[theta][exception][resource]")
+{
+ constexpr std::uint8_t lg_k = 8;
+ constexpr std::uint64_t seed = 9001;
+ constexpr float p = 0.125F;
+ const auto keys = make_keys(1000, 0xa110cULL);
+
+ ::cuda::stream allocation_stream{::cuda::devices[0]};
+ ::cuda::stream operation_stream{::cuda::devices[0]};
+ allocation_failure failure;
+ failing_resource mr{::cuda::device_default_memory_pool(::cuda::devices[0]),
&failure};
+ datasketches::cuda::theta_sketch<std::uint64_t, failing_resource> sketch(
+ allocation_stream, mr, lg_k, seed, p);
+
+ SECTION("first allocation") { failure.fail_at = 0; }
+ SECTION("final allocation after screening and sorting")
+ {
+ // Learn the successful path's allocation count instead of depending on how
+ // many temporary allocations this CUB version makes.
+ failure.allocations = 0;
+ sketch.update(operation_stream, keys.begin(), keys.end());
+ REQUIRE(failure.allocations > 1);
+ const auto final_allocation = failure.allocations - 1;
+ sketch.reset(operation_stream);
+ failure.fail_at = final_allocation;
+ }
+ const auto before = sketch.serialize_compact(operation_stream);
+
+ failure.allocations = 0;
+ // CUB translates allocator exceptions into CUDA errors.
+ REQUIRE_THROWS(sketch.update(operation_stream, keys.begin(), keys.end()));
+ REQUIRE(failure.allocations == failure.fail_at + 1);
+ failure.fail_at = std::numeric_limits<std::size_t>::max();
+ REQUIRE(sketch.is_empty());
+ REQUIRE(sketch.serialize_compact(operation_stream) == before);
+
+ sketch.update(operation_stream, keys.begin(), keys.end());
+ REQUIRE(sketch.serialize_compact(operation_stream) == cpu_image(keys, lg_k,
seed, p));
+}
+
+TEST_CASE("Theta compact v3 bytes match trimmed CPU in estimation mode",
+ "[theta][parity][serialization]")
+{
+ constexpr std::uint8_t lg_k = 10;
+ constexpr std::uint64_t seed = 123456789;
+ auto keys = make_keys(100000, 0xabcdefULL);
+
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ datasketches::cuda::theta_sketch<std::uint64_t> gpu(stream, mr, lg_k, seed);
+ gpu.update(stream, keys.begin(), keys.end());
+
+ REQUIRE(gpu.is_estimation_mode());
+ REQUIRE(gpu.get_num_retained() == (std::size_t{1} << lg_k));
+ REQUIRE(gpu.serialize_compact(stream) == cpu_image(keys, lg_k, seed));
+
+ const double estimate = gpu.get_estimate();
+ const double lower = gpu.get_lower_bound(2);
+ const double upper = gpu.get_upper_bound(2);
+ REQUIRE(lower <= estimate);
+ REQUIRE(estimate <= upper);
+}
+
+TEST_CASE("Theta p-sampling and custom seed match CPU",
"[theta][parity][sampling]")
+{
+ constexpr std::uint8_t lg_k = 12;
+ constexpr std::uint64_t seed = 0xdecafbadULL;
+ constexpr float p = 0.125F;
+ auto keys = make_keys(10000, 0xf00dULL);
+
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ datasketches::cuda::theta_sketch<std::uint64_t> gpu(stream, mr, lg_k, seed,
p);
+ gpu.update(stream, keys.begin(), keys.end());
+
+ const auto bytes = gpu.serialize_compact(stream);
+ REQUIRE(bytes == cpu_image(keys, lg_k, seed, p));
+
+ gpu.reset(stream);
+ REQUIRE(gpu.is_empty());
+ REQUIRE(gpu.get_num_retained() == 0);
+ gpu.update(stream, keys.begin(), keys.end());
+ REQUIRE(gpu.serialize_compact(stream) == bytes);
+}
+
+TEST_CASE("Theta imports and updates CPU compact v3 images",
"[theta][serialization][parity]")
+{
+ constexpr std::uint8_t lg_k = 8;
+ constexpr std::uint64_t seed = 0xc0ffeeULL;
+ using sketch_type =
datasketches::cuda::theta_sketch<std::uint64_t>;
+ std::vector<std::uint64_t> keys;
+ float p = 1.0F;
+ SECTION("empty") {}
+ SECTION("single entry") { keys = {42}; }
+ SECTION("exact") { keys.resize(100); }
+ SECTION("estimation") { keys.resize(10000); }
+ SECTION("sampled with zero retained entries")
+ {
+ keys = {42};
+ p = 0x1p-63F;
+ }
+
+ if (keys.size() > 1) std::iota(keys.begin(), keys.end(), std::uint64_t{1});
+ const auto bytes = cpu_image(keys, lg_k, seed, p);
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ auto sketch = sketch_type::deserialize(
+ stream, ::cuda::std::span<const std::uint8_t>{bytes.data(), bytes.size()},
mr, lg_k, seed);
+ REQUIRE(sketch.serialize_compact(stream) == bytes);
+ REQUIRE(sketch.is_empty() == keys.empty());
+
+ const std::vector<std::uint64_t> additional{42, 43, 44};
+ thrust::device_vector<std::uint64_t> device = additional;
+ sketch.update(stream, device.begin(), device.end());
+ keys.insert(keys.end(), additional.begin(), additional.end());
+ REQUIRE(sketch.serialize_compact(stream) == cpu_image(keys, lg_k, seed, p));
+}
+
+TEST_CASE("Theta validates constructor and compact image",
"[theta][validation]")
+{
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+
+ REQUIRE_THROWS_AS((datasketches::cuda::theta_sketch<std::uint64_t>(stream,
mr, 4)),
+ std::invalid_argument);
+ REQUIRE_THROWS_AS((datasketches::cuda::theta_sketch<std::uint64_t>(stream,
mr, 12, 9001, 0.0F)),
+ std::invalid_argument);
+ // Below 2^-63 the hash threshold rounds to zero.
+ REQUIRE_THROWS_AS((datasketches::cuda::theta_sketch<std::uint64_t>(
+ stream, mr, 12, 9001, std::nextafter(0x1p-63F, 0.0F))),
+ std::invalid_argument);
+
+ std::vector<std::uint8_t> invalid(8, 0);
+ invalid[0] = 1;
+ invalid[1] = 4;
+ invalid[2] = 3;
+ REQUIRE_THROWS_AS(
+ datasketches::cuda::theta_sketch<std::uint64_t>::deserialize(
+ stream, ::cuda::std::span<const std::uint8_t>{invalid.data(),
invalid.size()}, mr),
+ std::invalid_argument);
+
+ std::vector<std::uint8_t> truncated(8, 0);
+ truncated[0] = 3;
+ truncated[1] = 3;
+ truncated[2] = 3;
+ truncated[5] = (1U << 1) | (1U << 3) | (1U << 4);
+ datasketches::cuda::theta_sketch<std::uint64_t> seed_source(stream, mr);
+ const auto seed_hash = seed_source.get_seed_hash();
+ truncated[6] = static_cast<std::uint8_t>(seed_hash);
+ truncated[7] = static_cast<std::uint8_t>(seed_hash >> 8);
+ REQUIRE_THROWS_AS(
+ datasketches::cuda::theta_sketch<std::uint64_t>::deserialize(
+ stream, ::cuda::std::span<const std::uint8_t>{truncated.data(),
truncated.size()}, mr),
+ std::invalid_argument);
+
+ auto trailing = cpu_image(std::vector<std::uint64_t>{42}, 12, 9001);
+ trailing.push_back(0);
+ REQUIRE_THROWS_AS(
+ datasketches::cuda::theta_sketch<std::uint64_t>::deserialize(
+ stream, ::cuda::std::span<const std::uint8_t>{trailing.data(),
trailing.size()}, mr),
+ std::invalid_argument);
+}
+
+TEST_CASE("Theta batch splitting does not change the result",
"[theta][parity][chunking]")
+{
+ // The same keys fed as one call and as several must produce identical
images,
+ // whether the split is the caller's or update()'s own.
+ constexpr std::uint8_t lg_k = 11;
+ constexpr std::uint64_t seed = 9001;
+ auto keys = make_keys(3'000'000, 0xfeedfaceULL);
+
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+
+ datasketches::cuda::theta_sketch<std::uint64_t> single(stream, mr, lg_k,
seed);
+ single.update(stream, keys.begin(), keys.end());
+
+ datasketches::cuda::theta_sketch<std::uint64_t> split(stream, mr, lg_k,
seed);
+ const std::size_t batch = 700'000;
+ for (std::size_t offset = 0; offset < keys.size(); offset += batch) {
+ const auto end = std::min(offset + batch, keys.size());
+ split.update(stream, keys.begin() + offset, keys.begin() + end);
+ }
+
+ REQUIRE(single.serialize_compact(stream) == split.serialize_compact(stream));
+ REQUIRE(single.serialize_compact(stream) == cpu_image(keys, lg_k, seed));
+}
+
+TEST_CASE("Theta grouped duplicates followed by distinct keys match CPU",
+ "[theta][parity][chunking][budget]")
+{
+ // An abrupt transition from grouped duplicates to distinct keys must remain
+ // correct regardless of how chunks are sized or retried.
+ using sketch_type =
datasketches::cuda::theta_sketch<std::uint64_t>;
+ constexpr std::uint8_t lg_k = 12;
+ constexpr std::uint64_t seed = 9001;
+ constexpr std::size_t run = 2048;
+ constexpr std::size_t grouped_distinct = 4096;
+ constexpr std::size_t grouped_keys = run * grouped_distinct; // 8M keys
+ constexpr std::size_t distinct_keys = std::size_t{1} << 24; // 16M keys
+ std::vector<std::uint64_t> keys(grouped_keys + distinct_keys);
+ for (std::size_t i = 0; i < grouped_keys; ++i)
+ keys[i] = 0x1000'0000'0000'0000ULL + i / run;
+ for (std::size_t i = 0; i < distinct_keys; ++i)
+ keys[grouped_keys + i] = 0x2000'0000'0000'0000ULL + i;
+
+ thrust::device_vector<std::uint64_t> device_keys = keys;
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ sketch_type gpu(stream, mr, lg_k, seed, 1.0F);
+ gpu.update(stream, device_keys.begin(), device_keys.end());
+
+ REQUIRE(gpu.serialize_compact(stream) == cpu_image(keys, lg_k, seed));
+}
+
+TEST_CASE("Theta screen overflow retries preserve the result",
"[theta][parity][chunking][budget]")
+{
+ // Force bounded screen writes, then compare retrying updates with the CPU,
+ // including when the caller also splits the input.
+ using impl_type =
datasketches::cuda::detail::theta::sketch_impl<std::uint64_t>;
+ constexpr std::uint8_t lg_k = 8;
+ constexpr std::uint64_t seed = 9001;
+ auto keys = make_keys(20'000, 0x0ff10adULL);
+ keys.resize(25'000);
+ thrust::copy_n(thrust::device, keys.begin(), 5'000, keys.begin() + 20'000);
// duplicates
+ thrust::shuffle(thrust::device, keys.begin(), keys.end(),
thrust::default_random_engine{0x5eed});
+
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+
+ impl_type tiny(stream, mr, lg_k, seed, 1.0F);
+ tiny.survivor_budget_ = 64;
+ {
+ const auto result = tiny.screen(stream, keys.begin(), keys.end() -
keys.begin());
+ REQUIRE(result.data.size() == 64);
+ REQUIRE(result.size > result.data.size());
+ }
+ tiny.update(stream, keys.begin(), keys.end());
+
+ const auto bytes = tiny.serialize_compact(stream);
+ REQUIRE(bytes == cpu_image(keys, lg_k, seed));
+
+ // The same holds when the keys arrive in several calls, so the retry path
+ // composes with the caller's own batching.
+ impl_type batched(stream, mr, lg_k, seed, 1.0F);
+ batched.survivor_budget_ = 64;
+ const std::size_t batch = 7'001;
+ for (std::size_t offset = 0; offset < keys.size(); offset += batch) {
+ const auto end = std::min(offset + batch, keys.size());
+ batched.update(stream, keys.begin() + offset, keys.begin() + end);
+ }
+ REQUIRE(batched.serialize_compact(stream) == bytes);
+}
+
+TEST_CASE("Theta batch far above the budget with few distinct keys matches
CPU",
+ "[theta][parity][chunking][budget]")
+{
+ // Fewer than k distinct keys keeps theta at its maximum even for a large
+ // input whose duplicates are scattered.
+ using sketch_type =
datasketches::cuda::theta_sketch<std::uint64_t>;
+ constexpr std::uint8_t lg_k = 12;
+ constexpr std::uint64_t seed = 9001;
+ constexpr std::size_t count = std::size_t{3} << 22;
+ constexpr std::size_t distinct = 3'000;
+ constexpr std::uint64_t first = 0xd15c0ULL;
+ std::vector<std::uint64_t> keys(count);
+ for (std::size_t i = 0; i < count; ++i)
+ keys[i] = first + (i * 7919) % distinct;
+
+ thrust::device_vector<std::uint64_t> device_keys = keys;
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ sketch_type gpu(stream, mr, lg_k, seed, 1.0F);
+ gpu.update(stream, device_keys.begin(), device_keys.end());
+
+ REQUIRE(gpu.get_estimate() == static_cast<double>(distinct));
+ REQUIRE(gpu.serialize_compact(stream) == cpu_image(keys, lg_k, seed));
+}
+
+TEST_CASE("Theta large lg_k matches CPU", "[theta][parity][chunking][budget]")
+{
+ using sketch_type =
datasketches::cuda::theta_sketch<std::uint64_t>;
+ constexpr std::uint8_t lg_k = 20;
+ constexpr std::uint64_t seed = 9001;
+ auto keys = make_keys(5'000'000, 0xb16b00b5ULL);
+
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ sketch_type gpu(stream, mr, lg_k, seed, 1.0F);
+ gpu.update(stream, keys.begin(), keys.end());
+
+ REQUIRE(gpu.serialize_compact(stream) == cpu_image(keys, lg_k, seed));
+}
diff --git a/test/theta/cpu_reference.cpp b/test/theta/cpu_reference.cpp
new file mode 100644
index 0000000..d471495
--- /dev/null
+++ b/test/theta/cpu_reference.cpp
@@ -0,0 +1,92 @@
+/*
+ * 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 "cpu_reference.hpp"
+
+#include <theta_a_not_b.hpp>
+#include <theta_intersection.hpp>
+#include <theta_sketch.hpp>
+#include <theta_union.hpp>
+
+namespace theta_test {
+
+template <class T>
+auto cpu_compact(const std::vector<T>& values,
+ std::uint8_t lg_k,
+ std::uint64_t seed,
+ float p = 1.0F)
+{
+ auto sketch =
+
::datasketches::update_theta_sketch::builder().set_lg_k(lg_k).set_seed(seed).set_p(p).build();
+ for (const auto value : values)
+ sketch.update(value);
+ sketch.trim();
+ return sketch.compact(true);
+}
+
+template <class T>
+std::vector<std::uint8_t> cpu_image(const std::vector<T>& values,
+ std::uint8_t lg_k,
+ std::uint64_t seed,
+ float p)
+{
+ return cpu_compact(values, lg_k, seed, p).serialize();
+}
+
+set_operation_images cpu_set_operation_images(const
std::vector<std::uint64_t>& a,
+ const
std::vector<std::uint64_t>& b,
+ std::uint8_t lg_k,
+ std::uint64_t seed)
+{
+ const auto a_compact = cpu_compact(a, lg_k, seed);
+ const auto b_compact = cpu_compact(b, lg_k, seed);
+
+ auto set_union =
::datasketches::theta_union::builder().set_lg_k(lg_k).set_seed(seed).build();
+ set_union.update(a_compact);
+ set_union.update(b_compact);
+
+ ::datasketches::theta_intersection intersection(seed);
+ intersection.update(a_compact);
+ intersection.update(b_compact);
+
+ ::datasketches::theta_a_not_b a_not_b(seed);
+ return set_operation_images{set_union.get_result(true).serialize(),
+ intersection.get_result(true).serialize(),
+ a_not_b.compute(a_compact, b_compact,
true).serialize()};
+}
+
+// Keep upstream Theta templates out of CUDA translation units: NVCC mishandles
+// their scalar pseudo-destructor calls.
+#define INSTANTIATE_CPU_IMAGE(T) \
+ template std::vector<std::uint8_t> cpu_image( \
+ const std::vector<T>&, std::uint8_t, std::uint64_t, float);
+
+INSTANTIATE_CPU_IMAGE(std::uint64_t)
+INSTANTIATE_CPU_IMAGE(std::int64_t)
+INSTANTIATE_CPU_IMAGE(std::uint32_t)
+INSTANTIATE_CPU_IMAGE(std::int32_t)
+INSTANTIATE_CPU_IMAGE(std::uint16_t)
+INSTANTIATE_CPU_IMAGE(std::int16_t)
+INSTANTIATE_CPU_IMAGE(std::uint8_t)
+INSTANTIATE_CPU_IMAGE(std::int8_t)
+INSTANTIATE_CPU_IMAGE(double)
+INSTANTIATE_CPU_IMAGE(float)
+#undef INSTANTIATE_CPU_IMAGE
+
+} // namespace theta_test
diff --git a/test/theta/cpu_reference.hpp b/test/theta/cpu_reference.hpp
new file mode 100644
index 0000000..8c91a65
--- /dev/null
+++ b/test/theta/cpu_reference.hpp
@@ -0,0 +1,44 @@
+/*
+ * 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 <cstdint>
+#include <vector>
+
+namespace theta_test {
+
+template <class T>
+std::vector<std::uint8_t> cpu_image(const std::vector<T>& values,
+ std::uint8_t lg_k,
+ std::uint64_t seed,
+ float p = 1.0F);
+
+struct set_operation_images {
+ std::vector<std::uint8_t> set_union;
+ std::vector<std::uint8_t> intersection;
+ std::vector<std::uint8_t> a_not_b;
+};
+
+set_operation_images cpu_set_operation_images(const
std::vector<std::uint64_t>& a,
+ const
std::vector<std::uint64_t>& b,
+ std::uint8_t lg_k,
+ std::uint64_t seed = 9001);
+
+} // namespace theta_test
diff --git a/test/theta/parity_typed_test.cu b/test/theta/parity_typed_test.cu
new file mode 100644
index 0000000..391a3ba
--- /dev/null
+++ b/test/theta/parity_typed_test.cu
@@ -0,0 +1,95 @@
+/*
+ * 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 <cstdint>
+#include <cuda/devices>
+#include <cuda/memory_pool>
+#include <cuda/stream>
+#include <limits>
+#include <random>
+#include <vector>
+
+#include <thrust/device_vector.h>
+
+#include <catch2/catch_test_macros.hpp>
+
+#include <datasketches/cuda/theta.hpp>
+
+#include "cpu_reference.hpp"
+
+namespace {
+
+template <class T>
+void compare_typed(const std::vector<T>& values)
+{
+ constexpr std::uint8_t lg_k = 12;
+ thrust::device_vector<T> device = values;
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ datasketches::cuda::theta_sketch<T> gpu(stream, mr, lg_k);
+ gpu.update(stream, device.begin(), device.end());
+ REQUIRE(gpu.serialize_compact(stream) == theta_test::cpu_image(values, lg_k,
9001));
+}
+
+template <class T>
+std::vector<T> random_integrals(std::size_t count, std::uint64_t seed)
+{
+ std::mt19937_64 rng(seed);
+ std::vector<T> values(count);
+ for (auto& value : values)
+ value = static_cast<T>(rng());
+ return values;
+}
+
+} // namespace
+
+TEST_CASE("Theta primitive integer hashing matches CPU",
"[theta][parity][typed]")
+{
+ compare_typed(random_integrals<std::uint64_t>(3000, 1));
+ compare_typed(random_integrals<std::int64_t>(3000, 2));
+ compare_typed(random_integrals<std::uint32_t>(3000, 3));
+ compare_typed(random_integrals<std::int32_t>(3000, 4));
+ compare_typed(random_integrals<std::uint16_t>(3000, 5));
+ compare_typed(random_integrals<std::int16_t>(3000, 6));
+ compare_typed(random_integrals<std::uint8_t>(3000, 7));
+ compare_typed(random_integrals<std::int8_t>(3000, 8));
+}
+
+TEST_CASE("Theta floating-point normalization matches CPU",
"[theta][parity][typed]")
+{
+ std::mt19937_64 rng(9);
+ std::uniform_real_distribution<double> doubles(-1e12, 1e12);
+ std::uniform_real_distribution<float> floats(-1e6F, 1e6F);
+
+ std::vector<double> double_values;
+ std::vector<float> float_values;
+ for (int i = 0; i < 3000; ++i) {
+ double_values.push_back(doubles(rng));
+ float_values.push_back(floats(rng));
+ }
+ double_values.push_back(0.0);
+ double_values.push_back(-0.0);
+ double_values.push_back(std::numeric_limits<double>::quiet_NaN());
+ float_values.push_back(0.0F);
+ float_values.push_back(-0.0F);
+ float_values.push_back(std::numeric_limits<float>::quiet_NaN());
+
+ compare_typed(double_values);
+ compare_typed(float_values);
+}
diff --git a/test/theta/set_operations_test.cu
b/test/theta/set_operations_test.cu
new file mode 100644
index 0000000..7f3c72c
--- /dev/null
+++ b/test/theta/set_operations_test.cu
@@ -0,0 +1,137 @@
+/*
+ * 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 <cstdint>
+#include <cuda/devices>
+#include <cuda/memory_pool>
+#include <cuda/std/span>
+#include <cuda/stream>
+#include <vector>
+
+#include <thrust/device_vector.h>
+
+#include <catch2/catch_test_macros.hpp>
+
+#include <datasketches/cuda/theta.hpp>
+
+#include "cpu_reference.hpp"
+
+namespace {
+
+using sketch_type = datasketches::cuda::theta_sketch<std::uint64_t>;
+
+std::vector<std::uint64_t> sequence(std::uint64_t first, std::uint64_t last)
+{
+ std::vector<std::uint64_t> values(last - first);
+ for (std::uint64_t i = first; i < last; ++i)
+ values[i - first] = i;
+ return values;
+}
+
+sketch_type clone(::cuda::stream_ref stream,
+ ::cuda::device_memory_pool_ref mr,
+ const sketch_type& source,
+ std::uint8_t lg_k)
+{
+ const auto bytes = source.serialize_compact(stream);
+ return sketch_type::deserialize(
+ stream, ::cuda::std::span<const std::uint8_t>{bytes.data(), bytes.size()},
mr, lg_k);
+}
+
+} // namespace
+
+TEST_CASE("Theta set operations reject seed mismatch", "[theta][setops][seed]")
+{
+ auto values = sequence(0, 100);
+ thrust::device_vector<std::uint64_t> device = values;
+
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ sketch_type a(stream, mr, 12, 1);
+ sketch_type b(stream, mr, 12, 2);
+ a.update(stream, device.begin(), device.end());
+ b.update(stream, device.begin(), device.end());
+
+ REQUIRE_THROWS_AS(a.merge(stream, b), std::invalid_argument);
+ REQUIRE_THROWS_AS(a.intersect(stream, b), std::invalid_argument);
+ REQUIRE_THROWS_AS(a.a_not_b(stream, b), std::invalid_argument);
+}
+
+TEST_CASE("Theta empty sampled union applies its configured p",
"[theta][setops][sampling][parity]")
+{
+ constexpr std::uint8_t lg_k = 12;
+ constexpr std::uint64_t seed = 0xdecafbadULL;
+ constexpr float p = 0.125F;
+ auto values = sequence(0, 10000);
+ thrust::device_vector<std::uint64_t> device = values;
+
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ sketch_type sampled_union(stream, mr, lg_k, seed, p);
+ sketch_type source(stream, mr, lg_k, seed);
+ source.update(stream, device.begin(), device.end());
+
+ sampled_union.merge(stream, source);
+ REQUIRE(sampled_union.serialize_compact(stream) ==
theta_test::cpu_image(values, lg_k, seed, p));
+}
+
+TEST_CASE("Theta set operations match CPU compact bytes",
"[theta][setops][parity][serialization]")
+{
+ std::uint8_t lg_k = 12;
+ std::vector<std::uint64_t> a_values;
+ std::vector<std::uint64_t> b_values;
+ SECTION("exact overlap")
+ {
+ a_values = sequence(0, 1000);
+ b_values = sequence(500, 1500);
+ }
+ SECTION("exact disjoint")
+ {
+ a_values = sequence(0, 100);
+ b_values = sequence(100, 200);
+ }
+ SECTION("estimated overlap")
+ {
+ lg_k = 8;
+ a_values = sequence(0, 10000);
+ b_values = sequence(5000, 15000);
+ }
+ const auto cpu = theta_test::cpu_set_operation_images(a_values, b_values,
lg_k);
+
+ thrust::device_vector<std::uint64_t> a_device = a_values;
+ thrust::device_vector<std::uint64_t> b_device = b_values;
+ ::cuda::stream stream{::cuda::devices[0]};
+ auto mr = ::cuda::device_default_memory_pool(::cuda::devices[0]);
+ sketch_type a(stream, mr, lg_k);
+ sketch_type b(stream, mr, lg_k);
+ a.update(stream, a_device.begin(), a_device.end());
+ b.update(stream, b_device.begin(), b_device.end());
+
+ auto set_union = clone(stream, mr, a, lg_k);
+ set_union.merge(stream, b);
+ REQUIRE(set_union.serialize_compact(stream) == cpu.set_union);
+
+ auto intersection = clone(stream, mr, a, lg_k);
+ intersection.intersect(stream, b);
+ REQUIRE(intersection.serialize_compact(stream) == cpu.intersection);
+
+ auto difference = clone(stream, mr, a, lg_k);
+ difference.a_not_b(stream, b);
+ REQUIRE(difference.serialize_compact(stream) == cpu.a_not_b);
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
