feat: support catalog labels on tables loaded from REST catalogs

Parse the optional `labels` field of the REST load-table response and
expose it through Table::labels(), mirroring apache/iceberg#18045 and
live on the Table instance instead of TableMetadata.

Part of #938.

Author: NihalJain

src/iceberg/catalog/rest/json_serde.cc

@@ -33,6 +33,7 @@
 #include "iceberg/expression/json_serde_internal.h"
 #include "iceberg/file_format.h"
 #include "iceberg/json_serde_internal.h"
+#include "iceberg/labels.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
@@ -72,6 +73,10 @@ constexpr std::string_view kDestination = "destination";
 constexpr std::string_view kMetadata = "metadata";
 constexpr std::string_view kConfig = "config";
 constexpr std::string_view kStorageCredentials = "storage-credentials";
+constexpr std::string_view kLabels = "labels";
+constexpr std::string_view kObjectLabels = "object-labels";
+constexpr std::string_view kFields = "fields";
+constexpr std::string_view kFieldId = "field-id";
 constexpr std::string_view kPrefix = "prefix";
 constexpr std::string_view kIdentifiers = "identifiers";
 constexpr std::string_view kOverrides = "overrides";
@@ -474,6 +479,59 @@ Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json
   return request;
 }
 
+nlohmann::json ToJson(const FieldLabel& field_label) {
+  nlohmann::json json;
+  json[kFieldId] = field_label.field_id;
+  json[kLabels] = field_label.labels;
+  return json;
+}
+
+Result<FieldLabel> FieldLabelFromJson(const nlohmann::json& json) {
+  if (!json.is_object()) {
+    return JsonParseError("Cannot parse field label from non-object: {}",
+                          SafeDumpJson(json));
+  }
+  FieldLabel field_label;
+  ICEBERG_ASSIGN_OR_RAISE(field_label.field_id, GetJsonInteger<int32_t>(json, kFieldId));
+  ICEBERG_ASSIGN_OR_RAISE(field_label.labels,
+                          GetJsonValue<decltype(field_label.labels)>(json, kLabels));
+  return field_label;
+}
+
+nlohmann::json ToJson(const Labels& labels) {
+  nlohmann::json json = nlohmann::json::object();
+  if (!labels.object_labels.empty()) {
+    json[kObjectLabels] = labels.object_labels;
+  }
+  if (!labels.fields.empty()) {
+    nlohmann::json fields = nlohmann::json::array();
+    for (const auto& field_label : labels.fields) {
+      fields.push_back(ToJson(field_label));
+    }
+    json[kFields] = std::move(fields);
+  }
+  return json;
+}
+
+// A non-object value is treated as absent labels rather than failing the load.
+Result<Labels> LabelsFromJson(const nlohmann::json& json) {
+  Labels labels;
+  ICEBERG_ASSIGN_OR_RAISE(
+      labels.object_labels,
+      GetJsonValueOrDefault<decltype(labels.object_labels)>(json, kObjectLabels));
+  if (auto it = json.find(kFields); it != json.end() && !it->is_null()) {
+    if (!it->is_array()) {
+      return JsonParseError("Cannot parse '{}' from non-array: {}", kFields,
+                            SafeDumpJson(*it));
+    }
+    for (const auto& entry : *it) {
+      ICEBERG_ASSIGN_OR_RAISE(auto field_label, FieldLabelFromJson(entry));
+      labels.fields.push_back(std::move(field_label));
+    }
+  }
+  return labels;
+}
+
 // LoadTableResult (used by CreateTableResponse, LoadTableResponse)
 Result<nlohmann::json> ToJson(const LoadTableResult& result) {
   nlohmann::json json;
@@ -488,6 +546,9 @@ Result<nlohmann::json> ToJson(const LoadTableResult& result) {
     }
     json[kStorageCredentials] = std::move(creds);
   }
+  if (!result.labels.empty()) {
+    json[kLabels] = ToJson(result.labels);
+  }
   return json;
 }
 
@@ -501,6 +562,9 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(result.config,
                           GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
   ICEBERG_ASSIGN_OR_RAISE(result.storage_credentials, StorageCredentialsFromJson(json));
+  if (auto it = json.find(kLabels); it != json.end() && !it->is_null()) {
+    ICEBERG_ASSIGN_OR_RAISE(result.labels, LabelsFromJson(*it));
+  }
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
   return result;
 }
