Add test and fix GetFieldId

Author: MisterRaindrop

src/iceberg/avro/avro_schema_util.cc

@@ -63,25 +63,26 @@ ::avro::CustomAttributes GetAttributesWithFieldId(int32_t field_id) {
   return attributes;
 }
 
+}  // namespace
+
 std::string SanitizeFieldName(std::string_view field_name) {
   if (field_name.empty()) {
     return "_empty";
   }
 
   std::string result;
-  result.reserve(field_name.size());
+  result.reserve(field_name.size() + 1);  // +1 for potential leading underscore
 
   // First character must be a letter or underscore
-  if (!std::isalpha(field_name[0]) && field_name[0] != '_') {
+  char first_char = field_name[0];
+  if (!std::isalpha(static_cast<unsigned char>(first_char)) && first_char != '_') {
     result.push_back('_');
-  } else {
-    result.push_back(field_name[0]);
   }
 
-  // Rest of characters must be letters, digits, or underscores
-  for (size_t i = 1; i < field_name.size(); ++i) {
+  // Process all characters
+  for (size_t i = 0; i < field_name.size(); ++i) {
     char c = field_name[i];
-    if (std::isalnum(c) || c == '_') {
+    if (std::isalnum(static_cast<unsigned char>(c)) || c == '_') {
       result.push_back(c);
     } else {
       result.push_back('_');
@@ -90,8 +91,6 @@ std::string SanitizeFieldName(std::string_view field_name) {
   return result;
 }
 
-}  // namespace
-
 std::string ToString(const ::avro::NodePtr& node) {
   std::stringstream ss;
   ss << *node;
@@ -217,14 +216,16 @@ Status ToAvroNodeVisitor::Visit(const StructType& type, ::avro::NodePtr* node) {
 
     // Sanitize field name to ensure it follows Avro field name requirements
     std::string sanitized_name = SanitizeFieldName(sub_field.name());
+
     // Store original name as a custom attribute if it was modified
     if (sanitized_name != sub_field.name()) {
       // Add custom attribute to preserve the original field name
-      ::avro::CustomAttributes attrs;
-      attrs.addAttribute(std::string(kIcebergFieldNameProp),
-                         std::string(sub_field.name()));
-      (*node)->addCustomAttributesForField(attrs);
+      ::avro::CustomAttributes name_attrs;
+      name_attrs.addAttribute(std::string(kIcebergFieldNameProp),
+                              std::string(sub_field.name()));
+      (*node)->addCustomAttributesForField(name_attrs);
     }
+
     (*node)->addName(sanitized_name);
     (*node)->addLeaf(field_node);
     (*node)->addCustomAttributesForField(GetAttributesWithFieldId(sub_field.field_id()));
@@ -475,7 +476,50 @@ Result<int32_t> GetValueId(const ::avro::NodePtr& node) {
 
 Result<int32_t> GetFieldId(const ::avro::NodePtr& node, size_t field_idx) {
   static const std::string kFieldIdKey{kFieldIdProp};
-  return GetId(node, kFieldIdKey, field_idx);
+
+  // When field names are sanitized, we add custom attributes in this order for each field:
+  // 1. If the field name was sanitized: iceberg-field-name attribute
+  // 2. Always: field-id attribute
+  // So for field i, we need to find the correct attribute index containing field-id
+
+  size_t attribute_search_start = 0;
+  for (size_t field = 0; field <= field_idx; ++field) {
+    if (field == field_idx) {
+      // For the target field, search for field-id in the remaining attributes
+      for (size_t attr_idx = attribute_search_start; attr_idx < node->customAttributes(); ++attr_idx) {
+        auto id_str = node->customAttributesAt(attr_idx).getAttribute(kFieldIdKey);
+        if (id_str.has_value()) {
+          try {
+            return std::stoi(id_str.value());
+          } catch (const std::exception& e) {
+            return InvalidSchema("Invalid {}: {}", kFieldIdKey, id_str.value());
+          }
+        }
+        // If this attribute doesn't have field-id, move to next
+        auto name_attr = node->customAttributesAt(attr_idx).getAttribute(std::string(kIcebergFieldNameProp));
+        if (name_attr.has_value()) {
+          // This is a name attribute, the next one should be field-id
+          continue;
+        }
+      }
+      break;
+    } else {
+      // For previous fields, count how many attributes they used
+      // Check if this field has a name attribute (means the field name was sanitized)
+      if (attribute_search_start < node->customAttributes()) {
+        auto name_attr = node->customAttributesAt(attribute_search_start).getAttribute(std::string(kIcebergFieldNameProp));
+        if (name_attr.has_value()) {
+          // This field has both name and id attributes
+          attribute_search_start += 2;
+        } else {
+          // This field only has id attribute
+          attribute_search_start += 1;
+        }
+      }
+    }
+  }
+
+  return InvalidSchema("Missing avro attribute: {}", kFieldIdKey);
 }
 
 Status ValidateAvroSchemaEvolution(const Type& expected_type,

src/iceberg/avro/avro_schema_util_internal.h

@@ -163,4 +163,9 @@ Result<::avro::NodePtr> MakeAvroNodeWithFieldIds(const ::avro::NodePtr& original
 Result<::avro::NodePtr> MakeAvroNodeWithFieldIds(const ::avro::NodePtr& original_node,
                                                  const NameMapping& mapping);
 
+/// \brief Sanitize a field name to make it compatible with Avro field name requirements.
+/// \param field_name The original field name to sanitize.
+/// \return A sanitized field name that follows Avro naming conventions.
+std::string SanitizeFieldName(std::string_view field_name);
+
 }  // namespace iceberg::avro