Add CLP connector native code

Author: wraymo

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

@@ -55,6 +55,7 @@ target_link_libraries(
   velox_abfs
   velox_aggregates
   velox_caching
+  velox_clp_connector
   velox_common_base
   velox_core
   velox_dwio_common_exception

presto-native-execution/presto_cpp/main/connectors/PrestoToVeloxConnector.cpp

@@ -15,11 +15,15 @@
 #include "presto_cpp/main/connectors/PrestoToVeloxConnector.h"
 #include "presto_cpp/main/types/PrestoToVeloxExpr.h"
 #include "presto_cpp/main/types/TypeParser.h"
+#include "presto_cpp/presto_protocol/connector/clp/ClpConnectorProtocol.h"
 #include "presto_cpp/presto_protocol/connector/hive/HiveConnectorProtocol.h"
 #include "presto_cpp/presto_protocol/connector/iceberg/IcebergConnectorProtocol.h"
 #include "presto_cpp/presto_protocol/connector/tpch/TpchConnectorProtocol.h"
 
 #include <velox/type/fbhive/HiveTypeParser.h>
+#include "velox/connectors/clp/ClpColumnHandle.h"
+#include "velox/connectors/clp/ClpConnectorSplit.h"
+#include "velox/connectors/clp/ClpTableHandle.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/HiveDataSink.h"
@@ -1412,9 +1416,11 @@ IcebergPrestoToVeloxConnector::toVeloxColumnHandle(
   //  constructor similar to how Hive Connector is handling for bucketing
   velox::type::fbhive::HiveTypeParser hiveTypeParser;
   auto type = stringToType(icebergColumn->type, typeParser);
-  connector::hive::HiveColumnHandle::ColumnParseParameters columnParseParameters;
+  connector::hive::HiveColumnHandle::ColumnParseParameters
+      columnParseParameters;
   if (type->isDate()) {
-    columnParseParameters.partitionDateValueFormat = connector::hive::HiveColumnHandle::ColumnParseParameters::kDaysSinceEpoch;
+    columnParseParameters.partitionDateValueFormat = connector::hive::
+        HiveColumnHandle::ColumnParseParameters::kDaysSinceEpoch;
   }
   return std::make_unique<connector::hive::HiveColumnHandle>(
       icebergColumn->columnIdentity.name,
@@ -1548,4 +1554,62 @@ std::unique_ptr<protocol::ConnectorProtocol>
 TpchPrestoToVeloxConnector::createConnectorProtocol() const {
   return std::make_unique<protocol::tpch::TpchConnectorProtocol>();
 }
+
+std::unique_ptr<velox::connector::ConnectorSplit>
+ClpPrestoToVeloxConnector::toVeloxSplit(
+    const protocol::ConnectorId& catalogId,
+    const protocol::ConnectorSplit* connectorSplit,
+    const protocol::SplitContext* splitContext) const {
+  auto clpSplit = dynamic_cast<const protocol::clp::ClpSplit*>(connectorSplit);
+  VELOX_CHECK_NOT_NULL(
+      clpSplit, "Unexpected split type {}", connectorSplit->_type);
+  return std::make_unique<connector::clp::ClpConnectorSplit>(
+      catalogId, clpSplit->path);
+}
+
+std::unique_ptr<velox::connector::ColumnHandle>
+ClpPrestoToVeloxConnector::toVeloxColumnHandle(
+    const protocol::ColumnHandle* column,
+    const TypeParser& typeParser) const {
+  auto clpColumn = dynamic_cast<const protocol::clp::ClpColumnHandle*>(column);
+  VELOX_CHECK_NOT_NULL(
+      clpColumn, "Unexpected column handle type {}", column->_type);
+  return std::make_unique<connector::clp::ClpColumnHandle>(
+      clpColumn->columnName,
+      clpColumn->originalColumnName,
+      typeParser.parse(clpColumn->columnType),
+      clpColumn->nullable);
+}
+
+std::unique_ptr<velox::connector::ConnectorTableHandle>
+ClpPrestoToVeloxConnector::toVeloxTableHandle(
+    const protocol::TableHandle& tableHandle,
+    const VeloxExprConverter& exprConverter,
+    const TypeParser& typeParser,
+    std::unordered_map<
+        std::string,
+        std::shared_ptr<velox::connector::ColumnHandle>>& assignments) const {
+  auto clpLayout =
+      std::dynamic_pointer_cast<const protocol::clp::ClpTableLayoutHandle>(
+          tableHandle.connectorTableLayout);
+  VELOX_CHECK_NOT_NULL(
+      clpLayout,
+      "Unexpected layout type {}",
+      tableHandle.connectorTableLayout->_type);
+  auto storageType =
+      (clpLayout->table.storageType == protocol::clp::StorageType::S3)
+      ? connector::clp::ClpTableHandle::StorageType::kS3
+      : connector::clp::ClpTableHandle::StorageType::kFS;
+  return std::make_unique<connector::clp::ClpTableHandle>(
+      tableHandle.connectorId,
+      clpLayout->table.schemaTableName.table,
+      storageType,
+      clpLayout->kqlQuery);
+}
+
+std::unique_ptr<protocol::ConnectorProtocol>
+ClpPrestoToVeloxConnector::createConnectorProtocol() const {
+  return std::make_unique<protocol::clp::ClpConnectorProtocol>();
+}
+
 } // namespace facebook::presto

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

@@ -13,9 +13,9 @@
  */
 #pragma once
 
+#include "presto_cpp/main/types/PrestoToVeloxExpr.h"
 #include "presto_cpp/presto_protocol/connector/hive/presto_protocol_hive.h"
 #include "presto_cpp/presto_protocol/core/ConnectorProtocol.h"
-#include "presto_cpp/main/types/PrestoToVeloxExpr.h"
 #include "velox/connectors/Connector.h"
 #include "velox/connectors/hive/TableHandle.h"
 #include "velox/core/PlanNode.h"
@@ -223,4 +223,31 @@ class TpchPrestoToVeloxConnector final : public PrestoToVeloxConnector {
   std::unique_ptr<protocol::ConnectorProtocol> createConnectorProtocol()
       const final;
 };
+
+class ClpPrestoToVeloxConnector final : public PrestoToVeloxConnector {
+ public:
+  explicit ClpPrestoToVeloxConnector(std::string connectorName)
+      : PrestoToVeloxConnector(std::move(connectorName)) {}
+
+  std::unique_ptr<velox::connector::ConnectorSplit> toVeloxSplit(
+      const protocol::ConnectorId& catalogId,
+      const protocol::ConnectorSplit* connectorSplit,
+      const protocol::SplitContext* splitContext) const final;
+
+  std::unique_ptr<velox::connector::ColumnHandle> toVeloxColumnHandle(
+      const protocol::ColumnHandle* column,
+      const TypeParser& typeParser) const final;
+
+  std::unique_ptr<velox::connector::ConnectorTableHandle> toVeloxTableHandle(
+      const protocol::TableHandle& tableHandle,
+      const VeloxExprConverter& exprConverter,
+      const TypeParser& typeParser,
+      std::unordered_map<
+          std::string,
+          std::shared_ptr<velox::connector::ColumnHandle>>& assignments)
+      const final;
+
+  std::unique_ptr<protocol::ConnectorProtocol> createConnectorProtocol()
+      const final;
+};
 } // namespace facebook::presto

presto-native-execution/presto_cpp/main/connectors/Registration.cpp

@@ -19,6 +19,7 @@
 #include "presto_cpp/main/connectors/arrow_flight/ArrowPrestoToVeloxConnector.h"
 #endif
 
+#include "velox/connectors/clp/ClpConnector.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/tpch/TpchConnector.h"
 
@@ -45,6 +46,12 @@ void registerConnectorFactories() {
         std::make_shared<velox::connector::tpch::TpchConnectorFactory>());
   }
 
