Add support for case-insensitive object names in Iceberg REST catalog

Certain Rest catalog implementation such as Polaris supports
case-sensitive object(namespace, table, view etc.) names. This
change allows querying mixed/upper case letter object names in rest
catalog from Trino.
`iceberg.rest-catalog.case-insensitive-name-matching` controls
whether to match lowercase object names in Trino with different case
object names in rest catalog with the limitations that only single
object name with the identical name is supported.

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

@@ -503,6 +503,11 @@ following properties:
 * - `iceberg.rest-catalog.nested-namespace-enabled`
   - Support querying objects under nested namespace.
     Defaults to `false`.
+* - `iceberg.rest-catalog.case-insensitive-name-matching`
+  - Match namespace, table, and view names case insensitively. Defaults to `false`.
+* - `iceberg.rest-catalog.case-insensitive-name-matching.cache-ttl`
+  - [Duration](prop-type-duration) for which case-insensitive namespace, table, 
+    and view names are cached. Defaults to `1m`.
   :::
 
 The following example shows a minimal catalog configuration using an Iceberg

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

@@ -15,12 +15,16 @@
 
 import io.airlift.configuration.Config;
 import io.airlift.configuration.ConfigDescription;
+import io.airlift.units.Duration;
+import io.airlift.units.MinDuration;
 import jakarta.validation.constraints.NotNull;
 import org.apache.iceberg.catalog.Namespace;
 
 import java.net.URI;
 import java.util.Optional;
 
+import static java.util.concurrent.TimeUnit.MINUTES;
+
 public class IcebergRestCatalogConfig
 {
     public enum Security
@@ -43,6 +47,8 @@ public enum SessionType
     private Security security = Security.NONE;
     private SessionType sessionType = SessionType.NONE;
     private boolean vendedCredentialsEnabled;
+    private boolean caseInsensitiveNameMatching;
+    private Duration caseInsensitiveNameMatchingCacheTtl = new Duration(1, MINUTES);
 
     @NotNull
     public URI getBaseUri()
@@ -152,4 +158,32 @@ public IcebergRestCatalogConfig setVendedCredentialsEnabled(boolean vendedCreden
         this.vendedCredentialsEnabled = vendedCredentialsEnabled;
         return this;
     }
+
+    public boolean isCaseInsensitiveNameMatching()
+    {
+        return caseInsensitiveNameMatching;
+    }
+
+    @Config("iceberg.rest-catalog.case-insensitive-name-matching")
+    @ConfigDescription("Match object names case-insensitively")
+    public IcebergRestCatalogConfig setCaseInsensitiveNameMatching(boolean caseInsensitiveNameMatching)
+    {
+        this.caseInsensitiveNameMatching = caseInsensitiveNameMatching;
+        return this;
+    }
+
+    @NotNull
+    @MinDuration("0ms")
+    public Duration getCaseInsensitiveNameMatchingCacheTtl()
+    {
+        return caseInsensitiveNameMatchingCacheTtl;
+    }
+
+    @Config("iceberg.rest-catalog.case-insensitive-name-matching.cache-ttl")
+    @ConfigDescription("Duration to keep case insensitive object mapping prior to eviction")
+    public IcebergRestCatalogConfig setCaseInsensitiveNameMatchingCacheTtl(Duration caseInsensitiveNameMatchingCacheTtl)
+    {
+        this.caseInsensitiveNameMatchingCacheTtl = caseInsensitiveNameMatchingCacheTtl;
+        return this;
+    }
 }

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

@@ -13,10 +13,12 @@
  */
 package io.trino.plugin.iceberg.catalog.rest;
 
+import com.google.common.cache.Cache;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import com.google.inject.Inject;
+import io.trino.cache.EvictableCacheBuilder;
 import io.trino.plugin.hive.NodeVersion;
 import io.trino.plugin.iceberg.IcebergConfig;
 import io.trino.plugin.iceberg.IcebergFileSystemFactory;
@@ -29,6 +31,7 @@
 import io.trino.spi.type.TypeManager;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.rest.HTTPClient;
 import org.apache.iceberg.rest.RESTSessionCatalog;
 
@@ -38,6 +41,7 @@
 import java.util.Set;
 
 import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.apache.iceberg.rest.auth.OAuth2Properties.CREDENTIAL;
 import static org.apache.iceberg.rest.auth.OAuth2Properties.TOKEN;
 
@@ -57,6 +61,9 @@ public class TrinoIcebergRestCatalogFactory
     private final SecurityProperties securityProperties;
     private final boolean uniqueTableLocation;
     private final TypeManager typeManager;
+    private final boolean caseInsensitiveNameMatching;
+    private final Cache<Namespace, Namespace> remoteNamespaceMappingCache;
+    private final Cache<TableIdentifier, TableIdentifier> remoteTableMappingCache;
 
     @GuardedBy("this")
     private RESTSessionCatalog icebergCatalog;
@@ -86,6 +93,15 @@ public TrinoIcebergRestCatalogFactory(
         requireNonNull(icebergConfig, "icebergConfig is null");
         this.uniqueTableLocation = icebergConfig.isUniqueTableLocation();
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
+        this.caseInsensitiveNameMatching = restConfig.isCaseInsensitiveNameMatching();
+        this.remoteNamespaceMappingCache = EvictableCacheBuilder.newBuilder()
+                .expireAfterWrite(restConfig.getCaseInsensitiveNameMatchingCacheTtl().toMillis(), MILLISECONDS)
+                .shareNothingWhenDisabled()
+                .build();
+        this.remoteTableMappingCache = EvictableCacheBuilder.newBuilder()
+                .expireAfterWrite(restConfig.getCaseInsensitiveNameMatchingCacheTtl().toMillis(), MILLISECONDS)
+                .shareNothingWhenDisabled()
+                .build();
     }
 
     @Override
@@ -132,6 +148,9 @@ public synchronized TrinoCatalog create(ConnectorIdentity identity)
                 nestedNamespaceEnabled,
                 trinoVersion,
                 typeManager,
-                uniqueTableLocation);
+                uniqueTableLocation,
+                caseInsensitiveNameMatching,
+                remoteNamespaceMappingCache,
+                remoteTableMappingCache);
     }
 }