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

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);
     }
 }