Add timestamp support for consolidate and vaccum helpers

R/TileDBArray.R

@@ -1180,33 +1180,83 @@ setReplaceMethod("datetimes_as_int64",
 
 #' Consolidate fragments of a TileDB Array
 #'
-#' This function invokes a consolidation operation. Parameters can be set via
-#' an option configuration object.
+#' This function invokes a consolidation operation. Parameters affecting the operation
+#' can be set via an optional configuration object. Start and end timestamps can also be
+#' set directly.
 #' @param uri A character value with the URI of a TileDB Array
+#' @param start_time An optional timestamp value, if missing config default is used
+#' @param end_time An optional timestamp value, if missing config default is used
 #' @param cfg An optional TileDB Configuration object
 #' @param ctx An option TileDB Context object
-#' @return \code{NULL} is returned invisibly
+#' @return NULL is returned invisibly
 #' @export
-array_consolidate <- function(uri, cfg = NULL, ctx = tiledb_get_context()) {
-  libtiledb_array_consolidate(ctx = ctx@ptr, uri = uri,
-                              # C++ code has Nullable and can instantiate but needs S4 XPtr
-                              cfgptr = if (is.null(cfg)) cfg else cfg@ptr)
+array_consolidate <- function(uri, cfg = NULL,
+                              start_time, end_time,
+                              ctx = tiledb_get_context()) {
+    if (is.null(cfg)) {
+        cfg <- tiledb_config()
+    }
+
+    if (!missing(start_time)) {
+        stopifnot(`start_time must be datetime object` = inherits(start_time, "POSIXt"),
+                  `TileDB 2.3.0 or later is required`  = tiledb_version(TRUE) >= "2.3.0")
+        start_time_int64 <- bit64::as.integer64(as.numeric(start_time) * 1000)
+        cfg["sm.consolidation.timestamp_start"] = as.character(start_time_int64)
+    }
+
+    if (!missing(end_time)) {
+        stopifnot(`end_time must be datetime object`  = inherits(end_time, "POSIXt"),
+                  `TileDB 2.3.0 or later is required` = tiledb_version(TRUE) >= "2.3.0")
+        end_time_int64 <- bit64::as.integer64(as.numeric(end_time) * 1000)
+        cfg["sm.consolidation.timestamp_end"] = as.character(end_time_int64)
+    }
+
+    ctx <- tiledb_ctx(cfg)
+
+    libtiledb_array_consolidate(ctx = ctx@ptr, uri = uri, cfgptr = cfg@ptr)
 }
 
-#' After consolidation, remove consilidated fragments of a TileDB Array
+#' After consolidation, remove consolidated fragments of a TileDB Array
 #'
 #' This function can remove fragments following a consolidation step. Note that vacuuming
 #' should \emph{not} be run if one intends to use the TileDB \emph{time-traveling} feature
-#' of opening arrays at particular timestamps
+#' of opening arrays at particular timestamps.
+#'
+#' Parameters affecting the operation can be set via an optional configuration object.
+#' Start and end timestamps can also be set directly.
+#'
 #' @param uri A character value with the URI of a TileDB Array
+#' @param start_time An optional timestamp value, if missing config default is used
+#' @param end_time An optional timestamp value, if missing config default is used
 #' @param cfg An optional TileDB Configuration object
 #' @param ctx An option TileDB Context object
-#' @return \code{NULL} is returned invisibly
+#' @return NULL is returned invisibly
 #' @export
-array_vacuum <- function(uri, cfg = NULL, ctx = tiledb_get_context()) {
-    libtiledb_array_vacuum(ctx = ctx@ptr, uri = uri,
-                           # C++ code has Nullable and can instantiate but needs S4 XPtr
-                           cfgptr = if (is.null(cfg)) cfg else cfg@ptr)
+array_vacuum <- function(uri, cfg = NULL,
+                         start_time, end_time,
+                         ctx = tiledb_get_context()) {
+
+    if (is.null(cfg)) {
+        cfg <- tiledb_config()
+    }
+
+    if (!missing(start_time)) {
+        stopifnot(`start_time must be datetime object` = inherits(start_time, "POSIXt"),
+                  `TileDB 2.3.0 or later is required`  = tiledb_version(TRUE) >= "2.3.0")
+        start_time_int64 <- bit64::as.integer64(as.numeric(start_time) * 1000)
+        cfg["sm.consolidation.timestamp_start"] = as.character(start_time_int64)
+    }
+
+    if (!missing(end_time)) {
+        stopifnot(`end_time must be datetime object` = inherits(end_time, "POSIXt"),
+                  `TileDB 2.3.0 or later is required` = tiledb_version(TRUE) >= "2.3.0")
+        end_time_int64 <- bit64::as.integer64(as.numeric(end_time) * 1000)
+        cfg["sm.consolidation.timestamp_end"] = as.character(end_time_int64)
+    }
+
+    ctx <- tiledb_ctx(cfg)
+
+    libtiledb_array_vacuum(ctx = ctx@ptr, uri = uri, cfgptr = cfg@ptr)
 }
 
 #' Get the non-empty domain from a TileDB Array by index

R/VFS.R

@@ -443,7 +443,6 @@ tiledb_vfs_write <- function(fh, vec, ctx = tiledb_get_context()) {
 #' @return The binary file content is returned as an integer vector.
 #' @export
 tiledb_vfs_read <- function(fh, offset, nbytes, ctx = tiledb_get_context()) {
-  if (!requireNamespace("bit64", quietly=TRUE)) stop("The 'bit64' package is needed.")
   if (missing(offset)) offset <- bit64::as.integer64(0)
   if (missing(nbytes)) stop("nbytes currently a required parameter")
   stopifnot(fh_argument=is(fh, "externalptr"),

inst/tinytest/test_tiledbarray.R

@@ -1216,3 +1216,16 @@ chk <- tiledb_array(uri = uri, as.data.frame=TRUE)
 res <- chk[]
 expect_equal(dim(res), c(100,6))
 expect_equal(colnames(res), c("rows", "cols", "time", "a", "b", "c"))
+
+## consolidate
+expect_equal(array_consolidate(uri), NULL)
+expect_error(array_consolidate(uri, start_time="abc")) # not a datetime
+expect_error(array_consolidate(uri, end_time="def"))   # not a datetime
+now <- Sys.time()
+expect_equal(array_consolidate(uri, start_time=now-60, end_time=now), NULL)
+
+## vaccum
+expect_equal(array_vacuum(uri), NULL)
+expect_error(array_vacuum(uri, start_time="abc")) # not a datetime
+expect_error(array_vacuum(uri, end_time="def"))   # not a datetime
+expect_equal(array_vacuum(uri, start_time=now-60, end_time=now), NULL)