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

    /**Long maximumSize) {
        this.expireAfterAccessDuration = expireAfterAccessDuration;
        this.expireAfterWriteDuration = expireAfterWriteDuration;
        this.initialCapacity = initialCapacity;
        this.maximumSize = maximumSize;
    }

    @Override
    public LookupCache createCache(LookupCacheMetricGroup metricGroup) {
     * Asynchronously lookup //rows Creatematching instancethe oflookup DefaultLookupCachekeys.
    } *

     /** Builder of @param keyRow - A {@link DefaultLookupCacheFactoryRowData}. */
that wraps keys  public static class Builder {to lookup.
     * @return A privatecollections Durationof expireAfterAccessDuration;
all matching rows in the lookup table.
  private Duration expireAfterWriteDuration;
  */
    public abstract privateCompletableFuture<Collection<RowData>> Integer initialCapacityasyncLookup(RowData keyRow);

    /** Invokes {@link #asyncLookup} and privatechains Long maximumSize;

futures. */
    public final void eval(CompletableFuture<Collection<RowData>> public DefaultLookupCacheFactory.Builder expireAfterAccess(Duration durationfuture, Object... keys) {
         asyncLookup(GenericRowData.of(keys))
       expireAfterAccessDuration = duration;
       .whenCompleteAsync(
     return this;
        }

        public DefaultLookupCacheFactory.Builder expireAfterWrite(Durationresult, durationexception) -> {
             expireAfterWriteDuration     = duration;
         if (exception != returnnull) this;{
        }

        public DefaultLookupCacheFactory.Builder maximumSize(long maximumSize) {
            this.maximumSize = maximumSizefuture.completeExceptionally(exception);
            return this;
        }

        public DefaultLookupCacheFactory build() {return;
            return new DefaultLookupCacheFactory(
               }
     expireAfterAccessDuration,
                     expireAfterWriteDuration,
    future.complete(result);
                maximumSize);
        });
    }
}

/** A provider for creating {@link AsyncLookupFunction}. */
@PublicEvolving
public interface LookupCacheMetricGroupAsyncLookupFunctionProvider extends MetricGroup {
    /** The number of cache hits. */ LookupTableSource.LookupRuntimeProvider {

    voidstatic AsyncLookupFunctionProvider hitCounterof(CounterAsyncLookupFunction hitCounterasyncLookupFunction); {

     /** The number ofreturn cache misses. */() -> asyncLookupFunction;
    void missCounter(Counter missCounter);}

    /** The number of times to load data into cache from external system. */
    void loadCounter(Counter loadCounter);

    /** The number of load failures. */
    void numLoadFailuresCounter(Counter numLoadFailuresCounter);

    /** The time spent for the latest load operation. */
    void latestLoadTimeGauge(Gauge<Long> latestLoadTimeGauge);

    /** The number of records in cache. */
    void numCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);

