req_float_sketch
diff --git a/Makefile b/Makefile
index 41bf70c..a4b8555 100644
--- a/Makefile
+++ b/Makefile
@@ -24,7 +24,8 @@
sql/datasketches_theta_sketch.sql \
sql/datasketches_frequent_strings_sketch.sql \
sql/datasketches_hll_sketch.sql \
- sql/datasketches_aod_sketch.sql
+ sql/datasketches_aod_sketch.sql \
+ sql/datasketches_req_float_sketch.sql
SQL_INSTALL = sql/$(EXTENSION)--$(EXTVERSION).sql
DATA = $(SQL_INSTALL)
@@ -36,7 +37,8 @@
src/theta_sketch_pg_functions.o src/theta_sketch_c_adapter.o \
src/frequent_strings_sketch_pg_functions.o src/frequent_strings_sketch_c_adapter.o \
src/hll_sketch_pg_functions.o src/hll_sketch_c_adapter.o \
- src/aod_sketch_pg_functions.o src/aod_sketch_c_adapter.o
+ src/aod_sketch_pg_functions.o src/aod_sketch_c_adapter.o \
+ src/req_float_sketch_pg_functions.o src/req_float_sketch_c_adapter.o
# assume a dir or link named "datasketches-cpp" in the current dir
CORE = datasketches-cpp
@@ -44,7 +46,14 @@
# assume a dir or link named "boost" in the current dir
BOOST = boost
-PG_CPPFLAGS = -fPIC -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 -I$(CORE)/tuple/include -I$(BOOST)
+PG_CPPFLAGS = -fPIC -I/usr/local/include -I$(BOOST) -I$(CORE)/common/include \
+ -I$(CORE)/kll/include \
+ -I$(CORE)/cpc/include \
+ -I$(CORE)/theta/include \
+ -I$(CORE)/fi/include \
+ -I$(CORE)/hll/include \
+ -I$(CORE)/tuple/include \
+ -I$(CORE)/req/include
PG_CXXFLAGS = -std=c++11
SHLIB_LINK = -lstdc++ -L/usr/local/lib
diff --git a/sql/datasketches_req_float_sketch.sql b/sql/datasketches_req_float_sketch.sql
new file mode 100644
index 0000000..c6fde9d
--- /dev/null
+++ b/sql/datasketches_req_float_sketch.sql
@@ -0,0 +1,135 @@
+-- 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.
+
+CREATE TYPE req_float_sketch;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_in(cstring) RETURNS req_float_sketch
+ AS '$libdir/datasketches', 'pg_sketch_in'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_out(req_float_sketch) RETURNS cstring
+ AS '$libdir/datasketches', 'pg_sketch_out'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE TYPE req_float_sketch (
+ INPUT = req_float_sketch_in,
+ OUTPUT = req_float_sketch_out,
+ STORAGE = EXTERNAL
+);
+
+CREATE CAST (bytea as req_float_sketch) WITHOUT FUNCTION AS ASSIGNMENT;
+CREATE CAST (req_float_sketch as bytea) WITHOUT FUNCTION AS ASSIGNMENT;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_add_item(internal, real) RETURNS internal
+ AS '$libdir/datasketches', 'pg_req_float_sketch_add_item'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_add_item(internal, real, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_req_float_sketch_add_item'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_add_item(internal, real, int, boolean) RETURNS internal
+ AS '$libdir/datasketches', 'pg_req_float_sketch_add_item'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_rank(req_float_sketch, real) RETURNS double precision
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_rank'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_quantile(req_float_sketch, double precision) RETURNS real
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_quantile'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_n(req_float_sketch) RETURNS bigint
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_n'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_to_string(req_float_sketch) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_req_float_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_merge(internal, req_float_sketch) RETURNS internal
+ AS '$libdir/datasketches', 'pg_req_float_sketch_merge'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_merge(internal, req_float_sketch, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_req_float_sketch_merge'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_merge(internal, req_float_sketch, int, boolean) RETURNS internal
+ AS '$libdir/datasketches', 'pg_req_float_sketch_merge'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_from_internal(internal) RETURNS req_float_sketch
+ AS '$libdir/datasketches', 'pg_req_float_sketch_from_internal'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE AGGREGATE req_float_sketch_build(real) (
+ sfunc = req_float_sketch_add_item,
+ stype = internal,
+ finalfunc = req_float_sketch_from_internal
+);
+
+CREATE AGGREGATE req_float_sketch_build(real, int) (
+ sfunc = req_float_sketch_add_item,
+ stype = internal,
+ finalfunc = req_float_sketch_from_internal
+);
+
+CREATE AGGREGATE req_float_sketch_build(real, int, boolean) (
+ sfunc = req_float_sketch_add_item,
+ stype = internal,
+ finalfunc = req_float_sketch_from_internal
+);
+
+CREATE AGGREGATE req_float_sketch_merge(req_float_sketch) (
+ sfunc = req_float_sketch_merge,
+ stype = internal,
+ finalfunc = req_float_sketch_from_internal
+);
+
+CREATE AGGREGATE req_float_sketch_merge(req_float_sketch, int) (
+ sfunc = req_float_sketch_merge,
+ stype = internal,
+ finalfunc = req_float_sketch_from_internal
+);
+
+CREATE AGGREGATE req_float_sketch_merge(req_float_sketch, int, boolean) (
+ sfunc = req_float_sketch_merge,
+ stype = internal,
+ finalfunc = req_float_sketch_from_internal
+);
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_pmf(req_float_sketch, real[]) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_pmf'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_cdf(req_float_sketch, real[]) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_cdf'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_quantiles(req_float_sketch, double precision[]) RETURNS real[]
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_quantiles'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_histogram(req_float_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_histogram'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION req_float_sketch_get_histogram(req_float_sketch, int) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_req_float_sketch_get_histogram'
+ LANGUAGE C STRICT IMMUTABLE;
diff --git a/src/req_float_sketch_c_adapter.cpp b/src/req_float_sketch_c_adapter.cpp
new file mode 100644
index 0000000..fe56e06
--- /dev/null
+++ b/src/req_float_sketch_c_adapter.cpp
@@ -0,0 +1,168 @@
+/*
+ * 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.
+ */
+
+#include "req_float_sketch_c_adapter.h"
+#include "allocator.h"
+#include "postgres_h_substitute.h"
+
+#include <req_sketch.hpp>
+
+using req_float_sketch = datasketches::req_sketch<float, std::less<float>, datasketches::serde<float>, palloc_allocator<float>>;
+
+void* req_float_sketch_new(unsigned k, bool hra) {
+ try {
+ return new (palloc(sizeof(req_float_sketch))) req_float_sketch(k, hra);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void req_float_sketch_delete(void* sketchptr) {
+ try {
+ static_cast<req_float_sketch*>(sketchptr)->~req_float_sketch();
+ pfree(sketchptr);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void req_float_sketch_update(void* sketchptr, float value) {
+ try {
+ static_cast<req_float_sketch*>(sketchptr)->update(value);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void req_float_sketch_merge(void* sketchptr1, void* sketchptr2) {
+ try {
+ static_cast<req_float_sketch*>(sketchptr1)->merge(std::move(*static_cast<req_float_sketch*>(sketchptr2)));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+double req_float_sketch_get_rank(const void* sketchptr, float value) {
+ try {
+ return static_cast<const req_float_sketch*>(sketchptr)->get_rank(value);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+float req_float_sketch_get_quantile(const void* sketchptr, double rank) {
+ try {
+ return static_cast<const req_float_sketch*>(sketchptr)->get_quantile(rank);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+unsigned long long req_float_sketch_get_n(const void* sketchptr) {
+ try {
+ return static_cast<const req_float_sketch*>(sketchptr)->get_n();
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+char* req_float_sketch_to_string(const void* sketchptr) {
+ try {
+ auto str = static_cast<const req_float_sketch*>(sketchptr)->to_string(true, true);
+ const size_t len = str.length() + 1;
+ char* buffer = (char*) palloc(len);
+ strncpy(buffer, str.c_str(), len);
+ return buffer;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+ptr_with_size req_float_sketch_serialize(const void* sketchptr, unsigned header_size) {
+ try {
+ ptr_with_size p;
+ auto bytes = new (palloc(sizeof(req_float_sketch::vector_bytes))) req_float_sketch::vector_bytes(
+ static_cast<const req_float_sketch*>(sketchptr)->serialize(header_size)
+ );
+ p.ptr = bytes->data();
+ p.size = bytes->size();
+ return p;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* req_float_sketch_deserialize(const char* buffer, unsigned length) {
+ try {
+ return new (palloc(sizeof(req_float_sketch))) req_float_sketch(req_float_sketch::deserialize(buffer, length));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+unsigned req_float_sketch_get_serialized_size_bytes(const void* sketchptr) {
+ try {
+ return static_cast<const req_float_sketch*>(sketchptr)->get_serialized_size_bytes();
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+Datum* req_float_sketch_get_pmf_or_cdf(const void* sketchptr, const float* split_points, unsigned num_split_points, bool is_cdf, bool scale) {
+ try {
+ auto array = is_cdf ?
+ static_cast<const req_float_sketch*>(sketchptr)->get_CDF(split_points, num_split_points) :
+ static_cast<const req_float_sketch*>(sketchptr)->get_PMF(split_points, num_split_points);
+ Datum* pmf = (Datum*) palloc(sizeof(Datum) * (num_split_points + 1));
+ const uint64_t n = static_cast<const req_float_sketch*>(sketchptr)->get_n();
+ for (unsigned i = 0; i < num_split_points + 1; i++) {
+ if (scale) {
+ pmf[i] = pg_float8_get_datum(array[i] * n);
+ } else {
+ pmf[i] = pg_float8_get_datum(array[i]);
+ }
+ }
+ return pmf;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+Datum* req_float_sketch_get_quantiles(const void* sketchptr, const double* fractions, unsigned num_fractions) {
+ try {
+ auto array = static_cast<const req_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] = pg_float4_get_datum(array[i]);
+ }
+ return quantiles;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
diff --git a/src/req_float_sketch_c_adapter.h b/src/req_float_sketch_c_adapter.h
new file mode 100644
index 0000000..4f07a8f
--- /dev/null
+++ b/src/req_float_sketch_c_adapter.h
@@ -0,0 +1,52 @@
+/*
+ * 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 REQ_FLOAT_SKETCH_C_ADAPTER_H
+#define REQ_FLOAT_SKETCH_C_ADAPTER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "ptr_with_size.h"
+
+static const unsigned DEFAULT_K = 12;
+
+void* req_float_sketch_new(unsigned k, bool hra);
+void req_float_sketch_delete(void* sketchptr);
+
+void req_float_sketch_update(void* sketchptr, float value);
+void req_float_sketch_merge(void* sketchptr1, void* sketchptr2);
+double req_float_sketch_get_rank(const void* sketchptr, float value);
+float req_float_sketch_get_quantile(const void* sketchptr, double rank);
+unsigned long long req_float_sketch_get_n(const void* sketchptr);
+char* req_float_sketch_to_string(const void* sketchptr);
+
+struct ptr_with_size req_float_sketch_serialize(const void* sketchptr, unsigned header_size);
+void* req_float_sketch_deserialize(const char* buffer, unsigned length);
+unsigned req_float_sketch_get_serialized_size_bytes(const void* sketchptr);
+
+void** req_float_sketch_get_pmf_or_cdf(const void* sketchptr, const float* split_points, unsigned num_split_points, bool is_cdf, bool scale);
+void** req_float_sketch_get_quantiles(const void* sketchptr, const double* fractions, unsigned num_fractions);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/src/req_float_sketch_pg_functions.c b/src/req_float_sketch_pg_functions.c
new file mode 100644
index 0000000..5edeba3
--- /dev/null
+++ b/src/req_float_sketch_pg_functions.c
@@ -0,0 +1,389 @@
+/*
+ * 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.
+ */
+
+#include <postgres.h>
+#include <fmgr.h>
+#include <utils/lsyscache.h>
+#include <utils/builtins.h>
+#include <utils/array.h>
+#include <catalog/pg_type.h>
+
+#include "req_float_sketch_c_adapter.h"
+#include "base64.h"
+
+/* PG_FUNCTION_INFO_V1 macro to pass functions to postgres */
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_add_item);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_get_rank);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_get_quantile);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_get_n);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_to_string);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_merge);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_from_internal);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_get_pmf);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_get_cdf);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_get_quantiles);
+PG_FUNCTION_INFO_V1(pg_req_float_sketch_get_histogram);
+
+/* function declarations */
+Datum pg_req_float_sketch_recv(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_send(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_add_item(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_get_rank(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_get_quantile(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_get_n(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_to_string(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_merge(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_from_internal(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_get_pmf(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_get_cdf(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_get_quantiles(PG_FUNCTION_ARGS);
+Datum pg_req_float_sketch_get_histogram(PG_FUNCTION_ARGS);
+
+static const unsigned DEFAULT_NUM_BINS = 10;
+
+Datum pg_req_float_sketch_add_item(PG_FUNCTION_ARGS) {
+ void* sketchptr;
+ float value;
+ int k;
+ bool hra;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0) && PG_ARGISNULL(1)) {
+ PG_RETURN_NULL();
+ } else if (PG_ARGISNULL(1)) {
+ PG_RETURN_POINTER(PG_GETARG_POINTER(0)); // no update value. return unmodified state
+ }
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "req_float_sketch_add_item called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ if (PG_ARGISNULL(0)) {
+ k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : DEFAULT_K;
+ hra = PG_NARGS() > 3 ? PG_GETARG_BOOL(3) : true;
+ sketchptr = req_float_sketch_new(k, hra);
+ } else {
+ sketchptr = PG_GETARG_POINTER(0);
+ }
+
+ value = PG_GETARG_FLOAT4(1);
+ req_float_sketch_update(sketchptr, value);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(sketchptr);
+}
+
+Datum pg_req_float_sketch_get_rank(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ float value;
+ double rank;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ value = PG_GETARG_FLOAT4(1);
+ rank = req_float_sketch_get_rank(sketchptr, value);
+ req_float_sketch_delete(sketchptr);
+ PG_RETURN_FLOAT8(rank);
+}
+
+Datum pg_req_float_sketch_get_quantile(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ float value;
+ double rank;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ rank = PG_GETARG_FLOAT8(1);
+ value = req_float_sketch_get_quantile(sketchptr, rank);
+ req_float_sketch_delete(sketchptr);
+ PG_RETURN_FLOAT4(value);
+}
+
+Datum pg_req_float_sketch_get_n(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ uint64 n;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ n = req_float_sketch_get_n(sketchptr);
+ req_float_sketch_delete(sketchptr);
+ PG_RETURN_INT64(n);
+}
+
+Datum pg_req_float_sketch_to_string(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ char* str;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ str = req_float_sketch_to_string(sketchptr);
+ req_float_sketch_delete(sketchptr);
+ PG_RETURN_TEXT_P(cstring_to_text(str));
+}
+
+Datum pg_req_float_sketch_merge(PG_FUNCTION_ARGS) {
+ void* unionptr;
+ bytea* sketch_bytes;
+ void* sketchptr;
+ int k;
+ bool hra;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0) && PG_ARGISNULL(1)) {
+ PG_RETURN_NULL();
+ } else if (PG_ARGISNULL(1)) {
+ PG_RETURN_POINTER(PG_GETARG_POINTER(0)); // no update value. return unmodified state
+ }
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "req_float_sketch_merge called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ if (PG_ARGISNULL(0)) {
+ k = PG_NARGS() > 2 ? PG_GETARG_INT32(2) : DEFAULT_K;
+ hra = PG_NARGS() > 3 ? PG_GETARG_BOOL(3) : true;
+ unionptr = req_float_sketch_new(k ? k : DEFAULT_K, hra);
+ } else {
+ unionptr = PG_GETARG_POINTER(0);
+ }
+
+ sketch_bytes = PG_GETARG_BYTEA_P(1);
+ sketchptr = req_float_sketch_deserialize(VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
+ req_float_sketch_merge(unionptr, sketchptr);
+ req_float_sketch_delete(sketchptr);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(unionptr);
+}
+
+Datum pg_req_float_sketch_from_internal(PG_FUNCTION_ARGS) {
+ void* sketchptr;
+ struct ptr_with_size bytes_out;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "req_float_sketch_from_internal called in non-aggregate context");
+ }
+ sketchptr = PG_GETARG_POINTER(0);
+ bytes_out = req_float_sketch_serialize(sketchptr, VARHDRSZ);
+ req_float_sketch_delete(sketchptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_req_float_sketch_get_pmf(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+
+ // input array of split points
+ ArrayType* arr_in;
+ Oid elmtype_in;
+ int16 elmlen_in;
+ bool elmbyval_in;
+ char elmalign_in;
+ Datum* data_in;
+ bool* nulls_in;
+ int arr_len_in;
+ float* split_points;
+
+ // output array of fractions
+ Datum* result;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+ int arr_len_out;
+
+ int i;
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+
+ arr_in = PG_GETARG_ARRAYTYPE_P(1);
+ elmtype_in = ARR_ELEMTYPE(arr_in);
+ get_typlenbyvalalign(elmtype_in, &elmlen_in, &elmbyval_in, &elmalign_in);
+ deconstruct_array(arr_in, elmtype_in, elmlen_in, elmbyval_in, elmalign_in, &data_in, &nulls_in, &arr_len_in);
+
+ split_points = palloc(sizeof(float) * arr_len_in);
+ for (i = 0; i < arr_len_in; i++) {
+ split_points[i] = DatumGetFloat4(data_in[i]);
+ }
+ result = (Datum*) req_float_sketch_get_pmf_or_cdf(sketchptr, split_points, arr_len_in, false, false);
+ pfree(split_points);
+
+ // construct output array of fractions
+ arr_len_out = arr_len_in + 1; // N split points devide the number line into N+1 intervals
+ get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(result, arr_len_out, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
+
+ req_float_sketch_delete(sketchptr);
+
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
+
+Datum pg_req_float_sketch_get_cdf(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+
+ // input array of split points
+ ArrayType* arr_in;
+ Oid elmtype_in;
+ int16 elmlen_in;
+ bool elmbyval_in;
+ char elmalign_in;
+ Datum* data_in;
+ bool* nulls_in;
+ int arr_len_in;
+ float* split_points;
+
+ // output array of fractions
+ Datum* result;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+ int arr_len_out;
+
+ int i;
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+
+ arr_in = PG_GETARG_ARRAYTYPE_P(1);
+ elmtype_in = ARR_ELEMTYPE(arr_in);
+ get_typlenbyvalalign(elmtype_in, &elmlen_in, &elmbyval_in, &elmalign_in);
+ deconstruct_array(arr_in, elmtype_in, elmlen_in, elmbyval_in, elmalign_in, &data_in, &nulls_in, &arr_len_in);
+
+ split_points = palloc(sizeof(float) * arr_len_in);
+ for (i = 0; i < arr_len_in; i++) {
+ split_points[i] = DatumGetFloat4(data_in[i]);
+ }
+ result = (Datum*) req_float_sketch_get_pmf_or_cdf(sketchptr, split_points, arr_len_in, true, false);
+ pfree(split_points);
+
+ // construct output array of fractions
+ arr_len_out = arr_len_in + 1; // N split points devide the number line into N+1 intervals
+ get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(result, arr_len_out, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
+
+ req_float_sketch_delete(sketchptr);
+
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
+
+Datum pg_req_float_sketch_get_quantiles(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+
+ // input array of fractions
+ ArrayType* arr_in;
+ Oid elmtype_in;
+ int16 elmlen_in;
+ bool elmbyval_in;
+ char elmalign_in;
+ Datum* data_in;
+ bool* nulls_in;
+ int arr_len;
+ double* fractions;
+
+ // output array of quantiles
+ Datum* quantiles;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+
+ int i;
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+
+ arr_in = PG_GETARG_ARRAYTYPE_P(1);
+ elmtype_in = ARR_ELEMTYPE(arr_in);
+ get_typlenbyvalalign(elmtype_in, &elmlen_in, &elmbyval_in, &elmalign_in);
+ deconstruct_array(arr_in, elmtype_in, elmlen_in, elmbyval_in, elmalign_in, &data_in, &nulls_in, &arr_len);
+
+ fractions = palloc(sizeof(double) * arr_len);
+ for (i = 0; i < arr_len; i++) {
+ fractions[i] = DatumGetFloat8(data_in[i]);
+ }
+ quantiles = (Datum*) req_float_sketch_get_quantiles(sketchptr, fractions, arr_len);
+ pfree(fractions);
+
+ // construct output array of quantiles
+ get_typlenbyvalalign(FLOAT4OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(quantiles, arr_len, FLOAT4OID, elmlen_out, elmbyval_out, elmalign_out);
+
+ req_float_sketch_delete(sketchptr);
+
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
+
+Datum pg_req_float_sketch_get_histogram(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ int num_bins;
+
+ // output array of bins
+ Datum* result;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+ int arr_len_out;
+
+ int i;
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = req_float_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+
+ num_bins = PG_GETARG_INT32(1);
+ if (num_bins == 0) num_bins = DEFAULT_NUM_BINS;
+ if (num_bins < 2) {
+ elog(ERROR, "at least two bins expected");
+ }
+
+ float* split_points = palloc(sizeof(float) * (num_bins - 1));
+ const float min_value = req_float_sketch_get_quantile(sketchptr, 0);
+ const float max_value = req_float_sketch_get_quantile(sketchptr, 1);
+ const float delta = (max_value - min_value) / num_bins;
+ for (i = 0; i < num_bins - 1; i++) {
+ split_points[i] = min_value + delta * (i + 1);
+ }
+ result = (Datum*) req_float_sketch_get_pmf_or_cdf(sketchptr, split_points, num_bins - 1, false, true);
+ pfree(split_points);
+
+ // construct output array
+ arr_len_out = num_bins;
+ get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(result, arr_len_out, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
+
+ req_float_sketch_delete(sketchptr);
+
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
