fix(metadata): drop removed schema ids from the builder's schema index

src/iceberg/table_metadata.cc

@@ -1006,6 +1006,9 @@ Status TableMetadataBuilder::Impl::RemoveSchemas(
                           return !schema_ids.contains(schema->schema_id());
                         }) |
                         std::ranges::to<std::vector<std::shared_ptr<Schema>>>();
+    for (const auto schema_id : schema_ids) {
+      schemas_by_id_.erase(schema_id);
+    }
     changes_.push_back(std::make_unique<table::RemoveSchemas>(schema_ids));
   }
 

src/iceberg/test/table_metadata_builder_test.cc

@@ -584,6 +584,51 @@ TEST(TableMetadataBuilderTest, SetDefaultSortOrderInvalid) {
   ASSERT_THAT(builder->Build(), HasErrorMessage("no sort order has been added"));
 }
 
+TEST(TableMetadataBuilderTest, RemoveSchemasDropsIndexEntry) {
+  auto base = CreateBaseMetadata();
+  auto builder = TableMetadataBuilder::BuildFrom(base.get());
+
+  auto field1 = SchemaField::MakeRequired(4, "field1", int64());
+  auto schema1 = std::make_shared<Schema>(std::vector<SchemaField>{field1}, 1);
+  builder->AddSchema(schema1);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
+  ASSERT_EQ(metadata->schemas.size(), 2);
+
+  builder = TableMetadataBuilder::BuildFrom(metadata.get());
+  builder->RemoveSchemas({1});
+
+  // The removed id is gone: setting it as current must fail at Build(),
+  // otherwise Build() would emit a current-schema-id dangling outside
+  // metadata.schemas.
+  builder->SetCurrentSchema(1);
+  auto result = builder->Build();
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(result, HasErrorMessage("Cannot set current schema to unknown schema: 1"));
+}
+
+TEST(TableMetadataBuilderTest, AddSchemaAfterRemoveNotConfusedByStaleIndex) {
+  auto base = CreateBaseMetadata();
+  auto builder = TableMetadataBuilder::BuildFrom(base.get());
+
+  auto field1 = SchemaField::MakeRequired(4, "field1", int64());
+  auto schema1 = std::make_shared<Schema>(std::vector<SchemaField>{field1}, 1);
+  builder->AddSchema(schema1);
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
+  ASSERT_EQ(metadata->schemas.size(), 2);
+
+  builder = TableMetadataBuilder::BuildFrom(metadata.get());
+  builder->RemoveSchemas({1});
+
+  // Re-adding the same schema shape must not be treated as a no-op reuse
+  // via a stale index entry: it has to land in metadata.schemas again.
+  auto schema_again = std::make_shared<Schema>(std::vector<SchemaField>{field1}, 1);
+  builder->AddSchema(schema_again);
+  ICEBERG_UNWRAP_OR_FAIL(auto rebuilt, builder->Build());
+  ASSERT_EQ(rebuilt->schemas.size(), 2);
+  EXPECT_EQ(rebuilt->schemas[1]->schema_id(), 1);
+}
+
 // Test AddSchema
 TEST(TableMetadataBuilderTest, AddSchemaBasic) {
   auto base = CreateBaseMetadata();

src/iceberg/test/transaction_test.cc

@@ -35,6 +35,7 @@
 #include "iceberg/test/update_test_base.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
+#include "iceberg/update/expire_snapshots.h"
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/update_properties.h"
@@ -115,6 +116,42 @@ TEST_F(TransactionTest, MultipleUpdatesInTransaction) {
   EXPECT_EQ(*sort_order, *expected_sort_order);
 }
 
+class MinimalTransactionTest : public MinimalUpdateTestBase {};
+
+TEST_F(MinimalTransactionTest, ExpireUnusedSchemaThenAddColumn) {
+  // Add a column and drop it again: the current schema goes back to id 0 and
+  // schema 1 is left unused.
+  ICEBERG_UNWRAP_OR_FAIL(auto add_column, table_->NewUpdateSchema());
+  add_column->AddColumn("dropped", int64());
+  ASSERT_THAT(add_column->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(table_, catalog_->LoadTable(table_ident_));
+  ICEBERG_UNWRAP_OR_FAIL(auto drop_column, table_->NewUpdateSchema());
+  drop_column->DeleteColumn("dropped");
+  ASSERT_THAT(drop_column->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(table_, catalog_->LoadTable(table_ident_));
+  ASSERT_EQ(table_->metadata()->current_schema_id, 0);
+  ASSERT_EQ(table_->metadata()->schemas.size(), 2);
+
+  // Expiring removes schema 1, and the schema change in the same transaction is
+  // assigned the freed id.
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto expire, txn->NewExpireSnapshots());
+  expire->CleanExpiredMetadata(true);
+  ASSERT_THAT(expire->Commit(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto update_schema, txn->NewUpdateSchema());
+  update_schema->AddColumn("added", int64());
+  ASSERT_THAT(update_schema->Commit(), IsOk());
+  ASSERT_THAT(txn->Commit(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto reloaded, catalog_->LoadTable(table_ident_));
+  ICEBERG_UNWRAP_OR_FAIL(auto schema, reloaded->schema());
+  EXPECT_EQ(schema->schema_id(), 1);
+  ICEBERG_UNWRAP_OR_FAIL(auto added, schema->FindFieldByName("added"));
+  EXPECT_TRUE(added.has_value());
+  ICEBERG_UNWRAP_OR_FAIL(auto dropped, schema->FindFieldByName("dropped"));
+  EXPECT_FALSE(dropped.has_value());
+}
+
 class TransactionRetryTest : public UpdateTestBase {
  protected:
   void SetUp() override {