fixup! Add support for case-insensitive object names in Iceberg REST …

…catalog - fix invalid cache state

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

@@ -412,7 +412,7 @@ public Table loadTable(ConnectorSession session, SchemaTableName schemaTableName
                     tableCache,
                     schemaTableName,
                     () -> {
-                        BaseTable baseTable = (BaseTable) restSessionCatalog.loadTable(convert(session), toRemoteTable(session, schemaTableName));
+                        BaseTable baseTable = (BaseTable) restSessionCatalog.loadTable(convert(session), toRemoteObject(session, schemaTableName));
                         // Creating a new base table is necessary to adhere to Trino's expectations for quoted table names
                         return new BaseTable(baseTable.operations(), quotedTableName(schemaTableName));
                     });
@@ -425,6 +425,22 @@ public Table loadTable(ConnectorSession session, SchemaTableName schemaTableName
         }
     }
 
+    private TableIdentifier toRemoteObject(ConnectorSession session, SchemaTableName schemaTableName)
+    {
+        TableIdentifier liveTable = toLiveRemoteTable(session, schemaTableName);
+        TableIdentifier liveView = toLiveRemoteView(session, schemaTableName);
+        if (!liveTable.name().equals(schemaTableName.getTableName())) {
+            return liveTable;
+        }
+        else if (!liveView.name().equals(schemaTableName.getTableName())) {
+            return liveView;
+        }
+        else if (liveView.name().equals(schemaTableName.getTableName()) && liveTable.name().equals(schemaTableName.getTableName())) {
+            return liveTable; // table name and view name are same
+        }
+        throw new RuntimeException("Unable to find remote object");
+    }
+
     @Override
     public Map<SchemaTableName, List<ColumnMetadata>> tryGetColumnMetadata(ConnectorSession session, List<SchemaTableName> tables)
     {
@@ -545,7 +561,7 @@ public Map<SchemaTableName, ConnectorViewDefinition> getViews(ConnectorSession s
     @Override
     public Optional<ConnectorViewDefinition> getView(ConnectorSession session, SchemaTableName viewName)
     {
-        return getIcebergView(session, viewName).flatMap(view -> {
+        return getIcebergView(session, viewName, true).flatMap(view -> {
             SQLViewRepresentation sqlView = view.sqlFor("trino");
             if (!sqlView.dialect().equalsIgnoreCase("trino")) {
                 throw new TrinoException(ICEBERG_UNSUPPORTED_VIEW_DIALECT, "Cannot read unsupported dialect '%s' for view '%s'".formatted(sqlView.dialect(), viewName));
@@ -565,9 +581,12 @@ public Optional<ConnectorViewDefinition> getView(ConnectorSession session, Schem
         });
     }
 
-    private Optional<View> getIcebergView(ConnectorSession session, SchemaTableName viewName)
+    private Optional<View> getIcebergView(ConnectorSession session, SchemaTableName viewName, boolean findLiveView)
     {
         try {
+            if (caseInsensitiveNameMatching) {
+                return Optional.of(restSessionCatalog.loadView(convert(session), findLiveView ? toLiveRemoteView(session, viewName) : toRemoteView(session, viewName)));
+            }
             return Optional.of(restSessionCatalog.loadView(convert(session), toRemoteView(session, viewName)));
         }
         catch (NoSuchViewException e) {
@@ -640,7 +659,7 @@ public Optional<CatalogSchemaTableName> redirectTable(ConnectorSession session,
     @Override
     public void updateViewComment(ConnectorSession session, SchemaTableName schemaViewName, Optional<String> comment)
     {
-        View view = getIcebergView(session, schemaViewName).orElseThrow(() -> new ViewNotFoundException(schemaViewName));
+        View view = getIcebergView(session, schemaViewName, false).orElseThrow(() -> new ViewNotFoundException(schemaViewName));
         UpdateViewProperties updateViewProperties = view.updateProperties();
         comment.ifPresentOrElse(
                 value -> updateViewProperties.set(COMMENT, value),
@@ -651,7 +670,7 @@ public void updateViewComment(ConnectorSession session, SchemaTableName schemaVi
     @Override
     public void updateViewColumnComment(ConnectorSession session, SchemaTableName schemaViewName, String columnName, Optional<String> comment)
     {
-        View view = getIcebergView(session, schemaViewName)
+        View view = getIcebergView(session, schemaViewName, false)
                 .orElseThrow(() -> new ViewNotFoundException(schemaViewName));
 
         ViewVersion current = view.currentVersion();
@@ -756,6 +775,15 @@ private TableIdentifier toRemoteTable(ConnectorSession session, SchemaTableName
         return tableIdentifier;
     }
 
+    private TableIdentifier toLiveRemoteTable(ConnectorSession session, SchemaTableName schemaTableName)
+    {
+        TableIdentifier tableIdentifier = toIdentifier(schemaTableName);
+        if (caseInsensitiveNameMatching) {
+            return findRemoteTable(session, tableIdentifier);
+        }
+        return tableIdentifier;
+    }
+
     private TableIdentifier findRemoteTable(ConnectorSession session, TableIdentifier tableIdentifier)
     {
         Namespace remoteNamespace = toRemoteNamespace(session, tableIdentifier.namespace());
@@ -786,6 +814,15 @@ private TableIdentifier toRemoteView(ConnectorSession session, SchemaTableName s
         return tableIdentifier;
     }
 
+    private TableIdentifier toLiveRemoteView(ConnectorSession session, SchemaTableName schemaViewName)
+    {
+        TableIdentifier tableIdentifier = toIdentifier(schemaViewName);
+        if (caseInsensitiveNameMatching) {
+            return findRemoteView(session, tableIdentifier);
+        }
+        return tableIdentifier;
+    }
+
     private TableIdentifier findRemoteView(ConnectorSession session, TableIdentifier tableIdentifier)
     {
         Namespace remoteNamespace = toRemoteNamespace(session, tableIdentifier.namespace());