[opt](iceberg) support glue rest catalog with s3tables

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

@@ -130,6 +130,32 @@ public class IcebergRestProperties extends AbstractIcebergProperties {
             description = "The cache TTL for case insensitive name matching in ms.")
     private String icebergRestCaseInsensitiveNameMatchingCacheTtlMs = "0";
 
+    // The following properties are specific to AWS Glue Rest Catalog
+    @ConnectorProperty(names = {"iceberg.rest.sigv4-enabled"},
+            required = false,
+            description = "True for Glue Rest Catalog")
+    private String icebergRestSigV4Enabled = "";
+
+    @ConnectorProperty(names = {"iceberg.rest.signing-name"},
+            required = false,
+            description = "The signing name for the iceberg rest catalog service.")
+    private String icebergRestSigningName = "";
+
+    @ConnectorProperty(names = {"iceberg.rest.signing-region"},
+            required = false,
+            description = "The signing region for the iceberg rest catalog service.")
+    private String icebergRestSigningRegion = "";
+
+    @ConnectorProperty(names = {"iceberg.rest.access-key-id"},
+            required = false,
+            description = "The access key ID for the iceberg rest catalog service.")
+    private String icebergRestAccessKeyId = "";
+
+    @ConnectorProperty(names = {"iceberg.rest.secret-access-key"},
+            required = false,
+            description = "The secret access key for the iceberg rest catalog service.")
+    private String icebergRestSecretAccessKey = "";
+
     protected IcebergRestProperties(Map<String, String> props) {
         super(props);
     }
@@ -196,6 +222,15 @@ private ParamRules buildRules() {
                 throw new IllegalArgumentException("OAuth2 requires either credential or token");
             }
         }
+
+        // Check for glue rest catalog specific properties
+        rules.requireIf(icebergRestSigningName, "glue",
+                new String[] {icebergRestSigningRegion,
+                        icebergRestAccessKeyId,
+                        icebergRestSecretAccessKey,
+                        icebergRestSigV4Enabled},
+                "Rest Catalog requires signing-region, access-key-id, secret-access-key "
+                        + "and sigv4-enabled set to true when signing-name is glue");
         return rules;
     }
 
@@ -207,6 +242,8 @@ private void initIcebergRestCatalogProperties() {
         addOptionalProperties();
         // Authentication properties
         addAuthenticationProperties();
+        // Glue Rest Catalog specific properties
+        addGlueRestCatalogProperties();
     }
 
     private void addCoreCatalogProperties() {
@@ -253,6 +290,17 @@ private void addOAuth2Properties() {
         }
     }
 
+    private void addGlueRestCatalogProperties() {
+        if (Strings.isNotBlank(icebergRestSigningName) && icebergRestSigningName.equalsIgnoreCase("glue")) {
+            icebergRestCatalogProperties.put("rest.signing-name", "glue");
+            icebergRestCatalogProperties.put("rest.sigv4-enabled", icebergRestSigV4Enabled);
+            icebergRestCatalogProperties.put("rest.access-key-id", icebergRestAccessKeyId);
+            icebergRestCatalogProperties.put("rest.secret-access-key", icebergRestSecretAccessKey);
+            icebergRestCatalogProperties.put("rest.signing-region", icebergRestSigningRegion);
+        }
+    }
+
+
     public Map<String, String> getIcebergRestCatalogProperties() {
         return Collections.unmodifiableMap(icebergRestCatalogProperties);
     }

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

@@ -239,12 +239,22 @@ private void setEndpointIfNotSet() {
             return;
         }
         String endpoint = S3PropertyUtils.constructEndpointFromUrl(origProps, usePathStyle, forceParsingByStandardUrl);
+        if (StringUtils.isBlank(endpoint)) {
+            endpoint = getEndpointFromRegion();
+        }
         if (StringUtils.isBlank(endpoint)) {
             throw new IllegalArgumentException("endpoint is required");
         }
         setEndpoint(endpoint);
     }
 
