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

    /**
      future.complete(result);
      * Asynchronously lookup rows matching the lookup keys.
     *
     * @param keyRow - A {@link   RowData});
 that   }
}

/** A provider for creating {@link LookupFunction}. */
@PublicEvolving
public interface LookupFunctionProvider extends LookupTableSource.LookupRuntimeProvider {

    static LookupFunctionProvider of(LookupFunction lookupFunction) {
        return () -> lookupFunction;
    }

    LookupFunction createLookupFunction();
}

/**
 @param* metricGroupA semi-persistent Themapping lookupfrom cachekeys metricto groupvalues in which the cache register predefined and
    for storing entries of lookup table.
 *
 * <p>The type of the customcaching metrics.
key is a {@link RowData} */
with lookup key fields LookupCache createCache(LookupCacheMetricGroup metricGroup);
}

/** Default implementation of {@link LookupCache}. */
@PublicEvolving
public class DefaultLookupCache implements LookupCache {
    private final Duration expireAfterAccessDuration;
    private final Duration expireAfterWriteDuration}

        public DefaultLookupCacheFactory.Builder maximumSize(long maximumSize) {
            this.maximumSize = maximumSize;
            return this;
    private final Long  }maximumSize;

    private final boolean cacheMissingKey;
 public DefaultLookupCacheFactory build() {
            return new DefaultLookupCacheFactory(
      private DefaultLookupCache(
            Duration  expireAfterAccessDuration,
            Duration expireAfterWriteDuration,
       expireAfterWriteDuration,
     Long maximumSize,
			boolean cacheMissingKey) {
        this.expireAfterAccessDuration =   maximumSize)expireAfterAccessDuration;
        }
this.expireAfterWriteDuration    }
}

/**
 * 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 returnundefined value of this function will be ignored.behavior.
 */
@PublicEvolving
public interface CacheMetricGroup extends MetricGroup {
     /**
 The number of cache hits. */
 @return true if avoid null or empty value should be stored in the cache.
     hitCounter(Counter hitCounter);

    /** The number of cache misses. */
    Optional<Boolean>void cacheMissingKeymissCounter(Counter missCounter);

      /** Builder class for {@link LookupFunctionProvider} The number of times to load data into cache from external system. */
    classvoid Builder {loadCounter(Counter loadCounter);

    /** The number of privateload LookupFunction lookupFunction;failures. */
    void numLoadFailuresCounter(Counter numLoadFailuresCounter);

   private LookupCacheFactory cacheFactory;
        private Boolean enableCacheMissingKey;

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

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

  public Builder withLookupFunction(LookupFunction lookupFunction) {
       /** 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}. this.lookupFunction = lookupFunction;
            return this;
        }

        /** Enables caching and sets the cache factory. */
        public Builder withCacheFactory(LookupCacheFactory cacheFactory) {
     */
    static PartialCachingLookupProvider of(LookupFunction lookupFunction, this.cacheFactory = cacheFactory;LookupCache cache) {
        return new PartialCachingLookupProvider() {

   return this;
        }
@Override
        /**
    public LookupCache getCache() {
  * Enables storing missing key in the cache.
        return *cache;
         * <p>See {@link LookupFunctionProvider#cacheMissingKey()}

 for more details.
         */@Override
            public BuilderLookupFunction enableCacheMissingKeycreateLookupFunction() {
            this.enableCacheMissingKey = true    return lookupFunction;
            return this;}
        };

    }

    public/** LookupFunctionProviderGet build() {
     a new instance of {@link LookupCache}. */
       // Build LookupFunctionProvider
        }
    }
}

/**
 * Provider for creating {@link AsyncLookupFunction} and {@link LookupCache} for storing lookup entries.
 */

@PublicEvolving
public interface AsyncLookupFunctionProviderPartialCachingAsyncLookupProvider extends LookupTableSource.LookupRuntimeProviderAsyncLookupFunctionProvider {

