/*

/**
 * A semi-persistentprovider mappingfor fromcreating {@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 {

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

    /*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 in the cache. If the cache
     * previously contained value associated with the key, the old value is replaced by the
     * specified value.
      *
     * @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(
   

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

   public static class Builder {
        private Duration expireAfterAccessDuration;
        private Duration expireAfterWriteDuration;
expireAfterAccessDuration,
           private LongDuration maximumSize;

expireAfterWriteDuration,
           public DefaultLookupCache.Builder expireAfterAccess(Duration durationLong maximumSize,
			boolean cacheMissingKey) {
            this.expireAfterAccessDuration = durationexpireAfterAccessDuration;
        this.expireAfterWriteDuration =   return thisexpireAfterWriteDuration;
        }
this.initialCapacity = initialCapacity;
        public DefaultLookupCache.Builder expireAfterWrite(Duration duration) {this.maximumSize = maximumSize;
		this.cacheMissingKey = cacheMissingKey;
    }
    
    public static expireAfterWriteDurationBuilder = duration;newBuilder() {
        return    return thisnew Builder();
      }  }

   public static class Builder {   public  DefaultLookupCache.Builder maximumSize(long maximumSize) {    
        private Duration expireAfterAccessDuration;
  this.maximumSize = maximumSize;
         private Duration expireAfterWriteDuration;
        private returnLong thismaximumSize;
        private Boolean }cacheMissingKey;

        public DefaultLookupCacheBuilder buildexpireAfterAccess(Duration duration) {
            returnexpireAfterAccessDuration new= DefaultLookupCache(duration;
            return this;
        expireAfterAccessDuration,}

        public Builder expireAfterWrite(Duration duration) {
        expireAfterWriteDuration,
    expireAfterWriteDuration = duration;
            return  maximumSize)this;
        }

    }     public   

}

/**
 * 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);

//**
 * Provider for creating {@link LookupFunction} and {@link LookupCache} if caching should be enabled
 * for the lookup table.
 */
@PublicEvolving
public interface PartialCachingLookupProvider extends LookupTableSource.LookupRuntimeProvider {

    /** Creates a builder of {@link PartialCachingLookupProvider}. */
    static Builder newBuilder() {
        return new Builder();
    }

    /** CreatesThe annumber {@linkof LookupFunction}cache instancemisses. */
    LookupFunctionvoid createLookupFunction()missCounter(Counter missCounter);

    /**
 The number of times *to load Getsdata theinto instancecache offrom {@linkexternal LookupCache}system. */
     *void loadCounter(Counter loadCounter);

     /** <p>ThisThe cachenumber willof beload initialized by {@link LookupCache#open} during runtime execution andfailures. */
    void numLoadFailuresCounter(Counter numLoadFailuresCounter);

     * used/** The time spent for optimizing the accesslatest toload external lookup table.
operation. */
    void  *latestLoadTimeGauge(Gauge<Long> latestLoadTimeGauge);

     /** @returnThe annumber {@linkof Optional}records ofin {@link LookupCache}, or an empty {@link Optional} if cachingcache. */
    void numCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);

    /** *The number of bytes used shouldn't be applies to the lookup table.
     */
    Optional<LookupCache> getCache();

    by cache. */
    void numCachedBytesGauge(Gauge<Long> numCachedBytesGauge);
}

/**
 * Provider for creating *{@link WhetherLookupFunction} theand missing{@link keyLookupCache} (keyfor fieldsstoring withoutlookup any matching value rows) should be stored in theentries.
 */
@PublicEvolving
public interface PartialCachingLookupProvider extends LookupFunctionProvider {

