yuqi1129 commented on code in PR #14741:
URL: https://github.com/apache/gravitino/pull/14741#discussion_r4494189546
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java:
##########
@@ -152,29 +166,59 @@ public void close() {
}
public Catalog getGravitinoCatalogInfo(String name) {
- GravitinoIdentity identity = currentIdentity();
try {
- return gravitinoCatalogs.get(cacheKey(identity, name), key ->
loadCatalog(identity, name));
+ return getGravitinoCatalogInfo(currentIdentity(), name);
} catch (Exception e) {
LOG.error(String.format("Load catalog %s failed", name), e);
throw new RuntimeException(e);
}
}
+ /**
+ * Returns the catalog for the given identity, loaded through that
identity's current client.
+ *
+ * <p>Spark catalogs call this before every Gravitino request instead of
keeping the returned
+ * catalog: the lookup refreshes the client's cache entry, so a client in
use is not evicted for
+ * being idle, and a catalog loaded through a client that has since been
evicted and closed is
+ * loaded again through the current one.
+ *
+ * <p>Gravitino exceptions are thrown as is, so callers can still tell them
apart.
+ *
+ * @param identity the identity the catalog is loaded for
+ * @param name the catalog name
+ * @return the catalog, bound to the identity's current client
+ */
+ Catalog getGravitinoCatalogInfo(GravitinoIdentity identity, String name) {
+ // Spark catalogs keep a reference to this manager, so fail instead of
building a new client.
+ Preconditions.checkState(!isClosed, "GravitinoCatalogManager is already
closed");
+ GravitinoClient client = getClient(identity);
+ String key = cacheKey(identity, name);
+ CachedCatalog cached =
+ gravitinoCatalogs.get(
+ key, k -> new CachedCatalog(client, loadCatalog(client, identity,
name)));
+ if (cached.client != client) {
+ cached = new CachedCatalog(client, loadCatalog(client, identity, name));
+ gravitinoCatalogs.put(key, cached);
Review Comment:
Minor, non-blocking: could we coalesce this refresh per cache key? Two
callers can obtain the same new client, both read the old CachedCatalog, and
each issue loadCatalog before either put completes. Cache.get only makes the
initial cache-miss load atomic; this refresh happens outside it. A two-thread
latch-based probe reproduced two reloads for the same key/client. An atomic
per-key refresh that rechecks cached.client inside it, plus a concurrent
load-count test, would avoid the duplicate requests.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java:
##########
@@ -192,12 +193,24 @@ public String name() {
return catalogName;
}
+ /**
+ * Returns the Gravitino catalog that serves schema, table and function
requests. Look it up for
+ * each request instead of keeping it: the lookup resets the idle timer of
the Gravitino client in
+ * the manager's cache, and returns a catalog bound to a live client after
the cache has evicted
+ * and closed an earlier one.
Review Comment:
Minor, non-blocking: could we clarify the in-flight eviction limitation here
and in the PR description? It applies to TTL expiry as well as size eviction: a
sufficiently long-running lookup/request can still lose its client after the
cache access. testCatalogLoadedThroughAnEvictedClientIsLoadedAgain already
demonstrates the boundary by expiring the client during the load: the first
lookup returns a catalog bound to C1, and the next lookup switches to C2. The
lookup restores subsequent requests but does not keep the client alive
throughout the current request; documenting that is sufficient for this PR.
--
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]
