Merge pull request #5 from apache/cpc_error_bounds

cpc error bounds

sql/datasketches_cpc_sketch.sql

@@ -46,6 +46,14 @@ CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate(cpc_sketch) RETURNS double pr
     AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate'
     LANGUAGE C STRICT IMMUTABLE;
 
+CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate_and_bounds(cpc_sketch) RETURNS double precision[]
+    AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate_and_bounds'
+    LANGUAGE C STRICT IMMUTABLE;
+
+CREATE OR REPLACE FUNCTION cpc_sketch_get_estimate_and_bounds(cpc_sketch, int) RETURNS double precision[]
+    AS '$libdir/datasketches', 'pg_cpc_sketch_get_estimate_and_bounds'
+    LANGUAGE C STRICT IMMUTABLE;
+
 CREATE OR REPLACE FUNCTION cpc_sketch_from_internal(internal) RETURNS cpc_sketch
     AS '$libdir/datasketches', 'pg_cpc_sketch_from_internal'
     LANGUAGE C STRICT IMMUTABLE;

src/cpc_sketch_c_adapter.cpp

@@ -19,10 +19,6 @@
 
 #include "cpc_sketch_c_adapter.h"
 
-extern "C" {
-#include <postgres.h>
-}
-
 #include <sstream>
 
 #include <cpc_sketch.hpp>
@@ -62,6 +58,18 @@ double cpc_sketch_get_estimate(const void* sketchptr) {
   }
 }
 
+Datum* cpc_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 datasketches::cpc_sketch*>(sketchptr)->get_estimate());
+    est_and_bounds[1] = Float8GetDatum(static_cast<const datasketches::cpc_sketch*>(sketchptr)->get_lower_bound(num_std_devs));
+    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());
+  }
+}
+
 void cpc_sketch_to_string(const void* sketchptr, char* buffer, unsigned length) {
   try {
     std::stringstream s;

src/cpc_sketch_c_adapter.h

@@ -24,12 +24,15 @@
 extern "C" {
 #endif
 
+#include <postgres.h>
+
 void* cpc_sketch_new(unsigned lg_k);
 void cpc_sketch_delete(void* sketchptr);
 
 void cpc_sketch_update(void* sketchptr, const void* data, unsigned length);
 void cpc_sketch_merge(void* sketchptr1, const void* sketchptr2);
 double cpc_sketch_get_estimate(const void* sketchptr);
+Datum* cpc_sketch_get_estimate_and_bounds(const void* sketchptr, unsigned num_std_devs);
 void cpc_sketch_to_string(const void* sketchptr, char* buffer, unsigned length);
 
 void* cpc_sketch_serialize(const void* sketchptr);

src/cpc_sketch_pg_functions.c

@@ -21,6 +21,8 @@
 #include <fmgr.h>
 #include <utils/lsyscache.h>
 #include <utils/builtins.h>
+#include <utils/array.h>
+#include <catalog/pg_type.h>
 
 #include "cpc_sketch_c_adapter.h"
 #include "base64.h"
@@ -30,6 +32,7 @@ const unsigned DEFAULT_LG_K = 11;
 /* PG_FUNCTION_INFO_V1 macro to pass functions to postgres */
 PG_FUNCTION_INFO_V1(pg_cpc_sketch_add_item);
 PG_FUNCTION_INFO_V1(pg_cpc_sketch_get_estimate);
+PG_FUNCTION_INFO_V1(pg_cpc_sketch_get_estimate_and_bounds);
 PG_FUNCTION_INFO_V1(pg_cpc_sketch_to_string);
 PG_FUNCTION_INFO_V1(pg_cpc_sketch_union_agg);
 PG_FUNCTION_INFO_V1(pg_cpc_sketch_from_internal);
@@ -42,6 +45,7 @@ Datum pg_cpc_sketch_recv(PG_FUNCTION_ARGS);
 Datum pg_cpc_sketch_send(PG_FUNCTION_ARGS);
 Datum pg_cpc_sketch_add_item(PG_FUNCTION_ARGS);
 Datum pg_cpc_sketch_get_estimate(PG_FUNCTION_ARGS);
+Datum pg_cpc_sketch_get_estimate_and_bounds(PG_FUNCTION_ARGS);
 Datum pg_cpc_sketch_to_string(PG_FUNCTION_ARGS);
 Datum pg_cpc_sketch_union_agg(PG_FUNCTION_ARGS);
 Datum pg_cpc_sketch_from_internal(PG_FUNCTION_ARGS);
@@ -111,6 +115,31 @@ Datum pg_cpc_sketch_get_estimate(PG_FUNCTION_ARGS) {
   PG_RETURN_FLOAT8(estimate);
 }
 
+Datum pg_cpc_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 = cpc_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 = cpc_sketch_get_estimate_and_bounds(sketchptr, num_std_devs);
+  cpc_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_cpc_sketch_to_string(PG_FUNCTION_ARGS) {
   const bytea* bytes_in;
   void* sketchptr;