WIP for aggregates

TileDB-Inc · Nov 21, 2023 · d322db6 · d322db6
1 parent 8ccc4cb
commit d322db6
Show file tree

Hide file tree

Showing 4 changed files with 64 additions and 7 deletions.
diff --git a/R/RcppExports.R b/R/RcppExports.R
@@ -814,6 +814,10 @@ libtiledb_query_get_ctx <- function(query) {
     .Call(`_tiledb_libtiledb_query_get_ctx`, query)
 }
 
+libtiledb_query_apply_aggregate <- function(query, attribute_name, operator_name, nullable = FALSE) {
+    .Call(`_tiledb_libtiledb_query_apply_aggregate`, query, attribute_name, operator_name, nullable)
+}
+
 libtiledb_query_condition <- function(ctx) {
     .Call(`_tiledb_libtiledb_query_condition`, ctx)
 }

diff --git a/inst/include/tiledb.h b/inst/include/tiledb.h
@@ -7,13 +7,9 @@
 
 #include <tiledb/tiledb>
 #if TILEDB_VERSION_MAJOR == 2 && TILEDB_VERSION_MINOR >= 4
-#include <tiledb/tiledb_experimental>
-#endif
-#if TILEDB_VERSION_MAJOR == 2 && TILEDB_VERSION_MINOR >= 17
-#include <tiledb/array_experimental.h>
-#include <tiledb/attribute_experimental.h>
-#include <tiledb/enumeration_experimental.h>
-#include <tiledb/query_condition_experimental.h>
+  // this header includes the other experimental headers
+  // condition on the appropriate version is still done in each function
+  #include <tiledb/tiledb_experimental>
 #endif
 
 // Use the 'finalizer on exit' toggle in the XPtr template to ensure

diff --git a/src/RcppExports.cpp b/src/RcppExports.cpp
@@ -2415,6 +2415,20 @@ BEGIN_RCPP
     return rcpp_result_gen;
 END_RCPP
 }
+// libtiledb_query_apply_aggregate
+SEXP libtiledb_query_apply_aggregate(XPtr<tiledb::Query> query, std::string attribute_name, std::string operator_name, bool nullable);
+RcppExport SEXP _tiledb_libtiledb_query_apply_aggregate(SEXP querySEXP, SEXP attribute_nameSEXP, SEXP operator_nameSEXP, SEXP nullableSEXP) {
+BEGIN_RCPP
+    Rcpp::RObject rcpp_result_gen;
+    Rcpp::RNGScope rcpp_rngScope_gen;
+    Rcpp::traits::input_parameter< XPtr<tiledb::Query> >::type query(querySEXP);
+    Rcpp::traits::input_parameter< std::string >::type attribute_name(attribute_nameSEXP);
+    Rcpp::traits::input_parameter< std::string >::type operator_name(operator_nameSEXP);
+    Rcpp::traits::input_parameter< bool >::type nullable(nullableSEXP);
+    rcpp_result_gen = Rcpp::wrap(libtiledb_query_apply_aggregate(query, attribute_name, operator_name, nullable));
+    return rcpp_result_gen;
+END_RCPP
+}
 // libtiledb_query_condition
 XPtr<tiledb::QueryCondition> libtiledb_query_condition(XPtr<tiledb::Context> ctx);
 RcppExport SEXP _tiledb_libtiledb_query_condition(SEXP ctxSEXP) {
@@ -3758,6 +3772,7 @@ static const R_CallMethodDef CallEntries[] = {
     {"_tiledb_libtiledb_query_get_schema", (DL_FUNC) &_tiledb_libtiledb_query_get_schema, 2},
     {"_tiledb_libtiledb_query_stats", (DL_FUNC) &_tiledb_libtiledb_query_stats, 1},
     {"_tiledb_libtiledb_query_get_ctx", (DL_FUNC) &_tiledb_libtiledb_query_get_ctx, 1},
+    {"_tiledb_libtiledb_query_apply_aggregate", (DL_FUNC) &_tiledb_libtiledb_query_apply_aggregate, 4},
     {"_tiledb_libtiledb_query_condition", (DL_FUNC) &_tiledb_libtiledb_query_condition, 1},
     {"_tiledb_libtiledb_query_condition_init", (DL_FUNC) &_tiledb_libtiledb_query_condition_init, 5},
     {"_tiledb_libtiledb_query_condition_combine", (DL_FUNC) &_tiledb_libtiledb_query_condition_combine, 3},

diff --git a/src/libtiledb.cpp b/src/libtiledb.cpp
@@ -3881,6 +3881,48 @@ XPtr<tiledb::Context> libtiledb_query_get_ctx(XPtr<tiledb::Query> query) {
     return make_xptr<tiledb::Context>(new tiledb::Context(ctx));
 }
 
+// [[Rcpp::export]]
+SEXP libtiledb_query_apply_aggregate(XPtr<tiledb::Query> query,
+                                     std::string attribute_name,
+                                     std::string operator_name,
+                                     bool nullable = false) {
+#if TILEDB_VERSION >= TileDB_Version(2,18,0)
+    check_xptr_tag<tiledb::Query>(query);
+    tiledb::QueryChannel channel = tiledb::QueryExperimental::get_default_channel(*query.get());
+    tiledb::ChannelOperation operation;
+    if (operator_name == "Sum") {
+        operation = tiledb::QueryExperimental::create_unary_aggregate<tiledb::SumOperator>(*query.get(), attribute_name);
+    } else if (operator_name == "Min") {
+        operation = tiledb::QueryExperimental::create_unary_aggregate<tiledb::MinOperator>(*query.get(), attribute_name);
+    } else if (operator_name == "Max") {
+        operation = tiledb::QueryExperimental::create_unary_aggregate<tiledb::MaxOperator>(*query.get(), attribute_name);
+    } else if (operator_name == "Mean") {
+        operation = tiledb::QueryExperimental::create_unary_aggregate<tiledb::MeanOperator>(*query.get(), attribute_name);
+    } else if (operator_name == "NullCount") {
+        operation = tiledb::QueryExperimental::create_unary_aggregate<tiledb::NullCountOperator>(*query.get(), attribute_name);
+    } else {
+        Rcpp::stop("Invalid aggregation operator '%s' specified.", operator_name.c_str());
+    }
+    channel.apply_aggregate(operator_name, operation);
+    std::vector<uint8_t> nulls = { 0 };
+    uint64_t size = 1;
+    if (operator_name != "NullCount") {
+        double result = 0;
+        query->set_data_buffer(operator_name, &result, size);
+        if (nullable) query->set_validity_buffer(operator_name, nulls);
+        query->submit();
+        return Rcpp::wrap(result);
+    } else {
+        uint64_t result = 0;
+        query->set_data_buffer(operator_name, &result, size);
+        // no validity buffer for NullCount
+        query->submit();
+        return Rcpp::wrap(result);
+    }
+#else
+    return Rcpp::wrap(R_NaReal);
+#endif
+}
 
 /**
  * Query Condition