/** A provider for creating {@link LookupFunction}. */
@PublicEvolving
public interface LookupFunctionProvider extends LookupTableSource.LookupRuntimeProvider {

    static LookupFunctionProvider of(LookupFunction lookupFunction) {
        return () -> lookupFunction;
    }

    LookupFunction createLookupFunction();
}

/** A provider for creating {@link AsyncLookupFunction}. */
@PublicEvolving
public interface AsyncLookupFunctionProvider extends LookupTableSource.LookupRuntimeProvider {

    static AsyncLookupFunctionProvider of(AsyncLookupFunction asyncLookupFunction) {
        return () -> asyncLookupFunction;
    }

    AsyncLookupFunction createAsyncLookupFunction();
}

/**
 * A semi-persistent mapping from keys to values for storing entries of lookup table.
 *
 * <p>The type of the caching key is a {@link RowData} with lookup key fields packed inside. The
 * type of value is a {@link Collection} of {@link RowData}, which are rows matching lookup key
 * fields.
 *
 * <p>Cache entries are manually added using {@link #put}, and are stored in the cache until either
 * evicted or manually invalidated.
 *
 * <p>Implementations of this interface are expected to be thread-safe, and can be safely accessed
 * by multiple concurrent threads.
 */
@PublicEvolving
public interface LookupCache extends AutoClosable, Serializable {

    

/**
 * A semi-persistent mapping from keys to values for storing entries of lookup table.
 *
 * <p>The type of the caching key is a {@link RowData} with lookup key fields packed inside. The
 * type of value is a {@link Collection} of {@link RowData}, which are rows matching lookup key
 * fields.
 *
 * <p>Cache entries are manually added using {@link #put}, and are stored in the cache until either
 * evicted or manually invalidated.
 *
 * <p>Implementations of this interface are expected to be thread-safe, and can be safely accessed
 * by multiple concurrent threads.
 */
@PublicEvolving
public interface LookupCache extends AutoClosable {

    /**
     * Initialize the cache.
     *
     * @param metricGroup the metric group to register cache related metrics.
     */
    void open(CacheMetricGroup metricGroup);

    /**
     * Returns the value associated with key in this cache, or null if there is no cached value for
     * key.
     */
    @Nullable
    Collection<RowData> getIfPresent(RowData key);

    /**
     * Associates the specified value rows with the specified key row inInitialize the cache. If the cache
     *
 previously contained value associated with* the@param key,metricGroup the metric oldgroup valueto isregister replacedcache byrelated themetrics.
     */
 specified value.   void open(CacheMetricGroup metricGroup);

      /**
     * @returnReturns the previous value rows associated with key in this cache, or null if there wasis no mappingcached value for
     * key.
     */
 @param key - key@Nullable
 row with which theCollection<RowData> specified value is to be associatedgetIfPresent(RowData key);

    /**
     * @paramAssociates valuethe –specified value rows to be associated with the specified key
 row in the cache. If */the cache
    Collection<RowData> put(RowData key, Collection<RowData> value);

    /** Discards any cached value for the specified key. */
    void invalidate(RowData key);

    /** Returns the number of key-value mappings in the cache. */
    long size();
}

/** Default implementation of {@link LookupCache}. */
@PublicEvolving
public class DefaultLookupCache implements LookupCache {
    private final Duration expireAfterAccessDuration;
    private final Duration expireAfterWriteDuration;
    private final Long maximumSize;
    private final boolean cacheMissingKey;
    
