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

/** 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,
            Duration expireAfterWriteDuration,
            Long maximumSize,
			boolean cacheMissingKey) {
        this.expireAfterAccessDuration = expireAfterAccessDuration;
        this.expireAfterWriteDuration = expireAfterWriteDuration;
        this.initialCapacity = initialCapacity;
        this.maximumSize = maximumSize;
		this.cacheMissingKey = cacheMissingKey;
    }
    
    public static Builder newBuilder() {
        return new Builder();
    } 

   public static class Builder {         
        private Duration expireAfterAccessDuration;
        private Duration expireAfterWriteDuration;
        private Long maximumSize;
        private Boolean cacheMissingKey;

        public Builder expireAfterAccess(Duration duration) {
            expireAfterAccessDuration = duration;try {
            return thislookup(GenericRowData.of(keys)).forEach(this::collect);
        }

        public Builder expireAfterWrite(Duration durationcatch (IOException e) {
            expireAfterWriteDurationthrow = duration;
            return thisnew RuntimeException("Failed to lookup values with given key", e);
        }

    }
}

/**
 * A wrapper publicclass Builder maximumSize(long maximumSize) {
            this.maximumSize = maximumSize;
            return this;
        }

        public Builder cacheMissingKey(boolean cacheMissingKey) {
            this.cacheMissingKey = cacheMissingKey;
            return this;
        }          

        public DefaultLookupCache build() {
            return new DefaultLookupCache(
                    expireAfterAccessDuration,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))
                .whenCompleteAsync(
    expireAfterWriteDuration,
                    maximumSize,
					cacheMissingKey);
     (result, exception) -> {
   }
    }     
}

/**
 A provider   * Gets the instance offor creating {@link LookupCacheAsyncLookupFunction}. */
@PublicEvolving
public interface AsyncLookupFunctionProvider extends LookupTableSource.LookupRuntimeProvider *{

