From f3d1d6fff990d3b20675af701b35b9e9dafee939 Mon Sep 17 00:00:00 2001
From: xylaaaaa <2392805527@qq.com>
Date: Fri, 14 Aug 2026 19:25:17 +0800
Subject: [PATCH] [fix](iceberg) Support Azure vended credentials

### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary: Databricks Unity Catalog returns Azure SAS credentials using Iceberg ADLS property names. Doris previously filtered those properties and did not package Iceberg ADLSFileIO, so catalog discovery could succeed while data reads failed. Passing the resulting Hadoop configuration to BE also risked logging the short-lived SAS value. This change normalizes account-scoped ADLS SAS credentials for Hadoop ABFS, packages the Iceberg Azure runtime, and stops logging HDFS configuration values.

### Release note

Support Azure SAS credentials vended by Iceberg REST catalogs such as Databricks Unity Catalog.

### Check List (For Author)

- Test:
    - Unit Test: CredentialUtilsTest, DefaultConnectorContextVendTest, and IcebergScanPlanProviderTest
    - Manual test: FE build and BE build with -j 8
- Behavior changed: Yes. Iceberg REST catalogs can use vended Azure SAS credentials for ADLS data access.
- Does this need documentation: No
---
 be/src/io/hdfs_builder.cpp                    |  3 +-
 fe/fe-connector/fe-connector-iceberg/pom.xml  | 12 +++++++
 .../iceberg/IcebergScanPlanProviderTest.java  | 13 +++++++
 .../connector/DefaultConnectorContext.java    |  8 ++---
 .../credentials/CredentialUtils.java          | 35 +++++++++++++++++++
 .../DefaultConnectorContextVendTest.java      | 27 ++++++++++++++
 .../credentials/CredentialUtilsTest.java      | 17 +++++++++
 fe/pom.xml                                    | 10 ++++++
 8 files changed, 120 insertions(+), 5 deletions(-)

