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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
