github-actions[bot] commented on code in PR #67545:
URL: https://github.com/apache/doris/pull/67545#discussion_r3942659423
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogFactory.java:
##########
@@ -174,6 +178,12 @@ private static boolean isStoragePrefixed(String key) {
return false;
}
+ private static void appendDlfOptions(Options options) {
+ options.set("metastore.client.class",
"com.aliyun.datalake.metastore.hive2.ProxyMetaStoreClient");
+ // Paimon's client pool is JVM-static, so the DLF catalog id must
remain part of its cache identity.
+ options.set("client-pool-cache.keys", "conf:dlf.catalog.id");
Review Comment:
[P1] Include the full DLF connection identity in the pool key
Paimon 1.3.1 keeps this pool in a JVM-static cache whose fixed key is the
client class, HMS URI, and `hive`; DLF has an empty HMS URI, so this option
distinguishes only `dlf.catalog.id`. Two catalogs—or an ALTER generation—with
the same/empty id but different endpoint, credentials, token, UID, or proxy
mode reuse the first catalog's HiveConf and send requests with stale or another
catalog's credentials. Please key all client-binding fields (or a
configuration/generation fingerprint) and test both same-id catalogs and ALTER.
##########
fe/fe-connector/fe-connector-metastore-spi/src/main/java/org/apache/doris/connector/metastore/spi/AbstractDlfMetaStoreProperties.java:
##########
@@ -0,0 +1,123 @@
+// 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.connector.metastore.spi;
+
+import org.apache.doris.connector.metastore.DlfMetaStoreProperties;
+import org.apache.doris.foundation.property.ConnectorProperty;
+
+import org.apache.commons.lang3.BooleanUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** Shared Aliyun DLF property binding and neutral catalog configuration. */
+public abstract class AbstractDlfMetaStoreProperties extends
AbstractMetaStoreProperties
+ implements DlfMetaStoreProperties {
+
+ @ConnectorProperty(names = {"dlf.access_key", "dlf.catalog.accessKeyId"},
required = false, sensitive = true,
+ description = "DLF access key id.")
+ private String accessKey = "";
+
+ @ConnectorProperty(names = {"dlf.secret_key",
"dlf.catalog.accessKeySecret"}, required = false, sensitive = true,
Review Comment:
[P1] Keep DLF credential aliases consistent with storage binding
This holder accepts `dlf.catalog.accessKeySecret` plus
`dlf.session_token`/`dlf.catalog.sessionToken`, but both OSS binders recognize
`dlf.catalog.secret_key` instead and neither recognizes the DLF token aliases.
A catalog using the camel-case secret passes DLF binding but fails the storage
AK/SK pair check; a catalog using temporary credentials silently loses its
token before Iceberg FileIO and BE scan credentials are built, so OSS requests
fail. Please normalize these aliases into the storage binding (for both OSS and
OSS-HDFS) and cover all three STS values end to end.
##########
fe/fe-connector/fe-connector-metastore-spi/src/main/java/org/apache/doris/connector/metastore/spi/AbstractDlfMetaStoreProperties.java:
##########
@@ -0,0 +1,123 @@
+// 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.connector.metastore.spi;
+
+import org.apache.doris.connector.metastore.DlfMetaStoreProperties;
+import org.apache.doris.foundation.property.ConnectorProperty;
+
+import org.apache.commons.lang3.BooleanUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** Shared Aliyun DLF property binding and neutral catalog configuration. */
+public abstract class AbstractDlfMetaStoreProperties extends
AbstractMetaStoreProperties
+ implements DlfMetaStoreProperties {
+
+ @ConnectorProperty(names = {"dlf.access_key", "dlf.catalog.accessKeyId"},
required = false, sensitive = true,
+ description = "DLF access key id.")
+ private String accessKey = "";
+
+ @ConnectorProperty(names = {"dlf.secret_key",
"dlf.catalog.accessKeySecret"}, required = false, sensitive = true,
+ description = "DLF access key secret.")
+ private String secretKey = "";
+
+ @ConnectorProperty(names = {"dlf.session_token",
"dlf.catalog.sessionToken"}, required = false, sensitive = true,
+ description = "DLF session/security token.")
+ private String sessionToken = "";
+
+ @ConnectorProperty(names = {"dlf.region"}, required = false,
+ description = "DLF region used to derive the endpoint when it is
not set.")
+ private String region = "";
+
+ @ConnectorProperty(names = {"dlf.endpoint", "dlf.catalog.endpoint"},
required = false,
+ description = "DLF endpoint.")
+ private String endpoint = "";
+
+ @ConnectorProperty(names = {"dlf.catalog.uid", "dlf.uid"}, required =
false,
+ description = "DLF account uid.")
+ private String uid = "";
+
+ @ConnectorProperty(names = {"dlf.catalog.id", "dlf.catalog_id"}, required
= false,
+ description = "DLF catalog id, defaulting to the uid.")
+ private String catalogId = "";
+
+ @ConnectorProperty(names = {"dlf.access.public",
"dlf.catalog.accessPublic"}, required = false,
+ description = "Whether to use the public DLF endpoint instead of
the VPC endpoint.")
+ private String accessPublic = "false";
+
+ @ConnectorProperty(names = {"dlf.catalog.proxyMode", "dlf.proxy.mode"},
required = false,
+ description = "DLF proxy mode.")
+ private String proxyMode = "DLF_ONLY";
+
+ private final Map<String, String> storageHadoopConfig;
+
+ protected AbstractDlfMetaStoreProperties(Map<String, String> raw,
Map<String, String> storageHadoopConfig) {
+ super(raw);
+ this.storageHadoopConfig = storageHadoopConfig;
+ }
+
+ @Override
+ public String providerName() {
+ return "DLF";
+ }
+
+ @Override
+ public boolean needsStorage() {
+ return true;
+ }
+
+ protected void validateConnection() {
+ if (StringUtils.isBlank(accessKey)) {
+ throw new IllegalArgumentException("dlf.access_key is required");
+ }
+ if (StringUtils.isBlank(secretKey)) {
+ throw new IllegalArgumentException("dlf.secret_key is required");
+ }
+ if (StringUtils.isBlank(endpoint) && StringUtils.isBlank(region)) {
Review Comment:
[P1] Make endpoint-only DLF configuration valid for OSS storage
This accepts `dlf.endpoint` without `dlf.region`, but the same raw endpoint
is also consumed by native `OssFileSystemProperties`. Its region extractor
recognizes only `oss-*`/`s3.oss-*` hosts, so `dlf.cn-hangzhou.aliyuncs.com`
leaves the storage region empty and catalog binding fails after DLF validation
succeeds. Either derive the native OSS region from DLF public/VPC endpoints (as
OSS-HDFS already does), or require `dlf.region` for this form; add
endpoint-only Iceberg and Paimon binding tests.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java:
##########
@@ -0,0 +1,166 @@
+// 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.connector.iceberg.dlf;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.iceberg.BaseMetastoreCatalog;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.FileIO;
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Base catalog for Hive-compatible metastores that need a custom client
pool. */
+public abstract class HiveCompatibleCatalog extends BaseMetastoreCatalog
implements SupportsNamespaces, Configurable {
+
+ protected Configuration conf;
+ protected ClientPool<IMetaStoreClient, TException> clients;
+ protected FileIO fileIO;
+ protected String catalogName;
+
+ public void initialize(String name, FileIO fileIO,
ClientPool<IMetaStoreClient, TException> clients) {
+ this.catalogName = name;
+ this.fileIO = fileIO;
+ this.clients = clients;
+ }
+
+ protected FileIO initializeFileIO(Map<String, String> properties,
Configuration hadoopConf) {
+ String fileIOImpl = properties.get(CatalogProperties.FILE_IO_IMPL);
+ if (fileIOImpl == null) {
+ FileIO io = new HadoopFileIO(hadoopConf);
+ io.initialize(properties);
+ return io;
+ }
+ return CatalogUtil.loadFileIO(fileIOImpl, properties, hadoopConf);
+ }
+
+ @Override
+ protected String defaultWarehouseLocation(TableIdentifier tableIdentifier)
{
+ return null;
+ }
+
+ @Override
+ protected boolean isValidIdentifier(TableIdentifier tableIdentifier) {
+ return tableIdentifier.namespace().levels().length == 1;
+ }
+
+ protected boolean isValidNamespace(Namespace namespace) {
+ return namespace.levels().length != 1;
+ }
+
+ @Override
+ public List<TableIdentifier> listTables(Namespace namespace) {
+ if (isValidNamespace(namespace)) {
+ throw new NoSuchTableException("Invalid namespace: %s", namespace);
+ }
+ String dbName = namespace.level(0);
+ try {
+ return clients.run(client -> client.getAllTables(dbName)).stream()
Review Comment:
[P1] Filter non-Iceberg tables from DLF listings
DLF databases can contain Hive/Paimon tables alongside Iceberg tables, but
this returns every metastore name and the connector publishes them all through
`SHOW TABLES`. Selecting one of those non-Iceberg entries then fails in
`HiveTableOperations` because it has no Iceberg table marker/metadata. Match
Iceberg `HiveCatalog`'s default filtering semantics (and its optional
`list-all-tables` behavior if supported), with a mixed-format DLF test.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java:
##########
@@ -0,0 +1,166 @@
+// 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.connector.iceberg.dlf;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.iceberg.BaseMetastoreCatalog;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.FileIO;
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Base catalog for Hive-compatible metastores that need a custom client
pool. */
+public abstract class HiveCompatibleCatalog extends BaseMetastoreCatalog
implements SupportsNamespaces, Configurable {
+
+ protected Configuration conf;
+ protected ClientPool<IMetaStoreClient, TException> clients;
+ protected FileIO fileIO;
+ protected String catalogName;
+
+ public void initialize(String name, FileIO fileIO,
ClientPool<IMetaStoreClient, TException> clients) {
+ this.catalogName = name;
+ this.fileIO = fileIO;
+ this.clients = clients;
+ }
+
+ protected FileIO initializeFileIO(Map<String, String> properties,
Configuration hadoopConf) {
+ String fileIOImpl = properties.get(CatalogProperties.FILE_IO_IMPL);
+ if (fileIOImpl == null) {
+ FileIO io = new HadoopFileIO(hadoopConf);
+ io.initialize(properties);
+ return io;
+ }
+ return CatalogUtil.loadFileIO(fileIOImpl, properties, hadoopConf);
+ }
+
+ @Override
+ protected String defaultWarehouseLocation(TableIdentifier tableIdentifier)
{
+ return null;
+ }
+
+ @Override
+ protected boolean isValidIdentifier(TableIdentifier tableIdentifier) {
+ return tableIdentifier.namespace().levels().length == 1;
+ }
+
+ protected boolean isValidNamespace(Namespace namespace) {
+ return namespace.levels().length != 1;
+ }
+
+ @Override
+ public List<TableIdentifier> listTables(Namespace namespace) {
+ if (isValidNamespace(namespace)) {
+ throw new NoSuchTableException("Invalid namespace: %s", namespace);
+ }
+ String dbName = namespace.level(0);
+ try {
+ return clients.run(client -> client.getAllTables(dbName)).stream()
+ .map(table -> TableIdentifier.of(dbName, table))
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public boolean dropTable(TableIdentifier tableIdentifier, boolean purge) {
+ throw new UnsupportedOperationException("Cannot drop table " +
tableIdentifier + ": not supported");
+ }
+
+ @Override
+ public void renameTable(TableIdentifier source, TableIdentifier target) {
+ throw new UnsupportedOperationException("Cannot rename table " +
source + ": not supported");
+ }
+
+ @Override
+ public void createNamespace(Namespace namespace, Map<String, String>
properties) {
+ throw new UnsupportedOperationException("Cannot create namespace " +
namespace + ": not supported");
+ }
+
+ @Override
+ public List<Namespace> listNamespaces(Namespace namespace) throws
NoSuchNamespaceException {
+ if (isValidNamespace(namespace) && !namespace.isEmpty()) {
+ throw new NoSuchNamespaceException("Namespace does not exist: %s",
namespace);
+ }
+ if (!namespace.isEmpty()) {
+ return new ArrayList<>();
+ }
+ try {
+ return clients.run(IMetaStoreClient::getAllDatabases).stream()
+ .map(Namespace::of)
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Map<String, String> loadNamespaceMetadata(Namespace namespace)
throws NoSuchNamespaceException {
+ if (isValidNamespace(namespace)) {
+ throw new NoSuchTableException("Invalid namespace: %s", namespace);
+ }
+ try {
+ return clients.run(client ->
client.getDatabase(namespace.level(0))).getParameters();
+ } catch (Exception e) {
Review Comment:
[P2] Translate missing DLF databases to `NoSuchNamespaceException`
`getDatabase` reports an absent database with Hive's
`NoSuchObjectException`/`UnknownDBException`, but this catch-all wraps it as
`RuntimeException`. The inherited `SupportsNamespaces.namespaceExists()` only
converts `NoSuchNamespaceException` to `false`, so a normal existence check
throws instead. Preserve HiveCatalog's exception mapping (and restore the
interrupt flag separately) and cover an absent namespace.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java:
##########
@@ -0,0 +1,166 @@
+// 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.connector.iceberg.dlf;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.iceberg.BaseMetastoreCatalog;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.FileIO;
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Base catalog for Hive-compatible metastores that need a custom client
pool. */
+public abstract class HiveCompatibleCatalog extends BaseMetastoreCatalog
implements SupportsNamespaces, Configurable {
+
+ protected Configuration conf;
+ protected ClientPool<IMetaStoreClient, TException> clients;
+ protected FileIO fileIO;
+ protected String catalogName;
+
+ public void initialize(String name, FileIO fileIO,
ClientPool<IMetaStoreClient, TException> clients) {
Review Comment:
[P1] Close the DLF-owned FileIO and client pool
These resources are catalog-owned, but this class inherits
`BaseMetastoreCatalog.close()`, which closes only the metrics reporter.
Consequently DROP/ALTER/connector replacement leaves the S3FileIO HTTP client
and the live `DLFClientPool` open; the Caffeine removal listener is not a
catalog-close path and may never run after the catalog becomes unreachable.
Please add an idempotent close that invalidates/closes the DLF pool, closes
FileIO, and invokes `super.close()`, including partial-initialization cleanup.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java:
##########
@@ -0,0 +1,166 @@
+// 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.connector.iceberg.dlf;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.iceberg.BaseMetastoreCatalog;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.FileIO;
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Base catalog for Hive-compatible metastores that need a custom client
pool. */
+public abstract class HiveCompatibleCatalog extends BaseMetastoreCatalog
implements SupportsNamespaces, Configurable {
+
+ protected Configuration conf;
+ protected ClientPool<IMetaStoreClient, TException> clients;
+ protected FileIO fileIO;
+ protected String catalogName;
+
+ public void initialize(String name, FileIO fileIO,
ClientPool<IMetaStoreClient, TException> clients) {
+ this.catalogName = name;
+ this.fileIO = fileIO;
+ this.clients = clients;
+ }
+
+ protected FileIO initializeFileIO(Map<String, String> properties,
Configuration hadoopConf) {
+ String fileIOImpl = properties.get(CatalogProperties.FILE_IO_IMPL);
+ if (fileIOImpl == null) {
+ FileIO io = new HadoopFileIO(hadoopConf);
+ io.initialize(properties);
+ return io;
+ }
+ return CatalogUtil.loadFileIO(fileIOImpl, properties, hadoopConf);
+ }
+
+ @Override
+ protected String defaultWarehouseLocation(TableIdentifier tableIdentifier)
{
Review Comment:
[P1] Preserve DLF's unsupported CREATE TABLE contract
The plugin-driven metadata now routes CREATE TABLE into inherited
`BaseMetastoreCatalog.createTable()`, while the removed DLF catalog explicitly
rejected it. With no explicit LOCATION this method supplies null, and Iceberg
1.10.1 formats the first metadata write as `null/metadata/...`, which S3FileIO
rejects with an unrelated URI/I/O failure. Either reject DLF table creation
explicitly like the other DLF DDL methods, or implement a valid
default-location policy before exposing it; please add a SQL-path test.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/DLFCatalog.java:
##########
@@ -0,0 +1,126 @@
+// 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.connector.iceberg.dlf;
+
+import org.apache.doris.connector.iceberg.dlf.client.DLFCachedClientPool;
+import org.apache.doris.filesystem.properties.S3CompatibleFileSystemProperties;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.aws.s3.S3FileIO;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+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.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.auth.signer.AwsS3V4Signer;
+import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
+import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption;
+import software.amazon.awssdk.core.retry.RetryPolicy;
+import software.amazon.awssdk.core.retry.backoff.EqualJitterBackoffStrategy;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.Map;
+
+/** Aliyun DLF Iceberg catalog backed by the DLF Hive-compatible metastore and
OSS. */
+public class DLFCatalog extends HiveCompatibleCatalog {
+
+ private final S3CompatibleFileSystemProperties ossStorage;
+
+ public DLFCatalog(S3CompatibleFileSystemProperties ossStorage) {
+ this.ossStorage = ossStorage;
+ }
+
+ @Override
+ public void initialize(String name, Map<String, String> properties) {
+ super.initialize(name, initializeFileIO(properties, conf), new
DLFCachedClientPool(conf, properties));
+ }
+
+ @Override
+ protected TableOperations newTableOps(TableIdentifier tableIdentifier) {
+ return new DLFTableOperations(conf, clients, fileIO, catalogName,
+ tableIdentifier.namespace().level(0), tableIdentifier.name());
+ }
+
+ @Override
+ protected FileIO initializeFileIO(Map<String, String> properties,
Configuration hadoopConf) {
+ String region = ossStorage.getRegion();
+ boolean usePathStyle =
Boolean.parseBoolean(ossStorage.getUsePathStyle());
+ URI endpoint =
URI.create(toS3CompatibleEndpoint(ossStorage.getEndpoint(), region));
+ AwsCredentialsProvider credentials = buildCredentials(ossStorage);
+ FileIO io = new S3FileIO(() -> buildOssS3Client(endpoint, region,
credentials, usePathStyle));
+ io.initialize(properties);
+ return io;
+ }
+
+ static String toS3CompatibleEndpoint(String endpoint, String region) {
+ String s3Endpoint = endpoint.replace("oss-" + region, "s3.oss-" +
region);
Review Comment:
[P1] Keep S3-compatible OSS endpoint conversion idempotent
`OssFileSystemProperties` accepts an already compatible endpoint such as
`https://s3.oss-cn-hangzhou.aliyuncs.com`, but this substring replacement turns
it into `https://s3.s3.oss-cn-hangzhou.aliyuncs.com`. That valid configuration
then fails DNS/connection setup. Parse the host or guard the existing `s3.oss-`
prefix before adding `s3.`, and test public/internal pre-prefixed endpoints.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java:
##########
@@ -967,6 +969,10 @@ private Catalog createCatalog() {
return createS3TablesCatalog(catalogName, chosenS3);
}
+ if (IcebergCatalogProperties.TYPE_DLF.equals(flavor)) {
Review Comment:
[P1] Probe DLF when `test_connection` is enabled
This new remote flavor is not included in `probesMetastore()`, and
`probeStorage()` separately recognizes only raw `s3.access_key`/`s3.endpoint`,
not the normal `dlf.*` aliases. Thus CREATE CATALOG with `test_connection=true`
can accept a bad DLF endpoint and credentials without performing either
advertised probe. Add DLF to the metastore probe and drive storage probing from
the already-bound OSS properties, with a failing DLF connectivity test.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java:
##########
@@ -0,0 +1,166 @@
+// 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.connector.iceberg.dlf;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.iceberg.BaseMetastoreCatalog;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.FileIO;
+import shade.doris.hive.org.apache.thrift.TException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Base catalog for Hive-compatible metastores that need a custom client
pool. */
+public abstract class HiveCompatibleCatalog extends BaseMetastoreCatalog
implements SupportsNamespaces, Configurable {
+
+ protected Configuration conf;
+ protected ClientPool<IMetaStoreClient, TException> clients;
+ protected FileIO fileIO;
+ protected String catalogName;
+
+ public void initialize(String name, FileIO fileIO,
ClientPool<IMetaStoreClient, TException> clients) {
+ this.catalogName = name;
+ this.fileIO = fileIO;
+ this.clients = clients;
+ }
+
+ protected FileIO initializeFileIO(Map<String, String> properties,
Configuration hadoopConf) {
+ String fileIOImpl = properties.get(CatalogProperties.FILE_IO_IMPL);
+ if (fileIOImpl == null) {
+ FileIO io = new HadoopFileIO(hadoopConf);
+ io.initialize(properties);
+ return io;
+ }
+ return CatalogUtil.loadFileIO(fileIOImpl, properties, hadoopConf);
+ }
+
+ @Override
+ protected String defaultWarehouseLocation(TableIdentifier tableIdentifier)
{
+ return null;
+ }
+
+ @Override
+ protected boolean isValidIdentifier(TableIdentifier tableIdentifier) {
+ return tableIdentifier.namespace().levels().length == 1;
+ }
+
+ protected boolean isValidNamespace(Namespace namespace) {
+ return namespace.levels().length != 1;
+ }
+
+ @Override
+ public List<TableIdentifier> listTables(Namespace namespace) {
+ if (isValidNamespace(namespace)) {
+ throw new NoSuchTableException("Invalid namespace: %s", namespace);
+ }
+ String dbName = namespace.level(0);
+ try {
+ return clients.run(client -> client.getAllTables(dbName)).stream()
+ .map(table -> TableIdentifier.of(dbName, table))
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public boolean dropTable(TableIdentifier tableIdentifier, boolean purge) {
+ throw new UnsupportedOperationException("Cannot drop table " +
tableIdentifier + ": not supported");
+ }
+
+ @Override
+ public void renameTable(TableIdentifier source, TableIdentifier target) {
+ throw new UnsupportedOperationException("Cannot rename table " +
source + ": not supported");
+ }
+
+ @Override
+ public void createNamespace(Namespace namespace, Map<String, String>
properties) {
+ throw new UnsupportedOperationException("Cannot create namespace " +
namespace + ": not supported");
+ }
+
+ @Override
+ public List<Namespace> listNamespaces(Namespace namespace) throws
NoSuchNamespaceException {
+ if (isValidNamespace(namespace) && !namespace.isEmpty()) {
+ throw new NoSuchNamespaceException("Namespace does not exist: %s",
namespace);
+ }
+ if (!namespace.isEmpty()) {
+ return new ArrayList<>();
+ }
+ try {
+ return clients.run(IMetaStoreClient::getAllDatabases).stream()
+ .map(Namespace::of)
+ .collect(Collectors.toList());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Map<String, String> loadNamespaceMetadata(Namespace namespace)
throws NoSuchNamespaceException {
+ if (isValidNamespace(namespace)) {
+ throw new NoSuchTableException("Invalid namespace: %s", namespace);
+ }
+ try {
+ return clients.run(client ->
client.getDatabase(namespace.level(0))).getParameters();
Review Comment:
[P2] Preserve the DLF database location in namespace metadata
Returning only `Database.getParameters()` drops the actual `locationUri`.
Downstream `loadNamespaceLocation()` looks for the reserved `location` key, so
`SHOW CREATE DATABASE` omits LOCATION for every normal DLF database unless a
user redundantly stored it as a parameter. Please convert the Database like
Iceberg `HiveCatalog` does (including `location`) and add a location-bearing
namespace 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]
