fix the case sensitivity bug

Author: gty404

src/iceberg/test/update_schema_test.cc

@@ -1456,6 +1456,85 @@ TEST_F(UpdateSchemaTest, DeleteRenameConflict) {
   EXPECT_THAT(result, HasErrorMessage("Cannot rename a column that will be deleted"));
 }
 
+// Test case insensitive add then update
+TEST_F(UpdateSchemaTest, CaseInsensitiveAddThenUpdate) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateSchema());
+  update->CaseSensitive(false)
+      .AddColumn("Foo", int32(), "A column with uppercase name")
+      .UpdateColumn("foo", int64());  // Update using lowercase
+
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+
+  // Verify the column was added and updated
+  ICEBERG_UNWRAP_OR_FAIL(auto field_opt, result.schema->FindFieldByName("Foo", false));
+  ASSERT_TRUE(field_opt.has_value());
+  EXPECT_EQ(*field_opt->get().type(), *int64());  // Type should be updated to int64
+}
+
+// Test case insensitive add then rename
+TEST_F(UpdateSchemaTest, CaseInsensitiveAddThenRename) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateSchema());
+  update->CaseSensitive(false)
+      .AddColumn("Foo", string(), "A column with uppercase name")
+      .RenameColumn("foo", "Bar");  // Rename using lowercase
+
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+
+  // Verify the column was renamed
+  ICEBERG_UNWRAP_OR_FAIL(auto foo_opt, result.schema->FindFieldByName("Foo", false));
+  ICEBERG_UNWRAP_OR_FAIL(auto bar_opt, result.schema->FindFieldByName("Bar", false));
+  EXPECT_FALSE(foo_opt.has_value());
+  ASSERT_TRUE(bar_opt.has_value());
+  EXPECT_EQ(*bar_opt->get().type(), *string());
+}
+
+// Test case insensitive add then update doc
+TEST_F(UpdateSchemaTest, CaseInsensitiveAddThenUpdateDoc) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateSchema());
+  update->CaseSensitive(false)
+      .AddColumn("Foo", int32(), "original doc")
+      .UpdateColumnDoc("foo", "updated doc");  // Update doc using lowercase
+
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+
+  // Verify the doc was updated
+  ICEBERG_UNWRAP_OR_FAIL(auto field_opt, result.schema->FindFieldByName("Foo", false));
+  ASSERT_TRUE(field_opt.has_value());
+  EXPECT_EQ(field_opt->get().doc(), "updated doc");
+}
+
+// Test case insensitive add then make optional
+TEST_F(UpdateSchemaTest, CaseInsensitiveAddThenMakeOptional) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateSchema());
+  update->CaseSensitive(false)
+      .AllowIncompatibleChanges()
+      .AddRequiredColumn("Foo", int32(), "required column")
+      .MakeColumnOptional("foo");  // Make optional using lowercase
+
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+
+  // Verify the column is now optional
+  ICEBERG_UNWRAP_OR_FAIL(auto field_opt, result.schema->FindFieldByName("Foo", false));
+  ASSERT_TRUE(field_opt.has_value());
+  EXPECT_TRUE(field_opt->get().optional());
+}
+
+// Test case insensitive add then require
+TEST_F(UpdateSchemaTest, CaseInsensitiveAddThenRequire) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewUpdateSchema());
+  update->CaseSensitive(false)
+      .AllowIncompatibleChanges()
+      .AddColumn("Foo", int32(), "optional column")
+      .RequireColumn("foo");  // Require using lowercase
+
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+
+  // Verify the column is now required
+  ICEBERG_UNWRAP_OR_FAIL(auto field_opt, result.schema->FindFieldByName("Foo", false));
+  ASSERT_TRUE(field_opt.has_value());
+  EXPECT_FALSE(field_opt->get().optional());
+}
+
 // Test mixed changes - comprehensive test combining multiple operations
 TEST_F(UpdateSchemaTest, MixedChanges) {
   // First, create a complex schema similar to Java's SCHEMA constant

src/iceberg/update/update_schema.cc

@@ -37,6 +37,7 @@
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/error_collector.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
 #include "iceberg/util/type_util.h"
 #include "iceberg/util/visit_type.h"
 
@@ -716,7 +717,7 @@ UpdateSchema& UpdateSchema::AddColumnInternal(std::optional<std::string_view> pa
   int32_t new_id = AssignNewColumnId();
 
   // Update tracking for moves
-  added_name_to_id_[full_name] = new_id;
+  added_name_to_id_[CaseSensitivityAwareName(full_name)] = new_id;
   if (parent_id != kTableRootId) {
     id_to_parent_[new_id] = parent_id;
   }
@@ -769,7 +770,7 @@ UpdateSchema::FindFieldForUpdate(std::string_view name) const {
   }
 
   // Field not in original schema (or is being deleted), check if it's a newly added field
-  auto added_it = added_name_to_id_.find(std::string(name));
+  auto added_it = added_name_to_id_.find(CaseSensitivityAwareName(name));
   if (added_it != added_name_to_id_.end()) {
     int32_t added_id = added_it->second;
     auto update_it = updates_.find(added_id);
@@ -783,4 +784,12 @@ UpdateSchema::FindFieldForUpdate(std::string_view name) const {
   return std::nullopt;
 }
 
+std::string UpdateSchema::CaseSensitivityAwareName(std::string_view name) const {
+  if (case_sensitive_) {
+    return std::string(name);
+  }
+  // Convert to lowercase for case-insensitive comparison
+  return StringUtils::ToLower(name);
+}
+
 }  // namespace iceberg

src/iceberg/update/update_schema.h

@@ -371,6 +371,15 @@ class ICEBERG_EXPORT UpdateSchema : public PendingUpdate {
   Result<std::optional<std::reference_wrapper<const SchemaField>>> FindFieldForUpdate(
       std::string_view name) const;
 
+  /// \brief Normalize a field name based on case sensitivity setting.
+  ///
+  /// If case_sensitive_ is true, returns the name as-is.
+  /// If case_sensitive_ is false, returns the lowercase version of the name.
+  ///
+  /// \param name The field name to normalize.
+  /// \return The normalized field name.
+  std::string CaseSensitivityAwareName(std::string_view name) const;
+
   // Internal state
   std::shared_ptr<Schema> schema_;
   int32_t last_column_id_;