feat: add support for avro to arrow data conversion

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,22 +399,6 @@ 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;

src/iceberg/avro/avro_schema_util.cc

@@ -135,4 +135,8 @@ 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);
+
 }  // 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) {
+static 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

@@ -71,7 +71,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)

test/CMakeLists.txt

@@ -0,0 +1,108 @@
+/*
+ * 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 <arrow/c/bridge.h>
+#include <arrow/json/from_string.h>
+#include <avro/Compiler.hh>
+#include <avro/Generic.hh>
+#include <avro/Node.hh>
+#include <avro/Types.hh>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <iceberg/arrow/arrow_error_transform_internal.h>
+#include <iceberg/avro/avro_data_util_internal.h>
+#include <iceberg/avro/avro_schema_util_internal.h>
+#include <iceberg/schema.h>
+#include <iceberg/schema_internal.h>
+#include <iceberg/schema_util.h>
+#include <iceberg/type.h>
+#include <iceberg/util/macros.h>
+
+#include "matchers.h"
+
+namespace iceberg::avro {
+
+void VerifyAppendDatumToBuilder(const Schema& iceberg_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(iceberg_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(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 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, iceberg_schema,
+                                     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));
+}
+
+TEST(AvroDataTest, AppendIntDatumToBuilder) {
+  Schema iceberg_schema(
+      {SchemaField::MakeRequired(/*field_id=*/1, "a", std::make_shared<IntType>())});
+  ::avro::NodePtr avro_node;
+  EXPECT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
+
+  std::vector<::avro::GenericDatum> avro_data;
+  for (int i = 0; i < 3; ++i) {
+    ::avro::GenericDatum avro_datum(avro_node);
+    avro_datum.value<::avro::GenericRecord>().fieldAt(0).value<int32_t>() = i;
+    avro_data.push_back(avro_datum);
+  }
+
+  ASSERT_NO_FATAL_FAILURE(VerifyAppendDatumToBuilder(
+      iceberg_schema, avro_node, avro_data, R"([{"a": 0}, {"a": 1}, {"a": 2}])"));
+}
+
+TEST(AvroDataTest, AppendStringDatumToBuilder) {
+  Schema iceberg_schema(
+      {SchemaField::MakeRequired(/*field_id=*/1, "a", std::make_shared<StringType>())});
+  ::avro::NodePtr avro_node;
+  EXPECT_THAT(ToAvroNodeVisitor{}.Visit(iceberg_schema, &avro_node), IsOk());
+
+  std::vector<::avro::GenericDatum> avro_data;
+  for (int i = 0; i < 3; ++i) {
+    ::avro::GenericDatum avro_datum(avro_node);
+    avro_datum.value<::avro::GenericRecord>().fieldAt(0).value<std::string>() =
+        "a" + std::to_string(i);
+    avro_data.push_back(avro_datum);
+  }
+
+  ASSERT_NO_FATAL_FAILURE(
+      VerifyAppendDatumToBuilder(iceberg_schema, avro_node, avro_data,
+                                 R"([{"a": "a0"}, {"a": "a1"}, {"a": "a2"}])"));
+}
+
+}  // namespace iceberg::avro