Merge pull request #30 from apache/aod_sketch
array of doubles sketch
diff --git a/Makefile b/Makefile
index 06b2ce0..41bf70c 100644
--- a/Makefile
+++ b/Makefile
@@ -19,7 +19,12 @@
EXTVERSION = $(shell grep default_version $(EXTENSION).control | sed -e "s/default_version[[:space:]]*=[[:space:]]*'\([^']*\)'/\1/")
MODULE_big = datasketches
-SQL_MODULES = 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
+SQL_MODULES = 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 \
+ sql/datasketches_aod_sketch.sql
SQL_INSTALL = sql/$(EXTENSION)--$(EXTVERSION).sql
DATA = $(SQL_INSTALL)
@@ -30,12 +35,16 @@
src/cpc_sketch_pg_functions.o src/cpc_sketch_c_adapter.o \
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/hll_sketch_pg_functions.o src/hll_sketch_c_adapter.o \
+ src/aod_sketch_pg_functions.o src/aod_sketch_c_adapter.o
-# assume a copy or link datasketches-cpp in the current dir
+# assume a dir or link named "datasketches-cpp" in the current dir
CORE = datasketches-cpp
-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
+# 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_CXXFLAGS = -std=c++11
SHLIB_LINK = -lstdc++ -L/usr/local/lib
@@ -53,7 +62,7 @@
if [ "$(with_llvm)" = "yes" ]; then $(LLVM_BINPATH)/opt -module-summary -f $@ -o $@; fi
# generate combined sql
-$(SQL_INSTALL): $(sort $(SQL_MODULES))
+$(SQL_INSTALL): $(SQL_MODULES)
cat $^ > $@
install: $(SQL_INSTALL)
diff --git a/sql/datasketches_aod_sketch.sql b/sql/datasketches_aod_sketch.sql
new file mode 100644
index 0000000..b737519
--- /dev/null
+++ b/sql/datasketches_aod_sketch.sql
@@ -0,0 +1,191 @@
+-- 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 aod_sketch;
+
+CREATE OR REPLACE FUNCTION aod_sketch_in(cstring) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_sketch_in'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_out(aod_sketch) RETURNS cstring
+ AS '$libdir/datasketches', 'pg_sketch_out'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE TYPE aod_sketch (
+ INPUT = aod_sketch_in,
+ OUTPUT = aod_sketch_out,
+ STORAGE = EXTERNAL
+);
+
+CREATE CAST (bytea as aod_sketch) WITHOUT FUNCTION AS ASSIGNMENT;
+CREATE CAST (aod_sketch as bytea) WITHOUT FUNCTION AS ASSIGNMENT;
+
+CREATE OR REPLACE FUNCTION aod_sketch_add_item(internal, anyelement, double precision[]) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_add_item'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_add_item(internal, anyelement, double precision[], int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_add_item'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_add_item(internal, anyelement, double precision[], int, real) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_add_item'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_get_estimate(aod_sketch) RETURNS double precision
+ AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_get_estimate_and_bounds(aod_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_get_estimate_and_bounds(aod_sketch, int) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_aod_sketch_get_estimate_and_bounds'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_from_internal(internal) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_from_internal'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_string(aod_sketch) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_string(aod_sketch, boolean) RETURNS TEXT
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_string'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_union_agg(internal, aod_sketch) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_union_agg'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_union_agg(internal, aod_sketch, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_union_agg'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_union_agg(internal, aod_sketch, int, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_union_agg'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_intersection_agg(internal, aod_sketch) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_intersection_agg'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_intersection_agg(internal, aod_sketch, int) RETURNS internal
+ AS '$libdir/datasketches', 'pg_aod_sketch_intersection_agg'
+ LANGUAGE C IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_union_get_result(internal) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_union_get_result'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_intersection_get_result(internal) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_intersection_get_result'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE AGGREGATE aod_sketch_build(anyelement, double precision[]) (
+ sfunc = aod_sketch_add_item,
+ stype = internal,
+ finalfunc = aod_sketch_from_internal
+);
+
+CREATE AGGREGATE aod_sketch_build(anyelement, double precision[], int) (
+ sfunc = aod_sketch_add_item,
+ stype = internal,
+ finalfunc = aod_sketch_from_internal
+);
+
+CREATE AGGREGATE aod_sketch_build(anyelement, double precision[], int, real) (
+ sfunc = aod_sketch_add_item,
+ stype = internal,
+ finalfunc = aod_sketch_from_internal
+);
+
+CREATE AGGREGATE aod_sketch_union(aod_sketch) (
+ sfunc = aod_sketch_union_agg,
+ stype = internal,
+ finalfunc = aod_union_get_result
+);
+
+CREATE AGGREGATE aod_sketch_union(aod_sketch, int) (
+ sfunc = aod_sketch_union_agg,
+ stype = internal,
+ finalfunc = aod_union_get_result
+);
+
+CREATE AGGREGATE aod_sketch_union(aod_sketch, int, int) (
+ sfunc = aod_sketch_union_agg,
+ stype = internal,
+ finalfunc = aod_union_get_result
+);
+
+CREATE AGGREGATE aod_sketch_intersection(aod_sketch) (
+ sfunc = aod_sketch_intersection_agg,
+ stype = internal,
+ finalfunc = aod_intersection_get_result
+);
+
+CREATE AGGREGATE aod_sketch_intersection(aod_sketch, int) (
+ sfunc = aod_sketch_intersection_agg,
+ stype = internal,
+ finalfunc = aod_intersection_get_result
+);
+
+CREATE OR REPLACE FUNCTION aod_sketch_union(aod_sketch, aod_sketch) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_union'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_union(aod_sketch, aod_sketch, int) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_union'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_intersection(aod_sketch, aod_sketch) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_intersection'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_intersection(aod_sketch, aod_sketch, int) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_intersection'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_a_not_b(aod_sketch, aod_sketch) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_a_not_b'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_a_not_b(aod_sketch, aod_sketch, int) RETURNS aod_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_a_not_b'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_kll_float_sketch(aod_sketch, int) RETURNS kll_float_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_kll_float_sketch'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_kll_float_sketch(aod_sketch, int, int) RETURNS kll_float_sketch
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_kll_float_sketch'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_students_t_test(aod_sketch, aod_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_aod_sketch_students_t_test'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_means(aod_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_means'
+ LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION aod_sketch_to_variances(aod_sketch) RETURNS double precision[]
+ AS '$libdir/datasketches', 'pg_aod_sketch_to_variances'
+ LANGUAGE C STRICT IMMUTABLE;
diff --git a/src/allocator.h b/src/allocator.h
index 1a49f4b..06b810c 100644
--- a/src/allocator.h
+++ b/src/allocator.h
@@ -42,11 +42,9 @@
template <class U>
struct rebind { typedef palloc_allocator<U> other; };
- palloc_allocator() {}
- palloc_allocator(const palloc_allocator&) {}
+ palloc_allocator() = default;
template <class U>
palloc_allocator(const palloc_allocator<U>&) {}
- ~palloc_allocator() {}
pointer address(reference x) const { return &x; }
const_pointer address(const_reference x) const {
@@ -70,9 +68,6 @@
new(p) value_type(std::forward<Args>(args)...);
}
void destroy(pointer p) { p->~value_type(); }
-
-private:
- void operator=(const palloc_allocator&);
};
template<> class palloc_allocator<void> {
diff --git a/src/aod_sketch_c_adapter.cpp b/src/aod_sketch_c_adapter.cpp
new file mode 100644
index 0000000..7db6b8f
--- /dev/null
+++ b/src/aod_sketch_c_adapter.cpp
@@ -0,0 +1,393 @@
+/*
+ * 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 "aod_sketch_c_adapter.h"
+#include "allocator.h"
+#include "postgres_h_substitute.h"
+#include "kll_float_sketch_c_adapter.h"
+
+#include <array_of_doubles_sketch.hpp>
+#include <array_of_doubles_union.hpp>
+#include <array_of_doubles_intersection.hpp>
+#include <array_of_doubles_a_not_b.hpp>
+
+#include <boost/accumulators/accumulators.hpp>
+#include <boost/accumulators/statistics/stats.hpp>
+#include <boost/accumulators/statistics/mean.hpp>
+#include <boost/accumulators/statistics/variance.hpp>
+#include <boost/math/distributions/students_t.hpp>
+
+using update_aod_sketch_pg = datasketches::update_array_of_doubles_sketch_alloc<palloc_allocator<double>>;
+using compact_aod_sketch_pg = datasketches::compact_array_of_doubles_sketch_alloc<palloc_allocator<double>>;
+using aod_union_pg = datasketches::array_of_doubles_union_alloc<palloc_allocator<double>>;
+// using the union policy in the intersection since this is how it is done in Druid
+using aod_intersection_pg = datasketches::array_of_doubles_intersection<datasketches::array_of_doubles_union_policy_alloc<palloc_allocator<double>>, palloc_allocator<double>>;
+using aod_a_not_b_pg = datasketches::array_of_doubles_a_not_b_alloc<palloc_allocator<double>>;
+
+std::ostream& operator<<(std::ostream& os, const datasketches::aod<palloc_allocator<double>>& v) {
+ os << "(";
+ for (size_t i = 0; i < v.size(); ++i) {
+ if (i != 0) os << ", ";
+ os << v[i];
+ }
+ os << ")";
+ return os;
+}
+
+void* aod_sketch_new(unsigned num_values) {
+ try {
+ return new (palloc(sizeof(update_aod_sketch_pg))) update_aod_sketch_pg(update_aod_sketch_pg::builder(num_values).build());
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* aod_sketch_new_lgk(unsigned num_values, unsigned lg_k) {
+ try {
+ return new (palloc(sizeof(update_aod_sketch_pg))) update_aod_sketch_pg(update_aod_sketch_pg::builder(num_values).set_lg_k(lg_k).build());
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* aod_sketch_new_lgk_p(unsigned num_values, unsigned lg_k, float p) {
+ try {
+ return new (palloc(sizeof(update_aod_sketch_pg))) update_aod_sketch_pg(update_aod_sketch_pg::builder(num_values).set_lg_k(lg_k).set_p(p).build());
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void update_aod_sketch_delete(void* sketchptr) {
+ try {
+ static_cast<update_aod_sketch_pg*>(sketchptr)->~update_aod_sketch_pg();
+ pfree(sketchptr);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void compact_aod_sketch_delete(void* sketchptr) {
+ try {
+ static_cast<compact_aod_sketch_pg*>(sketchptr)->~compact_aod_sketch_pg();
+ pfree(sketchptr);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void aod_sketch_update(void* sketchptr, const void* data, unsigned length, const double* values) {
+ try {
+ static_cast<update_aod_sketch_pg*>(sketchptr)->update(data, length, values);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void* aod_sketch_compact(void* sketchptr) {
+ try {
+ auto newptr = new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(static_cast<update_aod_sketch_pg*>(sketchptr)->compact());
+ static_cast<update_aod_sketch_pg*>(sketchptr)->~update_aod_sketch_pg();
+ pfree(sketchptr);
+ return newptr;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+double update_aod_sketch_get_estimate(const void* sketchptr) {
+ try {
+ return static_cast<const update_aod_sketch_pg*>(sketchptr)->get_estimate();
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+double compact_aod_sketch_get_estimate(const void* sketchptr) {
+ try {
+ return static_cast<const compact_aod_sketch_pg*>(sketchptr)->get_estimate();
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+Datum* aod_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] = pg_float8_get_datum(static_cast<const compact_aod_sketch_pg*>(sketchptr)->get_estimate());
+ est_and_bounds[1] = pg_float8_get_datum(static_cast<const compact_aod_sketch_pg*>(sketchptr)->get_lower_bound(num_std_devs));
+ est_and_bounds[2] = pg_float8_get_datum(static_cast<const compact_aod_sketch_pg*>(sketchptr)->get_upper_bound(num_std_devs));
+ return est_and_bounds;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+char* aod_sketch_to_string(const void* sketchptr, bool print_entries) {
+ try {
+ auto str = static_cast<const compact_aod_sketch_pg*>(sketchptr)->to_string(print_entries);
+ 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 aod_sketch_serialize(const void* sketchptr, unsigned header_size) {
+ try {
+ ptr_with_size p;
+ auto bytes = new (palloc(sizeof(compact_aod_sketch_pg::vector_bytes))) compact_aod_sketch_pg::vector_bytes(
+ static_cast<const compact_aod_sketch_pg*>(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* aod_sketch_deserialize(const char* buffer, unsigned length) {
+ try {
+ return new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(compact_aod_sketch_pg::deserialize(buffer, length));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* aod_union_new(unsigned num_values) {
+ try {
+ return new (palloc(sizeof(aod_union_pg))) aod_union_pg(aod_union_pg::builder(num_values).build());
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* aod_union_new_lgk(unsigned num_values, unsigned lg_k) {
+ try {
+ return new (palloc(sizeof(aod_union_pg))) aod_union_pg(aod_union_pg::builder(num_values).set_lg_k(lg_k).build());
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void aod_union_delete(void* unionptr) {
+ try {
+ static_cast<aod_union_pg*>(unionptr)->~aod_union_pg();
+ pfree(unionptr);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void aod_union_update(void* unionptr, const void* sketchptr) {
+ try {
+ static_cast<aod_union_pg*>(unionptr)->update(std::move(*static_cast<const compact_aod_sketch_pg*>(sketchptr)));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void* aod_union_get_result(const void* unionptr) {
+ try {
+ return new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(static_cast<const aod_union_pg*>(unionptr)->get_result());
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* aod_intersection_new(unsigned num_values) {
+ try {
+ return new (palloc(sizeof(aod_intersection_pg))) aod_intersection_pg(datasketches::DEFAULT_SEED, num_values);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void aod_intersection_delete(void* interptr) {
+ try {
+ static_cast<aod_intersection_pg*>(interptr)->~aod_intersection_pg();
+ pfree(interptr);
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void aod_intersection_update(void* interptr, const void* sketchptr) {
+ try {
+ static_cast<aod_intersection_pg*>(interptr)->update(*static_cast<const compact_aod_sketch_pg*>(sketchptr));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+}
+
+void* aod_intersection_get_result(const void* interptr) {
+ try {
+ return new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(static_cast<const aod_intersection_pg*>(interptr)->get_result());
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* aod_a_not_b(const void* sketchptr1, const void* sketchptr2) {
+ try {
+ aod_a_not_b_pg a_not_b;
+ return new (palloc(sizeof(compact_aod_sketch_pg))) compact_aod_sketch_pg(a_not_b.compute(
+ *static_cast<const compact_aod_sketch_pg*>(sketchptr1),
+ *static_cast<const compact_aod_sketch_pg*>(sketchptr2)
+ ));
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+void* aod_sketch_to_kll_float_sketch(const void* sketchptr, unsigned column_index, unsigned k) {
+ try {
+ auto kllptr = kll_float_sketch_new(k);
+ for (const auto& entry: *static_cast<const compact_aod_sketch_pg*>(sketchptr)) {
+ kll_float_sketch_update(kllptr, entry.second[column_index]);
+ }
+ return kllptr;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+double t_test_unequal_sd(double m1, double v1, uint64_t n1, double m2, double v2, uint64_t n2) {
+ double degrees_of_freedom = v1 / n1 + v2 / n2;
+ degrees_of_freedom *= degrees_of_freedom;
+ double t1 = v1 / n1;
+ t1 *= t1;
+ t1 /= (n1 - 1);
+ double t2 = v2 / n2;
+ t2 *= t2;
+ t2 /= (n2 - 1);
+ degrees_of_freedom /= (t1 + t2);
+ double t_stat = (m1 - m2) / sqrt(v1 / n1 + v2 / n2);
+ using boost::math::students_t;
+ students_t distribution(degrees_of_freedom);
+ return 2 * cdf(complement(distribution, fabs(t_stat))); // double to match 2-sided test in Java (commons-math3)
+}
+
+Datum* aod_sketch_students_t_test(const void* sketchptr1, const void* sketchptr2, unsigned* arr_len_out) {
+ try {
+ const auto& sketch1 = *static_cast<const compact_aod_sketch_pg*>(sketchptr1);
+ const auto& sketch2 = *static_cast<const compact_aod_sketch_pg*>(sketchptr2);
+ if (sketch1.get_num_values() != sketch2.get_num_values()) pg_error("aod_sketch_students_t_test: number of values mismatch");
+ unsigned num_values = sketch1.get_num_values();
+ Datum* p_values = (Datum*) palloc(sizeof(Datum) * num_values);
+ *arr_len_out = num_values;
+
+ using namespace boost::accumulators;
+ using Accum = accumulator_set<double, stats<tag::mean, tag::variance>>;
+
+ std::vector<Accum, palloc_allocator<Accum>> stats1(num_values);
+ for (const auto& entry: sketch1) {
+ for (unsigned i = 0; i < num_values; ++i) stats1[i](entry.second[i]);
+ }
+
+ std::vector<Accum, palloc_allocator<Accum>> stats2(num_values);
+ for (const auto& entry: sketch2) {
+ for (unsigned i = 0; i < num_values; ++i) stats2[i](entry.second[i]);
+ }
+
+ for (unsigned i = 0; i < num_values; ++i) {
+ p_values[i] = pg_float8_get_datum(t_test_unequal_sd(
+ mean(stats1[i]), variance(stats1[i]), sketch1.get_num_retained(),
+ mean(stats2[i]), variance(stats2[i]), sketch2.get_num_retained()
+ ));
+ }
+
+ return p_values;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+Datum* aod_sketch_to_means(const void* sketchptr, unsigned* arr_len_out) {
+ try {
+ const auto& sketch = *static_cast<const compact_aod_sketch_pg*>(sketchptr);
+ unsigned num_values = sketch.get_num_values();
+ Datum* means = (Datum*) palloc(sizeof(Datum) * num_values);
+ *arr_len_out = num_values;
+
+ using namespace boost::accumulators;
+ using Accum = accumulator_set<double, stats<tag::mean>>;
+
+ std::vector<Accum, palloc_allocator<Accum>> stats(num_values);
+ for (const auto& entry: sketch) {
+ for (unsigned i = 0; i < num_values; ++i) stats[i](entry.second[i]);
+ }
+
+ for (unsigned i = 0; i < num_values; ++i) {
+ means[i] = pg_float8_get_datum(mean(stats[i]));
+ }
+
+ return means;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
+
+Datum* aod_sketch_to_variances(const void* sketchptr, unsigned* arr_len_out) {
+ try {
+ const auto& sketch = *static_cast<const compact_aod_sketch_pg*>(sketchptr);
+ unsigned num_values = sketch.get_num_values();
+ Datum* variances = (Datum*) palloc(sizeof(Datum) * num_values);
+ *arr_len_out = num_values;
+
+ using namespace boost::accumulators;
+ using Accum = accumulator_set<double, stats<tag::variance>>;
+
+ std::vector<Accum, palloc_allocator<Accum>> stats(num_values);
+ for (const auto& entry: sketch) {
+ for (unsigned i = 0; i < num_values; ++i) stats[i](entry.second[i]);
+ }
+
+ for (unsigned i = 0; i < num_values; ++i) {
+ variances[i] = pg_float8_get_datum(variance(stats[i]));
+ }
+
+ return variances;
+ } catch (std::exception& e) {
+ pg_error(e.what());
+ }
+ pg_unreachable();
+}
diff --git a/src/aod_sketch_c_adapter.h b/src/aod_sketch_c_adapter.h
new file mode 100644
index 0000000..9f36b10
--- /dev/null
+++ b/src/aod_sketch_c_adapter.h
@@ -0,0 +1,69 @@
+/*
+ * 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 AOD_SKETCH_C_ADAPTER_H
+#define AOD_SKETCH_C_ADAPTER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "ptr_with_size.h"
+
+void* aod_sketch_new(unsigned num_values);
+void* aod_sketch_new_lgk(unsigned num_values, unsigned lg_k);
+void* aod_sketch_new_lgk_p(unsigned num_values, unsigned lg_k, float p);
+void update_aod_sketch_delete(void* sketchptr);
+void compact_aod_sketch_delete(void* sketchptr);
+
+void aod_sketch_update(void* sketchptr, const void* data, unsigned length, const double* values);
+void* aod_sketch_compact(void* sketchptr);
+void aod_sketch_union(void* sketchptr1, const void* sketchptr2);
+double update_aod_sketch_get_estimate(const void* sketchptr);
+double compact_aod_sketch_get_estimate(const void* sketchptr);
+void** aod_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
+char* aod_sketch_to_string(const void* sketchptr, bool print_entries);
+
+struct ptr_with_size aod_sketch_serialize(const void* sketchptr, unsigned header_size);
+void* aod_sketch_deserialize(const char* buffer, unsigned length);
+
+void* aod_union_new(unsigned num_values);
+void* aod_union_new_lgk(unsigned num_values, unsigned lg_k);
+void aod_union_delete(void* unionptr);
+void aod_union_update(void* unionptr, const void* sketchptr);
+void* aod_union_get_result(const void* unionptr);
+
+void* aod_intersection_new(unsigned num_values);
+void aod_intersection_delete(void* interptr);
+void aod_intersection_update(void* interptr, const void* sketchptr);
+void* aod_intersection_get_result(const void* interptr);
+
+void* aod_a_not_b(const void* sketchptr1, const void* sketchptr2);
+
+void* aod_sketch_to_kll_float_sketch(const void* sketchptr, unsigned column_index, unsigned k);
+
+void** aod_sketch_students_t_test(const void* sketchptr1, const void* sketchptr2, unsigned* arr_len_out);
+void** aod_sketch_to_means(const void* sketchptr, unsigned* arr_len_out);
+void** aod_sketch_to_variances(const void* sketchptr, unsigned* arr_len_out);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/src/aod_sketch_pg_functions.c b/src/aod_sketch_pg_functions.c
new file mode 100644
index 0000000..98766ef
--- /dev/null
+++ b/src/aod_sketch_pg_functions.c
@@ -0,0 +1,587 @@
+/*
+ * 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 "aod_sketch_c_adapter.h"
+#include "base64.h"
+#include "kll_float_sketch_c_adapter.h"
+
+/* PG_FUNCTION_INFO_V1 macro to pass functions to postgres */
+PG_FUNCTION_INFO_V1(pg_aod_sketch_add_item);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate_and_bounds);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_to_string);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_union_agg);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_intersection_agg);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_from_internal);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_get_estimate_from_internal);
+PG_FUNCTION_INFO_V1(pg_aod_union_get_result);
+PG_FUNCTION_INFO_V1(pg_aod_intersection_get_result);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_union);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_intersection);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_a_not_b);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_to_kll_float_sketch);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_students_t_test);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_to_means);
+PG_FUNCTION_INFO_V1(pg_aod_sketch_to_variances);
+
+/* function declarations */
+Datum pg_aod_sketch_recv(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_send(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_add_item(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_get_estimate(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_to_string(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_union_agg(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_intersection_agg(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_from_internal(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS);
+Datum pg_aod_union_get_result(PG_FUNCTION_ARGS);
+Datum pg_aod_intersection_get_result(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_union(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_intersection(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_a_not_b(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_to_kll_float_sketch(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_students_t_test(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_to_means(PG_FUNCTION_ARGS);
+Datum pg_aod_sketch_to_variances(PG_FUNCTION_ARGS);
+
+Datum pg_aod_sketch_add_item(PG_FUNCTION_ARGS) {
+ void* sketchptr;
+ int lg_k;
+ float p;
+
+ // anyelement
+ Oid element_type;
+ Datum element;
+ int16 typlen;
+ bool typbyval;
+ char typalign;
+
+ // input array of doubles
+ 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* values;
+ int i;
+
+ 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, "aod_sketch_add_item called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ // look at the array of values first to know the array length in case we need to create a new sketch
+ arr_in = PG_GETARG_ARRAYTYPE_P(2);
+ 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);
+
+ values = palloc(sizeof(double) * arr_len);
+ for (i = 0; i < arr_len; i++) {
+ values[i] = DatumGetFloat8(data_in[i]);
+ }
+
+ if (PG_ARGISNULL(0)) {
+ lg_k = PG_GETARG_INT32(3);
+ p = PG_GETARG_FLOAT4(4);
+ if (lg_k) {
+ sketchptr = p ? aod_sketch_new_lgk_p(arr_len, lg_k, p) : aod_sketch_new_lgk(arr_len, lg_k);
+ } else {
+ sketchptr = aod_sketch_new(arr_len);
+ }
+ } else {
+ sketchptr = PG_GETARG_POINTER(0);
+ }
+
+ element_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
+ element = PG_GETARG_DATUM(1);
+ get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);
+ if (typlen == -1) {
+ // varlena
+ aod_sketch_update(sketchptr, VARDATA_ANY(element), VARSIZE_ANY_EXHDR(element), values);
+ } else if (typbyval) {
+ // fixed-length passed by value
+ aod_sketch_update(sketchptr, &element, typlen, values);
+ } else {
+ // fixed-length passed by reference
+ aod_sketch_update(sketchptr, (void*)element, typlen, values);
+ }
+ pfree(values);
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(sketchptr);
+}
+
+Datum pg_aod_sketch_get_estimate(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ double estimate;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ estimate = compact_aod_sketch_get_estimate(sketchptr);
+ compact_aod_sketch_delete(sketchptr);
+ PG_RETURN_FLOAT8(estimate);
+}
+
+Datum pg_aod_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 = aod_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 = (Datum*) aod_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+ compact_aod_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_aod_sketch_to_string(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+ bool print_entries;
+ char* str;
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ if (PG_NARGS() > 1) {
+ print_entries = PG_GETARG_BOOL(1);
+ } else {
+ print_entries = false;
+ }
+ sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ str = aod_sketch_to_string(sketchptr, print_entries);
+ compact_aod_sketch_delete(sketchptr);
+ PG_RETURN_TEXT_P(cstring_to_text(str));
+}
+
+Datum pg_aod_sketch_intersection_agg(PG_FUNCTION_ARGS) {
+ void* interptr;
+ bytea* sketch_bytes;
+ void* sketchptr;
+ int num_values;
+
+ 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, "aod_sketch_intersect called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ if (PG_ARGISNULL(0)) {
+ num_values = PG_GETARG_INT32(2);
+ if (num_values == 0) num_values = 1;
+ interptr = aod_intersection_new(num_values);
+ } else {
+ interptr = PG_GETARG_POINTER(0);
+ }
+
+ sketch_bytes = PG_GETARG_BYTEA_P(1);
+ sketchptr = aod_sketch_deserialize(VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
+ aod_intersection_update(interptr, sketchptr);
+ compact_aod_sketch_delete(sketchptr);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(interptr);
+}
+
+Datum pg_aod_sketch_union_agg(PG_FUNCTION_ARGS) {
+ void* unionptr;
+ bytea* sketch_bytes;
+ void* sketchptr;
+ int num_values;
+ int lg_k;
+
+ 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, "aod_sketch_merge called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ if (PG_ARGISNULL(0)) {
+ num_values = PG_GETARG_INT32(2);
+ if (num_values == 0) num_values = 1;
+ lg_k = PG_GETARG_INT32(3);
+ unionptr = lg_k ? aod_union_new_lgk(num_values, lg_k) : aod_union_new(num_values);
+ } else {
+ unionptr = PG_GETARG_POINTER(0);
+ }
+
+ sketch_bytes = PG_GETARG_BYTEA_P(1);
+ sketchptr = aod_sketch_deserialize(VARDATA(sketch_bytes), VARSIZE(sketch_bytes) - VARHDRSZ);
+ aod_union_update(unionptr, sketchptr);
+ compact_aod_sketch_delete(sketchptr);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_POINTER(unionptr);
+}
+
+Datum pg_aod_sketch_from_internal(PG_FUNCTION_ARGS) {
+ void* sketchptr;
+ struct ptr_with_size bytes_out;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "aod_sketch_from_internal called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ sketchptr = PG_GETARG_POINTER(0);
+ sketchptr = aod_sketch_compact(sketchptr);
+ bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
+ compact_aod_sketch_delete(sketchptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_aod_sketch_get_estimate_from_internal(PG_FUNCTION_ARGS) {
+ void* sketchptr;
+ double estimate;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "aod_sketch_from_internal called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ sketchptr = PG_GETARG_POINTER(0);
+ estimate = update_aod_sketch_get_estimate(sketchptr);
+ update_aod_sketch_delete(sketchptr);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_FLOAT8(estimate);
+}
+
+Datum pg_aod_union_get_result(PG_FUNCTION_ARGS) {
+ void* unionptr;
+ void* sketchptr;
+ struct ptr_with_size bytes_out;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "aod_union_get_result called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ unionptr = PG_GETARG_POINTER(0);
+ sketchptr = aod_union_get_result(unionptr);
+ bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
+ compact_aod_sketch_delete(sketchptr);
+ aod_union_delete(unionptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_aod_intersection_get_result(PG_FUNCTION_ARGS) {
+ void* interptr;
+ void* sketchptr;
+ struct ptr_with_size bytes_out;
+
+ MemoryContext oldcontext;
+ MemoryContext aggcontext;
+
+ if (PG_ARGISNULL(0)) PG_RETURN_NULL();
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext)) {
+ elog(ERROR, "aod_intersection_get_result called in non-aggregate context");
+ }
+ oldcontext = MemoryContextSwitchTo(aggcontext);
+
+ interptr = PG_GETARG_POINTER(0);
+ sketchptr = aod_intersection_get_result(interptr);
+ bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
+ compact_aod_sketch_delete(sketchptr);
+ aod_intersection_delete(interptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+
+ MemoryContextSwitchTo(oldcontext);
+
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_aod_sketch_union(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in1;
+ const bytea* bytes_in2;
+ void* sketchptr1;
+ void* sketchptr2;
+ void* unionptr;
+ void* sketchptr;
+ struct ptr_with_size bytes_out;
+ int num_values;
+ int lg_k;
+
+ num_values = PG_GETARG_INT32(2);
+ if (num_values == 0) num_values = 1;
+ lg_k = PG_GETARG_INT32(3);
+ unionptr = lg_k ? aod_union_new_lgk(num_values, lg_k) : aod_union_new(num_values);
+ if (!PG_ARGISNULL(0)) {
+ bytes_in1 = PG_GETARG_BYTEA_P(0);
+ sketchptr1 = aod_sketch_deserialize(VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
+ aod_union_update(unionptr, sketchptr1);
+ compact_aod_sketch_delete(sketchptr1);
+ }
+ if (!PG_ARGISNULL(1)) {
+ bytes_in2 = PG_GETARG_BYTEA_P(1);
+ sketchptr2 = aod_sketch_deserialize(VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
+ aod_union_update(unionptr, sketchptr2);
+ compact_aod_sketch_delete(sketchptr2);
+ }
+ sketchptr = aod_union_get_result(unionptr);
+ aod_union_delete(unionptr);
+ bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
+ compact_aod_sketch_delete(sketchptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_aod_sketch_intersection(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in1;
+ const bytea* bytes_in2;
+ void* sketchptr1;
+ void* sketchptr2;
+ void* interptr;
+ void* sketchptr;
+ struct ptr_with_size bytes_out;
+ int num_values;
+
+ num_values = PG_GETARG_INT32(2);
+ if (num_values == 0) num_values = 1;
+ interptr = aod_intersection_new(num_values);
+ if (!PG_ARGISNULL(0)) {
+ bytes_in1 = PG_GETARG_BYTEA_P(0);
+ sketchptr1 = aod_sketch_deserialize(VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
+ aod_intersection_update(interptr, sketchptr1);
+ compact_aod_sketch_delete(sketchptr1);
+ }
+ if (!PG_ARGISNULL(1)) {
+ bytes_in2 = PG_GETARG_BYTEA_P(1);
+ sketchptr2 = aod_sketch_deserialize(VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
+ aod_intersection_update(interptr, sketchptr2);
+ compact_aod_sketch_delete(sketchptr2);
+ }
+ sketchptr = aod_intersection_get_result(interptr);
+ aod_intersection_delete(interptr);
+ bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
+ compact_aod_sketch_delete(sketchptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_aod_sketch_a_not_b(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in1;
+ const bytea* bytes_in2;
+ void* sketchptr1;
+ void* sketchptr2;
+ void* sketchptr;
+ struct ptr_with_size bytes_out;
+
+ if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
+ elog(ERROR, "aod_a_not_b expects two valid aod sketches");
+ }
+
+ bytes_in1 = PG_GETARG_BYTEA_P(0);
+ sketchptr1 = aod_sketch_deserialize(VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
+ bytes_in2 = PG_GETARG_BYTEA_P(1);
+ sketchptr2 = aod_sketch_deserialize(VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
+ sketchptr = aod_a_not_b(sketchptr1, sketchptr2);
+ compact_aod_sketch_delete(sketchptr1);
+ compact_aod_sketch_delete(sketchptr2);
+ bytes_out = aod_sketch_serialize(sketchptr, VARHDRSZ);
+ compact_aod_sketch_delete(sketchptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_aod_sketch_to_kll_float_sketch(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* aodptr;
+ int column_index;
+ int k;
+ void* kllptr;
+ struct ptr_with_size bytes_out;
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ aodptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+ column_index = PG_GETARG_INT32(1);
+ k = PG_GETARG_INT32(2);
+ if (k == 0) k = DEFAULT_K;
+ kllptr = aod_sketch_to_kll_float_sketch(aodptr, column_index, k);
+ bytes_out = kll_float_sketch_serialize(kllptr, VARHDRSZ);
+ kll_float_sketch_delete(kllptr);
+ compact_aod_sketch_delete(aodptr);
+ SET_VARSIZE(bytes_out.ptr, bytes_out.size);
+ PG_RETURN_BYTEA_P(bytes_out.ptr);
+}
+
+Datum pg_aod_sketch_students_t_test(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in1;
+ const bytea* bytes_in2;
+ void* sketchptr1;
+ void* sketchptr2;
+
+ // output array of p-values
+ Datum* p_values;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+ unsigned arr_len_out;
+
+ if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
+ elog(ERROR, "aod_a_not_b expects two valid aod sketches");
+ }
+
+ bytes_in1 = PG_GETARG_BYTEA_P(0);
+ sketchptr1 = aod_sketch_deserialize(VARDATA(bytes_in1), VARSIZE(bytes_in1) - VARHDRSZ);
+ bytes_in2 = PG_GETARG_BYTEA_P(1);
+ sketchptr2 = aod_sketch_deserialize(VARDATA(bytes_in2), VARSIZE(bytes_in2) - VARHDRSZ);
+ p_values = (Datum*) aod_sketch_students_t_test(sketchptr1, sketchptr2, &arr_len_out);
+ compact_aod_sketch_delete(sketchptr1);
+ compact_aod_sketch_delete(sketchptr2);
+
+ // construct output array of p-values
+ get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(p_values, arr_len_out, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
+
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
+
+Datum pg_aod_sketch_to_means(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+
+ // output array
+ Datum* means;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+ unsigned arr_len_out;
+
+ if (PG_ARGISNULL(0)) {
+ elog(ERROR, "aod_sketch_to_means expects a valid aod sketch");
+ }
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+
+ means = (Datum*) aod_sketch_to_means(sketchptr, &arr_len_out);
+ compact_aod_sketch_delete(sketchptr);
+
+ // construct output array
+ get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(means, arr_len_out, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
+
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
+
+Datum pg_aod_sketch_to_variances(PG_FUNCTION_ARGS) {
+ const bytea* bytes_in;
+ void* sketchptr;
+
+ // output array
+ Datum* variances;
+ ArrayType* arr_out;
+ int16 elmlen_out;
+ bool elmbyval_out;
+ char elmalign_out;
+ unsigned arr_len_out;
+
+ if (PG_ARGISNULL(0)) {
+ elog(ERROR, "aod_sketch_to_variances expects a valid aod sketch");
+ }
+
+ bytes_in = PG_GETARG_BYTEA_P(0);
+ sketchptr = aod_sketch_deserialize(VARDATA(bytes_in), VARSIZE(bytes_in) - VARHDRSZ);
+
+ variances = (Datum*) aod_sketch_to_variances(sketchptr, &arr_len_out);
+ compact_aod_sketch_delete(sketchptr);
+
+ // construct output array
+ get_typlenbyvalalign(FLOAT8OID, &elmlen_out, &elmbyval_out, &elmalign_out);
+ arr_out = construct_array(variances, arr_len_out, FLOAT8OID, elmlen_out, elmbyval_out, elmalign_out);
+
+ PG_RETURN_ARRAYTYPE_P(arr_out);
+}
diff --git a/src/cpc_sketch_c_adapter.h b/src/cpc_sketch_c_adapter.h
index 550f8e3..8df971d 100644
--- a/src/cpc_sketch_c_adapter.h
+++ b/src/cpc_sketch_c_adapter.h
@@ -24,6 +24,8 @@
extern "C" {
#endif
+#include "ptr_with_size.h"
+
void cpc_init();
void cpc_cleanup();
@@ -36,11 +38,6 @@
void** cpc_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
char* cpc_sketch_to_string(const void* sketchptr);
-struct ptr_with_size {
- void* ptr;
- unsigned long long size;
-};
-
struct ptr_with_size cpc_sketch_serialize(const void* sketchptr, unsigned header_size);
void* cpc_sketch_deserialize(const char* buffer, unsigned length);
diff --git a/src/frequent_strings_sketch_c_adapter.h b/src/frequent_strings_sketch_c_adapter.h
index ab0fd3b..c6a8f7d 100644
--- a/src/frequent_strings_sketch_c_adapter.h
+++ b/src/frequent_strings_sketch_c_adapter.h
@@ -24,6 +24,8 @@
extern "C" {
#endif
+#include "ptr_with_size.h"
+
void* frequent_strings_sketch_new(unsigned lg_k);
void frequent_strings_sketch_delete(void* sketchptr);
@@ -31,11 +33,6 @@
void frequent_strings_sketch_merge(void* sketchptr1, const void* sketchptr2);
char* frequent_strings_sketch_to_string(const void* sketchptr, bool print_items);
-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/hll_sketch_c_adapter.h b/src/hll_sketch_c_adapter.h
index 08c2317..21ef1bd 100644
--- a/src/hll_sketch_c_adapter.h
+++ b/src/hll_sketch_c_adapter.h
@@ -24,6 +24,8 @@
extern "C" {
#endif
+#include "ptr_with_size.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);
@@ -34,11 +36,6 @@
void** hll_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
char* hll_sketch_to_string(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);
diff --git a/src/kll_float_sketch_c_adapter.h b/src/kll_float_sketch_c_adapter.h
index df98366..f175906 100644
--- a/src/kll_float_sketch_c_adapter.h
+++ b/src/kll_float_sketch_c_adapter.h
@@ -24,6 +24,10 @@
extern "C" {
#endif
+#include "ptr_with_size.h"
+
+static const unsigned DEFAULT_K = 200;
+
void* kll_float_sketch_new(unsigned k);
void kll_float_sketch_delete(void* sketchptr);
@@ -34,11 +38,6 @@
unsigned long long kll_float_sketch_get_n(const void* sketchptr);
char* kll_float_sketch_to_string(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);
diff --git a/src/kll_float_sketch_pg_functions.c b/src/kll_float_sketch_pg_functions.c
index 64df873..87426c8 100644
--- a/src/kll_float_sketch_pg_functions.c
+++ b/src/kll_float_sketch_pg_functions.c
@@ -55,7 +55,6 @@
Datum pg_kll_float_sketch_get_quantiles(PG_FUNCTION_ARGS);
Datum pg_kll_float_sketch_get_histogram(PG_FUNCTION_ARGS);
-static const unsigned DEFAULT_K = 200;
static const unsigned DEFAULT_NUM_BINS = 10;
Datum pg_kll_float_sketch_add_item(PG_FUNCTION_ARGS) {
diff --git a/src/ptr_with_size.h b/src/ptr_with_size.h
new file mode 100644
index 0000000..f130444
--- /dev/null
+++ b/src/ptr_with_size.h
@@ -0,0 +1,28 @@
+/*
+ * 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 PTR_WITH_SIZE_H
+#define PTR_WITH_SIZE_H
+
+struct ptr_with_size {
+ void* ptr;
+ unsigned long long size;
+};
+
+#endif
diff --git a/src/theta_sketch_c_adapter.h b/src/theta_sketch_c_adapter.h
index 36e2b3a..2a14f98 100644
--- a/src/theta_sketch_c_adapter.h
+++ b/src/theta_sketch_c_adapter.h
@@ -24,6 +24,8 @@
extern "C" {
#endif
+#include "ptr_with_size.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);
@@ -36,11 +38,6 @@
void** theta_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
char* theta_sketch_to_string(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);
