This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 62240804b7 [#13697] improvement(trino-connector): Load catalogs with a
single detailed listing on refresh (#13698)
62240804b7 is described below
commit 62240804b7c4a79da147b7f659b840ff50ce934a
Author: Yuhui <[email protected]>
AuthorDate: Thu Oct 8 20:48:11 2026 +0800
[#13697] improvement(trino-connector): Load catalogs with a single detailed
listing on refresh (#13698)
### What changes were proposed in this pull request?
`CatalogConnectorManager` now reads all catalogs of a metalake with one
`listCatalogsInfo()` call per refresh, instead of `listCatalogs()`
followed by `loadCatalog()` for each catalog.
### Why are the changes needed?
Each refresh sent 1 + N requests per metalake and wrote N + 1 audit
records, even when no catalog had changed.
Fix: #13697
### Does this PR introduce _any_ user-facing change?
A catalog that the server cannot resolve, or whose type the client does
not know, now fails the listing of its metalake, which is reported as a
metalake error in the load status, instead of failing only that catalog.
Registered catalogs stay registered meanwhile.
### How was this patch tested?
Updated unit tests in `TestCatalogConnectorManager`. Existing tests
pass.
---
.../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 a7280cf53f..1446d97dbc 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
@@ -27,6 +27,7 @@ import io.trino.spi.connector.ConnectorContext;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -454,9 +455,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.
@@ -469,7 +474,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)) {
@@ -562,7 +567,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);
// Registration deliberately carries only the visible properties. The
resolved secrets are
// added by each node in createCatalogConnectorContext(), so that they
never reach the
// CREATE CATALOG statement, the catalog properties file Trino
persists from it, or
@@ -602,14 +607,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 2586d526dd..3a848acd57 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
@@ -263,15 +263,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);
@@ -324,7 +323,7 @@ public class TestCatalogConnectorManager {
Mockito.doThrow(new RuntimeException("Connection refused"))
.when(fixture.metalake)
- .listCatalogs();
+ .listCatalogsInfo();
manager.loadMetalakeSync();
Map<String, String> metalakeErrors = manager.getMetalakeErrors();
@@ -453,7 +452,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.
@@ -488,7 +487,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
@@ -544,9 +543,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
@@ -558,21 +555,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"));
}
@@ -634,6 +629,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();
@@ -650,7 +667,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");
@@ -797,7 +814,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());
@@ -830,7 +847,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();
@@ -878,8 +895,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();
@@ -969,7 +985,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();
@@ -1151,7 +1167,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());
@@ -1171,7 +1187,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"));
@@ -1691,7 +1707,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) {
@@ -1728,7 +1744,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;
