jerryshao commented on code in PR #12547:
URL: https://github.com/apache/gravitino/pull/12547#discussion_r3841838795
##########
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:
**[Confirmed bug]** An `UnsupportedOperationException` thrown by
`metalake.loadCatalog()` for a catalog type the client library doesn't
recognize is now recorded as `FAILED` (with escalating `failure_count` and an
ERROR log) instead of being handled as `UNSUPPORTED`.
`DTOConverters.toCatalog` (clients/client-java) throws
`UnsupportedOperationException("Unsupported catalog type: " + catalog.type())`
for a catalog type it doesn't map. The old code caught this specifically and
logged a WARN with no persisted state. This new loop wraps
`metalake.loadCatalog(catalogName)` inside the same try whose only catch is the
generic `catch (Exception e)` below, which calls
`recordCatalogState(CatalogRegistrationState.failed(...))`. Because the
type/provider checks run only *after* `loadCatalog()` succeeds, this catalog is
permanently misreported as FAILED (not UNSUPPORTED) in
`gravitino.system.catalog_status`, defeating the PR's own goal of letting users
self-diagnose via that table.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalog.java:
##########
@@ -54,23 +56,29 @@ public class GravitinoSystemTableCatalog extends
GravitinoSystemTable {
ColumnMetadata.builder().setName("properties").setType(VARCHAR).build()));
private final CatalogConnectorManager catalogConnectorManager;
+ private final String metalake;
/**
* Constructs a new GravitinoSystemTableCatalog.
*
* @param catalogConnectorManager the manager for catalog connectors
+ * @param metalake the metalake to report on
*/
- public GravitinoSystemTableCatalog(CatalogConnectorManager
catalogConnectorManager) {
+ public GravitinoSystemTableCatalog(
+ CatalogConnectorManager catalogConnectorManager, String metalake) {
this.catalogConnectorManager = catalogConnectorManager;
+ this.metalake = metalake;
}
@Override
public Page loadPageData() {
List<GravitinoCatalog> gravitinoCatalogs = new ArrayList<>();
// retrieve catalogs form the Gravitino server with the configuration
metalakes,
// the catalogConnectorManager does not manager catalogs in worker nodes
- catalogConnectorManager
- .getUsedMetalakes()
+ // Only the metalake this connector is configured with: the manager is
shared by every entry
+ // catalog in this Trino.
+ catalogConnectorManager.getUsedMetalakes().stream()
+ .filter(metalake::equals)
.forEach(
(metalakeName) -> {
GravitinoMetalake metalake =
catalogConnectorManager.getMetalake(metalakeName);
Review Comment:
**[Confirmed bug, a few lines below]** `loadPageData()` calls
`catalogConnectorManager.skipCatalog(catalog.name())` (unchanged line just
below this hunk) with the bare Gravitino catalog name, while the load loop
(`CatalogConnectorManager.loadCatalogs`) matches skip patterns against the
Trino-qualified name via `skipCatalog(trinoCatalogName)`.
In multi-metalake mode, `getTrinoCatalogName` returns a quoted
`"metalake.catalog"` form. If `gravitino.trino.skip-catalog-patterns` is
written against that qualified form (e.g. `prod\..*`), the load loop correctly
skips/unregisters matching catalogs (and now reports them SKIPPED in
catalog_status), but this table still evaluates `skipCatalog()` against the
bare catalog name, which never matches a metalake-qualified pattern — so
`gravitino.system.catalog` keeps listing a catalog that is actually SKIPPED and
absent from `SHOW CATALOGS`, misleading a user into thinking it is usable.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -310,6 +531,7 @@ private void unloadCatalog(GravitinoCatalog catalog) {
String catalogFullName = getTrinoCatalogName(catalog);
catalogRegister.unregisterCatalog(catalogFullName);
catalogConnectors.remove(catalogFullName);
+ catalogStates.remove(catalogFullName);
Review Comment:
**[Confirmed bug]** A catalog that newly starts matching
`gravitino.trino.skip-catalog-patterns` has its just-recorded `SKIPPED` state
deleted in the very same load cycle by this method, which unconditionally
removes the `catalogStates` entry.
When a previously-REGISTERED catalog starts matching a skip pattern: (1) the
load loop calls `recordCatalogState(skipped(...))` for it; (2) since it's now
excluded from `catalogNames`, the unload loop sees it still in
`catalogConnectors` and calls `unloadCatalog(...)`; (3) `unloadCatalog()`
unregisters it from Trino correctly but also does
`catalogStates.remove(catalogFullName)` here, wiping the SKIPPED row just
written in step 1. For that one refresh cycle,
`gravitino.system.catalog_status` has no row at all for the catalog,
contradicting the PR's own documentation claim that catalog_status "covers
every catalog the connector considered, including the ones catalog filters
out." It reappears correctly on the next tick.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableLoadStatus.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.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.catalog.CatalogConnectorManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 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 {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(GravitinoSystemTableLoadStatus.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ /** 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;
+ private final String metalake;
+
+ /**
+ * Constructs a new GravitinoSystemTableLoadStatus.
+ *
+ * @param catalogConnectorManager the manager for catalog connectors
+ * @param metalake the metalake to report errors for
+ */
+ public GravitinoSystemTableLoadStatus(
+ CatalogConnectorManager catalogConnectorManager, String metalake) {
+ this.catalogConnectorManager = catalogConnectorManager;
+ this.metalake = metalake;
+ }
+
+ @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());
Review Comment:
**[Plausible]** `loadPageData()` makes six independent, unsynchronized reads
of `CatalogConnectorManager` (`isTrinoStarted`, `getLastLoadAttemptTimeMs`,
`getLastSuccessfulLoadTimeMs`, `getConsecutiveLoadFailures`,
`getLastLoadError`, `getMetalakeErrors`) that the load loop updates as a group,
with no atomic snapshot.
The background load loop writes these separate fields sequentially over the
course of one cycle. A query thread landing mid-cycle can read
`trino_started=true` (already updated) together with a stale `last_error` from
the previous failed cycle and a `consecutive_load_failures` count not yet
reset, producing a self-contradictory `load_status` row. Notably,
`GravitinoSystemTableCatalogStatus` explicitly takes a single snapshot via
`getCatalogRegistrationStates()` for exactly this reason, but that discipline
wasn't applied here.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoConnectorFactory.java:
##########
@@ -52,8 +53,6 @@ public class GravitinoConnectorFactory implements
ConnectorFactory {
public static final String DEFAULT_CONNECTOR_NAME = "gravitino";
@SuppressWarnings("UnusedVariable")
Review Comment:
**[Confirmed]** `@SuppressWarnings("UnusedVariable")` is left on this field
after the field it originally annotated (the deleted
`gravitinoSystemTableFactory` field) was removed by this diff;
`catalogConnectorManager` is actively assigned and read and is not unused.
This stale suppression now silences an unused-variable warning that would
never fire for this field, and misleads a reviewer into thinking the field is
intentionally write-only. Should simply be deleted.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -186,14 +216,102 @@ private void loadMetalake() {
LOG.debug("Load metalake: {}", usedMetalake);
loadCatalogs(metalake);
} catch (Exception e) {
- LOG.error("Load Metalake {} failed.", usedMetalake, e);
+ recordMetalakeError(usedMetalake, e);
}
}
- } catch (Exception e) {
- LOG.error("Error when loading metalake", e);
+
+ pruneMissingMetalakes(usedMetalakes);
+
+ if (metalakeErrors.isEmpty()) {
+ lastSuccessfulLoadTimeMs = System.currentTimeMillis();
+ recordLoadSuccess();
+ } else {
+ // Some metalake failed. The loop reaching its last line is not a
health signal, so do not
+ // advance the success time or clear the error, or load_status would
report a healthy loop
+ // while no catalog is being registered at all.
+ recordLoadFailure(
+ String.format(
+ "%d of %d metalakes failed to load: %s",
+ metalakeErrors.size(), usedMetalakes.size(), new
TreeMap<>(metalakeErrors)),
+ null);
+ }
+ } catch (Throwable t) {
+ // Catch Throwable, not Exception: scheduleWithFixedDelay silently
cancels the task forever
+ // the first time the runnable throws, and loading a Trino connector
plugin can raise
+ // NoClassDefFoundError. A dead loop must not look like a healthy one.
+ recordLoadFailure(toErrorMessage(t), t);
}
}
+ private void pruneMissingMetalakes(Set<String> usedMetalakes) {
+ // A metalake that was deleted, or that dropped out of the configuration,
leaves its catalog
+ // rows behind. Without this they keep reporting REGISTERED for catalogs
that no longer exist.
+ catalogStates.values().removeIf(state ->
!usedMetalakes.contains(state.getMetalake()));
+ metalakeErrors.keySet().removeIf(metalakeName ->
!usedMetalakes.contains(metalakeName));
+ }
+
+ private void recordLoadSuccess() {
+ if (lastLoadError != null) {
+ LOG.info("The Gravitino catalog load loop recovered.");
+ }
+ lastLoadError = null;
+ consecutiveLoadFailures.set(0);
+ }
+
+ private void recordLoadFailure(String message, Throwable cause) {
+ boolean changed = !Objects.equals(lastLoadError, message);
+ lastLoadError = message;
+ consecutiveLoadFailures.incrementAndGet();
+ if (!changed) {
+ LOG.debug("Failed to load catalogs from the Gravitino server: {}",
message, cause);
+ } else if (trinoStarted) {
+ LOG.error("Failed to load catalogs from the Gravitino server: {}",
message, cause);
+ } else {
+ // Trino not being up yet is the normal state during startup, not an
error.
+ LOG.info("{}", message);
+ }
+ }
+
+ private void recordMetalakeError(String metalakeName, Throwable 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(Throwable e) {
Review Comment:
**[Plausible simplification]** `toErrorMessage(Throwable)` hand-rolls a
bounded (`MAX_CAUSE_DEPTH=32`) root-cause walk with ad hoc cycle detection,
duplicating Guava's `Throwables.getRootCause()`, which is already used
elsewhere in the codebase and has proper cycle detection.
A cause chain longer than 32 links (unlikely but possible with deeply
wrapped exceptions) is silently truncated at an arbitrary depth instead of
reaching the true root, producing a less useful error message in
catalog_status/load_status than Guava's well-tested implementation would.
Replacing this loop with `Throwables.getRootCause(e)` removes the hand-rolled
code and its arbitrary bound.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogConnectorManager.java:
##########
@@ -382,6 +604,100 @@ public String getTrinoCatalogName(GravitinoCatalog
catalog) {
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() {
Review Comment:
**[Plausible efficiency]** `getCatalogRegistrationStates()` does
`List.copyOf(catalogStates.values())` over ALL metalakes' catalog states on
every call, but its only caller
(`GravitinoSystemTableCatalogStatus.loadPageData`) immediately discards
everything outside the current connector's own metalake.
On every query against `gravitino.system.catalog_status`, a Trino cluster
with many entry catalogs/metalakes and many catalogs per metalake copies the
entire cluster-wide catalog-state collection just to keep the subset belonging
to one metalake — cost scales with total cluster-wide catalog count rather than
the querying connector's own. Filtering while streaming
`catalogStates.values()` directly, or exposing a metalake-scoped accessor,
would avoid materializing the full collection per query.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/GravitinoSystemConnector.java:
##########
@@ -183,14 +217,33 @@ public SchemaTableName getTableName() {
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. Set
once by the
+ // coordinator's GravitinoConnectorFactory.create(), which necessarily
runs before any query
+ // can reach the scheduler, and read by the scheduler through
isRemotelyAccessible() and
+ // getAddresses() below. On a real worker JVM it is never set and the
split keeps the previous
+ // remotely accessible behaviour; in a single JVM test runner the static
is shared with the
+ // coordinator, which is harmless because only the coordinator's scheduler
reads it.
+ private static volatile HostAddress coordinatorAddress;
Review Comment:
**[Plausible / CLAUDE.md violation]** This new static field
`coordinatorAddress` is declared after the instance field `tableName`, the
constructor, and the `getTableName()` method in the `Split` class, violating
the project's Class Member Ordering rule (static constants → static fields →
instance fields → constructors → methods).
This newly added static field sits after an instance field, a constructor,
and a method instead of being grouped with the class's other fields before the
constructor — a direct, quotable violation of the stated ordering rule in
CLAUDE.md.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/catalog/CatalogRegistrationState.java:
##########
@@ -0,0 +1,299 @@
+/*
+ * 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(
Review Comment:
**[Plausible simplification]** The four static factories (`succeeded`,
`failed`, `skipped`, `unsupported`) are ~90% identical, each calling the same
9-arg private constructor and differing only in the `Status` literal and a
couple of field values.
Adding a 10th field to this class requires editing the same field list in
all four factories plus `withHistoryOf`'s reconstruction, with no compiler
assistance if one call site is missed since the diffs all look alike — a
copy-paste-and-forget-one-field bug is easy to introduce and hard to catch in
review. A shared private helper taking `(metalake, catalogName,
trinoCatalogName, provider, status, reason)` would collapse three of the four
factories to one-line callers.
##########
trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/system/table/GravitinoSystemTableCatalogStatus.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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;
+ private final String metalake;
+
+ /**
+ * Constructs a new GravitinoSystemTableCatalogStatus.
+ *
+ * @param catalogConnectorManager the manager for catalog connectors
+ * @param metalake the metalake to report on
+ */
+ public GravitinoSystemTableCatalogStatus(
+ CatalogConnectorManager catalogConnectorManager, String metalake) {
+ this.catalogConnectorManager = catalogConnectorManager;
+ this.metalake = metalake;
+ }
+
+ @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.
+ // The load loop is shared by every entry catalog in this Trino, so report
only the metalake
+ // this connector is configured with.
+ List<CatalogRegistrationState> states =
+ catalogConnectorManager.getCatalogRegistrationStates().stream()
+ .filter(state -> state.getMetalake().equals(metalake))
Review Comment:
**[Plausible]** This metalake scoping filter compares the raw
`gravitino.metalake` connector-config string via `.equals()` against canonical
metalake names collected by the load loop, with no normalization of case or
whitespace.
`GravitinoConnectorFactory.create()` passes `config.getMetalake()` (the
literal config value) into `GravitinoSystemTableFactory`, used unmodified in
the `.filter(state -> state.getMetalake().equals(metalake))` /
`.filter(metalake::equals)` checks across all three per-metalake tables. If the
configured value differs even slightly (trailing whitespace, case) from the
canonical name recorded in `CatalogRegistrationState`/`getUsedMetalakes()`,
every row is silently filtered out — `gravitino.system.catalog`,
`catalog_status`, and `load_status.metalake_errors` all report empty/zero with
no error, even while catalogs are actively failing to load.
--
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]
