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

    /**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 valuegroup rowsto associatedregister withcache key, or null if there was no mapping for keyrelated metrics.
     */
 @param key - key row with whichvoid open(CacheMetricGroup metricGroup);

    /**
     * Returns the specified value isassociated towith bekey associated
in 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);

    /**
 Discards any cached value for * Associates the specified value rows with the specified key. */
row in the cache. voidIf invalidate(RowData key);

the cache
     /** 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 expireAfterWriteDurationprivate final Integer initialCapacity;
    private final Long maximumSize;

    public static DefaultLookupCacheFactory.Builder newBuilder() {
        return new Builder();
    }

private final Long maximumSize;
    private DefaultLookupCacheFactory(
final boolean cacheMissingKey;
    
      Duration expireAfterAccessDuration,private DefaultLookupCache(
            Duration expireAfterWriteDurationexpireAfterAccessDuration,
            IntegerDuration initialCapacityexpireAfterWriteDuration,
            Long maximumSize,
			boolean cacheMissingKey) {
        // Validation
        this.expireAfterAccessDuration this.expireAfterAccessDuration = expireAfterAccessDuration;
        this.expireAfterWriteDuration = expireAfterWriteDuration;
        this.initialCapacity = initialCapacity;
        this.maximumSize = maximumSize;
		this.cacheMissingKey = cacheMissingKey;
    }

    @Override
    public static publicBuilder LookupCache createCache(LookupCacheMetricGroup metricGroupnewBuilder() {
        //return Create instance of DefaultLookupCachenew Builder();
    } 

   public static /**class Builder of {@link DefaultLookupCacheFactory}. */
    public static class Builder {        
        private Duration expireAfterAccessDuration;
        private Duration expireAfterWriteDuration;
        private IntegerLong initialCapacitymaximumSize;
        private LongBoolean maximumSizecacheMissingKey;

        public DefaultLookupCacheFactory.Builder expireAfterAccess(Duration duration) {
            expireAfterAccessDuration = duration;
            return this;
        }

        public DefaultLookupCacheFactory.Builder expireAfterWrite(Duration duration) {
            expireAfterWriteDuration = duration;
            return this;
        }

        public DefaultLookupCacheFactory.Builder initialCapacitymaximumSize(intlong initialCapacitymaximumSize) {
            this.initialCapacitymaximumSize = initialCapacitymaximumSize;
            return this;
        }

        public DefaultLookupCacheFactory.Builder maximumSizecacheMissingKey(longboolean maximumSizecacheMissingKey) {
            this.maximumSizecacheMissingKey = maximumSizecacheMissingKey;
            return this;
        }          

        public DefaultLookupCacheFactoryDefaultLookupCache build() {
            return new DefaultLookupCacheFactoryDefaultLookupCache(
                    expireAfterAccessDuration,
                    expireAfterWriteDuration,
                    initialCapacity,
                    maximumSizemaximumSize,
					cacheMissingKey);
        }
    }     
}

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

    /**
     * instance. If the cache factory is empty, the return value of this function will be ignoredBuild a {@link PartialCachingLookupProvider} from the specified {@link AsyncLookupFunction} and
     * {@link LookupCache}.
     */
    static *PartialCachingLookupProvider @return true if a null or empty value should be stored in the cache.of(AsyncLookupFunction asyncLookupFunction, LookupCache cache) {
        return new PartialCachingAsyncLookupProvider() {

     */
      Optional<Boolean> cacheMissingKey();

    /** Builder class for {@link LookupFunctionProvider}. */
@Override
            public classLookupCache BuildergetCache() {

        private LookupFunction lookupFunction;
       return privatecache;
 LookupCacheFactory cacheFactory;
        private Boolean enableCacheMissingKey;}

        /** Sets lookup function. */
    @Override
            public BuilderAsyncLookupFunction withLookupFunctioncreateAsyncLookupFunction(LookupFunction lookupFunction) {
            this.lookupFunction = lookupFunction    return asyncLookupFunction;
            return this;}
        }
;
    }

    /** EnablesGet cachinga andnew setsinstance theof cache{@link factoryLookupCache}. */
    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(
    public Builder withCacheFactory(LookupCacheFactory cacheFactory) {
            this.cacheFactory = cacheFactory;
            return this;
        }

        /**
         * Enables storing missing key in the cache.
         *
         * <p>See {@link LookupFunctionProvider#cacheMissingKey()} for more detailsScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
         */
   CacheReloadTrigger cacheReloadTrigger) {
   public Builder enableCacheMissingKey() {
   return new FullCachingLookupProvider() {
            @Override
    this.enableCacheMissingKey = true;
        public ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider() {
                return thisscanRuntimeProvider;
            }

            @Override
            public LookupFunctionProviderCacheReloadTrigger buildgetCacheReloadTrigger() {
            // Build LookupFunctionProvider    return cacheReloadTrigger;
        }
    }
}

/** 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)  return Optional.empty()throws Exception;

    /**
     * Context of {@link CacheReloadTrigger}

 for getting information about times and
     * triggering @Overridereload.
     */
    interface Context {

 public Optional<Boolean> cacheMissingKey() {
    /** Get current processing time. */
        returnlong Optional.emptycurrentProcessingTime();

        /** Get current watermark }
on the main stream. */
    };
    long }currentWatermark();

    /**
     /** Creates {@link AsyncLookupFunctionProvider} withTrigger a reload operation on the givenfull {@link AsyncLookupFunction} andcache. */
     * enable caching with specified {@link LookupCacheFactory}.CompletableFuture<Void> triggerReload();
    }
}

