diff --git a/NEWS.md b/NEWS.md
index e6da9c0..73be078 100644
--- a/NEWS.md
+++ b/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()`
diff --git a/R/compare_datasets_from_yaml.R b/R/compare_datasets_from_yaml.R
index 021f5a3..91f3565 100644
--- a/R/compare_datasets_from_yaml.R
+++ b/R/compare_datasets_from_yaml.R
@@ -690,40 +690,45 @@ 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),
@@ -731,20 +736,31 @@ compare_datasets_from_yaml <- function(data_reference,
),
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,
diff --git a/R/constants.R b/R/constants.R
index 97d7dd9..91789fd 100644
--- a/R/constants.R
+++ b/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)
}
diff --git a/R/coverage.R b/R/coverage.R
index b656024..0807d09 100644
--- a/R/coverage.R
+++ b/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(
= 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) {
diff --git a/tests/testthat/test-lazy-aggregates.R b/tests/testthat/test-lazy-aggregates.R
new file mode 100644
index 0000000..d385e37
--- /dev/null
+++ b/tests/testthat/test-lazy-aggregates.R
@@ -0,0 +1,136 @@
+# Lazy verdict via SQL aggregates: the coverage counts come from one aggregate
+# scan in the database, and the boolean slim table is only collected for the
+# FAILING columns. A green lazy comparison must not load N rows of booleans
+# into R ("without loading data into R memory" is the documented promise).
+
+test_that("a green lazy comparison does not collect the boolean table", {
+ skip_if_not_installed("duckdb")
+ skip_if_not_installed("dbplyr")
+ con <- DBI::dbConnect(duckdb::duckdb())
+ on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
+
+ n <- 5e4
+ df <- data.frame(id = seq_len(n), a = rnorm(n), b = rnorm(n),
+ s = sample(letters, n, replace = TRUE))
+ duckdb::dbWriteTable(con, "ref_agg", df)
+ duckdb::dbWriteTable(con, "cand_agg", df)
+
+ res <- suppressMessages(compare_datasets_from_yaml(
+ dplyr::tbl(con, "ref_agg"), dplyr::tbl(con, "cand_agg"), key = "id"
+ ))
+ expect_true(res$all_passed)
+ # Coverage counts are intact (they now come from the SQL aggregates)
+ cov <- res$coverage
+ expect_identical(as.integer(cov$n[cov$column == "a" & cov$check == "tolerance"]), as.integer(n))
+ expect_identical(as.integer(sum(cov$n_failed)), 0L)
+ # The agent is the constant-size pass placeholder, not N collected rows
+ expect_lte(nrow(res$reponse$tbl), 1L)
+})
+
+test_that("a red lazy comparison collects only the failing boolean columns", {
+ skip_if_not_installed("duckdb")
+ skip_if_not_installed("dbplyr")
+ con <- DBI::dbConnect(duckdb::duckdb())
+ on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
+
+ n <- 1e4
+ df <- data.frame(id = seq_len(n), a = rnorm(n), b = rnorm(n),
+ s = sample(letters, n, replace = TRUE))
+ df2 <- df
+ df2$a[c(3, 7)] <- 999
+ duckdb::dbWriteTable(con, "ref_red2", df)
+ duckdb::dbWriteTable(con, "cand_red2", df2)
+
+ res <- suppressMessages(compare_datasets_from_yaml(
+ dplyr::tbl(con, "ref_red2"), dplyr::tbl(con, "cand_red2"), key = "id"
+ ))
+ expect_false(res$all_passed)
+
+ # Only the failing tolerance column (plus the row-count flag consumed by
+ # its validation step) reaches the local agent: b__ok and s__eq stay in
+ # the database
+ agent_cols <- names(res$reponse$tbl)
+ expect_true("a__ok" %in% agent_cols)
+ expect_false("b__ok" %in% agent_cols)
+ expect_false("s__eq" %in% agent_cols)
+
+ # The verdict details are unchanged: 2 failing rows on a, everything else
+ # green in the coverage
+ cov <- res$coverage
+ expect_identical(as.integer(cov$n_failed[cov$column == "a" & cov$check == "tolerance"]), 2L)
+ expect_identical(as.integer(cov$n_failed[cov$column == "b" & cov$check == "tolerance"]), 0L)
+ expect_identical(as.integer(cov$n_failed[cov$column == "s" & cov$check == "equality"]), 0L)
+
+ # And the extracts still surface the failing rows
+ ex <- pointblank::get_data_extracts(res$reponse)
+ expect_gte(length(ex), 1L)
+})
+
+test_that("SQL-aggregated counts match the collected booleans on SQLite too", {
+ skip_if_not_installed("RSQLite")
+ skip_if_not_installed("dbplyr")
+ con <- DBI::dbConnect(RSQLite::SQLite(), ":memory:")
+ on.exit(DBI::dbDisconnect(con), add = TRUE)
+
+ df <- data.frame(id = 1:50, x = c(rep(1, 45), rep(2, 5)),
+ s = rep(c("u", "v"), 25))
+ ref <- data.frame(id = 1:50, x = rep(1, 50), s = rep(c("u", "v"), 25))
+ DBI::dbWriteTable(con, "ref_sq", ref)
+ DBI::dbWriteTable(con, "cand_sq", df)
+
+ res <- suppressMessages(compare_datasets_from_yaml(
+ dplyr::tbl(con, "ref_sq"), dplyr::tbl(con, "cand_sq"), key = "id"
+ ))
+ expect_false(res$all_passed)
+ cov <- res$coverage
+ expect_identical(as.integer(cov$n_failed[cov$column == "x" & cov$check == "tolerance"]), 5L)
+ expect_identical(as.integer(cov$n_failed[cov$column == "s" & cov$check == "equality"]), 0L)
+})
+
+test_that("structural-only lazy failure keeps a failing verdict (no value checks)", {
+ skip_if_not_installed("duckdb")
+ skip_if_not_installed("dbplyr")
+ con <- DBI::dbConnect(duckdb::duckdb())
+ on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
+
+ duckdb::dbWriteTable(con, "ref_struct", data.frame(id = 1:3, x = c(1, 2, 3), extra_ref = 1:3))
+ duckdb::dbWriteTable(con, "cand_struct", data.frame(id = 1:3, x = c(1, 2, 3)))
+
+ p <- tempfile(fileext = ".yaml")
+ on.exit(unlink(p), add = TRUE)
+ writeLines(
+ "version: 1\ndefaults: {keys: [id], na_equal: yes}\nrow_validation: {check_count: no}\nby_type:\n numeric: {abs: 0.001}\n",
+ con = p
+ )
+ res <- suppressMessages(suppressWarnings(compare_datasets_from_yaml(
+ dplyr::tbl(con, "ref_struct"), dplyr::tbl(con, "cand_struct"),
+ key = "id", path = p
+ )))
+ # The missing column is the only failure: the agent verdict must agree
+ # with the coverage, not pass on a 0-unit dummy step
+ expect_false(res$all_passed)
+ expect_false(pointblank::all_passed(res$reponse))
+ expect_identical(res$missing_in_candidate, "extra_ref")
+})
+
+test_that("key-only lazy comparison runs without value columns", {
+ skip_if_not_installed("duckdb")
+ skip_if_not_installed("dbplyr")
+ con <- DBI::dbConnect(duckdb::duckdb())
+ on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
+
+ duckdb::dbWriteTable(con, "ref_keys", data.frame(id = 1:3))
+ duckdb::dbWriteTable(con, "cand_keys", data.frame(id = 1:3))
+
+ p <- tempfile(fileext = ".yaml")
+ on.exit(unlink(p), add = TRUE)
+ writeLines(
+ "version: 1\ndefaults: {keys: [id], na_equal: yes}\nrow_validation: {check_count: no}\n",
+ con = p
+ )
+ res <- suppressMessages(compare_datasets_from_yaml(
+ dplyr::tbl(con, "ref_keys"), dplyr::tbl(con, "cand_keys"),
+ key = "id", path = p
+ ))
+ expect_true(res$all_passed)
+})
diff --git a/vignettes/datadiff.Rmd b/vignettes/datadiff.Rmd
index ddb84ad..530fbf5 100644
--- a/vignettes/datadiff.Rmd
+++ b/vignettes/datadiff.Rmd
@@ -165,6 +165,10 @@ failed_rows
The `type = "pass"` variant returns rows that passed all steps. This is
useful to understand the scope of the problem before investigating further.
+Note: on an all-pass LAZY comparison, `reponse` carries a constant-size
+placeholder rather than the source rows (the verdict is computed by SQL
+aggregates without loading the data into R), so sundering is only
+meaningful on a failing comparison.
For each failing tolerance column, the step extracts returned by
`pointblank::get_data_extracts()` (and the CSV download in the HTML report)