Current section
Files
Jump to
Current section
Files
c_src/adbc_arrow_array.hpp
#ifndef ADBC_ARROW_ARRAY_HPP
#pragma once
#include <stdio.h>
#include <cmath>
#include <cstdbool>
#include <cstdint>
#include <vector>
#include <adbc.h>
#include <erl_nif.h>
#include "adbc_half_float.hpp"
static int arrow_array_to_nif_term(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level, std::vector<ERL_NIF_TERM> &out_terms, ERL_NIF_TERM &value_type, ERL_NIF_TERM &metadata, ERL_NIF_TERM &error, bool *end_of_series = nullptr);
static int arrow_array_to_nif_term(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, int64_t level, std::vector<ERL_NIF_TERM> &out_terms, ERL_NIF_TERM &value_type, ERL_NIF_TERM &metadata, ERL_NIF_TERM &error, bool *end_of_series = nullptr);
static int get_arrow_array_children_as_list(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level, std::vector<ERL_NIF_TERM> &children, ERL_NIF_TERM &error);
static int get_arrow_array_children_as_list(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level, std::vector<ERL_NIF_TERM> &children, ERL_NIF_TERM &error);
static ERL_NIF_TERM get_arrow_array_map_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level);
static ERL_NIF_TERM get_arrow_array_map_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level);
static ERL_NIF_TERM get_arrow_array_list_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level, ArrowType list_type, unsigned n_items = 0);
static ERL_NIF_TERM get_arrow_array_list_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level, ArrowType list_type, unsigned n_items = 0);
static ERL_NIF_TERM get_arrow_array_dense_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level);
static ERL_NIF_TERM get_arrow_array_dense_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level);
static ERL_NIF_TERM get_arrow_array_sparse_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level);
static ERL_NIF_TERM get_arrow_array_sparse_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level);
template <typename T, typename M> static ERL_NIF_TERM values_from_buffer(ErlNifEnv *env, int64_t offset, int64_t count, const uint8_t * validity_bitmap, const T * value_buffer, const M& value_to_nif) {
std::vector<ERL_NIF_TERM> values(count);
if (validity_bitmap == nullptr) {
for (int64_t i = offset; i < offset + count; i++) {
values[i - offset] = value_to_nif(env, value_buffer[i]);
}
} else {
for (int64_t i = offset; i < offset + count; i++) {
uint8_t vbyte = validity_bitmap[i/8];
if (vbyte & (1 << (i % 8))) {
values[i - offset] = value_to_nif(env, value_buffer[i]);
} else {
values[i - offset] = kAtomNil;
}
}
}
return enif_make_list_from_array(env, values.data(), (unsigned)values.size());
}
template <typename T, typename M> static ERL_NIF_TERM values_from_buffer(ErlNifEnv *env, int64_t length, const uint8_t * validity_bitmap, const T * value_buffer, const M& value_to_nif) {
return values_from_buffer(env, 0, length, validity_bitmap, value_buffer, value_to_nif);
}
template <typename M, typename OffsetT> static ERL_NIF_TERM strings_from_buffer(
ErlNifEnv *env,
int64_t element_offset,
int64_t element_count,
const uint8_t * validity_bitmap,
const OffsetT * offsets_buffer,
const uint8_t* value_buffer,
const M& value_to_nif) {
OffsetT offset = offsets_buffer[element_offset];
std::vector<ERL_NIF_TERM> values(element_count);
if (validity_bitmap == nullptr) {
for (int64_t i = element_offset; i < element_offset + element_count; i++) {
OffsetT end_index = offsets_buffer[i + 1];
size_t nbytes = end_index - offset;
if (nbytes == 0) {
values[i - element_offset] = kAtomNil;
} else {
values[i - element_offset] = value_to_nif(env, value_buffer, offset, nbytes);
}
offset = end_index;
}
} else {
for (int64_t i = element_offset; i < element_offset + element_count; i++) {
uint8_t vbyte = validity_bitmap[i / 8];
OffsetT end_index = offsets_buffer[i + 1];
size_t nbytes = end_index - offset;
if (nbytes > 0 && vbyte & (1 << (i % 8))) {
values[i - element_offset] = value_to_nif(env, value_buffer, offset, nbytes);
} else {
values[i - element_offset] = kAtomNil;
}
offset = end_index;
}
}
return enif_make_list_from_array(env, values.data(), (unsigned)values.size());
}
template <typename M, typename OffsetT> static ERL_NIF_TERM strings_from_buffer(
ErlNifEnv *env,
int64_t length,
const uint8_t * validity_bitmap,
const OffsetT * offsets_buffer,
const uint8_t* value_buffer,
const M& value_to_nif) {
return strings_from_buffer(env, 0, length, validity_bitmap, offsets_buffer, value_buffer, value_to_nif);
}
template <typename M>
static ERL_NIF_TERM fixed_size_binary_from_buffer(
ErlNifEnv *env,
int64_t element_offset,
int64_t element_count,
size_t element_bytes,
const uint8_t * validity_bitmap,
const uint8_t* value_buffer,
const M& value_to_nif) {
std::vector<ERL_NIF_TERM> values(element_count);
if (validity_bitmap == nullptr) {
for (int64_t i = element_offset; i < element_offset + element_count; i++) {
values[i - element_offset] = value_to_nif(env, &value_buffer[element_bytes * i]);
}
} else {
for (int64_t i = element_offset; i < element_offset + element_count; i++) {
uint8_t vbyte = validity_bitmap[i / 8];
if (vbyte & (1 << (i % 8))) {
values[i - element_offset] = value_to_nif(env, &value_buffer[element_bytes * i]);
} else {
values[i - element_offset] = kAtomNil;
}
}
}
return enif_make_list_from_array(env, values.data(), (unsigned)values.size());
}
template <typename M> static ERL_NIF_TERM fixed_size_binary_from_buffer(
ErlNifEnv *env,
int64_t length,
size_t element_bytes,
const uint8_t * validity_bitmap,
const uint8_t* value_buffer,
const M& value_to_nif) {
return fixed_size_binary_from_buffer(env, 0, length, element_bytes, validity_bitmap, value_buffer, value_to_nif);
}
int get_arrow_array_children_as_list(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level, std::vector<ERL_NIF_TERM> &children, ERL_NIF_TERM &error) {
if (schema->n_children > 0 && schema->children == nullptr) {
error = erlang::nif::error(env, "invalid ArrowSchema, schema->children == nullptr, however, schema->n_children > 0");
return 1;
}
if (values->n_children > 0 && values->children == nullptr) {
error = erlang::nif::error(env, "invalid ArrowArray, values->children == nullptr, however, values->n_children > 0");
return 1;
}
if (values->n_children != schema->n_children) {
error = erlang::nif::error(env, "invalid ArrowArray or ArrowSchema, values->n_children != schema->n_children");
return 1;
}
if (offset < 0 || (offset != 0 && offset >= values->n_children)) {
error = erlang::nif::error(env, "invalid offset value, offset < 0 || offset >= values->n_children");
return 1;
}
if ((offset + count) > values->n_children) {
error = erlang::nif::error(env, "invalid offset or count value, (offset + count) > values->n_children");
return 1;
}
constexpr int64_t bitmap_buffer_index = 0;
const uint8_t * bitmap_buffer = (const uint8_t *)values->buffers[bitmap_buffer_index];
children.resize(count);
for (int64_t child_i = offset; child_i < offset + count; child_i++) {
if (bitmap_buffer && ((schema->flags & ARROW_FLAG_NULLABLE) || (values->null_count > 0))) {
uint8_t vbyte = bitmap_buffer[child_i / 8];
if (!(vbyte & (1 << (child_i % 8)))) {
children[child_i - offset] = kAtomNil;
continue;
}
}
struct ArrowSchema * child_schema = schema->children[child_i];
struct ArrowArray * child_values = values->children[child_i];
std::vector<ERL_NIF_TERM> childrens;
ERL_NIF_TERM child_type;
ERL_NIF_TERM child_metadata;
if (arrow_array_to_nif_term(env, child_schema, child_values, level + 1, childrens, child_type, child_metadata, error) == 1) {
return 1;
}
if (childrens.size() == 1) {
children[child_i - offset] = childrens[0];
} else {
bool nullable = (child_schema->flags & ARROW_FLAG_NULLABLE) || (child_values->null_count > 0);
if (enif_is_identical(childrens[1], kAtomNil)) {
children[child_i - offset] = kAtomNil;
} else {
children[child_i - offset] = make_adbc_column(env, childrens[0], child_type, nullable, child_metadata, childrens[1]);
}
}
}
return 0;
}
int get_arrow_array_children_as_list(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level, std::vector<ERL_NIF_TERM> &children, ERL_NIF_TERM &error) {
return get_arrow_array_children_as_list(env, schema, values, 0, values->n_children, level, children, error);
}
ERL_NIF_TERM get_arrow_array_map_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level) {
// From https://arrow.apache.org/docs/format/CDataInterface.html#data-type-description-format-strings
//
// As specified in the Arrow columnar format, the map type has a single child type named entries,
// itself a 2-child struct type of (key, value).
ERL_NIF_TERM error{}, map_out{};
if (schema->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowSchema (map), schema->children == nullptr");
}
if (schema->n_children != 1) {
return erlang::nif::error(env, "invalid ArrowSchema (map), schema->n_children != 1");
}
if (values->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowArray (map), values->children == nullptr");
}
if (values->n_children != 1) {
return erlang::nif::error(env, "invalid ArrowArray (map), values->n_children != 1");
}
struct ArrowSchema * entries_schema = schema->children[0];
struct ArrowArray * entries_values = values->children[0];
if (strncmp("entries", entries_schema->name, 7) != 0) {
return erlang::nif::error(env, "invalid ArrowSchema (map), its single child is not named entries");
}
if (count == -1) count = entries_values->n_children;
std::vector<ERL_NIF_TERM> nif_keys, nif_values;
bool failed = false;
for (int64_t child_i = offset; child_i < offset + count; child_i++) {
struct ArrowSchema * entry_schema = entries_schema->children[child_i];
struct ArrowArray * entry_values = entries_values->children[child_i];
if (strncmp("key", entry_schema->name, 3) == 0) {
if (get_arrow_array_children_as_list(env, entry_schema, entry_values, level + 1, nif_keys, error) == 1) {
failed = true;
break;
}
} else if (strncmp("value", entry_schema->name, 5) == 0 && entry_schema->n_children == 1) {
struct ArrowSchema * item_schema = entry_schema->children[0];
struct ArrowArray * item_values = entry_values->children[0];
if (get_arrow_array_children_as_list(env, item_schema, item_values, level + 1, nif_values, error) == 1) {
failed = true;
break;
}
} else {
failed = true;
}
}
if (!failed) {
if (nif_keys.size() != nif_values.size()) {
return erlang::nif::error(env, "number of keys and values doesn't match");
}
if (!enif_make_map_from_arrays(env, nif_keys.data(), nif_values.data(), (unsigned)nif_keys.size(), &map_out)) {
return erlang::nif::error(env, "map contains duplicated keys");
} else {
return map_out;
}
} else {
return erlang::nif::error(env, "invalid map");
}
}
ERL_NIF_TERM get_arrow_array_map_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level) {
return get_arrow_array_map_children(env, schema, values, 0, -1, level);
}
ERL_NIF_TERM get_arrow_array_dense_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level) {
ERL_NIF_TERM error{};
if (schema->n_children > 0 && schema->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowSchema (dense union), schema->children == nullptr while schema->n_children > 0 ");
}
if (values->n_children > 0 && values->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowArray (dense union), values->children == nullptr while values->n_children > 0");
}
if (offset < 0 || offset >= values->length) {
return erlang::nif::error(env, "invalid offset value when parsing ArrowArray (dense union), offset < 0 || offset >= values->length");
}
if (count == -1) count = values->length;
if ((offset + count) > values->length) {
return erlang::nif::error(env, "invalid offset or count value when parsing ArrowArray (dense union), (offset + count) > values->length");
}
if (values->n_buffers != 2) {
return erlang::nif::error(env, "invalid ArrowArray (dense union), values->n_buffers != 2");
}
constexpr int64_t types_buffer_index = 0;
constexpr int64_t offset_buffer_index = 1;
const uint8_t * types_buffer = (const uint8_t *)values->buffers[types_buffer_index];
const int32_t * offsets_buffer = (const int32_t *)values->buffers[offset_buffer_index];
std::vector<ERL_NIF_TERM> elements(count);
for (int64_t child_i = offset; child_i < offset + count; child_i++) {
uint8_t child_type = types_buffer[child_i];
int32_t child_offset = offsets_buffer[child_i];
if (child_type >= schema->n_children || child_type >= values->n_children) {
return erlang::nif::error(env, "invalid child type when parsing ArrowArray (dense union), child_type >= schema->n_children || child_type >= values->n_children");
}
struct ArrowSchema * field_schema = schema->children[child_type];
struct ArrowArray * field_array = values->children[child_type];
std::vector<ERL_NIF_TERM> union_element_name(1), union_element_value(1);
std::vector<ERL_NIF_TERM> field_values;
union_element_name[0] = erlang::nif::make_binary(env, field_schema->name);
ERL_NIF_TERM field_type;
ERL_NIF_TERM field_metadata;
if (arrow_array_to_nif_term(env, field_schema, field_array, child_offset, 1, level + 1, field_values, field_type, field_metadata, error) == 1) {
return error;
}
if (field_values.size() == 1) {
union_element_value[0] = field_values[0];
} else if (field_values.size() == 2) {
union_element_value[0] = field_values[1];
} else {
return erlang::nif::error(env, "invalid dense union field value");
}
ERL_NIF_TERM element{};
if (!enif_make_map_from_arrays(env, union_element_name.data(), union_element_value.data(), 1, &element)) {
return erlang::nif::error(env, "failed to enif_make_map_from_arrays when parsing ArrowSchema (dense union)");
}
elements[child_i - offset] = element;
}
return enif_make_list_from_array(env, elements.data(), (unsigned)elements.size());
}
ERL_NIF_TERM get_arrow_array_dense_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level) {
return get_arrow_array_dense_union_children(env, schema, values, 0, -1, level);
}
ERL_NIF_TERM get_arrow_array_sparse_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level) {
ERL_NIF_TERM error{};
if (schema->n_children > 0 && schema->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowSchema (sparse union), schema->children == nullptr while schema->n_children > 0 ");
}
if (values->n_children > 0 && values->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowArray (sparse union), values->children == nullptr while values->n_children > 0");
}
if (offset < 0 || offset >= values->length) {
return erlang::nif::error(env, "invalid offset value when parsing ArrowArray (sparse union), offset < 0 || offset >= values->length");
}
if (count == -1) count = values->length;
if ((offset + count) > values->length) {
return erlang::nif::error(env, "invalid offset or count value when parsing ArrowArray (sparse union), (offset + count) > values->length");
}
if (values->n_buffers != 1) {
return erlang::nif::error(env, "invalid ArrowArray (sparse union), values->n_buffers != 1");
}
constexpr int64_t types_buffer_index = 0;
const uint8_t * types_buffer = (const uint8_t *)values->buffers[types_buffer_index];
std::vector<ERL_NIF_TERM> elements(count);
for (int64_t child_i = offset; child_i < offset + count; child_i++) {
uint8_t child_type = types_buffer[child_i];
if (child_type >= schema->n_children || child_type >= values->n_children) {
return erlang::nif::error(env, "invalid child type when parsing ArrowArray (sparse union), child_type >= schema->n_children || child_type >= values->n_children");
}
struct ArrowSchema * field_schema = schema->children[child_type];
struct ArrowArray * field_array = values->children[child_type];
std::vector<ERL_NIF_TERM> union_element_name(1), union_element_value(1);
std::vector<ERL_NIF_TERM> field_values;
union_element_name[0] = erlang::nif::make_binary(env, field_schema->name);
ERL_NIF_TERM field_type;
// todo: use field_metadata
ERL_NIF_TERM field_metadata;
if (arrow_array_to_nif_term(env, field_schema, field_array, child_i, 1, level + 1, field_values, field_type, field_metadata, error) == 1) {
return error;
}
if (field_values.size() == 1) {
union_element_value[0] = field_values[0];
} else if (field_values.size() == 2) {
union_element_value[0] = field_values[1];
} else {
return erlang::nif::error(env, "invalid sparse union field value");
}
ERL_NIF_TERM element{};
if (!enif_make_map_from_arrays(env, union_element_name.data(), union_element_value.data(), 1, &element)) {
return erlang::nif::error(env, "failed to enif_make_map_from_arrays when parsing ArrowSchema (sparse union)");
}
elements[child_i - offset] = element;
}
return enif_make_list_from_array(env, elements.data(), (unsigned)elements.size());
}
ERL_NIF_TERM get_arrow_array_sparse_union_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level) {
return get_arrow_array_sparse_union_children(env, schema, values, 0, -1, level);
}
ERL_NIF_TERM get_arrow_array_list_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, uint64_t level, ArrowType list_type, unsigned n_items) {
ERL_NIF_TERM error{};
if (schema->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowSchema (list), schema->children == nullptr");
}
if (schema->n_children != 1) {
return erlang::nif::error(env, "invalid ArrowSchema (list), schema->n_children != 1");
}
if (values->children == nullptr) {
return erlang::nif::error(env, "invalid ArrowArray (list), values->children == nullptr");
}
if (values->n_children != 1) {
return erlang::nif::error(env, "invalid ArrowArray (list), values->n_children != 1");
}
if (list_type != NANOARROW_TYPE_LIST && list_type != NANOARROW_TYPE_LARGE_LIST && list_type != NANOARROW_TYPE_FIXED_SIZE_LIST) {
return erlang::nif::error(env, "invalid ArrowArray (list), internal error: unexpected list type");
}
constexpr int64_t bitmap_buffer_index = 0;
const uint8_t * bitmap_buffer = (const uint8_t *)values->buffers[bitmap_buffer_index];
struct ArrowSchema * items_schema = schema->children[0];
struct ArrowArray * items_values = values->children[0];
if (strncmp("item", items_schema->name, 4) != 0) {
return erlang::nif::error(env, "invalid ArrowSchema (list), its single child is not named item");
}
std::vector<ERL_NIF_TERM> children;
if (items_values->n_children > 0) {
if (offset < 0 || offset >= items_values->n_children) {
return erlang::nif::error(env, "invalid offset for ArrowArray (list), offset < 0 || offset >= items_values->n_children");
}
if (count == -1) count = items_values->n_children;
if ((offset + count) > items_values->n_children) {
return erlang::nif::error(env, "invalid offset for ArrowArray (list), (offset + count) > items_values->n_children");
}
children.resize(count);
bool items_nullable = (items_schema->flags & ARROW_FLAG_NULLABLE) || (items_values->null_count > 0);
for (int64_t child_i = offset; child_i < offset + count; child_i++) {
if (bitmap_buffer && items_nullable) {
uint8_t vbyte = bitmap_buffer[child_i / 8];
if (!(vbyte & (1 << (child_i % 8)))) {
children[child_i - offset] = kAtomNil;
continue;
}
}
struct ArrowSchema * item_schema = items_schema->children[child_i];
struct ArrowArray * item_values = items_values->children[child_i];
std::vector<ERL_NIF_TERM> childrens;
ERL_NIF_TERM item_type;
ERL_NIF_TERM item_metadata;
if (arrow_array_to_nif_term(env, item_schema, item_values, level + 1, childrens, item_type, item_metadata, error) == 1) {
return error;
}
if (childrens.size() == 1) {
children[child_i - offset] = childrens[0];
} else {
bool children_nullable = (item_schema->flags & ARROW_FLAG_NULLABLE) || (item_values->null_count > 0);
if (enif_is_identical(childrens[1], kAtomNil)) {
children[child_i - offset] = kAtomNil;
} else {
children[child_i - offset] = make_adbc_column(env, childrens[0], item_type, children_nullable, item_metadata, childrens[1]);
}
}
}
return enif_make_list_from_array(env, children.data(), (unsigned)children.size());
} else {
if (list_type == NANOARROW_TYPE_LIST || list_type == NANOARROW_TYPE_LARGE_LIST) {
constexpr int64_t offset_buffer_index = 1;
const void * offsets_ptr = (const void *)values->buffers[offset_buffer_index];
if (offsets_ptr == nullptr) return erlang::nif::error(env, "invalid ArrowArray (list), offsets == nullptr");
if (count == -1) count = values->length;
bool items_nullable = (schema->flags & ARROW_FLAG_NULLABLE) || (values->null_count > 0);
int has_error = 0;
auto get_list_children_with_offsets = [&](auto offsets) -> void {
for (int64_t i = offset; i < offset + count; i++) {
if (bitmap_buffer && items_nullable) {
uint8_t vbyte = bitmap_buffer[i / 8];
if (!(vbyte & (1 << (i % 8)))) {
children.emplace_back(kAtomNil);
continue;
}
}
std::vector<ERL_NIF_TERM> childrens;
ERL_NIF_TERM children_type;
ERL_NIF_TERM children_metadata;
if (arrow_array_to_nif_term(env, items_schema, items_values, offsets[i], offsets[i+1] - offsets[i], level + 1, childrens, children_type, children_metadata, error) == 1) {
has_error = 1;
return;
}
if (childrens.size() == 1) {
children.emplace_back(childrens[0]);
} else {
bool children_nullable = (schema->flags & ARROW_FLAG_NULLABLE) || (values->null_count > 0);
if (enif_is_identical(childrens[1], kAtomNil)) {
children.emplace_back(kAtomNil);
} else {
children.emplace_back(make_adbc_column(env, childrens[0], children_type, children_nullable, children_metadata, childrens[1]));
}
}
}
};
if (list_type == NANOARROW_TYPE_LIST) {
get_list_children_with_offsets((const int32_t *)offsets_ptr);
if (has_error) return error;
} else if (list_type == NANOARROW_TYPE_LARGE_LIST) {
get_list_children_with_offsets((const int64_t *)offsets_ptr);
if (has_error) return error;
}
} else {
// NANOARROW_TYPE_FIXED_SIZE_LIST
if (count == -1) count = values->length;
bool items_nullable = (schema->flags & ARROW_FLAG_NULLABLE) || (values->null_count > 0);
for (int64_t child_i = offset; child_i < offset + count; child_i++) {
if (bitmap_buffer && items_nullable) {
uint8_t vbyte = bitmap_buffer[child_i / 8];
if (!(vbyte & (1 << (child_i % 8)))) {
children.emplace_back(kAtomNil);
continue;
}
}
std::vector<ERL_NIF_TERM> childrens;
ERL_NIF_TERM children_type;
ERL_NIF_TERM children_metadata;
if (arrow_array_to_nif_term(env, items_schema, items_values, child_i * n_items, n_items, level + 1, childrens, children_type, children_metadata, error)) {
return error;
}
if (childrens.size() == 1) {
children.emplace_back(childrens[0]);
} else {
bool children_nullable = (schema->flags & ARROW_FLAG_NULLABLE) || (values->null_count > 0);
if (enif_is_identical(childrens[1], kAtomNil)) {
children.emplace_back(kAtomNil);
} else {
children.emplace_back(make_adbc_column(env, childrens[0], children_type, children_nullable, children_metadata, childrens[1]));
}
}
}
}
return enif_make_list_from_array(env, children.data(), (unsigned)children.size());
}
}
ERL_NIF_TERM get_arrow_array_list_children(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level, ArrowType list_type, unsigned n_items) {
return get_arrow_array_list_children(env, schema, values, 0, -1, level, list_type, n_items);
}
int arrow_array_to_nif_term(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, int64_t offset, int64_t count, int64_t level, std::vector<ERL_NIF_TERM> &out_terms, ERL_NIF_TERM &term_type, ERL_NIF_TERM &arrow_metadata, ERL_NIF_TERM &error, bool *end_of_series) {
if (schema == nullptr) {
error = erlang::nif::error(env, "invalid ArrowSchema (nullptr) when invoking next");
return 1;
}
if (values == nullptr) {
error = erlang::nif::error(env, "invalid ArrowArray (nullptr) when invoking next");
return 1;
}
char err_msg_buf[256] = { '\0' };
const char* format = schema->format ? schema->format : "";
const char* name = schema->name ? schema->name : "";
term_type = kAtomNil;
arrow_metadata = kAtomNil;
ERL_NIF_TERM current_term{}, children_term{};
std::vector<ERL_NIF_TERM> children;
constexpr int64_t bitmap_buffer_index = 0;
int64_t data_buffer_index = 1;
int64_t offset_buffer_index = 2;
std::vector<ERL_NIF_TERM> metadata_keys, metadata_values;
if (schema->metadata) {
struct ArrowMetadataReader metadata_reader{};
struct ArrowStringView key;
struct ArrowStringView value;
if (ArrowMetadataReaderInit(&metadata_reader, schema->metadata) == NANOARROW_OK) {
while (ArrowMetadataReaderRead(&metadata_reader, &key, &value) == NANOARROW_OK) {
// printf("key: %.*s, value: %.*s\n", (int)key.size_bytes, key.data, (int)value.size_bytes, value.data);
metadata_keys.push_back(erlang::nif::make_binary(env, key.data, (size_t)key.size_bytes));
metadata_values.push_back(erlang::nif::make_binary(env, value.data, (size_t)value.size_bytes));
}
if (metadata_keys.size() > 0) {
enif_make_map_from_arrays(env, metadata_keys.data(), metadata_values.data(), (unsigned)metadata_keys.size(), &arrow_metadata);
}
}
}
bool is_struct = false;
size_t format_len = strlen(format);
bool format_processed = true;
if (format_len == 1) {
if (format[0] == 'n') {
term_type = kAtomNil;
if (count == -1) count = values->length;
std::vector<ERL_NIF_TERM> nils(count);
for (int64_t i = offset; i < offset + count; i++) {
nils.push_back(kAtomNil);
}
current_term = kAtomNil;
} else if (format[0] == 'l') {
// NANOARROW_TYPE_INT64
using value_type = int64_t;
term_type = kAdbcColumnTypeI64;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=l), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_int64
);
} else if (format[0] == 'c') {
// NANOARROW_TYPE_INT8
using value_type = int8_t;
term_type = kAdbcColumnTypeI8;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=c), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_int64
);
} else if (format[0] == 's') {
// NANOARROW_TYPE_INT16
using value_type = int16_t;
term_type = kAdbcColumnTypeI16;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=s), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_int64
);
} else if (format[0] == 'i') {
// NANOARROW_TYPE_INT32
using value_type = int32_t;
term_type = kAdbcColumnTypeI32;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=i), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_int64
);
} else if (format[0] == 'L') {
// NANOARROW_TYPE_UINT64
using value_type = uint64_t;
term_type = kAdbcColumnTypeU64;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=L), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_uint64
);
} else if (format[0] == 'C') {
// NANOARROW_TYPE_UINT8
using value_type = uint8_t;
term_type = kAdbcColumnTypeU8;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=C), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_uint64
);
} else if (format[0] == 'S') {
// NANOARROW_TYPE_UINT16
using value_type = uint16_t;
term_type = kAdbcColumnTypeU16;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=S), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_uint64
);
} else if (format[0] == 'I') {
// NANOARROW_TYPE_UINT32
using value_type = uint32_t;
term_type = kAdbcColumnTypeU32;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=I), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_uint64
);
} else if (format[0] == 'e') {
// NANOARROW_TYPE_HALF_FLOAT
using value_type = uint16_t;
term_type = kAdbcColumnTypeF16;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=e), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, const uint16_t u16) -> ERL_NIF_TERM {
float val = float16_to_float(u16);
if (std::isnan(val)) {
return kAtomNaN;
} else if (std::isinf(val)) {
if (val > 0) {
return kAtomInfinity;
} else {
return kAtomNegInfinity;
}
} else {
return enif_make_double(env, val);
}
}
);
} else if (format[0] == 'f') {
// NANOARROW_TYPE_FLOAT
using value_type = float;
term_type = kAdbcColumnTypeF32;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=f), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, double val) -> ERL_NIF_TERM {
if (std::isnan(val)) {
return kAtomNaN;
} else if (std::isinf(val)) {
if (val > 0) {
return kAtomInfinity;
} else {
return kAtomNegInfinity;
}
} else {
return enif_make_double(env, val);
}
}
);
} else if (format[0] == 'g') {
// NANOARROW_TYPE_DOUBLE
using value_type = double;
term_type = kAdbcColumnTypeF64;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=g), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, double val) -> ERL_NIF_TERM {
if (std::isnan(val)) {
return kAtomNaN;
} else if (std::isinf(val)) {
if (val > 0) {
return kAtomInfinity;
} else {
return kAtomNegInfinity;
}
} else {
return enif_make_double(env, val);
}
}
);
} else if (format[0] == 'b') {
// NANOARROW_TYPE_BOOL
using value_type = bool;
term_type = kAdbcColumnTypeBool;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=b), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, bool val) -> ERL_NIF_TERM {
if (val) {
return kAtomTrue;
}
return kAtomFalse;
}
);
} else if (format[0] == 'u' || format[0] == 'z') {
// NANOARROW_TYPE_BINARY
// NANOARROW_TYPE_STRING
if (format[0] == 'z') {
term_type = kAdbcColumnTypeBinary;
} else {
term_type = kAdbcColumnTypeString;
}
offset_buffer_index = 1;
data_buffer_index = 2;
if (count == -1) count = values->length;
if (values->n_buffers != 3) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=u or format=z), values->n_buffers != 3");
return 1;
}
current_term = strings_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const int32_t *)values->buffers[offset_buffer_index],
(const uint8_t *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, const uint8_t * string_buffers, int32_t offset, size_t nbytes) -> ERL_NIF_TERM {
return erlang::nif::make_binary(env, (const char *)(string_buffers + offset), nbytes);
}
);
} else if (format[0] == 'U' || format[0] == 'Z') {
// NANOARROW_TYPE_LARGE_STRING
// NANOARROW_TYPE_LARGE_BINARY
if (format[0] == 'Z') {
term_type = kAdbcColumnTypeLargeBinary;
} else {
term_type = kAdbcColumnTypeLargeString;
}
offset_buffer_index = 1;
data_buffer_index = 2;
if (count == -1) count = values->length;
if (values->n_buffers != 3) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=U or format=Z), values->n_buffers != 3");
return 1;
}
current_term = strings_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const int64_t *)values->buffers[offset_buffer_index],
(const uint8_t *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, const uint8_t * string_buffers, int64_t offset, size_t nbytes) -> ERL_NIF_TERM {
return erlang::nif::make_binary(env, (const char *)(string_buffers + offset), nbytes);
}
);
} else {
format_processed = false;
}
} else if (format_len == 2) {
if (strncmp("+s", format, 2) == 0) {
// NANOARROW_TYPE_STRUCT
// only handle and return children if this is a struct
is_struct = true;
term_type = kAdbcColumnTypeStruct;
if (values->length > 0 || values->release != nullptr) {
if (count == -1) count = values->n_children;
if (get_arrow_array_children_as_list(env, schema, values, offset, count, level, children, error) == 1) {
return 1;
}
children_term = enif_make_list_from_array(env, children.data(), (unsigned)count);
} else {
if (end_of_series) {
*end_of_series = true;
}
children_term = kAtomEndOfSeries;
}
} else if (strncmp("+m", format, 2) == 0) {
// NANOARROW_TYPE_MAP
term_type = kAdbcColumnTypeMap;
children_term = get_arrow_array_map_children(env, schema, values, offset, count, level);
} else if (strncmp("+l", format, 2) == 0) {
// NANOARROW_TYPE_LIST
term_type = kAdbcColumnTypeList;
children_term = get_arrow_array_list_children(env, schema, values, offset, count, level, NANOARROW_TYPE_LIST);
} else if (strncmp("+L", format, 2) == 0) {
// NANOARROW_TYPE_LARGE_LIST
term_type = kAdbcColumnTypeLargeList;
children_term = get_arrow_array_list_children(env, schema, values, offset, count, level, NANOARROW_TYPE_LARGE_LIST);
} else {
format_processed = false;
}
} else if (format_len >= 3) {
if (strncmp("+w:", format, 3) == 0) {
// NANOARROW_TYPE_FIXED_SIZE_LIST
unsigned n_items = 0;
for (size_t i = 3; i < format_len; i++) {
n_items = n_items * 10 + (format[i] - '0');
}
term_type = kAdbcColumnTypeFixedSizeList(n_items);
children_term = get_arrow_array_list_children(env, schema, values, offset, count, level, NANOARROW_TYPE_FIXED_SIZE_LIST, n_items);
} else if (format_len >= 3 && strncmp("w:", format, 2) == 0) {
// NANOARROW_TYPE_FIXED_SIZE_BINARY
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
snprintf(err_msg_buf, 255, "invalid n_buffers value for ArrowArray (format=%s), values->n_buffers != 2", schema->format);
error = erlang::nif::error(env, erlang::nif::make_binary(env, err_msg_buf));
return 1;
}
size_t nbytes = 0;
for (size_t i = 2; i < format_len; i++) {
nbytes = nbytes * 10 + (format[i] - '0');
}
term_type = kAdbcColumnTypeFixedSizeBinary(nbytes);
current_term = fixed_size_binary_from_buffer(
env,
offset,
count,
nbytes,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const uint8_t *)values->buffers[data_buffer_index],
[&](ErlNifEnv *env, const uint8_t * val) -> ERL_NIF_TERM {
return erlang::nif::make_binary(env, (const char *)val, nbytes);
}
);
} else if (format_len > 4 && (strncmp("+ud:", format, 4) == 0)) {
// NANOARROW_TYPE_DENSE_UNION
term_type = kAdbcColumnTypeDenseUnion;
children_term = get_arrow_array_dense_union_children(env, schema, values, offset, count, level);
} else if (format_len > 4 && (strncmp("+us:", format, 4) == 0)) {
// NANOARROW_TYPE_SPARSE_UNION
term_type = kAdbcColumnTypeSparseUnion;
children_term = get_arrow_array_sparse_union_children(env, schema, values, offset, count, level);
} else if (strncmp("d:", format, 2) == 0) {
// NANOARROW_TYPE_DECIMAL128
// NANOARROW_TYPE_DECIMAL256
//
// format should match `d:P,S[,N]`
// where P is precision, S is scale, N is bits
// N is optional and defaults to 128
int precision = 0;
int scale = 0;
int bits = 128;
int * d[3] = {&precision, &scale, &bits};
int index = 0;
for (size_t i = 2; i < format_len; i++) {
if (format[i] == ',') {
if (index < 2) {
index++;
} else {
format_processed = false;
break;
}
continue;
}
*d[index] = *d[index] * 10 + (format[i] - '0');
}
if (format_processed) {
term_type = kAdbcColumnTypeDecimal(bits, precision, scale);
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
snprintf(err_msg_buf, 255, "invalid n_buffers value for ArrowArray (format=%s), values->n_buffers != 2", schema->format);
error = erlang::nif::error(env, erlang::nif::make_binary(env, err_msg_buf));
return 1;
}
current_term = fixed_size_binary_from_buffer(
env,
offset,
count,
bits / 8,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const uint8_t *)values->buffers[data_buffer_index],
[&](ErlNifEnv *env, const uint8_t * val) -> ERL_NIF_TERM {
return erlang::nif::make_binary(env, (const char *)val, bits / 8);
}
);
}
} else if (format_len == 3 && strncmp("td", format, 2) == 0) {
// possible format strings:
// tdD - date32 [days]
// tdm - date64 [milliseconds]
char unit = format[2];
if (unit == 'D' || unit == 'm') {
// NANOARROW_TYPE_DATE32
// NANOARROW_TYPE_DATE64
ERL_NIF_TERM date_module = kAtomDateModule;
ERL_NIF_TERM calendar_iso = kAtomCalendarISO;
ERL_NIF_TERM keys[] = {
kAtomStructKey,
kAtomCalendarKey,
kAtomYearKey,
kAtomMonthKey,
kAtomDayKey,
};
auto convert = [unit, date_module, calendar_iso, &keys](ErlNifEnv *env, uint64_t val) -> ERL_NIF_TERM {
uint64_t seconds;
if (unit == 'D') {
seconds = val * 24 * 60 * 60; // days
} else {
seconds = val / 1000; // milliseconds
}
time_t t = (time_t)seconds;
tm* time = gmtime(&t);
ERL_NIF_TERM ex_date;
ERL_NIF_TERM values[] = {
date_module,
calendar_iso,
enif_make_int(env, time->tm_year + 1900),
enif_make_int(env, time->tm_mon + 1),
enif_make_int(env, time->tm_mday)
};
enif_make_map_from_arrays(env, keys, values, 5, &ex_date);
return ex_date;
};
if (unit == 'D') {
using value_type = uint32_t;
term_type = kAdbcColumnTypeDate32;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=tdD), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
convert
);
} else {
using value_type = uint64_t;
term_type = kAdbcColumnTypeDate64;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=tdm), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
convert
);
}
} else {
format_processed = false;
}
// time
} else if (format_len == 3 && strncmp("tt", format, 2) == 0) {
// possible format strings:
// tts - time32 [seconds]
// ttm - time32 [milliseconds]
// ttu - time64 [microseconds]
// ttn - time64 [nanoseconds]
uint64_t unit;
uint8_t us_precision;
switch (format[2]) {
case 's': // seconds
// NANOARROW_TYPE_TIME32
unit = 1000000000;
us_precision = 0;
term_type = kAdbcColumnTypeTime32Seconds;
break;
case 'm': // milliseconds
// NANOARROW_TYPE_TIME32
unit = 1000000;
us_precision = 3;
term_type = kAdbcColumnTypeTime32Milliseconds;
break;
case 'u': // microseconds
// NANOARROW_TYPE_TIME64
unit = 1000;
us_precision = 6;
term_type = kAdbcColumnTypeTime64Microseconds;
break;
case 'n': // nanoseconds
// NANOARROW_TYPE_TIME64
unit = 1;
us_precision = 6;
term_type = kAdbcColumnTypeTime64Nanoseconds;
break;
default:
format_processed = false;
}
if (format_processed) {
using value_type = uint64_t;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=tt), values->n_buffers != 2");
return 1;
}
ERL_NIF_TERM keys[] = {
kAtomStructKey,
kAtomCalendarKey,
kAtomHourKey,
kAtomMinuteKey,
kAtomSecondKey,
kAtomMicrosecondKey,
};
ERL_NIF_TERM time_module = kAtomTimeModule;
ERL_NIF_TERM calendar_iso = kAtomCalendarISO;
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[unit, us_precision, time_module, calendar_iso, &keys](ErlNifEnv *env, uint64_t val) -> ERL_NIF_TERM {
// Elixir only supports microsecond precision
uint64_t us = val * unit / 1000;
time_t s = (time_t)(us / 1000000);
tm* time = gmtime(&s);
us = us % 1000000;
ERL_NIF_TERM ex_time;
ERL_NIF_TERM values[] = {
time_module,
calendar_iso,
enif_make_int(env, time->tm_hour),
enif_make_int(env, time->tm_min),
enif_make_int(env, time->tm_sec),
enif_make_tuple2(env, enif_make_int(env, us), enif_make_int(env, us_precision))
};
enif_make_map_from_arrays(env, keys, values, 6, &ex_time);
return ex_time;
}
);
}
// timestamp
} else if (strncmp("ts", format, 2) == 0) {
// possible format strings:
// tss - timestamp [seconds]
// tsm - timestamp [milliseconds]
// tsu - timestamp [microseconds]
// tsn - timestamp [nanoseconds]
//
// if there're any timezone infomation
// it should be in the format like `tsu:timezone`
// NANOARROW_TYPE_TIMESTAMP
uint64_t unit;
uint8_t us_precision;
ERL_NIF_TERM term_unit;
ERL_NIF_TERM term_timezone = kAtomNil;
switch (format[2]) {
case 's': // seconds
unit = 1000000000;
us_precision = 0;
term_unit = kAtomSeconds;
break;
case 'm': // milliseconds
unit = 1000000;
us_precision = 3;
term_unit = kAtomMilliseconds;
break;
case 'u': // microseconds
unit = 1000;
us_precision = 6;
term_unit = kAtomMicroseconds;
break;
case 'n': // nanoseconds
unit = 1;
us_precision = 6;
term_unit = kAtomNanoseconds;
break;
default:
format_processed = false;
}
if (format_processed) {
if (format_len > 4 && format[3] == ':') {
std::string timezone(&format[4]);
term_timezone = erlang::nif::make_binary(env, timezone);
}
term_type = enif_make_tuple3(env, kAtomTimestamp, term_unit, term_timezone);
using value_type = int64_t;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=ts), values->n_buffers != 2");
return 1;
}
ERL_NIF_TERM naive_dt_module = kAtomNaiveDateTimeModule;
ERL_NIF_TERM calendar_iso = kAtomCalendarISO;
ERL_NIF_TERM keys[] = {
kAtomStructKey,
kAtomCalendarKey,
kAtomYearKey,
kAtomMonthKey,
kAtomDayKey,
kAtomHourKey,
kAtomMinuteKey,
kAtomSecondKey,
kAtomMicrosecondKey,
};
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[unit, us_precision, naive_dt_module, calendar_iso, &keys](ErlNifEnv *env, int64_t val) -> ERL_NIF_TERM {
// Elixir only supports microsecond precision
int64_t us = val * unit / 1000;
time_t t = (time_t)(us / 1000000);
tm* time = gmtime(&t);
us = us % 1000000;
ERL_NIF_TERM ex_dt;
ERL_NIF_TERM values[] = {
naive_dt_module,
calendar_iso,
enif_make_int(env, time->tm_year + 1900),
enif_make_int(env, time->tm_mon + 1),
enif_make_int(env, time->tm_mday),
enif_make_int(env, time->tm_hour),
enif_make_int(env, time->tm_min),
enif_make_int(env, time->tm_sec),
enif_make_tuple2(env, enif_make_int(env, us), enif_make_int(env, us_precision))
};
enif_make_map_from_arrays(env, keys, values, 9, &ex_dt);
return ex_dt;
}
);
}
} else if (format_len == 3 && strncmp("tD", format, 2) == 0) {
// possible format strings:
// tDs - duration [seconds]
// tDm - duration [milliseconds]
// tDu - duration [microseconds]
// tDn - duration [nanoseconds]
// NANOARROW_TYPE_DURATION
switch (format[2]) {
case 's': // seconds
term_type = kAdbcColumnTypeDurationSeconds;
break;
case 'm': // milliseconds
term_type = kAdbcColumnTypeDurationMilliseconds;
break;
case 'u': // microseconds
term_type = kAdbcColumnTypeDurationMicroseconds;
break;
case 'n': // nanoseconds
term_type = kAdbcColumnTypeDurationNanoseconds;
break;
default:
format_processed = false;
}
if (format_processed) {
using value_type = int64_t;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=tD), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_int64
);
}
} else if (format_len == 3 && strncmp("ti", format, 2) == 0) {
// possible format strings:
// tiM - interval [months]
// tiD - interval [days, time]
// tin - interval [month, day, nanoseconds]
// NANOARROW_TYPE_INTERVAL
switch (format[2]) {
case 'M': // months
term_type = kAdbcColumnTypeIntervalMonth;
break;
case 'D': // days, time
term_type = kAdbcColumnTypeIntervalDayTime;
break;
case 'n': // month, day, nanoseconds
term_type = kAdbcColumnTypeIntervalMonthDayNano;
break;
default:
format_processed = false;
}
if (format_processed) {
if (format[2] == 'M') {
using value_type = int32_t;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=tiM), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
enif_make_int64
);
} else if (format[2] == 'D') {
using value_type = int64_t;
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=tiD), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, int64_t val) -> ERL_NIF_TERM {
int32_t days = val & 0xFFFFFFFF;
int32_t time = val >> 32;
return enif_make_tuple2(env, enif_make_int(env, days), enif_make_int(env, time));
}
);
} else {
using value_type = struct {
int64_t data[2];
};
if (count == -1) count = values->length;
if (values->n_buffers != 2) {
error = erlang::nif::error(env, "invalid n_buffers value for ArrowArray (format=tin), values->n_buffers != 2");
return 1;
}
current_term = values_from_buffer(
env,
offset,
count,
(const uint8_t *)values->buffers[bitmap_buffer_index],
(const value_type *)values->buffers[data_buffer_index],
[](ErlNifEnv *env, value_type val) -> ERL_NIF_TERM {
int32_t months = val.data[0] & 0xFFFFFFFF;
int32_t days = val.data[0] >> 32;
return enif_make_tuple3(env, enif_make_int64(env, months), enif_make_int64(env, days), enif_make_int64(env, val.data[1]));
}
);
}
}
} else {
format_processed = false;
}
} else {
format_processed = false;
}
if (!format_processed) {
snprintf(err_msg_buf, 255, "not yet implemented for format: `%s`", schema->format);
error = erlang::nif::error(env, erlang::nif::make_binary(env, err_msg_buf));
return 1;
// printf("not implemented for format: `%s`\r\n", schema->format);
// printf("length: %lld\r\n", values->length);
// printf("null_count: %lld\r\n", values->null_count);
// printf("offset: %lld\r\n", values->offset);
// printf("n_buffers: %lld\r\n", values->n_buffers);
// printf("n_children: %lld\r\n", values->n_children);
// printf("buffers: %p\r\n", values->buffers);
}
out_terms.clear();
if (is_struct) {
out_terms.emplace_back(children_term);
} else {
if (schema->children) {
out_terms.emplace_back(erlang::nif::make_binary(env, name));
out_terms.emplace_back(children_term);
} else {
out_terms.emplace_back(erlang::nif::make_binary(env, name));
out_terms.emplace_back(current_term);
}
}
return 0;
}
int arrow_array_to_nif_term(ErlNifEnv *env, struct ArrowSchema * schema, struct ArrowArray * values, uint64_t level, std::vector<ERL_NIF_TERM> &out_terms, ERL_NIF_TERM &out_type, ERL_NIF_TERM &metadata, ERL_NIF_TERM &error, bool *end_of_series) {
return arrow_array_to_nif_term(env, schema, values, 0, -1, level, out_terms, out_type, metadata, error, end_of_series);
}
#endif // ADBC_ARROW_ARRAY_HPP