refactor: adapt table metadata to use table properties

src/iceberg/json_internal.cc

@@ -39,6 +39,7 @@
 #include "iceberg/statistics_file.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/table_properties.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
@@ -796,7 +797,9 @@ nlohmann::json ToJson(const TableMetadata& table_metadata) {
   json[kSortOrders] = ToJsonList(table_metadata.sort_orders);
 
   // write properties map
-  json[kProperties] = table_metadata.properties;
+  if (table_metadata.properties) {
+    json[kProperties] = table_metadata.properties->configs();
+  }
 
   if (std::ranges::find_if(table_metadata.snapshots, [&](const auto& snapshot) {
         return snapshot->snapshot_id == table_metadata.current_snapshot_id;
@@ -1037,7 +1040,10 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
       table_metadata->default_sort_order_id, table_metadata->sort_orders));
 
   if (json.contains(kProperties)) {
-    ICEBERG_ASSIGN_OR_RAISE(table_metadata->properties, FromJsonMap(json, kProperties));
+    ICEBERG_ASSIGN_OR_RAISE(auto properties, FromJsonMap(json, kProperties));
+    table_metadata->properties = TableProperties::FromMap(std::move(properties));
+  } else {
+    table_metadata->properties = TableProperties::default_properties();
   }
 
   // This field is optional, but internally we set this to -1 when not set

src/iceberg/table.cc

@@ -41,7 +41,6 @@ Table::Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata
       metadata_location_(std::move(metadata_location)),
       io_(std::move(io)),
       catalog_(std::move(catalog)),
-      properties_(TableProperties::FromMap(metadata_->properties)),
       metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())) {}
 
 const std::string& Table::uuid() const { return metadata_->table_uuid; }
@@ -56,7 +55,6 @@ Status Table::Refresh() {
     metadata_ = std::move(refreshed_table->metadata_);
     metadata_location_ = std::move(refreshed_table->metadata_location_);
     io_ = std::move(refreshed_table->io_);
-    properties_ = std::move(refreshed_table->properties_);
     metadata_cache_ = std::make_unique<TableMetadataCache>(metadata_.get());
   }
   return {};
@@ -89,7 +87,9 @@ Table::sort_orders() const {
   return metadata_cache_->GetSortOrdersById();
 }
 
-const TableProperties& Table::properties() const { return *properties_; }
+const std::shared_ptr<TableProperties>& Table::properties() const {
+  return metadata_->properties;
+}
 
 const std::string& Table::location() const { return metadata_->location; }
 

src/iceberg/table.h

