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

    /**
 Invoke {@link #lookup} and handle exceptions. */
 Synchronously  lookup rowspublic matchingfinal the lookup keys.void eval(Object... keys) {
     *   try {
     *  @param keys - Keys to lookup(keys).forEach(this::collect);
     * @return A collections} ofcatch all(IOException matchinge) rows{
 in the lookup table.
     */
   throw public abstract Collection<RowData> lookup(Object... keys) throws IOException;

new RuntimeException("Failed to lookup values with given key", e);
     /** Invoke {@link #lookup}
 and handle exceptions. */
    public final void eval(Object... keys) {
        try {
            lookup( }
}

/**
 *@return A semi-persistentcollections mapping from keys to values. Cache entries are manually added using {@link
 * #put(Object, Object)}, and are stored in the cache until either evicted or manually invalidated.
 *
 * <p>Implementations of this interfaceof all matching rows in the lookup table.
     */
    public abstract CompletableFuture<Collection<RowData>> asyncLookup(Object... keys);

    /** Invokes {@link #asyncLookup} and chains futures. */
    public final void eval(CompletableFuture<Collection<RowData>> future, Object... keys) {
        asyncLookup(keys)
                .whenCompleteAsync(
                        (result, exception) -> {
                            if (exception != null) {
                                future.completeExceptionally(exception);
                                return;
                            }
                            future.complete(result);
                        });
    }
}

/** Pre-defined metrics for {@code Cache}. */
public interface CacheMetricGroup {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);

    /**
 The number of cache hits. */
 Copies all the Counter getHitCounter();

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

mappings from the specified map to the cache. The effect of this call is
     /** equivalent Theto numberthat of timescalling to load data into cache from external system. */
{@code put(k, v)} on this map once for each mapping from key
     Counter getLoadCounter();

    /** The number of load failures. */
    Counter getNumLoadFailureCounter();
 
    /** The time spent for the latest load operation.* {@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 setLatestLoadTimeGauge(Gauge<Long> latestLoadTimeGauge); putAll(Map<? extends RowData, ? extends Collection<RowData>> m);

      /** The number of records in cacheDiscards any cached value for the specified key. */
    void setNumCachedRecordsGaugeinvalidate(Gauge<Long>RowData numCachedRecordsGaugekey);

    /** TheDiscards numberall ofentries bytesin usedthe by cache. */
    void setNumCachedBytesGauge(Gauge<Long> numCachedBytesGauge);
}

public class CacheConfigOptions// {Validation

    public static final ConfigOption<CacheStrategy> CACHE_STRATEGY this.expireAfterAccessDuration = expireAfterAccessDuration;
        this.expireAfterWriteDuration    ConfigOptions.key("lookup.cache.strategy")= expireAfterWriteDuration;
        this.initialCapacity = initialCapacity;
        this.maximumSize  .enumType(CacheStrategy.class)= maximumSize;
    }

    @Override
    public LookupCache createCache(LookupCacheMetricGroup metricGroup) {
    .noDefaultValue()
    // Create instance of DefaultLookupCache
    }

    /** Builder of {@link .withDescription(DefaultLookupCacheFactory}. */
    public static class Builder {
        private Duration expireAfterAccessDuration;
        private Duration "StrategyexpireAfterWriteDuration;
 of the cache to load and evict entries,private including ALL, LRU and NONE")Integer initialCapacity;

     public static final ConfigOption<Duration>private CACHE_TTL =Long maximumSize;

        public DefaultLookupCacheFactory.Builder expireAfterAccess(Duration  ConfigOptions.key("lookup.cache.ttl")duration) {
            expireAfterAccessDuration = duration;
      .durationType()
      return this;
        }

     .noDefaultValue()
   public DefaultLookupCacheFactory.Builder expireAfterWrite(Duration duration) {
             .withDescription("Time-to-live (TTL) of entries in cache");

expireAfterWriteDuration = duration;
        public static final ConfigOption<Long> CACHE_MAX_ENTRIES =return this;
        }

    ConfigOptions.key("lookup.cache.max-entries")
    public DefaultLookupCacheFactory.Builder initialCapacity(int initialCapacity) {
            .longType()this.initialCapacity = initialCapacity;
            return this;
        .noDefaultValue()}

        public DefaultLookupCacheFactory.Builder maximumSize(long maximumSize) {
        .withDescription("The  maximum number ofthis.maximumSize entries= thatmaximumSize;
 the cache could contain");

    public static final ConfigOption<Boolean> CACHE_ALLOW_EMPTY_VALUES = return this;
        }

     ConfigOptions.key("lookup.cache.allow-empty-values")
   public DefaultLookupCacheFactory build() {
            return new .booleanTypeDefaultLookupCacheFactory()
                    .noDefaultValue()expireAfterAccessDuration,
                    .withDescription(expireAfterWriteDuration,
                    initialCapacity,
         "Whether to allow empty values associated with a key in the cache"maximumSize);

     public enum CacheStrategy {}
        ALL,
        LRU,
        NONE
    }
}

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

    // Interval to reload all entries in cache
    private final Duration reloadInterval;

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

            @Override
            public Optional<LookupCacheFactory> getCacheFactory() {
                return Optional.empty();
            }

    public AllCachingTableFunction(Cache<RowData, List<RowData>> cache, Duration reloadInterval) {
  @Override
      this.cache = cache;
    public Optional<Boolean> cacheMissingKey() {
 this.reloadInterval = reloadInterval;
    }

    @Override
    public voidreturn open(FunctionContext context) throws Exception {
Optional.empty();
            super.open(context);}
        scheduleCacheReload(this::reloadAllEntries, reloadInterval)};
    }

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

    /**/**
     * Creates {@link LookupFunctionProvider} with the given {@link LookupFunction} and enable
     * Reloadcaching allwith entriesspecified in the cache{@link LookupCacheFactory}.
     */
    static LookupFunctionProvider *of(
 <p>This function will be invoked periodically in a scheduled executor, which invalidatesLookupFunction alllookupFunction,
     * entries in the cache and reloads allLookupCacheFactory entriescacheFactory,
 from external system.
     *
    boolean *cacheMissingKey) @param{
 cache - the cache in which all entriesreturn tonew beLookupFunctionProvider() reloaded{
     * @throws IOException - if any exception was@Override
 thrown while reloading the value
     */
  public LookupFunction public abstract void reloadAllEntries(Cache<RowData, List<RowData>> cache) throws IOException;
}

/**
 * Base class for asynchronous lookup function that supports caching.
 */
@PublicEvolving
public abstract class CachingAsyncTableFunction extends AsyncTableFunction<RowData> {createLookupFunction() {
                return lookupFunction;
            }

    private final Cache<RowData, List<RowData>> cache;

    /**@Override
     *   Use default cache implementation withpublic specifiedOptional<LookupCacheFactory> cache configuration.getCacheFactory() {
     *
     * @param cacheConfig - configuration of the defaultreturn cacheOptional.of(cacheFactory);
     */
     public CachingTableFunction(CacheConfig cacheConfig) {
 }

         cache = new DefaultCache<>(cacheConfig);@Override
    }

    /**
    public *Optional<Boolean> Use custom cache implementation.
cacheMissingKey() {
       *
     * @param cache - custom cache implementationreturn Optional.of(cacheMissingKey);
     */
    public CachingTableFunction(Cache<RowData, List<RowData>> cache) {}
        this.cache = cache;
};
    }

    @Override
    public void open(FunctionContext context) throws Exception { /** Creates an {@link LookupFunction} instance. */
    LookupFunction createLookupFunction();

    /**
    cache.open(new InternalCacheMetricGroup(context.getMetricGroup()));
    }

	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. * 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 @parambe keyapplies -to the keylookup rowtable.
 whose associated value rows is to be returned  */
    Optional<LookupCacheFactory> getCacheFactory();

    /**
     */
  Whether  publicthe abstractmissing CompletableFuture<List<RowData>>key lookupAsync(RowData key);
}