This is an automated email from the ASF dual-hosted git repository. diqiu50 pushed a commit to branch trino-irc-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 8e01e964063ed446c94b0586188bc3c8e5790f48 Author: diqiu50 <[email protected]> AuthorDate: Fri Aug 21 14:11:08 2026 +0800 [#12546] improvement(trino-connector): Report catalog registration status through system tables Record the registration state of every catalog and the health of the load loop, and expose both through the gravitino.system.catalog_status and gravitino.system.load_status system tables. Errors now report the root cause rather than the outer wrapper. --- docs/trino-connector/supported-catalog.md | 78 +++++ .../testsets/jdbc-mysql/00015_catalog_status.sql | 3 + .../testsets/jdbc-mysql/00015_catalog_status.txt | 3 + .../trino/connector/GravitinoConnectorFactory.java | 19 ++ .../connector/catalog/CatalogConnectorManager.java | 347 ++++++++++++++++++--- .../catalog/CatalogRegistrationState.java | 247 +++++++++++++++ .../connector/system/GravitinoSystemConnector.java | 31 +- .../AlterCatalogStoredProcedure.java | 8 +- .../CreateCatalogStoredProcedure.java | 7 +- .../system/table/GravitinoSystemTable.java | 35 +++ .../table/GravitinoSystemTableCatalogStatus.java | 116 +++++++ .../system/table/GravitinoSystemTableFactory.java | 6 + .../table/GravitinoSystemTableLoadStatus.java | 118 +++++++ .../trino/connector/TestGravitinoConnector.java | 36 ++- .../catalog/TestCatalogConnectorManager.java | 250 +++++++++++++++ .../table/TestGravitinoSystemStatusTables.java | 135 ++++++++ 16 files changed, 1382 insertions(+), 57 deletions(-) diff --git a/docs/trino-connector/supported-catalog.md b/docs/trino-connector/supported-catalog.md index 80a0380f30..8762ce145b 100644 --- a/docs/trino-connector/supported-catalog.md +++ b/docs/trino-connector/supported-catalog.md @@ -80,6 +80,66 @@ 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 catalogs the Gravitino server knows about. A catalog listed +there is not necessarily usable in Trino: registering it is a separate step that can fail. When a +catalog does not show up in `SHOW CATALOGS`, `gravitino.system.catalog_status` says why. + +```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 | +|----------------------|----------------------------------------------------------------------------------------------| +| `metalake` | The metalake the catalog belongs to. | +| `catalog_name` | The name of the catalog in Gravitino. | +| `trino_catalog_name` | The name the catalog is registered under in Trino, as it appears in `SHOW CATALOGS`. | +| `provider` | The catalog provider, for example `hive` or `lakehouse-iceberg`. | +| `status` | One of `REGISTERED`, `FAILED`, `UNSUPPORTED` or `SKIPPED`. See the table below. | +| `last_error` | The reason the catalog is not registered, `NULL` when it is. | +| `last_attempt_time` | When the catalog was last processed, as an ISO-8601 UTC timestamp. | +| `last_success_time` | When the catalog was last registered successfully, `NULL` if it never was. | +| `failure_count` | The number of consecutive failed attempts, `0` when the last attempt succeeded. | + +| Status | Meaning | +|---------------|-----------------------------------------------------------------------------------------------| +| `REGISTERED` | The catalog is registered in Trino and appears in `SHOW CATALOGS`. | +| `FAILED` | The last registration attempt failed, `last_error` carries the reason. Retried every refresh. | +| `UNSUPPORTED` | The catalog is not relational, or its provider is not supported by the connector. | +| `SKIPPED` | The catalog matches `gravitino.trino.skip-catalog-patterns` and is deliberately not registered. | + +A failure that stops the connector before it can list catalogs at all, such as an unreachable +Gravitino server, leaves no row to attach itself to. `gravitino.system.load_status` reports the +health of the loop itself, and always has exactly one row. + +```sql +select * from gravitino.system.load_status; +``` + +| Column | Description | +|------------------------|-------------------------------------------------------------------------------------------| +| `trino_started` | Whether the connector can reach the Trino server over JDBC. No catalog is registered until it can. | +| `last_attempt_time` | When the loop last ran, as an ISO-8601 UTC timestamp. | +| `last_success_time` | When the loop last completed successfully, `NULL` if it never did. | +| `consecutive_failures` | The number of consecutive failed runs, `0` when the last run succeeded. | +| `last_error` | The reason the last run failed, `NULL` when it succeeded. | +| `metalake_errors` | A JSON map of metalake name to its last error, `NULL` when every metalake loaded. | + +Both tables are served by the coordinator and reflect the last refresh, which runs every +`gravitino.metadata.refresh-interval-second` seconds (10 by default). A catalog created moments ago +may not have been processed yet. + Example: Run the following SQL to create a catalog named `mysql` with `jdbc-mysql` provider. @@ -181,3 +241,21 @@ Hive does not support `TIME` data type. | Struct | ROW | For more about Trino data types, refer to [Trino data types](https://trino.io/docs/current/language/types.html) and Gravitino data types, refer to [Gravitino data types](../tables-and-views.md#table-column-type). + +## Troubleshooting + +Registration happens in the background, so a catalog that fails to register simply never appears in +`SHOW CATALOGS`. Start from `gravitino.system.catalog_status` rather than the coordinator log. + +| Symptom | Likely cause | +|--------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------------------| +| A catalog is missing from `SHOW CATALOGS` | Query `gravitino.system.catalog_status` and read `status` and `last_error`, then follow the rows below | +| `status = FAILED`, `last_error` mentions `Access Denied` | The `gravitino.trino.user` lacks a Trino system role permitted to run `CREATE CATALOG` | +| `status = FAILED`, `last_error` mentions a configuration property | A `trino.bypass.` property is not accepted by the underlying Trino connector | +| `status = UNSUPPORTED` | The catalog is not relational, or its provider is outside the supported list. `last_error` names the supported providers | +| `status = SKIPPED` | The catalog matches `gravitino.trino.skip-catalog-patterns` | +| The catalog has no row in `catalog_status` at all | The load loop never reached it. Check `gravitino.system.load_status` | +| `load_status.trino_started = false` | The connector cannot reach Trino over JDBC. Check `discovery.uri`, `gravitino.trino.user` and `gravitino.trino.password` | +| `load_status.last_error` mentions connection refused | The Gravitino server is unreachable. Check `gravitino.uri` | +| `load_status.metalake_errors` names a metalake | That metalake could not be listed, the other metalakes are unaffected | +| Querying `gravitino.system.catalog_status` itself fails | The entry catalog did not initialise. The error is reported when the entry catalog is created, check the Trino server log at startup | diff --git a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_catalog_status.sql b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_catalog_status.sql new file mode 100644 index 0000000000..53e3f32237 --- /dev/null +++ b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_catalog_status.sql @@ -0,0 +1,3 @@ +select catalog_name, status, failure_count from gravitino.system.catalog_status where catalog_name = 'gt_mysql'; + +select cast(trino_started as varchar), consecutive_failures from gravitino.system.load_status; diff --git a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_catalog_status.txt b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_catalog_status.txt new file mode 100644 index 0000000000..dce3462b9c --- /dev/null +++ b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_catalog_status.txt @@ -0,0 +1,3 @@ +"gt_mysql","REGISTERED","0" + +"true","0" diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java index a748c0ab58..917d9f3928 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java @@ -25,6 +25,7 @@ import static org.apache.gravitino.trino.connector.GravitinoErrorCode.GRAVITINO_ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.base.Strings; +import io.trino.spi.HostAddress; import io.trino.spi.TrinoException; import io.trino.spi.connector.Connector; import io.trino.spi.connector.ConnectorContext; @@ -132,6 +133,13 @@ public class GravitinoConnectorFactory implements ConnectorFactory { catalogConnectorManager.config(config, client); gravitinoSystemTableFactory = new GravitinoSystemTableFactory(catalogConnectorManager); + if (isCoordinator(trinoConnectorContext)) { + // Pin the system table splits here: the registration state the system tables report + // is only recorded on the coordinator by the load loop started below. Starting the + // manager remains deferred until the static connector supplies its JDBC settings. + GravitinoSystemConnector.Split.setCoordinatorAddress( + getCurrentNodeAddress(trinoConnectorContext)); + } } // The `trino.jdbc.*` settings that CatalogRegister needs to connect back to the @@ -289,6 +297,17 @@ public class GravitinoConnectorFactory implements ConnectorFactory { return connectorContext.getNodeManager().getCurrentNode().isCoordinator(); } + /** + * Retrieves the address of the Trino node this connector is running on. + * + * @param connectorContext the Trino connector context + * @return the host and port of the current node + */ + @SuppressWarnings("deprecation") + protected HostAddress getCurrentNodeAddress(ConnectorContext connectorContext) { + return connectorContext.getNodeManager().getCurrentNode().getHostAndPort(); + } + private CatalogConnectorFactory createCatalogConnectorFactory(GravitinoConfig config) { // Create a CatalogConnectorFactory. If we specify a customized class name for the // CatalogConnectorFactory, 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 380c5214f8..f55752be30 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 @@ -22,18 +22,20 @@ import com.google.common.base.Preconditions; import com.google.common.util.concurrent.ThreadFactoryBuilder; import io.trino.spi.TrinoException; import io.trino.spi.connector.ConnectorContext; -import java.util.Arrays; +import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Pattern; -import java.util.stream.Collectors; +import javax.annotation.Nullable; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.Catalog; import org.apache.gravitino.client.GravitinoAdminClient; @@ -72,6 +74,20 @@ public class CatalogConnectorManager { private final ConcurrentHashMap<String, CatalogConnectorContext> catalogConnectors = new ConcurrentHashMap<>(); + // The registration state of every catalog seen by the load loop, keyed by the Trino catalog + // name. Written only by the load loop thread, read by query threads through the system tables. + private final ConcurrentHashMap<String, CatalogRegistrationState> catalogStates = + new ConcurrentHashMap<>(); + + // The last error reported by each metalake, keyed by the metalake name. + private final ConcurrentHashMap<String, String> metalakeErrors = new ConcurrentHashMap<>(); + + private volatile boolean trinoStarted = false; + private volatile long lastLoadAttemptTimeMs = 0L; + private volatile long lastSuccessfulLoadTimeMs = 0L; + private volatile String lastLoadError = null; + private final AtomicLong consecutiveLoadFailures = new AtomicLong(); + private String targetMetalake; private final Map<String, GravitinoMetalake> metalakes = new ConcurrentHashMap<>(); // Tracks which metalakes' Iceberg REST discovery is currently failing, so a failure is logged @@ -179,11 +195,18 @@ public class CatalogConnectorManager { } private void loadMetalake() { + lastLoadAttemptTimeMs = System.currentTimeMillis(); try { if (!catalogRegister.isTrinoStarted()) { - LOG.info("Waiting for the Trino started."); + String message = "Waiting for the Trino server to start"; + if (!Objects.equals(lastLoadError, message)) { + LOG.info("{}.", message); + } + lastLoadError = message; + trinoStarted = false; return; } + trinoStarted = true; Set<String> usedMetalakes = new HashSet<>(); if (config.singleMetalakeMode()) { @@ -207,12 +230,61 @@ public class CatalogConnectorManager { } loadCatalogs(metalake); } catch (Exception e) { - LOG.error("Load Metalake {} failed.", usedMetalake, e); + recordMetalakeError(usedMetalake, e); } } + + lastSuccessfulLoadTimeMs = System.currentTimeMillis(); + recordLoadSuccess(); } catch (Exception e) { - LOG.error("Error when loading metalake", e); + recordLoadFailure(toErrorMessage(e), e); + } + } + + private void recordLoadSuccess() { + if (lastLoadError != null) { + LOG.info("The Gravitino catalog load loop recovered."); + } + lastLoadError = null; + consecutiveLoadFailures.set(0); + } + + private void recordLoadFailure(String message, Exception cause) { + boolean changed = !Objects.equals(lastLoadError, message); + lastLoadError = message; + consecutiveLoadFailures.incrementAndGet(); + if (changed) { + LOG.error("Failed to load catalogs from the Gravitino server: {}", message, cause); + } else { + LOG.debug("Failed to load catalogs from the Gravitino server: {}", message, cause); + } + } + + private void recordMetalakeError(String metalakeName, Exception cause) { + String message = toErrorMessage(cause); + String previous = metalakeErrors.put(metalakeName, message); + if (!Objects.equals(previous, message)) { + LOG.error("Load metalake {} failed: {}", metalakeName, message, cause); + } else { + LOG.debug("Load metalake {} failed: {}", metalakeName, message, cause); + } + } + + private static String toErrorMessage(Exception e) { + // Report the root cause: the actual reason a registration failed, such as "Access Denied: + // Cannot create catalog", is wrapped in several layers of TrinoException by the time it gets + // here, and the outer messages say nothing a user can act on. Do not use + // GravitinoErrorCode.toSimpleErrorMessage(), it throws a NullPointerException when the + // exception carries no message. + Throwable rootCause = e; + while (rootCause.getCause() != null && rootCause.getCause() != rootCause) { + rootCause = rootCause.getCause(); } + String message = rootCause.getMessage(); + if (StringUtils.isBlank(message)) { + message = e.getMessage(); + } + return StringUtils.isBlank(message) ? e.getClass().getName() : message; } /** @@ -262,30 +334,53 @@ public class CatalogConnectorManager { } 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.notLoaded( + metalakeName, + catalogName, + trinoCatalogName, + null, + CatalogRegistrationState.Status.SKIPPED, + "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) { @@ -294,35 +389,111 @@ public class CatalogConnectorManager { } } + // Drop the states of catalogs that no longer exist in the Gravitino server, including the + // states of catalogs that never had a connector. + catalogStates + .values() + .removeIf( + state -> + state.getMetalake().equals(metalakeName) + && !presentTrinoNames.contains(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); + GravitinoCatalog gravitinoCatalog = new GravitinoCatalog(metalakeName, catalog); + provider = gravitinoCatalog.getProvider(); + if (catalogConnectors.containsKey(trinoCatalogName)) { + // Reload catalogs that have been updated in Gravitino server. + reloadCatalog(gravitinoCatalog); + recordCatalogState( + CatalogRegistrationState.succeeded(gravitinoCatalog, trinoCatalogName), null); + } else if (catalog.type() != Catalog.Type.RELATIONAL) { + recordCatalogState( + CatalogRegistrationState.notLoaded( + metalakeName, + catalogName, + trinoCatalogName, + gravitinoCatalog.getProvider(), + CatalogRegistrationState.Status.UNSUPPORTED, + String.format( + "Only relational catalogs are supported, the catalog type is %s", + catalog.type())), + null); + } else if (!catalogConnectorFactory + .getSupportedCatalogProviders() + .contains(gravitinoCatalog.getProvider())) { + recordCatalogState( + CatalogRegistrationState.notLoaded( + metalakeName, + catalogName, + trinoCatalogName, + gravitinoCatalog.getProvider(), + CatalogRegistrationState.Status.UNSUPPORTED, + String.format( + "The catalog provider %s is not supported, the supported providers are %s", + gravitinoCatalog.getProvider(), + catalogConnectorFactory.getSupportedCatalogProviders())), + null); + } else { + loadCatalog(gravitinoCatalog); + recordCatalogState( + CatalogRegistrationState.succeeded(gravitinoCatalog, trinoCatalogName), null); + } + } catch (UnsupportedOperationException e) { + recordCatalogState( + CatalogRegistrationState.notLoaded( + metalakeName, + catalogName, + trinoCatalogName, + provider, + CatalogRegistrationState.Status.UNSUPPORTED, + toErrorMessage(e)), + null); + } catch (Exception e) { + recordCatalogState( + CatalogRegistrationState.failed( + metalakeName, + catalogName, + trinoCatalogName, + provider, + toErrorMessage(e), + catalogStates.get(trinoCatalogName)), + e); + } + } + } + + private void recordCatalogState(CatalogRegistrationState state, Exception cause) { + CatalogRegistrationState previous = catalogStates.put(state.getTrinoCatalogName(), state); + boolean changed = + previous == null + || previous.getStatus() != state.getStatus() + || !Objects.equals(previous.getLastError(), state.getLastError()); + if (!changed) { + LOG.debug("Catalog {} registration state unchanged: {}", state.getTrinoCatalogName(), state); + return; + } + + if (state.getStatus() == CatalogRegistrationState.Status.REGISTERED) { + LOG.info("Catalog {} is registered in Trino.", state.getTrinoCatalogName()); + } else if (state.getStatus() == CatalogRegistrationState.Status.FAILED) { + LOG.error( + "Failed to register catalog {} in Trino: {}", + state.getTrinoCatalogName(), + state.getLastError(), + cause); + } else { + LOG.warn( + "Catalog {} is not registered in Trino ({}): {}", + state.getTrinoCatalogName(), + state.getStatus(), + state.getLastError()); + } } private void reloadCatalog(GravitinoCatalog catalog) { @@ -355,7 +526,7 @@ public class CatalogConnectorManager { } catch (Exception e) { String message = String.format("Failed to create internal catalog connector. The catalog is: %s", catalog); - LOG.error(message, e); + LOG.debug(message, e); throw new TrinoException( GravitinoErrorCode.GRAVITINO_CREATE_INTERNAL_CONNECTOR_ERROR, message, e); } @@ -365,6 +536,7 @@ public class CatalogConnectorManager { String catalogFullName = getTrinoCatalogName(catalog); catalogRegister.unregisterCatalog(catalogFullName); catalogConnectors.remove(catalogFullName); + catalogStates.remove(catalogFullName); LOG.info( "Remove catalog '{}' in metalake {} successfully.", catalog.getName(), @@ -437,6 +609,89 @@ public class CatalogConnectorManager { return getTrinoCatalogName(catalog.getMetalake(), catalog.getName()); } + /** + * Retrieves a snapshot of the registration state of every Gravitino catalog seen by the load + * loop. + * + * @return the registration states + */ + public List<CatalogRegistrationState> getCatalogRegistrationStates() { + return List.copyOf(catalogStates.values()); + } + + /** + * Checks whether the Trino server has become reachable over JDBC. No catalog can be registered + * before it does. + * + * @return true if the Trino server is started, false otherwise + */ + public boolean isTrinoStarted() { + return trinoStarted; + } + + /** + * Retrieves the time of the last catalog load attempt. + * + * @return the time in milliseconds since the epoch, 0 if the load loop never ran + */ + public long getLastLoadAttemptTimeMs() { + return lastLoadAttemptTimeMs; + } + + /** + * Retrieves the time of the last successful catalog load. + * + * @return the time in milliseconds since the epoch, 0 if the load loop never succeeded + */ + public long getLastSuccessfulLoadTimeMs() { + return lastSuccessfulLoadTimeMs; + } + + /** + * Retrieves the error that made the last catalog load fail. + * + * @return the error message, null if the last load succeeded + */ + @Nullable + public String getLastLoadError() { + return lastLoadError; + } + + /** + * Retrieves the number of consecutive failed catalog loads. + * + * @return the failure count, 0 if the last load succeeded + */ + public long getConsecutiveLoadFailures() { + return consecutiveLoadFailures.get(); + } + + /** + * Retrieves the last error reported by each metalake, keyed by the metalake name. + * + * @return the metalake errors, empty if every metalake was loaded successfully + */ + public Map<String, String> getMetalakeErrors() { + return Map.copyOf(metalakeErrors); + } + + /** + * Describes why a catalog is not registered in Trino, for use in error messages. + * + * @param trinoCatalogName the name the catalog would be registered under in Trino + * @return a human readable explanation + */ + public String describeRegistrationFailure(String trinoCatalogName) { + CatalogRegistrationState state = catalogStates.get(trinoCatalogName); + if (state != null && state.getLastError() != null) { + return String.format("%s: %s", state.getStatus(), state.getLastError()); + } + if (lastLoadError != null) { + return lastLoadError; + } + return "The catalog has not been loaded yet, please retry later."; + } + /** * Retrieves the set of metalakes that have been used. * diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java new file mode 100644 index 0000000000..3ebb5dcb02 --- /dev/null +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java @@ -0,0 +1,247 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.trino.connector.catalog; + +import javax.annotation.Nullable; +import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; + +/** + * The registration state of a single Apache Gravitino catalog in Trino. + * + * <p>Instances are immutable and are always replaced as a whole, so a reader never observes a + * partially updated state. + */ +public final class CatalogRegistrationState { + + /** The registration status of a catalog. */ + public enum Status { + /** The catalog was registered in Trino with a CREATE CATALOG statement. */ + REGISTERED, + /** The last registration attempt failed. */ + FAILED, + /** The catalog matches `gravitino.trino.skip-catalog-patterns` and is not registered. */ + SKIPPED, + /** The catalog type or provider is not supported by the connector. */ + UNSUPPORTED + } + + private final String metalake; + private final String catalogName; + private final String trinoCatalogName; + private final String provider; + private final Status status; + private final String lastError; + private final long lastAttemptTimeMs; + private final long lastSuccessTimeMs; + private final long failureCount; + + private CatalogRegistrationState( + String metalake, + String catalogName, + String trinoCatalogName, + String provider, + Status status, + String lastError, + long lastAttemptTimeMs, + long lastSuccessTimeMs, + long failureCount) { + this.metalake = metalake; + this.catalogName = catalogName; + this.trinoCatalogName = trinoCatalogName; + this.provider = provider; + this.status = status; + this.lastError = lastError; + this.lastAttemptTimeMs = lastAttemptTimeMs; + this.lastSuccessTimeMs = lastSuccessTimeMs; + this.failureCount = failureCount; + } + + /** + * Creates a state for a catalog that was registered in Trino successfully. + * + * @param catalog the Gravitino catalog + * @param trinoCatalogName the name the catalog is registered under in Trino + * @return the registration state + */ + public static CatalogRegistrationState succeeded( + GravitinoCatalog catalog, String trinoCatalogName) { + long now = System.currentTimeMillis(); + return new CatalogRegistrationState( + catalog.getMetalake(), + catalog.getName(), + trinoCatalogName, + catalog.getProvider(), + Status.REGISTERED, + null, + now, + now, + 0); + } + + /** + * Creates a state for a catalog whose registration attempt failed. + * + * @param metalake the name of the metalake the catalog belongs to + * @param catalogName the name of the catalog in Gravitino + * @param trinoCatalogName the name the catalog would be registered under in Trino + * @param provider the catalog provider, null if it could not be determined + * @param error the error that prevented the registration + * @param previous the previous state of the catalog, null if the catalog was never seen before + * @return the registration state + */ + public static CatalogRegistrationState failed( + String metalake, + String catalogName, + String trinoCatalogName, + @Nullable String provider, + String error, + @Nullable CatalogRegistrationState previous) { + return new CatalogRegistrationState( + metalake, + catalogName, + trinoCatalogName, + provider, + Status.FAILED, + error, + System.currentTimeMillis(), + previous == null ? 0 : previous.lastSuccessTimeMs, + previous == null ? 1 : previous.failureCount + 1); + } + + /** + * Creates a state for a catalog that is intentionally not registered in Trino. + * + * @param metalake the name of the metalake the catalog belongs to + * @param catalogName the name of the catalog in Gravitino + * @param trinoCatalogName the name the catalog would be registered under in Trino + * @param provider the catalog provider, null if it could not be determined + * @param status the reason category, either {@link Status#SKIPPED} or {@link Status#UNSUPPORTED} + * @param reason a human readable explanation of why the catalog is not registered + * @return the registration state + */ + public static CatalogRegistrationState notLoaded( + String metalake, + String catalogName, + String trinoCatalogName, + @Nullable String provider, + Status status, + String reason) { + return new CatalogRegistrationState( + metalake, + catalogName, + trinoCatalogName, + provider, + status, + reason, + System.currentTimeMillis(), + 0, + 0); + } + + /** + * Retrieves the name of the metalake the catalog belongs to. + * + * @return the metalake name + */ + public String getMetalake() { + return metalake; + } + + /** + * Retrieves the name of the catalog in Gravitino. + * + * @return the catalog name + */ + public String getCatalogName() { + return catalogName; + } + + /** + * Retrieves the name the catalog is registered under in Trino. + * + * @return the Trino catalog name + */ + public String getTrinoCatalogName() { + return trinoCatalogName; + } + + /** + * Retrieves the catalog provider. + * + * @return the provider, null if it could not be determined + */ + @Nullable + public String getProvider() { + return provider; + } + + /** + * Retrieves the registration status of the catalog. + * + * @return the status + */ + public Status getStatus() { + return status; + } + + /** + * Retrieves the error or the reason why the catalog is not registered. + * + * @return the message, null if the catalog is registered + */ + @Nullable + public String getLastError() { + return lastError; + } + + /** + * Retrieves the time of the last registration attempt. + * + * @return the time in milliseconds since the epoch + */ + public long getLastAttemptTimeMs() { + return lastAttemptTimeMs; + } + + /** + * Retrieves the time of the last successful registration. + * + * @return the time in milliseconds since the epoch, 0 if the catalog was never registered + */ + public long getLastSuccessTimeMs() { + return lastSuccessTimeMs; + } + + /** + * Retrieves the number of consecutive failed registration attempts. + * + * @return the failure count, 0 if the last attempt succeeded + */ + public long getFailureCount() { + return failureCount; + } + + @Override + public String toString() { + return String.format( + "CatalogRegistrationState{metalake=%s, catalog=%s, trinoCatalog=%s, provider=%s," + + " status=%s, lastError=%s, failureCount=%d}", + metalake, catalogName, trinoCatalogName, provider, status, lastError, failureCount); + } +} diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java index 57ee95c0ea..8f5052ec93 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java @@ -129,7 +129,16 @@ public class GravitinoSystemConnector implements Connector { SchemaTableName tableName = ((GravitinoSystemConnectorMetadata.SystemTableHandle) table).getName(); - return createPageSource(GravitinoSystemTableFactory.loadPageData(tableName)); + Page page = GravitinoSystemTableFactory.loadPageData(tableName); + + // Project the page down to the requested columns. Trino only expects the columns it asked + // for, so handing it the whole row breaks any query that is not a SELECT *. + int[] channels = new int[columns.size()]; + for (int i = 0; i < channels.length; i++) { + channels[i] = + ((GravitinoSystemConnectorMetadata.SystemColumnHandle) columns.get(i)).getIndex(); + } + return createPageSource(page.getColumns(channels)); } protected ConnectorPageSource createPageSource(Page page) { @@ -183,14 +192,30 @@ public class GravitinoSystemConnector implements Connector { return tableName; } + // The system table data lives on the coordinator only: the catalog load loop runs there, and + // the registration state it records is never replicated to workers. Splits are built in + // SplitManager.getSplits(), which also runs on the coordinator, so this is always set by the + // time it is read. It stays null on worker JVMs, where the behaviour is unchanged. + private static volatile HostAddress coordinatorAddress; + + /** + * Sets the coordinator address that system table splits are pinned to. + * + * @param address the host and port of the Trino coordinator + */ + public static void setCoordinatorAddress(HostAddress address) { + coordinatorAddress = address; + } + @Override public boolean isRemotelyAccessible() { - return true; + return coordinatorAddress == null; } @Override public List<HostAddress> getAddresses() { - return Collections.emptyList(); + HostAddress address = coordinatorAddress; + return address == null ? Collections.emptyList() : List.of(address); } } diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java index 7126568911..caf469e06f 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/AlterCatalogStoredProcedure.java @@ -137,15 +137,15 @@ public class AlterCatalogStoredProcedure extends GravitinoStoredProcedure { .alterCatalog(catalogName, changes.toArray(changes.toArray(new CatalogChange[0]))); catalogConnectorManager.loadMetalakeSync(); - catalogConnectorContext = - catalogConnectorManager.getCatalogConnector( - catalogConnectorManager.getTrinoCatalogName(metalake, catalogName)); + String trinoCatalogName = catalogConnectorManager.getTrinoCatalogName(metalake, catalogName); + catalogConnectorContext = catalogConnectorManager.getCatalogConnector(trinoCatalogName); if (catalogConnectorContext == null || catalogConnectorContext.getCatalog().getLastModifiedTime() == oldCatalog.getLastModifiedTime()) { throw new TrinoException( GravitinoErrorCode.GRAVITINO_OPERATION_FAILED, - "Update catalog failed due to the reloading process fails"); + "Update catalog failed due to the reloading process fails. " + + catalogConnectorManager.describeRegistrationFailure(trinoCatalogName)); } LOG.info("Alter catalog {} in metalake {} successfully.", catalogName, metalake); diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java index 0de2383d95..4fb8438a3a 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/storedprocedure/CreateCatalogStoredProcedure.java @@ -115,11 +115,12 @@ public class CreateCatalogStoredProcedure extends GravitinoStoredProcedure { catalogName, Catalog.Type.RELATIONAL, provider, "Trino created", properties); catalogConnectorManager.loadMetalakeSync(); - if (!catalogConnectorManager.catalogConnectorExist( - catalogConnectorManager.getTrinoCatalogName(metalake, catalogName))) { + String trinoCatalogName = catalogConnectorManager.getTrinoCatalogName(metalake, catalogName); + if (!catalogConnectorManager.catalogConnectorExist(trinoCatalogName)) { throw new TrinoException( GravitinoErrorCode.GRAVITINO_OPERATION_FAILED, - "Create catalog failed due to the loading process fails"); + "Create catalog failed due to the loading process fails. " + + catalogConnectorManager.describeRegistrationFailure(trinoCatalogName)); } LOG.info("Create catalog {} in metalake {} successfully.", catalogName, metalake); diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTable.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTable.java index eb2328b0b5..136ec9290c 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTable.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTable.java @@ -18,8 +18,13 @@ */ package org.apache.gravitino.trino.connector.system.table; +import static io.trino.spi.type.VarcharType.VARCHAR; + import io.trino.spi.Page; +import io.trino.spi.block.BlockBuilder; import io.trino.spi.connector.ConnectorTableMetadata; +import java.time.Instant; +import javax.annotation.Nullable; /** Gravitino System table interfaces */ public abstract class GravitinoSystemTable { @@ -40,4 +45,34 @@ public abstract class GravitinoSystemTable { * @return a Page object containing all the table data */ public abstract Page loadPageData(); + + /** + * Appends a string to a VARCHAR column, writing a null when the value is absent. + * + * @param builder the column builder to append to + * @param value the value to append, may be null + */ + protected static void writeNullableString(BlockBuilder builder, @Nullable String value) { + if (value == null) { + builder.appendNull(); + } else { + VARCHAR.writeString(builder, value); + } + } + + /** + * Appends a timestamp to a VARCHAR column as an ISO-8601 UTC string, writing a null when the + * timestamp is unset. Timestamps are rendered as strings because the block encoding of + * TimestampType differs across the supported Trino versions. + * + * @param builder the column builder to append to + * @param timeMs the time in milliseconds since the epoch, 0 when unset + */ + protected static void writeTime(BlockBuilder builder, long timeMs) { + if (timeMs == 0) { + builder.appendNull(); + } else { + VARCHAR.writeString(builder, Instant.ofEpochMilli(timeMs).toString()); + } + } } diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java new file mode 100644 index 0000000000..2d2722f506 --- /dev/null +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java @@ -0,0 +1,116 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.trino.connector.system.table; + +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.VarcharType.VARCHAR; + +import io.trino.spi.Page; +import io.trino.spi.block.BlockBuilder; +import io.trino.spi.connector.ColumnMetadata; +import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.SchemaTableName; +import java.util.List; +import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager; +import org.apache.gravitino.trino.connector.catalog.CatalogRegistrationState; + +/** + * An implementation of the catalog status system table. + * + * <p>It reports why every Apache Gravitino catalog is or is not registered in Trino, so that a + * catalog missing from SHOW CATALOGS can be diagnosed without reading the coordinator log. + */ +public class GravitinoSystemTableCatalogStatus extends GravitinoSystemTable { + + /** The name of the catalog status system table. */ + public static final SchemaTableName TABLE_NAME = + new SchemaTableName(SYSTEM_TABLE_SCHEMA_NAME, "catalog_status"); + + private static final ConnectorTableMetadata TABLE_METADATA = + new ConnectorTableMetadata( + TABLE_NAME, + List.of( + ColumnMetadata.builder().setName("metalake").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("catalog_name").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("trino_catalog_name").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("provider").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("status").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("last_error").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("last_attempt_time").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("last_success_time").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("failure_count").setType(BIGINT).build())); + + private final CatalogConnectorManager catalogConnectorManager; + + /** + * Constructs a new GravitinoSystemTableCatalogStatus. + * + * @param catalogConnectorManager the manager for catalog connectors + */ + public GravitinoSystemTableCatalogStatus(CatalogConnectorManager catalogConnectorManager) { + this.catalogConnectorManager = catalogConnectorManager; + } + + @Override + public Page loadPageData() { + // Take a snapshot first, the load loop writes these states concurrently and the column + // builders must all end up with the same number of positions. + List<CatalogRegistrationState> states = catalogConnectorManager.getCatalogRegistrationStates(); + int size = states.size(); + + BlockBuilder metalakeColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder catalogNameColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder trinoCatalogNameColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder providerColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder statusColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder lastErrorColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder lastAttemptTimeColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder lastSuccessTimeColumnBuilder = VARCHAR.createBlockBuilder(null, size); + BlockBuilder failureCountColumnBuilder = BIGINT.createBlockBuilder(null, size); + + for (CatalogRegistrationState state : states) { + VARCHAR.writeString(metalakeColumnBuilder, state.getMetalake()); + VARCHAR.writeString(catalogNameColumnBuilder, state.getCatalogName()); + VARCHAR.writeString(trinoCatalogNameColumnBuilder, state.getTrinoCatalogName()); + writeNullableString(providerColumnBuilder, state.getProvider()); + VARCHAR.writeString(statusColumnBuilder, state.getStatus().name()); + writeNullableString(lastErrorColumnBuilder, state.getLastError()); + writeTime(lastAttemptTimeColumnBuilder, state.getLastAttemptTimeMs()); + writeTime(lastSuccessTimeColumnBuilder, state.getLastSuccessTimeMs()); + BIGINT.writeLong(failureCountColumnBuilder, state.getFailureCount()); + } + + return new Page( + size, + metalakeColumnBuilder.build(), + catalogNameColumnBuilder.build(), + trinoCatalogNameColumnBuilder.build(), + providerColumnBuilder.build(), + statusColumnBuilder.build(), + lastErrorColumnBuilder.build(), + lastAttemptTimeColumnBuilder.build(), + lastSuccessTimeColumnBuilder.build(), + failureCountColumnBuilder.build()); + } + + @Override + public ConnectorTableMetadata getTableMetaData() { + return TABLE_METADATA; + } +} diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java index 4995e69821..832e89b78d 100644 --- a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableFactory.java @@ -49,6 +49,12 @@ public class GravitinoSystemTableFactory { SYSTEM_TABLES.put( GravitinoSystemTableCatalog.TABLE_NAME, new GravitinoSystemTableCatalog(catalogConnectorManager)); + SYSTEM_TABLES.put( + GravitinoSystemTableCatalogStatus.TABLE_NAME, + new GravitinoSystemTableCatalogStatus(catalogConnectorManager)); + SYSTEM_TABLES.put( + GravitinoSystemTableLoadStatus.TABLE_NAME, + new GravitinoSystemTableLoadStatus(catalogConnectorManager)); } /** diff --git a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java new file mode 100644 index 0000000000..dc4cbfe882 --- /dev/null +++ b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java @@ -0,0 +1,118 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.trino.connector.system.table; + +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.VarcharType.VARCHAR; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.trino.spi.Page; +import io.trino.spi.TrinoException; +import io.trino.spi.block.BlockBuilder; +import io.trino.spi.connector.ColumnMetadata; +import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.SchemaTableName; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import org.apache.gravitino.trino.connector.GravitinoErrorCode; +import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager; + +/** + * An implementation of the load status system table. + * + * <p>It reports the health of the loop that registers Apache Gravitino catalogs into Trino. A + * failure that prevents the loop from listing catalogs at all, such as an unreachable Gravitino + * server, has no catalog to attach itself to and is only visible here. + */ +public class GravitinoSystemTableLoadStatus extends GravitinoSystemTable { + + /** The name of the load status system table. */ + public static final SchemaTableName TABLE_NAME = + new SchemaTableName(SYSTEM_TABLE_SCHEMA_NAME, "load_status"); + + private static final ConnectorTableMetadata TABLE_METADATA = + new ConnectorTableMetadata( + TABLE_NAME, + List.of( + ColumnMetadata.builder().setName("trino_started").setType(BOOLEAN).build(), + ColumnMetadata.builder().setName("last_attempt_time").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("last_success_time").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("consecutive_failures").setType(BIGINT).build(), + ColumnMetadata.builder().setName("last_error").setType(VARCHAR).build(), + ColumnMetadata.builder().setName("metalake_errors").setType(VARCHAR).build())); + + private final CatalogConnectorManager catalogConnectorManager; + + /** + * Constructs a new GravitinoSystemTableLoadStatus. + * + * @param catalogConnectorManager the manager for catalog connectors + */ + public GravitinoSystemTableLoadStatus(CatalogConnectorManager catalogConnectorManager) { + this.catalogConnectorManager = catalogConnectorManager; + } + + @Override + public Page loadPageData() { + BlockBuilder trinoStartedColumnBuilder = BOOLEAN.createBlockBuilder(null, 1); + BlockBuilder lastAttemptTimeColumnBuilder = VARCHAR.createBlockBuilder(null, 1); + BlockBuilder lastSuccessTimeColumnBuilder = VARCHAR.createBlockBuilder(null, 1); + BlockBuilder consecutiveFailuresColumnBuilder = BIGINT.createBlockBuilder(null, 1); + BlockBuilder lastErrorColumnBuilder = VARCHAR.createBlockBuilder(null, 1); + BlockBuilder metalakeErrorsColumnBuilder = VARCHAR.createBlockBuilder(null, 1); + + BOOLEAN.writeBoolean(trinoStartedColumnBuilder, catalogConnectorManager.isTrinoStarted()); + writeTime(lastAttemptTimeColumnBuilder, catalogConnectorManager.getLastLoadAttemptTimeMs()); + writeTime(lastSuccessTimeColumnBuilder, catalogConnectorManager.getLastSuccessfulLoadTimeMs()); + BIGINT.writeLong( + consecutiveFailuresColumnBuilder, catalogConnectorManager.getConsecutiveLoadFailures()); + writeNullableString(lastErrorColumnBuilder, catalogConnectorManager.getLastLoadError()); + + Map<String, String> metalakeErrors = catalogConnectorManager.getMetalakeErrors(); + if (metalakeErrors.isEmpty()) { + metalakeErrorsColumnBuilder.appendNull(); + } else { + try { + VARCHAR.writeString( + metalakeErrorsColumnBuilder, + new ObjectMapper().writeValueAsString(new TreeMap<>(metalakeErrors))); + } catch (JsonProcessingException e) { + throw new TrinoException( + GravitinoErrorCode.GRAVITINO_ILLEGAL_ARGUMENT, "Invalid metalake error format", e); + } + } + + return new Page( + 1, + trinoStartedColumnBuilder.build(), + lastAttemptTimeColumnBuilder.build(), + lastSuccessTimeColumnBuilder.build(), + consecutiveFailuresColumnBuilder.build(), + lastErrorColumnBuilder.build(), + metalakeErrorsColumnBuilder.build()); + } + + @Override + public ConnectorTableMetadata getTableMetaData() { + return TABLE_METADATA; + } +} diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnector.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnector.java index fecc1fd9cc..2fbd12c4c1 100644 --- a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnector.java +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoConnector.java @@ -21,6 +21,7 @@ package org.apache.gravitino.trino.connector; import static java.lang.String.format; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import com.google.common.base.Preconditions; import io.trino.testing.DistributedQueryRunner; @@ -282,7 +283,10 @@ public abstract class TestGravitinoConnector extends AbstractGravitinoConnectorT assertQueryFails( "call gravitino.system.create_catalog(" + "catalog=>'memory1', provider=>'memory', properties => Map(array['trino.bypass.unknown-direct-key'], array['10']))", - format("Create catalog failed. Create catalog failed due to the loading process fails")); + // The message must carry the reason the registration failed, not just the fact that it + // did. (?s) lets .* span the newlines of the underlying configuration error. + "(?s)Create catalog failed. Create catalog failed due to the loading process fails\\." + + ".*unknown-direct-key.*"); assertThat(computeActual("show catalogs").getOnlyColumnAsSet()).doesNotContain("memory1"); assertUpdate( @@ -306,6 +310,36 @@ public abstract class TestGravitinoConnector extends AbstractGravitinoConnectorT assertEquals(row.getField(2), "{\"max_ttl\":\"10\"}"); } + @Test + public void testCatalogStatusSystemTable() throws Exception { + MaterializedResult result = + computeActual( + "select metalake, catalog_name, trino_catalog_name, provider, status, last_error," + + " failure_count from gravitino.system.catalog_status"); + assertEquals(result.getRowCount(), 1); + MaterializedRow row = result.getMaterializedRows().get(0); + assertEquals(row.getField(1), "memory"); + assertEquals(row.getField(2), "memory"); + assertEquals(row.getField(3), "memory"); + assertEquals(row.getField(4), "REGISTERED"); + assertNull(row.getField(5)); + assertEquals(row.getField(6), 0L); + } + + @Test + public void testLoadStatusSystemTable() throws Exception { + MaterializedResult result = + computeActual( + "select trino_started, consecutive_failures, last_error, metalake_errors" + + " from gravitino.system.load_status"); + assertEquals(result.getRowCount(), 1); + MaterializedRow row = result.getMaterializedRows().get(0); + assertEquals(row.getField(0), true); + assertEquals(row.getField(1), 0L); + assertNull(row.getField(2)); + assertNull(row.getField(3)); + } + private TableName createTestTable(String fullTableName) throws Exception { TableName tableName = new TableName(fullTableName); 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 995aedc29d..bc6f42f7ea 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 @@ -21,9 +21,12 @@ package org.apache.gravitino.trino.connector.catalog; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -31,12 +34,18 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; import io.trino.spi.TrinoException; import io.trino.spi.connector.ConnectorContext; import java.util.Optional; import org.apache.gravitino.client.GravitinoAdminClient; import org.apache.gravitino.client.GravitinoMetalake; import org.apache.gravitino.exceptions.RESTException; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import org.apache.gravitino.Audit; +import org.apache.gravitino.Catalog; import org.apache.gravitino.trino.connector.GravitinoConfig; import org.apache.gravitino.trino.connector.GravitinoErrorCode; import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; @@ -269,6 +278,185 @@ public class TestCatalogConnectorManager { assertEquals("http://irc-host:9001/iceberg";, config.getDiscoveredIcebergRestUri("test")); } + @Test + public void testSuccessfulRegistrationIsRecorded() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.REGISTERED, state.getStatus()); + assertEquals("test", state.getMetalake()); + assertEquals("memory", state.getCatalogName()); + assertEquals("memory", state.getTrinoCatalogName()); + assertEquals("memory", state.getProvider()); + assertNull(state.getLastError()); + assertEquals(0, state.getFailureCount()); + assertTrue(state.getLastSuccessTimeMs() > 0); + + assertTrue(manager.isTrinoStarted()); + assertNull(manager.getLastLoadError()); + assertEquals(0, manager.getConsecutiveLoadFailures()); + assertTrue(manager.getMetalakeErrors().isEmpty()); + } + + @Test + public void testRegistrationFailureIsRecorded() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + doThrow( + new TrinoException( + GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR, + "Access Denied: Cannot create catalog memory")) + .when(fixture.catalogRegister) + .registerCatalog(any(), any()); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.FAILED, state.getStatus()); + assertTrue(state.getLastError().contains("Access Denied")); + assertEquals(1, state.getFailureCount()); + assertEquals(0, state.getLastSuccessTimeMs()); + + // A second failing round must accumulate rather than reset the failure count. + manager.loadMetalakeSync(); + state = singleState(manager); + assertEquals(2, state.getFailureCount()); + assertEquals(0, state.getLastSuccessTimeMs()); + } + + @Test + public void testRegistrationFailureIsReportedByStoredProcedureMessage() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + doThrow( + new TrinoException( + GravitinoErrorCode.GRAVITINO_RUNTIME_ERROR, + "Access Denied: Cannot create catalog memory")) + .when(fixture.catalogRegister) + .registerCatalog(any(), any()); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + String description = manager.describeRegistrationFailure("memory"); + assertTrue(description.contains("FAILED")); + assertTrue(description.contains("Access Denied")); + } + + @Test + public void testNonRelationalCatalogIsRecorded() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("files", "hadoop", Catalog.Type.FILESET)); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.UNSUPPORTED, state.getStatus()); + assertTrue(state.getLastError().contains("FILESET")); + } + + @Test + public void testUnsupportedProviderIsRecorded() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("other", "unknown", Catalog.Type.RELATIONAL)); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.UNSUPPORTED, state.getStatus()); + assertTrue(state.getLastError().contains("unknown")); + assertEquals("unknown", state.getProvider()); + } + + @Test + public void testSkippedCatalogIsRecorded() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + + CatalogConnectorManager manager = + fixture.createManager(ImmutableMap.of("gravitino.trino.skip-catalog-patterns", "mem.*")); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals(CatalogRegistrationState.Status.SKIPPED, state.getStatus()); + assertTrue(state.getLastError().contains("skip-catalog-patterns")); + } + + @Test + public void testStaleStateIsPruned() throws Exception { + LoadFixture fixture = new LoadFixture(); + Catalog first = mockCatalog("a", "memory", Catalog.Type.RELATIONAL); + Catalog second = mockCatalog("b", "memory", Catalog.Type.RELATIONAL); + fixture.withCatalogs(first, second); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + assertEquals(2, manager.getCatalogRegistrationStates().size()); + + // The second catalog is dropped in the Gravitino server, its state must not linger. + fixture.withCatalogs(first); + manager.loadMetalakeSync(); + + CatalogRegistrationState state = singleState(manager); + assertEquals("a", state.getCatalogName()); + } + + @Test + public void testListCatalogsFailureIsRecordedPerMetalake() throws Exception { + LoadFixture fixture = new LoadFixture(); + fixture.withCatalogs(mockCatalog("memory", "memory", Catalog.Type.RELATIONAL)); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + assertEquals(CatalogRegistrationState.Status.REGISTERED, singleState(manager).getStatus()); + + when(fixture.metalake.listCatalogs()).thenThrow(new RuntimeException("Connection refused")); + manager.loadMetalakeSync(); + + Map<String, String> metalakeErrors = manager.getMetalakeErrors(); + assertEquals(1, metalakeErrors.size()); + assertTrue(metalakeErrors.get("test").contains("Connection refused")); + // A transient listing failure must not turn a healthy catalog into a failed one. + assertEquals(CatalogRegistrationState.Status.REGISTERED, singleState(manager).getStatus()); + } + + @Test + public void testUnreachableServerIsRecordedInLoadStatus() throws Exception { + LoadFixture fixture = new LoadFixture(); + when(fixture.client.loadMetalake(any())).thenThrow(new RuntimeException("Connection refused")); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + assertNotNull(manager.getLastLoadError()); + assertTrue(manager.getLastLoadError().contains("Connection refused")); + assertEquals(1, manager.getConsecutiveLoadFailures()); + assertEquals(0, manager.getLastSuccessfulLoadTimeMs()); + assertTrue(manager.getLastLoadAttemptTimeMs() > 0); + assertTrue(manager.getCatalogRegistrationStates().isEmpty()); + } + + @Test + public void testTrinoNotStartedIsRecordedInLoadStatus() throws Exception { + LoadFixture fixture = new LoadFixture(); + when(fixture.catalogRegister.isTrinoStarted()).thenReturn(false); + + CatalogConnectorManager manager = fixture.createManager(ImmutableMap.of()); + manager.loadMetalakeSync(); + + assertFalse(manager.isTrinoStarted()); + assertNotNull(manager.getLastLoadError()); + assertTrue(manager.getLastLoadError().contains("Waiting for the Trino server")); + assertTrue(manager.getCatalogRegistrationStates().isEmpty()); + } + private CatalogConnectorManager createManager(ImmutableMap<String, String> configMap) throws Exception { return createManager(createCatalogConnectorFactory(), configMap); @@ -323,4 +511,66 @@ public class TestCatalogConnectorManager { private static ConnectorContext mockContext() { return mock(ConnectorContext.class); } + + private static CatalogRegistrationState singleState(CatalogConnectorManager manager) { + List<CatalogRegistrationState> states = manager.getCatalogRegistrationStates(); + assertEquals(1, states.size()); + return states.get(0); + } + + private static Catalog mockCatalog(String name, String provider, Catalog.Type type) { + Catalog catalog = mock(Catalog.class); + when(catalog.name()).thenReturn(name); + when(catalog.provider()).thenReturn(provider); + when(catalog.type()).thenReturn(type); + when(catalog.properties()).thenReturn(ImmutableMap.of()); + Audit audit = mock(Audit.class); + when(audit.createTime()).thenReturn(Instant.now()); + when(audit.lastModifiedTime()).thenReturn(null); + when(catalog.auditInfo()).thenReturn(audit); + return catalog; + } + + /** Wires up a manager whose load loop can be driven with {@code loadMetalakeSync()}. */ + private static class LoadFixture { + private final CatalogRegister catalogRegister = mock(CatalogRegister.class); + private final GravitinoAdminClient client = mock(GravitinoAdminClient.class); + private final GravitinoMetalake metalake = mock(GravitinoMetalake.class); + private final CatalogConnectorFactory catalogFactory = mock(CatalogConnectorFactory.class); + + LoadFixture() throws Exception { + when(catalogRegister.isTrinoStarted()).thenReturn(true); + when(metalake.name()).thenReturn("test"); + when(client.loadMetalake(any())).thenReturn(metalake); + when(catalogFactory.getSupportedCatalogProviders()).thenReturn(ImmutableSet.of("memory")); + CatalogConnectorContext.Builder builder = mock(CatalogConnectorContext.Builder.class); + when(catalogFactory.createCatalogConnectorContextBuilder(any())).thenReturn(builder); + when(builder.withMetalake(any())).thenReturn(builder); + when(builder.withContext(any())).thenReturn(builder); + when(builder.build()).thenReturn(mock(CatalogConnectorContext.class)); + } + + void withCatalogs(Catalog... catalogs) { + String[] names = new String[catalogs.length]; + for (int i = 0; i < catalogs.length; i++) { + names[i] = catalogs[i].name(); + when(metalake.loadCatalog(names[i])).thenReturn(catalogs[i]); + } + when(metalake.listCatalogs()).thenReturn(names); + } + + CatalogConnectorManager createManager(Map<String, String> extraConfig) { + ImmutableMap<String, String> configMap = + ImmutableMap.<String, String>builder() + .put("gravitino.uri", "http://127.0.0.1:8090";) + .put("gravitino.metalake", "test") + .put("gravitino.use-single-metalake", "true") + .putAll(extraConfig) + .build(); + CatalogConnectorManager manager = + new CatalogConnectorManager(catalogRegister, catalogFactory, null); + manager.config(new GravitinoConfig(configMap), client); + return manager; + } + } } diff --git a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java new file mode 100644 index 0000000000..7139f1a163 --- /dev/null +++ b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/system/table/TestGravitinoSystemStatusTables.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.trino.connector.system.table; + +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableMap; +import io.trino.spi.Page; +import io.trino.spi.block.Block; +import java.util.List; +import java.util.Map; +import org.apache.gravitino.trino.connector.catalog.CatalogConnectorManager; +import org.apache.gravitino.trino.connector.catalog.CatalogRegistrationState; +import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog; +import org.junit.jupiter.api.Test; + +public class TestGravitinoSystemStatusTables { + + @Test + public void testCatalogStatusTableRendersRegisteredCatalog() { + GravitinoCatalog catalog = + new GravitinoCatalog("test", "memory", "memory", ImmutableMap.of(), 0L); + CatalogRegistrationState state = CatalogRegistrationState.succeeded(catalog, "memory"); + + Page page = loadCatalogStatusPage(List.of(state)); + + assertEquals(1, page.getPositionCount()); + assertEquals(9, page.getChannelCount()); + assertEquals("test", varchar(page, 0)); + assertEquals("memory", varchar(page, 1)); + assertEquals("memory", varchar(page, 2)); + assertEquals("memory", varchar(page, 3)); + assertEquals("REGISTERED", varchar(page, 4)); + // last_error is null for a registered catalog, last_success_time is set. + assertTrue(page.getBlock(5).isNull(0)); + assertFalse(page.getBlock(7).isNull(0)); + assertEquals(0, BIGINT.getLong(page.getBlock(8), 0)); + } + + @Test + public void testCatalogStatusTableRendersFailedCatalog() { + CatalogRegistrationState state = + CatalogRegistrationState.failed( + "test", "memory", "memory", null, "Access Denied: Cannot create catalog memory", null); + + Page page = loadCatalogStatusPage(List.of(state)); + + assertEquals("FAILED", varchar(page, 4)); + assertTrue(varchar(page, 5).contains("Access Denied")); + // The provider is unknown on this path, and the catalog was never registered. + assertTrue(page.getBlock(3).isNull(0)); + assertTrue(page.getBlock(7).isNull(0)); + assertEquals(1, BIGINT.getLong(page.getBlock(8), 0)); + } + + @Test + public void testCatalogStatusTableIsEmptyWhenNoCatalogWasSeen() { + Page page = loadCatalogStatusPage(List.of()); + assertEquals(0, page.getPositionCount()); + } + + @Test + public void testLoadStatusTableRendersHealthyLoop() { + CatalogConnectorManager manager = mock(CatalogConnectorManager.class); + when(manager.isTrinoStarted()).thenReturn(true); + when(manager.getLastLoadAttemptTimeMs()).thenReturn(1000L); + when(manager.getLastSuccessfulLoadTimeMs()).thenReturn(1000L); + when(manager.getConsecutiveLoadFailures()).thenReturn(0L); + when(manager.getLastLoadError()).thenReturn(null); + when(manager.getMetalakeErrors()).thenReturn(Map.of()); + + Page page = new GravitinoSystemTableLoadStatus(manager).loadPageData(); + + assertEquals(1, page.getPositionCount()); + assertEquals(6, page.getChannelCount()); + assertTrue(BOOLEAN.getBoolean(page.getBlock(0), 0)); + assertEquals("1970-01-01T00:00:01Z", varchar(page, 1)); + assertEquals(0, BIGINT.getLong(page.getBlock(3), 0)); + assertTrue(page.getBlock(4).isNull(0)); + assertTrue(page.getBlock(5).isNull(0)); + } + + @Test + public void testLoadStatusTableRendersUnreachableServer() { + CatalogConnectorManager manager = mock(CatalogConnectorManager.class); + when(manager.isTrinoStarted()).thenReturn(true); + when(manager.getLastLoadAttemptTimeMs()).thenReturn(2000L); + when(manager.getLastSuccessfulLoadTimeMs()).thenReturn(0L); + when(manager.getConsecutiveLoadFailures()).thenReturn(3L); + when(manager.getLastLoadError()).thenReturn("Connection refused"); + when(manager.getMetalakeErrors()).thenReturn(Map.of("test", "Connection refused")); + + Page page = new GravitinoSystemTableLoadStatus(manager).loadPageData(); + + // last_success_time stays null while the server is unreachable. + assertTrue(page.getBlock(2).isNull(0)); + assertEquals(3, BIGINT.getLong(page.getBlock(3), 0)); + assertEquals("Connection refused", varchar(page, 4)); + assertEquals("{\"test\":\"Connection refused\"}", varchar(page, 5)); + } + + private static Page loadCatalogStatusPage(List<CatalogRegistrationState> states) { + CatalogConnectorManager manager = mock(CatalogConnectorManager.class); + when(manager.getCatalogRegistrationStates()).thenReturn(states); + return new GravitinoSystemTableCatalogStatus(manager).loadPageData(); + } + + private static String varchar(Page page, int channel) { + Block block = page.getBlock(channel); + return VARCHAR.getSlice(block, 0).toStringUtf8(); + } +}
