Repository navigation
feat(r): add a function for lazy R callbacks - #975
Conversation
eitsupi
changed the title
|
Feedback requested on the evaluation boundary: the implementation uses This isolation also means the caller's calling handlers do not apply inside the callback, and warnings are handled in the isolated top-level context unless the callback handles them itself. These semantics are documented. Is this boundary and its handler behavior appropriate for the R API, or would you prefer another approach using public R APIs? |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #975 +/- ##
==========================================
+ Coverage 78.29% 78.44% +0.14%
==========================================
Files 106 106
Lines 16914 17044 +130
Branches 2014 2016 +2
==========================================
+ Hits 13243 13370 +127
- Misses 2429 2432 +3
Partials 1242 1242 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
paleolimbot
left a comment
paleolimbot
left a comment
There was a problem hiding this comment.
Thank you for this!
I have some suggestions about how to simplify the evaluation approach a little, although I think this implementation will also work and is close!
| if (data->release_pending) { | ||
| if (out->release != NULL) { | ||
| out->release(out); | ||
| } | ||
| out->release = NULL; | ||
| // The stream itself may already have been freed by a finalizer during evaluation. | ||
| release_function_array_stream_data(data); | ||
| return EINVAL; | ||
| } |
There was a problem hiding this comment.
Unless I'm misunderstanding here, this is definitely not something that should be happening...a caller is required by the array stream contract to serialize calls and wait until a get_next result before releasing.
There was a problem hiding this comment.
I removed the deferred-release handling.
The docs now state that fun must not release the stream while it is being evaluated.
|
|
||
| struct FunctionArrayStreamData { | ||
| struct ArrowSchema schema; | ||
| SEXP fun; |
There was a problem hiding this comment.
Instead of storing the function, I think it is easier to store the expression and the environment. You can construct both at the R level...create a throwaway environment inheriting from the caller's environment; assign fun to it, then construct an expression that is a call to fun(). In C, then all you have to do is Rf_eval().
| SEXP result = | ||
| PROTECT(R_tryCatch(&function_array_stream_eval_call, callback, classes, | ||
| &function_array_stream_condition_handler, call, NULL, NULL)); |
There was a problem hiding this comment.
You can also do this at the R level (wrap the user function in an R tryCatch or withcallinghandlers before assigning it into the env that you save), which is a little simpler.
| // R_tryCatch() does not catch restart jumps or failures in conditionMessage(). | ||
| // All R API calls must stay inside this outer boundary. | ||
| if (!R_ToplevelExec(&function_array_stream_eval, &call)) { | ||
| function_array_stream_set_error( | ||
| data, EIO, "R callback exited non-locally (e.g., via an abort restart)"); | ||
| call.code = data->error_code; | ||
| } | ||
| data->evaluating = 0; |
There was a problem hiding this comment.
I'm not familiar with the distiction between R_TopLevelExec and a straight Rf_eval()...if we do need it, this comment would be a good place to explain it!
| function_array_stream_get_next_wrapper <- function(fun, schema, validate) { | ||
| result <- fun() |
There was a problem hiding this comment.
This is where I think you can do the tryCatch wrapper. At the C level you may still need an R_UnwindProtect in the event something about the tryCatch itself goes sidways; however at least for the first implementation I think it is better to keep the error handling in cushy R code (it can be moved to C later if based on a benchmark it's too slow).
Build the callback environment and its tryCatch() wrapper in R so the C stream only preserves and evaluates an expression. Keep R_ToplevelExec() as the boundary that stops other non-local exits, and drop support for releasing the stream from its own callback.
3 participants
Add
function_array_stream(fun, schema, validate = TRUE)to create an Arrow array stream that requests batches lazily from an R function. Construction and schema access do not call the function; returningNULLends the stream. The explicit schema is copied at construction, and batches are converted and optionally validated against it.validatemust be a single non-missing logical value.The stream retains callback state across export and garbage collection and supports
array_stream_set_finalizer(). The callback is wrapped in R withtryCatch(), so errors and interrupts become terminal stream errors. C evaluates the wrapper insideR_ToplevelExec(), so other non-local exits such as anabortrestart cannot escape through the C stream interface. Reentrant calls are rejected, and callback evaluation is restricted to the R main thread.funmust not release the stream while it is being evaluated. Tests cover empty and multi-batch streams, coercion and validation, errors and interrupts, abort restarts, warnings, export, finalizers, release/GC behavior, and release from a non-R thread.Related to #386
Assisted-by: Opus 5.5 and GPT-6.1 Sol