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 03:55:19 UTC
[datasketches-postgresql] branch partial_agg updated (0b41511 -> c0ce01f)
This is an automated email from the ASF dual-hosted git repository.
alsay pushed a change to branch partial_agg
in repository https://gitbox.apache.org/repos/asf/datasketches-postgresql.git
from 0b41511 aod sketch partial aggregation support
new 51bdf2c removed duplicate line
new c0ce01f fixes
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
sql/datasketches_theta_sketch.sql | 24 +++++++-----
src/aod_sketch_pg_functions.c | 3 --
src/theta_sketch_c_adapter.cpp | 4 +-
src/theta_sketch_pg_functions.c | 77 ++++++++++++++++++++++++++-------------
4 files changed, 68 insertions(+), 40 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org
[datasketches-postgresql] 02/02: fixes
Posted by al...@apache.org.
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 c0ce01f7891c98b7236a33971686f9127395febd
Author: AlexanderSaydakov <Al...@users.noreply.github.com>
AuthorDate: Wed Apr 19 20:55:10 2023 -0700
fixes
---
sql/datasketches_theta_sketch.sql | 24 +++++++-----
src/aod_sketch_pg_functions.c | 1 -
src/theta_sketch_c_adapter.cpp | 4 +-
src/theta_sketch_pg_functions.c | 77 ++++++++++++++++++++++++++-------------
4 files changed, 68 insertions(+), 38 deletions(-)
diff --git a/sql/datasketches_theta_sketch.sql b/sql/datasketches_theta_sketch.sql
index 9ed5ad7..bd5ca11 100644
--- a/sql/datasketches_theta_sketch.sql
+++ b/sql/datasketches_theta_sketch.sql
@@ -66,8 +66,12 @@ CREATE OR REPLACE FUNCTION theta_sketch_intersection_agg(internal, theta_sketch)
AS '$libdir/datasketches', 'pg_theta_sketch_intersection_agg'
LANGUAGE C IMMUTABLE PARALLEL SAFE;
-CREATE OR REPLACE FUNCTION theta_sketch_combine(internal, internal) RETURNS internal
- AS '$libdir/datasketches', 'pg_theta_sketch_combine'
+CREATE OR REPLACE FUNCTION theta_sketch_union_combine(internal, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_theta_sketch_union_combine'
+ LANGUAGE C IMMUTABLE PARALLEL SAFE;
+
+CREATE OR REPLACE FUNCTION theta_sketch_intersection_combine(internal, internal) RETURNS internal
+ AS '$libdir/datasketches', 'pg_theta_sketch_intersection_combine'
LANGUAGE C IMMUTABLE PARALLEL SAFE;
CREATE OR REPLACE FUNCTION theta_sketch_serialize_state(internal) RETURNS bytea
@@ -81,7 +85,7 @@ CREATE OR REPLACE FUNCTION theta_sketch_deserialize_state(bytea, internal) RETUR
CREATE OR REPLACE AGGREGATE theta_sketch_distinct(anyelement) (
STYPE = internal,
SFUNC = theta_sketch_build_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_union_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_get_estimate_from_internal,
@@ -91,7 +95,7 @@ CREATE OR REPLACE AGGREGATE theta_sketch_distinct(anyelement) (
CREATE OR REPLACE AGGREGATE theta_sketch_distinct(anyelement, int) (
STYPE = internal,
SFUNC = theta_sketch_build_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_union_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_get_estimate_from_internal,
@@ -101,7 +105,7 @@ CREATE OR REPLACE AGGREGATE theta_sketch_distinct(anyelement, int) (
CREATE AGGREGATE theta_sketch_build(anyelement) (
STYPE = internal,
SFUNC = theta_sketch_build_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_union_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_from_internal,
@@ -111,7 +115,7 @@ CREATE AGGREGATE theta_sketch_build(anyelement) (
CREATE OR REPLACE AGGREGATE theta_sketch_build(anyelement, int) (
STYPE = internal,
SFUNC = theta_sketch_build_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_union_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_from_internal,
@@ -121,7 +125,7 @@ CREATE OR REPLACE AGGREGATE theta_sketch_build(anyelement, int) (
CREATE OR REPLACE AGGREGATE theta_sketch_build(anyelement, int, real) (
STYPE = internal,
SFUNC = theta_sketch_build_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_union_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_from_internal,
@@ -131,7 +135,7 @@ CREATE OR REPLACE AGGREGATE theta_sketch_build(anyelement, int, real) (
CREATE OR REPLACE AGGREGATE theta_sketch_union(theta_sketch) (
STYPE = internal,
SFUNC = theta_sketch_union_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_union_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_from_internal,
@@ -141,7 +145,7 @@ CREATE OR REPLACE AGGREGATE theta_sketch_union(theta_sketch) (
CREATE OR REPLACE AGGREGATE theta_sketch_union(theta_sketch, int) (
STYPE = internal,
SFUNC = theta_sketch_union_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_union_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_from_internal,
@@ -151,7 +155,7 @@ CREATE OR REPLACE AGGREGATE theta_sketch_union(theta_sketch, int) (
CREATE OR REPLACE AGGREGATE theta_sketch_intersection(theta_sketch) (
STYPE = internal,
SFUNC = theta_sketch_intersection_agg,
- COMBINEFUNC = theta_sketch_combine,
+ COMBINEFUNC = theta_sketch_intersection_combine,
SERIALFUNC = theta_sketch_serialize_state,
DESERIALFUNC = theta_sketch_deserialize_state,
FINALFUNC = theta_sketch_from_internal,
diff --git a/src/aod_sketch_pg_functions.c b/src/aod_sketch_pg_functions.c
index c4a2a2b..1b6df01 100644
--- a/src/aod_sketch_pg_functions.c
+++ b/src/aod_sketch_pg_functions.c
@@ -338,7 +338,6 @@ Datum pg_aod_sketch_intersection_combine(PG_FUNCTION_ARGS) {
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) {
diff --git a/src/theta_sketch_c_adapter.cpp b/src/theta_sketch_c_adapter.cpp
index d6e573b..757a0ac 100644
--- a/src/theta_sketch_c_adapter.cpp
+++ b/src/theta_sketch_c_adapter.cpp
@@ -195,7 +195,7 @@ void theta_union_update_with_bytes(void* unionptr, const void* buffer, unsigned
void* theta_union_get_result(void* unionptr) {
try {
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();
+ static_cast<theta_union_pg*>(unionptr)->~theta_union_pg();
pfree(unionptr);
return sketchptr;
} catch (std::exception& e) {
@@ -241,7 +241,7 @@ void theta_intersection_update_with_bytes(void* interptr, const void* buffer, un
void* theta_intersection_get_result(void* interptr) {
try {
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();
+ static_cast<theta_intersection_pg*>(interptr)->~theta_intersection_pg();
pfree(interptr);
return sketchptr;
} catch (std::exception& e) {
diff --git a/src/theta_sketch_pg_functions.c b/src/theta_sketch_pg_functions.c
index c74f5fe..c4cb5b8 100644
--- a/src/theta_sketch_pg_functions.c
+++ b/src/theta_sketch_pg_functions.c
@@ -34,7 +34,8 @@ 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_sketch_combine);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_union_combine);
+PG_FUNCTION_INFO_V1(pg_theta_sketch_intersection_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);
@@ -50,7 +51,8 @@ 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_sketch_combine(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_union_combine(PG_FUNCTION_ARGS);
+Datum pg_theta_sketch_intersection_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);
@@ -243,7 +245,7 @@ Datum pg_theta_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
PG_RETURN_FLOAT8(estimate);
}
-Datum pg_theta_sketch_combine(PG_FUNCTION_ARGS) {
+Datum pg_theta_sketch_union_combine(PG_FUNCTION_ARGS) {
struct agg_state* stateptr1;
struct agg_state* stateptr2;
struct agg_state* stateptr;
@@ -262,45 +264,70 @@ Datum pg_theta_sketch_combine(PG_FUNCTION_ARGS) {
stateptr2 = (struct agg_state*) PG_GETARG_POINTER(1);
stateptr = palloc(sizeof(struct agg_state));
- stateptr->type = stateptr1 ? stateptr1->type : stateptr2->type;
+ stateptr->type = IMMUTABLE_SKETCH;
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();
- }
+ 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_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_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->ptr = theta_union_get_result(stateptr->ptr);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(stateptr);
+}
+
+Datum pg_theta_sketch_intersection_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 = IMMUTABLE_SKETCH;
+ stateptr->ptr = theta_intersection_new_default();
+ if (stateptr1) {
+ if (stateptr1->type == INTERSECTION) {
+ stateptr1->ptr = theta_intersection_get_result(stateptr1->ptr);
+ }
+ theta_intersection_update_with_sketch(stateptr->ptr, stateptr1->ptr);
+ theta_sketch_delete(stateptr1->ptr);
+ pfree(stateptr1);
+ }
+ if (stateptr2) {
+ if (stateptr2->type == INTERSECTION) {
+ stateptr2->ptr = theta_intersection_get_result(stateptr2->ptr);
+ }
+ theta_intersection_update_with_sketch(stateptr->ptr, stateptr2->ptr);
+ theta_sketch_delete(stateptr2->ptr);
+ pfree(stateptr2);
+ }
+ stateptr->ptr = theta_intersection_get_result(stateptr->ptr);
MemoryContextSwitchTo(oldcontext);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org
[datasketches-postgresql] 01/02: removed duplicate line
Posted by al...@apache.org.
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 51bdf2c1137b20490f5061675e99c9577fd81598
Author: AlexanderSaydakov <Al...@users.noreply.github.com>
AuthorDate: Wed Apr 19 17:40:41 2023 -0700
removed duplicate line
---
src/aod_sketch_pg_functions.c | 2 --
1 file changed, 2 deletions(-)
diff --git a/src/aod_sketch_pg_functions.c b/src/aod_sketch_pg_functions.c
index 7a951b5..c4a2a2b 100644
--- a/src/aod_sketch_pg_functions.c
+++ b/src/aod_sketch_pg_functions.c
@@ -312,7 +312,6 @@ Datum pg_aod_sketch_union_combine(PG_FUNCTION_ARGS) {
pfree(stateptr2);
}
stateptr->ptr = aod_union_get_result(stateptr->ptr);
- stateptr->type = IMMUTABLE_SKETCH;
MemoryContextSwitchTo(oldcontext);
@@ -359,7 +358,6 @@ Datum pg_aod_sketch_intersection_combine(PG_FUNCTION_ARGS) {
pfree(stateptr2);
}
stateptr->ptr = aod_intersection_get_result(stateptr->ptr);
- stateptr->type = IMMUTABLE_SKETCH;
MemoryContextSwitchTo(oldcontext);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org