+    // This method should be overridden by subclasses to provide a default endpoint based on the region.
+    // Because for aws s3, only region is needed, the endpoint can be constructed from the region.
+    // But for other s3 compatible storage, the endpoint may need to be specified explicitly.
+    protected String getEndpointFromRegion() {
+        return "";
+    }
+
     @Override
     public String validateAndNormalizeUri(String uri) throws UserException {
         return S3PropertyUtils.validateAndNormalizeUri(uri, getUsePathStyle(), getForceParsingByStandardUrl());

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

@@ -46,10 +46,15 @@
 
 public class S3Properties extends AbstractS3CompatibleProperties {
 
-    private static final String[] ENDPOINT_NAMES = {
+    private static final String[] ENDPOINT_NAMES_FOR_GUESSING = {
             "s3.endpoint", "AWS_ENDPOINT", "endpoint", "ENDPOINT", "aws.endpoint", "glue.endpoint",
             "aws.glue.endpoint"
     };
+
+    private static final String[] REGION_NAMES_FOR_GUESSING = {
+            "s3.region", "glue.region", "aws.glue.region", "iceberg.rest.signing-region"
+    };
+
     @Setter
     @Getter
     @ConnectorProperty(names = {"s3.endpoint", "AWS_ENDPOINT", "endpoint", "ENDPOINT", "aws.endpoint", "glue.endpoint",
@@ -61,21 +66,21 @@ public class S3Properties extends AbstractS3CompatibleProperties {
     @Setter
     @Getter
     @ConnectorProperty(names = {"s3.region", "AWS_REGION", "region", "REGION", "aws.region", "glue.region",
-            "aws.glue.region"},
+            "aws.glue.region", "iceberg.rest.signing-region"},
             required = false,
             description = "The region of S3.")
     protected String region = "";
 
     @Getter
     @ConnectorProperty(names = {"s3.access_key", "AWS_ACCESS_KEY", "access_key", "ACCESS_KEY", "glue.access_key",
-            "aws.glue.access-key", "client.credentials-provider.glue.access_key"},
+            "aws.glue.access-key", "client.credentials-provider.glue.access_key", "iceberg.rest.access-key-id"},
             required = false,
             description = "The access key of S3. Optional for anonymous access to public datasets.")
     protected String accessKey = "";
 
     @Getter
     @ConnectorProperty(names = {"s3.secret_key", "AWS_SECRET_KEY", "secret_key", "SECRET_KEY", "glue.secret_key",
-            "aws.glue.secret-key", "client.credentials-provider.glue.secret_key"},
+            "aws.glue.secret-key", "client.credentials-provider.glue.secret_key", "iceberg.rest.secret-access-key"},
             required = false,
             description = "The secret key of S3. Optional for anonymous access to public datasets.")
     protected String secretKey = "";
@@ -189,7 +194,7 @@ public void initNormalizeAndCheckProps() {
      * @return
      */
     protected static boolean guessIsMe(Map<String, String> origProps) {
-        String endpoint = Stream.of(ENDPOINT_NAMES)
+        String endpoint = Stream.of(ENDPOINT_NAMES_FOR_GUESSING)
                 .map(origProps::get)
                 .filter(Objects::nonNull)
                 .findFirst()
@@ -203,11 +208,26 @@ protected static boolean guessIsMe(Map<String, String> origProps) {
         if (!Strings.isNullOrEmpty(endpoint)) {
             return endpoint.contains("amazonaws.com");
         }
+
+        // guess from URI
         Optional<String> uriValue = origProps.entrySet().stream()
                 .filter(e -> e.getKey().equalsIgnoreCase("uri"))
                 .map(Map.Entry::getValue)
                 .findFirst();
-        return uriValue.isPresent() && uriValue.get().contains("amazonaws.com");
+        if (uriValue.isPresent()) {
+            return uriValue.get().contains("amazonaws.com");
+        }
+
+        // guess from region
+        String region = Stream.of(REGION_NAMES_FOR_GUESSING)
+                .map(origProps::get)
+                .filter(Objects::nonNull)
+                .findFirst()
+                .orElse(null);
+        if (!Strings.isNullOrEmpty(region)) {
+            return true;
+        }
+        return false;
     }
 
     @Override
@@ -278,4 +298,16 @@ public void initializeHadoopStorageConfig() {
                     "org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider");
         }
     }
+
+    @Override
+    protected String getEndpointFromRegion() {
+        if (!StringUtils.isBlank(endpoint)) {
+            return endpoint;
+        }
+        if (StringUtils.isBlank(region)) {
+            return "";
+        }
+        return "https://s3." + region + ".amazonaws.com";
+    }
 }
+

fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/IcebergRestPropertiesTest.java

@@ -231,4 +231,189 @@ public void testImmutablePropertiesMap() {
             catalogProps.put("test", "value");
         });
     }