diff --git a/be/src/io/hdfs_builder.cpp b/be/src/io/hdfs_builder.cpp
index cb9c75fb29debc..b004d41de53afa 100644
--- a/be/src/io/hdfs_builder.cpp
+++ b/be/src/io/hdfs_builder.cpp
@@ -210,7 +210,8 @@ Status create_hdfs_builder(const THdfsParams& hdfsParams, const std::string& fs_
         // set other conf
         for (const THdfsConf& conf : hdfsParams.hdfs_conf) {
             builder->set_hdfs_conf(conf.key, conf.value);
-            LOG(INFO) << "set hdfs config key: " << conf.key << ", value: " << conf.value;
+            // HDFS configuration may contain short-lived storage credentials such as an ADLS SAS token.
+            LOG(INFO) << "set hdfs config key: " << conf.key;
             if (strcmp(conf.key.c_str(), "hadoop.security.authentication") == 0) {
                 auth_type = conf.value;
             }
diff --git a/fe/fe-connector/fe-connector-iceberg/pom.xml b/fe/fe-connector/fe-connector-iceberg/pom.xml
index e3432510b22b49..5896b1f32e50a8 100644
--- a/fe/fe-connector/fe-connector-iceberg/pom.xml
+++ b/fe/fe-connector/fe-connector-iceberg/pom.xml
@@ -150,6 +150,18 @@ under the License.
             <version>${iceberg.version}</version>
         </dependency>
 
+        <!-- Iceberg Azure integration: ADLSFileIO plus its shaded Azure SDK dependencies. -->
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-azure</artifactId>
+            <version>${iceberg.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iceberg</groupId>
+            <artifactId>iceberg-azure-bundle</artifactId>
+            <version>${iceberg.version}</version>
+        </dependency>
+
         <!-- The hms-flavor org.apache.iceberg.hive.HiveCatalog is NOT a direct dependency here: it is
              bundled (with the shared relocated thrift shade.doris.hive.org.apache.thrift) inside
              fe-connector-hms-hive-shade and reaches this plugin transitively via fe-connector-hms above —
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index 9dc3c42da3311c..0defb916502781 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -65,6 +65,7 @@
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.ResolvingFileIO;
 import org.apache.iceberg.io.StorageCredential;
 import org.apache.iceberg.io.SupportsStorageCredentials;
 import org.apache.iceberg.types.Conversions;
@@ -2979,6 +2980,18 @@ public void extractVendedTokenReturnsIoPropsWhenFileIoHasNoStorageCredentials()
                 IcebergScanPlanProvider.extractVendedToken(table, true));
     }
 
+    @Test
+    public void adlsFileIoIsAvailableForAbfssLocations() {
+        ResolvingFileIO fileIO = new ResolvingFileIO();
+        fileIO.initialize(Collections.emptyMap());
+        try {
+            Assertions.assertEquals("org.apache.iceberg.azure.adlsv2.ADLSFileIO",
+                    fileIO.ioClass("abfss://container@account.dfs.core.windows.net/table").getName());
+        } finally {
+            fileIO.close();
+        }
+    }
+
     @Test
     public void extractVendedTokenEmptyWhenFlagDisabled() {
         FakeIcebergTable table = fakeTable("t1");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/connector/DefaultConnectorContext.java b/fe/fe-core/src/main/java/org/apache/doris/connector/DefaultConnectorContext.java
index b0728d2d389fdf..3ba57dd97f5193 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/connector/DefaultConnectorContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/connector/DefaultConnectorContext.java
@@ -235,7 +235,7 @@ public Map<String, String> vendStorageCredentials(Map<String, String> rawVendedC
     }
 
     /**
-     * Builds the vended {@link StorageAdapter} typed map from a raw per-table token: filter to
+     * Builds the vended {@link StorageAdapter} typed map from a raw per-table token: normalize
      * cloud-storage props, run {@link StorageAdapter#ofAll} (normalizes arbitrary token key
      * shapes + derives region/endpoint), then index by {@link StorageTypeId}. Mirrors the
      * legacy vended-credentials normalization tail exactly, so the BE-credential overlay
@@ -250,11 +250,11 @@ private Map<StorageTypeId, StorageAdapter> buildVendedStorageMap(
             return null;
         }
         try {
-            Map<String, String> filtered = CredentialUtils.filterCloudStorageProperties(rawVendedCredentials);
-            if (filtered.isEmpty()) {
+            Map<String, String> normalized = CredentialUtils.normalizeCloudStorageProperties(rawVendedCredentials);
+            if (normalized.isEmpty()) {
                 return null;
             }
-            List<StorageAdapter> vended = StorageAdapter.ofAll(filtered);
+            List<StorageAdapter> vended = StorageAdapter.ofAll(normalized);
             return vended.stream()
                     .collect(Collectors.toMap(StorageAdapter::getType, Function.identity()));
         } catch (Exception e) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/credentials/CredentialUtils.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/credentials/CredentialUtils.java
index 2c5e734d21962b..99ac1d08404c81 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/credentials/CredentialUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/credentials/CredentialUtils.java
@@ -31,6 +31,11 @@
  */
 public class CredentialUtils {
 
+    private static final String ADLS_SAS_TOKEN_PREFIX = "adls.sas-token.";
+    private static final String ADLS_PROPERTY_PREFIX = "adls.";
+    private static final String HADOOP_AZURE_ACCOUNT_AUTH_TYPE_PREFIX = "fs.azure.account.auth.type.";
+    private static final String HADOOP_AZURE_FIXED_SAS_TOKEN_PREFIX = "fs.azure.sas.fixed.token.";
+
     /**
      * Supported cloud storage prefixes for filtering vended credentials
      */
@@ -42,6 +47,7 @@ public class CredentialUtils {
             "obs.",          // Huawei OBS
             "gs.",           // Google Cloud Storage
             "azure.",        // Microsoft Azure
+            "adls.",         // Iceberg Azure ADLS vended credentials
             "client.",       // Iceberg client properties (e.g., client.region)
             "iceberg.rest."  // Iceberg REST catalog properties (e.g., iceberg.rest.access-key-id)
     ));
@@ -67,6 +73,35 @@ public static Map<String, String> filterCloudStorageProperties(Map<String, Strin
         return filtered;
     }
 
+    /**
+     * Convert cloud storage credentials to the properties consumed by Doris storage adapters.
+     * Databricks Unity Catalog returns Azure SAS credentials using Iceberg's
+     * {@code adls.sas-token.<account-host>} property. Hadoop ABFS instead consumes an account-scoped
+     * authentication type and fixed SAS token, so translate that representation before selecting the
+     * storage adapter. Remove the raw Iceberg {@code adls.*} property names after translating them to
+     * the equivalent backend configuration.
+     *
+     * @param rawVendedCredentials Raw vended credentials map
+     * @return Normalized cloud storage properties
+     */
+    public static Map<String, String> normalizeCloudStorageProperties(
+            Map<String, String> rawVendedCredentials) {
+        Map<String, String> normalized = filterCloudStorageProperties(rawVendedCredentials);
+        Map<String, String> adlsSasTokens = new HashMap<>();
+        normalized.forEach((key, value) -> {
+            if (key.startsWith(ADLS_SAS_TOKEN_PREFIX)) {
+                String accountHost = key.substring(ADLS_SAS_TOKEN_PREFIX.length());
+                adlsSasTokens.put(accountHost, value);
+            }
+        });
+        normalized.keySet().removeIf(key -> key.startsWith(ADLS_PROPERTY_PREFIX));
+        adlsSasTokens.forEach((accountHost, sasToken) -> {
+            normalized.put(HADOOP_AZURE_ACCOUNT_AUTH_TYPE_PREFIX + accountHost, "SAS");
+            normalized.put(HADOOP_AZURE_FIXED_SAS_TOKEN_PREFIX + accountHost, sasToken);
+        });
+        return normalized;
+    }
+
     /**
      * Extract backend properties from StorageAdapter map
      * Reference: CatalogProperty.getBackendStorageProperties()
diff --git a/fe/fe-core/src/test/java/org/apache/doris/connector/DefaultConnectorContextVendTest.java b/fe/fe-core/src/test/java/org/apache/doris/connector/DefaultConnectorContextVendTest.java
index b8314046dd5512..d9f11481d4a193 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/connector/DefaultConnectorContextVendTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/connector/DefaultConnectorContextVendTest.java
@@ -17,6 +17,9 @@
 
 package org.apache.doris.connector;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
+import org.apache.hadoop.fs.azurebfs.services.AuthType;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -60,6 +63,30 @@ public void normalizesOssTokenToBackendAwsProps() {
         Assertions.assertEquals("testSessionToken789", be.get("AWS_TOKEN"));
     }
 
+    @Test
+    public void normalizesAdlsTokenToBackendAbfsProps() throws Exception {
+        String accountHost = "account.dfs.core.windows.net";
+        String sasToken = "testSasToken";
+        Map<String, String> token = Map.of(
+                "adls.sas-token." + accountHost, sasToken,
+                "adls.sas-token-expires-at-ms." + accountHost, "4102444800000");
+
+        Map<String, String> be = context().vendStorageCredentials(token);
+
+        String authTypeKey = "fs.azure.account.auth.type." + accountHost;
+        String fixedSasTokenKey = "fs.azure.sas.fixed.token." + accountHost;
+        Assertions.assertEquals("SAS", be.get(authTypeKey));
+        Assertions.assertEquals(sasToken, be.get(fixedSasTokenKey));
+        Assertions.assertTrue(be.keySet().stream().noneMatch(key -> key.startsWith("adls.")));
+
+        Configuration configuration = new Configuration(false);
+        be.forEach(configuration::set);
+        AbfsConfiguration abfsConfiguration = new AbfsConfiguration(configuration, accountHost);
+        Assertions.assertEquals(AuthType.SAS, abfsConfiguration.getAuthType(accountHost));
+        Assertions.assertEquals(sasToken,
+                abfsConfiguration.getSASTokenProvider().getSASToken(accountHost, "container", "/table", "read"));
+    }
+
     @Test
     public void emptyOrNullInputYieldsEmpty() {
         // WHY: a non-REST / no-token table passes an empty map; the bridge must short-circuit to
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/credentials/CredentialUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/credentials/CredentialUtilsTest.java
index 67856553076e32..5bda89f1dc4e66 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/datasource/credentials/CredentialUtilsTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/credentials/CredentialUtilsTest.java
@@ -92,6 +92,23 @@ public void testFilterCloudStoragePropertiesWithMultipleCloudTypes() {
         Assertions.assertFalse(filtered.containsKey("hdfs.namenode"));
     }
 
+    @Test
+    public void testFilterCloudStoragePropertiesWithAdlsVendedCredentials() {
+        String accountHost = "account.dfs.core.windows.net";
+        Map<String, String> rawCredentials = new HashMap<>();
+        rawCredentials.put("adls.sas-token." + accountHost, "testSasToken");
+        rawCredentials.put("adls.sas-token-expires-at-ms." + accountHost, "4102444800000");
+        rawCredentials.put("table.name", "test_table");
+
+        Map<String, String> filtered = CredentialUtils.filterCloudStorageProperties(rawCredentials);
+
+        Assertions.assertEquals(2, filtered.size());
+        Assertions.assertEquals("testSasToken", filtered.get("adls.sas-token." + accountHost));
+        Assertions.assertEquals("4102444800000",
+                filtered.get("adls.sas-token-expires-at-ms." + accountHost));
+        Assertions.assertFalse(filtered.containsKey("table.name"));
+    }
+
     @Test
     public void testFilterCloudStoragePropertiesWithEmptyInput() {
         Map<String, String> filtered = CredentialUtils.filterCloudStorageProperties(new HashMap<>());
diff --git a/fe/pom.xml b/fe/pom.xml
index 1d124c69c26640..dda667087443ef 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -1485,6 +1485,16 @@ under the License.
                 <artifactId>iceberg-aws</artifactId>
                 <version>${iceberg.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.iceberg</groupId>
+                <artifactId>iceberg-azure</artifactId>
+                <version>${iceberg.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.iceberg</groupId>
+                <artifactId>iceberg-azure-bundle</artifactId>
+                <version>${iceberg.version}</version>
+            </dependency>
             <!-- S3Tables catalog impl for the iceberg s3tables flavor; not part of the AWS SDK BOM. -->
             <dependency>
                 <groupId>software.amazon.s3tables</groupId>