Google Git Google Git
Sign in
apache/iceberg-cpp/HEAD/./src/iceberg/test/avro_data_test.cc
blob: 9ec0221dae910ecf862902cbbf5ea84752287b05 [file]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this 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 <memory>
#include <ranges>
#include <vector>
#include <arrow/array/builder_decimal.h>
#include <arrow/array/builder_nested.h>
#include <arrow/array/builder_primitive.h>
#include <arrow/c/bridge.h>
#include <arrow/json/from_string.h>
#include <arrow/memory_pool.h>
#include <arrow/type.h>
#include <arrow/util/decimal.h>
#include <avro/Compiler.hh>
#include <avro/Generic.hh>
#include <avro/Node.hh>
#include <avro/Specific.hh>
#include <avro/Stream.hh>
#include <avro/Types.hh>
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include "iceberg/avro/avro_data_util_internal.h"
#include "iceberg/avro/avro_direct_decoder_internal.h"
#include "iceberg/avro/avro_direct_encoder_internal.h"
#include "iceberg/avro/avro_schema_util_internal.h"
#include "iceberg/expression/literal.h"
#include "iceberg/schema.h"
#include "iceberg/schema_internal.h"
#include "iceberg/schema_util.h"
#include "iceberg/test/matchers.h"
#include "iceberg/type.h"
#include "iceberg/util/string_util.h"
namespace iceberg::avro {
class AvroDecimalEncodingTest : public ::testing::TestWithParam<bool> {};
TEST_P(AvroDecimalEncodingTest, FixedWidthBigEndianBytes) {
struct TestCase {
int32_t precision;
std::string unscaled;
std::string expected_hex;
};
const std::vector<TestCase> cases{
{.precision = 1, .unscaled = "0", .expected_hex = "00"},
{.precision = 1, .unscaled = "1", .expected_hex = "01"},
{.precision = 1, .unscaled = "-1", .expected_hex = "ff"},
{.precision = 2, .unscaled = "99", .expected_hex = "63"},
{.precision = 2, .unscaled = "-99", .expected_hex = "9d"},
{.precision = 3, .unscaled = "127", .expected_hex = "007f"},
{.precision = 3, .unscaled = "128", .expected_hex = "0080"},
{.precision = 3, .unscaled = "-128", .expected_hex = "ff80"},
{.precision = 3, .unscaled = "-129", .expected_hex = "ff7f"},
{.precision = 5, .unscaled = "32767", .expected_hex = "007fff"},
{.precision = 5, .unscaled = "32768", .expected_hex = "008000"},
{.precision = 5, .unscaled = "-32768", .expected_hex = "ff8000"},
{.precision = 5, .unscaled = "-32769", .expected_hex = "ff7fff"},
{.precision = 10, .unscaled = "1001", .expected_hex = "00000003e9"},
{.precision = 10, .unscaled = "-1001", .expected_hex = "fffffffc17"},
{.precision = 38,
.unscaled = "0",
.expected_hex = "00000000000000000000000000000000"},
{.precision = 38,
.unscaled = "1",
.expected_hex = "00000000000000000000000000000001"},
{.precision = 38,
.unscaled = "-1",
.expected_hex = "ffffffffffffffffffffffffffffffff"},
{.precision = 38,
.unscaled = "128",
.expected_hex = "00000000000000000000000000000080"},
{.precision = 38,
.unscaled = "-129",
.expected_hex = "ffffffffffffffffffffffffffffff7f"},
{.precision = 38,
.unscaled = "1208925819614629174706299",
.expected_hex = "0000000000010000000000000000007b"},
{.precision = 38,
.unscaled = "-1208925819614629174706299",
.expected_hex = "fffffffffffeffffffffffffffffff85"},
{.precision = 38,
.unscaled = "99999999999999999999999999999999999999",
.expected_hex = "4b3b4ca85a86c47a098a223fffffffff"},
{.precision = 38,
.unscaled = "-99999999999999999999999999999999999999",
.expected_hex = "b4c4b357a5793b85f675ddc000000001"},
};
for (const auto& test : cases) {
SCOPED_TRACE(test.unscaled);
auto type = decimal(test.precision, 0);
::avro::NodePtr node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(*type, &node), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto expected,
StringUtils::HexStringToBytes(test.expected_hex));
ASSERT_EQ(node->fixedSize(), expected.size());
::arrow::Decimal128Builder builder(::arrow::decimal128(test.precision, 0),
::arrow::default_memory_pool());
ASSERT_TRUE(builder.Append(::arrow::Decimal128(test.unscaled)).ok());
auto array = builder.Finish().ValueOrDie();
auto stream = ::avro::memoryOutputStream();
auto encoder = ::avro::binaryEncoder();
encoder->init(*stream);
if (GetParam()) {
EncodeContext context;
ASSERT_THAT(EncodeArrowToAvro(node, *encoder, *type, *array, 0, context), IsOk());
} else {
::avro::GenericDatum datum(node);
ASSERT_THAT(ExtractDatumFromArray(*array, 0, &datum), IsOk());
EXPECT_EQ(datum.value<::avro::GenericFixed>().value(), expected);
::avro::encode(*encoder, datum);
}
encoder->flush();
EXPECT_EQ(*::avro::snapshot(*stream), expected);
}
}
TEST_P(AvroDecimalEncodingTest, RejectsFixedWidthOverflow) {
auto type = decimal(2, 0);
::avro::NodePtr node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(*type, &node), IsOk());
ASSERT_EQ(node->fixedSize(), 1);
// Arrow's builder accepts these invalid values; do not silently truncate them.
for (const auto& unscaled : {"128", "-129", "256", "-257"}) {
SCOPED_TRACE(unscaled);
::arrow::Decimal128Builder builder(::arrow::decimal128(2, 0),
::arrow::default_memory_pool());
ASSERT_TRUE(builder.Append(::arrow::Decimal128(std::string(unscaled))).ok());
auto array = builder.Finish().ValueOrDie();
auto stream = ::avro::memoryOutputStream();
auto encoder = ::avro::binaryEncoder();
encoder->init(*stream);
if (GetParam()) {
EncodeContext context;
EXPECT_THAT(EncodeArrowToAvro(node, *encoder, *type, *array, 0, context),
IsError(ErrorKind::kInvalidArgument));
} else {
::avro::GenericDatum datum(node);
EXPECT_THAT(ExtractDatumFromArray(*array, 0, &datum),
IsError(ErrorKind::kInvalidArgument));
}
encoder->flush();
EXPECT_TRUE(::avro::snapshot(*stream)->empty());
}
}
TEST(AvroDecimalFixedTest, SignedBoundariesAndPadding) {
std::vector<uint8_t> bytes;
ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(127), 1, &bytes), IsOk());
EXPECT_EQ(bytes, (std::vector<uint8_t>{0x7f}));
ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(-128), 1, &bytes), IsOk());
EXPECT_EQ(bytes, (std::vector<uint8_t>{0x80}));
ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(1), 17, &bytes), IsOk());
ICEBERG_UNWRAP_OR_FAIL(
auto positive, StringUtils::HexStringToBytes("0000000000000000000000000000000001"));
EXPECT_EQ(bytes, positive);
ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(-1), 17, &bytes), IsOk());
ICEBERG_UNWRAP_OR_FAIL(
auto negative, StringUtils::HexStringToBytes("ffffffffffffffffffffffffffffffffff"));
EXPECT_EQ(bytes, negative);
EXPECT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(0), 0, &bytes),
IsError(ErrorKind::kInvalidArgument));
}
INSTANTIATE_TEST_SUITE_P(WriterBackends, AvroDecimalEncodingTest,
::testing::Values(true, false),
[](const ::testing::TestParamInfo<bool>& info) {
return info.param ? "DirectEncoder" : "GenericDatum";
});
/// \brief Test case structure for parameterized primitive type tests
struct AppendDatumParam {
std::string name;
std::shared_ptr<Type> projected_type;
std::shared_ptr<Type> source_type;
std::function<void(::avro::GenericDatum&, int)> value_setter;
std::string expected_json;
};
/// \brief Helper function to create test data for a primitive type
std::vector<::avro::GenericDatum> CreateTestData(
const ::avro::NodePtr& avro_node,
const std::function<void(::avro::GenericDatum&, int)>& value_setter, int count = 3) {
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < count; ++i) {
::avro::GenericDatum avro_datum(avro_node);
value_setter(avro_datum, i);
avro_data.push_back(avro_datum);
}
return avro_data;
}
/// \brief Utility function to verify AppendDatumToBuilder behavior
void VerifyAppendDatumToBuilder(const Schema& projected_schema,
const ::avro::NodePtr& avro_node,
const std::vector<::avro::GenericDatum>& avro_data,
std::string_view expected_array_json) {
// Create 1 to 1 projection
auto projection_result = Project(projected_schema, avro_node, /*prune_source=*/false);
ASSERT_THAT(projection_result, IsOk());
auto projection = std::move(projection_result.value());
// Create arrow schema and array builder
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(projected_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
auto builder = ::arrow::MakeBuilder(arrow_struct_type).ValueOrDie();
// Call AppendDatumToBuilder repeatedly to append the datum
for (const auto& avro_datum : avro_data) {
ASSERT_THAT(AppendDatumToBuilder(avro_node, avro_datum, projection, projected_schema,
/*metadata_context=*/{}, builder.get()),
IsOk());
}
// Verify the result
auto array = builder->Finish().ValueOrDie();
auto expected_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, expected_array_json)
.ValueOrDie();
ASSERT_TRUE(array->Equals(*expected_array))
<< "array: " << array->ToString()
<< "\nexpected_array: " << expected_array->ToString();
}
/// \brief Test class for primitive types using parameterized tests
class AppendDatumToBuilderTest : public ::testing::TestWithParam<AppendDatumParam> {};
TEST_P(AppendDatumToBuilderTest, PrimitiveType) {
const auto& test_case = GetParam();
Schema projected_schema({SchemaField::MakeRequired(
/*field_id=*/1, /*name=*/"a", test_case.projected_type)});
Schema source_schema({SchemaField::MakeRequired(
/*field_id=*/1, /*name=*/"a", test_case.source_type)});
::avro::NodePtr avro_node;
EXPECT_THAT(ToAvroNodeVisitor{}.Visit(source_schema, &avro_node), IsOk());
auto avro_data = CreateTestData(avro_node, test_case.value_setter);
ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(projected_schema, avro_node,
avro_data, test_case.expected_json));
}
// Define test cases for all primitive types
const std::vector<AppendDatumParam> kPrimitiveTestCases = {
{
.name = "Boolean",
.projected_type = iceberg::boolean(),
.source_type = iceberg::boolean(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<bool>() =
(i % 2 == 0);
},
.expected_json = R"([{"a": true}, {"a": false}, {"a": true}])",
},
{
.name = "Int",
.projected_type = iceberg::int32(),
.source_type = iceberg::int32(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<int32_t>() = i * 100;
},
.expected_json = R"([{"a": 0}, {"a": 100}, {"a": 200}])",
},
{
.name = "Long",
.projected_type = iceberg::int64(),
.source_type = iceberg::int64(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<int64_t>() =
i * 1000000LL;
},
.expected_json = R"([{"a": 0}, {"a": 1000000}, {"a": 2000000}])",
},
{
.name = "Float",
.projected_type = iceberg::float32(),
.source_type = iceberg::float32(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<float>() = i * 3.14f;
},
.expected_json = R"([{"a": 0.0}, {"a": 3.14}, {"a": 6.28}])",
},
{
.name = "Double",
.projected_type = iceberg::float64(),
.source_type = iceberg::float64(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<double>() =
i * 1.234567890;
},
.expected_json = R"([{"a": 0.0}, {"a": 1.234567890}, {"a": 2.469135780}])",
},
{
.name = "String",
.projected_type = iceberg::string(),
.source_type = iceberg::string(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<std::string>() =
"test_string_" + std::to_string(i);
},
.expected_json =
R"([{"a": "test_string_0"}, {"a": "test_string_1"}, {"a": "test_string_2"}])",
},
{
.name = "Binary",
.projected_type = iceberg::binary(),
.source_type = iceberg::binary(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>()
.fieldAt(0)
.value<std::vector<uint8_t>>() = {static_cast<uint8_t>('a' + i),
static_cast<uint8_t>('b' + i),
static_cast<uint8_t>('c' + i)};
},
.expected_json = R"([{"a": "abc"}, {"a": "bcd"}, {"a": "cde"}])",
},
{
.name = "Fixed",
.projected_type = iceberg::fixed(4),
.source_type = iceberg::fixed(4),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>()
.fieldAt(0)
.value<::avro::GenericFixed>()
.value() = {
static_cast<uint8_t>('a' + i), static_cast<uint8_t>('b' + i),
static_cast<uint8_t>('c' + i), static_cast<uint8_t>('d' + i)};
},
.expected_json = R"([{"a": "abcd"}, {"a": "bcde"}, {"a": "cdef"}])",
},
/// FIXME: NotImplemented: MakeBuilder: cannot construct builder for type
/// extension<arrow.uuid>. Need to fix this in the upstream Arrow.
// {
// .name = "UUID",
// .projected_type = iceberg::uuid(),
// .source_type = iceberg::uuid(),
// .value_setter =
// [](::avro::GenericDatum& datum, int i) {
// datum.value<::avro::GenericRecord>()
// .fieldAt(0)
// .value<::avro::GenericFixed>()
// .value() = {'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h',
// 'i', 'j', 'k', 'l', 'm', 'n', 'o',
// static_cast<uint8_t>(i)};
// },
// .expected_json = R"([{"a": "abcdefghijklmnop"}, {"a": "bcdefghijklmnopq"},
// {"a": "cdefghijklmnopqr"}])",
// },
{
.name = "Decimal",
.projected_type = iceberg::decimal(10, 2),
.source_type = iceberg::decimal(10, 2),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
int32_t decimal_value = i * 1000 + i;
std::vector<uint8_t>& fixed = datum.value<::avro::GenericRecord>()
.fieldAt(0)
.value<::avro::GenericFixed>()
.value();
// The byte array must contain the two's-complement representation of
// the unscaled integer value in big-endian byte order.
for (uint8_t& rvalue : std::ranges::reverse_view(fixed)) {
rvalue = static_cast<uint8_t>(decimal_value & 0xFF);
decimal_value >>= 8;
}
},
.expected_json = R"([{"a": "0.00"}, {"a": "10.01"}, {"a": "20.02"}])",
},
{
.name = "Date",
.projected_type = iceberg::date(),
.source_type = iceberg::date(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
// Date as days since epoch (1970-01-01)
// 0 = 1970-01-01, 1 = 1970-01-02, etc.
datum.value<::avro::GenericRecord>().fieldAt(0).value<int32_t>() =
18000 + i; // ~2019-04-11 + i days
},
.expected_json = R"([{"a": 18000}, {"a": 18001}, {"a": 18002}])",
},
{
.name = "Time",
.projected_type = iceberg::time(),
.source_type = iceberg::time(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
// Time as microseconds since midnight
// 12:30:45.123456 + i seconds = 45045123456 + i*1000000 microseconds
datum.value<::avro::GenericRecord>().fieldAt(0).value<int64_t>() =
45045123456LL + i * 1000000LL;
},
.expected_json =
R"([{"a": 45045123456}, {"a": 45046123456}, {"a": 45047123456}])",
},
{
.name = "Timestamp",
.projected_type = iceberg::timestamp(),
.source_type = iceberg::timestamp(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<int64_t>() =
i * 1000000LL;
},
.expected_json = R"([{"a": 0}, {"a": 1000000}, {"a": 2000000}])",
},
{
.name = "TimestampTz",
.projected_type = std::make_shared<TimestampTzType>(),
.source_type = std::make_shared<TimestampTzType>(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<int64_t>() =
1672531200000000LL + i * 1000000LL;
},
.expected_json =
R"([{"a": 1672531200000000}, {"a": 1672531201000000}, {"a": 1672531202000000}])",
},
{
.name = "TimestampNs",
.projected_type = iceberg::timestamp_ns(),
.source_type = iceberg::timestamp_ns(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<int64_t>() =
1672531200000000000LL + i * 1000000000LL + i;
},
.expected_json =
R"([{"a": 1672531200000000000}, {"a": 1672531201000000001}, {"a": 1672531202000000002}])",
},
{
.name = "TimestampTzNs",
.projected_type = iceberg::timestamptz_ns(),
.source_type = iceberg::timestamptz_ns(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<int64_t>() =
1672531200000000000LL + i * 1000000000LL + i;
},
.expected_json =
R"([{"a": 1672531200000000000}, {"a": 1672531201000000001}, {"a": 1672531202000000002}])",
},
{
.name = "IntToLongPromotion",
.projected_type = iceberg::int64(),
.source_type = iceberg::int32(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<int32_t>() = i * 100;
},
.expected_json = R"([{"a": 0}, {"a": 100}, {"a": 200}])",
},
{
.name = "FloatToDoublePromotion",
.projected_type = iceberg::float64(),
.source_type = iceberg::float32(),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
datum.value<::avro::GenericRecord>().fieldAt(0).value<float>() = i * 1.0f;
},
.expected_json = R"([{"a": 0.0}, {"a": 1.0}, {"a": 2.0}])",
},
{
.name = "DecimalPrecisionPromotion",
.projected_type = iceberg::decimal(10, 2),
.source_type = iceberg::decimal(6, 2),
.value_setter =
[](::avro::GenericDatum& datum, int i) {
int32_t decimal_value = i * 1000 + i;
std::vector<uint8_t>& fixed = datum.value<::avro::GenericRecord>()
.fieldAt(0)
.value<::avro::GenericFixed>()
.value();
for (uint8_t& rvalue : std::ranges::reverse_view(fixed)) {
rvalue = static_cast<uint8_t>(decimal_value & 0xFF);
decimal_value >>= 8;
}
},
.expected_json = R"([{"a": "0.00"}, {"a": "10.01"}, {"a": "20.02"}])",
},
};
INSTANTIATE_TEST_SUITE_P(AllPrimitiveTypes, AppendDatumToBuilderTest,
::testing::ValuesIn(kPrimitiveTestCases),
[](const ::testing::TestParamInfo<AppendDatumParam>& info) {
return info.param.name;
});
TEST(AppendDatumToBuilderTest, StructWithTwoFields) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", iceberg::int32()),
SchemaField::MakeRequired(2, "name", iceberg::string()),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
std::vector<::avro::GenericDatum> avro_data;
::avro::GenericDatum avro_datum(avro_node);
auto& record = avro_datum.value<::avro::GenericRecord>();
record.fieldAt(0).value<int32_t>() = 42;
record.fieldAt(1).value<std::string>() = "test";
avro_data.push_back(avro_datum);
ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data,
R"([{"id": 42, "name": "test"}])"));
}
TEST(AppendDatumToBuilderTest, NestedStruct) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", iceberg::int32()),
SchemaField::MakeRequired(
2, "person",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(3, "name", iceberg::string()),
SchemaField::MakeRequired(4, "age", iceberg::int32()),
})),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_node);
auto& record = avro_datum.value<::avro::GenericRecord>();
// Set id field
record.fieldAt(0).value<int32_t>() = i + 1;
// Set nested person struct
auto& person_record = record.fieldAt(1).value<::avro::GenericRecord>();
person_record.fieldAt(0).value<std::string>() = "Person" + std::to_string(i);
person_record.fieldAt(1).value<int32_t>() = 25 + i;
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"id": 1, "person": {"name": "Person0", "age": 25}},
{"id": 2, "person": {"name": "Person1", "age": 26}}
])";
ASSERT_NO_FATAL_FAILURE(
VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, ListOfIntegers) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "numbers",
std::make_shared<ListType>(SchemaField::MakeRequired(
2, "element", iceberg::int32()))),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_node);
auto& record = avro_datum.value<::avro::GenericRecord>();
// Create array with values [i*10, i*10+1, i*10+2]
auto& array = record.fieldAt(0).value<::avro::GenericArray>();
for (int j = 0; j < 3; ++j) {
::avro::GenericDatum element(avro_node->leafAt(0)->leafAt(0));
element.value<int32_t>() = i * 10 + j;
array.value().push_back(element);
}
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"numbers": [0, 1, 2]},
{"numbers": [10, 11, 12]}
])";
ASSERT_NO_FATAL_FAILURE(
VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, ListOfStructs) {
Schema iceberg_schema({
SchemaField::MakeRequired(
1, "people",
std::make_shared<ListType>(SchemaField::MakeRequired(
2, "element",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(3, "name", iceberg::string()),
SchemaField::MakeRequired(4, "age", iceberg::int32()),
})))),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_node);
auto& record = avro_datum.value<::avro::GenericRecord>();
auto& array = record.fieldAt(0).value<::avro::GenericArray>();
for (int j = 0; j < 2; ++j) {
::avro::GenericDatum element(avro_node->leafAt(0)->leafAt(0));
auto& person_record = element.value<::avro::GenericRecord>();
person_record.fieldAt(0).value<std::string>() =
"Person" + std::to_string(i) + "_" + std::to_string(j);
person_record.fieldAt(1).value<int32_t>() = 20 + i * 10 + j;
array.value().push_back(element);
}
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"people": [
{"name": "Person0_0", "age": 20},
{"name": "Person0_1", "age": 21}
]},
{"people": [
{"name": "Person1_0", "age": 30},
{"name": "Person1_1", "age": 31}
]}
])";
ASSERT_NO_FATAL_FAILURE(
VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, MapStringToInt) {
Schema iceberg_schema({
SchemaField::MakeRequired(
1, "scores",
std::make_shared<MapType>(
SchemaField::MakeRequired(2, "key", iceberg::string()),
SchemaField::MakeRequired(3, "value", iceberg::int32()))),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_node);
auto& record = avro_datum.value<::avro::GenericRecord>();
auto& map = record.fieldAt(0).value<::avro::GenericMap>();
auto& map_container = map.value();
map_container.emplace_back("score_" + std::to_string(i * 2),
::avro::GenericDatum(static_cast<int32_t>(100 + i * 10)));
map_container.emplace_back(
"score_" + std::to_string(i * 2 + 1),
::avro::GenericDatum(static_cast<int32_t>(100 + i * 10 + 5)));
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"scores": [["score_0", 100], ["score_1", 105]]},
{"scores": [["score_2", 110], ["score_3", 115]]}
])";
ASSERT_NO_FATAL_FAILURE(
VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, MapIntToStringAsArray) {
Schema iceberg_schema({
SchemaField::MakeRequired(
1, "names",
std::make_shared<MapType>(
SchemaField::MakeRequired(2, "key", iceberg::int32()),
SchemaField::MakeRequired(3, "value", iceberg::string()))),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_node);
auto& record = avro_datum.value<::avro::GenericRecord>();
auto& array = record.fieldAt(0).value<::avro::GenericArray>();
for (int j = 0; j < 2; ++j) {
::avro::GenericDatum kv_pair(avro_node->leafAt(0)->leafAt(0));
auto& kv_record = kv_pair.value<::avro::GenericRecord>();
kv_record.fieldAt(0).value<int32_t>() = i * 10 + j;
kv_record.fieldAt(1).value<std::string>() = "name_" + std::to_string(i * 10 + j);
array.value().push_back(kv_pair);
}
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"names": [[0, "name_0"], [1, "name_1"]]},
{"names": [[10, "name_10"], [11, "name_11"]]}
])";
ASSERT_NO_FATAL_FAILURE(
VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, MapStringToStruct) {
Schema iceberg_schema({
SchemaField::MakeRequired(
1, "users",
std::make_shared<MapType>(
SchemaField::MakeRequired(2, "key", iceberg::string()),
SchemaField::MakeRequired(
3, "value",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(4, "id", iceberg::int32()),
SchemaField::MakeRequired(5, "email", iceberg::string()),
})))),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_node);
auto& record = avro_datum.value<::avro::GenericRecord>();
auto& map = record.fieldAt(0).value<::avro::GenericMap>();
auto& map_container = map.value();
::avro::GenericDatum struct_value(avro_node->leafAt(0)->leafAt(1));
auto& struct_record = struct_value.value<::avro::GenericRecord>();
struct_record.fieldAt(0).value<int32_t>() = 1000 + i;
struct_record.fieldAt(1).value<std::string>() =
"user" + std::to_string(i) + "@example.com";
map_container.emplace_back("user_" + std::to_string(i), std::move(struct_value));
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"users": [["user_0", {"id": 1000, "email": "user0@example.com"}]]},
{"users": [["user_1", {"id": 1001, "email": "user1@example.com"}]]}
])";
ASSERT_NO_FATAL_FAILURE(
VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, StructWithMissingOptionalField) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", iceberg::int32()),
SchemaField::MakeRequired(2, "name", iceberg::string()),
SchemaField::MakeOptional(3, "age",
iceberg::int32()), // Missing in Avro
SchemaField::MakeOptional(4, "email",
iceberg::string()), // Missing in Avro
});
// Create Avro schema that only has id and name fields (missing age and email)
std::string avro_schema_json = R"({
"type": "record",
"name": "person",
"fields": [
{"name": "id", "type": "int", "field-id": 1},
{"name": "name", "type": "string", "field-id": 2}
]
})";
auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_schema.root());
auto& record = avro_datum.value<::avro::GenericRecord>();
record.fieldAt(0).value<int32_t>() = i + 1;
record.fieldAt(1).value<std::string>() = "Person" + std::to_string(i);
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"id": 1, "name": "Person0", "age": null, "email": null},
{"id": 2, "name": "Person1", "age": null, "email": null}
])";
ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(iceberg_schema, avro_schema.root(),
avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, StructWithMissingDefaultFields) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", iceberg::int32()),
// Missing required field with an initial-default: filled with the default.
SchemaField(2, "score", iceberg::int64(), /*optional=*/false, /*doc=*/{},
std::make_shared<const Literal>(Literal::Long(100))),
// Missing optional field with an initial-default: also filled, not null.
SchemaField(3, "grade", iceberg::string(), /*optional=*/true, /*doc=*/{},
std::make_shared<const Literal>(Literal::String("A"))),
});
// Create Avro schema that only has the id field (missing score and grade).
std::string avro_schema_json = R"({
"type": "record",
"name": "person",
"fields": [
{"name": "id", "type": "int", "field-id": 1}
]
})";
auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_schema.root());
auto& record = avro_datum.value<::avro::GenericRecord>();
record.fieldAt(0).value<int32_t>() = i + 1;
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"id": 1, "score": 100, "grade": "A"},
{"id": 2, "score": 100, "grade": "A"}
])";
ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(iceberg_schema, avro_schema.root(),
avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, NestedListOfListStructWithMissingDefaultField) {
// A default on a struct nested two collection levels deep must still be filled, which
// requires recursing through both list builders rather than only an immediate struct.
auto inner_struct = std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(5, "x", iceberg::int32()),
SchemaField(6, "y", iceberg::int64(), /*optional=*/false, /*doc=*/{},
std::make_shared<const Literal>(Literal::Long(7))),
});
auto inner_list =
std::make_shared<ListType>(SchemaField::MakeRequired(4, "element", inner_struct));
auto outer_list =
std::make_shared<ListType>(SchemaField::MakeRequired(3, "element", inner_list));
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", iceberg::int32()),
SchemaField::MakeRequired(2, "matrix", outer_list),
});
// The Avro schema's innermost record only has `x`, so `y` comes from its default.
std::string avro_schema_json = R"({
"type": "record",
"name": "outer",
"fields": [
{"name": "id", "type": "int", "field-id": 1},
{"name": "matrix", "field-id": 2, "type": {
"type": "array", "element-id": 3, "items": {
"type": "array", "element-id": 4, "items": {
"type": "record", "name": "point",
"fields": [{"name": "x", "type": "int", "field-id": 5}]
}
}
}}
]
})";
auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
std::vector<::avro::GenericDatum> avro_data;
::avro::GenericDatum avro_datum(avro_schema.root());
auto& record = avro_datum.value<::avro::GenericRecord>();
record.fieldAt(0).value<int32_t>() = 1;
auto& outer_array = record.fieldAt(1).value<::avro::GenericArray>();
::avro::GenericDatum inner_datum(avro_schema.root()->leafAt(1)->leafAt(0));
auto& inner_array = inner_datum.value<::avro::GenericArray>();
for (int32_t x : {10, 20}) {
::avro::GenericDatum point_datum(avro_schema.root()->leafAt(1)->leafAt(0)->leafAt(0));
point_datum.value<::avro::GenericRecord>().fieldAt(0).value<int32_t>() = x;
inner_array.value().push_back(point_datum);
}
outer_array.value().push_back(inner_datum);
avro_data.push_back(avro_datum);
const std::string expected_json = R"([
{"id": 1, "matrix": [[{"x": 10, "y": 7}, {"x": 20, "y": 7}]]}
])";
ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(iceberg_schema, avro_schema.root(),
avro_data, expected_json));
}
TEST(AppendDefaultToBuilderTest, AppendsValue) {
::arrow::Int64Builder builder;
ASSERT_THAT(AppendDefaultToBuilder(Literal::Long(42), &builder), IsOk());
ASSERT_THAT(AppendDefaultToBuilder(Literal::Long(42), &builder), IsOk());
std::shared_ptr<::arrow::Array> array;
ASSERT_TRUE(builder.Finish(&array).ok());
ASSERT_EQ(array->length(), 2);
const auto& long_array = static_cast<const ::arrow::Int64Array&>(*array);
ASSERT_EQ(long_array.Value(0), 42);
ASSERT_EQ(long_array.Value(1), 42);
}
TEST(AppendDefaultToBuilderTest, CastsToBuilderType) {
// The literal's natural type (int32) differs from the builder type (int64); the value
// is cast to the builder type.
::arrow::Int64Builder builder;
ASSERT_THAT(AppendDefaultToBuilder(Literal::Int(7), &builder), IsOk());
std::shared_ptr<::arrow::Array> array;
ASSERT_TRUE(builder.Finish(&array).ok());
ASSERT_EQ(array->length(), 1);
ASSERT_EQ(static_cast<const ::arrow::Int64Array&>(*array).Value(0), 7);
}
TEST(AppendDefaultToBuilderTest, ReusesPreparedScalar) {
auto pool = ::arrow::default_memory_pool();
auto child = std::make_shared<::arrow::Int64Builder>(pool);
::arrow::StructBuilder struct_builder(
::arrow::struct_({::arrow::field("d", ::arrow::int64())}), pool, {child});
FieldProjection projection;
projection.kind = FieldProjection::Kind::kDefault;
projection.from = Literal::Long(42);
SchemaProjection schema_projection;
schema_projection.fields.push_back(projection);
ASSERT_THAT(PrepareDefaultScalars(schema_projection, &struct_builder), IsOk());
ASSERT_NE(dynamic_cast<const AvroExtraAttributes*>(
schema_projection.fields[0].attributes.get()),
nullptr);
auto* field_builder = struct_builder.field_builder(0);
ASSERT_THAT(AppendDefaultToBuilder(schema_projection.fields[0], field_builder), IsOk());
ASSERT_THAT(AppendDefaultToBuilder(schema_projection.fields[0], field_builder), IsOk());
std::shared_ptr<::arrow::Array> array;
ASSERT_TRUE(field_builder->Finish(&array).ok());
ASSERT_EQ(array->length(), 2);
const auto& long_array = static_cast<const ::arrow::Int64Array&>(*array);
ASSERT_EQ(long_array.Value(0), 42);
ASSERT_EQ(long_array.Value(1), 42);
}
TEST(AppendDefaultToBuilderTest, PreparesScalarWhenContainerAlreadyExists) {
// The Avro attributes container may already exist (e.g. attached by another Avro
// attribute) with default_scalar unset. Preparation must still fill default_scalar
// rather than skip on the container's mere presence, otherwise the append path falls
// back to rebuilding the scalar per row.
auto pool = ::arrow::default_memory_pool();
auto child = std::make_shared<::arrow::Int64Builder>(pool);
::arrow::StructBuilder struct_builder(
::arrow::struct_({::arrow::field("d", ::arrow::int64())}), pool, {child});
FieldProjection projection;
projection.kind = FieldProjection::Kind::kDefault;
projection.from = Literal::Long(42);
// Pre-attach an empty container, simulating another attribute having created it.
projection.attributes = std::make_shared<AvroExtraAttributes>();
SchemaProjection schema_projection;
schema_projection.fields.push_back(projection);
ASSERT_THAT(PrepareDefaultScalars(schema_projection, &struct_builder), IsOk());
const auto* attrs = dynamic_cast<const AvroExtraAttributes*>(
schema_projection.fields[0].attributes.get());
ASSERT_NE(attrs, nullptr);
ASSERT_NE(attrs->default_scalar, nullptr);
}
TEST(AppendDefaultToBuilderTest, PreparesScalarUnderNestedCollections) {
// A default under `list<list<struct<...>>>` must be prepared too, so decoding reuses
// the cached scalar instead of rebuilding it for every element.
auto pool = ::arrow::default_memory_pool();
auto leaf = std::make_shared<::arrow::Int64Builder>(pool);
auto point_builder = std::make_shared<::arrow::StructBuilder>(
::arrow::struct_({::arrow::field("y", ::arrow::int64())}), pool,
std::vector<std::shared_ptr<::arrow::ArrayBuilder>>{leaf});
auto inner_list = std::make_shared<::arrow::ListBuilder>(pool, point_builder);
auto outer_list = std::make_shared<::arrow::ListBuilder>(pool, inner_list);
::arrow::StructBuilder root_builder(
::arrow::struct_({::arrow::field("matrix", outer_list->type())}), pool,
std::vector<std::shared_ptr<::arrow::ArrayBuilder>>{outer_list});
FieldProjection leaf_default;
leaf_default.kind = FieldProjection::Kind::kDefault;
leaf_default.from = Literal::Long(7);
FieldProjection point_projection;
point_projection.kind = FieldProjection::Kind::kProjected;
point_projection.children.push_back(leaf_default);
FieldProjection inner_element;
inner_element.kind = FieldProjection::Kind::kProjected;
inner_element.children.push_back(point_projection);
FieldProjection outer_element;
outer_element.kind = FieldProjection::Kind::kProjected;
outer_element.children.push_back(inner_element);
SchemaProjection schema_projection;
schema_projection.fields.push_back(outer_element);
ASSERT_THAT(PrepareDefaultScalars(schema_projection, &root_builder), IsOk());
const auto& prepared = schema_projection.fields[0].children[0].children[0].children[0];
ASSERT_NE(dynamic_cast<const AvroExtraAttributes*>(prepared.attributes.get()), nullptr);
}
TEST(AppendDatumToBuilderTest, NestedStructWithMissingOptionalFields) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", iceberg::int32()),
SchemaField::MakeRequired(
2, "person",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(3, "name", iceberg::string()),
SchemaField::MakeOptional(4, "age",
iceberg::int32()), // Missing
SchemaField::MakeOptional(5, "phone",
iceberg::string()), // Missing
})),
SchemaField::MakeOptional(6, "department",
iceberg::string()), // Missing
});
// Create Avro schema with only id, person.name fields
std::string avro_schema_json = R"({
"type": "record",
"name": "employee",
"fields": [
{"name": "id", "type": "int", "field-id": 1},
{"name": "person", "type": {
"type": "record",
"name": "person_info",
"fields": [
{"name": "name", "type": "string", "field-id": 3}
]
}, "field-id": 2}
]
})";
auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_schema.root());
auto& record = avro_datum.value<::avro::GenericRecord>();
record.fieldAt(0).value<int32_t>() = i + 100;
auto& person_record = record.fieldAt(1).value<::avro::GenericRecord>();
person_record.fieldAt(0).value<std::string>() = "Employee" + std::to_string(i);
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"id": 100, "person": {"name": "Employee0", "age": null, "phone": null}, "department": null},
{"id": 101, "person": {"name": "Employee1", "age": null, "phone": null}, "department": null}
])";
ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(iceberg_schema, avro_schema.root(),
avro_data, expected_json));
}
TEST(AppendDatumToBuilderTest, ListWithMissingOptionalElementFields) {
Schema iceberg_schema({
SchemaField::MakeRequired(
1, "people",
std::make_shared<ListType>(SchemaField::MakeRequired(
2, "element",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(3, "name", iceberg::string()),
SchemaField::MakeOptional(4, "age",
iceberg::int32()), // Missing in Avro
SchemaField::MakeOptional(5, "email",
iceberg::string()), // Missing in Avro
})))),
});
// Create Avro schema with list of structs that only have name field
std::string avro_schema_json = R"({
"type": "record",
"name": "people_list",
"fields": [
{"name": "people", "type": {
"type": "array",
"items": {
"type": "record",
"name": "person",
"fields": [
{"name": "name", "type": "string", "field-id": 3}
]
},
"element-id": 2
}, "field-id": 1}
]
})";
auto avro_schema = ::avro::compileJsonSchemaFromString(avro_schema_json);
std::vector<::avro::GenericDatum> avro_data;
for (int i = 0; i < 2; ++i) {
::avro::GenericDatum avro_datum(avro_schema.root());
auto& record = avro_datum.value<::avro::GenericRecord>();
auto& array = record.fieldAt(0).value<::avro::GenericArray>();
for (int j = 0; j < 2; ++j) {
::avro::GenericDatum element(avro_schema.root()->leafAt(0)->leafAt(0));
auto& person_record = element.value<::avro::GenericRecord>();
person_record.fieldAt(0).value<std::string>() =
"Person" + std::to_string(i) + "_" + std::to_string(j);
array.value().push_back(element);
}
avro_data.push_back(avro_datum);
}
const std::string expected_json = R"([
{"people": [
{"name": "Person0_0", "age": null, "email": null},
{"name": "Person0_1", "age": null, "email": null}
]},
{"people": [
{"name": "Person1_0", "age": null, "email": null},
{"name": "Person1_1", "age": null, "email": null}
]}
])";
ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(iceberg_schema, avro_schema.root(),
avro_data, expected_json));
}
struct ExtractDatumParam {
std::string name;
std::shared_ptr<Type> iceberg_type;
std::string arrow_json;
std::function<void(const ::avro::GenericDatum&, int)> value_verifier;
};
void VerifyExtractDatumFromArray(const ExtractDatumParam& test_case) {
Schema iceberg_schema({SchemaField::MakeRequired(
/*field_id=*/1, /*name=*/"a", test_case.iceberg_type)});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
auto arrow_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, test_case.arrow_json)
.ValueOrDie();
for (int64_t i = 0; i < arrow_array->length(); ++i) {
::avro::GenericDatum extracted_datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, i, &extracted_datum), IsOk())
<< "Failed to extract at index " << i;
test_case.value_verifier(extracted_datum, static_cast<int>(i));
}
}
class ExtractDatumFromArrayTest : public ::testing::TestWithParam<ExtractDatumParam> {};
TEST_P(ExtractDatumFromArrayTest, PrimitiveType) {
ASSERT_NO_FATAL_FAILURE(VerifyExtractDatumFromArray(GetParam()));
}
const std::vector<ExtractDatumParam> kExtractDatumTestCases = {
{
.name = "Boolean",
.iceberg_type = boolean(),
.arrow_json = R"([{"a": true}, {"a": false}, {"a": true}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
bool expected = (i % 2 == 0);
EXPECT_EQ(record.fieldAt(0).value<bool>(), expected);
},
},
{
.name = "Int",
.iceberg_type = int32(),
.arrow_json = R"([{"a": 0}, {"a": 100}, {"a": 200}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int32_t>(), i * 100);
},
},
{
.name = "Long",
.iceberg_type = int64(),
.arrow_json = R"([{"a": 0}, {"a": 1000000}, {"a": 2000000}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int64_t>(), i * 1000000LL);
},
},
{
.name = "Float",
.iceberg_type = float32(),
.arrow_json = R"([{"a": 0.0}, {"a": 3.14}, {"a": 6.28}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_FLOAT_EQ(record.fieldAt(0).value<float>(), i * 3.14f);
},
},
{
.name = "Double",
.iceberg_type = float64(),
.arrow_json = R"([{"a": 0.0}, {"a": 1.234567890}, {"a": 2.469135780}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_DOUBLE_EQ(record.fieldAt(0).value<double>(), i * 1.234567890);
},
},
{
.name = "String",
.iceberg_type = string(),
.arrow_json =
R"([{"a": "test_string_0"}, {"a": "test_string_1"}, {"a": "test_string_2"}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
std::string expected = "test_string_" + std::to_string(i);
EXPECT_EQ(record.fieldAt(0).value<std::string>(), expected);
},
},
{
.name = "Binary",
.iceberg_type = binary(),
.arrow_json = R"([{"a": "abc"}, {"a": "bcd"}, {"a": "cde"}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
const auto& bytes = record.fieldAt(0).value<std::vector<uint8_t>>();
EXPECT_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));
},
},
{
.name = "Fixed",
.iceberg_type = fixed(4),
.arrow_json = R"([{"a": "abcd"}, {"a": "bcde"}, {"a": "cdef"}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
const auto& fixed = record.fieldAt(0).value<::avro::GenericFixed>();
EXPECT_EQ(fixed.value().size(), 4);
EXPECT_EQ(static_cast<char>(fixed.value()[0]), static_cast<char>('a' + i));
EXPECT_EQ(static_cast<char>(fixed.value()[1]), static_cast<char>('b' + i));
EXPECT_EQ(static_cast<char>(fixed.value()[2]), static_cast<char>('c' + i));
EXPECT_EQ(static_cast<char>(fixed.value()[3]), static_cast<char>('d' + i));
},
},
{
.name = "Decimal",
.iceberg_type = decimal(10, 2),
.arrow_json = R"([{"a": "0.00"}, {"a": "10.01"}, {"a": "20.02"}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
const auto& fixed = record.fieldAt(0).value<::avro::GenericFixed>();
const auto& bytes = fixed.value();
auto decimal =
::arrow::Decimal128::FromBigEndian(
reinterpret_cast<const uint8_t*>(bytes.data()), bytes.size())
.ValueOrDie();
int64_t expected_unscaled = i * 1000 + i;
EXPECT_EQ(decimal.low_bits(), static_cast<uint64_t>(expected_unscaled));
EXPECT_EQ(decimal.high_bits(), 0);
},
},
{
.name = "Date",
.iceberg_type = date(),
.arrow_json = R"([{"a": 18000}, {"a": 18001}, {"a": 18002}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int32_t>(), 18000 + i);
},
},
{
.name = "Time",
.iceberg_type = time(),
.arrow_json = R"([{"a": 45045123456}, {"a": 45046123456}, {"a": 45047123456}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int64_t>(),
45045123456LL + i * 1000000LL);
},
},
{
.name = "Timestamp",
.iceberg_type = timestamp(),
.arrow_json = R"([{"a": 0}, {"a": 1000000}, {"a": 2000000}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int64_t>(), i * 1000000LL);
},
},
{
.name = "TimestampTz",
.iceberg_type = timestamp_tz(),
.arrow_json =
R"([{"a": 1672531200000000}, {"a": 1672531201000000}, {"a": 1672531202000000}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int64_t>(),
1672531200000000LL + i * 1000000LL);
},
},
{
.name = "TimestampNs",
.iceberg_type = timestamp_ns(),
.arrow_json =
R"([{"a": 1672531200000000000}, {"a": 1672531201000000001}, {"a": 1672531202000000002}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int64_t>(),
1672531200000000000LL + i * 1000000000LL + i);
},
},
{
.name = "TimestampTzNs",
.iceberg_type = timestamptz_ns(),
.arrow_json =
R"([{"a": 1672531200000000000}, {"a": 1672531201000000001}, {"a": 1672531202000000002}])",
.value_verifier =
[](const ::avro::GenericDatum& datum, int i) {
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int64_t>(),
1672531200000000000LL + i * 1000000000LL + i);
},
},
};
INSTANTIATE_TEST_SUITE_P(AllPrimitiveTypes, ExtractDatumFromArrayTest,
::testing::ValuesIn(kExtractDatumTestCases),
[](const ::testing::TestParamInfo<ExtractDatumParam>& info) {
return info.param.name;
});
TEST(ExtractDatumFromArrayTest, StructWithTwoFields) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", int32()),
SchemaField::MakeRequired(2, "name", string()),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
auto arrow_array = ::arrow::json::ArrayFromJSONString(arrow_struct_type,
R"([
{"id": 42, "name": "Alice"},
{"id": 43, "name": "Bob"},
{"id": 44, "name": "Charlie"}
])")
.ValueOrDie();
struct ExpectedData {
int32_t id;
std::string name;
};
std::vector<ExpectedData> expected = {{.id = 42, .name = "Alice"},
{.id = 43, .name = "Bob"},
{.id = 44, .name = "Charlie"}};
auto verify_record = [&](int64_t index, const ExpectedData& expected_data) {
::avro::GenericDatum extracted_datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, index, &extracted_datum), IsOk());
const auto& record = extracted_datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int32_t>(), expected_data.id);
EXPECT_EQ(record.fieldAt(1).value<std::string>(), expected_data.name);
};
for (size_t i = 0; i < expected.size(); ++i) {
verify_record(i, expected[i]);
}
}
TEST(ExtractDatumFromArrayTest, NestedStruct) {
Schema iceberg_schema({
SchemaField::MakeRequired(1, "id", int32()),
SchemaField::MakeRequired(2, "person",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(3, "name", string()),
SchemaField::MakeRequired(4, "age", int32()),
})),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
const std::string arrow_json = R"([
{"id": 1, "person": {"name": "Alice", "age": 25}},
{"id": 2, "person": {"name": "Bob", "age": 30}},
{"id": 3, "person": {"name": "Charlie", "age": 35}}
])";
auto arrow_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, arrow_json).ValueOrDie();
struct ExpectedData {
int32_t id;
std::string name;
int32_t age;
};
std::vector<ExpectedData> expected = {{.id = 1, .name = "Alice", .age = 25},
{.id = 2, .name = "Bob", .age = 30},
{.id = 3, .name = "Charlie", .age = 35}};
auto verify_record = [&](int64_t index, const ExpectedData& expected_data) {
::avro::GenericDatum extracted_datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, index, &extracted_datum), IsOk());
const auto& record = extracted_datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).value<int32_t>(), expected_data.id);
const auto& person_record = record.fieldAt(1).value<::avro::GenericRecord>();
EXPECT_EQ(person_record.fieldAt(0).value<std::string>(), expected_data.name);
EXPECT_EQ(person_record.fieldAt(1).value<int32_t>(), expected_data.age);
};
for (size_t i = 0; i < expected.size(); ++i) {
verify_record(i, expected[i]);
}
}
TEST(ExtractDatumFromArrayTest, ListOfIntegers) {
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());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
const std::string arrow_json = R"([
{"numbers": [10, 11, 12]},
{"numbers": [20, 21]},
{"numbers": [30, 31, 32, 33]}
])";
auto arrow_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, arrow_json).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& array = record.fieldAt(0).value<::avro::GenericArray>();
const auto& elements = array.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, MapStringToInt) {
Schema iceberg_schema({
SchemaField::MakeRequired(
1, "scores",
std::make_shared<MapType>(SchemaField::MakeRequired(2, "key", string()),
SchemaField::MakeRequired(3, "value", int32()))),
});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
const std::string arrow_json = R"([
{"scores": [["alice", 95], ["bob", 87]]},
{"scores": [["charlie", 92], ["diana", 98], ["eve", 89]]},
{"scores": [["frank", 91]]}
])";
auto arrow_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, arrow_json).ValueOrDie();
using MapEntry = std::pair<std::string, int32_t>;
std::vector<std::vector<MapEntry>> expected = {
{{"alice", 95}, {"bob", 87}},
{{"charlie", 92}, {"diana", 98}, {"eve", 89}},
{{"frank", 91}}};
auto verify_record = [&](int64_t index, const std::vector<MapEntry>& expected_entries) {
::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& map = record.fieldAt(0).value<::avro::GenericMap>();
const auto& entries = map.value();
ASSERT_EQ(entries.size(), expected_entries.size());
for (size_t i = 0; i < expected_entries.size(); ++i) {
EXPECT_EQ(entries[i].first, expected_entries[i].first);
EXPECT_EQ(entries[i].second.value<int32_t>(), expected_entries[i].second);
}
};
for (size_t i = 0; i < expected.size(); ++i) {
verify_record(i, expected[i]);
}
}
TEST(ExtractDatumFromArrayTest, ErrorHandling) {
Schema iceberg_schema({SchemaField::MakeRequired(1, "a", int32())});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
auto arrow_array = ::arrow::json::ArrayFromJSONString(
arrow_struct_type, R"([{"a": 1}, {"a": 2}, {"a": 3}])")
.ValueOrDie();
::avro::GenericDatum datum(avro_node);
// Test negative index
EXPECT_THAT(ExtractDatumFromArray(*arrow_array, -1, &datum),
HasErrorMessage("Cannot extract datum from array at index -1"));
// Test index beyond array length
EXPECT_THAT(ExtractDatumFromArray(*arrow_array, 3, &datum),
HasErrorMessage("Cannot extract datum from array at index 3"));
}
TEST(ExtractDatumFromArrayTest, NullHandling) {
Schema iceberg_schema({SchemaField::MakeOptional(1, "a", int32())});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
auto arrow_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, R"([{"a": 42}, {"a": null}])")
.ValueOrDie();
::avro::GenericDatum datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, 0, &datum), IsOk());
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).unionBranch(), 1);
EXPECT_EQ(record.fieldAt(0).type(), ::avro::AVRO_INT);
EXPECT_EQ(record.fieldAt(0).value<int32_t>(), 42);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, 1, &datum), IsOk());
const auto& record2 = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record2.fieldAt(0).unionBranch(), 0);
EXPECT_EQ(record2.fieldAt(0).type(), ::avro::AVRO_NULL);
}
TEST(ExtractDatumFromArrayTest, UnknownType) {
Schema iceberg_schema({SchemaField::MakeOptional(1, "a", unknown())});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
auto arrow_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, R"([{"a": null}])")
.ValueOrDie();
::avro::GenericDatum datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*arrow_array, 0, &datum), IsOk());
const auto& record = datum.value<::avro::GenericRecord>();
EXPECT_EQ(record.fieldAt(0).type(), ::avro::AVRO_NULL);
}
struct RoundTripParam {
std::string name;
std::shared_ptr<Schema> iceberg_schema;
std::string arrow_json;
};
void VerifyRoundTripConversion(const RoundTripParam& test_case) {
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(*test_case.iceberg_schema, &avro_node), IsOk());
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(*test_case.iceberg_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto arrow_struct_type = std::make_shared<::arrow::StructType>(arrow_schema->fields());
auto original_array =
::arrow::json::ArrayFromJSONString(arrow_struct_type, test_case.arrow_json)
.ValueOrDie();
std::vector<::avro::GenericDatum> extracted_data;
for (int64_t i = 0; i < original_array->length(); ++i) {
::avro::GenericDatum datum(avro_node);
ASSERT_THAT(ExtractDatumFromArray(*original_array, i, &datum), IsOk())
<< "Failed to extract datum at index " << i;
extracted_data.push_back(datum);
}
auto projection_result =
Project(*test_case.iceberg_schema, avro_node, /*prune_source=*/false);
ASSERT_THAT(projection_result, IsOk());
auto projection = std::move(projection_result.value());
auto builder = ::arrow::MakeBuilder(arrow_struct_type).ValueOrDie();
for (const auto& datum : extracted_data) {
ASSERT_THAT(
AppendDatumToBuilder(avro_node, datum, projection, *test_case.iceberg_schema,
/*metadata_context=*/{}, builder.get()),
IsOk());
}
auto rebuilt_array = builder->Finish().ValueOrDie();
ASSERT_TRUE(original_array->Equals(*rebuilt_array))
<< "Round-trip consistency failed!\n"
<< "Original array: " << original_array->ToString() << "\n"
<< "Rebuilt array: " << rebuilt_array->ToString();
}
class AvroRoundTripConversionTest : public ::testing::TestWithParam<RoundTripParam> {};
TEST_P(AvroRoundTripConversionTest, ConvertTypes) {
ASSERT_NO_FATAL_FAILURE(VerifyRoundTripConversion(GetParam()));
}
const std::vector<RoundTripParam> kRoundTripTestCases = {
{
.name = "SimpleStruct",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "id", int32()),
SchemaField::MakeRequired(2, "name", string()),
SchemaField::MakeOptional(3, "age", int32()),
}),
.arrow_json = R"([
{"id": 100, "name": "Alice", "age": 25},
{"id": 101, "name": "Bob", "age": null},
{"id": 102, "name": "Charlie", "age": 35}
])",
},
{
.name = "PrimitiveTypes",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "bool_field", boolean()),
SchemaField::MakeRequired(2, "int_field", int32()),
SchemaField::MakeRequired(3, "long_field", int64()),
SchemaField::MakeRequired(4, "float_field", float32()),
SchemaField::MakeRequired(5, "double_field", float64()),
SchemaField::MakeRequired(6, "string_field", string()),
}),
.arrow_json = R"([
{"bool_field": true, "int_field": 42, "long_field": 1000000, "float_field": 3.14, "double_field": 2.718281828, "string_field": "hello"},
{"bool_field": false, "int_field": -42, "long_field": -1000000, "float_field": -3.14, "double_field": -2.718281828, "string_field": "world"}
])",
},
{
.name = "NestedStruct",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "id", int32()),
SchemaField::MakeRequired(
2, "person",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(3, "name", string()),
SchemaField::MakeRequired(4, "age", int32()),
})),
}),
.arrow_json = R"([
{"id": 1, "person": {"name": "Alice", "age": 30}},
{"id": 2, "person": {"name": "Bob", "age": 25}}
])",
},
{
.name = "ListOfIntegers",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(
1, "numbers",
std::make_shared<ListType>(
SchemaField::MakeRequired(2, "element", int32()))),
}),
.arrow_json = R"([
{"numbers": [1, 2, 3]},
{"numbers": [10, 20]},
{"numbers": []}
])",
},
{
.name = "MapStringToInt",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(
1, "scores",
std::make_shared<MapType>(
SchemaField::MakeRequired(2, "key", string()),
SchemaField::MakeRequired(3, "value", int32()))),
}),
.arrow_json = R"([
{"scores": [["alice", 95], ["bob", 87]]},
{"scores": [["charlie", 92]]},
{"scores": []}
])",
},
{
.name = "ComplexNested",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(
1, "data",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(2, "id", int32()),
SchemaField::MakeRequired(
3, "tags",
std::make_shared<ListType>(
SchemaField::MakeRequired(4, "element", string()))),
SchemaField::MakeOptional(
5, "metadata",
std::make_shared<MapType>(
SchemaField::MakeRequired(6, "key", string()),
SchemaField::MakeRequired(7, "value", string()))),
})),
}),
.arrow_json = R"([
{"data": {"id": 1, "tags": ["tag1", "tag2"], "metadata": [["key1", "value1"]]}},
{"data": {"id": 2, "tags": [], "metadata": null}}
])",
},
{
.name = "NullablePrimitives",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeOptional(1, "optional_bool", boolean()),
SchemaField::MakeOptional(2, "optional_int", int32()),
SchemaField::MakeOptional(3, "optional_long", int64()),
SchemaField::MakeOptional(4, "optional_string", string()),
SchemaField::MakeRequired(5, "required_id", int32()),
}),
.arrow_json = R"([
{"optional_bool": true, "optional_int": 42, "optional_long": 1000000, "optional_string": "hello", "required_id": 1},
{"optional_bool": null, "optional_int": null, "optional_long": null, "optional_string": null, "required_id": 2},
{"optional_bool": false, "optional_int": null, "optional_long": 2000000, "optional_string": null, "required_id": 3},
{"optional_bool": null, "optional_int": 123, "optional_long": null, "optional_string": "world", "required_id": 4}
])",
},
{
.name = "NullableNestedStruct",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "id", int32()),
SchemaField::MakeOptional(
2, "person",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(3, "name", string()),
SchemaField::MakeOptional(4, "age", int32()),
SchemaField::MakeOptional(5, "email", string()),
})),
SchemaField::MakeOptional(6, "department", string()),
}),
.arrow_json = R"([
{"id": 1, "person": {"name": "Alice", "age": 30, "email": "alice@example.com"}, "department": "Engineering"},
{"id": 2, "person": null, "department": null},
{"id": 3, "person": {"name": "Bob", "age": null, "email": null}, "department": "Sales"},
{"id": 4, "person": {"name": "Charlie", "age": 25, "email": null}, "department": null}
])",
},
{
.name = "NullableListElements",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "id", int32()),
SchemaField::MakeOptional(
2, "numbers",
std::make_shared<ListType>(
SchemaField::MakeOptional(3, "element", int32()))),
SchemaField::MakeRequired(
4, "tags",
std::make_shared<ListType>(
SchemaField::MakeOptional(5, "element", string()))),
}),
.arrow_json = R"([
{"id": 1, "numbers": [1, null, 3], "tags": ["tag1", null, "tag3"]},
{"id": 2, "numbers": null, "tags": ["only_tag"]},
{"id": 3, "numbers": [null, null], "tags": [null, null, null]},
{"id": 4, "numbers": [], "tags": []}
])",
},
{
.name = "NullableMapValues",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(1, "id", int32()),
SchemaField::MakeOptional(
2, "scores",
std::make_shared<MapType>(
SchemaField::MakeRequired(3, "key", string()),
SchemaField::MakeOptional(4, "value", int32()))),
SchemaField::MakeRequired(
5, "metadata",
std::make_shared<MapType>(
SchemaField::MakeRequired(6, "key", string()),
SchemaField::MakeOptional(7, "value", string()))),
}),
.arrow_json = R"([
{"id": 1, "scores": [["alice", 95], ["bob", null]], "metadata": [["key1", "value1"], ["key2", null]]},
{"id": 2, "scores": null, "metadata": [["key3", null]]},
{"id": 3, "scores": [["charlie", null], ["diana", 98]], "metadata": []},
{"id": 4, "scores": [], "metadata": [["key4", null], ["key5", "value5"]]}
])",
},
{
.name = "DeeplyNestedWithNulls",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeRequired(
1, "root",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeRequired(2, "id", int32()),
SchemaField::MakeOptional(
3, "nested",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeOptional(4, "name", string()),
SchemaField::MakeOptional(
5, "values",
std::make_shared<ListType>(
SchemaField::MakeOptional(6, "element", int32()))),
})),
SchemaField::MakeOptional(
7, "tags",
std::make_shared<ListType>(
SchemaField::MakeOptional(8, "element", string()))),
})),
}),
.arrow_json = R"([
{"root": {"id": 1, "nested": {"name": "test", "values": [1, null, 3]}, "tags": ["a", "b"]}},
{"root": {"id": 2, "nested": null, "tags": null}},
{"root": {"id": 3, "nested": {"name": null, "values": null}, "tags": [null, "c"]}},
{"root": {"id": 4, "nested": {"name": "empty", "values": []}, "tags": []}}
])",
},
{
.name = "AllNullsVariations",
.iceberg_schema = std::make_shared<Schema>(std::vector<SchemaField>{
SchemaField::MakeOptional(1, "always_null", string()),
SchemaField::MakeOptional(2, "sometimes_null", int32()),
SchemaField::MakeOptional(
3, "nested_struct",
std::make_shared<StructType>(std::vector<SchemaField>{
SchemaField::MakeOptional(4, "inner_null", string()),
SchemaField::MakeRequired(5, "inner_required", boolean()),
})),
SchemaField::MakeRequired(6, "id", int32()),
}),
.arrow_json = R"([
{"always_null": null, "sometimes_null": 42, "nested_struct": {"inner_null": "value", "inner_required": true}, "id": 1},
{"always_null": null, "sometimes_null": null, "nested_struct": null, "id": 2},
{"always_null": null, "sometimes_null": 123, "nested_struct": {"inner_null": null, "inner_required": false}, "id": 3},
{"always_null": null, "sometimes_null": null, "nested_struct": {"inner_null": null, "inner_required": true}, "id": 4}
])",
},
};
TEST(DirectDecoderTest, ProjectionCacheKeyedPerAvroNodeForEmptyStructs) {
auto make_struct = [](int32_t first_id, int count) {
std::vector<SchemaField> fields;
for (int i = 0; i < count; ++i) {
fields.push_back(SchemaField::MakeRequired(first_id + i, "f" + std::to_string(i),
std::make_shared<IntType>()));
}
return std::make_shared<StructType>(std::move(fields));
};
Schema file_schema({SchemaField::MakeRequired(1, "narrow", make_struct(10, 1)),
SchemaField::MakeRequired(2, "wide", make_struct(20, 6))});
Schema read_schema({SchemaField::MakeRequired(1, "narrow", make_struct(10, 0)),
SchemaField::MakeRequired(2, "wide", make_struct(20, 0))});
::avro::NodePtr avro_node;
ASSERT_THAT(ToAvroNodeVisitor{}.Visit(file_schema, &avro_node), IsOk());
ICEBERG_UNWRAP_OR_FAIL(auto projection,
Project(read_schema, avro_node, /*prune_source=*/false));
auto out = ::avro::memoryOutputStream();
auto encoder = ::avro::binaryEncoder();
encoder->init(*out);
for (int i = 0; i < 7; ++i) {
encoder->encodeInt(i);
}
encoder->flush();
auto in = ::avro::memoryInputStream(*out);
auto decoder = ::avro::binaryDecoder();
decoder->init(*in);
ArrowSchema arrow_c_schema;
ASSERT_THAT(ToArrowSchema(read_schema, &arrow_c_schema), IsOk());
auto arrow_schema = ::arrow::ImportSchema(&arrow_c_schema).ValueOrDie();
auto builder =
::arrow::MakeBuilder(std::make_shared<::arrow::StructType>(arrow_schema->fields()))
.ValueOrDie();
DecodeContext ctx;
ASSERT_THAT(DecodeAvroToBuilder(avro_node, *decoder, projection, read_schema,
/*metadata_context=*/{}, builder.get(), ctx),
IsOk());
std::vector<size_t> mapping_sizes;
for (const auto& [key, mapping] : ctx.avro_to_projection_cache) {
mapping_sizes.push_back(mapping.size());
}
EXPECT_THAT(mapping_sizes, ::testing::UnorderedElementsAre(1, 6, 2));
}
INSTANTIATE_TEST_SUITE_P(AllTypes, AvroRoundTripConversionTest,
::testing::ValuesIn(kRoundTripTestCases),
[](const ::testing::TestParamInfo<RoundTripParam>& info) {
return info.param.name;
});
} // namespace iceberg::avro