    private DefaultLookupCache(
            Duration expireAfterAccessDuration,
   

/** Default implementation of {@link LookupCache}. */
@PublicEvolving
public class DefaultLookupCache implements LookupCache {
    private final Duration expireAfterAccessDuration;
    private final Duration expireAfterWriteDuration;
    private final Long maximumSize;
    private final boolean cacheMissingKey;
    
    private DefaultLookupCache(
            Duration expireAfterAccessDuration,
            Duration expireAfterWriteDuration,
            Long maximumSize,
			boolean cacheMissingKey) {
        this.expireAfterAccessDuration = expireAfterAccessDuration;
        this.expireAfterWriteDuration = expireAfterWriteDuration;
        this.initialCapacity = initialCapacity;
        this.maximumSize = maximumSize;
		this.cacheMissingKey = cacheMissingKey;
    }
    
    public static Builder newBuilder() {
        return new Builder();
    } 

   public static class Builder {         
        private Duration expireAfterAccessDuration;
        private Duration expireAfterWriteDuration;,
        private    Long maximumSize;,
			boolean cacheMissingKey) {
      private Boolean cacheMissingKey;

   this.expireAfterAccessDuration = expireAfterAccessDuration;
      public Builder expireAfterAccess(Duration duration) { this.expireAfterWriteDuration = expireAfterWriteDuration;
            expireAfterAccessDuration this.initialCapacity = durationinitialCapacity;
        this.maximumSize = maximumSize;
		this.cacheMissingKey = return thiscacheMissingKey;
        }

    
    public  publicstatic Builder expireAfterWritenewBuilder(Duration duration) {
        return new Builder();
  expireAfterWriteDuration  =} duration;

   public static class Builder {          return this;

        private Duration expireAfterAccessDuration;
        private Duration expireAfterWriteDuration;
        private Long maximumSize;
        private Boolean }cacheMissingKey;

        public Builder maximumSizeexpireAfterAccess(longDuration maximumSizeduration) {
            this.maximumSizeexpireAfterAccessDuration = maximumSizeduration;
            return this;
        }

        public Builder cacheMissingKeyexpireAfterWrite(booleanDuration cacheMissingKeyduration) {
            this.cacheMissingKeyexpireAfterWriteDuration = cacheMissingKeyduration;
            return this;
        }          

        public DefaultLookupCacheBuilder buildmaximumSize(long maximumSize) {
            returnthis.maximumSize new= DefaultLookupCache(maximumSize;
            return this;
       expireAfterAccessDuration, }

        public Builder cacheMissingKey(boolean cacheMissingKey) {
        expireAfterWriteDuration,
    this.cacheMissingKey = cacheMissingKey;
              maximumSize,
					cacheMissingKey)return this;
        }          

     }     
}

/**
 * Pre-defined metrics for cache.
 *
 * <p>Please note that these methods should only be invoked once. Registering a metric with same
 * name for multiple times would lead to an undefined behavior.
 */
@PublicEvolving
public interface CacheMetricGroup extends MetricGroup {

/**
 * Pre-defined metrics for cache.
 *
 * <p>Please note that these methods should only be invoked once. Registering a metric with same
 * name for multiple times would lead to an undefined behavior.
 */
@PublicEvolving
public interface CacheMetricGroup extends MetricGroup {
    /** The number of cache hits. */
    void hitCounter(Counter hitCounter);

    /** The number of cache misses. */
    void missCounter(Counter missCounter);

    /** The number of times to load data into cache from external system. */
    void loadCounter(Counter loadCounter);

    /** The number of load failures. */
    void numLoadFailuresCounter(Counter numLoadFailuresCounter);

    /** The time spent for the latest load operation. */
    void latestLoadTimeGauge(Gauge<Long> latestLoadTimeGauge);

    /** The number of recordscache in cachehits. */
    void numCachedRecordsGaugehitCounter(Gauge<Long>Counter numCachedRecordsGaugehitCounter);

    /** The number of bytes used by cachecache misses. */
    void numCachedBytesGaugemissCounter(Gauge<Long>Counter numCachedBytesGauge);
}

);

    /**
 * Provider for creating *{@link GetsLookupFunction} theand instance of {@link LookupCache} for storing lookup entries.
 */
@PublicEvolving
public interface PartialCachingLookupProvider extends *LookupFunctionProvider {

    /**
 * <p>This cache will be* initializedBuild bya {@link LookupCache#openPartialCachingLookupProvider} during runtime executionfrom the specified {@link LookupFunction} and
     * used for optimizing the access to external lookup table{@link LookupCache}.
     */
    static * @return an {@link Optional} of {@link LookupCache}, or an empty {@link Optional} if cachingPartialCachingLookupProvider of(LookupFunction lookupFunction, LookupCache cache) {
        return new PartialCachingLookupProvider() {

     *     shouldn't be applies@Override
 to the lookup table.
     */
   public Optional<LookupCache>LookupCache getCache(); {

     /** Builder class for {@link PartialCachingLookupProvider}. */
    class Builderreturn {cache;

         private LookupFunction lookupFunction; }

        private LookupCache cache;
        private Boolean cacheMissingKey;

@Override
        /** Sets lookup function. */
        public BuilderLookupFunction withLookupFunctioncreateLookupFunction(LookupFunction lookupFunction) {
            this.lookupFunction =    return lookupFunction;
            return this;}
        };

    }

    /** EnablesGet cachinga andnew setsinstance theof cache{@link factoryLookupCache}. */
    LookupCache getCache();
}

/**
 * Provider for creating {@link AsyncLookupFunction} and {@link LookupCache} for storing lookup entries.
 */
@PublicEvolving
public interface PartialCachingAsyncLookupProvider extends AsyncLookupFunctionProvider {

    /**    public Builder withCache(LookupCache cache, boolean cacheMissingKey) {
            this.cache = cache;
     * Build a {@link PartialCachingLookupProvider} from the this.cacheMissingKey = cacheMissingKey;specified {@link AsyncLookupFunction} and
     * {@link LookupCache}.
     */
 return this;
  static PartialCachingLookupProvider of(AsyncLookupFunction asyncLookupFunction, LookupCache cache) }{

        publicreturn PartialCachingLookupProvidernew buildPartialCachingAsyncLookupProvider() {

           ... @Override
        }
    public LookupCache getCache() }
}

/**
 optimizing the access to external* A {@link CachingLookupProvider} that never lookup table.
in external system on cache *
miss and provides a
  * @returncache anfor {@linkholding Optional}all ofentries {@link LookupCache}, or an empty {@link Optional} if caching
     *     shouldn't be applies to the lookup table.
     */
    Optional<LookupCache> getCache();

    /** Builder class for {@link AsyncPartialCachingLookupProvider}. */
    class Builder {

in the external system. The cache will be fully reloaded from the
 * external system by the {@link ScanTableSource.ScanRuntimeProvider} and reload operations will be
 * triggered by the {@link CacheReloadTrigger}.
 */
@PublicEvolving
public interface FullCachingLookupProvider extends LookupFunctionProvider {
    static FullCachingLookupProvider of(
            ScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
            CacheReloadTrigger cacheReloadTrigger) private{
 AsyncLookupFunction asyncLookupFunction;
      return  private LookupCache cache;
new FullCachingLookupProvider() {
          private Boolean cacheMissingKey;@Override

          /** Sets lookuppublic functionScanTableSource. */ScanRuntimeProvider getScanRuntimeProvider() {
        public AsyncPartialCachingLookupProvider.Builder withLookupFunction(
      return scanRuntimeProvider;
         AsyncLookupFunction  asyncLookupFunction) {}

            this.asyncLookupFunction@Override
 = asyncLookupFunction;
          public CacheReloadTrigger return this;getCacheReloadTrigger() {
        }

         /** Enables caching and sets the cache factory. */
   return cacheReloadTrigger;
            }

     public AsyncPartialCachingLookupProvider.Builder withCache(
     @Override
           LookupCache cache,public booleanLookupFunction cacheMissingKeycreateLookupFunction() {
            this.cache = cache;
  return keyRow -> null;
       this.cacheMissingKey = cacheMissingKey;
    }
        return this};
    }

    }/**

     * Get a  public AsyncPartialCachingLookupProvider build() {{@link ScanTableSource.ScanRuntimeProvider} for scanning all entries from the external
     * lookup table and load into the cache...
     */
     }ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();

    }
}

/** A @Override
trigger that reloads all entries periodically with specified interval or delay.  */
public FullCachingReloadTriggerclass getReloadTrigger() PeriodicCacheReloadTrigger implements CacheReloadTrigger {

    private final Duration reloadInterval;
    private final ScheduleMode scheduleMode;

     return fullCachingReloadTrigger;public PeriodicCacheReloadTrigger(Duration reloadInterval, ScheduleMode scheduleMode) {
        this.reloadInterval = reloadInterval;
  }
      this.scheduleMode = }scheduleMode;
    }

    /**@Override
    public * Gets the {@link ScanTableSource.ScanRuntimeProvider} for executing the periodically reload.
     */void open(CacheReloadTrigger.Context context) {
        // Register periodic reload task
    }

    @Override
    ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();

public void close() throws Exception {
        /** Get the {@link FullCachingReloadTrigger} for triggering a full caching reload operation. */// Dispose resources
    }

    public enum ScheduleMode {
        FIXED_DELAY,
    FullCachingReloadTrigger getReloadTrigger();
}