private synchronized void checkAndLoad() {
    if (nextLoadTime > System.currentTimeMillis()) {
      return;
    }
    nextLoadTime = System.currentTimeMillis() + 1000 * reloadIntervalSeconds;

    long batchStart = System.currentTimeMillis();
    while (incrementalLoader.hasNext()) {
      long start = System.currentTimeMillis();
      mixedTable
          .io()
          .doAs(
              () -> {
                try (CloseableIterator<T> iterator = incrementalLoader.next()) {
                  if (kvTable.initialized()) {
                    kvTable.upsert(iterator);
                  } else {
                    LOG.info(
                        "This table {} is still under initialization progress.", mixedTable.name());
                    kvTable.initialize(iterator);
                  }
                }
                return null;
              });
      LOG.info("Split task fetched, cost {}ms.", System.currentTimeMillis() - start);
    }
    if (!kvTable.initialized()) {
      kvTable.waitInitializationCompleted();
    }
    lookupLoadingTimeMs.set(System.currentTimeMillis() - batchStart);

    LOG.info(
        "{} table lookup loading, these batch tasks completed, cost {}ms.",
        mixedTable.name(),
        lookupLoadingTimeMs.get());
  }
Check whether it is time to periodically load data to kvTable. Support to use {@link
Expression} filters to filter the data.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/BasicLookupFunction.java