Introduce Arctic on Flink Lookup Join

[YesOrNo828]

Arctic on Flink Lookup Join

Introduction:

This proposal aims to outline the implementation of an Arctic on Flink Lookup Table Join that will enable fast and efficient retrieval of data from the Arctic Lakehouse Mixed format table. The current system uses Flink event time temporal join technology, leading to a slow and inefficient data initialization process. With the implementation of the Arctic on Flink Lookup Table Join, we will be able to improve the speed and efficiency of data retrieval, improving the system’s performance.

Background & Motivation:

The user encountered some problems while using the current lookup join scheme:

1. All primary keys must be used as join keys, but the actual business expects to match multiple data sets through partial primary keys or non-primary keys;
2. Based on event time temporal join, sorting of consumed data will be done after data loading is complete, and this time-consuming process is uncontrollable, which can easily cause checkpoint timeouts.
3. The user is not very familiar with the event time temporal join syntax, which is different from the lookup join syntax, causing confusion in the actual usage process.

There are several advantages to using Lookup Join Syntax:

1. Faster and more efficient data retrieval, the current Arctic MoR has been implemented, so it is possible to quickly read full data through engines such as Flink and Spark[1].
2. Increased flexibility in querying data through partial primary key or non-primary key comparisons.
3. Elimination of event time temporal join technology, improving data initialization processes, and reducing checkpoint timeouts.
4. Improved user experience through the use of familiar Lookup Join Syntax.
5. The Flink community's follow-up development of lookup join features could be utilized by Arctic on Flink lookup join, e.g. Shuffle hash lookup join, solve data skew issue, and load part of the cache.[2]

GOALS

1. Reduce the initialization time of the dimension table in Flink tasks and improve lookup join performance.
2. Support non-primary keys as join keys.

NON-GOALS

1. Support the left side of the lookup join to do a hash partitioner to raise the cache hint ratio
2. Each task could only load part of the cache instead of loading all cache.
3. The Flink Community has some relative proposals for these non-goals[2].

Proposal:

We propose the implementation of the Arctic on Flink Lookup Table as a replacement for the existing indexing method. The new system will be implemented using Apache Flink, which is a distributed computing framework that can process large volumes of data in real-time. The Arctic on Flink Lookup Table will store data in the form of key-value pairs through JVM or RocksDB.

SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
  JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
    ON o.customer_id = c.id;

Reduce Initialization Time

The proposed methods for reducing initialization time are as follows:

1. Using MOR to read the table's historical data, then seamlessly fetching the changestore data.
2. Support projection and filter pushdown.
3. Optimizing bulk load in the rocksDB: parallelizing writes, turning off compactions, and executing one big compaction in the end.[3]
4. The use of the Flink BinaryRowSerializer utility can enhance the serialization and deserialization process.

Project pushdown does not work for the lookup source[4]. Fixed in 1.16.0, 1.14.6, 1.15.3 [3].

Support the non-primary keys as join keys

We propose to support non-primary keys as join keys in the Arctic on Flink Lookup Table Join. This will increase the flexibility of querying data, allowing for a partial primary key or non-primary key comparisons. With this improvement, users will no longer be restricted to using only primary keys as join keys, expanding the use cases for the system.

KVTable

/**
 * The KVTable interface is used for lookup join in Arctic on Flink.
 * It includes methods for initializing and updating the lookup table,
 * as well as getting results by key and cleaning up the cache.
 */
public interface KVTable extends Serializable, Closeable {

  /**
   * Initialize the lookup table
   */
  void open();

  /**
   * Get the result by the key.
   * @throws IOException Serialize the rowData failed.
   */
  List<RowData> get(RowData key) throws IOException;

  /**
   * Upsert the {@link KVTable} by the Change table dataStream.
   * @throws IOException Serialize the rowData failed.
   */
  void upsert(Iterator<RowData> dataStream) throws IOException;

  /**
   * Initialize the {@link KVTable} by the MoR dataStream.
   * @param dataStream
   * @throws IOException
   */
  void initialize(Iterator<RowData> dataStream) throws IOException;

  /**
   * @return if initialization is completed.
   */
  boolean initialized();

