@Override
  public void initialize(Iterator<RowData> dataStream) throws IOException {
    while (dataStream.hasNext()) {
      RowData value = dataStream.next();
      if (filter(value)) {
        continue;
      }
      RowData uniqueKey = new KeyRowData(uniqueKeyIndexMapping, value);
      RowData joinKey = new KeyRowData(secondaryKeyIndexMapping, value);
      byte[] uniqueKeyBytes = recordState.serializeKey(uniqueKey);

      recordState.asyncWrite(value.getRowKind(), uniqueKeyBytes, value);
      setState.asyncWrite(joinKey, uniqueKeyBytes);
    }
    recordState.checkConcurrentFailed();
    setState.checkConcurrentFailed();
  }
Use secondary index to lookup. Working for the situation where the join keys don't match the
mixed-format table's primary keys.

<p>Example: <code>
SELECT * FROM t1 JOIN t2 for system_time as of t1.pt as dim ON t1.user_name = dim.user_name
</code>

<p>t2 as an mixed-format table with primary keys: user_name, city_name.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/SecondaryIndexTable.java