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 2019/07/03 01:10:02 UTC
[incubator-datasketches-postgresql] 01/01: GCC 4.8.5 compatibility
This is an automated email from the ASF dual-hosted git repository.
alsay pushed a commit to branch gcc_4_8_5
in repository https://gitbox.apache.org/repos/asf/incubator-datasketches-postgresql.git
commit e0f508be73bb6510f5405f4e04ab175ea26c71df
Author: Alexander Saydakov <Al...@users.noreply.github.com>
AuthorDate: Wed Jul 3 00:47:33 2019 +0000
GCC 4.8.5 compatibility
---
Makefile | 2 +-
src/allocator.h | 9 +++-
src/common.c | 26 +++++++++-
src/cpc_sketch_c_adapter.cpp | 26 +++++-----
src/frequent_strings_sketch_c_adapter.cpp | 37 ++++++++------
src/frequent_strings_sketch_c_adapter.h | 9 ++--
src/frequent_strings_sketch_pg_functions.c | 7 +--
src/hll_sketch_c_adapter.cpp | 58 ++++++++++++---------
src/hll_sketch_c_adapter.h | 11 ++--
src/hll_sketch_pg_functions.c | 24 +++++----
src/kll_float_sketch_c_adapter.cpp | 56 ++++++++++++---------
src/kll_float_sketch_c_adapter.h | 15 +++---
src/kll_float_sketch_pg_functions.c | 12 +++--
src/postgres_h_substitute.h | 49 ++++++++++++++++++
src/theta_sketch_c_adapter.cpp | 81 +++++++++++++++++-------------
src/theta_sketch_c_adapter.h | 11 ++--
src/theta_sketch_pg_functions.c | 37 ++++++++------
17 files changed, 306 insertions(+), 164 deletions(-)
diff --git a/Makefile b/Makefile
index 2d24a99..d5995c8 100644
--- a/Makefile
+++ b/Makefile
@@ -15,7 +15,7 @@ OBJS += $(CPC)/cpc_sketch.o $(CPC)/fm85.o $(CPC)/fm85Compression.o $(CPC)/fm85Co
DATA = sql/datasketches_cpc_sketch.sql sql/datasketches_kll_float_sketch.sql sql/datasketches_theta_sketch.sql sql/datasketches_frequent_strings_sketch.sql sql/datasketches_hll_sketch.sql
-PG_CPPFLAGS = -I/usr/local/include -I$(CORE)/kll/include -I$(CORE)/common/include -I$(CORE)/cpc/include -I$(CORE)/theta/include -I$(CORE)/fi/include -I$(CORE)/hll/include
+PG_CPPFLAGS = -std=c++11 -I/usr/local/include -I$(CORE)/kll/include -I$(CORE)/common/include -I$(CORE)/cpc/include -I$(CORE)/theta/include -I$(CORE)/fi/include -I$(CORE)/hll/include
SHLIB_LINK = -lstdc++ -L/usr/local/lib
PG_CONFIG = pg_config
diff --git a/src/allocator.h b/src/allocator.h
index 4d7ed00..c66a0a9 100644
--- a/src/allocator.h
+++ b/src/allocator.h
@@ -17,8 +17,13 @@
* under the License.
*/
+#ifndef PALLOC_ALLOCATOR_H
+#define PALLOC_ALLOCATOR_H
+
+// these extern declarations are to avoid including postgres.h in C++ code
extern "C" {
-#include <postgres.h>
+extern void *palloc(unsigned long long size);
+extern void pfree(void *pointer);
}
#include <new>
@@ -90,3 +95,5 @@ template <class T>
inline bool operator!=(const palloc_allocator<T>&, const palloc_allocator<T>&) {
return false;
}
+
+#endif
diff --git a/src/common.c b/src/common.c
index ffd2838..f9a11f1 100644
--- a/src/common.c
+++ b/src/common.c
@@ -30,6 +30,10 @@ PG_FUNCTION_INFO_V1(pg_sketch_out);
Datum pg_sketch_in(PG_FUNCTION_ARGS);
Datum pg_sketch_out(PG_FUNCTION_ARGS);
+void pg_error(const char* message);
+Datum pg_float4_get_datum(float x);
+Datum pg_float8_get_datum(double x);
+
// cstring to type
Datum pg_sketch_in(PG_FUNCTION_ARGS) {
// not invoked for nulls
@@ -52,5 +56,25 @@ Datum pg_sketch_out(PG_FUNCTION_ARGS) {
encoded = palloc(encoded_length + 1);
b64_encode(VARDATA(bytes), VARSIZE(bytes) - VARHDRSZ, encoded);
encoded[encoded_length] = '\0';
- PG_RETURN_CSTRING(encoded); // who should free it?
+ PG_RETURN_CSTRING(encoded);
+}
+
+// These are implementations of redirects defined in postgres_h_substitute.h
+
+Datum pg_float4_get_datum(float x) {
+ return Float4GetDatum(x);
+}
+
+Datum pg_float8_get_datum(double x) {
+ return Float8GetDatum(x);
+}
+
+void pg_error(const char* message) {
+ ereport(
+ ERROR,
+ (
+ errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("%s", message)
+ )
+ );
}
diff --git a/src/cpc_sketch_c_adapter.cpp b/src/cpc_sketch_c_adapter.cpp
index 6900380..d6d4520 100644
--- a/src/cpc_sketch_c_adapter.cpp
+++ b/src/cpc_sketch_c_adapter.cpp
@@ -29,7 +29,7 @@ void* cpc_sketch_new(unsigned lg_k) {
datasketches::cpc_init(&palloc, &pfree);
return new (palloc(sizeof(datasketches::cpc_sketch))) datasketches::cpc_sketch(lg_k, datasketches::DEFAULT_SEED);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -38,7 +38,7 @@ void cpc_sketch_delete(void* sketchptr) {
static_cast<datasketches::cpc_sketch*>(sketchptr)->~cpc_sketch();
pfree(sketchptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -46,7 +46,7 @@ void cpc_sketch_update(void* sketchptr, const void* data, unsigned length) {
try {
static_cast<datasketches::cpc_sketch*>(sketchptr)->update(data, length);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -54,7 +54,7 @@ double cpc_sketch_get_estimate(const void* sketchptr) {
try {
return static_cast<const datasketches::cpc_sketch*>(sketchptr)->get_estimate();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -66,7 +66,7 @@ Datum* cpc_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_st
est_and_bounds[2] = Float8GetDatum(static_cast<const datasketches::cpc_sketch*>(sketchptr)->get_upper_bound(num_std_devs));
return est_and_bounds;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -74,9 +74,9 @@ void cpc_sketch_to_string(const void* sketchptr, char* buffer, unsigned length)
try {
std::stringstream s;
s << *(static_cast<const datasketches::cpc_sketch*>(sketchptr));
- snprintf(buffer, length, s.str().c_str());
+ snprintf(buffer, length, "%s", s.str().c_str());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -88,7 +88,7 @@ void* cpc_sketch_serialize(const void* sketchptr) {
SET_VARSIZE(buffer, length);
return buffer;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -98,7 +98,7 @@ void* cpc_sketch_deserialize(const char* buffer, unsigned length) {
auto ptr = datasketches::cpc_sketch::deserialize(buffer, length, datasketches::DEFAULT_SEED);
return ptr.release();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -107,7 +107,7 @@ void* cpc_union_new(unsigned lg_k) {
datasketches::cpc_init(&palloc, &pfree);
return new (palloc(sizeof(datasketches::cpc_union))) datasketches::cpc_union(lg_k, datasketches::DEFAULT_SEED);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -116,7 +116,7 @@ void cpc_union_delete(void* unionptr) {
static_cast<datasketches::cpc_union*>(unionptr)->~cpc_union();
pfree(unionptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -124,7 +124,7 @@ void cpc_union_update(void* unionptr, const void* sketchptr) {
try {
static_cast<datasketches::cpc_union*>(unionptr)->update(*static_cast<const datasketches::cpc_sketch*>(sketchptr));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
@@ -133,6 +133,6 @@ void* cpc_union_get_result(void* unionptr) {
auto ptr = static_cast<datasketches::cpc_union*>(unionptr)->get_result();
return ptr.release();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ elog(ERROR, "%s", e.what());
}
}
diff --git a/src/frequent_strings_sketch_c_adapter.cpp b/src/frequent_strings_sketch_c_adapter.cpp
index a1c4d4e..c2687d8 100644
--- a/src/frequent_strings_sketch_c_adapter.cpp
+++ b/src/frequent_strings_sketch_c_adapter.cpp
@@ -19,6 +19,7 @@
#include "frequent_strings_sketch_c_adapter.h"
#include "allocator.h"
+#include "postgres_h_substitute.h"
#include <string>
#include <sstream>
@@ -90,8 +91,9 @@ void* frequent_strings_sketch_new(unsigned lg_k) {
try {
return new (palloc(sizeof(frequent_strings_sketch))) frequent_strings_sketch(lg_k);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void frequent_strings_sketch_delete(void* sketchptr) {
@@ -99,7 +101,7 @@ void frequent_strings_sketch_delete(void* sketchptr) {
static_cast<frequent_strings_sketch*>(sketchptr)->~frequent_strings_sketch();
pfree(sketchptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -107,7 +109,7 @@ void frequent_strings_sketch_update(void* sketchptr, const char* str, unsigned l
try {
static_cast<frequent_strings_sketch*>(sketchptr)->update(string(str, length), weight);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -115,7 +117,7 @@ void frequent_strings_sketch_merge(void* sketchptr1, const void* sketchptr2) {
try {
static_cast<frequent_strings_sketch*>(sketchptr1)->merge(*static_cast<const frequent_strings_sketch*>(sketchptr2));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -128,20 +130,22 @@ char* frequent_strings_sketch_to_string(const void* sketchptr, bool print_items)
strncpy(buffer, s.str().c_str(), len);
return buffer;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
-void* frequent_strings_sketch_serialize(const void* sketchptr) {
+ptr_with_size frequent_strings_sketch_serialize(const void* sketchptr, unsigned header_size) {
try {
- auto data = static_cast<const frequent_strings_sketch*>(sketchptr)->serialize(VARHDRSZ);
- bytea* buffer = (bytea*) data.first.release();
- const size_t length = data.second;
- SET_VARSIZE(buffer, length);
- return buffer;
+ ptr_with_size p;
+ auto data = static_cast<const frequent_strings_sketch*>(sketchptr)->serialize(header_size);
+ p.ptr = data.first.release();
+ p.size = data.second;
+ return p;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* frequent_strings_sketch_deserialize(const char* buffer, unsigned length) {
@@ -150,16 +154,18 @@ void* frequent_strings_sketch_deserialize(const char* buffer, unsigned length) {
frequent_strings_sketch(frequent_strings_sketch::deserialize(buffer, length));
return sketchptr;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
unsigned frequent_strings_sketch_get_serialized_size_bytes(const void* sketchptr) {
try {
return static_cast<const frequent_strings_sketch*>(sketchptr)->get_serialized_size_bytes();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
frequent_strings_sketch_result* frequent_strings_sketch_get_frequent_items(void* sketchptr, bool no_false_positives, unsigned long long threshold) {
@@ -184,6 +190,7 @@ frequent_strings_sketch_result* frequent_strings_sketch_get_frequent_items(void*
result->num = data.size();
return result;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
diff --git a/src/frequent_strings_sketch_c_adapter.h b/src/frequent_strings_sketch_c_adapter.h
index 549c21d..ab0fd3b 100644
--- a/src/frequent_strings_sketch_c_adapter.h
+++ b/src/frequent_strings_sketch_c_adapter.h
@@ -24,8 +24,6 @@
extern "C" {
#endif
-#include <postgres.h>
-
void* frequent_strings_sketch_new(unsigned lg_k);
void frequent_strings_sketch_delete(void* sketchptr);
@@ -33,7 +31,12 @@ void frequent_strings_sketch_update(void* sketchptr, const char* str, unsigned l
void frequent_strings_sketch_merge(void* sketchptr1, const void* sketchptr2);
char* frequent_strings_sketch_to_string(const void* sketchptr, bool print_items);
-void* frequent_strings_sketch_serialize(const void* sketchptr);
+struct ptr_with_size {
+ void* ptr;
+ unsigned long long size;
+};
+
+struct ptr_with_size frequent_strings_sketch_serialize(const void* sketchptr, unsigned header_size);
void* frequent_strings_sketch_deserialize(const char* buffer, unsigned length);
unsigned frequent_strings_sketch_get_serialized_size_bytes(const void* sketchptr);
diff --git a/src/frequent_strings_sketch_pg_functions.c b/src/frequent_strings_sketch_pg_functions.c
index c9c44ac..dced821 100644
--- a/src/frequent_strings_sketch_pg_functions.c
+++ b/src/frequent_strings_sketch_pg_functions.c
@@ -131,16 +131,17 @@ Datum pg_frequent_strings_sketch_merge(PG_FUNCTION_ARGS) {
Datum pg_frequent_strings_sketch_from_internal(PG_FUNCTION_ARGS) {
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
MemoryContext aggcontext;
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
elog(ERROR, "frequent_strings_sketch_from_internal called in non-aggregate context");
}
sketchptr = PG_GETARG_POINTER(0);
- bytes_out = frequent_strings_sketch_serialize(sketchptr);
+ bytes_out = frequent_strings_sketch_serialize(sketchptr, VARHDRSZ);
frequent_strings_sketch_delete(sketchptr);
- PG_RETURN_BYTEA_P(bytes_out);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_frequent_strings_sketch_to_string(PG_FUNCTION_ARGS) {
diff --git a/src/hll_sketch_c_adapter.cpp b/src/hll_sketch_c_adapter.cpp
index 53320f9..b830099 100644
--- a/src/hll_sketch_c_adapter.cpp
+++ b/src/hll_sketch_c_adapter.cpp
@@ -19,6 +19,7 @@
#include "hll_sketch_c_adapter.h"
#include "allocator.h"
+#include "postgres_h_substitute.h"
#include <sstream>
@@ -31,8 +32,9 @@ void* hll_sketch_new(unsigned lg_k) {
try {
return new (palloc(sizeof(hll_sketch))) hll_sketch(lg_k);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* hll_sketch_new_tgt_type(unsigned lg_k, unsigned tgt_type) {
@@ -42,8 +44,9 @@ void* hll_sketch_new_tgt_type(unsigned lg_k, unsigned tgt_type) {
tgt_type == 4 ? datasketches::TgtHllType::HLL_4 : tgt_type == 6 ? datasketches::TgtHllType::HLL_6 : datasketches::TgtHllType::HLL_8
);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void hll_sketch_delete(void* sketchptr) {
@@ -51,7 +54,7 @@ void hll_sketch_delete(void* sketchptr) {
static_cast<hll_sketch*>(sketchptr)->~hll_sketch();
pfree(sketchptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -59,7 +62,7 @@ void hll_sketch_update(void* sketchptr, const void* data, unsigned length) {
try {
static_cast<hll_sketch*>(sketchptr)->update(data, length);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -67,42 +70,45 @@ double hll_sketch_get_estimate(const void* sketchptr) {
try {
return static_cast<const hll_sketch*>(sketchptr)->getEstimate();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
Datum* hll_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs) {
try {
Datum* est_and_bounds = (Datum*) palloc(sizeof(Datum) * 3);
- est_and_bounds[0] = Float8GetDatum(static_cast<const hll_sketch*>(sketchptr)->getEstimate());
- est_and_bounds[1] = Float8GetDatum(static_cast<const hll_sketch*>(sketchptr)->getLowerBound(num_std_devs));
- est_and_bounds[2] = Float8GetDatum(static_cast<const hll_sketch*>(sketchptr)->getUpperBound(num_std_devs));
+ est_and_bounds[0] = pg_float8_get_datum(static_cast<const hll_sketch*>(sketchptr)->getEstimate());
+ est_and_bounds[1] = pg_float8_get_datum(static_cast<const hll_sketch*>(sketchptr)->getLowerBound(num_std_devs));
+ est_and_bounds[2] = pg_float8_get_datum(static_cast<const hll_sketch*>(sketchptr)->getUpperBound(num_std_devs));
return est_and_bounds;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void hll_sketch_to_string(const void* sketchptr, char* buffer, unsigned length) {
try {
std::stringstream s;
static_cast<const hll_sketch*>(sketchptr)->to_string(s);
- snprintf(buffer, length, s.str().c_str());
+ snprintf(buffer, length, "%s", s.str().c_str());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
-void* hll_sketch_serialize(const void* sketchptr) {
+ptr_with_size hll_sketch_serialize(const void* sketchptr, unsigned header_size) {
try {
- auto data = static_cast<const hll_sketch*>(sketchptr)->serializeCompact(VARHDRSZ);
- bytea* buffer = (bytea*) data.first.release();
- const size_t length = data.second;
- SET_VARSIZE(buffer, length);
- return buffer;
+ ptr_with_size p;
+ auto data = static_cast<const hll_sketch*>(sketchptr)->serializeCompact(header_size);
+ p.ptr = data.first.release();
+ p.size = data.second;
+ return p;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* hll_sketch_deserialize(const char* buffer, unsigned length) {
@@ -110,16 +116,18 @@ void* hll_sketch_deserialize(const char* buffer, unsigned length) {
hll_sketch* sketchptr = new (palloc(sizeof(hll_sketch))) hll_sketch(hll_sketch::deserialize(buffer, length));
return sketchptr;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* hll_union_new(unsigned lg_k) {
try {
return new (palloc(sizeof(hll_union))) hll_union(lg_k);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void hll_union_delete(void* unionptr) {
@@ -127,7 +135,7 @@ void hll_union_delete(void* unionptr) {
static_cast<hll_union*>(unionptr)->~hll_union();
pfree(unionptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -135,7 +143,7 @@ void hll_union_update(void* unionptr, const void* sketchptr) {
try {
static_cast<hll_union*>(unionptr)->update(*static_cast<const hll_sketch*>(sketchptr));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -143,8 +151,9 @@ void* hll_union_get_result(void* unionptr) {
try {
return new (palloc(sizeof(hll_sketch))) hll_sketch(static_cast<hll_union*>(unionptr)->getResult());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* hll_union_get_result_tgt_type(void* unionptr, unsigned tgt_type) {
@@ -153,6 +162,7 @@ void* hll_union_get_result_tgt_type(void* unionptr, unsigned tgt_type) {
tgt_type == 4 ? datasketches::TgtHllType::HLL_4 : tgt_type == 6 ? datasketches::TgtHllType::HLL_6 : datasketches::TgtHllType::HLL_8
));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
diff --git a/src/hll_sketch_c_adapter.h b/src/hll_sketch_c_adapter.h
index 1f291a1..394fab2 100644
--- a/src/hll_sketch_c_adapter.h
+++ b/src/hll_sketch_c_adapter.h
@@ -24,8 +24,6 @@
extern "C" {
#endif
-#include <postgres.h>
-
void* hll_sketch_new(unsigned lg_k);
void* hll_sketch_new_tgt_type(unsigned lg_k, unsigned tgt_type);
void hll_sketch_delete(void* sketchptr);
@@ -33,10 +31,15 @@ void hll_sketch_delete(void* sketchptr);
void hll_sketch_update(void* sketchptr, const void* data, unsigned length);
void hll_sketch_merge(void* sketchptr1, const void* sketchptr2);
double hll_sketch_get_estimate(const void* sketchptr);
-Datum* hll_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
+void** hll_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
void hll_sketch_to_string(const void* sketchptr, char* buffer, unsigned length);
-void* hll_sketch_serialize(const void* sketchptr);
+struct ptr_with_size {
+ void* ptr;
+ unsigned long long size;
+};
+
+struct ptr_with_size hll_sketch_serialize(const void* sketchptr, unsigned header_size);
void* hll_sketch_deserialize(const char* buffer, unsigned length);
void* hll_union_new(unsigned lg_k);
diff --git a/src/hll_sketch_pg_functions.c b/src/hll_sketch_pg_functions.c
index 59cdbe2..8cea03b 100644
--- a/src/hll_sketch_pg_functions.c
+++ b/src/hll_sketch_pg_functions.c
@@ -141,7 +141,7 @@ Datum pg_hll_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS) {
sketchptr = hll_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
num_std_devs = PG_GETARG_INT32(1);
if (num_std_devs == 0) num_std_devs = 1; // default
- est_and_bounds = hll_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+ est_and_bounds = (Datum*) hll_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
hll_sketch_delete(sketchptr);
// construct output array
@@ -212,7 +212,7 @@ Datum pg_hll_sketch_union_agg(PG_FUNCTION_ARGS) {
Datum pg_hll_sketch_from_internal(PG_FUNCTION_ARGS) {
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -225,12 +225,13 @@ Datum pg_hll_sketch_from_internal(PG_FUNCTION_ARGS) {
oldcontext = MemoryContextSwitchTo(aggcontext);
sketchptr = PG_GETARG_POINTER(0);
- bytes_out = hll_sketch_serialize(sketchptr);
+ bytes_out = hll_sketch_serialize(sketchptr, VARHDRSZ);
hll_sketch_delete(sketchptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_BYTEA_P(bytes_out);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_hll_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
@@ -259,7 +260,7 @@ Datum pg_hll_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
Datum pg_hll_union_get_result(PG_FUNCTION_ARGS) {
struct hll_union_state* stateptr;
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -277,14 +278,15 @@ Datum pg_hll_union_get_result(PG_FUNCTION_ARGS) {
} else {
sketchptr = hll_union_get_result(stateptr->unionptr);
}
- bytes_out = hll_sketch_serialize(sketchptr);
+ bytes_out = hll_sketch_serialize(sketchptr, VARHDRSZ);
hll_sketch_delete(sketchptr);
hll_union_delete(stateptr->unionptr);
pfree(stateptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_BYTEA_P(bytes_out);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_hll_sketch_union(PG_FUNCTION_ARGS) {
@@ -294,7 +296,7 @@ Datum pg_hll_sketch_union(PG_FUNCTION_ARGS) {
void* sketchptr2;
void* unionptr;
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
unsigned lg_k;
unsigned tgt_type;
@@ -324,7 +326,9 @@ Datum pg_hll_sketch_union(PG_FUNCTION_ARGS) {
sketchptr = hll_union_get_result(unionptr);
}
hll_union_delete(unionptr);
- bytes_out = hll_sketch_serialize(sketchptr);
+ bytes_out = hll_sketch_serialize(sketchptr, VARHDRSZ);
hll_sketch_delete(sketchptr);
- PG_RETURN_BYTEA_P(bytes_out);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
diff --git a/src/kll_float_sketch_c_adapter.cpp b/src/kll_float_sketch_c_adapter.cpp
index 5285882..c9db67c 100644
--- a/src/kll_float_sketch_c_adapter.cpp
+++ b/src/kll_float_sketch_c_adapter.cpp
@@ -19,6 +19,7 @@
#include "kll_float_sketch_c_adapter.h"
#include "allocator.h"
+#include "postgres_h_substitute.h"
#include <sstream>
@@ -30,8 +31,9 @@ void* kll_float_sketch_new(unsigned k) {
try {
return new (palloc(sizeof(kll_float_sketch))) kll_float_sketch(k);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void kll_float_sketch_delete(void* sketchptr) {
@@ -39,7 +41,7 @@ void kll_float_sketch_delete(void* sketchptr) {
static_cast<kll_float_sketch*>(sketchptr)->~kll_float_sketch();
pfree(sketchptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -47,7 +49,7 @@ void kll_float_sketch_update(void* sketchptr, float value) {
try {
static_cast<kll_float_sketch*>(sketchptr)->update(value);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -55,7 +57,7 @@ void kll_float_sketch_merge(void* sketchptr1, const void* sketchptr2) {
try {
static_cast<kll_float_sketch*>(sketchptr1)->merge(*static_cast<const kll_float_sketch*>(sketchptr2));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -63,62 +65,68 @@ double kll_float_sketch_get_rank(const void* sketchptr, float value) {
try {
return static_cast<const kll_float_sketch*>(sketchptr)->get_rank(value);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
float kll_float_sketch_get_quantile(const void* sketchptr, double rank) {
try {
return static_cast<const kll_float_sketch*>(sketchptr)->get_quantile(rank);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
-uint64 kll_float_sketch_get_n(const void* sketchptr) {
+unsigned long long kll_float_sketch_get_n(const void* sketchptr) {
try {
return static_cast<const kll_float_sketch*>(sketchptr)->get_n();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void kll_float_sketch_to_string(const void* sketchptr, char* buffer, unsigned length) {
try {
std::stringstream s;
static_cast<const kll_float_sketch*>(sketchptr)->to_stream(s);
- snprintf(buffer, length, s.str().c_str());
+ snprintf(buffer, length, "%s", s.str().c_str());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
-void* kll_float_sketch_serialize(const void* sketchptr) {
+ptr_with_size kll_float_sketch_serialize(const void* sketchptr, unsigned header_size) {
try {
- auto data = static_cast<const kll_float_sketch*>(sketchptr)->serialize(VARHDRSZ);
- bytea* buffer = (bytea*) data.first.release();
- const size_t length = data.second;
- SET_VARSIZE(buffer, length);
- return buffer;
+ ptr_with_size p;
+ auto data = static_cast<const kll_float_sketch*>(sketchptr)->serialize(header_size);
+ p.ptr = data.first.release();
+ p.size = data.second;
+ return p;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* kll_float_sketch_deserialize(const char* buffer, unsigned length) {
try {
return new (palloc(sizeof(kll_float_sketch))) kll_float_sketch(kll_float_sketch::deserialize(buffer, length));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
unsigned kll_float_sketch_get_serialized_size_bytes(const void* sketchptr) {
try {
return static_cast<const kll_float_sketch*>(sketchptr)->get_serialized_size_bytes();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
Datum* kll_float_sketch_get_pmf(const void* sketchptr, const float* split_points, unsigned num_split_points) {
@@ -126,12 +134,13 @@ Datum* kll_float_sketch_get_pmf(const void* sketchptr, const float* split_points
auto ptr = static_cast<const kll_float_sketch*>(sketchptr)->get_PMF(split_points, num_split_points);
Datum* pmf = (Datum*) palloc(sizeof(Datum) * (num_split_points + 1));
for (unsigned i = 0; i < num_split_points + 1; i++) {
- pmf[i] = Float8GetDatum(ptr[i]);
+ pmf[i] = pg_float8_get_datum(ptr[i]);
}
return pmf;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
Datum* kll_float_sketch_get_quantiles(const void* sketchptr, const double* fractions, unsigned num_fractions) {
@@ -139,10 +148,11 @@ Datum* kll_float_sketch_get_quantiles(const void* sketchptr, const double* fract
auto ptr = static_cast<const kll_float_sketch*>(sketchptr)->get_quantiles(fractions, num_fractions);
Datum* quantiles = (Datum*) palloc(sizeof(Datum) * num_fractions);
for (unsigned i = 0; i < num_fractions; i++) {
- quantiles[i] = Float4GetDatum(ptr[i]);
+ quantiles[i] = pg_float4_get_datum(ptr[i]);
}
return quantiles;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
diff --git a/src/kll_float_sketch_c_adapter.h b/src/kll_float_sketch_c_adapter.h
index eb573a5..9aefc91 100644
--- a/src/kll_float_sketch_c_adapter.h
+++ b/src/kll_float_sketch_c_adapter.h
@@ -24,8 +24,6 @@
extern "C" {
#endif
-#include <postgres.h>
-
void* kll_float_sketch_new(unsigned k);
void kll_float_sketch_delete(void* sketchptr);
@@ -33,15 +31,20 @@ void kll_float_sketch_update(void* sketchptr, float value);
void kll_float_sketch_merge(void* sketchptr1, const void* sketchptr2);
double kll_float_sketch_get_rank(const void* sketchptr, float value);
float kll_float_sketch_get_quantile(const void* sketchptr, double rank);
-uint64 kll_float_sketch_get_n(const void* sketchptr);
+unsigned long long kll_float_sketch_get_n(const void* sketchptr);
void kll_float_sketch_to_string(const void* sketchptr, char* buffer, unsigned length);
-void* kll_float_sketch_serialize(const void* sketchptr);
+struct ptr_with_size {
+ void* ptr;
+ unsigned long long size;
+};
+
+struct ptr_with_size kll_float_sketch_serialize(const void* sketchptr, unsigned header_size);
void* kll_float_sketch_deserialize(const char* buffer, unsigned length);
unsigned kll_float_sketch_get_serialized_size_bytes(const void* sketchptr);
-Datum* kll_float_sketch_get_pmf(const void* sketchptr, const float* split_points, unsigned num_split_points);
-Datum* kll_float_sketch_get_quantiles(const void* sketchptr, const double* fractions, unsigned num_fractions);
+void** kll_float_sketch_get_pmf(const void* sketchptr, const float* split_points, unsigned num_split_points);
+void** kll_float_sketch_get_quantiles(const void* sketchptr, const double* fractions, unsigned num_fractions);
#ifdef __cplusplus
}
diff --git a/src/kll_float_sketch_pg_functions.c b/src/kll_float_sketch_pg_functions.c
index 5ed3f4d..3284bb4 100644
--- a/src/kll_float_sketch_pg_functions.c
+++ b/src/kll_float_sketch_pg_functions.c
@@ -174,16 +174,18 @@ Datum pg_kll_float_sketch_merge(PG_FUNCTION_ARGS) {
Datum pg_kll_float_sketch_from_internal(PG_FUNCTION_ARGS) {
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
MemoryContext aggcontext;
+
if (PG_ARGISNULL(0)) PG_RETURN_NULL();
if (!AggCheckCallContext(fcinfo, &aggcontext)) {
elog(ERROR, "kll_float_sketch_from_internal called in non-aggregate context");
}
sketchptr = PG_GETARG_POINTER(0);
- bytes_out = kll_float_sketch_serialize(sketchptr);
+ bytes_out = kll_float_sketch_serialize(sketchptr, VARHDRSZ);
kll_float_sketch_delete(sketchptr);
- PG_RETURN_BYTEA_P(bytes_out);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_kll_float_sketch_get_pmf(PG_FUNCTION_ARGS) {
@@ -223,7 +225,7 @@ Datum pg_kll_float_sketch_get_pmf(PG_FUNCTION_ARGS) {
for (i = 0; i < arr_len_in; i++) {
split_points[i] = DatumGetFloat4(data_in[i]);
}
- pmf = kll_float_sketch_get_pmf(sketchptr, split_points, arr_len_in);
+ pmf = (Datum*) kll_float_sketch_get_pmf(sketchptr, split_points, arr_len_in);
pfree(split_points);
// construct output array of fractions
@@ -272,7 +274,7 @@ Datum pg_kll_float_sketch_get_quantiles(PG_FUNCTION_ARGS) {
for (i = 0; i < arr_len; i++) {
fractions[i] = DatumGetFloat8(data_in[i]);
}
- quantiles = kll_float_sketch_get_quantiles(sketchptr, fractions, arr_len);
+ quantiles = (Datum*) kll_float_sketch_get_quantiles(sketchptr, fractions, arr_len);
pfree(fractions);
// construct output array of quantiles
diff --git a/src/postgres_h_substitute.h b/src/postgres_h_substitute.h
new file mode 100644
index 0000000..b3aa320
--- /dev/null
+++ b/src/postgres_h_substitute.h
@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+
+// There is some problem compiling C++ code using GCC 4.8.5 (standard on current RHEL)
+// with postgres.h included. This is to avoid including postgres.h
+
+#ifndef POSTGRES_H_SUBSTITUTE
+#define POSTGRES_H_SUBSTITUTE
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef void* Datum;
+extern Datum pg_float4_get_datum(float x);
+extern Datum pg_float8_get_datum(double x);
+
+extern void pg_error(const char* message);
+
+// from postgresql c.h to hint compiler and avoid warnings
+#if defined(HAVE__BUILTIN_UNREACHABLE) && !defined(USE_ASSERT_CHECKING)
+#define pg_unreachable() __builtin_unreachable()
+#elif defined(_MSC_VER) && !defined(USE_ASSERT_CHECKING)
+#define pg_unreachable() __assume(0)
+#else
+#define pg_unreachable() abort()
+#endif
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/src/theta_sketch_c_adapter.cpp b/src/theta_sketch_c_adapter.cpp
index 120dc62..4056759 100644
--- a/src/theta_sketch_c_adapter.cpp
+++ b/src/theta_sketch_c_adapter.cpp
@@ -19,10 +19,7 @@
#include "theta_sketch_c_adapter.h"
#include "allocator.h"
-
-extern "C" {
-#include <postgres.h>
-}
+#include "postgres_h_substitute.h"
#include <sstream>
@@ -42,24 +39,27 @@ void* theta_sketch_new_default() {
try {
return new (palloc(sizeof(update_theta_sketch_pg))) update_theta_sketch_pg(update_theta_sketch_pg::builder().build());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* theta_sketch_new_lgk(unsigned lg_k) {
try {
return new (palloc(sizeof(update_theta_sketch_pg))) update_theta_sketch_pg(update_theta_sketch_pg::builder().set_lg_k(lg_k).build());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* theta_sketch_new_lgk_p(unsigned lg_k, float p) {
try {
return new (palloc(sizeof(update_theta_sketch_pg))) update_theta_sketch_pg(update_theta_sketch_pg::builder().set_lg_k(lg_k).set_p(p).build());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void theta_sketch_delete(void* sketchptr) {
@@ -67,7 +67,7 @@ void theta_sketch_delete(void* sketchptr) {
static_cast<theta_sketch_pg*>(sketchptr)->~theta_sketch_pg();
pfree(sketchptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -75,7 +75,7 @@ void theta_sketch_update(void* sketchptr, const void* data, unsigned length) {
try {
static_cast<update_theta_sketch_pg*>(sketchptr)->update(data, length);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -86,50 +86,54 @@ void* theta_sketch_compact(void* sketchptr) {
pfree(sketchptr);
return newptr;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
double theta_sketch_get_estimate(const void* sketchptr) {
try {
return static_cast<const theta_sketch_pg*>(sketchptr)->get_estimate();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
Datum* theta_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs) {
try {
Datum* est_and_bounds = (Datum*) palloc(sizeof(Datum) * 3);
- est_and_bounds[0] = Float8GetDatum(static_cast<const theta_sketch_pg*>(sketchptr)->get_estimate());
- est_and_bounds[1] = Float8GetDatum(static_cast<const theta_sketch_pg*>(sketchptr)->get_lower_bound(num_std_devs));
- est_and_bounds[2] = Float8GetDatum(static_cast<const theta_sketch_pg*>(sketchptr)->get_upper_bound(num_std_devs));
+ est_and_bounds[0] = pg_float8_get_datum(static_cast<const theta_sketch_pg*>(sketchptr)->get_estimate());
+ est_and_bounds[1] = pg_float8_get_datum(static_cast<const theta_sketch_pg*>(sketchptr)->get_lower_bound(num_std_devs));
+ est_and_bounds[2] = pg_float8_get_datum(static_cast<const theta_sketch_pg*>(sketchptr)->get_upper_bound(num_std_devs));
return est_and_bounds;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void theta_sketch_to_string(const void* sketchptr, char* buffer, unsigned length) {
try {
std::stringstream s;
static_cast<const theta_sketch_pg*>(sketchptr)->to_stream(s);
- snprintf(buffer, length, s.str().c_str());
+ snprintf(buffer, length, "%s", s.str().c_str());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
-void* theta_sketch_serialize(const void* sketchptr) {
+ptr_with_size theta_sketch_serialize(const void* sketchptr, unsigned header_size) {
try {
- auto data = static_cast<const theta_sketch_pg*>(sketchptr)->serialize(VARHDRSZ);
- bytea* buffer = (bytea*) data.first.release();
- const size_t length = data.second;
- SET_VARSIZE(buffer, length);
- return buffer;
+ ptr_with_size p;
+ auto data = static_cast<const theta_sketch_pg*>(sketchptr)->serialize(header_size);
+ p.ptr = data.first.release();
+ p.size = data.second;
+ return p;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* theta_sketch_deserialize(const char* buffer, unsigned length) {
@@ -137,24 +141,27 @@ void* theta_sketch_deserialize(const char* buffer, unsigned length) {
auto ptr = theta_sketch_pg::deserialize(buffer, length);
return ptr.release();
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* theta_union_new_default() {
try {
return new (palloc(sizeof(theta_union_pg))) theta_union_pg(theta_union_pg::builder().build());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* theta_union_new(unsigned lg_k) {
try {
return new (palloc(sizeof(theta_union_pg))) theta_union_pg(theta_union_pg::builder().set_lg_k(lg_k).build());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void theta_union_delete(void* unionptr) {
@@ -162,7 +169,7 @@ void theta_union_delete(void* unionptr) {
static_cast<theta_union_pg*>(unionptr)->~theta_union_pg();
pfree(unionptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -170,7 +177,7 @@ void theta_union_update(void* unionptr, const void* sketchptr) {
try {
static_cast<theta_union_pg*>(unionptr)->update(*static_cast<const theta_sketch_pg*>(sketchptr));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -178,16 +185,18 @@ void* theta_union_get_result(const void* unionptr) {
try {
return new (palloc(sizeof(compact_theta_sketch_pg))) compact_theta_sketch_pg(static_cast<const theta_union_pg*>(unionptr)->get_result());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* theta_intersection_new_default() {
try {
return new (palloc(sizeof(theta_intersection_pg))) theta_intersection_pg;
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void theta_intersection_delete(void* interptr) {
@@ -195,7 +204,7 @@ void theta_intersection_delete(void* interptr) {
static_cast<theta_intersection_pg*>(interptr)->~theta_intersection_pg();
pfree(interptr);
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -203,7 +212,7 @@ void theta_intersection_update(void* interptr, const void* sketchptr) {
try {
static_cast<theta_intersection_pg*>(interptr)->update(*static_cast<const theta_sketch_pg*>(sketchptr));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
}
@@ -211,8 +220,9 @@ void* theta_intersection_get_result(const void* interptr) {
try {
return new (palloc(sizeof(compact_theta_sketch_pg))) compact_theta_sketch_pg(static_cast<const theta_intersection_pg*>(interptr)->get_result());
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
void* theta_a_not_b(const void* sketchptr1, const void* sketchptr2) {
@@ -223,6 +233,7 @@ void* theta_a_not_b(const void* sketchptr1, const void* sketchptr2) {
*static_cast<const theta_sketch_pg*>(sketchptr2)
));
} catch (std::exception& e) {
- elog(ERROR, e.what());
+ pg_error(e.what());
}
+ pg_unreachable();
}
diff --git a/src/theta_sketch_c_adapter.h b/src/theta_sketch_c_adapter.h
index 1cc1d5b..4a411d9 100644
--- a/src/theta_sketch_c_adapter.h
+++ b/src/theta_sketch_c_adapter.h
@@ -24,8 +24,6 @@
extern "C" {
#endif
-#include <postgres.h>
-
void* theta_sketch_new_default();
void* theta_sketch_new_lgk(unsigned lg_k);
void* theta_sketch_new_lgk_p(unsigned lg_k, float p);
@@ -35,10 +33,15 @@ void theta_sketch_update(void* sketchptr, const void* data, unsigned length);
void* theta_sketch_compact(void* sketchptr);
void theta_sketch_union(void* sketchptr1, const void* sketchptr2);
double theta_sketch_get_estimate(const void* sketchptr);
-Datum* theta_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
+void** theta_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
void theta_sketch_to_string(const void* sketchptr, char* buffer, unsigned length);
-void* theta_sketch_serialize(const void* sketchptr);
+struct ptr_with_size {
+ void* ptr;
+ unsigned long long size;
+};
+
+struct ptr_with_size theta_sketch_serialize(const void* sketchptr, unsigned header_size);
void* theta_sketch_deserialize(const char* buffer, unsigned length);
void* theta_union_new_default();
diff --git a/src/theta_sketch_pg_functions.c b/src/theta_sketch_pg_functions.c
index feb2583..ff99d7b 100644
--- a/src/theta_sketch_pg_functions.c
+++ b/src/theta_sketch_pg_functions.c
@@ -139,7 +139,7 @@ Datum pg_theta_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS) {
sketchptr = theta_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
num_std_devs = PG_GETARG_INT32(1);
if (num_std_devs == 0) num_std_devs = 1; // default
- est_and_bounds = theta_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+ est_and_bounds = (Datum*) theta_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
theta_sketch_delete(sketchptr);
// construct output array
@@ -198,7 +198,7 @@ Datum pg_theta_sketch_union_agg(PG_FUNCTION_ARGS) {
Datum pg_theta_sketch_from_internal(PG_FUNCTION_ARGS) {
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -212,12 +212,13 @@ Datum pg_theta_sketch_from_internal(PG_FUNCTION_ARGS) {
sketchptr = PG_GETARG_POINTER(0);
sketchptr = theta_sketch_compact(sketchptr);
- bytes_out = theta_sketch_serialize(sketchptr);
+ bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
theta_sketch_delete(sketchptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_BYTEA_P(bytes_out);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_theta_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
@@ -246,7 +247,7 @@ Datum pg_theta_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
Datum pg_theta_union_get_result(PG_FUNCTION_ARGS) {
void* unionptr;
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
MemoryContext oldcontext;
MemoryContext aggcontext;
@@ -260,13 +261,14 @@ Datum pg_theta_union_get_result(PG_FUNCTION_ARGS) {
unionptr = PG_GETARG_POINTER(0);
sketchptr = theta_union_get_result(unionptr);
- bytes_out = theta_sketch_serialize(sketchptr);
+ bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
theta_sketch_delete(sketchptr);
theta_union_delete(unionptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
MemoryContextSwitchTo(oldcontext);
- PG_RETURN_BYTEA_P(bytes_out);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_theta_sketch_union(PG_FUNCTION_ARGS) {
@@ -276,7 +278,7 @@ Datum pg_theta_sketch_union(PG_FUNCTION_ARGS) {
void* sketchptr2;
void* unionptr;
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
int lg_k;
lg_k = PG_GETARG_INT32(2);
@@ -295,9 +297,10 @@ Datum pg_theta_sketch_union(PG_FUNCTION_ARGS) {
}
sketchptr = theta_union_get_result(unionptr);
theta_union_delete(unionptr);
- bytes_out = theta_sketch_serialize(sketchptr);
+ bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
theta_sketch_delete(sketchptr);
- PG_RETURN_BYTEA_P(bytes_out);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_theta_sketch_intersection(PG_FUNCTION_ARGS) {
@@ -307,7 +310,7 @@ Datum pg_theta_sketch_intersection(PG_FUNCTION_ARGS) {
void* sketchptr2;
void* interptr;
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
interptr = theta_intersection_new_default();
if (!PG_ARGISNULL(0)) {
@@ -324,9 +327,10 @@ Datum pg_theta_sketch_intersection(PG_FUNCTION_ARGS) {
}
sketchptr = theta_intersection_get_result(interptr);
theta_intersection_delete(interptr);
- bytes_out = theta_sketch_serialize(sketchptr);
+ bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
theta_sketch_delete(sketchptr);
- PG_RETURN_BYTEA_P(bytes_out);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
Datum pg_theta_sketch_a_not_b(PG_FUNCTION_ARGS) {
@@ -335,7 +339,7 @@ Datum pg_theta_sketch_a_not_b(PG_FUNCTION_ARGS) {
void* sketchptr1;
void* sketchptr2;
void* sketchptr;
- bytea* bytes_out;
+ struct ptr_with_size bytes_out;
if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
elog(ERROR, "theta_a_not_b expects two valid theta sketches");
@@ -348,7 +352,8 @@ Datum pg_theta_sketch_a_not_b(PG_FUNCTION_ARGS) {
sketchptr = theta_a_not_b(sketchptr1, sketchptr2);
theta_sketch_delete(sketchptr1);
theta_sketch_delete(sketchptr2);
- bytes_out = theta_sketch_serialize(sketchptr);
+ bytes_out = theta_sketch_serialize(sketchptr, VARHDRSZ);
theta_sketch_delete(sketchptr);
- PG_RETURN_BYTEA_P(bytes_out);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org