/** 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 {

    /**
     * Initialize the cache.
     

/**
 * 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 {

    /**
     * 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.
      *
     * @return@param metricGroup the previousmetric valuegroup rowsto associatedregister withcache key, or null if there was no mapping for key.related metrics.
     */
    void open(CacheMetricGroup metricGroup);

    /**
 @param key - key row* with whichReturns the specifiedvalue valueassociated iswith tokey bein associated
this cache, or null if *there @paramis valueno –cached value rowsfor
 to be associated with the specified* key.
     */
    @Nullable
    Collection<RowData> putgetIfPresent(RowData key, Collection<RowData> value);

    /**
     * Copies allAssociates the mappings fromspecified value rows with the specified mapkey row toin the cache. TheIf effect of this call isthe cache
     * equivalentpreviously tocontained thatvalue ofassociated callingwith {@code put(kthe key, v)}the onold thisvalue mapis oncereplaced for each mapping from keyby the
     * {@code k} to value {@code v} in the specified map. The behavior of this operation is
     * undefined if the specified map is modified while the operation is in progress.
     */
    void putAll(Map<? extends RowData, ? extends Collection<RowData>> m);

    /** Discards any cached value for the specified key.specified value.
      *
     * @return the previous value rows associated with key, or null if there was no mapping for key.
     * @param key - key row with which the specified value is to be associated
     * @param value – value rows to be associated with the specified key
     */
    voidCollection<RowData> invalidateput(RowData key, Collection<RowData> value);

    /** Discards allany cached entriesvalue infor the cachespecified key. */
    void invalidateAllinvalidate(RowData key);

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

/**
 * 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 records in cache. */
    void numCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);

/** Pre-defined metrics for {@code LookupCache}. */
@PublicEvolving
public interface LookupCacheMetricGroup extends MetricGroup {
    /** The number of cache hitsbytes used by cache. */
    void setHitCounternumCachedBytesGauge(CounterGauge<Long> hitCounternumCachedBytesGauge);
}

/**
 * Provider  /** The number of cache misses. */
    void setMissCounter(Counter missCounter);for creating {@link LookupFunction} and {@link LookupCache} for storing lookup entries.
 */
@PublicEvolving
public interface PartialCachingLookupProvider extends LookupFunctionProvider {

    /**
 The number of times to* loadBuild dataa into{@link cachePartialCachingLookupProvider} from external system. */
the specified {@link LookupFunction} and
     void setLoadCounter(Counter loadCounter);

* {@link LookupCache}.
     /** The number of load failures. */*/
    static PartialCachingLookupProvider of(LookupFunction lookupFunction, LookupCache cache) {
    void setNumLoadFailuresCounter(Counter numLoadFailuresCounter);    return new PartialCachingLookupProvider() {

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

    /** The number of recordspublic in cache. */LookupCache getCache() {
    void setNumCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);

    /** The number of bytes used byreturn cache. */;
    void setNumCachedBytesGauge(Gauge<Long> numCachedBytesGauge);
}

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

     return lookupFunction;/**
     * Build a {@link PartialCachingLookupProvider} from the }

specified {@link AsyncLookupFunction} and
     *    @Override{@link LookupCache}.
     */
    static PartialCachingLookupProvider of(AsyncLookupFunction publicasyncLookupFunction, Optional<LookupCacheFactory>LookupCache getCacheFactory(cache) {
        return        return Optional.emptynew PartialCachingAsyncLookupProvider();
            } {

            @Override
            public Optional<Boolean>LookupCache cacheMissingKeygetCache() {
                return Optional.empty()cache;
            }

        };
    }@Override

    /**
     * Creates {@link LookupFunctionProvider}public withAsyncLookupFunction thecreateAsyncLookupFunction() given {@link
 LookupFunction} and enable
     * caching with specified {@link LookupCacheFactory}.
   return asyncLookupFunction;
 */
    static LookupFunctionProvider of(
     }
       LookupFunction lookupFunction,};
    }

    /** Get a new LookupCacheFactoryinstance cacheFactory,
of {@link LookupCache}. */
    LookupCache getCache();
}

