PrestoToVeloxQueryPlan

Author: xin-zhang2

presto-native-execution/presto_cpp/main/types/CMakeLists.txt

@@ -21,6 +21,7 @@ add_library(presto_types PrestoToVeloxQueryPlan.cpp VeloxPlanValidator.cpp
                          PrestoToVeloxSplit.cpp)
 target_link_libraries(
   presto_types
+  $<TARGET_OBJECTS:presto_tvf_core>
   presto_velox_expr_conversion
   presto_connectors
   presto_operators

presto-native-execution/presto_cpp/main/types/PrestoToVeloxQueryPlan.cpp

@@ -1769,6 +1769,49 @@ VeloxQueryPlanConverterBase::toVeloxQueryPlan(
       toVeloxQueryPlan(node->source, tableWriteInfo, taskId));
 }
 
+std::shared_ptr<const tvf::TableFunctionProcessorNode>
+VeloxQueryPlanConverterBase::toVeloxQueryPlan(
+    const std::shared_ptr<const protocol::TableFunctionProcessorNode>& node,
+    const std::shared_ptr<protocol::TableWriteInfo>& tableWriteInfo,
+    const protocol::TaskId& taskId) {
+  const auto outputType = toRowType(node->properOutputs, typeParser_);
+
+  std::vector<column_index_t> requiredColumns;
+  std::vector<core::PlanNodePtr> sources;
+  if (node->source) {
+    const auto sourceNode = toVeloxQueryPlan(*node->source, tableWriteInfo, taskId);
+    const auto inputType = sourceNode->outputType();
+    for (const auto& variables : node->requiredVariables) {
+      for (const auto& expr : toVeloxExprs(variables)) {
+        requiredColumns.push_back(exprToChannel(expr.get(), inputType));
+      }
+    }
+  }
+
+  // TODO: deserialize node->handle.functionHandle to construct the handle
+  const auto tableFunctionHandlePtr = nullptr;
+
+  auto partitionKeys = toVeloxExprs(node->specification->partitionBy);
+
+  std::vector<core::FieldAccessTypedExprPtr> sortingKeys;
+  std::vector<core::SortOrder> sortingOrders;
+  for (const auto& orderby : node->specification->orderingScheme->orderBy) {
+    sortingKeys.emplace_back(exprConverter_.toVeloxExpr(orderby.variable));
+    sortingOrders.push_back(toVeloxSortOrder(orderby.sortOrder));
+  }
+
+  return std::make_shared<tvf::TableFunctionProcessorNode>(
+    node->id,
+    node->name,
+    tableFunctionHandlePtr,
+    partitionKeys,
+    sortingKeys,
+    sortingOrders,
+    outputType,
+    requiredColumns,
+    sources);
+}
+
 core::PlanNodePtr VeloxQueryPlanConverterBase::toVeloxQueryPlan(
     const std::shared_ptr<const protocol::PlanNode>& node,
     const std::shared_ptr<protocol::TableWriteInfo>& tableWriteInfo,

presto-native-execution/presto_cpp/main/types/PrestoToVeloxQueryPlan.h

@@ -15,6 +15,7 @@
 #include <stdexcept>
 #include <vector>
 #include "presto_cpp/main/operators/ShuffleInterface.h"
+#include "presto_cpp/main/tvf/core/TableFunctionProcessorNode.h"
 #include "presto_cpp/presto_protocol/core/presto_protocol_core.h"
 #include "velox/core/Expressions.h"
 #include "velox/core/PlanFragment.h"
@@ -190,6 +191,11 @@ class VeloxQueryPlanConverterBase {
       const std::shared_ptr<protocol::TableWriteInfo>& tableWriteInfo,
       const protocol::TaskId& taskId);
 
+  std::shared_ptr<const tvf::TableFunctionProcessorNode> toVeloxQueryPlan(
+      const std::shared_ptr<const protocol::TableFunctionProcessorNode>& node,
+      const std::shared_ptr<protocol::TableWriteInfo>& tableWriteInfo,
+      const protocol::TaskId& taskId);
+
   std::vector<velox::core::FieldAccessTypedExprPtr> toVeloxExprs(
       const std::vector<protocol::VariableReferenceExpression>& variables);
 

presto-native-execution/presto_cpp/main/types/tests/CMakeLists.txt

@@ -51,6 +51,7 @@ target_link_libraries(
   GTest::gtest_main
   presto_connectors
   $<TARGET_OBJECTS:presto_protocol>
+  $<TARGET_OBJECTS:presto_tvf_core>
   $<TARGET_OBJECTS:presto_type_converter>
   $<TARGET_OBJECTS:presto_types>
   presto_operators