feat: Add native_pushdown_integer_upcasts_to_source session properties

This commit adds a new native session property native_pushdown_integer_
upcasts_to_scan, which when set to true, would rewrite the local plan
and push down integer upcasts to the source operaters.

Author: yingsu00

.gitmodules

@@ -1,3 +1,4 @@
 [submodule "presto-native-execution/velox"]
 	path = presto-native-execution/velox
-	url = https://github.com/facebookincubator/velox.git
+	url = https://github.com/yingsu00/velox.git
+	branch = pushdownCastToSourceLocal

presto-main-base/src/main/java/com/facebook/presto/sessionpropertyproviders/NativeWorkerSessionPropertyProvider.java

@@ -88,6 +88,7 @@ public class NativeWorkerSessionPropertyProvider
     public static final String NATIVE_INDEX_LOOKUP_JOIN_SPLIT_OUTPUT = "native_index_lookup_join_split_output";
     public static final String NATIVE_UNNEST_SPLIT_OUTPUT = "native_unnest_split_output";
     public static final String NATIVE_USE_VELOX_GEOSPATIAL_JOIN = "native_use_velox_geospatial_join";
+    public static final String NATIVE_PUSHDOWN_INTEGER_UPCASTS_TO_SOURCE = "native_pushdown_integer_upcasts_to_source";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -427,11 +428,19 @@ public NativeWorkerSessionPropertyProvider(FeaturesConfig featuresConfig)
                         true,
                         !nativeExecution),
                 booleanProperty(
+
                         NATIVE_USE_VELOX_GEOSPATIAL_JOIN,
                         "If this is true, then the protocol::SpatialJoinNode is converted to a " +
                                 "velox::core::SpatialJoinNode. Otherwise, it is converted to a " +
                                 "velox::core::NestedLoopJoinNode.",
                         true,
+                        !nativeExecution),
+
+                booleanProperty(
+                        NATIVE_PUSHDOWN_INTEGER_UPCASTS_TO_SOURCE,
+                        "Native Execution only. Pushdown integer type upcasts to scan if they are " +
+                                "immediately after scan",
+                        false,
                         !nativeExecution));
     }
 

presto-main-base/src/main/java/com/facebook/presto/sql/Optimizer.java

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.sql;
 
+import com.facebook.airlift.log.Logger;
 import com.facebook.presto.Session;
 import com.facebook.presto.SystemSessionProperties;
 import com.facebook.presto.cost.CachingCostProvider;
