fix build error and set name_mapping to ReaderOptions

MisterRaindrop · MisterRaindrop · commit 883255b56a4c · 2025-06-29T17:22:05.000+08:00
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
@@ -98,8 +98,7 @@ if(ICEBERG_BUILD_BUNDLE)
       avro/avro_data_util.cc
       avro/avro_reader.cc
       avro/avro_schema_util.cc
-      avro/avro_stream_internal.cc
-      json_internal.cc)
+      avro/avro_stream_internal.cc)
 
   # Libraries to link with exported libiceberg_bundle.{so,a}.
   set(ICEBERG_BUNDLE_STATIC_BUILD_INTERFACE_LIBS)
diff --git a/src/iceberg/avro/avro_constants.h b/src/iceberg/avro/avro_constants.h
@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <string_view>
+
+namespace iceberg::avro {
+
+// Avro field ID attribute constants
+constexpr std::string_view kFieldId = "field-id";
+constexpr std::string_view kElementId = "element-id";
+constexpr std::string_view kKeyId = "key-id";
+constexpr std::string_view kValueId = "value-id";
+
+// Avro logical type constants
+constexpr std::string_view kMapLogicalType = "map";
+
+// Name mapping field constants
+constexpr std::string_view kElement = "element";
+constexpr std::string_view kKey = "key";
+constexpr std::string_view kValue = "value";
+
+}  // namespace iceberg::avro
diff --git a/src/iceberg/avro/avro_reader.cc b/src/iceberg/avro/avro_reader.cc
@@ -33,10 +33,10 @@
 #include <nlohmann/json.hpp>
 
 #include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/avro/avro_constants.h"
 #include "iceberg/avro/avro_data_util_internal.h"
 #include "iceberg/avro/avro_schema_util_internal.h"
 #include "iceberg/avro/avro_stream_internal.h"
-#include "iceberg/json_internal.h"
 #include "iceberg/name_mapping.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/util/checked_cast.h"