/**
 * A {@link CachingLookupProvider} that never lookup in external system on cache miss and provides a
 * cache for holding all entries 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(     boolean cacheMissingKey) {
        return new LookupFunctionProvider() {
            @Override
            public LookupFunction createLookupFunction() {
                return lookupFunction;
            }

            @OverrideScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
            public Optional<LookupCacheFactory> getCacheFactory(CacheReloadTrigger cacheReloadTrigger) {
        return        return Optional.of(cacheFactory);
            }

new FullCachingLookupProvider() {
            @Override
            public Optional<Boolean>ScanTableSource.ScanRuntimeProvider cacheMissingKeygetScanRuntimeProvider() {
                return Optional.of(cacheMissingKey)scanRuntimeProvider;
            }

        };
    }@Override

    /** Creates an {@link LookupFunction} instance. */
  public CacheReloadTrigger LookupFunction createLookupFunctiongetCacheReloadTrigger(); {

    /**
     * Gets the {@link LookupCacheFactory} for creating lookupreturn cache.cacheReloadTrigger;
     *
     * <p>This factory will be used for creating an instance of cache during runtime execution for }

            @Override
     * optimizing the access to external lookup table.
public LookupFunction createLookupFunction() {
  *
     * @return an {@link Optional} of {@link LookupCacheFactory}, or anreturn emptykeyRow {@link Optional} if-> null;
     *     caching shouldn't be}
 applies to the lookup table.
     */};
    Optional<LookupCacheFactory> getCacheFactory();}

    /**
     * Get Whethera the missing key (key fields without any matching value rows) should be stored in the
     *{@link ScanTableSource.ScanRuntimeProvider} for scanning all entries from the external
     * lookup table and load into the cache.
     */
     * <p>Please note that this option is required if {@link #getCacheFactory()} returns a non-emptyScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();

    /** Get a {@link CacheReloadTrigger} for triggering the reload operation. */
    CacheReloadTrigger getCacheReloadTrigger();
}

/** instance.Customized Iftrigger thefor cachereloading factoryall islookup empty,table theentries returnin valuefull of this function will be ignored.caching mode. */
@PublicEvolving
public interface CachingReloadTrigger extends AutoCloseable, Serializable {

     /**
 Open the trigger.  */
 @return true if avoid null or empty value should be stored in the cache.
     */
    Optional<Boolean> cacheMissingKey();
}

/** 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;
        this.scheduleMode = scheduleMode;
    }

    @Override
    public Optional<Boolean>void cacheMissingKeyopen(CacheReloadTrigger.Context context) {
        // Register periodic reload task
    return Optional.empty();
}

    @Override
    public void close() throws Exception }{
         };// Dispose resources
    }

    /**public enum ScheduleMode {
     * Creates {@link AsyncLookupFunctionProvider} with the given {@link AsyncLookupFunction} and   FIXED_DELAY,
        FIXED_RATE
     }
}

/** enableA cachingtrigger withthat specifiedreloads {@link LookupCacheFactory}.
     */
    static AsyncLookupFunctionProvider of(
       at a specific local time and repeat for the given interval in days. */ 
public class TimedCacheReloadTrigger implements CacheReloadTrigger {

    private AsyncLookupFunction asyncLookupFunction,final LocalTime reloadTime;
    private final int reloadIntervalInDays;

    public LookupCacheFactoryTimedCacheReloadTrigger(LocalTime cacheBuilderreloadTime,
            boolean cacheMissingKeyint reloadIntervalInDays) {
        returnthis.reloadTime new AsyncLookupFunctionProvider() {= reloadTime;
        this.reloadIntervalInDays =   @OverridereloadIntervalInDays;
    }

    @Override
    public AsyncLookupFunctionvoid createAsyncLookupFunctionopen(Context context) {
		// Register periodic              return asyncLookupFunction;
       reload task
     }

            @Override
            public Optional<LookupCacheFactory>void getCacheFactoryclose() throws Exception {
		// Dispose resources
              return Optional.of(cacheBuilder);
            }

            @Override
            public Optional<Boolean> cacheMissingKey() {
                return Optional.of(cacheMissingKey);
            }
        };
    }

    /** Creates an {@link AsyncLookupFunction} instance. */
    AsyncLookupFunction createAsyncLookupFunction();

    /**
     * Gets the {@link LookupCacheFactory} for creating lookup cache.
     *
     * <p>This factory will be used for creating an instance of cache during runtime execution for
     * optimizing the access to external lookup table.
     *
     * @return an {@link Optional} of {@link LookupCacheFactory}, or an empty {@link Optional} if
     *     caching shouldn't be applies to the lookup table.
     */
    Optional<LookupCacheFactory> getCacheFactory();

    /**
     * Whether the missing key (key fields without any matching value rows) should be stored in the
     * cache.
     *
     * <p>Please note that this option is required if {@link #getCacheFactory()} returns a non-empty
     * instance. If the cache factory is empty, the return value of this function will be ignored.
     *
     * @return true if a null or empty value should be stored in the cache.
     */
    Optional<Boolean> cacheMissingKey();
}