refactor: allocate table labels only when present

Clarify label lifecycle and cover refresh, empty transactions, and REST load/create/register in both planning modes.

Author: wgtmac

src/iceberg/table.cc

@@ -93,7 +93,8 @@ Table::Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata
       catalog_(std::move(catalog)),
       reporter_(std::move(reporter)),
       metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())),
-      labels_(std::move(labels)) {}
+      labels_(labels.empty() ? nullptr
+                             : std::make_unique<const Labels>(std::move(labels))) {}
 
 const std::string& Table::uuid() const { return metadata_->table_uuid; }
 
@@ -167,7 +168,13 @@ const std::shared_ptr<Catalog>& Table::catalog() const { return catalog_; }
 
 const std::shared_ptr<MetricsReporter>& Table::reporter() const { return reporter_; }
 
-const Labels& Table::labels() const { return labels_; }
+const Labels& Table::labels() const {
+  if (labels_) {
+    return *labels_;
+  }
+  static const Labels kEmpty;
+  return kEmpty;
+}
 
 Result<std::unique_ptr<LocationProvider>> Table::location_provider() const {
   return LocationProvider::Make(metadata_->location, metadata_->properties);

src/iceberg/table.h

@@ -134,9 +134,8 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
 
   /// \brief Returns the catalog-provided labels for this table.
   ///
-  /// Labels come only from the catalog's load, create or register response and are fixed
-  /// for the life of this table: Refresh() does not update them, and a table returned by
-  /// Transaction::Commit() has none. Load the table again to get current labels.
+  /// Labels are supplied at construction and are not updated by Refresh().
+  /// Reload the table to obtain current catalog labels.
   const Labels& labels() const;
 
   /// \brief Returns a LocationProvider for this table
@@ -235,7 +234,7 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   std::shared_ptr<Catalog> catalog_;
   std::shared_ptr<MetricsReporter> reporter_;
   std::unique_ptr<class TableMetadataCache> metadata_cache_;
-  const Labels labels_;
+  const std::unique_ptr<const Labels> labels_;
 };
 
 /// \brief A table created by stage-create and not yet committed.

src/iceberg/test/rest_catalog_unit_test.cc

@@ -39,6 +39,8 @@
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_io_registry.h"
+#include "iceberg/labels.h"
+#include "iceberg/schema.h"
 #include "iceberg/storage_credential.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
@@ -214,6 +216,76 @@ class RestCatalogLoadTableTest : public ::testing::Test {
   std::unordered_set<Endpoint> no_plan_endpoint_set_;
 };
 
