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

    /**
     * Synchronously lookup rows matching the lookup keys.
     *
     * @param keyRow - A {@link RowData} that wraps keys to lookup.
     * @return A collections of all matching rows in the lookup table.
     */
    public abstract Collection<RowData> lookup(RowData keyRow) throws IOException;

    /** Invoke {@link #lookup} and handle exceptions. */
    public final void eval(Object... keys) {
        try {
            lookup(GenericRowData.of(keys)).forEach(this::collect);
        } catch (IOException e) {
            throw new RuntimeException("Failed to lookup values with given key", e);
        }
    }
}

/**
 * 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 of all matching rows in the lookup table.
     */
    public abstract CompletableFuture<Collection<RowData>> asyncLookup(RowData keyRow);

    /** Invokes {@link #asyncLookup} and chains futures. */
    public final void eval(CompletableFuture<Collection<RowData>> future, Object... keys) {
         asyncLookup(GenericRowData.of(keys))

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

/**
 * 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 LookupCacheAsyncLookupFunctionProvider extends LookupTableSource.LookupRuntimeProvider {

    /**
static AsyncLookupFunctionProvider of(AsyncLookupFunction asyncLookupFunction)  *{
 Returns the value associated with key in thisreturn cache, or null if there is no cached value for() -> 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 {

    /**
     * Initialize the cache. * 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.* @param metricGroup the metric group to register cache related metrics.
     */
    void invalidateopen(RowDataCacheMetricGroup keymetricGroup);

    /**
     /** Returns the numbervalue associated of key-value mappings in the cache. with key in this cache, or null if there is no cached value for
     * key.
     */
    long size();
}

/** Default Durationimplementation expireAfterAccessDuration,
of {@link LookupCache}. */
@PublicEvolving
public class DefaultLookupCache implements LookupCache {
    private final Duration expireAfterWriteDuration,expireAfterAccessDuration;
    private final Duration expireAfterWriteDuration;
    private final IntegerLong initialCapacity,maximumSize;
    private final boolean cacheMissingKey;
    
  Long  maximumSize)private {DefaultLookupCache(
           // Validation
 Duration expireAfterAccessDuration,
         this.expireAfterAccessDuration = expireAfterAccessDuration;   Duration expireAfterWriteDuration,
        this.expireAfterWriteDuration = expireAfterWriteDuration;    Long maximumSize,
			boolean cacheMissingKey) {
        this.initialCapacityexpireAfterAccessDuration = initialCapacityexpireAfterAccessDuration;
        this.maximumSizeexpireAfterWriteDuration = maximumSizeexpireAfterWriteDuration;
    }

    @Override
this.initialCapacity    public LookupCache createCache(LookupCacheMetricGroup metricGroup) {= initialCapacity;
        // Create instance of DefaultLookupCachethis.maximumSize = maximumSize;
		this.cacheMissingKey = cacheMissingKey;
    }

    /** Builder of {@link DefaultLookupCacheFactory}. */
     public public static classBuilder BuildernewBuilder() {
        privatereturn Durationnew expireAfterAccessDurationBuilder();
    } 

   public static class Builder {         
        private Duration expireAfterWriteDurationexpireAfterAccessDuration;
        private IntegerDuration initialCapacityexpireAfterWriteDuration;
        private Long maximumSize;
        private Boolean cacheMissingKey;

        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 {

    /**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.Build Ifa the cache factory is empty, the return value of this function will be ignored{@link PartialCachingLookupProvider} from the specified {@link AsyncLookupFunction} and
     * {@link LookupCache}.
     */
     * @return true if a null or empty value should be stored in the cache.static PartialCachingLookupProvider of(AsyncLookupFunction asyncLookupFunction, LookupCache cache) {
        return new PartialCachingAsyncLookupProvider() {

     */
      Optional<Boolean> cacheMissingKey();
@Override
    /** Builder  class for {@link LookupFunctionProvider}. */
   public classLookupCache BuildergetCache() {

        private LookupFunction lookupFunction;
      return cache;
 private 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 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(
         *
   ScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
          * <p>See {@linkCacheReloadTrigger LookupFunctionProvider#cacheMissingKey(cacheReloadTrigger)} for{
 more details.
      return new FullCachingLookupProvider() */{
        public  Builder enableCacheMissingKey() {@Override
            public thisScanTableSource.enableCacheMissingKeyScanRuntimeProvider = true;
getScanRuntimeProvider() {
                return thisscanRuntimeProvider;
            }

        public  LookupFunctionProvider build() {@Override
            // Build LookupFunctionProviderpublic CacheReloadTrigger getCacheReloadTrigger() {
        }
     }
}

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

    /** Open @Override
the trigger. */
    void open(Context context) throws Exception;

  public Optional<Boolean> cacheMissingKey() { /**
     * Context of {@link CacheReloadTrigger} for getting information about   return Optional.empty();times and
     * triggering reload.
     }*/
    interface    };Context {

    }

    /**
 Get current processing time. */
 Creates {@link AsyncLookupFunctionProvider} with the given {@link AsyncLookupFunction} and
long currentProcessingTime();

     * enable caching with/** specifiedGet {@link LookupCacheFactory}.
    current watermark on the main stream. */
    static AsyncLookupFunctionProvider of(
  long currentWatermark();

        /** AsyncLookupFunctionTrigger asyncLookupFunction,
a reload operation on the full       LookupCacheFactory cacheBuilder,cache. */
            boolean cacheMissingKey) {CompletableFuture<Void> triggerReload();
    }
}

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

    private final Duration reloadInterval;    return new AsyncLookupFunctionProvider() {
            @Override
            public AsyncLookupFunction createAsyncLookupFunction() {
    private final ScheduleMode scheduleMode;

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

        this.scheduleMode =   @OverridescheduleMode;
    }

    @Override
    public Optional<LookupCacheFactory>void getCacheFactoryopen(CacheReloadTrigger.Context context) {
        // Register periodic reload task
    return Optional.of(cacheBuilder);}

    @Override
    public void close() throws Exception }{

        // Dispose resources
  @Override  }

    public enum ScheduleMode {
     public Optional<Boolean> cacheMissingKey() { FIXED_DELAY,
        FIXED_RATE
    }
}

/** A trigger that reloads at 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();
    
    /** 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
        }
    }
}