  /**
	 * Waiting for the initialization completed, and enable auto compaction at the end of the initialization.
   */
  void waitInitializationCompleted();

  /**
   * Try to clean up the cache manually, due to the lookup_cache.ttl-after-write configuration.
	 * <p>lookup_cache.ttl-after-writ</p> Only works in SecondaryIndexTable.
   */
  default void cleanUp() {
  }

  void close();
}

UniqueIndexTable

public class UniqueIndexTable implements KVTable {
    private static final long serialVersionUID = 1L;
    protected final RocksDBRecordState recordState;
    protected int[] uniqueKeyIndexMapping;
    ......
}

SecondaryIndexTable

public class SecondaryIndexTable extends UniqueIndexTable {
    private static final long serialVersionUID = 1L;
    private final int[] secondaryKeyIndexMapping;
    private final RocksDBSetMemoryState setState;
    ......
}

RocksDBCacheState:

/**
 * This is an abstract state backed by RocksDB and Guava cache for storing and retrieving key-value pairs of byte arrays.
 *
 * @param <V> the type of the cache's values, which are not permitted to be null
 */
public abstract class RocksDBCacheState<V> {
    protected RocksDBBackend rocksDB;
    protected final String columnFamilyName;
    protected final ColumnFamilyHandle columnFamilyHandle;

    protected Cache<ByteArrayWrapper, V> guavaCache;

    private final AtomicBoolean initialized = new AtomicBoolean(false);
    protected Queue<LookupRecord> lookupRecordsQueue;

    private List<Future<?>> writeRocksDBThreadFutures;
    private final AtomicReference<Throwable> writingThreadException = new AtomicReference<>();

    public void open() {
        ......
    }

    /**
     * Waiting for the writing threads completed.
     */
    public void waitWriteRocksDBDone() {
        ......
    }

    public boolean initialized() {
        return initialized.get();
    }

    /**
     * Closes the RocksDB instance and cleans up the Guava cache.
     * <p>Additionally, it shuts down the writing service and clears the RocksDB record queue if they exist.
     */
    public void close() {
	......
    }

    public void initializationCompleted() {
        ......
        LOG.info("set db options[disable_auto_compactions={}]", false);
    }

    protected void checkConcurrentFailed() {
        if (writingThreadException.get() != null) {
            LOG.error("Check concurrent writing threads.", writingThreadException.get());
            throw new FlinkRuntimeException(writingThreadException.get());
        }
    }

    /**
     * This task is running during the initialization phase to write data{@link LookupRecord} to RocksDB.
     * <p>During the initialization phase, the Merge-on-Read approach is used to retrieve data,
     * which will only return INSERT data.
     * When there are multiple entries with the same primary key, only one entry will be returned.
     * <p>During the initialization phase, the incremental pull approach is also used to retrieve data that include
     * four {@link RowKind} rowKinds, -D, +I, -U, and +U.
     */
    class WriteRocksDBTask implements Runnable {

        private final String name;

        public WriteRocksDBTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            LOG.info("{} starting.", name);
            try {
                while (!initialized.get()) {
                    LookupRecord record = lookupRecordsQueue.poll();
                    ......
                }
            } catch (Throwable e) {
                LOG.error("writing failed:", e);
                writingThreadException.set(e);
            }
            LOG.info("{} stopping.", name);
        }
    }
}

RocksDBRecordState:

/**
 * A class used to store the state of a lookup record. For {@link UniqueIndexTable}.
 */
public class RocksDBRecordState extends RocksDBCacheState<byte[]> {

    /**
     * Retrieve the RowData from the guava cache first, if the value is null, fetch the value from the rocksDB.
     *
     * @param key try to find the record via this key.
     * @throws IOException if serialize the RowData variable <code>key</code> failed.
     */
    public Optional<RowData> get(RowData key) throws IOException {
        ......
    }

    /**
     * Putting the serialized RowData key and value into the rocksDB and cache.
     *
     * @throws IOException if serialize the RowData variable key and value failed.
     */
    public void put(RowData key, RowData value) throws IOException {
				......
    }

    /**
     * Delete the record in the rocksDB and cache if it exists.
     *
     * @throws IOException if serialize the RowData variable <code>key</code> failed.
     */
    public void delete(RowData key) throws IOException {
				......
    }
}

