[feature](catalog) Support Alibaba Cloud OSS Tables REST catalog

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java

@@ -38,6 +38,7 @@
 import org.apache.iceberg.rest.auth.OAuth2Properties;
 
 import java.io.File;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
@@ -108,8 +109,6 @@ public final class IcebergCatalogFactory {
     private static final String REST_SIGV4_ENABLED_KEY = "rest.sigv4-enabled";
     private static final String REST_SIGNING_REGION_KEY = "rest.signing-region";
     private static final String SECURITY_TYPE_OAUTH2 = "oauth2";
-    private static final String SIGNING_NAME_GLUE = "glue";
-    private static final String SIGNING_NAME_S3TABLES = "s3tables";
 
     // GLUE.
     private static final String GLUE_CREDENTIALS_PROVIDER_KEY = "client.credentials-provider";
@@ -213,6 +212,25 @@ public static Optional<S3CompatibleFileSystemProperties> chooseS3Compatible(
         return Optional.ofNullable(target != null ? target : fallback);
     }
 
+    /**
+     * Selects the storage bindings Iceberg should consume together. All non-S3-compatible bindings are
+     * preserved, while the S3-compatible family is reduced to the same single binding selected for S3FileIO:
+     * a cloud-specific provider such as OSS/COS/OBS wins over the generic S3 fallback. Raw-property routing can
+     * legitimately bind both (for legacy parity), but merging both maps would let a later generic S3 binding
+     * overwrite the explicit provider's endpoint, credentials, and path-style setting.
+     */
+    public static List<StorageProperties> selectEffectiveStorages(
+            List<? extends StorageProperties> storages) {
+        S3CompatibleFileSystemProperties chosenS3 = chooseS3Compatible(storages).orElse(null);
+        List<StorageProperties> selected = new ArrayList<>();
+        for (StorageProperties storage : storages) {
+            if (!(storage instanceof S3CompatibleFileSystemProperties) || storage == chosenS3) {
+                selected.add(storage);
+            }
+        }
+        return selected;
+    }
+
     /**
      * Emits the iceberg {@code S3FileIO} catalog properties from the chosen fe-filesystem S3-compatible
      * storage, mirroring legacy {@code AbstractIcebergProperties.toS3FileIOProperties} (D-061): the
@@ -461,8 +479,8 @@ public static String resolveCatalogName(IcebergCatalogProperties catalogProps, S
     /**
      * Mirrors legacy {@code IcebergRestProperties}: core ({@code uri} always, default empty), optional
      * ({@code prefix} / vended-credentials header / the two effectively-always timeouts), oauth2, and the glue
-     * sigv4 signing block (with credentials sourced from the chosen S3 store for glue/s3tables, else from the
-     * {@code iceberg.rest.*} aliases). PURE.
+     * sigv4 signing block (with credentials sourced from the chosen S3 store for managed signing names, else
+     * from the {@code iceberg.rest.*} aliases). PURE.
      *
      * <p>Every {@code iceberg.rest.*} value is read off the BOUND {@code rest} holder, which declares the alias
      * set once. {@code props} is still needed for the credential-provider mode, whose alias set spans the
@@ -515,9 +533,8 @@ private static void appendRestSigningProperties(Map<String, String> opts, Iceber
         opts.put(REST_SIGNING_NAME_KEY, signingName);
         opts.put(REST_SIGV4_ENABLED_KEY, rest.getSigV4Enabled());
         opts.put(REST_SIGNING_REGION_KEY, rest.getSigningRegion());
-        if (SIGNING_NAME_GLUE.equals(signingName)
-                || SIGNING_NAME_S3TABLES.equals(signingName)) {
-            // glue/s3tables: credentials come from the chosen S3 store, switching on its credential type
+        if (rest.usesS3CredentialsForRestSigning()) {
+            // glue/s3tables/osstables: credentials come from the chosen S3-compatible store, switching on its type
             // (legacy getCredentialType precedence: EXPLICIT before ASSUME_ROLE before PROVIDER_CHAIN).
             if (chosenS3.isPresent()) {
                 S3CompatibleFileSystemProperties s3 = chosenS3.get();

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java

@@ -1219,7 +1219,8 @@ static Map<String, String> deriveHdfsDefaultFsFromWarehouse(String warehouse) {
      */
     private Map<String, String> buildStorageHadoopConfig() {
         Map<String, String> merged = new HashMap<>();
-        for (StorageProperties sp : storage().getStorageProperties()) {
+        for (StorageProperties sp : IcebergCatalogFactory.selectEffectiveStorages(
+                storage().getStorageProperties())) {
             sp.toHadoopProperties().ifPresent(h -> merged.putAll(h.toHadoopConfigurationMap()));
         }
         return merged;

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java

@@ -1654,7 +1654,8 @@ public Map<String, String> getScanNodeProperties(
         // overlay, just the vended one below.
         if (context != null) {
             Map<String, String> backendStorageProps = new HashMap<>();
-            for (StorageProperties sp : storage().getStorageProperties()) {
+            for (StorageProperties sp : IcebergCatalogFactory.selectEffectiveStorages(
+                    storage().getStorageProperties())) {
                 sp.toBackendProperties().ifPresent(b -> backendStorageProps.putAll(b.toMap()));
             }
             backendStorageProps.forEach((k, v) -> props.put(ScanNodePropertyKeys.LOCATION_PREFIX + k, v));

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java

@@ -929,7 +929,8 @@ private Map<String, String> buildHadoopConfig(Table table) {
             // getBackendStorageProperties() second parse). The BE S3 sink (s3_util.cpp
             // convert_properties_to_s3_conf) reads ONLY AWS_*, so the fs.s3a.* hadoop form (correct for the FE
             // iceberg-catalog Configuration) would leave the BE writer with no creds.
-            for (StorageProperties sp : storage().getStorageProperties()) {
+            for (StorageProperties sp : IcebergCatalogFactory.selectEffectiveStorages(
+                    storage().getStorageProperties())) {
                 sp.toBackendProperties().ifPresent(b -> merged.putAll(b.toMap()));
             }
             // REST per-table vended overlay (colliding key takes the vended value — legacy/scan precedence): a

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeS3CompatibleStorageProperties.java

@@ -18,11 +18,15 @@
 package org.apache.doris.connector.iceberg;
 
 import org.apache.doris.filesystem.FileSystemType;
+import org.apache.doris.filesystem.properties.BackendStorageKind;
+import org.apache.doris.filesystem.properties.BackendStorageProperties;
 import org.apache.doris.filesystem.properties.S3CompatibleFileSystemProperties;
 import org.apache.doris.filesystem.properties.StorageKind;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 /**
@@ -43,6 +47,7 @@ final class FakeS3CompatibleStorageProperties implements S3CompatibleFileSystemP
     private String roleArn = "";
     private String externalId = "";
     private String usePathStyle = "";
+    private Map<String, String> backendProperties = Collections.emptyMap();
 
     FakeS3CompatibleStorageProperties(String providerName) {
         this.providerName = providerName;
@@ -88,6 +93,11 @@ FakeS3CompatibleStorageProperties usePathStyle(String v) {
         return this;
     }
 
+    FakeS3CompatibleStorageProperties backendProperties(Map<String, String> v) {
+        this.backendProperties = Collections.unmodifiableMap(new HashMap<>(v));
+        return this;
+    }
+
     @Override
     public String providerName() {
         return providerName;
@@ -183,4 +193,19 @@ public Set<String> getSupportedSchemes() {
         // Mirrors the real S3 provider (this fake's type() is FileSystemType.S3); no test asserts on it.
         return Set.of("s3", "s3a", "s3n");
     }
+
+    @Override
+    public Optional<BackendStorageProperties> toBackendProperties() {
+        return Optional.of(new BackendStorageProperties() {
+            @Override
+            public BackendStorageKind backendKind() {
+                return BackendStorageKind.S3_COMPATIBLE;
+            }
+
+            @Override
+            public Map<String, String> toMap() {
+                return backendProperties;
+            }
+        });
+    }
 }

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java

@@ -325,6 +325,17 @@ public void chooseS3CompatibleFallsBackToGenericS3() {
         Assertions.assertEquals("S3", chosen.get().providerName());
     }
 
+    @Test
+    public void selectEffectiveStoragesDropsGenericS3WhenOssIsPresent() {
+        FakeS3CompatibleStorageProperties genericS3 = new FakeS3CompatibleStorageProperties("S3");
+        FakeS3CompatibleStorageProperties oss = new FakeS3CompatibleStorageProperties("OSS");
+
+        List<StorageProperties> selected = IcebergCatalogFactory.selectEffectiveStorages(
+                Arrays.asList(genericS3, oss));
+
+        Assertions.assertEquals(Collections.singletonList(oss), selected);
+    }
+
     @Test
     public void chooseS3CompatibleEmptyWhenNoS3Storage() {
         // WHY: a credential-less / HDFS-only catalog has no S3-compatible storage, so no S3FileIO props
@@ -436,7 +447,7 @@ public void appendRestOAuth2NotAppliedWhenSecurityNotOauth2() {
     @Test
     public void appendRestSigningBlockEmitsSigningKeysAndS3ExplicitCredentials() {
         // WHY: when signing-name is set, legacy emits rest.signing-name/sigv4-enabled/signing-region; for
-        // glue/s3tables the credentials come from the chosen S3 store, EXPLICIT (static AK/SK) -> rest.* creds
+        // managed signing names get credentials from the chosen S3 store, EXPLICIT (static AK/SK) -> rest.* creds
         // (AwsProperties.REST_*). MUTATION: wrong signing keys, or sourcing creds from the wrong place -> red.
         Map<String, String> opts = new HashMap<>();
         appendRest(opts,
@@ -452,10 +463,49 @@ public void appendRestSigningBlockEmitsSigningKeysAndS3ExplicitCredentials() {
         Assertions.assertEquals("TK", opts.get("rest.session-token"));
     }
 
+    @Test
+    public void buildRestCatalogForOssTablesUsesSharedS3Credentials() {
+        Map<String, String> opts = IcebergCatalogFactory.buildCatalogProperties(
+                IcebergCatalogProperties.of(props("type", "iceberg",
+                        "iceberg.catalog.type", "rest",
+                        "iceberg.rest.uri", "https://cn-hangzhou.oss-tables.aliyuncs.com/iceberg",
+                        "warehouse", "acs:osstables:cn-hangzhou:1234567890:bucket/my-table-bucket",
+                        "iceberg.rest.signing-name", "osstables",
+                        "iceberg.rest.signing-region", "cn-hangzhou",
+                        "iceberg.rest.sigv4-enabled", "true",
+                        "iceberg.rest.view-enabled", "false",
+                        "io-impl", "org.apache.iceberg.aws.s3.S3FileIO")),
+                Optional.of(new FakeS3CompatibleStorageProperties("OSS")
+                        .endpoint("https://oss-cn-hangzhou.aliyuncs.com")
+                        .region("cn-hangzhou")
+                        .accessKey("OSS_AK")
+                        .secretKey("OSS_SK")
+                        .sessionToken("OSS_TOKEN")
+                        .usePathStyle("false")));
+
+        Assertions.assertEquals("https://cn-hangzhou.oss-tables.aliyuncs.com/iceberg", opts.get("uri"));
+        Assertions.assertEquals("acs:osstables:cn-hangzhou:1234567890:bucket/my-table-bucket",
+                opts.get("warehouse"));
+        Assertions.assertEquals("org.apache.iceberg.aws.s3.S3FileIO", opts.get("io-impl"));
+        Assertions.assertEquals("osstables", opts.get("rest.signing-name"));
+        Assertions.assertEquals("cn-hangzhou", opts.get("rest.signing-region"));
+        Assertions.assertEquals("true", opts.get("rest.sigv4-enabled"));
+        Assertions.assertEquals("OSS_AK", opts.get("rest.access-key-id"));
+        Assertions.assertEquals("OSS_SK", opts.get("rest.secret-access-key"));
+        Assertions.assertEquals("OSS_TOKEN", opts.get("rest.session-token"));
+        Assertions.assertEquals("https://oss-cn-hangzhou.aliyuncs.com", opts.get("s3.endpoint"));
+        Assertions.assertEquals("cn-hangzhou", opts.get("client.region"));
+        Assertions.assertEquals("OSS_AK", opts.get("s3.access-key-id"));
+        Assertions.assertEquals("OSS_SK", opts.get("s3.secret-access-key"));
+        Assertions.assertEquals("OSS_TOKEN", opts.get("s3.session-token"));
+        Assertions.assertEquals("false", opts.get("s3.path-style-access"));
+        Assertions.assertNull(opts.get("type"));
+    }
+
     @Test
     public void appendRestSigningGlueAssumeRoleWhenNoStaticCreds() {
         // WHY: legacy getCredentialType precedence is EXPLICIT then ASSUME_ROLE; with no static AK/SK but a role
-        // ARN the glue/s3tables signing path emits the assume-role block (client.factory + client.assume-role.*).
+        // ARN the managed signing path emits the assume-role block (client.factory + client.assume-role.*).
         // MUTATION: emitting rest.access-key-id from a blank AK, or skipping assume-role -> red.
         Map<String, String> opts = new HashMap<>();
         appendRest(opts,
@@ -470,7 +520,7 @@ public void appendRestSigningGlueAssumeRoleWhenNoStaticCreds() {
 
     @Test
     public void appendRestSigningOtherNameUsesIcebergRestCredentials() {
-        // WHY: a signing-name NOT in {glue,s3tables} uses the iceberg.rest.* explicit creds (not the S3 store).
+        // WHY: a non-managed signing name uses the iceberg.rest.* explicit creds (not the S3 store).
         // MUTATION: reading the S3 store here -> red.
         Map<String, String> opts = new HashMap<>();
         appendRest(opts,
@@ -484,7 +534,7 @@ public void appendRestSigningOtherNameUsesIcebergRestCredentials() {
 
     @Test
     public void appendRestSigningGlueProviderChainPinsNonDefaultProvider() {
-        // F14: glue/s3tables signing with NO static creds and NO role -> PROVIDER_CHAIN. A non-DEFAULT
+        // F14: managed signing with NO static creds and NO role -> PROVIDER_CHAIN. A non-DEFAULT
         // s3.credentials_provider_type must pin client.credentials-provider to that provider class (was silently
         // dropped). MUTATION: dropping the else branch -> the key is absent -> red.
         Map<String, String> opts = new HashMap<>();
@@ -505,7 +555,7 @@ public void appendRestSigningGlueProviderChainPinsNonDefaultProvider() {
 
     @Test
     public void appendRestSigningOtherNameProviderChainPinsNonDefaultProvider() {
-        // F14: a non-glue/s3tables signing-name with NO explicit iceberg.rest.* creds falls to PROVIDER_CHAIN;
+        // F14: a non-managed signing name with NO explicit iceberg.rest.* creds falls to PROVIDER_CHAIN;
         // iceberg.rest.credentials_provider_type pins the provider class. MUTATION: dropping the else -> absent.
         Map<String, String> opts = new HashMap<>();
         appendRest(opts,

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java

@@ -42,6 +42,7 @@
 import org.apache.doris.thrift.TTableFormatFileDesc;
 import org.apache.doris.thrift.schema.external.TFieldPtr;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
@@ -2788,6 +2789,33 @@ public void getScanNodePropertiesEmitsStaticStorageCredsAsLocation() {
         Assertions.assertEquals("ep", props.get("location.AWS_ENDPOINT"));
     }
 
+    @Test
+    public void getScanNodePropertiesPrefersOssOverGenericS3Fallback() {
+        FakeIcebergTable table = fakeTable("t1");
+        RecordingConnectorContext context = new RecordingConnectorContext();
+        context.storageProperties = Arrays.asList(
+                new FakeS3CompatibleStorageProperties("OSS").backendProperties(ImmutableMap.of(
+                        "AWS_ENDPOINT", "https://oss-cn-beijing.aliyuncs.com",
+                        "AWS_REGION", "cn-beijing",
+                        "use_path_style", "false")),
+                new FakeS3CompatibleStorageProperties("S3").backendProperties(ImmutableMap.of(
+                        "AWS_ENDPOINT", "https://s3.cn-beijing.amazonaws.com",
+                        "AWS_REGION", "cn-beijing",
+                        "use_path_style", "true",
+                        "AWS_CREDENTIALS_PROVIDER_TYPE", "DEFAULT")));
+        IcebergScanPlanProvider provider =
+                new IcebergScanPlanProvider(IcebergCatalogProperties.of(Collections.emptyMap()), opsReturning(table), context);
+
+        Map<String, String> props = provider.getScanNodeProperties(
+                null, new IcebergTableHandle("db1", "t1"), Collections.emptyList(), Optional.empty());
+
+        Assertions.assertEquals("https://oss-cn-beijing.aliyuncs.com",
+                props.get("location.AWS_ENDPOINT"));
+        Assertions.assertEquals("false", props.get("location.use_path_style"));
+        Assertions.assertNull(props.get("location.AWS_CREDENTIALS_PROVIDER_TYPE"),
+                "generic S3-only properties must not leak into an explicitly matched OSS scan");
+    }
+
     @Test
     public void getScanNodePropertiesOverlaysVendedCredsOverStatic() {
         FakeIcebergTable table = fakeTable("t1");

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java

@@ -744,6 +744,32 @@ public void planWriteMergesStorageHadoopConfig() {
                 "the sink must not ship the fs.s3a.* hadoop form (BE cannot read it)");
     }
 
+    @Test
+    public void planWritePrefersOssBackendConfigOverGenericS3Fallback() {
+        Table table = partitionedSortedTable(freshCatalog());
+        RecordingConnectorContext ctx = new RecordingConnectorContext();
+        ctx.backendFileType = TFileType.FILE_S3;
+        ctx.storageProperties = Arrays.asList(
+                new FakeS3CompatibleStorageProperties("OSS").backendProperties(ImmutableMap.of(
+                        "AWS_ENDPOINT", "https://oss-cn-beijing.aliyuncs.com",
+                        "AWS_REGION", "cn-beijing",
+                        "use_path_style", "false")),
+                new FakeS3CompatibleStorageProperties("S3").backendProperties(ImmutableMap.of(
+                        "AWS_ENDPOINT", "https://s3.cn-beijing.amazonaws.com",
+                        "AWS_REGION", "cn-beijing",
+                        "use_path_style", "true",
+                        "AWS_CREDENTIALS_PROVIDER_TYPE", "DEFAULT")));
+
+        TIcebergTableSink sink = planSink(table, ctx,
+                new WriteHandle(new IcebergTableHandle("db1", "t1")));
+
+        Assertions.assertEquals("https://oss-cn-beijing.aliyuncs.com",
+                sink.getHadoopConfig().get("AWS_ENDPOINT"));
+        Assertions.assertEquals("false", sink.getHadoopConfig().get("use_path_style"));
+        Assertions.assertNull(sink.getHadoopConfig().get("AWS_CREDENTIALS_PROVIDER_TYPE"),
+                "generic S3-only properties must not leak into an explicitly matched OSS write");
+    }
+
     // ───────────────────────────── broker backend (ofs:// / gfs:// -> FILE_BROKER) ─────────────────────────────
     //
     // WHY: SchemaTypeMapper maps ofs/gfs to FILE_BROKER; the sink must then carry the catalog's broker