+
+    @Test
+    public void testGlueRestCatalogValidConfiguration() {
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "glue");
+        props.put("iceberg.rest.signing-region", "us-east-1");
+        props.put("iceberg.rest.access-key-id", "AKIAIOSFODNN7EXAMPLE");
+        props.put("iceberg.rest.secret-access-key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+        props.put("iceberg.rest.sigv4-enabled", "true");
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        restProps.initNormalizeAndCheckProps();
+
+        Map<String, String> catalogProps = restProps.getIcebergRestCatalogProperties();
+        Assertions.assertEquals("glue", catalogProps.get("rest.signing-name"));
+        Assertions.assertEquals("us-east-1", catalogProps.get("rest.signing-region"));
+        Assertions.assertEquals("AKIAIOSFODNN7EXAMPLE", catalogProps.get("rest.access-key-id"));
+        Assertions.assertEquals("wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+                catalogProps.get("rest.secret-access-key"));
+        Assertions.assertEquals("true", catalogProps.get("rest.sigv4-enabled"));
+    }
+
+    @Test
+    public void testGlueRestCatalogCaseInsensitive() {
+        // Test that "GLUE" is also recognized (case insensitive)
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "GLUE");
+        props.put("iceberg.rest.signing-region", "us-west-2");
+        props.put("iceberg.rest.access-key-id", "AKIAIOSFODNN7EXAMPLE");
+        props.put("iceberg.rest.secret-access-key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+        props.put("iceberg.rest.sigv4-enabled", "true");
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        restProps.initNormalizeAndCheckProps();
+
+        Map<String, String> catalogProps = restProps.getIcebergRestCatalogProperties();
+        Assertions.assertEquals("glue", catalogProps.get("rest.signing-name"));
+        Assertions.assertEquals("us-west-2", catalogProps.get("rest.signing-region"));
+    }
+
+    @Test
+    public void testGlueRestCatalogMissingSigningRegion() {
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "glue");
+        props.put("iceberg.rest.access-key-id", "AKIAIOSFODNN7EXAMPLE");
+        props.put("iceberg.rest.secret-access-key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+        props.put("iceberg.rest.sigv4-enabled", "true");
+        // Missing signing-region
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        Assertions.assertThrows(IllegalArgumentException.class, restProps::initNormalizeAndCheckProps);
+    }
+
+    @Test
+    public void testGlueRestCatalogMissingAccessKeyId() {
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "glue");
+        props.put("iceberg.rest.signing-region", "us-east-1");
+        props.put("iceberg.rest.secret-access-key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+        props.put("iceberg.rest.sigv4-enabled", "true");
+        // Missing access-key-id
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        Assertions.assertThrows(IllegalArgumentException.class, restProps::initNormalizeAndCheckProps);
+    }
+
+    @Test
+    public void testGlueRestCatalogMissingSecretAccessKey() {
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "glue");
+        props.put("iceberg.rest.signing-region", "us-east-1");
+        props.put("iceberg.rest.access-key-id", "AKIAIOSFODNN7EXAMPLE");
+        props.put("iceberg.rest.sigv4-enabled", "true");
+        // Missing secret-access-key
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        Assertions.assertThrows(IllegalArgumentException.class, restProps::initNormalizeAndCheckProps);
+    }
+
+    @Test
+    public void testGlueRestCatalogMissingSigV4Enabled() {
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "glue");
+        props.put("iceberg.rest.signing-region", "us-east-1");
+        props.put("iceberg.rest.access-key-id", "AKIAIOSFODNN7EXAMPLE");
+        props.put("iceberg.rest.secret-access-key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+        // Missing sigv4-enabled
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        Assertions.assertThrows(IllegalArgumentException.class, restProps::initNormalizeAndCheckProps);
+    }
+
+    @Test
+    public void testNonGlueSigningNameDoesNotRequireAdditionalProperties() {
+        // Test that non-glue signing names don't require additional properties
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "custom-service");
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        restProps.initNormalizeAndCheckProps(); // Should not throw
+
+        Map<String, String> catalogProps = restProps.getIcebergRestCatalogProperties();
+        // Should not contain glue-specific properties
+        Assertions.assertFalse(catalogProps.containsKey("rest.signing-name"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.signing-region"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.access-key-id"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.secret-access-key"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.sigv4-enabled"));
+    }
+
+    @Test
+    public void testEmptySigningNameDoesNotAddGlueProperties() {
+        // Test that empty signing name doesn't add glue properties
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-region", "us-east-1");
+        props.put("iceberg.rest.access-key-id", "AKIAIOSFODNN7EXAMPLE");
+        props.put("iceberg.rest.secret-access-key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+        props.put("iceberg.rest.sigv4-enabled", "true");
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        restProps.initNormalizeAndCheckProps(); // Should not throw
+
+        Map<String, String> catalogProps = restProps.getIcebergRestCatalogProperties();
+        // Should not contain glue-specific properties since signing-name is not "glue"
+        Assertions.assertFalse(catalogProps.containsKey("rest.signing-name"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.signing-region"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.access-key-id"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.secret-access-key"));
+        Assertions.assertFalse(catalogProps.containsKey("rest.sigv4-enabled"));
+    }
+
+    @Test
+    public void testGlueRestCatalogWithOAuth2() {
+        // Test that Glue properties can be combined with OAuth2
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.security.type", "oauth2");
+        props.put("iceberg.rest.oauth2.token", "my-access-token");
+        props.put("iceberg.rest.signing-name", "glue");
+        props.put("iceberg.rest.signing-region", "us-east-1");
+        props.put("iceberg.rest.access-key-id", "AKIAIOSFODNN7EXAMPLE");
+        props.put("iceberg.rest.secret-access-key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
+        props.put("iceberg.rest.sigv4-enabled", "true");
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        restProps.initNormalizeAndCheckProps();
+
+        Map<String, String> catalogProps = restProps.getIcebergRestCatalogProperties();
+        // Should have both OAuth2 and Glue properties
+        Assertions.assertEquals("my-access-token", catalogProps.get(OAuth2Properties.TOKEN));
+        Assertions.assertEquals("glue", catalogProps.get("rest.signing-name"));
+        Assertions.assertEquals("us-east-1", catalogProps.get("rest.signing-region"));
+        Assertions.assertEquals("AKIAIOSFODNN7EXAMPLE", catalogProps.get("rest.access-key-id"));
+        Assertions.assertEquals("wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+                catalogProps.get("rest.secret-access-key"));
+        Assertions.assertEquals("true", catalogProps.get("rest.sigv4-enabled"));
+    }
+
+    @Test
+    public void testGlueRestCatalogMissingMultipleProperties() {
+        // Test error message when multiple required properties are missing
+        Map<String, String> props = new HashMap<>();
+        props.put("iceberg.rest.uri", "http://localhost:8080");
+        props.put("iceberg.rest.signing-name", "glue");
+        // Missing all required properties
+
+        IcebergRestProperties restProps = new IcebergRestProperties(props);
+        IllegalArgumentException exception = Assertions.assertThrows(
+                IllegalArgumentException.class, restProps::initNormalizeAndCheckProps);
+
+        // The error message should mention the required properties
+        String errorMessage = exception.getMessage();
+        Assertions.assertTrue(errorMessage.contains("signing-region")
+                || errorMessage.contains("access-key-id")
+                || errorMessage.contains("secret-access-key")
+                || errorMessage.contains("sigv4-enabled"));
+    }
 }