Skip to content

feat(avro): support LARGE_STRING, LARGE_BINARY, and LARGE_LIST in datum writer #1105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
akshatgupta08 wants to merge 7 commits into
base: main
Choose a base branch
from
+164 −3
Open

feat(avro): support LARGE_STRING, LARGE_BINARY, and LARGE_LIST in datum writer #1105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
04648e9
support_large_types_arrowToAvro
akshatgupta08 Nov 9, 2026
119346e
test(avro): add tests for LARGE_STRING, LARGE_BINARY, and LARGE_LIST …
akshatgupta08 Nov 10, 2026
104f58d
Revert "test(avro): add tests for LARGE_STRING, LARGE_BINARY, and LAR…
akshatgupta08 Nov 10, 2026
754fab5
test(avro): add LargeListOfIntegers test for ExtractDatumFromArray
akshatgupta08 Nov 10, 2026
0598520
added_test_for_large_strings
akshatgupta08 Nov 10, 2026
94e541c
added_tests_largeBinary
akshatgupta08 Nov 10, 2026
27f3ff5
fixed_signed_unsigned_int_comparison
akshatgupta08 Nov 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 37 additions & 3 deletions src/iceberg/avro/avro_data_util.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -748,15 +748,20 @@ Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index,
return {};
}

// TODO(gangwu): support LARGE_STRING.
case ::arrow::Type::STRING: {
const auto& string_array =
internal::checked_cast<const ::arrow::StringArray&>(array);
datum->value<std::string>() = string_array.GetString(index);
return {};
}

// TODO(gangwu): support LARGE_BINARY.
case ::arrow::Type::LARGE_STRING: {
const auto& large_string_array =
internal::checked_cast<const ::arrow::LargeStringArray&>(array);
datum->value<std::string>() = large_string_array.GetString(index);
return {};
}

case ::arrow::Type::BINARY: {
const auto& binary_array =
internal::checked_cast<const ::arrow::BinaryArray&>(array);
Expand All @@ -767,6 +772,16 @@ Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index,
return {};
}

case ::arrow::Type::LARGE_BINARY: {
const auto& large_binary_array =
internal::checked_cast<const ::arrow::LargeBinaryArray&>(array);
std::string_view value = large_binary_array.GetView(index);
datum->value<std::vector<uint8_t>>().assign(
reinterpret_cast<const uint8_t*>(value.data()),
reinterpret_cast<const uint8_t*>(value.data()) + value.size());
return {};
}

case ::arrow::Type::FIXED_SIZE_BINARY: {
const auto& fixed_array =
internal::checked_cast<const ::arrow::FixedSizeBinaryArray&>(array);
Expand Down Expand Up @@ -835,7 +850,6 @@ Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index,
return {};
}

// TODO(gangwu): support LARGE_LIST.
case ::arrow::Type::LIST: {
const auto& list_array = internal::checked_cast<const ::arrow::ListArray&>(array);
auto& avro_array = datum->value<::avro::GenericArray>();
Expand All @@ -855,6 +869,26 @@ Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index,
return {};
}

case ::arrow::Type::LARGE_LIST: {
const auto& large_list_array =
internal::checked_cast<const ::arrow::LargeListArray&>(array);
auto& avro_array = datum->value<::avro::GenericArray>();
auto& elements = avro_array.value();

auto start = large_list_array.value_offset(index);
auto end = large_list_array.value_offset(index + 1);
auto length = end - start;

auto values = large_list_array.values();
elements.resize(length, ::avro::GenericDatum(avro_array.schema()->leafAt(0)));

for (int64_t i = 0; i < length; ++i) {
ICEBERG_RETURN_UNEXPECTED(
ExtractDatumFromArray(*values, start + i, &elements[i]));
}
return {};
}

case ::arrow::Type::MAP: {
const auto& map_array = internal::checked_cast<const ::arrow::MapArray&>(array);
auto start = map_array.value_offset(index);
Expand Down
127 changes: 127 additions & 0 deletions src/iceberg/test/avro_data_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
#include <ranges>
#include <vector>

#include <arrow/array/array_nested.h>
#include <arrow/array/builder_binary.h>
#include <arrow/array/builder_decimal.h>
#include <arrow/array/builder_nested.h>
#include <arrow/array/builder_primitive.h>
Expand Down Expand Up @@ -1547,6 +1549,131 @@ TEST(ExtractDatumFromArrayTest, MapStringToInt) {
}
}

TEST(ExtractDatumFromArrayTest, LargeListOfIntegers) {
Schema iceberg_schema({
SchemaField::MakeRequired(
1, "numbers",
std::make_shared<ListType>(SchemaField::MakeRequired(2, "element", int32()))),
});

::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());

