public void init(FunctionContext context) {
    LOG.info("lookup function row data predicate: {}.", predicate);
    MetricGroup metricGroup = context.getMetricGroup().addGroup(LookupMetrics.GROUP_NAME_LOOKUP);
    if (mixedTable == null) {
      mixedTable = loadMixedTable(loader).asKeyedTable();
    }
    mixedTable.refresh();

    lookupLoadingTimeMs = new AtomicLong();
    metricGroup.gauge(LookupMetrics.LOADING_TIME_MS, () -> lookupLoadingTimeMs.get());

    LOG.info("projected schema {}.\n table schema {}.", projectSchema, mixedTable.schema());
    kvTable =
        kvTableFactory.create(
            new RowDataStateFactory(generateRocksDBPath(context, mixedTable.name()), metricGroup),
            mixedTable.asKeyedTable().primaryKeySpec().fieldNames(),
            joinKeys,
            projectSchema,
            config,
            predicate);
    kvTable.open();

    this.incrementalLoader =
        new MixedIncrementalLoader<>(
            new MergeOnReadIncrementalPlanner(loader), flinkMORDataReader, readerFunction, filters);
  }
Initialize the mixed-format table, kvTable and incrementalLoader.

@param context
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/BasicLookupFunction.java