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

    /**
     * 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
     * specifiedInitialize the valuecache.
        *
     * @return@param metricGroup the previousmetric group valueto rowsregister associatedcache 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 keyrelated 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> putgetIfPresent(RowData key, Collection<RowData> value);

    /**
     * CopiesAssociates all 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(k, v)} on this map once for each mapping from key
     * {@code k} to value {@code v} in the specified map. The behavior of this operation isthe key, the old value is replaced by the
     * specified value.
      *
     * @return the previous value rows associated with key, or null if there was no mapping for key.
     * undefined@param ifkey the- specifiedkey maprow iswith modifiedwhich whilethe thespecified operationvalue is in progress.to be associated
     */
 @param value – void putAll(Map<? extends RowData, ? extends Collection<RowData>> m);

    /** Discards any cached value for the specified key. 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 thiscache.maximumSize = maximumSize;
 *
 * <p>Please note that these methods should only be invoked once. returnRegistering this;
a metric with same
 * name for  }

        public DefaultLookupCacheFactory build()multiple times would lead to an undefined behavior.
 */
@PublicEvolving
public interface CacheMetricGroup extends MetricGroup {
    /** The number of cache hits. */
  return  newvoid DefaultLookupCacheFactoryhitCounter(Counter hitCounter);

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

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

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

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

     maximumSize);
        }
    }
}

/**
 The* numberProvider offor cachecreating misses. */
    void setMissCounter(Counter missCounter);{@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  static PartialCachingLookupProvider of load failures. */
(LookupFunction lookupFunction, LookupCache cache) {
        return voidnew setNumLoadFailuresCounterPartialCachingLookupProvider(Counter) numLoadFailuresCounter);{

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

    /** Thepublic numberLookupCache of records in cache. */
getCache() {
       void setNumCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);         return cache;
            }

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

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

    /**
     * Build a {@link PartialCachingLookupProvider} from the specified {@link AsyncLookupFunction} and
     * {@link LookupCache}.
     */
    static PartialCachingLookupProvider of(AsyncLookupFunction asyncLookupFunction, LookupCache cache) {

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

    /**
     * Creates {@link LookupFunctionProvider} with the given {@link LookupFunction} and disable
     * lookup table caching.
     */
    static LookupFunctionProvider of(LookupFunction lookupFunction) {
        return new LookupFunctionProvider() {
            @Override
           return publicnew LookupFunction createLookupFunctionPartialCachingAsyncLookupProvider() {

                return lookupFunction;
            }

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

            @Override
            public Optional<Boolean>AsyncLookupFunction cacheMissingKeycreateAsyncLookupFunction() {
                return Optional.empty()asyncLookupFunction;
            }
        };
    }

    /**
 Get a new instance *of Creates {@link LookupFunctionProviderLookupCache}. with*/
 the given {@link LookupFunction} and enable
     * caching with specified {@link LookupCacheFactory}.
     */
    static LookupFunctionProvider of(
            LookupFunction lookupFunction,
            LookupCacheFactory cacheFactory,
            boolean cacheMissingKey)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    return new LookupFunctionProvider() {
FullCachingLookupProvider of(
            ScanTableSource.ScanRuntimeProvider @OverridescanRuntimeProvider,
            public LookupFunction createLookupFunction(CacheReloadTrigger cacheReloadTrigger) {
                return lookupFunction;
            }

new FullCachingLookupProvider() {
            @Override
            public Optional<LookupCacheFactory>ScanTableSource.ScanRuntimeProvider getCacheFactorygetScanRuntimeProvider() {
                return Optional.of(cacheFactory)scanRuntimeProvider;
            }

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

        };
    }@Override

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

    /**
     * Gets the {@link LookupCacheFactory} for creating lookupreturn cache.
keyRow     *-> null;
     * <p>This factory will be used for creating}
 an instance of cache during runtime execution for};
     * optimizing the access to external lookup table.}

     /**
     * @returnGet ana {@link OptionalScanTableSource.ScanRuntimeProvider} offor {@link LookupCacheFactory}, or an empty {@link Optional} ifscanning all entries from the external
     * lookup table and load caching shouldn't be applies to the lookup tableinto the cache.
     */
    Optional<LookupCacheFactory>ScanTableSource.ScanRuntimeProvider getCacheFactorygetScanRuntimeProvider();

    /**
 Get a {@link CacheReloadTrigger} *for Whethertriggering the missingreload key (key fields without any matching value rows) should be stored in the
     * cache.
     *operation. */
    CacheReloadTrigger getCacheReloadTrigger();
}

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

     /** <p>PleaseOpen notethe that this option is required if {@link #getCacheFactory()} returns a non-emptytrigger. */
    void open(Context context) throws Exception;

    /**
     * instance. If the cache factory is empty, the return value of this function will be ignoredContext of {@link CacheReloadTrigger} for getting information about times and
     * triggering reload.
     */
    interface *Context @return{

 true if a null or empty value should/** beGet storedcurrent inprocessing the cache.time. */
     */
   long Optional<Boolean> cacheMissingKeycurrentProcessingTime();
}

/** A trigger that returnreloads newall AsyncLookupFunctionProvider() {
            @Overrideentries periodically with specified interval or delay. */
public class PeriodicCacheReloadTrigger implements CacheReloadTrigger {

    private final Duration reloadInterval;
    private publicfinal AsyncLookupFunction createAsyncLookupFunction() {ScheduleMode scheduleMode;

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

    @Override
    public void open(CacheReloadTrigger.Context context) @Override{
        // Register periodic reload task
 public Optional<LookupCacheFactory> getCacheFactory() {}

    @Override
    public void close() throws Exception {
   return Optional.empty();
      // Dispose resources
    }

    public enum ScheduleMode {
     @Override
   FIXED_DELAY,
        FIXED_RATE
 public Optional<Boolean> cacheMissingKey() {
                return Optional.empty();
            }}
}

/** A trigger that reloads at a specific local time and repeat for the given interval in days. */ 
public class TimedCacheReloadTrigger implements CacheReloadTrigger {

    private final LocalTime reloadTime;
    private final int  }reloadIntervalInDays;

     }

    /**public TimedCacheReloadTrigger(LocalTime reloadTime, int reloadIntervalInDays) {
     * Creates {@link AsyncLookupFunctionProvider} with the given {@link AsyncLookupFunction} and
 this.reloadTime = reloadTime;
        this.reloadIntervalInDays *= enablereloadIntervalInDays;
 caching with specified {@link LookupCacheFactory}.

     */@Override
    staticpublic AsyncLookupFunctionProvidervoid ofopen(
Context context) {
		// Register periodic reload task
      AsyncLookupFunction asyncLookupFunction,}

            LookupCacheFactory cacheBuilder,@Override
    public void close() throws     boolean cacheMissingKey) Exception {
		// Dispose resources
      return new AsyncLookupFunctionProvider() {
            @Override
            public AsyncLookupFunction createAsyncLookupFunction() {
                return asyncLookupFunction;
            }

            @Override
            public Optional<LookupCacheFactory> getCacheFactory() {
                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();
}}
}