R/project.R

In rquery: Relational Query Generator for Data Manipulation at Scale

#' project data by grouping, and adding aggregate columns.
#'
#' @param source source to select from.
#' @param ... force later arugments to bind by name.
#' @param groupby grouping columns.
#' @param parsed new column assignment expressions.
#' @return project node.
#'
#' @noRd
#'
project_impl <- function(source, ...,
                         groupby, parsed) {
  wrapr::stop_if_dot_args(substitute(list(...)),
                          "rquery:::project_impl")
  if(length(groupby)!=length(unique(groupby))) {
    stop("rquery:::project_impl duplicate groupby columns")
  }
  have <- column_names(source)
  required_cols <- sort(unique(c(
    merge_fld(parsed, "symbols_used"),
    merge_fld(parsed, "free_symbols"),
    groupby
  )))
  check_have_cols(have, required_cols, "rquery::project")
  assignments <- list()
  if(length(parsed)>0) {
    produced <- vapply(parsed, function(pi) pi$symbols_produced, character(1))
    if(length(produced)!=length(unique(produced))) {
      stop("rquery:::project_impl can not produce the same column twice in a project")
    }
    parts <- partition_assignments(parsed)
    if(length(parts)>1) {
      stop("rquery:::project_impl can not use produced columns during a project")
    }
    assignments <- unpack_assignments(source, parsed)
  }
  # producing <- names(assignments)
  # overlap <- intersect(have, producing)
  # if(length(overlap)>0) {
  #   stop(paste("rquery:::project_impl produced columns must be disjoint from incoming table: ",
  #              paste(overlap, collapse = ", ")))
  # }
  r <- list(source = list(source),
            table_name = NULL,
            parsed = parsed,
            groupby = groupby,
            columns = c(groupby, names(assignments)),
            required_cols = required_cols,
            assignments = assignments)
  r <- relop_decorate("relop_project", r)
  r
}

#' project data by grouping, and adding aggregate columns.
#'
#' @param source source to select from.
#' @param assignments new column assignment expressions.
#' @param ... not used, force later arguments to be by name
#' @param groupby grouping columns.
#' @param env environment to look for values in.
#' @return project node.
#'
#' @examples
#'
#' if (requireNamespace("DBI", quietly = TRUE) && requireNamespace("RSQLite", quietly = TRUE)) {
#'   my_db <- DBI::dbConnect(RSQLite::SQLite(),
#'                           ":memory:")
#'   d <- rq_copy_to(
#'     my_db, 'd',
#'     data.frame(group = c('a', 'a', 'b', 'b'),
#'                val = 1:4,
#'                stringsAsFactors = FALSE))
#'
#'   op_tree <- d %.>%
#'     project_se(., groupby = "group", "vmax" %:=% "max(val)")
#'   cat(format(op_tree))
#'   sql <- to_sql(op_tree, my_db)
#'   cat(sql)
#'   execute(my_db, op_tree) %.>%
#'      print(.)
#'
#'   op_tree <- d %.>%
#'     project_se(., groupby = NULL, "vmax" %:=% "max(val)")
#'   cat(format(op_tree))
#'   sql <- to_sql(op_tree, my_db)
#'   cat(sql)
#'   execute(my_db, op_tree) %.>%
#'      print(.)
#'
#'   DBI::dbDisconnect(my_db)
#' }
#'
#' @export
#'
project_se <- function(source,
                       assignments,
                       ...,
                       groupby = c(),
                       env = parent.frame()) {
  force(env)
  wrapr::stop_if_dot_args(substitute(list(...)), 'project_se')
  UseMethod("project_se", source)
}

#' @export
#' @rdname project_se
aggregate_se <- project_se

#' @export
project_se.relop <- function(source,
                             assignments,
                             ...,
                             groupby=c(),
                             env = parent.frame()) {
  force(env)
  wrapr::stop_if_dot_args(substitute(list(...)), 'project_se.relop')
  parsed <- parse_se(source, assignments, env = env, allow_empty = TRUE)
  project_impl(source, groupby = groupby, parsed = parsed)
}

