From b94b345622b007fed428778a7ae2459554412961 Mon Sep 17 00:00:00 2001 From: infra-bot Date: Fri, 3 Jul 2026 01:19:33 +0200 Subject: [PATCH] perf: verdict lazy par agregats SQL, plus de collect integral des booleens (issue #22) Le collect() de la table slim chargeait N x (T+E) logicals en RAM R (~2 Go pour 4M lignes x 125 colonnes sur le cas vert), contredisant la promesse without loading data into R memory du commentaire - alors que le verdict tient dans (T+E) paires (n, n_failed). - lazy_boolean_counts() : un scan d'agregats SQL sur la temp table (SUM(CASE WHEN ok THEN 0 ELSE 1 END) compte FALSE et NULL, semantique exacte des reducteurs locaux ; SUM sur table vide -> NULL -> 0) - build_coverage(counts =) : les comptes precalcules remplacent les accesseurs locaux sur le chemin lazy - cas vert : plus aucun collect, l'agent placeholder est de taille constante (nrow(res$reponse$tbl) <= 1 au lieu de N) - cas rouge : seules les colonnes booleennes EN ECHEC sont collectees pour l'agent (+ row_count_ok si check actif), extraits inchanges - les helpers datadiff_ok_col/eq_col deviennent length-guarded : le fantome recycle0 (paste0(character(0), suffixe) == suffixe) a mordu une 2e fois via lazy_boolean_counts ; la garde centrale immunise la classe et le suffix_all local devient inutile --- NEWS.md | 10 ++ R/compare_datasets_from_yaml.R | 88 ++++++++++++----- R/constants.R | 8 ++ R/coverage.R | 69 +++++++++++-- tests/testthat/test-lazy-aggregates.R | 136 ++++++++++++++++++++++++++ vignettes/datadiff.Rmd | 4 + 6 files changed, 285 insertions(+), 30 deletions(-) create mode 100644 tests/testthat/test-lazy-aggregates.R 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)