refactor(native): Decouple Iceberg from HivePrestoToVeloxConnector

Author: hantangwangd

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

@@ -226,23 +226,40 @@ velox::connector::hive::HiveBucketConversion toVeloxBucketConversion(
   return veloxBucketConversion;
 }
 
-} // namespace
-
-connector::hive::HiveColumnHandle::ColumnType toHiveColumnType(
-    protocol::hive::ColumnType type) {
-  switch (type) {
-    case protocol::hive::ColumnType::PARTITION_KEY:
-      return connector::hive::HiveColumnHandle::ColumnType::kPartitionKey;
-    case protocol::hive::ColumnType::REGULAR:
-      return connector::hive::HiveColumnHandle::ColumnType::kRegular;
-    case protocol::hive::ColumnType::SYNTHESIZED:
-      return connector::hive::HiveColumnHandle::ColumnType::kSynthesized;
-    default:
-      VELOX_UNSUPPORTED(
-          "Unsupported Hive column type: {}.", toJsonString(type));
+dwio::common::FileFormat toVeloxFileFormat(
+    const presto::protocol::hive::StorageFormat& format) {
+  if (format.inputFormat == "com.facebook.hive.orc.OrcInputFormat") {
+    return dwio::common::FileFormat::DWRF;
+  } else if (
+      format.inputFormat == "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat") {
+    return dwio::common::FileFormat::ORC;
+  } else if (
+      format.inputFormat ==
+      "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat") {
+    return dwio::common::FileFormat::PARQUET;
+  } else if (format.inputFormat == "org.apache.hadoop.mapred.TextInputFormat") {
+    if (format.serDe == "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe") {
+      return dwio::common::FileFormat::TEXT;
+    } else if (format.serDe == "org.apache.hive.hcatalog.data.JsonSerDe") {
+      return dwio::common::FileFormat::JSON;
+    }
+  } else if (
+      format.inputFormat ==
+      "org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat") {
+    if (format.serDe ==
+        "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe") {
+      return dwio::common::FileFormat::PARQUET;
+    }
+  } else if (format.inputFormat == "com.facebook.alpha.AlphaInputFormat") {
+    // ALPHA has been renamed in Velox to NIMBLE.
+    return dwio::common::FileFormat::NIMBLE;
   }
+  VELOX_UNSUPPORTED(
+      "Unsupported file format: {} {}", format.inputFormat, format.serDe);
 }
 
+} // namespace
+
 std::unique_ptr<velox::connector::ConnectorSplit>
 HivePrestoToVeloxConnector::toVeloxSplit(
     const protocol::ConnectorId& catalogId,

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

@@ -24,9 +24,6 @@
 
 namespace facebook::presto {
 
-velox::connector::hive::HiveColumnHandle::ColumnType toHiveColumnType(
-    protocol::hive::ColumnType type);
-
 class HivePrestoToVeloxConnector final : public PrestoToVeloxConnector {
  public:
   explicit HivePrestoToVeloxConnector(std::string connectorName)

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

@@ -13,7 +13,6 @@
  */
 
 #include "presto_cpp/main/connectors/IcebergPrestoToVeloxConnector.h"
-#include "presto_cpp/main/connectors/HivePrestoToVeloxConnector.h"
 #include "presto_cpp/main/connectors/PrestoToVeloxConnectorUtils.h"
 
 #include "presto_cpp/presto_protocol/connector/iceberg/IcebergConnectorProtocol.h"

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

@@ -728,36 +728,19 @@ velox::common::CompressionKind toFileCompressionKind(
   }
 }
 
-dwio::common::FileFormat toVeloxFileFormat(
-    const presto::protocol::hive::StorageFormat& format) {
-  if (format.inputFormat == "com.facebook.hive.orc.OrcInputFormat") {
-    return dwio::common::FileFormat::DWRF;
-  } else if (
-      format.inputFormat == "org.apache.hadoop.hive.ql.io.orc.OrcInputFormat") {
-    return dwio::common::FileFormat::ORC;
-  } else if (
-      format.inputFormat ==
-      "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat") {
-    return dwio::common::FileFormat::PARQUET;
-  } else if (format.inputFormat == "org.apache.hadoop.mapred.TextInputFormat") {
-    if (format.serDe == "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe") {
-      return dwio::common::FileFormat::TEXT;
-    } else if (format.serDe == "org.apache.hive.hcatalog.data.JsonSerDe") {
-      return dwio::common::FileFormat::JSON;
-    }
-  } else if (
-      format.inputFormat ==
-      "org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat") {
-    if (format.serDe ==
-        "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe") {
-      return dwio::common::FileFormat::PARQUET;
-    }
-  } else if (format.inputFormat == "com.facebook.alpha.AlphaInputFormat") {
-    // ALPHA has been renamed in Velox to NIMBLE.
-    return dwio::common::FileFormat::NIMBLE;
+connector::hive::HiveColumnHandle::ColumnType toHiveColumnType(
+    protocol::hive::ColumnType type) {
+  switch (type) {
+    case protocol::hive::ColumnType::PARTITION_KEY:
+      return connector::hive::HiveColumnHandle::ColumnType::kPartitionKey;
+    case protocol::hive::ColumnType::REGULAR:
+      return connector::hive::HiveColumnHandle::ColumnType::kRegular;
+    case protocol::hive::ColumnType::SYNTHESIZED:
+      return connector::hive::HiveColumnHandle::ColumnType::kSynthesized;
+    default:
+      VELOX_UNSUPPORTED(
+          "Unsupported Hive column type: {}.", toJsonString(type));
   }
-  VELOX_UNSUPPORTED(
-      "Unsupported file format: {} {}", format.inputFormat, format.serDe);
 }
 
 std::unique_ptr<velox::connector::ConnectorTableHandle> toHiveTableHandle(

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

@@ -17,6 +17,7 @@
 #include "presto_cpp/presto_protocol/connector/hive/presto_protocol_hive.h"
 #include "presto_cpp/presto_protocol/core/presto_protocol_core.h"
 #include "velox/connectors/Connector.h"
+#include "velox/connectors/hive/TableHandle.h"
 #include "velox/dwio/common/Options.h"
 #include "velox/type/Filter.h"
 #include "velox/type/Type.h"
@@ -46,8 +47,8 @@ std::vector<velox::common::Subfield> toRequiredSubfields(
 velox::common::CompressionKind toFileCompressionKind(
     const protocol::hive::HiveCompressionCodec& hiveCompressionCodec);
 
-velox::dwio::common::FileFormat toVeloxFileFormat(
-    const presto::protocol::hive::StorageFormat& format);
+velox::connector::hive::HiveColumnHandle::ColumnType toHiveColumnType(
+    protocol::hive::ColumnType type);
 
 std::unique_ptr<velox::connector::ConnectorTableHandle> toHiveTableHandle(
     const protocol::TupleDomain<protocol::Subfield>& domainPredicate,