@@ -983,5 +1047,7 @@ ICEBERG_DEFINE_FROM_JSON(CommitTableResponse)
 ICEBERG_DEFINE_FROM_JSON(OAuthTokenResponse)
 ICEBERG_DEFINE_FROM_JSON(PlanTableScanRequest)
 ICEBERG_DEFINE_FROM_JSON(FetchScanTasksRequest)
+ICEBERG_DEFINE_FROM_JSON(FieldLabel)
+ICEBERG_DEFINE_FROM_JSON(Labels)
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/json_serde_internal.h

@@ -60,6 +60,8 @@ ICEBERG_DECLARE_JSON_SERDE(ListTablesResponse)
 ICEBERG_DECLARE_JSON_SERDE(RegisterTableRequest)
 ICEBERG_DECLARE_JSON_SERDE(RenameTableRequest)
 ICEBERG_DECLARE_JSON_SERDE(OAuthTokenResponse)
+ICEBERG_DECLARE_JSON_SERDE(FieldLabel)
+ICEBERG_DECLARE_JSON_SERDE(Labels)
 
 #undef ICEBERG_DECLARE_JSON_SERDE
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -975,7 +975,7 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
   return Table::Make(identifier, std::move(result.metadata),
                      std::move(result.metadata_location), std::move(table_io),
                      std::move(table_catalog), RestTableName(name_, identifier),
-                     std::move(reporter));
+                     std::move(reporter), std::move(result.labels));
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromCommitResponse(

src/iceberg/catalog/rest/types.cc

@@ -87,7 +87,7 @@ bool CreateTableRequest::operator==(const CreateTableRequest& other) const {
 
 bool LoadTableResult::operator==(const LoadTableResult& other) const {
   if (metadata_location != other.metadata_location || config != other.config ||
-      storage_credentials != other.storage_credentials) {
+      storage_credentials != other.storage_credentials || labels != other.labels) {
     return false;
   }
 

src/iceberg/catalog/rest/types.h

@@ -29,6 +29,7 @@
 
 #include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/labels.h"
 #include "iceberg/result.h"
 #include "iceberg/storage_credential.h"
 #include "iceberg/table_identifier.h"
@@ -188,6 +189,8 @@ struct ICEBERG_REST_EXPORT LoadTableResult {
   std::unordered_map<std::string, std::string> config;
   /// \brief Vended storage credentials, one per URI prefix; empty if none.
   std::vector<StorageCredential> storage_credentials;
+  /// \brief Catalog-provided labels; empty if none.
+  Labels labels;
 
   /// \brief Validates the LoadTableResult.
   Status Validate() const {

src/iceberg/cpp20_compatibility_internal.h

@@ -81,6 +81,7 @@
 #include "iceberg/inspect/history_table.h"
 #include "iceberg/inspect/metadata_table.h"
 #include "iceberg/inspect/snapshots_table.h"
+#include "iceberg/labels.h"
 #include "iceberg/location_provider.h"
 #include "iceberg/logging/cerr_logger.h"
 #include "iceberg/logging/log_level.h"

src/iceberg/labels.h

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/labels.h
+/// \brief Catalog-provided labels for tables and their fields.
+
+#include <cstdint>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+/// \brief Catalog-provided labels for a single field, identified by its field ID.
+struct ICEBERG_EXPORT FieldLabel {
+  int32_t field_id = 0;
+  std::unordered_map<std::string, std::string> labels;
+
+  bool operator==(const FieldLabel&) const = default;
+};
+
+/// \brief Optional catalog-provided labels returned when a table is loaded.
+///
+/// Labels are advisory enrichment, not table state: they are not persisted with the
+/// table and may be absent.
+struct ICEBERG_EXPORT Labels {
+  std::unordered_map<std::string, std::string> object_labels;
+  std::vector<FieldLabel> fields;
+
+  /// \brief Returns true when there are neither object-level nor field-level labels.
+  bool empty() const { return object_labels.empty() && fields.empty(); }
+
+  bool operator==(const Labels&) const = default;
+};
+
+}  // namespace iceberg

src/iceberg/table.cc

@@ -56,13 +56,11 @@
 
 namespace iceberg {
 
-Result<std::shared_ptr<Table>> Table::Make(TableIdentifier identifier,
-                                           std::shared_ptr<TableMetadata> metadata,
-                                           std::string metadata_location,
-                                           std::shared_ptr<FileIO> io,
-                                           std::shared_ptr<Catalog> catalog,
-                                           std::string full_name,
-                                           std::shared_ptr<MetricsReporter> reporter) {
+Result<std::shared_ptr<Table>> Table::Make(
+    TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
+    std::string metadata_location, std::shared_ptr<FileIO> io,
+    std::shared_ptr<Catalog> catalog, std::string full_name,
+    std::shared_ptr<MetricsReporter> reporter, Labels labels) {
   if (metadata == nullptr) [[unlikely]] {
     return InvalidArgument("Metadata cannot be null");
   }
@@ -75,25 +73,27 @@ Result<std::shared_ptr<Table>> Table::Make(TableIdentifier identifier,
   if (catalog == nullptr) [[unlikely]] {
     return InvalidArgument("Catalog cannot be null");
   }
-  return std::shared_ptr<Table>(new Table(
-      std::move(identifier), std::move(metadata), std::move(metadata_location),
-      std::move(io), std::move(catalog), std::move(full_name), std::move(reporter)));
+  return std::shared_ptr<Table>(new Table(std::move(identifier), std::move(metadata),
+                                          std::move(metadata_location), std::move(io),
+                                          std::move(catalog), std::move(full_name),
+                                          std::move(reporter), std::move(labels)));
 }
 
 Table::~Table() = default;
 
 Table::Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
              std::string metadata_location, std::shared_ptr<FileIO> io,
              std::shared_ptr<Catalog> catalog, std::string full_name,
-             std::shared_ptr<MetricsReporter> reporter)
+             std::shared_ptr<MetricsReporter> reporter, Labels labels)
     : identifier_(std::move(identifier)),
       full_name_(full_name.empty() ? identifier_.ToString() : std::move(full_name)),
       metadata_(std::move(metadata)),
       metadata_location_(std::move(metadata_location)),
       io_(std::move(io)),
       catalog_(std::move(catalog)),
       reporter_(std::move(reporter)),
-      metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())) {}
+      metadata_cache_(std::make_unique<TableMetadataCache>(metadata_.get())),
+      labels_(std::move(labels)) {}
 
 const std::string& Table::uuid() const { return metadata_->table_uuid; }
 
@@ -167,6 +167,8 @@ 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_; }
+
 Result<std::unique_ptr<LocationProvider>> Table::location_provider() const {
   return LocationProvider::Make(metadata_->location, metadata_->properties);
 }

src/iceberg/table.h

@@ -29,6 +29,7 @@
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
+#include "iceberg/labels.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/type_fwd.h"
@@ -50,11 +51,12 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   ///            string representation of identifier when empty.
   /// \param[in] reporter Optional metrics reporter for this table. Defaults to nullptr
   ///            (noop).
+  /// \param[in] labels Catalog-provided labels for this table. Defaults to empty.
   static Result<std::shared_ptr<Table>> Make(
       TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
       std::string metadata_location, std::shared_ptr<FileIO> io,
       std::shared_ptr<Catalog> catalog, std::string full_name = "",
-      std::shared_ptr<MetricsReporter> reporter = nullptr);
+      std::shared_ptr<MetricsReporter> reporter = nullptr, Labels labels = {});
 
   virtual ~Table();
 
@@ -130,6 +132,13 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Returns the metrics reporter for this table.
   const std::shared_ptr<MetricsReporter>& reporter() const;
 
+  /// \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.
+  const Labels& labels() const;
+
   /// \brief Returns a LocationProvider for this table
   Result<std::unique_ptr<LocationProvider>> location_provider() const;
 
@@ -216,7 +225,7 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
         std::string metadata_location, std::shared_ptr<FileIO> io,
         std::shared_ptr<Catalog> catalog, std::string full_name,
-        std::shared_ptr<MetricsReporter> reporter = nullptr);
+        std::shared_ptr<MetricsReporter> reporter = nullptr, Labels labels = {});
 
   const TableIdentifier identifier_;
   const std::string full_name_;
@@ -226,6 +235,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_;
 };
 
 /// \brief A table created by stage-create and not yet committed.