branch-4.1: [feature](catalog) Support Alibaba Cloud OSS Tables REST catalog

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

@@ -304,6 +304,7 @@ public void checkWhenCreating() throws DdlException {
             CatalogConnectivityTestCoordinator testCoordinator = new CatalogConnectivityTestCoordinator(
                     name,
                     catalogProperty.getMetastoreProperties(),
+                    catalogProperty.getOrderedStoragePropertiesList(),
                     catalogProperty.getStoragePropertiesMap()
             );
             testCoordinator.runTests();

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

@@ -35,6 +35,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -47,16 +48,19 @@ public class CatalogConnectivityTestCoordinator {
 
     private final String catalogName;
     private final MetastoreProperties metastoreProperties;
+    private final List<StorageProperties> storagePropertiesList;
     private final Map<StorageProperties.Type, StorageProperties> storagePropertiesMap;
 
     private String warehouseLocation;
 
     public CatalogConnectivityTestCoordinator(
             String catalogName,
             MetastoreProperties metastoreProperties,
+            List<StorageProperties> storagePropertiesList,
             Map<StorageProperties.Type, StorageProperties> storagePropertiesMap) {
         this.catalogName = catalogName;
         this.metastoreProperties = metastoreProperties;
+        this.storagePropertiesList = storagePropertiesList;
         this.storagePropertiesMap = storagePropertiesMap;
     }
 
@@ -295,7 +299,7 @@ private MetaConnectivityTester createMetaTester(MetastoreProperties props) {
 
         // Iceberg REST
         if (props instanceof IcebergRestProperties) {
-            return new IcebergRestConnectivityTester((IcebergRestProperties) props);
+            return new IcebergRestConnectivityTester((IcebergRestProperties) props, storagePropertiesList);
         }
 
         // Iceberg S3Table

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

@@ -18,21 +18,25 @@
 package org.apache.doris.datasource.connectivity;
 
 import org.apache.doris.datasource.property.metastore.AbstractIcebergProperties;
-import org.apache.doris.datasource.property.metastore.IcebergRestProperties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
 
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.rest.RESTCatalog;
 
+import java.util.List;
 import java.util.Map;
 
 public class IcebergRestConnectivityTester extends AbstractIcebergConnectivityTester {
     // For Polaris REST catalog compatibility
     private static final String DEFAULT_BASE_LOCATION = "default-base-location";
 
     private String warehouseLocation;
+    private final List<StorageProperties> storagePropertiesList;
 
-    public IcebergRestConnectivityTester(AbstractIcebergProperties properties) {
+    public IcebergRestConnectivityTester(AbstractIcebergProperties properties,
+            List<StorageProperties> storagePropertiesList) {
         super(properties);
+        this.storagePropertiesList = storagePropertiesList;
     }
 
     @Override
@@ -48,10 +52,8 @@ public String getErrorHint() {
 
     @Override
     public void testConnection() throws Exception {
-        Map<String, String> restProps = ((IcebergRestProperties) properties).getIcebergRestCatalogProperties();
-
-        try (RESTCatalog catalog = new RESTCatalog()) {
-            catalog.initialize("connectivity-test", restProps);
+        try (RESTCatalog catalog = (RESTCatalog) properties.initializeCatalog(
+                "connectivity-test", storagePropertiesList)) {
 
             // Validate connection by listing namespaces.
             // This verifies authentication and warehouse configuration.

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

@@ -59,6 +59,9 @@
 import org.apache.doris.datasource.mvcc.MvccSnapshot;
 import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.datasource.property.metastore.HMSBaseProperties;
+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 org.apache.doris.nereids.exceptions.NotSupportedException;
 import org.apache.doris.nereids.trees.expressions.literal.Result;
 import org.apache.doris.nereids.types.VarBinaryType;
@@ -145,6 +148,7 @@
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.Comparator;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -177,6 +181,56 @@ public Integer initialValue() {
     public static final String TOTAL_POSITION_DELETES = "total-position-deletes";
     public static final String TOTAL_EQUALITY_DELETES = "total-equality-deletes";
 
+    /**
+     * Selects the storage bindings Iceberg should consume together. Iceberg can configure only one
+     * S3-compatible data plane, so a concrete provider such as OSS takes precedence over the generic
+     * S3 fallback while unrelated storage bindings are preserved.
+     */
+    public static List<StorageProperties> selectEffectiveStorageProperties(
+            List<StorageProperties> storagePropertiesList) {
+        StorageProperties chosenS3 = chooseS3CompatibleStorage(storagePropertiesList);
+        List<StorageProperties> selected = new ArrayList<>();
+        for (StorageProperties storageProperties : storagePropertiesList) {
+            if (!(storageProperties instanceof AbstractS3CompatibleProperties)
+                    || storageProperties == chosenS3) {
+                selected.add(storageProperties);
+            }
+        }
+        return selected;
+    }
+
+    public static Map<StorageProperties.Type, StorageProperties> selectEffectiveStorageProperties(
+            Map<StorageProperties.Type, StorageProperties> storagePropertiesMap) {
+        List<StorageProperties> ordered = new ArrayList<>();
+        for (StorageProperties.Type type : StorageProperties.Type.values()) {
+            StorageProperties storageProperties = storagePropertiesMap.get(type);
+            if (storageProperties != null) {
+                ordered.add(storageProperties);
+            }
+        }
+
+        Map<StorageProperties.Type, StorageProperties> selected = new EnumMap<>(StorageProperties.Type.class);
+        for (StorageProperties storageProperties : selectEffectiveStorageProperties(ordered)) {
+            selected.put(storageProperties.getType(), storageProperties);
+        }
+        return selected;
+    }
+
+    private static StorageProperties chooseS3CompatibleStorage(List<StorageProperties> storagePropertiesList) {
+        StorageProperties fallback = null;
+        for (StorageProperties storageProperties : storagePropertiesList) {
+            if (storageProperties instanceof AbstractS3CompatibleProperties) {
+                if (fallback == null) {
+                    fallback = storageProperties;
+                }
+                if (!(storageProperties instanceof S3Properties)) {
+                    return storageProperties;
+                }
+            }
+        }
+        return fallback;
+    }
+
     // nickname in flink and spark
     public static final String WRITE_FORMAT = "write-format";
     public static final String COMPRESSION_CODEC = "compression-codec";

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

@@ -273,6 +273,7 @@ protected void doInitialize() throws UserException {
                     source.getCatalog().getCatalogProperty().getStoragePropertiesMap(),
                     icebergTable
             );
+            storagePropertiesMap = IcebergUtils.selectEffectiveStorageProperties(storagePropertiesMap);
             backendStorageProperties = CredentialUtils.getBackendPropertiesFromStorageMap(storagePropertiesMap);
         } finally {
             if (getSummaryProfile() != null) {

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

@@ -19,6 +19,7 @@
 
 import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
 import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
+import org.apache.doris.datasource.iceberg.IcebergUtils;
 import org.apache.doris.datasource.metacache.CacheSpec;
 import org.apache.doris.datasource.property.common.IcebergAwsAssumeRoleProperties;
 import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties;
@@ -210,28 +211,18 @@ protected abstract Catalog initCatalog(
      */
     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) {
+        for (StorageProperties storageProperties :
+                IcebergUtils.selectEffectiveStorageProperties(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;
-                }
+                s3Target = (AbstractS3CompatibleProperties) storageProperties;
             }
         }
-        AbstractS3CompatibleProperties chosen = s3Target != null ? s3Target : s3Fallback;
-        if (chosen != null) {
-            toS3FileIOProperties(chosen, fileIOProperties);
+        if (s3Target != null) {
+            toS3FileIOProperties(s3Target, fileIOProperties);
         } else {
             String region = AbstractS3CompatibleProperties.getRegionFromProperties(fileIOProperties);
             if (!Strings.isNullOrEmpty(region)) {

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

@@ -20,6 +20,7 @@
 import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
 import org.apache.doris.datasource.property.common.AwsCredentialsProviderMode;
 import org.apache.doris.datasource.property.common.IcebergAwsClientCredentialsProperties;
+import org.apache.doris.datasource.property.storage.OSSProperties;
 import org.apache.doris.datasource.property.storage.S3Properties;
 import org.apache.doris.datasource.property.storage.StorageProperties;
 import org.apache.doris.foundation.property.ConnectorProperty;
@@ -49,6 +50,7 @@ public class IcebergRestProperties extends AbstractIcebergProperties {
 
     private Map<String, String> icebergRestCatalogProperties;
     private S3Properties s3Properties;
+    private OSSProperties ossProperties;
 
     @Getter
     @ConnectorProperty(names = {"iceberg.rest.uri", "uri"},
@@ -214,7 +216,11 @@ public void initNormalizeAndCheckProps() {
                 AwsCredentialsProviderMode.fromString(icebergRestCredentialsProviderType);
         buildRules().validate();
         if (shouldUseS3PropertiesForRestCredentials()) {
-            s3Properties = S3Properties.of(origProps);
+            if (isOssTables()) {
+                ossProperties = OSSProperties.of(origProps);
+            } else {
+                s3Properties = S3Properties.of(origProps);
+            }
         }
         initIcebergRestCatalogProperties();
     }
@@ -251,13 +257,19 @@ private ParamRules buildRules() {
             }
         }
 
-        // When signing-name is glue or s3tables: require signing-region and sigv4-enabled
+        // SigV4-backed REST catalogs require a signing region and SigV4 to be enabled.
         rules.requireIf(icebergRestSigningName, "glue",
                 new String[] {icebergRestSigningRegion, icebergRestSigV4Enabled},
                 "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is glue");
         rules.requireIf(icebergRestSigningName, "s3tables",
                 new String[] {icebergRestSigningRegion, icebergRestSigV4Enabled},
                 "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is s3tables");
+        rules.requireIf(icebergRestSigningName, "osstables",
+                new String[] {icebergRestSigningRegion, icebergRestSigV4Enabled},
+                "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is osstables");
+        rules.check(() -> shouldUseS3PropertiesForRestCredentials()
+                        && !"true".equalsIgnoreCase(icebergRestSigV4Enabled),
+                "Rest Catalog requires sigv4-enabled set to true when signing-name is " + icebergRestSigningName);
 
         rejectUnsupportedAwsAssumeRoleProperty(ICEBERG_REST_ROLE_ARN);
         rejectUnsupportedAwsAssumeRoleProperty(ICEBERG_REST_EXTERNAL_ID);
@@ -350,8 +362,14 @@ private void addGlueRestCatalogProperties() {
             icebergRestCatalogProperties.put("rest.signing-region", icebergRestSigningRegion);
 
             if (shouldUseS3PropertiesForRestCredentials()) {
-                IcebergAwsClientCredentialsProperties.putCredentialProviderProperties(
-                        icebergRestCatalogProperties, s3Properties);
+                if (isOssTables()) {
+                    IcebergAwsClientCredentialsProperties.putCredentialProviderProperties(
+                            icebergRestCatalogProperties, ossProperties.getAccessKey(), ossProperties.getSecretKey(),
+                            ossProperties.getSessionToken(), icebergRestCredentialsProviderMode);
+                } else {
+                    IcebergAwsClientCredentialsProperties.putCredentialProviderProperties(
+                            icebergRestCatalogProperties, s3Properties);
+                }
             } else {
                 IcebergAwsClientCredentialsProperties.putCredentialProviderProperties(
                         icebergRestCatalogProperties, icebergRestAccessKeyId,
@@ -362,7 +380,12 @@ private void addGlueRestCatalogProperties() {
 
     private boolean shouldUseS3PropertiesForRestCredentials() {
         return "glue".equals(icebergRestSigningName)
-                || "s3tables".equals(icebergRestSigningName);
+                || "s3tables".equals(icebergRestSigningName)
+                || "osstables".equals(icebergRestSigningName);
+    }
+
+    private boolean isOssTables() {
+        return "osstables".equals(icebergRestSigningName);
     }
 
     public Map<String, String> getIcebergRestCatalogProperties() {

fe/fe-core/src/main/java/org/apache/doris/planner/IcebergTableSink.java

@@ -79,10 +79,11 @@ public IcebergTableSink(IcebergExternalTable targetTable) {
         this.targetTable = targetTable;
         this.icebergTable = targetTable.getIcebergTable();
         IcebergExternalCatalog catalog = (IcebergExternalCatalog) targetTable.getCatalog();
-        storagePropertiesMap = VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
-                catalog.getCatalogProperty().getMetastoreProperties(),
-                catalog.getCatalogProperty().getStoragePropertiesMap(),
-                icebergTable);
+        storagePropertiesMap = IcebergUtils.selectEffectiveStorageProperties(
+                VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
+                        catalog.getCatalogProperty().getMetastoreProperties(),
+                        catalog.getCatalogProperty().getStoragePropertiesMap(),
+                        icebergTable));
     }
 
     public IcebergTableSink(IcebergExternalTable targetTable, Table icebergTable) {
@@ -94,10 +95,11 @@ public IcebergTableSink(IcebergExternalTable targetTable, Table icebergTable) {
         // Keep credentials and every writer option on the metadata generation pinned during analysis.
         this.icebergTable = Objects.requireNonNull(icebergTable, "icebergTable is not null");
         IcebergExternalCatalog catalog = (IcebergExternalCatalog) targetTable.getCatalog();
-        storagePropertiesMap = VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
-                catalog.getCatalogProperty().getMetastoreProperties(),
-                catalog.getCatalogProperty().getStoragePropertiesMap(),
-                icebergTable);
+        storagePropertiesMap = IcebergUtils.selectEffectiveStorageProperties(
+                VendedCredentialsFactory.getStoragePropertiesMapWithVendedCredentials(
+                        catalog.getCatalogProperty().getMetastoreProperties(),
+                        catalog.getCatalogProperty().getStoragePropertiesMap(),
+                        icebergTable));
     }
 
     @Override

fe/fe-core/src/test/java/org/apache/doris/datasource/connectivity/IcebergRestConnectivityTesterTest.java

@@ -0,0 +1,57 @@
+// 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.doris.datasource.connectivity;
+
+import org.apache.doris.datasource.property.metastore.IcebergRestProperties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class IcebergRestConnectivityTesterTest {
+
+    @Test
+    public void testUsesNormalCatalogInitializationPath() throws Exception {
+        IcebergRestProperties properties = Mockito.mock(IcebergRestProperties.class);
+        RESTCatalog catalog = Mockito.mock(RESTCatalog.class);
+        StorageProperties storageProperties = Mockito.mock(StorageProperties.class);
+        List<StorageProperties> storagePropertiesList = Collections.singletonList(storageProperties);
+        Map<String, String> catalogProperties = new HashMap<>();
+        catalogProperties.put(CatalogProperties.WAREHOUSE_LOCATION, "s3://warehouse/path");
+
+        Mockito.when(properties.initializeCatalog("connectivity-test", storagePropertiesList)).thenReturn(catalog);
+        Mockito.when(catalog.properties()).thenReturn(catalogProperties);
+
+        IcebergRestConnectivityTester tester = new IcebergRestConnectivityTester(
+                properties, storagePropertiesList);
+        tester.testConnection();
+
+        Mockito.verify(properties).initializeCatalog("connectivity-test", storagePropertiesList);
+        Mockito.verify(catalog).listNamespaces();
+        Mockito.verify(catalog).close();
+        Assertions.assertEquals("s3://warehouse/path", tester.getTestLocation());
+    }
+}