refactor: Introduce ConnectorObjectFactory and HiveObjectFactory

velox/connectors/Connector.cpp

@@ -169,4 +169,8 @@ folly::dynamic ConnectorTableHandle::serialize() const {
   return serializeBase("ConnectorTableHandle");
 }
 
+// Allow the compiler to emit vtable and typeinfo for ConnectorLocationHandle at
+// a single point instead of every translation unit.
+ConnectorLocationHandle::~ConnectorLocationHandle() = default;
+
 } // namespace facebook::velox::connector

velox/connectors/Connector.h

@@ -165,6 +165,33 @@ class ConnectorInsertTableHandle : public ISerializable {
   }
 };
 
+class ConnectorLocationHandle : public ISerializable {
+ public:
+  enum class TableType { kNew, kExisting, kTemp };
+
+  ConnectorLocationHandle(const std::string& connectorId, TableType tableType)
+      : connectorId_{connectorId}, tableType_{tableType} {}
+
+  virtual ~ConnectorLocationHandle();
+
+  const std::string& connectorId() const {
+    return connectorId_;
+  }
+
+  /// New vs existing vs temp.
+  TableType tableType() const {
+    return tableType_;
+  }
+
+  virtual std::string toString() const = 0;
+
+  virtual folly::dynamic serialize() const = 0;
+
+ private:
+  const std::string connectorId_;
+  const TableType tableType_;
+};
+
 using ConnectorInsertTableHandlePtr =
     std::shared_ptr<const ConnectorInsertTableHandle>;
 

velox/connectors/ConnectorNames.h

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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
+
+namespace facebook::velox::connector {
+
+// TODO: Add a demo connector plugin
+
+constexpr const char* kFuzzerConnectorName = "fuzzer";
+constexpr const char* kHiveConnectorName = "hive";
+constexpr const char* kIcebergConnectorName = "iceberg";
+constexpr const char* kTpchConnectorName = "tpch";
+
+} // namespace facebook::velox::connector

velox/connectors/ConnectorObjectFactory.cpp

@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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.
+ */
+
+#include "velox/connectors/ConnectorObjectFactory.h"
+
+namespace facebook::velox::connector {
+
+ConnectorObjectFactory::~ConnectorObjectFactory() = default;
+
+std::unordered_map<std::string, std::shared_ptr<ConnectorObjectFactory>>&
+connectorObjectFactories() {
+  static std::
+      unordered_map<std::string, std::shared_ptr<ConnectorObjectFactory>>
+          factories;
+  return factories;
+}
+
+bool registerConnectorObjectFactory(
+    std::shared_ptr<ConnectorObjectFactory> factory) {
+  bool ok =
+      connectorObjectFactories().insert({factory->name(), factory}).second;
+  VELOX_CHECK(
+      ok,
+      "ConnectorObjectFactory with name '{}' is already registered",
+      factory->name());
+  return true;
+}
+
+bool hasConnectorObjectFactory(const std::string& connectorName) {
+  return connectorObjectFactories().count(connectorName) == 1;
+}
+
+bool unregisterConnectorObjectFactory(const std::string& connectorName) {
+  auto count = connectorObjectFactories().erase(connectorName);
+  return count == 1;
+}
+
+std::shared_ptr<ConnectorObjectFactory> getConnectorObjectFactory(
+    const std::string& connectorName) {
+  auto it = connectorObjectFactories().find(connectorName);
+  VELOX_CHECK(
+      it != connectorObjectFactories().end(),
+      "ConnectorObjectFactory with name '{}' not registered",
+      connectorName);
+  return it->second;
+}
+
+} // namespace facebook::velox::connector

velox/connectors/ConnectorObjectFactory.h

@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * 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 <shared_mutex>
+#include <string>
+
+#include "velox/connectors/Connector.h"
+
+namespace facebook::velox::connector {
+
+class ConnectorObjectFactory {
+ public:
+  ConnectorObjectFactory(const std::string& name) : name_(name) {}
+
+  virtual ~ConnectorObjectFactory();
+
+  const std::string& name() const {
+    return name_;
+  }
+
+  virtual std::shared_ptr<ConnectorSplit> makeSplit(
+      const std::string& connectorId,
+      const folly::dynamic& options = {}) const {
+    VELOX_UNSUPPORTED("ConnectorSplit not supported by connector", connectorId);
+  }
+
+  virtual std::shared_ptr<connector::ColumnHandle> makeColumnHandle(
+      const std::string& connectorId,
+      const std::string& name,
+      const TypePtr& type,
+      const folly::dynamic& options = {}) const {
+    VELOX_UNSUPPORTED(
+        "connector::ColumnHandle not supported by connector", connectorId);
+  }
+
+  virtual std::shared_ptr<ConnectorTableHandle> makeTableHandle(
+      const std::string& connectorId,
+      const std::string& tableName,
+      std::vector<ColumnHandlePtr> columnHandles,
+      const folly::dynamic& options) const {
+    VELOX_UNSUPPORTED(
+        "ConnectorTableHandle not supported by connector", connectorId);
+  }
+
+  virtual std::shared_ptr<ConnectorInsertTableHandle> makeInsertTableHandle(
+      const std::string& connectorId,
+      std::vector<ColumnHandlePtr> inputColumns,
+      std::shared_ptr<const ConnectorLocationHandle> locationHandle,
+      const folly::dynamic& options = {}) const {
+    VELOX_UNSUPPORTED(
+        "ConnectorInsertTableHandle not supported by connector", connectorId);
+  }
+
+  virtual std::shared_ptr<ConnectorLocationHandle> makeLocationHandle(
+      const std::string& connectorId,
+      ConnectorLocationHandle::TableType tableType =
+          ConnectorLocationHandle::TableType::kNew,
+      const folly::dynamic& options = {}) const {
+    VELOX_UNSUPPORTED(
+        "ConnectorLocationHandle not supported by connector", connectorId);
+  }
+
+ private:
+  const std::string name_;
+};
+
+bool registerConnectorObjectFactory(
+    std::shared_ptr<ConnectorObjectFactory> factory);
+
+bool hasConnectorObjectFactory(const std::string& connectorName);
+
+bool unregisterConnectorObjectFactory(const std::string& connectorName);
+
+std::shared_ptr<ConnectorObjectFactory> getConnectorObjectFactory(
+    const std::string& connectorName);
+
+} // namespace facebook::velox::connector

