[feat](iceberg) support aliyun dlf iceberg rest catalog

build.sh

@@ -474,6 +474,12 @@ if [[ "$(echo "${DISABLE_BUILD_AZURE}" | tr '[:lower:]' '[:upper:]')" == "ON" ]]
     BUILD_AZURE='OFF'
 fi
 
+if [[ "$(echo "${DISABLE_BUILD_JINDOFS}" | tr '[:lower:]' '[:upper:]')" == "ON" ]]; then
+    BUILD_JINDOFS='OFF'
+else
+    BUILD_JINDOFS='ON'
+fi
+
 if [[ -z "${ENABLE_INJECTION_POINT}" ]]; then
     ENABLE_INJECTION_POINT='OFF'
 fi
@@ -795,7 +801,9 @@ if [[ "${BUILD_FE}" -eq 1 ]]; then
     cp -r -p "${DORIS_HOME}/conf/ldap.conf" "${DORIS_OUTPUT}/fe/conf"/
     cp -r -p "${DORIS_HOME}/conf/mysql_ssl_default_certificate" "${DORIS_OUTPUT}/fe/"/
     rm -rf "${DORIS_OUTPUT}/fe/lib"/*
-    install -d "${DORIS_OUTPUT}/fe/lib/jindofs"
+    if [[ "${BUILD_JINDOFS}" == "ON" ]]; then
+        install -d "${DORIS_OUTPUT}/fe/lib/jindofs"
+    fi
     cp -r -p "${DORIS_HOME}/fe/fe-core/target/lib"/* "${DORIS_OUTPUT}/fe/lib"/
     cp -r -p "${DORIS_HOME}/fe/fe-core/target/doris-fe.jar" "${DORIS_OUTPUT}/fe/lib"/
     if [[ "${WITH_TDE_DIR}" != "" ]]; then
@@ -805,13 +813,15 @@ if [[ "${BUILD_FE}" -eq 1 ]]; then
     #cp -r -p "${DORIS_HOME}/docs/build/help-resource.zip" "${DORIS_OUTPUT}/fe/lib"/
 
     # copy jindofs jars, only support for Linux x64 or arm
-    if [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "${TARGET_ARCH}" == 'x86_64' ]]; then
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-ubuntu22-x86_64-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
-    elif [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "${TARGET_ARCH}" == 'aarch64' ]]; then
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-el7-aarch64-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
+    if [[ "${BUILD_JINDOFS}" == "ON" ]]; then
+        if [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "${TARGET_ARCH}" == 'x86_64' ]]; then
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-ubuntu22-x86_64-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
+        elif [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "${TARGET_ARCH}" == 'aarch64' ]]; then
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-el7-aarch64-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/fe/lib/jindofs"/
+        fi
     fi
 
     cp -r -p "${DORIS_HOME}/minidump" "${DORIS_OUTPUT}/fe"/
@@ -981,14 +991,16 @@ EOF
     done        
 
     # copy jindofs jars, only support for Linux x64 or arm
-    install -d "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
-    if [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "$TARGET_ARCH" == 'x86_64' ]]; then
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-ubuntu22-x86_64-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
-    elif [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "$TARGET_ARCH" == 'aarch64' ]]; then
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-el7-aarch64-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
-        cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
+    if [[ "${BUILD_JINDOFS}" == "ON" ]]; then
+        install -d "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
+        if [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "$TARGET_ARCH" == 'x86_64' ]]; then
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-ubuntu22-x86_64-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
+        elif [[ "${TARGET_SYSTEM}" == 'Linux' ]] && [[ "$TARGET_ARCH" == 'aarch64' ]]; then
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-core-linux-el7-aarch64-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
+            cp -r -p "${DORIS_THIRDPARTY}"/installed/jindofs_libs/jindo-sdk-[0-9]*.jar "${DORIS_OUTPUT}/be/lib/java_extensions/jindofs"/
+        fi
     fi
 
     cp -r -p "${DORIS_THIRDPARTY}/installed/webroot"/* "${DORIS_OUTPUT}/be/www"/

fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogProperty.java

@@ -18,10 +18,13 @@
 package org.apache.doris.datasource;
 
 import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.credentials.AbstractVendedCredentialsProvider;
+import org.apache.doris.datasource.credentials.VendedCredentialsFactory;
 import org.apache.doris.datasource.property.metastore.MetastoreProperties;
 import org.apache.doris.datasource.property.storage.StorageProperties;
 
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.gson.annotations.SerializedName;
 import org.apache.commons.collections4.MapUtils;
@@ -174,9 +177,20 @@ private void initStorageProperties() {
             synchronized (this) {
                 if (storagePropertiesMap == null) {
                     try {
-                        this.orderedStoragePropertiesList = StorageProperties.createAll(getProperties());
-                        this.storagePropertiesMap = orderedStoragePropertiesList.stream()
-                                .collect(Collectors.toMap(StorageProperties::getType, Function.identity()));
+                        boolean checkStorageProperties = true;
+                        AbstractVendedCredentialsProvider provider =
+                                VendedCredentialsFactory.getProviderType(getMetastoreProperties());
+                        if (provider != null) {
+                            checkStorageProperties = !provider.isVendedCredentialsEnabled(getMetastoreProperties());
+                        }
+                        if (checkStorageProperties) {
+                            this.orderedStoragePropertiesList = StorageProperties.createAll(getProperties());
+                            this.storagePropertiesMap = orderedStoragePropertiesList.stream()
+                                    .collect(Collectors.toMap(StorageProperties::getType, Function.identity()));
+                        } else {
+                            this.orderedStoragePropertiesList = Lists.newArrayList();
+                            this.storagePropertiesMap = Maps.newHashMap();
+                        }
                     } catch (UserException e) {
                         LOG.warn("Failed to initialize catalog storage properties", e);
                         throw new RuntimeException("Failed to initialize storage properties, error: "
@@ -199,16 +213,13 @@ public List<StorageProperties> getOrderedStoragePropertiesList() {
 
     public void checkMetaStoreAndStorageProperties(Class msClass) {
         MetastoreProperties msProperties;
-        List<StorageProperties> storageProperties;
         try {
             msProperties = MetastoreProperties.create(getProperties());
-            storageProperties = StorageProperties.createAll(getProperties());
+            initStorageProperties();
         } catch (UserException e) {
             throw new RuntimeException("Failed to initialize Catalog properties, error: "
                     + ExceptionUtils.getRootCauseMessage(e), e);
         }
-        Preconditions.checkNotNull(storageProperties,
-                "Storage properties are not configured properly");
         Preconditions.checkNotNull(msProperties, "Metastore properties are not configured properly");
         Preconditions.checkArgument(
                 msClass.isInstance(msProperties),

fe/fe-core/src/main/java/org/apache/doris/datasource/credentials/AbstractVendedCredentialsProvider.java

@@ -85,7 +85,7 @@ public final <T> Map<StorageProperties.Type, StorageProperties> getStorageProper
     /**
      * Check whether to enable vendor credentials (subclass implementation)
      */
