Function vector_from_r_memory_impl

r/src/r_to_arrow.cpp:1357–1410

//       leverage concurrency. Maybe some refactoring needed.
template <typename RVector, typename Type>
bool vector_from_r_memory_impl(SEXP x, const std::shared_ptr<DataType>& type,
                               std::vector<std::shared_ptr<arrow::ChunkedArray>>& columns,
                               int j, RTasks& tasks) {
  RVector vec(x);
  using value_type = typename arrow::TypeTraits<Type>::ArrayType::value_type;
  auto buffer = std::make_shared<RBuffer<RVector>>(vec);

  tasks.Append(true, [buffer, x, &columns, j]() {
    std::vector<std::shared_ptr<Buffer>> buffers{nullptr, buffer};

    auto n = XLENGTH(x);
    auto p_x_start = reinterpret_cast<const value_type*>(DATAPTR_RO(x));
    auto p_x_end = p_x_start + n;

    int null_count = 0;
    auto first_na = std::find_if(p_x_start, p_x_end, is_NA<value_type>);

    if (first_na < p_x_end) {
      auto null_bitmap =
          ValueOrStop(AllocateBuffer(bit_util::BytesForBits(n), gc_memory_pool()));
      internal::FirstTimeBitmapWriter bitmap_writer(null_bitmap->mutable_data(), 0, n);

      // first loop to clear all the bits before the first NA
      auto k = std::distance(p_x_start, first_na);
      int i = 0;
      for (; i < k; i++, bitmap_writer.Next()) {
        bitmap_writer.Set();
      }

      auto p_vec = first_na;
      // then finish
      for (; i < n; i++, bitmap_writer.Next(), ++p_vec) {
        if (is_NA<value_type>(*p_vec)) {
          bitmap_writer.Clear();
          null_count++;
        } else {
          bitmap_writer.Set();
        }
      }

      bitmap_writer.Finish();
      buffers[0] = std::move(null_bitmap);
    }

    auto data = ArrayData::Make(std::make_shared<Type>(), n, std::move(buffers),
                                null_count, 0 /*offset*/);
    auto array = std::make_shared<typename TypeTraits<Type>::ArrayType>(data);
    columns[j] = std::make_shared<arrow::ChunkedArray>(array);

    return Status::OK();
  });

  return true;
}

bool vector_from_r_memory(SEXP x, const std::shared_ptr<DataType>& type,
                          std::vector<std::shared_ptr<arrow::ChunkedArray>>& columns,