velox/connectors/hive/CMakeLists.txt

@@ -27,6 +27,7 @@ velox_add_library(
   HiveConnectorSplit.cpp
   HiveDataSink.cpp
   HiveDataSource.cpp
+  HiveObjectFactory.cpp
   HivePartitionUtil.cpp
   PartitionIdGenerator.cpp
   SplitReader.cpp

velox/connectors/hive/HiveConnectorSplit.h

@@ -17,6 +17,7 @@
 
 #include <optional>
 #include <unordered_map>
+
 #include "velox/connectors/Connector.h"
 #include "velox/connectors/hive/FileProperties.h"
 #include "velox/connectors/hive/TableHandle.h"

velox/connectors/hive/HiveDataSink.cpp

@@ -1205,7 +1205,7 @@ std::string LocationHandle::toString() const {
       "LocationHandle [targetPath: {}, writePath: {}, tableType: {}, tableFileName: {}]",
       targetPath_,
       writePath_,
-      tableTypeName(tableType_),
+      tableTypeName(tableType()),
       targetFileName_);
 }
 
@@ -1219,7 +1219,7 @@ folly::dynamic LocationHandle::serialize() const {
   obj["name"] = "LocationHandle";
   obj["targetPath"] = targetPath_;
   obj["writePath"] = writePath_;
-  obj["tableType"] = tableTypeName(tableType_);
+  obj["tableType"] = tableTypeName(tableType());
   obj["targetFileName"] = targetFileName_;
   return obj;
 }

velox/connectors/hive/HiveDataSink.h

@@ -17,6 +17,7 @@
 
 #include "velox/common/compression/Compression.h"
 #include "velox/connectors/Connector.h"
+#include "velox/connectors/ConnectorNames.h"
 #include "velox/connectors/hive/HiveConfig.h"
 #include "velox/connectors/hive/PartitionIdGenerator.h"
 #include "velox/connectors/hive/TableHandle.h"
@@ -31,24 +32,18 @@ class LocationHandle;
 using LocationHandlePtr = std::shared_ptr<const LocationHandle>;
 
 /// Location related properties of the Hive table to be written.
-class LocationHandle : public ISerializable {
+class LocationHandle : public ConnectorLocationHandle {
  public:
-  enum class TableType {
-    /// Write to a new table to be created.
-    kNew,
-    /// Write to an existing table.
-    kExisting,
-  };
-
   LocationHandle(
       std::string targetPath,
       std::string writePath,
       TableType tableType,
-      std::string targetFileName = "")
-      : targetPath_(std::move(targetPath)),
+      std::string targetFileName = "",
+      const std::string& connectorId = kHiveConnectorName)
+      : ConnectorLocationHandle(connectorId, tableType),
+        targetPath_(std::move(targetPath)),
         targetFileName_(std::move(targetFileName)),
-        writePath_(std::move(writePath)),
-        tableType_(tableType) {}
+        writePath_(std::move(writePath)) {}
 
   const std::string& targetPath() const {
     return targetPath_;
@@ -62,16 +57,12 @@ class LocationHandle : public ISerializable {
     return writePath_;
   }
 
-  TableType tableType() const {
-    return tableType_;
-  }
+  std::string toString() const override;
 
-  std::string toString() const;
+  folly::dynamic serialize() const override;
 
   static void registerSerDe();
 
-  folly::dynamic serialize() const override;
-
   static LocationHandlePtr create(const folly::dynamic& obj);
 
   static const std::string tableTypeName(LocationHandle::TableType type);
@@ -85,8 +76,6 @@ class LocationHandle : public ISerializable {
   const std::string targetFileName_;
   // Staging directory path.
   const std::string writePath_;
-  // Whether the table to be written is new, already existing or temporary.
-  const TableType tableType_;
 };
 
 class HiveSortingColumn : public ISerializable {