You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by np...@apache.org on 2019/11/16 03:49:34 UTC
[arrow] branch master updated: ARROW-7186: [R] Add inline comments
to document the dplyr code
This is an automated email from the ASF dual-hosted git repository.
npr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new e869581 ARROW-7186: [R] Add inline comments to document the dplyr code
e869581 is described below
commit e869581e6872043a1ab5525a55608ef854060604
Author: Neal Richardson <ne...@gmail.com>
AuthorDate: Fri Nov 15 19:49:18 2019 -0800
ARROW-7186: [R] Add inline comments to document the dplyr code
In addition to adding comments to help explain how the machinery works, this PR also improves the messaging when a user tries to filter Arrow data with an expression that isn't yet supported, and it raises an error when that happens on a Dataset instead of a warning, under the assumption that you want to explicitly control when to pull data from a big multi-file dataset. (https://issues.apache.org/jira/browse/ARROW-7095 is a followup to be more clever about unsupported expressions.)
Closes #5847 from nealrichardson/dplyr-inline-docs and squashes the following commits:
42a7371bb <Neal Richardson> Apply any good filters before collecting and applying bad ones
236fc7605 <Neal Richardson> Better error message when filter expression isn't supported
8d1420227 <Neal Richardson> Add a bunch of comments to dplyr.R
Authored-by: Neal Richardson <ne...@gmail.com>
Signed-off-by: Neal Richardson <ne...@gmail.com>
---
r/NAMESPACE | 2 +
r/R/arrow-package.R | 4 +-
r/R/dplyr.R | 122 +++++++++++++++++++++++++------------
r/R/util.R | 4 +-
r/tests/testthat/test-dataset.R | 9 +++
r/tests/testthat/test-dplyr.R | 3 +-
r/tests/testthat/test-expression.R | 3 +-
7 files changed, 102 insertions(+), 45 deletions(-)
diff --git a/r/NAMESPACE b/r/NAMESPACE
index 798b939..77aa722 100644
--- a/r/NAMESPACE
+++ b/r/NAMESPACE
@@ -190,10 +190,12 @@ importFrom(bit64,str.integer64)
importFrom(methods,as)
importFrom(purrr,map)
importFrom(purrr,map2)
+importFrom(purrr,map_chr)
importFrom(purrr,map_int)
importFrom(purrr,map_lgl)
importFrom(rlang,"%||%")
importFrom(rlang,abort)
+importFrom(rlang,as_label)
importFrom(rlang,dots_n)
importFrom(rlang,enquo)
importFrom(rlang,enquos)
diff --git a/r/R/arrow-package.R b/r/R/arrow-package.R
index da6a018..26f1233 100644
--- a/r/R/arrow-package.R
+++ b/r/R/arrow-package.R
@@ -16,9 +16,9 @@
# under the License.
#' @importFrom R6 R6Class
-#' @importFrom purrr map map_int map_lgl map2
+#' @importFrom purrr map map_int map_lgl map_chr map2
#' @importFrom assertthat assert_that
-#' @importFrom rlang list2 %||% is_false abort dots_n warn enquo quo_is_null enquos is_integerish quos eval_tidy new_data_mask syms env
+#' @importFrom rlang list2 %||% is_false abort dots_n warn enquo quo_is_null enquos is_integerish quos eval_tidy new_data_mask syms env as_label
#' @importFrom Rcpp sourceCpp
#' @importFrom tidyselect vars_select
#' @useDynLib arrow, .registration = TRUE
diff --git a/r/R/dplyr.R b/r/R/dplyr.R
index c244ac9..22b2755 100644
--- a/r/R/dplyr.R
+++ b/r/R/dplyr.R
@@ -19,39 +19,60 @@
#' @include table.R
arrow_dplyr_query <- function(.data) {
+ # An arrow_dplyr_query is a container for an Arrow data object (Table,
+ # RecordBatch, or Dataset) and the state of the user's dplyr query--things
+ # like selected colums, filters, and group vars.
+
+ # For most dplyr methods,
+ # method.Table == method.RecordBatch == method.Dataset == method.arrow_dplyr_query
+ # This works because the functions all pass .data through arrow_dplyr_query()
if (inherits(.data, "arrow_dplyr_query")) {
return(.data)
}
structure(
list(
.data = .data$clone(),
+ # selected_columns is a named character vector:
+ # * vector contents are the names of the columns in the data
+ # * vector names are the names they should be in the end (i.e. this
+ # records any renaming)
selected_columns = stats::setNames(names(.data), names(.data)),
+ # filtered_rows will be a ComparisonExpression
filtered_rows = TRUE,
+ # group_by_vars is a character vector of columns (as renamed)
+ # in the data. They will be kept when data is pulled into R.
group_by_vars = character()
),
class = "arrow_dplyr_query"
)
}
+# These are the names reflecting all select/rename, not what is in Arrow
+names.arrow_dplyr_query <- function(x) names(x$selected_columns)
+
+# The following S3 methods are registered on load if dplyr is present
select.arrow_dplyr_query <- function(.data, ...) {
- # This S3 method is registered on load if dplyr is present
column_select(arrow_dplyr_query(.data), !!!enquos(...))
}
select.Dataset <- select.Table <- select.RecordBatch <- select.arrow_dplyr_query
#' @importFrom tidyselect vars_rename
rename.arrow_dplyr_query <- function(.data, ...) {
- # This S3 method is registered on load if dplyr is present
column_select(arrow_dplyr_query(.data), !!!enquos(...), .FUN = vars_rename)
}
rename.Dataset <- rename.Table <- rename.RecordBatch <- rename.arrow_dplyr_query
column_select <- function(.data, ..., .FUN = vars_select) {
- out <- .FUN(names(.data$selected_columns), !!!enquos(...))
- # Make sure that the resulting selected columns map back to the original data
+ # .FUN is either tidyselect::vars_select or tidyselect::vars_rename
+ # It operates on the names() of selected_columns, i.e. the column names
+ # factoring in any renaming that may already have happened
+ out <- .FUN(names(.data), !!!enquos(...))
+ # Make sure that the resulting selected columns map back to the original data,
# as in when there are multiple renaming steps
.data$selected_columns <- stats::setNames(.data$selected_columns[out], names(out))
+ # If we've renamed columns, we need to project that renaming into other
+ # query parameters we've collected
renamed <- out[names(out) != out]
if (length(renamed)) {
# Massage group_by
@@ -59,69 +80,91 @@ column_select <- function(.data, ..., .FUN = vars_select) {
renamed_groups <- gbv %in% renamed
gbv[renamed_groups] <- names(renamed)[match(gbv[renamed_groups], renamed)]
.data$group_by_vars <- gbv
- # No need to massage filters because those contain pointers to Arrays
+ # No need to massage filters because those contain references to Arrow objects
}
.data
}
filter.arrow_dplyr_query <- function(.data, ..., .preserve = FALSE) {
- # This S3 method is registered on load if dplyr is present
# TODO something with the .preserve argument
filts <- quos(...)
if (length(filts) == 0) {
# Nothing to do
return(.data)
}
+
.data <- arrow_dplyr_query(.data)
- # Eval filters to generate Expressions with references to Arrays.
- filter_data <- env()
data_is_dataset <- inherits(.data$.data, "Dataset")
+
+ # The filter() method works by evaluating the filters to generate Expressions
+ # with references to Arrays (if .data is Table/RecordBatch) or Fields (if
+ # .data is a Dataset).
+ filter_data <- env()
for (v in unique(unlist(lapply(filts, all.vars)))) {
- # Map any renamed vars to their name in the underlying Arrow schema
- if (!(v %in% names(.data$selected_columns))) {
+ if (!(v %in% names(.data))) {
stop("object '", v, "' not found", call. = FALSE)
}
+ # Map any renamed vars to their name in the underlying Arrow schema
old_var_name <- .data$selected_columns[v]
if (data_is_dataset) {
# Make a FieldExpression
this <- FieldExpression$create(old_var_name)
} else {
- # Get the Array
+ # Get the (reference to the) Array
this <- .data$.data[[old_var_name]]
}
assign(v, this, envir = filter_data)
}
dm <- new_data_mask(filter_data)
- filters <- try(lapply(filts, function (f) {
- # This should yield an Expression
- eval_tidy(f, dm)
- }), silent = FALSE)
- # If that errored, bail out and collect(), with a warning
- # TODO: consider re-evaling with the as.vector Arrays and yielding logical vector
- if (inherits(filters, "try-error")) {
- # TODO: only show this in some debug mode?
- # TODO: if data_is_dataset, don't auto-collect?
- warning(
- "Filter expression not implemented in arrow, pulling data into R",
- immediate. = TRUE,
- call. = FALSE
- )
- return(dplyr::filter(dplyr::collect(.data), ...))
+ filters <- lapply(filts, function (f) {
+ # This should yield an Expression as long as the filter function(s) are
+ # implemented in Arrow.
+ try(eval_tidy(f, dm), silent = TRUE)
+ })
+
+ bad_filters <- map_lgl(filters, ~inherits(., "try-error"))
+ if (any(bad_filters)) {
+ # TODO: consider ways not to bail out here, e.g. defer evaluating the
+ # unsupported filters until after the data has been pulled to R.
+ # See https://issues.apache.org/jira/browse/ARROW-7095
+ bads <- oxford_paste(map_chr(filts, as_label)[bad_filters], quote = FALSE)
+ if (data_is_dataset) {
+ # Abort. We don't want to auto-collect if this is a Dataset because that
+ # could blow up, too big.
+ stop(
+ "Filter expression not supported for Arrow Datasets: ", bads,
+ call. = FALSE
+ )
+ } else {
+ # TODO: only show this in some debug mode?
+ warning(
+ "Filter expression not implemented in Arrow: ", bads, "; pulling data into R",
+ immediate. = TRUE,
+ call. = FALSE
+ )
+ # Set any valid filters first, then collect and then apply the invalid ones in R
+ .data <- set_filters(.data, filters[!bad_filters])
+ return(dplyr::filter(dplyr::collect(.data), !!!filts[bad_filters]))
+ }
}
- # filters is a list of Expressions. AND them together and return
- new_filter <- Reduce("&", filters)
+ set_filters(.data, filters)
+}
+filter.Dataset <- filter.Table <- filter.RecordBatch <- filter.arrow_dplyr_query
+
+set_filters <- function(.data, expressions) {
+ # expressions is a list of Expressions. AND them together and set them on .data
+ new_filter <- Reduce("&", expressions)
if (isTRUE(.data$filtered_rows)) {
+ # TRUE is default (i.e. no filter yet), so we don't need to & with it
.data$filtered_rows <- new_filter
} else {
.data$filtered_rows <- .data$filtered_rows & new_filter
}
.data
}
-filter.Dataset <- filter.Table <- filter.RecordBatch <- filter.arrow_dplyr_query
collect.arrow_dplyr_query <- function(x, ...) {
- # This S3 method is registered on load if dplyr is present
colnames <- x$selected_columns
# Be sure to retain any group_by vars
gv <- setdiff(dplyr::group_vars(x), names(colnames))
@@ -131,6 +174,7 @@ collect.arrow_dplyr_query <- function(x, ...) {
# Pull only the selected rows and cols into R
if (inherits(x$.data, "Dataset")) {
+ # See dataset.R for Dataset and Scanner(Builder) classes
scanner_builder <- x$.data$NewScan()
scanner_builder$UseThreads()
scanner_builder$Project(colnames)
@@ -139,6 +183,7 @@ collect.arrow_dplyr_query <- function(x, ...) {
}
df <- as.data.frame(scanner_builder$Finish()$ToTable())
} else {
+ # This is a Table/RecordBatch. See record-batch.R for the [ method
df <- as.data.frame(x$.data[x$filtered_rows, colnames])
}
# In case variables were renamed, apply those names
@@ -152,20 +197,20 @@ collect.arrow_dplyr_query <- function(x, ...) {
}
collect.Table <- as.data.frame.Table
collect.RecordBatch <- as.data.frame.RecordBatch
+# We probably don't want someone to try to pull a big multi-file Dataset into
+# R without first filtering/selecting
collect.Dataset <- function(x, ...) stop("not implemented")
#' @importFrom tidyselect vars_pull
pull.arrow_dplyr_query <- function(.data, var = -1) {
- # This S3 method is registered on load if dplyr is present
.data <- arrow_dplyr_query(.data)
- var <- vars_pull(names(.data$selected_columns), !!enquo(var))
+ var <- vars_pull(names(.data), !!enquo(var))
.data$selected_columns <- stats::setNames(.data$selected_columns[var], var)
dplyr::collect(.data)[[1]]
}
pull.Dataset <- pull.Table <- pull.RecordBatch <- pull.arrow_dplyr_query
summarise.arrow_dplyr_query <- function(.data, ...) {
- # This S3 method is registered on load if dplyr is present
# Only retain the columns we need to do our aggregations
vars_to_keep <- unique(c(
unlist(lapply(quos(...), all.vars)), # vars referenced in summarise
@@ -178,37 +223,36 @@ summarise.arrow_dplyr_query <- function(.data, ...) {
summarise.Dataset <- summarise.Table <- summarise.RecordBatch <- summarise.arrow_dplyr_query
group_by.arrow_dplyr_query <- function(.data, ..., add = FALSE) {
- # This S3 method is registered on load if dplyr is present
.data <- arrow_dplyr_query(.data)
.data$group_by_vars <- dplyr::group_by_prepare(.data, ..., add = add)$group_names
.data
}
group_by.Dataset <- group_by.Table <- group_by.RecordBatch <- group_by.arrow_dplyr_query
-# This S3 method is registered on load if dplyr is present
groups.arrow_dplyr_query <- function(x) syms(dplyr::group_vars(x))
groups.Dataset <- groups.Table <- groups.RecordBatch <- function(x) NULL
-# This S3 method is registered on load if dplyr is present
group_vars.arrow_dplyr_query <- function(x) x$group_by_vars
group_vars.Dataset <- group_vars.Table <- group_vars.RecordBatch <- function(x) NULL
ungroup.arrow_dplyr_query <- function(x, ...) {
- # This S3 method is registered on load if dplyr is present
x$group_by_vars <- character()
x
}
ungroup.Dataset <- ungroup.Table <- ungroup.RecordBatch <- force
mutate.arrow_dplyr_query <- function(.data, ...) {
- # This S3 method is registered on load if dplyr is present
+ # TODO: see if we can defer evaluating the expressions and not collect here.
+ # It's different from filters (as currently implemented) because the basic
+ # vector transformation functions aren't yet implemented in Arrow C++.
dplyr::mutate(dplyr::collect(arrow_dplyr_query(.data)), ...)
}
mutate.Table <- mutate.RecordBatch <- mutate.arrow_dplyr_query
mutate.Dataset <- function(.data, ...) stop("not implemented")
+# transmute() "just works" because it calls mutate() internally
+# TODO: add transmute() that does what summarise() does (select only the vars we need)
arrange.arrow_dplyr_query <- function(.data, ...) {
- # This S3 method is registered on load if dplyr is present
dplyr::arrange(dplyr::collect(arrow_dplyr_query(.data)), ...)
}
arrange.Table <- arrange.RecordBatch <- arrange.arrow_dplyr_query
diff --git a/r/R/util.R b/r/R/util.R
index 4318997..da22a4e 100644
--- a/r/R/util.R
+++ b/r/R/util.R
@@ -15,8 +15,8 @@
# specific language governing permissions and limitations
# under the License.
-oxford_paste <- function(x, conjunction = "and") {
- if (is.character(x)) {
+oxford_paste <- function(x, conjunction = "and", quote = TRUE) {
+ if (quote && is.character(x)) {
x <- paste0('"', x, '"')
}
if (length(x) < 2) {
diff --git a/r/tests/testthat/test-dataset.R b/r/tests/testthat/test-dataset.R
index 8ff40af..fd35fa0 100644
--- a/r/tests/testthat/test-dataset.R
+++ b/r/tests/testthat/test-dataset.R
@@ -97,6 +97,15 @@ test_that("Hive partitioning", {
)
})
+test_that("filter() on a dataset won't auto-collect", {
+ ds <- open_dataset(dataset_dir)
+ expect_error(
+ ds %>% filter(int > 6L, dbl > max(dbl)),
+ "Filter expression not supported for Arrow Datasets: dbl > max(dbl)",
+ fixed = TRUE
+ )
+})
+
test_that("Assembling a Dataset manually and getting a Table", {
fs <- LocalFileSystem$create()
selector <- Selector$create(dataset_dir, recursive = TRUE)
diff --git a/r/tests/testthat/test-dplyr.R b/r/tests/testthat/test-dplyr.R
index ae093e0..cbe8c55 100644
--- a/r/tests/testthat/test-dplyr.R
+++ b/r/tests/testthat/test-dplyr.R
@@ -97,7 +97,8 @@ test_that("Filtering with a function that doesn't have an Array/expr method stil
collect(),
tbl
),
- "Filter expression not implemented in arrow, pulling data into R"
+ 'Filter expression not implemented in Arrow: chr %in% c("d", "f"); pulling data into R',
+ fixed = TRUE
)
})
diff --git a/r/tests/testthat/test-expression.R b/r/tests/testthat/test-expression.R
index 476eed5..9d43ae4 100644
--- a/r/tests/testthat/test-expression.R
+++ b/r/tests/testthat/test-expression.R
@@ -55,7 +55,8 @@ test_that("C++ expressions", {
expect_output(
print(f > 4),
# We can do better than this right?
- 'ComparisonExpression\nGREATER(field(f), scalar<double>(4))',
+ # TODO: re-add the detail after ARROW-7172
+ 'ComparisonExpression', #\nGREATER(field(f), scalar<double>(4))',
fixed = TRUE
)
})