[native] Table writer 1: Pass LocationHandle to HiveInsertTableHandle

presto-native-execution/CMakeLists.txt

@@ -165,6 +165,7 @@ include_directories(.)
 include_directories(velox)
 include_directories(velox/velox/external/xxhash)
 include_directories(${VELOX_ROOT})
+include_directories(${CMAKE_BINARY_DIR})
 
 add_subdirectory(velox)
 

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

@@ -25,20 +25,8 @@
 #include "presto_cpp/main/types/TypeSignatureTypeConverter.h"
 // clang-format on
 
-#include <boost/uuid/uuid.hpp>
-#include <boost/uuid/uuid_generators.hpp>
-#include <boost/uuid/uuid_io.hpp>
-
 #include <folly/container/F14Set.h>
 
-#if __has_include("filesystem")
-#include <filesystem>
-namespace fs = std::filesystem;
-#else
-#include <experimental/filesystem>
-namespace fs = std::experimental::filesystem;
-#endif
-
 using namespace facebook::velox;
 
 namespace facebook::presto {
@@ -112,6 +100,46 @@ std::shared_ptr<connector::ColumnHandle> toColumnHandle(
   throw std::invalid_argument("Unknown column handle type: " + column->_type);
 }
 
+connector::hive::LocationHandle::TableType toTableType(
+    protocol::TableType tableType) {
+  switch (tableType) {
+    case protocol::TableType::NEW:
+      return connector::hive::LocationHandle::TableType::kNew;
+    case protocol::TableType::EXISTING:
+      return connector::hive::LocationHandle::TableType::kExisting;
+    case protocol::TableType::TEMPORARY:
+      return connector::hive::LocationHandle::TableType::kTemporary;
+    default:
+      throw std::invalid_argument("Unknown table type");
+  }
+}
+
+connector::hive::LocationHandle::WriteMode toWriteMode(
+    protocol::WriteMode writeMode) {
+  switch (writeMode) {
+    case protocol::WriteMode::STAGE_AND_MOVE_TO_TARGET_DIRECTORY:
+      return connector::hive::LocationHandle::WriteMode::
+          kStageAndMoveToTargetDirectory;
+    case protocol::WriteMode::DIRECT_TO_TARGET_NEW_DIRECTORY:
+      return connector::hive::LocationHandle::WriteMode::
+          kDirectToTargetNewDirectory;
+    case protocol::WriteMode::DIRECT_TO_TARGET_EXISTING_DIRECTORY:
+      return connector::hive::LocationHandle::WriteMode::
+          kDirectToTargetExistingDirectory;
+    default:
+      throw std::invalid_argument("Unknown write mode");
+  }
+}
+
+std::shared_ptr<connector::hive::LocationHandle> toLocationHandle(
+    const protocol::LocationHandle& locationHandle) {
+  return std::make_shared<connector::hive::LocationHandle>(
+      locationHandle.targetPath,
+      locationHandle.writePath,
+      toTableType(locationHandle.tableType),
+      toWriteMode(locationHandle.writeMode));
+}
+
 int64_t toInt64(
     const std::shared_ptr<protocol::Block>& block,
     const VeloxExprConverter& exprConverter,
@@ -1632,24 +1660,51 @@ VeloxQueryPlanConverter::toVeloxQueryPlan(
     const std::shared_ptr<const protocol::TableWriterNode>& node,
     const std::shared_ptr<protocol::TableWriteInfo>& tableWriteInfo,
     const protocol::TaskId& taskId) {
-  auto outputTableHandle = std::dynamic_pointer_cast<protocol::CreateHandle>(
-                               tableWriteInfo->writerTarget)
-                               ->handle;
-
-  auto hiveOutputTableHandle =
-      std::dynamic_pointer_cast<protocol::HiveOutputTableHandle>(
-          outputTableHandle.connectorHandle);
+  std::string connectorId;
+  std::vector<std::shared_ptr<const connector::hive::HiveColumnHandle>>
+      inputColumns;
+  std::shared_ptr<connector::ConnectorInsertTableHandle> hiveTableHandle;
+  if (auto createHandle = std::dynamic_pointer_cast<protocol::CreateHandle>(
+          tableWriteInfo->writerTarget)) {
+    connectorId = createHandle->handle.connectorId;
+
+    auto hiveOutputTableHandle =
+        std::dynamic_pointer_cast<protocol::HiveOutputTableHandle>(
+            createHandle->handle.connectorHandle);
+
+    for (const auto& columnHandle : hiveOutputTableHandle->inputColumns) {
+      inputColumns.emplace_back(
+          std::dynamic_pointer_cast<connector::hive::HiveColumnHandle>(
+              toColumnHandle(&columnHandle)));
+    }
 
-  auto uuid = boost::uuids::random_generator()();
-  auto fileName = boost::uuids::to_string(uuid);
+    hiveTableHandle = std::make_shared<connector::hive::HiveInsertTableHandle>(
+        inputColumns, toLocationHandle(hiveOutputTableHandle->locationHandle));
+  } else if (
+      auto insertHandle = std::dynamic_pointer_cast<protocol::InsertHandle>(
+          tableWriteInfo->writerTarget)) {
+    connectorId = insertHandle->handle.connectorId;
+
+    auto hiveInsertTableHandle =
+        std::dynamic_pointer_cast<protocol::HiveInsertTableHandle>(
+            insertHandle->handle.connectorHandle);
+
+    for (const auto& columnHandle : hiveInsertTableHandle->inputColumns) {
+      inputColumns.emplace_back(
+          std::dynamic_pointer_cast<connector::hive::HiveColumnHandle>(
+              toColumnHandle(&columnHandle)));
+    }
 
-  auto filePath =
-      fs::path(hiveOutputTableHandle->locationHandle.writePath) / fileName;
+    hiveTableHandle = std::make_shared<connector::hive::HiveInsertTableHandle>(
+        inputColumns, toLocationHandle(hiveInsertTableHandle->locationHandle));
+  } else {
+    VELOX_UNSUPPORTED(
+        "Unsupported table writer handle: {}",
+        toJsonString(tableWriteInfo->writerTarget));
+  }
 
-  auto hiveTableHandle =
-      std::make_shared<velox::connector::hive::HiveInsertTableHandle>(filePath);
-  auto insertTableHandle = std::make_shared<core::InsertTableHandle>(
-      outputTableHandle.connectorId, hiveTableHandle);
+  auto insertTableHandle =
+      std::make_shared<core::InsertTableHandle>(connectorId, hiveTableHandle);
 
   auto outputType = toRowType(
       {node->rowCountVariable,

presto-native-execution/presto_cpp/presto_protocol/special/ConnectorInsertTableHandle.cpp.inc

@@ -0,0 +1,49 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+namespace facebook::presto::protocol {
+void to_json(json& j, const std::shared_ptr<ConnectorInsertTableHandle>& p) {
+  if (p == nullptr) {
+    return;
+  }
+  String type = p->_type;
+
+  if (getConnectorKey(type) == "hive") {
+    j = *std::static_pointer_cast<HiveInsertTableHandle>(p);
+    return;
+  }
+
+  throw TypeError(type + " no abstract type ConnectorInsertTableHandle ");
+}
+
+void from_json(const json& j, std::shared_ptr<ConnectorInsertTableHandle>& p) {
+  String type;
+  try {
+    type = p->getSubclassKey(j);
+  } catch (json::parse_error& e) {
+    throw ParseError(
+        std::string(e.what()) +
+        " ConnectorInsertTableHandle  ConnectorInsertTableHandle");
+  }
+
+  if (getConnectorKey(type) == "hive") {
+    std::shared_ptr<HiveInsertTableHandle> k =
+        std::make_shared<HiveInsertTableHandle>();
+    j.get_to(*k);
+    p = std::static_pointer_cast<ConnectorInsertTableHandle>(k);
+    return;
+  }
+
+  throw TypeError(type + " no abstract type ConnectorInsertTableHandle ");
+}
+} // namespace facebook::presto::protocol