R/resolve_cols.R

In mintyr: Grouped and Nested Data Pipelines Built on 'data.table'

# WARNING - Generated by {fusen} from dev/flat_internal.Rmd: do not edit by hand # nolint: line_length_linter.

#' Resolve a column specification to a character vector of column names
#'
#' Accepts integer positions or character names, validates them against
#' `data_names`, and returns unique column names in the order supplied.
#'
#' @param x Numeric indices or character names (or `NULL` if `allow_null`).
#' @param data_names Character vector of available column names.
#' @param arg_name Argument name used in error messages.
#' @param allow_null Logical; if `TRUE`, `NULL` is returned unchanged.
#' @return Character vector of column names, or `NULL`.
#' @import data.table
#' @importFrom stats quantile
#' @noRd
.resolve_cols <- function(x, data_names, arg_name, allow_null = FALSE) {
  if (is.null(x)) {
    if (allow_null) return(NULL)
    stop(sprintf("`%s` must not be NULL.", arg_name), call. = FALSE)
  }
  if (length(x) == 0L)
    stop(sprintf("`%s` must not be empty.", arg_name), call. = FALSE)
  if (anyNA(x))
    stop(sprintf("`%s` must not contain NA.", arg_name), call. = FALSE)

  if (is.numeric(x)) {
    if (any(x != floor(x)))
      stop(sprintf("`%s` must contain whole-number column indices.", arg_name),
           call. = FALSE)
    if (any(x < 1 | x > length(data_names)))
      stop(sprintf(
        "`%s` contains out-of-bounds index/indices (valid range: 1 to %d).",
        arg_name, length(data_names)), call. = FALSE)
    x <- data_names[as.integer(x)]
  } else if (is.character(x)) {
    missing_cols <- x[!x %chin% data_names]
    if (length(missing_cols) > 0L)
      stop(sprintf("`%s` contains column(s) not found in `data`: %s",
                   arg_name, paste(missing_cols, collapse = ", ")), call. = FALSE)
  } else {
    stop(sprintf(paste0("`%s` must be a numeric index vector or a character ",
                        "column-name vector; got: %s"), arg_name, class(x)[1L]),
         call. = FALSE)
  }

  if (anyDuplicated(x)) {
    warning(sprintf("`%s` contains duplicated column(s); duplicates were removed: %s",
                    arg_name, paste(unique(x[duplicated(x)]), collapse = ", ")),
            call. = FALSE)
    x <- unique(x)
  }
  x
}

#' Validate a single choice from a fixed set of strings
#' @noRd
.check_choice <- function(x, choices, arg_name) {
  if (!is.character(x) || length(x) != 1L || is.na(x) || !x %in% choices)
    stop(sprintf("`%s` must be one of %s.", arg_name,
                 paste0("\"", choices, "\"", collapse = ", ")), call. = FALSE)
  x
}

#' Validate a single non-NA logical flag
#' @noRd
.check_flag <- function(x, arg_name) {
  if (!is.logical(x) || length(x) != 1L || is.na(x))
    stop(sprintf("`%s` must be a single TRUE or FALSE.", arg_name), call. = FALSE)
  x
}

#' Validate a single whole number >= `min` and return it as integer
#' @noRd
.check_count <- function(x, arg_name, min = 1L) {
  if (!is.numeric(x) || length(x) != 1L || is.na(x) || x != floor(x) || x < min)
    stop(sprintf("`%s` must be a single integer >= %d.", arg_name, as.integer(min)),
         call. = FALSE)
  as.integer(x)
}

#' Strip the last file extension, path-aware
#'
#' Only the final path component is affected: a dot is treated as the start
#' of an extension only when it is preceded by a character that is not a
#' separator or another dot, and followed by at least one character that is
#' neither a dot nor a separator. Dot-files such as `.bashrc` are left intact,
#' and dots in directory names (e.g. `v1.2/file`) are never touched.
#' @noRd
.strip_ext <- function(x) {
  out <- x
  ok  <- !is.na(x) & nzchar(x)
  out[ok] <- sub("(?<=[^/\\\\.])\\.[^./\\\\]+$", "", x[ok], perl = TRUE)
  out
}

