[#10717] fix(trino-connector): Enable multi-metalake support for Trino 469+

Enable supportCatalogNameWithMetalake for Trino 469-472 and 473-478 by
removing the false override, allowing them to inherit the base class
default (true). Enable the corresponding MultiMetalake tests.

For unsupported versions (452-468), change the hard TrinoException to a
LOG.warn so multi-metalake mode degrades gracefully instead of failing.

Enable the MultiMetalake test for 446-451 to match its factory
(supportCatalogNameWithMetalake=true). Override testCreateCatalog to
skip drop assertions due to a Trino 446 bug where DROP CATALOG does
not unquote quoted identifiers.

Author: sachinnn99

docs/trino-connector/configuration.md

@@ -16,11 +16,14 @@ license: "This software is licensed under the Apache License version 2."
 | gravitino.trino.skip-version-validation     | boolean | false                 | The `gravitino.trino.skip-version-validation` defines whether to skip Trino version validation. Gravitino supports Trino versions between 435 and 478. If this option is `true`, unsupported Trino versions can still be used, but compatibility is not guaranteed.                                                 | No       | 1.0.0         |
 | gravitino.client.                           | string  | (none)                | The configuration key prefix for the Gravitino client config.                                                                                                                                                                                                                                                       | No       | 1.0.0         |
 | gravitino.trino.skip-catalog-patterns       | string  | (none)                | The `gravitino.trino.skip-catalog-patterns` defines a comma-separated list of catalog name regex patterns that should be excluded from loading. For example, `test_.*, .*_tmp` excludes all catalogs starting with `test_` or ending with `_tmp`.                                                                   | No       | 1.2.0         |
+| gravitino.use-single-metalake               | boolean | true                  | If `true`, only one metalake is used and catalogs are identified by `<catalog_name>`. If `false`, multi-metalake mode is enabled and catalogs are identified by `<metalake_name>.<catalog_name>`. | No       | 1.2.0         |
 
 To configure the Gravitino client, use properties prefixed with `gravitino.client.`. These properties will directly passed to the Gravitino client.
 
 **Note:** Invalid configuration properties will result in exceptions. Please see [Gravitino Java client configurations](../how-to-use-gravitino-client.md#gravitino-java-client-configuration) for more support client configuration.
 
+Multi-metalake mode (`gravitino.use-single-metalake=false`) is supported on Trino connector versions 435-451 and 469-478. On versions 452-468, a warning is logged and the connector initializes, but the mode is not fully supported and some operations may fail.
+
 ## Authentication
 
 The Gravitino Trino connector supports authenticating to the Gravitino server using Simple, OAuth, and Kerberos authentication. For detailed authentication configuration, please refer to [Trino Connector Authentication](./authentication.md).

trino-connector/trino-connector-446-451/src/test/java/TestGravitinoConnector446.java

@@ -18,6 +18,7 @@
  */
 
 import static io.trino.testing.TestingSession.testSessionBuilder;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import io.trino.Session;
 import io.trino.testing.DistributedQueryRunner;
@@ -26,8 +27,8 @@
 import org.apache.gravitino.trino.connector.GravitinoPlugin446;
 import org.apache.gravitino.trino.connector.TestGravitinoConnector;
 import org.apache.gravitino.trino.connector.TestGravitinoConnectorWithMetalakeCatalogName;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
 
 public class TestGravitinoConnector446 {
   @Nested
@@ -45,7 +46,6 @@ protected DistributedQueryRunner createTrinoQueryRunner() throws Exception {
   }
 
   @Nested
-  @Disabled
   class MultiMetalake extends TestGravitinoConnectorWithMetalakeCatalogName {
     @Override
     protected GravitinoPlugin createGravitinoPlugin(GravitinoAdminClient client) {
@@ -57,5 +57,33 @@ protected DistributedQueryRunner createTrinoQueryRunner() throws Exception {
       Session session = testSessionBuilder().setCatalog("gravitino").build();
       return DistributedQueryRunner.builder(session).setWorkerCount(1).build();
     }
+
+    @Override
+    protected String getTrinoCliCatalogName(String metalake, String catalog) {
+      return metalake + "." + catalog;
+    }
+
+    @Override
+    protected String getTrinoSqlCatalogName(String metalakeName, String catalogName) {
+      return "\"" + metalakeName + "." + catalogName + "\"";
+    }
+
+    // Trino 446-451 has a bug where DROP CATALOG does not unquote identifiers,
+    // so DROP CATALOG "test.memory1" looks up the literal name '"test.memory1"'
+    // instead of 'test.memory1'. Override to test only creation (the core
+    // multi-metalake functionality) and skip the drop-and-verify assertions.
+    @Override
+    @Test
+    public void testCreateCatalog() throws Exception {
+      assertThat(computeActual("show catalogs").getOnlyColumnAsSet()).contains("gravitino");
+      assertThat(computeActual("show catalogs").getOnlyColumnAsSet())
+          .contains(getTrinoCliCatalogName("test", "memory"));
+
+      assertThat(computeActual("select * from system.jdbc.tables").getRowCount()).isGreaterThan(1);
+
+      assertUpdate("call gravitino.system.create_catalog('memory1', 'memory', Map())");
+      assertThat(computeActual("show catalogs").getOnlyColumnAsSet())
+          .contains(getTrinoCliCatalogName("test", "memory1"));
+    }
   }
 }

trino-connector/trino-connector-452-468/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory452.java

@@ -39,11 +39,6 @@ protected int getMaxSupportTrinoSpiVersion() {
     return 468;
   }
 
-  @Override
-  protected String getTrinoCatalogName(String metalake, String catalog) {
-    return "\"" + metalake + "." + catalog + "\"";
-  }
-
   @Override
   protected boolean supportCatalogNameWithMetalake() {
     return false;

trino-connector/trino-connector-469-472/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory469.java

@@ -39,16 +39,6 @@ protected int getMaxSupportTrinoSpiVersion() {
     return 472;
   }
 
-  @Override
-  protected String getTrinoCatalogName(String metalake, String catalog) {
-    return "\"" + metalake + "." + catalog + "\"";
-  }
-
-  @Override
-  protected boolean supportCatalogNameWithMetalake() {
-    return false;
-  }
-
   @Override
   protected GravitinoConnector createConnector(CatalogConnectorContext connectorContext) {
     return new GravitinoConnector469(connectorContext);

trino-connector/trino-connector-469-472/src/test/java/TestGravitinoConnector469.java

@@ -26,7 +26,6 @@
 import org.apache.gravitino.trino.connector.GravitinoPlugin469;
 import org.apache.gravitino.trino.connector.TestGravitinoConnector;
 import org.apache.gravitino.trino.connector.TestGravitinoConnectorWithMetalakeCatalogName;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Nested;
 
 public class TestGravitinoConnector469 {
@@ -45,7 +44,6 @@ protected DistributedQueryRunner createTrinoQueryRunner() throws Exception {
   }
 
   @Nested
-  @Disabled
   class MultiMetalake extends TestGravitinoConnectorWithMetalakeCatalogName {
     @Override
     protected GravitinoPlugin createGravitinoPlugin(GravitinoAdminClient client) {

trino-connector/trino-connector-473-478/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory478.java

@@ -39,16 +39,6 @@ protected int getMaxSupportTrinoSpiVersion() {
     return 478;
   }
 
-  @Override
-  protected String getTrinoCatalogName(String metalake, String catalog) {
-    return "\"" + metalake + "." + catalog + "\"";
-  }
-
-  @Override
-  protected boolean supportCatalogNameWithMetalake() {
-    return false;
-  }
-
   @Override
   protected GravitinoConnector createConnector(CatalogConnectorContext connectorContext) {
     return new GravitinoConnector478(connectorContext);

trino-connector/trino-connector-473-478/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnector478.java

@@ -23,7 +23,6 @@
 import io.trino.Session;
 import io.trino.testing.DistributedQueryRunner;
 import org.apache.gravitino.client.GravitinoAdminClient;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Nested;
 
 public class TestGravitinoConnector478 {
@@ -42,7 +41,6 @@ protected DistributedQueryRunner createTrinoQueryRunner() throws Exception {
   }
 
   @Nested
-  @Disabled
   class MultiMetalake extends TestGravitinoConnectorWithMetalakeCatalogName {
     @Override
     protected GravitinoPlugin createGravitinoPlugin(GravitinoAdminClient client) {
@@ -54,5 +52,15 @@ protected DistributedQueryRunner createTrinoQueryRunner() throws Exception {
       Session session = testSessionBuilder().setCatalog("gravitino").build();
       return DistributedQueryRunner.builder(session).setWorkerCount(0).build();
     }
+
+    @Override
+    protected String getTrinoCliCatalogName(String metalake, String catalog) {
+      return metalake + "." + catalog;
+    }
+
+    @Override
+    protected String getTrinoSqlCatalogName(String metalakeName, String catalogName) {
+      return "\"" + metalakeName + "." + catalogName + "\"";
+    }
   }
 }

trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java

@@ -161,11 +161,11 @@ private void checkTrinoSpiVersion(ConnectorContext context, GravitinoConfig conf
 
     // check catalog name with metalake are supported in this trino version
     if (!config.singleMetalakeMode() && !supportCatalogNameWithMetalake()) {
-      String errmsg =
-          String.format(
-              "The trino-connector-%s-%s does not support catalog name with metalake.",
-              getMinSupportTrinoSpiVersion(), getMaxSupportTrinoSpiVersion());
-      throw new TrinoException(GravitinoErrorCode.GRAVITINO_UNSUPPORTED_TRINO_VERSION, errmsg);
+      LOG.warn(
+          "The trino-connector-{}-{} does not fully support catalog name with metalake. "
+              + "The DROP CATALOG operation may not work correctly in multi-metalake mode.",
+          getMinSupportTrinoSpiVersion(),
+          getMaxSupportTrinoSpiVersion());
     }
 
     // skip version validation