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/18 02:57:16 UTC
[datasketches-postgresql] 03/03: HLL 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
commit ea8306d878324ebf92e9f451fb88d081f4e99961
Author: AlexanderSaydakov <Al...@users.noreply.github.com>
AuthorDate: Mon Apr 17 19:56:57 2023 -0700
HLL sketch partial aggregation support
---
sql/datasketches_hll_sketch.sql | 164 +++++++++++++--------
src/hll_sketch_c_adapter.cpp | 10 +-
src/hll_sketch_pg_functions.c | 317 ++++++++++++++++++++++++++--------------
3 files changed, 319 insertions(+), 172 deletions(-)
diff --git a/sql/datasketches_hll_sketch.sql b/sql/datasketches_hll_sketch.sql
index e19f972..f8c4a64 100644
--- a/sql/datasketches_hll_sketch.sql
+++ b/sql/datasketches_hll_sketch.sql
@@ -19,11 +19,11 @@ CREATE TYPE hll_sketch;
CREATE OR REPLACE FUNCTION hll_sketch_in(cstring) RETURNS hll_sketch
AS '$libdir/datasketches', 'pg_sketch_in'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_out(hll_sketch) RETURNS cstring
AS '$libdir/datasketches', 'pg_sketch_out'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE TYPE hll_sketch (
INPUT = hll_sketch_in,
@@ -34,114 +34,154 @@ CREATE TYPE hll_sketch (
CREATE CAST (bytea as hll_sketch) WITHOUT FUNCTION AS ASSIGNMENT;
CREATE CAST (hll_sketch as bytea) WITHOUT FUNCTION AS ASSIGNMENT;
-CREATE OR REPLACE FUNCTION hll_sketch_add_item(internal, anyelement) RETURNS internal
- AS '$libdir/datasketches', 'pg_hll_sketch_add_item'
- LANGUAGE C IMMUTABLE;
+CREATE OR REPLACE FUNCTION hll_sketch_build_agg(internal, anyelement) RETURNS internal
+ AS '$libdir/datasketches', 'pg_hll_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION hll_sketch_add_item(internal, anyelement, int) RETURNS internal
- AS '$libdir/datasketches', 'pg_hll_sketch_add_item'
- LANGUAGE C IMMUTABLE;
+CREATE OR REPLACE FUNCTION hll_sketch_build_agg(internal, anyelement, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_hll_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION hll_sketch_add_item(internal, anyelement, int, int) RETURNS internal
- AS '$libdir/datasketches', 'pg_hll_sketch_add_item'
- LANGUAGE C IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION hll_sketch_get_estimate(hll_sketch) RETURNS double precision
- AS '$libdir/datasketches', 'pg_hll_sketch_get_estimate'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION hll_sketch_get_estimate_and_bounds(hll_sketch) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_hll_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION hll_sketch_get_estimate_and_bounds(hll_sketch, int) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_hll_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION hll_sketch_build_agg(internal, anyelement, int, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_hll_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_from_internal(internal) RETURNS hll_sketch
AS '$libdir/datasketches', 'pg_hll_sketch_from_internal'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_get_estimate_from_internal(internal) RETURNS double precision
AS '$libdir/datasketches', 'pg_hll_sketch_get_estimate_from_internal'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION hll_sketch_to_string(hll_sketch) RETURNS TEXT
- AS '$libdir/datasketches', 'pg_hll_sketch_to_string'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_union_agg(internal, hll_sketch) RETURNS internal
AS '$libdir/datasketches', 'pg_hll_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_union_agg(internal, hll_sketch, int) RETURNS internal
AS '$libdir/datasketches', 'pg_hll_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_union_agg(internal, hll_sketch, int, int) RETURNS internal
AS '$libdir/datasketches', 'pg_hll_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION hll_sketch_combine(internal, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_hll_sketch_combine'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION hll_union_get_result(internal) RETURNS hll_sketch
- AS '$libdir/datasketches', 'pg_hll_union_get_result'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION hll_sketch_serialize_state(internal) RETURNS bytea
+ AS '$libdir/datasketches', 'pg_hll_sketch_serialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION hll_sketch_deserialize_state(bytea, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_hll_sketch_deserialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE AGGREGATE hll_sketch_distinct(anyelement) (
- sfunc = hll_sketch_add_item,
- stype = internal,
- finalfunc = hll_sketch_get_estimate_from_internal
+ STYPE = internal,
+ SFUNC = hll_sketch_build_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_get_estimate_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE hll_sketch_distinct(anyelement, int) (
- sfunc = hll_sketch_add_item,
- stype = internal,
- finalfunc = hll_sketch_get_estimate_from_internal
+ STYPE = internal,
+ SFUNC = hll_sketch_build_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_get_estimate_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE hll_sketch_build(anyelement) (
- sfunc = hll_sketch_add_item,
- stype = internal,
- finalfunc = hll_sketch_from_internal
+ STYPE = internal,
+ SFUNC = hll_sketch_build_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE hll_sketch_build(anyelement, int) (
- sfunc = hll_sketch_add_item,
- stype = internal,
- finalfunc = hll_sketch_from_internal
+ STYPE = internal,
+ SFUNC = hll_sketch_build_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE hll_sketch_build(anyelement, int, int) (
- sfunc = hll_sketch_add_item,
- stype = internal,
- finalfunc = hll_sketch_from_internal
+ STYPE = internal,
+ SFUNC = hll_sketch_build_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE hll_sketch_union(hll_sketch) (
- sfunc = hll_sketch_union_agg,
- stype = internal,
- finalfunc = hll_union_get_result
+ STYPE = internal,
+ SFUNC = hll_sketch_union_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE hll_sketch_union(hll_sketch, int) (
- sfunc = hll_sketch_union_agg,
- stype = internal,
- finalfunc = hll_union_get_result
+ STYPE = internal,
+ SFUNC = hll_sketch_union_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE hll_sketch_union(hll_sketch, int, int) (
- sfunc = hll_sketch_union_agg,
- stype = internal,
- finalfunc = hll_union_get_result
+ STYPE = internal,
+ SFUNC = hll_sketch_union_agg,
+ COMBINEFUNC = hll_sketch_combine,
+ SERIALFUNC = hll_sketch_serialize_state,
+ DESERIALFUNC = hll_sketch_deserialize_state,
+ FINALFUNC = hll_sketch_from_internal,
+ PARALLEL = SAFE
);
+CREATE OR REPLACE FUNCTION hll_sketch_get_estimate(hll_sketch) RETURNS double precision
+ AS '$libdir/datasketches', 'pg_hll_sketch_get_estimate'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION hll_sketch_get_estimate_and_bounds(hll_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_hll_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION hll_sketch_get_estimate_and_bounds(hll_sketch, int) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_hll_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION hll_sketch_to_string(hll_sketch) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_hll_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
CREATE OR REPLACE FUNCTION hll_sketch_union(hll_sketch, hll_sketch) RETURNS hll_sketch
AS '$libdir/datasketches', 'pg_hll_sketch_union'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_union(hll_sketch, hll_sketch, int) RETURNS hll_sketch
AS '$libdir/datasketches', 'pg_hll_sketch_union'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION hll_sketch_union(hll_sketch, hll_sketch, int, int) RETURNS hll_sketch
AS '$libdir/datasketches', 'pg_hll_sketch_union'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
diff --git a/src/hll_sketch_c_adapter.cpp b/src/hll_sketch_c_adapter.cpp
index 7ed28ab..f045054 100644
--- a/src/hll_sketch_c_adapter.cpp
+++ b/src/hll_sketch_c_adapter.cpp
@@ -152,7 +152,10 @@ void hll_union_update(void* unionptr, const void* sketchptr) {
void* hll_union_get_result(void* unionptr) {
try {
- return new (palloc(sizeof(hll_sketch_pg))) hll_sketch_pg(static_cast<hll_union_pg*>(unionptr)->get_result());
+ auto sketchptr = new (palloc(sizeof(hll_sketch_pg))) hll_sketch_pg(static_cast<hll_union_pg*>(unionptr)->get_result());
+ static_cast<hll_union_pg*>(unionptr)->~hll_union_pg();
+ pfree(unionptr);
+ return sketchptr;
} catch (std::exception& e) {
pg_error(e.what());
}
@@ -161,9 +164,12 @@ void* hll_union_get_result(void* unionptr) {
void* hll_union_get_result_tgt_type(void* unionptr, unsigned tgt_type) {
try {
- return new (palloc(sizeof(hll_sketch_pg))) hll_sketch_pg(static_cast<hll_union_pg*>(unionptr)->get_result(
+ auto sketchptr = new (palloc(sizeof(hll_sketch_pg))) hll_sketch_pg(static_cast<hll_union_pg*>(unionptr)->get_result(
tgt_type == 4 ? datasketches::target_hll_type::HLL_4 : tgt_type == 6 ? datasketches::target_hll_type::HLL_6 : datasketches::target_hll_type::HLL_8
));
+ static_cast<hll_union_pg*>(unionptr)->~hll_union_pg();
+ pfree(unionptr);
+ return sketchptr;
} catch (std::exception& e) {
pg_error(e.what());
}
diff --git a/src/hll_sketch_pg_functions.c b/src/hll_sketch_pg_functions.c
index 743cced..8905a8b 100644
--- a/src/hll_sketch_pg_functions.c
+++ b/src/hll_sketch_pg_functions.c
@@ -27,36 +27,45 @@
#include "hll_sketch_c_adapter.h"
#include "base64.h"
+enum hll_agg_state_type { SKETCH, UNION };
+
+struct hll_agg_state {
+ enum hll_agg_state_type type;
+ unsigned lg_k;
+ unsigned tgt_type;
+ void* ptr;
+};
+
const unsigned HLL_DEFAULT_LG_K = 12;
/* PG_FUNCTION_INFO_V1 macro to pass functions to postgres */
-PG_FUNCTION_INFO_V1(pg_hll_sketch_add_item);
-PG_FUNCTION_INFO_V1(pg_hll_sketch_get_estimate);
-PG_FUNCTION_INFO_V1(pg_hll_sketch_get_estimate_and_bounds);
-PG_FUNCTION_INFO_V1(pg_hll_sketch_to_string);
+PG_FUNCTION_INFO_V1(pg_hll_sketch_build_agg);
PG_FUNCTION_INFO_V1(pg_hll_sketch_union_agg);
PG_FUNCTION_INFO_V1(pg_hll_sketch_from_internal);
PG_FUNCTION_INFO_V1(pg_hll_sketch_get_estimate_from_internal);
-PG_FUNCTION_INFO_V1(pg_hll_union_get_result);
+PG_FUNCTION_INFO_V1(pg_hll_sketch_combine);
+PG_FUNCTION_INFO_V1(pg_hll_sketch_serialize_state);
+PG_FUNCTION_INFO_V1(pg_hll_sketch_deserialize_state);
+PG_FUNCTION_INFO_V1(pg_hll_sketch_get_estimate);
+PG_FUNCTION_INFO_V1(pg_hll_sketch_get_estimate_and_bounds);
+PG_FUNCTION_INFO_V1(pg_hll_sketch_to_string);
PG_FUNCTION_INFO_V1(pg_hll_sketch_union);
/* function declarations */
-Datum pg_hll_sketch_recv(PG_FUNCTION_ARGS);
-Datum pg_hll_sketch_send(PG_FUNCTION_ARGS);
-Datum pg_hll_sketch_add_item(PG_FUNCTION_ARGS);
-Datum pg_hll_sketch_get_estimate(PG_FUNCTION_ARGS);
-Datum pg_hll_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
-Datum pg_hll_sketch_to_string(PG_FUNCTION_ARGS);
+Datum pg_hll_sketch_build_agg(PG_FUNCTION_ARGS);
Datum pg_hll_sketch_union_agg(PG_FUNCTION_ARGS);
Datum pg_hll_sketch_from_internal(PG_FUNCTION_ARGS);
Datum pg_hll_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS);
-Datum pg_hll_union_get_result(PG_FUNCTION_ARGS);
+Datum pg_hll_sketch_combine(PG_FUNCTION_ARGS);
+Datum pg_hll_sketch_serialize_state(PG_FUNCTION_ARGS);
+Datum pg_hll_sketch_deserialize_state(PG_FUNCTION_ARGS);
+Datum pg_hll_sketch_get_estimate(PG_FUNCTION_ARGS);
+Datum pg_hll_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
+Datum pg_hll_sketch_to_string(PG_FUNCTION_ARGS);
Datum pg_hll_sketch_union(PG_FUNCTION_ARGS);
-Datum pg_hll_sketch_add_item(PG_FUNCTION_ARGS) {
- void* sketchptr;
- unsigned lg_k;
- unsigned tgt_type;
+Datum pg_hll_sketch_build_agg(PG_FUNCTION_ARGS) {
+ struct hll_agg_state* stateptr;
// anyelement
Oid element_type;
@@ -75,23 +84,25 @@ Datum pg_hll_sketch_add_item(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "hll_sketch_add_item called in non-aggregate context");
+ elog(ERROR, "hll_sketch_build_agg called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : HLL_DEFAULT_LG_K;
- tgt_type = PG_NARGS() > 3 ? PG_GETARG_INT32(3) : 0;
- if (tgt_type) {
- if ((tgt_type != 4) && (tgt_type != 6) && (tgt_type != 8)) {
- elog(ERROR, "hll_sketch_add_item: unsupported target type, must be 4, 6 or 8");
+ stateptr = palloc(sizeof(struct hll_agg_state));
+ stateptr->type = SKETCH;
+ stateptr->lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : HLL_DEFAULT_LG_K;
+ stateptr->tgt_type = PG_NARGS() > 3 ? PG_GETARG_INT32(3) : 0;
+ if (stateptr->tgt_type) {
+ if ((stateptr->tgt_type != 4) && (stateptr->tgt_type != 6) && (stateptr->tgt_type != 8)) {
+ elog(ERROR, "hll_sketch_build_agg: unsupported target type, must be 4, 6 or 8");
}
- sketchptr = hll_sketch_new_tgt_type(lg_k, tgt_type);
+ stateptr->ptr = hll_sketch_new_tgt_type(stateptr->lg_k, stateptr->tgt_type);
} else {
- sketchptr = hll_sketch_new(lg_k);
+ stateptr->ptr = hll_sketch_new(stateptr->lg_k);
}
} else {
- sketchptr = PG_GETARG_POINTER(0);
+ stateptr = (struct hll_agg_state*) PG_GETARG_POINTER(0);
}
element_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
@@ -99,76 +110,24 @@ Datum pg_hll_sketch_add_item(PG_FUNCTION_ARGS) {
get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);
if (typlen == -1) {
// varlena
- hll_sketch_update(sketchptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element));
+ hll_sketch_update(stateptr->ptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element));
} else if (typbyval) {
// fixed-length passed by value
- hll_sketch_update(sketchptr, &element, typlen);
+ hll_sketch_update(stateptr->ptr, &element, typlen);
} else {
// fixed-length passed by reference
- hll_sketch_update(sketchptr, (void*)element, typlen);
+ hll_sketch_update(stateptr->ptr, (void*)element, typlen);
}
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(sketchptr);
-}
-
-Datum pg_hll_sketch_get_estimate(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- double estimate;
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = hll_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- estimate = hll_sketch_get_estimate(sketchptr);
- hll_sketch_delete(sketchptr);
- PG_RETURN_FLOAT8(estimate);
-}
-
-Datum pg_hll_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 = hll_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
- est_and_bounds = (Datum*) hll_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
- hll_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_hll_sketch_to_string(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- char* str;
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = hll_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- str = hll_sketch_to_string(sketchptr);
- hll_sketch_delete(sketchptr);
- PG_RETURN_TEXT_P(cstring_to_text(str));
+ PG_RETURN_POINTER(stateptr);
}
-struct hll_union_state {
- void* unionptr;
- unsigned tgt_type;
-};
-
Datum pg_hll_sketch_union_agg(PG_FUNCTION_ARGS) {
- struct hll_union_state* stateptr;
+ struct hll_agg_state* stateptr;
bytea* sketch_bytes;
void* sketchptr;
- unsigned lg_k;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -185,22 +144,23 @@ Datum pg_hll_sketch_union_agg(PG_FUNCTION_ARGS) {
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : HLL_DEFAULT_LG_K;
- stateptr = palloc(sizeof(struct hll_union_state));
- stateptr->unionptr = hll_union_new(lg_k);
+ stateptr = palloc(sizeof(struct hll_agg_state));
+ stateptr->type = UNION;
+ stateptr->lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : HLL_DEFAULT_LG_K;
stateptr->tgt_type = PG_NARGS() > 3 ? PG_GETARG_INT32(3) : 0;
if (stateptr->tgt_type) {
if ((stateptr->tgt_type != 4) && (stateptr->tgt_type != 6) && (stateptr->tgt_type != 8)) {
elog(ERROR, "hll_sketch_union_agg: unsupported target type, must be 4, 6 or 8");
}
}
+ stateptr->ptr = hll_union_new(stateptr->lg_k);
} else {
- stateptr = (struct hll_union_state*) PG_GETARG_POINTER(0);
+ stateptr = (struct hll_agg_state*) PG_GETARG_POINTER(0);
}
sketch_bytes = PG_GETARG_BYTEA_P(1);
sketchptr = hll_sketch_deserialize(VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
- hll_union_update(stateptr->unionptr, sketchptr);
+ hll_union_update(stateptr->ptr, sketchptr);
hll_sketch_delete(sketchptr);
MemoryContextSwitchTo(oldcontext);
@@ -209,7 +169,7 @@ Datum pg_hll_sketch_union_agg(PG_FUNCTION_ARGS) {
}
Datum pg_hll_sketch_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
+ struct hll_agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -222,9 +182,17 @@ Datum pg_hll_sketch_from_internal(PG_FUNCTION_ARGS) {
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- bytes_out = hll_sketch_serialize(sketchptr, VARHDRSZ);
- hll_sketch_delete(sketchptr);
+ stateptr = (struct hll_agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == UNION) {
+ if (stateptr->tgt_type) {
+ stateptr->ptr = hll_union_get_result_tgt_type(stateptr->ptr, stateptr->tgt_type);
+ } else {
+ stateptr->ptr = hll_union_get_result(stateptr->ptr);
+ }
+ }
+ bytes_out = hll_sketch_serialize(stateptr->ptr, VARHDRSZ);
+ hll_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
@@ -233,7 +201,7 @@ Datum pg_hll_sketch_from_internal(PG_FUNCTION_ARGS) {
}
Datum pg_hll_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
+ struct hll_agg_state* stateptr;
double estimate;
MemoryContext oldcontext;
@@ -242,22 +210,81 @@ Datum pg_hll_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "hll_sketch_from_internal called in non-aggregate context");
+ elog(ERROR, "hll_sketch_get_estimate_from_internal called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- estimate = hll_sketch_get_estimate(sketchptr);
- hll_sketch_delete(sketchptr);
+ stateptr = (struct hll_agg_state*) PG_GETARG_POINTER(0);
+ estimate = hll_sketch_get_estimate(stateptr->ptr);
+ hll_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
MemoryContextSwitchTo(oldcontext);
PG_RETURN_FLOAT8(estimate);
}
+Datum pg_hll_sketch_combine(PG_FUNCTION_ARGS) {
+ struct hll_agg_state* stateptr1;
+ struct hll_agg_state* stateptr2;
+ struct hll_agg_state* stateptr;
-Datum pg_hll_union_get_result(PG_FUNCTION_ARGS) {
- struct hll_union_state* stateptr;
- void* sketchptr;
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0) && PG_ARGISNULL(1)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "hll_sketch_from_internal called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ stateptr1 = (struct hll_agg_state*) PG_GETARG_POINTER(0);
+ stateptr2 = (struct hll_agg_state*) PG_GETARG_POINTER(1);
+
+ stateptr = palloc(sizeof(struct hll_agg_state));
+ stateptr->type = SKETCH;
+ stateptr->lg_k = stateptr1 ? stateptr1->lg_k : stateptr2->lg_k;
+ stateptr->tgt_type = stateptr1 ? stateptr1->tgt_type : stateptr2->tgt_type;
+ stateptr->ptr = hll_union_new(stateptr->lg_k);
+
+ if (stateptr1) {
+ if (stateptr1->type == UNION) {
+ if (stateptr1->tgt_type) {
+ stateptr1->ptr = hll_union_get_result_tgt_type(stateptr1->ptr, stateptr1->tgt_type);
+ } else {
+ stateptr1->ptr = hll_union_get_result(stateptr1->ptr);
+ }
+ }
+ hll_union_update(stateptr->ptr, stateptr1->ptr);
+ hll_sketch_delete(stateptr1->ptr);
+ pfree(stateptr1);
+ }
+ if (stateptr2) {
+ if (stateptr2->type == UNION) {
+ if (stateptr2->tgt_type) {
+ stateptr2->ptr = hll_union_get_result_tgt_type(stateptr2->ptr, stateptr2->tgt_type);
+ } else {
+ stateptr2->ptr = hll_union_get_result(stateptr2->ptr);
+ }
+ }
+ hll_union_update(stateptr->ptr, stateptr2->ptr);
+ hll_sketch_delete(stateptr2->ptr);
+ pfree(stateptr2);
+ }
+ if (stateptr->tgt_type) {
+ stateptr->ptr = hll_union_get_result_tgt_type(stateptr->ptr, stateptr->tgt_type);
+ } else {
+ stateptr->ptr = hll_union_get_result(stateptr->ptr);
+ }
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(stateptr);
+}
+
+
+Datum pg_hll_sketch_serialize_state(PG_FUNCTION_ARGS) {
+ struct hll_agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -266,19 +293,22 @@ Datum pg_hll_union_get_result(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "hll_union_get_result called in non-aggregate context");
+ elog(ERROR, "hll_sketch_serialize_state called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- stateptr = (struct hll_union_state*) PG_GETARG_POINTER(0);
- if (stateptr->tgt_type) {
- sketchptr = hll_union_get_result_tgt_type(stateptr->unionptr, stateptr->tgt_type);
- } else {
- sketchptr = hll_union_get_result(stateptr->unionptr);
+ stateptr = (struct hll_agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == UNION) {
+ if (stateptr->tgt_type) {
+ stateptr->ptr = hll_union_get_result_tgt_type(stateptr->ptr, stateptr->tgt_type);
+ } else {
+ stateptr->ptr = hll_union_get_result(stateptr->ptr);
+ }
}
- bytes_out = hll_sketch_serialize(sketchptr, VARHDRSZ);
- hll_sketch_delete(sketchptr);
- hll_union_delete(stateptr->unionptr);
+ bytes_out = hll_sketch_serialize(stateptr->ptr, VARHDRSZ + 2);
+ ((char*)bytes_out.ptr)[VARHDRSZ] = stateptr->lg_k;
+ ((char*)bytes_out.ptr)[VARHDRSZ + 1] = stateptr->tgt_type;
+ hll_sketch_delete(stateptr->ptr);
pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
@@ -287,6 +317,77 @@ Datum pg_hll_union_get_result(PG_FUNCTION_ARGS) {
PG_RETURN_BYTEA_P(bytes_out.ptr);
}
+Datum pg_hll_sketch_deserialize_state(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ struct hll_agg_state* stateptr;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "hll_sketch_deserialize_state called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ stateptr = palloc(sizeof(struct hll_agg_state));
+ stateptr->lg_k = *VARDATA(bytes_in);
+ stateptr->tgt_type = *(VARDATA(bytes_in) + 1);
+ stateptr->ptr = hll_sketch_deserialize(VARDATA(bytes_in) + 2, VARSIZE(bytes_in) - VARHDRSZ - 2);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(stateptr);
+}
+
+Datum pg_hll_sketch_get_estimate(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ double estimate;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = hll_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ estimate = hll_sketch_get_estimate(sketchptr);
+ hll_sketch_delete(sketchptr);
+ PG_RETURN_FLOAT8(estimate);
+}
+
+Datum pg_hll_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 = hll_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
+ est_and_bounds = (Datum*) hll_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+ hll_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_hll_sketch_to_string(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ char* str;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = hll_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ str = hll_sketch_to_string(sketchptr);
+ hll_sketch_delete(sketchptr);
+ PG_RETURN_TEXT_P(cstring_to_text(str));
+}
+
Datum pg_hll_sketch_union(PG_FUNCTION_ARGS) {
const bytea* bytes_in1;
const bytea* bytes_in2;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org