You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datasketches.apache.org by al...@apache.org on 2023/04/20 00:36:19 UTC
[datasketches-postgresql] branch partial_agg updated: aod sketch partial aggregation support
This is an automated email from the ASF dual-hosted git repository.
alsay pushed a commit to branch partial_agg
in repository https://gitbox.apache.org/repos/asf/datasketches-postgresql.git
The following commit(s) were added to refs/heads/partial_agg by this push:
new 0b41511 aod sketch partial aggregation support
0b41511 is described below
commit 0b415115efbbda89903a610c4d6f84d65d50aaf0
Author: AlexanderSaydakov <Al...@users.noreply.github.com>
AuthorDate: Wed Apr 19 17:36:09 2023 -0700
aod sketch partial aggregation support
---
sql/datasketches_aod_sketch.sql | 196 +++++++++++++---------
src/aod_sketch_c_adapter.cpp | 14 +-
src/aod_sketch_c_adapter.h | 4 +-
src/aod_sketch_pg_functions.c | 364 +++++++++++++++++++++++++---------------
4 files changed, 362 insertions(+), 216 deletions(-)
diff --git a/sql/datasketches_aod_sketch.sql b/sql/datasketches_aod_sketch.sql
index b737519..870d42b 100644
--- a/sql/datasketches_aod_sketch.sql
+++ b/sql/datasketches_aod_sketch.sql
@@ -19,11 +19,11 @@ CREATE TYPE aod_sketch;
CREATE OR REPLACE FUNCTION aod_sketch_in(cstring) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_sketch_in'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_out(aod_sketch) RETURNS cstring
AS '$libdir/datasketches', 'pg_sketch_out'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE TYPE aod_sketch (
INPUT = aod_sketch_in,
@@ -34,158 +34,198 @@ CREATE TYPE aod_sketch (
CREATE CAST (bytea as aod_sketch) WITHOUT FUNCTION AS ASSIGNMENT;
CREATE CAST (aod_sketch as bytea) WITHOUT FUNCTION AS ASSIGNMENT;
-CREATE OR REPLACE FUNCTION aod_sketch_add_item(internal, anyelement, double precision[]) RETURNS internal
- AS '$libdir/datasketches', 'pg_aod_sketch_add_item'
- LANGUAGE C IMMUTABLE;
+CREATE OR REPLACE FUNCTION aod_sketch_build_agg(internal, anyelement, double precision[]) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION aod_sketch_add_item(internal, anyelement, double precision[], int) RETURNS internal
- AS '$libdir/datasketches', 'pg_aod_sketch_add_item'
- LANGUAGE C IMMUTABLE;
+CREATE OR REPLACE FUNCTION aod_sketch_build_agg(internal, anyelement, double precision[], int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION aod_sketch_add_item(internal, anyelement, double precision[], int, real) RETURNS internal
- AS '$libdir/datasketches', 'pg_aod_sketch_add_item'
- LANGUAGE C IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION aod_sketch_get_estimate(aod_sketch) RETURNS double precision
- AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION aod_sketch_get_estimate_and_bounds(aod_sketch) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION aod_sketch_get_estimate_and_bounds(aod_sketch, int) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION aod_sketch_build_agg(internal, anyelement, double precision[], int, real) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_from_internal(internal) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_from_internal'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION aod_sketch_to_string(aod_sketch) RETURNS TEXT
- AS '$libdir/datasketches', 'pg_aod_sketch_to_string'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION aod_sketch_to_string(aod_sketch, boolean) RETURNS TEXT
- AS '$libdir/datasketches', 'pg_aod_sketch_to_string'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_union_agg(internal, aod_sketch) RETURNS internal
AS '$libdir/datasketches', 'pg_aod_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_union_agg(internal, aod_sketch, int) RETURNS internal
AS '$libdir/datasketches', 'pg_aod_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_union_agg(internal, aod_sketch, int, int) RETURNS internal
AS '$libdir/datasketches', 'pg_aod_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_intersection_agg(internal, aod_sketch) RETURNS internal
AS '$libdir/datasketches', 'pg_aod_sketch_intersection_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_intersection_agg(internal, aod_sketch, int) RETURNS internal
AS '$libdir/datasketches', 'pg_aod_sketch_intersection_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_union_combine(internal, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_union_combine'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION aod_union_get_result(internal) RETURNS aod_sketch
- AS '$libdir/datasketches', 'pg_aod_union_get_result'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION aod_sketch_intersection_combine(internal, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_intersection_combine'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION aod_intersection_get_result(internal) RETURNS aod_sketch
- AS '$libdir/datasketches', 'pg_aod_intersection_get_result'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION aod_sketch_serialize_state(internal) RETURNS bytea
+ AS '$libdir/datasketches', 'pg_aod_sketch_serialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_deserialize_state(bytea, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_deserialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE AGGREGATE aod_sketch_build(anyelement, double precision[]) (
- sfunc = aod_sketch_add_item,
- stype = internal,
- finalfunc = aod_sketch_from_internal
+ STYPE = internal,
+ SFUNC = aod_sketch_build_agg,
+ COMBINEFUNC = aod_sketch_union_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE aod_sketch_build(anyelement, double precision[], int) (
- sfunc = aod_sketch_add_item,
- stype = internal,
- finalfunc = aod_sketch_from_internal
+ STYPE = internal,
+ SFUNC = aod_sketch_build_agg,
+ COMBINEFUNC = aod_sketch_union_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE aod_sketch_build(anyelement, double precision[], int, real) (
- sfunc = aod_sketch_add_item,
- stype = internal,
- finalfunc = aod_sketch_from_internal
+ STYPE = internal,
+ SFUNC = aod_sketch_build_agg,
+ COMBINEFUNC = aod_sketch_union_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE aod_sketch_union(aod_sketch) (
- sfunc = aod_sketch_union_agg,
- stype = internal,
- finalfunc = aod_union_get_result
+ STYPE = internal,
+ SFUNC = aod_sketch_union_agg,
+ COMBINEFUNC = aod_sketch_union_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE aod_sketch_union(aod_sketch, int) (
- sfunc = aod_sketch_union_agg,
- stype = internal,
- finalfunc = aod_union_get_result
+ STYPE = internal,
+ SFUNC = aod_sketch_union_agg,
+ COMBINEFUNC = aod_sketch_union_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE aod_sketch_union(aod_sketch, int, int) (
- sfunc = aod_sketch_union_agg,
- stype = internal,
- finalfunc = aod_union_get_result
+ STYPE = internal,
+ SFUNC = aod_sketch_union_agg,
+ COMBINEFUNC = aod_sketch_union_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE aod_sketch_intersection(aod_sketch) (
- sfunc = aod_sketch_intersection_agg,
- stype = internal,
- finalfunc = aod_intersection_get_result
+ STYPE = internal,
+ SFUNC = aod_sketch_intersection_agg,
+ COMBINEFUNC = aod_sketch_intersection_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE aod_sketch_intersection(aod_sketch, int) (
- sfunc = aod_sketch_intersection_agg,
- stype = internal,
- finalfunc = aod_intersection_get_result
+ STYPE = internal,
+ SFUNC = aod_sketch_intersection_agg,
+ COMBINEFUNC = aod_sketch_intersection_combine,
+ SERIALFUNC = aod_sketch_serialize_state,
+ DESERIALFUNC = aod_sketch_deserialize_state,
+ FINALFUNC = aod_sketch_from_internal,
+ PARALLEL = SAFE
);
+CREATE OR REPLACE FUNCTION aod_sketch_get_estimate(aod_sketch) RETURNS double precision
+ AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_get_estimate_and_bounds(aod_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_get_estimate_and_bounds(aod_sketch, int) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_string(aod_sketch) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_string(aod_sketch, boolean) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
CREATE OR REPLACE FUNCTION aod_sketch_union(aod_sketch, aod_sketch) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_union'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_union(aod_sketch, aod_sketch, int) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_union'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_intersection(aod_sketch, aod_sketch) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_intersection'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_intersection(aod_sketch, aod_sketch, int) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_intersection'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_a_not_b(aod_sketch, aod_sketch) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_a_not_b'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_a_not_b(aod_sketch, aod_sketch, int) RETURNS aod_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_a_not_b'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_to_kll_float_sketch(aod_sketch, int) RETURNS kll_float_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_to_kll_float_sketch'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_to_kll_float_sketch(aod_sketch, int, int) RETURNS kll_float_sketch
AS '$libdir/datasketches', 'pg_aod_sketch_to_kll_float_sketch'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_students_t_test(aod_sketch, aod_sketch) RETURNS double precision[]
AS '$libdir/datasketches', 'pg_aod_sketch_students_t_test'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_to_means(aod_sketch) RETURNS double precision[]
AS '$libdir/datasketches', 'pg_aod_sketch_to_means'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION aod_sketch_to_variances(aod_sketch) RETURNS double precision[]
AS '$libdir/datasketches', 'pg_aod_sketch_to_variances'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
diff --git a/src/aod_sketch_c_adapter.cpp b/src/aod_sketch_c_adapter.cpp
index ccc05de..0147c51 100644
--- a/src/aod_sketch_c_adapter.cpp
+++ b/src/aod_sketch_c_adapter.cpp
@@ -218,9 +218,12 @@ void aod_union_update(void* unionptr, const void* sketchptr) {
}
}
-void* aod_union_get_result(const void* unionptr) {
+void* aod_union_get_result(void* unionptr) {
try {
- return new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(static_cast<const aod_union_pg*>(unionptr)->get_result());
+ auto sketchptr = new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(static_cast<const aod_union_pg*>(unionptr)->get_result());
+ static_cast<aod_union_pg*>(unionptr)->~aod_union_pg();
+ pfree(unionptr);
+ return sketchptr;
} catch (std::exception& e) {
pg_error(e.what());
}
@@ -253,9 +256,12 @@ void aod_intersection_update(void* interptr, const void* sketchptr) {
}
}
-void* aod_intersection_get_result(const void* interptr) {
+void* aod_intersection_get_result(void* interptr) {
try {
- return new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(static_cast<const aod_intersection_pg*>(interptr)->get_result());
+ auto sketchptr = new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(static_cast<aod_intersection_pg*>(interptr)->get_result());
+ static_cast<aod_intersection_pg*>(interptr)->~aod_intersection_pg();
+ pfree(interptr);
+ return sketchptr;
} catch (std::exception& e) {
pg_error(e.what());
}
diff --git a/src/aod_sketch_c_adapter.h b/src/aod_sketch_c_adapter.h
index 9f36b10..7918c24 100644
--- a/src/aod_sketch_c_adapter.h
+++ b/src/aod_sketch_c_adapter.h
@@ -47,12 +47,12 @@ void* aod_union_new(unsigned num_values);
void* aod_union_new_lgk(unsigned num_values, unsigned lg_k);
void aod_union_delete(void* unionptr);
void aod_union_update(void* unionptr, const void* sketchptr);
-void* aod_union_get_result(const void* unionptr);
+void* aod_union_get_result(void* unionptr);
void* aod_intersection_new(unsigned num_values);
void aod_intersection_delete(void* interptr);
void aod_intersection_update(void* interptr, const void* sketchptr);
-void* aod_intersection_get_result(const void* interptr);
+void* aod_intersection_get_result(void* interptr);
void* aod_a_not_b(const void* sketchptr1, const void* sketchptr2);
diff --git a/src/aod_sketch_pg_functions.c b/src/aod_sketch_pg_functions.c
index d3d27cc..7a951b5 100644
--- a/src/aod_sketch_pg_functions.c
+++ b/src/aod_sketch_pg_functions.c
@@ -28,17 +28,27 @@
#include "base64.h"
#include "kll_float_sketch_c_adapter.h"
+enum aod_agg_state_type { MUTABLE_SKETCH, IMMUTABLE_SKETCH, UNION, INTERSECTION };
+
+struct aod_agg_state {
+ enum aod_agg_state_type type;
+ unsigned lg_k;
+ unsigned num_values;
+ void* ptr;
+};
+
/* PG_FUNCTION_INFO_V1 macro to pass functions to postgres */
-PG_FUNCTION_INFO_V1(pg_aod_sketch_add_item);
-PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate);
-PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate_and_bounds);
-PG_FUNCTION_INFO_V1(pg_aod_sketch_to_string);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_build_agg);
PG_FUNCTION_INFO_V1(pg_aod_sketch_union_agg);
PG_FUNCTION_INFO_V1(pg_aod_sketch_intersection_agg);
PG_FUNCTION_INFO_V1(pg_aod_sketch_from_internal);
-PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate_from_internal);
-PG_FUNCTION_INFO_V1(pg_aod_union_get_result);
-PG_FUNCTION_INFO_V1(pg_aod_intersection_get_result);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_union_combine);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_intersection_combine);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_serialize_state);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_deserialize_state);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate_and_bounds);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_to_string);
PG_FUNCTION_INFO_V1(pg_aod_sketch_union);
PG_FUNCTION_INFO_V1(pg_aod_sketch_intersection);
PG_FUNCTION_INFO_V1(pg_aod_sketch_a_not_b);
@@ -48,18 +58,17 @@ PG_FUNCTION_INFO_V1(pg_aod_sketch_to_means);
PG_FUNCTION_INFO_V1(pg_aod_sketch_to_variances);
/* function declarations */
-Datum pg_aod_sketch_recv(PG_FUNCTION_ARGS);
-Datum pg_aod_sketch_send(PG_FUNCTION_ARGS);
-Datum pg_aod_sketch_add_item(PG_FUNCTION_ARGS);
-Datum pg_aod_sketch_get_estimate(PG_FUNCTION_ARGS);
-Datum pg_aod_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
-Datum pg_aod_sketch_to_string(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_build_agg(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_union_agg(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_intersection_agg(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_from_internal(PG_FUNCTION_ARGS);
-Datum pg_aod_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS);
-Datum pg_aod_union_get_result(PG_FUNCTION_ARGS);
-Datum pg_aod_intersection_get_result(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_union_combine(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_intersection_combine(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_serialize_state(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_deserialize_state(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_get_estimate(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_to_string(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_union(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_intersection(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_a_not_b(PG_FUNCTION_ARGS);
@@ -68,9 +77,8 @@ Datum pg_aod_sketch_students_t_test(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_to_means(PG_FUNCTION_ARGS);
Datum pg_aod_sketch_to_variances(PG_FUNCTION_ARGS);
-Datum pg_aod_sketch_add_item(PG_FUNCTION_ARGS) {
- void* sketchptr;
- int lg_k;
+Datum pg_aod_sketch_build_agg(PG_FUNCTION_ARGS) {
+ struct aod_agg_state* stateptr;
float p;
// anyelement
@@ -102,7 +110,7 @@ Datum pg_aod_sketch_add_item(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "aod_sketch_add_item called in non-aggregate context");
+ elog(ERROR, "aod_sketch_build_agg called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
@@ -118,15 +126,18 @@ Datum pg_aod_sketch_add_item(PG_FUNCTION_ARGS) {
}
if (PG_ARGISNULL(0)) {
- lg_k = PG_NARGS() > 3 ? PG_GETARG_INT32(3) : 0;
+ stateptr = palloc(sizeof(struct aod_agg_state));
+ stateptr->type = MUTABLE_SKETCH;
+ stateptr->lg_k = PG_NARGS() > 3 ? PG_GETARG_INT32(3) : 0;
+ stateptr->num_values = arr_len;
p = PG_NARGS() > 4 ? PG_GETARG_FLOAT4(4) : 1;
- if (lg_k) {
- sketchptr = p ? aod_sketch_new_lgk_p(arr_len, lg_k, p) : aod_sketch_new_lgk(arr_len, lg_k);
+ if (stateptr->lg_k) {
+ stateptr->ptr = p ? aod_sketch_new_lgk_p(arr_len, stateptr->lg_k, p) : aod_sketch_new_lgk(arr_len, stateptr->lg_k);
} else {
- sketchptr = aod_sketch_new(arr_len);
+ stateptr->ptr = aod_sketch_new(arr_len);
}
} else {
- sketchptr = PG_GETARG_POINTER(0);
+ stateptr = (struct aod_agg_state*) PG_GETARG_POINTER(0);
}
element_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
@@ -134,73 +145,24 @@ Datum pg_aod_sketch_add_item(PG_FUNCTION_ARGS) {
get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);
if (typlen == -1) {
// varlena
- aod_sketch_update(sketchptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element), values);
+ aod_sketch_update(stateptr->ptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element), values);
} else if (typbyval) {
// fixed-length passed by value
- aod_sketch_update(sketchptr, &element, typlen, values);
+ aod_sketch_update(stateptr->ptr, &element, typlen, values);
} else {
// fixed-length passed by reference
- aod_sketch_update(sketchptr, (void*)element, typlen, values);
+ aod_sketch_update(stateptr->ptr, (void*)element, typlen, values);
}
pfree(values);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(sketchptr);
+ PG_RETURN_POINTER(stateptr);
}
-Datum pg_aod_sketch_get_estimate(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- double estimate;
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- estimate = compact_aod_sketch_get_estimate(sketchptr);
- compact_aod_sketch_delete(sketchptr);
- PG_RETURN_FLOAT8(estimate);
-}
-
-Datum pg_aod_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- int num_std_devs;
-
- // output array
- Datum* est_and_bounds;
- ArrayType* arr_out;
- int16 elmlen_out;
- bool elmbyval_out;
- char elmalign_out;
-
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
- est_and_bounds = (Datum*) aod_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
- compact_aod_sketch_delete(sketchptr);
-
- // construct output array
- get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
- arr_out = construct_array(est_and_bounds, 3, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
- PG_RETURN_ARRAYTYPE_P(arr_out);
-}
-
-Datum pg_aod_sketch_to_string(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- bool print_entries;
- char* str;
- bytes_in = PG_GETARG_BYTEA_P(0);
- print_entries = PG_NARGS() > 1 ? PG_GETARG_BOOL(1) : false;
- sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- str = aod_sketch_to_string(sketchptr, print_entries);
- compact_aod_sketch_delete(sketchptr);
- PG_RETURN_TEXT_P(cstring_to_text(str));
-}
-
-Datum pg_aod_sketch_intersection_agg(PG_FUNCTION_ARGS) {
- void* interptr;
+Datum pg_aod_sketch_union_agg(PG_FUNCTION_ARGS) {
+ struct aod_agg_state* stateptr;
bytea* sketch_bytes;
void* sketchptr;
- int num_values;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -212,33 +174,34 @@ Datum pg_aod_sketch_intersection_agg(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "aod_sketch_intersect called in non-aggregate context");
+ elog(ERROR, "aod_sketch_union_agg called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- num_values = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 1;
- interptr = aod_intersection_new(num_values);
+ stateptr = palloc(sizeof(struct aod_agg_state));
+ stateptr->type = UNION;
+ stateptr->num_values = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 1;
+ stateptr->lg_k = PG_NARGS() > 3 ? PG_GETARG_INT32(3) : 0;
+ stateptr->ptr = stateptr->lg_k ? aod_union_new_lgk(stateptr->num_values, stateptr->lg_k) : aod_union_new(stateptr->num_values);
} else {
- interptr = PG_GETARG_POINTER(0);
+ stateptr = (struct aod_agg_state*) PG_GETARG_POINTER(0);
}
sketch_bytes = PG_GETARG_BYTEA_P(1);
sketchptr = aod_sketch_deserialize(VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
- aod_intersection_update(interptr, sketchptr);
+ aod_union_update(stateptr->ptr, sketchptr);
compact_aod_sketch_delete(sketchptr);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(interptr);
+ PG_RETURN_POINTER(stateptr);
}
-Datum pg_aod_sketch_union_agg(PG_FUNCTION_ARGS) {
- void* unionptr;
+Datum pg_aod_sketch_intersection_agg(PG_FUNCTION_ARGS) {
+ struct aod_agg_state* stateptr;
bytea* sketch_bytes;
void* sketchptr;
- int num_values;
- int lg_k;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -250,30 +213,31 @@ Datum pg_aod_sketch_union_agg(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "aod_sketch_merge called in non-aggregate context");
+ elog(ERROR, "aod_sketch_intersection_agg called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- num_values = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 1;
- lg_k = PG_NARGS() > 3 ? PG_GETARG_INT32(3) : 0;
- unionptr = lg_k ? aod_union_new_lgk(num_values, lg_k) : aod_union_new(num_values);
+ stateptr = palloc(sizeof(struct aod_agg_state));
+ stateptr->type = INTERSECTION;
+ stateptr->num_values = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 1;
+ stateptr->ptr = aod_intersection_new(stateptr->num_values);
} else {
- unionptr = PG_GETARG_POINTER(0);
+ stateptr = (struct aod_agg_state*) PG_GETARG_POINTER(0);
}
sketch_bytes = PG_GETARG_BYTEA_P(1);
sketchptr = aod_sketch_deserialize(VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
- aod_union_update(unionptr, sketchptr);
+ aod_intersection_update(stateptr->ptr, sketchptr);
compact_aod_sketch_delete(sketchptr);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(unionptr);
+ PG_RETURN_POINTER(stateptr);
}
Datum pg_aod_sketch_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
+ struct aod_agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -286,10 +250,17 @@ Datum pg_aod_sketch_from_internal(PG_FUNCTION_ARGS) {
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- sketchptr = aod_sketch_compact(sketchptr);
- bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
- compact_aod_sketch_delete(sketchptr);
+ stateptr = (struct aod_agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == MUTABLE_SKETCH) {
+ stateptr->ptr = aod_sketch_compact(stateptr->ptr);
+ } else if (stateptr->type == UNION) {
+ stateptr->ptr = aod_union_get_result(stateptr->ptr);
+ } else if (stateptr->type == INTERSECTION) {
+ stateptr->ptr = aod_intersection_get_result(stateptr->ptr);
+ }
+ bytes_out = aod_sketch_serialize(stateptr->ptr, VARHDRSZ);
+ compact_aod_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
@@ -297,32 +268,106 @@ Datum pg_aod_sketch_from_internal(PG_FUNCTION_ARGS) {
PG_RETURN_BYTEA_P(bytes_out.ptr);
}
-Datum pg_aod_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
- double estimate;
+Datum pg_aod_sketch_union_combine(PG_FUNCTION_ARGS) {
+ struct aod_agg_state* stateptr1;
+ struct aod_agg_state* stateptr2;
+ struct aod_agg_state* stateptr;
MemoryContext oldcontext;
MemoryContext aggcontext;
- if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+ if (PG_ARGISNULL(0) && PG_ARGISNULL(1)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "aod_sketch_from_internal called in non-aggregate context");
+ elog(ERROR, "aod_sketch_union_combine called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ stateptr1 = (struct aod_agg_state*) PG_GETARG_POINTER(0);
+ stateptr2 = (struct aod_agg_state*) PG_GETARG_POINTER(1);
+
+ stateptr = palloc(sizeof(struct aod_agg_state));
+ stateptr->type = IMMUTABLE_SKETCH;
+ stateptr->lg_k = stateptr1 ? stateptr1->lg_k : stateptr2->lg_k;
+ stateptr->num_values = stateptr1 ? stateptr1->num_values : stateptr2->num_values;
+ stateptr->ptr = stateptr->lg_k ? aod_union_new_lgk(stateptr->num_values, stateptr->lg_k) : aod_union_new(stateptr->num_values);
+ if (stateptr1) {
+ if (stateptr1->type == UNION) {
+ stateptr1->ptr = aod_union_get_result(stateptr1->ptr);
+ } else if (stateptr1->type == MUTABLE_SKETCH) {
+ stateptr1->ptr = aod_sketch_compact(stateptr1->ptr);
+ }
+ aod_union_update(stateptr->ptr, stateptr1->ptr);
+ compact_aod_sketch_delete(stateptr1->ptr);
+ pfree(stateptr1);
+ }
+ if (stateptr2) {
+ if (stateptr2->type == UNION) {
+ stateptr2->ptr = aod_union_get_result(stateptr2->ptr);
+ } else if (stateptr2->type == MUTABLE_SKETCH) {
+ stateptr2->ptr = aod_sketch_compact(stateptr2->ptr);
+ }
+ aod_union_update(stateptr->ptr, stateptr2->ptr);
+ compact_aod_sketch_delete(stateptr2->ptr);
+ pfree(stateptr2);
+ }
+ stateptr->ptr = aod_union_get_result(stateptr->ptr);
+ stateptr->type = IMMUTABLE_SKETCH;
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(stateptr);
+}
+
+Datum pg_aod_sketch_intersection_combine(PG_FUNCTION_ARGS) {
+ struct aod_agg_state* stateptr1;
+ struct aod_agg_state* stateptr2;
+ struct aod_agg_state* stateptr;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0) && PG_ARGISNULL(1)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "aod_sketch_combine called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- estimate = update_aod_sketch_get_estimate(sketchptr);
- update_aod_sketch_delete(sketchptr);
+ stateptr1 = (struct aod_agg_state*) PG_GETARG_POINTER(0);
+ stateptr2 = (struct aod_agg_state*) PG_GETARG_POINTER(1);
+
+ stateptr = palloc(sizeof(struct aod_agg_state));
+ stateptr->type = IMMUTABLE_SKETCH;
+ stateptr->lg_k = stateptr1 ? stateptr1->lg_k : stateptr2->lg_k;
+ stateptr->num_values = stateptr1 ? stateptr1->num_values : stateptr2->num_values;
+ stateptr->ptr = aod_intersection_new(stateptr->num_values);
+ if (stateptr1) {
+ if (stateptr1->type == INTERSECTION) {
+ stateptr1->ptr = aod_intersection_get_result(stateptr1->ptr);
+ }
+ aod_intersection_update(stateptr->ptr, stateptr1->ptr);
+ compact_aod_sketch_delete(stateptr1->ptr);
+ pfree(stateptr1);
+ }
+ if (stateptr2) {
+ if (stateptr2->type == INTERSECTION) {
+ stateptr2->ptr = aod_intersection_get_result(stateptr2->ptr);
+ }
+ aod_intersection_update(stateptr->ptr, stateptr2->ptr);
+ compact_aod_sketch_delete(stateptr2->ptr);
+ pfree(stateptr2);
+ }
+ stateptr->ptr = aod_intersection_get_result(stateptr->ptr);
+ stateptr->type = IMMUTABLE_SKETCH;
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_FLOAT8(estimate);
+ PG_RETURN_POINTER(stateptr);
}
-Datum pg_aod_union_get_result(PG_FUNCTION_ARGS) {
- void* unionptr;
- void* sketchptr;
+Datum pg_aod_sketch_serialize_state(PG_FUNCTION_ARGS) {
+ struct aod_agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -331,15 +376,23 @@ Datum pg_aod_union_get_result(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "aod_union_get_result called in non-aggregate context");
+ elog(ERROR, "aod_sketch_serialize_state called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- unionptr = PG_GETARG_POINTER(0);
- sketchptr = aod_union_get_result(unionptr);
- bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
- compact_aod_sketch_delete(sketchptr);
- aod_union_delete(unionptr);
+ stateptr = (struct aod_agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == MUTABLE_SKETCH) {
+ stateptr->ptr = aod_sketch_compact(stateptr->ptr);
+ } else if (stateptr->type == UNION) {
+ stateptr->ptr = aod_union_get_result(stateptr->ptr);
+ } else if (stateptr->type == INTERSECTION) {
+ stateptr->ptr = aod_intersection_get_result(stateptr->ptr);
+ }
+ bytes_out = aod_sketch_serialize(stateptr->ptr, VARHDRSZ + 2);
+ ((char*)bytes_out.ptr)[VARHDRSZ] = stateptr->lg_k;
+ ((char*)bytes_out.ptr)[VARHDRSZ + 1] = stateptr->num_values;
+ compact_aod_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
@@ -347,10 +400,9 @@ Datum pg_aod_union_get_result(PG_FUNCTION_ARGS) {
PG_RETURN_BYTEA_P(bytes_out.ptr);
}
-Datum pg_aod_intersection_get_result(PG_FUNCTION_ARGS) {
- void* interptr;
- void* sketchptr;
- struct ptr_with_size bytes_out;
+Datum pg_aod_sketch_deserialize_state(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ struct aod_agg_state* stateptr;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -358,20 +410,68 @@ Datum pg_aod_intersection_get_result(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "aod_intersection_get_result called in non-aggregate context");
+ elog(ERROR, "aod_sketch_deserialize_state called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- interptr = PG_GETARG_POINTER(0);
- sketchptr = aod_intersection_get_result(interptr);
- bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
- compact_aod_sketch_delete(sketchptr);
- aod_intersection_delete(interptr);
- SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ stateptr = palloc(sizeof(struct aod_agg_state));
+ stateptr->type = IMMUTABLE_SKETCH;
+ stateptr->lg_k = *VARDATA(bytes_in);
+ stateptr->num_values = *(VARDATA(bytes_in) + 1);
+ stateptr->ptr = aod_sketch_deserialize(VARDATA(bytes_in) + 2, VARSIZE(bytes_in) - VARHDRSZ - 2);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_BYTEA_P(bytes_out.ptr);
+ PG_RETURN_POINTER(stateptr);
+}
+
+Datum pg_aod_sketch_get_estimate(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ double estimate;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ estimate = compact_aod_sketch_get_estimate(sketchptr);
+ compact_aod_sketch_delete(sketchptr);
+ PG_RETURN_FLOAT8(estimate);
+}
+
+Datum pg_aod_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ int num_std_devs;
+
+ // output array
+ Datum* est_and_bounds;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
+ est_and_bounds = (Datum*) aod_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+ compact_aod_sketch_delete(sketchptr);
+
+ // construct output array
+ get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(est_and_bounds, 3, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
+
+Datum pg_aod_sketch_to_string(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ bool print_entries;
+ char* str;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ print_entries = PG_NARGS() > 1 ? PG_GETARG_BOOL(1) : false;
+ sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ str = aod_sketch_to_string(sketchptr, print_entries);
+ compact_aod_sketch_delete(sketchptr);
+ PG_RETURN_TEXT_P(cstring_to_text(str));
}
Datum pg_aod_sketch_union(PG_FUNCTION_ARGS) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org