@@ -82,7 +82,7 @@ class ICEBERG_EXPORT Table {
   sort_orders() const;
 
   /// \brief Return a map of string properties for this table
-  const TableProperties& properties() const;
+  const std::shared_ptr<TableProperties>& properties() const;
 
   /// \brief Return the table's base location
   const std::string& location() const;
@@ -130,7 +130,6 @@ class ICEBERG_EXPORT Table {
   std::string metadata_location_;
   std::shared_ptr<FileIO> io_;
   std::shared_ptr<Catalog> catalog_;
-  std::unique_ptr<TableProperties> properties_;
   std::unique_ptr<class TableMetadataCache> metadata_cache_;
 };
 

src/iceberg/table_metadata.cc

@@ -38,6 +38,7 @@
 #include "iceberg/schema.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/sort_order.h"
+#include "iceberg/table_properties.h"
 #include "iceberg/table_update.h"
 #include "iceberg/util/gzip_internal.h"
 #include "iceberg/util/macros.h"
@@ -141,6 +142,19 @@ bool SnapshotRefEquals(
   return true;
 }
 
+bool TablePropertiesEquals(const std::shared_ptr<TableProperties>& lhs,
+                           const std::shared_ptr<TableProperties>& rhs) {
+  bool left_is_empty = lhs == nullptr || lhs->configs().empty();
+  bool right_is_empty = rhs == nullptr || rhs->configs().empty();
+  if (left_is_empty && right_is_empty) {
+    return true;
+  }
+  if (left_is_empty || right_is_empty) {
+    return false;
+  }
+  return lhs->configs() == rhs->configs();
+}
+
 }  // namespace
 
 bool operator==(const TableMetadata& lhs, const TableMetadata& rhs) {
@@ -153,7 +167,7 @@ bool operator==(const TableMetadata& lhs, const TableMetadata& rhs) {
          SharedPtrVectorEquals(lhs.schemas, rhs.schemas) &&
          lhs.default_spec_id == rhs.default_spec_id &&
          lhs.last_partition_id == rhs.last_partition_id &&
-         lhs.properties == rhs.properties &&
+         TablePropertiesEquals(lhs.properties, rhs.properties) &&
          lhs.current_snapshot_id == rhs.current_snapshot_id &&
          SharedPtrVectorEquals(lhs.snapshots, rhs.snapshots) &&
          lhs.snapshot_log == rhs.snapshot_log && lhs.metadata_log == rhs.metadata_log &&
@@ -300,6 +314,7 @@ struct TableMetadataBuilder::Impl {
     metadata.current_snapshot_id = Snapshot::kInvalidSnapshotId;
     metadata.default_sort_order_id = SortOrder::kInitialSortOrderId;
     metadata.next_row_id = TableMetadata::kInitialRowId;
+    metadata.properties = TableProperties::default_properties();
   }
 
   // Constructor from existing metadata
@@ -611,7 +626,7 @@ TableMetadataBuilder& TableMetadataBuilder::SetProperties(
 
   // Add all updated properties to the metadata properties
   for (const auto& [key, value] : updated) {
-    impl_->metadata.properties[key] = value;
+    impl_->metadata.properties->mutable_configs()[key] = value;
   }
 
   // Record the change
@@ -629,7 +644,7 @@ TableMetadataBuilder& TableMetadataBuilder::RemoveProperties(
 
   // Remove each property from the metadata properties
   for (const auto& key : removed) {
-    impl_->metadata.properties.erase(key);
+    impl_->metadata.properties->mutable_configs().erase(key);
   }
 
   // Record the change

src/iceberg/table_metadata.h

@@ -99,7 +99,7 @@ struct ICEBERG_EXPORT TableMetadata {
   /// The highest assigned partition field ID across all partition specs for the table
   int32_t last_partition_id;
   /// A string to string map of table properties
-  std::unordered_map<std::string, std::string> properties;
+  std::shared_ptr<TableProperties> properties;
   /// ID of the current table snapshot
   int64_t current_snapshot_id;
   /// A list of valid snapshots

src/iceberg/table_properties.cc

@@ -36,11 +36,9 @@ std::unique_ptr<TableProperties> TableProperties::default_properties() {
 }
 
 std::unique_ptr<TableProperties> TableProperties::FromMap(
-    const std::unordered_map<std::string, std::string>& properties) {
+    std::unordered_map<std::string, std::string> properties) {
   auto table_properties = std::unique_ptr<TableProperties>(new TableProperties());
-  for (const auto& [key, value] : properties) {  // NOLINT(modernize-type-traits)
-    table_properties->configs_[key] = value;
-  }
+  table_properties->configs_ = std::move(properties);
   return table_properties;
 }
 

src/iceberg/table_properties.h

@@ -296,7 +296,7 @@ class ICEBERG_EXPORT TableProperties : public ConfigBase<TableProperties> {
   /// \param properties The map containing property key-value pairs
   /// \return A unique pointer to a TableProperties instance
   static std::unique_ptr<TableProperties> FromMap(
-      const std::unordered_map<std::string, std::string>& properties);
+      std::unordered_map<std::string, std::string> properties);
 
  private:
   TableProperties() = default;

src/iceberg/test/metadata_io_test.cc

@@ -30,6 +30,7 @@
 #include "iceberg/schema.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/table_properties.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/temp_file_test_base.h"
 
@@ -58,7 +59,7 @@ class MetadataIOTest : public TempFileTestBase {
                          .current_schema_id = 1,
                          .default_spec_id = 0,
                          .last_partition_id = 0,
-                         .properties = {{"key", "value"}},
+                         .properties = TableProperties::FromMap({{"key", "value"}}),
                          .current_snapshot_id = 3051729675574597004,
                          .snapshots = {std::make_shared<Snapshot>(Snapshot{
                              .snapshot_id = 3051729675574597004,

src/iceberg/test/table_metadata_builder_test.cc

@@ -29,6 +29,7 @@
 #include "iceberg/sort_field.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/table_properties.h"
 #include "iceberg/table_update.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/transform.h"
@@ -63,6 +64,7 @@ std::unique_ptr<TableMetadata> CreateBaseMetadata() {
   metadata->default_sort_order_id = SortOrder::kInitialSortOrderId;
   metadata->sort_orders.push_back(SortOrder::Unsorted());
   metadata->next_row_id = TableMetadata::kInitialRowId;
+  metadata->properties = TableProperties::default_properties();
   return metadata;
 }
 
@@ -145,19 +147,19 @@ TEST(TableMetadataBuilderTest, SetProperties) {
   builder->SetProperties({{"key1", "value1"}, {"key2", "value2"}});
 
   ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
-  EXPECT_EQ(metadata->properties.size(), 2);
-  EXPECT_EQ(metadata->properties["key1"], "value1");
-  EXPECT_EQ(metadata->properties["key2"], "value2");
+  EXPECT_EQ(metadata->properties->configs().size(), 2);
+  EXPECT_EQ(metadata->properties->configs().at("key1"), "value1");
+  EXPECT_EQ(metadata->properties->configs().at("key2"), "value2");
 
   // Update existing property and add new one
   builder = TableMetadataBuilder::BuildFromEmpty(2);
   builder->SetProperties({{"key1", "value1"}});
   builder->SetProperties({{"key1", "new_value1"}, {"key3", "value3"}});
 
   ICEBERG_UNWRAP_OR_FAIL(metadata, builder->Build());
-  EXPECT_EQ(metadata->properties.size(), 2);
-  EXPECT_EQ(metadata->properties["key1"], "new_value1");
-  EXPECT_EQ(metadata->properties["key3"], "value3");
+  EXPECT_EQ(metadata->properties->configs().size(), 2);
+  EXPECT_EQ(metadata->properties->configs().at("key1"), "new_value1");
+  EXPECT_EQ(metadata->properties->configs().at("key3"), "value3");
 }
 
 TEST(TableMetadataBuilderTest, RemoveProperties) {
@@ -166,9 +168,9 @@ TEST(TableMetadataBuilderTest, RemoveProperties) {
   builder->RemoveProperties({"key2", "key4"});  // key4 does not exist
 
   ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
-  EXPECT_EQ(metadata->properties.size(), 2);
-  EXPECT_EQ(metadata->properties["key1"], "value1");
-  EXPECT_EQ(metadata->properties["key3"], "value3");
+  EXPECT_EQ(metadata->properties->configs().size(), 2);
+  EXPECT_EQ(metadata->properties->configs().at("key1"), "value1");
+  EXPECT_EQ(metadata->properties->configs().at("key3"), "value3");
 }
 
 TEST(TableMetadataBuilderTest, UpgradeFormatVersion) {

src/iceberg/util/config.h

@@ -112,6 +112,8 @@ class ConfigBase {
 
   const std::unordered_map<std::string, std::string>& configs() const { return configs_; }
 
+  std::unordered_map<std::string, std::string>& mutable_configs() { return configs_; }
+
   /// \brief Extracts the prefix from the configuration.
   /// \param prefix The prefix to extract.
   /// \return A map of entries that match the prefix with prefix removed.