This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new cbb31202e89 [feature](catalog) Support Alibaba Cloud OSS Tables REST
catalog (#66567)
cbb31202e89 is described below
commit cbb31202e8958000f9f82c4c1980a08b29f26337
Author: Chenjunwei <[email protected]>
AuthorDate: Wed Aug 12 15:36:46 2026 +0800
[feature](catalog) Support Alibaba Cloud OSS Tables REST catalog (#66567)
### 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. The
plugin-driven Iceberg connector only recognized Glue and S3 Tables as
managed signed REST catalogs. Consequently, an OSS Tables catalog did
not reuse the selected OSS credentials for REST signing, and its
required signing region and SigV4 settings were not fully validated.
This change:
- recognizes `osstables` as a managed SigV4 signing name;
- reuses the selected OSS/S3-compatible AK, SK, and STS token for both
the REST control plane and S3FileIO data plane;
- requires a signing region and `sigv4-enabled=true`;
- adds coverage for the official OSS Tables endpoint shape, ACS
warehouse ARN, OSS endpoint, and session token mapping.
The current connector's connection check already initializes the real
catalog through the full property-building path, so no separate
connectivity-only implementation is needed.
---
.../connector/iceberg/IcebergCatalogFactory.java | 31 +++++++---
.../doris/connector/iceberg/IcebergConnector.java | 3 +-
.../connector/iceberg/IcebergScanPlanProvider.java | 3 +-
.../iceberg/IcebergWritePlanProvider.java | 3 +-
.../iceberg/FakeS3CompatibleStorageProperties.java | 25 ++++++++
.../iceberg/IcebergCatalogFactoryTest.java | 60 ++++++++++++++++++--
.../iceberg/IcebergScanPlanProviderTest.java | 28 +++++++++
.../iceberg/IcebergWritePlanProviderTest.java | 26 +++++++++
.../rest/IcebergRestMetaStoreProperties.java | 17 +++++-
.../rest/IcebergRestMetaStorePropertiesTest.java | 16 +++++-
.../doris/connector/DefaultConnectorContext.java | 14 +++++
.../apache/doris/datasource/CatalogFactory.java | 3 +-
.../CatalogFactoryPluginRoutingTest.java | 66 ++++++++++++++++++++++
13 files changed, 276 insertions(+), 19 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 232623da99b..9cc82b71b0f 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.aws.s3.S3FileIOProperties;
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 final class IcebergCatalogFactory {
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 final class IcebergCatalogFactory {
/**
* 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 @@ public final class IcebergCatalogFactory {
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 47df2d3f9c6..c32f8bb6d8c 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 @@ public class IcebergConnector implements Connector {
*/
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;
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 09f8fa44615..4f0f8417c66 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
@@ -1656,7 +1656,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// 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));
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 16b6ea50c19..c6b9a85656a 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
@@ -951,7 +951,8 @@ public class IcebergWritePlanProvider implements
ConnectorWritePlanProvider {
// 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 eab46928e51..4195232ba21 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<String, String> backendProperties = Collections.emptyMap();
FakeS3CompatibleStorageProperties(String providerName) {
this.providerName = providerName;
@@ -88,6 +93,11 @@ final class FakeS3CompatibleStorageProperties implements
S3CompatibleFileSystemP
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 @@ final class FakeS3CompatibleStorageProperties implements
S3CompatibleFileSystemP
// 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;
+ }
+ });
+ }
}
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 b86600b248d..e6bf7c0ca7d 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 class IcebergCatalogFactoryTest {
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 class IcebergCatalogFactoryTest {
@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 class IcebergCatalogFactoryTest {
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 class IcebergCatalogFactoryTest {
@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 class IcebergCatalogFactoryTest {
@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 class IcebergCatalogFactoryTest {
@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,
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 26fa913062a..9d0041bc602 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.TIcebergFileDesc;
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;
@@ -2903,6 +2904,33 @@ public class IcebergScanPlanProviderTest {
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");
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 4ab51cbe4e7..bcd7f07b129 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
@@ -758,6 +758,32 @@ public class IcebergWritePlanProviderTest {
"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 538d52406c2..90e079640ec 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 @@ import java.util.Map;
* <li>OAuth2 credential/token mutually exclusive (ParamRules)</li>
* <li>signing-name=glue requires signing-region + sigv4-enabled
(ParamRules)</li>
* <li>signing-name=s3tables requires signing-region + sigv4-enabled
(ParamRules)</li>
+ * <li>signing-name=osstables requires signing-region + sigv4-enabled
(ParamRules)</li>
+ * <li>managed signing names require sigv4-enabled=true (ParamRules)</li>
* <li>access-key-id + secret-access-key set together (ParamRules)</li>
* </ol>
* 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 final class IcebergRestMetaStoreProperties extends
AbstractMetaStorePrope
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 final class IcebergRestMetaStoreProperties extends
AbstractMetaStorePrope
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 8e29f5f67cc..a2280f37d9b 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 class IcebergRestMetaStorePropertiesTest {
}
@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 class IcebergRestMetaStorePropertiesTest {
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 d5a70b233e9..b0728d2d389 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 class DefaultConnectorContext implements
ConnectorContext, ConnectorStora
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<String, String> rawStorageProperties) {
+ Map<String, String> 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<ExecutionAuthenticator> 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 a781fcc9eaa..cf410288032 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 @@ public class CatalogFactory {
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 7fd68249309..2707e406acb 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.ConnectorContext;
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 @@ public class CatalogFactoryPluginRoutingTest {
@AfterEach
void tearDown() {
ConnectorFactory.initPluginManager(new ConnectorPluginManager());
+ FileSystemFactory.initPluginManager(null);
}
@Test
@@ -156,6 +161,67 @@ public class CatalogFactoryPluginRoutingTest {
}
}
+ @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<String, String> 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<String, String> 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();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
