Add support for unique-table-location

CHANGELOG.md

@@ -41,6 +41,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
     - Names containing any of these characters: <code>/\:*?"<>|#+`</code>
 
 ### New Features
+- Added support for Iceberg's `unique-table-location` catalog property for default table locations (direct and staged create, including prefixed warehouse layouts). View default locations are unchanged.
 - Added `SESSION_NAME_FIELDS_IN_SUBSCOPED_CREDENTIAL` feature flag for AWS credential vending. Operators can now configure an ordered list of fields (`realm`, `catalog`, `namespace`, `table`, `principal`) to compose structured STS role session names (e.g. `p-acme-hr_catalog-employee-etl_writer`). Session names are sanitized and proportionally truncated to the AWS 64-character limit. When unset, existing `INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL` behaviour is preserved.
 - Added `hostUsers` support in Helm chart.
 - Added documentation for BigQuery Metastore Catalog federation. Build with `-PNonRESTCatalogs=BIGQUERY` to include the BigQueryMetastoreCatalog federation extension. See `site/content/in-dev/unreleased/federation/bigquery-metastore-federation.md`.

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -49,6 +49,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogProperties;
@@ -60,6 +61,7 @@
 import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -183,6 +185,14 @@ public class IcebergCatalog extends BaseMetastoreViewCatalog
   private final PolarisEventMetadataFactory eventMetadataFactory;
   private final AtomicBoolean loggedPrefixOverlapWarning = new AtomicBoolean(false);
 
+  /**
+   * Set while a {@link TableBuilder} is deriving a default table location so {@link
+   * #defaultWarehouseLocation} can honor {@link CatalogProperties#UNIQUE_TABLE_LOCATION} without
+   * applying unique suffixes to view default locations (views use the same Iceberg hook).
+   */
+  private final ThreadLocal<Boolean> deriveTableDefaultLocationWithUniqueSuffix =
+      ThreadLocal.withInitial(() -> false);
+
   private String ioImplClassName;
   private FileIO catalogFileIO;
   private CloseableGroup closeableGroup;