@@ -102,14 +102,9 @@ class AvroBatchReader::Impl {
 
     if (has_id_visitor.HasNoIds()) {
       // Apply field IDs based on name mapping if available
-      auto name_mapping_iter = options.properties.find("name_mapping");
-      if (name_mapping_iter != options.properties.end()) {
-        // Parse name mapping from JSON string
-        ICEBERG_ASSIGN_OR_RAISE(auto name_mapping,
-                                iceberg::NameMappingFromJson(
-                                    nlohmann::json::parse(name_mapping_iter->second)));
-        ICEBERG_RETURN_UNEXPECTED(
-            ApplyFieldIdsFromNameMapping(file_schema.root(), *name_mapping));
+      if (options.name_mapping) {
+        ICEBERG_RETURN_UNEXPECTED(ApplyFieldIdsFromNameMapping(*options.name_mapping,
+                                                               file_schema.root().get()));
       } else {
         return NotImplemented(
             "Avro file schema has no field IDs and no name mapping provided");
@@ -210,8 +205,8 @@ class AvroBatchReader::Impl {
   }
 
   // Apply field IDs to Avro schema nodes based on name mapping
-  Status ApplyFieldIdsFromNameMapping(const ::avro::NodePtr& node,
-                                      const NameMapping& name_mapping) {
+  Status ApplyFieldIdsFromNameMapping(const NameMapping& name_mapping,
+                                      ::avro::Node* node) {
     switch (node->type()) {
       case ::avro::AVRO_RECORD:
         return ApplyFieldIdsToRecord(node, name_mapping);
@@ -238,19 +233,19 @@ class AvroBatchReader::Impl {
     }
   }
 
-  Status ApplyFieldIdsToRecord(const ::avro::NodePtr& node,
-                               const NameMapping& name_mapping) {
+  Status ApplyFieldIdsToRecord(::avro::Node* node, const NameMapping& name_mapping) {
     for (size_t i = 0; i < node->leaves(); ++i) {
       const std::string& field_name = node->nameAt(i);
-      ::avro::NodePtr field_node = node->leafAt(i);
+      ::avro::Node* field_node = node->leafAt(i).get();
 
       // Try to find field ID by name in the name mapping
       if (auto field_ref = name_mapping.Find(field_name)) {
         if (field_ref->get().field_id.has_value()) {
           // Add field ID attribute to the node
           ::avro::CustomAttributes attributes;
-          attributes.addAttribute(
-              "field-id", std::to_string(field_ref->get().field_id.value()), false);
+          attributes.addAttribute(std::string(kFieldId),
+                                  std::to_string(field_ref->get().field_id.value()),
+                                  false);
           node->addCustomAttributesForField(attributes);
         }
 
@@ -262,92 +257,101 @@ class AvroBatchReader::Impl {
           std::vector<MappedField> fields_vector(fields_span.begin(), fields_span.end());
           auto nested_name_mapping = NameMapping::Make(std::move(fields_vector));
           ICEBERG_RETURN_UNEXPECTED(
-              ApplyFieldIdsFromNameMapping(field_node, *nested_name_mapping));
+              ApplyFieldIdsFromNameMapping(*nested_name_mapping, field_node));
         } else {
           // Recursively apply field IDs to child nodes (only if not already handled by
           // nested mapping)
           ICEBERG_RETURN_UNEXPECTED(
-              ApplyFieldIdsFromNameMapping(field_node, name_mapping));
+              ApplyFieldIdsFromNameMapping(name_mapping, field_node));
         }
       } else {
         // Recursively apply field IDs to child nodes even if no mapping found
-        ICEBERG_RETURN_UNEXPECTED(ApplyFieldIdsFromNameMapping(field_node, name_mapping));
+        ICEBERG_RETURN_UNEXPECTED(ApplyFieldIdsFromNameMapping(name_mapping, field_node));
       }
     }
     return {};
   }
 
-  Status ApplyFieldIdsToArray(const ::avro::NodePtr& node,
-                              const NameMapping& name_mapping) {
+  Status ApplyFieldIdsToArray(::avro::Node* node, const NameMapping& name_mapping) {
+    // TODO(liuxiaoyu): Add debug logging to print node information for troubleshooting
+    // when array type validation fails
     if (node->leaves() != 1) {
       return InvalidSchema("Array type must have exactly one leaf");
     }
 
     // Check if this is a map represented as array
     if (node->logicalType().type() == ::avro::LogicalType::CUSTOM &&
         node->logicalType().customLogicalType() != nullptr &&
-        node->logicalType().customLogicalType()->name() == "map") {
-      return ApplyFieldIdsFromNameMapping(node->leafAt(0), name_mapping);
+        node->logicalType().customLogicalType()->name() == kMapLogicalType) {
+      return ApplyFieldIdsFromNameMapping(name_mapping, node->leafAt(0).get());
     }
 
     // For regular arrays, try to find element field ID
-    if (auto element_field = name_mapping.Find("element")) {
+    if (auto element_field = name_mapping.Find(std::string(kElement))) {
       if (element_field->get().field_id.has_value()) {
         ::avro::CustomAttributes attributes;
-        attributes.addAttribute(
-            "element-id", std::to_string(element_field->get().field_id.value()), false);
+        attributes.addAttribute(std::string(kElementId),
+                                std::to_string(element_field->get().field_id.value()),
+                                false);
         node->addCustomAttributesForField(attributes);
       }
     }
 
-    return ApplyFieldIdsFromNameMapping(node->leafAt(0), name_mapping);
+    return ApplyFieldIdsFromNameMapping(name_mapping, node->leafAt(0).get());
   }
 
-  Status ApplyFieldIdsToMap(const ::avro::NodePtr& node,
-                            const NameMapping& name_mapping) {
+  Status ApplyFieldIdsToMap(::avro::Node* node, const NameMapping& name_mapping) {
     if (node->leaves() != 2) {
       return InvalidSchema("Map type must have exactly two leaves");
     }
 
     // Try to find key and value field IDs
-    if (auto key_field = name_mapping.Find("key")) {
+    if (auto key_field = name_mapping.Find(std::string(kKey))) {
       if (key_field->get().field_id.has_value()) {
         ::avro::CustomAttributes attributes;
-        attributes.addAttribute("key-id",
+        attributes.addAttribute(std::string(kKeyId),
                                 std::to_string(key_field->get().field_id.value()), false);
         node->addCustomAttributesForField(attributes);
       }
     }
 
-    if (auto value_field = name_mapping.Find("value")) {
+    if (auto value_field = name_mapping.Find(std::string(kValue))) {
       if (value_field->get().field_id.has_value()) {
         ::avro::CustomAttributes attributes;
-        attributes.addAttribute(
-            "value-id", std::to_string(value_field->get().field_id.value()), false);
+        attributes.addAttribute(std::string(kValueId),
+                                std::to_string(value_field->get().field_id.value()),
+                                false);
         node->addCustomAttributesForField(attributes);
       }
     }
 
-    return ApplyFieldIdsFromNameMapping(node->leafAt(1), name_mapping);
+    return ApplyFieldIdsFromNameMapping(name_mapping, node->leafAt(1).get());
   }
 
-  Status ApplyFieldIdsToUnion(const ::avro::NodePtr& node,
-                              const NameMapping& name_mapping) {
+  Status ApplyFieldIdsToUnion(::avro::Node* node, const NameMapping& name_mapping) {
     if (node->leaves() != 2) {
       return InvalidSchema("Union type must have exactly two branches");
     }
 
     const auto& branch_0 = node->leafAt(0);
     const auto& branch_1 = node->leafAt(1);
 
-    if (branch_0->type() == ::avro::AVRO_NULL) {
-      return ApplyFieldIdsFromNameMapping(branch_1, name_mapping);
+    bool branch_0_is_null = (branch_0->type() == ::avro::AVRO_NULL);
+    bool branch_1_is_null = (branch_1->type() == ::avro::AVRO_NULL);
+
+    if (branch_0_is_null && !branch_1_is_null) {
+      // branch_0 is null, branch_1 is not null
+      return ApplyFieldIdsFromNameMapping(name_mapping, branch_1.get());
+    } else if (!branch_0_is_null && branch_1_is_null) {
+      // branch_0 is not null, branch_1 is null
+      return ApplyFieldIdsFromNameMapping(name_mapping, branch_0.get());
+    } else if (branch_0_is_null && branch_1_is_null) {
+      // Both branches are null - this is invalid
+      return InvalidSchema("Union type cannot have two null branches");
+    } else {
+      // Neither branch is null - this is invalid
+      return InvalidSchema("Union type must have exactly one null branch");
     }
-    if (branch_1->type() == ::avro::AVRO_NULL) {
-      return ApplyFieldIdsFromNameMapping(branch_0, name_mapping);
-    }
-
-    return InvalidSchema("Union type must have exactly one null branch");
   }
 
  private:
diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h
@@ -29,6 +29,7 @@
 
 #include "iceberg/arrow_c_data.h"
 #include "iceberg/file_format.h"
+#include "iceberg/name_mapping.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
 
@@ -140,6 +141,9 @@ struct ICEBERG_EXPORT ReaderOptions {
   /// \brief The filter to apply to the data. Reader implementations may ignore this if
   /// the file format does not support filtering.
   std::shared_ptr<class Expression> filter;
+  /// \brief Name mapping for schema evolution compatibility. Used when reading files
+  /// that may have different field names than the current schema.
+  std::shared_ptr<class NameMapping> name_mapping;
   /// \brief Format-specific or implementation-specific properties.
   std::unordered_map<std::string, std::string> properties;
 };
diff --git a/test/avro_reader_test.cc b/test/avro_reader_test.cc
@@ -173,4 +173,25 @@ TEST_F(AvroReaderTest, NameMappingWithNestedFields) {
               testing::UnorderedElementsAre("description"));
 }
 
+TEST_F(AvroReaderTest, NameMappingFromReaderOptionsWorks) {
+  // Create a name mapping
+  auto name_mapping = CreateTestNameMapping();
+  ASSERT_TRUE(name_mapping != nullptr);
+  EXPECT_EQ(name_mapping->AsMappedFields().Size(), 3);
+
+  // Create reader options with name mapping
+  ReaderOptions options;
+  options.name_mapping = std::move(name_mapping);
+
+  // Verify that the name mapping is accessible
+  ASSERT_TRUE(options.name_mapping != nullptr);
+  EXPECT_EQ(options.name_mapping->AsMappedFields().Size(), 3);
+
+  // Test that the name mapping works correctly
+  auto field_by_id = options.name_mapping->Find(1);
+  ASSERT_TRUE(field_by_id.has_value());
+  EXPECT_EQ(field_by_id->get().field_id, 1);
+  EXPECT_THAT(field_by_id->get().names, testing::UnorderedElementsAre("id"));
+}
+
 }  // namespace iceberg::avro