    static AsyncLookupFunctionProvider of(AsyncLookupFunction *asyncLookupFunction) <p>This{
 cache will be initialized by {@link LookupCache#open} duringreturn runtime() execution-> andasyncLookupFunction;
     * used for optimizing the access to external lookup table.
     *
     * @return an {@link Optional} of {@link LookupCache}, or an empty {@link Optional} if caching
     *     shouldn't be applies to the lookup table.
     */
    Optional<LookupCache> getCache();

    /** Builder class for {@link PartialCachingLookupProvider}. */
    class Builder {

        private LookupFunction lookupFunction;
        private LookupCache cache;
        private Boolean cacheMissingKey;

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

        /** Enables caching and sets the cache factory. */
        public Builder withCache(LookupCache cache, boolean cacheMissingKey) {
            this.cache = cache;
            this.cacheMissingKey = cacheMissingKey;
            return this;
        }

        public PartialCachingLookupProvider build() {
           ...
        }
    }
}

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

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

/**
 * Provider for creating {@link AsyncLookupFunction} and {@link LookupCache} if caching should be
 * enabled for the lookup table.
 */
@PublicEvolving
public interface AsyncPartialCachingLookupProvider extends LookupTableSource.LookupRuntimeProvider {

    /** Creates a builder of {@link AsyncPartialCachingLookupProvider}. */
    static AsyncPartialCachingLookupProvider.Builder newBuilder() {
        return new AsyncPartialCachingLookupProvider.Builder();
    }

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

    /**
 Discards any cached value *for Gets the {@link LookupCache} for creating lookup cache.
     *
 specified key. */
    void invalidate(RowData key);

    /** <p>ThisReturns factorythe willnumber beof usedkey-value formappings creating an instance ofin the cache during runtime execution for. */
    long size();
}

/** Default implementation  * optimizing the access to external lookup table.
     *
     * @return an {@link Optional} of {@link LookupCache}, or an empty {@link Optional} if caching
     *. */
@PublicEvolving
public class DefaultLookupCache implements LookupCache {
    private final Duration expireAfterAccessDuration;
    private shouldn'tfinal be applies to the lookup table.
  Duration expireAfterWriteDuration;
   */
 private final Long Optional<LookupCache> getCache()maximumSize;

    private /**final Builderboolean classcacheMissingKey;
 for {@link AsyncPartialCachingLookupProvider}. */
    class Builder {
private DefaultLookupCache(
        private AsyncLookupFunction asyncLookupFunction;
  Duration expireAfterAccessDuration,
     private LookupCache cache;
     Duration expireAfterWriteDuration,
  private Boolean cacheMissingKey;

        /** Sets lookup function. */Long maximumSize,
			boolean cacheMissingKey) {
        public AsyncPartialCachingLookupProvider.Builder withLookupFunction(this.expireAfterAccessDuration = expireAfterAccessDuration;
        this.expireAfterWriteDuration = expireAfterWriteDuration;
      AsyncLookupFunction asyncLookupFunction) {
   this.initialCapacity = initialCapacity;
        this.maximumSize = maximumSize;
		this.asyncLookupFunctioncacheMissingKey = asyncLookupFunctioncacheMissingKey;
    }
    
    public static Builder newBuilder() {
        return thisnew Builder();
    } 

   public static class }

Builder {         
  /** Enables caching and sets the cacheprivate factory. */Duration expireAfterAccessDuration;
        publicprivate AsyncPartialCachingLookupProvider.Builder withCache(Duration expireAfterWriteDuration;
        private Long maximumSize;
       LookupCache cache,private booleanBoolean cacheMissingKey) {;

        public Builder expireAfterAccess(Duration  this.cache = cache;duration) {
            this.cacheMissingKeyexpireAfterAccessDuration = cacheMissingKeyduration;
            return this;
        }

        public AsyncPartialCachingLookupProvider build(Builder expireAfterWrite(Duration duration) {
             ...expireAfterWriteDuration = duration;
        }
    }
}

/**
     * GetsPre-defined the {@link ScanTableSource.ScanRuntimeProvider} for executing the periodically reloadmetrics for cache.
     */
 * <p>Please note ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();

    /** Get the {@link FullCachingReloadTrigger} for triggering a full caching reload operation. */
    FullCachingReloadTrigger getReloadTrigger();
}

/**
 A* Provider triggerfor thatcreating reloads{@link allLookupFunction} entriesand periodically{@link withLookupCache} specifiedfor intervalstoring orlookup delayentries.
 */
@PublicEvolving
public classinterface PeriodicFullCachingReloadTriggerPartialCachingLookupProvider implementsextends FullCachingReloadTriggerLookupFunctionProvider { 

    private final DurationLookupFunction reloadIntervallookupFunction;
    private final ScheduleMode scheduleMode;

    private ScheduledExecutorService scheduledExecutor LookupCache cache;

    public PeriodicFullCachingReloadTriggerPartialCachingLookupProvider(DurationLookupFunction reloadIntervallookupFunction, ScheduleModeLookupCache scheduleModecache) {
        this.reloadIntervallookupFunction = reloadIntervallookupFunction;
        this.scheduleModecache = scheduleModecache;
    }

    @Override
    public voidLookupFunction opencreateLookupFunction(FullCachingReloadTrigger.Context context) {
        return lookupFunction;
    }
}

/**
 * Provider for creating {@link AsyncLookupFunction} and {@link LookupCache} for scheduledExecutorstoring =lookup Executorsentries.newSingleThreadScheduledExecutor();
 */
@PublicEvolving
public interface AsyncPartialCachingLookupProvider extends AsyncLookupFunctionProvider { 
 switch (scheduleMode) {
 private final AsyncLookupFunction asyncLookupFunction;
    private final LookupCache cache;

 case FIXED_RATE:
  public AsyncPartialCachingLookupProvider(AsyncLookupFunction asyncLookupFunction, LookupCache cache) {
         scheduledExecutor.scheduleAtFixedRate(
  this.asyncLookupFunction = asyncLookupFunction;
        this.cache = cache;
       }

     context::triggerReload,@Override
    public AsyncLookupFunction createAsyncLookupFunction() {
        return asyncLookupFunction;
    }
}

/**
 * A {@link LookupFunctionProvider} 0,
that never lookup in external system on cache miss and provides a
 * cache for holding all entries in the external system. The  reloadInterval.toMillis(),
                        TimeUnit.MILLISECONDS);cache will be fully reloaded from the
 * external system and reload operations will be triggered by the {@link CacheReloadTrigger}.
 */
@PublicEvolving
public class FullCachingLookupProvider implements LookupFunctionProvider {

    private final ScanTableSource.ScanRuntimeProvider scanRuntimeProvider;
    private final CacheReloadTrigger   breakreloadTrigger;

    public FullCachingLookupProvider(
       case  FIXED_DELAY:
   ScanTableSource.ScanRuntimeProvider scanRuntimeProvider,
            scheduledExecutor.scheduleWithFixedDelay(
  CacheReloadTrigger reloadTrigger) {
        this.scanRuntimeProvider = scanRuntimeProvider;
        this.reloadTrigger = reloadTrigger;
  () -> {}

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

/** 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. */
 try {
                                context.triggerReload().get();
                            } catch (Exception e) {
                                throw new RuntimeException(
                   long currentProcessingTime();

        /** Get current watermark on the main stream. */
    "Uncaught exception during the reload", elong 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 = 0,reloadInterval;
        this.scheduleMode = scheduleMode;
    }

    @Override
    public void reloadInterval.toMillis(),open(CacheReloadTrigger.Context context) {
        // Register periodic reload task
    }

    @Override
    public void TimeUnit.MILLISECONDS);
  close() throws Exception {
        // Dispose resources
    break;}

    public enum ScheduleMode {
     default:
   FIXED_DELAY,
        FIXED_RATE
     throw new IllegalArgumentException(
                        String.format("Unrecognized schedule mode \"%s\"", scheduleMode))}
}

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

 }
    }

    @Overridepublic TimedCacheReloadTrigger(LocalTime reloadTime, int reloadIntervalInDays) {
    public  void close() throws Exception { this.reloadTime = reloadTime;
        if (scheduledExecutor != null) {
  this.reloadIntervalInDays = reloadIntervalInDays;
    }

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

    public@Override
 enum ScheduleMode {
 public void close() throws    FIXED_DELAY,
        FIXED_RATEException {
		// Dispose resources
    }
}