#' @export
project_se.data.frame <- function(source,
                                  assignments,
                                  ...,
                                  groupby=c(),
                                  env = parent.frame()) {
  force(env)
  wrapr::stop_if_dot_args(substitute(list(...)), 'project_se.data.frame')
  tmp_name <- mk_tmp_name_source("rquery_tmp")()
  dnode <- mk_td(tmp_name, colnames(source))
  enode <- project_se(dnode, assignments,
                      groupby = groupby,
                      env = env)
  rquery_apply_to_data_frame(source, enode, env = env)
}




#' project data by grouping, and adding aggregate columns.
#'
#' Supports \code{bquote()} \code{.()}-style name abstraction including \code{.(-)} notation to promote strings to names
#' (please see here: \url{https://github.com/WinVector/rquery/blob/master/Examples/Substitution/Substitution.md}).
#'
#' @param source source to select from.
#' @param ... new column assignment expressions.
#' @param groupby grouping columns.
#' @param env environment to look for values in.
#' @return project node.
#'
#' @examples
#'
#' if (requireNamespace("DBI", quietly = TRUE) && requireNamespace("RSQLite", quietly = TRUE)) {
#'   my_db <- DBI::dbConnect(RSQLite::SQLite(),
#'                           ":memory:")
#'   d <- rq_copy_to(
#'     my_db, 'd',
#'     data.frame(group = c('a', 'a', 'b', 'b'),
#'                val = 1:4,
#'                stringsAsFactors = FALSE))
#'
#'   op_tree <- d %.>%
#'     project(., groupby = "group", vmax %:=% max(val))
#'   cat(format(op_tree))
#'   sql <- to_sql(op_tree, my_db)
#'   cat(sql)
#'   execute(my_db, op_tree) %.>%
#'      print(.)
#'
#'   op_tree <- d %.>%
#'     project(., groupby = NULL, vmax %:=% max(val))
#'   cat(format(op_tree))
#'   sql <- to_sql(op_tree, my_db)
#'   cat(sql)
#'   execute(my_db, op_tree) %.>%
#'     print(.)
#'
#'   DBI::dbDisconnect(my_db)
#' }
#'
#' @export
#'
project <- function(source,
                    ...,
                    groupby = c(),
                    env = parent.frame()) {
  force(env)
  UseMethod("project", source)
}

#' @rdname project
#' @export
#'
project_nse <- project


#' @export
#' @rdname project
aggregate_nse <- project

#' @export
project.relop <- function(source,
                          ...,
                          groupby=c(),
                          env = parent.frame()) {
  force(env)
  # Recommend way to caputre ... unevalauted from
  # http://adv-r.had.co.nz/Computing-on-the-language.html#substitute "Capturing unevaluated ..."
  exprs <-  eval(substitute(alist(...)))
  exprs <- lapply_bquote_to_langauge_list(exprs, env)
  parsed <- parse_nse(source, exprs, env = env, allow_empty = TRUE)
  project_impl(source, groupby = groupby, parsed = parsed)
}

#' @export
project.data.frame <- function(source,
                               ...,
                               groupby=c(),
                               env = parent.frame()) {
  force(env)
  tmp_name <- mk_tmp_name_source("rquery_tmp")()
  dnode <- mk_td(tmp_name, colnames(source))
  enode <- project(dnode, ...,
                   groupby = groupby,
                   env = env)
  rquery_apply_to_data_frame(source, enode, env = env)
}


#' @export
column_names.relop_project <- function (x, ...) {
  wrapr::stop_if_dot_args(substitute(list(...)), "rquery::column_names.relop_project")
  x$columns
}


#' @export
format_node.relop_project <- function(node) {
  origTerms <- vapply(node$parsed,
                      function(pi) {
                        paste(as.character(pi$presentation), collapse = ' ')
                      }, character(1))
  aterms <- paste(origTerms, collapse = ", ")
  paste0("project(., ",
         aterms,
         ",\n  groupby = ", wrapr::map_to_char(node$groupby),
         ")",
         "\n")
}

