Skip to content

Commit 4e66c11

Browse files
committed
Add serialize_to sink API for count-min sketch
1 parent 6029803 commit 4e66c11

3 files changed

Lines changed: 66 additions & 50 deletions

File tree

count/include/count_min.hpp

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -326,6 +326,15 @@ class count_min_sketch{
326326
*/
327327
void serialize(std::ostream& os) const;
328328

329+
/**
330+
* This method serializes the sketch by passing binary fragments to a callback.
331+
* The callback must be callable as write_bytes(const void* data, size_t size).
332+
* The data pointer is valid only until the callback returns.
333+
* @return size in bytes written to the callback
334+
*/
335+
template<typename WriteBytes>
336+
size_t serialize_to(WriteBytes&& write_bytes) const;
337+
329338
// This is a convenience alias for users
330339
// The type returned by the following serialize method
331340
using vector_bytes = std::vector<uint8_t, typename std::allocator_traits<Allocator>::template rebind_alloc<uint8_t>>;

count/include/count_min_impl.hpp

Lines changed: 34 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -268,39 +268,54 @@ return _sketch_array.end();
268268

269269
template<typename W, typename A>
270270
void count_min_sketch<W,A>::serialize(std::ostream& os) const {
271+
serialize_to([&os](const void* data, size_t size) {
272+
os.write(static_cast<const char*>(data), size);
273+
});
274+
}
275+
276+
template<typename WriteBytes, typename T>
277+
static inline void write_count_min_value(WriteBytes& write_bytes, size_t& bytes_written, const T& value) {
278+
write_bytes(&value, sizeof(value));
279+
bytes_written += sizeof(value);
280+
}
281+
282+
template<typename W, typename A>
283+
template<typename WriteBytes>
284+
size_t count_min_sketch<W,A>::serialize_to(WriteBytes&& write_bytes) const {
285+
size_t bytes_written = 0;
286+
271287
// Long 0
272288
//const uint8_t preamble_longs = is_empty() ? PREAMBLE_LONGS_SHORT : PREAMBLE_LONGS_FULL;
273289
const uint8_t preamble_longs = PREAMBLE_LONGS_SHORT;
274290
const uint8_t ser_ver = SERIAL_VERSION_1;
275291
const uint8_t family_id = FAMILY_ID;
276292
const uint8_t flags_byte = (is_empty() ? 1 << flags::IS_EMPTY : 0);
277293
const uint32_t unused32 = NULL_32;
278-
write(os, preamble_longs);
279-
write(os, ser_ver);
280-
write(os, family_id);
281-
write(os, flags_byte);
282-
write(os, unused32);
294+
write_count_min_value(write_bytes, bytes_written, preamble_longs);
295+
write_count_min_value(write_bytes, bytes_written, ser_ver);
296+
write_count_min_value(write_bytes, bytes_written, family_id);
297+
write_count_min_value(write_bytes, bytes_written, flags_byte);
298+
write_count_min_value(write_bytes, bytes_written, unused32);
283299

284300
// Long 1
285301
const uint32_t nbuckets = _num_buckets;
286302
const uint8_t nhashes = _num_hashes;
287303
const uint16_t seed_hash(compute_seed_hash(_seed));
288304
const uint8_t unused8 = NULL_8;
289-
write(os, nbuckets);
290-
write(os, nhashes);
291-
write(os, seed_hash);
292-
write(os, unused8);
293-
if (is_empty()) return; // sketch is empty, no need to write further bytes.
305+
write_count_min_value(write_bytes, bytes_written, nbuckets);
306+
write_count_min_value(write_bytes, bytes_written, nhashes);
307+
write_count_min_value(write_bytes, bytes_written, seed_hash);
308+
write_count_min_value(write_bytes, bytes_written, unused8);
309+
if (is_empty()) return bytes_written; // sketch is empty, no need to write further bytes.
294310

295311
// Long 2
296-
write(os, _total_weight);
312+
const W t_weight = _total_weight;
313+
write_count_min_value(write_bytes, bytes_written, t_weight);
297314

298315
// Long 3 onwards: remaining bytes are consumed by writing the weight and the array values.
299-
auto it = _sketch_array.begin();
300-
while (it != _sketch_array.end()) {
301-
write(os, *it);
302-
++it;
303-
}
316+
for (const auto& value: _sketch_array) write_count_min_value(write_bytes, bytes_written, value);
317+
318+
return bytes_written;
304319
}
305320

306321
template<typename W, typename A>
@@ -351,40 +366,9 @@ template<typename W, typename A>
351366
auto count_min_sketch<W,A>::serialize(unsigned header_size_bytes) const -> vector_bytes {
352367
vector_bytes bytes(header_size_bytes + get_serialized_size_bytes(), 0, _allocator);
353368
uint8_t *ptr = bytes.data() + header_size_bytes;
354-
355-
// Long 0
356-
const uint8_t preamble_longs = PREAMBLE_LONGS_SHORT;
357-
ptr += copy_to_mem(preamble_longs, ptr);
358-
const uint8_t ser_ver = SERIAL_VERSION_1;
359-
ptr += copy_to_mem(ser_ver, ptr);
360-
const uint8_t family_id = FAMILY_ID;
361-
ptr += copy_to_mem(family_id, ptr);
362-
const uint8_t flags_byte = (is_empty() ? 1 << flags::IS_EMPTY : 0);
363-
ptr += copy_to_mem(flags_byte, ptr);
364-
const uint32_t unused32 = NULL_32;
365-
ptr += copy_to_mem(unused32, ptr);
366-
367-
// Long 1
368-
const uint32_t nbuckets = _num_buckets;
369-
const uint8_t nhashes = _num_hashes;
370-
const uint16_t seed_hash(compute_seed_hash(_seed));
371-
const uint8_t null_characters_8 = NULL_8;
372-
ptr += copy_to_mem(nbuckets, ptr);
373-
ptr += copy_to_mem(nhashes, ptr);
374-
ptr += copy_to_mem(seed_hash, ptr);
375-
ptr += copy_to_mem(null_characters_8, ptr);
376-
if (is_empty()) return bytes; // sketch is empty, no need to write further bytes.
377-
378-
// Long 2
379-
const W t_weight = _total_weight;
380-
ptr += copy_to_mem(t_weight, ptr);
381-
382-
// Long 3 onwards: remaining bytes are consumed by writing the weight and the array values.
383-
auto it = _sketch_array.begin();
384-
while (it != _sketch_array.end()) {
385-
ptr += copy_to_mem(*it, ptr);
386-
++it;
387-
}
369+
serialize_to([&ptr](const void* data, size_t size) {
370+
ptr += copy_to_mem(data, ptr, size);
371+
});
388372

389373
return bytes;
390374
}

count/test/count_min_test.cpp

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -300,4 +300,27 @@ TEST_CASE("CountMin sketch: bytes serialize-deserialize non-empty", "[cm_sketch]
300300

301301
}
302302

303+
TEST_CASE("CountMin sketch: sink serialize matches bytes", "[cm_sketch]") {
304+
auto check_sink_serialize = [](const count_min_sketch<uint64_t>& c) {
305+
auto bytes = c.serialize();
306+
std::vector<uint8_t> sink_bytes;
307+
sink_bytes.reserve(c.get_serialized_size_bytes());
308+
309+
const size_t bytes_written = c.serialize_to([&sink_bytes](const void* data, size_t size) {
310+
const auto* begin = static_cast<const uint8_t*>(data);
311+
sink_bytes.insert(sink_bytes.end(), begin, begin + size);
312+
});
313+
314+
REQUIRE(bytes_written == bytes.size());
315+
REQUIRE(sink_bytes == bytes);
316+
};
317+
318+
count_min_sketch<uint64_t> empty(3, 32);
319+
check_sink_serialize(empty);
320+
321+
count_min_sketch<uint64_t> non_empty(5, 64);
322+
for (uint64_t i=0; i < 10; ++i) non_empty.update(i, 10 * i * i);
323+
check_sink_serialize(non_empty);
324+
}
325+
303326
} /* namespace datasketches */

0 commit comments

Comments
 (0)