#' Sanitise values for use as a single file or directory name
#'
#' Replaces characters that are illegal on Windows/macOS/Linux, removes
#' trailing dots/spaces, neutralises `.`/`..`, and prefixes Windows reserved
#' device names. Uniqueness is NOT enforced here; see `.make_unique()`.
#' @noRd
.safe_path_segment <- function(x) {
  x <- as.character(x)
  x[is.na(x)] <- "NA"
  x <- gsub("[<>:\"/\\\\|?*[:cntrl:]]", "_", x)
  x <- trimws(x)
  x <- sub("[. ]+$", "", x)            # Windows forbids trailing dots/spaces
  x[!nzchar(x)] <- "unnamed"
  reserved <- grepl("^(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(\\..*)?$", x,
                    ignore.case = TRUE)
  x[reserved] <- paste0("_", x[reserved])
  x
}

#' Make names unique (case-insensitively), keeping each within `max_len`
#' @noRd
.make_unique <- function(x, max_len) {
  out  <- x
  seen <- new.env(hash = TRUE, parent = emptyenv())
  for (i in seq_along(out)) {
    cand <- out[[i]]
    k    <- 1L
    while (exists(tolower(cand), envir = seen, inherits = FALSE)) {
      k      <- k + 1L
      suffix <- paste0("_", k)
      cand   <- paste0(substr(x[[i]], 1L, max_len - nchar(suffix)), suffix)
    }
    out[[i]] <- cand
    assign(tolower(cand), TRUE, envir = seen)
  }
  out
}

#' Evaluate an expression under a temporary RNG seed
#'
#' The global `.Random.seed` is restored on exit, so calling code that sets
#' `seed` does not disturb the user's random-number stream.
#' @noRd
.with_seed <- function(seed, expr) {
  if (is.null(seed)) return(expr)
  if (!is.numeric(seed) || length(seed) != 1L || is.na(seed))
    stop("`seed` must be a single number or NULL.", call. = FALSE)
  genv     <- globalenv()
  had_seed <- exists(".Random.seed", envir = genv, inherits = FALSE)
  old_seed <- if (had_seed) get(".Random.seed", envir = genv, inherits = FALSE)
  on.exit({
    if (had_seed) {
      assign(".Random.seed", old_seed, envir = genv)
    } else if (exists(".Random.seed", envir = genv, inherits = FALSE)) {
      rm(".Random.seed", envir = genv)
    }
  }, add = TRUE)
  set.seed(seed)
  expr                                   # promise is forced after set.seed()
}

#' Validate `...` forwarded to fwrite(): forbid arguments managed internally
#' @noRd
.check_fwrite_dots <- function(dots, fn) {
  reserved <- intersect(names(dots), c("x", "file", "sep", "na", "quote"))
  if (length(reserved))
    stop(sprintf("[ %s ] Do not pass %s via `...`; use the dedicated argument(s).",
                 fn, paste0("`", reserved, "`", collapse = ", ")), call. = FALSE)
  invisible(TRUE)
}

#' Build stratification groups (numeric binning + pooling of small strata)
#'
#' Numeric variables with more than `breaks` distinct values are cut at
#' quantiles into at most `breaks` bins. Afterwards, while the smallest
#' stratum holds less than `pool` of the rows, it is merged into the
#' next-smallest one.
#' @noRd
.make_strata <- function(x, breaks, pool) {
  if (anyNA(x))
    stop("The `strata` column contains NA values.", call. = FALSE)
  n <- length(x)
  if (is.numeric(x) && length(unique(x)) > breaks) {
    q <- unique(stats::quantile(x, probs = seq(0, 1, length.out = breaks + 1L),
                                names = FALSE))
    x <- cut(x, breaks = q, include.lowest = TRUE)
  }
  s <- as.character(x)
  repeat {
    tab <- sort(table(s))
    if (length(tab) <= 1L || tab[[1L]] / n >= pool) break
    s[s == names(tab)[1L]] <- names(tab)[2L]
  }
  s
}

