array_to_vector.cpp

// Licensed to the Apache Software Foundation (ASF) under one
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// or more contributor license agreements. See the NOTICE 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_types.h"

#include<arrow/array.h>
#include<arrow/builder.h>
#include<arrow/datum.h>
#include<arrow/table.h>
#include<arrow/util/bitmap_reader.h>
#include<arrow/util/bitmap_writer.h>
#include<arrow/util/int_util.h>

#include<type_traits>

#include"./extension.h"
#include"./r_task_group.h"

namespacearrow {

using internal::checked_cast;
using internal::IntegersCanFit;

namespacer {

classConverter {
public:
explicitConverter(const std::shared_ptr<ChunkedArray>& chunked_array)
 : chunked_array_(std::move(chunked_array)) {}

virtual~Converter() {}

// Allocate a vector of the right R type for this converter
virtualSEXPAllocate(R_xlen_t n) const = 0;

// data[ start:(start + n) ] = NA
virtual Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const = 0;

// ingest the values from the array into data[ start : (start + n)]
//
// chunk_index indicates which of the chunk is being ingested into data. This is
// ignored by most implementations and currently only used with Dictionary
// arrays.
virtual Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n,
size_t chunk_index) const = 0;

// can this run in parallel ?
virtualboolParallel() const { returntrue; }

// converter is passed as self to outlive the scope of Converter::Convert()
SEXPScheduleConvertTasks(RTasks& tasks, std::shared_ptr<Converter> self) {
// try altrep first
SEXP alt = altrep::MakeAltrepVector(chunked_array_);
if (!Rf_isNull(alt)) {
return alt;
 }

// otherwise use the Converter api:

// allocating the R vector upfront
SEXP out = PROTECT(Allocate(chunked_array_->length()));

// for each array, fill the relevant slice of `out`, potentially in parallel
 R_xlen_t k = 0, i = 0;
for (constauto& array : chunked_array_->chunks()) {
auto n_chunk = array->length();

 tasks.Append(Parallel(), [=] {
if (array->null_count() == n_chunk) {
return self->Ingest_all_nulls(out, k, n_chunk);
 } else {
return self->Ingest_some_nulls(out, array, k, n_chunk, i);
 }
 });

 k += n_chunk;
 i++;
 }

UNPROTECT(1);
return out;
 }

// Converter factory
static std::shared_ptr<Converter> Make(
const std::shared_ptr<ChunkedArray>& chunked_array);

staticSEXPLazyConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
 RTasks& tasks) {
auto converter = Make(chunked_array);
return converter->ScheduleConvertTasks(tasks, converter);
 }

staticSEXPConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
bool use_threads) {
 RTasks tasks(use_threads);
SEXP out = PROTECT(Converter::LazyConvert(chunked_array, tasks));
StopIfNotOk(tasks.Finish());

UNPROTECT(1);
return out;
 }

staticSEXPConvert(const std::shared_ptr<Array>& array) {
returnConvert(std::make_shared<ChunkedArray>(array), false);
 }

SEXPMaybeAltrep() { returnaltrep::MakeAltrepVector(chunked_array_); }

protected:
 std::shared_ptr<ChunkedArray> chunked_array_;
};

template <typename SetNonNull, typename SetNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null, SetNull&& set_null) {
if (array->null_count()) {
 internal::BitmapReader bitmap_reader(array->null_bitmap()->data(), array->offset(),
 n);

for (R_xlen_t i = 0; i < n; i++, bitmap_reader.Next()) {
if (bitmap_reader.IsSet()) {
RETURN_NOT_OK(set_non_null(i));
 } else {
RETURN_NOT_OK(set_null(i));
 }
 }

 } else {
for (R_xlen_t i = 0; i < n; i++) {
RETURN_NOT_OK(set_non_null(i));
 }
 }

returnStatus::OK();
}

template <typename SetNonNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null) {
auto nothing = [](R_xlen_t i) { returnStatus::OK(); };
returnIngestSome(array, n, std::forward<SetNonNull>(set_non_null), nothing);
}

std::shared_ptr<Array> CreateEmptyArray(const std::shared_ptr<DataType>& array_type) {
 std::unique_ptr<arrow::ArrayBuilder> builder;
StopIfNotOk(arrow::MakeBuilder(gc_memory_pool(), array_type, &builder));

 std::shared_ptr<arrow::Array> array;
StopIfNotOk(builder->Finish(&array));
return array;
}

template <typename Type>
classConverter_Int : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Int(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(INTSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = INTEGER(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename Type>
classConverter_Double : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Double(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<value_type>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

classConverter_Date32 : publicConverter {
public:
explicitConverter_Date32(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP data = PROTECT(Rf_allocVector(REALSXP, n));
Rf_classgets(data, Rf_mkString("Date"));
UNPROTECT(1);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<int>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename StringArrayType>
structConverter_String : publicConverter {
public:
explicitConverter_String(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(STRSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
for (R_xlen_t i = 0; i < n; i++) {
SET_STRING_ELT(data, i + start, NA_STRING);
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_offset = array->data()->GetValues<int32_t>(1);
if (!p_offset) {
returnStatus::Invalid("Invalid offset buffer");
 }
auto p_strings = array->data()->GetValues<char>(2, *p_offset);
if (!p_strings) {
// There is an offset buffer, but the data buffer is null
// There is at least one value in the array and not all the values are null
// That means all values are either empty strings or nulls so there is nothing to do

if (array->null_count()) {
 arrow::internal::BitmapReader null_reader(array->null_bitmap_data(),
 array->offset(), n);
for (int i = 0; i < n; i++, null_reader.Next()) {
if (null_reader.IsNotSet()) {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 }
returnStatus::OK();
 }

 StringArrayType* string_array = static_cast<StringArrayType*>(array.get());

constbool all_valid = array->null_count() == 0;
constbool strip_out_nuls = GetBoolOption("arrow.skip_nul", false);

bool nul_was_stripped = false;

if (all_valid) {
// no need to watch for missing strings
cpp11::unwind_protect([&] {
if (strip_out_nuls) {
for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 }
return;
 }

for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 }
 });
 } else {
cpp11::unwind_protect([&] {
 arrow::internal::BitmapReader validity_reader(array->null_bitmap_data(),
 array->offset(), n);

if (strip_out_nuls) {
for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
return;
 }

for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 });
 }

if (nul_was_stripped) {
 cpp11::safe[Rf_warning]("Stripping '\\0' (nul) from character vector");
 }

returnStatus::OK();
 }

boolParallel() const { returnfalse; }

private:
staticSEXPr_string_from_view(std::string_view view) {
returnRf_mkCharLenCE(view.data(), static_cast<int>(view.size()), CE_UTF8);
 }

staticSEXPr_string_from_view_strip_nul(std::string_view view,
bool* nul_was_stripped) {
constchar* old_string = view.data();

 std::string stripped_string;
size_t stripped_len = 0, nul_count = 0;

for (size_t i = 0; i < view.size(); i++) {
if (old_string[i] == '\0') {
 ++nul_count;

if (nul_count == 1) {
// first nul spotted: allocate stripped string storage
 stripped_string = std::string(view);
 stripped_len = i;
 }

// don't copy old_string[i] (which is \0) into stripped_string
continue;
 }

if (nul_count > 0) {
 stripped_string[stripped_len++] = old_string[i];
 }
 }

if (nul_count > 0) {
 *nul_was_stripped = true;
 stripped_string.resize(stripped_len);
returnr_string_from_view(stripped_string);
 }

returnr_string_from_view(view);
 }
};

classConverter_Boolean : publicConverter {
public:
explicitConverter_Boolean(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(LGLSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(LOGICAL(data) + start, n, NA_LOGICAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = LOGICAL(data) + start;
auto p_bools = array->data()->GetValues<uint8_t>(1, 0);
if (!p_bools) {
returnStatus::Invalid("Invalid data buffer");
 }

 arrow::internal::BitmapReader data_reader(p_bools, array->offset(), n);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = data_reader.IsSet();
 data_reader.Next();
returnStatus::OK();
 };

auto null_one = [&](R_xlen_t i) {
 data_reader.Next();
 p_data[i] = NA_LOGICAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename ArrayType>
classConverter_Binary : publicConverter {
public:
using offset_type = typename ArrayType::offset_type;
explicitConverter_Binary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
if (std::is_same<ArrayType, BinaryArray>::value) {
Rf_classgets(res, data::classes_arrow_binary);
 } else {
Rf_classgets(res, data::classes_arrow_large_binary);
 }
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const ArrayType* binary_array = checked_cast<const ArrayType*>(array.get());

auto ingest_one = [&](R_xlen_t i) {
 offset_type ni;
auto value = binary_array->GetValue(i, &ni);
if (ni > R_XLEN_T_MAX) {
returnStatus::RError("Array too big to be represented as a raw vector");
 }
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, ni));
std::copy(value, value + ni, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }
};

classConverter_FixedSizeBinary : publicConverter {
public:
explicitConverter_FixedSizeBinary(const std::shared_ptr<ChunkedArray>& chunked_array,
int byte_width)
 : Converter(chunked_array), byte_width_(byte_width) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
Rf_classgets(res, data::classes_arrow_fixed_size_binary);
Rf_setAttrib(res, symbols::byte_width, Rf_ScalarInteger(byte_width_));
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const FixedSizeBinaryArray* binary_array =
 checked_cast<const FixedSizeBinaryArray*>(array.get());

int byte_width = binary_array->byte_width();
auto ingest_one = [&, byte_width](R_xlen_t i) {
auto value = binary_array->GetValue(i);
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, byte_width));
std::copy(value, value + byte_width, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }

private:
int byte_width_;
};

boolDictionaryChunkArrayNeedUnification(
const std::shared_ptr<ChunkedArray>& chunked_array) {
int n = chunked_array->num_chunks();
if (n < 2) {
returnfalse;
 }
constauto& arr_first =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
for (int i = 1; i < n; i++) {
constauto& arr =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(i));
if (!(arr_first.dictionary()->Equals(arr.dictionary()))) {
returntrue;
 }
 }
returnfalse;
}

classConverter_Dictionary : publicConverter {
private:
bool need_unification_;
 std::unique_ptr<arrow::DictionaryUnifier> unifier_;
 std::vector<std::shared_ptr<Buffer>> arrays_transpose_;
 std::shared_ptr<DataType> out_type_;
 std::shared_ptr<Array> dictionary_;

public:
explicitConverter_Dictionary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array),
 need_unification_(DictionaryChunkArrayNeedUnification(chunked_array)) {
if (need_unification_) {
constauto& arr_type = checked_cast<const DictionaryType&>(*chunked_array->type());
 unifier_ = ValueOrStop(DictionaryUnifier::Make(arr_type.value_type()));

int n_arrays = chunked_array->num_chunks();
 arrays_transpose_.resize(n_arrays);

for (int i = 0; i < n_arrays; i++) {
constauto& dict_i =
 *checked_cast<const DictionaryArray&>(*chunked_array->chunk(i)).dictionary();
StopIfNotOk(unifier_->Unify(dict_i, &arrays_transpose_[i]));
 }

StopIfNotOk(unifier_->GetResult(&out_type_, &dictionary_));
 } else {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());

constauto& indices_type = *dict_type.index_type();
switch (indices_type.id()) {
case Type::UINT8:
case Type::INT8:
case Type::UINT16:
case Type::INT16:
case Type::INT32:
case Type::UINT32:
case Type::INT64:
case Type::UINT64:
break;
default:
cpp11::stop("Cannot convert Dictionary Array of type `%s` to R",
 dict_type.ToString().c_str());
 }

if (chunked_array->num_chunks() > 0) {
// DictionaryChunkArrayNeedUnification() returned false so we can safely assume
// the dictionary of the first chunk applies everywhere
constauto& dict_array =
 checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
 dictionary_ = dict_array.dictionary();
 } else {
 dictionary_ = CreateEmptyArray(dict_type.value_type());
 }
 }

// R factors store their codes in 32-bit integers, so dictionary arrays with
// more levels than that cannot be represented safely.
if (dictionary_->length() > std::numeric_limits<int>::max()) {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());
cpp11::stop(
"Cannot convert Dictionary Array of type `%s` to R: dictionary has "
"more levels than an R factor can represent",
 dict_type.ToString().c_str());
 }
 }

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::integers data(n);
 data.attr("levels") = GetLevels();
if (GetOrdered()) {
Rf_classgets(data, arrow::r::data::classes_ordered);
 } else {
Rf_classgets(data, arrow::r::data::classes_factor);
 }
return data;
 }

virtualboolParallel() const { returnfalse; }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const DictionaryArray& dict_array =
 checked_cast<const DictionaryArray&>(*array.get());
auto indices = dict_array.indices();
switch (indices->type_id()) {
case Type::UINT8:
return Ingest_some_nulls_Impl<arrow::UInt8Type>(data, array, start, n,
 chunk_index);
case Type::INT8:
return Ingest_some_nulls_Impl<arrow::Int8Type>(data, array, start, n,
 chunk_index);
case Type::UINT16:
return Ingest_some_nulls_Impl<arrow::UInt16Type>(data, array, start, n,
 chunk_index);
case Type::INT16:
return Ingest_some_nulls_Impl<arrow::Int16Type>(data, array, start, n,
 chunk_index);
case Type::INT32:
return Ingest_some_nulls_Impl<arrow::Int32Type>(data, array, start, n,
 chunk_index);
case Type::UINT32:
return Ingest_some_nulls_Impl<arrow::UInt32Type>(data, array, start, n,
 chunk_index);
case Type::INT64:
return Ingest_some_nulls_Impl<arrow::Int64Type>(data, array, start, n,
 chunk_index);
case Type::UINT64:
return Ingest_some_nulls_Impl<arrow::UInt64Type>(data, array, start, n,
 chunk_index);
default:
break;
 }
returnStatus::OK();
 }

private:
template <typename Type>
 Status Ingest_some_nulls_Impl(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
using index_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
auto indices = checked_cast<const DictionaryArray&>(*array).indices();
auto raw_indices = indices->data()->GetValues<index_type>(1);

auto p_data = INTEGER(data) + start;
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

// convert the 0-based indices from the arrow Array
// to 1-based indices used in R factors
if (need_unification_) {
// transpose the indices before converting
auto transposed =
reinterpret_cast<constint32_t*>(arrays_transpose_[chunk_index]->data());

auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = transposed[raw_indices[i]] + 1;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 } else {
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(raw_indices[i]) + 1;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
 }

boolGetOrdered() const {
return checked_cast<const DictionaryType&>(*chunked_array_->type()).ordered();
 }

SEXPGetLevels() const {
// R factor levels must be type "character" so coerce `dict` to STRSXP
// TODO (npr): this coercion should be optional, "dictionariesAsFactors" ;)
// Alternative: preserve the logical type of the dictionary values
// (e.g. if dict is timestamp, return a POSIXt R vector, not factor)
if (dictionary_->type_id() != Type::STRING &&
 dictionary_->type_id() != Type::LARGE_STRING) {
 cpp11::safe[Rf_warning]("Coercing dictionary values to R character factor levels");
 }

SEXP vec = PROTECT(Converter::Convert(dictionary_));
SEXP strings_vec = PROTECT(Rf_coerceVector(vec, STRSXP));
UNPROTECT(2);
return strings_vec;
 }
};

classConverter_Struct : publicConverter {
public:
explicitConverter_Struct(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array), converters() {
constauto& struct_type =
 checked_cast<const arrow::StructType&>(*chunked_array->type());

int nf = struct_type.num_fields();

 std::shared_ptr<arrow::Table> array_as_table =
ValueOrStop(arrow::Table::FromChunkedStructArray(chunked_array));
for (int i = 0; i < nf; i++) {
 converters.push_back(Converter::Make(array_as_table->column(i)));
 }
 }

SEXPAllocate(R_xlen_t n) const {
// allocate a data frame column to host each array
// If possible, a column is dealt with directly with altrep
auto type =
 checked_cast<const arrow::StructType*>(this->chunked_array_->type().get());
auto out =
arrow::r::to_r_list(converters, [n](const std::shared_ptr<Converter>& converter) {
SEXP out = converter->MaybeAltrep();
if (Rf_isNull(out)) {
 out = converter->Allocate(n);
 }
return out;
 });
auto colnames = arrow::r::to_r_strings(
 type->fields(),
 [](const std::shared_ptr<Field>& field) { return field->name(); });
 out.attr(symbols::row_names) = arrow::r::short_row_names(static_cast<int>(n));
 out.attr(R_NamesSymbol) = colnames;
 out.attr(R_ClassSymbol) = arrow::r::data::classes_tbl_df;

return out;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
int nf = static_cast<int>(converters.size());
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_all_nulls(data_i, start, n));
 }
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto struct_array = checked_cast<const arrow::StructArray*>(array.get());
int nf = static_cast<int>(converters.size());
// Flatten() deals with merging of nulls
auto arrays = ValueOrStop(struct_array->Flatten(gc_memory_pool()));
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_some_nulls(VECTOR_ELT(data, i), arrays[i],
 start, n, chunk_index));
 }
 }

returnStatus::OK();
 }

virtualboolParallel() const {
// this can only run in parallel if all the
// inner converters can
for (constauto& converter : converters) {
if (!converter->Parallel()) returnfalse;
 }
returntrue;
 }

private:
 std::vector<std::shared_ptr<Converter>> converters;
};

