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/19 03:24:59 UTC
[datasketches-postgresql] 02/02: CPC 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 360aa174ccf2043c25815bfa1243603d7e000520
Author: AlexanderSaydakov <Al...@users.noreply.github.com>
AuthorDate: Tue Apr 18 20:24:44 2023 -0700
CPC sketch partial aggregation support
---
sql/datasketches_cpc_sketch.sql | 134 ++++++++++++--------
src/cpc_sketch_c_adapter.cpp | 9 +-
src/cpc_sketch_pg_functions.c | 272 ++++++++++++++++++++++++++--------------
3 files changed, 266 insertions(+), 149 deletions(-)
diff --git a/sql/datasketches_cpc_sketch.sql b/sql/datasketches_cpc_sketch.sql
index 4c293e3..e506e34 100644
--- a/sql/datasketches_cpc_sketch.sql
+++ b/sql/datasketches_cpc_sketch.sql
@@ -19,11 +19,11 @@ CREATE TYPE cpc_sketch;
CREATE OR REPLACE FUNCTION cpc_sketch_in(cstring) RETURNS cpc_sketch
AS '$libdir/datasketches', 'pg_sketch_in'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION cpc_sketch_out(cpc_sketch) RETURNS cstring
AS '$libdir/datasketches', 'pg_sketch_out'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE TYPE cpc_sketch (
INPUT = cpc_sketch_in,
@@ -34,90 +34,122 @@ CREATE TYPE cpc_sketch (
CREATE CAST (bytea as cpc_sketch) WITHOUT FUNCTION AS ASSIGNMENT;
CREATE CAST (cpc_sketch as bytea) WITHOUT FUNCTION AS ASSIGNMENT;
-CREATE OR REPLACE FUNCTION cpc_sketch_add_item(internal, anyelement) RETURNS internal
- AS '$libdir/datasketches', 'pg_cpc_sketch_add_item'
- LANGUAGE C IMMUTABLE;
+CREATE OR REPLACE FUNCTION cpc_sketch_build_agg(internal, anyelement) RETURNS internal
+ AS '$libdir/datasketches', 'pg_cpc_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION cpc_sketch_add_item(internal, anyelement, int) RETURNS internal
- AS '$libdir/datasketches', 'pg_cpc_sketch_add_item'
- LANGUAGE C IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate(cpc_sketch) RETURNS double precision
- AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate_and_bounds(cpc_sketch) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate_and_bounds(cpc_sketch, int) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION cpc_sketch_build_agg(internal, anyelement, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_cpc_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION cpc_sketch_from_internal(internal) RETURNS cpc_sketch
AS '$libdir/datasketches', 'pg_cpc_sketch_from_internal'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate_from_internal(internal) RETURNS double precision
AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate_from_internal'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION cpc_sketch_to_string(cpc_sketch) RETURNS TEXT
- AS '$libdir/datasketches', 'pg_cpc_sketch_to_string'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION cpc_sketch_union_agg(internal, cpc_sketch) RETURNS internal
AS '$libdir/datasketches', 'pg_cpc_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION cpc_sketch_union_agg(internal, cpc_sketch, int) RETURNS internal
AS '$libdir/datasketches', 'pg_cpc_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION cpc_sketch_combine(internal, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_cpc_sketch_combine'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION cpc_union_get_result(internal) RETURNS cpc_sketch
- AS '$libdir/datasketches', 'pg_cpc_union_get_result'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION cpc_sketch_serialize_state(internal) RETURNS bytea
+ AS '$libdir/datasketches', 'pg_cpc_sketch_serialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION cpc_sketch_deserialize_state(bytea, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_cpc_sketch_deserialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE AGGREGATE cpc_sketch_distinct(anyelement) (
- sfunc = cpc_sketch_add_item,
- stype = internal,
- finalfunc = cpc_sketch_get_estimate_from_internal
+ STYPE = internal,
+ SFUNC = cpc_sketch_build_agg,
+ COMBINEFUNC = cpc_sketch_combine,
+ SERIALFUNC = cpc_sketch_serialize_state,
+ DESERIALFUNC = cpc_sketch_deserialize_state,
+ FINALFUNC = cpc_sketch_get_estimate_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE cpc_sketch_distinct(anyelement, int) (
- sfunc = cpc_sketch_add_item,
- stype = internal,
- finalfunc = cpc_sketch_get_estimate_from_internal
+ STYPE = internal,
+ SFUNC = cpc_sketch_build_agg,
+ COMBINEFUNC = cpc_sketch_combine,
+ SERIALFUNC = cpc_sketch_serialize_state,
+ DESERIALFUNC = cpc_sketch_deserialize_state,
+ FINALFUNC = cpc_sketch_get_estimate_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE cpc_sketch_build(anyelement) (
- sfunc = cpc_sketch_add_item,
- stype = internal,
- finalfunc = cpc_sketch_from_internal
+ STYPE = internal,
+ SFUNC = cpc_sketch_build_agg,
+ COMBINEFUNC = cpc_sketch_combine,
+ SERIALFUNC = cpc_sketch_serialize_state,
+ DESERIALFUNC = cpc_sketch_deserialize_state,
+ FINALFUNC = cpc_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE cpc_sketch_build(anyelement, int) (
- sfunc = cpc_sketch_add_item,
- stype = internal,
- finalfunc = cpc_sketch_from_internal
+ STYPE = internal,
+ SFUNC = cpc_sketch_build_agg,
+ COMBINEFUNC = cpc_sketch_combine,
+ SERIALFUNC = cpc_sketch_serialize_state,
+ DESERIALFUNC = cpc_sketch_deserialize_state,
+ FINALFUNC = cpc_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE cpc_sketch_union(cpc_sketch) (
- sfunc = cpc_sketch_union_agg,
- stype = internal,
- finalfunc = cpc_union_get_result
+ STYPE = internal,
+ SFUNC = cpc_sketch_union_agg,
+ COMBINEFUNC = cpc_sketch_combine,
+ SERIALFUNC = cpc_sketch_serialize_state,
+ DESERIALFUNC = cpc_sketch_deserialize_state,
+ FINALFUNC = cpc_sketch_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE cpc_sketch_union(cpc_sketch, int) (
- sfunc = cpc_sketch_union_agg,
- stype = internal,
- finalfunc = cpc_union_get_result
+ STYPE = internal,
+ SFUNC = cpc_sketch_union_agg,
+ COMBINEFUNC = cpc_sketch_combine,
+ SERIALFUNC = cpc_sketch_serialize_state,
+ DESERIALFUNC = cpc_sketch_deserialize_state,
+ FINALFUNC = cpc_sketch_from_internal,
+ PARALLEL = SAFE
);
+CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate(cpc_sketch) RETURNS double precision
+ AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate_and_bounds(cpc_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate_and_bounds(cpc_sketch, int) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION cpc_sketch_to_string(cpc_sketch) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_cpc_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
CREATE OR REPLACE FUNCTION cpc_sketch_union(cpc_sketch, cpc_sketch) RETURNS cpc_sketch
AS '$libdir/datasketches', 'pg_cpc_sketch_union'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION cpc_sketch_union(cpc_sketch, cpc_sketch, int) RETURNS cpc_sketch
AS '$libdir/datasketches', 'pg_cpc_sketch_union'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
diff --git a/src/cpc_sketch_c_adapter.cpp b/src/cpc_sketch_c_adapter.cpp
index df1ae3f..00b59cb 100644
--- a/src/cpc_sketch_c_adapter.cpp
+++ b/src/cpc_sketch_c_adapter.cpp
@@ -24,8 +24,8 @@
#include <cpc_sketch.hpp>
#include <cpc_union.hpp>
-typedef datasketches::cpc_sketch_alloc<palloc_allocator<char>> cpc_sketch_pg;
-typedef datasketches::cpc_union_alloc<palloc_allocator<char>> cpc_union_pg;
+using cpc_sketch_pg = datasketches::cpc_sketch_alloc<palloc_allocator<char>>;
+using cpc_union_pg = datasketches::cpc_union_alloc<palloc_allocator<char>>;
void cpc_init() {
datasketches::cpc_init<palloc_allocator<char>>();
@@ -147,7 +147,10 @@ void cpc_union_update(void* unionptr, const void* sketchptr) {
void* cpc_union_get_result(void* unionptr) {
try {
- return new (palloc(sizeof(cpc_sketch_pg))) cpc_sketch_pg(static_cast<cpc_union_pg*>(unionptr)->get_result());
+ auto sketchptr = new (palloc(sizeof(cpc_sketch_pg))) cpc_sketch_pg(static_cast<cpc_union_pg*>(unionptr)->get_result());
+ static_cast<cpc_union_pg*>(unionptr)->~cpc_union_pg();
+ pfree(unionptr);
+ return sketchptr;
} catch (std::exception& e) {
pg_error(e.what());
}
diff --git a/src/cpc_sketch_pg_functions.c b/src/cpc_sketch_pg_functions.c
index 2d0d626..6a6082d 100644
--- a/src/cpc_sketch_pg_functions.c
+++ b/src/cpc_sketch_pg_functions.c
@@ -26,36 +26,38 @@
#include "cpc_sketch_c_adapter.h"
#include "base64.h"
+#include "agg_state.h"
const unsigned CPC_DEFAULT_LG_K = 11;
/* PG_FUNCTION_INFO_V1 macro to pass functions to postgres */
-PG_FUNCTION_INFO_V1(pg_cpc_sketch_add_item);
-PG_FUNCTION_INFO_V1(pg_cpc_sketch_get_estimate);
-PG_FUNCTION_INFO_V1(pg_cpc_sketch_get_estimate_and_bounds);
-PG_FUNCTION_INFO_V1(pg_cpc_sketch_to_string);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_build_agg);
PG_FUNCTION_INFO_V1(pg_cpc_sketch_union_agg);
PG_FUNCTION_INFO_V1(pg_cpc_sketch_from_internal);
PG_FUNCTION_INFO_V1(pg_cpc_sketch_get_estimate_from_internal);
-PG_FUNCTION_INFO_V1(pg_cpc_union_get_result);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_combine);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_serialize_state);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_deserialize_state);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_get_estimate);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_get_estimate_and_bounds);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_to_string);
PG_FUNCTION_INFO_V1(pg_cpc_sketch_union);
/* function declarations */
-Datum pg_cpc_sketch_recv(PG_FUNCTION_ARGS);
-Datum pg_cpc_sketch_send(PG_FUNCTION_ARGS);
-Datum pg_cpc_sketch_add_item(PG_FUNCTION_ARGS);
-Datum pg_cpc_sketch_get_estimate(PG_FUNCTION_ARGS);
-Datum pg_cpc_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
-Datum pg_cpc_sketch_to_string(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_build_agg(PG_FUNCTION_ARGS);
Datum pg_cpc_sketch_union_agg(PG_FUNCTION_ARGS);
Datum pg_cpc_sketch_from_internal(PG_FUNCTION_ARGS);
Datum pg_cpc_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS);
-Datum pg_cpc_union_get_result(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_combine(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_serialize_state(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_deserialize_state(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_get_estimate(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_to_string(PG_FUNCTION_ARGS);
Datum pg_cpc_sketch_union(PG_FUNCTION_ARGS);
-Datum pg_cpc_sketch_add_item(PG_FUNCTION_ARGS) {
- void* sketchptr;
- int lg_k;
+Datum pg_cpc_sketch_build_agg(PG_FUNCTION_ARGS) {
+ struct agg_state* stateptr;
// anyelement
Oid element_type;
@@ -74,15 +76,17 @@ Datum pg_cpc_sketch_add_item(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "cpc_sketch_add_item called in non-aggregate context");
+ elog(ERROR, "cpc_sketch_build_agg called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : CPC_DEFAULT_LG_K;
- sketchptr = cpc_sketch_new(lg_k);
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = MUTABLE_SKETCH;
+ stateptr->lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : CPC_DEFAULT_LG_K;
+ stateptr->ptr = cpc_sketch_new(stateptr->lg_k);
} else {
- sketchptr = PG_GETARG_POINTER(0);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
}
element_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
@@ -90,71 +94,24 @@ Datum pg_cpc_sketch_add_item(PG_FUNCTION_ARGS) {
get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);
if (typlen == -1) {
// varlena
- cpc_sketch_update(sketchptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element));
+ cpc_sketch_update(stateptr->ptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element));
} else if (typbyval) {
// fixed-length passed by value
- cpc_sketch_update(sketchptr, &element, typlen);
+ cpc_sketch_update(stateptr->ptr, &element, typlen);
} else {
// fixed-length passed by reference
- cpc_sketch_update(sketchptr, (void*)element, typlen);
+ cpc_sketch_update(stateptr->ptr, (void*)element, typlen);
}
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(sketchptr);
-}
-
-Datum pg_cpc_sketch_get_estimate(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- double estimate;
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = cpc_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- estimate = cpc_sketch_get_estimate(sketchptr);
- cpc_sketch_delete(sketchptr);
- PG_RETURN_FLOAT8(estimate);
-}
-
-Datum pg_cpc_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 = cpc_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
- est_and_bounds = (Datum*) cpc_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
- cpc_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_cpc_sketch_to_string(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- char* str;
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = cpc_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- str = cpc_sketch_to_string(sketchptr);
- cpc_sketch_delete(sketchptr);
- PG_RETURN_TEXT_P(cstring_to_text(str));
+ PG_RETURN_POINTER(stateptr);
}
Datum pg_cpc_sketch_union_agg(PG_FUNCTION_ARGS) {
- void* unionptr;
+ struct agg_state* stateptr;
bytea* sketch_bytes;
void* sketchptr;
- int lg_k;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -171,24 +128,26 @@ Datum pg_cpc_sketch_union_agg(PG_FUNCTION_ARGS) {
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : CPC_DEFAULT_LG_K;
- unionptr = cpc_union_new(lg_k);
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = UNION;
+ stateptr->lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : CPC_DEFAULT_LG_K;
+ stateptr->ptr = cpc_union_new(stateptr->lg_k);
} else {
- unionptr = PG_GETARG_POINTER(0);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
}
sketch_bytes = PG_GETARG_BYTEA_P(1);
sketchptr = cpc_sketch_deserialize(VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
- cpc_union_update(unionptr, sketchptr);
+ cpc_union_update(stateptr->ptr, sketchptr);
cpc_sketch_delete(sketchptr);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(unionptr);
+ PG_RETURN_POINTER(stateptr);
}
Datum pg_cpc_sketch_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
+ struct agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -201,9 +160,13 @@ Datum pg_cpc_sketch_from_internal(PG_FUNCTION_ARGS) {
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- bytes_out = cpc_sketch_serialize(sketchptr, VARHDRSZ);
- cpc_sketch_delete(sketchptr);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == UNION) {
+ stateptr->ptr = cpc_union_get_result(stateptr->ptr);
+ }
+ bytes_out = cpc_sketch_serialize(stateptr->ptr, VARHDRSZ);
+ cpc_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
@@ -212,7 +175,7 @@ Datum pg_cpc_sketch_from_internal(PG_FUNCTION_ARGS) {
}
Datum pg_cpc_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
+ struct agg_state* stateptr;
double estimate;
MemoryContext oldcontext;
@@ -221,22 +184,68 @@ Datum pg_cpc_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "cpc_sketch_from_internal called in non-aggregate context");
+ elog(ERROR, "cpc_sketch_get_estimate_from_internal called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- estimate = cpc_sketch_get_estimate(sketchptr);
- cpc_sketch_delete(sketchptr);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+ estimate = cpc_sketch_get_estimate(stateptr->ptr);
+ cpc_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
MemoryContextSwitchTo(oldcontext);
PG_RETURN_FLOAT8(estimate);
}
-Datum pg_cpc_union_get_result(PG_FUNCTION_ARGS) {
- void* unionptr;
- void* sketchptr;
+Datum pg_cpc_sketch_combine(PG_FUNCTION_ARGS) {
+ struct agg_state* stateptr1;
+ struct agg_state* stateptr2;
+ struct agg_state* stateptr;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0) && PG_ARGISNULL(1)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "cpc_sketch_from_combine called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ stateptr1 = (struct agg_state*) PG_GETARG_POINTER(0);
+ stateptr2 = (struct agg_state*) PG_GETARG_POINTER(1);
+
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = MUTABLE_SKETCH;
+ stateptr->lg_k = stateptr1 ? stateptr1->lg_k : stateptr2->lg_k;
+ stateptr->ptr = cpc_union_new(stateptr->lg_k);
+
+ if (stateptr1) {
+ if (stateptr1->type == UNION) {
+ stateptr1->ptr = cpc_union_get_result(stateptr1->ptr);
+ }
+ cpc_union_update(stateptr->ptr, stateptr1->ptr);
+ cpc_sketch_delete(stateptr1->ptr);
+ pfree(stateptr1);
+ }
+ if (stateptr2) {
+ if (stateptr2->type == UNION) {
+ stateptr2->ptr = cpc_union_get_result(stateptr2->ptr);
+ }
+ cpc_union_update(stateptr->ptr, stateptr2->ptr);
+ cpc_sketch_delete(stateptr2->ptr);
+ pfree(stateptr2);
+ }
+ stateptr->ptr = cpc_union_get_result(stateptr->ptr);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(stateptr);
+}
+
+Datum pg_cpc_sketch_serialize_state(PG_FUNCTION_ARGS) {
+ struct agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -245,15 +254,18 @@ Datum pg_cpc_union_get_result(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "cpc_union_get_result called in non-aggregate context");
+ elog(ERROR, "cpc_sketch_serialize_state called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- unionptr = PG_GETARG_POINTER(0);
- sketchptr = cpc_union_get_result(unionptr);
- bytes_out = cpc_sketch_serialize(sketchptr, VARHDRSZ);
- cpc_sketch_delete(sketchptr);
- cpc_union_delete(unionptr);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == UNION) {
+ stateptr->ptr = cpc_union_get_result(stateptr->ptr);
+ }
+ bytes_out = cpc_sketch_serialize(stateptr->ptr, VARHDRSZ + 1);
+ ((char*)bytes_out.ptr)[VARHDRSZ] = stateptr->lg_k;
+ cpc_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
@@ -261,6 +273,77 @@ Datum pg_cpc_union_get_result(PG_FUNCTION_ARGS) {
PG_RETURN_BYTEA_P(bytes_out.ptr);
}
+Datum pg_cpc_sketch_deserialize_state(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ struct agg_state* stateptr;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "cpc_sketch_deserialize_state called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = MUTABLE_SKETCH;
+ stateptr->lg_k = *VARDATA(bytes_in);
+ stateptr->ptr = cpc_sketch_deserialize(VARDATA(bytes_in) + 1, VARSIZE(bytes_in) - VARHDRSZ - 1);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(stateptr);
+}
+
+Datum pg_cpc_sketch_get_estimate(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ double estimate;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = cpc_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ estimate = cpc_sketch_get_estimate(sketchptr);
+ cpc_sketch_delete(sketchptr);
+ PG_RETURN_FLOAT8(estimate);
+}
+
+Datum pg_cpc_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 = cpc_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
+ est_and_bounds = (Datum*) cpc_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+ cpc_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_cpc_sketch_to_string(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ char* str;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = cpc_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ str = cpc_sketch_to_string(sketchptr);
+ cpc_sketch_delete(sketchptr);
+ PG_RETURN_TEXT_P(cstring_to_text(str));
+}
+
Datum pg_cpc_sketch_union(PG_FUNCTION_ARGS) {
const bytea* bytes_in1;
const bytea* bytes_in2;
@@ -286,7 +369,6 @@ Datum pg_cpc_sketch_union(PG_FUNCTION_ARGS) {
cpc_sketch_delete(sketchptr2);
}
sketchptr = cpc_union_get_result(unionptr);
- cpc_union_delete(unionptr);
bytes_out = cpc_sketch_serialize(sketchptr, VARHDRSZ);
cpc_sketch_delete(sketchptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org