From d8c8be6b2fde7cb13e1bfa558578348fc1fb6794 Mon Sep 17 00:00:00 2001 From: xylaaaaa <2392805527@qq.com> Date: Fri, 7 Aug 2026 10:54:14 +0800 Subject: [PATCH 1/2] [feature](catalog) Support Alibaba Cloud OSS Tables REST catalog ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Alibaba Cloud OSS Tables exposes an Iceberg REST Catalog that uses the osstables SigV4 service name and an S3-compatible data plane. Doris only treated Glue and S3 Tables as managed signed REST catalogs, so an OSS Tables catalog did not reuse the selected OSS credentials for REST signing and did not validate its required signing settings. Recognize osstables, reuse the same OSS credentials for the REST control plane and S3FileIO data plane, require a signing region with SigV4 enabled, and cover the official endpoint, ACS warehouse ARN, and STS token mapping. ### Release note Support Alibaba Cloud OSS Tables through the Iceberg REST Catalog. ### Check List (For Author) - Test: Unit Test and FE build - ./run-fe-ut.sh --run org.apache.doris.connector.metastore.iceberg.rest.IcebergRestMetaStorePropertiesTest - ./run-fe-ut.sh --run org.apache.doris.connector.iceberg.IcebergCatalogFactoryTest - DISABLE_BUILD_UI=ON ./build.sh --fe - Behavior changed: Yes. signing-name=osstables now reuses OSS/S3-compatible credentials and requires a signing region with SigV4 enabled. Managed Glue and S3 Tables catalogs also reject an explicit sigv4-enabled=false value during validation instead of failing later at runtime. - Does this need documentation: Yes. Follow-up documentation should include the OSS Tables REST endpoint, ACS warehouse ARN, osstables signing name, and OSS S3FileIO endpoint. --- .../iceberg/IcebergCatalogFactory.java | 11 ++--- .../iceberg/IcebergCatalogFactoryTest.java | 49 +++++++++++++++++-- .../rest/IcebergRestMetaStoreProperties.java | 17 ++++++- .../IcebergRestMetaStorePropertiesTest.java | 16 +++++- 4 files changed, 78 insertions(+), 15 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java index 232623da99ba9d..f3d8219565d498 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java @@ -108,8 +108,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"; @@ -461,8 +459,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. * *

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 +513,8 @@ private static void appendRestSigningProperties(Map 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(); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java index b86600b248d7ed..eed8cd1377b7b6 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java @@ -436,7 +436,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 opts = new HashMap<>(); appendRest(opts, @@ -452,10 +452,49 @@ public void appendRestSigningBlockEmitsSigningKeysAndS3ExplicitCredentials() { Assertions.assertEquals("TK", opts.get("rest.session-token")); } + @Test + public void buildRestCatalogForOssTablesUsesSharedS3Credentials() { + Map 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 opts = new HashMap<>(); appendRest(opts, @@ -470,7 +509,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 opts = new HashMap<>(); appendRest(opts, @@ -484,7 +523,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 opts = new HashMap<>(); @@ -505,7 +544,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 opts = new HashMap<>(); appendRest(opts, diff --git a/fe/fe-connector/fe-connector-metastore-iceberg/src/main/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStoreProperties.java b/fe/fe-connector/fe-connector-metastore-iceberg/src/main/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStoreProperties.java index 538d52406c2c3f..90e079640ecaf0 100644 --- a/fe/fe-connector/fe-connector-metastore-iceberg/src/main/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStoreProperties.java +++ b/fe/fe-connector/fe-connector-metastore-iceberg/src/main/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStoreProperties.java @@ -41,6 +41,8 @@ *

  • OAuth2 credential/token mutually exclusive (ParamRules)
    • *
  • signing-name=glue requires signing-region + sigv4-enabled (ParamRules)
    • *
  • signing-name=s3tables requires signing-region + sigv4-enabled (ParamRules)
    • + *
  • signing-name=osstables requires signing-region + sigv4-enabled (ParamRules)
    • + *
  • managed signing names require sigv4-enabled=true (ParamRules)
    • *
  • access-key-id + secret-access-key set together (ParamRules)
    • * * No uri/warehouse requirement. The {@code Security}/{@code AwsCredentialsProviderMode} enum checks are @@ -121,7 +123,7 @@ public final class IcebergRestMetaStoreProperties extends AbstractMetaStorePrope private String signingRegion = ""; @ConnectorProperty(names = {"iceberg.rest.sigv4-enabled"}, required = false, - description = "True for Glue/S3Tables Rest Catalog.") + description = "True for Glue/S3Tables/OSS Tables Rest Catalog.") private String sigV4Enabled = ""; @ConnectorProperty(names = {"iceberg.rest.access-key-id"}, required = false, @@ -242,6 +244,13 @@ public String getSigV4Enabled() { return sigV4Enabled; } + /** Whether REST signing reuses the selected S3-compatible storage credentials. */ + public boolean usesS3CredentialsForRestSigning() { + return "glue".equals(signingName) + || "s3tables".equals(signingName) + || "osstables".equals(signingName); + } + public String getAccessKeyId() { return accessKeyId; } @@ -307,11 +316,15 @@ public void validate() { throw new IllegalArgumentException("OAuth2 requires either credential or token"); } } - // When signing-name is glue or s3tables: require signing-region and sigv4-enabled (registered). + // SigV4-backed REST catalogs require a signing region and SigV4 to be enabled (registered). rules.requireIf(signingName, "glue", new String[] {signingRegion, sigV4Enabled}, "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is glue"); rules.requireIf(signingName, "s3tables", new String[] {signingRegion, sigV4Enabled}, "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is s3tables"); + rules.requireIf(signingName, "osstables", new String[] {signingRegion, sigV4Enabled}, + "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is osstables"); + rules.check(() -> usesS3CredentialsForRestSigning() && !"true".equalsIgnoreCase(sigV4Enabled), + "Rest Catalog requires sigv4-enabled set to true when signing-name is " + signingName); // AWS assume-role properties are not supported for the Iceberg REST catalog (eager). rejectUnsupportedAwsAssumeRoleProperty(ICEBERG_REST_ROLE_ARN); rejectUnsupportedAwsAssumeRoleProperty(ICEBERG_REST_EXTERNAL_ID); diff --git a/fe/fe-connector/fe-connector-metastore-iceberg/src/test/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStorePropertiesTest.java b/fe/fe-connector/fe-connector-metastore-iceberg/src/test/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStorePropertiesTest.java index 8e29f5f67cc708..a2280f37d9b6ca 100644 --- a/fe/fe-connector/fe-connector-metastore-iceberg/src/test/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStorePropertiesTest.java +++ b/fe/fe-connector/fe-connector-metastore-iceberg/src/test/java/org/apache/doris/connector/metastore/iceberg/rest/IcebergRestMetaStorePropertiesTest.java @@ -107,13 +107,16 @@ public void rule7OAuth2MutuallyExclusiveCredentialAndToken() { } @Test - public void rule8And9SigningNameRequiresRegionAndSigV4() { + public void managedSigningNamesRequireRegionAndSigV4() { Assertions.assertEquals( "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is glue", validateError(raw("iceberg.rest.signing-name", "glue"))); Assertions.assertEquals( "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is s3tables", validateError(raw("iceberg.rest.signing-name", "s3tables"))); + Assertions.assertEquals( + "Rest Catalog requires signing-region and sigv4-enabled set to true when signing-name is osstables", + validateError(raw("iceberg.rest.signing-name", "osstables"))); // satisfied when both region + sigv4-enabled present. IcebergRestMetaStoreProperties.of(raw("iceberg.rest.signing-name", "glue", "iceberg.rest.signing-region", "us-east-1", "iceberg.rest.sigv4-enabled", "true")).validate(); @@ -122,6 +125,17 @@ public void rule8And9SigningNameRequiresRegionAndSigV4() { IcebergRestMetaStoreProperties.of(raw("iceberg.rest.signing-name", "Glue")).validate(); } + @Test + public void managedSigningNamesRejectDisabledSigV4() { + for (String signingName : new String[] {"glue", "s3tables", "osstables"}) { + Assertions.assertEquals( + "Rest Catalog requires sigv4-enabled set to true when signing-name is " + signingName, + validateError(raw("iceberg.rest.signing-name", signingName, + "iceberg.rest.signing-region", "us-east-1", + "iceberg.rest.sigv4-enabled", "false"))); + } + } + @Test public void rule10AccessKeyAndSecretMustBeSetTogether() { Assertions.assertEquals("iceberg.rest.access-key-id and iceberg.rest.secret-access-key must be set together", From aef2570679f7152ab752e959358c0e71fdd99cac Mon Sep 17 00:00:00 2001 From: xylaaaaa <2392805527@qq.com> Date: Fri, 7 Aug 2026 16:25:54 +0800 Subject: [PATCH 2/2] [fix](catalog) Preserve OSS storage settings in Iceberg catalogs ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: CREATE CATALOG connectivity validation constructed a connector context without the raw storage properties, so OSS credentials were unavailable during the validation request. At runtime, auto-detection could also bind both OSS and generic S3, and scan/write configuration merging allowed the generic S3 endpoint to overwrite the explicit OSS endpoint. Pass a snapshot of raw properties to the creation context and consistently select the cloud-specific S3-compatible binding across Iceberg catalog, scan, and write paths. ### Release note Fix OSS Tables catalog creation and data access when OSS storage is auto-detected without fs.oss.support. ### Check List (For Author) - Test: Unit Test and Manual test - IcebergCatalogFactoryTest, IcebergScanPlanProviderTest, IcebergWritePlanProviderTest, and CatalogFactoryPluginRoutingTest - Created an OSS Tables REST catalog without fs.oss.support, then created a database/table and verified INSERT and SELECT - Behavior changed: Yes. Explicit OSS settings now take precedence over the generic S3 fallback for Iceberg catalog, scan, and write configuration. - Does this need documentation: No --- .../iceberg/IcebergCatalogFactory.java | 20 ++++++ .../connector/iceberg/IcebergConnector.java | 3 +- .../iceberg/IcebergScanPlanProvider.java | 3 +- .../iceberg/IcebergWritePlanProvider.java | 3 +- .../FakeS3CompatibleStorageProperties.java | 25 +++++++ .../iceberg/IcebergCatalogFactoryTest.java | 11 ++++ .../iceberg/IcebergScanPlanProviderTest.java | 28 ++++++++ .../iceberg/IcebergWritePlanProviderTest.java | 26 ++++++++ .../connector/DefaultConnectorContext.java | 14 ++++ .../doris/datasource/CatalogFactory.java | 3 +- .../CatalogFactoryPluginRoutingTest.java | 66 +++++++++++++++++++ 11 files changed, 198 insertions(+), 4 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java index f3d8219565d498..9cc82b71b0f9d0 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java +++ b/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; @@ -211,6 +212,25 @@ public static Optional 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 selectEffectiveStorages( + List storages) { + S3CompatibleFileSystemProperties chosenS3 = chooseS3Compatible(storages).orElse(null); + List 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 diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java index 47df2d3f9c6f06..c32f8bb6d8ce09 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java @@ -1219,7 +1219,8 @@ static Map deriveHdfsDefaultFsFromWarehouse(String warehouse) { */ private Map buildStorageHadoopConfig() { Map 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; diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java index a38af307a3d5a2..f6e1dde65ad488 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java @@ -1654,7 +1654,8 @@ public Map getScanNodeProperties( // overlay, just the vended one below. if (context != null) { Map 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)); diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java index 02bb47ac08c544..758d1ab51f8bc9 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java @@ -929,7 +929,8 @@ private Map 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 diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeS3CompatibleStorageProperties.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeS3CompatibleStorageProperties.java index eab46928e514d9..4195232ba216da 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/FakeS3CompatibleStorageProperties.java +++ b/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 backendProperties = Collections.emptyMap(); FakeS3CompatibleStorageProperties(String providerName) { this.providerName = providerName; @@ -88,6 +93,11 @@ FakeS3CompatibleStorageProperties usePathStyle(String v) { return this; } + FakeS3CompatibleStorageProperties backendProperties(Map v) { + this.backendProperties = Collections.unmodifiableMap(new HashMap<>(v)); + return this; + } + @Override public String providerName() { return providerName; @@ -183,4 +193,19 @@ public Set 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 toBackendProperties() { + return Optional.of(new BackendStorageProperties() { + @Override + public BackendStorageKind backendKind() { + return BackendStorageKind.S3_COMPATIBLE; + } + + @Override + public Map toMap() { + return backendProperties; + } + }); + } } diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java index eed8cd1377b7b6..e6bf7c0ca7ddf2 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java +++ b/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 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 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 55f6c087f08931..4bcba4b6e607ea 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 @@ -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 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"); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java index bcfaf94a932a44..0d8a1131baba1a 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java +++ b/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 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 d5a70b233e9ea1..b0728d2d389fdf 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 @@ -127,6 +127,20 @@ public DefaultConnectorContext(String catalogName, long catalogId) { this(catalogName, catalogId, () -> NOOP_AUTH); } + /** + * Creates the lightweight pre-initialization context used by the catalog factory for CREATE validation and + * edit-log replay. It exposes a snapshot of the raw storage properties so connector construction and + * connectivity checks can bind credentials, but it does not provide runtime authentication, + * connector-derived storage defaults, backend adapters, or a filesystem. + */ + public static DefaultConnectorContext forCatalogCreationValidation(String catalogName, long catalogId, + Map rawStorageProperties) { + Map rawSnapshot = Collections.unmodifiableMap( + new HashMap<>(Objects.requireNonNull(rawStorageProperties, "rawStorageProperties"))); + return new DefaultConnectorContext(catalogName, catalogId, () -> NOOP_AUTH, + Collections::emptyMap, () -> new HashMap<>(rawSnapshot)); + } + public DefaultConnectorContext(String catalogName, long catalogId, Supplier authSupplier) { this(catalogName, catalogId, authSupplier, Collections::emptyMap); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java index a781fcc9eaa7ee..cf4102880321c4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java @@ -114,7 +114,8 @@ private static CatalogIf createCatalog(long catalogId, String name, String resou Connector spiConnector; try { spiConnector = ConnectorFactory.createStandaloneCatalogConnector( - catalogType, props, new DefaultConnectorContext(name, catalogId)); + catalogType, props, + DefaultConnectorContext.forCatalogCreationValidation(name, catalogId, props)); } catch (RuntimeException | Error e) { if (!isReplay) { // Creating a catalog interactively must still fail loud: the user is waiting for the error. diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryPluginRoutingTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryPluginRoutingTest.java index 7fd68249309618..2707e406acb13d 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryPluginRoutingTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/CatalogFactoryPluginRoutingTest.java @@ -26,8 +26,12 @@ import org.apache.doris.connector.spi.ConnectorMetadata; import org.apache.doris.connector.spi.ConnectorProvider; import org.apache.doris.connector.spi.ConnectorSession; +import org.apache.doris.connector.spi.ConnectorTestResult; import org.apache.doris.datasource.log.CatalogLog; import org.apache.doris.datasource.plugin.PluginDrivenExternalCatalog; +import org.apache.doris.filesystem.properties.S3CompatibleFileSystemProperties; +import org.apache.doris.fs.FileSystemFactory; +import org.apache.doris.fs.FileSystemPluginManager; import org.apache.doris.nereids.trees.plans.commands.CreateCatalogCommand; import com.google.common.collect.Maps; @@ -64,6 +68,7 @@ void setUp() { @AfterEach void tearDown() { ConnectorFactory.initPluginManager(new ConnectorPluginManager()); + FileSystemFactory.initPluginManager(null); } @Test @@ -156,6 +161,67 @@ void everyReservedTypeNameIsStillServedByTheEngineItself() throws Exception { } } + @Test + void createConnectivityTestSeesRawStorageProperties() throws Exception { + FileSystemPluginManager fileSystemManager = new FileSystemPluginManager(); + fileSystemManager.loadBuiltins(); + FileSystemFactory.initPluginManager(fileSystemManager); + + AtomicInteger connectivityTests = new AtomicInteger(); + ConnectorPluginManager connectorManager = new ConnectorPluginManager(); + connectorManager.registerProvider(new ConnectorProvider() { + @Override + public String getType() { + return THIRD_PARTY_TYPE; + } + + @Override + public boolean isStandaloneCatalogType() { + return true; + } + + @Override + public Connector create(Map properties, ConnectorContext context) { + return new Connector() { + @Override + public ConnectorMetadata getMetadata(ConnectorSession session) { + return null; + } + + @Override + public boolean defaultTestConnection() { + return true; + } + + @Override + public ConnectorTestResult testConnection(ConnectorSession session) { + connectivityTests.incrementAndGet(); + boolean found = context.getStorageContext().getStorageProperties().stream() + .filter(S3CompatibleFileSystemProperties.class::isInstance) + .map(S3CompatibleFileSystemProperties.class::cast) + .anyMatch(storage -> "OSS".equals(storage.providerName()) + && "create-ak".equals(storage.getAccessKey()) + && "create-sk".equals(storage.getSecretKey())); + return found ? ConnectorTestResult.success() + : ConnectorTestResult.failure("CREATE context did not expose raw storage props"); + } + }; + } + }); + ConnectorFactory.initPluginManager(connectorManager); + + Map properties = props(THIRD_PARTY_TYPE); + properties.put("oss.endpoint", "https://oss-cn-beijing.aliyuncs.com"); + properties.put("oss.region", "cn-beijing"); + properties.put("oss.access_key", "create-ak"); + properties.put("oss.secret_key", "create-sk"); + CatalogIf catalog = CatalogFactory.createFromCommand( + 1L, command("storage_ctx_ctl", properties)); + + Assertions.assertInstanceOf(PluginDrivenExternalCatalog.class, catalog); + Assertions.assertEquals(1, connectivityTests.get()); + } + /** Registers a single fake provider and returns a counter of how often it was asked to build a connector. */ private static AtomicInteger registerProvider(String type, boolean standalone) { AtomicInteger consulted = new AtomicInteger();