fix(r): Collect array streams in C (not R) before conversion

r/R/convert-array-stream.R

@@ -72,31 +72,21 @@ convert_array_stream <- function(array_stream, to = NULL, size = NULL, n = Inf)
   } else {
     # Otherwise, we need to collect all batches and calculate the total length
     # before calling nanoarrow_c_convert_array_stream().
-    batches <- collect_array_stream(
-      array_stream,
-      n,
-      schema = schema,
-      validate = FALSE
-    )
+    batch_info <- .Call(nanoarrow_c_collect_array_stream, array_stream, n)
 
     # If there is exactly one batch, use convert_array(). Converting a single
     # array currently takes a more efficient code path for types that can be
     # converted as ALTREP (e.g., strings).
-    if (length(batches) == 1L) {
-      return(.Call(nanoarrow_c_convert_array, batches[[1]], to))
+    if (batch_info$n == 1L) {
+      array <- batch_info$stream$get_next(schema)
+      return(.Call(nanoarrow_c_convert_array, array, to))
     }
 
-    # Otherwise, compute the final size, create another array stream,
-    # and call convert_array_stream() with a known size. Using .Call()
-    # directly because we have already type checked the inputs.
-    size <- .Call(nanoarrow_c_array_list_total_length, batches)
-    basic_stream <- .Call(nanoarrow_c_basic_array_stream, batches, schema, FALSE)
-
     .Call(
       nanoarrow_c_convert_array_stream,
-      basic_stream,
+      batch_info$stream,
       to,
-      as.double(size),
+      as.double(batch_info$size),
       Inf
     )
   }

r/bootstrap.R

@@ -60,7 +60,3 @@ stopifnot(file.exists("../CMakeLists.txt") && run_bundler())
 f <- "src/flatcc/portable/pdiagnostic.h"
 lines <- readLines(f)
 writeLines(gsub("^#pragma", "/**/#pragma", lines), f)
-
-# Remove unused files
-unused_files <- list.files("src", "\\.hpp$", full.names = TRUE)
-unlink(unused_files)

r/src/init.c

@@ -61,6 +61,7 @@ extern SEXP nanoarrow_c_ipc_array_reader_buffer(SEXP buffer_xptr);
 extern SEXP nanoarrow_c_ipc_array_reader_connection(SEXP con);
 extern SEXP nanoarrow_c_ipc_writer_connection(SEXP con);
 extern SEXP nanoarrow_c_ipc_writer_write_stream(SEXP writer_xptr, SEXP array_stream_xptr);
+extern SEXP nanoarrow_c_collect_array_stream(SEXP array_stream_xptr, SEXP n_sexp);
 extern SEXP nanoarrow_c_allocate_schema(void);
 extern SEXP nanoarrow_c_allocate_array(void);
 extern SEXP nanoarrow_c_allocate_array_stream(void);
@@ -144,6 +145,7 @@ static const R_CallMethodDef CallEntries[] = {
     {"nanoarrow_c_ipc_writer_connection", (DL_FUNC)&nanoarrow_c_ipc_writer_connection, 1},
     {"nanoarrow_c_ipc_writer_write_stream", (DL_FUNC)&nanoarrow_c_ipc_writer_write_stream,
      2},
+    {"nanoarrow_c_collect_array_stream", (DL_FUNC)&nanoarrow_c_collect_array_stream, 2},
     {"nanoarrow_c_allocate_schema", (DL_FUNC)&nanoarrow_c_allocate_schema, 0},
     {"nanoarrow_c_allocate_array", (DL_FUNC)&nanoarrow_c_allocate_array, 0},
     {"nanoarrow_c_allocate_array_stream", (DL_FUNC)&nanoarrow_c_allocate_array_stream, 0},