[native] Expose index join perf related session properties

Expose following Velox index join performance optimizations related query configs in Prestissimo as session properties:
- native_index_lookup_join_max_prefetch_batches: 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.
- native_index_lookup_join_split_output: If this is true, then the index join operator might split output for each input
batch based on the output batch size control. Otherwise, it tries to produce a single output for each input batch.
- native_unnest_split_output: If this is true, then the unnest operator might split output for each input batch based on the output batch size control. Otherwise, it produces a single output for each input batch.

Author: zacw7

presto-docs/src/main/sphinx/presto_cpp/properties-session.rst

@@ -487,3 +487,32 @@ This is used in global arbitration victim selection.
 
 Maximum number of splits to listen to by the SplitListener per table scan node per
 native worker.
+
+``native_index_lookup_join_max_prefetch_batches``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``integer``
+* **Default value:** ``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.
+
+``native_index_lookup_join_split_output``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``true``
+
+If this is true, then the index join operator might split output for each input
+batch based on the output batch size control. Otherwise, it tries to produce a
+single output for each input batch.
+
+``native_unnest_split_output``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``true``
+
+If this is true, then the unnest operator might split output for each input
+batch based on the output batch size control. Otherwise, it produces a single
+output for each input batch.

presto-native-execution/presto_cpp/main/SessionProperties.cpp

@@ -509,6 +509,35 @@ SessionProperties::SessionProperties() {
       0,
       QueryConfig::kMaxNumSplitsListenedTo,
       std::to_string(c.maxNumSplitsListenedTo()));
+
+  addSessionProperty(
+      kIndexLookupJoinMaxPrefetchBatches,
+      "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.",
+      INTEGER(),
+      false,
+      QueryConfig::kIndexLookupJoinMaxPrefetchBatches,
+      std::to_string(c.indexLookupJoinMaxPrefetchBatches()));
+
+  addSessionProperty(
+      kIndexLookupJoinSplitOutput,
+      "If this is true, then the index join operator might split output for"
+      "each input batch based on the output batch size control. Otherwise, it tries to"
+      "produce a single output for each input batch.",
+      BOOLEAN(),
+      false,
+      QueryConfig::kIndexLookupJoinSplitOutput,
+      std::to_string(c.indexLookupJoinSplitOutput()));
+
+  addSessionProperty(
+      kUnnestSplitOutput,
+      "In streaming aggregation, wait until we have enough number of output"
+      "rows to produce a batch of size specified by this. If set to 0, then"
+      "Operator::outputBatchRows will be used as the min output batch rows.",
+      BOOLEAN(),
+      false,
+      QueryConfig::kUnnestSplitOutput,
+      std::to_string(c.unnestSplitOutput()));
 }
 
 const std::unordered_map<std::string, std::shared_ptr<SessionProperty>>&

presto-native-execution/presto_cpp/main/SessionProperties.h

@@ -231,8 +231,7 @@ class SessionProperties {
   static constexpr const char* kQueryTraceDir = "native_query_trace_dir";
 
   /// The plan node id whose input data will be traced.
-  static constexpr const char* kQueryTraceNodeId =
-      "native_query_trace_node_id";
+  static constexpr const char* kQueryTraceNodeId = "native_query_trace_node_id";
 
   /// The max trace bytes limit. Tracing is disabled if zero.
   static constexpr const char* kQueryTraceMaxBytes =
@@ -322,10 +321,27 @@ class SessionProperties {
   static constexpr const char* kMaxNumSplitsListenedTo =
       "native_max_num_splits_listened_to";
 
+  /// 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 =
+      "native_index_lookup_join_max_prefetch_batches";
+
+  /// If this is true, then the index join operator might split output for each
+  /// input batch based on the output batch size control. Otherwise, it tries to
+  /// produce a single output for each input batch.
+  static constexpr const char* kIndexLookupJoinSplitOutput =
+      "native_index_lookup_join_split_output";
+
+  /// If this is true, then the unnest operator might split output for each
+  /// input batch based on the output batch size control. Otherwise, it produces
+  /// a single output for each input batch.
+  static constexpr const char* kUnnestSplitOutput =
+      "native_unnest_split_output";
+
   static SessionProperties* instance();
 
   SessionProperties();
-  
+
   /// Utility function to translate a config name in Presto to its equivalent in
   /// Velox. Returns 'name' as is if there is no mapping.
   const std::string toVeloxConfig(const std::string& name) const;

presto-native-execution/presto_cpp/main/tests/SessionPropertiesTest.cpp

@@ -117,7 +117,13 @@ TEST_F(SessionPropertiesTest, validateMapping) {
       {SessionProperties::kNativeQueryMemoryReclaimerPriority,
        core::QueryConfig::kQueryMemoryReclaimerPriority},
       {SessionProperties::kMaxNumSplitsListenedTo,
-       core::QueryConfig::kMaxNumSplitsListenedTo}};
+       core::QueryConfig::kMaxNumSplitsListenedTo},
+      {SessionProperties::kIndexLookupJoinMaxPrefetchBatches,
+       core::QueryConfig::kIndexLookupJoinMaxPrefetchBatches},
+      {SessionProperties::kIndexLookupJoinSplitOutput,
+       core::QueryConfig::kIndexLookupJoinSplitOutput},
+      {SessionProperties::kUnnestSplitOutput,
+       core::QueryConfig::kUnnestSplitOutput}};
 
   const auto& sessionProperties =
       SessionProperties::instance()->testingSessionProperties();