feat: add support for avro to arrow data conversion (#124)

Preliminary avro datum to arrow array data conversion support:
- Support projecting all primitive and nested types.
- Support missing fields (reading into null values).
- Support int->long and float->double promotion.
- Add test cases for all primitive and nested types.

Author: wgtmac

cmake_modules/IcebergThirdpartyToolchain.cmake

@@ -71,6 +71,9 @@ function(resolve_arrow_dependency)
   set(ARROW_FILESYSTEM
       ON
       CACHE BOOL "" FORCE)
+  set(ARROW_JSON
+      ON
+      CACHE BOOL "" FORCE)
   set(ARROW_PARQUET
       ON
       CACHE BOOL "" FORCE)
@@ -95,8 +98,8 @@ function(resolve_arrow_dependency)
 
   fetchcontent_declare(VendoredArrow
                        ${FC_DECLARE_COMMON_OPTIONS}
-                       GIT_REPOSITORY https://github.com/wgtmac/arrow.git
-                       GIT_TAG 7d50c4ac803ad983734de5f418b7cd18f25b0dc9
+                       GIT_REPOSITORY https://github.com/apache/arrow.git
+                       GIT_TAG 5f0aeb5de53fb25b59a52661a80071faef99a4a4
                        #URL ${ARROW_SOURCE_URL}
                        #URL_HASH "SHA256=${ICEBERG_ARROW_BUILD_SHA256_CHECKSUM}"
                        SOURCE_SUBDIR

src/iceberg/avro/avro_data_util.cc

@@ -26,10 +26,21 @@
 
 namespace iceberg::avro {
 
+/// \brief Append an Avro datum to an Arrow array builder.
+///
+/// This function handles schema evolution by using the provided projection to map
+/// fields from the Avro data to the expected Arrow schema.
+///
+/// \param avro_node The Avro schema node (must be a record at root level)
+/// \param avro_datum The Avro data to append
+/// \param projection Schema projection from `projected_schema` to `avro_node`
+/// \param projected_schema The projected schema
+/// \param array_builder The Arrow array builder to append to (must be a struct builder)
+/// \return Status indicating success or failure
 Status AppendDatumToBuilder(const ::avro::NodePtr& avro_node,
                             const ::avro::GenericDatum& avro_datum,
                             const SchemaProjection& projection,
-                            const Schema& arrow_schema,
+                            const Schema& projected_schema,
                             ::arrow::ArrayBuilder* array_builder);
 
 }  // namespace iceberg::avro

src/iceberg/avro/avro_data_util_internal.h

@@ -73,6 +73,22 @@ ::avro::CustomAttributes GetAttributesWithFieldId(int32_t field_id) {
 
 }  // namespace
 
+std::string ToString(const ::avro::NodePtr& node) {
+  std::stringstream ss;
+  ss << *node;
+  return ss.str();
+}
+
+std::string ToString(const ::avro::LogicalType& logical_type) {
+  std::stringstream ss;
+  logical_type.printJson(ss);
+  return ss.str();
+}
+
+std::string ToString(const ::avro::LogicalType::Type& logical_type) {
+  return ToString(::avro::LogicalType(logical_type));
+}
+
 Status ToAvroNodeVisitor::Visit(const BooleanType& type, ::avro::NodePtr* node) {
   *node = std::make_shared<::avro::NodePrimitive>(::avro::AVRO_BOOL);
   return {};
@@ -383,33 +399,11 @@ Status HasIdVisitor::Visit(const ::avro::Schema& schema) { return Visit(schema.r
 
 namespace {
 
-std::string ToString(const ::avro::NodePtr& node) {
-  std::stringstream ss;
-  ss << *node;
-  return ss.str();
-}
-
-std::string ToString(const ::avro::LogicalType& logical_type) {
-  std::stringstream ss;
-  logical_type.printJson(ss);
-  return ss.str();
-}
-
-std::string ToString(const ::avro::LogicalType::Type& logical_type) {
-  return ToString(::avro::LogicalType(logical_type));
-}
-
 bool HasLogicalType(const ::avro::NodePtr& node,
                     ::avro::LogicalType::Type expected_type) {
   return node->logicalType().type() == expected_type;
 }
 
-bool HasMapLogicalType(const ::avro::NodePtr& node) {
-  return node->logicalType().type() == ::avro::LogicalType::CUSTOM &&
-         node->logicalType().customLogicalType() != nullptr &&
-         node->logicalType().customLogicalType()->name() == "map";
-}
-
 std::optional<std::string> GetAdjustToUtc(const ::avro::NodePtr& node) {
   if (node->customAttributes() == 0) {
     return std::nullopt;
@@ -501,7 +495,7 @@ Status ValidateAvroSchemaEvolution(const Type& expected_type,
     case TypeId::kTimestamp:
       if (avro_node->type() == ::avro::AVRO_LONG &&
           HasLogicalType(avro_node, ::avro::LogicalType::TIMESTAMP_MICROS) &&
-          GetAdjustToUtc(avro_node).value_or("false") == "true") {
+          GetAdjustToUtc(avro_node).value_or("false") == "false") {
         return {};
       }
       break;
@@ -676,6 +670,10 @@ Result<FieldProjection> ProjectList(const ListType& list_type,
         ValidateAvroSchemaEvolution(*expected_element_field.type(), element_node));
   }
 
+  // Set the element projection metadata but preserve its children
+  element_projection.kind = FieldProjection::Kind::kProjected;
+  element_projection.from = FieldProjection::SourceFieldIndex{0};
+
   FieldProjection result;
   result.children.emplace_back(std::move(element_projection));
   return result;
@@ -771,6 +769,12 @@ Result<FieldProjection> ProjectNested(const Type& expected_type,
 
 }  // namespace
 
+bool HasMapLogicalType(const ::avro::NodePtr& node) {
+  return node->logicalType().type() == ::avro::LogicalType::CUSTOM &&
+         node->logicalType().customLogicalType() != nullptr &&
+         node->logicalType().customLogicalType()->name() == "map";
+}
+
 Result<SchemaProjection> Project(const Schema& expected_schema,
                                  const ::avro::NodePtr& avro_node, bool prune_source) {
   ICEBERG_ASSIGN_OR_RAISE(

src/iceberg/avro/avro_schema_util.cc

@@ -135,4 +135,13 @@ class HasIdVisitor {
 Result<SchemaProjection> Project(const Schema& expected_schema,
                                  const ::avro::NodePtr& avro_node, bool prune_source);
 
+std::string ToString(const ::avro::NodePtr& node);
+std::string ToString(const ::avro::LogicalType& logical_type);
+std::string ToString(const ::avro::LogicalType::Type& logical_type);
+
+/// \brief Check if an Avro node has a map logical type.
+/// \param node The Avro node to check.
+/// \return True if the node has a map logical type, false otherwise.
+bool HasMapLogicalType(const ::avro::NodePtr& node);
+
 }  // namespace iceberg::avro

src/iceberg/avro/avro_schema_util_internal.h

@@ -26,7 +26,7 @@
 namespace iceberg {
 
 // Fix `from` field of `FieldProjection` to use pruned field index.
-void PruneFieldProjection(FieldProjection& field_projection) {
+inline void PruneFieldProjection(FieldProjection& field_projection) {
   std::map<size_t, size_t> local_index_to_pruned_index;
   for (const auto& child_projection : field_projection.children) {
     if (child_projection.kind == FieldProjection::Kind::kProjected) {

src/iceberg/schema_util_internal.h

@@ -78,7 +78,8 @@ add_test(NAME util_test COMMAND util_test)
 
 if(ICEBERG_BUILD_BUNDLE)
   add_executable(avro_test)
-  target_sources(avro_test PRIVATE avro_test.cc avro_schema_test.cc avro_stream_test.cc)
+  target_sources(avro_test PRIVATE avro_data_test.cc avro_test.cc avro_schema_test.cc
+                                   avro_stream_test.cc)
   target_link_libraries(avro_test PRIVATE iceberg_bundle_static GTest::gtest_main
                                           GTest::gmock)
   add_test(NAME avro_test COMMAND avro_test)