[native] Add TextReader and TextWriter

presto-native-execution/pom.xml

@@ -351,7 +351,7 @@
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-surefire-plugin</artifactId>
                         <configuration>
-                            <excludedGroups combine.self="override">writer,parquet,remote-function,textfile_reader,no_textfile_reader,async_data_cache</excludedGroups>
+                            <excludedGroups combine.self="override">writer,parquet,remote-function,textfile_reader,async_data_cache</excludedGroups>
                         </configuration>
                     </plugin>
                 </plugins>

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

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

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

@@ -58,6 +58,8 @@
 #include "velox/dwio/orc/reader/OrcReader.h"
 #include "velox/dwio/parquet/RegisterParquetReader.h"
 #include "velox/dwio/parquet/RegisterParquetWriter.h"
+#include "velox/dwio/text/RegisterTextReader.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"
@@ -1397,13 +1399,17 @@ void PrestoServer::registerFileReadersAndWriters() {
   velox::orc::registerOrcReaderFactory();
   velox::parquet::registerParquetReaderFactory();
   velox::parquet::registerParquetWriterFactory();
+  velox::text::registerTextReaderFactory();
+  velox::text::registerTextWriterFactory();
 }
 
 void PrestoServer::unregisterFileReadersAndWriters() {
   velox::dwrf::unregisterDwrfReaderFactory();
   velox::dwrf::unregisterDwrfWriterFactory();
   velox::parquet::unregisterParquetReaderFactory();
   velox::parquet::unregisterParquetWriterFactory();
+  velox::text::unregisterTextReaderFactory();
+  velox::text::unregisterTextWriterFactory();
 }
 
 void PrestoServer::registerStatsCounters() {

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

@@ -910,6 +910,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 {}: {}.",

presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeGeneralQueries.java

@@ -403,6 +403,64 @@ public void testDateFilter()
         }
     }
 
+    @Test(groups = {"textfile_reader"})
+    public void testTableWrite()
+    {
+        String tmpTableName = generateRandomTableName();
+        try {
+            getQueryRunner().execute(String.format(
+                    "CREATE TABLE %s (" +
+                            "id BIGINT," +
+                            "name VARCHAR," +
+                            "is_active BOOLEAN," +
+                            "score DOUBLE," +
+                            "created_at TIMESTAMP," +
+                            "tags ARRAY<VARCHAR>," +
+                            "metrics ARRAY<DOUBLE>," +
+                            "properties MAP<VARCHAR, VARCHAR>," +
+                            "flags MAP<TINYINT, BOOLEAN>," +
+                            "nested_struct ROW(sub_id INTEGER, sub_name VARCHAR, sub_scores ARRAY<REAL>, sub_map MAP<SMALLINT, VARCHAR>)," +
+                            "ds VARCHAR" +
+                            ") WITH (format = 'TEXTFILE', oncall = 'presto', partitioned_by = ARRAY['ds'], retention_days = 30)", tmpTableName));
+
+            getQueryRunner().execute(String.format(
+                    "INSERT INTO %s (" +
+                            "id," +
+                            "name," +
+                            "is_active," +
+                            "score," +
+                            "created_at," +
+                            "tags," +
+                            "metrics," +
+                            "properties," +
+                            "flags," +
+                            "nested_struct," +
+                            "ds" +
+                            ") VALUES (" +
+                            "1001," +
+                            "'Jane Doe'," +
+                            "TRUE," +
+                            "88.5," +
+                            "TIMESTAMP '2025-07-23 10:00:00'," +
+                            "ARRAY['alpha', 'beta', 'gamma']," +
+                            "ARRAY[3.14, 2.71, 1.41]," +
+                            "MAP(ARRAY['color', 'size'], ARRAY['blue', 'large'])," +
+                            "MAP(ARRAY[TINYINT '1', TINYINT '2'], ARRAY[TRUE, FALSE])," +
+                            "ROW(" +
+                            "42," +
+                            "'sub_jane'," +
+                            "ARRAY[REAL '1.1', REAL '2.2', REAL '3.3']," +
+                            "MAP(ARRAY[SMALLINT '10', SMALLINT '20'], ARRAY['foo', 'bar'])" +
+                            ")," +
+                            "'2025-07-01'" +
+                            ")", tmpTableName));
+            assertQueryResultCount(String.format("SELECT count(*) FROM %s", tmpTableName), 1);
+        }
+        finally {
+            dropTableIfExists(tmpTableName);
+        }
+    }
+
     @Test
     public void testOrderBy()
     {
@@ -1259,18 +1317,6 @@ public void testReadTableWithUnsupportedJsonFormat()
         assertQueryFails("SELECT * FROM nation_json", ".*ReaderFactory is not registered for format json.*");
     }
 
