Support deletion of fragments

NAMESPACE

@@ -83,6 +83,7 @@ export(tile_order)
 export(tiledb_array)
 export(tiledb_array_close)
 export(tiledb_array_create)
+export(tiledb_array_delete_fragments)
 export(tiledb_array_get_non_empty_domain_from_index)
 export(tiledb_array_get_non_empty_domain_from_name)
 export(tiledb_array_is_heterogeneous)

NEWS.md

@@ -10,6 +10,8 @@
 
 * Support for XOR filters has been added (#472)
 
+* Support for deletion of fragments has been added (#473)
+
 ## Bug Fixes
 
 * Treatment of character columns with missing values has been corrected (#454)

R/Array.R

@@ -52,12 +52,15 @@ tiledb_array_create <- function(uri, schema, encryption_key) {
 ##'
 ##' @param arr A TileDB Array object as for example returned by `tiledb_array()`
 ##' @param type A character value that must be either \sQuote{READ}, \sQuote{WRITE}
-##' or (for TileDB 2.12.0 or later) \sQuote{DELETE}
+##' or (for TileDB 2.12.0 or later) \sQuote{DELETE} or \sQuote{MODIFY_EXCLUSIVE}
 ##' @return The TileDB Array object but opened for reading or writing
 ##' @importFrom methods .hasSlot
 ##' @export
 tiledb_array_open <- function(arr,
-                              type = if (tiledb_version(TRUE) >= "2.12.0") c("READ", "WRITE", "DELETE") else c("READ", "WRITE")) {
+                              type = if (tiledb_version(TRUE) >= "2.12.0")
+                                         c("READ", "WRITE", "DELETE", "MODIFY_EXCLUSIVE")
+                                     else
+                                         c("READ", "WRITE")) {
   stopifnot("The 'arr' argument must be a tiledb_array object" = .isArray(arr))
   type <- match.arg(type)
 
@@ -143,3 +146,18 @@ tiledb_array_is_heterogeneous <- function(arr) {
   n <- length(unique(domaintype))
   n > 1
 }
+
+##' Delete fragments written between the start and end times given
+##'
+##' @param arr A TileDB Array object as for example returned by \code{tiledb_array()}
+##' @param ts_start A Datetime object that will be converted to millisecond granularity
+##' @param ts_end A Datetime object that will be converted to millisecond granularity
+##' @return A boolean indicating success
+##' @export
+tiledb_array_delete_fragments <- function(arr, ts_start, ts_end) {
+    stopifnot("The 'arr' argument must be a tiledb_array object" = .isArray(arr),
+              "The 'ts_start' argument must a time object" = inherits(ts_start, "POSIXct"),
+              "The 'ts_end' argument must a time object" = inherits(ts_end, "POSIXct"))
+    libtiledb_array_delete_fragments(arr@ptr, ts_start, ts_end)
+    invisible(TRUE)
+}

R/Query.R

@@ -39,7 +39,10 @@ setClass("tiledb_query",
 #' @return 'tiledb_query' object
 #' @export tiledb_query
 tiledb_query <- function(array,
-                         type = if (tiledb_version(TRUE) >= "2.12.0") c("READ", "WRITE", "DELETE") else c("READ", "WRITE"),
+                         type = if (tiledb_version(TRUE) >= "2.12.0")
+                                    c("READ", "WRITE", "DELETE", "MODIFY_EXCLUSIVE")
+                                else
+                                    c("READ", "WRITE"),
                          ctx = tiledb_get_context()) {
   stopifnot(`Argument 'arr' must be a tiledb_array object` = .isArray(array))
   type <- match.arg(type)

R/RcppExports.R

@@ -532,6 +532,10 @@ libtiledb_array_open_timestamp_end <- function(array) {
     .Call(`_tiledb_libtiledb_array_open_timestamp_end`, array)
 }
 
+libtiledb_array_delete_fragments <- function(array, tstamp_start, tstamp_end) {
+    invisible(.Call(`_tiledb_libtiledb_array_delete_fragments`, array, tstamp_start, tstamp_end))
+}
+
 libtiledb_query <- function(ctx, array, type) {
     .Call(`_tiledb_libtiledb_query`, ctx, array, type)
 }

inst/tinytest/test_query.R

@@ -160,7 +160,7 @@ schema <- tiledb_array_schema(dom,
 tiledb_array_create(tmp, schema)
 arr <- tiledb_array(tmp)
 qry <- tiledb_query(arr, "WRITE")
-qry <- tiledb_query_set_layout(qry, "ROW_MAJOR")
+if (tiledb_version(TRUE) < "2.12.0") qry <- tiledb_query_set_layout(qry, "ROW_MAJOR")
 
 rows <- 1:10
 qry <- tiledb_query_set_buffer(qry, "rows", rows)

inst/tinytest/test_tiledbarray_extra.R

@@ -44,3 +44,36 @@ expect_equal(chk[,"cols"], c(3L,4L))
 expect_equal(chk[,"a"], c(3L,2L))
 
 unlink(tmp, recursive = TRUE)
+
+
+
+## delete fragment (2.12.0 or later)
+if (tiledb_version(TRUE) < "2.12.0") exit_file("Remainder needs 2.12.0 or later")
+N <- 5
+ts <- rep(Sys.time(), N)
+tmp <- tempfile()
+dir.create(tmp)
+uri <- file.path(tmp, "array")
+D <- data.frame(index = paste0("A", format(trunc(runif(10)*1000))), value = cumsum(runif(10)))
+fromDataFrame(D, uri, col_index=1, sparse=TRUE)
+ts[1] <- Sys.time()
+for (i in 2:N) {
+    Sys.sleep(0.25)
+    D <- data.frame(index = paste0(LETTERS[i], format(trunc(runif(10)*1000))), value = cumsum(runif(10)))
+    fromDataFrame(D, uri, col_index=1, mode="append", sparse=TRUE)
+    ts[i] <- Sys.time()
+}
+
+fraginfo <- tiledb_fragment_info(uri)
+expect_equal(tiledb_fragment_info_get_num(fraginfo), N) # N (ie 5) before deletion
+
+arr <- tiledb_array(uri)
+arr <- tiledb_array_open(arr, "MODIFY_EXCLUSIVE")
+expect_true(tiledb_array_is_open(arr))
+expect_true(is(arr, "tiledb_array"))
+expect_true(tiledb_array_delete_fragments(arr, ts[2]-0.1, ts[4]+0.1))
+arr <- tiledb_array_close(arr)
+fraginfo <- tiledb_fragment_info(uri)
+expect_equal(tiledb_fragment_info_get_num(fraginfo), 2) # 2 after three deleted
+
+unlink(tmp, recursive = TRUE)

src/RcppExports.cpp

@@ -1520,6 +1520,18 @@ BEGIN_RCPP
     return rcpp_result_gen;
 END_RCPP
 }
+// libtiledb_array_delete_fragments
+void libtiledb_array_delete_fragments(XPtr<tiledb::Array> array, Rcpp::Datetime tstamp_start, Rcpp::Datetime tstamp_end);
+RcppExport SEXP _tiledb_libtiledb_array_delete_fragments(SEXP arraySEXP, SEXP tstamp_startSEXP, SEXP tstamp_endSEXP) {
+BEGIN_RCPP
+    Rcpp::RNGScope rcpp_rngScope_gen;
+    Rcpp::traits::input_parameter< XPtr<tiledb::Array> >::type array(arraySEXP);
+    Rcpp::traits::input_parameter< Rcpp::Datetime >::type tstamp_start(tstamp_startSEXP);
+    Rcpp::traits::input_parameter< Rcpp::Datetime >::type tstamp_end(tstamp_endSEXP);
+    libtiledb_array_delete_fragments(array, tstamp_start, tstamp_end);
+    return R_NilValue;
+END_RCPP
+}
 // libtiledb_query
 XPtr<tiledb::Query> libtiledb_query(XPtr<tiledb::Context> ctx, XPtr<tiledb::Array> array, std::string type);
 RcppExport SEXP _tiledb_libtiledb_query(SEXP ctxSEXP, SEXP arraySEXP, SEXP typeSEXP) {
@@ -3248,6 +3260,7 @@ static const R_CallMethodDef CallEntries[] = {
     {"_tiledb_libtiledb_array_open_timestamp_start", (DL_FUNC) &_tiledb_libtiledb_array_open_timestamp_start, 1},
     {"_tiledb_libtiledb_array_set_open_timestamp_end", (DL_FUNC) &_tiledb_libtiledb_array_set_open_timestamp_end, 2},
     {"_tiledb_libtiledb_array_open_timestamp_end", (DL_FUNC) &_tiledb_libtiledb_array_open_timestamp_end, 1},
+    {"_tiledb_libtiledb_array_delete_fragments", (DL_FUNC) &_tiledb_libtiledb_array_delete_fragments, 3},
     {"_tiledb_libtiledb_query", (DL_FUNC) &_tiledb_libtiledb_query, 3},
     {"_tiledb_libtiledb_query_type", (DL_FUNC) &_tiledb_libtiledb_query_type, 1},
     {"_tiledb_libtiledb_query_set_layout", (DL_FUNC) &_tiledb_libtiledb_query_set_layout, 2},

src/libtiledb.cpp

@@ -407,6 +407,8 @@ tiledb_query_type_t _string_to_tiledb_query_type(std::string qtstr) {
   } else if (qtstr == "WRITE") {
     return TILEDB_WRITE;
 #if TILEDB_VERSION >= TileDB_Version(2,12,0)
+  } else if (qtstr == "MODIFY_EXCLUSIVE") {
+    return TILEDB_MODIFY_EXCLUSIVE;
   } else if (qtstr == "DELETE") {
     return TILEDB_DELETE;
 #endif
@@ -424,6 +426,8 @@ std::string _tiledb_query_type_to_string(tiledb_query_type_t qtype) {
 #if TILEDB_VERSION >= TileDB_Version(2,12,0)
     case TILEDB_DELETE:
       return "DELETE";
+    case TILEDB_MODIFY_EXCLUSIVE:
+      return "MODIFY_EXCLUSIVE";
 #endif
     default:
       Rcpp::stop("unknown tiledb_query_type_t (%d)", qtype);
@@ -2488,6 +2492,19 @@ Rcpp::Datetime libtiledb_array_open_timestamp_end(XPtr<tiledb::Array> array) {
 #endif
 }
 
+// [[Rcpp::export]]
+void libtiledb_array_delete_fragments(XPtr<tiledb::Array> array,
+                                      Rcpp::Datetime tstamp_start, Rcpp::Datetime tstamp_end) {
+#if TILEDB_VERSION >= TileDB_Version(2,12,0)
+    check_xptr_tag<tiledb::Array>(array);
+    const std::string uri = array->uri();
+    uint64_t ts_ms_st = static_cast<uint64_t>(std::round(tstamp_start.getFractionalTimestamp() * 1000));
+    uint64_t ts_ms_en = static_cast<uint64_t>(std::round(tstamp_end.getFractionalTimestamp() * 1000));
+    array->delete_fragments(uri, ts_ms_st, ts_ms_en);
+#endif
+}
+
+
 /**
  * Query
  */