RocksDBSetMemoryState:

/**
 * A class that stores the secondary index in the cache. For {@link SecondaryIndexTable}.
 * <p>Support update the secondary index in the cache.
 */
public class RocksDBSetMemoryState extends RocksDBCacheState<Set<ByteArrayWrapper>> {
    protected BinaryRowDataSerializerWrapper joinKeySerializer;

    /**
     * Serialize the join key to bytes and put the join key bytes and unique key bytes in the cache.
     *
     * @param joinKey        the join key
     * @param uniqueKeyBytes the unique key bytes
     * @throws IOException if serializing the RowData variable failed.
     */
    public void put(RowData joinKey, byte[] uniqueKeyBytes) throws IOException {
        ......
    }

    /**
     * Delete the secondary index in the cache.
     * @param joinKey the join key
     * @param uniqueKeyBytes the unique key bytes
     * @throws IOException if serializing the RowData variable failed.
     */
    public void delete(RowData joinKey, byte[] uniqueKeyBytes) throws IOException {
        ......
    }

    /**
     * Retrieve the elements of the key.
     * <p>Fetch the Collection from the guava cache,
     * if not present, return an empty list.
     * if present, just return the result.
     *
     * @return is not null but may be empty.
     */
    public Collection<ByteArrayWrapper> get(RowData key) throws IOException {
        ......
    }

BinaryRowDataSerializerWrapper:

/**
 * This is a wrapper for {@link BinaryRowDataSerializer}. It is used to serialize and deserialize RowData.
 */
public class BinaryRowDataSerializerWrapper implements Serializable

MixedIncrementalLoader:

/**
 * This is a mixed-format table(mixed iceberg, mixed-hive) incremental loader.
 * This loader is used to load data by the merge-on-read approach first,
 * then by the incremental pull approach.
 * <p>Merge-on-read approach only contains INSERT rows.
 * <p>Incremental pull approach contains INSERT, DELETE, UPDATE_BEFORE, and UPDATE_AFTER rows.
 * <p>Support projection and filter push-down to speed up the loading process.
 */
public class MixedIncrementalLoader<T> implements AutoCloseable {
  

  public boolean hasNext() {
    ......
  }

  public CloseableIterator<T> next() {
    ......
  }

  @Override
  public void close() throws Exception {
    ......
  }
}

ArcticLookupFunction:

/**
 * This is a lookup function for Arctic table.
 */
public class ArcticLookupFunction extends TableFunction<RowData> {
    private static final long serialVersionUID = 1L;
    private ArcticTable arcticTable;
    private KVTable kvTable;
    private MixedIncrementalLoader<RowData> incrementalLoader;

    /**
     * Open the lookup function, e.g.: create {@link KVTable} kvTable, and load data.
     *
     * @throws IOException If serialize or deserialize failed
     */
    @Override
    public void open(FunctionContext context) throws IOException {
        ......
        checkAndLoad();
    }

    public void eval(Object... values) {
        try {
            checkAndLoad();
            RowData lookupKey = GenericRowData.of(values);
            List<RowData> results = kvTable.get(lookupKey);
            results.forEach(this::collect);
        } catch (Exception e) {
            throw new FlinkRuntimeException(e);
        }
    }

    /**
     * Check whether it is time to periodically load data to kvTable.
     * Support to use {@link Expression} filters to filter the data.
     *
     * @throws IOException If serialize or deserialize failed.
     */
    private synchronized void checkAndLoad() throws IOException {
        ......
    }

    @Override
    public void close() throws Exception {
        if (kvTable != null) {
            kvTable.close();
        }
    }

Test Plan

Each new feature would be covered by unit tests.

Reference:

[1]: Support MOR with Flink Engine in runtime batch mode

[2]: FLIP-204: Introduce Hash Lookup Join

[3]: optimizing-bulk-load-in-rocksdb

[4]: FLINK-29138

[YesOrNo828]

google docs link: https://docs.google.com/document/d/1euYt7rbNGjrr2JFw2vK_epuHIjZ9AJV99FXC75yXSfw/edit?usp=sharing