|
| 1 | +// Licensed to the Apache Software Foundation (ASF) under one |
| 2 | +// or more contributor license agreements. See the NOTICE file |
| 3 | +// distributed with this work for additional information |
| 4 | +// regarding copyright ownership. The ASF licenses this file |
| 5 | +// to you under the Apache License, Version 2.0 (the |
| 6 | +// "License"); you may not use this file except in compliance |
| 7 | +// with the License. You may obtain a copy of the License at |
| 8 | +// |
| 9 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | +// |
| 11 | +// Unless required by applicable law or agreed to in writing, |
| 12 | +// software distributed under the License is distributed on an |
| 13 | +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 14 | +// KIND, either express or implied. See the License for the |
| 15 | +// specific language governing permissions and limitations |
| 16 | +// under the License. |
| 17 | + |
| 18 | +package org.apache.doris.connector.iceberg.dlf; |
| 19 | + |
| 20 | +import org.apache.doris.connector.iceberg.dlf.client.DLFCachedClientPool; |
| 21 | +import org.apache.doris.filesystem.properties.S3CompatibleFileSystemProperties; |
| 22 | + |
| 23 | +import org.apache.commons.lang3.StringUtils; |
| 24 | +import org.apache.hadoop.conf.Configuration; |
| 25 | +import org.apache.iceberg.TableOperations; |
| 26 | +import org.apache.iceberg.aws.s3.S3FileIO; |
| 27 | +import org.apache.iceberg.catalog.TableIdentifier; |
| 28 | +import org.apache.iceberg.io.FileIO; |
| 29 | +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; |
| 30 | +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; |
| 31 | +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; |
| 32 | +import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider; |
| 33 | +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; |
| 34 | +import software.amazon.awssdk.auth.signer.AwsS3V4Signer; |
| 35 | +import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; |
| 36 | +import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption; |
| 37 | +import software.amazon.awssdk.core.retry.RetryPolicy; |
| 38 | +import software.amazon.awssdk.core.retry.backoff.EqualJitterBackoffStrategy; |
| 39 | +import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient; |
| 40 | +import software.amazon.awssdk.regions.Region; |
| 41 | +import software.amazon.awssdk.services.s3.S3Client; |
| 42 | +import software.amazon.awssdk.services.s3.S3Configuration; |
| 43 | + |
| 44 | +import java.net.URI; |
| 45 | +import java.time.Duration; |
| 46 | +import java.util.Map; |
| 47 | + |
| 48 | +/** Aliyun DLF Iceberg catalog backed by the DLF Hive-compatible metastore and OSS. */ |
| 49 | +public class DLFCatalog extends HiveCompatibleCatalog { |
| 50 | + |
| 51 | + private final S3CompatibleFileSystemProperties ossStorage; |
| 52 | + |
| 53 | + public DLFCatalog(S3CompatibleFileSystemProperties ossStorage) { |
| 54 | + this.ossStorage = ossStorage; |
| 55 | + } |
| 56 | + |
| 57 | + @Override |
| 58 | + public void initialize(String name, Map<String, String> properties) { |
| 59 | + super.initialize(name, initializeFileIO(properties, conf), new DLFCachedClientPool(conf, properties)); |
| 60 | + } |
| 61 | + |
| 62 | + @Override |
| 63 | + protected TableOperations newTableOps(TableIdentifier tableIdentifier) { |
| 64 | + return new DLFTableOperations(conf, clients, fileIO, catalogName, |
| 65 | + tableIdentifier.namespace().level(0), tableIdentifier.name()); |
| 66 | + } |
| 67 | + |
| 68 | + @Override |
| 69 | + protected FileIO initializeFileIO(Map<String, String> properties, Configuration hadoopConf) { |
| 70 | + String region = ossStorage.getRegion(); |
| 71 | + boolean usePathStyle = Boolean.parseBoolean(ossStorage.getUsePathStyle()); |
| 72 | + URI endpoint = URI.create(toS3CompatibleEndpoint(ossStorage.getEndpoint(), region)); |
| 73 | + AwsCredentialsProvider credentials = buildCredentials(ossStorage); |
| 74 | + FileIO io = new S3FileIO(() -> buildOssS3Client(endpoint, region, credentials, usePathStyle)); |
| 75 | + io.initialize(properties); |
| 76 | + return io; |
| 77 | + } |
| 78 | + |
| 79 | + static String toS3CompatibleEndpoint(String endpoint, String region) { |
| 80 | + String s3Endpoint = endpoint.replace("oss-" + region, "s3.oss-" + region); |
| 81 | + if (!s3Endpoint.contains("://")) { |
| 82 | + s3Endpoint = "http://" + s3Endpoint; |
| 83 | + } |
| 84 | + return s3Endpoint; |
| 85 | + } |
| 86 | + |
| 87 | + private static AwsCredentialsProvider buildCredentials(S3CompatibleFileSystemProperties oss) { |
| 88 | + if (!oss.hasStaticCredentials()) { |
| 89 | + return DefaultCredentialsProvider.create(); |
| 90 | + } |
| 91 | + if (StringUtils.isBlank(oss.getSessionToken())) { |
| 92 | + return StaticCredentialsProvider.create(AwsBasicCredentials.create(oss.getAccessKey(), oss.getSecretKey())); |
| 93 | + } |
| 94 | + return StaticCredentialsProvider.create( |
| 95 | + AwsSessionCredentials.create(oss.getAccessKey(), oss.getSecretKey(), oss.getSessionToken())); |
| 96 | + } |
| 97 | + |
| 98 | + private static S3Client buildOssS3Client(URI endpoint, String region, AwsCredentialsProvider credentials, |
| 99 | + boolean usePathStyle) { |
| 100 | + RetryPolicy retryPolicy = RetryPolicy.builder() |
| 101 | + .numRetries(3) |
| 102 | + .backoffStrategy(EqualJitterBackoffStrategy.builder() |
| 103 | + .baseDelay(Duration.ofSeconds(1)) |
| 104 | + .maxBackoffTime(Duration.ofMinutes(1)) |
| 105 | + .build()) |
| 106 | + .build(); |
| 107 | + ClientOverrideConfiguration clientConf = ClientOverrideConfiguration.builder() |
| 108 | + .retryPolicy(retryPolicy) |
| 109 | + .putAdvancedOption(SdkAdvancedClientOption.SIGNER, AwsS3V4Signer.create()) |
| 110 | + .build(); |
| 111 | + return S3Client.builder() |
| 112 | + .httpClient(UrlConnectionHttpClient.builder() |
| 113 | + .socketTimeout(Duration.ofSeconds(30)) |
| 114 | + .connectionTimeout(Duration.ofSeconds(30)) |
| 115 | + .build()) |
| 116 | + .endpointOverride(endpoint) |
| 117 | + .credentialsProvider(credentials) |
| 118 | + .region(Region.of(region)) |
| 119 | + .overrideConfiguration(clientConf) |
| 120 | + .serviceConfiguration(S3Configuration.builder() |
| 121 | + .chunkedEncodingEnabled(false) |
| 122 | + .pathStyleAccessEnabled(usePathStyle) |
| 123 | + .build()) |
| 124 | + .build(); |
| 125 | + } |
| 126 | +} |
0 commit comments