R/quantile.R

In WinVector/rquery: Relational Query Generator for Data Manipulation at Scale

quantile_col <- function(db, incoming_table_name, probs, ci, qualifiers) {
  q_table_name <- quote_table_name(db, incoming_table_name, qualifiers = qualifiers)
  qcount <- paste0("
     SELECT
        COUNT(1)
     FROM
        ", q_table_name, "
     WHERE
        ", quote_identifier(db, ci), " IS NOT NULL")
  nrows <- as.numeric(rq_get_query(db, qcount)[[1]][[1]])
  # was coming back s integer64 and messing up pmax(), pmin()
  if(nrows<1) {
    return(rep(NA, length(probs)))
  }
  indexes <- round((nrows+0.5)*probs)
  indexes <- pmax(1, indexes)
  indexes <- pmin(nrows, indexes)
  # deal with repeated indexes
  # also make sure index 1 is present so we get something back
  # if there is only one value
  uindexes <- sort(unique(c(1, indexes, nrows)))
  indexes_str <- paste(uindexes, collapse = ", ")
  unpack <- match(indexes, uindexes)
  q <- paste0("
     SELECT
        *
      FROM (
        SELECT
           ", quote_identifier(db, ci), ",
           ROW_NUMBER() OVER (ORDER BY ", quote_identifier(db, ci), ")  AS idx_rquery
        FROM
           ", q_table_name, "
        WHERE
           ", quote_identifier(db, ci), " IS NOT NULL
      ) rquery_qtable
     WHERE
        idx_rquery IN (", indexes_str, ")
     ORDER BY
        idx_rquery")
  r <- rq_get_query(db, q)
  r[[ci]][unpack]
}

# same semantics as DB fn.
quantile_col_d <- function(d, probs, ci) {
  dcol <- d[[ci]][!is.na(d[[ci]])]
  nrows <- length(dcol)
  # was coming back s integer64 and messing up pmax(), pmin()
  if(nrows<1) {
    return(rep(NA, length(probs)))
  }
  indexes <- round((nrows+0.5)*probs)
  indexes <- pmax(1, indexes)
  indexes <- pmin(nrows, indexes)
  # deal with repeated indexes
  # also make sure index 1 is present so we get something back
  # if there is only one value
  uindexes <- sort(unique(c(1, indexes, nrows)))
  indexes_str <- paste(uindexes, collapse = ", ")
  unpack <- match(indexes, uindexes)
  r <- sort(dcol)[uindexes]
  r[unpack]
}

#' Compute quantiles of specified columns
#' (without interpolation, needs a database with window functions).
#'
#' @param db database connection
#' @param incoming_table_name name of table to compute quantiles of
#' @param ... force later arguments to bind by name
#' @param probs numeric, probabilities to compute quantiles of
#' @param probs_name character name for probability column
#' @param cols character, columns to compute quantiles of
#' @param qualifiers optional named ordered vector of strings carrying additional db hierarchy terms, such as schema.
#' @return data.frame of quantiles
#'
#' @seealso \code{\link{quantile_node}}, \code{\link{rsummary}}
#'
#' @export
#'
quantile_cols <- function(db, incoming_table_name,
                          ...,
                          probs = seq(0, 1, 0.25),
                          probs_name = "quantile_probability",
                          cols = rq_colnames(db, incoming_table_name),
                          qualifiers = NULL) {
  wrapr::stop_if_dot_args(substitute(list(...)), "rquery::quantile_cols")
  qtable <- data.frame(probs = probs)
  colnames(qtable) <- probs_name
  for(ci in cols) {
    qi <- quantile_col(db, incoming_table_name, probs, ci,
                       qualifiers = qualifiers)
    qtable[[ci]] <- qi
  }
  qtable
}

quantile_cols_d <- function(d,
                            ...,
                            probs = seq(0, 1, 0.25),
                            probs_name = "quantile_probability",
                            cols = column_names(d)) {
  if(!is.data.frame(d)) {
    stop("rquery::quantile_cols_d: d must be a data.frame")
  }
  wrapr::stop_if_dot_args(substitute(list(...)), "rquery::quantile_cols_d")
  qtable <- data.frame(probs = probs)
  colnames(qtable) <- probs_name
  for(ci in cols) {
    qi <- quantile_col_d(d, probs, ci)
    qtable[[ci]] <- qi
  }
  qtable
}


#' Compute quantiles over non-NULL values
#' (without interpolation, needs a database with window functions).
#'
#' Please see \url{https://github.com/WinVector/rquery/blob/master/extras/Summary_Example.md} for an example.
#'
#' This is a non_sql_node, so please see \code{\link{non_sql_node}} for some of the issues for this node type.
#'
#' @param source source to select from (relop or data.frame).
#' @param cols character, compute quantiles for these columns (NULL indicates all columns).
#' @param ... force later arguments to be bound by name
#' @param probs_name character, column name to write probs in.
#' @param probs numeric quantiles to compute
#' @param tmp_name_source wrapr::mk_tmp_name_source(), temporary name generator.
#' @param temporary logical, if TRUE use temporary tables
#' @param qualifiers optional named ordered vector of strings carrying additional db hierarchy terms, such as schema.
#' @return table of quantiles
#'
#' @seealso \code{\link{quantile_cols}}, \code{\link{rsummary}}, \code{\link{non_sql_node}}
#'
#'
#' @export
#'
quantile_node <- function(source,
                          cols = NULL,
                          ...,
                          probs_name = "quantile_probability",
                          probs = seq(0, 1, 0.25),
                          tmp_name_source = wrapr::mk_tmp_name_source("qn"),
                          temporary = TRUE,
                          qualifiers = NULL) {
  wrapr::stop_if_dot_args(substitute(list(...)), "rquery::quantile_node.relop")
  if(probs_name %in% cols) {
    stop("rquery::quantile_node.relop probs_name must be disjoint from cols")
  }
  have <- column_names(source)
  if(!is.null(cols)) {
    check_have_cols(have, cols, "rquery::quantile_node.relop cols")
  } else {
    cols <- have
  }
  force(cols)
  force(probs_name)
  force(probs)
  force(temporary)
  incoming_table_name = tmp_name_source()
  outgoing_table_name = tmp_name_source()
  f_db <- function(db,
                   incoming_table_name,
                   outgoing_table_name,
                   nd = NULL,
                   ...) {
    qtable <- quantile_cols(db, incoming_table_name,
                            probs = probs,
                            probs_name = probs_name,
                            cols = cols,
                            qualifiers = qualifiers)
    rq_copy_to(db,
               table_name = outgoing_table_name,
               d = qtable,
               overwrite = TRUE,
               temporary = temporary,
               qualifiers = qualifiers)
  }
  f_df <- function(d, nd = NULL) {
    quantile_cols_d(d,
                    probs = probs,
                    probs_name = probs_name,
                    cols = cols)
  }
  nd <- non_sql_node(source,
                     f_db = f_db,
                     f_df = f_df,
                     f_dt = NULL,
                     incoming_table_name = incoming_table_name,
                     incoming_qualifiers = qualifiers,
                     outgoing_table_name = outgoing_table_name,
                     outgoing_qualifiers = qualifiers,
                     columns_produced = c(probs_name, cols),
                     display_form = paste0("quantile_node(.)"),
                     orig_columns = FALSE,
                     temporary = temporary)
  nd
}