yuqi1129 commented on code in PR #12547:
URL: https://github.com/apache/gravitino/pull/12547#discussion_r3922031713
##########
docs/trino-connector/supported-catalog.md:
##########
@@ -80,6 +80,68 @@ The result is like:
gt_hive | hive |
{gravitino.bypass.hive.metastore.client.capability.check=false,
metastore.uris=thrift://trino-ci-hive:9083}
```
+Check catalog registration status:
+
+`gravitino.system.catalog` lists the relational catalogs the Gravitino server
knows about, minus any
+that match `gravitino.trino.skip-catalog-patterns`. A catalog listed there is
not necessarily usable
+in Trino: registering it is a separate step that can fail.
`gravitino.system.catalog_status` covers
+every catalog the connector considered, including the ones `catalog` filters
out, and says why each
+one is or is not registered.
+
+```sql
+select catalog_name, status, last_error from gravitino.system.catalog_status;
+```
+
+The result is like:
+
+```test
+ catalog_name | status | last_error
+--------------+------------+-------------------------------------------------
+ gt_hive | REGISTERED | NULL
+ gt_iceberg | FAILED | Access Denied: Cannot create catalog gt_iceberg
+ gt_files | UNSUPPORTED| Only relational catalogs are supported, the
catalog type is FILESET
+```
+
+| Column | Description
|
+|----------------------|----------------------------------------------------------------------------------------------|
Review Comment:
Format the table.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -232,67 +401,197 @@ public GravitinoMetalake retrieveMetalake(String
metalakeName) {
}
private void loadCatalogs(GravitinoMetalake metalake) {
- List<String> catalogNames;
+ String metalakeName = metalake.name();
+ String[] allCatalogNames;
try {
- catalogNames =
- Arrays.stream(metalake.listCatalogs())
- .filter(id -> !skipCatalog(getTrinoCatalogName(metalake.name(),
id)))
- .collect(Collectors.toList());
+ allCatalogNames = metalake.listCatalogs();
} catch (Exception e) {
- LOG.error(e, "Failed to list catalogs in metalake %s.", metalake.name());
+ // 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.
+ recordMetalakeError(metalakeName, e);
return;
}
+ metalakeErrors.remove(metalakeName);
+
+ // The Trino names of every catalog the Gravitino server currently
reports, including the
+ // catalogs that are intentionally not registered.
+ Set<String> presentTrinoNames = new HashSet<>();
+ List<String> catalogNames = new ArrayList<>();
+ for (String catalogName : allCatalogNames) {
+ String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
+ presentTrinoNames.add(trinoCatalogName);
+ if (skipCatalog(trinoCatalogName)) {
+ recordCatalogState(
+ CatalogRegistrationState.skipped(
+ metalakeName,
+ catalogName,
+ trinoCatalogName,
+ "Matched gravitino.trino.skip-catalog-patterns"),
+ null);
+ continue;
+ }
+ catalogNames.add(catalogName);
+ }
- LOG.debug("Load metalake %s's catalogs. catalogs: %s.", metalake.name(),
catalogNames);
+ LOG.debug("Load metalake %s's catalogs. catalogs: %s.", metalakeName,
catalogNames);
// Delete those catalogs that have been deleted in Gravitino server
- Set<String> catalogNameStrings =
- catalogNames.stream()
- .map(id -> getTrinoCatalogName(metalake.name(), id))
- .collect(Collectors.toSet());
+ Set<String> catalogNameStrings = new HashSet<>();
+ for (String catalogName : catalogNames) {
+ catalogNameStrings.add(getTrinoCatalogName(metalakeName, catalogName));
+ }
for (Map.Entry<String, CatalogConnectorContext> entry :
catalogConnectors.entrySet()) {
if (!catalogNameStrings.contains(entry.getKey())
&&
// Skip the catalog doesn't belong to this metalake.
- entry.getValue().getMetalake().name().equals(metalake.name())) {
+ entry.getValue().getMetalake().name().equals(metalakeName)) {
try {
unloadCatalog(entry.getValue().getCatalog());
} catch (Exception e) {
- LOG.error(e, "Failed to remove catalog %s.", entry.getKey());
+ // The catalog is gone from Gravitino but is still registered in
Trino. Record it, or
+ // the pruning below would drop the row and the table would report
nothing at all about
+ // a catalog that still shows up in SHOW CATALOGS.
+ GravitinoCatalog catalog = entry.getValue().getCatalog();
+ recordCatalogState(
+ CatalogRegistrationState.failed(
+ metalakeName,
+ catalog.getName(),
+ entry.getKey(),
+ catalog.getProvider(),
+ "The catalog was deleted in Gravitino but could not be
unregistered from Trino: "
+ + toErrorMessage(e)),
+ e);
}
}
}
+ // Drop the states of catalogs that no longer exist in the Gravitino
server, including the
+ // states of catalogs that never had a connector. A catalog whose
connector could not be
+ // removed from Trino is kept, so that its failure stays visible for as
long as it is real.
+ catalogStates
+ .values()
+ .removeIf(
+ state ->
+ state.getMetalake().equals(metalakeName)
+ && !presentTrinoNames.contains(state.getTrinoCatalogName())
+ &&
!catalogConnectors.containsKey(state.getTrinoCatalogName()));
+
// Load new catalogs belows to the metalake.
- catalogNames.stream()
- .forEach(
- (String catalogName) -> {
- try {
- Catalog catalog = metalake.loadCatalog(catalogName);
- Map<String, String> properties = propsWithSecrets(catalog);
- GravitinoCatalog gravitinoCatalog =
- new GravitinoCatalog(metalake.name(), catalog, properties);
- if
(catalogConnectors.containsKey(getTrinoCatalogName(gravitinoCatalog))) {
- // Reload catalogs that have been updated in Gravitino
server.
- reloadCatalog(gravitinoCatalog);
- } else {
- if (catalog.type() == Catalog.Type.RELATIONAL
- && catalogConnectorFactory
- .getSupportedCatalogProviders()
- .contains(gravitinoCatalog.getProvider())) {
- loadCatalog(gravitinoCatalog);
- }
- }
- } catch (UnsupportedOperationException e) {
- LOG.warn(
- "Unsupported catalog type for catalog %s in metalake %s:
%s",
- catalogName, metalake.name(), e.getMessage());
- } catch (Exception e) {
- LOG.error(
- e, "Failed to load metalake %s's catalog %s.",
metalake.name(), catalogName);
- }
+ for (String catalogName : catalogNames) {
+ String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
+ // Known before the catalog is even loaded, since it only depends on the
name.
+ boolean alreadyRegistered =
catalogConnectors.containsKey(trinoCatalogName);
+ // 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);
+ // 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
+ // anything that quotes either of them back.
+ GravitinoCatalog gravitinoCatalog =
+ new GravitinoCatalog(metalakeName, catalog, visibleProps(catalog));
+ provider = gravitinoCatalog.getProvider();
+ if (alreadyRegistered) {
+ // Reload catalogs that have been updated in Gravitino server.
+ reloadCatalog(gravitinoCatalog);
+ recordCatalogState(
+ CatalogRegistrationState.succeeded(gravitinoCatalog,
trinoCatalogName), null);
+ } else if (catalog.type() != Catalog.Type.RELATIONAL) {
Review Comment:
[Confirmed correctness gap] The eligibility checks occur only after the
alreadyRegistered branch. If a live catalog is deleted and recreated under the
same name with an unsupported type/provider, it is reloaded (or treated as an
unchanged REGISTERED catalog) instead of transitioning to UNSUPPORTED.
testUnsupportedCatalogKeepsLastSuccessTime does not exercise this path because
it never creates a live connector context, so alreadyRegistered remains false.
Please determine the desired state before choosing reload/register/unload and
add live REGISTERED -> UNSUPPORTED transition coverage.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -232,67 +401,197 @@ public GravitinoMetalake retrieveMetalake(String
metalakeName) {
}
private void loadCatalogs(GravitinoMetalake metalake) {
- List<String> catalogNames;
+ String metalakeName = metalake.name();
+ String[] allCatalogNames;
try {
- catalogNames =
- Arrays.stream(metalake.listCatalogs())
- .filter(id -> !skipCatalog(getTrinoCatalogName(metalake.name(),
id)))
- .collect(Collectors.toList());
+ allCatalogNames = metalake.listCatalogs();
} catch (Exception e) {
- LOG.error(e, "Failed to list catalogs in metalake %s.", metalake.name());
+ // 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.
+ recordMetalakeError(metalakeName, e);
return;
}
+ metalakeErrors.remove(metalakeName);
+
+ // The Trino names of every catalog the Gravitino server currently
reports, including the
+ // catalogs that are intentionally not registered.
+ Set<String> presentTrinoNames = new HashSet<>();
+ List<String> catalogNames = new ArrayList<>();
+ for (String catalogName : allCatalogNames) {
+ String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
+ presentTrinoNames.add(trinoCatalogName);
+ if (skipCatalog(trinoCatalogName)) {
+ recordCatalogState(
+ CatalogRegistrationState.skipped(
+ metalakeName,
+ catalogName,
+ trinoCatalogName,
+ "Matched gravitino.trino.skip-catalog-patterns"),
+ null);
+ continue;
+ }
+ catalogNames.add(catalogName);
+ }
- LOG.debug("Load metalake %s's catalogs. catalogs: %s.", metalake.name(),
catalogNames);
+ LOG.debug("Load metalake %s's catalogs. catalogs: %s.", metalakeName,
catalogNames);
// Delete those catalogs that have been deleted in Gravitino server
- Set<String> catalogNameStrings =
- catalogNames.stream()
- .map(id -> getTrinoCatalogName(metalake.name(), id))
- .collect(Collectors.toSet());
+ Set<String> catalogNameStrings = new HashSet<>();
+ for (String catalogName : catalogNames) {
+ catalogNameStrings.add(getTrinoCatalogName(metalakeName, catalogName));
+ }
for (Map.Entry<String, CatalogConnectorContext> entry :
catalogConnectors.entrySet()) {
if (!catalogNameStrings.contains(entry.getKey())
&&
// Skip the catalog doesn't belong to this metalake.
- entry.getValue().getMetalake().name().equals(metalake.name())) {
+ entry.getValue().getMetalake().name().equals(metalakeName)) {
try {
unloadCatalog(entry.getValue().getCatalog());
} catch (Exception e) {
- LOG.error(e, "Failed to remove catalog %s.", entry.getKey());
+ // The catalog is gone from Gravitino but is still registered in
Trino. Record it, or
+ // the pruning below would drop the row and the table would report
nothing at all about
+ // a catalog that still shows up in SHOW CATALOGS.
+ GravitinoCatalog catalog = entry.getValue().getCatalog();
+ recordCatalogState(
+ CatalogRegistrationState.failed(
+ metalakeName,
+ catalog.getName(),
+ entry.getKey(),
+ catalog.getProvider(),
+ "The catalog was deleted in Gravitino but could not be
unregistered from Trino: "
Review Comment:
This unload branch handles both deleted catalogs and live catalogs that now
match skip-catalog-patterns, because catalogNameStrings excludes skipped names.
If unregistering a skipped catalog fails, this overwrites SKIPPED with FAILED
and incorrectly says the catalog was deleted in Gravitino. presentTrinoNames
can distinguish the two causes. Please add a live REGISTERED -> SKIPPED test
for both successful and failed unregister.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
