kbendick commented on a change in pull request #3543:
URL: https://github.com/apache/iceberg/pull/3543#discussion_r748914866
##########
File path: core/src/main/java/org/apache/iceberg/CachingCatalog.java
##########
@@ -20,33 +20,143 @@
package org.apache.iceberg;
import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.CacheWriter;
import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.RemovalCause;
+import com.github.benmanes.caffeine.cache.RemovalListener;
+import com.github.benmanes.caffeine.cache.Ticker;
+import java.time.Duration;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
+import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class CachingCatalog implements Catalog {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CachingCatalog.class);
+ private static final RemovalListener<TableIdentifier, Table>
keyLoggingRemovalListener =
+ (key, value, cause) -> LOG.info("Expired {} from the TableCache", key);
+
public static Catalog wrap(Catalog catalog) {
- return wrap(catalog, true);
+ return wrap(catalog, false, 0);
+ }
+
+ public static Catalog wrap(Catalog catalog, boolean expirationEnabled, long
expirationIntervalMilllis) {
+ return wrap(catalog, true, expirationEnabled, expirationIntervalMilllis);
+ }
+
+ public static Catalog wrap(Catalog catalog, boolean caseSensitive, boolean
expirationEnabled,
+ long expirationIntervalMillis) {
+ return new CachingCatalog(catalog, caseSensitive, expirationEnabled,
expirationIntervalMillis);
}
- public static Catalog wrap(Catalog catalog, boolean caseSensitive) {
- return new CachingCatalog(catalog, caseSensitive);
+ @VisibleForTesting
+ static Catalog wrap(Catalog catalog, boolean expirationEnabled, long
expirationIntervalMillis, Ticker ticker) {
+ return new CachingCatalog(catalog, true, expirationEnabled,
expirationIntervalMillis, ticker);
}
- private final Cache<TableIdentifier, Table> tableCache =
Caffeine.newBuilder().softValues().build();
private final Catalog catalog;
private final boolean caseSensitive;
+ private final boolean expirationEnabled;
+ private final long expirationIntervalMillis;
+ private final Cache<TableIdentifier, Table> tableCache;
- private CachingCatalog(Catalog catalog, boolean caseSensitive) {
+ private CachingCatalog(Catalog catalog, boolean caseSensitive, boolean
isExpirationEnabled,
+ long expirationIntervalInMillis) {
+ this(catalog, caseSensitive, isExpirationEnabled,
expirationIntervalInMillis, Ticker.systemTicker());
+ }
+
+ private CachingCatalog(Catalog catalog, boolean caseSensitive, boolean
isExpirationEnabled,
+ long expirationIntervalMillis, Ticker ticker) {
this.catalog = catalog;
this.caseSensitive = caseSensitive;
+ this.expirationEnabled = isExpirationEnabled;
+ this.expirationIntervalMillis = expirationIntervalMillis;
+
+ this.tableCache = createTableCache(ticker);
+ }
+
+ /**
+ * Return the age of an entry in the cache.
+ * <p>
+ * This method is only visible for testing the cache expiration policy, as
cache invalidation is handled
+ * by the catalog and not the cache itself.
+ * <p>
+ * Returns the age of the cache entry corresponding to the identifier, or
{@code Optional.empty} if the table
+ * is not present in the cache or if no expireAfterAccess policy is present
in this CachingCatalog.
+ */
+ @VisibleForTesting
+ Optional<Duration> cachedEntryAge(TableIdentifier identifier) {
+ return tableCache.policy()
+ .expireAfterAccess()
+ .flatMap(tableExpiration -> tableExpiration.ageOf(identifier));
+ }
+
+ // Returns the cached Table entry corresponding to the given identifier iff
+ // it's in the cache. Grabs the table in a way that doesn't count as an
access
+ // and thus won't affect the cached entry's ttl (if enabled).
+ @VisibleForTesting
+ Optional<Table> tableFromCacheQuietly(TableIdentifier identifier) {
+ // Ensure async cleanup actions have happened.
+ tableCache.cleanUp();
+ return
Optional.ofNullable(tableCache.policy().getIfPresentQuietly(identifier));
+ }
+
+ @VisibleForTesting
+ Cache<TableIdentifier, Table> cache() {
+ return tableCache;
+ }
+
+ @VisibleForTesting
+ Optional<Duration> getTimeToTTL(TableIdentifier identifier) {
+ return tableCache
+ .policy()
+ .expireAfterAccess() // Assumes expireAfterAccess, which is what we
set at cache level.
+ .flatMap(tableExpiration -> tableExpiration.ageOf(identifier)) // Get
the time the table has been cached.
+ .map(age -> Duration.ofMillis(expirationIntervalMillis).minus(age));
+ }
+
+ private Cache<TableIdentifier, Table> createTableCache(Ticker ticker) {
+ Caffeine<TableIdentifier, Table> cacheBuilder = Caffeine
+ .newBuilder()
+ .softValues()
+ .removalListener(keyLoggingRemovalListener)
+ .writer(new CacheWriter<TableIdentifier, Table>() {
+ @Override
+ // TODO - Consider expiring and syncing any metadata tables that
have a different snapshotId
+ // upon write.
+ public void write(TableIdentifier tableIdentifier, Table table) {
+ LOG.info("Table {} was written to the catalog with snapshotId {}",
tableIdentifier,
+ table.currentSnapshot() == null ? null :
table.currentSnapshot().snapshotId());
+ }
+
+ @Override
+ public void delete(TableIdentifier tableIdentifier, Table table,
RemovalCause cause) {
+ // On expiration, remove any associated metadata tables so that
subsequent catalog loads won't
+ // return stale metadata tables w.r.t. the underlying data tables
they would return.
+ //
+ // TODO - Should we put metadata tables back into the catalog if
their associated table is still
+ // cached to keep tables and metadata tables on the same
snapshot?
+ if (expirationEnabled &&
!MetadataTableUtils.hasMetadataTableName(tableIdentifier)) {
+ onTableExpiration(tableIdentifier);
+ }
+ }
+ });
+
+ if (expirationEnabled) {
+ // Expire after write isn't going to work as we need to invalidate after
writes.
+ return
cacheBuilder.expireAfterAccess(Duration.ofMillis(expirationIntervalMillis)).ticker(ticker).build();
+ }
+ return cacheBuilder.build();
Review comment:
Added.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
