fix(catalog): honor format-version when creating tables

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -467,6 +467,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
         format_version =
             internal::checked_cast<const table::UpgradeFormatVersion&>(*update)
                 .format_version();
+        // Start at the initial version and replay later upgrades in order.
+        break;
       }
     }
     builder = TableMetadataBuilder::BuildFromEmpty(format_version);

src/iceberg/catalog/sql/sql_catalog.cc

@@ -449,6 +449,8 @@ Result<std::shared_ptr<Table>> SqlCatalog::UpdateTable(
         format_version =
             iceberg::internal::checked_cast<const table::UpgradeFormatVersion&>(*update)
                 .format_version();
+        // Start at the initial version and replay later upgrades in order.
+        break;
       }
     }
     builder = TableMetadataBuilder::BuildFromEmpty(format_version);

src/iceberg/table_metadata.cc

@@ -54,6 +54,7 @@
 #include "iceberg/util/location_util.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/property_util.h"
+#include "iceberg/util/string_util.h"
 #include "iceberg/util/timepoint.h"
 #include "iceberg/util/type_util.h"
 #include "iceberg/util/uuid.h"
@@ -194,10 +195,32 @@ std::string ToString(const MetadataLogEntry& entry) {
                      entry.metadata_file);
 }
 
+Result<std::unique_ptr<TableMetadata>> TableMetadata::Make(
+    const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
+    const iceberg::SortOrder& sort_order, const std::string& location,
+    const std::unordered_map<std::string, std::string>& properties) {
+  auto persisted_properties = properties;
+  int format_version = kDefaultTableFormatVersion;
+  auto iter = persisted_properties.find(TableProperties::kFormatVersion.key());
+  if (iter != persisted_properties.end()) {
+    auto parsed_version = StringUtils::ParseNumber<int>(iter->second);
+    if (!parsed_version.has_value()) {
+      return InvalidArgument("Invalid format-version property: '{}'", iter->second);
+    }
+    format_version = parsed_version.value();
+    persisted_properties.erase(iter);
+  }
+  return Make(schema, spec, sort_order, location, persisted_properties, format_version);
+}
+
 Result<std::unique_ptr<TableMetadata>> TableMetadata::Make(
     const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
     const iceberg::SortOrder& sort_order, const std::string& location,
     const std::unordered_map<std::string, std::string>& properties, int format_version) {
+  if (format_version < 1 || format_version > kSupportedTableFormatVersion) {
+    return InvalidArgument("Unsupported table format version: {} (supported: 1-{})",
+                           format_version, kSupportedTableFormatVersion);
+  }
   for (const auto& [key, _] : properties) {
     if (TableProperties::reserved_properties().contains(key)) {
       return InvalidArgument(
@@ -227,7 +250,7 @@ Result<std::unique_ptr<TableMetadata>> TableMetadata::Make(
 
   ICEBERG_RETURN_UNEXPECTED(PropertyUtil::ValidateCommitProperties(properties));
 
-  return TableMetadataBuilder::BuildFromEmpty(format_version)
+  return TableMetadataBuilder::BuildFromEmpty(static_cast<int8_t>(format_version))
       ->AssignUUID()
       .SetLocation(location)
       .SetCurrentSchema(std::move(fresh_schema), last_column_id)
@@ -567,6 +590,8 @@ class TableMetadataBuilder::Impl {
   // Constructor for new table
   explicit Impl(int8_t format_version) : base_(nullptr), metadata_{} {
     metadata_.format_version = format_version;
+    // Create commits must carry the initial version even when no upgrade occurs.
+    changes_.push_back(std::make_unique<table::UpgradeFormatVersion>(format_version));
     metadata_.last_sequence_number = TableMetadata::kInitialSequenceNumber;
     metadata_.last_updated_ms = kInvalidLastUpdatedMs;
     metadata_.last_column_id = Schema::kInvalidColumnId;

src/iceberg/table_metadata.h

@@ -132,11 +132,14 @@ struct ICEBERG_EXPORT TableMetadata {
   /// A list of encrypted table key entries
   std::vector<EncryptedKey> encryption_keys;
 
+  /// \brief Create table metadata from user-supplied properties.
+  /// \note The reserved format-version property selects the table format version,
+  /// defaulting to kDefaultTableFormatVersion, and is not persisted. Other reserved
+  /// properties are rejected. The input properties are not modified.
   static Result<std::unique_ptr<TableMetadata>> Make(
       const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
       const iceberg::SortOrder& sort_order, const std::string& location,
-      const std::unordered_map<std::string, std::string>& properties,
-      int format_version = kDefaultTableFormatVersion);
+      const std::unordered_map<std::string, std::string>& properties);
 
   /// \brief Get the current schema, return NotFoundError if not found
   /// \note The returned schema is guaranteed to be not null
@@ -168,6 +171,14 @@ struct ICEBERG_EXPORT TableMetadata {
 
   ICEBERG_EXPORT friend bool operator==(const TableMetadata& lhs,
                                         const TableMetadata& rhs);
+
+ private:
+  /// \brief Create table metadata with an explicit supported format version.
+  /// \note All reserved properties, including format-version, are rejected.
+  static Result<std::unique_ptr<TableMetadata>> Make(
+      const iceberg::Schema& schema, const iceberg::PartitionSpec& spec,
+      const iceberg::SortOrder& sort_order, const std::string& location,
+      const std::unordered_map<std::string, std::string>& properties, int format_version);
 };
 
 // Cache for table metadata mappings to facilitate fast lookups.

src/iceberg/test/in_memory_catalog_test.cc

@@ -21,6 +21,7 @@
 
 #include <filesystem>
 #include <string>
+#include <tuple>
 #include <unordered_map>
 
 #include <arrow/filesystem/localfs.h>
@@ -140,6 +141,115 @@ TEST_F(InMemoryCatalogTest, CreateTable) {
   EXPECT_THAT(table2, IsError(ErrorKind::kAlreadyExists));
 }
 
+class InMemoryCatalogFormatVersionTest
+    : public InMemoryCatalogTest,
+      public ::testing::WithParamInterface<std::tuple<bool, int>> {};
+
+TEST_P(InMemoryCatalogFormatVersionTest, CreateTable) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "x", int64())});
+  const auto [staged, version] = GetParam();
+  TableIdentifier ident{.ns = {}, .name = "versioned_table"};
+  const auto location = GenerateTestTableLocation(ident.name);
+  const std::unordered_map<std::string, std::string> properties = {
+      {"format-version", std::to_string(version)}, {"owner", "alice"}};
+  std::shared_ptr<Table> created;
+  if (staged) {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto transaction,
+        catalog_->StageCreateTable(ident, schema, PartitionSpec::Unpartitioned(),
+                                   SortOrder::Unsorted(), location, properties));
+    EXPECT_EQ(transaction->current().format_version, version);
+    EXPECT_THAT(catalog_->TableExists(ident), HasValue(::testing::Eq(false)));
+    ICEBERG_UNWRAP_OR_FAIL(created, transaction->Commit());
+  } else {
+    ICEBERG_UNWRAP_OR_FAIL(
+        created, catalog_->CreateTable(ident, schema, PartitionSpec::Unpartitioned(),
+                                       SortOrder::Unsorted(), location, properties));
+  }
+  EXPECT_EQ(created->metadata()->format_version, version);
+  ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog_->LoadTable(ident));
+  EXPECT_EQ(loaded->metadata()->format_version, version);
+  EXPECT_THAT(loaded->properties().configs(),
+              ::testing::UnorderedElementsAre(::testing::Pair("owner", "alice")));
+}
+
+INSTANTIATE_TEST_SUITE_P(SupportedVersions, InMemoryCatalogFormatVersionTest,
+                         ::testing::Combine(::testing::Bool(),
+                                            ::testing::Values(1, 2, 3)),
+                         [](const ::testing::TestParamInfo<std::tuple<bool, int>>& info) {
+                           return std::string(std::get<0>(info.param) ? "StagedV"
+                                                                      : "CreateV") +
+                                  std::to_string(std::get<1>(info.param));
+                         });
+
+class InMemoryCatalogStagedFormatUpgradeTest : public InMemoryCatalogTest,
+                                               public ::testing::WithParamInterface<int> {
+};
+
+TEST_P(InMemoryCatalogStagedFormatUpgradeTest, UpgradeBeforeCreateCommit) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "x", int64())});
+  TableIdentifier ident{.ns = {}, .name = "upgraded_table"};
+  ICEBERG_UNWRAP_OR_FAIL(auto transaction,
+                         catalog_->StageCreateTable(
+                             ident, schema, PartitionSpec::Unpartitioned(),
+                             SortOrder::Unsorted(), GenerateTestTableLocation(ident.name),
+                             {{"format-version", std::to_string(GetParam())}}));
+  ICEBERG_UNWRAP_OR_FAIL(auto update, transaction->NewUpdateProperties());
+  ASSERT_THAT(update->Set("format-version", "3").Set("owner", "alice").Commit(), IsOk());
+  EXPECT_EQ(transaction->current().format_version, 3);
+  EXPECT_THAT(catalog_->TableExists(ident), HasValue(::testing::Eq(false)));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto created, transaction->Commit());
+  EXPECT_EQ(created->metadata()->format_version, 3);
+  ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog_->LoadTable(ident));
+  EXPECT_EQ(loaded->metadata()->format_version, 3);
+  EXPECT_THAT(loaded->properties().configs(),
+              ::testing::UnorderedElementsAre(::testing::Pair("owner", "alice")));
+}
+
+INSTANTIATE_TEST_SUITE_P(StagedUpgrades, InMemoryCatalogStagedFormatUpgradeTest,
+                         ::testing::Values(1, 2),
+                         [](const ::testing::TestParamInfo<int>& info) {
+                           return "V" + std::to_string(info.param) + "ToV3";
+                         });
+
+class InMemoryCatalogInvalidFormatVersionTest
+    : public InMemoryCatalogTest,
+      public ::testing::WithParamInterface<std::tuple<bool, std::string>> {};
+
+TEST_P(InMemoryCatalogInvalidFormatVersionTest, DoesNotCreateTableOrMetadata) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "x", int64())});
+  const auto& [staged, version] = GetParam();
+  TableIdentifier ident{.ns = {}, .name = "invalid_version"};
+  const auto location = GenerateTestTableLocation(ident.name);
+  const std::unordered_map<std::string, std::string> properties = {
+      {"format-version", version}};
+  if (staged) {
+    EXPECT_THAT(catalog_->StageCreateTable(ident, schema, PartitionSpec::Unpartitioned(),
+                                           SortOrder::Unsorted(), location, properties),
+                IsError(ErrorKind::kInvalidArgument));
+  } else {
+    EXPECT_THAT(catalog_->CreateTable(ident, schema, PartitionSpec::Unpartitioned(),
+                                      SortOrder::Unsorted(), location, properties),
+                IsError(ErrorKind::kInvalidArgument));
+  }
+  EXPECT_THAT(catalog_->TableExists(ident), HasValue(::testing::Eq(false)));
+  EXPECT_TRUE(std::filesystem::is_empty(std::filesystem::path(location) / "metadata"));
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    InvalidVersions, InMemoryCatalogInvalidFormatVersionTest,
+    ::testing::Combine(::testing::Bool(),
+                       ::testing::Values("", "abc", "2x", "1.5", "2147483648", "0", "-1",
+                                         "4", "257")),
+    [](const ::testing::TestParamInfo<std::tuple<bool, std::string>>& info) {
+      return std::string(std::get<0>(info.param) ? "StagedCase" : "CreateCase") +
+             std::to_string(info.index);
+    });
+
 TEST_F(InMemoryCatalogTest, RegisterTable) {
   TableIdentifier tableIdent{.ns = {}, .name = "t1"};
 

src/iceberg/test/sql_catalog_test.cc

@@ -22,15 +22,18 @@
 #include <sqlite3.h>
 
 #include <filesystem>
+#include <fstream>
 #include <memory>
 #include <optional>
 #include <string>
 #include <string_view>
+#include <tuple>
 #include <unordered_map>
 #include <vector>
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
+#include <nlohmann/json.hpp>
 
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/catalog/sql/catalog_store.h"
@@ -41,8 +44,11 @@
 #include "iceberg/sort_order.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/test/matchers.h"
+#include "iceberg/transaction.h"
 #include "iceberg/type.h"
+#include "iceberg/update/update_properties.h"
 
 namespace iceberg::sql {
 
@@ -291,6 +297,126 @@ TEST_F(SqlCatalogTest, TableLifecycle) {
   EXPECT_THAT(catalog_->DropTable(renamed, false), IsError(ErrorKind::kNoSuchTable));
 }
 
+class SqlCatalogFormatVersionTest
+    : public SqlCatalogTest,
+      public ::testing::WithParamInterface<std::tuple<bool, int>> {};
+
+TEST_P(SqlCatalogFormatVersionTest, CreateTable) {
+  Namespace ns{{"db"}};
+  ASSERT_THAT(catalog_->CreateNamespace(ns, {}), IsOk());
+  const auto [staged, version] = GetParam();
+  TableIdentifier ident{.ns = ns, .name = "versioned_table"};
+  const auto location = MakeTableLocation(ident.name);
+  const std::unordered_map<std::string, std::string> properties = {
+      {"format-version", std::to_string(version)}, {"owner", "alice"}};
+  std::shared_ptr<Table> created;
+  if (staged) {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto transaction,
+        catalog_->StageCreateTable(ident, MakeSchema(), PartitionSpec::Unpartitioned(),
+                                   SortOrder::Unsorted(), location, properties));
+    EXPECT_EQ(transaction->current().format_version, version);
+    EXPECT_THAT(catalog_->TableExists(ident), HasValue(::testing::Eq(false)));
+    ICEBERG_UNWRAP_OR_FAIL(created, transaction->Commit());
+  } else {
+    ICEBERG_UNWRAP_OR_FAIL(
+        created,
+        catalog_->CreateTable(ident, MakeSchema(), PartitionSpec::Unpartitioned(),
+                              SortOrder::Unsorted(), location, properties));
+  }
+  EXPECT_EQ(created->metadata()->format_version, version);
+  ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog_->LoadTable(ident));
+  EXPECT_EQ(loaded->metadata()->format_version, version);
+  EXPECT_THAT(loaded->properties().configs(),
+              ::testing::UnorderedElementsAre(::testing::Pair("owner", "alice")));
+  std::ifstream input(std::string(loaded->metadata_file_location()));
+  ASSERT_TRUE(input.is_open());
+  const auto json = nlohmann::json::parse(input);
+  EXPECT_EQ(json.at("format-version"), version);
+  EXPECT_FALSE(json.at("properties").contains("format-version"));
+}
+
+INSTANTIATE_TEST_SUITE_P(SupportedVersions, SqlCatalogFormatVersionTest,
+                         ::testing::Combine(::testing::Bool(),
+                                            ::testing::Values(1, 2, 3)),
+                         [](const ::testing::TestParamInfo<std::tuple<bool, int>>& info) {
+                           return std::string(std::get<0>(info.param) ? "StagedV"
+                                                                      : "CreateV") +
+                                  std::to_string(std::get<1>(info.param));
+                         });
+
+class SqlCatalogStagedFormatUpgradeTest : public SqlCatalogTest,
+                                          public ::testing::WithParamInterface<int> {};
+
+TEST_P(SqlCatalogStagedFormatUpgradeTest, UpgradeBeforeCreateCommit) {
+  Namespace ns{{"db"}};
+  ASSERT_THAT(catalog_->CreateNamespace(ns, {}), IsOk());
+  TableIdentifier ident{.ns = ns, .name = "upgraded_table"};
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto transaction,
+      catalog_->StageCreateTable(ident, MakeSchema(), PartitionSpec::Unpartitioned(),
+                                 SortOrder::Unsorted(), MakeTableLocation(ident.name),
+                                 {{"format-version", std::to_string(GetParam())}}));
+  ICEBERG_UNWRAP_OR_FAIL(auto update, transaction->NewUpdateProperties());
+  ASSERT_THAT(update->Set("format-version", "3").Set("owner", "alice").Commit(), IsOk());
+  EXPECT_EQ(transaction->current().format_version, 3);
+  EXPECT_THAT(catalog_->TableExists(ident), HasValue(::testing::Eq(false)));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto created, transaction->Commit());
+  EXPECT_EQ(created->metadata()->format_version, 3);
+  ICEBERG_UNWRAP_OR_FAIL(auto loaded, catalog_->LoadTable(ident));
+  EXPECT_EQ(loaded->metadata()->format_version, 3);
+  EXPECT_THAT(loaded->properties().configs(),
+              ::testing::UnorderedElementsAre(::testing::Pair("owner", "alice")));
+  std::ifstream input(std::string(loaded->metadata_file_location()));
+  ASSERT_TRUE(input.is_open());
+  const auto json = nlohmann::json::parse(input);
+  EXPECT_EQ(json.at("format-version"), 3);
+  EXPECT_FALSE(json.at("properties").contains("format-version"));
+}
+
+INSTANTIATE_TEST_SUITE_P(StagedUpgrades, SqlCatalogStagedFormatUpgradeTest,
+                         ::testing::Values(1, 2),
+                         [](const ::testing::TestParamInfo<int>& info) {
+                           return "V" + std::to_string(info.param) + "ToV3";
+                         });
+
+class SqlCatalogInvalidFormatVersionTest
+    : public SqlCatalogTest,
+      public ::testing::WithParamInterface<std::tuple<bool, std::string>> {};
+
+TEST_P(SqlCatalogInvalidFormatVersionTest, DoesNotCreateTableOrMetadata) {
+  Namespace ns{{"db"}};
+  ASSERT_THAT(catalog_->CreateNamespace(ns, {}), IsOk());
+  const auto& [staged, version] = GetParam();
+  TableIdentifier ident{.ns = ns, .name = "invalid_version"};
+  const auto location = MakeTableLocation(ident.name);
+  const std::unordered_map<std::string, std::string> properties = {
+      {"format-version", version}};
+  if (staged) {
+    EXPECT_THAT(
+        catalog_->StageCreateTable(ident, MakeSchema(), PartitionSpec::Unpartitioned(),
+                                   SortOrder::Unsorted(), location, properties),
+        IsError(ErrorKind::kInvalidArgument));
+  } else {
+    EXPECT_THAT(catalog_->CreateTable(ident, MakeSchema(), PartitionSpec::Unpartitioned(),
+                                      SortOrder::Unsorted(), location, properties),
+                IsError(ErrorKind::kInvalidArgument));
+  }
+  EXPECT_THAT(catalog_->TableExists(ident), HasValue(::testing::Eq(false)));
+  EXPECT_TRUE(std::filesystem::is_empty(std::filesystem::path(location) / "metadata"));
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    InvalidVersions, SqlCatalogInvalidFormatVersionTest,
+    ::testing::Combine(::testing::Bool(),
+                       ::testing::Values("", "abc", "2x", "1.5", "2147483648", "0", "-1",
+                                         "4", "257")),
+    [](const ::testing::TestParamInfo<std::tuple<bool, std::string>>& info) {
+      return std::string(std::get<0>(info.param) ? "StagedCase" : "CreateCase") +
+             std::to_string(info.index);
+    });
+
 TEST_F(SqlCatalogTest, CreateTableRequiresNamespace) {
   TableIdentifier ident{.ns = Namespace{{"missing"}}, .name = "t1"};
   EXPECT_THAT(catalog_->CreateTable(ident, MakeSchema(), PartitionSpec::Unpartitioned(),