/**
 * A wrapper class of {@link TableFunction} for synchronously lookup rows matching the lookup keys
 * from external system.
 *
 * <p>The output type of this table function is fixed as {@link RowData}.
 */
@PublicEvolving
public abstract class LookupFunction extends TableFunction<RowData> {

    

/**
 * A wrapper class of {@link AsyncTableFunction} for asynchronously lookup rows matching the lookup
 * keys from external system.
 *
 * <p>The output type of this table function is fixed as {@link RowData}.
 */
@PublicEvolving
public abstract class AsyncLookupFunction extends AsyncTableFunction<RowData> {

    /**
     * Asynchronously lookup rows matching the lookup keys.
     *
     * @param keyRow - A {@link RowData} that wraps keys to lookup.
     */
 @return A collections Vof put(K key, V value);

    /**all matching rows in the lookup table.
     */
 Copies all the mappingspublic fromabstract theCompletableFuture<Collection<RowData>> specified map to the cache. The effect of this call is
     * equivalent to that of calling {@code put(k, v)} on this map once for each mapping from key
asyncLookup(RowData keyRow);

    /** Invokes {@link #asyncLookup} and chains futures. */
    public final void eval(CompletableFuture<Collection<RowData>> future, Object... keys) {
         asyncLookup(GenericRowData.of(keys))
         * {@code k} to value {@code v} in the specified map. The behavior of this operation is
 .whenCompleteAsync(
              * undefined if the specified map is modified while the operation is in progress.(result, exception) -> {
     */
    void putAll(Map<? extends K, ? extends V> m);

          /** Delete all entriesif in(exception the cache. */!= null) {
    void clean();

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

/** TheA numberprovider offor recordscreating in{@link cacheLookupFunction}. */
@PublicEvolving
public interface LookupFunctionProvider extends void setNumCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);LookupTableSource.LookupRuntimeProvider {

    /**static TheLookupFunctionProvider number of(LookupFunction byteslookupFunction) used{
 by cache. */      return () -> lookupFunction;
    }

    voidLookupFunction setNumCachedBytesGaugecreateLookupFunction(Gauge<Long> numCachedBytesGauge);
}

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

    /**
     * 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 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
     */
    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(
            Duration expireAfterAccessDuration,

public class CacheConfigOptions {

    public static final ConfigOption<CacheStrategy> CACHE_STRATEGY =
            ConfigOptions.key("lookup.cache.strategy")
                    .enumType(CacheStrategy.class)
                    .noDefaultValue()
                    .withDescription(
                            "Strategy of the cache to load and evict entries, including ALL, LRU and NONE");

    public static final ConfigOption<Duration> CACHE_TTL =
            ConfigOptions.key("lookup.cache.ttl")
                    .durationType()
                    .noDefaultValue()
                    .withDescription("Time-to-live (TTL) of entries in cache");

    public static final ConfigOption<Long> CACHE_MAX_ENTRIES =
            ConfigOptions.key("lookup.cache.max-entries")
                    .longType()
                    .noDefaultValue()Duration expireAfterWriteDuration,
            Long maximumSize,
			boolean cacheMissingKey) {
      .withDescription("The maximum numberthis.expireAfterAccessDuration of entries that the cache could contain")= expireAfterAccessDuration;

    public static final ConfigOption<Boolean> CACHE_ALLOW_EMPTY_VALUES = this.expireAfterWriteDuration = expireAfterWriteDuration;
        this.initialCapacity    ConfigOptions.key("lookup.cache.allow-empty-values")= initialCapacity;
        this.maximumSize       = maximumSize;
		this.cacheMissingKey = cacheMissingKey;
     .booleanType()}
    
    public static Builder newBuilder() {
        return new .noDefaultValueBuilder();
    } 

   public static class Builder {         
    .withDescription(
    private Duration expireAfterAccessDuration;
        private Duration expireAfterWriteDuration;
        private Long maximumSize;
  "Whether to allow empty values associated withprivate a key in the cache");

Boolean cacheMissingKey;

        public enum CacheStrategyBuilder expireAfterAccess(Duration duration) {
        ALL,    expireAfterAccessDuration = duration;
        LRU,
    return this;
    NONE
    }
}

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

/**
 * Base class for lookup function that loads all entries from external system and schedules
 * reloading by the specified TTL.
 */
public abstract class AllCachingTableFunction extends TableFunction<RowData> {
 	private Cache<RowData, List<RowData>> cache;

    //** IntervalThe tonumber reloadof allcache entries in cachemisses. */
    privatevoid final Duration reloadIntervalmissCounter(Counter missCounter);

	public AllCachingTableFunction(CacheConfig cacheConfig) {
		cache = new DefaultCache(cacheConfig);
		reloadInterval = cacheConfig.get(TTL);
	}

    public AllCachingTableFunction(Cache<RowData, List<RowData>> cache, Duration reloadInterval) { /** The number of times to load data into cache from external system. */
    void    this.cache = cacheloadCounter(Counter loadCounter);

    /** The number of load this.reloadInterval = reloadInterval;
    }failures. */
    void numLoadFailuresCounter(Counter numLoadFailuresCounter);

    @Override
/** The time spent publicfor voidthe open(FunctionContext context) throws Exception {latest load operation. */
    void    super.open(contextlatestLoadTimeGauge(Gauge<Long> latestLoadTimeGauge);

    /** The number  scheduleCacheReload(this::reloadAllEntries, reloadInterval);
    }

of records in cache. */
    public void eval(Object... keys) {
		// Only check the cache here
	}

numCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);

    /**
 The number of bytes *used Reloadby all entries in the cache.
     *
     * <p>This function will be invoked periodically in a scheduled executor, which invalidates allcache. */
    void numCachedBytesGauge(Gauge<Long> numCachedBytesGauge);
}

/**
 * Provider for creating {@link LookupFunction} and {@link LookupCache} for storing lookup entries.
 */
@PublicEvolving
public interface PartialCachingLookupProvider extends LookupFunctionProvider {

    /**
     * entries inBuild a {@link PartialCachingLookupProvider} from the cache specified {@link LookupFunction} and
 reloads  all entries from* external{@link systemLookupCache}.
     */
    static *PartialCachingLookupProvider @param cache - the cache in which all entries to be reloaded
of(LookupFunction lookupFunction, LookupCache cache) {
        return new PartialCachingLookupProvider() {

        * @throws IOException - if any exception was thrown while reloading the value
    @Override
            public LookupCache getCache() {
      */
    public abstract void reloadAllEntries(Cache<RowData, List<RowData>> cache) throwsreturn IOException;
}

/**
 * Provider for creating {@link AsyncLookupFunction} and {@link LookupCache} for storing lookup entries.
 */
@PublicEvolving
public interface PartialCachingAsyncLookupProvider extends AsyncLookupFunctionProvider { * @param cache - custom cache implementation
     */
    public CachingTableFunction(Cache<RowData, List<RowData>> cache) {
        this.cache = cache;
    }

    @Override/**
    public void open(FunctionContext context) throws Exception {
   * Build a {@link PartialCachingLookupProvider} from the specified {@link AsyncLookupFunction} and
      cache.open(new InternalCacheMetricGroup(context.getMetricGroup()));* {@link LookupCache}.
    }

	public void eval(CompletableFuture<Collection<RowData>> result, Object... keys){
		// Check the cache first and complete the result directly if cache hit, 
		// or load asynchronously via lookupAsync() if the cache doesn't contains the key
    }
   
    /**
     * Lookup rows associated with the key row asynchronously from external system and return the
     * {@link CompletableFuture} of the lookup result.
     *
     * @param key - the key row whose associated value rows is to be returned
     */
    public abstract CompletableFuture<List<RowData>> lookupAsync(RowData key);
}

/**
 * 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(
            ScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
            CacheReloadTrigger cacheReloadTrigger) {
        return new FullCachingLookupProvider() {
            @Override
            public ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider() {
                return scanRuntimeProvider;
            }

            @Override
            public CacheReloadTrigger getCacheReloadTrigger() {
                return cacheReloadTrigger;
            }

            @Override
            public LookupFunction createLookupFunction() {
                return keyRow -> null;
            }
        };
    }

    /**
     * Get a {@link ScanTableSource.ScanRuntimeProvider} for scanning all entries from the external
     * lookup table and load into the cache.
     */
    ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();

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

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

    /**
     * Context of {@link CacheReloadTrigger} for getting information about times and
     * triggering reload.
     */
    interface Context {

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

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

        /** Trigger a reload operation on the full cache. */
        CompletableFuture<Void> triggerReload();
    }
}

/** 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 void open(CacheReloadTrigger.Context context) {
        // Register periodic reload task
    }

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

    public enum ScheduleMode {
        FIXED_DELAY,
        FIXED_RATE
    }
}

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

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

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