feat: Hunt for and clean up intermediate tables

R/digest_to_checksum.R

@@ -73,11 +73,11 @@ digest_to_checksum.default <- function(
       dplyr::across(tidyselect::all_of(col), openssl::md5)
     ) %>%
     dplyr::copy_to(dbplyr::remote_con(.data), df = ., name = unique_table_name("SCDB_digest_to_checksum_helper"))
+  defer_db_cleanup(checksums)
 
   .data <- .data %>%
     dplyr::mutate(id__ = dplyr::row_number()) %>%
-    dplyr::left_join(checksums, by = "id__", copy = TRUE) %>%
-    dplyr::select(!"id__")
+    dplyr::left_join(checksums, by = "id__") %>%
     dplyr::select(!"id__") %>%
     dplyr::compute(unique_table_name("SCDB_digest_to_checksum"))
 

R/interlace.R

@@ -102,6 +102,7 @@ interlace.tbl_sql <- function(tables, by = NULL, colnames = NULL) {
     dplyr::select(!".row") %>%
     dplyr::ungroup() %>%
     dplyr::compute(name = unique_table_name("SCDB_interlace_t"))
+  defer_db_cleanup(t)
 
 
   # Merge data onto the new validities using non-equi joins

R/locks.R

@@ -91,6 +91,7 @@ lock_table <- function(conn, db_table, schema = NULL) {
         ),
         name = unique_table_name("SCDB_lock")
       )
+      defer_db_cleanup(lock)
 
       dplyr::rows_insert(db_lock_table, lock, by = c("schema", "table"), conflict = "ignore", in_place = TRUE)
     },
@@ -174,6 +175,7 @@ unlock_table <- function(conn, db_table, schema = NULL, pid = Sys.getpid()) {
         ),
         name = unique_table_name("SCDB_lock")
       )
+      defer_db_cleanup(lock)
 
       dplyr::rows_delete(db_lock_table, lock, by = c("schema", "table", "pid"), unmatched = "ignore", in_place = TRUE)
     },

R/update_snapshot.R

@@ -173,7 +173,8 @@ update_snapshot <- function(.data, conn, db_table, timestamp, filters = NULL, me
   }
 
   # Once we ensure .data is on the same connection as the target, we compute the checksums
-  .data <- dplyr::compute(digest_to_checksum(.data, col = "checksum"))
+  .data <- digest_to_checksum(.data, col = "checksum")
+  if (!inherits(conn, "SQLiteConnection")) .data <- dplyr::compute(.data) # SQLite was computed in digest_to_checksum
   defer_db_cleanup(.data)
 
   ### Determine the next timestamp in the data (can be NA if none is found)