feat: Support HLL

chore: working

chore: deserialize

chore: working

chore: working

chore: more tests

chore: support target hll type

chore: support target hll type for new sketch

Author: ovr

build.rs

@@ -9,6 +9,7 @@ fn main() {
     bridge
         .files(&[
             datasketches.join("cpc.cpp"),
+            datasketches.join("hll.cpp"),
             datasketches.join("theta.cpp"),
             datasketches.join("hh.cpp"),
         ])

datasketches-cpp/hll.cpp

@@ -0,0 +1,78 @@
+#include <cstdint>
+#include <ios>
+#include <sstream>
+#include <iostream>
+
+#include "rust/cxx.h"
+#include "hll/include/hll.hpp"
+
+#include "hll.hpp"
+
+OpaqueHLLSketch::OpaqueHLLSketch(unsigned lg_k, datasketches::target_hll_type tgt_type):
+  inner_{ datasketches::hll_sketch(lg_k, tgt_type) } {
+}
+
+OpaqueHLLSketch::OpaqueHLLSketch(datasketches::hll_sketch&& hll):
+  inner_{std::move(hll)} {
+}
+
+OpaqueHLLSketch::OpaqueHLLSketch(std::istream& is):
+  inner_{datasketches::hll_sketch::deserialize(is)} {
+}
+
+double OpaqueHLLSketch::estimate() const {
+  return this->inner_.get_estimate();
+}
+
+void OpaqueHLLSketch::update(rust::Slice<const uint8_t> buf) {
+  this->inner_.update(buf.data(), buf.size());
+}
+
+void OpaqueHLLSketch::update_u64(uint64_t value) {
+  this->inner_.update(value);
+}
+
+std::unique_ptr<std::vector<uint8_t>> OpaqueHLLSketch::serialize() const {
+  // TODO: could use a custom streambuf to avoid the
+  // stream -> vec copy https://stackoverflow.com/a/13059195/1779853
+  std::stringstream s{};
+  auto start = s.tellg();
+  this->inner_.serialize_compact(s);
+  s.seekg(0, std::ios::end);
+  auto stop = s.tellg();
+
+  std::vector<uint8_t> v(std::size_t(stop-start));
+  s.seekg(0, std::ios::beg);
+  s.read(reinterpret_cast<char*>(v.data()), std::streamsize(v.size()));
+
+  return std::unique_ptr<std::vector<uint8_t>>(new std::vector<uint8_t>(std::move(v)));
+}
+
+std::unique_ptr<OpaqueHLLSketch> new_opaque_hll_sketch(unsigned lg_k, datasketches::target_hll_type tgt_type) {
+  return std::unique_ptr<OpaqueHLLSketch>(new OpaqueHLLSketch { lg_k, tgt_type });
+}
+
+std::unique_ptr<OpaqueHLLSketch> deserialize_opaque_hll_sketch(rust::Slice<const uint8_t> buf) {
+    // TODO: could use a custom streambuf to avoid the slice -> stream copy
+    std::stringstream s{};
+    s.write(const_cast<char*>(reinterpret_cast<const char*>(buf.data())), std::streamsize(buf.size()));
+    s.seekg(0, std::ios::beg);
+    return std::unique_ptr<OpaqueHLLSketch>(new OpaqueHLLSketch{s});
+}
+
+OpaqueHLLUnion::OpaqueHLLUnion(uint8_t lg_max_k):
+  inner_{ datasketches::hll_union(lg_max_k) } {
+}
+
+std::unique_ptr<OpaqueHLLSketch> OpaqueHLLUnion::sketch(datasketches::target_hll_type tgt_type) const {
+  return std::unique_ptr<OpaqueHLLSketch>(new OpaqueHLLSketch{this->inner_.get_result(tgt_type)});
+}
+
+void OpaqueHLLUnion::merge(std::unique_ptr<OpaqueHLLSketch> to_add) {
+  this->inner_.update(std::move(to_add->inner_));
+}
+
+
+std::unique_ptr<OpaqueHLLUnion> new_opaque_hll_union(uint8_t lg_max_k) {
+  return std::unique_ptr<OpaqueHLLUnion>(new OpaqueHLLUnion{ lg_max_k });
+}

datasketches-cpp/hll.hpp

@@ -0,0 +1,41 @@
+#pragma once
+
+#include <cstdint>
+#include <iostream>
+#include <vector>
+#include <memory>
+
+#include "rust/cxx.h"
+#include "hll/include/hll.hpp"
+
+// alias
+typedef datasketches::target_hll_type target_hll_type;
+
+class OpaqueHLLSketch {
+public:
+  double estimate() const;
+  void update(rust::Slice<const uint8_t> buf);
+  void update_u64(uint64_t value);
+  std::unique_ptr<std::vector<uint8_t>> serialize() const;
+  friend std::unique_ptr<OpaqueHLLSketch> deserialize_opaque_hll_sketch(rust::Slice<const uint8_t> buf);
+  OpaqueHLLSketch(unsigned lg_k, datasketches::target_hll_type tgt_type);
+private:
+  OpaqueHLLSketch(datasketches::hll_sketch&& hll);
+  OpaqueHLLSketch(std::istream& is);
+  friend class OpaqueHLLUnion;
+  datasketches::hll_sketch inner_;
+};
+
+std::unique_ptr<OpaqueHLLSketch> new_opaque_hll_sketch(unsigned lg_k, datasketches::target_hll_type tgt_type);
+std::unique_ptr<OpaqueHLLSketch> deserialize_opaque_hll_sketch(rust::Slice<const uint8_t> buf);
+
+class OpaqueHLLUnion {
+public:
+  std::unique_ptr<OpaqueHLLSketch> sketch(datasketches::target_hll_type tgt_type) const;
+  void merge(std::unique_ptr<OpaqueHLLSketch> to_add);
+  OpaqueHLLUnion(uint8_t lg_max_k);
+private:
+  datasketches::hll_union inner_;
+};
+
+std::unique_ptr<OpaqueHLLUnion> new_opaque_hll_union(uint8_t lg_max_k);

src/bridge.rs

@@ -18,6 +18,13 @@ pub(crate) mod ffi {
         unsafe fn remove_from_hashset(hashset_addr: usize, addr: usize);
     }
 
+    #[repr(i32)]
+    enum target_hll_type {
+        HLL_4,
+        HLL_6,
+        HLL_8,
+    }
+
     unsafe extern "C++" {
         include!("dsrs/datasketches-cpp/cpc.hpp");
 
@@ -38,6 +45,31 @@ pub(crate) mod ffi {
         pub(crate) fn sketch(self: &OpaqueCpcUnion) -> UniquePtr<OpaqueCpcSketch>;
         pub(crate) fn merge(self: Pin<&mut OpaqueCpcUnion>, to_add: UniquePtr<OpaqueCpcSketch>);
 
+        include!("dsrs/datasketches-cpp/hll.hpp");
+
+        type target_hll_type;
+
+        pub(crate) type OpaqueHLLSketch;
+        pub(crate) fn estimate(self: &OpaqueHLLSketch) -> f64;
+        pub(crate) fn update(self: Pin<&mut OpaqueHLLSketch>, buf: &[u8]);
+        pub(crate) fn update_u64(self: Pin<&mut OpaqueHLLSketch>, value: u64);
+        pub(crate) fn serialize(self: &OpaqueHLLSketch) -> UniquePtr<CxxVector<u8>>;
+
+        pub(crate) fn new_opaque_hll_sketch(
+            lg_k: u32,
+            tgt_type: target_hll_type,
+        ) -> UniquePtr<OpaqueHLLSketch>;
+        pub(crate) fn deserialize_opaque_hll_sketch(buf: &[u8]) -> UniquePtr<OpaqueHLLSketch>;
+
+        pub(crate) type OpaqueHLLUnion;
+
+        pub(crate) fn new_opaque_hll_union(lg_max_k: u8) -> UniquePtr<OpaqueHLLUnion>;
+        pub(crate) fn sketch(
+            self: &OpaqueHLLUnion,
+            tgt_type: target_hll_type,
+        ) -> UniquePtr<OpaqueHLLSketch>;
+        pub(crate) fn merge(self: Pin<&mut OpaqueHLLUnion>, to_add: UniquePtr<OpaqueHLLSketch>);
+
         include!("dsrs/datasketches-cpp/theta.hpp");
 
         pub(crate) type OpaqueThetaSketch;

src/lib.rs

@@ -7,10 +7,4 @@ pub mod stream_reducer;
 mod wrapper;
 
 pub use error::DataSketchesError;
-pub use wrapper::CpcSketch;
-pub use wrapper::CpcUnion;
-pub use wrapper::HhSketch;
-pub use wrapper::StaticThetaSketch;
-pub use wrapper::ThetaIntersection;
-pub use wrapper::ThetaSketch;
-pub use wrapper::ThetaUnion;
+pub use wrapper::*;

src/wrapper.rs

@@ -7,8 +7,10 @@
 
 mod cpc;
 pub(crate) mod hh;
+mod hll;
 mod theta;
 
 pub use cpc::{CpcSketch, CpcUnion};
 pub use hh::HhSketch;
+pub use hll::{HLLSketch, HLLType, HLLUnion};
 pub use theta::{StaticThetaSketch, ThetaIntersection, ThetaSketch, ThetaUnion};