@@ -104,9 +105,13 @@ public Plan validateAndOptimizePlan(PlanNode root, PlanStage stage)
     {
         validateIntermediatePlanWithRuntimeStats(root);
 
+        Logger log = Logger.get(Optimizer.class);
+
         boolean enableVerboseRuntimeStats = SystemSessionProperties.isVerboseRuntimeStatsEnabled(session);
         if (stage.ordinal() >= OPTIMIZED.ordinal()) {
+            int i = 0;
             for (PlanOptimizer optimizer : planOptimizers) {
+                log.info("optimizer %s, name ", i++, optimizer.toString());
                 if (Thread.currentThread().isInterrupted()) {
                     throw new PrestoException(QUERY_PLANNING_TIMEOUT, String.format("The query optimizer exceeded the timeout of %s.", getQueryAnalyzerTimeout(session).toString()));
                 }

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

@@ -586,6 +586,14 @@ SessionProperties::SessionProperties() {
       false,
       std::nullopt,
       "true");
+
+  addSessionProperty(
+      kPushdownIntegerUpcastsToSource,
+      "Enable pushdown of integer upcasts to the source operators.",
+      BOOLEAN(),
+      false,
+      QueryConfig::kPushdownIntegerUpcastsToSource,
+      boolToString(c.pushdownIntegerUpcastsToSource()));
 }
 
 const std::string SessionProperties::toVeloxConfig(

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

@@ -378,6 +378,9 @@ class SessionProperties {
   static constexpr const char* kUseVeloxGeospatialJoin =
       "native_use_velox_geospatial_join";
 
+  static constexpr const char* kPushdownIntegerUpcastsToSource =
+      "native_pushdown_integer_upcasts_to_source";
+
   inline bool hasVeloxConfig(const std::string& key) {
     auto sessionProperty = sessionProperties_.find(key);
     if (sessionProperty == sessionProperties_.end()) {

presto-native-execution/presto_cpp/main/connectors/SystemConnector.h

@@ -151,7 +151,8 @@ class SystemConnector : public velox::connector::Connector {
       const velox::RowTypePtr& outputType,
       const velox::connector::ConnectorTableHandlePtr& tableHandle,
       const velox::connector::ColumnHandleMap& columnHandles,
-      velox::connector::ConnectorQueryCtx* connectorQueryCtx) override final {
+      velox::connector::ConnectorQueryCtx* connectorQueryCtx,
+      bool pushdownCasts = false) override final {
     VELOX_CHECK(taskManager_);
     return std::make_unique<SystemDataSource>(
         outputType,

presto-native-execution/presto_cpp/main/connectors/arrow_flight/ArrowFlightConnector.cpp

@@ -220,7 +220,8 @@ ArrowFlightConnector::createDataSource(
     const velox::RowTypePtr& outputType,
     const velox::connector::ConnectorTableHandlePtr& tableHandle,
     const velox::connector::ColumnHandleMap& columnHandles,
-    velox::connector::ConnectorQueryCtx* connectorQueryCtx) {
+    velox::connector::ConnectorQueryCtx* connectorQueryCtx,
+    bool pushdownCasts) {
   return std::make_unique<ArrowFlightDataSource>(
       outputType,
       columnHandles,

presto-native-execution/presto_cpp/main/connectors/arrow_flight/ArrowFlightConnector.h

@@ -139,7 +139,8 @@ class ArrowFlightConnector : public velox::connector::Connector {
       const velox::RowTypePtr& outputType,
       const velox::connector::ConnectorTableHandlePtr& tableHandle,
       const velox::connector::ColumnHandleMap& columnHandles,
-      velox::connector::ConnectorQueryCtx* connectorQueryCtx) override;
+      velox::connector::ConnectorQueryCtx* connectorQueryCtx,
+      bool pushdownCasts = false) override;
 
   std::unique_ptr<velox::connector::DataSink> createDataSink(
       velox::RowTypePtr inputType,

presto-native-execution/presto_cpp/main/operators/PartitionAndSerialize.cpp

@@ -155,7 +155,7 @@ class PartitionAndSerializeOperator : public Operator {
     serializeKeys(
         nextOutputRow_, endOutputRow, keyOutputBufferSize, *keyVector);
 
-    // Extract slice from output_ and construct the output vector.
+    // Extract slice from outputWithoutUpcasts_ and construct the output vector.
     std::vector<VectorPtr> childrenVectors;
     childrenVectors.push_back(
         output_->childAt(0)->slice(nextOutputRow_, batchSize));

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

@@ -66,6 +66,7 @@ TEST_F(QueryContextManagerTest, nativeSessionProperties) {
           {"native_expression_max_array_size_in_reduce", "99999"},
           {"native_expression_max_compiled_regexes", "54321"},
           {"request_data_sizes_max_wait_sec", "20"},
+          {"native_pushdown_integer_upcasts_to_source", "true"},
       }};
   protocol::TaskUpdateRequest updateRequest;
   updateRequest.session = session;
@@ -89,6 +90,7 @@ TEST_F(QueryContextManagerTest, nativeSessionProperties) {
   EXPECT_EQ(queryCtx->queryConfig().exprMaxArraySizeInReduce(), 99999);
   EXPECT_EQ(queryCtx->queryConfig().exprMaxCompiledRegexes(), 54321);
   EXPECT_EQ(queryCtx->queryConfig().requestDataSizesMaxWaitSec(), 20);
+  EXPECT_TRUE(queryCtx->queryConfig().pushdownIntegerUpcastsToSource());
 }
 
 TEST_F(QueryContextManagerTest, nativeConnectorSessionProperties) {