Background execution and R process pool

DESCRIPTION

@@ -82,20 +82,20 @@ Collate:
     'groupby.R'
     'info.R'
     'ipc.R'
-    'lazyframe__background.R'
     'lazyframe__groupby.R'
     'lazyframe__lazy.R'
     'namespace.R'
     'options.R'
     'parquet.R'
     'pkg-knitr.R'
     'pkg-nanoarrow.R'
+    'rbackground.R'
     'rlang.R'
     'rust_result.R'
     's3_methods.R'
     'series__series.R'
     'translation.R'
     'vctrs.R'
     'zzz.R'
-Config/rextendr/version: 0.3.1
+Config/rextendr/version: 0.3.1.9000
 VignetteBuilder: knitr

Makefile

@@ -47,14 +47,18 @@ requirements-rs:
 .PHONY: build
 build: ## Compile polars R package with all features and generate Rd files
 	export RPOLARS_ALL_FEATURES=true \
-	&& export RPOLARS_PROFILE=release-optimized \
 	&& Rscript -e 'if (!(require(arrow)&&require(nanoarrow))) warning("could not load arrow/nanoarrow, igonore changes to nanoarrow.Rd"); rextendr::document()'
 
+.PHONY: install
+install:
+	export RPOLARS_ALL_FEATURES=true \
+	&& R CMD INSTALL --no-multiarch --with-keep.source .
+
 .PHONY: all
 all: fmt build test README.md LICENSE.note ## build -> test -> Update README.md, LICENSE.note
 
 .PHONY: docs
-docs: build README.md docs/docs/reference_home.md ## Generate docs
+docs: build install README.md docs/docs/reference_home.md ## Generate docs
 	cp docs/mkdocs.orig.yml docs/mkdocs.yml
 	Rscript -e 'altdoc::update_docs(custom_reference = "docs/make-docs.R")'
 	cd docs && ../$(VENV_BIN)/python3 -m mkdocs build
@@ -76,10 +80,6 @@ LICENSE.note: src/rust/Cargo.lock ## Update LICENSE.note
 test: build ## Run fast unittests
 	Rscript -e 'devtools::load_all(); devtools::test()'
 
-.PHONY: install
-install: ## Install this R package locally
-	Rscript -e 'devtools::install(pkg = ".", dependencies = TRUE)'
-
 .PHONY: fmt
 fmt: fmt-rs fmt-r ## Format files
 

NAMESPACE

@@ -18,12 +18,12 @@ S3method("$",FeatureInfo)
 S3method("$",GroupBy)
 S3method("$",LazyFrame)
 S3method("$",LazyGroupBy)
-S3method("$",PolarsBackgroundHandle)
 S3method("$",ProtoExprArray)
 S3method("$",RField)
 S3method("$",RNullValues)
 S3method("$",RPolarsDataType)
 S3method("$",RPolarsErr)
+S3method("$",RThreadHandle)
 S3method("$",Series)
 S3method("$",VecDataFrame)
 S3method("$",When)
@@ -63,12 +63,12 @@ S3method("[[",FeatureInfo)
 S3method("[[",GroupBy)
 S3method("[[",LazyFrame)
 S3method("[[",LazyGroupBy)
-S3method("[[",PolarsBackgroundHandle)
 S3method("[[",ProtoExprArray)
 S3method("[[",RField)
 S3method("[[",RNullValues)
 S3method("[[",RPolarsDataType)
 S3method("[[",RPolarsErr)
+S3method("[[",RThreadHandle)
 S3method("[[",Series)
 S3method("[[",VecDataFrame)
 S3method("[[",When)
@@ -80,9 +80,9 @@ S3method(.DollarNames,DataFrame)
 S3method(.DollarNames,Expr)
 S3method(.DollarNames,GroupBy)
 S3method(.DollarNames,LazyFrame)
-S3method(.DollarNames,PolarsBackgroundHandle)
 S3method(.DollarNames,RField)
 S3method(.DollarNames,RPolarsErr)
+S3method(.DollarNames,RThreadHandle)
 S3method(.DollarNames,Series)
 S3method(.DollarNames,VecDataFrame)
 S3method(.DollarNames,When)