@@ -364,9 +374,23 @@ protected TableOperations newTableOps(TableIdentifier tableIdentifier) {
 
   @Override
   protected String defaultWarehouseLocation(TableIdentifier tableIdentifier) {
+    boolean useUniqueTableLocation =
+        useUniqueTableLocation() && deriveTableDefaultLocationWithUniqueSuffix.get();
+    return resolveDefaultWarehouseLocation(tableIdentifier, useUniqueTableLocation);
+  }
+
+  private boolean useUniqueTableLocation() {
+    return PropertyUtil.propertyAsBoolean(
+        properties(),
+        CatalogProperties.UNIQUE_TABLE_LOCATION,
+        CatalogProperties.UNIQUE_TABLE_LOCATION_DEFAULT);
+  }
+
+  private String resolveDefaultWarehouseLocation(
+      TableIdentifier tableIdentifier, boolean useUniqueTableLocation) {
+    String tableLocation = LocationUtil.tableLocation(tableIdentifier, useUniqueTableLocation);
     if (tableIdentifier.namespace().isEmpty()) {
-      return SLASH.join(
-          defaultNamespaceLocation(tableIdentifier.namespace()), tableIdentifier.name());
+      return SLASH.join(defaultNamespaceLocation(tableIdentifier.namespace()), tableLocation);
     } else {
       PolarisResolvedPathWrapper resolvedNamespace =
           resolvedEntityView.getResolvedPath(
@@ -376,7 +400,7 @@ protected String defaultWarehouseLocation(TableIdentifier tableIdentifier) {
       }
       List<PolarisEntity> namespacePath = resolvedNamespace.getRawFullPath();
       String namespaceLocation = resolveLocationForPath(diagnostics, namespacePath);
-      return SLASH.join(namespaceLocation, tableIdentifier.name());
+      return SLASH.join(namespaceLocation, tableLocation);
     }
   }
 
@@ -1000,7 +1024,10 @@ private String buildPrefixedLocation(TableIdentifier tableIdentifier) {
     }
     locationBuilder
         .append("/")
-        .append(URLEncoder.encode(tableIdentifier.name(), Charset.defaultCharset()))
+        .append(
+            URLEncoder.encode(
+                LocationUtil.tableLocation(tableIdentifier, useUniqueTableLocation()),
+                Charset.defaultCharset()))
         .append("/");
     return locationBuilder.toString();
   }
@@ -1407,6 +1434,35 @@ public PolarisIcebergCatalogTableBuilder(TableIdentifier identifier, Schema sche
     public TableBuilder withLocation(String newLocation) {
       return super.withLocation(transformTableLikeLocation(identifier, newLocation));
     }
+
+    @Override
+    public Table create() {
+      return withUniqueTableDefaultLocation(super::create);
+    }
+
+    @Override
+    public Transaction createTransaction() {
+      return withUniqueTableDefaultLocation(super::createTransaction);
+    }
+
+    @Override
+    public Transaction replaceTransaction() {
+      return withUniqueTableDefaultLocation(super::replaceTransaction);
+    }
+
+    @Override
+    public Transaction createOrReplaceTransaction() {
+      return withUniqueTableDefaultLocation(super::createOrReplaceTransaction);
+    }
+  }
+
+  private <T> T withUniqueTableDefaultLocation(Supplier<T> action) {
+    deriveTableDefaultLocationWithUniqueSuffix.set(true);
+    try {
+      return action.get();
+    } finally {
+      deriveTableDefaultLocationWithUniqueSuffix.remove();
+    }
   }
 
   private class PolarisIcebergCatalogViewBuilder extends BaseMetastoreViewCatalog.BaseViewBuilder {

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java

@@ -2799,14 +2799,51 @@ public void testPaginatedListNamespaces() {
   }
 
   @Test
-  @Disabled("Test is not compatible with REST catalogs")
-  @Override
-  public void testLoadTableWithMissingMetadataFile(@TempDir Path tempDir) {}
+  public void stagedCreateUsesUniqueTableLocationWhenEnabled() {
+    IcebergCatalog uniqueCatalog =
+        initCatalog(
+            "staged_unique_catalog",
+            ImmutableMap.of(CatalogProperties.UNIQUE_TABLE_LOCATION, "true"));
+    if (requiresNamespaceCreate()) {
+      uniqueCatalog.createNamespace(NS);
+    }
+
+    String stagedLocation =
+        uniqueCatalog
+            .buildTable(TableIdentifier.of(NS, "staged_unique_table"), SCHEMA)
+            .createTransaction()
+            .table()
+            .location();
+
+    assertThat(stagedLocation).contains("staged_unique_table-");
+    assertThat(stagedLocation).doesNotEndWith("/staged_unique_table");
+  }
+
+  @Test
+  public void viewDefaultLocationIgnoresUniqueTableLocationProperty() {
+    IcebergCatalog uniqueCatalog =
+        initCatalog(
+            "view_unique_catalog",
+            ImmutableMap.of(CatalogProperties.UNIQUE_TABLE_LOCATION, "true"));
+    uniqueCatalog.createNamespace(NS);
+
+    TableIdentifier viewId = TableIdentifier.of(NS, "unique_loc_view");
+    uniqueCatalog
+        .buildView(viewId)
+        .withSchema(SCHEMA)
+        .withDefaultNamespace(NS)
+        .withQuery("spark", VIEW_QUERY)
+        .create();
+
+    assertThat(uniqueCatalog.loadView(viewId).location()).endsWith("/unique_loc_view");
+    assertThat(uniqueCatalog.loadView(viewId).location())
+        .doesNotMatch(".*\\/unique_loc_view-[0-9a-f]+");
+  }
 
   @Test
-  @Disabled("Feature is not implemented yet")
+  @Disabled("Test is not compatible with REST catalogs")
   @Override
-  public void createTableInUniqueLocation() {}
+  public void testLoadTableWithMissingMetadataFile(@TempDir Path tempDir) {}
 
   @Test
   @Disabled(