-    @Test(groups = {"no_textfile_reader"})
-    public void testReadTableWithUnsupportedTextfileFormat()
-    {
-        assertQueryFails("SELECT * FROM nation_text", ".*ReaderFactory is not registered for format text.*");
-    }
-
-    @Test(groups = {"textfile_reader"})
-    public void testReadTableWithTextfileFormat()
-    {
-        assertQuery("SELECT * FROM nation_text");
-    }
-
     private void dropTableIfExists(String tableName)
     {
         // An ugly workaround for the lack of getExpectedQueryRunner()
@@ -1671,10 +1717,10 @@ public void testSelectFieldsWithCapitalLetters()
                     ColumnMetadata.builder()
                             .setName("col")
                             .setType(RowType.from(ImmutableList.of(
-                            new RowType.Field(Optional.of("NationKey"), BIGINT),
-                            new RowType.Field(Optional.of("NAME"), VARCHAR),
-                            new RowType.Field(Optional.of("ReGiOnKeY"), BIGINT),
-                            new RowType.Field(Optional.of("commenT"), VARCHAR))))
+                                    new RowType.Field(Optional.of("NationKey"), BIGINT),
+                                    new RowType.Field(Optional.of("NAME"), VARCHAR),
+                                    new RowType.Field(Optional.of("ReGiOnKeY"), BIGINT),
+                                    new RowType.Field(Optional.of("commenT"), VARCHAR))))
                             .build()),
                     tableProperties);
             transaction(queryRunner.getTransactionManager(), queryRunner.getAccessControl())
@@ -1830,15 +1876,16 @@ public void testUnicodeInJson()
     {
         // Test casting to JSON returning the same results for all unicode characters in the
         // entire range.
-        List<int[]> unicodeRanges = new ArrayList<int[]>() {
+        List<int[]> unicodeRanges = new ArrayList<int[]>()
+        {
             {
-                add(new int[]{0, 0x7F});
-                add(new int[]{0x80, 0xD7FF});
-                add(new int[]{0xE000, 0xFFFF});
+                add(new int[] {0, 0x7F});
+                add(new int[] {0x80, 0xD7FF});
+                add(new int[] {0xE000, 0xFFFF});
             }
         };
         for (int start = 0x10000; start < 0x110000; start += 0x10000) {
-            unicodeRanges.add(new int[]{start, start + 0xFFFF});
+            unicodeRanges.add(new int[] {start, start + 0xFFFF});
         }
         List<String> unicodeStrings = unicodeRanges.stream().map(range -> {
             StringBuilder unicodeString = new StringBuilder();

presto-native-execution/src/test/java/com/facebook/presto/spark/TestPrestoSparkNativeGeneralQueries.java

@@ -66,9 +66,9 @@ public void testUnicodeInJson()
     @Ignore
     public void testDistributedSortSingleNode() {}
 
-    // Disable: Text file reader is not supported. This test is also disabled in pom.xml through disabling groups "textfile_reader".
+    // Disable: Text file reader and writer is not supported.
     @Override
-    public void testReadTableWithTextfileFormat() {}
+    public void testTableWrite() {}
 
     // Disable: Not supporte by POS
     @Override