github-actions[bot] commented on code in PR #67545: URL: https://github.com/apache/doris/pull/67545#discussion_r3943032124
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java: ########## @@ -0,0 +1,274 @@ +// 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.hadoop.hive.metastore.api.Database; +import org.apache.hadoop.hive.metastore.api.NoSuchObjectException; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.hive.metastore.api.UnknownDBException; +import org.apache.iceberg.BaseMetastoreCatalog; +import org.apache.iceberg.BaseMetastoreTableOperations; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.ClientPool; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.Catalog.TableBuilder; +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.hadoop.HadoopFileIO; +import org.apache.iceberg.hive.HiveCatalog; +import org.apache.iceberg.io.FileIO; +import shade.doris.hive.org.apache.thrift.TException; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +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; + private boolean listAllTables; + private boolean closed; + + public void initialize(String name, FileIO fileIO, ClientPool<IMetaStoreClient, TException> clients) { + initialize(name, fileIO, clients, Map.of()); + } + + public void initialize(String name, FileIO fileIO, ClientPool<IMetaStoreClient, TException> clients, + Map<String, String> properties) { + this.catalogName = name; + this.fileIO = fileIO; + this.clients = clients; + this.listAllTables = Boolean.parseBoolean(properties.getOrDefault( + HiveCatalog.LIST_ALL_TABLES, HiveCatalog.LIST_ALL_TABLES_DEFAULT)); + } + + 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 NoSuchNamespaceException("Namespace does not exist: %s", namespace); + } + String dbName = namespace.level(0); + try { + List<String> tableNames = clients.run(client -> client.getAllTables(dbName)); + if (listAllTables) { + return tableNames.stream() + .map(table -> TableIdentifier.of(dbName, table)) + .collect(Collectors.toList()); + } + // DLF namespaces are format-shared; publishing non-Iceberg names creates unusable Doris tables. + List<Table> tables = clients.run(client -> client.getTableObjectsByName(dbName, tableNames)); Review Comment: [P2] Bound the full-table metadata fetch used by SHOW TABLES The exact DLF 0.2.14 client forwards this entire list as one `BatchGetTablesRequest`; it does not paginate or chunk it. Thus the default filtering fix makes `SHOW TABLES` on a large shared namespace send and materialize every table's full schema/storage descriptor in one request, making its network and peak-memory cost scale with all table metadata rather than the name list. Please fetch and filter in bounded chunks (or use a paginated full-object API) and add a multi-chunk test. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalCatalog.java: ########## @@ -522,6 +522,14 @@ public void overlayMetaCacheConfig(Map<String, String> metaCacheProperties) { */ @Override public boolean createTable(CreateTableInfo createTableInfo) throws UserException { + Map<String, String> properties = getProperties(); + try { + // Unsupported configuration-specific DDL must fail before initialization can touch a remote service. + ConnectorFactory.findProvider(getType(), properties) + .ifPresent(provider -> provider.validateCreateTable(properties)); Review Comment: [P2] Preserve both preflight and IF NOT EXISTS ordering This unconditional hook has incompatible placement on the two CTAS branches. Non-`IF NOT EXISTS` CTAS performs `targetTableExists()` first, so it can initialize and read DLF before the intended configuration-only rejection. But once this hook is reached, it rejects `IF NOT EXISTS` before the catalog can return the required existing-target no-op. Make the preflight branch-aware so a missing-target DLF creation is rejected before remote lookup while an existing `IF NOT EXISTS` target still short-circuits, and add both CTAS cases plus the plain existing-target case. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java: ########## @@ -0,0 +1,274 @@ +// 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.hadoop.hive.metastore.api.Database; +import org.apache.hadoop.hive.metastore.api.NoSuchObjectException; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.hive.metastore.api.UnknownDBException; +import org.apache.iceberg.BaseMetastoreCatalog; +import org.apache.iceberg.BaseMetastoreTableOperations; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.ClientPool; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.Catalog.TableBuilder; +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.hadoop.HadoopFileIO; +import org.apache.iceberg.hive.HiveCatalog; +import org.apache.iceberg.io.FileIO; +import shade.doris.hive.org.apache.thrift.TException; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +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; + private boolean listAllTables; + private boolean closed; + + public void initialize(String name, FileIO fileIO, ClientPool<IMetaStoreClient, TException> clients) { + initialize(name, fileIO, clients, Map.of()); + } + + public void initialize(String name, FileIO fileIO, ClientPool<IMetaStoreClient, TException> clients, Review Comment: [P2] Retain catalog properties used by the Iceberg base class This initializer receives the full copy-all Iceberg option map but keeps only `list-all-tables`. Because the class then inherits `BaseMetastoreCatalog.properties()`'s empty map, options consumed by the base class are silently lost; in particular `metrics-reporter-impl` can never select the configured reporter for DLF tables. Store an immutable copy and override `properties()` as Iceberg's `HiveCatalog` does, with a reporter initialization/close test. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/plugin/PluginDrivenExternalCatalog.java: ########## @@ -522,6 +522,14 @@ public void overlayMetaCacheConfig(Map<String, String> metaCacheProperties) { */ @Override public boolean createTable(CreateTableInfo createTableInfo) throws UserException { + Map<String, String> properties = getProperties(); + try { + // Unsupported configuration-specific DDL must fail before initialization can touch a remote service. + ConnectorFactory.findProvider(getType(), properties) Review Comment: [P2] Invoke the provider preflight under its plugin classloader This invokes the new provider callback under the FE caller's TCCL. Directory providers are child-loaded, and the neighboring ALTER validation path explicitly pins each provider's defining loader around both `supports` and validation because callbacks can resolve plugin-local helpers through TCCL. A compatible API-v7 provider can therefore fail this CREATE preflight with a helper/linkage error before its connector exists. Route this through a plugin-manager entry point that pins and restores the provider loader, and extend the temporary directory-plugin fixture to exercise the new callback. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java: ########## @@ -465,11 +466,32 @@ private Catalog createCatalog() { hmsAuth, storageHadoopConfig, "Failed to create Paimon catalog with HMS metastore"); } + case PaimonCatalogProperties.DLF: { + // Legacy DLF catalogs often expose OSS only through dlf.* aliases and an oss:// warehouse. + // Check the resolved storage bindings here so those catalogs remain valid while non-OSS + // backends cannot be passed to Paimon's DLF Hive catalog. + if (!hasDlfCompatibleStorage(storage().getStorageProperties())) { + throw new IllegalStateException("Paimon DLF metastore requires OSS storage properties."); + } + DlfMetaStoreProperties dlf = (DlfMetaStoreProperties) + MetaStoreProviders.bind(catalogProps.getRaw(), storageHadoopConfig); + Map<String, String> dlfConf = new HashMap<>(dlf.toDlfCatalogConf()); + dlfConf.put(PaimonCatalogFactory.DLF_CLIENT_POOL_IDENTITY, + PaimonCatalogFactory.dlfClientPoolIdentity(dlfConf)); + HiveConf hc = PaimonCatalogFactory.assembleHiveConf(null, dlfConf); + return createCatalogFromContext(CatalogContext.create(options, hc), flavor, + "Failed to create Paimon catalog with DLF metastore"); + } default: throw new IllegalArgumentException("Unknown paimon.catalog.type value: " + flavor); } } + static boolean hasDlfCompatibleStorage(List<StorageProperties> storageProperties) { + return storageProperties.stream().anyMatch(storage -> "OSS".equals(storage.providerName()) Review Comment: [P2] Do not accept a misconverted mixed-case DLF OSS-HDFS binding This newly restores OSS-HDFS for Paimon DLF, but that binder matches `https://DLF-VPC.cn-beijing.aliyuncs.com` after lowercasing it and then uses case-sensitive `endpoint.contains("dlf")` for conversion. The binding is therefore accepted here while `fs.oss.endpoint` still points at the DLF metastore host, so data I/O fails. Make the DLF-to-OSS-DLS conversion use the same case-insensitive parsed match and cover a mixed-case hostname. ########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnector.java: ########## @@ -465,11 +466,32 @@ private Catalog createCatalog() { hmsAuth, storageHadoopConfig, "Failed to create Paimon catalog with HMS metastore"); } + case PaimonCatalogProperties.DLF: { Review Comment: [P1] Make test_connection exercise the restored Paimon DLF path `PaimonConnector` still inherits the SPI's unconditional-success `testConnection`, while this DLF catalog is created only lazily through `ensureCatalog`. Consequently an explicit `test_connection=true` accepts an unreachable DLF endpoint and invalid OSS connection without entering this branch, then fails on first metadata access. Please add a TCCL/authenticated DLF metadata probe plus the required storage leg, and an unreachable Paimon DLF regression; the existing negative case covers only Iceberg. ########## fe/fe-connector/fe-connector-metastore-spi/src/main/java/org/apache/doris/connector/metastore/spi/AbstractDlfMetaStoreProperties.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.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 { + + // These aliases must stay aligned with both OSS binders so one credential set reaches metadata and storage. + @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.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", "dlf.catalog.securityToken"}, Review Comment: [P1] Add the new token alias to provider-independent masking `dlf.catalog.securityToken` is accepted here as a credential, but fe-core's unconditional `DatasourcePrintableMap` inventory still contains only the four older DLF secret/token spellings. Normal startup happens to mask it through the OSS provider's dynamic registration; if that provider is missing or rejected, a persisted catalog remains listable and `SHOW CREATE CATALOG` prints this token in clear text. Please add this alias to the static DLF sensitive-key set and its direct masking test. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/HiveCompatibleCatalog.java: ########## @@ -0,0 +1,274 @@ +// 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.hadoop.hive.metastore.api.Database; +import org.apache.hadoop.hive.metastore.api.NoSuchObjectException; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.hive.metastore.api.UnknownDBException; +import org.apache.iceberg.BaseMetastoreCatalog; +import org.apache.iceberg.BaseMetastoreTableOperations; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.ClientPool; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.Catalog.TableBuilder; +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.hadoop.HadoopFileIO; +import org.apache.iceberg.hive.HiveCatalog; +import org.apache.iceberg.io.FileIO; +import shade.doris.hive.org.apache.thrift.TException; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +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; + private boolean listAllTables; + private boolean closed; + + public void initialize(String name, FileIO fileIO, ClientPool<IMetaStoreClient, TException> clients) { + initialize(name, fileIO, clients, Map.of()); + } + + public void initialize(String name, FileIO fileIO, ClientPool<IMetaStoreClient, TException> clients, + Map<String, String> properties) { + this.catalogName = name; + this.fileIO = fileIO; + this.clients = clients; + this.listAllTables = Boolean.parseBoolean(properties.getOrDefault( + HiveCatalog.LIST_ALL_TABLES, HiveCatalog.LIST_ALL_TABLES_DEFAULT)); + } + + 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 NoSuchNamespaceException("Namespace does not exist: %s", namespace); + } + String dbName = namespace.level(0); + try { + List<String> tableNames = clients.run(client -> client.getAllTables(dbName)); + if (listAllTables) { + return tableNames.stream() + .map(table -> TableIdentifier.of(dbName, table)) + .collect(Collectors.toList()); + } + // DLF namespaces are format-shared; publishing non-Iceberg names creates unusable Doris tables. + List<Table> tables = clients.run(client -> client.getTableObjectsByName(dbName, tableNames)); + return tables.stream() + .filter(table -> table.getParameters() != null + && BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE.equalsIgnoreCase( + table.getParameters().get(BaseMetastoreTableOperations.TABLE_TYPE_PROP))) + .map(Table::getTableName) + .map(table -> TableIdentifier.of(dbName, table)) + .collect(Collectors.toList()); + } catch (UnknownDBException e) { + throw new NoSuchNamespaceException(e, "Namespace does not exist: %s", namespace); + } catch (TException e) { + throw new RuntimeException("Failed to list tables under namespace " + namespace, e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted in call to listTables", e); + } + } + + @Override + public TableBuilder buildTable(TableIdentifier identifier, Schema schema) { + // DLF metadata writes were never supported; reject before BaseMetastoreCatalog builds a null location. + throw new UnsupportedOperationException("Cannot create table " + identifier + ": not supported"); + } + + @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()) { Review Comment: [P2] Reject the unsupported external catalog root for DLF `external_catalog.name` is honored for every Iceberg flavor, so its value reaches this method as the listing root. This returns an empty leaf without consulting DLF, while subsequent database/table resolution appends the same value and produces a two-level namespace that this adapter rejects. A DLF catalog with this accepted property therefore lists no databases and cannot resolve direct table names. Either reject the property for DLF during configuration validation or implement one consistent namespace mapping, and cover both listing and table lookup. ########## fe/fe-connector/fe-connector-metastore-spi/src/main/java/org/apache/doris/connector/metastore/spi/AbstractDlfMetaStoreProperties.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.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 { + + // These aliases must stay aligned with both OSS binders so one credential set reaches metadata and storage. + @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.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", "dlf.catalog.securityToken"}, + 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)) { + throw new IllegalArgumentException("dlf.endpoint is required."); + } + } + + @Override + public Map<String, String> toDlfCatalogConf() { + String resolvedEndpoint = endpoint; + if (StringUtils.isBlank(resolvedEndpoint) && StringUtils.isNotBlank(region)) { + resolvedEndpoint = BooleanUtils.toBoolean(accessPublic) Review Comment: [P2] Parse dlf.access.public consistently for both connection legs This uses `BooleanUtils.toBoolean`, so values such as `yes` or `on` select the public DLF endpoint. The same raw property is rebound by native OSS with `Boolean.parseBoolean`, which treats those values as false and derives an internal data endpoint instead. A catalog can therefore authenticate metadata publicly but send data traffic to the VPC host. Use one validated parser/representation for both binders and test the non-literal truthy spellings end to end. -- 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]
