yuqi1129 commented on code in PR #12547:
URL: https://github.com/apache/gravitino/pull/12547#discussion_r3841961350
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -212,67 +330,170 @@ 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("Failed to list catalogs in metalake {}.", metalake.name(), 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.
+ 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 catalogs. catalogs: {}.", metalake.name(),
catalogNames);
+ LOG.debug("Load metalake {}'s catalogs. catalogs: {}.", 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("Failed to remove catalog {}.", entry.getKey(), e);
+ // 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);
- GravitinoCatalog gravitinoCatalog = new
GravitinoCatalog(metalake.name(), catalog);
- 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 {} in metalake {}:
{}",
- catalogName,
- metalake.name(),
- e.getMessage());
- } catch (Exception e) {
- LOG.error(
- "Failed to load metalake {}'s catalog {}.",
metalake.name(), catalogName, e);
- }
+ for (String catalogName : catalogNames) {
+ String trinoCatalogName = getTrinoCatalogName(metalakeName, catalogName);
+ // 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);
Review Comment:
If a catalog that has been registered and can be seen via `SHOW CATALOGs`
fails to load catalog info in a
certain rounds, the status will be marked as `FAILED`
However, as the doc says
- `REGISTERED` means it can be shown by `SHOW CATALOG`
- `FAILED` means it fails to register.
So we need to handle the case where a catalog has been registered
successfully and fails to refresh.
--
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]