#' Assign rows to `v` folds of (near) equal size, optionally stratified
#'
#' Unstratified: a random permutation of `rep_len(1:v, n)`.
#' Stratified: rows are shuffled within each stratum, strata are laid out one
#' after another and fold labels are dealt out cyclically, so every stratum
#' is spread evenly across folds and fold sizes differ by at most one.
#' @noRd
.assign_folds <- function(n, v, strata = NULL) {
  if (is.null(strata)) return(sample(rep_len(seq_len(v), n)))
  ord <- unlist(lapply(split(seq_len(n), strata),
                       function(i) i[sample.int(length(i))]), use.names = FALSE)
  folds <- integer(n)
  folds[ord] <- rep_len(sample.int(v), n)
  folds
}

#' V-fold cross-validation index table for one dataset
#'
#' @return data.table with `id` (and `id2` when repeats > 1), `train_idx`,
#'   `validate_idx` and, if `materialize`, `train` / `validate` (data.tables,
#'   or plain data.frames when `out_type = "df"`).
#' @noRd
.vfold_table <- function(x, v, repeats, strata, breaks, pool, materialize,
                         out_type = "dt") {
  n <- nrow(x)
  strata_vec <- if (!is.null(strata)) .make_strata(x[[strata]], breaks, pool)
  fold_lab   <- sprintf(paste0("Fold%0", nchar(v), "d"), seq_len(v))
  rep_lab    <- sprintf(paste0("Repeat%0", nchar(repeats), "d"), seq_len(repeats))

  pieces <- lapply(seq_len(repeats), function(r) {
    f <- .assign_folds(n, v, strata_vec)
    data.table::data.table(
      .rep         = r,
      .fold        = seq_len(v),
      train_idx    = lapply(seq_len(v), function(k) which(f != k)),
      validate_idx = lapply(seq_len(v), function(k) which(f == k))
    )
  })
  out <- data.table::rbindlist(pieces)

  if (repeats == 1L) {
    data.table::set(out, j = "id", value = fold_lab[out$.fold])
  } else {
    data.table::set(out, j = "id",  value = rep_lab[out$.rep])
    data.table::set(out, j = "id2", value = fold_lab[out$.fold])
  }
  data.table::set(out, j = c(".rep", ".fold"), value = NULL)

  if (materialize) {
    # x[i] is a fresh copy, so setDF() can convert it in place (no extra copy).
    take <- if (out_type == "df") function(i) data.table::setDF(x[i]) else function(i) x[i]
    data.table::set(out, j = "train",    value = lapply(out$train_idx,    take))
    data.table::set(out, j = "validate", value = lapply(out$validate_idx, take))
  }
  data.table::setcolorder(out, intersect(c("id", "id2"), names(out)))
  out[]
}

