This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 81fa21ad27 [Cherry-pick to branch-1.3] [#13697]
improvement(trino-connector): Load catalogs with a single detailed listing on
refresh (#13698) (#13704)
81fa21ad27 is described below
commit 81fa21ad2751b8393581138085f50ccdd1df64a5
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 9 15:08:29 2026 +0800
[Cherry-pick to branch-1.3] [#13697] improvement(trino-connector): Load
catalogs with a single detailed listing on refresh (#13698) (#13704)
**Cherry-pick Information:**
- Original commit: 62240804b7c4a79da147b7f659b840ff50ce934a
- Target branch: `branch-1.3`
- Status: ⚠️ **Has conflicts - manual resolution required**
---------
Co-authored-by: Yuhui <[email protected]>
Co-authored-by: diqiu50 <[email protected]>
---
.../connector/catalog/CatalogConnectorManager.java | 21 +++----
.../catalog/TestCatalogConnectorManager.java | 72 +++++++++++++---------
2 files changed, 53 insertions(+), 40 deletions(-)
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
index d7fed43f55..cdc4cac2ba 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java
@@ -26,6 +26,7 @@ import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorContext;
import java.util.ArrayList;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -450,9 +451,13 @@ public class CatalogConnectorManager {
private void loadCatalogs(GravitinoMetalake metalake) {
String metalakeName = metalake.name();
- String[] allCatalogNames;
+ // Keyed by name in listing order. Read with their details in a single
request, so that a
+ // refresh costs the server one call instead of one per catalog.
+ Map<String, Catalog> allCatalogs = new LinkedHashMap<>();
try {
- allCatalogNames = metalake.listCatalogs();
+ for (Catalog catalog : metalake.listCatalogsInfo()) {
+ allCatalogs.put(catalog.name(), catalog);
+ }
} catch (Exception e) {
// Keep the existing catalog states untouched, a transient listing
failure must not turn
// healthy catalogs into failed ones. The load status system table
reports the cause.
@@ -465,7 +470,7 @@ public class CatalogConnectorManager {
// catalogs that are intentionally not registered.
Set<String> presentTrinoNames = new HashSet<>();
List<String> catalogNames = new ArrayList<>();
- for (String catalogName : allCatalogNames) {
+ for (String catalogName : allCatalogs.keySet()) {
String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
presentTrinoNames.add(trinoCatalogName);
if (skipCatalog(trinoCatalogName)) {
@@ -558,7 +563,7 @@ public class CatalogConnectorManager {
// Tracked outside the try so that a failure can still report the
provider it knows about.
String provider = null;
try {
- Catalog catalog = metalake.loadCatalog(catalogName);
+ Catalog catalog = allCatalogs.get(catalogName);
GravitinoCatalog gravitinoCatalog = new GravitinoCatalog(metalakeName,
catalog);
provider = gravitinoCatalog.getProvider();
// Checked before the already-registered path: a catalog can be
dropped and recreated
@@ -593,14 +598,6 @@ public class CatalogConnectorManager {
recordCatalogState(
CatalogRegistrationState.succeeded(gravitinoCatalog,
trinoCatalogName), null);
}
- } catch (UnsupportedOperationException e) {
- // The client library does not recognize this catalog's type, e.g.
DTOConverters.toCatalog
- // throws for a type it cannot map. This is the same "we know about
it, we just don't
- // support it" case as the type/provider checks above, not a
registration failure.
- recordCatalogState(
- CatalogRegistrationState.unsupported(
- metalakeName, catalogName, trinoCatalogName, provider,
toErrorMessage(e)),
- e);
} catch (Exception e) {
// reloadCatalog() unregisters the old connector before
re-registering; if the
// re-register then fails, the catalog is no longer in
catalogConnectors even though
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
index 8bea849ee3..2e904d0a3a 100644
---
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/TestCatalogConnectorManager.java
@@ -258,15 +258,14 @@ public class TestCatalogConnectorManager {
@Test
public void testFailureBeforeProviderIsKnownKeepsPreviousProvider() throws
Exception {
LoadFixture fixture = new LoadFixture();
- fixture.withCatalogs(mockCatalog("memory", "memory",
Catalog.Type.RELATIONAL));
+ Catalog catalog = mockCatalog("memory", "memory", Catalog.Type.RELATIONAL);
+ fixture.withCatalogs(catalog);
CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
manager.loadMetalakeSync();
assertEquals("memory", singleState(manager).getProvider());
// The next attempt fails before the provider can even be read off the
catalog.
- Mockito.doThrow(new RuntimeException("Connection reset"))
- .when(fixture.metalake)
- .loadCatalog("memory");
+ when(catalog.properties()).thenThrow(new RuntimeException("simulated:
invalid properties"));
manager.loadMetalakeSync();
CatalogRegistrationState state = singleState(manager);
@@ -319,7 +318,7 @@ public class TestCatalogConnectorManager {
Mockito.doThrow(new RuntimeException("Connection refused"))
.when(fixture.metalake)
- .listCatalogs();
+ .listCatalogsInfo();
manager.loadMetalakeSync();
Map<String, String> metalakeErrors = manager.getMetalakeErrors();
@@ -448,7 +447,7 @@ public class TestCatalogConnectorManager {
Mockito.doThrow(new RuntimeException("Connection refused"))
.when(fixture.metalake)
- .listCatalogs();
+ .listCatalogsInfo();
manager.loadMetalakeSync();
assertEquals(1, manager.getMetalakeErrors().size());
// A metalake that cannot be listed is not a healthy loop.
@@ -483,7 +482,7 @@ public class TestCatalogConnectorManager {
doThrow(new TrinoException(GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR,
"Access Denied"))
.when(fixture.catalogRegister)
.unregisterCatalog(any());
- Mockito.doReturn(new String[0]).when(fixture.metalake).listCatalogs();
+ Mockito.doReturn(new Catalog[0]).when(fixture.metalake).listCatalogsInfo();
manager.loadMetalakeSync();
// The catalog is gone from Gravitino but still registered in Trino: the
state must say so
@@ -539,9 +538,7 @@ public class TestCatalogConnectorManager {
when(context.getCatalog())
.thenReturn(new GravitinoCatalog("test", "memory", "memory",
ImmutableMap.of(), 0L));
- Mockito.doThrow(new RESTException("simulated: the server rejected the
load"))
- .when(fixture.metalake)
- .loadCatalog("memory");
+ when(catalog.properties()).thenThrow(new RuntimeException("simulated:
invalid properties"));
manager.loadMetalakeSync();
// The catalog is still registered and queryable, so a failed refresh must
not flip it to
@@ -553,21 +550,19 @@ public class TestCatalogConnectorManager {
}
@Test
- public void testUnknownCatalogTypeIsRecordedAsUnsupported() throws Exception
{
+ public void testUnknownCatalogTypeFailsTheMetalakeListing() throws Exception
{
LoadFixture fixture = new LoadFixture();
- fixture.withCatalogs(mockCatalog("memory", "memory",
Catalog.Type.RELATIONAL));
CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
- // The client library cannot map this catalog's type, e.g.
DTOConverters.toCatalog throwing
- // for a type it does not know. That is a "not supported", not a
registration failure.
+ // The client library cannot map a listed catalog's type, e.g.
DTOConverters.toCatalog throwing
+ // for a type it does not know, which fails the listing of the whole
metalake.
Mockito.doThrow(new UnsupportedOperationException("Unsupported catalog
type: UNKNOWN"))
.when(fixture.metalake)
- .loadCatalog("memory");
+ .listCatalogsInfo();
manager.loadMetalakeSync();
- CatalogRegistrationState state = singleState(manager);
- assertEquals(CatalogRegistrationState.Status.UNSUPPORTED,
state.getStatus());
- assertTrue(state.getLastError().contains("Unsupported catalog type"));
+ assertTrue(manager.getMetalakeErrors().get("test").contains("Unsupported
catalog type"));
+ assertTrue(manager.getCatalogRegistrationStates().isEmpty());
assertFalse(manager.catalogConnectorExist("memory"));
}
@@ -629,6 +624,28 @@ public class TestCatalogConnectorManager {
assertNull(state.getLastError());
}
+ @Test
+ public void testCatalogsAreLoadedWithOneDetailedListing() throws Exception {
+ LoadFixture fixture = new LoadFixture();
+ Catalog first = mockCatalog("a", "memory", Catalog.Type.RELATIONAL);
+ Catalog second = mockCatalog("b", "memory", Catalog.Type.RELATIONAL);
+ Mockito.doReturn(new Catalog[] {first,
second}).when(fixture.metalake).listCatalogsInfo();
+ CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
+
+ manager.loadMetalakeSync();
+ manager.loadMetalakeSync();
+
+ // Every refresh reads all catalogs with one request instead of one
request per catalog.
+ verify(fixture.metalake, times(2)).listCatalogsInfo();
+ verify(fixture.metalake, never()).listCatalogs();
+ verify(fixture.metalake, never()).loadCatalog(any());
+ List<CatalogRegistrationState> states =
manager.getCatalogRegistrationStates();
+ assertEquals(2, states.size());
+ for (CatalogRegistrationState state : states) {
+ assertEquals(CatalogRegistrationState.Status.REGISTERED,
state.getStatus());
+ }
+ }
+
@Test
public void testDeletedCatalogWithConnectorIsRemoved() throws Exception {
LoadFixture fixture = new LoadFixture();
@@ -645,7 +662,7 @@ public class TestCatalogConnectorManager {
.thenReturn(new GravitinoCatalog("test", "memory", "memory",
ImmutableMap.of(), 0L));
// The catalog is deleted in Gravitino and unregisters cleanly from Trino.
- Mockito.doReturn(new String[0]).when(fixture.metalake).listCatalogs();
+ Mockito.doReturn(new Catalog[0]).when(fixture.metalake).listCatalogsInfo();
manager.loadMetalakeSync();
verify(fixture.catalogRegister, times(1)).unregisterCatalog("memory");
@@ -792,7 +809,7 @@ public class TestCatalogConnectorManager {
// The losing catalog is deleted in Gravitino. The winner still holds the
Trino name, which
// must not keep the loser's row alive.
- Mockito.doReturn(new String[0]).when(dev).listCatalogs();
+ Mockito.doReturn(new Catalog[0]).when(dev).listCatalogsInfo();
manager.loadMetalakeSync();
assertTrue(manager.getCatalogRegistrationStates("dev").isEmpty());
@@ -825,7 +842,7 @@ public class TestCatalogConnectorManager {
assertNull(manager.getCatalogConnector("dev", "memory"));
// The winning catalog is deleted in Gravitino, which frees the Trino name
for the loser.
- Mockito.doReturn(new String[0]).when(fixture.metalake).listCatalogs();
+ Mockito.doReturn(new Catalog[0]).when(fixture.metalake).listCatalogsInfo();
owner[0] = "dev";
manager.loadMetalakeSync();
@@ -872,8 +889,7 @@ public class TestCatalogConnectorManager {
GravitinoMetalake devMetalake = mock(GravitinoMetalake.class);
when(devMetalake.name()).thenReturn("dev");
Catalog devCatalog = mockCatalog("sandbox", "memory",
Catalog.Type.RELATIONAL);
- Mockito.doReturn(new String[]
{"sandbox"}).when(devMetalake).listCatalogs();
- Mockito.doReturn(devCatalog).when(devMetalake).loadCatalog("sandbox");
+ Mockito.doReturn(new Catalog[]
{devCatalog}).when(devMetalake).listCatalogsInfo();
Mockito.doReturn(new GravitinoMetalake[] {fixture.metalake, devMetalake})
.when(fixture.client)
.listMetalakes();
@@ -963,7 +979,7 @@ public class TestCatalogConnectorManager {
LoadFixture fixture = new LoadFixture();
Mockito.doThrow(new RESTException("simulated: listing failed"))
.when(fixture.metalake)
- .listCatalogs();
+ .listCatalogsInfo();
CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
manager.loadMetalakeSync();
@@ -1145,7 +1161,7 @@ public class TestCatalogConnectorManager {
LoadFixture fixture = new LoadFixture();
Mockito.doThrow(new RESTException("simulated: listing failed"))
.when(fixture.metalake)
- .listCatalogs();
+ .listCatalogsInfo();
CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
manager.loadMetalakeSync();
assertFalse(manager.getLoadOutcome().getMetalakeErrors().isEmpty());
@@ -1165,7 +1181,7 @@ public class TestCatalogConnectorManager {
LoadFixture fixture = new LoadFixture();
Mockito.doThrow(new RESTException("simulated: listing failed"))
.when(fixture.metalake)
- .listCatalogs();
+ .listCatalogsInfo();
CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of());
manager.loadMetalakeSync();
assertTrue(manager.describeRegistrationFailure("test",
"memory").contains("listing failed"));
@@ -1560,7 +1576,7 @@ public class TestCatalogConnectorManager {
names[i] = catalogs[i].name();
Mockito.doReturn(catalogs[i]).when(metalake).loadCatalog(names[i]);
}
- Mockito.doReturn(names).when(metalake).listCatalogs();
+ Mockito.doReturn(catalogs).when(metalake).listCatalogsInfo();
}
CatalogConnectorManager createManager(Map<String, String> extraConfig) {
@@ -1597,7 +1613,7 @@ public class TestCatalogConnectorManager {
names[i] = catalogs[i].name();
Mockito.doReturn(catalogs[i]).when(second).loadCatalog(names[i]);
}
- Mockito.doReturn(names).when(second).listCatalogs();
+ Mockito.doReturn(catalogs).when(second).listCatalogsInfo();
Mockito.doReturn(second).when(client).loadMetalake(name);
Mockito.doReturn(new GravitinoMetalake[] {metalake,
second}).when(client).listMetalakes();
return second;
