Core: expose labels on the loaded table via SupportsLabels

core/src/main/java/org/apache/iceberg/BaseTable.java

@@ -29,6 +29,7 @@
 import org.apache.iceberg.metrics.MetricsReporter;
 import org.apache.iceberg.metrics.MetricsReporters;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.rest.labels.Labels;
 
 /**
  * Base {@link Table} implementation.
@@ -40,20 +41,34 @@
  * when reading the table data after deserialization.
  */
 public class BaseTable
-    implements Table, HasTableOperations, Serializable, SupportsDistributedScanPlanning {
+    implements Table,
+        HasTableOperations,
+        Serializable,
+        SupportsDistributedScanPlanning,
+        SupportsLabels {
   private final TableOperations ops;
   private final String name;
   private MetricsReporter reporter;
+  // Ephemeral catalog enrichment, not table state. Transient because it is not preserved across
+  // serialization: writeReplace() swaps in a SerializableTable for Java serialization, and any
+  // path that bypasses writeReplace (e.g. Kryo) leaves this null, handled by labels().
+  private final transient Labels labels;
 
   public BaseTable(TableOperations ops, String name) {
     this(ops, name, LoggingMetricsReporter.instance());
   }
 
   public BaseTable(TableOperations ops, String name, MetricsReporter reporter) {
+    this(ops, name, reporter, Labels.empty());
+  }
+
+  public BaseTable(TableOperations ops, String name, MetricsReporter reporter, Labels labels) {
     Preconditions.checkNotNull(reporter, "reporter cannot be null");
+    Preconditions.checkNotNull(labels, "labels cannot be null");
     this.ops = ops;
     this.name = name;
     this.reporter = reporter;
+    this.labels = labels;
   }
 
   public MetricsReporter reporter() {
@@ -74,6 +89,12 @@ public String name() {
     return name;
   }
 
+  @Override
+  public Labels labels() {
+    // labels is null only when this instance was deserialized bypassing writeReplace (e.g. Kryo).
+    return labels != null ? labels : Labels.empty();
+  }
+
   @Override
   public void refresh() {
     ops.refresh();

core/src/main/java/org/apache/iceberg/SupportsLabels.java

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+package org.apache.iceberg;
+
+import org.apache.iceberg.rest.labels.Labels;
+
+/**
+ * Implemented by tables that can expose catalog-provided labels obtained from the load response.
+ *
+ * <p>Labels are optional, catalog-provided metadata enrichment. They are not part of table state
+ * and are not preserved when the table is serialized.
+ */
+public interface SupportsLabels {
+  /**
+   * Returns the catalog-provided labels for this table, or an empty instance when there are none.
+   */
+  Labels labels();
+}

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -79,6 +79,7 @@
 import org.apache.iceberg.rest.auth.AuthManagers;
 import org.apache.iceberg.rest.auth.AuthSession;
 import org.apache.iceberg.rest.credentials.Credential;
+import org.apache.iceberg.rest.labels.Labels;
 import org.apache.iceberg.rest.requests.CommitTransactionRequest;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
@@ -557,10 +558,11 @@ public Table loadTable(SessionContext context, TableIdentifier identifier) {
     }
 
     List<Credential> credentials = response.credentials();
+    Labels labels = response.labels();
     RESTClient tableClient = client.withAuthSession(tableSession);
     Supplier<BaseTable> tableSupplier =
         createTableSupplier(
-            finalIdentifier, tableMetadata, context, tableClient, tableConf, credentials);
+            finalIdentifier, tableMetadata, context, tableClient, tableConf, credentials, labels);
 
     String eTag = responseHeaders.getOrDefault(HttpHeaders.ETAG, null);
     if (eTag != null) {
@@ -580,7 +582,8 @@ private Supplier<BaseTable> createTableSupplier(
       SessionContext context,
       RESTClient tableClient,
       Map<String, String> tableConf,
-      List<Credential> credentials) {
+      List<Credential> credentials,
+      Labels labels) {
     return () -> {
       RESTTableOperations ops =
           newTableOps(
@@ -594,21 +597,25 @@ private Supplier<BaseTable> createTableSupplier(
 
       trackFileIO(ops);
 
-      RESTTable table = restTableForScanPlanning(ops, identifier, tableClient, tableConf);
+      RESTTable table = restTableForScanPlanning(ops, identifier, tableClient, tableConf, labels);
       if (table != null) {
         return table;
       }
 
       return new BaseTable(
-          ops, fullTableName(identifier), metricsReporter(paths.metrics(identifier), tableClient));
+          ops,
+          fullTableName(identifier),
+          metricsReporter(paths.metrics(identifier), tableClient),
+          labels);
     };
   }
 
   private RESTTable restTableForScanPlanning(
       TableOperations ops,
       TableIdentifier finalIdentifier,
       RESTClient restClient,
-      Map<String, String> tableConf) {
+      Map<String, String> tableConf,
+      Labels labels) {
     String planningModeServerConfig = tableConf.get(RESTCatalogProperties.SCAN_PLANNING_MODE);
     ScanPlanningMode serverScanPlanningMode =
         planningModeServerConfig == null
@@ -653,7 +660,8 @@ private RESTTable restTableForScanPlanning(
           paths,
           endpoints,
           properties(),
-          conf);
+          conf,
+          labels);
     }
 
     // Default to client-side planning
@@ -740,13 +748,17 @@ public Table registerTable(
 
     trackFileIO(ops);
 
-    RESTTable restTable = restTableForScanPlanning(ops, ident, tableClient, tableConf);
+    RESTTable restTable =
+        restTableForScanPlanning(ops, ident, tableClient, tableConf, response.labels());
     if (restTable != null) {
       return restTable;
     }
 
     return new BaseTable(
-        ops, fullTableName(ident), metricsReporter(paths.metrics(ident), tableClient));
+        ops,
+        fullTableName(ident),
+        metricsReporter(paths.metrics(ident), tableClient),
+        response.labels());
   }
 
   @Override
@@ -1009,13 +1021,17 @@ public Table create() {
 
       trackFileIO(ops);
 
-      RESTTable restTable = restTableForScanPlanning(ops, ident, tableClient, tableConf);
+      RESTTable restTable =
+          restTableForScanPlanning(ops, ident, tableClient, tableConf, response.labels());
       if (restTable != null) {
         return restTable;
       }
 
       return new BaseTable(
-          ops, fullTableName(ident), metricsReporter(paths.metrics(ident), tableClient));
+          ops,
+          fullTableName(ident),
+          metricsReporter(paths.metrics(ident), tableClient),
+          response.labels());
     }
 
     @Override

core/src/main/java/org/apache/iceberg/rest/RESTTable.java

@@ -30,6 +30,7 @@
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.metrics.MetricsReporter;
+import org.apache.iceberg.rest.labels.Labels;
 
 class RESTTable extends BaseTable implements SupportsDistributedScanPlanning {
   private final RESTClient client;
@@ -51,8 +52,9 @@ class RESTTable extends BaseTable implements SupportsDistributedScanPlanning {
       ResourcePaths resourcePaths,
       Set<Endpoint> supportedEndpoints,
       Map<String, String> catalogProperties,
-      Object hadoopConf) {
-    super(ops, name, reporter);
+      Object hadoopConf,
+      Labels labels) {
+    super(ops, name, reporter, labels);
     this.reporter = reporter;
     this.client = client;
     this.headers = headers;

core/src/main/java/org/apache/iceberg/rest/labels/FieldLabels.java

@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+package org.apache.iceberg.rest.labels;
+
+import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.immutables.value.Value;
+
+/** Labels attached to a single field, identified by its field id. See {@link Labels}. */
+@Value.Immutable
+public interface FieldLabels {
+  int fieldId();
+
+  Map<String, String> labels();
+
+  @Value.Check
+  default void validate() {
+    Preconditions.checkArgument(fieldId() >= 1, "Invalid field id: must be >= 1");
+    Preconditions.checkArgument(!labels().isEmpty(), "Invalid labels: must be non-empty");
+  }
+}

core/src/main/java/org/apache/iceberg/rest/labels/FieldLabelsParser.java

@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.iceberg.rest.labels;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+public class FieldLabelsParser {
+  private static final String FIELD_ID = "field-id";
+  private static final String LABELS = "labels";
+
+  private FieldLabelsParser() {}
+
+  public static String toJson(FieldLabels fieldLabels) {
+    return toJson(fieldLabels, false);
+  }
+
+  public static String toJson(FieldLabels fieldLabels, boolean pretty) {
+    return JsonUtil.generate(gen -> toJson(fieldLabels, gen), pretty);
+  }
+
+  public static void toJson(FieldLabels fieldLabels, JsonGenerator gen) throws IOException {
+    Preconditions.checkArgument(null != fieldLabels, "Invalid field labels: null");
+
+    gen.writeStartObject();
+
+    gen.writeNumberField(FIELD_ID, fieldLabels.fieldId());
+    JsonUtil.writeStringMap(LABELS, fieldLabels.labels(), gen);
+
+    gen.writeEndObject();
+  }
+
+  public static FieldLabels fromJson(String json) {
+    return JsonUtil.parse(json, FieldLabelsParser::fromJson);
+  }
+
+  public static FieldLabels fromJson(JsonNode json) {
+    Preconditions.checkArgument(null != json, "Cannot parse field labels from null object");
+
+    return ImmutableFieldLabels.builder()
+        .fieldId(JsonUtil.getInt(FIELD_ID, json))
+        .labels(JsonUtil.getStringMap(LABELS, json))
+        .build();
+  }
+}

core/src/main/java/org/apache/iceberg/rest/labels/Labels.java

@@ -0,0 +1,45 @@
+/*
+ * 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.
+ */
+package org.apache.iceberg.rest.labels;
+
+import java.util.List;
+import java.util.Map;
+import org.immutables.value.Value;
+
+/** Optional catalog-provided labels returned on a load response. */
+@Value.Immutable
+public interface Labels {
+  Labels EMPTY = ImmutableLabels.builder().build();
+
+  /** Object-level labels. */
+  Map<String, String> object();
+
+  /** Field-level labels, keyed by field id. */
+  List<FieldLabels> fields();
+
+  /** Returns true when there are neither object-level nor field-level labels. */
+  default boolean isEmpty() {
+    return object().isEmpty() && fields().isEmpty();
+  }
+
+  /** Returns a shared empty instance. */
+  static Labels empty() {
+    return EMPTY;
+  }
+}