/**
 * A semi-persistent 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 interface are expected to be thread-safe, and can be safely accessed
 * by multiple concurrent threads.
 *
 * @param <K> key
 * @param <V> value
 */
public interface Cache<K, V> extends AutoCloseable {

    /**
     * Initialize the cache.
     *
     * @param cacheMetricGroup - metric group for reporting metrics of the cache
     */
    void open(CacheMetricGroup cacheMetricGroup);

    /**
     * Returns the value associated with key in this cache, or null if there is no cached value for
     * key.
     */
    V getIfPresent(K key);

    /**
     * Returns the value associated with the key in the cache, or obtain the value by loader if the
     * key doesn't exist.
     *
     * @param key key - the key whose associated value is to be returned
     * @param loader - the loader for loading associated values from external system
     * @return The value associated with the key
     * @throws IOException if any exception was thrown while loading the value
     */
    V get(K key, Callable<? extends V> loader) throws IOException;

    /**
     * Associates the specified value with the specified key in the cache. If the cache previously
     * contained a value associated with the key, the old value is replaced by the specified value.
     *
     * @return the previous value associated with key, or null if there was no mapping for key.
     * @param key - key with which the specified value is to be associated
     * @param value – value to be associated with the specified key
     */
    V put(K key, V value);

    /**
     * Copies all the mappings from the specified map to the cache. The effect of this call is
     * equivalent to that of calling {@code put(k, v)} on this map once for each mapping from key
     * {@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 putAll(Map<? extends K, ? extends V> m);

    /** Delete all entries in the cache. */
    void clean();

    /** Returns the number of key-value mappings in the cache. */
    long size();
}

/** Pre-defined metrics for {@code Cache}. */
public interface CacheMetricGroup {

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

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

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

    /** The number of load failures. */
    Counter getNumLoadFailureCounter();
 
    /** The time spent for the latest load operation. */
    void setLatestLoadTimeGauge(Gauge<Long> latestLoadTimeGauge); 

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

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

public class CacheConfig implements ReadableConfig, WritableConfig {

    private final Configuration configuration = new Configuration();

    public CacheConfig setCacheStrategy(CacheConfigOptions.CacheStrategy strategy) {
        configuration.set(CacheConfigOptions.CACHE_STRATEGY, strategy);
        return this;
    }

    public CacheConfig setTTL(Duration ttl) {
        configuration.set(CacheConfigOptions.CACHE_TTL, ttl);
        return this;
    }

    public CacheConfig setMaxEntries(long maxEntries) {
        configuration.set(CacheConfigOptions.CACHE_MAX_ENTRIES, maxEntries);
        return this;
    }

    public CacheConfig setAllowEmptyValues(boolean allowEmptyValues) {
        configuration.set(CacheConfigOptions.CACHE_ALLOW_EMPTY_VALUES, allowEmptyValues);
        return this;
    }

    @Override
    public <T> T get(ConfigOption<T> option) {
        return configuration.get(option);
    }

    @Override
    public <T> Optional<T> getOptional(ConfigOption<T> option) {
        return configuration.getOptional(option);
    }

    @Override
    public <T> WritableConfig set(ConfigOption<T> option, T value) {
        return configuration.set(option, value);
    }
}  

public class CacheConfigOptions {

    public static final ConfigOption<CacheStrategy> CACHE_STRATEGY =
            ConfigOptions.key("lookup.cache.strategy")
                    .enumType(CacheStrategy.class)
                    .noDefaultValue()
                    .withDescription(
                            "Strategy of the cache to load and evict entries, including ALL, LRU and NONE");

    public static final ConfigOption<Duration> CACHE_TTL =
            ConfigOptions.key("lookup.cache.ttl")
                    .durationType()
                    .noDefaultValue()
                    .withDescription("Time-to-live (TTL) of entries in cache");

    public static final ConfigOption<Long> CACHE_MAX_ENTRIES =
            ConfigOptions.key("lookup.cache.max-entries")
                    .longType()
                    .noDefaultValue()
                    .withDescription("The maximum number of entries that the cache could contain");

    public static final ConfigOption<Boolean> CACHE_ALLOW_EMPTY_VALUES =
            ConfigOptions.key("lookup.cache.allow-empty-values")
                    .booleanType()
                    .noDefaultValue()
                    .withDescription(
                            "Whether to allow empty values associated with a key in the cache");

    public enum CacheStrategy {
        ALL,
        LRU,
        NONE
    }
}

/**
 * Base class for lookup function that supports caching.
 */
@PublicEvolving
public abstract class CachingTableFunction extends TableFunction<RowData> {
    private final Cache<RowData, List<RowData>> cache;

    /**
     * Use default cache implementation with specified cache configuration.
     *
     * @param cacheConfig - configuration of the default cache.
     */
    public CachingTableFunction(CacheConfig cacheConfig) {
        cache = new DefaultCache<>(cacheConfig);
    }

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

    @Override
    public void open(FunctionContext context) throws Exception {
        cache.open(new InternalCacheMetricGroup(context.getMetricGroup()));
    }

	public void eval(Object... keys){
		// Check the cache first, or load via lookup() if the cache doesn't contains the key
    }
   
    /**
     * Lookup rows associated with the key row from external system.
     *
     * @param key the key row containing lookup fields
     * @throws IOException if any exception was thrown while looking up the value
     */
    public abstract List<RowData> lookup(RowData key) throws IOException; 
}

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

    public AllCachingTableFunction(Cache<RowData, List<RowData>> cache, Duration reloadInterval) {
        this.cache = cache;
        this.reloadInterval = reloadInterval;
    }

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

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

    /**
     * Reload all entries in the cache.
     *
     * <p>This function will be invoked periodically in a scheduled executor, which invalidates all
     * entries in the cache and reloads all entries from external system.
     *
     * @param cache - the cache in which all entries to be reloaded
     * @throws IOException - if any exception was thrown while reloading the value
     */
    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> {
    private final Cache<RowData, List<RowData>> cache;

    /**
     * Use default cache implementation with specified cache configuration.
     *
     * @param cacheConfig - configuration of the default cache.
     */
    public CachingTableFunction(CacheConfig cacheConfig) {
        cache = new DefaultCache<>(cacheConfig);
    }

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

    @Override
    public void open(FunctionContext context) throws Exception {
        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.
     *
     * @param key - the key row whose associated value rows is to be returned
     */
    public abstract CompletableFuture<List<RowData>> lookupAsync(RowData key);
}