     /** cache.
     *
 Build a {@link PartialCachingLookupProvider} from * <p>Please note that this option is required ifthe specified {@link LookupFunction} and
     * {@link #getCache} returns a non-emptyLookupCache}.
     */
 instance. If the cachestatic factoryPartialCachingLookupProvider isof(LookupFunction emptylookupFunction, theLookupCache returncache) value{
 of this function will be ignored.
  return new PartialCachingLookupProvider() *{

     * @return true if a null or empty@Override
 value should be stored in the cache.
     */
public LookupCache   Optional<Boolean> cacheMissingKeygetCache(); {

     /** 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 publicfor Builder withCache(LookupCache cache, boolean cacheMissingKey) {
            this.cache = cache;creating {@link AsyncLookupFunction} and {@link LookupCache} for storing lookup entries.
 */
@PublicEvolving
public interface PartialCachingAsyncLookupProvider extends AsyncLookupFunctionProvider {

    /**
     * Build a {@link PartialCachingLookupProvider} from  this.cacheMissingKey = cacheMissingKey;the 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();

    /**
     * Whether the missing key (key fields without any matching value rows) should be stored in thein 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) {
     * cache.
  return new FullCachingLookupProvider() *{
     *  <p>Please note that this option is@Override
 required if {@link #getCache} returns a non-empty
     *public instanceScanTableSource.ScanRuntimeProvider If the cache factory is empty, the return value of this function will be ignored.
     *getScanRuntimeProvider() {
                return scanRuntimeProvider;
     * @return true if a null or empty}

 value should be stored in the cache.
     */@Override
    Optional<Boolean> cacheMissingKey();

    /** Builder class forpublic {@link AsyncPartialCachingLookupProvider}. */
CacheReloadTrigger getCacheReloadTrigger() {
     class Builder {

        private AsyncLookupFunctionreturn asyncLookupFunctioncacheReloadTrigger;
          private LookupCache cache;}

        private   Boolean cacheMissingKey;@Override

        /** Sets lookup function. */
public LookupFunction createLookupFunction() {
     public AsyncPartialCachingLookupProvider.Builder withLookupFunction(
         return keyRow -> null;
    AsyncLookupFunction asyncLookupFunction) {
      }
      this.asyncLookupFunction = asyncLookupFunction};
    }

    /**
    return this;
* Get a {@link ScanTableSource.ScanRuntimeProvider} for scanning all }

entries from the external
     /** Enableslookup cachingtable and setsload into the cache factory. */
     */
   public AsyncPartialCachingLookupProviderScanTableSource.BuilderScanRuntimeProvider withCachegetScanRuntimeProvider();

    /** Get a {@link CacheReloadTrigger} for triggering the reload operation. */
  LookupCache cache, booleanCacheReloadTrigger cacheMissingKeygetCacheReloadTrigger() {
            this.cache = cache;
            this.cacheMissingKey = cacheMissingKey;;
}

/** Customized trigger for reloading all lookup table entries in full caching mode. */
@PublicEvolving
public interface CachingReloadTrigger extends AutoCloseable, Serializable {

    /** Open the trigger. */
    void open(Context context) throws Exception;

    return this;/**
     * Context of {@link CacheReloadTrigger}

 for getting information about times and
  public   AsyncPartialCachingLookupProvider* build() {triggering reload.
     */
    interface Context ...{

        }
/** Get current  }
}

/** A trigger that reloads all entries periodically with specified interval or delay. */
public class PeriodicCacheReloadTrigger implements CacheReloadTrigger {

    private final Duration reloadInterval;
    private final ScheduleMode scheduleMode;

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

/**
 * Runtime provider for fully loading and periodically reloading all entries of the lookup table and
 * storing the table locally for lookup.
 *
 * <p>Implementations should provide a {@link ScanTableSource.ScanRuntimeProvider} in order to reuse
 * the ability of scanning for loading all entries from the lookup table.
 */
@PublicEvolving
public interface FullCachingLookupProvider extends LookupTableSource.LookupRuntimeProvider {

    static FullCachingLookupProvider of(
            ScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
            FullCachingReloadTrigger fullCachingReloadTrigger) {
        return new FullCachingLookupProvider() {
            @Override
            public ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider() {
               this.scheduleMode return= scanRuntimeProviderscheduleMode;
    }

    @Override
    public void  }

open(CacheReloadTrigger.Context context) {
        // Register periodic reload @Overridetask
    }

    @Override
    public FullCachingReloadTriggervoid getReloadTriggerclose() throws Exception {
        // Dispose resources
    }

  return fullCachingReloadTrigger;
 public enum ScheduleMode {
        }FIXED_DELAY,
        };FIXED_RATE
    }

    }

/**
 A trigger that reloads at a *specific Getslocal thetime {@link ScanTableSource.ScanRuntimeProvider}and repeat for executing the periodicallygiven reload.
interval in days.   */ 
public class TimedCacheReloadTrigger implements ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();CacheReloadTrigger {

    /**private Getfinal the {@link FullCachingReloadTrigger} for triggering a full caching reload operation. */LocalTime reloadTime;
    private final int reloadIntervalInDays;

    FullCachingReloadTriggerpublic getReloadTrigger();
}