ARROW-8938: [R] Provide binding for arrow::compute::CallFunction

r/DESCRIPTION

@@ -85,6 +85,7 @@ Collate:
     'record-batch-writer.R'
     'reexports-bit64.R'
     'reexports-tidyselect.R'
+    'scalar.R'
     'schema.R'
     'struct.R'
     'util.R'

r/NAMESPACE

@@ -22,6 +22,7 @@ S3method(as.list,Table)
 S3method(as.raw,Buffer)
 S3method(as.vector,Array)
 S3method(as.vector,ChunkedArray)
+S3method(as.vector,Scalar)
 S3method(as.vector,array_expression)
 S3method(c,Dataset)
 S3method(dim,Dataset)
@@ -37,9 +38,11 @@ S3method(head,Table)
 S3method(is.na,Array)
 S3method(is.na,ChunkedArray)
 S3method(is.na,Expression)
+S3method(is.na,Scalar)
 S3method(is.na,array_expression)
 S3method(length,Array)
 S3method(length,ChunkedArray)
+S3method(length,Scalar)
 S3method(length,Schema)
 S3method(names,Dataset)
 S3method(names,RecordBatch)
@@ -120,6 +123,7 @@ export(RecordBatchFileWriter)
 export(RecordBatchStreamReader)
 export(RecordBatchStreamWriter)
 export(S3FileSystem)
+export(Scalar)
 export(Scanner)
 export(ScannerBuilder)
 export(Schema)

r/R/array.R

@@ -128,17 +128,19 @@ Array <- R6Class("Array",
         i <- Array$create(i)
       }
       if (inherits(i, "ChunkedArray")) {
+        # Invalid: Kernel does not support chunked array arguments
+        # so use the old method
         return(shared_ptr(ChunkedArray, Array__TakeChunked(self, i)))
       }
       assert_is(i, "Array")
-      Array$create(Array__Take(self, i))
+      Array$create(call_function("take", self, i))
     },
     Filter = function(i, keep_na = TRUE) {
       if (is.logical(i)) {
         i <- Array$create(i)
       }
       assert_is(i, "Array")
-      Array$create(Array__Filter(self, i, keep_na))
+      Array$create(call_function("filter", self, i, options = list(keep_na = keep_na)))
     },
     RangeEquals = function(other, start_idx, end_idx, other_start_idx = 0L) {
       assert_is(other, "Array")

r/R/chunked-array.R

@@ -75,6 +75,8 @@ ChunkedArray <- R6Class("ChunkedArray", inherit = ArrowObject,
       if (is.integer(i)) {
         i <- Array$create(i)
       }
+      # Invalid: Kernel does not support chunked array arguments
+      # so use the old method for both cases
       if (inherits(i, "ChunkedArray")) {
         return(shared_ptr(ChunkedArray, ChunkedArray__TakeChunked(self, i)))
       }

r/R/compute.R

@@ -17,6 +17,11 @@
 
 #' @include array.R
 
+call_function <- function(function_name, ..., options = list()) {
+  assert_that(is.string(function_name))
+  compute__CallFunction(function_name, list(...), options)
+}
+
 CastOptions <- R6Class("CastOptions", inherit = ArrowObject)
 
 #' Cast options

r/R/expression.R

@@ -83,8 +83,7 @@ Expression$field_ref <- function(name) {
   shared_ptr(Expression, dataset___expr__field_ref(name))
 }
 Expression$scalar <- function(x) {
-  stopifnot(vec_size(x) == 1L || is.null(x))
-  shared_ptr(Expression, dataset___expr__scalar(x))
+  shared_ptr(Expression, dataset___expr__scalar(Scalar$create(x)))
 }
 Expression$compare <- function(OP, e1, e2) {
   comp_func <- comparison_function_map[[OP]]

r/R/record-batch.R

@@ -119,6 +119,8 @@ RecordBatch <- R6Class("RecordBatch", inherit = ArrowObject,
         i <- Array$create(i)
       }
       assert_is(i, "Array")
+      # Invalid: Tried executing function with non-value type: RecordBatch
+      # so use old methods
       shared_ptr(RecordBatch, RecordBatch__Take(self, i))
     },
     Filter = function(i, keep_na = TRUE) {

r/R/scalar.R

@@ -0,0 +1,64 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+#' @include arrow-package.R
+
+#' @title Arrow scalars
+#' @usage NULL
+#' @format NULL
+#' @docType class
+#'
+#' @description A `Scalar` holds a single value of an Arrow type.
+#'
+#' @name Scalar
+#' @rdname Scalar
+#' @export
+Scalar <- R6Class("Scalar", inherit = ArrowObject,
+  # TODO: document the methods
+  public = list(
+    ToString = function() Scalar__ToString(self),
+    cast = function(target_type) {
+      Scalar$create(Scalar__CastTo(self, as_type(target_type)))
+    },
+    as_vector = function() Scalar__as_vector(self)
+  ),
+  active = list(
+    is_valid = function() Scalar__is_valid(self),
+    type = function() DataType$create(Scalar__type(self))
+  )
+)
+Scalar$create <- function(x, type = NULL) {
+  if (!inherits(x, "externalptr")) {
+    if (is.null(x)) {
+      x <- vctrs::unspecified(1)
+    } else if (length(x) != 1 && !is.data.frame(x)) {
+      # Wrap in a list type
+      x <- list(x)
+    }
+    x <- Array__GetScalar(Array$create(x, type = type), 0)
+  }
+  shared_ptr(Scalar, x)
+}
+
+#' @export
+length.Scalar <- function(x) 1L
+
+#' @export
+is.na.Scalar <- function(x) !x$is_valid
+
+#' @export
+as.vector.Scalar <- function(x, mode) x$as_vector()