/** 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.
        *
     * @param @returnmetricGroup the previousmetric valuegroup rowsto associatedregister withcache 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);

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

value rows with the specified key row in the cache. If the cache
     /** Returns* previously contained value associated with the number of key-value mappings in the cache. */key, the old value is replaced by the
     * specified value.
      *
    long size();
}

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

     /** cache.
     *
 Build a {@link PartialCachingLookupProvider} from * <p>Please note that this option is required ifthe specified {@link AsyncLookupFunction} and
     * {@link #getCacheFactory()LookupCache}.
 returns   a non-empty*/
    static * instance. If the cache factory is empty, the return value of this function will be ignored.
PartialCachingLookupProvider of(AsyncLookupFunction asyncLookupFunction, LookupCache cache) {
        return new PartialCachingAsyncLookupProvider() {

      *
      @Override
 * @return true if a null or empty value should be storedpublic inLookupCache the cache.getCache() {
     */
    Optional<Boolean> cacheMissingKey();

    /** Builder class for {@link LookupFunctionProvider}. */return cache;
    class Builder {

      }

  private LookupFunction lookupFunction;
        private LookupCacheFactory cacheFactory;
 @Override
       private Boolean enableCacheMissingKey;

    public AsyncLookupFunction createAsyncLookupFunction() {
 /** Sets lookup function. */
        public Builder withLookupFunction(LookupFunction lookupFunction)return {asyncLookupFunction;
            this.lookupFunction = lookupFunction;}
        };
    return this;
        }

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

/**
 * A public{@link Builder withCacheFactory(LookupCacheFactory cacheFactory) {
            this.cacheFactory = cacheFactory;
            return this;
        }

        /**
         * Enables storing missing key in the cache.
         *
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(
            * <p>See {@link LookupFunctionProvider#cacheMissingKey()} for more details.
ScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
            CacheReloadTrigger cacheReloadTrigger) */{
        publicreturn Buildernew enableCacheMissingKeyFullCachingLookupProvider() {
            this.enableCacheMissingKey@Override
 = true;
          public  return this;ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider() {
        }

        return publicscanRuntimeProvider;
 LookupFunctionProvider build() {
         }

   //  Build LookupFunctionProvider
      @Override
  }
    }
}

/** Customized trigger for reloading all lookup publictable Optional<Boolean>entries cacheMissingKey() {
         in full caching mode. */
@PublicEvolving
public interface CachingReloadTrigger extends AutoCloseable, Serializable {

    /** Open the return Optional.empty();
            }trigger. */
    void open(Context context) throws }Exception;
    }

    /**
     * Context Createsof {@link AsyncLookupFunctionProviderCacheReloadTrigger} withfor thegetting giveninformation {@linkabout AsyncLookupFunction}times and
     * enable caching with specified {@link LookupCacheFactory}triggering reload.
     */
    staticinterface AsyncLookupFunctionProviderContext of({

        /** Get current processing AsyncLookupFunction asyncLookupFunction,time. */
          long currentProcessingTime();

  LookupCacheFactory cacheBuilder,
     /** Get current watermark on the main boolean cacheMissingKey) {stream. */
        returnlong new AsyncLookupFunctionProvidercurrentWatermark() {;

        /** Trigger a reload @Override
operation on the full cache. */
       public AsyncLookupFunctionCompletableFuture<Void> createAsyncLookupFunctiontriggerReload() {;
    }
}

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

    private final Duration }reloadInterval;

    private final ScheduleMode  scheduleMode;

    @Override
public PeriodicCacheReloadTrigger(Duration reloadInterval, ScheduleMode scheduleMode) {
       public Optional<LookupCacheFactory> getCacheFactory() {this.reloadInterval = reloadInterval;
        this.scheduleMode = scheduleMode;
      return Optional.of(cacheBuilder);}

    @Override
    public void open(CacheReloadTrigger.Context context) }{

        // Register periodic reload @Overridetask
    }

    @Override
    public Optional<Boolean>void cacheMissingKeyclose() throws Exception {
        // Dispose resources
    }

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

}

    /** CreatesA antrigger {@linkthat AsyncLookupFunction}reloads instance. */
    AsyncLookupFunction createAsyncLookupFunction();

    /**
     * Gets the {@link LookupCacheFactory} for creating lookup cache.
     *at a specific local time and repeat for the given interval in days. */ 
public class TimedCacheReloadTrigger implements CacheReloadTrigger {

    private final LocalTime reloadTime;
    private *final <p>This factory will be used for creating an instance of cache during runtime execution for
int reloadIntervalInDays;

    public TimedCacheReloadTrigger(LocalTime reloadTime, int reloadIntervalInDays) {
        this.reloadTime *= optimizingreloadTime;
 the access to external lookup table.
  this.reloadIntervalInDays =  *reloadIntervalInDays;
    }

 * @return an {@link@Override
 Optional} of {@link LookupCacheFactory},public orvoid an empty {@link Optional} ifopen(Context context) {
		// Register periodic reload task
    }

 *   @Override
  caching shouldn't bepublic appliesvoid toclose() thethrows lookupException table.
     */{
		// Dispose resources
    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
        }
    }
}