fix: address PR feedback for table requirement validations

Changes based on review feedback from @wgtmac:

1. AssertRefSnapshotID: Updated to match Java implementation logic
   - No longer requires base metadata (allows null for new tables)
   - Updated error messages to match Java version
   - Fixed logic flow to match Java's behavior

2. AssertLastAssignedFieldId: Allow null base metadata
   - Null base is now valid (for new tables)
   - Only validates if base exists

3. AssertLastAssignedPartitionId: Allow null base metadata
   - Null base is now valid (for new tables)
   - Only validates if base exists

Updated all related tests to reflect the new behavior.

Author: shangxinli

src/iceberg/table_requirement.cc

@@ -55,32 +55,32 @@ Status AssertRefSnapshotID::Validate(const TableMetadata* base) const {
   // Validate that a reference (branch or tag) points to the expected snapshot ID
 
   if (base == nullptr) {
-    return CommitFailed("Requirement failed: current table metadata is missing");
-  }
-
-  auto it = base->refs.find(ref_name_);
-
-  // If snapshot_id is nullopt, the reference should not exist
-  if (!snapshot_id_.has_value()) {
-    if (it != base->refs.end()) {
+    // For new tables (null base), only succeed if we're not expecting a specific snapshot
+    if (snapshot_id_.has_value()) {
       return CommitFailed(
-          "Requirement failed: reference '{}' should not exist but found snapshot ID {}",
-          ref_name_, it->second->snapshot_id);
+          "Requirement failed: branch or tag '{}' is missing, expected {}", ref_name_,
+          snapshot_id_.value());
     }
     return {};
   }
 
-  // snapshot_id has a value, so the reference should exist and match
-  if (it == base->refs.end()) {
-    return CommitFailed("Requirement failed: reference '{}' is missing in table metadata",
-                        ref_name_);
-  }
-
-  if (it->second->snapshot_id != snapshot_id_.value()) {
+  auto it = base->refs.find(ref_name_);
+  if (it != base->refs.end()) {
+    // Reference exists
+    if (!snapshot_id_.has_value()) {
+      // A null snapshot ID means the ref should not exist already
+      return CommitFailed(
+          "Requirement failed: reference '{}' was created concurrently", ref_name_);
+    } else if (snapshot_id_.value() != it->second->snapshot_id) {
+      return CommitFailed(
+          "Requirement failed: reference '{}' has changed: expected id {} != {}",
+          ref_name_, snapshot_id_.value(), it->second->snapshot_id);
+    }
+  } else if (snapshot_id_.has_value()) {
+    // Reference does not exist but snapshot_id is specified
     return CommitFailed(
-        "Requirement failed: reference '{}' snapshot ID does not match (expected={}, "
-        "actual={})",
-        ref_name_, snapshot_id_.value(), it->second->snapshot_id);
+        "Requirement failed: branch or tag '{}' is missing, expected {}", ref_name_,
+        snapshot_id_.value());
   }
 
   return {};
@@ -89,11 +89,7 @@ Status AssertRefSnapshotID::Validate(const TableMetadata* base) const {
 Status AssertLastAssignedFieldId::Validate(const TableMetadata* base) const {
   // Validate that the last assigned field ID matches the expected value
 
-  if (base == nullptr) {
-    return CommitFailed("Requirement failed: current table metadata is missing");
-  }
-
-  if (base->last_column_id != last_assigned_field_id_) {
+  if (base && base->last_column_id != last_assigned_field_id_) {
     return CommitFailed(
         "Requirement failed: last assigned field ID does not match (expected={}, "
         "actual={})",
@@ -127,11 +123,7 @@ Status AssertCurrentSchemaID::Validate(const TableMetadata* base) const {
 Status AssertLastAssignedPartitionId::Validate(const TableMetadata* base) const {
   // Validate that the last assigned partition ID matches the expected value
 
-  if (base == nullptr) {
-    return CommitFailed("Requirement failed: current table metadata is missing");
-  }
-
-  if (base->last_partition_id != last_assigned_partition_id_) {
+  if (base && base->last_partition_id != last_assigned_partition_id_) {
     return CommitFailed(
         "Requirement failed: last assigned partition ID does not match (expected={}, "
         "actual={})",

src/iceberg/test/table_metadata_builder_test.cc

@@ -299,23 +299,23 @@ TEST_F(TableMetadataBuilderTest, TableRequirementAssertRefSnapshotIDMismatch) {
 
   auto status = requirement.Validate(base_metadata_.get());
   EXPECT_THAT(status, IsError(ErrorKind::kCommitFailed));
-  EXPECT_THAT(status, HasErrorMessage("snapshot ID does not match"));
+  EXPECT_THAT(status, HasErrorMessage("has changed"));
 }
 
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertRefSnapshotIDRefMissing) {
   table::AssertRefSnapshotID requirement("missing-ref", 100);
 
   auto status = requirement.Validate(base_metadata_.get());
   EXPECT_THAT(status, IsError(ErrorKind::kCommitFailed));
-  EXPECT_THAT(status, HasErrorMessage("missing in table metadata"));
+  EXPECT_THAT(status, HasErrorMessage("is missing"));
 }
 
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertRefSnapshotIDNullBase) {
   table::AssertRefSnapshotID requirement("main", 100);
 
   auto status = requirement.Validate(nullptr);
   EXPECT_THAT(status, IsError(ErrorKind::kCommitFailed));
-  EXPECT_THAT(status, HasErrorMessage("metadata is missing"));
+  EXPECT_THAT(status, HasErrorMessage("is missing"));
 }
 
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertRefSnapshotIDNulloptSuccess) {
@@ -336,7 +336,7 @@ TEST_F(TableMetadataBuilderTest, TableRequirementAssertRefSnapshotIDNulloptButEx
 
   auto status = requirement.Validate(base_metadata_.get());
   EXPECT_THAT(status, IsError(ErrorKind::kCommitFailed));
-  EXPECT_THAT(status, HasErrorMessage("should not exist"));
+  EXPECT_THAT(status, HasErrorMessage("created concurrently"));
 }
 
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertLastAssignedFieldIdSuccess) {
@@ -358,9 +358,8 @@ TEST_F(TableMetadataBuilderTest, TableRequirementAssertLastAssignedFieldIdMismat
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertLastAssignedFieldIdNullBase) {
   table::AssertLastAssignedFieldId requirement(10);
 
-  auto status = requirement.Validate(nullptr);
-  EXPECT_THAT(status, IsError(ErrorKind::kCommitFailed));
-  EXPECT_THAT(status, HasErrorMessage("metadata is missing"));
+  // Null base is allowed (for new tables)
+  ASSERT_THAT(requirement.Validate(nullptr), IsOk());
 }
 
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertLastAssignedPartitionIdSuccess) {
@@ -382,9 +381,8 @@ TEST_F(TableMetadataBuilderTest, TableRequirementAssertLastAssignedPartitionIdMi
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertLastAssignedPartitionIdNullBase) {
   table::AssertLastAssignedPartitionId requirement(5);
 
-  auto status = requirement.Validate(nullptr);
-  EXPECT_THAT(status, IsError(ErrorKind::kCommitFailed));
-  EXPECT_THAT(status, HasErrorMessage("metadata is missing"));
+  // Null base is allowed (for new tables)
+  ASSERT_THAT(requirement.Validate(nullptr), IsOk());
 }
 
 TEST_F(TableMetadataBuilderTest, TableRequirementAssertDefaultSpecIDSuccess) {