/** A trigger that reloads all entries periodically with specified interval or delay. */
public class PeriodicCacheReloadTrigger implements CacheReloadTrigger {
 */
    static AsyncLookupFunctionProvider of(
            AsyncLookupFunction asyncLookupFunction,
            LookupCacheFactory cacheBuilder,
    private final Duration reloadInterval;
    private booleanfinal cacheMissingKey) {ScheduleMode scheduleMode;

    public PeriodicCacheReloadTrigger(Duration reloadInterval,  return new AsyncLookupFunctionProvider(ScheduleMode scheduleMode) {
        this.reloadInterval =   @OverridereloadInterval;
        this.scheduleMode = scheduleMode;
  public AsyncLookupFunction createAsyncLookupFunction() { }

    @Override
    public void open(CacheReloadTrigger.Context context) {
    return asyncLookupFunction;
   // Register periodic reload task
     }

    @Override
    public void close() throws Exception @Override{
        // Dispose resources
  public Optional<LookupCacheFactory> getCacheFactory() { }

    public enum ScheduleMode {
         return Optional.of(cacheBuilder);FIXED_DELAY,
        FIXED_RATE
    }

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

/** 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) {
        this.reloadTime = reloadTime;
  }
      this.reloadIntervalInDays = }reloadIntervalInDays;
    }

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

    /**@Override
    public void open(Context context) {
		// Register periodic reload task
    }

    @Override
 * Gets the {@linkpublic LookupCacheFactory} for creating lookup cache.
     *void close() throws Exception {
		// Dispose resources
     * <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();
    
    /** Builder class for {@link AsyncLookupFunctionProvider}. */
    class Builder {

        private AsyncLookupFunction asyncLookupFunction;
        private LookupCacheFactory cacheFactory;
        private Boolean enableCacheMissingKey;

        /** Sets lookup function. */
        public Builder withAsyncLookupFunction(AsyncLookupFunction asyncLookupFunction) {
            this.asyncLookupFunction = asyncLookupFunction;
            return this;
        }

        /** Enables caching and sets the cache factory. */
        public Builder withCacheFactory(LookupCacheFactory cacheFactory) {
            this.cacheFactory = cacheFactory;
            return this;
        }

        /**
         * Enables storing missing key in the cache.
         *
         * <p>See {@link AsyncLookupFunctionProvider#cacheMissingKey()} for more details.
         */
        public Builder enableCacheMissingKey() {
            this.enableCacheMissingKey = true;
            return this;
        }

        public AsyncLookupFunctionProvider build() {
            // Build AsyncLookupFunctionProvider
        }
    }
}