You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "eitsupi (via GitHub)" <gi...@apache.org> on 2026/10/18 01:15:58 UTC
[PR] feat(r): Add function_array_stream() for lazy R callbacks [arrow-nanoarrow]
eitsupi opened a new pull request, #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975
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; returning `NULL` ends the stream. The explicit schema is copied at construction, and batches are converted and optionally validated against it.
The stream retains callback state across export and garbage collection and supports `array_stream_set_finalizer()`. Errors remain terminal, reentrant calls are rejected, and callback evaluation is restricted to the R main thread. Tests cover empty and multi-batch streams, coercion and validation, errors and interrupts, abort restarts, warnings, export, finalizers, and release/GC behavior.
Feedback requested on the evaluation boundary: the implementation uses `R_ToplevelExec()` around `R_tryCatch()` for errors and interrupts. `R_tryCatch()` alone does not contain non-local exits such as `invokeRestart("abort")`; allowing those jumps across an Arrow C Stream callback can bypass a C/C++ consumer's cleanup. The outer boundary converts these exits to EIO.
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?
Validation on R 4.6.1 / Linux: the package test suite passes, and `R CMD check --as-cran --no-manual` reports 0 errors, 0 warnings, and the same compiler-flag NOTE as main. The check test suite reports 1718 passes and 7 skips for unavailable optional reticulate/lz4 support. An additional installed-package lifecycle check completes 100 export/finalizer cycles and collects 100 unreferenced streams with no remaining preserved R objects.
Related to #386.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(r): add a function for lazy R callbacks [arrow-nanoarrow]
Posted by "eitsupi (via GitHub)" <gi...@apache.org>.
eitsupi commented on PR #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975#issuecomment-6344367112
Thanks for merging.
I've opened r-dbi/DBI#769
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(r): add a function for lazy R callbacks [arrow-nanoarrow]
Posted by "eitsupi (via GitHub)" <gi...@apache.org>.
eitsupi commented on code in PR #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975#discussion_r4315916390
##########
r/src/array_stream.c:
##########
@@ -18,13 +18,209 @@
#define R_NO_REMAP
#include <R.h>
#include <Rinternals.h>
+#include <errno.h>
+#include <string.h>
#include "array.h"
#include "array_stream.h"
#include "nanoarrow.h"
#include "schema.h"
#include "util.h"
+struct FunctionArrayStreamData {
+ struct ArrowSchema schema;
+ SEXP fun;
+ SEXP schema_xptr;
+ int validate;
+ struct ArrowError error;
+ int error_code;
+ int finished;
+ int errored;
+ int evaluating;
+ int release_pending;
+};
+
+struct FunctionArrayStreamCall {
+ struct FunctionArrayStreamData* data;
+ struct ArrowArray* out;
+ int code;
+ int caught;
+};
+
+static void release_function_array_stream_data(struct FunctionArrayStreamData* data) {
+ if (data->schema.release != NULL) {
+ data->schema.release(&data->schema);
+ }
+ nanoarrow_release_sexp(data->fun);
+ nanoarrow_release_sexp(data->schema_xptr);
+ ArrowFree(data);
+}
+
+static void function_array_stream_set_error(struct FunctionArrayStreamData* data,
+ int code, const char* message) {
+ if (!data->errored) {
+ ArrowErrorSet(&data->error, "%s", message);
+ data->error_code = code;
+ data->errored = 1;
+ }
+}
+
+static void release_function_array_stream(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ stream->release = NULL;
+ stream->private_data = NULL;
+ if (data->evaluating) {
+ // The callback may release its own stream. Keep its state alive until it returns.
+ data->release_pending = 1;
+ return;
+ }
+ release_function_array_stream_data(data);
+}
+
+static const char* function_array_stream_get_last_error(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return data == NULL ? "Array stream has been released"
+ : ArrowErrorMessage(&data->error);
+}
+
+static int function_array_stream_get_schema(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return ArrowSchemaDeepCopy(&data->schema, out);
+}
+
+static SEXP function_array_stream_condition_handler(SEXP condition, void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ call->caught = 1;
+ if (Rf_inherits(condition, "interrupt")) {
+ call->caught = 2;
+ }
+ SEXP message_call = PROTECT(Rf_lang2(Rf_install("conditionMessage"), condition));
+ SEXP message = Rf_eval(message_call, R_BaseEnv);
+ UNPROTECT(1);
+ return message;
+}
+
+static SEXP function_array_stream_eval_call(void* opaque) {
+ return Rf_eval((SEXP)opaque, nanoarrow_ns_pkg);
+}
+
+static void function_array_stream_eval(void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ struct FunctionArrayStreamData* data = call->data;
+ SEXP classes = PROTECT(Rf_allocVector(STRSXP, 2));
+ SET_STRING_ELT(classes, 0, Rf_mkChar("error"));
+ SET_STRING_ELT(classes, 1, Rf_mkChar("interrupt"));
+ SEXP validate = PROTECT(Rf_ScalarLogical(data->validate));
+ SEXP callback = PROTECT(Rf_lang4(Rf_install("function_array_stream_get_next_wrapper"),
+ data->fun, data->schema_xptr, validate));
+ SEXP result =
+ PROTECT(R_tryCatch(&function_array_stream_eval_call, callback, classes,
+ &function_array_stream_condition_handler, call, NULL, NULL));
+ if (data->errored) {
+ call->code = data->error_code;
+ } else if (call->caught) {
+ const char* message = Rf_translateCharUTF8(STRING_ELT(result, 0));
+ if (call->caught == 2) {
+ struct ArrowError error;
+ ArrowErrorSet(&error, "R callback interrupted: %s", message);
+ function_array_stream_set_error(data, EIO, ArrowErrorMessage(&error));
+ } else {
+ function_array_stream_set_error(data, EIO, message);
+ }
+ call->code = data->error_code;
+ } else if (result == R_NilValue) {
+ data->finished = 1;
+ call->code = NANOARROW_OK;
+ } else {
+ // Let the external pointer clean up a partial export if R jumps during allocation.
+ SEXP array_xptr = PROTECT(nanoarrow_array_owning_xptr());
+ struct ArrowArray* array = nanoarrow_output_array_from_xptr(array_xptr);
+ array_export(result, array);
+ ArrowArrayMove(array, call->out);
+ UNPROTECT(1);
+ call->code = NANOARROW_OK;
+ }
+ UNPROTECT(4);
+}
+
+static int function_array_stream_get_next(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ out->release = NULL;
+ if (data == NULL || data->finished) {
+ return NANOARROW_OK;
+ }
+ if (data->errored) {
+ return data->error_code;
+ }
+ if (!nanoarrow_is_main_thread()) {
+ function_array_stream_set_error(data, EIO,
+ "Can't evaluate R callback on a non-R thread");
+ return data->error_code;
+ }
+ if (data->evaluating) {
+ function_array_stream_set_error(data, EINVAL,
+ "Re-entrant call to function array stream");
+ return data->error_code;
+ }
+
+ data->evaluating = 1;
+ struct FunctionArrayStreamCall call = {data, out, NANOARROW_OK, 0};
+ // 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;
+
+ 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;
+ }
Review Comment:
I removed the deferred-release handling.
The docs now state that fun must not release the stream while it is being evaluated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(r): add a function for lazy R callbacks [arrow-nanoarrow]
Posted by "paleolimbot (via GitHub)" <gi...@apache.org>.
paleolimbot merged PR #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(r): add a function for lazy R callbacks [arrow-nanoarrow]
Posted by "paleolimbot (via GitHub)" <gi...@apache.org>.
paleolimbot commented on code in PR #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975#discussion_r4309979638
##########
r/src/array_stream.c:
##########
@@ -18,13 +18,209 @@
#define R_NO_REMAP
#include <R.h>
#include <Rinternals.h>
+#include <errno.h>
+#include <string.h>
#include "array.h"
#include "array_stream.h"
#include "nanoarrow.h"
#include "schema.h"
#include "util.h"
+struct FunctionArrayStreamData {
+ struct ArrowSchema schema;
+ SEXP fun;
+ SEXP schema_xptr;
+ int validate;
+ struct ArrowError error;
+ int error_code;
+ int finished;
+ int errored;
+ int evaluating;
+ int release_pending;
+};
+
+struct FunctionArrayStreamCall {
+ struct FunctionArrayStreamData* data;
+ struct ArrowArray* out;
+ int code;
+ int caught;
+};
+
+static void release_function_array_stream_data(struct FunctionArrayStreamData* data) {
+ if (data->schema.release != NULL) {
+ data->schema.release(&data->schema);
+ }
+ nanoarrow_release_sexp(data->fun);
+ nanoarrow_release_sexp(data->schema_xptr);
+ ArrowFree(data);
+}
+
+static void function_array_stream_set_error(struct FunctionArrayStreamData* data,
+ int code, const char* message) {
+ if (!data->errored) {
+ ArrowErrorSet(&data->error, "%s", message);
+ data->error_code = code;
+ data->errored = 1;
+ }
+}
+
+static void release_function_array_stream(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ stream->release = NULL;
+ stream->private_data = NULL;
+ if (data->evaluating) {
+ // The callback may release its own stream. Keep its state alive until it returns.
+ data->release_pending = 1;
+ return;
+ }
+ release_function_array_stream_data(data);
+}
+
+static const char* function_array_stream_get_last_error(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return data == NULL ? "Array stream has been released"
+ : ArrowErrorMessage(&data->error);
+}
+
+static int function_array_stream_get_schema(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return ArrowSchemaDeepCopy(&data->schema, out);
+}
+
+static SEXP function_array_stream_condition_handler(SEXP condition, void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ call->caught = 1;
+ if (Rf_inherits(condition, "interrupt")) {
+ call->caught = 2;
+ }
+ SEXP message_call = PROTECT(Rf_lang2(Rf_install("conditionMessage"), condition));
+ SEXP message = Rf_eval(message_call, R_BaseEnv);
+ UNPROTECT(1);
+ return message;
+}
+
+static SEXP function_array_stream_eval_call(void* opaque) {
+ return Rf_eval((SEXP)opaque, nanoarrow_ns_pkg);
+}
+
+static void function_array_stream_eval(void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ struct FunctionArrayStreamData* data = call->data;
+ SEXP classes = PROTECT(Rf_allocVector(STRSXP, 2));
+ SET_STRING_ELT(classes, 0, Rf_mkChar("error"));
+ SET_STRING_ELT(classes, 1, Rf_mkChar("interrupt"));
+ SEXP validate = PROTECT(Rf_ScalarLogical(data->validate));
+ SEXP callback = PROTECT(Rf_lang4(Rf_install("function_array_stream_get_next_wrapper"),
+ data->fun, data->schema_xptr, validate));
+ SEXP result =
+ PROTECT(R_tryCatch(&function_array_stream_eval_call, callback, classes,
+ &function_array_stream_condition_handler, call, NULL, NULL));
+ if (data->errored) {
+ call->code = data->error_code;
+ } else if (call->caught) {
+ const char* message = Rf_translateCharUTF8(STRING_ELT(result, 0));
+ if (call->caught == 2) {
+ struct ArrowError error;
+ ArrowErrorSet(&error, "R callback interrupted: %s", message);
+ function_array_stream_set_error(data, EIO, ArrowErrorMessage(&error));
+ } else {
+ function_array_stream_set_error(data, EIO, message);
+ }
+ call->code = data->error_code;
+ } else if (result == R_NilValue) {
+ data->finished = 1;
+ call->code = NANOARROW_OK;
+ } else {
+ // Let the external pointer clean up a partial export if R jumps during allocation.
+ SEXP array_xptr = PROTECT(nanoarrow_array_owning_xptr());
+ struct ArrowArray* array = nanoarrow_output_array_from_xptr(array_xptr);
+ array_export(result, array);
+ ArrowArrayMove(array, call->out);
+ UNPROTECT(1);
+ call->code = NANOARROW_OK;
+ }
+ UNPROTECT(4);
+}
+
+static int function_array_stream_get_next(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ out->release = NULL;
+ if (data == NULL || data->finished) {
+ return NANOARROW_OK;
+ }
+ if (data->errored) {
+ return data->error_code;
+ }
+ if (!nanoarrow_is_main_thread()) {
+ function_array_stream_set_error(data, EIO,
+ "Can't evaluate R callback on a non-R thread");
+ return data->error_code;
+ }
+ if (data->evaluating) {
+ function_array_stream_set_error(data, EINVAL,
+ "Re-entrant call to function array stream");
+ return data->error_code;
+ }
+
+ data->evaluating = 1;
+ struct FunctionArrayStreamCall call = {data, out, NANOARROW_OK, 0};
+ // 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;
+
+ 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;
+ }
Review 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.
##########
r/src/array_stream.c:
##########
@@ -18,13 +18,209 @@
#define R_NO_REMAP
#include <R.h>
#include <Rinternals.h>
+#include <errno.h>
+#include <string.h>
#include "array.h"
#include "array_stream.h"
#include "nanoarrow.h"
#include "schema.h"
#include "util.h"
+struct FunctionArrayStreamData {
+ struct ArrowSchema schema;
+ SEXP fun;
Review 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()`.
##########
r/src/array_stream.c:
##########
@@ -18,13 +18,209 @@
#define R_NO_REMAP
#include <R.h>
#include <Rinternals.h>
+#include <errno.h>
+#include <string.h>
#include "array.h"
#include "array_stream.h"
#include "nanoarrow.h"
#include "schema.h"
#include "util.h"
+struct FunctionArrayStreamData {
+ struct ArrowSchema schema;
+ SEXP fun;
+ SEXP schema_xptr;
+ int validate;
+ struct ArrowError error;
+ int error_code;
+ int finished;
+ int errored;
+ int evaluating;
+ int release_pending;
+};
+
+struct FunctionArrayStreamCall {
+ struct FunctionArrayStreamData* data;
+ struct ArrowArray* out;
+ int code;
+ int caught;
+};
+
+static void release_function_array_stream_data(struct FunctionArrayStreamData* data) {
+ if (data->schema.release != NULL) {
+ data->schema.release(&data->schema);
+ }
+ nanoarrow_release_sexp(data->fun);
+ nanoarrow_release_sexp(data->schema_xptr);
+ ArrowFree(data);
+}
+
+static void function_array_stream_set_error(struct FunctionArrayStreamData* data,
+ int code, const char* message) {
+ if (!data->errored) {
+ ArrowErrorSet(&data->error, "%s", message);
+ data->error_code = code;
+ data->errored = 1;
+ }
+}
+
+static void release_function_array_stream(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ stream->release = NULL;
+ stream->private_data = NULL;
+ if (data->evaluating) {
+ // The callback may release its own stream. Keep its state alive until it returns.
+ data->release_pending = 1;
+ return;
+ }
+ release_function_array_stream_data(data);
+}
+
+static const char* function_array_stream_get_last_error(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return data == NULL ? "Array stream has been released"
+ : ArrowErrorMessage(&data->error);
+}
+
+static int function_array_stream_get_schema(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return ArrowSchemaDeepCopy(&data->schema, out);
+}
+
+static SEXP function_array_stream_condition_handler(SEXP condition, void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ call->caught = 1;
+ if (Rf_inherits(condition, "interrupt")) {
+ call->caught = 2;
+ }
+ SEXP message_call = PROTECT(Rf_lang2(Rf_install("conditionMessage"), condition));
+ SEXP message = Rf_eval(message_call, R_BaseEnv);
+ UNPROTECT(1);
+ return message;
+}
+
+static SEXP function_array_stream_eval_call(void* opaque) {
+ return Rf_eval((SEXP)opaque, nanoarrow_ns_pkg);
+}
+
+static void function_array_stream_eval(void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ struct FunctionArrayStreamData* data = call->data;
+ SEXP classes = PROTECT(Rf_allocVector(STRSXP, 2));
+ SET_STRING_ELT(classes, 0, Rf_mkChar("error"));
+ SET_STRING_ELT(classes, 1, Rf_mkChar("interrupt"));
+ SEXP validate = PROTECT(Rf_ScalarLogical(data->validate));
+ SEXP callback = PROTECT(Rf_lang4(Rf_install("function_array_stream_get_next_wrapper"),
+ data->fun, data->schema_xptr, validate));
+ SEXP result =
+ PROTECT(R_tryCatch(&function_array_stream_eval_call, callback, classes,
+ &function_array_stream_condition_handler, call, NULL, NULL));
Review 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/src/array_stream.c:
##########
@@ -18,13 +18,209 @@
#define R_NO_REMAP
#include <R.h>
#include <Rinternals.h>
+#include <errno.h>
+#include <string.h>
#include "array.h"
#include "array_stream.h"
#include "nanoarrow.h"
#include "schema.h"
#include "util.h"
+struct FunctionArrayStreamData {
+ struct ArrowSchema schema;
+ SEXP fun;
+ SEXP schema_xptr;
+ int validate;
+ struct ArrowError error;
+ int error_code;
+ int finished;
+ int errored;
+ int evaluating;
+ int release_pending;
+};
+
+struct FunctionArrayStreamCall {
+ struct FunctionArrayStreamData* data;
+ struct ArrowArray* out;
+ int code;
+ int caught;
+};
+
+static void release_function_array_stream_data(struct FunctionArrayStreamData* data) {
+ if (data->schema.release != NULL) {
+ data->schema.release(&data->schema);
+ }
+ nanoarrow_release_sexp(data->fun);
+ nanoarrow_release_sexp(data->schema_xptr);
+ ArrowFree(data);
+}
+
+static void function_array_stream_set_error(struct FunctionArrayStreamData* data,
+ int code, const char* message) {
+ if (!data->errored) {
+ ArrowErrorSet(&data->error, "%s", message);
+ data->error_code = code;
+ data->errored = 1;
+ }
+}
+
+static void release_function_array_stream(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ stream->release = NULL;
+ stream->private_data = NULL;
+ if (data->evaluating) {
+ // The callback may release its own stream. Keep its state alive until it returns.
+ data->release_pending = 1;
+ return;
+ }
+ release_function_array_stream_data(data);
+}
+
+static const char* function_array_stream_get_last_error(struct ArrowArrayStream* stream) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return data == NULL ? "Array stream has been released"
+ : ArrowErrorMessage(&data->error);
+}
+
+static int function_array_stream_get_schema(struct ArrowArrayStream* stream,
+ struct ArrowSchema* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ return ArrowSchemaDeepCopy(&data->schema, out);
+}
+
+static SEXP function_array_stream_condition_handler(SEXP condition, void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ call->caught = 1;
+ if (Rf_inherits(condition, "interrupt")) {
+ call->caught = 2;
+ }
+ SEXP message_call = PROTECT(Rf_lang2(Rf_install("conditionMessage"), condition));
+ SEXP message = Rf_eval(message_call, R_BaseEnv);
+ UNPROTECT(1);
+ return message;
+}
+
+static SEXP function_array_stream_eval_call(void* opaque) {
+ return Rf_eval((SEXP)opaque, nanoarrow_ns_pkg);
+}
+
+static void function_array_stream_eval(void* opaque) {
+ struct FunctionArrayStreamCall* call = (struct FunctionArrayStreamCall*)opaque;
+ struct FunctionArrayStreamData* data = call->data;
+ SEXP classes = PROTECT(Rf_allocVector(STRSXP, 2));
+ SET_STRING_ELT(classes, 0, Rf_mkChar("error"));
+ SET_STRING_ELT(classes, 1, Rf_mkChar("interrupt"));
+ SEXP validate = PROTECT(Rf_ScalarLogical(data->validate));
+ SEXP callback = PROTECT(Rf_lang4(Rf_install("function_array_stream_get_next_wrapper"),
+ data->fun, data->schema_xptr, validate));
+ SEXP result =
+ PROTECT(R_tryCatch(&function_array_stream_eval_call, callback, classes,
+ &function_array_stream_condition_handler, call, NULL, NULL));
+ if (data->errored) {
+ call->code = data->error_code;
+ } else if (call->caught) {
+ const char* message = Rf_translateCharUTF8(STRING_ELT(result, 0));
+ if (call->caught == 2) {
+ struct ArrowError error;
+ ArrowErrorSet(&error, "R callback interrupted: %s", message);
+ function_array_stream_set_error(data, EIO, ArrowErrorMessage(&error));
+ } else {
+ function_array_stream_set_error(data, EIO, message);
+ }
+ call->code = data->error_code;
+ } else if (result == R_NilValue) {
+ data->finished = 1;
+ call->code = NANOARROW_OK;
+ } else {
+ // Let the external pointer clean up a partial export if R jumps during allocation.
+ SEXP array_xptr = PROTECT(nanoarrow_array_owning_xptr());
+ struct ArrowArray* array = nanoarrow_output_array_from_xptr(array_xptr);
+ array_export(result, array);
+ ArrowArrayMove(array, call->out);
+ UNPROTECT(1);
+ call->code = NANOARROW_OK;
+ }
+ UNPROTECT(4);
+}
+
+static int function_array_stream_get_next(struct ArrowArrayStream* stream,
+ struct ArrowArray* out) {
+ struct FunctionArrayStreamData* data =
+ (struct FunctionArrayStreamData*)stream->private_data;
+ out->release = NULL;
+ if (data == NULL || data->finished) {
+ return NANOARROW_OK;
+ }
+ if (data->errored) {
+ return data->error_code;
+ }
+ if (!nanoarrow_is_main_thread()) {
+ function_array_stream_set_error(data, EIO,
+ "Can't evaluate R callback on a non-R thread");
+ return data->error_code;
+ }
+ if (data->evaluating) {
+ function_array_stream_set_error(data, EINVAL,
+ "Re-entrant call to function array stream");
+ return data->error_code;
+ }
+
+ data->evaluating = 1;
+ struct FunctionArrayStreamCall call = {data, out, NANOARROW_OK, 0};
+ // 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;
Review 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!
##########
r/R/array-stream.R:
##########
@@ -50,6 +50,73 @@ basic_array_stream <- function(batches, schema = NULL, validate = TRUE) {
.Call(nanoarrow_c_basic_array_stream, batches, schema, validate)
}
+#' Create an array stream from an R function
+#'
+#' The function is called once for each requested array. It should return an
+#' object coercible by [as_nanoarrow_array()], or `NULL` to indicate end of
+#' stream. The schema is required because it is available before the first
+#' array is requested. Neither construction nor schema access calls `fun`.
+#' After `NULL` or an error is returned, the callback is not called again.
+#'
+#' Evaluation is isolated in a top-level R context so non-local exits (for
+#' example, invoking an `abort` restart) cannot escape through the C stream
+#' interface. Calling handlers in the caller are therefore not applied to
+#' callback evaluation. The callback is evaluated only on R's main thread.
+#' Warnings are handled in this isolated context, unless handled within `fun`.
+#' The stream retains `fun` until it is released; explicitly release streams
+#' when `fun` or its environment refers to the stream itself.
+#'
+#' @param fun A function called with no arguments to produce each array.
+#' @param schema A schema coercible by [as_nanoarrow_schema()], used to coerce
+#' and validate each result.
+#' @param validate A single `TRUE` or `FALSE`. Use `FALSE` to skip validation
+#' within the callback. The stream's `get_next()` method can validate the
+#' returned array independently.
+#'
+#' @return An [nanoarrow_array_stream][as_nanoarrow_array_stream]
+#' @export
+#'
+#' @examples
+#' i <- 0L
+#' stream <- function_array_stream(
+#' function() {
+#' i <<- i + 1L
+#' if (i > 2L) NULL else mtcars[i, , drop = FALSE]
+#' },
+#' infer_nanoarrow_schema(mtcars)
+#' )
+#' as.data.frame(stream$get_next())
+#' as.data.frame(stream$get_next())
+#' stream$get_next()
+#' stream$release()
+#'
+function_array_stream <- function(fun, schema, validate = TRUE) {
+ stopifnot(is.function(fun))
+ schema <- as_nanoarrow_schema(schema)
+ if (!is.logical(validate) || length(validate) != 1L || is.na(validate)) {
+ stop("`validate` must be a single non-NA logical value")
+ }
+
+ .Call(nanoarrow_c_function_array_stream, fun, schema, validate)
+}
+
+function_array_stream_get_next_wrapper <- function(fun, schema, validate) {
+ result <- fun()
Review 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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(r): add a function for lazy R callbacks [arrow-nanoarrow]
Posted by "eitsupi (via GitHub)" <gi...@apache.org>.
eitsupi commented on PR #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975#issuecomment-6244143750
Feedback requested on the evaluation boundary: the implementation uses `R_ToplevelExec()` around `R_tryCatch()` for errors and interrupts. `R_tryCatch()` alone does not contain non-local exits such as `invokeRestart("abort")`; allowing those jumps across an Arrow C Stream callback can bypass a C/C++ consumer's cleanup. The outer boundary converts these exits to EIO.
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(r): add a function for lazy R callbacks [arrow-nanoarrow]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #975:
URL: https://github.com/apache/arrow-nanoarrow/pull/975#issuecomment-6244158306
## [Codecov](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `93.50649%` with `10 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 78.43%. Comparing base ([`cc67c6b`](https://app.codecov.io/gh/apache/arrow-nanoarrow/commit/cc67c6becadd11d3890f36049fd2b41db3394fa7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`f794df9`](https://app.codecov.io/gh/apache/arrow-nanoarrow/commit/f794df9f54227dfd7b378a03714847f532532cec?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [r/src/array\_stream.c](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?src=pr&el=tree&filepath=r%2Fsrc%2Farray_stream.c&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ci9zcmMvYXJyYXlfc3RyZWFtLmM=) | 93.93% | [8 Missing :warning: ](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [r/R/array-stream.R](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?src=pr&el=tree&filepath=r%2FR%2Farray-stream.R&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ci9SL2FycmF5LXN0cmVhbS5S) | 92.85% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [r/src/nanoarrow\_cpp.cc](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?src=pr&el=tree&filepath=r%2Fsrc%2Fnanoarrow_cpp.cc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ci9zcmMvbmFub2Fycm93X2NwcC5jYw==) | 87.50% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #975 +/- ##
==========================================
+ Coverage 78.29% 78.43% +0.13%
==========================================
Files 106 106
Lines 16914 17068 +154
Branches 2014 2014
==========================================
+ Hits 13243 13387 +144
- Misses 2429 2439 +10
Partials 1242 1242
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/arrow-nanoarrow/pull/975?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org