perf: verdict lazy par agregats SQL, plus de collect integral des booleens

NEWS.md

@@ -10,6 +10,16 @@
 
 ## Performance
 
+* The lazy verdict no longer loads the boolean table into R: the per-column
+  (n, n_failed) counts come from ONE SQL aggregate scan over the computed
+  slim table (`SUM(CASE WHEN ok THEN 0 ELSE 1 END)`, counting FALSE and NULL
+  alike, exactly the local reducers' semantics), and only the FAILING
+  columns' booleans are collected for the pointblank agent. A green lazy
+  comparison previously collected N x columns logicals (~2 GB of R memory
+  for 4M rows x 125 columns, 32x the figure the code comment promised); it
+  now keeps O(columns) in R, and `res$reponse` carries a constant-size
+  placeholder instead of N collected rows (issue #22).
+
 * Four avoidable scans and transfers are gone from the lazy path (issue #24):
   the two systematic `COUNT(*)` full scans are skipped when nothing consumes
   them (keyed comparison with `check_count: false`; `validate_row_counts()`

R/compare_datasets_from_yaml.R

@@ -690,61 +690,77 @@ compare_datasets_from_yaml <- function(data_reference,
   is_lazy <- is_non_local(cmp)
   cmp_for_agent <- cmp
   if (is_lazy) {
-    # Length guards: paste0(character(0), "__ok") yields "__ok" (recycle0 is
-    # FALSE by default), a phantom name the previous any_of() silently ate
-    suffix_all <- function(cols, suffix) {
-      if (length(cols) == 0) {
-        return(character(0))
-      }
-      paste0(cols, suffix)
-    }
+    # The name helpers are length-guarded against the recycle0 phantom
+    # (paste0(character(0), "__ok") yields "__ok"); all_of() below then
+    # fails loudly instead of silently dropping a phantom name
     val_cols <- c(
-      suffix_all(tol_cols, suffix = datadiff_suffix_ok),
-      suffix_all(eq_cols, suffix = datadiff_suffix_eq),
+      datadiff_ok_col(tol_cols),
+      datadiff_eq_col(eq_cols),
       if (isTRUE(row_validation_info$check_count)) {
         "row_count_ok"
       } else {
         character(0)
       }
     )
+    if (length(val_cols) == 0) {
+      # Nothing to compute or aggregate (no value checks, no row-count flag):
+      # the verdict rests on the structural coverage rows alone, and a
+      # zero-column CREATE TABLE AS SELECT is invalid SQL on some backends
+      lazy_counts   <- list()
+      cmp_for_agent <- data.frame()
+    } else {
     # all_of(): a validation column silently dropped here would later read as
     # NULL by the boolean accessors, i.e. a false all-pass; fail loudly instead
     cmp_slim      <- dplyr::select(cmp, dplyr::all_of(val_cols))
     tmp_tbl_name  <- datadiff_tmp_table_name()
     # compute() sends CREATE TEMP TABLE AS SELECT ... to DuckDB: all computation
     # (join, boolean expressions) happens inside DuckDB's process, with disk
-    # spilling available for the large join.  We then collect() the slim boolean
-    # result into R so that pointblank receives a plain data.frame - avoiding
-    # DuckDB connection-state issues (is_tbl_mssql crash) during interrogation.
+    # spilling available for the large join. The verdict is then derived from
+    # an SQL aggregate scan of this table; only the FAILING boolean columns
+    # are later collected into R (red path) so that pointblank receives a
+    # plain data.frame - avoiding DuckDB connection-state issues
+    # (is_tbl_mssql crash) during interrogation.
     cmp_slim_computed <- dplyr::compute(cmp_slim, name = tmp_tbl_name, temporary = TRUE)
-    # The slim table only feeds the collect() below. Drop it at exit so that
-    # repeated calls on a user-supplied connection do not accumulate temp
-    # tables for the lifetime of that connection. after = FALSE runs the drop
-    # BEFORE the exit handlers registered earlier, in particular before the
-    # dbDisconnect of the private connection on the Arrow path (on.exit
-    # add = TRUE fires FIFO by default, which would drop on a closed
-    # connection); the try() then only masks genuine failures.
+    # The slim table only feeds the aggregate scan and the failing-column
+    # collect below. Drop it at exit so that repeated calls on a user-supplied
+    # connection do not accumulate temp tables for the lifetime of that
+    # connection. after = FALSE runs the drop BEFORE the exit handlers
+    # registered earlier, in particular before the dbDisconnect of the private
+    # connection on the Arrow path (on.exit add = TRUE fires FIFO by default,
+    # which would drop on a closed connection); the try() then only masks
+    # genuine failures.
     on.exit(
       try(
         DBI::dbRemoveTable(dbplyr::remote_con(cmp_slim_computed), tmp_tbl_name),
         silent = TRUE
       ),
       add = TRUE, after = FALSE
     )
-    cmp_for_agent     <- dplyr::collect(cmp_slim_computed)
+    # The verdict needs only (n, n_failed) per column: one aggregate scan in
+    # the database replaces the collect of N x columns booleans into R (the
+    # collect made the green verdict cost O(rows) of R memory, contradicting
+    # the documented promise). Only the FAILING columns are collected later,
+    # on the red path, for the pointblank agent.
+    lazy_counts   <- lazy_boolean_counts(
+      cmp_slim_computed, tol_cols = tol_cols, eq_cols = eq_cols
+    )
+    cmp_for_agent <- data.frame()
+    }
   }
 
   # Faithful, O(columns) record of every check performed, built from the
   # boolean validation columns computed above. It is the SINGLE pass over the
   # data: the verdict and the failing-column sets below both derive from it,
   # so the same booleans are no longer scanned two or three times (and the
-  # local equality booleans no longer recomputed once per scan).
+  # local equality booleans no longer recomputed once per scan). On the lazy
+  # path the counts come from the SQL aggregate scan.
   coverage <- build_coverage(
     tbl = cmp_for_agent, tol_cols = tol_cols, eq_cols = eq_cols,
     missing_in_candidate = missing_in_candidate,
     type_mismatch_cols = type_mismatch_cols,
     row_validation_info = row_validation_info, row_count_ok = row_count_ok,
-    ref_suffix = ref_suffix, na_equal = na_equal
+    ref_suffix = ref_suffix, na_equal = na_equal,
+    counts = if (is_lazy) lazy_counts else NULL
   )
 
   # Fast all-pass short-circuit.
@@ -802,6 +818,32 @@ compare_datasets_from_yaml <- function(data_reference,
         cmp_for_agent, fail$tol, col_rules, ref_suffix, na_equal
       )
     }
+    # Lazy path: collect ONLY the failing boolean columns (plus the row-count
+    # flag consumed by its validation step) for the agent; the passing
+    # majority stays in the database.
+    if (is_lazy) {
+      red_cols <- c(
+        datadiff_ok_col(fail$tol),
+        datadiff_eq_col(fail$eq),
+        if (isTRUE(row_validation_info$check_count)) {
+          "row_count_ok"
+        } else {
+          character(0)
+        }
+      )
+      cmp_for_agent <- if (length(red_cols) > 0) {
+        dplyr::collect(
+          dplyr::select(cmp_slim_computed, dplyr::all_of(red_cols))
+        )
+      } else {
+        # Structural-only failure (missing columns / type mismatches with no
+        # failing value column and no row-count check): seed ONE row so the
+        # dummy FALSE columns added by setup_pointblank_agent carry a failing
+        # unit - a 0-row dummy step would interrogate 0 units and pass,
+        # flipping pointblank::all_passed() against the coverage verdict
+        data.frame(.datadiff_structural = FALSE)
+      }
+    }
     agent <- setup_pointblank_agent(
       cmp_for_agent,
       cols_reference,

R/constants.R

@@ -17,10 +17,18 @@ datadiff_prefix_missing_col <- "__missing_col_"
 datadiff_prefix_type_mismatch <- "__type_mismatch_"
 
 # Column name helpers: the only supported way to build these names.
+# Length-guarded: paste0(character(0), suffix) yields the bare suffix
+# (recycle0 is FALSE by default), a phantom column name.
 datadiff_ok_col <- function(col) {
+  if (length(col) == 0) {
+    return(character(0))
+  }
   paste0(col, datadiff_suffix_ok)
 }
 datadiff_eq_col <- function(col) {
+  if (length(col) == 0) {
+    return(character(0))
+  }
   paste0(col, datadiff_suffix_eq)
 }
 

R/coverage.R

@@ -18,33 +18,87 @@ eq_col_counts <- function(tbl, col, ref_suffix, na_equal) {
   list(n = length(b), n_failed = sum(!b))
 }
 
+# (n, n_failed) for every boolean validation column of a LAZY table, computed
+# in ONE SQL aggregate scan: SUM(CASE WHEN col THEN 0 ELSE 1 END) counts the
+# not-TRUE values (FALSE and NULL alike), matching the local reducers
+# (tolerance: is.na | !ok; equality: !b, NA-free by construction). This is
+# what lets the lazy verdict avoid collecting N x columns booleans into R.
+# Returns a named list keyed by the UNDERLYING column name:
+# list(<col> = list(n, n_failed), ...).
+lazy_boolean_counts <- function(tbl_lazy, tol_cols, eq_cols) {
+  con <- dbplyr::remote_con(tbl_lazy)
+  sub <- dbplyr::sql_render(tbl_lazy)
+  q   <- function(x) as.character(DBI::dbQuoteIdentifier(con, x = x))
+
+  bool_cols <- c(datadiff_ok_col(tol_cols), datadiff_eq_col(eq_cols))
+  underlying <- c(tol_cols, eq_cols)
+  nf_exprs <- vapply(X = seq_along(bool_cols), FUN = function(i) {
+    sprintf("SUM(CASE WHEN %s THEN 0 ELSE 1 END) AS %s",
+            q(bool_cols[i]), q(sprintf("..datadiff_nf_%d", i)))
+  }, FUN.VALUE = character(1))
+
+  sql <- sprintf(
+    "SELECT COUNT(*) AS %s%s FROM (%s) AS %s",
+    q("..datadiff_total"),
+    if (length(nf_exprs) > 0) {
+      paste0(", ", paste(nf_exprs, collapse = ", "))
+    } else {
+      ""
+    },
+    sub, q("datadiff_counts")
+  )
+  row <- DBI::dbGetQuery(con, statement = sql)
+
+  total <- as.numeric(row[["..datadiff_total"]])
+  counts <- vector("list", length(underlying))
+  names(counts) <- underlying
+  for (i in seq_along(underlying)) {
+    nf <- as.numeric(row[[sprintf("..datadiff_nf_%d", i)]])
+    # SUM over an empty table is NULL: zero rows means zero failures
+    counts[[i]] <- list(n = total, n_failed = if (is.na(nf)) {
+      0
+    } else {
+      nf
+    })
+  }
+  counts
+}
+
 #' Build a faithful coverage summary of the checks performed
 #'
 #' @param tbl Local data.frame holding the boolean validation columns (and, on
 #'   the local path, the raw + reference columns for equality recomputation).
+#'   Ignored when `counts` is supplied.
 #' @param tol_cols,eq_cols Character vectors of tolerance / equality columns.
 #' @param missing_in_candidate,type_mismatch_cols Structural failures.
 #' @param row_validation_info List with `check_count`.
 #' @param row_count_ok Logical row-count outcome.
 #' @param ref_suffix Suffix identifying reference columns.
 #' @param na_equal Logical; NA equality semantics for equality columns.
+#' @param counts Optional precomputed per-column counts (named list
+#'   `col -> list(n, n_failed)` covering every tolerance and equality column),
+#'   as produced by `lazy_boolean_counts()`: the lazy path aggregates in SQL
+#'   instead of collecting the booleans.
 #' @return A `datadiff_coverage` data.frame with columns `column`, `check`,
 #'   `n`, `n_failed`, `status` (one row per check performed).
 #' @noRd
 build_coverage <- function(tbl, tol_cols, eq_cols,
                            missing_in_candidate, type_mismatch_cols,
                            row_validation_info, row_count_ok,
-                           ref_suffix, na_equal) {
+                           ref_suffix, na_equal, counts = NULL) {
   columns  <- character(0)
   checks   <- character(0)
-  ns       <- integer(0)
-  n_failed <- integer(0)
+  ns       <- numeric(0)
+  n_failed <- numeric(0)
 
   add <- function(column, check, n, nf) {
     columns  <<- c(columns, column)
     checks   <<- c(checks, check)
-    ns       <<- c(ns, as.integer(n))
-    n_failed <<- c(n_failed, as.integer(nf))
+    # numeric, not integer: the lazy counts come from SQL aggregates and can
+    # exceed 2^31 - 1, where as.integer() would overflow to NA and silently
+    # corrupt the coverage and the verdict
+    ns       <<- c(ns, as.numeric(n))
+    n_failed <<- c(n_failed, as.numeric(nf))
   }
 
   # Existence checks: every common (non-type-mismatch) column gets a col_exists
@@ -54,11 +108,12 @@ build_coverage <- function(tbl, tol_cols, eq_cols,
     add(c, "col_exists", 1L, 0L)
   }
   for (c in tol_cols) {
-    cnt <- tol_col_counts(tbl, col = c)
+    cnt <- counts[[c]] %||% tol_col_counts(tbl, col = c)
     add(c, "tolerance", cnt$n, cnt$n_failed)
   }
   for (c in eq_cols) {
-    cnt <- eq_col_counts(tbl, col = c, ref_suffix = ref_suffix, na_equal = na_equal)
+    cnt <- counts[[c]] %||%
+      eq_col_counts(tbl, col = c, ref_suffix = ref_suffix, na_equal = na_equal)
     add(c, "equality", cnt$n, cnt$n_failed)
   }
   for (c in missing_in_candidate) {