#' Read merged-cell ranges of one worksheet of an .xlsx/.xlsm file
#'
#' Base R only (zip access through `unz()`): the sheet is located via
#' `xl/workbook.xml` and `xl/_rels/workbook.xml.rels`, then the sheet XML is
#' scanned in 1 MB chunks; merge definitions always follow `</sheetData>`, so
#' only that tail is kept in memory. Self-contained (nested helpers only), so
#' it can be shipped to PSOCK workers.
#'
#' @return Integer matrix with columns r1, c1, r2, c2 (one row per merged
#'   range; zero rows if there are none).
#' @noRd
.xlsx_merges <- function(path, sheet) {
  read_entry <- function(entry, chunk = 1048576L) {
    con <- unz(path, entry, open = "rb")
    on.exit(close(con))
    out <- list()
    repeat {
      b <- readBin(con, "raw", chunk)
      if (!length(b)) break
      out[[length(out) + 1L]] <- b
    }
    rawToChar(do.call(c, out))
  }
  tags <- function(txt, el) {
    regmatches(txt, gregexpr(sprintf("<(?:[A-Za-z0-9_]+:)?%s\\s[^>]*>", el), txt,
                             perl = TRUE, useBytes = TRUE))[[1L]]
  }
  attr_of <- function(tg, at) {
    m <- regmatches(tg, regexec(sprintf("\\s%s=\"([^\"]*)\"", at), tg, useBytes = TRUE))
    vapply(m, function(x) if (length(x) == 2L) x[[2L]] else NA_character_, character(1L))
  }
  unescape <- function(x) {
    for (p in list(c("&lt;", "<"), c("&gt;", ">"), c("&quot;", "\""),
                   c("&apos;", "'"), c("&amp;", "&")))
      x <- gsub(p[[1L]], p[[2L]], x, fixed = TRUE)
    x
  }
  none <- matrix(integer(0L), ncol = 4L, dimnames = list(NULL, c("r1", "c1", "r2", "c2")))

  # 1. sheet name -> relationship id -> XML entry
  wb_tags <- tags(read_entry("xl/workbook.xml"), "sheet")
  rid     <- attr_of(wb_tags, "[A-Za-z0-9_]+:id")[match(sheet, unescape(attr_of(wb_tags, "name")))]
  rel     <- tags(read_entry("xl/_rels/workbook.xml.rels"), "Relationship")
  target  <- attr_of(rel, "Target")[match(rid, attr_of(rel, "Id"))]
  if (is.na(target)) stop("worksheet '", sheet, "' not found in the workbook XML")
  entry <- if (startsWith(target, "/")) sub("^/", "", target) else paste0("xl/", target)

  # 2. stream the sheet; keep only what follows </sheetData>
  marker <- charToRaw("</sheetData>")
  con <- unz(path, entry, open = "rb")
  on.exit(close(con))
  carry <- raw(0L)
  tail_parts <- NULL
  repeat {
    b <- readBin(con, "raw", 1048576L)
    if (!length(b)) break
    if (is.null(tail_parts)) {
      b   <- c(carry, b)
      pos <- grepRaw(marker, b, fixed = TRUE)
      if (length(pos)) {
        tail_parts <- list(b[pos:length(b)])
      } else {
        carry <- b[max(1L, length(b) - length(marker) + 2L):length(b)]
      }
    } else {
      tail_parts[[length(tail_parts) + 1L]] <- b
    }
  }
  if (is.null(tail_parts)) return(none)
  refs <- attr_of(tags(rawToChar(do.call(c, tail_parts)), "mergeCell"), "ref")
  refs <- refs[!is.na(refs)]
  if (!length(refs)) return(none)

  # 3. "D2:E3" -> row/column numbers
  col_num <- function(letters) {
    vapply(strsplit(letters, ""), function(ch) {
      v <- 0L
      for (x in match(ch, LETTERS)) v <- v * 26L + x
      v
    }, integer(1L))
  }
  ends <- do.call(rbind, strsplit(sub("^([^:]+)$", "\\1:\\1", refs), ":", fixed = TRUE))
  cell <- function(x) cbind(as.integer(sub("^[A-Z]+", "", x)), col_num(sub("[0-9]+$", "", x)))
  tl <- cell(ends[, 1L])
  br <- cell(ends[, 2L])
  out <- cbind(tl[, 1L], tl[, 2L], br[, 1L], br[, 2L])
  storage.mode(out) <- "integer"
  dimnames(out) <- dimnames(none)
  out
}

