[feature](iceberg) support s3 tables

be/src/util/s3_util.cpp

@@ -103,6 +103,7 @@ constexpr char S3_TOKEN[] = "AWS_TOKEN";
 constexpr char S3_MAX_CONN_SIZE[] = "AWS_MAX_CONN_SIZE";
 constexpr char S3_REQUEST_TIMEOUT_MS[] = "AWS_REQUEST_TIMEOUT_MS";
 constexpr char S3_CONN_TIMEOUT_MS[] = "AWS_CONNECTION_TIMEOUT_MS";
+constexpr char S3_NEED_OVERRIDE_ENDPOINT[] = "AWS_NEED_OVERRIDE_ENDPOINT";
 
 auto metric_func_factory(bvar::Adder<int64_t>& ns_bvar, bvar::Adder<int64_t>& req_num_bvar) {
     return [&](int64_t ns) {
@@ -272,7 +273,9 @@ std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(
             "s3_client_factory::create",
             std::make_shared<io::S3ObjStorageClient>(std::make_shared<Aws::S3::S3Client>()));
     Aws::Client::ClientConfiguration aws_config = S3ClientFactory::getClientConfiguration();
-    aws_config.endpointOverride = s3_conf.endpoint;
+    if (s3_conf.need_override_endpoint) {
+        aws_config.endpointOverride = s3_conf.endpoint;
+    }
     aws_config.region = s3_conf.region;
     std::string ca_cert = get_valid_ca_cert_path();
     if ("" != _ca_cert_file_path) {
@@ -350,6 +353,9 @@ Status S3ClientFactory::convert_properties_to_s3_conf(
     if (auto it = properties.find(S3_ENDPOINT); it != properties.end()) {
         s3_conf->client_conf.endpoint = it->second;
     }
+    if (auto it = properties.find(S3_NEED_OVERRIDE_ENDPOINT); it != properties.end()) {
+        s3_conf->client_conf.need_override_endpoint = (it->second == "true");
+    }
     if (auto it = properties.find(S3_REGION); it != properties.end()) {
         s3_conf->client_conf.region = it->second;
     }

be/src/util/s3_util.h

@@ -75,6 +75,8 @@ struct S3ClientConf {
     int request_timeout_ms = -1;
     int connect_timeout_ms = -1;
     bool use_virtual_addressing = true;
+    // For aws s3, no need to override endpoint
+    bool need_override_endpoint = true;
 
     uint64_t get_hash() const {
         uint64_t hash_code = 0;

fe/fe-core/pom.xml

@@ -928,6 +928,22 @@ under the License.
             <artifactId>azure-storage-blob-batch</artifactId>
             <version>${azure.sdk.batch.version}</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>s3tables</artifactId>
+            <version>${awssdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.s3tables</groupId>
+            <artifactId>s3-tables-catalog-for-iceberg</artifactId>
+            <version>${s3tables.catalog.version}</version>
+        </dependency>
+        <!-- AWS SDK Core -->
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>sdk-core</artifactId>
+            <version>${awssdk.version}</version>
+        </dependency>
     </dependencies>
     <repositories>
         <!-- for huawei obs sdk -->

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java

@@ -43,6 +43,7 @@ public abstract class IcebergExternalCatalog extends ExternalCatalog {
     public static final String ICEBERG_HADOOP = "hadoop";
     public static final String ICEBERG_GLUE = "glue";
     public static final String ICEBERG_DLF = "dlf";
+    public static final String ICEBERG_S3_TABLES = "s3tables";
     public static final String EXTERNAL_CATALOG_NAME = "external_catalog.name";
     protected String icebergCatalogType;
     protected Catalog catalog;

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalogFactory.java

@@ -41,6 +41,8 @@ public static ExternalCatalog createCatalog(long catalogId, String name, String
                 return new IcebergDLFExternalCatalog(catalogId, name, resource, props, comment);
             case IcebergExternalCatalog.ICEBERG_HADOOP:
                 return new IcebergHadoopExternalCatalog(catalogId, name, resource, props, comment);
+            case IcebergExternalCatalog.ICEBERG_S3_TABLES:
+                return new IcebergS3TablesExternalCatalog(catalogId, name, resource, props, comment);
             default:
                 throw new DdlException("Unknown " + IcebergExternalCatalog.ICEBERG_CATALOG_TYPE
                     + " value: " + catalogType);

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergS3TablesExternalCatalog.java

@@ -0,0 +1,66 @@
+// 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.iceberg;
+
+import org.apache.doris.datasource.CatalogProperty;
+import org.apache.doris.datasource.iceberg.s3tables.CustomAwsCredentialsProvider;
+import org.apache.doris.datasource.property.PropertyConverter;
+import org.apache.doris.datasource.property.constants.S3Properties;
+
+import com.google.common.collect.Maps;
+import org.apache.iceberg.CatalogProperties;
+import software.amazon.s3tables.iceberg.S3TablesCatalog;
+
+import java.util.Map;
+
+public class IcebergS3TablesExternalCatalog extends IcebergExternalCatalog {
+
+    public IcebergS3TablesExternalCatalog(long catalogId, String name, String resource, Map<String, String> props,
+            String comment) {
+        super(catalogId, name, comment);
+        props = PropertyConverter.convertToMetaProperties(props);
+        catalogProperty = new CatalogProperty(resource, props);
+    }
+
+    @Override
+    protected void initCatalog() {
+        icebergCatalogType = ICEBERG_S3_TABLES;
+        S3TablesCatalog s3TablesCatalog = new S3TablesCatalog();
+        Map<String, String> s3TablesCatalogProperties = convertToS3TablesCatalogProperties();
+        String warehouse = catalogProperty.getHadoopProperties().get(CatalogProperties.WAREHOUSE_LOCATION);
+        s3TablesCatalogProperties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
+        s3TablesCatalog.initialize(getName(), s3TablesCatalogProperties);
+        catalog = s3TablesCatalog;
+    }
+
+    private Map<String, String> convertToS3TablesCatalogProperties() {
+        Map<String, String> props = catalogProperty.getProperties();
+        Map<String, String> s3Properties = Maps.newHashMap();
+        s3Properties.put("client.credentials-provider", CustomAwsCredentialsProvider.class.getName());
+        if (props.containsKey(S3Properties.ACCESS_KEY)) {
+            s3Properties.put("client.credentials-provider.s3.access-key-id", props.get(S3Properties.ACCESS_KEY));
+        }
+        if (props.containsKey(S3Properties.SECRET_KEY)) {
+            s3Properties.put("client.credentials-provider.s3.secret-access-key", props.get(S3Properties.SECRET_KEY));
+        }
+        if (props.containsKey(S3Properties.REGION)) {
+            s3Properties.put("client.region", props.get(S3Properties.REGION));
+        }
+        return s3Properties;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/s3tables/CustomAwsCredentialsProvider.java

@@ -0,0 +1,43 @@
+// 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.iceberg.s3tables;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+
+import java.util.Map;
+
+public class CustomAwsCredentialsProvider implements AwsCredentialsProvider {
+    private final String accessKeyId;
+    private final String secretAccessKey;
+
+    public CustomAwsCredentialsProvider(String accessKeyId, String secretAccessKey) {
+        this.accessKeyId = accessKeyId;
+        this.secretAccessKey = secretAccessKey;
+    }
+
+    @Override
+    public AwsCredentials resolveCredentials() {
+        return AwsBasicCredentials.create(accessKeyId, secretAccessKey);
+    }
+
+    public static CustomAwsCredentialsProvider create(Map<String, String> props) {
+        return new CustomAwsCredentialsProvider(props.get("s3.access-key-id"), props.get("s3.secret-access-key"));
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java

@@ -121,6 +121,7 @@ public IcebergScanNode(PlanNodeId id, TupleDescriptor desc, boolean needCheckCol
                 case IcebergExternalCatalog.ICEBERG_DLF:
                 case IcebergExternalCatalog.ICEBERG_GLUE:
                 case IcebergExternalCatalog.ICEBERG_HADOOP:
+                case IcebergExternalCatalog.ICEBERG_S3_TABLES:
                     source = new IcebergApiSource((IcebergExternalTable) table, desc, columnNameToRange);
                     break;
                 default:

fe/fe-core/src/main/java/org/apache/doris/datasource/property/S3ClientBEProperties.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.datasource.property;
 
+import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
 import org.apache.doris.datasource.property.constants.CosProperties;
 import org.apache.doris.datasource.property.constants.GCSProperties;
 import org.apache.doris.datasource.property.constants.MinioProperties;
@@ -53,6 +54,13 @@ public static Map<String, String> getBeFSProperties(Map<String, String> properti
 
     private static Map<String, String> getBeAWSPropertiesFromS3(Map<String, String> properties) {
         Map<String, String> beProperties = new HashMap<>();
+        if (properties.containsKey(IcebergExternalCatalog.ICEBERG_CATALOG_TYPE)
+                && properties.get(IcebergExternalCatalog.ICEBERG_CATALOG_TYPE).equals(
+                    IcebergExternalCatalog.ICEBERG_S3_TABLES)) {
+            beProperties.put(Env.NEED_OVERRIDE_ENDPOINT, "false");
+        } else {
+            beProperties.put(Env.NEED_OVERRIDE_ENDPOINT, "true");
+        }
         String endpoint = properties.get(S3Properties.ENDPOINT);
         beProperties.put(S3Properties.Env.ENDPOINT, endpoint);
         String region = PropertyConverter.checkRegion(endpoint, properties.get(S3Properties.Env.REGION),

fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/S3Properties.java

@@ -119,6 +119,8 @@ public static class Env {
         public static final String DEFAULT_MAX_CONNECTIONS = "50";
         public static final String DEFAULT_REQUEST_TIMEOUT_MS = "3000";
         public static final String DEFAULT_CONNECTION_TIMEOUT_MS = "1000";
+        public static final String NEED_OVERRIDE_ENDPOINT = "AWS_NEED_OVERRIDE_ENDPOINT";
+
         public static final List<String> REQUIRED_FIELDS = Arrays.asList(ENDPOINT);
         public static final List<String> FS_KEYS = Arrays.asList(ENDPOINT, REGION, ACCESS_KEY, SECRET_KEY, TOKEN,
                 ROOT_PATH, BUCKET, MAX_CONNECTIONS, REQUEST_TIMEOUT_MS, CONNECTION_TIMEOUT_MS);

fe/fe-core/src/main/java/org/apache/doris/persist/gson/GsonUtils.java

@@ -146,6 +146,7 @@
 import org.apache.doris.datasource.iceberg.IcebergHMSExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergHadoopExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergRestExternalCatalog;
+import org.apache.doris.datasource.iceberg.IcebergS3TablesExternalCatalog;
 import org.apache.doris.datasource.infoschema.ExternalInfoSchemaDatabase;
 import org.apache.doris.datasource.infoschema.ExternalInfoSchemaTable;
 import org.apache.doris.datasource.infoschema.ExternalMysqlDatabase;
@@ -417,6 +418,8 @@ public class GsonUtils {
                 .registerSubtype(IcebergRestExternalCatalog.class, IcebergRestExternalCatalog.class.getSimpleName())
                 .registerSubtype(IcebergDLFExternalCatalog.class, IcebergDLFExternalCatalog.class.getSimpleName())
                 .registerSubtype(IcebergHadoopExternalCatalog.class, IcebergHadoopExternalCatalog.class.getSimpleName())
+                .registerSubtype(IcebergS3TablesExternalCatalog.class,
+                        IcebergS3TablesExternalCatalog.class.getSimpleName())
                 .registerSubtype(PaimonExternalCatalog.class, PaimonExternalCatalog.class.getSimpleName())
                 .registerSubtype(PaimonHMSExternalCatalog.class, PaimonHMSExternalCatalog.class.getSimpleName())
                 .registerSubtype(PaimonFileExternalCatalog.class, PaimonFileExternalCatalog.class.getSimpleName())