github-actions[bot] commented on code in PR #66004:
URL: https://github.com/apache/doris/pull/66004#discussion_r3654979415
##########
fe/fe-core/src/main/java/org/apache/doris/common/util/LocationPath.java:
##########
@@ -150,14 +152,14 @@ public static LocationPath of(String location,
schema = Strings.nullToEmpty(uri.getScheme());
}
- return new LocationPath(schema, normalizedLocation, fsIdentifier,
storageProperties);
+ return new LocationPath(schema, normalizedLocation, fsIdentifier,
storageAdapter);
}
- public static StorageProperties.Type fromSchemaWithContext(String
location, String schema) {
+ public static StorageTypeId fromSchemaWithContext(String location, String
schema) {
if (isHdfsOnOssEndpoint(location)) {
- return StorageProperties.Type.OSS_HDFS;
+ return StorageTypeId.OSS_HDFS;
}
- return SchemaTypeMapper.fromSchema(schema); // fallback to default
+ return StorageRegistry.fromScheme(schema); // fallback to default
Review Comment:
[P1] Route provider-declared schemes through the bound adapter
`FileSystemProperties.getSupportedSchemes()` is documented as the routing
source of truth, and the new typed `XyzProps` test advertises `{s3, xyz}`, but
this lookup only knows the static built-in registry. For `xyz://bucket/key`,
`type` is null, `findStorageAdapter` misses every fallback, and resolution
fails before the typed provider can normalize or create a filesystem; the later
BE/filesystem type lookups are static as well. This makes the advertised
zero-fe-core-change native scheme unusable through catalogs and
`SpiSwitchingFileSystem`. Please resolve unknown schemes from the bound
adapters and carry the selected adapter's S3-family type through
normalization/BE dispatch, with an end-to-end `xyz://` test.
##########
fe/fe-filesystem/fe-filesystem-hdfs-base/src/main/java/org/apache/doris/filesystem/hdfs/properties/HdfsCompatibleProperties.java:
##########
@@ -85,6 +267,11 @@ protected void checkRequiredProperties() {
}
}
}
+ if ("kerberos".equalsIgnoreCase(hdfsAuthenticationType) &&
(Strings.isNullOrEmpty(hdfsKerberosPrincipal)
Review Comment:
[P1] Keep OSS-HDFS out of the shared Kerberos validation
In the legacy hierarchy, the Kerberos fields, validation, and authenticator
selection lived only in `HdfsProperties`; `OSSHdfsProperties` inherited the
simple-auth base. `OssHdfsProperties` now inherits this check unconditionally,
so an otherwise valid OSS-HDFS catalog carrying a shared
`hadoop.security.authentication=kerberos` setting fails unless HDFS
principal/keytab values are present, and with those values it selects a
Kerberos authenticator instead of the legacy simple one. This also contradicts
the JFS-thread follow-up that OSS-HDFS stayed unchanged. Please make the shared
auth model opt-in for HDFS/JFS (or explicitly opt OSS-HDFS out) and add an
OSS-HDFS parity case.
##########
fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemPluginManager.java:
##########
@@ -141,6 +158,197 @@ public void registerProvider(FileSystemProvider provider)
{
DatasourcePrintableMap.registerSensitiveKeys(provider.sensitivePropertyKeys());
}
+ /** Returns an unmodifiable view of the loaded providers, in registration
order. */
+ public List<FileSystemProvider> getProviders() {
+ return Collections.unmodifiableList(providers);
+ }
+
+ /**
+ * Selection priority for raw-user-props binding = {@code
StorageRegistry.Provider} declaration
+ * order (single source of truth; mirrors the legacy
StorageProperties.PROVIDERS registry
+ * order, with JFS hoisted ahead of HDFS so a jfs:// uri beats HDFS's
key-hint guess).
+ */
+
+ private static final String DEPRECATED_OSS_HDFS_SUPPORT =
"oss.hdfs.enabled";
+
+ /**
+ * Binds the first matching provider for raw user properties, mirroring
+ * {@code StorageProperties.createPrimary}: fixed priority, explicit
{@code fs.<x>.support}
+ * flags first-class, heuristic guesses globally disabled once any
explicit flag is present,
+ * and no default fallback — throws when nothing matches.
+ */
+ public FileSystemProperties bindPrimary(Map<String, String> properties) {
+ boolean useGuess = !hasAnyExplicitFsSupport(properties);
+ Map<String, String> probeView = withProbeContext(properties);
+ for (StorageRegistry.Provider meta :
StorageRegistry.Provider.values()) {
+ FileSystemProvider provider = providerByName(meta.name());
+ if (provider == null) {
+ continue;
+ }
+ if (provider.supportsExplicit(properties) || (useGuess &&
provider.supportsGuess(probeView))) {
+ return provider.bind(properties);
+ }
+ }
+ // Providers not in the StorageRegistry.Provider table (out-of-tree
plugins) are consulted
+ // after every known provider, in registration order — adding a plugin
needs no fe-core
+ // priority-list change and can never preempt the built-in set.
+ for (FileSystemProvider provider : providers) {
+ if (StorageRegistry.Provider.byName(provider.name()).isPresent()) {
+ continue;
+ }
+ FileSystemProperties bound = tryBindUnlisted(provider, properties,
useGuess, probeView);
+ if (bound != null) {
+ return bound;
+ }
+ }
+ throw new StoragePropertiesException(
+ "No supported storage type found. Please check your
configuration."
+ + " Loaded filesystem providers: " + providerNames());
+ }
+
+ /**
+ * Binds every matching provider for raw user properties, mirroring
+ * {@code StorageProperties.createAll}: multi-hit is allowed, OSS-HDFS and
OSS stay mutually
+ * exclusive, and when no explicit flag is present and no HDFS-family
provider matched, a
+ * default HDFS binding is inserted at index 0 (fe-core's default-HDFS
fallback).
+ */
+ public List<FileSystemProperties> bindAll(Map<String, String> properties) {
+ boolean useGuess = !hasAnyExplicitFsSupport(properties);
+ Map<String, String> probeView = withProbeContext(properties);
+ List<FileSystemProperties> result = new ArrayList<>();
+ boolean ossHdfsMatched = false;
+ boolean hdfsFamilyMatched = false;
+ for (StorageRegistry.Provider meta :
StorageRegistry.Provider.values()) {
+ FileSystemProvider provider = providerByName(meta.name());
+ if (provider == null) {
+ continue;
+ }
+ if (meta == StorageRegistry.Provider.OSS && ossHdfsMatched) {
+ continue;
+ }
+ // JFS and HDFS were ONE typed class in fe-core (jfs rode
HdfsProperties), so a map
+ // matching both (jfs fs.defaultFS also trips HDFS's key-hint
guess) produced a
+ // single instance there. The plugin split makes them two
providers sharing
+ // StorageTypeId.HDFS — bind at most one or the type-keyed catalog
map collides
+ // ("Duplicate storage type: HDFS", caught by external_table_p0
jfs catalog tests).
+ if (meta == StorageRegistry.Provider.HDFS && hdfsFamilyMatched) {
+ continue;
+ }
+ if (provider.supportsExplicit(properties) || (useGuess &&
provider.supportsGuess(probeView))) {
+ result.add(provider.bind(properties));
+ if (meta == StorageRegistry.Provider.OSS_HDFS) {
+ ossHdfsMatched = true;
+ }
+ // fe-core adds its default-HDFS fallback only when no
HdfsProperties instance is
+ // in the result; jfs rode HdfsProperties there, so JFS counts
as HDFS here.
+ // OSS_HDFS does NOT: fe-core's OSSHdfsProperties is not an
HdfsProperties, and
+ // createAll happily prepends the HDFS fallback next to it.
+ if (meta == StorageRegistry.Provider.HDFS || meta ==
StorageRegistry.Provider.JFS) {
+ hdfsFamilyMatched = true;
+ }
+ }
+ }
+ // Unlisted (out-of-tree) providers append after the known set,
registration order.
+ for (FileSystemProvider provider : providers) {
+ if (StorageRegistry.Provider.byName(provider.name()).isPresent()) {
+ continue;
+ }
+ FileSystemProperties bound = tryBindUnlisted(provider, properties,
useGuess, probeView);
+ if (bound != null) {
+ result.add(bound);
+ }
+ }
+ if (useGuess && !hdfsFamilyMatched) {
+ FileSystemProvider hdfs = providerByName("HDFS");
+ if (hdfs != null) {
+ result.add(0, hdfs.bind(properties));
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Probe-context key carrying {@code Config.azure_blob_host_suffixes} into
provider guess
+ * probes. Providers cannot see fe-core Config; routing runs here in
fe-core, so the live
+ * (admin-extensible) suffix list is injected into a COPY used only for
+ * {@code supportsGuess} probes — the map handed to {@code bind()} stays
untouched, so the
+ * marker can never leak into bound properties, backend maps, or persisted
state.
+ */
+ public static final String AZURE_HOST_SUFFIXES_PROBE_KEY =
"_AZURE_HOST_SUFFIXES_";
+
+ /**
+ * Typed-binding attempt for a provider outside the built-in registry.
Typed providers
+ * (supportsExplicit/supportsGuess + bind) are selected normally. A legacy
raw-SPI provider
+ * — one that only implements {@code supports(Map)}/{@code create(Map)}
and inherits the
+ * throwing default {@code bind()} — is still CONSULTED via {@code
supports()} so it is not
+ * silently invisible, but typed binding cannot represent it: it is
rejected with an
+ * actionable warning (implement the typed hooks) and remains usable
through the raw
+ * {@code createFileSystem(Map)} entry point.
+ */
+ private FileSystemProperties tryBindUnlisted(FileSystemProvider provider,
+ Map<String, String> properties, boolean useGuess, Map<String,
String> probeView) {
+ boolean typedMatch = provider.supportsExplicit(properties)
+ || (useGuess && provider.supportsGuess(probeView));
+ if (!typedMatch) {
+ if (useGuess && provider.supports(properties)) {
+ LOG.warn("Filesystem provider '{}' matches these properties
via the legacy"
+ + " supports() hook only; typed binding requires
supportsExplicit/"
+ + "supportsGuess/bind — implement them to make the
provider selectable"
+ + " for catalogs/TVFs. It remains usable via
createFileSystem(Map).",
+ provider.name());
+ }
+ return null;
+ }
+ try {
+ return provider.bind(properties);
+ } catch (UnsupportedOperationException e) {
+ LOG.warn("Filesystem provider '{}' matched but does not implement
typed bind();"
+ + " skipping it for typed binding. Implement bind() to
integrate with the"
+ + " storage facade.", provider.name());
+ return null;
+ }
+ }
+
+ /** Builds the guess-probe view of the raw properties (see {@link
#AZURE_HOST_SUFFIXES_PROBE_KEY}). */
+ public static Map<String, String> withProbeContext(Map<String, String>
properties) {
+ String[] suffixes = Config.azure_blob_host_suffixes;
+ if (suffixes == null || suffixes.length == 0) {
+ return properties;
+ }
+ Map<String, String> probeView = new HashMap<>(properties);
+ probeView.put(AZURE_HOST_SUFFIXES_PROBE_KEY, String.join(",",
suffixes));
+ return probeView;
+ }
+
+ private FileSystemProvider providerByName(String name) {
+ for (FileSystemProvider p : providers) {
+ if (name.equalsIgnoreCase(p.name())) {
+ return p;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * True when the user explicitly declared any filesystem via {@code
fs.<x>.support=true} or the
+ * legacy {@code oss.hdfs.enabled=true}. Counterpart of
+ * {@code StorageProperties.hasAnyExplicitFsSupport}; checked structurally
so this class never
+ * needs the full dialect list.
+ */
+ private static boolean hasAnyExplicitFsSupport(Map<String, String>
properties) {
+ if
("true".equalsIgnoreCase(properties.getOrDefault(DEPRECATED_OSS_HDFS_SUPPORT,
"false"))) {
+ return true;
+ }
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ String key = entry.getKey();
+ if (key != null && key.startsWith("fs.") &&
key.endsWith(".support")
Review Comment:
[P1] Do not let unrelated fs.* flags suppress GCS guessing
This structural wildcard treats `fs.gs.support=true` as an explicit Doris
provider selection, but `GcsFileSystemProvider.supportsExplicit()` recognizes
only `fs.gcs.support`. The existing Paimon catalog regression uses
`fs.gs.support=true` with both native GCS and S3-style Google endpoints; the
legacy gate did not recognize that spelling, so endpoint guessing still
selected GCS. Here the wildcard disables every guess and leaves those catalogs
with no GCS binding. Please derive the gate from providers that actually report
`supportsExplicit(properties)`, or preserve/alias the recognized compatibility
keys, and cover both exercised `fs.gs.support` cases.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/storage/StorageUriUtils.java:
##########
@@ -0,0 +1,360 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.storage;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.S3URI;
+import org.apache.doris.foundation.property.StoragePropertiesException;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLDecoder;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Pattern;
+
+/**
+ * fe-core-only URI validation/normalization glue for {@link StorageAdapter}.
+ *
+ * <p>Ports the URI subset of the legacy {@code S3PropertyUtils} and {@code
AzurePropertyUtils}
+ * pure functions so the facade does not depend on the to-be-deleted
+ * legacy typed storage package (deleted in Phase D). Behavior is kept
identical, with one deliberate
+ * exception: the legacy full-parse path could throw the checked {@code
UserException} (from
+ * {@link S3URI#create}); the facade surface is unchecked, so that exception
is rethrown as
+ * {@link StoragePropertiesException} with the same message.</p>
+ */
+public final class StorageUriUtils {
+
+ private static final String URI_KEY = "uri";
+ private static final String SCHEME_DELIM = "://";
+ private static final String S3_SCHEME_PREFIX = "s3://";
+
+ // S3-compatible schemes that can be converted to s3:// with simple string
replacement
+ // Format: scheme://bucket/key -> s3://bucket/key
+ private static final String[] SIMPLE_S3_COMPATIBLE_SCHEMES = {
+ "s3a", "s3n", "oss", "cos", "cosn", "obs", "bos", "gs"
+ };
+
+ private static final Pattern ONELAKE_PATTERN = Pattern.compile(
+
"abfs[s]?://([^@]+)@([^/]+)\\.dfs\\.fabric\\.microsoft\\.com(/.*)?",
Pattern.CASE_INSENSITIVE);
+
+ private StorageUriUtils() {
+ }
+
+ /**
+ * Port of legacy {@code S3PropertyUtils.validateAndNormalizeUri}:
validates and normalizes
+ * the given path into a standard {@code s3://bucket/key} URI.
+ *
+ * <p>Also carries the legacy {@code
OSSProperties.validateAndNormalizeUri} override when
+ * {@code applyOssBucketRewrite} is true (the caller's binding is the OSS
provider): a
+ * virtual-hosted {@code scheme://bucket.endpoint/key} authority is
rewritten to
+ * {@code scheme://bucket/key} before the base normalization,
byte-identical to the legacy
+ * {@code rewriteOssBucketIfNecessary}. Non-OSS bindings never rewrite — a
dotted bucket
+ * name under an explicit S3 binding (e.g. {@code oss://logs.prod/…} with
+ * {@code fs.s3.support=true}) passes through unchanged, exactly like
legacy.</p>
+ */
+ public static String validateAndNormalizeS3Uri(String path,
+ String stringUsePathStyle,
+ String stringForceParsingByStandardUri,
+ boolean applyOssBucketRewrite) {
+ if (StringUtils.isBlank(path)) {
+ throw new StoragePropertiesException("path is null");
+ }
+
+ // Legacy OSSProperties applied the bucket-domain rewrite before the
base normalization.
+ if (applyOssBucketRewrite) {
+ path = rewriteOssBucketIfNecessary(path);
+ }
+
+ // Fast path 1: s3:// paths are already in the normalized format
expected by BE
+ if (path.startsWith(S3_SCHEME_PREFIX)) {
+ return path;
+ }
+
+ // Fast path 2: simple S3-compatible schemes (oss://, cos://, s3a://,
etc.)
+ // can be converted with simple string replacement:
scheme://bucket/key -> s3://bucket/key
+ String normalized = trySimpleSchemeConversion(path);
+ if (normalized != null) {
+ return normalized;
+ }
+
+ // Full parsing path: for HTTP URLs and other complex formats
+ boolean usePathStyle = Boolean.parseBoolean(stringUsePathStyle);
+ boolean forceParsingByStandardUri =
Boolean.parseBoolean(stringForceParsingByStandardUri);
+ S3URI s3uri;
+ try {
+ s3uri = S3URI.create(path, usePathStyle,
forceParsingByStandardUri);
+ } catch (UserException e) {
+ // Facade surface is unchecked; keep the legacy message.
+ throw new StoragePropertiesException(e.getMessage(), e);
Review Comment:
[P2] Preserve user-error classification for malformed S3 URLs
The legacy S3 helper let `S3URI.create()` propagate `UserException`, and
`BrokerDesc.getFileLocation()` still advertises that checked contract. Wrapping
it here changes a reachable REST path: `ImportAction.fileReview()` catches only
`UserException` and returns `okWithCommonError`, so a malformed full-parse
location such as `not a uri` now escapes to the generic exception handler as an
internal error. Please preserve/translate the checked user error at the facade
or broker boundary (or update every checked-error consumer), and cover the
file-review response.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/storage/StorageAdapter.java:
##########
@@ -0,0 +1,878 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.storage;
+
+import org.apache.doris.common.Config;
+import
org.apache.doris.datasource.property.common.AwsCredentialsProviderFactory;
+import org.apache.doris.datasource.property.common.AwsCredentialsProviderMode;
+import org.apache.doris.filesystem.properties.FileSystemProperties;
+import org.apache.doris.filesystem.properties.HadoopStorageProperties;
+import org.apache.doris.filesystem.properties.S3CompatibleFileSystemProperties;
+import org.apache.doris.filesystem.spi.FileSystemProvider;
+import org.apache.doris.foundation.property.StoragePropertiesException;
+import org.apache.doris.foundation.security.ExecutionAuthenticator;
+import org.apache.doris.fs.FileSystemPluginManager;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.lang3.BooleanUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import
software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import
software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * fe-core facade over the fe-filesystem SPI typed properties (Phase B3 of the
+ * StorageProperties removal plan).
+ *
+ * <p>One adapter wraps one {@link FileSystemProperties} binding produced by
+ * {@link FileSystemPluginManager#bindPrimary}/{@code bindAll}. Its public
surface mirrors the
+ * legacy typed storage-properties contract exactly — backend map,
+ * Hadoop configuration, storage name, schemas, type — so consumers can
migrate mechanically.
+ * Every known SPI-vs-fe-core drift from the master plan's §2.4 parity ledger
is reconciled here
+ * (or in the SPI implementation, where noted); each reconciliation carries a
+ * "align fe-core" comment referencing the ledger item.</p>
+ */
+public final class StorageAdapter {
+
+ private static final Logger LOG =
LogManager.getLogger(StorageAdapter.class);
+
+ /**
+ * Plugin manager injected at FE startup (see {@code
FileSystemFactory.initPluginManager},
+ * which forwards here). Held locally so the facade does not reach back
into the {@code fs}
+ * runtime layer — the dependency direction is fs → datasource.storage
only.
+ */
+ private static volatile FileSystemPluginManager pluginManager;
+
+ /**
+ * Lazy fallback registry for environments where FE startup did not run
+ * {@link #initPluginManager} (unit tests, migration phase).
+ */
+ private static volatile FileSystemPluginManager fallbackManager;
+
+ /** Called once at FE startup, before any {@code of}/{@code ofAll}
binding. */
+ public static void initPluginManager(FileSystemPluginManager manager) {
+ pluginManager = manager;
+ }
+
+ private static final String SIMPLE_AWS_CREDENTIALS_PROVIDER =
+ "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider";
+ private static final String ASSUMED_ROLE_CREDENTIAL_PROVIDER =
+ "org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider";
+
+ /**
+ * Hadoop keys the facade re-derives for the S3 provider instead of taking
the SPI values:
+ * fe-core gates them on accessKey-blank and on
Config.aws_credentials_provider_version,
+ * which the SPI layer cannot see (align fe-core, ledger 2.4-3).
+ */
+ private static final Set<String> S3_CREDENTIAL_KEYS = ImmutableSet.of(
+ "fs.s3a.aws.credentials.provider",
+ "fs.s3a.assumed.role.arn",
+ "fs.s3a.assumed.role.credentials.provider",
+ "fs.s3a.assumed.role.external.id");
+
+ /**
+ * Legacy per-dialect alias prefix for {@code
force_parsing_by_standard_uri} (fe-core
+ * {@code AbstractS3CompatibleProperties} subclasses declared
+ * {@code <prefix>.force_parsing_by_standard_uri} plus the generic key;
plain S3 had the
+ * generic key only). Read from the raw props because the SPI api
interface does not expose
+ * this flag.
+ */
+ private static final Map<String, String> FORCE_PARSING_PREFIXES =
ImmutableMap.<String, String>builder()
+ .put("OSS", "oss")
+ .put("COS", "cos")
+ .put("OBS", "obs")
+ .put("GCS", "gs")
+ .put("MINIO", "minio")
+ .put("OZONE", "ozone")
+ .build();
+
+ private final FileSystemProperties spi;
+ private final Map<String, String> origProps;
+ private final String providerKey;
+ private final StorageTypeId type;
+ /** Explicit broker name for adapters built via {@link #ofBroker}; mirrors
the legacy
+ * {@code BrokerProperties.of(name, props)} where the name is NOT part of
the raw props. */
+ private final String brokerNameOverride;
+ /** fe-core S3Properties credentials-provider mode; only resolved for the
S3 provider. */
+ private final AwsCredentialsProviderMode s3CredentialsMode;
+ /** Resolved once at construction (legacy bound it at init); read per file
in listing loops. */
+ private final String forceParsingByStandardUriValue;
+ /**
+ * Lazily built (legacy {@code X.of()} factories never built one; {@code
new Configuration()}
+ * re-parses the Hadoop XML defaults and is too expensive for
per-statement bindings that
+ * never read it, e.g. cloud COPY INTO). Volatile double-checked;
construction is a pure
+ * function of final fields.
+ */
+ private volatile Configuration hadoopStorageConfig;
+ private volatile boolean hadoopStorageConfigBuilt;
+ /**
+ * Adapters are shared across threads (catalog adapter map, FS caches), so
the lazily cached
+ * map must be safely published — legacy classes were immune (eager init
or fresh map per
+ * call); the cache is a facade addition.
+ */
+ private volatile Map<String, String> backendConfigProperties;
+
+ private StorageAdapter(FileSystemProperties spi, Map<String, String>
origProps) {
+ this(spi, origProps, null);
+ }
+
+ private StorageAdapter(FileSystemProperties spi, Map<String, String>
origProps, String brokerNameOverride) {
+ this.spi = spi;
+ this.origProps = origProps;
+ this.brokerNameOverride = brokerNameOverride;
+ this.providerKey = spi.providerName().toUpperCase(Locale.ROOT);
+ // Providers outside the StorageRegistry.Provider table (out-of-tree
plugins): an
+ // S3-compatible dialect joins the S3 family id so every
TypeId-dispatched consumer
+ // (TVF dispatch, connectivity, credentials) treats it like the
built-in dialects —
+ // no new enum value or case additions needed. Note the type-keyed
catalog map can
+ // hold one S3-family binding at a time, so one catalog cannot combine
such a plugin
+ // with another S3-family storage. Non-S3 protocols stay UNKNOWN (they
need the
+ // full scenario-B integration incl. BE support).
+ this.type = StorageRegistry.Provider.byName(providerKey)
+ .map(StorageRegistry.Provider::typeId)
+ .orElse(spi instanceof S3CompatibleFileSystemProperties
+ ? StorageTypeId.S3 : StorageTypeId.UNKNOWN);
+ // Align fe-core S3Properties.initNormalizeAndCheckProps: the mode
string is parsed with
+ // the fe-core enum (which rejects spellings the SPI tolerates) at
construction time.
+ this.s3CredentialsMode = "S3".equals(providerKey)
+ ?
AwsCredentialsProviderMode.fromString(feCoreS3CredentialsProviderType())
+ : null;
+ this.forceParsingByStandardUriValue =
resolveForceParsingByStandardUri();
+ checkAzureOauth2OnlyForIcebergRest();
+ }
+
+ /**
+ * Binds the primary provider for the raw user properties, mirroring the
legacy
+ * {@code StorageProperties.createPrimary}.
+ */
+ public static StorageAdapter of(Map<String, String> origProps) {
+ return new
StorageAdapter(manager().bindPrimary(withHadoopConfigDir(origProps)),
origProps);
+ }
+
+ /**
+ * Binds every matching provider, mirroring the legacy {@code
StorageProperties.createAll}
+ * (multi-hit, OSS-HDFS/OSS exclusivity, default HDFS fallback at index 0).
+ */
+ public static List<StorageAdapter> ofAll(Map<String, String> origProps) {
+ List<FileSystemProperties> bindings =
manager().bindAll(withHadoopConfigDir(origProps));
+ List<StorageAdapter> result = new ArrayList<>(bindings.size());
+ for (FileSystemProperties binding : bindings) {
+ result.add(new StorageAdapter(binding, origProps));
+ }
+ return result;
+ }
+
+ /**
+ * Binds one specific provider by name, bypassing routing entirely — the
facade twin of the
+ * legacy per-dialect factories ({@code S3Properties.of}, {@code
OSSProperties.of},
+ * {@code AzureProperties.of}, ...) that constructed and validated a fixed
typed class no
+ * matter what routing would have chosen for the same map.
+ */
+ public static StorageAdapter ofProvider(String providerName, Map<String,
String> origProps) {
+ Map<String, String> props = origProps == null ? new HashMap<>() :
origProps;
+ for (FileSystemProvider<?> provider : manager().getProviders()) {
+ if (provider.name().equalsIgnoreCase(providerName)) {
+ return new
StorageAdapter(provider.bind(withHadoopConfigDir(props)), props);
+ }
+ }
+ throw new StoragePropertiesException(
+ "Filesystem provider '" + providerName + "' is not available");
+ }
+
+ /**
+ * Returns whether the named provider's guess heuristics claim the raw
properties — the
+ * facade twin of the legacy per-dialect {@code guessIsMe} statics (guess
only, explicit
+ * {@code fs.<x>.support} flags deliberately NOT consulted, matching the
legacy call sites).
+ */
+ public static boolean matchesProviderGuess(String providerName,
Map<String, String> origProps) {
+ Map<String, String> props = origProps == null ? new HashMap<>() :
origProps;
+ // Same probe view the bind registry uses: carries Config-derived
guess context
+ // (e.g. the admin-extensible Azure host suffixes) into the provider.
+ Map<String, String> probeView =
FileSystemPluginManager.withProbeContext(props);
+ for (FileSystemProvider<?> provider : manager().getProviders()) {
+ if (provider.name().equalsIgnoreCase(providerName)) {
+ return provider.supportsGuess(probeView);
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Quirk preservation (converter parity): fe-filesystem resolves relative
+ * {@code hadoop.config.resources} entries against the injected {@code
_HADOOP_CONFIG_DIR_}
+ * context key because it cannot see fe-core {@code
Config.hadoop_config_dir}; the legacy
+ * conversion layer injected that key for every HDFS-family conversion.
The marker is added
+ * only to the copy handed to {@code bind()} — {@link #getOrigProps()} and
the
+ * Broker/Local/Http verbatim backend maps stay clean — and only when a
+ * {@code *config.resources} key is present (the only consumer of the
marker).
+ */
+ private static Map<String, String> withHadoopConfigDir(Map<String, String>
props) {
+ if (props == null || props.containsKey("_HADOOP_CONFIG_DIR_")
+ || StringUtils.isBlank(Config.hadoop_config_dir)) {
+ return props;
+ }
+ boolean hasConfigResources = false;
+ for (Map.Entry<String, String> entry : props.entrySet()) {
+ if (entry.getKey() != null &&
entry.getKey().endsWith("config.resources")
+ && StringUtils.isNotBlank(entry.getValue())) {
+ hasConfigResources = true;
+ break;
+ }
+ }
+ if (!hasConfigResources) {
+ return props;
+ }
+ Map<String, String> copy = new HashMap<>(props);
+ copy.put("_HADOOP_CONFIG_DIR_", Config.hadoop_config_dir);
+ return copy;
+ }
+
+ /**
+ * Builds a BROKER adapter with an explicitly supplied broker name,
mirroring the legacy
+ * {@code BrokerProperties.of(name, props)}: routing is bypassed (the
props of a
+ * {@code WITH BROKER "name"} clause carry no {@code broker.name} key),
the raw props are
+ * NOT mutated, and {@link #getBrokerName()} returns the supplied name.
+ */
+ public static StorageAdapter ofBroker(String brokerName, Map<String,
String> origProps) {
+ Map<String, String> props = origProps == null ? new HashMap<>() :
origProps;
+ for (FileSystemProvider<?> provider : manager().getProviders()) {
+ if ("BROKER".equalsIgnoreCase(provider.name())) {
+ return new StorageAdapter(provider.bind(props), props,
brokerName);
+ }
+ }
+ throw new StoragePropertiesException("Broker filesystem provider is
not available");
+ }
+
+ private static FileSystemPluginManager manager() {
+ FileSystemPluginManager managerFromStartup = pluginManager;
+ if (managerFromStartup != null) {
+ return managerFromStartup;
+ }
+ FileSystemPluginManager local = fallbackManager;
+ if (local == null) {
+ synchronized (StorageAdapter.class) {
+ local = fallbackManager;
+ if (local == null) {
+ LOG.warn("StorageAdapter used before initPluginManager;
falling back to"
+ + " built-in providers loaded from the classpath.
Expected in unit"
+ + " tests only — in production this means the FE
startup sequence"
+ + " skipped filesystem plugin initialization and
bindings may not"
+ + " match the configured plugin directory.");
+ local = new FileSystemPluginManager();
+ local.loadBuiltins();
+ fallbackManager = local;
+ }
+ }
+ }
+ return local;
+ }
+
+ /** The underlying SPI typed properties (for callers migrating deeper
integrations). */
+ public FileSystemProperties getSpiProperties() {
+ return spi;
+ }
+
+ /** The raw user properties this adapter was bound from. */
+ public Map<String, String> getOrigProps() {
+ return origProps;
+ }
+
+ public StorageTypeId getType() {
+ return type;
+ }
+
+ /** Legacy getStorageName() (2.4-1 mapping: every S3-compatible dialect
reports "S3"). */
+ public String getStorageName() {
+ // provider-declared family: the 2.4-1 "every S3 dialect reports S3"
mapping lives in the SPI
+ return spi.storageFamilyName();
+ }
+
+ /** Legacy schemas() (drives ensureDisableCache): provider-declared legacy
scheme set. */
+ public Set<String> schemas() {
+ // provider-declared legacy scheme set (ledger 2.4-6 note lives on
legacyCacheSchemes)
+ return spi.legacyCacheSchemes();
+ }
+
+ /** Hadoop configuration equivalent of legacy getHadoopStorageConfig();
null for BROKER/HTTP. */
+ public Configuration getHadoopStorageConfig() {
+ if (!hadoopStorageConfigBuilt) {
+ synchronized (this) {
+ if (!hadoopStorageConfigBuilt) {
+ hadoopStorageConfig = buildHadoopStorageConfig();
+ hadoopStorageConfigBuilt = true;
+ }
+ }
+ }
+ return hadoopStorageConfig;
+ }
+
+ /** Legacy isKerberos() — meaningful for the HDFS family only. */
+ public boolean isKerberos() {
+ return
spi.toHadoopProperties().map(HadoopStorageProperties::isKerberos).orElse(false);
+ }
+
+ /** Authenticator for HDFS-family doAs execution; DIRECT passthrough for
everything else. */
+ public ExecutionAuthenticator getExecutionAuthenticator() {
+ return spi.toHadoopProperties()
+ .map(HadoopStorageProperties::getExecutionAuthenticator)
+ .orElse(ExecutionAuthenticator.DIRECT);
+ }
+
+ /**
+ * Legacy {@code HdfsProperties.isExplicitlyConfigured()}: {@code false}
only for the
+ * default-HDFS fallback binding that {@code bindAll}/legacy {@code
createAll} auto-prepends
+ * when nothing HDFS-like matched. A binding whose own provider matches
the raw props
+ * (explicit {@code fs.x.support} flag or guess heuristics) is explicit;
non-HDFS types were
+ * always explicit in fe-core.
+ */
+ public boolean isExplicitlyConfigured() {
+ if (type != StorageTypeId.HDFS) {
+ return true;
+ }
+ for (FileSystemProvider provider : manager().getProviders()) {
+ if (providerKey.equalsIgnoreCase(provider.name())) {
+ return provider.supportsExplicit(origProps) ||
provider.supportsGuess(origProps);
+ }
+ }
+ return true;
+ }
+
+ /**
+ * fe-core S3Properties credentials-provider mode ({@code
s3.credentials_provider_type}
+ * alias family, default DEFAULT). Non-null only for the S3 provider,
mirroring the legacy
+ * class layout where only the generic S3 typed class carried the mode.
+ */
+ public AwsCredentialsProviderMode getAwsCredentialsProviderMode() {
+ return s3CredentialsMode;
+ }
+
+ /**
+ * fe-core-only AWS SDK credentials accessor, mirroring the legacy typed
classes'
+ * {@code getAwsCredentialsProvider()} overrides exactly: static
(optionally session)
+ * credentials first for every S3-compatible dialect; the S3 provider adds
assume-role and
+ * the v1/v2 chain selection per {@code
Config.aws_credentials_provider_version};
+ * OSS/GCS/COS/OBS fall back to anonymous credentials when both AK and SK
are blank;
+ * everything else (Minio/Ozone/Azure and non-S3 types) returns {@code
null}.
+ */
+ public AwsCredentialsProvider getAwsCredentialsProvider() {
+ if (!(spi instanceof S3CompatibleFileSystemProperties)) {
+ return null;
+ }
+ S3CompatibleFileSystemProperties s3 =
(S3CompatibleFileSystemProperties) spi;
+ AwsCredentialsProvider staticProvider =
staticAwsCredentialsProvider(s3);
+ if ("S3".equals(providerKey)) {
+ return s3AwsCredentialsProvider(s3, staticProvider);
+ }
+ if (staticProvider != null) {
+ return staticProvider;
+ }
+ switch (providerKey) {
+ case "OSS":
+ case "GCS":
+ case "COS":
+ case "OBS":
+ // Align fe-core OSS/GCS/COS/OBS Properties: anonymous access
when unauthenticated.
+ if (StringUtils.isBlank(s3.getAccessKey()) &&
StringUtils.isBlank(s3.getSecretKey())) {
+ return AnonymousCredentialsProvider.create();
+ }
+ return null;
+ default:
+ return null;
+ }
+ }
+
+ /** Align fe-core AbstractS3CompatibleProperties.getAwsCredentialsProvider
(static creds only). */
+ private static AwsCredentialsProvider
staticAwsCredentialsProvider(S3CompatibleFileSystemProperties s3) {
+ if (StringUtils.isNotBlank(s3.getAccessKey()) &&
StringUtils.isNotBlank(s3.getSecretKey())) {
+ if (StringUtils.isEmpty(s3.getSessionToken())) {
+ return StaticCredentialsProvider.create(
+ AwsBasicCredentials.create(s3.getAccessKey(),
s3.getSecretKey()));
+ }
+ return
StaticCredentialsProvider.create(AwsSessionCredentials.create(
+ s3.getAccessKey(), s3.getSecretKey(),
s3.getSessionToken()));
+ }
+ return null;
+ }
+
+ /** Align fe-core S3Properties.getAwsCredentialsProviderV1/V2 (assume-role
+ chain selection). */
+ private AwsCredentialsProvider
s3AwsCredentialsProvider(S3CompatibleFileSystemProperties s3,
+ AwsCredentialsProvider staticProvider) {
+ if (staticProvider != null) {
+ return staticProvider;
+ }
+ boolean v2 =
Config.aws_credentials_provider_version.equalsIgnoreCase("v2");
+ if (StringUtils.isNotBlank(s3.getRoleArn())) {
+ StsClient stsClient = StsClient.builder()
+ .region(Region.of(s3.getRegion()))
+ .credentialsProvider(v2
+ ?
AwsCredentialsProviderFactory.createV2(s3CredentialsMode, false)
+ : InstanceProfileCredentialsProvider.create())
+ .build();
+ return StsAssumeRoleCredentialsProvider.builder()
+ .stsClient(stsClient)
+ .refreshRequest(builder -> {
+
builder.roleArn(s3.getRoleArn()).roleSessionName("aws-sdk-java-v2-fe");
+ if (StringUtils.isNotBlank(s3.getExternalId())) {
+ builder.externalId(s3.getExternalId());
+ }
+ }).build();
+ }
+ // For anonymous access (no credentials required) v1 uses the
anonymous provider; v2
+ // delegates to the factory's default chain (which may include
anonymous).
+ return v2 ? AwsCredentialsProviderFactory.createV2(s3CredentialsMode,
true)
+ : AnonymousCredentialsProvider.create();
+ }
+
+ public String validateAndNormalizeUri(String uri) {
+ // Align fe-core AbstractS3CompatibleProperties/AzureProperties: the
SPI S3/Azure typed
+ // props do not normalize URIs (compat schemes like cos:// must become
s3:// before the
+ // path reaches BE or a concrete filesystem), so the facade owns the
legacy logic.
+ if (spi instanceof S3CompatibleFileSystemProperties) {
+ return StorageUriUtils.validateAndNormalizeS3Uri(uri,
+ ((S3CompatibleFileSystemProperties) spi).getUsePathStyle(),
+ forceParsingByStandardUriValue,
+ "OSS".equals(providerKey));
+ }
+ if ("AZURE".equals(providerKey)) {
+ return StorageUriUtils.validateAndNormalizeAzureUri(uri);
+ }
+ if (type == StorageTypeId.HDFS && StorageUriUtils.isJfsLocation(uri)) {
+ // Align fe-core: the legacy HDFS typed class accepted {hdfs,
viewfs, jfs} while the
+ // HDFS plugin's scheme identity excludes jfs (it has its own
plugin), so the facade
+ // owns the jfs leg of an HDFS binding.
+ return StorageUriUtils.validateAndNormalizeJfsUri(uri);
+ }
+ return spi.validateAndNormalizeUri(uri);
+ }
+
+ public String validateAndGetUri(Map<String, String> loadProps) {
+ // Align fe-core: the S3/Azure legacy classes return the RAW uri value
(no normalization)
+ // and throw when the map is empty or has no uri key; the SPI default
would silently
+ // return null instead.
+ if (spi instanceof S3CompatibleFileSystemProperties) {
+ return StorageUriUtils.validateAndGetS3Uri(loadProps);
+ }
+ if ("AZURE".equals(providerKey)) {
+ return StorageUriUtils.validateAndGetAzureUri(loadProps);
+ }
+ if (type == StorageTypeId.HDFS && loadProps != null) {
+ // jfs leg of the HDFS binding (see validateAndNormalizeUri
above): a single
+ // case-insensitive `uri` entry with a jfs scheme is normalized by
the facade.
+ String uriValue = loadProps.entrySet().stream()
+ .filter(e -> "uri".equalsIgnoreCase(e.getKey()))
+ .map(Map.Entry::getValue)
+ .filter(StringUtils::isNotBlank)
+ .findFirst()
+ .orElse(null);
+ if (uriValue != null && uriValue.split(",").length == 1
+ && StorageUriUtils.isJfsLocation(uriValue)) {
+ return StorageUriUtils.validateAndNormalizeJfsUri(uriValue);
+ }
+ }
+ return spi.validateAndGetUri(loadProps);
+ }
+
+ /** fe-core S3 family alias resolution for force_parsing_by_standard_uri
(see map above). */
+ private String resolveForceParsingByStandardUri() {
+ String prefix = FORCE_PARSING_PREFIXES.get(providerKey);
+ String value = prefix == null ? null : getOrigPropIgnoreCase(prefix +
".force_parsing_by_standard_uri");
+ if (StringUtils.isBlank(value)) {
+ value = getOrigPropIgnoreCase("force_parsing_by_standard_uri");
+ }
+ return StringUtils.isNotBlank(value) ? value : "false";
+ }
+
+ private String getOrigPropIgnoreCase(String key) {
+ for (Map.Entry<String, String> entry : origProps.entrySet()) {
+ if (key.equalsIgnoreCase(entry.getKey())) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /** Legacy BrokerProperties.getBrokerName(): the explicit name from {@link
#ofBroker} wins,
+ * otherwise a case-insensitive {@code broker.name} lookup on the raw
props. */
+ public String getBrokerName() {
+ if (brokerNameOverride != null) {
+ return brokerNameOverride;
+ }
+ return getOrigPropIgnoreCase("broker.name");
+ }
+
+ /** Legacy BrokerProperties.getBrokerParams(): {@code broker.}-prefixed
keys, prefix stripped. */
+ public Map<String, String> getBrokerParams() {
+ Map<String, String> params = new HashMap<>();
+ for (Map.Entry<String, String> entry : origProps.entrySet()) {
+ if (entry.getKey().startsWith("broker.")) {
+ params.put(entry.getKey().substring("broker.".length()),
entry.getValue());
+ }
+ }
+ return params;
+ }
+
+ /** Backend (BE) configuration map, key-for-key equal to the legacy typed
classes. */
+ public Map<String, String> getBackendConfigProperties() {
+ if (backendConfigProperties == null) {
+ backendConfigProperties = computeBackendConfigProperties();
+ }
+ return backendConfigProperties;
+ }
+
+ private Map<String, String> computeBackendConfigProperties() {
+ switch (type) {
+ case BROKER:
+ case LOCAL:
+ case HTTP:
+ // Align fe-core: Broker/Local/Http return the raw user
properties verbatim.
+ return origProps;
+ case AZURE:
+ return azureBackendConfigProperties();
+ default:
+ break;
+ }
+ Map<String, String> base = spi.toBackendProperties()
+ .orElseThrow(() -> new IllegalStateException(
+ "Provider " + providerKey + " exposes no backend
properties"))
+ .toMap();
+ if (spi instanceof S3CompatibleFileSystemProperties) {
+ return alignS3FamilyBackendMap((S3CompatibleFileSystemProperties)
spi, base);
+ }
+ if ("JFS".equals(providerKey)) {
+ return alignJfsBackendMap(base);
+ }
+ return base;
+ }
+
+ /**
+ * Align fe-core, ledger 2.4-7: with OAuth2 the legacy AzureProperties
dumps the ENTIRE
+ * Hadoop Configuration (hadoop defaults + fs.azure.* + user fs.*
passthrough +
+ * disable-cache keys) as the backend map; shared-key uses the SPI's exact
7-key map.
+ */
+ private Map<String, String> azureBackendConfigProperties() {
+ if (!isAzureOauth2()) {
+ return spi.toBackendProperties().orElseThrow().toMap();
+ }
+ Map<String, String> dump = new HashMap<>();
+ getHadoopStorageConfig().forEach(entry -> dump.put(entry.getKey(),
entry.getValue()));
+ return dump;
+ }
+
+ /**
+ * Backend-map reconciliation for every S3-compatible provider
(S3/OSS/OBS/COS/GCS/MinIO/Ozone).
+ */
+ private Map<String, String>
alignS3FamilyBackendMap(S3CompatibleFileSystemProperties s3,
+ Map<String, String> base) {
+ Map<String, String> aligned = new HashMap<>(base);
+ // Align fe-core, ledger 2.4-5: fe-core never emits AWS_BUCKET /
AWS_ROOT_PATH to BE.
+ aligned.remove("AWS_BUCKET");
+ aligned.remove("AWS_ROOT_PATH");
+ // Align fe-core, ledger 2.4-5: endpoint/region/AK/SK are emitted
unconditionally —
+ // anonymous access sends empty-string AWS_ACCESS_KEY/AWS_SECRET_KEY,
not absent keys.
+ aligned.put("AWS_ENDPOINT", s3.getEndpoint());
+ aligned.put("AWS_REGION", s3.getRegion());
+ aligned.put("AWS_ACCESS_KEY", s3.getAccessKey());
+ aligned.put("AWS_SECRET_KEY", s3.getSecretKey());
+ if ("S3".equals(providerKey)) {
+ // Align fe-core, ledger 2.4-3/5: S3Properties always emits the
mode name (default
+ // DEFAULT, even with static AK/SK), parsed by the fe-core enum.
AWS_ROLE_ARN /
+ // AWS_EXTERNAL_ID pass through unconditionally when set (the SPI
already emits them).
+ aligned.put("AWS_CREDENTIALS_PROVIDER_TYPE",
s3CredentialsMode.getMode());
+ } else {
Review Comment:
[P1] Do not rewrite unlisted S3 roles as anonymous
Every unlisted `S3CompatibleFileSystemProperties` binding is assigned
`StorageTypeId.S3`, but any provider key other than the literal built-in `S3`
enters this branch. For a typed third-party provider with a role ARN/external
ID and no static keys, this removes both role fields and sends
`AWS_CREDENTIALS_PROVIDER_TYPE=ANONYMOUS`; `getAwsCredentialsProvider()`
likewise returns null for that role-bearing adapter. Thus even its standard
`s3://` path fails in scan/sink and FE connectivity flows. Please apply the
legacy non-S3 policy only to the known built-in dialects, and retain generic S3
AssumeRole/default-chain behavior (or make credential policy explicit in the
SPI), with a role-based unlisted-provider test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