-    protected abstract boolean isVendedCredentialsEnabled(MetastoreProperties metastoreProperties);
+    public abstract boolean isVendedCredentialsEnabled(MetastoreProperties metastoreProperties);
 
     /**
      * Extract original vendored credentials from table objects (subclass implementation)

fe/fe-core/src/main/java/org/apache/doris/datasource/credentials/VendedCredentialsFactory.java

@@ -53,7 +53,7 @@ public static <T> Map<Type, StorageProperties> getStoragePropertiesMapWithVended
     /**
      * Select the right provider according to the MetastoreProperties type
      */
-    private static AbstractVendedCredentialsProvider getProviderType(MetastoreProperties metastoreProperties) {
+    public static AbstractVendedCredentialsProvider getProviderType(MetastoreProperties metastoreProperties) {
         if (metastoreProperties == null) {
             return null;
         }

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java

@@ -154,11 +154,6 @@ protected void initLocalObjectsImpl() {
         metadataOps = hiveOps;
     }
 
-    @Override
-    public synchronized void resetToUninitialized(boolean invalidCache) {
-        super.resetToUninitialized(invalidCache);
-    }
-
     @Override
     public void onClose() {
         super.onClose();

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java

@@ -75,9 +75,7 @@ public IcebergExternalCatalog(long catalogId, String name, String comment) {
     protected void initCatalog() {
         try {
             msProperties = (AbstractIcebergProperties) catalogProperty.getMetastoreProperties();
-            this.catalog = msProperties.initializeCatalog(getName(), catalogProperty
-                    .getOrderedStoragePropertiesList());
-
+            this.catalog = msProperties.initializeCatalog(getName(), catalogProperty.getOrderedStoragePropertiesList());
             this.icebergCatalogType = msProperties.getIcebergCatalogType();
         } catch (ClassCastException e) {
             throw new RuntimeException("Invalid properties for Iceberg catalog: " + getProperties(), e);

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergVendedCredentialsProvider.java

@@ -23,6 +23,9 @@
 
 import com.google.common.collect.Maps;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
 
 import java.util.Map;
 
@@ -38,7 +41,7 @@ public static IcebergVendedCredentialsProvider getInstance() {
     }
 
     @Override
-    protected boolean isVendedCredentialsEnabled(MetastoreProperties metastoreProperties) {
+    public boolean isVendedCredentialsEnabled(MetastoreProperties metastoreProperties) {
         if (metastoreProperties instanceof IcebergRestProperties) {
             return ((IcebergRestProperties) metastoreProperties).isIcebergRestVendedCredentialsEnabled();
         }
@@ -57,7 +60,15 @@ protected <T> Map<String, String> extractRawVendedCredentials(T tableObject) {
         }
 
         // Return table.io().properties() directly, and let StorageProperties.createAll() to convert the format
-        return table.io().properties();
+        FileIO fileIO = table.io();
+        Map<String, String> ioProps = Maps.newHashMap(fileIO.properties());
+        if (fileIO instanceof SupportsStorageCredentials) {
+            SupportsStorageCredentials ssc = (SupportsStorageCredentials) fileIO;
+            for (StorageCredential storageCredential : ssc.credentials()) {
+                ioProps.putAll(storageCredential.config());
+            }
+        }
+        return ioProps;
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonVendedCredentialsProvider.java

@@ -40,7 +40,7 @@ public static PaimonVendedCredentialsProvider getInstance() {
     }
 
     @Override
-    protected boolean isVendedCredentialsEnabled(MetastoreProperties metastoreProperties) {
+    public boolean isVendedCredentialsEnabled(MetastoreProperties metastoreProperties) {
         // Paimon REST catalog always supports vended credentials if it's REST type
         return metastoreProperties instanceof PaimonRestMetaStoreProperties;
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractIcebergProperties.java

@@ -21,11 +21,18 @@
 import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
 import org.apache.doris.datasource.metacache.CacheSpec;
 import org.apache.doris.datasource.property.ConnectorProperty;
+import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties;
+import org.apache.doris.datasource.property.storage.S3Properties;
 import org.apache.doris.datasource.property.storage.StorageProperties;
 
+import com.google.common.base.Preconditions;
 import lombok.Getter;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.catalog.Catalog;
 
 import java.util.HashMap;
@@ -82,6 +89,14 @@ public abstract class AbstractIcebergProperties extends MetastoreProperties {
     )
     protected String ioManifestCacheMaxContentLength;
 
+    @Getter
+    @ConnectorProperty(
+            names = {CatalogProperties.FILE_IO_IMPL},
+            required = false,
+            description = "Custom io impl for iceberg"
+    )
+    protected String ioImpl;
+
     @Getter
     protected ExecutionAuthenticator executionAuthenticator = new ExecutionAuthenticator(){};
 
@@ -114,7 +129,7 @@ protected AbstractIcebergProperties(Map<String, String> props) {
      * and deleting Iceberg tables.
      */
     public final Catalog initializeCatalog(String catalogName,
-                                                        List<StorageProperties> storagePropertiesList) {
+                                           List<StorageProperties> storagePropertiesList) {
         Map<String, String> catalogProps = new HashMap<>(getOrigProps());
         if (StringUtils.isNotBlank(warehouse)) {
             catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
@@ -179,4 +194,80 @@ protected abstract Catalog initCatalog(
             Map<String, String> catalogProps,
             List<StorageProperties> storagePropertiesList
     );
+
+    /**
+     * Unified method to configure FileIO properties for Iceberg catalog.
+     * This method handles all storage types (HDFS, S3, MinIO, etc.) by:
+     * 1. Adding all storage properties to Hadoop Configuration (for HadoopFileIO / S3A access).
+     * 2. Extracting S3-compatible properties into fileIOProperties map (for Iceberg S3FileIO).
+     *
+     * @param storagePropertiesList list of storage properties
+     * @param fileIOProperties options map to be populated with S3 FileIO properties
+     * @return Hadoop Configuration populated with all storage properties
+     */
+    public void toFileIOProperties(List<StorageProperties> storagePropertiesList,
+            Map<String, String> fileIOProperties, Configuration conf) {
+        // We only support one S3-compatible storage property for FileIO configuration.
+        // When multiple AbstractS3CompatibleProperties exist, prefer the first non-S3Properties one,
+        // because a non-S3 type (e.g. OSSProperties, COSProperties) indicates the user has explicitly
+        // specified a concrete S3-compatible storage, which should take priority over the generic S3Properties.
+        AbstractS3CompatibleProperties s3Fallback = null;
+        AbstractS3CompatibleProperties s3Target = null;
+        for (StorageProperties storageProperties : storagePropertiesList) {
+            if (conf != null && storageProperties.getHadoopStorageConfig() != null) {
+                conf.addResource(storageProperties.getHadoopStorageConfig());
+            }
+            if (storageProperties instanceof AbstractS3CompatibleProperties) {
+                if (s3Fallback == null) {
+                    s3Fallback = (AbstractS3CompatibleProperties) storageProperties;
+                }
+                if (s3Target == null && !(storageProperties instanceof S3Properties)) {
+                    s3Target = (AbstractS3CompatibleProperties) storageProperties;
+                }
+            }
+        }
+        AbstractS3CompatibleProperties chosen = s3Target != null ? s3Target : s3Fallback;
+        if (chosen != null) {
+            toS3FileIOProperties(chosen, fileIOProperties);
+        }
+    }
+
+    /**
+     * Configure S3 FileIO properties for all S3-compatible storage types (S3, MinIO, etc.)
+     * This method provides a unified way to convert S3-compatible properties to Iceberg S3FileIO format.
+     *
+     * @param s3Properties S3-compatible properties
+     * @param options Options map to be populated with S3 FileIO properties
+     */
+    private void toS3FileIOProperties(AbstractS3CompatibleProperties s3Properties, Map<String, String> options) {
+        // Common properties - only set if not blank
+        if (StringUtils.isNotBlank(s3Properties.getEndpoint())) {
+            options.put(S3FileIOProperties.ENDPOINT, s3Properties.getEndpoint());
+        }
+        if (StringUtils.isNotBlank(s3Properties.getUsePathStyle())) {
+            options.put(S3FileIOProperties.PATH_STYLE_ACCESS, s3Properties.getUsePathStyle());
+        }
+        if (StringUtils.isNotBlank(s3Properties.getRegion())) {
+            options.put(AwsClientProperties.CLIENT_REGION, s3Properties.getRegion());
+        }
+        if (StringUtils.isNotBlank(s3Properties.getAccessKey())) {
+            options.put(S3FileIOProperties.ACCESS_KEY_ID, s3Properties.getAccessKey());
+        }
+        if (StringUtils.isNotBlank(s3Properties.getSecretKey())) {
+            options.put(S3FileIOProperties.SECRET_ACCESS_KEY, s3Properties.getSecretKey());
+        }
+        if (StringUtils.isNotBlank(s3Properties.getSessionToken())) {
+            options.put(S3FileIOProperties.SESSION_TOKEN, s3Properties.getSessionToken());
+        }
+    }
+
+    protected Catalog buildIcebergCatalog(String catalogName, Map<String, String> options, Configuration conf) {
+        // For Iceberg SDK, "type" means catalog type, such as hive, jdbc, rest.
+        // But in Doris, "type" is "iceberg".
+        // And Iceberg SDK does not allow with both "type" and "catalog-impl" properties,
+        // So here we remove "type" and make sure "catalog-impl" is set.
+        options.remove(CatalogUtil.ICEBERG_CATALOG_TYPE);
+        Preconditions.checkArgument(options.containsKey(CatalogProperties.CATALOG_IMPL));
+        return CatalogUtil.buildIcebergCatalog(catalogName, options, conf);
+    }
 }