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/17 18:29:30 UTC
[datasketches-postgresql] 01/01: theta 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 eeef3a923e243a5ee33b2cf700197b43b7ac4e5f
Author: AlexanderSaydakov <Al...@users.noreply.github.com>
AuthorDate: Mon Apr 17 11:29:11 2023 -0700
theta sketch partial aggregation support
---
sql/datasketches_theta_sketch.sql | 200 +++++++++++--------
src/agg_state.h | 31 +++
src/theta_sketch_c_adapter.cpp | 34 +++-
src/theta_sketch_c_adapter.h | 10 +-
src/theta_sketch_pg_functions.c | 394 +++++++++++++++++++++++++-------------
5 files changed, 452 insertions(+), 217 deletions(-)
diff --git a/sql/datasketches_theta_sketch.sql b/sql/datasketches_theta_sketch.sql
index dfd4250..d12a7d2 100644
--- a/sql/datasketches_theta_sketch.sql
+++ b/sql/datasketches_theta_sketch.sql
@@ -18,12 +18,12 @@
CREATE TYPE theta_sketch;
CREATE OR REPLACE FUNCTION theta_sketch_in(cstring) RETURNS theta_sketch
- AS '$libdir/datasketches', 'pg_sketch_in'
- LANGUAGE C STRICT IMMUTABLE;
+ AS '$libdir/datasketches', 'pg_sketch_in'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_out(theta_sketch) RETURNS cstring
- AS '$libdir/datasketches', 'pg_sketch_out'
- LANGUAGE C STRICT IMMUTABLE;
+ AS '$libdir/datasketches', 'pg_sketch_out'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE TYPE theta_sketch (
INPUT = theta_sketch_in,
@@ -34,122 +34,166 @@ CREATE TYPE theta_sketch (
CREATE CAST (bytea as theta_sketch) WITHOUT FUNCTION AS ASSIGNMENT;
CREATE CAST (theta_sketch as bytea) WITHOUT FUNCTION AS ASSIGNMENT;
-CREATE OR REPLACE FUNCTION theta_sketch_add_item(internal, anyelement) RETURNS internal
- AS '$libdir/datasketches', 'pg_theta_sketch_add_item'
- LANGUAGE C IMMUTABLE;
+CREATE OR REPLACE FUNCTION theta_sketch_build_agg(internal, anyelement) RETURNS internal
+ AS '$libdir/datasketches', 'pg_theta_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION theta_sketch_add_item(internal, anyelement, int) RETURNS internal
- AS '$libdir/datasketches', 'pg_theta_sketch_add_item'
- LANGUAGE C IMMUTABLE;
+CREATE OR REPLACE FUNCTION theta_sketch_build_agg(internal, anyelement, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_theta_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION theta_sketch_add_item(internal, anyelement, int, real) RETURNS internal
- AS '$libdir/datasketches', 'pg_theta_sketch_add_item'
- LANGUAGE C IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION theta_sketch_get_estimate(theta_sketch) RETURNS double precision
- AS '$libdir/datasketches', 'pg_theta_sketch_get_estimate'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION theta_sketch_get_estimate_and_bounds(theta_sketch) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_theta_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION theta_sketch_get_estimate_and_bounds(theta_sketch, int) RETURNS double precision[]
- AS '$libdir/datasketches', 'pg_theta_sketch_get_estimate_and_bounds'
- LANGUAGE C STRICT IMMUTABLE;
+CREATE OR REPLACE FUNCTION theta_sketch_build_agg(internal, anyelement, int, real) RETURNS internal
+ AS '$libdir/datasketches', 'pg_theta_sketch_build_agg'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_from_internal(internal) RETURNS theta_sketch
AS '$libdir/datasketches', 'pg_theta_sketch_from_internal'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_get_estimate_from_internal(internal) RETURNS double precision
AS '$libdir/datasketches', 'pg_theta_sketch_get_estimate_from_internal'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION theta_sketch_to_string(theta_sketch) RETURNS TEXT
- AS '$libdir/datasketches', 'pg_theta_sketch_to_string'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_union_agg(internal, theta_sketch) RETURNS internal
AS '$libdir/datasketches', 'pg_theta_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_union_agg(internal, theta_sketch, int) RETURNS internal
AS '$libdir/datasketches', 'pg_theta_sketch_union_agg'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_intersection_agg(internal, theta_sketch) RETURNS internal
AS '$libdir/datasketches', 'pg_theta_sketch_intersection_agg'
- LANGUAGE C IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION theta_union_get_result(internal) RETURNS theta_sketch
- AS '$libdir/datasketches', 'pg_theta_union_get_result'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE OR REPLACE FUNCTION theta_intersection_get_result(internal) RETURNS theta_sketch
- AS '$libdir/datasketches', 'pg_theta_intersection_get_result'
- LANGUAGE C STRICT IMMUTABLE;
-
-CREATE AGGREGATE theta_sketch_distinct(anyelement) (
- sfunc = theta_sketch_add_item,
- stype = internal,
- finalfunc = theta_sketch_get_estimate_from_internal
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
+
+-- CREATE OR REPLACE FUNCTION theta_union_get_result(internal) RETURNS theta_sketch
+-- AS '$libdir/datasketches', 'pg_theta_union_get_result'
+-- LANGUAGE C STRICT IMMUTABLE;
+
+-- CREATE OR REPLACE FUNCTION theta_intersection_get_result(internal) RETURNS theta_sketch
+-- AS '$libdir/datasketches', 'pg_theta_intersection_get_result'
+-- LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION theta_sketch_combine(internal, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_theta_sketch_combine'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION theta_sketch_serialize_state(internal) RETURNS bytea
+ AS '$libdir/datasketches', 'pg_theta_sketch_serialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION theta_sketch_deserialize_state(bytea, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_theta_sketch_deserialize_state'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE AGGREGATE theta_sketch_distinct(anyelement) (
+ STYPE = internal,
+ SFUNC = theta_sketch_build_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_get_estimate_from_internal,
+ PARALLEL = SAFE
);
-CREATE AGGREGATE theta_sketch_distinct(anyelement, int) (
- sfunc = theta_sketch_add_item,
- stype = internal,
- finalfunc = theta_sketch_get_estimate_from_internal
+CREATE OR REPLACE AGGREGATE theta_sketch_distinct(anyelement, int) (
+ STYPE = internal,
+ SFUNC = theta_sketch_build_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_get_estimate_from_internal,
+ PARALLEL = SAFE
);
CREATE AGGREGATE theta_sketch_build(anyelement) (
- sfunc = theta_sketch_add_item,
- stype = internal,
- finalfunc = theta_sketch_from_internal
+ STYPE = internal,
+ SFUNC = theta_sketch_build_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_from_internal,
+ PARALLEL = SAFE
);
-CREATE AGGREGATE theta_sketch_build(anyelement, int) (
- sfunc = theta_sketch_add_item,
- stype = internal,
- finalfunc = theta_sketch_from_internal
+CREATE OR REPLACE AGGREGATE theta_sketch_build(anyelement, int) (
+ STYPE = internal,
+ SFUNC = theta_sketch_build_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_from_internal,
+ PARALLEL = SAFE
);
-CREATE AGGREGATE theta_sketch_build(anyelement, int, real) (
- sfunc = theta_sketch_add_item,
- stype = internal,
- finalfunc = theta_sketch_from_internal
+CREATE OR REPLACE AGGREGATE theta_sketch_build(anyelement, int, real) (
+ STYPE = internal,
+ SFUNC = theta_sketch_build_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_from_internal,
+ PARALLEL = SAFE
);
-CREATE AGGREGATE theta_sketch_union(theta_sketch) (
- sfunc = theta_sketch_union_agg,
- stype = internal,
- finalfunc = theta_union_get_result
+CREATE OR REPLACE AGGREGATE theta_sketch_union(theta_sketch) (
+ STYPE = internal,
+ SFUNC = theta_sketch_union_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_from_internal,
+ PARALLEL = SAFE
);
-CREATE AGGREGATE theta_sketch_union(theta_sketch, int) (
- sfunc = theta_sketch_union_agg,
- stype = internal,
- finalfunc = theta_union_get_result
+CREATE OR REPLACE AGGREGATE theta_sketch_union(theta_sketch, int) (
+ STYPE = internal,
+ SFUNC = theta_sketch_union_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_from_internal,
+ PARALLEL = SAFE
);
-CREATE AGGREGATE theta_sketch_intersection(theta_sketch) (
- sfunc = theta_sketch_intersection_agg,
- stype = internal,
- finalfunc = theta_intersection_get_result
+CREATE OR REPLACE AGGREGATE theta_sketch_intersection(theta_sketch) (
+ STYPE = internal,
+ SFUNC = theta_sketch_intersection_agg,
+ COMBINEFUNC = theta_sketch_combine,
+ SERIALFUNC = theta_sketch_serialize_state,
+ DESERIALFUNC = theta_sketch_deserialize_state,
+ FINALFUNC = theta_sketch_from_internal,
+ PARALLEL = SAFE
);
+CREATE OR REPLACE FUNCTION theta_sketch_get_estimate(theta_sketch) RETURNS double precision
+ AS '$libdir/datasketches', 'pg_theta_sketch_get_estimate'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION theta_sketch_get_estimate_and_bounds(theta_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_theta_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION theta_sketch_get_estimate_and_bounds(theta_sketch, int) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_theta_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION theta_sketch_to_string(theta_sketch) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_theta_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
+
CREATE OR REPLACE FUNCTION theta_sketch_union(theta_sketch, theta_sketch) RETURNS theta_sketch
AS '$libdir/datasketches', 'pg_theta_sketch_union'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_union(theta_sketch, theta_sketch, int) RETURNS theta_sketch
AS '$libdir/datasketches', 'pg_theta_sketch_union'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_intersection(theta_sketch, theta_sketch) RETURNS theta_sketch
AS '$libdir/datasketches', 'pg_theta_sketch_intersection'
- LANGUAGE C IMMUTABLE;
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_a_not_b(theta_sketch, theta_sketch) RETURNS theta_sketch
AS '$libdir/datasketches', 'pg_theta_sketch_a_not_b'
- LANGUAGE C STRICT IMMUTABLE;
+ LANGUAGE C STRICT IMMUTABLE PARALLEL SAFE;
diff --git a/src/agg_state.h b/src/agg_state.h
new file mode 100644
index 0000000..5a4a1e9
--- /dev/null
+++ b/src/agg_state.h
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#ifndef AGG_STATE_H
+#define AGG_STATE_H
+
+enum agg_state_type { MUTABLE_SKETCH, IMMUTABLE_SKETCH, UNION, INTERSECTION };
+
+struct agg_state {
+ enum agg_state_type type;
+ unsigned lg_k;
+ void* ptr;
+};
+
+#endif
diff --git a/src/theta_sketch_c_adapter.cpp b/src/theta_sketch_c_adapter.cpp
index dd635e9..0bcf966 100644
--- a/src/theta_sketch_c_adapter.cpp
+++ b/src/theta_sketch_c_adapter.cpp
@@ -176,7 +176,15 @@ void theta_union_delete(void* unionptr) {
}
}
-void theta_union_update(void* unionptr, const void* buffer, unsigned length) {
+void theta_union_update_with_sketch(void* unionptr, const void* sketchptr) {
+ try {
+ static_cast<theta_union_pg*>(unionptr)->update(*static_cast<const theta_sketch_pg*>(sketchptr));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void theta_union_update_with_bytes(void* unionptr, const void* buffer, unsigned length) {
try {
static_cast<theta_union_pg*>(unionptr)->update(wrapped_compact_theta_sketch_pg::wrap(buffer, length));
} catch (std::exception& e) {
@@ -184,9 +192,12 @@ void theta_union_update(void* unionptr, const void* buffer, unsigned length) {
}
}
-void* theta_union_get_result(const void* unionptr) {
+void* theta_union_get_result(void* unionptr) {
try {
- return new (palloc(sizeof(compact_theta_sketch_pg))) compact_theta_sketch_pg(static_cast<const theta_union_pg*>(unionptr)->get_result());
+ auto sketchptr = new (palloc(sizeof(compact_theta_sketch_pg))) compact_theta_sketch_pg(static_cast<const theta_union_pg*>(unionptr)->get_result());
+ static_cast<const theta_union_pg*>(unionptr)->~theta_union_pg();
+ pfree(unionptr);
+ return sketchptr;
} catch (std::exception& e) {
pg_error(e.what());
}
@@ -211,7 +222,15 @@ void theta_intersection_delete(void* interptr) {
}
}
-void theta_intersection_update(void* interptr, const void* buffer, unsigned length) {
+void theta_intersection_update_with_sketch(void* interptr, const void* sketchptr) {
+ try {
+ static_cast<theta_intersection_pg*>(interptr)->update(*static_cast<const theta_sketch_pg*>(sketchptr));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void theta_intersection_update_with_bytes(void* interptr, const void* buffer, unsigned length) {
try {
static_cast<theta_intersection_pg*>(interptr)->update(wrapped_compact_theta_sketch_pg::wrap(buffer, length));
} catch (std::exception& e) {
@@ -219,9 +238,12 @@ void theta_intersection_update(void* interptr, const void* buffer, unsigned leng
}
}
-void* theta_intersection_get_result(const void* interptr) {
+void* theta_intersection_get_result(void* interptr) {
try {
- return new (palloc(sizeof(compact_theta_sketch_pg))) compact_theta_sketch_pg(static_cast<const theta_intersection_pg*>(interptr)->get_result());
+ auto sketchptr = new (palloc(sizeof(compact_theta_sketch_pg))) compact_theta_sketch_pg(static_cast<const theta_intersection_pg*>(interptr)->get_result());
+ static_cast<const theta_intersection_pg*>(interptr)->~theta_intersection_pg();
+ pfree(interptr);
+ return sketchptr;
} catch (std::exception& e) {
pg_error(e.what());
}
diff --git a/src/theta_sketch_c_adapter.h b/src/theta_sketch_c_adapter.h
index 5d6eef1..7f44045 100644
--- a/src/theta_sketch_c_adapter.h
+++ b/src/theta_sketch_c_adapter.h
@@ -44,13 +44,15 @@ void* theta_sketch_deserialize(const char* buffer, unsigned length);
void* theta_union_new_default();
void* theta_union_new(unsigned lg_k);
void theta_union_delete(void* unionptr);
-void theta_union_update(void* unionptr, const void* buffer, unsigned length);
-void* theta_union_get_result(const void* unionptr);
+void theta_union_update_with_sketch(void* unionptr, const void* sketchptr);
+void theta_union_update_with_bytes(void* unionptr, const void* buffer, unsigned length);
+void* theta_union_get_result(void* unionptr);
void* theta_intersection_new_default();
void theta_intersection_delete(void* interptr);
-void theta_intersection_update(void* interptr, const void* buffer, unsigned length);
-void* theta_intersection_get_result(const void* interptr);
+void theta_intersection_update_with_sketch(void* interptr, const void* sketchptr);
+void theta_intersection_update_with_bytes(void* interptr, const void* buffer, unsigned length);
+void* theta_intersection_get_result(void* interptr);
void* theta_a_not_b(const void* buffer1, unsigned length1, const void* buffer2, unsigned length2);
diff --git a/src/theta_sketch_pg_functions.c b/src/theta_sketch_pg_functions.c
index 27ca259..44f23c3 100644
--- a/src/theta_sketch_pg_functions.c
+++ b/src/theta_sketch_pg_functions.c
@@ -26,42 +26,46 @@
#include "theta_sketch_c_adapter.h"
#include "base64.h"
+#include "agg_state.h"
/* PG_FUNCTION_INFO_V1 macro to pass functions to postgres */
-PG_FUNCTION_INFO_V1(pg_theta_sketch_add_item);
-PG_FUNCTION_INFO_V1(pg_theta_sketch_get_estimate);
-PG_FUNCTION_INFO_V1(pg_theta_sketch_get_estimate_and_bounds);
-PG_FUNCTION_INFO_V1(pg_theta_sketch_to_string);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_build_agg);
PG_FUNCTION_INFO_V1(pg_theta_sketch_union_agg);
PG_FUNCTION_INFO_V1(pg_theta_sketch_intersection_agg);
PG_FUNCTION_INFO_V1(pg_theta_sketch_from_internal);
PG_FUNCTION_INFO_V1(pg_theta_sketch_get_estimate_from_internal);
-PG_FUNCTION_INFO_V1(pg_theta_union_get_result);
-PG_FUNCTION_INFO_V1(pg_theta_intersection_get_result);
+//PG_FUNCTION_INFO_V1(pg_theta_union_get_result);
+//PG_FUNCTION_INFO_V1(pg_theta_intersection_get_result);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_combine);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_serialize_state);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_deserialize_state);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_get_estimate);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_get_estimate_and_bounds);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_to_string);
PG_FUNCTION_INFO_V1(pg_theta_sketch_union);
PG_FUNCTION_INFO_V1(pg_theta_sketch_intersection);
PG_FUNCTION_INFO_V1(pg_theta_sketch_a_not_b);
/* function declarations */
-Datum pg_theta_sketch_recv(PG_FUNCTION_ARGS);
-Datum pg_theta_sketch_send(PG_FUNCTION_ARGS);
-Datum pg_theta_sketch_add_item(PG_FUNCTION_ARGS);
-Datum pg_theta_sketch_get_estimate(PG_FUNCTION_ARGS);
-Datum pg_theta_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
-Datum pg_theta_sketch_to_string(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_build_agg(PG_FUNCTION_ARGS);
Datum pg_theta_sketch_union_agg(PG_FUNCTION_ARGS);
Datum pg_theta_sketch_intersection_agg(PG_FUNCTION_ARGS);
Datum pg_theta_sketch_from_internal(PG_FUNCTION_ARGS);
Datum pg_theta_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS);
-Datum pg_theta_union_get_result(PG_FUNCTION_ARGS);
-Datum pg_theta_intersection_get_result(PG_FUNCTION_ARGS);
+//Datum pg_theta_union_get_result(PG_FUNCTION_ARGS);
+//Datum pg_theta_intersection_get_result(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_combine(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_serialize_state(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_deserialize_state(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_get_estimate(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_to_string(PG_FUNCTION_ARGS);
Datum pg_theta_sketch_union(PG_FUNCTION_ARGS);
Datum pg_theta_sketch_intersection(PG_FUNCTION_ARGS);
Datum pg_theta_sketch_a_not_b(PG_FUNCTION_ARGS);
-Datum pg_theta_sketch_add_item(PG_FUNCTION_ARGS) {
- void* sketchptr;
- int lg_k;
+Datum pg_theta_sketch_build_agg(PG_FUNCTION_ARGS) {
+ struct agg_state* stateptr;
float p;
// anyelement
@@ -81,20 +85,22 @@ Datum pg_theta_sketch_add_item(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "theta_sketch_add_item called in non-aggregate context");
+ elog(ERROR, "theta_sketch_build_agg called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 0;
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = MUTABLE_SKETCH;
+ stateptr->lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 0;
p = PG_NARGS() > 3 ? PG_GETARG_FLOAT4(3) : 1;
- if (lg_k) {
- sketchptr = p ? theta_sketch_new_lgk_p(lg_k, p) : theta_sketch_new_lgk(lg_k);
+ if (stateptr->lg_k) {
+ stateptr->ptr = p ? theta_sketch_new_lgk_p(stateptr->lg_k, p) : theta_sketch_new_lgk(stateptr->lg_k);
} else {
- sketchptr = theta_sketch_new_default();
+ stateptr->ptr = theta_sketch_new_default();
}
} else {
- sketchptr = PG_GETARG_POINTER(0);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
}
element_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
@@ -102,68 +108,22 @@ Datum pg_theta_sketch_add_item(PG_FUNCTION_ARGS) {
get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);
if (typlen == -1) {
// varlena
- theta_sketch_update(sketchptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element));
+ theta_sketch_update(stateptr->ptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element));
} else if (typbyval) {
// fixed-length passed by value
- theta_sketch_update(sketchptr, &element, typlen);
+ theta_sketch_update(stateptr->ptr, &element, typlen);
} else {
// fixed-length passed by reference
- theta_sketch_update(sketchptr, (void*)element, typlen);
+ theta_sketch_update(stateptr->ptr, (void*)element, typlen);
}
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(sketchptr);
+ PG_RETURN_POINTER(stateptr);
}
-Datum pg_theta_sketch_get_estimate(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- double estimate;
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = theta_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- estimate = theta_sketch_get_estimate(sketchptr);
- theta_sketch_delete(sketchptr);
- PG_RETURN_FLOAT8(estimate);
-}
-
-Datum pg_theta_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 = theta_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
- est_and_bounds = (Datum*) theta_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
- theta_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_theta_sketch_to_string(PG_FUNCTION_ARGS) {
- const bytea* bytes_in;
- void* sketchptr;
- char* str;
- bytes_in = PG_GETARG_BYTEA_P(0);
- sketchptr = theta_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
- str = theta_sketch_to_string(sketchptr);
- theta_sketch_delete(sketchptr);
- PG_RETURN_TEXT_P(cstring_to_text(str));
-}
-
-Datum pg_theta_sketch_intersection_agg(PG_FUNCTION_ARGS) {
- void* interptr;
+Datum pg_theta_sketch_union_agg(PG_FUNCTION_ARGS) {
+ struct agg_state* stateptr;
bytea* sketch_bytes;
MemoryContext oldcontext;
@@ -176,28 +136,30 @@ Datum pg_theta_sketch_intersection_agg(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "theta_sketch_intersect called in non-aggregate context");
+ elog(ERROR, "theta_sketch_union called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- interptr = theta_intersection_new_default();
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = UNION;
+ stateptr->lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 0;
+ stateptr->ptr = stateptr->lg_k ? theta_union_new(stateptr->lg_k) : theta_union_new_default();
} else {
- interptr = PG_GETARG_POINTER(0);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
}
sketch_bytes = PG_GETARG_BYTEA_P(1);
- theta_intersection_update(interptr, VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
+ theta_union_update_with_bytes(stateptr->ptr, VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(interptr);
+ PG_RETURN_POINTER(stateptr);
}
-Datum pg_theta_sketch_union_agg(PG_FUNCTION_ARGS) {
- void* unionptr;
+Datum pg_theta_sketch_intersection_agg(PG_FUNCTION_ARGS) {
+ struct agg_state* stateptr;
bytea* sketch_bytes;
- int lg_k;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -209,27 +171,28 @@ Datum pg_theta_sketch_union_agg(PG_FUNCTION_ARGS) {
}
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "theta_sketch_merge called in non-aggregate context");
+ elog(ERROR, "theta_sketch_intersection called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
if (PG_ARGISNULL(0)) {
- lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 0;
- unionptr = lg_k ? theta_union_new(lg_k) : theta_union_new_default();
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = INTERSECTION;
+ stateptr->ptr = theta_intersection_new_default();
} else {
- unionptr = PG_GETARG_POINTER(0);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
}
sketch_bytes = PG_GETARG_BYTEA_P(1);
- theta_union_update(unionptr, VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
+ theta_intersection_update_with_bytes(stateptr->ptr, VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_POINTER(unionptr);
+ PG_RETURN_POINTER(stateptr->ptr);
}
Datum pg_theta_sketch_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
+ struct agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -242,10 +205,17 @@ Datum pg_theta_sketch_from_internal(PG_FUNCTION_ARGS) {
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- sketchptr = theta_sketch_compact(sketchptr);
- bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
- theta_sketch_delete(sketchptr);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == MUTABLE_SKETCH) {
+ stateptr->ptr = theta_sketch_compact(stateptr->ptr);
+ } else if (stateptr->type == UNION) {
+ stateptr->ptr = theta_union_get_result(stateptr->ptr);
+ } else if (stateptr->type == INTERSECTION) {
+ stateptr->ptr = theta_intersection_get_result(stateptr->ptr);
+ }
+ bytes_out = theta_sketch_serialize(stateptr->ptr, VARHDRSZ);
+ theta_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
@@ -254,7 +224,7 @@ Datum pg_theta_sketch_from_internal(PG_FUNCTION_ARGS) {
}
Datum pg_theta_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
- void* sketchptr;
+ struct agg_state* stateptr;
double estimate;
MemoryContext oldcontext;
@@ -263,22 +233,138 @@ Datum pg_theta_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "theta_sketch_from_internal called in non-aggregate context");
+ elog(ERROR, "theta_sketch_get_estimate_from_internal called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- sketchptr = PG_GETARG_POINTER(0);
- estimate = theta_sketch_get_estimate(sketchptr);
- theta_sketch_delete(sketchptr);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+ estimate = theta_sketch_get_estimate(stateptr->ptr);
+ theta_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
MemoryContextSwitchTo(oldcontext);
PG_RETURN_FLOAT8(estimate);
}
-Datum pg_theta_union_get_result(PG_FUNCTION_ARGS) {
- void* unionptr;
- void* sketchptr;
+//Datum pg_theta_union_get_result(PG_FUNCTION_ARGS) {
+// struct agg_state* stateptr;
+// struct ptr_with_size bytes_out;
+//
+// MemoryContext oldcontext;
+// MemoryContext aggcontext;
+//
+// if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+//
+// if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+// elog(ERROR, "theta_union_get_result called in non-aggregate context");
+// }
+// oldcontext = MemoryContextSwitchTo(aggcontext);
+//
+// stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+// if (stateptr->type == PHASE1) stateptr->ptr = theta_union_get_result(stateptr->ptr);
+// bytes_out = theta_sketch_serialize(stateptr->ptr, VARHDRSZ);
+// theta_sketch_delete(stateptr->ptr);
+// pfree(stateptr);
+// SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+//
+// MemoryContextSwitchTo(oldcontext);
+//
+// PG_RETURN_BYTEA_P(bytes_out.ptr);
+//}
+//
+//Datum pg_theta_intersection_get_result(PG_FUNCTION_ARGS) {
+// struct agg_state* stateptr;
+// struct ptr_with_size bytes_out;
+//
+// MemoryContext oldcontext;
+// MemoryContext aggcontext;
+//
+// if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+//
+// if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+// elog(ERROR, "theta_intersection_get_result called in non-aggregate context");
+// }
+// oldcontext = MemoryContextSwitchTo(aggcontext);
+//
+// stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+// if (stateptr->type == PHASE1) stateptr->ptr = theta_intersection_get_result(stateptr->ptr);
+// bytes_out = theta_sketch_serialize(stateptr->ptr, VARHDRSZ);
+// theta_sketch_delete(stateptr->ptr);
+// pfree(stateptr);
+// SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+//
+// MemoryContextSwitchTo(oldcontext);
+//
+// PG_RETURN_BYTEA_P(bytes_out.ptr);
+//}
+
+Datum pg_theta_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, "theta_sketch_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 = stateptr1 ? stateptr1->type : stateptr2->type;
+ stateptr->lg_k = stateptr1 ? stateptr1->lg_k : stateptr2->lg_k;
+ if (stateptr->type == INTERSECTION) {
+ stateptr->ptr = theta_intersection_new_default();
+ } else {
+ stateptr->ptr = stateptr->lg_k ? theta_union_new(stateptr->lg_k) : theta_union_new_default();
+ }
+ if (stateptr1) {
+ if (stateptr1->type == UNION) {
+ stateptr1->ptr = theta_union_get_result(stateptr1->ptr);
+ theta_union_update_with_sketch(stateptr->ptr, stateptr1->ptr);
+ } else if (stateptr1->type == INTERSECTION) {
+ stateptr1->ptr = theta_intersection_get_result(stateptr1->ptr);
+ theta_intersection_update_with_sketch(stateptr->ptr, stateptr1->ptr);
+ } else {
+ theta_union_update_with_sketch(stateptr->ptr, stateptr1->ptr);
+ }
+ theta_sketch_delete(stateptr1->ptr);
+ pfree(stateptr1);
+ }
+ if (stateptr2) {
+ if (stateptr2->type == UNION) {
+ stateptr2->ptr = theta_union_get_result(stateptr2->ptr);
+ theta_union_update_with_sketch(stateptr->ptr, stateptr2->ptr);
+ } else if (stateptr2->type == INTERSECTION) {
+ stateptr2->ptr = theta_intersection_get_result(stateptr2->ptr);
+ theta_intersection_update_with_sketch(stateptr->ptr, stateptr2->ptr);
+ } else {
+ theta_union_update_with_sketch(stateptr->ptr, stateptr2->ptr);
+ }
+ theta_sketch_delete(stateptr2->ptr);
+ pfree(stateptr2);
+ }
+ if (stateptr->type == INTERSECTION) {
+ stateptr->ptr = theta_intersection_get_result(stateptr->ptr);
+ } else {
+ stateptr->ptr = theta_union_get_result(stateptr->ptr);
+ }
+ stateptr->type = IMMUTABLE_SKETCH;
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(stateptr);
+}
+
+Datum pg_theta_sketch_serialize_state(PG_FUNCTION_ARGS) {
+ struct agg_state* stateptr;
struct ptr_with_size bytes_out;
MemoryContext oldcontext;
@@ -287,15 +373,23 @@ Datum pg_theta_union_get_result(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "theta_union_get_result called in non-aggregate context");
+ elog(ERROR, "theta_sketch_serialize_state called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- unionptr = PG_GETARG_POINTER(0);
- sketchptr = theta_union_get_result(unionptr);
- bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
- theta_sketch_delete(sketchptr);
- theta_union_delete(unionptr);
+ stateptr = (struct agg_state*) PG_GETARG_POINTER(0);
+ if (stateptr->type == MUTABLE_SKETCH) {
+ stateptr->ptr = theta_sketch_compact(stateptr->ptr);
+ } else if (stateptr->type == UNION) {
+ stateptr->ptr = theta_union_get_result(stateptr->ptr);
+ } else if (stateptr->type == INTERSECTION) {
+ stateptr->ptr = theta_intersection_get_result(stateptr->ptr);
+ }
+ bytes_out = theta_sketch_serialize(stateptr->ptr, VARHDRSZ + 2);
+ ((char*)bytes_out.ptr)[VARHDRSZ] = IMMUTABLE_SKETCH;
+ ((char*)bytes_out.ptr)[VARHDRSZ + 1] = stateptr->lg_k;
+ theta_sketch_delete(stateptr->ptr);
+ pfree(stateptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
@@ -303,10 +397,9 @@ Datum pg_theta_union_get_result(PG_FUNCTION_ARGS) {
PG_RETURN_BYTEA_P(bytes_out.ptr);
}
-Datum pg_theta_intersection_get_result(PG_FUNCTION_ARGS) {
- void* interptr;
- void* sketchptr;
- struct ptr_with_size bytes_out;
+Datum pg_theta_sketch_deserialize_state(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ struct agg_state* stateptr;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -314,20 +407,65 @@ Datum pg_theta_intersection_get_result(PG_FUNCTION_ARGS) {
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
- elog(ERROR, "theta_intersection_get_result called in non-aggregate context");
+ elog(ERROR, "theta_sketch_deserialize_state called in non-aggregate context");
}
oldcontext = MemoryContextSwitchTo(aggcontext);
- interptr = PG_GETARG_POINTER(0);
- sketchptr = theta_intersection_get_result(interptr);
- bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
- theta_sketch_delete(sketchptr);
- theta_intersection_delete(interptr);
- SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ stateptr = palloc(sizeof(struct agg_state));
+ stateptr->type = *VARDATA(bytes_in);
+ stateptr->lg_k = *(VARDATA(bytes_in) + 1);
+ stateptr->ptr = theta_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_theta_sketch_get_estimate(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ double estimate;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = theta_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ estimate = theta_sketch_get_estimate(sketchptr);
+ theta_sketch_delete(sketchptr);
+ PG_RETURN_FLOAT8(estimate);
+}
+
+Datum pg_theta_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 = theta_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ num_std_devs = PG_NARGS() > 1 ? PG_GETARG_INT32(1) : 1;
+ est_and_bounds = (Datum*) theta_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+ theta_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_theta_sketch_to_string(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ char* str;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = theta_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ str = theta_sketch_to_string(sketchptr);
+ theta_sketch_delete(sketchptr);
+ PG_RETURN_TEXT_P(cstring_to_text(str));
}
Datum pg_theta_sketch_union(PG_FUNCTION_ARGS) {
@@ -336,20 +474,19 @@ Datum pg_theta_sketch_union(PG_FUNCTION_ARGS) {
void* unionptr;
void* sketchptr;
struct ptr_with_size bytes_out;
- int lg_k;
-
+ unsigned lg_k;
+
lg_k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : 0;
unionptr = lg_k ? theta_union_new(lg_k) : theta_union_new_default();
if (!PG_ARGISNULL(0)) {
bytes_in1 = PG_GETARG_BYTEA_P(0);
- theta_union_update(unionptr, VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
+ theta_union_update_with_bytes(unionptr, VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
}
if (!PG_ARGISNULL(1)) {
bytes_in2 = PG_GETARG_BYTEA_P(1);
- theta_union_update(unionptr, VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
+ theta_union_update_with_bytes(unionptr, VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
}
sketchptr = theta_union_get_result(unionptr);
- theta_union_delete(unionptr);
bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
theta_sketch_delete(sketchptr);
SET_VARSIZE(bytes_out.ptr, bytes_out.size);
@@ -362,18 +499,17 @@ Datum pg_theta_sketch_intersection(PG_FUNCTION_ARGS) {
void* interptr;
void* sketchptr;
struct ptr_with_size bytes_out;
-
+
interptr = theta_intersection_new_default();
if (!PG_ARGISNULL(0)) {
bytes_in1 = PG_GETARG_BYTEA_P(0);
- theta_intersection_update(interptr, VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
+ theta_intersection_update_with_bytes(interptr, VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
}
if (!PG_ARGISNULL(1)) {
bytes_in2 = PG_GETARG_BYTEA_P(1);
- theta_intersection_update(interptr, VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
+ theta_intersection_update_with_bytes(interptr, VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
}
sketchptr = theta_intersection_get_result(interptr);
- theta_intersection_delete(interptr);
bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
theta_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