      /** Creates a builder of
     * Build a {@link PartialCachingLookupProvider} from the specified {@link AsyncLookupFunction} and
     * {@link LookupFunctionProviderLookupCache}.
     */
    static BuilderPartialCachingLookupProvider newBuilder(of(AsyncLookupFunction asyncLookupFunction, LookupCache cache) {
        return new BuilderPartialCachingAsyncLookupProvider(); {

    }

         @Override
   /** Creates an {@link AsyncLookupFunction} instance. */
   public AsyncLookupFunctionLookupCache createAsyncLookupFunctiongetCache(); {

    /**
     * Gets the {@link LookupCacheFactory} for creating lookupreturn cache.
;
            *}

     *   <p>This factory will be used@Override
 for creating an instance of cache during runtime execution for
  public AsyncLookupFunction createAsyncLookupFunction() *{
 optimizing the access to external lookup table.
     *
    return *asyncLookupFunction;
 @return an {@link Optional} of {@link LookupCacheFactory}, or an empty {@link Optional} if
     *   };
  caching shouldn't be applies to the lookup table.
      }

    /** Get a new instance of {@link LookupCache}. */
    Optional<LookupCacheFactory>LookupCache getCacheFactorygetCache();
}

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

{@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() {
        private AsyncLookupFunction asyncLookupFunction;
      return scanRuntimeProvider;
 private LookupCacheFactory cacheFactory;
        private Boolean enableCacheMissingKey;}

        /** Sets lookup function. */@Override
            public BuilderCacheReloadTrigger withAsyncLookupFunctiongetCacheReloadTrigger(AsyncLookupFunction asyncLookupFunction) {
            this.asyncLookupFunction = asyncLookupFunction;
         return cacheReloadTrigger;
   return this;
        }

        /** Enables caching and sets@Override
 the cache factory. */
        public BuilderLookupFunction withCacheFactorycreateLookupFunction(LookupCacheFactory cacheFactory) {
            this.cacheFactory = cacheFactory    return keyRow -> null;
            return this;}
        };

    }

    /**
     * Get a {@link ScanTableSource.ScanRuntimeProvider} for scanning all entries from the external
     * Enableslookup storingtable missingand keyload ininto the cache.
     */
    ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider();

    /*
* Get a {@link CacheReloadTrigger} for triggering the reload operation. */
  <p>See  {@linkCacheReloadTrigger AsyncLookupFunctionProvider#cacheMissingKey()}getCacheReloadTrigger();
}

/** Customized trigger for morereloading details.
all lookup table entries in full caching  mode. */
@PublicEvolving
public interface CachingReloadTrigger extends AutoCloseable,    public Builder enableCacheMissingKey() {Serializable {

    /** Open       this.enableCacheMissingKey = true;the trigger. */
    void open(Context context) throws Exception;

    return this;
/**
     * Context of {@link CacheReloadTrigger}

  for getting information about times and
 public AsyncLookupFunctionProvider build() {
 * triggering reload.
     */
    //interface BuildContext AsyncLookupFunctionProvider{

        }
/** Get current  }
}

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

/**
 * Runtime provider for fully loading and periodically reloading all entries of the lookup table and
 * storing the table locally for lookup.
 *
 * <p>Implementations should provide a {@link ScanTableSource.ScanRuntimeProvider} in order to reuse
 * the ability of scanning for loading all entries from the lookup table.
 */
@PublicEvolving
public interface FullCachingLookupProvider extends LookupTableSource.LookupRuntimeProvider {

    /**
     * Creates {@link FullCachingLookupProvider} with provided scan runtime provider and rescan
     * interval.
     */
    static FullCachingLookupProvider of(
            ScanTableSourcethis.ScanRuntimeProviderreloadInterval scanRuntimeProvider,= Duration reloadInterval) {;
        returnthis.scheduleMode new FullCachingLookupProvider() {
= scheduleMode;
    }

    @Override
    public @Override
   void open(CacheReloadTrigger.Context context) {
        // publicRegister ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider() {periodic reload task
    }

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

    public enum ScheduleMode {
     @Override
   FIXED_DELAY,
        FIXED_RATE
 public Duration getReloadInterval() {
                return reloadInterval;
            }}
}

/** 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) {
     *  Gets the {@link org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider}this.reloadTime = reloadTime;
     * for executing thethis.reloadIntervalInDays periodically= reload.
 reloadIntervalInDays;
    }

    */@Override
    public ScanTableSource.ScanRuntimeProvidervoid getScanRuntimeProvideropen();

 Context context) {
		// Register periodic reload task
    }

    @Override
   /** Getspublic thevoid intervalclose() betweenthrows two reload operations. */Exception {
		// Dispose resources
    Duration getReloadInterval();}
}