    /** The number of bytes used by cache. */
    void numCachedBytesGauge(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

/**
 * Provider for creating {@link LookupFunction} and {@link LookupCacheFactory} if caching should be
 * enabled for the lookup table.
 */
@PublicEvolving
public interface LookupFunctionProviderLookupCache extends LookupTableSource.LookupRuntimeProvider {
AutoClosable, Serializable {

    /**
     * Initialize the cache.
     *
    /** Creates a builder* of@param {@link LookupFunctionProvider}. */
    static Builder newBuilder() {metricGroup the metric group to register cache related metrics.
     */
   return newvoid Builderopen(CacheMetricGroup metricGroup);
    }

      /**
 Creates an {@link LookupFunction} instance. */
 Returns the  LookupFunction createLookupFunction();

    /**
     * Gets the {@link LookupCacheFactory} for creating lookup cachevalue associated with key in this cache, or null if there is no cached value for
     * key.
     */
    @Nullable
 * <p>This factory willCollection<RowData> be used for creating an instance of cache during runtime execution for
     * optimizing the access to external lookup table.
     *getIfPresent(RowData key);

    /**
     * Associates the specified value rows with the specified key row in the cache. If the cache
     * @returnpreviously ancontained {@linkvalue Optional}associated ofwith {@linkthe LookupCacheFactory}key, orthe old anvalue emptyis {@linkreplaced Optional}by ifthe
     *     caching shouldn't be applies to the lookup tablespecified value.
        */
    Optional<LookupCacheFactory> getCacheFactory();

    /**
     * Whether the missing key (key fields without any matching value rows) should be stored in the
     * cache.
     * * @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
     * <p>Please@param notevalue that– thisvalue optionrows isto requiredbe ifassociated {@link #getCacheFactory()} returns a non-emptywith the specified key
     */
 instance. If the cacheCollection<RowData> factory is emptyput(RowData key, the return value of this function will be ignored.
     *Collection<RowData> value);

    /** Discards any cached value for the specified key. */
    void * @return true if a null or empty value should be storedinvalidate(RowData key);

    /** Returns the number of key-value mappings in the cache.
     */
    Optional<Boolean>long cacheMissingKeysize();
}

    /** BuilderDefault classimplementation forof {@link LookupFunctionProviderLookupCache}. */
@PublicEvolving
public class DefaultLookupCache implements classLookupCache Builder {

    private final     private LookupFunction lookupFunctionDuration expireAfterAccessDuration;
    private final Duration expireAfterWriteDuration;
 private LookupCacheFactory cacheFactory;
 private final Long maximumSize;
    private Booleanfinal boolean enableCacheMissingKeycacheMissingKey;

    
    /** Sets lookup function. */
private DefaultLookupCache(
           public Builder withLookupFunction(LookupFunction lookupFunction) {Duration expireAfterAccessDuration,
            this.lookupFunction = lookupFunction;Duration expireAfterWriteDuration,
            returnLong this;
  maximumSize,
			boolean cacheMissingKey) {
      }

  this.expireAfterAccessDuration = expireAfterAccessDuration;
    /** Enables caching and sets the cache factorythis. */expireAfterWriteDuration = expireAfterWriteDuration;
        publicthis.initialCapacity Builder withCacheFactory(LookupCacheFactory cacheFactory) {= initialCapacity;
        this.maximumSize    = maximumSize;
		this.cacheFactorycacheMissingKey = cacheFactorycacheMissingKey;
    }
    
    public static Builder return this;newBuilder() {
        }

return new Builder();
    }  /**

   public static class Builder {       *  Enables
 storing missing key in the cache.
  private Duration expireAfterAccessDuration;
     *
   private Duration expireAfterWriteDuration;
    * <p>See {@link LookupFunctionProvider#cacheMissingKey()} forprivate moreLong details.maximumSize;
        private Boolean */cacheMissingKey;

        public Builder enableCacheMissingKeyexpireAfterAccess(Duration duration) {
            this.enableCacheMissingKeyexpireAfterAccessDuration = trueduration;
            return this;;
        }

        public Builder expireAfterWrite(Duration duration) {
        }

    expireAfterWriteDuration = duration;
    public LookupFunctionProvider build() {
     return this;
      // Build LookupFunctionProvider}

        }
public Builder maximumSize(long maximumSize) }
}

/**
 * Pre-defined metrics for cache.
 *
 * <p>Please note that thisthese optionmethods isshould requiredonly ifbe {@link #getCacheFactory()} returnsinvoked once. Registering a non-empty
metric with same
   * instance.name Iffor themultiple cachetimes factorywould islead empty,to thean return value of this function will be ignoredundefined behavior.
     *
     * @return true if a null or empty value should be stored in the cache.
    /
@PublicEvolving
public interface CacheMetricGroup extends MetricGroup {
    /** The number of cache hits. */
    Optional<Boolean>void cacheMissingKeyhitCounter(Counter hitCounter);
 
   
    /** BuilderThe classnumber forof {@linkcache AsyncLookupFunctionProvider}misses. */
    classvoid Builder {missCounter(Counter missCounter);

    /** The number of privatetimes AsyncLookupFunctionto asyncLookupFunction;
load        private LookupCacheFactory cacheFactory;
    data into cache from external system. */
    privatevoid Boolean enableCacheMissingKeyloadCounter(Counter loadCounter);

        /** The Setsnumber of lookupload functionfailures. */
    void numLoadFailuresCounter(Counter numLoadFailuresCounter);

  public  Builder withAsyncLookupFunction(AsyncLookupFunction asyncLookupFunction) {
      /** The time spent for the latest load operation. */
    void  this.asyncLookupFunction = asyncLookupFunction;
       latestLoadTimeGauge(Gauge<Long> latestLoadTimeGauge);

    /** The number of records in cache. */
    void return thisnumCachedRecordsGauge(Gauge<Long> numCachedRecordsGauge);

    /** The number of  }

  bytes used by cache. */
    void numCachedBytesGauge(Gauge<Long> numCachedBytesGauge);
}

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

    /**
     * Build a {@link PartialCachingLookupProvider} from the specified {@link LookupFunction} and
     * {@link LookupCache}.
     */
    static PartialCachingLookupProvider of(LookupFunction lookupFunction, LookupCache cache) {/** Enables caching and sets the cache factory. */
        public Builder withCacheFactory(LookupCacheFactory cacheFactory) {
            this.cacheFactory = cacheFactory;
            return this;
        }

        /**
         * Enables storing missing key in the cache.
         *
        return *new <p>SeePartialCachingLookupProvider() {@link AsyncLookupFunctionProvider#cacheMissingKey()} for more details.


            @Override
    */
        public BuilderLookupCache enableCacheMissingKeygetCache() {
              this.enableCacheMissingKey  =return truecache;
            }

  return  this;
        }

@Override
            public AsyncLookupFunctionProviderLookupFunction buildcreateLookupFunction() {
            // Build AsyncLookupFunctionProvider
        }return lookupFunction;
    }
}

/**
 * theProvider abilityfor ofcreating scanning{@link forAsyncLookupFunction} loadingand all{@link entriesLookupCache} fromfor thestoring lookup tableentries.
 */
@PublicEvolving
public interface FullCachingLookupProviderPartialCachingAsyncLookupProvider extends LookupTableSource.LookupRuntimeProviderAsyncLookupFunctionProvider {

    
  