#' Combine a multi-row Excel header into one name per column
#'
#' `hdr` holds the header rows (one row per level, read as text, column 1 =
#' Excel column A). Two ways to resolve merged cells, whose value is only
#' stored in their top-left cell:
#' * `merges` (from `.xlsx_merges()`) given: every merged range starting in the
#'   header block is filled with its value, horizontally and vertically.
#'   Cells that are genuinely empty stay empty. `first_row` is the Excel row
#'   number of the first header row.
#' * otherwise `fill = "right"`: every level except the last is filled to the
#'   right, restarting wherever a higher level starts a new label (heuristic).
#'   `fill = "none"` disables filling.
#' Per column, empty parts and consecutive duplicates are dropped and the rest
#' joined with `sep`. Base R only, so it can be shipped to PSOCK workers.
#' @noRd
.combine_headers <- function(hdr, sep = "_", fill = "right", merges = NULL,
                             first_row = 1L) {
  m <- as.matrix(as.data.frame(hdr, stringsAsFactors = FALSE))
  if (length(m) == 0L && (is.null(merges) || !nrow(merges))) return(character(0L))
  storage.mode(m) <- "character"
  m <- trimws(gsub("[\r\n]+", " ", m))
  m[!is.na(m) & !nzchar(m)] <- NA_character_
  n_lev <- nrow(m)

  if (!is.null(merges)) {
    last_row <- first_row + n_lev - 1L
    inside   <- merges[, "r1"] >= first_row & merges[, "r1"] <= last_row
    merges   <- merges[inside, , drop = FALSE]
    if (nrow(merges)) {
      width <- max(ncol(m), merges[, "c2"])
      if (width > ncol(m))
        m <- cbind(m, matrix(NA_character_, n_lev, width - ncol(m)))
      for (k in seq_len(nrow(merges))) {
        rr <- (merges[k, "r1"]:min(merges[k, "r2"], last_row)) - first_row + 1L
        cc <- merges[k, "c1"]:merges[k, "c2"]
        m[rr, cc] <- m[rr[1L], cc[1L]]
      }
    }
  } else if (fill == "right" && n_lev > 1L) {
    n_col <- ncol(m)
    orig  <- !is.na(m)
    for (k in seq_len(n_lev - 1L)) {
      starts <- if (k == 1L) logical(n_col)
                else colSums(orig[seq_len(k - 1L), , drop = FALSE]) > 0
      cur <- NA_character_
      for (j in seq_len(n_col)) {
        if (starts[j]) cur <- NA_character_
        if (is.na(m[k, j])) m[k, j] <- cur else cur <- m[k, j]
      }
    }
  }

  nm <- vapply(seq_len(ncol(m)), function(j) {
    p <- m[, j]
    p <- p[!is.na(p)]
    if (length(p) > 1L) p <- p[c(TRUE, p[-1L] != p[-length(p)])]
    paste(p, collapse = sep)
  }, character(1L))
  empty <- !nzchar(nm)
  nm[empty] <- paste0("col_", which(empty))
  make.unique(nm, sep = sep)
}

