Add scope field for Iceberg REST metastore

docs/src/main/sphinx/object-storage/metastores.md

@@ -483,6 +483,9 @@ following properties:
   - The credential to exchange for a token in the OAuth2 client credentials flow
     with the server. A `token` or `credential` is required for `OAUTH2`
     security. Example: `AbCdEf123456`
+* - `iceberg.rest-catalog.oauth2.scope`
+  - Scope to be used when communicating with the REST Catalog. Applicable only
+    when using `credential`.
 :::
 
 The following example shows a minimal catalog configuration using an Iceberg

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/OAuth2SecurityConfig.java

@@ -23,6 +23,7 @@
 public class OAuth2SecurityConfig
 {
     private String credential;
+    private String scope;
     private String token;
 
     public Optional<String> getCredential()
@@ -39,6 +40,19 @@ public OAuth2SecurityConfig setCredential(String credential)
         return this;
     }
 
+    public Optional<String> getScope()
+    {
+        return Optional.ofNullable(scope);
+    }
+
+    @Config("iceberg.rest-catalog.oauth2.scope")
+    @ConfigDescription("The scope which will be used for interactions with the server")
+    public OAuth2SecurityConfig setScope(String scope)
+    {
+        this.scope = scope;
+        return this;
+    }
+
     public Optional<String> getToken()
     {
         return Optional.ofNullable(token);
@@ -58,4 +72,10 @@ public boolean credentialOrTokenPresent()
     {
         return credential != null || token != null;
     }
+
+    @AssertTrue(message = "Scope is applicable only when using credential")
+    public boolean scopePresentOnlyWithCredential()
+    {
+        return !(token != null && scope != null);
+    }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/OAuth2SecurityProperties.java

@@ -33,7 +33,11 @@ public OAuth2SecurityProperties(OAuth2SecurityConfig securityConfig)
 
         ImmutableMap.Builder<String, String> propertiesBuilder = ImmutableMap.builder();
         securityConfig.getCredential().ifPresent(
-                value -> propertiesBuilder.put(OAuth2Properties.CREDENTIAL, value));
+                credential -> {
+                    propertiesBuilder.put(OAuth2Properties.CREDENTIAL, credential);
+                    securityConfig.getScope()
+                            .ifPresent(scope -> propertiesBuilder.put(OAuth2Properties.SCOPE, scope));
+                });
         securityConfig.getToken().ifPresent(
                 value -> propertiesBuilder.put(OAuth2Properties.TOKEN, value));
 

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/IcebergQueryRunner.java

@@ -218,7 +218,8 @@ public static void main(String[] args)
                     .addIcebergProperty("iceberg.rest-catalog.uri", polarisCatalog.restUri() + "/api/catalog")
                     .addIcebergProperty("iceberg.rest-catalog.warehouse", TestingPolarisCatalog.WAREHOUSE)
                     .addIcebergProperty("iceberg.rest-catalog.security", "OAUTH2")
-                    .addIcebergProperty("iceberg.rest-catalog.oauth2.token", polarisCatalog.oauth2Token())
+                    .addIcebergProperty("iceberg.rest-catalog.oauth2.credential", polarisCatalog.oauth2Credentials())
+                    .addIcebergProperty("iceberg.rest-catalog.oauth2.scope", "PRINCIPAL_ROLE:ALL")
                     .setInitialTables(TpchTable.getTables())
                     .build();
 

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/rest/TestIcebergPolarisCatalogConnectorSmokeTest.java

@@ -84,7 +84,8 @@ protected QueryRunner createQueryRunner()
                 .addIcebergProperty("iceberg.rest-catalog.uri", polarisCatalog.restUri() + "/api/catalog")
                 .addIcebergProperty("iceberg.rest-catalog.warehouse", TestingPolarisCatalog.WAREHOUSE)
                 .addIcebergProperty("iceberg.rest-catalog.security", "OAUTH2")
-                .addIcebergProperty("iceberg.rest-catalog.oauth2.token", polarisCatalog.oauth2Token())
+                .addIcebergProperty("iceberg.rest-catalog.oauth2.credential", polarisCatalog.oauth2Credentials())
+                .addIcebergProperty("iceberg.rest-catalog.oauth2.scope", "PRINCIPAL_ROLE:ALL")
                 .setInitialTables(REQUIRED_TPCH_TABLES)
                 .build();
     }

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/rest/TestOAuth2SecurityConfig.java

@@ -30,7 +30,8 @@ public void testDefaults()
     {
         assertRecordedDefaults(recordDefaults(OAuth2SecurityConfig.class)
                 .setCredential(null)
-                .setToken(null));
+                .setToken(null)
+                .setScope(null));
     }
 
     @Test
@@ -39,12 +40,15 @@ public void testExplicitPropertyMappings()
         Map<String, String> properties = ImmutableMap.<String, String>builder()
                 .put("iceberg.rest-catalog.oauth2.token", "token")
                 .put("iceberg.rest-catalog.oauth2.credential", "credential")
+                .put("iceberg.rest-catalog.oauth2.scope", "scope")
                 .buildOrThrow();
 
         OAuth2SecurityConfig expected = new OAuth2SecurityConfig()
                 .setCredential("credential")
-                .setToken("token");
+                .setToken("token")
+                .setScope("scope");
         assertThat(expected.credentialOrTokenPresent()).isTrue();
+        assertThat(expected.scopePresentOnlyWithCredential()).isFalse();
         assertFullMapping(properties, expected);
     }
 }

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/rest/TestingPolarisCatalog.java

@@ -39,11 +39,12 @@ public final class TestingPolarisCatalog
 {
     public static final String WAREHOUSE = "polaris";
     private static final int POLARIS_PORT = 8181;
-    private static final Pattern PATTERN = Pattern.compile("realm: default-realm root principal credentials: [a-f0-9]+:(?<secret>[a-f0-9]+)");
+    private static final Pattern PATTERN = Pattern.compile("realm: default-realm root principal credentials: (?<id>[a-f0-9]+):(?<secret>[a-f0-9]+)");
 
     private static final HttpClient HTTP_CLIENT = new JettyHttpClient();
 
     private final GenericContainer<?> polarisCatalog;
+    private final String clientId;
     private final String clientSecret;
     private final String warehouseLocation;
 
@@ -58,11 +59,21 @@ public TestingPolarisCatalog(String warehouseLocation)
         polarisCatalog.waitingFor(new LogMessageWaitStrategy().withRegEx(".*o.eclipse.jetty.server.Server: Started.*"));
         polarisCatalog.start();
 
+        clientId = findClientId();
         clientSecret = findClientSecret();
         createCatalog();
         grantPrivilege();
     }
 
+    private String findClientId()
+    {
+        return Stream.of(polarisCatalog.getLogs().split("\n"))
+                .map(PATTERN::matcher)
+                .filter(Matcher::find)
+                .map(matcher -> matcher.group("id"))
+                .collect(onlyElement());
+    }
+
     private String findClientSecret()
     {
         return Stream.of(polarisCatalog.getLogs().split("\n"))
@@ -124,6 +135,11 @@ public String oauth2Token()
         return "principal:root;password:%s;realm:default-realm;role:ALL".formatted(clientSecret);
     }
 
+    public String oauth2Credentials()
+    {
+        return "%s:%s".formatted(clientId, clientSecret);
+    }
+
     @Override
     public void close()
     {