// Build a LargeListArray manually — JSON parsing always produces regular ListArray,
// so we use builders to get the large-offset variant.
auto int32_builder = std::make_shared<::arrow::Int32Builder>();
::arrow::LargeListBuilder list_builder(::arrow::default_memory_pool(), int32_builder);

ASSERT_TRUE(list_builder.Append().ok());
ASSERT_TRUE(int32_builder->Append(10).ok());
ASSERT_TRUE(int32_builder->Append(11).ok());
ASSERT_TRUE(int32_builder->Append(12).ok());

ASSERT_TRUE(list_builder.Append().ok());
ASSERT_TRUE(int32_builder->Append(20).ok());
ASSERT_TRUE(int32_builder->Append(21).ok());

ASSERT_TRUE(list_builder.Append().ok());
ASSERT_TRUE(int32_builder->Append(30).ok());
ASSERT_TRUE(int32_builder->Append(31).ok());
ASSERT_TRUE(int32_builder->Append(32).ok());
ASSERT_TRUE(int32_builder->Append(33).ok());

std::shared_ptr<::arrow::Array> large_list_array;
ASSERT_TRUE(list_builder.Finish(&large_list_array).ok());

auto arrow_array = ::arrow::StructArray::Make(
::arrow::ArrayVector{large_list_array},
std::vector<std::shared_ptr<::arrow::Field>>{::arrow::field(
"numbers", ::arrow::large_list(::arrow::int32()))})
.ValueOrDie();

std::vector<std::vector<int32_t>> expected = {{10, 11, 12}, {20, 21}, {30, 31, 32, 33}};

auto verify_record = [&](int64_t index, const std::vector<int32_t>& expected_numbers) {
::avro::GenericDatum extracted_datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, index, &extracted_datum), IsOk());
const auto& record = extracted_datum.value<::avro::GenericRecord>();
const auto& avro_arr = record.fieldAt(0).value<::avro::GenericArray>();
const auto& elements = avro_arr.value();

ASSERT_EQ(elements.size(), expected_numbers.size());
for (size_t i = 0; i < expected_numbers.size(); ++i) {
EXPECT_EQ(elements[i].value<int32_t>(), expected_numbers[i]);
}
};

for (size_t i = 0; i < expected.size(); ++i) {
verify_record(i, expected[i]);
}
}

TEST(ExtractDatumFromArrayTest, LargeString) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "name", string()),
});

::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());

// Build a LargeStringArray manually
::arrow::LargeStringBuilder builder;
ASSERT_TRUE(builder.Append("hello").ok());
ASSERT_TRUE(builder.Append("world").ok());
ASSERT_TRUE(builder.Append("iceberg").ok());
std::shared_ptr<::arrow::Array> large_string_array;
ASSERT_TRUE(builder.Finish(&large_string_array).ok());

auto arrow_array =
::arrow::StructArray::Make(::arrow::ArrayVector{large_string_array},
std::vector<std::shared_ptr<::arrow::Field>>{
::arrow::field("name", ::arrow::large_utf8())})
.ValueOrDie();

std::vector<std::string> expected = {"hello", "world", "iceberg"};

for (int64_t i = 0; std::cmp_less(i, expected.size()); ++i) {
::avro::GenericDatum extracted_datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, i, &extracted_datum), IsOk());
const auto& record = extracted_datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<std::string>(), expected[i]);
}
}

TEST(ExtractDatumFromArrayTest, LargeBinary) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "data", binary()),
});

::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());

// Build a LargeBinaryArray manually
::arrow::LargeBinaryBuilder builder;
ASSERT_TRUE(builder.Append("abc").ok());
ASSERT_TRUE(builder.Append("bcd").ok());
ASSERT_TRUE(builder.Append("cde").ok());
std::shared_ptr<::arrow::Array> large_binary_array;
ASSERT_TRUE(builder.Finish(&large_binary_array).ok());

auto arrow_array =
::arrow::StructArray::Make(::arrow::ArrayVector{large_binary_array},
std::vector<std::shared_ptr<::arrow::Field>>{
::arrow::field("data", ::arrow::large_binary())})
.ValueOrDie();

for (int64_t i = 0; i < 3; ++i) {
::avro::GenericDatum extracted_datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, i, &extracted_datum), IsOk());
const auto& record = extracted_datum.value<::avro::GenericRecord>();
const auto& bytes = record.fieldAt(0).value<std::vector<uint8_t>>();
ASSERT_EQ(bytes.size(), 3);
EXPECT_EQ(bytes[0], static_cast<uint8_t>('a' + i));
EXPECT_EQ(bytes[1], static_cast<uint8_t>('b' + i));
EXPECT_EQ(bytes[2], static_cast<uint8_t>('c' + i));
}
}

TEST(ExtractDatumFromArrayTest, ErrorHandling) {
Schema iceberg_schema({SchemaField::MakeRequired(1, "a", int32())});
::avro::NodePtr avro_node;
Expand Down
Loading