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..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; @@ -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 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 @@ -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. * *

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 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/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 b86600b248d7ed..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 @@ -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 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 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 +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 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 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 opts = new HashMap<>(); appendRest(opts, 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-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", 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();