#' Statistics per group and variable: shared engine of desc_stats() and
#' top_perc() (internal)
#'
#' Works column by column on the wide table (no melt, so memory stays at
#' the size of the input). `n`, `mean`, `sd`, `min`, `max` and `median` are
#' computed with data.table's GForce (C-level grouped functions); the other
#' statistics need the group values and are computed in R only when
#' requested. Derived statistics (se, ci, cv, ...) come from the GForce ones.
#'
#' @param dt data.table (read only).
#' @param cols numeric columns to summarise.
#' @param by grouping columns or NULL.
#' @param stats requested statistics (see desc_stats()); "quantile" expands
#'   to one column per `probs`.
#' @param groups optional table of all group keys to report (groups without
#'   any value get n = 0); default: the groups present in `dt`.
#' @return data.table: `by` columns, `.variable`, `.var_ord`, statistics.
#' @noRd
.desc_engine <- function(dt, cols, by, stats, probs = numeric(0), conf_level = 0.95,
                         na_rm = TRUE, min_n = 1L, groups = NULL) {
  q_names <- paste0("q", format(probs * 100, trim = TRUE, drop0trailing = TRUE))
  out_stats <- unlist(lapply(stats, function(s) if (s == "quantile") q_names else s))

  # GForce part: everything the requested and derived statistics need
  need_g <- unique(c(
    intersect(stats, c("mean", "sd", "min", "max", "median")),
    if (any(c("se", "ci", "ci_low", "ci_high", "cv") %in% stats)) "sd",
    if (any(c("ci_low", "ci_high", "cv") %in% stats)) "mean",
    if ("cv" %in% stats) "min"
  ))
  r_stats <- intersect(stats, c("q1", "q3", "iqr", "mad", "skew", "kurt", "quantile"))
  crit <- function(n) ifelse(n > 1, stats::qt((1 + conf_level) / 2, df = pmax(n - 1, 1)),
                             NA_real_)

  if (is.null(groups) && length(by)) groups <- unique(dt[, by, with = FALSE])

  res <- lapply(seq_along(cols), function(k) {
    x <- as.name(cols[k])
    .i_keep <- if (na_rm) bquote(!is.na(.(x))) else TRUE
    g_calls <- lapply(need_g, function(f) as.call(list(as.name(f), x)))
    .j_g <- as.call(c(list(as.name("list"), n = quote(.N)), stats::setNames(g_calls, need_g)))
    r <- if (!length(by) && na_rm && !any(!is.na(dt[[cols[k]]]))) {
      data.table::data.table(n = 0L)        # nothing to summarise (avoids min() warnings)
    } else {
      dt[eval(.i_keep), eval(.j_g), by = by]
    }

    # Every group gets a row first (groups without any value: n = 0), so
    # the extra statistics below can be added with update joins.
    if (length(by)) {
      r <- r[groups, on = by]
    } else if (nrow(r) == 0L) {
      r <- data.table::data.table(n = 0L)
    }
    add_cols <- function(r, extra) {
      if (!length(by)) {
        if (nrow(extra)) for (cn in names(extra)) data.table::set(r, j = cn, value = extra[[cn]])
        return(r)
      }
      new <- setdiff(names(extra), by)
      for (cn in new) data.table::set(r, j = cn, value = rep(extra[[cn]][NA_integer_], nrow(r)))
      m <- r[extra, on = by, which = TRUE]
      for (cn in new) data.table::set(r, i = m, j = cn, value = extra[[cn]])
      r
    }
    if (length(r_stats)) {
      extra <- .grouped_extra(dt, cols[k], by, r_stats, probs, q_names)
      if (!na_rm && nrow(extra)) {
        # With na_rm = FALSE a group containing NA has undefined statistics.
        .i_na <- bquote(is.na(.(x)))
        bad <- dt[eval(.i_na), list(.bad = .N), by = by]
        bad <- bad[bad$.bad > 0L]          # without `by`, 0 rows still give one row
        if (nrow(bad)) {
          hit <- if (length(by)) extra[bad, on = by, which = TRUE, nomatch = 0L] else seq_len(nrow(extra))
          for (st in setdiff(names(extra), by)) data.table::set(extra, i = hit, j = st, value = NA_real_)
        }
      }
      r <- add_cols(r, extra)
    }
    if ("n_miss" %in% stats) {
      .i_na <- bquote(is.na(.(x)))
      r <- add_cols(r, dt[eval(.i_na), list(n_miss = .N), by = by])
    }

    data.table::set(r, i = which(is.na(r$n)), j = "n", value = 0L)
    if ("n_miss" %in% stats) {
      if (!"n_miss" %in% names(r)) data.table::set(r, j = "n_miss", value = 0L)
      data.table::set(r, i = which(is.na(r$n_miss)), j = "n_miss", value = 0L)
    }
    for (st in setdiff(c(need_g, setdiff(out_stats, c("n", "n_miss"))), names(r)))
      data.table::set(r, j = st, value = NA_real_)
    for (st in intersect(c("min", "max", "median"), names(r)))
      if (!is.double(r[[st]])) data.table::set(r, j = st, value = as.numeric(r[[st]]))

    # Derived statistics
    n <- r$n
    if ("se" %in% stats)      data.table::set(r, j = "se", value = r$sd / sqrt(n))
    if (any(c("ci", "ci_low", "ci_high") %in% stats)) {
      half <- crit(n) * r$sd / sqrt(n)
      if ("ci" %in% stats)      data.table::set(r, j = "ci", value = half)
      if ("ci_low" %in% stats)  data.table::set(r, j = "ci_low", value = r$mean - half)
      if ("ci_high" %in% stats) data.table::set(r, j = "ci_high", value = r$mean + half)
    }
    # The CV is only meaningful for strictly positive (ratio-scale) data.
    if ("cv" %in% stats)
      data.table::set(r, j = "cv", value = ifelse(!is.na(r$min) & r$min > 0,
                                                  r$sd / r$mean, NA_real_))

    # Groups without values, or below min_n: keep the counts, hide statistics.
    small <- which(n < max(min_n, 1L))
    if (length(small))
      for (st in setdiff(out_stats, c("n", "n_miss")))
        data.table::set(r, i = small, j = st, value = NA_real_)

    keep <- c(by, out_stats)
    r <- r[, keep, with = FALSE]
    data.table::set(r, j = ".variable", value = rep(cols[k], nrow(r)))
    data.table::set(r, j = ".var_ord", value = rep(k, nrow(r)))
    r
  })
  out <- data.table::rbindlist(res, use.names = TRUE)
  data.table::setcolorder(out, c(by, ".variable", ".var_ord"))
  out[]
}