@@ -91,6 +91,7 @@ S3method(.DollarNames,WhenThenThen)
 S3method(.DollarNames,method_environment)
 S3method(.DollarNames,polars_option_list)
 S3method(as.character,RPolarsErr)
+S3method(as.character,RThreadHandle)
 S3method(as.character,Series)
 S3method(as.data.frame,DataFrame)
 S3method(as.data.frame,LazyFrame)
@@ -128,10 +129,10 @@ S3method(print,GroupBy)
 S3method(print,LazyFrame)
 S3method(print,LazyGroupBy)
 S3method(print,PTime)
-S3method(print,PolarsBackgroundHandle)
 S3method(print,RField)
 S3method(print,RPolarsDataType)
 S3method(print,RPolarsErr)
+S3method(print,RThreadHandle)
 S3method(print,Series)
 S3method(print,When)
 S3method(print,WhenThen)
@@ -149,6 +150,8 @@ S3method(unique,LazyFrame)
 export("%**%")
 export("%**%.Expr")
 export(.pr)
+export(RThreadHandle_is_finished)
+export(RThreadHandle_join)
 export(knit_print.DataFrame)
 export(pl)
 importFrom(stats,median)

R/after-wrappers.R

@@ -82,7 +82,6 @@ extendr_method_to_pure_functions = function(env, class_name = NULL) {
 .pr$GroupBy = NULL # derived from DataFrame in R, has no rust calls
 .pr$LazyFrame = extendr_method_to_pure_functions(LazyFrame)
 .pr$LazyGroupBy = extendr_method_to_pure_functions(LazyGroupBy)
-.pr$PolarsBackgroundHandle = extendr_method_to_pure_functions(PolarsBackgroundHandle)
 .pr$DataType = extendr_method_to_pure_functions(RPolarsDataType)
 .pr$DataTypeVector = extendr_method_to_pure_functions(DataTypeVector)
 .pr$RField = extendr_method_to_pure_functions(RField)
@@ -94,6 +93,7 @@ extendr_method_to_pure_functions = function(env, class_name = NULL) {
 .pr$VecDataFrame = extendr_method_to_pure_functions(VecDataFrame)
 .pr$RNullValues = extendr_method_to_pure_functions(RNullValues)
 .pr$RPolarsErr = extendr_method_to_pure_functions(RPolarsErr)
+.pr$RThreadHandle = extendr_method_to_pure_functions(RThreadHandle)
 
 
 
@@ -265,7 +265,7 @@ DataType = clone_env_one_level_deep(RPolarsDataType)
 # used for printing public environment
 pl_class_names = sort(
   c(
-    "LazyFrame", "Series", "LazyGroupBy", "DataType", "Expr", "DataFrame", "PolarsBackgroundHandle",
+    "LazyFrame", "Series", "LazyGroupBy", "DataType", "Expr", "DataFrame",
     "When", "WhenThen", "WhenThenThen"
   )
 ) # TODO discover all public class automatically

R/expr__expr.R

@@ -644,31 +644,40 @@ construct_ProtoExprArray = function(...) {
 ## TODO Contribute polars, seems polars now prefer word f or function in map/apply/rolling/apply
 # over lambda. However lambda is still in examples.
 ## TODO Better explain aggregate list
-#' Expr_map
+#' Map an expression with an R function.
 #' @keywords Expr
 #'
-#' @param f a function mapping a series
+#' @param f a function to map with
 #' @param output_type NULL or one of pl$dtypes$..., the output datatype, NULL is the same as input.
+#' This is used to inform schema of the actual return type of the R function. Setting this wrong
+#' could theoretically have some downstream implications to the query.
 #' @param agg_list Aggregate list. Map from vector to group in groupby context.
-#' Likely not so useful.
+#' @param in_background Boolean. Whether to execute the map in a background R process. Combined wit
+#' setting e.g. `pl$set_global_rpool_cap(4)` it can speed up some slow R functions as they can run
+#' in parallel R sessions. The communication speed between processes is quite slower than between
+#' threads. Will likely only give a speed-up in a "low IO - high CPU" usecase. A single map will not
+#'be paralleled, only in case of multiple `$map`(s) in the query these can be run in parallel.
 #'
-#' @rdname Expr_map
 #' @return Expr
-#' @aliases Expr_map
-#' @details user function return should be a series or any Robj convertible into a Series.
-#' In PyPolars likely return must be Series. User functions do fully support `browser()`, helpful to
-#'  investigate.
+#' @details Sometime some specific R function is just necessary to perform a column transformation.
+#' Using R maps is slower than native polars. User function must take one polars `Series` as input
+#' and the return should be a `Series` or any Robj convertible into a `Series` (e.g. vectors).
+#' Map fully supports `browser()`. If `in_background = FALSE` the function can access any global
+#' variable of the R session. But all R maps in the query sequentially share the same main R
+#' session. Any native polars computations can still be executed meanwhile. In
+#' `in_background = TRUE` the map will run in one or more other R sessions and will not have access
+#' to global variables. Use `pl$set_global_rpool_cap(4)` and `pl$get_global_rpool_cap()` to see and
+#' view number of parallel R sessions.
 #' @name Expr_map
 #' @examples
 #' pl$DataFrame(iris)$select(pl$col("Sepal.Length")$map(\(x) {
 #'   paste("cheese", as.character(x$to_vector()))
 #' }, pl$dtypes$Utf8))
-Expr_map = function(f, output_type = NULL, agg_list = FALSE) {
-  .pr$Expr$map(self, f, output_type, agg_list)
+Expr_map = function(f, output_type = NULL, agg_list = FALSE, in_background = FALSE) {
+  map_fn = ifelse(in_background, .pr$Expr$map_in_background, .pr$Expr$map)
+  map_fn(self, f, output_type, agg_list)
 }
 
-
-
 #' Expr_apply
 #' @keywords Expr
 #'
@@ -683,6 +692,11 @@ Expr_map = function(f, output_type = NULL, agg_list = FALSE) {
 #' if FALSE will convert to a Polars Null and carry on.
 #' @param allow_fail_eval  bool (default FALSE), if TRUE will not raise user function error
 #' but convert result to a polars Null and carry on.
+#' @param in_background Boolean. Whether to execute the map in a background R process. Combined wit
+#' setting e.g. `pl$set_global_rpool_cap(4)` it can speed up some slow R functions as they can run
+#' in parallel R sessions. The communication speed between processes is quite slower than between
+#' threads. Will likely only give a speed-up in a "low IO - high CPU" usecase. A single map will not
+#'be paralleled, only in case of multiple `$map`(s) in the query these can be run in parallel.
 #'
 #' @details
 #'
@@ -773,7 +787,11 @@ Expr_map = function(f, output_type = NULL, agg_list = FALSE) {
 #' system.time({
 #'   r_vec * 2L
 #' })
-Expr_apply = function(f, return_type = NULL, strict_return_type = TRUE, allow_fail_eval = FALSE) {
+Expr_apply = function(f, return_type = NULL, strict_return_type = TRUE, allow_fail_eval = FALSE, in_background = FALSE) {
+  if (in_background) {
+    return(.pr$Expr$apply_in_background(self, f, return_type))
+  }
+
   # use series apply
   wrap_f = function(s) {
     s$apply(f, return_type, strict_return_type, allow_fail_eval)

R/extendr-wrappers.R

@@ -11,8 +11,6 @@
 #' @useDynLib polars, .registration = TRUE
 NULL
 
-test_rpolarserr <- function() .Call(wrap__test_rpolarserr)
-
 rlazy_csv_reader <- function(path, sep, has_header, ignore_errors, skip_rows, n_rows, cache, overwrite_dtype, low_memory, comment_char, quote_char, null_values, infer_schema_length, skip_rows_after_header, encoding, row_count_name, row_count_offset, parse_dates) .Call(wrap__rlazy_csv_reader, path, sep, has_header, ignore_errors, skip_rows, n_rows, cache, overwrite_dtype, low_memory, comment_char, quote_char, null_values, infer_schema_length, skip_rows_after_header, encoding, row_count_name, row_count_offset, parse_dates)
 
 import_arrow_ipc <- function(path, n_rows, cache, rechunk, row_name, row_count, memmap) .Call(wrap__import_arrow_ipc, path, n_rows, cache, rechunk, row_name, row_count, memmap)
@@ -59,43 +57,21 @@ test_robj_to_u32 <- function(robj) .Call(wrap__test_robj_to_u32, robj)
 
 test_print_string <- function(s) invisible(.Call(wrap__test_print_string, s))
 
-RPolarsErr <- new.env(parent = emptyenv())
-
-RPolarsErr$new <- function() .Call(wrap__RPolarsErr__new)
-
-RPolarsErr$contexts <- function() .Call(wrap__RPolarsErr__contexts, self)
-
-RPolarsErr$pretty_msg <- function() .Call(wrap__RPolarsErr__pretty_msg, self)
-
-RPolarsErr$bad_arg <- function(s) .Call(wrap__RPolarsErr__bad_arg, self, s)
-
-RPolarsErr$bad_robj <- function(r) .Call(wrap__RPolarsErr__bad_robj, self, r)
-
-RPolarsErr$bad_val <- function(s) .Call(wrap__RPolarsErr__bad_val, self, s)
-
-RPolarsErr$hint <- function(s) .Call(wrap__RPolarsErr__hint, self, s)
-
-RPolarsErr$mistyped <- function(s) .Call(wrap__RPolarsErr__mistyped, self, s)
-
-RPolarsErr$misvalued <- function(s) .Call(wrap__RPolarsErr__misvalued, self, s)
-
-RPolarsErr$plain <- function(s) .Call(wrap__RPolarsErr__plain, self, s)
+test_rpolarserr <- function() .Call(wrap__test_rpolarserr)
 
-RPolarsErr$rcall <- function(c) .Call(wrap__RPolarsErr__rcall, self, c)
+setup_renv <- function() .Call(wrap__setup_renv)
 
-RPolarsErr$get_rcall <- function() .Call(wrap__RPolarsErr__get_rcall, self)
+set_global_rpool_cap <- function(c) .Call(wrap__set_global_rpool_cap, c)
 
-RPolarsErr$rinfo <- function(i) .Call(wrap__RPolarsErr__rinfo, self, i)
+get_global_rpool_cap <- function() .Call(wrap__get_global_rpool_cap)
 
-RPolarsErr$get_rinfo <- function() .Call(wrap__RPolarsErr__get_rinfo, self)
+handle_background_request <- function(server_name) .Call(wrap__handle_background_request, server_name)
 
-RPolarsErr$when <- function(s) .Call(wrap__RPolarsErr__when, self, s)
+test_rbackgroundhandler <- function(lambda, arg) .Call(wrap__test_rbackgroundhandler, lambda, arg)
 
-#' @export
-`$.RPolarsErr` <- function (self, name) { func <- RPolarsErr[[name]]; environment(func) <- environment(); func }
+test_rthreadhandle <- function() .Call(wrap__test_rthreadhandle)
 
-#' @export
-`[[.RPolarsErr` <- `$.RPolarsErr`
+test_serde_df <- function(df) .Call(wrap__test_serde_df, df)
 
 DataFrame <- new.env(parent = emptyenv())
 
@@ -271,6 +247,58 @@ RField$set_datatype_mut <- function(datatype) invisible(.Call(wrap__RField__set_
 #' @export
 `[[.RField` <- `$.RField`
 
+RPolarsErr <- new.env(parent = emptyenv())
+
+RPolarsErr$new <- function() .Call(wrap__RPolarsErr__new)
+
+RPolarsErr$contexts <- function() .Call(wrap__RPolarsErr__contexts, self)
+
+RPolarsErr$pretty_msg <- function() .Call(wrap__RPolarsErr__pretty_msg, self)
+
+RPolarsErr$bad_arg <- function(s) .Call(wrap__RPolarsErr__bad_arg, self, s)
+
+RPolarsErr$bad_robj <- function(r) .Call(wrap__RPolarsErr__bad_robj, self, r)
+
+RPolarsErr$bad_val <- function(s) .Call(wrap__RPolarsErr__bad_val, self, s)
+
+RPolarsErr$hint <- function(s) .Call(wrap__RPolarsErr__hint, self, s)
+
+RPolarsErr$mistyped <- function(s) .Call(wrap__RPolarsErr__mistyped, self, s)
+
+RPolarsErr$misvalued <- function(s) .Call(wrap__RPolarsErr__misvalued, self, s)
+
+RPolarsErr$plain <- function(s) .Call(wrap__RPolarsErr__plain, self, s)
+
+RPolarsErr$rcall <- function(c) .Call(wrap__RPolarsErr__rcall, self, c)
+
+RPolarsErr$get_rcall <- function() .Call(wrap__RPolarsErr__get_rcall, self)
+
+RPolarsErr$rinfo <- function(i) .Call(wrap__RPolarsErr__rinfo, self, i)
+
+RPolarsErr$get_rinfo <- function() .Call(wrap__RPolarsErr__get_rinfo, self)
+
+RPolarsErr$when <- function(s) .Call(wrap__RPolarsErr__when, self, s)
+
+#' @export
+`$.RPolarsErr` <- function (self, name) { func <- RPolarsErr[[name]]; environment(func) <- environment(); func }
+
+#' @export
+`[[.RPolarsErr` <- `$.RPolarsErr`
+
+RThreadHandle <- new.env(parent = emptyenv())
+
+RThreadHandle$join <- function() .Call(wrap__RThreadHandle__join, self)
+
+RThreadHandle$is_finished <- function() .Call(wrap__RThreadHandle__is_finished, self)
+
+RThreadHandle$thread_description <- function() .Call(wrap__RThreadHandle__thread_description, self)
+
+#' @export
+`$.RThreadHandle` <- function (self, name) { func <- RThreadHandle[[name]]; environment(func) <- environment(); func }
+
+#' @export
+`[[.RThreadHandle` <- `$.RThreadHandle`
+
 Expr <- new.env(parent = emptyenv())
 
 Expr$col <- function(name) .Call(wrap__Expr__col, name)
@@ -687,6 +715,10 @@ Expr$print <- function() invisible(.Call(wrap__Expr__print, self))
 
 Expr$map <- function(lambda, output_type, agg_list) .Call(wrap__Expr__map, self, lambda, output_type, agg_list)
 
+Expr$map_in_background <- function(lambda, output_type, agg_list) .Call(wrap__Expr__map_in_background, self, lambda, output_type, agg_list)
+
+Expr$apply_in_background <- function(lambda, output_type) .Call(wrap__Expr__apply_in_background, self, lambda, output_type)
+
 Expr$is_unique <- function() .Call(wrap__Expr__is_unique, self)
 
 Expr$approx_unique <- function() .Call(wrap__Expr__approx_unique, self)
@@ -877,12 +909,12 @@ LazyFrame$debug_plan <- function() .Call(wrap__LazyFrame__debug_plan, self)
 
 LazyFrame$describe_optimized_plan <- function() .Call(wrap__LazyFrame__describe_optimized_plan, self)
 
-LazyFrame$collect_background <- function() .Call(wrap__LazyFrame__collect_background, self)
-
 LazyFrame$collect <- function() .Call(wrap__LazyFrame__collect, self)
 
 LazyFrame$collect_handled <- function() .Call(wrap__LazyFrame__collect_handled, self)
 
+LazyFrame$collect_in_background <- function() .Call(wrap__LazyFrame__collect_in_background, self)
+
 LazyFrame$first <- function() .Call(wrap__LazyFrame__first, self)
 
 LazyFrame$last <- function() .Call(wrap__LazyFrame__last, self)
@@ -1079,20 +1111,6 @@ Series$from_arrow <- function(name, array) .Call(wrap__Series__from_arrow, name,
 #' @export
 `[[.Series` <- `$.Series`
 
-PolarsBackgroundHandle <- new.env(parent = emptyenv())
-
-PolarsBackgroundHandle$new <- function(lazy_df) .Call(wrap__PolarsBackgroundHandle__new, lazy_df)
-
-PolarsBackgroundHandle$join <- function() .Call(wrap__PolarsBackgroundHandle__join, self)
-
-PolarsBackgroundHandle$is_exhausted <- function() .Call(wrap__PolarsBackgroundHandle__is_exhausted, self)
-
-#' @export
-`$.PolarsBackgroundHandle` <- function (self, name) { func <- PolarsBackgroundHandle[[name]]; environment(func) <- environment(); func }
-
-#' @export
-`[[.PolarsBackgroundHandle` <- `$.PolarsBackgroundHandle`
-
 FeatureInfo <- new.env(parent = emptyenv())
 
 FeatureInfo$new <- function() .Call(wrap__FeatureInfo__new)