+  if (!velox::connector::hasConnectorFactory(
+          velox::connector::clp::ClpConnectorFactory::kClpConnectorName)) {
+    velox::connector::registerConnectorFactory(
+        std::make_shared<velox::connector::clp::ClpConnectorFactory>());
+  }
+
   // Register Velox connector factory for iceberg.
   // The iceberg catalog is handled by the hive connector factory.
   if (!velox::connector::hasConnectorFactory(kIcebergConnectorName)) {
@@ -74,6 +81,8 @@ void registerConnectors() {
       std::make_unique<IcebergPrestoToVeloxConnector>(kIcebergConnectorName));
   registerPrestoToVeloxConnector(std::make_unique<TpchPrestoToVeloxConnector>(
       velox::connector::tpch::TpchConnectorFactory::kTpchConnectorName));
+  registerPrestoToVeloxConnector(std::make_unique<ClpPrestoToVeloxConnector>(
+      velox::connector::clp::ClpConnectorFactory::kClpConnectorName));
 
   // Presto server uses system catalog or system schema in other catalogs
   // in different places in the code. All these resolve to the SystemConnector.

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

@@ -47,6 +47,7 @@ target_link_libraries(
   $<TARGET_OBJECTS:presto_types>
   velox_hive_connector
   velox_tpch_connector
+  velox_clp_connector
   velox_presto_serializer
   velox_functions_prestosql
   velox_aggregates

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

@@ -27,6 +27,7 @@ target_link_libraries(
   velox_dwio_orc_reader
   velox_hive_connector
   velox_tpch_connector
+  velox_clp_connector
   velox_exec
   velox_dwio_common_exception
   presto_type_converter
@@ -64,6 +65,7 @@ target_link_libraries(
   velox_functions_lib
   velox_hive_connector
   velox_tpch_connector
+  velox_clp_connector
   velox_hive_partition_function
   velox_presto_serializer
   velox_serialization
@@ -96,6 +98,7 @@ target_link_libraries(
   velox_dwio_common
   velox_hive_connector
   velox_tpch_connector
+  velox_clp_connector
   GTest::gtest
   GTest::gtest_main)
 

presto-native-execution/presto_cpp/main/types/tests/PrestoToVeloxConnectorTest.cpp

@@ -33,6 +33,8 @@ TEST_F(PrestoToVeloxConnectorTest, registerVariousConnectors) {
       "iceberg", std::make_unique<IcebergPrestoToVeloxConnector>("iceberg")));
   connectorList.emplace_back(
       std::pair("tpch", std::make_unique<HivePrestoToVeloxConnector>("tpch")));
+  connectorList.emplace_back(
+      std::pair("clp", std::make_unique<ClpPrestoToVeloxConnector>("clp")));
 
   for (auto& [connectorName, connector] : connectorList) {
     registerPrestoToVeloxConnector(std::move(connector));

presto-native-execution/presto_cpp/presto_protocol/Makefile

@@ -52,16 +52,25 @@ presto_protocol-cpp: presto_protocol-json
 	chevron -d  connector/arrow_flight/presto_protocol_arrow_flight.json connector/arrow_flight/presto_protocol-json-hpp.mustache >> connector/arrow_flight/presto_protocol_arrow_flight.h
 	clang-format -style=file -i connector/arrow_flight/presto_protocol_arrow_flight.h connector/arrow_flight/presto_protocol_arrow_flight.cpp
 
-presto_protocol-json: 
+    # build clp connector related structs
+    echo "// DO NOT EDIT : This file is generated by chevron" > connector/clp/presto_protocol_clp.cpp
+    chevron -d  connector/clp/presto_protocol_clp.json connector/clp/presto_protocol-json-cpp.mustache >> connector/clp/presto_protocol_clp.cpp
+    echo "// DO NOT EDIT : This file is generated by chevron" > connector/clp/presto_protocol_clp.h
+    chevron -d  connector/clp/presto_protocol_clp.json connector/clp/presto_protocol-json-hpp.mustache >> connector/clp/presto_protocol_clp.h
+    clang-format -style=file -i connector/clp/presto_protocol_clp.h connector/clp/presto_protocol_clp.cpp
+
+presto_protocol-json:
 	./java-to-struct-json.py --config core/presto_protocol_core.yml core/special/*.java core/special/*.inc -j | jq . > core/presto_protocol_core.json
 	./java-to-struct-json.py --config connector/hive/presto_protocol_hive.yml connector/hive/special/*.inc -j | jq . > connector/hive/presto_protocol_hive.json
 	./java-to-struct-json.py --config connector/iceberg/presto_protocol_iceberg.yml connector/iceberg/special/*.inc -j | jq . > connector/iceberg/presto_protocol_iceberg.json
 	./java-to-struct-json.py --config connector/tpch/presto_protocol_tpch.yml connector/tpch/special/*.inc -j | jq . > connector/tpch/presto_protocol_tpch.json
 	./java-to-struct-json.py --config connector/arrow_flight/presto_protocol_arrow_flight.yml connector/arrow_flight/special/*.inc -j | jq . > connector/arrow_flight/presto_protocol_arrow_flight.json
+	./java-to-struct-json.py --config connector/clp/presto_protocol_clp.yml connector/clp/special/*.inc -j | jq . > connector/clp/presto_protocol_clp.json
 
 presto_protocol.proto: presto_protocol-json
 	pystache presto_protocol-protobuf.mustache core/presto_protocol_core.json > core/presto_protocol_core.proto
 	pystache presto_protocol-protobuf.mustache connector/hive/presto_protocol_hive.json > connector/hive/presto_protocol_hive.proto
 	pystache presto_protocol-protobuf.mustache connector/iceberg/presto_protocol_iceberg.json > connector/iceberg/presto_protocol_iceberg.proto
 	pystache presto_protocol-protobuf.mustache connector/tpch/presto_protocol_tpch.json > connector/tpch/presto_protocol_tpch.proto
 	pystache presto_protocol-protobuf.mustache connector/arrow_flight/presto_protocol_arrow_flight.json > connector/arrow_flight/presto_protocol_arrow_flight.proto
+	pystache presto_protocol-protobuf.mustache connector/clp/presto_protocol_clp.json > connector/clp/presto_protocol_clp.proto

presto-native-execution/presto_cpp/presto_protocol/connector/clp/ClpConnectorProtocol.h

@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+#pragma once
+#include "presto_cpp/presto_protocol/connector/clp/presto_protocol_clp.h"
+#include "presto_cpp/presto_protocol/core/ConnectorProtocol.h"
+
+namespace facebook::presto::protocol::clp {
+using ClpConnectorProtocol = ConnectorProtocolTemplate<
+    ClpTableHandle,
+    ClpTableLayoutHandle,
+    ClpColumnHandle,
+    NotImplemented,
+    NotImplemented,
+    ClpSplit,
+    NotImplemented,
+    ClpTransactionHandle,
+    NotImplemented>;
+} // namespace facebook::presto::protocol::clp