+class RestCatalogLabelsTest : public RestCatalogLoadTableTest,
+                              public ::testing::WithParamInterface<bool> {
+ protected:
+  std::string LabelsResponse() const {
+    auto response = nlohmann::json::parse(kLoadTableDefaultResponse);
+    response["config"]["scan-planning-mode"] = GetParam() ? "server" : "client";
+    response["labels"] = nlohmann::json::parse(
+        R"({"object-labels":{"owner":"team-a"},"fields":[{"field-id":1,"labels":{"classification":"pii"}}]})");
+    return response.dump();
+  }
+
+  Result<std::shared_ptr<Catalog>> MakeLabelsCatalog() {
+    auto endpoints = all_plan_endpoints_;
+    endpoints.insert(Endpoint::CreateTable());
+    endpoints.insert(Endpoint::RegisterTable());
+    ICEBERG_ASSIGN_OR_RAISE(auto catalog, MakeCatalog(endpoints));
+    return catalog->AsCatalog();
+  }
+
+  void CheckLabels(const std::shared_ptr<Table>& table) const {
+    ASSERT_NE(table, nullptr);
+    EXPECT_EQ(table->labels(), labels_);
+    EXPECT_EQ(std::dynamic_pointer_cast<RestTable>(table) != nullptr, GetParam());
+  }
+
+  const Labels labels_{
+      .object_labels = {{"owner", "team-a"}},
+      .fields = {{.field_id = 1, .labels = {{"classification", "pii"}}}}};
+};
+
+TEST_P(RestCatalogLabelsTest, LoadExposesLabels) {
+  EXPECT_CALL(*mock_client_, Get(::testing::EndsWith("/tables/my_table"), _, _, _, _))
+      .WillOnce(Return(HttpResponse(200, LabelsResponse())));
+  ICEBERG_UNWRAP_OR_FAIL(auto catalog, MakeLabelsCatalog());
+  ICEBERG_UNWRAP_OR_FAIL(auto table, catalog->LoadTable(identifier_));
+
+  CheckLabels(table);
+}
+
+TEST_P(RestCatalogLabelsTest, CreateExposesLabels) {
+  EXPECT_CALL(*mock_client_,
+              Post(::testing::EndsWith("/namespaces/default/tables"), _, _, _, _))
+      .WillOnce(Return(HttpResponse(200, LabelsResponse())));
+  ICEBERG_UNWRAP_OR_FAIL(auto catalog, MakeLabelsCatalog());
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32())}, 1);
+  ICEBERG_UNWRAP_OR_FAIL(auto table,
+                         catalog->CreateTable(identifier_, schema, /*spec=*/nullptr,
+                                              /*order=*/nullptr, /*location=*/"",
+                                              /*properties=*/{}));
+
+  CheckLabels(table);
+}
+
+TEST_P(RestCatalogLabelsTest, RegisterExposesLabels) {
+  EXPECT_CALL(*mock_client_,
+              Post(::testing::EndsWith("/namespaces/default/register"), _, _, _, _))
+      .WillOnce(Return(HttpResponse(200, LabelsResponse())));
+  ICEBERG_UNWRAP_OR_FAIL(auto catalog, MakeLabelsCatalog());
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto table, catalog->RegisterTable(identifier_, "s3://bucket/metadata/v1.json"));
+
+  CheckLabels(table);
+}
+
+INSTANTIATE_TEST_SUITE_P(PlanningModes, RestCatalogLabelsTest, ::testing::Bool(),
+                         [](const ::testing::TestParamInfo<bool>& info) {
+                           return info.param ? "Server" : "Client";
+                         });
+
 TEST_F(RestCatalogLoadTableTest, InitializationPropagatesClientFactoryErrors) {
   auto config = RestCatalogProperties::FromMap(
       {{"uri", "http://test-server"}, {"io-impl", "rest-catalog-unit-test-noop"}});

src/iceberg/test/table_test.cc

@@ -29,6 +29,7 @@
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
 #include "iceberg/test/mock_io.h"
+#include "iceberg/transaction.h"
 
 namespace iceberg {
 
@@ -175,27 +176,72 @@ class TableLabelsTest : public ::testing::Test {
   std::shared_ptr<TableMetadata> metadata_;
 };
 
-TEST_F(TableLabelsTest, ExposesProvidedLabels) {
+TEST_F(TableLabelsTest, EmptyLabelsShareFallback) {
   ICEBERG_UNWRAP_OR_FAIL(
-      auto table, Table::Make(ident_, metadata_, "s3://bucket/meta.json", io_, catalog_,
-                              /*full_name=*/"", /*reporter=*/nullptr, labels_));
+      auto first, Table::Make(ident_, metadata_, "s3://bucket/meta.json", io_, catalog_));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto second, Table::Make(ident_, metadata_, "s3://bucket/meta.json", io_, catalog_,
+                               /*full_name=*/"", /*reporter=*/nullptr, Labels{}));
 
-  EXPECT_EQ(table->labels(), labels_);
+  EXPECT_TRUE(first->labels().empty());
+  EXPECT_TRUE(second->labels().empty());
+  EXPECT_EQ(&first->labels(), &second->labels());
 }
 
-TEST_F(TableLabelsTest, RefreshKeepsLabels) {
+TEST_F(TableLabelsTest, ExposesProvidedLabels) {
+  for (const auto& labels : {labels_, Labels{.object_labels = {{"owner", "team-a"}}},
+                             Labels{.fields = {{.field_id = 1}}}}) {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto table, Table::Make(ident_, metadata_, "s3://bucket/meta.json", io_, catalog_,
+                                /*full_name=*/"", /*reporter=*/nullptr, labels));
+
+    EXPECT_EQ(table->labels(), labels);
+  }
+}
+
+class TableLabelsRefreshTest : public TableLabelsTest,
+                               public ::testing::WithParamInterface<bool> {};
+
+TEST_P(TableLabelsRefreshTest, RefreshKeepsLabels) {
+  const std::string metadata_location =
+      GetParam() ? "s3://bucket/meta2.json" : "s3://bucket/meta.json";
+  const Labels changed_labels{
+      .object_labels = {{"owner", "team-b"}},
+      .fields = {{.field_id = 1, .labels = {{"classification", "public"}}}}};
+  for (const auto& refreshed_labels : {Labels{}, changed_labels}) {
+    SCOPED_TRACE(refreshed_labels.empty() ? "Absent labels" : "Changed labels");
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto table, Table::Make(ident_, metadata_, "s3://bucket/meta.json", io_, catalog_,
+                                /*full_name=*/"", /*reporter=*/nullptr, labels_));
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto refreshed,
+        Table::Make(ident_, metadata_, metadata_location, io_, catalog_,
+                    /*full_name=*/"", /*reporter=*/nullptr, refreshed_labels));
+    EXPECT_CALL(*catalog_, LoadTable(::testing::_))
+        .WillOnce(::testing::Return(refreshed));
+
+    ASSERT_THAT(table->Refresh(), IsOk());
+
+    EXPECT_EQ(table->metadata_file_location(), metadata_location);
+    EXPECT_EQ(table->labels(), labels_);
+  }
+}
+
+INSTANTIATE_TEST_SUITE_P(MetadataLocation, TableLabelsRefreshTest, ::testing::Bool(),
+                         [](const ::testing::TestParamInfo<bool>& info) {
+                           return info.param ? "Changed" : "Unchanged";
+                         });
+
+TEST_F(TableLabelsTest, EmptyTransactionKeepsLabels) {
   ICEBERG_UNWRAP_OR_FAIL(
       auto table, Table::Make(ident_, metadata_, "s3://bucket/meta.json", io_, catalog_,
                               /*full_name=*/"", /*reporter=*/nullptr, labels_));
-  ICEBERG_UNWRAP_OR_FAIL(
-      auto refreshed,
-      Table::Make(ident_, metadata_, "s3://bucket/meta2.json", io_, catalog_));
-  EXPECT_CALL(*catalog_, LoadTable(::testing::_)).WillOnce(::testing::Return(refreshed));
-
-  ASSERT_THAT(table->Refresh(), IsOk());
+  EXPECT_CALL(*catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_)).Times(0);
+  ICEBERG_UNWRAP_OR_FAIL(auto transaction, table->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto committed, transaction->Commit());
 
-  EXPECT_EQ(table->metadata_file_location(), "s3://bucket/meta2.json");
-  EXPECT_EQ(table->labels(), labels_);
+  ASSERT_EQ(committed, table);
+  EXPECT_EQ(committed->labels(), labels_);
 }
 
 TEST(StaticTableTest, NewMutatingOperationsAreNotSupported) {