| 76 | std::shared_ptr<arrow::Schema> schema() const { return schema_; } |
| 77 | |
| 78 | arrow::Status ReadNext(std::shared_ptr<arrow::RecordBatch>* batch_out) { |
| 79 | auto batch = SafeCallIntoR<std::shared_ptr<arrow::RecordBatch>>([&]() { |
| 80 | cpp11::sexp result_sexp = cpp11::function(fun_)(); |
| 81 | if (result_sexp == R_NilValue) { |
| 82 | return std::shared_ptr<arrow::RecordBatch>(nullptr); |
| 83 | } else if (!Rf_inherits(result_sexp, "RecordBatch")) { |
| 84 | cpp11::stop("Expected fun() to return an arrow::RecordBatch"); |
| 85 | } |
| 86 | |
| 87 | return cpp11::as_cpp<std::shared_ptr<arrow::RecordBatch>>(result_sexp); |
| 88 | }); |
| 89 | |
| 90 | RETURN_NOT_OK(batch); |
| 91 | |
| 92 | if (batch.ValueUnsafe().get() != nullptr && |
| 93 | !batch.ValueUnsafe()->schema()->Equals(schema_)) { |
| 94 | return arrow::Status::Invalid("Expected fun() to return batch with schema '", |
| 95 | schema_->ToString(), "' but got batch with schema '", |
| 96 | batch.ValueUnsafe()->schema()->ToString(), "'"); |
| 97 | } |
| 98 | |
| 99 | *batch_out = batch.ValueUnsafe(); |
| 100 | return arrow::Status::OK(); |
| 101 | } |
| 102 | |
| 103 | private: |
| 104 | cpp11::sexp fun_; |