feat: Support prefetch in index lookup join (#12611)

Summary:

This PR adds prefetch for index join with query config. The index join operator can prefetch up to the
configured prefetch limit to enable: (1) parallel prefetches  at backend for parallel execution (1-1) in case
of multiple backend shards or (1-2) enable backend to batch multiple requests to improve throughput;
(2) pipeline the table scan and index lookup execution in the same driver pipeline. The table scan is sync
executed while index lookup is async. This achieve pipelining without relying exchange which might cause
non-deterministic execution and Meta internal use case needs deterministic execution for checkpointing.
With Meta internal testing, this can achieve 2x throughput improvement (measured in rows per second)
with 33% memory overhead with up to 4 batches prefetch.

The follow is to add memory based prefetch throttling to integrate with Meta internal
ML use case and memory pool wiring to ease performance (memory overhead) analysis

Reviewed By: wenqiwooo

Differential Revision: D70909786

Author: xiaoxmeng

velox/core/QueryConfig.h

@@ -511,6 +511,11 @@ class QueryConfig {
   static constexpr const char* kThrowExceptionOnDuplicateMapKeys =
       "throw_exception_on_duplicate_map_keys";
 
+  /// Specifies the max number of input batches to prefetch to do index lookup
+  /// ahead. If it is zero, then process one input batch at a time.
+  static constexpr const char* kIndexLookupJoinMaxPrefetchBatches =
+      "index_lookup_join_max_prefetch_batches";
+
   bool selectiveNimbleReaderEnabled() const {
     return get<bool>(kSelectiveNimbleReaderEnabled, false);
   }
@@ -932,6 +937,10 @@ class QueryConfig {
     return get<double>(kTableScanScaleUpMemoryUsageRatio, 0.7);
   }
 
+  uint32_t indexLookupJoinMaxPrefetchBatches() const {
+    return get<uint32_t>(kIndexLookupJoinMaxPrefetchBatches, 4);
+  }
+
   std::string shuffleCompressionKind() const {
     return get<std::string>(kShuffleCompressionKind, "none");
   }

velox/docs/configs.rst

@@ -166,6 +166,11 @@ Generic Configuration
      - false
      - By default, if a key is found in multiple given maps, that key's value in the resulting map comes from the last one of those maps.
        If true, throws exception when duplicate keys are found. This configuration is needed by Spark functions `CreateMap`, `MapFromArrays`, `MapFromEntries`, `StringToMap`, `MapConcat`, `TransformKeys`.
+   * - index_lookup_join_max_prefetch_batches
+     - integer
+     - 0
+     - Specifies the max number of input batches to prefetch to do index lookup ahead. If it is zero,
+       then process one input batch at a time.
 
 .. _expression-evaluation-conf: