Support hierarchical namespace in Azure

CHANGELOG.md

@@ -62,6 +62,9 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - Added `priorityClassName` support in Helm chart.
 - Added support for including principal name in subscoped credentials. `INCLUDE_PRINCIPAL_NAME_IN_SUBSCOPED_CREDENTIAL` (default: false) can be used to toggle this feature. If enabled, cached credentials issued to one principal will no longer be available for others.
 - Added support for [Kubernetes Gateway API](https://gateway-api.sigs.k8s.io/) to the Helm Chart. 
+- Added `hierarchical` flag to `AzureStorageConfigInfo` to allow more precise SAS token down-scoping in ADLS when
+  the [hierarchical namespace](https://learn.microsoft.com/en-us/azure/storage/blobs/data-lake-storage-namespace)
+  feature is enabled in Azure.
 
 ### Changes
 

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogAdlsIntegrationTestBase.java

@@ -19,6 +19,7 @@
 package org.apache.polaris.service.it.test;
 
 import java.util.List;
+import java.util.Optional;
 import org.apache.polaris.core.admin.model.AzureStorageConfigInfo;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 
@@ -27,13 +28,15 @@ public abstract class PolarisRestCatalogAdlsIntegrationTestBase
     extends PolarisRestCatalogIntegrationBase {
   public static final String TENANT_ID = System.getenv("INTEGRATION_TEST_AZURE_TENANT_ID");
   public static final String BASE_LOCATION = System.getenv("INTEGRATION_TEST_AZURE_PATH");
+  public static final String HIERARCHICAL = System.getenv("INTEGRATION_TEST_AZURE_HIERARCHICAL");
 
   @Override
   protected StorageConfigInfo getStorageConfigInfo() {
     return AzureStorageConfigInfo.builder()
         .setTenantId(TENANT_ID)
         .setStorageType(StorageConfigInfo.StorageTypeEnum.AZURE)
         .setAllowedLocations(List.of(BASE_LOCATION))
+        .setHierarchical(Optional.ofNullable(HIERARCHICAL).map(Boolean::parseBoolean).orElse(null))
         .build();
   }
 }

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogViewAdlsIntegrationTestBase.java

@@ -21,6 +21,7 @@
 import com.google.common.base.Strings;
 import java.nio.file.Path;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Stream;
 import org.apache.polaris.core.admin.model.AzureStorageConfigInfo;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
@@ -32,13 +33,15 @@ public abstract class PolarisRestCatalogViewAdlsIntegrationTestBase
     extends PolarisRestCatalogViewIntegrationBase {
   public static final String TENANT_ID = System.getenv("INTEGRATION_TEST_AZURE_TENANT_ID");
   public static final String BASE_LOCATION = System.getenv("INTEGRATION_TEST_AZURE_PATH");
+  public static final String HIERARCHICAL = System.getenv("INTEGRATION_TEST_AZURE_HIERARCHICAL");
 
   @Override
   protected StorageConfigInfo getStorageConfigInfo() {
     return AzureStorageConfigInfo.builder()
         .setTenantId(TENANT_ID)
         .setStorageType(StorageConfigInfo.StorageTypeEnum.AZURE)
         .setAllowedLocations(List.of(BASE_LOCATION))
+        .setHierarchical(Optional.ofNullable(HIERARCHICAL).map(Boolean::parseBoolean).orElse(null))
         .build();
   }
 

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -181,6 +181,7 @@ private StorageConfigInfo getStorageInfo(Map<String, String> internalProperties)
             .setConsentUrl(azureConfig.getConsentUrl())
             .setStorageType(AZURE)
             .setAllowedLocations(azureConfig.getAllowedLocations())
+            .setHierarchical(azureConfig.isHierarchical())
             .build();
       }
       if (configInfo instanceof GcpStorageConfigurationInfo) {
@@ -330,6 +331,7 @@ public Builder setStorageConfigurationInfo(
                     .tenantId(azureConfigModel.getTenantId())
                     .multiTenantAppName(azureConfigModel.getMultiTenantAppName())
                     .consentUrl(azureConfigModel.getConsentUrl())
+                    .hierarchical(azureConfigModel.getHierarchical())
                     .build();
             break;
           case GCS:

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureCredentialsStorageIntegration.java

@@ -36,12 +36,14 @@
 import com.azure.storage.blob.sas.BlobSasPermission;
 import com.azure.storage.blob.sas.BlobServiceSasSignatureValues;
 import com.azure.storage.file.datalake.DataLakeFileSystemClientBuilder;
+import com.azure.storage.file.datalake.DataLakePathClientBuilder;
 import com.azure.storage.file.datalake.DataLakeServiceClient;
 import com.azure.storage.file.datalake.DataLakeServiceClientBuilder;
 import com.azure.storage.file.datalake.models.DataLakeStorageException;
 import com.azure.storage.file.datalake.sas.DataLakeServiceSasSignatureValues;
 import com.azure.storage.file.datalake.sas.PathSasPermission;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import jakarta.annotation.Nonnull;
 import java.time.Duration;
 import java.time.Instant;
@@ -169,6 +171,17 @@ public StorageAccessConfig getSubscopedCreds(
               blobSasPermission,
               Mono.just(accessToken));
     } else if (location.getEndpoint().equalsIgnoreCase(AzureLocation.ADLS_ENDPOINT)) {
+      String path = null;
+      if (Boolean.TRUE.equals(config().isHierarchical())) {
+        Preconditions.checkArgument(
+            allowedReadLocations.size() <= 1,
+            "Allowed read locations must not have more that one entry");
+        Preconditions.checkArgument(
+            allowedWriteLocations.size() <= 1,
+            "Allowed write locations must not have more that one entry");
+        path = location.getFilePath();
+      }
+
       sasToken =
           getAdlsUserDelegationSas(
               startTime,
@@ -177,6 +190,7 @@ public StorageAccessConfig getSubscopedCreds(
               storageDnsName,
               location.getContainer(),
               pathSasPermission,
+              path,
               Mono.just(accessToken));
     } else {
       throw new RuntimeException(
@@ -275,6 +289,7 @@ private String getAdlsUserDelegationSas(
       String storageDnsName,
       String fileSystemNameOrContainer,
       PathSasPermission pathSasPermission,
+      String path,
       Mono<AccessToken> accessTokenMono) {
     String endpoint = "https://" + storageDnsName;
     try {
@@ -289,11 +304,21 @@ private String getAdlsUserDelegationSas(
       DataLakeServiceSasSignatureValues signatureValues =
           new DataLakeServiceSasSignatureValues(sasExpiry, pathSasPermission);
 
-      return new DataLakeFileSystemClientBuilder()
-          .endpoint(endpoint)
-          .fileSystemName(fileSystemNameOrContainer)
-          .buildClient()
-          .generateUserDelegationSas(signatureValues, userDelegationKey);
+      if (path != null) {
+        return new DataLakePathClientBuilder()
+            .endpoint(endpoint)
+            .fileSystemName(fileSystemNameOrContainer)
+            .pathName(path)
+            .buildDirectoryClient()
+            .generateUserDelegationSas(signatureValues, userDelegationKey);
+
+      } else {
+        return new DataLakeFileSystemClientBuilder()
+            .endpoint(endpoint)
+            .fileSystemName(fileSystemNameOrContainer)
+            .buildClient()
+            .generateUserDelegationSas(signatureValues, userDelegationKey);
+      }
     } catch (DataLakeStorageException ex) {
       LOGGER.debug(
           "Azure DataLakeStorageException for getAdlsUserDelegationSas. keyStart={} keyEnd={}, storageDns={}, fileSystemName={}",

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureStorageConfigurationInfo.java

@@ -58,6 +58,10 @@ public StorageType getStorageType() {
   @Nullable
   public abstract String getConsentUrl();
 
+  /** The flag indicating whether the storage account supports hierarchical namespaces. */
+  @Nullable
+  public abstract Boolean isHierarchical();
+
   @Override
   public void validatePrefixForStorageType(String loc) {
     AzureLocation location = new AzureLocation(loc);

runtime/service/src/cloudTest/java/org/apache/polaris/service/it/CredentialVendingProfile.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.polaris.service.it;
+
+import com.google.common.collect.ImmutableMap;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import java.util.Map;
+
+public class CredentialVendingProfile implements QuarkusTestProfile {
+
+  @Override
+  public Map<String, String> getConfigOverrides() {
+    return ImmutableMap.<String, String>builder()
+        .put("polaris.features.\"SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION\"", "false")
+        .build();
+  }
+}

runtime/service/src/cloudTest/java/org/apache/polaris/service/it/RestCatalogAdlsCredentialVendingIT.java

@@ -0,0 +1,29 @@
+/*
+ * 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.polaris.service.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import io.quarkus.test.junit.TestProfile;
+import org.apache.polaris.service.it.test.PolarisRestCatalogAdlsIntegrationTestBase;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+@QuarkusIntegrationTest
+@TestProfile(CredentialVendingProfile.class)
+@EnabledIfEnvironmentVariable(named = "INTEGRATION_TEST_AZURE_PATH", matches = ".+")
+public class RestCatalogAdlsCredentialVendingIT extends PolarisRestCatalogAdlsIntegrationTestBase {}

runtime/service/src/cloudTest/java/org/apache/polaris/service/it/RestCatalogViewAdlsCredentialVendingIT.java

@@ -0,0 +1,30 @@
+/*
+ * 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.polaris.service.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import io.quarkus.test.junit.TestProfile;
+import org.apache.polaris.service.it.test.PolarisRestCatalogViewAdlsIntegrationTestBase;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+@QuarkusIntegrationTest
+@TestProfile(CredentialVendingProfile.class)
+@EnabledIfEnvironmentVariable(named = "INTEGRATION_TEST_AZURE_PATH", matches = ".+")
+public class RestCatalogViewAdlsCredentialVendingIT
+    extends PolarisRestCatalogViewAdlsIntegrationTestBase {}

runtime/service/src/test/java/org/apache/polaris/service/entity/CatalogEntityTest.java

@@ -405,11 +405,11 @@ public void testAwsConfigJsonPropertiesPresence() throws JsonProcessingException
 
   @ParameterizedTest
   @MethodSource
-  public void testAwsConfigRoundTrip(AwsStorageConfigInfo config) throws JsonProcessingException {
+  public void testStorageConfigRoundTrip(StorageConfigInfo config) throws JsonProcessingException {
     String configStr = MAPPER.writeValueAsString(config);
     CatalogEntity catalogEntity =
         new CatalogEntity.Builder()
-            .setName("testAwsConfigRoundTrip")
+            .setName("testStorageConfigRoundTrip")
             .setDefaultBaseLocation(config.getAllowedLocations().getFirst())
             .setCatalogType(Catalog.TypeEnum.INTERNAL.name())
             .setStorageConfigurationInfo(
@@ -482,18 +482,36 @@ icebergRestConnectionConfigInfoModel, null, new AwsIamServiceIdentityInfoDpo(nul
         .isEqualTo("arn:aws:iam::123456789012:user/test-user");
   }
 
-  public static Stream<Arguments> testAwsConfigRoundTrip() {
+  public static Stream<Arguments> testStorageConfigRoundTrip() {
     AwsStorageConfigInfo.Builder b =
         AwsStorageConfigInfo.builder()
             .setStorageType(StorageConfigInfo.StorageTypeEnum.S3)
             .setAllowedLocations(List.of("s3://example.com"))
             .setRoleArn("arn:aws:iam::012345678901:role/test-role");
+    AzureStorageConfigInfo.Builder a =
+        AzureStorageConfigInfo.builder()
+            .setStorageType(StorageConfigInfo.StorageTypeEnum.AZURE)
+            .setTenantId("test-tenant")
+            .setAllowedLocations(List.of("abfss://test@example.dfs.core.windows.net/"));
     return Stream.of(
         Arguments.of(b.build()),
         Arguments.of(b.setExternalId("ex1").build()),
         Arguments.of(b.setRegion("us-west-2").build()),
         Arguments.of(b.setEndpoint("http://s3.example.com:1234").build()),
         Arguments.of(b.setStsEndpoint("http://sts.example.com:1234").build()),
-        Arguments.of(b.setPathStyleAccess(true).build()));
+        Arguments.of(b.setPathStyleAccess(true).build()),
+        Arguments.of(a.build()),
+        Arguments.of(a.setHierarchical(true).build()));
+  }
+
+  @Test
+  public void testAzureConfigJsonPropertiesPresence() throws JsonProcessingException {
+    AzureStorageConfigInfo.Builder b =
+        AzureStorageConfigInfo.builder().setStorageType(StorageConfigInfo.StorageTypeEnum.AZURE);
+    assertThat(MAPPER.writeValueAsString(b.build())).contains("storageType");
+    assertThat(MAPPER.writeValueAsString(b.build())).doesNotContain("hierarchical");
+
+    b.setHierarchical(true);
+    assertThat(MAPPER.writeValueAsString(b.build())).contains("hierarchical");
   }
 }

spec/polaris-management-service.yml

@@ -1166,6 +1166,13 @@ components:
             consentUrl:
               type: string
               description: URL to the Azure permissions request page
+            hierarchical:
+              type: boolean
+              description: >-
+                If set to `true`, instructs Polaris Servers to scope SAS tokens down to the most specific path
+                in the storage container (in most cases the table's base location). This flag should be set only
+                if hierarchical namespace is enabled in the Azure storage account. Using this feature with 
+                non-hierarchical storage will lead to storage authorization errors in runtime in most cases.
           required:
             - tenantId