calc_used_relop_project <- function (x,
                                     using = NULL) {
  expressions <- x$parsed
  if(length(using)>0) {
    want_expr <- vapply(x$parsed,
                        function(pi) {
                          length(intersect(pi$symbols_produced, using))>0
                        }, logical(1))
    expressions <- x$parsed[want_expr]
  }
  consuming <- merge_fld(expressions, "symbols_used")
  subusing <- unique(c(consuming, x$groupby, x$orderby))
  subusing
}

#' @export
columns_used.relop_project <- function (x, ...,
                                        using = NULL) {
  wrapr::stop_if_dot_args(substitute(list(...)), "rquery::columns_used.relop_project")
  cols <- calc_used_relop_project(x,
                                  using = using)
  columns_used(x$source[[1]],
               using = cols)
}


#' @export
to_sql.relop_project <- function (x,
                                  db,
                                  ...,
                                  limit = NULL,
                                  source_limit = NULL,
                                  indent_level = 0,
                                  tnum = mk_tmp_name_source('tsql'),
                                  append_cr = TRUE,
                                  using = NULL) {
  wrapr::stop_if_dot_args(substitute(list(...)), "rquery::to_sql.relop_project")
  dispatch_to_sql_method(
    method_name = "to_sql.relop_project",
    x = x,
    db = db,
    limit = limit,
    source_limit = source_limit,
    indent_level = indent_level,
    tnum = tnum,
    append_cr = append_cr,
    using = using)
}


to_sql_relop_project <- function(
  x,
  db,
  ...,
  limit = NULL,
  source_limit = NULL,
  indent_level = 0,
  tnum = mk_tmp_name_source('tsql'),
  append_cr = TRUE,
  using = NULL) {
  wrapr::stop_if_dot_args(substitute(list(...)), "rquery::to_sql_relop_project")
  # re-quote expr
  parsed <- x$parsed
  if(length(using)>0) {
    want_expr <- vapply(x$parsed,
                        function(pi) {
                          length(intersect(pi$symbols_produced, using))>0
                        }, logical(1))
    parsed <- x$parsed[want_expr]
  }
  re_quoted <- redo_parse_quoting(parsed, db)
  re_assignments <- unpack_assignments(x$source[[1]], re_quoted)
  # work on query
  using_incoming <- calc_used_relop_project(x,
                                            using = using)
  subsql_list <- to_sql(x$source[[1]],
                        db = db,
                        source_limit = source_limit,
                        indent_level = indent_level + 1,
                        tnum = tnum,
                        append_cr = FALSE,
                        using = using_incoming)
  subsql <- subsql_list[[length(subsql_list)]]
  grouping_cols <- x$groupby
  if(length(x$groupby)>0) {
    grouping_cols <- vapply(x$groupby,
                            function(ci) {
                              quote_identifier(db, ci)
                            }, character(1))
  }
  extra_cols <- x$groupby
  if(length(using)>0) {
    extra_cols <- intersect(x$groupby, using)
  }
  if(length(extra_cols)>0) {
    extra_cols <- vapply(extra_cols,
                         function(ci) {
                           quote_identifier(db, ci)
                         }, character(1))
  }
  derived <- NULL
  if(length(re_assignments)>0) {
    derived <- vapply(names(re_assignments),
                      function(ni) {
                        ei <- re_assignments[[ni]]
                        paste(ei, "AS", quote_identifier(db, ni))
                      }, character(1))
  }
  tab <- tnum()
  prefix <- paste(rep(' ', indent_level), collapse = '')
  q <- paste0(prefix, "SELECT ",
              paste(c(extra_cols, derived), collapse = ", "),
              " FROM (\n",
              subsql, "\n",
              prefix, " ) ", tab)
  if(length(grouping_cols)>0) {
    q <- paste0(q,
                "\n",
                prefix, "GROUP BY\n",
                prefix, " ", paste(grouping_cols, collapse = ", "))
  }
  if(!is.null(limit)) {
    q <- paste(q, "LIMIT",
               format(ceiling(limit), scientific = FALSE))
  }
  if(append_cr) {
    q <- paste0(q, "\n")
  }
  c(subsql_list[-length(subsql_list)], q)
}