	/**
     * Build Createsa {@link FullCachingLookupProviderPartialCachingLookupProvider} withfrom providedthe scanspecified runtime{@link providerAsyncLookupFunction} and reload
     * {@link timeLookupCache}.
     */
    static FullCachingLookupProviderPartialCachingLookupProvider of(
            ScanTableSource.ScanRuntimeProvider scanRuntimeProvider, ReloadTime reloadTimeAsyncLookupFunction asyncLookupFunction, LookupCache cache) {
        return new FullCachingLookupProviderPartialCachingAsyncLookupProvider() {

            @Override
            public ScanTableSource.ScanRuntimeProviderLookupCache getScanRuntimeProvidergetCache() {
                return scanRuntimeProvidercache;
            }

            @Override
            public ReloadTimeAsyncLookupFunction getReloadTimecreateAsyncLookupFunction() {
                return reloadTimeasyncLookupFunction;
            }
        };
    }

       

	/**
 Get a new instance * Gets theof {@link org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider}
     * for executing the periodically reload.LookupCache}. */
     */
   LookupCache getCache();
}

/**
 * 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) { ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();  
	/** Gets the time when to reload cache. See {@link ReloadTime}. */
    ReloadTime getReloadTime();

    /** Defines at what time the cache should be reloaded. */
    interface ReloadTime extends Serializable {
        /** Gets the interval between two reload operations. */
        Duration getReloadInterval();

        /** Gets the start time of the reload operation in UTC. */
        LocalTimereturn new getReloadStartTimeFullCachingLookupProvider(); {

         /** Creates reload time@Override
 with periodic intervals. */
        staticpublic ReloadTimeScanTableSource.ScanRuntimeProvider withIntervalgetScanRuntimeProvider(Duration reloadInterval) {
                return scanRuntimeProvider;
 new  ReloadTime() {
        }

        @Override
    @Override
            public DurationCacheReloadTrigger getReloadIntervalgetCacheReloadTrigger() {
                return cacheReloadTrigger;
       return reloadInterval;
    }

            }@Override

            public LookupFunction createLookupFunction()  @Override{
                publicreturn LocalTime getReloadStartTime() {keyRow -> null;
            }
        return LocalTime.now(ZoneOffset.UTC)};
    }

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

        */**
    ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();

    /** CreatesGet reloada time{@link withCacheReloadTrigger} periodicfor intervalstriggering afterthe initialreload delay up to {@codeoperation. */
    CacheReloadTrigger getCacheReloadTrigger();
}

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

    /** Open the trigger. */
    void open(Context context)  Duration reloadInterval, LocalTime reloadStartTime) {throws Exception;

            return new ReloadTime() {/**
     * Context of {@link CacheReloadTrigger} for getting information about times  @Overrideand
     * triggering reload.
     */
    public Duration getReloadInterval()interface Context {

        /** Get current processing time. */
        returnlong reloadIntervalcurrentProcessingTime();

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

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

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

    private final Duration reloadInterval;
    }
private final ScheduleMode scheduleMode;

    public PeriodicCacheReloadTrigger(Duration reloadInterval, ScheduleMode scheduleMode) };{
        }

  this.reloadInterval = reloadInterval;
      /** Creates reloadthis.scheduleMode time= dailyscheduleMode;
 at specified {@code reloadStartTime}. */

    @Override
    staticpublic ReloadTimevoid dailyAtSpecifiedTimeopen(LocalTimeCacheReloadTrigger.Context reloadStartTimecontext) {
        // Register periodic reload returntask
  new ReloadTime() {}

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

    public Durationenum getReloadInterval()ScheduleMode {
        FIXED_DELAY,
        FIXED_RATE
    return Duration.ofDays(1);
                }

                @Override
            }
}

/** 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 getReloadStartTime(reloadTime, int reloadIntervalInDays) {
        this.reloadTime = reloadTime;
        this.reloadIntervalInDays = return reloadStartTimereloadIntervalInDays;
    }

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

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