Spark 4.1: expose catalog labels via metadata table and DESCRIBE

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/LabelsTable.java

@@ -0,0 +1,120 @@
+/*
+ * 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 java.util.List;
+import java.util.Map;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.rest.labels.FieldLabels;
+import org.apache.iceberg.rest.labels.Labels;
+import org.apache.iceberg.types.Types;
+
+/**
+ * A {@link Table} implementation that exposes the catalog-provided labels of its base table as
+ * rows.
+ *
+ * <p>Each label key-value pair is a row. Object-level labels have {@code scope = "object"} and a
+ * null {@code field_id}; field-level labels have {@code scope = "field"} and the field id they are
+ * attached to. Labels are catalog-provided enrichment obtained at load time; a re-scan re-reads the
+ * base table's labels and may observe different values.
+ */
+public class LabelsTable extends BaseMetadataTable {
+  private static final String OBJECT_SCOPE = "object";
+  private static final String FIELD_SCOPE = "field";
+
+  private static final Schema LABELS_SCHEMA =
+      new Schema(
+          Types.NestedField.required(1, "scope", Types.StringType.get()),
+          Types.NestedField.optional(2, "field_id", Types.IntegerType.get()),
+          Types.NestedField.required(3, "key", Types.StringType.get()),
+          Types.NestedField.required(4, "value", Types.StringType.get()));
+
+  LabelsTable(Table table) {
+    this(table, table.name() + ".labels");
+  }
+
+  LabelsTable(Table table, String name) {
+    super(table, name);
+  }
+
+  @Override
+  MetadataTableType metadataTableType() {
+    return MetadataTableType.LABELS;
+  }
+
+  @Override
+  public TableScan newScan() {
+    return new LabelsScan(table());
+  }
+
+  @Override
+  public Schema schema() {
+    return LABELS_SCHEMA;
+  }
+
+  private Labels labels() {
+    return table() instanceof SupportsLabels
+        ? ((SupportsLabels) table()).labels()
+        : Labels.empty();
+  }
+
+  private DataTask task(TableScan scan) {
+    Labels labels = labels();
+    List<StaticDataTask.Row> rows = Lists.newArrayList();
+    for (Map.Entry<String, String> entry : labels.object().entrySet()) {
+      rows.add(StaticDataTask.Row.of(OBJECT_SCOPE, null, entry.getKey(), entry.getValue()));
+    }
+
+    for (FieldLabels fieldLabels : labels.fields()) {
+      for (Map.Entry<String, String> entry : fieldLabels.labels().entrySet()) {
+        rows.add(
+            StaticDataTask.Row.of(
+                FIELD_SCOPE, fieldLabels.fieldId(), entry.getKey(), entry.getValue()));
+      }
+    }
+
+    return StaticDataTask.of(
+        table().io().newInputFile(table().operations().current().metadataFileLocation()),
+        schema(),
+        scan.schema(),
+        rows,
+        row -> row);
+  }
+
+  private class LabelsScan extends StaticTableScan {
+    LabelsScan(Table table) {
+      super(table, LABELS_SCHEMA, MetadataTableType.LABELS, LabelsTable.this::task);
+    }
+
+    LabelsScan(Table table, TableScanContext context) {
+      super(table, LABELS_SCHEMA, MetadataTableType.LABELS, LabelsTable.this::task, context);
+    }
+
+    @Override
+    protected TableScan newRefinedScan(Table table, Schema schema, TableScanContext context) {
+      return new LabelsScan(table, context);
+    }
+
+    @Override
+    public CloseableIterable<FileScanTask> planFiles() {
+      return CloseableIterable.withNoopClose(LabelsTable.this.task(this));
+    }
+  }
+}

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

@@ -36,7 +36,8 @@ public enum MetadataTableType {
   ALL_FILES,
   ALL_MANIFESTS,
   ALL_ENTRIES,
-  POSITION_DELETES;
+  POSITION_DELETES,
+  LABELS;
 
   public static MetadataTableType from(String name) {
     try {

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

@@ -88,6 +88,8 @@ private static Table createMetadataTableInstance(
         return new AllEntriesTable(baseTable, metadataTableName);
       case POSITION_DELETES:
         return new PositionDeletesTable(baseTable, metadataTableName);
+      case LABELS:
+        return new LabelsTable(baseTable, metadataTableName);
       default:
         throw new NoSuchTableException(
             "Unknown metadata table type: %s for %s", type, metadataTableName);

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");
+  }
+}