Add remove_time_ranges_before function (issue #30)

NAMESPACE

@@ -12,3 +12,4 @@ export(get_download_ranges)
 export(insert_location_info)
 export(list_doc)
 export(remove_doc)
+export(remove_time_ranges_before)

R/remove_time_ranges_before.R

@@ -0,0 +1,80 @@
+#' Remove time ranges downloaded before a given date
+#'
+#' Remove measurements from the database based on the moment they were
+#' downloaded. Every time a time range is stored in the caching table,
+#' the download time is recorded in the `time_dl` field. This function
+#' removes all cached time ranges that were downloaded on or before the
+#' given date, together with the measurements that belong to those
+#' ranges.
+#'
+#' For each cached time range that was downloaded on or before `date`,
+#' the function removes the measurements of the station between the
+#' start and end time of that range, including the measurements exactly
+#' at the start and end time. If, after removing these measurements, no
+#' other measurements for that station remain, the station is also
+#' removed from the location table. The processed record is removed
+#' from the caching table.
+#'
+#' @param date a datetime (POSIXct) object. All cached time ranges with
+#'   a download time before or equal to this date are removed.
+#' @param conn database connection object
+#'
+#' @return a named list with one element per removed time range. Each
+#'   element is named after the station and is itself a list with the
+#'   `start` and `end` time of the removed range (as POSIXct) and a
+#'   boolean `removed` indicating whether the station was removed from
+#'   the location table.
+#'
+#' @export
+
+remove_time_ranges_before <- function(date, conn) {
+
+    if(!"POSIXct" %in% class(date)) {
+        stop("ERROR remove_time_ranges_before: date is not a datetime (POSIXct) class")
+    }
+
+    cutoff <- as.numeric(date)
+
+    qry <- glue::glue_sql("SELECT id FROM cache WHERE time_dl <= {cutoff};",
+                          .con = conn)
+    ids <- pool::dbGetQuery(conn, qry)$id
+
+    result <- list()
+
+    for(id in ids) {
+
+        qry <- glue::glue_sql("SELECT station, start, end FROM cache WHERE id = {id};",
+                              .con = conn)
+        record <- pool::dbGetQuery(conn, qry)
+
+        station <- record$station
+        start <- record$start
+        end <- record$end
+
+        qry <- glue::glue_sql("DELETE FROM measurements WHERE station = {station} AND timestamp >= {start} AND timestamp <= {end};",
+                              .con = conn)
+        pool::dbExecute(conn, qry)
+
+        qry <- glue::glue_sql("SELECT count(*) AS n FROM measurements WHERE station = {station};",
+                              .con = conn)
+        remaining <- pool::dbGetQuery(conn, qry)$n
+
+        removed <- FALSE
+        if(remaining == 0) {
+            qry <- glue::glue_sql("DELETE FROM location WHERE station = {station};",
+                                  .con = conn)
+            pool::dbExecute(conn, qry)
+            removed <- TRUE
+        }
+
+        qry <- glue::glue_sql("DELETE FROM cache WHERE id = {id};",
+                              .con = conn)
+        pool::dbExecute(conn, qry)
+
+        result[[station]] <- list(start = lubridate::as_datetime(start),
+                                  end = lubridate::as_datetime(end),
+                                  removed = removed)
+    }
+
+    return(result)
+}

tests/testthat/helper.R

@@ -0,0 +1,27 @@
+######################################################################
+# test helper functions
+######################################################################
+
+# insert a cache record with an explicit download time
+seed_cache <- function(station, start, end, time_dl, conn) {
+    qry <- glue::glue_sql(
+        "INSERT INTO cache (station, start, end, time_dl) VALUES ({station}, {start}, {end}, {time_dl});",
+        .con = conn)
+    pool::dbExecute(conn, qry)
+}
+
+# insert a single measurement
+seed_measurement <- function(station, timestamp, conn) {
+    qry <- glue::glue_sql(
+        "INSERT INTO measurements (station, parameter, value, aggregation, timestamp) VALUES ({station}, 'pm25', 1.0, 3600, {timestamp});",
+        .con = conn)
+    pool::dbExecute(conn, qry)
+}
+
+# insert a location record
+seed_location <- function(station, conn) {
+    qry <- glue::glue_sql(
+        "INSERT INTO location (station, lat, lon, timestamp) VALUES ({station}, 52.0, 5.0, 0);",
+        .con = conn)
+    pool::dbExecute(conn, qry)
+}

tests/testthat/test-remove_time_ranges_before.R

@@ -0,0 +1,156 @@
+######################################################################
+# tests for remove_time_ranges_before
+#
+# the seed_* helper functions used below live in helper.R
+######################################################################
+
+
+test_that("remove_time_ranges_before checks input", {
+    create_database_tables(dbconn)
+    expect_error(remove_time_ranges_before("2022-01-01", conn = dbconn))
+    drop_database_tables(dbconn)
+})
+
+
+test_that("remove_time_ranges_before removes measurements inclusive of endpoints", {
+    create_database_tables(dbconn)
+
+    start <- as.numeric(lubridate::as_datetime("2022-02-01 00:00:00"))
+    end <- as.numeric(lubridate::as_datetime("2022-02-02 00:00:00"))
+    dl <- as.numeric(lubridate::as_datetime("2022-03-01 00:00:00"))
+
+    seed_location("test-1", dbconn)
+    seed_cache("test-1", start, end, dl, dbconn)
+
+    # measurement before range (kept), at start (removed), inside (removed),
+    # at end (removed) and after range (kept)
+    seed_measurement("test-1", start - 3600, dbconn)
+    seed_measurement("test-1", start, dbconn)
+    seed_measurement("test-1", start + 3600, dbconn)
+    seed_measurement("test-1", end, dbconn)
+    seed_measurement("test-1", end + 3600, dbconn)
+
+    res <- remove_time_ranges_before(lubridate::as_datetime("2022-04-01"), conn = dbconn)
+
+    tbls <- get_db_tables(dbconn)
+    remaining <- tbls$measurements$timestamp
+
+    expect_true((start - 3600) %in% remaining)
+    expect_true((end + 3600) %in% remaining)
+    expect_false(start %in% remaining)
+    expect_false((start + 3600) %in% remaining)
+    expect_false(end %in% remaining)
+    expect_equal(length(remaining), 2)
+
+    # station still has measurements, so it is not removed from location
+    expect_true("test-1" %in% tbls$location$station)
+    expect_false(res[["test-1"]]$removed)
+
+    drop_database_tables(dbconn)
+})
+
+
+test_that("remove_time_ranges_before removes station from location when empty", {
+    create_database_tables(dbconn)
+
+    start <- as.numeric(lubridate::as_datetime("2022-02-01 00:00:00"))
+    end <- as.numeric(lubridate::as_datetime("2022-02-02 00:00:00"))
+    dl <- as.numeric(lubridate::as_datetime("2022-03-01 00:00:00"))
+
+    seed_location("test-1", dbconn)
+    seed_cache("test-1", start, end, dl, dbconn)
+
+    # all measurements fall within the removed range
+    seed_measurement("test-1", start, dbconn)
+    seed_measurement("test-1", start + 3600, dbconn)
+    seed_measurement("test-1", end, dbconn)
+
+    res <- remove_time_ranges_before(lubridate::as_datetime("2022-04-01"), conn = dbconn)
+
+    tbls <- get_db_tables(dbconn)
+    expect_equal(nrow(tbls$measurements), 0)
+    expect_false("test-1" %in% tbls$location$station)
+    expect_true(res[["test-1"]]$removed)
+
+    # cache record is removed as well
+    expect_equal(nrow(tbls$cache), 0)
+
+    drop_database_tables(dbconn)
+})
+
+
+test_that("remove_time_ranges_before respects the cutoff date", {
+    create_database_tables(dbconn)
+
+    start <- as.numeric(lubridate::as_datetime("2022-02-01 00:00:00"))
+    end <- as.numeric(lubridate::as_datetime("2022-02-02 00:00:00"))
+    dl_new <- as.numeric(lubridate::as_datetime("2022-05-01 00:00:00"))
+
+    seed_location("test-1", dbconn)
+    seed_cache("test-1", start, end, dl_new, dbconn)
+    seed_measurement("test-1", start + 3600, dbconn)
+
+    # cutoff is before the download time, nothing should be removed
+    res <- remove_time_ranges_before(lubridate::as_datetime("2022-04-01"), conn = dbconn)
+
+    tbls <- get_db_tables(dbconn)
+    expect_equal(length(res), 0)
+    expect_equal(nrow(tbls$measurements), 1)
+    expect_true("test-1" %in% tbls$location$station)
+    expect_equal(nrow(tbls$cache), 1)
+
+    drop_database_tables(dbconn)
+})
+
+
+test_that("remove_time_ranges_before includes ranges downloaded exactly at the cutoff", {
+    create_database_tables(dbconn)
+
+    start <- as.numeric(lubridate::as_datetime("2022-02-01 00:00:00"))
+    end <- as.numeric(lubridate::as_datetime("2022-02-02 00:00:00"))
+    cutoff <- lubridate::as_datetime("2022-04-01")
+    dl_equal <- as.numeric(cutoff)
+
+    seed_location("test-1", dbconn)
+    seed_cache("test-1", start, end, dl_equal, dbconn)
+    seed_measurement("test-1", start + 3600, dbconn)
+
+    # download time is exactly the cutoff, so the range must be removed
+    res <- remove_time_ranges_before(cutoff, conn = dbconn)
+
+    tbls <- get_db_tables(dbconn)
+    expect_equal(length(res), 1)
+    expect_true(res[["test-1"]]$removed)
+    expect_equal(nrow(tbls$measurements), 0)
+    expect_equal(nrow(tbls$cache), 0)
+
+    drop_database_tables(dbconn)
+})
+
+
+test_that("remove_time_ranges_before returns a named list keyed by station", {
+    create_database_tables(dbconn)
+
+    start1 <- as.numeric(lubridate::as_datetime("2022-02-01 00:00:00"))
+    end1 <- as.numeric(lubridate::as_datetime("2022-02-02 00:00:00"))
+    start2 <- as.numeric(lubridate::as_datetime("2022-02-10 00:00:00"))
+    end2 <- as.numeric(lubridate::as_datetime("2022-02-11 00:00:00"))
+    dl <- as.numeric(lubridate::as_datetime("2022-03-01 00:00:00"))
+
+    seed_location("test-1", dbconn)
+    seed_location("test-2", dbconn)
+    seed_cache("test-1", start1, end1, dl, dbconn)
+    seed_cache("test-2", start2, end2, dl, dbconn)
+    seed_measurement("test-1", start1 + 3600, dbconn)
+    seed_measurement("test-2", start2 + 3600, dbconn)
+
+    res <- remove_time_ranges_before(lubridate::as_datetime("2022-04-01"), conn = dbconn)
+
+    expect_setequal(names(res), c("test-1", "test-2"))
+    expect_true(all(c("start", "end", "removed") %in% names(res[["test-1"]])))
+    expect_equal(as.numeric(res[["test-1"]]$start), start1)
+    expect_equal(as.numeric(res[["test-1"]]$end), end1)
+    expect_true(res[["test-1"]]$removed)
+
+    drop_database_tables(dbconn)
+})

vignettes/ATdatabase.Rmd

@@ -656,6 +656,70 @@ function. A few examples of this function are shown above.
 Stay away from the cache table. Let the package handle this table
 
 
+# Remove data
+
+Sometimes data that was downloaded earlier needs to be removed from the
+database, for example when a data source is found to be unreliable or
+when you want to force a fresh download of a certain period. The
+`remove_time_ranges_before` function lets you remove data based on the
+moment it was downloaded.
+
+Every time a time range is stored in the caching table, the download
+time is recorded in the `time_dl` field. The
+`remove_time_ranges_before` function takes a single date as argument
+and removes all cached time ranges (and their measurements) that were
+downloaded *before or equal to* that date.
+
+For each cached time range that was downloaded before or equal to the
+given date, the function:
+
+* removes the measurements of the station between the start and end
+  time of that range, including the measurements exactly at the start
+  and end time;
+* checks whether any measurements for that station remain. If none
+  remain, the station is also removed from the location table;
+* removes the processed record from the caching table.
+
+The function returns a named list, one element per station, containing
+the start and end time of the removed range and a boolean indicating
+whether the station was removed from the location table.
+
+## Example
+
+Suppose we want to remove everything that was downloaded before a
+certain moment. We pass that moment as a datetime object:
+
+```{r eval = FALSE}
+cutoff <- lubridate::now()
+removed <- remove_time_ranges_before(cutoff, conn = dbconn)
+removed
+```
+
+The result is a named list, for example:
+
+```
+$SSK_LH003
+$SSK_LH003$start
+[1] "2022-02-01 00:00:00 UTC"
+
+$SSK_LH003$end
+[1] "2022-02-05 10:00:00 UTC"
+
+$SSK_LH003$removed
+[1] TRUE
+```
+
+Here the measurements for station `SSK_LH003` between the start and end
+time were removed, and because no other measurements for that station
+were left, it was also removed from the location table
+(`removed = TRUE`). If measurements for the station remain outside the
+removed range, `removed` is `FALSE` and the station stays in the
+location table.
+
+Be careful: this operation is destructive and cannot be undone. Make
+sure you really want to remove the data before calling this function.
+
+
 ```{r echo = FALSE, messages = FALSE}
 # closing stuff
 pool::poolClose(dbconn)