doublems_to_seconds(int64_t ms) { returnstatic_cast<double>(ms) / 1000; }

classConverter_Date64 : publicConverter {
public:
explicitConverter_Date64(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<int64_t>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i] / 1000);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename value_type, typename unit_type = TimeType>
classConverter_Time : publicConverter {
public:
explicitConverter_Time(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = cpp11::writable::strings({"hms", "difftime"});

// hms difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// hms difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type, typename unit_type = DurationType>
classConverter_Duration : publicConverter {
public:
explicitConverter_Duration(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = "difftime";

// difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type>
classConverter_Timestamp : publicConverter_Time<value_type, TimestampType> {
public:
explicitConverter_Timestamp(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter_Time<value_type, TimestampType>(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
auto array_type =
 checked_cast<const TimestampType*>(this->chunked_array_->type().get());
 std::string tzone = array_type->timezone();
if (tzone.size() > 0) {
 data.attr("tzone") = tzone;
 }
return data;
 }
};

template <typename Type>
classConverter_Decimal : publicConverter {
public:
explicitConverter_Decimal(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
// Licensed to the Apache Software Foundation (ASF) under one
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// or more contributor license agreements. See the NOTICE 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_types.h"

#include<arrow/array.h>
#include<arrow/builder.h>
#include<arrow/datum.h>
#include<arrow/table.h>
#include<arrow/util/bitmap_reader.h>
#include<arrow/util/bitmap_writer.h>
#include<arrow/util/int_util.h>

#include<type_traits>

#include"./extension.h"
#include"./r_task_group.h"

namespacearrow {

using internal::checked_cast;
using internal::IntegersCanFit;

namespacer {

classConverter {
public:
explicitConverter(const std::shared_ptr<ChunkedArray>& chunked_array)
 : chunked_array_(std::move(chunked_array)) {}

virtual~Converter() {}

// Allocate a vector of the right R type for this converter
virtualSEXPAllocate(R_xlen_t n) const = 0;

// data[ start:(start + n) ] = NA
virtual Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const = 0;

// ingest the values from the array into data[ start : (start + n)]
//
// chunk_index indicates which of the chunk is being ingested into data. This is
// ignored by most implementations and currently only used with Dictionary
// arrays.
virtual Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n,
size_t chunk_index) const = 0;

// can this run in parallel ?
virtualboolParallel() const { returntrue; }

// converter is passed as self to outlive the scope of Converter::Convert()
SEXPScheduleConvertTasks(RTasks& tasks, std::shared_ptr<Converter> self) {
// try altrep first
SEXP alt = altrep::MakeAltrepVector(chunked_array_);
if (!Rf_isNull(alt)) {
return alt;
 }

// otherwise use the Converter api:

// allocating the R vector upfront
SEXP out = PROTECT(Allocate(chunked_array_->length()));

// for each array, fill the relevant slice of `out`, potentially in parallel
 R_xlen_t k = 0, i = 0;
for (constauto& array : chunked_array_->chunks()) {
auto n_chunk = array->length();

 tasks.Append(Parallel(), [=] {
if (array->null_count() == n_chunk) {
return self->Ingest_all_nulls(out, k, n_chunk);
 } else {
return self->Ingest_some_nulls(out, array, k, n_chunk, i);
 }
 });

 k += n_chunk;
 i++;
 }

UNPROTECT(1);
return out;
 }

// Converter factory
static std::shared_ptr<Converter> Make(
const std::shared_ptr<ChunkedArray>& chunked_array);

staticSEXPLazyConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
 RTasks& tasks) {
auto converter = Make(chunked_array);
return converter->ScheduleConvertTasks(tasks, converter);
 }

staticSEXPConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
bool use_threads) {
 RTasks tasks(use_threads);
SEXP out = PROTECT(Converter::LazyConvert(chunked_array, tasks));
StopIfNotOk(tasks.Finish());

UNPROTECT(1);
return out;
 }

staticSEXPConvert(const std::shared_ptr<Array>& array) {
returnConvert(std::make_shared<ChunkedArray>(array), false);
 }

SEXPMaybeAltrep() { returnaltrep::MakeAltrepVector(chunked_array_); }

protected:
 std::shared_ptr<ChunkedArray> chunked_array_;
};

template <typename SetNonNull, typename SetNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null, SetNull&& set_null) {
if (array->null_count()) {
 internal::BitmapReader bitmap_reader(array->null_bitmap()->data(), array->offset(),
 n);

for (R_xlen_t i = 0; i < n; i++, bitmap_reader.Next()) {
if (bitmap_reader.IsSet()) {
RETURN_NOT_OK(set_non_null(i));
 } else {
RETURN_NOT_OK(set_null(i));
 }
 }

 } else {
for (R_xlen_t i = 0; i < n; i++) {
RETURN_NOT_OK(set_non_null(i));
 }
 }

returnStatus::OK();
}

template <typename SetNonNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null) {
auto nothing = [](R_xlen_t i) { returnStatus::OK(); };
returnIngestSome(array, n, std::forward<SetNonNull>(set_non_null), nothing);
}

std::shared_ptr<Array> CreateEmptyArray(const std::shared_ptr<DataType>& array_type) {
 std::unique_ptr<arrow::ArrayBuilder> builder;
StopIfNotOk(arrow::MakeBuilder(gc_memory_pool(), array_type, &builder));

 std::shared_ptr<arrow::Array> array;
StopIfNotOk(builder->Finish(&array));
return array;
}

template <typename Type>
classConverter_Int : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Int(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(INTSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = INTEGER(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename Type>
classConverter_Double : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Double(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<value_type>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

classConverter_Date32 : publicConverter {
public:
explicitConverter_Date32(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP data = PROTECT(Rf_allocVector(REALSXP, n));
Rf_classgets(data, Rf_mkString("Date"));
UNPROTECT(1);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<int>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename StringArrayType>
structConverter_String : publicConverter {
public:
explicitConverter_String(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(STRSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
for (R_xlen_t i = 0; i < n; i++) {
SET_STRING_ELT(data, i + start, NA_STRING);
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_offset = array->data()->GetValues<int32_t>(1);
if (!p_offset) {
returnStatus::Invalid("Invalid offset buffer");
 }
auto p_strings = array->data()->GetValues<char>(2, *p_offset);
if (!p_strings) {
// There is an offset buffer, but the data buffer is null
// There is at least one value in the array and not all the values are null
// That means all values are either empty strings or nulls so there is nothing to do

if (array->null_count()) {
 arrow::internal::BitmapReader null_reader(array->null_bitmap_data(),
 array->offset(), n);
for (int i = 0; i < n; i++, null_reader.Next()) {
if (null_reader.IsNotSet()) {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 }
returnStatus::OK();
 }

 StringArrayType* string_array = static_cast<StringArrayType*>(array.get());

constbool all_valid = array->null_count() == 0;
constbool strip_out_nuls = GetBoolOption("arrow.skip_nul", false);

bool nul_was_stripped = false;

if (all_valid) {
// no need to watch for missing strings
cpp11::unwind_protect([&] {
if (strip_out_nuls) {
for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 }
return;
 }

for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 }
 });
 } else {
cpp11::unwind_protect([&] {
 arrow::internal::BitmapReader validity_reader(array->null_bitmap_data(),
 array->offset(), n);

if (strip_out_nuls) {
for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
return;
 }

for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 });
 }

if (nul_was_stripped) {
 cpp11::safe[Rf_warning]("Stripping '\\0' (nul) from character vector");
 }

returnStatus::OK();
 }

boolParallel() const { returnfalse; }

private:
staticSEXPr_string_from_view(std::string_view view) {
returnRf_mkCharLenCE(view.data(), static_cast<int>(view.size()), CE_UTF8);
 }

staticSEXPr_string_from_view_strip_nul(std::string_view view,
bool* nul_was_stripped) {
constchar* old_string = view.data();

 std::string stripped_string;
size_t stripped_len = 0, nul_count = 0;

for (size_t i = 0; i < view.size(); i++) {
if (old_string[i] == '\0') {
 ++nul_count;

if (nul_count == 1) {
// first nul spotted: allocate stripped string storage
 stripped_string = std::string(view);
 stripped_len = i;
 }

// don't copy old_string[i] (which is \0) into stripped_string
continue;
 }

if (nul_count > 0) {
 stripped_string[stripped_len++] = old_string[i];
 }
 }

if (nul_count > 0) {
 *nul_was_stripped = true;
 stripped_string.resize(stripped_len);
returnr_string_from_view(stripped_string);
 }

returnr_string_from_view(view);
 }
};

classConverter_Boolean : publicConverter {
public:
explicitConverter_Boolean(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(LGLSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(LOGICAL(data) + start, n, NA_LOGICAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = LOGICAL(data) + start;
auto p_bools = array->data()->GetValues<uint8_t>(1, 0);
if (!p_bools) {
returnStatus::Invalid("Invalid data buffer");
 }

 arrow::internal::BitmapReader data_reader(p_bools, array->offset(), n);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = data_reader.IsSet();
 data_reader.Next();
returnStatus::OK();
 };

auto null_one = [&](R_xlen_t i) {
 data_reader.Next();
 p_data[i] = NA_LOGICAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename ArrayType>
classConverter_Binary : publicConverter {
public:
using offset_type = typename ArrayType::offset_type;
explicitConverter_Binary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
if (std::is_same<ArrayType, BinaryArray>::value) {
Rf_classgets(res, data::classes_arrow_binary);
 } else {
Rf_classgets(res, data::classes_arrow_large_binary);
 }
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const ArrayType* binary_array = checked_cast<const ArrayType*>(array.get());

auto ingest_one = [&](R_xlen_t i) {
 offset_type ni;
auto value = binary_array->GetValue(i, &ni);
if (ni > R_XLEN_T_MAX) {
returnStatus::RError("Array too big to be represented as a raw vector");
 }
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, ni));
std::copy(value, value + ni, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }
};

classConverter_FixedSizeBinary : publicConverter {
public:
explicitConverter_FixedSizeBinary(const std::shared_ptr<ChunkedArray>& chunked_array,
int byte_width)
 : Converter(chunked_array), byte_width_(byte_width) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
Rf_classgets(res, data::classes_arrow_fixed_size_binary);
Rf_setAttrib(res, symbols::byte_width, Rf_ScalarInteger(byte_width_));
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const FixedSizeBinaryArray* binary_array =
 checked_cast<const FixedSizeBinaryArray*>(array.get());

int byte_width = binary_array->byte_width();
auto ingest_one = [&, byte_width](R_xlen_t i) {
auto value = binary_array->GetValue(i);
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, byte_width));
std::copy(value, value + byte_width, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }

private:
int byte_width_;
};

boolDictionaryChunkArrayNeedUnification(
const std::shared_ptr<ChunkedArray>& chunked_array) {
int n = chunked_array->num_chunks();
if (n < 2) {
returnfalse;
 }
constauto& arr_first =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
for (int i = 1; i < n; i++) {
constauto& arr =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(i));
if (!(arr_first.dictionary()->Equals(arr.dictionary()))) {
returntrue;
 }
 }
returnfalse;
}

classConverter_Dictionary : publicConverter {
private:
bool need_unification_;
 std::unique_ptr<arrow::DictionaryUnifier> unifier_;
 std::vector<std::shared_ptr<Buffer>> arrays_transpose_;
 std::shared_ptr<DataType> out_type_;
 std::shared_ptr<Array> dictionary_;

public:
explicitConverter_Dictionary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array),
 need_unification_(DictionaryChunkArrayNeedUnification(chunked_array)) {
if (need_unification_) {
constauto& arr_type = checked_cast<const DictionaryType&>(*chunked_array->type());
 unifier_ = ValueOrStop(DictionaryUnifier::Make(arr_type.value_type()));

int n_arrays = chunked_array->num_chunks();
 arrays_transpose_.resize(n_arrays);

for (int i = 0; i < n_arrays; i++) {
constauto& dict_i =
 *checked_cast<const DictionaryArray&>(*chunked_array->chunk(i)).dictionary();
StopIfNotOk(unifier_->Unify(dict_i, &arrays_transpose_[i]));
 }

StopIfNotOk(unifier_->GetResult(&out_type_, &dictionary_));
 } else {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());

constauto& indices_type = *dict_type.index_type();
switch (indices_type.id()) {
case Type::UINT8:
case Type::INT8:
case Type::UINT16:
case Type::INT16:
case Type::INT32:
case Type::UINT32:
case Type::INT64:
case Type::UINT64:
break;
default:
cpp11::stop("Cannot convert Dictionary Array of type `%s` to R",
 dict_type.ToString().c_str());
 }

if (chunked_array->num_chunks() > 0) {
// DictionaryChunkArrayNeedUnification() returned false so we can safely assume
// the dictionary of the first chunk applies everywhere
constauto& dict_array =
 checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
 dictionary_ = dict_array.dictionary();
 } else {
 dictionary_ = CreateEmptyArray(dict_type.value_type());
 }
 }

// R factors store their codes in 32-bit integers, so dictionary arrays with
// more levels than that cannot be represented safely.
if (dictionary_->length() > std::numeric_limits<int>::max()) {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());
cpp11::stop(
"Cannot convert Dictionary Array of type `%s` to R: dictionary has "
"more levels than an R factor can represent",
 dict_type.ToString().c_str());
 }
 }

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::integers data(n);
 data.attr("levels") = GetLevels();
if (GetOrdered()) {
Rf_classgets(data, arrow::r::data::classes_ordered);
 } else {
Rf_classgets(data, arrow::r::data::classes_factor);
 }
return data;
 }

virtualboolParallel() const { returnfalse; }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const DictionaryArray& dict_array =
 checked_cast<const DictionaryArray&>(*array.get());
auto indices = dict_array.indices();
switch (indices->type_id()) {
case Type::UINT8:
return Ingest_some_nulls_Impl<arrow::UInt8Type>(data, array, start, n,
 chunk_index);
case Type::INT8:
return Ingest_some_nulls_Impl<arrow::Int8Type>(data, array, start, n,
 chunk_index);
case Type::UINT16:
return Ingest_some_nulls_Impl<arrow::UInt16Type>(data, array, start, n,
 chunk_index);
case Type::INT16:
return Ingest_some_nulls_Impl<arrow::Int16Type>(data, array, start, n,
 chunk_index);
case Type::INT32:
return Ingest_some_nulls_Impl<arrow::Int32Type>(data, array, start, n,
 chunk_index);
case Type::UINT32:
return Ingest_some_nulls_Impl<arrow::UInt32Type>(data, array, start, n,
 chunk_index);
case Type::INT64:
return Ingest_some_nulls_Impl<arrow::Int64Type>(data, array, start, n,
 chunk_index);
case Type::UINT64:
return Ingest_some_nulls_Impl<arrow::UInt64Type>(data, array, start, n,
 chunk_index);
default:
break;
 }
returnStatus::OK();
 }

private:
template <typename Type>
 Status Ingest_some_nulls_Impl(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
using index_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
auto indices = checked_cast<const DictionaryArray&>(*array).indices();
auto raw_indices = indices->data()->GetValues<index_type>(1);

auto p_data = INTEGER(data) + start;
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

// convert the 0-based indices from the arrow Array
// to 1-based indices used in R factors
if (need_unification_) {
// transpose the indices before converting
auto transposed =
reinterpret_cast<constint32_t*>(arrays_transpose_[chunk_index]->data());

auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = transposed[raw_indices[i]] + 1;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 } else {
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(raw_indices[i]) + 1;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
 }

boolGetOrdered() const {
return checked_cast<const DictionaryType&>(*chunked_array_->type()).ordered();
 }

SEXPGetLevels() const {
// R factor levels must be type "character" so coerce `dict` to STRSXP
// TODO (npr): this coercion should be optional, "dictionariesAsFactors" ;)
// Alternative: preserve the logical type of the dictionary values
// (e.g. if dict is timestamp, return a POSIXt R vector, not factor)
if (dictionary_->type_id() != Type::STRING &&
 dictionary_->type_id() != Type::LARGE_STRING) {
 cpp11::safe[Rf_warning]("Coercing dictionary values to R character factor levels");
 }

SEXP vec = PROTECT(Converter::Convert(dictionary_));
SEXP strings_vec = PROTECT(Rf_coerceVector(vec, STRSXP));
UNPROTECT(2);
return strings_vec;
 }
};

classConverter_Struct : publicConverter {
public:
explicitConverter_Struct(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array), converters() {
constauto& struct_type =
 checked_cast<const arrow::StructType&>(*chunked_array->type());

int nf = struct_type.num_fields();

 std::shared_ptr<arrow::Table> array_as_table =
ValueOrStop(arrow::Table::FromChunkedStructArray(chunked_array));
for (int i = 0; i < nf; i++) {
 converters.push_back(Converter::Make(array_as_table->column(i)));
 }
 }

SEXPAllocate(R_xlen_t n) const {
// allocate a data frame column to host each array
// If possible, a column is dealt with directly with altrep
auto type =
 checked_cast<const arrow::StructType*>(this->chunked_array_->type().get());
auto out =
arrow::r::to_r_list(converters, [n](const std::shared_ptr<Converter>& converter) {
SEXP out = converter->MaybeAltrep();
if (Rf_isNull(out)) {
 out = converter->Allocate(n);
 }
return out;
 });
auto colnames = arrow::r::to_r_strings(
 type->fields(),
 [](const std::shared_ptr<Field>& field) { return field->name(); });
 out.attr(symbols::row_names) = arrow::r::short_row_names(static_cast<int>(n));
 out.attr(R_NamesSymbol) = colnames;
 out.attr(R_ClassSymbol) = arrow::r::data::classes_tbl_df;

return out;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
int nf = static_cast<int>(converters.size());
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_all_nulls(data_i, start, n));
 }
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto struct_array = checked_cast<const arrow::StructArray*>(array.get());
int nf = static_cast<int>(converters.size());
// Flatten() deals with merging of nulls
auto arrays = ValueOrStop(struct_array->Flatten(gc_memory_pool()));
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_some_nulls(VECTOR_ELT(data, i), arrays[i],
 start, n, chunk_index));
 }
 }

returnStatus::OK();
 }

virtualboolParallel() const {
// this can only run in parallel if all the
// inner converters can
for (constauto& converter : converters) {
if (!converter->Parallel()) returnfalse;
 }
returntrue;
 }

private:
 std::vector<std::shared_ptr<Converter>> converters;
};

doublems_to_seconds(int64_t ms) { returnstatic_cast<double>(ms) / 1000; }

classConverter_Date64 : publicConverter {
public:
explicitConverter_Date64(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<int64_t>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i] / 1000);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename value_type, typename unit_type = TimeType>
classConverter_Time : publicConverter {
public:
explicitConverter_Time(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = cpp11::writable::strings({"hms", "difftime"});

// hms difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// hms difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type, typename unit_type = DurationType>
classConverter_Duration : publicConverter {
public:
explicitConverter_Duration(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = "difftime";

// difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type>
classConverter_Timestamp : publicConverter_Time<value_type, TimestampType> {
public:
explicitConverter_Timestamp(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter_Time<value_type, TimestampType>(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
auto array_type =
 checked_cast<const TimestampType*>(this->chunked_array_->type().get());
 std::string tzone = array_type->timezone();
if (tzone.size() > 0) {
 data.attr("tzone") = tzone;
 }
return data;
 }
};

template <typename Type>
classConverter_Decimal : publicConverter {
public:
explicitConverter_Decimal(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
// Licensed to the Apache Software Foundation (ASF) under one
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// or more contributor license agreements. See the NOTICE 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_types.h"

#include<arrow/array.h>
#include<arrow/builder.h>
#include<arrow/datum.h>
#include<arrow/table.h>
#include<arrow/util/bitmap_reader.h>
#include<arrow/util/bitmap_writer.h>
#include<arrow/util/int_util.h>

#include<type_traits>

#include"./extension.h"
#include"./r_task_group.h"

namespacearrow {

using internal::checked_cast;
using internal::IntegersCanFit;

namespacer {

classConverter {
public:
explicitConverter(const std::shared_ptr<ChunkedArray>& chunked_array)
 : chunked_array_(std::move(chunked_array)) {}

virtual~Converter() {}

// Allocate a vector of the right R type for this converter
virtualSEXPAllocate(R_xlen_t n) const = 0;

// data[ start:(start + n) ] = NA
virtual Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const = 0;

// ingest the values from the array into data[ start : (start + n)]
//
// chunk_index indicates which of the chunk is being ingested into data. This is
// ignored by most implementations and currently only used with Dictionary
// arrays.
virtual Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n,
size_t chunk_index) const = 0;

// can this run in parallel ?
virtualboolParallel() const { returntrue; }

// converter is passed as self to outlive the scope of Converter::Convert()
SEXPScheduleConvertTasks(RTasks& tasks, std::shared_ptr<Converter> self) {
// try altrep first
SEXP alt = altrep::MakeAltrepVector(chunked_array_);
if (!Rf_isNull(alt)) {
return alt;
 }

// otherwise use the Converter api:

// allocating the R vector upfront
SEXP out = PROTECT(Allocate(chunked_array_->length()));

// for each array, fill the relevant slice of `out`, potentially in parallel
 R_xlen_t k = 0, i = 0;
for (constauto& array : chunked_array_->chunks()) {
auto n_chunk = array->length();

 tasks.Append(Parallel(), [=] {
if (array->null_count() == n_chunk) {
return self->Ingest_all_nulls(out, k, n_chunk);
 } else {
return self->Ingest_some_nulls(out, array, k, n_chunk, i);
 }
 });

 k += n_chunk;
 i++;
 }

UNPROTECT(1);
return out;
 }

// Converter factory
static std::shared_ptr<Converter> Make(
const std::shared_ptr<ChunkedArray>& chunked_array);

staticSEXPLazyConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
 RTasks& tasks) {
auto converter = Make(chunked_array);
return converter->ScheduleConvertTasks(tasks, converter);
 }

staticSEXPConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
bool use_threads) {
 RTasks tasks(use_threads);
SEXP out = PROTECT(Converter::LazyConvert(chunked_array, tasks));
StopIfNotOk(tasks.Finish());

UNPROTECT(1);
return out;
 }

staticSEXPConvert(const std::shared_ptr<Array>& array) {
returnConvert(std::make_shared<ChunkedArray>(array), false);
 }

SEXPMaybeAltrep() { returnaltrep::MakeAltrepVector(chunked_array_); }

protected:
 std::shared_ptr<ChunkedArray> chunked_array_;
};

template <typename SetNonNull, typename SetNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null, SetNull&& set_null) {
if (array->null_count()) {
 internal::BitmapReader bitmap_reader(array->null_bitmap()->data(), array->offset(),
 n);

for (R_xlen_t i = 0; i < n; i++, bitmap_reader.Next()) {
if (bitmap_reader.IsSet()) {
RETURN_NOT_OK(set_non_null(i));
 } else {
RETURN_NOT_OK(set_null(i));
 }
 }

 } else {
for (R_xlen_t i = 0; i < n; i++) {
RETURN_NOT_OK(set_non_null(i));
 }
 }

returnStatus::OK();
}

template <typename SetNonNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null) {
auto nothing = [](R_xlen_t i) { returnStatus::OK(); };
returnIngestSome(array, n, std::forward<SetNonNull>(set_non_null), nothing);
}

std::shared_ptr<Array> CreateEmptyArray(const std::shared_ptr<DataType>& array_type) {
 std::unique_ptr<arrow::ArrayBuilder> builder;
StopIfNotOk(arrow::MakeBuilder(gc_memory_pool(), array_type, &builder));

 std::shared_ptr<arrow::Array> array;
StopIfNotOk(builder->Finish(&array));
return array;
}

template <typename Type>
classConverter_Int : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Int(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(INTSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = INTEGER(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename Type>
classConverter_Double : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Double(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<value_type>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

classConverter_Date32 : publicConverter {
public:
explicitConverter_Date32(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP data = PROTECT(Rf_allocVector(REALSXP, n));
Rf_classgets(data, Rf_mkString("Date"));
UNPROTECT(1);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<int>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename StringArrayType>
structConverter_String : publicConverter {
public:
explicitConverter_String(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(STRSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
for (R_xlen_t i = 0; i < n; i++) {
SET_STRING_ELT(data, i + start, NA_STRING);
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_offset = array->data()->GetValues<int32_t>(1);
if (!p_offset) {
returnStatus::Invalid("Invalid offset buffer");
 }
auto p_strings = array->data()->GetValues<char>(2, *p_offset);
if (!p_strings) {
// There is an offset buffer, but the data buffer is null
// There is at least one value in the array and not all the values are null
// That means all values are either empty strings or nulls so there is nothing to do

if (array->null_count()) {
 arrow::internal::BitmapReader null_reader(array->null_bitmap_data(),
 array->offset(), n);
for (int i = 0; i < n; i++, null_reader.Next()) {
if (null_reader.IsNotSet()) {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 }
returnStatus::OK();
 }

 StringArrayType* string_array = static_cast<StringArrayType*>(array.get());

constbool all_valid = array->null_count() == 0;
constbool strip_out_nuls = GetBoolOption("arrow.skip_nul", false);

bool nul_was_stripped = false;

if (all_valid) {
// no need to watch for missing strings
cpp11::unwind_protect([&] {
if (strip_out_nuls) {
for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 }
return;
 }

for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 }
 });
 } else {
cpp11::unwind_protect([&] {
 arrow::internal::BitmapReader validity_reader(array->null_bitmap_data(),
 array->offset(), n);

if (strip_out_nuls) {
for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
return;
 }

for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 });
 }

if (nul_was_stripped) {
 cpp11::safe[Rf_warning]("Stripping '\\0' (nul) from character vector");
 }

returnStatus::OK();
 }

boolParallel() const { returnfalse; }

private:
staticSEXPr_string_from_view(std::string_view view) {
returnRf_mkCharLenCE(view.data(), static_cast<int>(view.size()), CE_UTF8);
 }

staticSEXPr_string_from_view_strip_nul(std::string_view view,
bool* nul_was_stripped) {
constchar* old_string = view.data();

 std::string stripped_string;
size_t stripped_len = 0, nul_count = 0;

for (size_t i = 0; i < view.size(); i++) {
if (old_string[i] == '\0') {
 ++nul_count;

if (nul_count == 1) {
// first nul spotted: allocate stripped string storage
 stripped_string = std::string(view);
 stripped_len = i;
 }

// don't copy old_string[i] (which is \0) into stripped_string
continue;
 }

if (nul_count > 0) {
 stripped_string[stripped_len++] = old_string[i];
 }
 }

if (nul_count > 0) {
 *nul_was_stripped = true;
 stripped_string.resize(stripped_len);
returnr_string_from_view(stripped_string);
 }

returnr_string_from_view(view);
 }
};

classConverter_Boolean : publicConverter {
public:
explicitConverter_Boolean(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(LGLSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(LOGICAL(data) + start, n, NA_LOGICAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = LOGICAL(data) + start;
auto p_bools = array->data()->GetValues<uint8_t>(1, 0);
if (!p_bools) {
returnStatus::Invalid("Invalid data buffer");
 }

 arrow::internal::BitmapReader data_reader(p_bools, array->offset(), n);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = data_reader.IsSet();
 data_reader.Next();
returnStatus::OK();
 };

auto null_one = [&](R_xlen_t i) {
 data_reader.Next();
 p_data[i] = NA_LOGICAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename ArrayType>
classConverter_Binary : publicConverter {
public:
using offset_type = typename ArrayType::offset_type;
explicitConverter_Binary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
if (std::is_same<ArrayType, BinaryArray>::value) {
Rf_classgets(res, data::classes_arrow_binary);
 } else {
Rf_classgets(res, data::classes_arrow_large_binary);
 }
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const ArrayType* binary_array = checked_cast<const ArrayType*>(array.get());

auto ingest_one = [&](R_xlen_t i) {
 offset_type ni;
auto value = binary_array->GetValue(i, &ni);
if (ni > R_XLEN_T_MAX) {
returnStatus::RError("Array too big to be represented as a raw vector");
 }
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, ni));
std::copy(value, value + ni, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }
};

classConverter_FixedSizeBinary : publicConverter {
public:
explicitConverter_FixedSizeBinary(const std::shared_ptr<ChunkedArray>& chunked_array,
int byte_width)
 : Converter(chunked_array), byte_width_(byte_width) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
Rf_classgets(res, data::classes_arrow_fixed_size_binary);
Rf_setAttrib(res, symbols::byte_width, Rf_ScalarInteger(byte_width_));
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const FixedSizeBinaryArray* binary_array =
 checked_cast<const FixedSizeBinaryArray*>(array.get());

int byte_width = binary_array->byte_width();
auto ingest_one = [&, byte_width](R_xlen_t i) {
auto value = binary_array->GetValue(i);
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, byte_width));
std::copy(value, value + byte_width, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }

private:
int byte_width_;
};

boolDictionaryChunkArrayNeedUnification(
const std::shared_ptr<ChunkedArray>& chunked_array) {
int n = chunked_array->num_chunks();
if (n < 2) {
returnfalse;
 }
constauto& arr_first =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
for (int i = 1; i < n; i++) {
constauto& arr =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(i));
if (!(arr_first.dictionary()->Equals(arr.dictionary()))) {
returntrue;
 }
 }
returnfalse;
}

classConverter_Dictionary : publicConverter {
private:
bool need_unification_;
 std::unique_ptr<arrow::DictionaryUnifier> unifier_;
 std::vector<std::shared_ptr<Buffer>> arrays_transpose_;
 std::shared_ptr<DataType> out_type_;
 std::shared_ptr<Array> dictionary_;

public:
explicitConverter_Dictionary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array),
 need_unification_(DictionaryChunkArrayNeedUnification(chunked_array)) {
if (need_unification_) {
constauto& arr_type = checked_cast<const DictionaryType&>(*chunked_array->type());
 unifier_ = ValueOrStop(DictionaryUnifier::Make(arr_type.value_type()));

int n_arrays = chunked_array->num_chunks();
 arrays_transpose_.resize(n_arrays);

for (int i = 0; i < n_arrays; i++) {
constauto& dict_i =
 *checked_cast<const DictionaryArray&>(*chunked_array->chunk(i)).dictionary();
StopIfNotOk(unifier_->Unify(dict_i, &arrays_transpose_[i]));
 }

StopIfNotOk(unifier_->GetResult(&out_type_, &dictionary_));
 } else {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());

constauto& indices_type = *dict_type.index_type();
switch (indices_type.id()) {
case Type::UINT8:
case Type::INT8:
case Type::UINT16:
case Type::INT16:
case Type::INT32:
case Type::UINT32:
case Type::INT64:
case Type::UINT64:
break;
default:
cpp11::stop("Cannot convert Dictionary Array of type `%s` to R",
 dict_type.ToString().c_str());
 }

if (chunked_array->num_chunks() > 0) {
// DictionaryChunkArrayNeedUnification() returned false so we can safely assume
// the dictionary of the first chunk applies everywhere
constauto& dict_array =
 checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
 dictionary_ = dict_array.dictionary();
 } else {
 dictionary_ = CreateEmptyArray(dict_type.value_type());
 }
 }

// R factors store their codes in 32-bit integers, so dictionary arrays with
// more levels than that cannot be represented safely.
if (dictionary_->length() > std::numeric_limits<int>::max()) {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());
cpp11::stop(
"Cannot convert Dictionary Array of type `%s` to R: dictionary has "
"more levels than an R factor can represent",
 dict_type.ToString().c_str());
 }
 }

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::integers data(n);
 data.attr("levels") = GetLevels();
if (GetOrdered()) {
Rf_classgets(data, arrow::r::data::classes_ordered);
 } else {
Rf_classgets(data, arrow::r::data::classes_factor);
 }
return data;
 }

virtualboolParallel() const { returnfalse; }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const DictionaryArray& dict_array =
 checked_cast<const DictionaryArray&>(*array.get());
auto indices = dict_array.indices();
switch (indices->type_id()) {
case Type::UINT8:
return Ingest_some_nulls_Impl<arrow::UInt8Type>(data, array, start, n,
 chunk_index);
case Type::INT8:
return Ingest_some_nulls_Impl<arrow::Int8Type>(data, array, start, n,
 chunk_index);
case Type::UINT16:
return Ingest_some_nulls_Impl<arrow::UInt16Type>(data, array, start, n,
 chunk_index);
case Type::INT16:
return Ingest_some_nulls_Impl<arrow::Int16Type>(data, array, start, n,
 chunk_index);
case Type::INT32:
return Ingest_some_nulls_Impl<arrow::Int32Type>(data, array, start, n,
 chunk_index);
case Type::UINT32:
return Ingest_some_nulls_Impl<arrow::UInt32Type>(data, array, start, n,
 chunk_index);
case Type::INT64:
return Ingest_some_nulls_Impl<arrow::Int64Type>(data, array, start, n,
 chunk_index);
case Type::UINT64:
return Ingest_some_nulls_Impl<arrow::UInt64Type>(data, array, start, n,
 chunk_index);
default:
break;
 }
returnStatus::OK();
 }

private:
template <typename Type>
 Status Ingest_some_nulls_Impl(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
using index_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
auto indices = checked_cast<const DictionaryArray&>(*array).indices();
auto raw_indices = indices->data()->GetValues<index_type>(1);

auto p_data = INTEGER(data) + start;
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

// convert the 0-based indices from the arrow Array
// to 1-based indices used in R factors
if (need_unification_) {
// transpose the indices before converting
auto transposed =
reinterpret_cast<constint32_t*>(arrays_transpose_[chunk_index]->data());

auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = transposed[raw_indices[i]] + 1;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 } else {
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(raw_indices[i]) + 1;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
 }

boolGetOrdered() const {
return checked_cast<const DictionaryType&>(*chunked_array_->type()).ordered();
 }

SEXPGetLevels() const {
// R factor levels must be type "character" so coerce `dict` to STRSXP
// TODO (npr): this coercion should be optional, "dictionariesAsFactors" ;)
// Alternative: preserve the logical type of the dictionary values
// (e.g. if dict is timestamp, return a POSIXt R vector, not factor)
if (dictionary_->type_id() != Type::STRING &&
 dictionary_->type_id() != Type::LARGE_STRING) {
 cpp11::safe[Rf_warning]("Coercing dictionary values to R character factor levels");
 }

SEXP vec = PROTECT(Converter::Convert(dictionary_));
SEXP strings_vec = PROTECT(Rf_coerceVector(vec, STRSXP));
UNPROTECT(2);
return strings_vec;
 }
};

classConverter_Struct : publicConverter {
public:
explicitConverter_Struct(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array), converters() {
constauto& struct_type =
 checked_cast<const arrow::StructType&>(*chunked_array->type());

int nf = struct_type.num_fields();

 std::shared_ptr<arrow::Table> array_as_table =
ValueOrStop(arrow::Table::FromChunkedStructArray(chunked_array));
for (int i = 0; i < nf; i++) {
 converters.push_back(Converter::Make(array_as_table->column(i)));
 }
 }

SEXPAllocate(R_xlen_t n) const {
// allocate a data frame column to host each array
// If possible, a column is dealt with directly with altrep
auto type =
 checked_cast<const arrow::StructType*>(this->chunked_array_->type().get());
auto out =
arrow::r::to_r_list(converters, [n](const std::shared_ptr<Converter>& converter) {
SEXP out = converter->MaybeAltrep();
if (Rf_isNull(out)) {
 out = converter->Allocate(n);
 }
return out;
 });
auto colnames = arrow::r::to_r_strings(
 type->fields(),
 [](const std::shared_ptr<Field>& field) { return field->name(); });
 out.attr(symbols::row_names) = arrow::r::short_row_names(static_cast<int>(n));
 out.attr(R_NamesSymbol) = colnames;
 out.attr(R_ClassSymbol) = arrow::r::data::classes_tbl_df;

return out;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
int nf = static_cast<int>(converters.size());
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_all_nulls(data_i, start, n));
 }
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto struct_array = checked_cast<const arrow::StructArray*>(array.get());
int nf = static_cast<int>(converters.size());
// Flatten() deals with merging of nulls
auto arrays = ValueOrStop(struct_array->Flatten(gc_memory_pool()));
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_some_nulls(VECTOR_ELT(data, i), arrays[i],
 start, n, chunk_index));
 }
 }

returnStatus::OK();
 }

virtualboolParallel() const {
// this can only run in parallel if all the
// inner converters can
for (constauto& converter : converters) {
if (!converter->Parallel()) returnfalse;
 }
returntrue;
 }

private:
 std::vector<std::shared_ptr<Converter>> converters;
};

doublems_to_seconds(int64_t ms) { returnstatic_cast<double>(ms) / 1000; }

classConverter_Date64 : publicConverter {
public:
explicitConverter_Date64(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<int64_t>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i] / 1000);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename value_type, typename unit_type = TimeType>
classConverter_Time : publicConverter {
public:
explicitConverter_Time(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = cpp11::writable::strings({"hms", "difftime"});

// hms difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// hms difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type, typename unit_type = DurationType>
classConverter_Duration : publicConverter {
public:
explicitConverter_Duration(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = "difftime";

// difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type>
classConverter_Timestamp : publicConverter_Time<value_type, TimestampType> {
public:
explicitConverter_Timestamp(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter_Time<value_type, TimestampType>(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
auto array_type =
 checked_cast<const TimestampType*>(this->chunked_array_->type().get());
 std::string tzone = array_type->timezone();
if (tzone.size() > 0) {
 data.attr("tzone") = tzone;
 }
return data;
 }
};

template <typename Type>
classConverter_Decimal : publicConverter {
public:
explicitConverter_Decimal(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
// Licensed to the Apache Software Foundation (ASF) under one
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// or more contributor license agreements. See the NOTICE 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_types.h"

#include<arrow/array.h>
#include<arrow/builder.h>
#include<arrow/datum.h>
#include<arrow/table.h>
#include<arrow/util/bitmap_reader.h>
#include<arrow/util/bitmap_writer.h>
#include<arrow/util/int_util.h>

#include<type_traits>

#include"./extension.h"
#include"./r_task_group.h"

namespacearrow {

using internal::checked_cast;
using internal::IntegersCanFit;

namespacer {

classConverter {
public:
explicitConverter(const std::shared_ptr<ChunkedArray>& chunked_array)
 : chunked_array_(std::move(chunked_array)) {}

virtual~Converter() {}

// Allocate a vector of the right R type for this converter
virtualSEXPAllocate(R_xlen_t n) const = 0;

// data[ start:(start + n) ] = NA
virtual Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const = 0;

// ingest the values from the array into data[ start : (start + n)]
//
// chunk_index indicates which of the chunk is being ingested into data. This is
// ignored by most implementations and currently only used with Dictionary
// arrays.
virtual Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n,
size_t chunk_index) const = 0;

// can this run in parallel ?
virtualboolParallel() const { returntrue; }

// converter is passed as self to outlive the scope of Converter::Convert()
SEXPScheduleConvertTasks(RTasks& tasks, std::shared_ptr<Converter> self) {
// try altrep first
SEXP alt = altrep::MakeAltrepVector(chunked_array_);
if (!Rf_isNull(alt)) {
return alt;
 }

// otherwise use the Converter api:

// allocating the R vector upfront
SEXP out = PROTECT(Allocate(chunked_array_->length()));

// for each array, fill the relevant slice of `out`, potentially in parallel
 R_xlen_t k = 0, i = 0;
for (constauto& array : chunked_array_->chunks()) {
auto n_chunk = array->length();

 tasks.Append(Parallel(), [=] {
if (array->null_count() == n_chunk) {
return self->Ingest_all_nulls(out, k, n_chunk);
 } else {
return self->Ingest_some_nulls(out, array, k, n_chunk, i);
 }
 });

 k += n_chunk;
 i++;
 }

UNPROTECT(1);
return out;
 }

// Converter factory
static std::shared_ptr<Converter> Make(
const std::shared_ptr<ChunkedArray>& chunked_array);

staticSEXPLazyConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
 RTasks& tasks) {
auto converter = Make(chunked_array);
return converter->ScheduleConvertTasks(tasks, converter);
 }

staticSEXPConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
bool use_threads) {
 RTasks tasks(use_threads);
SEXP out = PROTECT(Converter::LazyConvert(chunked_array, tasks));
StopIfNotOk(tasks.Finish());

UNPROTECT(1);
return out;
 }

staticSEXPConvert(const std::shared_ptr<Array>& array) {
returnConvert(std::make_shared<ChunkedArray>(array), false);
 }

SEXPMaybeAltrep() { returnaltrep::MakeAltrepVector(chunked_array_); }

protected:
 std::shared_ptr<ChunkedArray> chunked_array_;
};

template <typename SetNonNull, typename SetNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null, SetNull&& set_null) {
if (array->null_count()) {
 internal::BitmapReader bitmap_reader(array->null_bitmap()->data(), array->offset(),
 n);

for (R_xlen_t i = 0; i < n; i++, bitmap_reader.Next()) {
if (bitmap_reader.IsSet()) {
RETURN_NOT_OK(set_non_null(i));
 } else {
RETURN_NOT_OK(set_null(i));
 }
 }

 } else {
for (R_xlen_t i = 0; i < n; i++) {
RETURN_NOT_OK(set_non_null(i));
 }
 }

returnStatus::OK();
}

template <typename SetNonNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null) {
auto nothing = [](R_xlen_t i) { returnStatus::OK(); };
returnIngestSome(array, n, std::forward<SetNonNull>(set_non_null), nothing);
}

std::shared_ptr<Array> CreateEmptyArray(const std::shared_ptr<DataType>& array_type) {
 std::unique_ptr<arrow::ArrayBuilder> builder;
StopIfNotOk(arrow::MakeBuilder(gc_memory_pool(), array_type, &builder));

 std::shared_ptr<arrow::Array> array;
StopIfNotOk(builder->Finish(&array));
return array;
}

template <typename Type>
classConverter_Int : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Int(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(INTSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = INTEGER(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename Type>
classConverter_Double : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Double(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<value_type>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

classConverter_Date32 : publicConverter {
public:
explicitConverter_Date32(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP data = PROTECT(Rf_allocVector(REALSXP, n));
Rf_classgets(data, Rf_mkString("Date"));
UNPROTECT(1);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<int>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename StringArrayType>
structConverter_String : publicConverter {
public:
explicitConverter_String(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(STRSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
for (R_xlen_t i = 0; i < n; i++) {
SET_STRING_ELT(data, i + start, NA_STRING);
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_offset = array->data()->GetValues<int32_t>(1);
if (!p_offset) {
returnStatus::Invalid("Invalid offset buffer");
 }
auto p_strings = array->data()->GetValues<char>(2, *p_offset);
if (!p_strings) {
// There is an offset buffer, but the data buffer is null
// There is at least one value in the array and not all the values are null
// That means all values are either empty strings or nulls so there is nothing to do

if (array->null_count()) {
 arrow::internal::BitmapReader null_reader(array->null_bitmap_data(),
 array->offset(), n);
for (int i = 0; i < n; i++, null_reader.Next()) {
if (null_reader.IsNotSet()) {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 }
returnStatus::OK();
 }

 StringArrayType* string_array = static_cast<StringArrayType*>(array.get());

constbool all_valid = array->null_count() == 0;
constbool strip_out_nuls = GetBoolOption("arrow.skip_nul", false);

bool nul_was_stripped = false;

if (all_valid) {
// no need to watch for missing strings
cpp11::unwind_protect([&] {
if (strip_out_nuls) {
for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 }
return;
 }

for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 }
 });
 } else {
cpp11::unwind_protect([&] {
 arrow::internal::BitmapReader validity_reader(array->null_bitmap_data(),
 array->offset(), n);

if (strip_out_nuls) {
for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
return;
 }

for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 });
 }

if (nul_was_stripped) {
 cpp11::safe[Rf_warning]("Stripping '\\0' (nul) from character vector");
 }

returnStatus::OK();
 }

boolParallel() const { returnfalse; }

private:
staticSEXPr_string_from_view(std::string_view view) {
returnRf_mkCharLenCE(view.data(), static_cast<int>(view.size()), CE_UTF8);
 }

staticSEXPr_string_from_view_strip_nul(std::string_view view,
bool* nul_was_stripped) {
constchar* old_string = view.data();

 std::string stripped_string;
size_t stripped_len = 0, nul_count = 0;

for (size_t i = 0; i < view.size(); i++) {
if (old_string[i] == '\0') {
 ++nul_count;

if (nul_count == 1) {
// first nul spotted: allocate stripped string storage
 stripped_string = std::string(view);
 stripped_len = i;
 }

// don't copy old_string[i] (which is \0) into stripped_string
continue;
 }

if (nul_count > 0) {
 stripped_string[stripped_len++] = old_string[i];
 }
 }

if (nul_count > 0) {
 *nul_was_stripped = true;
 stripped_string.resize(stripped_len);
returnr_string_from_view(stripped_string);
 }

returnr_string_from_view(view);
 }
};

classConverter_Boolean : publicConverter {
public:
explicitConverter_Boolean(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(LGLSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(LOGICAL(data) + start, n, NA_LOGICAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = LOGICAL(data) + start;
auto p_bools = array->data()->GetValues<uint8_t>(1, 0);
if (!p_bools) {
returnStatus::Invalid("Invalid data buffer");
 }

 arrow::internal::BitmapReader data_reader(p_bools, array->offset(), n);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = data_reader.IsSet();
 data_reader.Next();
returnStatus::OK();
 };

auto null_one = [&](R_xlen_t i) {
 data_reader.Next();
 p_data[i] = NA_LOGICAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename ArrayType>
classConverter_Binary : publicConverter {
public:
using offset_type = typename ArrayType::offset_type;
explicitConverter_Binary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
if (std::is_same<ArrayType, BinaryArray>::value) {
Rf_classgets(res, data::classes_arrow_binary);
 } else {
Rf_classgets(res, data::classes_arrow_large_binary);
 }
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const ArrayType* binary_array = checked_cast<const ArrayType*>(array.get());

auto ingest_one = [&](R_xlen_t i) {
 offset_type ni;
auto value = binary_array->GetValue(i, &ni);
if (ni > R_XLEN_T_MAX) {
returnStatus::RError("Array too big to be represented as a raw vector");
 }
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, ni));
std::copy(value, value + ni, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }
};

classConverter_FixedSizeBinary : publicConverter {
public:
explicitConverter_FixedSizeBinary(const std::shared_ptr<ChunkedArray>& chunked_array,
int byte_width)
 : Converter(chunked_array), byte_width_(byte_width) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
Rf_classgets(res, data::classes_arrow_fixed_size_binary);
Rf_setAttrib(res, symbols::byte_width, Rf_ScalarInteger(byte_width_));
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const FixedSizeBinaryArray* binary_array =
 checked_cast<const FixedSizeBinaryArray*>(array.get());

int byte_width = binary_array->byte_width();
auto ingest_one = [&, byte_width](R_xlen_t i) {
auto value = binary_array->GetValue(i);
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, byte_width));
std::copy(value, value + byte_width, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }

private:
int byte_width_;
};

boolDictionaryChunkArrayNeedUnification(
const std::shared_ptr<ChunkedArray>& chunked_array) {
int n = chunked_array->num_chunks();
if (n < 2) {
returnfalse;
 }
constauto& arr_first =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
for (int i = 1; i < n; i++) {
constauto& arr =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(i));
if (!(arr_first.dictionary()->Equals(arr.dictionary()))) {
returntrue;
 }
 }
returnfalse;
}

classConverter_Dictionary : publicConverter {
private:
bool need_unification_;
 std::unique_ptr<arrow::DictionaryUnifier> unifier_;
 std::vector<std::shared_ptr<Buffer>> arrays_transpose_;
 std::shared_ptr<DataType> out_type_;
 std::shared_ptr<Array> dictionary_;

public:
explicitConverter_Dictionary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array),
 need_unification_(DictionaryChunkArrayNeedUnification(chunked_array)) {
if (need_unification_) {
constauto& arr_type = checked_cast<const DictionaryType&>(*chunked_array->type());
 unifier_ = ValueOrStop(DictionaryUnifier::Make(arr_type.value_type()));

int n_arrays = chunked_array->num_chunks();
 arrays_transpose_.resize(n_arrays);

for (int i = 0; i < n_arrays; i++) {
constauto& dict_i =
 *checked_cast<const DictionaryArray&>(*chunked_array->chunk(i)).dictionary();
StopIfNotOk(unifier_->Unify(dict_i, &arrays_transpose_[i]));
 }

StopIfNotOk(unifier_->GetResult(&out_type_, &dictionary_));
 } else {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());

constauto& indices_type = *dict_type.index_type();
switch (indices_type.id()) {
case Type::UINT8:
case Type::INT8:
case Type::UINT16:
case Type::INT16:
case Type::INT32:
case Type::UINT32:
case Type::INT64:
case Type::UINT64:
break;
default:
cpp11::stop("Cannot convert Dictionary Array of type `%s` to R",
 dict_type.ToString().c_str());
 }

if (chunked_array->num_chunks() > 0) {
// DictionaryChunkArrayNeedUnification() returned false so we can safely assume
// the dictionary of the first chunk applies everywhere
constauto& dict_array =
 checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
 dictionary_ = dict_array.dictionary();
 } else {
 dictionary_ = CreateEmptyArray(dict_type.value_type());
 }
 }

// R factors store their codes in 32-bit integers, so dictionary arrays with
// more levels than that cannot be represented safely.
if (dictionary_->length() > std::numeric_limits<int>::max()) {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());
cpp11::stop(
"Cannot convert Dictionary Array of type `%s` to R: dictionary has "
"more levels than an R factor can represent",
 dict_type.ToString().c_str());
 }
 }

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::integers data(n);
 data.attr("levels") = GetLevels();
if (GetOrdered()) {
Rf_classgets(data, arrow::r::data::classes_ordered);
 } else {
Rf_classgets(data, arrow::r::data::classes_factor);
 }
return data;
 }

virtualboolParallel() const { returnfalse; }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const DictionaryArray& dict_array =
 checked_cast<const DictionaryArray&>(*array.get());
auto indices = dict_array.indices();
switch (indices->type_id()) {
case Type::UINT8:
return Ingest_some_nulls_Impl<arrow::UInt8Type>(data, array, start, n,
 chunk_index);
case Type::INT8:
return Ingest_some_nulls_Impl<arrow::Int8Type>(data, array, start, n,
 chunk_index);
case Type::UINT16:
return Ingest_some_nulls_Impl<arrow::UInt16Type>(data, array, start, n,
 chunk_index);
case Type::INT16:
return Ingest_some_nulls_Impl<arrow::Int16Type>(data, array, start, n,
 chunk_index);
case Type::INT32:
return Ingest_some_nulls_Impl<arrow::Int32Type>(data, array, start, n,
 chunk_index);
case Type::UINT32:
return Ingest_some_nulls_Impl<arrow::UInt32Type>(data, array, start, n,
 chunk_index);
case Type::INT64:
return Ingest_some_nulls_Impl<arrow::Int64Type>(data, array, start, n,
 chunk_index);
case Type::UINT64:
return Ingest_some_nulls_Impl<arrow::UInt64Type>(data, array, start, n,
 chunk_index);
default:
break;
 }
returnStatus::OK();
 }

private:
template <typename Type>
 Status Ingest_some_nulls_Impl(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
using index_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
auto indices = checked_cast<const DictionaryArray&>(*array).indices();
auto raw_indices = indices->data()->GetValues<index_type>(1);

auto p_data = INTEGER(data) + start;
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

// convert the 0-based indices from the arrow Array
// to 1-based indices used in R factors
if (need_unification_) {
// transpose the indices before converting
auto transposed =
reinterpret_cast<constint32_t*>(arrays_transpose_[chunk_index]->data());

auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = transposed[raw_indices[i]] + 1;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 } else {
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(raw_indices[i]) + 1;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
 }

boolGetOrdered() const {
return checked_cast<const DictionaryType&>(*chunked_array_->type()).ordered();
 }

SEXPGetLevels() const {
// R factor levels must be type "character" so coerce `dict` to STRSXP
// TODO (npr): this coercion should be optional, "dictionariesAsFactors" ;)
// Alternative: preserve the logical type of the dictionary values
// (e.g. if dict is timestamp, return a POSIXt R vector, not factor)
if (dictionary_->type_id() != Type::STRING &&
 dictionary_->type_id() != Type::LARGE_STRING) {
 cpp11::safe[Rf_warning]("Coercing dictionary values to R character factor levels");
 }

SEXP vec = PROTECT(Converter::Convert(dictionary_));
SEXP strings_vec = PROTECT(Rf_coerceVector(vec, STRSXP));
UNPROTECT(2);
return strings_vec;
 }
};

classConverter_Struct : publicConverter {
public:
explicitConverter_Struct(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array), converters() {
constauto& struct_type =
 checked_cast<const arrow::StructType&>(*chunked_array->type());

int nf = struct_type.num_fields();

 std::shared_ptr<arrow::Table> array_as_table =
ValueOrStop(arrow::Table::FromChunkedStructArray(chunked_array));
for (int i = 0; i < nf; i++) {
 converters.push_back(Converter::Make(array_as_table->column(i)));
 }
 }

SEXPAllocate(R_xlen_t n) const {
// allocate a data frame column to host each array
// If possible, a column is dealt with directly with altrep
auto type =
 checked_cast<const arrow::StructType*>(this->chunked_array_->type().get());
auto out =
arrow::r::to_r_list(converters, [n](const std::shared_ptr<Converter>& converter) {
SEXP out = converter->MaybeAltrep();
if (Rf_isNull(out)) {
 out = converter->Allocate(n);
 }
return out;
 });
auto colnames = arrow::r::to_r_strings(
 type->fields(),
 [](const std::shared_ptr<Field>& field) { return field->name(); });
 out.attr(symbols::row_names) = arrow::r::short_row_names(static_cast<int>(n));
 out.attr(R_NamesSymbol) = colnames;
 out.attr(R_ClassSymbol) = arrow::r::data::classes_tbl_df;

return out;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
int nf = static_cast<int>(converters.size());
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_all_nulls(data_i, start, n));
 }
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto struct_array = checked_cast<const arrow::StructArray*>(array.get());
int nf = static_cast<int>(converters.size());
// Flatten() deals with merging of nulls
auto arrays = ValueOrStop(struct_array->Flatten(gc_memory_pool()));
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_some_nulls(VECTOR_ELT(data, i), arrays[i],
 start, n, chunk_index));
 }
 }

returnStatus::OK();
 }

virtualboolParallel() const {
// this can only run in parallel if all the
// inner converters can
for (constauto& converter : converters) {
if (!converter->Parallel()) returnfalse;
 }
returntrue;
 }

private:
 std::vector<std::shared_ptr<Converter>> converters;
};

doublems_to_seconds(int64_t ms) { returnstatic_cast<double>(ms) / 1000; }

classConverter_Date64 : publicConverter {
public:
explicitConverter_Date64(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<int64_t>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i] / 1000);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename value_type, typename unit_type = TimeType>
classConverter_Time : publicConverter {
public:
explicitConverter_Time(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = cpp11::writable::strings({"hms", "difftime"});

// hms difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// hms difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type, typename unit_type = DurationType>
classConverter_Duration : publicConverter {
public:
explicitConverter_Duration(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = "difftime";

// difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type>
classConverter_Timestamp : publicConverter_Time<value_type, TimestampType> {
public:
explicitConverter_Timestamp(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter_Time<value_type, TimestampType>(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
auto array_type =
 checked_cast<const TimestampType*>(this->chunked_array_->type().get());
 std::string tzone = array_type->timezone();
if (tzone.size() > 0) {
 data.attr("tzone") = tzone;
 }
return data;
 }
};

template <typename Type>
classConverter_Decimal : publicConverter {
public:
explicitConverter_Decimal(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
// Licensed to the Apache Software Foundation (ASF) under one
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// or more contributor license agreements. See the NOTICE 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_types.h"

#include<arrow/array.h>
#include<arrow/builder.h>
#include<arrow/datum.h>
#include<arrow/table.h>
#include<arrow/util/bitmap_reader.h>
#include<arrow/util/bitmap_writer.h>
#include<arrow/util/int_util.h>

#include<type_traits>

#include"./extension.h"
#include"./r_task_group.h"

namespacearrow {

using internal::checked_cast;
using internal::IntegersCanFit;

namespacer {

classConverter {
public:
explicitConverter(const std::shared_ptr<ChunkedArray>& chunked_array)
 : chunked_array_(std::move(chunked_array)) {}

virtual~Converter() {}

// Allocate a vector of the right R type for this converter
virtualSEXPAllocate(R_xlen_t n) const = 0;

// data[ start:(start + n) ] = NA
virtual Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const = 0;

// ingest the values from the array into data[ start : (start + n)]
//
// chunk_index indicates which of the chunk is being ingested into data. This is
// ignored by most implementations and currently only used with Dictionary
// arrays.
virtual Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n,
size_t chunk_index) const = 0;

// can this run in parallel ?
virtualboolParallel() const { returntrue; }

// converter is passed as self to outlive the scope of Converter::Convert()
SEXPScheduleConvertTasks(RTasks& tasks, std::shared_ptr<Converter> self) {
// try altrep first
SEXP alt = altrep::MakeAltrepVector(chunked_array_);
if (!Rf_isNull(alt)) {
return alt;
 }

// otherwise use the Converter api:

// allocating the R vector upfront
SEXP out = PROTECT(Allocate(chunked_array_->length()));

// for each array, fill the relevant slice of `out`, potentially in parallel
 R_xlen_t k = 0, i = 0;
for (constauto& array : chunked_array_->chunks()) {
auto n_chunk = array->length();

 tasks.Append(Parallel(), [=] {
if (array->null_count() == n_chunk) {
return self->Ingest_all_nulls(out, k, n_chunk);
 } else {
return self->Ingest_some_nulls(out, array, k, n_chunk, i);
 }
 });

 k += n_chunk;
 i++;
 }

UNPROTECT(1);
return out;
 }

// Converter factory
static std::shared_ptr<Converter> Make(
const std::shared_ptr<ChunkedArray>& chunked_array);

staticSEXPLazyConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
 RTasks& tasks) {
auto converter = Make(chunked_array);
return converter->ScheduleConvertTasks(tasks, converter);
 }

staticSEXPConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
bool use_threads) {
 RTasks tasks(use_threads);
SEXP out = PROTECT(Converter::LazyConvert(chunked_array, tasks));
StopIfNotOk(tasks.Finish());

UNPROTECT(1);
return out;
 }

staticSEXPConvert(const std::shared_ptr<Array>& array) {
returnConvert(std::make_shared<ChunkedArray>(array), false);
 }

SEXPMaybeAltrep() { returnaltrep::MakeAltrepVector(chunked_array_); }

protected:
 std::shared_ptr<ChunkedArray> chunked_array_;
};

template <typename SetNonNull, typename SetNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null, SetNull&& set_null) {
if (array->null_count()) {
 internal::BitmapReader bitmap_reader(array->null_bitmap()->data(), array->offset(),
 n);

for (R_xlen_t i = 0; i < n; i++, bitmap_reader.Next()) {
if (bitmap_reader.IsSet()) {
RETURN_NOT_OK(set_non_null(i));
 } else {
RETURN_NOT_OK(set_null(i));
 }
 }

 } else {
for (R_xlen_t i = 0; i < n; i++) {
RETURN_NOT_OK(set_non_null(i));
 }
 }

returnStatus::OK();
}

template <typename SetNonNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null) {
auto nothing = [](R_xlen_t i) { returnStatus::OK(); };
returnIngestSome(array, n, std::forward<SetNonNull>(set_non_null), nothing);
}

std::shared_ptr<Array> CreateEmptyArray(const std::shared_ptr<DataType>& array_type) {
 std::unique_ptr<arrow::ArrayBuilder> builder;
StopIfNotOk(arrow::MakeBuilder(gc_memory_pool(), array_type, &builder));

 std::shared_ptr<arrow::Array> array;
StopIfNotOk(builder->Finish(&array));
return array;
}

template <typename Type>
classConverter_Int : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Int(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(INTSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = INTEGER(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename Type>
classConverter_Double : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Double(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<value_type>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

classConverter_Date32 : publicConverter {
public:
explicitConverter_Date32(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP data = PROTECT(Rf_allocVector(REALSXP, n));
Rf_classgets(data, Rf_mkString("Date"));
UNPROTECT(1);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<int>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename StringArrayType>
structConverter_String : publicConverter {
public:
explicitConverter_String(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(STRSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
for (R_xlen_t i = 0; i < n; i++) {
SET_STRING_ELT(data, i + start, NA_STRING);
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_offset = array->data()->GetValues<int32_t>(1);
if (!p_offset) {
returnStatus::Invalid("Invalid offset buffer");
 }
auto p_strings = array->data()->GetValues<char>(2, *p_offset);
if (!p_strings) {
// There is an offset buffer, but the data buffer is null
// There is at least one value in the array and not all the values are null
// That means all values are either empty strings or nulls so there is nothing to do

if (array->null_count()) {
 arrow::internal::BitmapReader null_reader(array->null_bitmap_data(),
 array->offset(), n);
for (int i = 0; i < n; i++, null_reader.Next()) {
if (null_reader.IsNotSet()) {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 }
returnStatus::OK();
 }

 StringArrayType* string_array = static_cast<StringArrayType*>(array.get());

constbool all_valid = array->null_count() == 0;
constbool strip_out_nuls = GetBoolOption("arrow.skip_nul", false);

bool nul_was_stripped = false;

if (all_valid) {
// no need to watch for missing strings
cpp11::unwind_protect([&] {
if (strip_out_nuls) {
for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 }
return;
 }

for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 }
 });
 } else {
cpp11::unwind_protect([&] {
 arrow::internal::BitmapReader validity_reader(array->null_bitmap_data(),
 array->offset(), n);

if (strip_out_nuls) {
for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
return;
 }

for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 });
 }

if (nul_was_stripped) {
 cpp11::safe[Rf_warning]("Stripping '\\0' (nul) from character vector");
 }

returnStatus::OK();
 }

boolParallel() const { returnfalse; }

private:
staticSEXPr_string_from_view(std::string_view view) {
returnRf_mkCharLenCE(view.data(), static_cast<int>(view.size()), CE_UTF8);
 }

staticSEXPr_string_from_view_strip_nul(std::string_view view,
bool* nul_was_stripped) {
constchar* old_string = view.data();

 std::string stripped_string;
size_t stripped_len = 0, nul_count = 0;

for (size_t i = 0; i < view.size(); i++) {
if (old_string[i] == '\0') {
 ++nul_count;

if (nul_count == 1) {
// first nul spotted: allocate stripped string storage
 stripped_string = std::string(view);
 stripped_len = i;
 }

// don't copy old_string[i] (which is \0) into stripped_string
continue;
 }

if (nul_count > 0) {
 stripped_string[stripped_len++] = old_string[i];
 }
 }

if (nul_count > 0) {
 *nul_was_stripped = true;
 stripped_string.resize(stripped_len);
returnr_string_from_view(stripped_string);
 }

returnr_string_from_view(view);
 }
};

classConverter_Boolean : publicConverter {
public:
explicitConverter_Boolean(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(LGLSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(LOGICAL(data) + start, n, NA_LOGICAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = LOGICAL(data) + start;
auto p_bools = array->data()->GetValues<uint8_t>(1, 0);
if (!p_bools) {
returnStatus::Invalid("Invalid data buffer");
 }

 arrow::internal::BitmapReader data_reader(p_bools, array->offset(), n);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = data_reader.IsSet();
 data_reader.Next();
returnStatus::OK();
 };

auto null_one = [&](R_xlen_t i) {
 data_reader.Next();
 p_data[i] = NA_LOGICAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename ArrayType>
classConverter_Binary : publicConverter {
public:
using offset_type = typename ArrayType::offset_type;
explicitConverter_Binary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
if (std::is_same<ArrayType, BinaryArray>::value) {
Rf_classgets(res, data::classes_arrow_binary);
 } else {
Rf_classgets(res, data::classes_arrow_large_binary);
 }
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const ArrayType* binary_array = checked_cast<const ArrayType*>(array.get());

auto ingest_one = [&](R_xlen_t i) {
 offset_type ni;
auto value = binary_array->GetValue(i, &ni);
if (ni > R_XLEN_T_MAX) {
returnStatus::RError("Array too big to be represented as a raw vector");
 }
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, ni));
std::copy(value, value + ni, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }
};

classConverter_FixedSizeBinary : publicConverter {
public:
explicitConverter_FixedSizeBinary(const std::shared_ptr<ChunkedArray>& chunked_array,
int byte_width)
 : Converter(chunked_array), byte_width_(byte_width) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
Rf_classgets(res, data::classes_arrow_fixed_size_binary);
Rf_setAttrib(res, symbols::byte_width, Rf_ScalarInteger(byte_width_));
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const FixedSizeBinaryArray* binary_array =
 checked_cast<const FixedSizeBinaryArray*>(array.get());

int byte_width = binary_array->byte_width();
auto ingest_one = [&, byte_width](R_xlen_t i) {
auto value = binary_array->GetValue(i);
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, byte_width));
std::copy(value, value + byte_width, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }

private:
int byte_width_;
};

boolDictionaryChunkArrayNeedUnification(
const std::shared_ptr<ChunkedArray>& chunked_array) {
int n = chunked_array->num_chunks();
if (n < 2) {
returnfalse;
 }
constauto& arr_first =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
for (int i = 1; i < n; i++) {
constauto& arr =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(i));
if (!(arr_first.dictionary()->Equals(arr.dictionary()))) {
returntrue;
 }
 }
returnfalse;
}

classConverter_Dictionary : publicConverter {
private:
bool need_unification_;
 std::unique_ptr<arrow::DictionaryUnifier> unifier_;
 std::vector<std::shared_ptr<Buffer>> arrays_transpose_;
 std::shared_ptr<DataType> out_type_;
 std::shared_ptr<Array> dictionary_;

public:
explicitConverter_Dictionary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array),
 need_unification_(DictionaryChunkArrayNeedUnification(chunked_array)) {
if (need_unification_) {
constauto& arr_type = checked_cast<const DictionaryType&>(*chunked_array->type());
 unifier_ = ValueOrStop(DictionaryUnifier::Make(arr_type.value_type()));

int n_arrays = chunked_array->num_chunks();
 arrays_transpose_.resize(n_arrays);

for (int i = 0; i < n_arrays; i++) {
constauto& dict_i =
 *checked_cast<const DictionaryArray&>(*chunked_array->chunk(i)).dictionary();
StopIfNotOk(unifier_->Unify(dict_i, &arrays_transpose_[i]));
 }

StopIfNotOk(unifier_->GetResult(&out_type_, &dictionary_));
 } else {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());

constauto& indices_type = *dict_type.index_type();
switch (indices_type.id()) {
case Type::UINT8:
case Type::INT8:
case Type::UINT16:
case Type::INT16:
case Type::INT32:
case Type::UINT32:
case Type::INT64:
case Type::UINT64:
break;
default:
cpp11::stop("Cannot convert Dictionary Array of type `%s` to R",
 dict_type.ToString().c_str());
 }

if (chunked_array->num_chunks() > 0) {
// DictionaryChunkArrayNeedUnification() returned false so we can safely assume
// the dictionary of the first chunk applies everywhere
constauto& dict_array =
 checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
 dictionary_ = dict_array.dictionary();
 } else {
 dictionary_ = CreateEmptyArray(dict_type.value_type());
 }
 }

// R factors store their codes in 32-bit integers, so dictionary arrays with
// more levels than that cannot be represented safely.
if (dictionary_->length() > std::numeric_limits<int>::max()) {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());
cpp11::stop(
"Cannot convert Dictionary Array of type `%s` to R: dictionary has "
"more levels than an R factor can represent",
 dict_type.ToString().c_str());
 }
 }

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::integers data(n);
 data.attr("levels") = GetLevels();
if (GetOrdered()) {
Rf_classgets(data, arrow::r::data::classes_ordered);
 } else {
Rf_classgets(data, arrow::r::data::classes_factor);
 }
return data;
 }

virtualboolParallel() const { returnfalse; }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const DictionaryArray& dict_array =
 checked_cast<const DictionaryArray&>(*array.get());
auto indices = dict_array.indices();
switch (indices->type_id()) {
case Type::UINT8:
return Ingest_some_nulls_Impl<arrow::UInt8Type>(data, array, start, n,
 chunk_index);
case Type::INT8:
return Ingest_some_nulls_Impl<arrow::Int8Type>(data, array, start, n,
 chunk_index);
case Type::UINT16:
return Ingest_some_nulls_Impl<arrow::UInt16Type>(data, array, start, n,
 chunk_index);
case Type::INT16:
return Ingest_some_nulls_Impl<arrow::Int16Type>(data, array, start, n,
 chunk_index);
case Type::INT32:
return Ingest_some_nulls_Impl<arrow::Int32Type>(data, array, start, n,
 chunk_index);
case Type::UINT32:
return Ingest_some_nulls_Impl<arrow::UInt32Type>(data, array, start, n,
 chunk_index);
case Type::INT64:
return Ingest_some_nulls_Impl<arrow::Int64Type>(data, array, start, n,
 chunk_index);
case Type::UINT64:
return Ingest_some_nulls_Impl<arrow::UInt64Type>(data, array, start, n,
 chunk_index);
default:
break;
 }
returnStatus::OK();
 }

private:
template <typename Type>
 Status Ingest_some_nulls_Impl(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
using index_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
auto indices = checked_cast<const DictionaryArray&>(*array).indices();
auto raw_indices = indices->data()->GetValues<index_type>(1);

auto p_data = INTEGER(data) + start;
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

// convert the 0-based indices from the arrow Array
// to 1-based indices used in R factors
if (need_unification_) {
// transpose the indices before converting
auto transposed =
reinterpret_cast<constint32_t*>(arrays_transpose_[chunk_index]->data());

auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = transposed[raw_indices[i]] + 1;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 } else {
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(raw_indices[i]) + 1;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
 }

boolGetOrdered() const {
return checked_cast<const DictionaryType&>(*chunked_array_->type()).ordered();
 }

SEXPGetLevels() const {
// R factor levels must be type "character" so coerce `dict` to STRSXP
// TODO (npr): this coercion should be optional, "dictionariesAsFactors" ;)
// Alternative: preserve the logical type of the dictionary values
// (e.g. if dict is timestamp, return a POSIXt R vector, not factor)
if (dictionary_->type_id() != Type::STRING &&
 dictionary_->type_id() != Type::LARGE_STRING) {
 cpp11::safe[Rf_warning]("Coercing dictionary values to R character factor levels");
 }

SEXP vec = PROTECT(Converter::Convert(dictionary_));
SEXP strings_vec = PROTECT(Rf_coerceVector(vec, STRSXP));
UNPROTECT(2);
return strings_vec;
 }
};

classConverter_Struct : publicConverter {
public:
explicitConverter_Struct(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array), converters() {
constauto& struct_type =
 checked_cast<const arrow::StructType&>(*chunked_array->type());

int nf = struct_type.num_fields();

 std::shared_ptr<arrow::Table> array_as_table =
ValueOrStop(arrow::Table::FromChunkedStructArray(chunked_array));
for (int i = 0; i < nf; i++) {
 converters.push_back(Converter::Make(array_as_table->column(i)));
 }
 }

SEXPAllocate(R_xlen_t n) const {
// allocate a data frame column to host each array
// If possible, a column is dealt with directly with altrep
auto type =
 checked_cast<const arrow::StructType*>(this->chunked_array_->type().get());
auto out =
arrow::r::to_r_list(converters, [n](const std::shared_ptr<Converter>& converter) {
SEXP out = converter->MaybeAltrep();
if (Rf_isNull(out)) {
 out = converter->Allocate(n);
 }
return out;
 });
auto colnames = arrow::r::to_r_strings(
 type->fields(),
 [](const std::shared_ptr<Field>& field) { return field->name(); });
 out.attr(symbols::row_names) = arrow::r::short_row_names(static_cast<int>(n));
 out.attr(R_NamesSymbol) = colnames;
 out.attr(R_ClassSymbol) = arrow::r::data::classes_tbl_df;

return out;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
int nf = static_cast<int>(converters.size());
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_all_nulls(data_i, start, n));
 }
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto struct_array = checked_cast<const arrow::StructArray*>(array.get());
int nf = static_cast<int>(converters.size());
// Flatten() deals with merging of nulls
auto arrays = ValueOrStop(struct_array->Flatten(gc_memory_pool()));
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_some_nulls(VECTOR_ELT(data, i), arrays[i],
 start, n, chunk_index));
 }
 }

returnStatus::OK();
 }

virtualboolParallel() const {
// this can only run in parallel if all the
// inner converters can
for (constauto& converter : converters) {
if (!converter->Parallel()) returnfalse;
 }
returntrue;
 }

private:
 std::vector<std::shared_ptr<Converter>> converters;
};

doublems_to_seconds(int64_t ms) { returnstatic_cast<double>(ms) / 1000; }

classConverter_Date64 : publicConverter {
public:
explicitConverter_Date64(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<int64_t>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i] / 1000);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename value_type, typename unit_type = TimeType>
classConverter_Time : publicConverter {
public:
explicitConverter_Time(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = cpp11::writable::strings({"hms", "difftime"});

// hms difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// hms difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type, typename unit_type = DurationType>
classConverter_Duration : publicConverter {
public:
explicitConverter_Duration(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = "difftime";

// difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type>
classConverter_Timestamp : publicConverter_Time<value_type, TimestampType> {
public:
explicitConverter_Timestamp(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter_Time<value_type, TimestampType>(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
auto array_type =
 checked_cast<const TimestampType*>(this->chunked_array_->type().get());
 std::string tzone = array_type->timezone();
if (tzone.size() > 0) {
 data.attr("tzone") = tzone;
 }
return data;
 }
};

template <typename Type>
classConverter_Decimal : publicConverter {
public:
explicitConverter_Decimal(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
// Licensed to the Apache Software Foundation (ASF) under one
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// or more contributor license agreements. See the NOTICE 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_types.h"

#include<arrow/array.h>
#include<arrow/builder.h>
#include<arrow/datum.h>
#include<arrow/table.h>
#include<arrow/util/bitmap_reader.h>
#include<arrow/util/bitmap_writer.h>
#include<arrow/util/int_util.h>

#include<type_traits>

#include"./extension.h"
#include"./r_task_group.h"

namespacearrow {

using internal::checked_cast;
using internal::IntegersCanFit;

namespacer {

classConverter {
public:
explicitConverter(const std::shared_ptr<ChunkedArray>& chunked_array)
 : chunked_array_(std::move(chunked_array)) {}

virtual~Converter() {}

// Allocate a vector of the right R type for this converter
virtualSEXPAllocate(R_xlen_t n) const = 0;

// data[ start:(start + n) ] = NA
virtual Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const = 0;

// ingest the values from the array into data[ start : (start + n)]
//
// chunk_index indicates which of the chunk is being ingested into data. This is
// ignored by most implementations and currently only used with Dictionary
// arrays.
virtual Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n,
size_t chunk_index) const = 0;

// can this run in parallel ?
virtualboolParallel() const { returntrue; }

// converter is passed as self to outlive the scope of Converter::Convert()
SEXPScheduleConvertTasks(RTasks& tasks, std::shared_ptr<Converter> self) {
// try altrep first
SEXP alt = altrep::MakeAltrepVector(chunked_array_);
if (!Rf_isNull(alt)) {
return alt;
 }

// otherwise use the Converter api:

// allocating the R vector upfront
SEXP out = PROTECT(Allocate(chunked_array_->length()));

// for each array, fill the relevant slice of `out`, potentially in parallel
 R_xlen_t k = 0, i = 0;
for (constauto& array : chunked_array_->chunks()) {
auto n_chunk = array->length();

 tasks.Append(Parallel(), [=] {
if (array->null_count() == n_chunk) {
return self->Ingest_all_nulls(out, k, n_chunk);
 } else {
return self->Ingest_some_nulls(out, array, k, n_chunk, i);
 }
 });

 k += n_chunk;
 i++;
 }

UNPROTECT(1);
return out;
 }

// Converter factory
static std::shared_ptr<Converter> Make(
const std::shared_ptr<ChunkedArray>& chunked_array);

staticSEXPLazyConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
 RTasks& tasks) {
auto converter = Make(chunked_array);
return converter->ScheduleConvertTasks(tasks, converter);
 }

staticSEXPConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
bool use_threads) {
 RTasks tasks(use_threads);
SEXP out = PROTECT(Converter::LazyConvert(chunked_array, tasks));
StopIfNotOk(tasks.Finish());

UNPROTECT(1);
return out;
 }

staticSEXPConvert(const std::shared_ptr<Array>& array) {
returnConvert(std::make_shared<ChunkedArray>(array), false);
 }

SEXPMaybeAltrep() { returnaltrep::MakeAltrepVector(chunked_array_); }

protected:
 std::shared_ptr<ChunkedArray> chunked_array_;
};

template <typename SetNonNull, typename SetNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null, SetNull&& set_null) {
if (array->null_count()) {
 internal::BitmapReader bitmap_reader(array->null_bitmap()->data(), array->offset(),
 n);

for (R_xlen_t i = 0; i < n; i++, bitmap_reader.Next()) {
if (bitmap_reader.IsSet()) {
RETURN_NOT_OK(set_non_null(i));
 } else {
RETURN_NOT_OK(set_null(i));
 }
 }

 } else {
for (R_xlen_t i = 0; i < n; i++) {
RETURN_NOT_OK(set_non_null(i));
 }
 }

returnStatus::OK();
}

template <typename SetNonNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null) {
auto nothing = [](R_xlen_t i) { returnStatus::OK(); };
returnIngestSome(array, n, std::forward<SetNonNull>(set_non_null), nothing);
}

std::shared_ptr<Array> CreateEmptyArray(const std::shared_ptr<DataType>& array_type) {
 std::unique_ptr<arrow::ArrayBuilder> builder;
StopIfNotOk(arrow::MakeBuilder(gc_memory_pool(), array_type, &builder));

 std::shared_ptr<arrow::Array> array;
StopIfNotOk(builder->Finish(&array));
return array;
}

template <typename Type>
classConverter_Int : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Int(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(INTSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = INTEGER(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename Type>
classConverter_Double : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Double(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<value_type>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

classConverter_Date32 : publicConverter {
public:
explicitConverter_Date32(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP data = PROTECT(Rf_allocVector(REALSXP, n));
Rf_classgets(data, Rf_mkString("Date"));
UNPROTECT(1);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<int>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename StringArrayType>
structConverter_String : publicConverter {
public:
explicitConverter_String(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(STRSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
for (R_xlen_t i = 0; i < n; i++) {
SET_STRING_ELT(data, i + start, NA_STRING);
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_offset = array->data()->GetValues<int32_t>(1);
if (!p_offset) {
returnStatus::Invalid("Invalid offset buffer");
 }
auto p_strings = array->data()->GetValues<char>(2, *p_offset);
if (!p_strings) {
// There is an offset buffer, but the data buffer is null
// There is at least one value in the array and not all the values are null
// That means all values are either empty strings or nulls so there is nothing to do

if (array->null_count()) {
 arrow::internal::BitmapReader null_reader(array->null_bitmap_data(),
 array->offset(), n);
for (int i = 0; i < n; i++, null_reader.Next()) {
if (null_reader.IsNotSet()) {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 }
returnStatus::OK();
 }

 StringArrayType* string_array = static_cast<StringArrayType*>(array.get());

constbool all_valid = array->null_count() == 0;
constbool strip_out_nuls = GetBoolOption("arrow.skip_nul", false);

bool nul_was_stripped = false;

if (all_valid) {
// no need to watch for missing strings
cpp11::unwind_protect([&] {
if (strip_out_nuls) {
for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 }
return;
 }

for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 }
 });
 } else {
cpp11::unwind_protect([&] {
 arrow::internal::BitmapReader validity_reader(array->null_bitmap_data(),
 array->offset(), n);

if (strip_out_nuls) {
for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
return;
 }

for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 });
 }

if (nul_was_stripped) {
 cpp11::safe[Rf_warning]("Stripping '\\0' (nul) from character vector");
 }

returnStatus::OK();
 }

boolParallel() const { returnfalse; }

private:
staticSEXPr_string_from_view(std::string_view view) {
returnRf_mkCharLenCE(view.data(), static_cast<int>(view.size()), CE_UTF8);
 }

staticSEXPr_string_from_view_strip_nul(std::string_view view,
bool* nul_was_stripped) {
constchar* old_string = view.data();

 std::string stripped_string;
size_t stripped_len = 0, nul_count = 0;

for (size_t i = 0; i < view.size(); i++) {
if (old_string[i] == '\0') {
 ++nul_count;

if (nul_count == 1) {
// first nul spotted: allocate stripped string storage
 stripped_string = std::string(view);
 stripped_len = i;
 }

// don't copy old_string[i] (which is \0) into stripped_string
continue;
 }

if (nul_count > 0) {
 stripped_string[stripped_len++] = old_string[i];
 }
 }

if (nul_count > 0) {
 *nul_was_stripped = true;
 stripped_string.resize(stripped_len);
returnr_string_from_view(stripped_string);
 }

returnr_string_from_view(view);
 }
};

classConverter_Boolean : publicConverter {
public:
explicitConverter_Boolean(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(LGLSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(LOGICAL(data) + start, n, NA_LOGICAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = LOGICAL(data) + start;
auto p_bools = array->data()->GetValues<uint8_t>(1, 0);
if (!p_bools) {
returnStatus::Invalid("Invalid data buffer");
 }

 arrow::internal::BitmapReader data_reader(p_bools, array->offset(), n);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = data_reader.IsSet();
 data_reader.Next();
returnStatus::OK();
 };

auto null_one = [&](R_xlen_t i) {
 data_reader.Next();
 p_data[i] = NA_LOGICAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename ArrayType>
classConverter_Binary : publicConverter {
public:
using offset_type = typename ArrayType::offset_type;
explicitConverter_Binary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
if (std::is_same<ArrayType, BinaryArray>::value) {
Rf_classgets(res, data::classes_arrow_binary);
 } else {
Rf_classgets(res, data::classes_arrow_large_binary);
 }
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const ArrayType* binary_array = checked_cast<const ArrayType*>(array.get());

auto ingest_one = [&](R_xlen_t i) {
 offset_type ni;
auto value = binary_array->GetValue(i, &ni);
if (ni > R_XLEN_T_MAX) {
returnStatus::RError("Array too big to be represented as a raw vector");
 }
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, ni));
std::copy(value, value + ni, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }
};

classConverter_FixedSizeBinary : publicConverter {
public:
explicitConverter_FixedSizeBinary(const std::shared_ptr<ChunkedArray>& chunked_array,
int byte_width)
 : Converter(chunked_array), byte_width_(byte_width) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
Rf_classgets(res, data::classes_arrow_fixed_size_binary);
Rf_setAttrib(res, symbols::byte_width, Rf_ScalarInteger(byte_width_));
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const FixedSizeBinaryArray* binary_array =
 checked_cast<const FixedSizeBinaryArray*>(array.get());

int byte_width = binary_array->byte_width();
auto ingest_one = [&, byte_width](R_xlen_t i) {
auto value = binary_array->GetValue(i);
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, byte_width));
std::copy(value, value + byte_width, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }

private:
int byte_width_;
};

boolDictionaryChunkArrayNeedUnification(
const std::shared_ptr<ChunkedArray>& chunked_array) {
int n = chunked_array->num_chunks();
if (n < 2) {
returnfalse;
 }
constauto& arr_first =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
for (int i = 1; i < n; i++) {
constauto& arr =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(i));
if (!(arr_first.dictionary()->Equals(arr.dictionary()))) {
returntrue;
 }
 }
returnfalse;
}

classConverter_Dictionary : publicConverter {
private:
bool need_unification_;
 std::unique_ptr<arrow::DictionaryUnifier> unifier_;
 std::vector<std::shared_ptr<Buffer>> arrays_transpose_;
 std::shared_ptr<DataType> out_type_;
 std::shared_ptr<Array> dictionary_;

public:
explicitConverter_Dictionary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array),
 need_unification_(DictionaryChunkArrayNeedUnification(chunked_array)) {
if (need_unification_) {
constauto& arr_type = checked_cast<const DictionaryType&>(*chunked_array->type());
 unifier_ = ValueOrStop(DictionaryUnifier::Make(arr_type.value_type()));

int n_arrays = chunked_array->num_chunks();
 arrays_transpose_.resize(n_arrays);

for (int i = 0; i < n_arrays; i++) {
constauto& dict_i =
 *checked_cast<const DictionaryArray&>(*chunked_array->chunk(i)).dictionary();
StopIfNotOk(unifier_->Unify(dict_i, &arrays_transpose_[i]));
 }

StopIfNotOk(unifier_->GetResult(&out_type_, &dictionary_));
 } else {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());

constauto& indices_type = *dict_type.index_type();
switch (indices_type.id()) {
case Type::UINT8:
case Type::INT8:
case Type::UINT16:
case Type::INT16:
case Type::INT32:
case Type::UINT32:
case Type::INT64:
case Type::UINT64:
break;
default:
cpp11::stop("Cannot convert Dictionary Array of type `%s` to R",
 dict_type.ToString().c_str());
 }

if (chunked_array->num_chunks() > 0) {
// DictionaryChunkArrayNeedUnification() returned false so we can safely assume
// the dictionary of the first chunk applies everywhere
constauto& dict_array =
 checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
 dictionary_ = dict_array.dictionary();
 } else {
 dictionary_ = CreateEmptyArray(dict_type.value_type());
 }
 }

// R factors store their codes in 32-bit integers, so dictionary arrays with
// more levels than that cannot be represented safely.
if (dictionary_->length() > std::numeric_limits<int>::max()) {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());
cpp11::stop(
"Cannot convert Dictionary Array of type `%s` to R: dictionary has "
"more levels than an R factor can represent",
 dict_type.ToString().c_str());
 }
 }

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::integers data(n);
 data.attr("levels") = GetLevels();
if (GetOrdered()) {
Rf_classgets(data, arrow::r::data::classes_ordered);
 } else {
Rf_classgets(data, arrow::r::data::classes_factor);
 }
return data;
 }

virtualboolParallel() const { returnfalse; }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const DictionaryArray& dict_array =
 checked_cast<const DictionaryArray&>(*array.get());
auto indices = dict_array.indices();
switch (indices->type_id()) {
case Type::UINT8:
return Ingest_some_nulls_Impl<arrow::UInt8Type>(data, array, start, n,
 chunk_index);
case Type::INT8:
return Ingest_some_nulls_Impl<arrow::Int8Type>(data, array, start, n,
 chunk_index);
case Type::UINT16:
return Ingest_some_nulls_Impl<arrow::UInt16Type>(data, array, start, n,
 chunk_index);
case Type::INT16:
return Ingest_some_nulls_Impl<arrow::Int16Type>(data, array, start, n,
 chunk_index);
case Type::INT32:
return Ingest_some_nulls_Impl<arrow::Int32Type>(data, array, start, n,
 chunk_index);
case Type::UINT32:
return Ingest_some_nulls_Impl<arrow::UInt32Type>(data, array, start, n,
 chunk_index);
case Type::INT64:
return Ingest_some_nulls_Impl<arrow::Int64Type>(data, array, start, n,
 chunk_index);
case Type::UINT64:
return Ingest_some_nulls_Impl<arrow::UInt64Type>(data, array, start, n,
 chunk_index);
default:
break;
 }
returnStatus::OK();
 }

private:
template <typename Type>
 Status Ingest_some_nulls_Impl(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
using index_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
auto indices = checked_cast<const DictionaryArray&>(*array).indices();
auto raw_indices = indices->data()->GetValues<index_type>(1);

auto p_data = INTEGER(data) + start;
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

// convert the 0-based indices from the arrow Array
// to 1-based indices used in R factors
if (need_unification_) {
// transpose the indices before converting
auto transposed =
reinterpret_cast<constint32_t*>(arrays_transpose_[chunk_index]->data());

auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = transposed[raw_indices[i]] + 1;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 } else {
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(raw_indices[i]) + 1;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
 }

boolGetOrdered() const {
return checked_cast<const DictionaryType&>(*chunked_array_->type()).ordered();
 }

SEXPGetLevels() const {
// R factor levels must be type "character" so coerce `dict` to STRSXP
// TODO (npr): this coercion should be optional, "dictionariesAsFactors" ;)
// Alternative: preserve the logical type of the dictionary values
// (e.g. if dict is timestamp, return a POSIXt R vector, not factor)
if (dictionary_->type_id() != Type::STRING &&
 dictionary_->type_id() != Type::LARGE_STRING) {
 cpp11::safe[Rf_warning]("Coercing dictionary values to R character factor levels");
 }

SEXP vec = PROTECT(Converter::Convert(dictionary_));
SEXP strings_vec = PROTECT(Rf_coerceVector(vec, STRSXP));
UNPROTECT(2);
return strings_vec;
 }
};

classConverter_Struct : publicConverter {
public:
explicitConverter_Struct(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array), converters() {
constauto& struct_type =
 checked_cast<const arrow::StructType&>(*chunked_array->type());

int nf = struct_type.num_fields();

 std::shared_ptr<arrow::Table> array_as_table =
ValueOrStop(arrow::Table::FromChunkedStructArray(chunked_array));
for (int i = 0; i < nf; i++) {
 converters.push_back(Converter::Make(array_as_table->column(i)));
 }
 }

SEXPAllocate(R_xlen_t n) const {
// allocate a data frame column to host each array
// If possible, a column is dealt with directly with altrep
auto type =
 checked_cast<const arrow::StructType*>(this->chunked_array_->type().get());
auto out =
arrow::r::to_r_list(converters, [n](const std::shared_ptr<Converter>& converter) {
SEXP out = converter->MaybeAltrep();
if (Rf_isNull(out)) {
 out = converter->Allocate(n);
 }
return out;
 });
auto colnames = arrow::r::to_r_strings(
 type->fields(),
 [](const std::shared_ptr<Field>& field) { return field->name(); });
 out.attr(symbols::row_names) = arrow::r::short_row_names(static_cast<int>(n));
 out.attr(R_NamesSymbol) = colnames;
 out.attr(R_ClassSymbol) = arrow::r::data::classes_tbl_df;

return out;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
int nf = static_cast<int>(converters.size());
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_all_nulls(data_i, start, n));
 }
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto struct_array = checked_cast<const arrow::StructArray*>(array.get());
int nf = static_cast<int>(converters.size());
// Flatten() deals with merging of nulls
auto arrays = ValueOrStop(struct_array->Flatten(gc_memory_pool()));
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_some_nulls(VECTOR_ELT(data, i), arrays[i],
 start, n, chunk_index));
 }
 }

returnStatus::OK();
 }

virtualboolParallel() const {
// this can only run in parallel if all the
// inner converters can
for (constauto& converter : converters) {
if (!converter->Parallel()) returnfalse;
 }
returntrue;
 }

private:
 std::vector<std::shared_ptr<Converter>> converters;
};

doublems_to_seconds(int64_t ms) { returnstatic_cast<double>(ms) / 1000; }

classConverter_Date64 : publicConverter {
public:
explicitConverter_Date64(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<int64_t>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i] / 1000);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename value_type, typename unit_type = TimeType>
classConverter_Time : publicConverter {
public:
explicitConverter_Time(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = cpp11::writable::strings({"hms", "difftime"});

// hms difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// hms difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type, typename unit_type = DurationType>
classConverter_Duration : publicConverter {
public:
explicitConverter_Duration(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = "difftime";

// difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type>
classConverter_Timestamp : publicConverter_Time<value_type, TimestampType> {
public:
explicitConverter_Timestamp(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter_Time<value_type, TimestampType>(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
auto array_type =
 checked_cast<const TimestampType*>(this->chunked_array_->type().get());
 std::string tzone = array_type->timezone();
if (tzone.size() > 0) {
 data.attr("tzone") = tzone;
 }
return data;
 }
};

template <typename Type>
classConverter_Decimal : publicConverter {
public:
explicitConverter_Decimal(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
// Licensed to the Apache Software Foundation (ASF) under one
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// or more contributor license agreements. See the NOTICE 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_types.h"

#include<arrow/array.h>
#include<arrow/builder.h>
#include<arrow/datum.h>
#include<arrow/table.h>
#include<arrow/util/bitmap_reader.h>
#include<arrow/util/bitmap_writer.h>
#include<arrow/util/int_util.h>

#include<type_traits>

#include"./extension.h"
#include"./r_task_group.h"

namespacearrow {

using internal::checked_cast;
using internal::IntegersCanFit;

namespacer {

classConverter {
public:
explicitConverter(const std::shared_ptr<ChunkedArray>& chunked_array)
 : chunked_array_(std::move(chunked_array)) {}

virtual~Converter() {}

// Allocate a vector of the right R type for this converter
virtualSEXPAllocate(R_xlen_t n) const = 0;

// data[ start:(start + n) ] = NA
virtual Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const = 0;

// ingest the values from the array into data[ start : (start + n)]
//
// chunk_index indicates which of the chunk is being ingested into data. This is
// ignored by most implementations and currently only used with Dictionary
// arrays.
virtual Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n,
size_t chunk_index) const = 0;

// can this run in parallel ?
virtualboolParallel() const { returntrue; }

// converter is passed as self to outlive the scope of Converter::Convert()
SEXPScheduleConvertTasks(RTasks& tasks, std::shared_ptr<Converter> self) {
// try altrep first
SEXP alt = altrep::MakeAltrepVector(chunked_array_);
if (!Rf_isNull(alt)) {
return alt;
 }

// otherwise use the Converter api:

// allocating the R vector upfront
SEXP out = PROTECT(Allocate(chunked_array_->length()));

// for each array, fill the relevant slice of `out`, potentially in parallel
 R_xlen_t k = 0, i = 0;
for (constauto& array : chunked_array_->chunks()) {
auto n_chunk = array->length();

 tasks.Append(Parallel(), [=] {
if (array->null_count() == n_chunk) {
return self->Ingest_all_nulls(out, k, n_chunk);
 } else {
return self->Ingest_some_nulls(out, array, k, n_chunk, i);
 }
 });

 k += n_chunk;
 i++;
 }

UNPROTECT(1);
return out;
 }

// Converter factory
static std::shared_ptr<Converter> Make(
const std::shared_ptr<ChunkedArray>& chunked_array);

staticSEXPLazyConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
 RTasks& tasks) {
auto converter = Make(chunked_array);
return converter->ScheduleConvertTasks(tasks, converter);
 }

staticSEXPConvert(const std::shared_ptr<ChunkedArray>& chunked_array,
bool use_threads) {
 RTasks tasks(use_threads);
SEXP out = PROTECT(Converter::LazyConvert(chunked_array, tasks));
StopIfNotOk(tasks.Finish());

UNPROTECT(1);
return out;
 }

staticSEXPConvert(const std::shared_ptr<Array>& array) {
returnConvert(std::make_shared<ChunkedArray>(array), false);
 }

SEXPMaybeAltrep() { returnaltrep::MakeAltrepVector(chunked_array_); }

protected:
 std::shared_ptr<ChunkedArray> chunked_array_;
};

template <typename SetNonNull, typename SetNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null, SetNull&& set_null) {
if (array->null_count()) {
 internal::BitmapReader bitmap_reader(array->null_bitmap()->data(), array->offset(),
 n);

for (R_xlen_t i = 0; i < n; i++, bitmap_reader.Next()) {
if (bitmap_reader.IsSet()) {
RETURN_NOT_OK(set_non_null(i));
 } else {
RETURN_NOT_OK(set_null(i));
 }
 }

 } else {
for (R_xlen_t i = 0; i < n; i++) {
RETURN_NOT_OK(set_non_null(i));
 }
 }

returnStatus::OK();
}

template <typename SetNonNull>
Status IngestSome(const std::shared_ptr<arrow::Array>& array, R_xlen_t n,
 SetNonNull&& set_non_null) {
auto nothing = [](R_xlen_t i) { returnStatus::OK(); };
returnIngestSome(array, n, std::forward<SetNonNull>(set_non_null), nothing);
}

std::shared_ptr<Array> CreateEmptyArray(const std::shared_ptr<DataType>& array_type) {
 std::unique_ptr<arrow::ArrayBuilder> builder;
StopIfNotOk(arrow::MakeBuilder(gc_memory_pool(), array_type, &builder));

 std::shared_ptr<arrow::Array> array;
StopIfNotOk(builder->Finish(&array));
return array;
}

template <typename Type>
classConverter_Int : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Int(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(INTSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = INTEGER(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename Type>
classConverter_Double : publicConverter {
using value_type = typename TypeTraits<Type>::ArrayType::value_type;

public:
explicitConverter_Double(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<value_type>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<value_type>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

classConverter_Date32 : publicConverter {
public:
explicitConverter_Date32(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP data = PROTECT(Rf_allocVector(REALSXP, n));
Rf_classgets(data, Rf_mkString("Date"));
UNPROTECT(1);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_values = array->data()->GetValues<int>(1);
if (!p_values) {
returnStatus::Invalid("Invalid data buffer");
 }
auto p_data = REAL(data) + start;
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename StringArrayType>
structConverter_String : publicConverter {
public:
explicitConverter_String(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(STRSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
for (R_xlen_t i = 0; i < n; i++) {
SET_STRING_ELT(data, i + start, NA_STRING);
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_offset = array->data()->GetValues<int32_t>(1);
if (!p_offset) {
returnStatus::Invalid("Invalid offset buffer");
 }
auto p_strings = array->data()->GetValues<char>(2, *p_offset);
if (!p_strings) {
// There is an offset buffer, but the data buffer is null
// There is at least one value in the array and not all the values are null
// That means all values are either empty strings or nulls so there is nothing to do

if (array->null_count()) {
 arrow::internal::BitmapReader null_reader(array->null_bitmap_data(),
 array->offset(), n);
for (int i = 0; i < n; i++, null_reader.Next()) {
if (null_reader.IsNotSet()) {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 }
returnStatus::OK();
 }

 StringArrayType* string_array = static_cast<StringArrayType*>(array.get());

constbool all_valid = array->null_count() == 0;
constbool strip_out_nuls = GetBoolOption("arrow.skip_nul", false);

bool nul_was_stripped = false;

if (all_valid) {
// no need to watch for missing strings
cpp11::unwind_protect([&] {
if (strip_out_nuls) {
for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 }
return;
 }

for (int i = 0; i < n; i++) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 }
 });
 } else {
cpp11::unwind_protect([&] {
 arrow::internal::BitmapReader validity_reader(array->null_bitmap_data(),
 array->offset(), n);

if (strip_out_nuls) {
for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i,
r_string_from_view_strip_nul(string_array->GetView(i),
 &nul_was_stripped));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
return;
 }

for (int i = 0; i < n; i++, validity_reader.Next()) {
if (validity_reader.IsSet()) {
SET_STRING_ELT(data, start + i, r_string_from_view(string_array->GetView(i)));
 } else {
SET_STRING_ELT(data, start + i, NA_STRING);
 }
 }
 });
 }

if (nul_was_stripped) {
 cpp11::safe[Rf_warning]("Stripping '\\0' (nul) from character vector");
 }

returnStatus::OK();
 }

boolParallel() const { returnfalse; }

private:
staticSEXPr_string_from_view(std::string_view view) {
returnRf_mkCharLenCE(view.data(), static_cast<int>(view.size()), CE_UTF8);
 }

staticSEXPr_string_from_view_strip_nul(std::string_view view,
bool* nul_was_stripped) {
constchar* old_string = view.data();

 std::string stripped_string;
size_t stripped_len = 0, nul_count = 0;

for (size_t i = 0; i < view.size(); i++) {
if (old_string[i] == '\0') {
 ++nul_count;

if (nul_count == 1) {
// first nul spotted: allocate stripped string storage
 stripped_string = std::string(view);
 stripped_len = i;
 }

// don't copy old_string[i] (which is \0) into stripped_string
continue;
 }

if (nul_count > 0) {
 stripped_string[stripped_len++] = old_string[i];
 }
 }

if (nul_count > 0) {
 *nul_was_stripped = true;
 stripped_string.resize(stripped_len);
returnr_string_from_view(stripped_string);
 }

returnr_string_from_view(view);
 }
};

classConverter_Boolean : publicConverter {
public:
explicitConverter_Boolean(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(LGLSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(LOGICAL(data) + start, n, NA_LOGICAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = LOGICAL(data) + start;
auto p_bools = array->data()->GetValues<uint8_t>(1, 0);
if (!p_bools) {
returnStatus::Invalid("Invalid data buffer");
 }

 arrow::internal::BitmapReader data_reader(p_bools, array->offset(), n);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = data_reader.IsSet();
 data_reader.Next();
returnStatus::OK();
 };

auto null_one = [&](R_xlen_t i) {
 data_reader.Next();
 p_data[i] = NA_LOGICAL;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename ArrayType>
classConverter_Binary : publicConverter {
public:
using offset_type = typename ArrayType::offset_type;
explicitConverter_Binary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
if (std::is_same<ArrayType, BinaryArray>::value) {
Rf_classgets(res, data::classes_arrow_binary);
 } else {
Rf_classgets(res, data::classes_arrow_large_binary);
 }
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const ArrayType* binary_array = checked_cast<const ArrayType*>(array.get());

auto ingest_one = [&](R_xlen_t i) {
 offset_type ni;
auto value = binary_array->GetValue(i, &ni);
if (ni > R_XLEN_T_MAX) {
returnStatus::RError("Array too big to be represented as a raw vector");
 }
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, ni));
std::copy(value, value + ni, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }
};

classConverter_FixedSizeBinary : publicConverter {
public:
explicitConverter_FixedSizeBinary(const std::shared_ptr<ChunkedArray>& chunked_array,
int byte_width)
 : Converter(chunked_array), byte_width_(byte_width) {}

SEXPAllocate(R_xlen_t n) const {
SEXP res = PROTECT(Rf_allocVector(VECSXP, n));
Rf_classgets(res, data::classes_arrow_fixed_size_binary);
Rf_setAttrib(res, symbols::byte_width, Rf_ScalarInteger(byte_width_));
UNPROTECT(1);
return res;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const FixedSizeBinaryArray* binary_array =
 checked_cast<const FixedSizeBinaryArray*>(array.get());

int byte_width = binary_array->byte_width();
auto ingest_one = [&, byte_width](R_xlen_t i) {
auto value = binary_array->GetValue(i);
SEXP raw = PROTECT(Rf_allocVector(RAWSXP, byte_width));
std::copy(value, value + byte_width, RAW(raw));

SET_VECTOR_ELT(data, i + start, raw);
UNPROTECT(1);

returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one);
 }

virtualboolParallel() const { returnfalse; }

private:
int byte_width_;
};

boolDictionaryChunkArrayNeedUnification(
const std::shared_ptr<ChunkedArray>& chunked_array) {
int n = chunked_array->num_chunks();
if (n < 2) {
returnfalse;
 }
constauto& arr_first =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
for (int i = 1; i < n; i++) {
constauto& arr =
 internal::checked_cast<const DictionaryArray&>(*chunked_array->chunk(i));
if (!(arr_first.dictionary()->Equals(arr.dictionary()))) {
returntrue;
 }
 }
returnfalse;
}

classConverter_Dictionary : publicConverter {
private:
bool need_unification_;
 std::unique_ptr<arrow::DictionaryUnifier> unifier_;
 std::vector<std::shared_ptr<Buffer>> arrays_transpose_;
 std::shared_ptr<DataType> out_type_;
 std::shared_ptr<Array> dictionary_;

public:
explicitConverter_Dictionary(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array),
 need_unification_(DictionaryChunkArrayNeedUnification(chunked_array)) {
if (need_unification_) {
constauto& arr_type = checked_cast<const DictionaryType&>(*chunked_array->type());
 unifier_ = ValueOrStop(DictionaryUnifier::Make(arr_type.value_type()));

int n_arrays = chunked_array->num_chunks();
 arrays_transpose_.resize(n_arrays);

for (int i = 0; i < n_arrays; i++) {
constauto& dict_i =
 *checked_cast<const DictionaryArray&>(*chunked_array->chunk(i)).dictionary();
StopIfNotOk(unifier_->Unify(dict_i, &arrays_transpose_[i]));
 }

StopIfNotOk(unifier_->GetResult(&out_type_, &dictionary_));
 } else {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());

constauto& indices_type = *dict_type.index_type();
switch (indices_type.id()) {
case Type::UINT8:
case Type::INT8:
case Type::UINT16:
case Type::INT16:
case Type::INT32:
case Type::UINT32:
case Type::INT64:
case Type::UINT64:
break;
default:
cpp11::stop("Cannot convert Dictionary Array of type `%s` to R",
 dict_type.ToString().c_str());
 }

if (chunked_array->num_chunks() > 0) {
// DictionaryChunkArrayNeedUnification() returned false so we can safely assume
// the dictionary of the first chunk applies everywhere
constauto& dict_array =
 checked_cast<const DictionaryArray&>(*chunked_array->chunk(0));
 dictionary_ = dict_array.dictionary();
 } else {
 dictionary_ = CreateEmptyArray(dict_type.value_type());
 }
 }

// R factors store their codes in 32-bit integers, so dictionary arrays with
// more levels than that cannot be represented safely.
if (dictionary_->length() > std::numeric_limits<int>::max()) {
constauto& dict_type = checked_cast<const DictionaryType&>(*chunked_array->type());
cpp11::stop(
"Cannot convert Dictionary Array of type `%s` to R: dictionary has "
"more levels than an R factor can represent",
 dict_type.ToString().c_str());
 }
 }

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::integers data(n);
 data.attr("levels") = GetLevels();
if (GetOrdered()) {
Rf_classgets(data, arrow::r::data::classes_ordered);
 } else {
Rf_classgets(data, arrow::r::data::classes_factor);
 }
return data;
 }

virtualboolParallel() const { returnfalse; }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(INTEGER(data) + start, n, NA_INTEGER);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
const DictionaryArray& dict_array =
 checked_cast<const DictionaryArray&>(*array.get());
auto indices = dict_array.indices();
switch (indices->type_id()) {
case Type::UINT8:
return Ingest_some_nulls_Impl<arrow::UInt8Type>(data, array, start, n,
 chunk_index);
case Type::INT8:
return Ingest_some_nulls_Impl<arrow::Int8Type>(data, array, start, n,
 chunk_index);
case Type::UINT16:
return Ingest_some_nulls_Impl<arrow::UInt16Type>(data, array, start, n,
 chunk_index);
case Type::INT16:
return Ingest_some_nulls_Impl<arrow::Int16Type>(data, array, start, n,
 chunk_index);
case Type::INT32:
return Ingest_some_nulls_Impl<arrow::Int32Type>(data, array, start, n,
 chunk_index);
case Type::UINT32:
return Ingest_some_nulls_Impl<arrow::UInt32Type>(data, array, start, n,
 chunk_index);
case Type::INT64:
return Ingest_some_nulls_Impl<arrow::Int64Type>(data, array, start, n,
 chunk_index);
case Type::UINT64:
return Ingest_some_nulls_Impl<arrow::UInt64Type>(data, array, start, n,
 chunk_index);
default:
break;
 }
returnStatus::OK();
 }

private:
template <typename Type>
 Status Ingest_some_nulls_Impl(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
using index_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
auto indices = checked_cast<const DictionaryArray&>(*array).indices();
auto raw_indices = indices->data()->GetValues<index_type>(1);

auto p_data = INTEGER(data) + start;
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_INTEGER;
returnStatus::OK();
 };

// convert the 0-based indices from the arrow Array
// to 1-based indices used in R factors
if (need_unification_) {
// transpose the indices before converting
auto transposed =
reinterpret_cast<constint32_t*>(arrays_transpose_[chunk_index]->data());

auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = transposed[raw_indices[i]] + 1;
returnStatus::OK();
 };

returnIngestSome(array, n, ingest_one, null_one);
 } else {
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<int>(raw_indices[i]) + 1;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
 }

boolGetOrdered() const {
return checked_cast<const DictionaryType&>(*chunked_array_->type()).ordered();
 }

SEXPGetLevels() const {
// R factor levels must be type "character" so coerce `dict` to STRSXP
// TODO (npr): this coercion should be optional, "dictionariesAsFactors" ;)
// Alternative: preserve the logical type of the dictionary values
// (e.g. if dict is timestamp, return a POSIXt R vector, not factor)
if (dictionary_->type_id() != Type::STRING &&
 dictionary_->type_id() != Type::LARGE_STRING) {
 cpp11::safe[Rf_warning]("Coercing dictionary values to R character factor levels");
 }

SEXP vec = PROTECT(Converter::Convert(dictionary_));
SEXP strings_vec = PROTECT(Rf_coerceVector(vec, STRSXP));
UNPROTECT(2);
return strings_vec;
 }
};

classConverter_Struct : publicConverter {
public:
explicitConverter_Struct(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array), converters() {
constauto& struct_type =
 checked_cast<const arrow::StructType&>(*chunked_array->type());

int nf = struct_type.num_fields();

 std::shared_ptr<arrow::Table> array_as_table =
ValueOrStop(arrow::Table::FromChunkedStructArray(chunked_array));
for (int i = 0; i < nf; i++) {
 converters.push_back(Converter::Make(array_as_table->column(i)));
 }
 }

SEXPAllocate(R_xlen_t n) const {
// allocate a data frame column to host each array
// If possible, a column is dealt with directly with altrep
auto type =
 checked_cast<const arrow::StructType*>(this->chunked_array_->type().get());
auto out =
arrow::r::to_r_list(converters, [n](const std::shared_ptr<Converter>& converter) {
SEXP out = converter->MaybeAltrep();
if (Rf_isNull(out)) {
 out = converter->Allocate(n);
 }
return out;
 });
auto colnames = arrow::r::to_r_strings(
 type->fields(),
 [](const std::shared_ptr<Field>& field) { return field->name(); });
 out.attr(symbols::row_names) = arrow::r::short_row_names(static_cast<int>(n));
 out.attr(R_NamesSymbol) = colnames;
 out.attr(R_ClassSymbol) = arrow::r::data::classes_tbl_df;

return out;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
int nf = static_cast<int>(converters.size());
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_all_nulls(data_i, start, n));
 }
 }
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto struct_array = checked_cast<const arrow::StructArray*>(array.get());
int nf = static_cast<int>(converters.size());
// Flatten() deals with merging of nulls
auto arrays = ValueOrStop(struct_array->Flatten(gc_memory_pool()));
for (int i = 0; i < nf; i++) {
SEXP data_i = VECTOR_ELT(data, i);

// only ingest if the column is not altrep
if (!altrep::is_unmaterialized_arrow_altrep(data_i)) {
StopIfNotOk(converters[i]->Ingest_some_nulls(VECTOR_ELT(data, i), arrays[i],
 start, n, chunk_index));
 }
 }

returnStatus::OK();
 }

virtualboolParallel() const {
// this can only run in parallel if all the
// inner converters can
for (constauto& converter : converters) {
if (!converter->Parallel()) returnfalse;
 }
returntrue;
 }

private:
 std::vector<std::shared_ptr<Converter>> converters;
};

doublems_to_seconds(int64_t ms) { returnstatic_cast<double>(ms) / 1000; }

classConverter_Date64 : publicConverter {
public:
explicitConverter_Date64(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<int64_t>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i] / 1000);
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }
};

template <typename value_type, typename unit_type = TimeType>
classConverter_Time : publicConverter {
public:
explicitConverter_Time(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = cpp11::writable::strings({"hms", "difftime"});

// hms difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// hms difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type, typename unit_type = DurationType>
classConverter_Duration : publicConverter {
public:
explicitConverter_Duration(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
 data.attr("class") = "difftime";

// difftime is always stored as "seconds"
 data.attr("units") = cpp11::writable::strings({"secs"});
return data;
 }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,
 R_xlen_t start, R_xlen_t n, size_t chunk_index) const {
int multiplier = TimeUnit_multiplier(array);

auto p_data = REAL(data) + start;
auto p_values = array->data()->GetValues<value_type>(1);
auto ingest_one = [&](R_xlen_t i) {
 p_data[i] = static_cast<double>(p_values[i]) / multiplier;
returnStatus::OK();
 };
auto null_one = [&](R_xlen_t i) {
 p_data[i] = NA_REAL;
returnStatus::OK();
 };
returnIngestSome(array, n, ingest_one, null_one);
 }

private:
intTimeUnit_multiplier(const std::shared_ptr<Array>& array) const {
// difftime is always "seconds", so multiply based on the Array's TimeUnit
switch (static_cast<unit_type*>(array->type().get())->unit()) {
case TimeUnit::SECOND:
return1;
case TimeUnit::MILLI:
return1000;
case TimeUnit::MICRO:
return1000000;
case TimeUnit::NANO:
return1000000000;
default:
return0;
 }
 }
};

template <typename value_type>
classConverter_Timestamp : publicConverter_Time<value_type, TimestampType> {
public:
explicitConverter_Timestamp(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter_Time<value_type, TimestampType>(chunked_array) {}

SEXPAllocate(R_xlen_t n) const {
 cpp11::writable::doubles data(n);
Rf_classgets(data, arrow::r::data::classes_POSIXct);
auto array_type =
 checked_cast<const TimestampType*>(this->chunked_array_->type().get());
 std::string tzone = array_type->timezone();
if (tzone.size() > 0) {
 data.attr("tzone") = tzone;
 }
return data;
 }
};

template <typename Type>
classConverter_Decimal : publicConverter {
public:
explicitConverter_Decimal(const std::shared_ptr<ChunkedArray>& chunked_array)
 : Converter(chunked_array) {}

SEXPAllocate(R_xlen_t n) const { returnRf_allocVector(REALSXP, n); }

 Status Ingest_all_nulls(SEXP data, R_xlen_t start, R_xlen_t n) const {
std::fill_n(REAL(data) + start, n, NA_REAL);
returnStatus::OK();
 }

 Status Ingest_some_nulls(SEXP data, const std::shared_ptr<arrow::Array>& array,