[Native] Register Text Writer

Author: kewang1024

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

@@ -64,6 +64,7 @@ target_link_libraries(
   velox_dwio_orc_reader
   velox_dwio_parquet_reader
   velox_dwio_parquet_writer
+  velox_dwio_text_writer_register
   velox_dynamic_library_loader
   velox_encode
   velox_exec

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -58,6 +58,7 @@
 #include "velox/dwio/orc/reader/OrcReader.h"
 #include "velox/dwio/parquet/RegisterParquetReader.h"
 #include "velox/dwio/parquet/RegisterParquetWriter.h"
+#include "velox/dwio/text/RegisterTextWriter.h"
 #include "velox/exec/OutputBufferManager.h"
 #include "velox/functions/prestosql/aggregates/RegisterAggregateFunctions.h"
 #include "velox/functions/prestosql/registration/RegistrationFunctions.h"
@@ -1413,13 +1414,19 @@ void PrestoServer::registerFileReadersAndWriters() {
   velox::orc::registerOrcReaderFactory();
   velox::parquet::registerParquetReaderFactory();
   velox::parquet::registerParquetWriterFactory();
+  if (SystemConfig::instance()->textWriterEnabled()) {
+    velox::text::registerTextWriterFactory();
+  }
 }
 
 void PrestoServer::unregisterFileReadersAndWriters() {
   velox::dwrf::unregisterDwrfReaderFactory();
   velox::dwrf::unregisterDwrfWriterFactory();
   velox::parquet::unregisterParquetReaderFactory();
   velox::parquet::unregisterParquetWriterFactory();
+  if (SystemConfig::instance()->textWriterEnabled()) {
+    velox::text::unregisterTextWriterFactory();
+  }
 }
 
 void PrestoServer::registerStatsCounters() {

presto-native-execution/presto_cpp/main/common/Configs.cpp

@@ -265,6 +265,7 @@ SystemConfig::SystemConfig() {
           NUM_PROP(kExchangeIoEvbViolationThresholdMs, 1000),
           NUM_PROP(kHttpSrvIoEvbViolationThresholdMs, 1000),
           NUM_PROP(kMaxLocalExchangePartitionBufferSize, 65536),
+        BOOL_PROP(kTextWriterEnabled, false),
       };
 }
 
@@ -919,6 +920,10 @@ uint64_t SystemConfig::maxLocalExchangePartitionBufferSize() const {
   return optionalProperty<uint64_t>(kMaxLocalExchangePartitionBufferSize).value();
 }
 
+bool SystemConfig::textWriterEnabled() const {
+  return optionalProperty<bool>(kTextWriterEnabled).value();
+}
+
 NodeConfig::NodeConfig() {
   registeredProps_ =
       std::unordered_map<std::string, folly::Optional<std::string>>{

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -763,6 +763,11 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kMaxLocalExchangePartitionBufferSize{
       "local-exchange.max-partition-buffer-size"};
 
+  // Add to temporarily help with gradual rollout for text writer
+  // TODO: remove once text writer is fully rolled out
+  static constexpr std::string_view kTextWriterEnabled{
+    "text-writer-enabled"};
+
   SystemConfig();
 
   virtual ~SystemConfig() = default;
@@ -1047,6 +1052,8 @@ class SystemConfig : public ConfigBase {
   int32_t httpSrvIoEvbViolationThresholdMs() const;
 
   uint64_t maxLocalExchangePartitionBufferSize() const;
+
+  bool textWriterEnabled() const;
 };
 
 /// Provides access to node properties defined in node.properties file.

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

@@ -20,6 +20,8 @@
 #include "presto_cpp/presto_protocol/connector/tpch/TpchConnectorProtocol.h"
 
 #include <velox/type/fbhive/HiveTypeParser.h>
+
+#include "presto_cpp/main/common/Configs.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/HiveDataSink.h"
@@ -1139,6 +1141,8 @@ dwio::common::FileFormat toFileFormat(
     case protocol::hive::HiveStorageFormat::ALPHA:
       // This has been renamed in Velox from ALPHA to NIMBLE.
       return dwio::common::FileFormat::NIMBLE;
+    case protocol::hive::HiveStorageFormat::TEXTFILE:
+      return dwio::common::FileFormat::TEXT;
     default:
       VELOX_UNSUPPORTED(
           "Unsupported file format in {}: {}.",