#' Quantile-based and moment-based statistics for all groups at once
#' (internal)
#'
#' Fully vectorised, no per-group R calls: the non-missing values of one
#' column are sorted within groups once, quantiles are read off by position
#' (type 7, as stats::quantile()), `mad` uses GForce medians of the absolute
#' deviations and `skew` / `kurt` use GForce sums of powered deviations.
#' @return data.table with the `by` columns and the requested statistics,
#'   one row per group that has at least one non-missing value.
#' @noRd
.grouped_extra <- function(dt, col, by, r_stats, probs, q_names) {
  .g <- .v <- .d <- .k <- d2 <- d3 <- d4 <- NULL
  x <- dt[[col]]
  keep <- which(!is.na(x))
  tmp <- if (length(by)) dt[keep, by, with = FALSE] else data.table::data.table(.k = rep(1L, length(keep)))
  keys_by <- if (length(by)) by else ".k"
  data.table::set(tmp, j = ".v", value = as.numeric(x[keep]))
  data.table::setorderv(tmp, c(keys_by, ".v"), na.last = TRUE)
  if (!nrow(tmp)) {
    out <- if (length(by)) dt[0L, by, with = FALSE] else data.table::data.table()
    return(out)
  }
  g     <- data.table::rleidv(tmp, cols = keys_by)
  start <- which(!duplicated(g))
  n     <- diff(c(start, length(g) + 1L))
  v     <- tmp$.v
  q7 <- function(p) {                       # type-7 quantile of every group
    h  <- (n - 1) * p + 1
    lo <- floor(h)
    hi <- pmin(lo + 1, n)
    a  <- v[start + lo - 1L]
    a + (h - lo) * (v[start + hi - 1L] - a)
  }
  out <- if (length(by)) tmp[start, by, with = FALSE] else data.table::data.table(.k = 1L)

  if (any(c("q1", "q3", "iqr") %in% r_stats)) {
    q1 <- q7(0.25); q3 <- q7(0.75)
    if ("q1" %in% r_stats)  data.table::set(out, j = "q1", value = q1)
    if ("q3" %in% r_stats)  data.table::set(out, j = "q3", value = q3)
    if ("iqr" %in% r_stats) data.table::set(out, j = "iqr", value = q3 - q1)
  }
  if ("quantile" %in% r_stats)
    for (i in seq_along(probs)) data.table::set(out, j = q_names[i], value = q7(probs[i]))
  if ("mad" %in% r_stats) {
    med <- q7(0.5)
    dev <- data.table::data.table(.g = g, .d = abs(v - med[g]))
    data.table::set(out, j = "mad", value = 1.4826 * dev[, list(m = stats::median(.d)), by = .g]$m)
  }
  if (any(c("skew", "kurt") %in% r_stats)) {
    mu <- data.table::data.table(.g = g, .v = v)[, list(m = mean(.v)), by = .g]$m
    d  <- v - mu[g]
    mom <- data.table::data.table(.g = g, d2 = d^2, d3 = d^3, d4 = d^4)[
      , list(s2 = sum(d2), s3 = sum(d3), s4 = sum(d4)), by = .g]
    m2 <- mom$s2 / n
    ok <- m2 > 0
    # Adjusted Fisher-Pearson skewness G1 and excess kurtosis G2 (as in SAS,
    # SPSS and Excel's SKEW() / KURT()).
    if ("skew" %in% r_stats) {
      sk <- ifelse(ok & n >= 3L, (mom$s3 / n) / m2^1.5 * sqrt(n * (n - 1)) / (n - 2), NA_real_)
      data.table::set(out, j = "skew", value = sk)
    }
    if ("kurt" %in% r_stats) {
      g2 <- (mom$s4 / n) / m2^2 - 3
      ku <- ifelse(ok & n >= 4L, ((n + 1) * g2 + 6) * (n - 1) / ((n - 2) * (n - 3)), NA_real_)
      data.table::set(out, j = "kurt", value = ku)
    }
  }
  if (!length(by)) out[, .k := NULL]
  out[]
}