Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhjwpku wants to merge 1 commit into
base: main
Choose a base branch
from
+418 −4
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/iceberg/catalog/memory/in_memory_catalog.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
2 changes: 2 additions & 0 deletions src/iceberg/catalog/sql/sql_catalog.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
27 changes: 26 additions & 1 deletion src/iceberg/table_metadata.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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;
Expand Down
14 changes: 12 additions & 2 deletions src/iceberg/table_metadata.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,11 +132,21 @@ 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 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);

@wgtmac wgtmac Nov 4, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we avoid a separate overload? Perhaps we can change it to std::optional<int>? How does the Java implementation deal with this?

@zhjwpku zhjwpku Nov 4, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Java impl uses a public overload to read format-version from properties(defaulting to V2), filters out reserved properties, and delegates to a package-private overload taking an explicit int formatVersion, see [1].

We can follow Java here. But the overload with format_version was already public before this PR, so I'm not sure if we should make it private. WDYT?

[1] https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/TableMetadata.java#L69-L119


/// \brief Get the current schema, return NotFoundError if not found
/// \note The returned schema is guaranteed to be not null
Expand Down
110 changes: 110 additions & 0 deletions src/iceberg/test/in_memory_catalog_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

#include <filesystem>
#include <string>
#include <tuple>
#include <unordered_map>

#include <arrow/filesystem/localfs.h>
Expand Down Expand Up @@ -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"};

Expand Down
126 changes: 126 additions & 0 deletions src/iceberg/test/sql_catalog_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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 {

Expand Down Expand Up @@ -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(),
Expand Down
Loading
Loading