[native] Add TextReader

xin-zhang2 · xin-zhang2 · commit 1ffcec3c36be · 2025-09-09T14:02:25.000+01:00
diff --git a/presto-native-execution/pom.xml b/presto-native-execution/pom.xml
@@ -382,7 +382,7 @@
                 <configuration>
                     <forkCount>1</forkCount>
                     <reuseForks>false</reuseForks>
-                    <excludedGroups>remote-function,textfile_reader</excludedGroups>
+                    <excludedGroups>remote-function,textfile</excludedGroups>
                     <systemPropertyVariables>
                         <PRESTO_SERVER>/root/project/build/debug/presto_cpp/main/presto_server</PRESTO_SERVER>
                         <DATA_DIR>/tmp/velox</DATA_DIR>
@@ -434,7 +434,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,async_data_cache</excludedGroups>
                         </configuration>
                     </plugin>
                 </plugins>
diff --git a/presto-native-execution/presto_cpp/main/CMakeLists.txt b/presto-native-execution/presto_cpp/main/CMakeLists.txt
@@ -65,6 +65,7 @@ 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
diff --git a/presto-native-execution/presto_cpp/main/PrestoServer.cpp b/presto-native-execution/presto_cpp/main/PrestoServer.cpp
@@ -59,6 +59,7 @@
 #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"
@@ -1412,8 +1413,9 @@ void PrestoServer::registerFileReadersAndWriters() {
   velox::orc::registerOrcReaderFactory();
   velox::parquet::registerParquetReaderFactory();
   velox::parquet::registerParquetWriterFactory();
-  if (SystemConfig::instance()->textWriterEnabled()) {
+  if (SystemConfig::instance()->textReaderWriterEnabled()) {
     velox::text::registerTextWriterFactory();
+    velox::text::registerTextReaderFactory();
   }
 }
 
@@ -1422,7 +1424,8 @@ void PrestoServer::unregisterFileReadersAndWriters() {
   velox::dwrf::unregisterDwrfWriterFactory();
   velox::parquet::unregisterParquetReaderFactory();
   velox::parquet::unregisterParquetWriterFactory();
-  if (SystemConfig::instance()->textWriterEnabled()) {
+  if (SystemConfig::instance()->textReaderWriterEnabled()) {
+    velox::text::unregisterTextReaderFactory();
     velox::text::unregisterTextWriterFactory();
   }
 }
diff --git a/presto-native-execution/presto_cpp/main/common/Configs.cpp b/presto-native-execution/presto_cpp/main/common/Configs.cpp
@@ -260,7 +260,7 @@ SystemConfig::SystemConfig() {
           NUM_PROP(kExchangeIoEvbViolationThresholdMs, 1000),
           NUM_PROP(kHttpSrvIoEvbViolationThresholdMs, 1000),
           NUM_PROP(kMaxLocalExchangePartitionBufferSize, 65536),
-          BOOL_PROP(kTextWriterEnabled, true),
+          BOOL_PROP(kTextReaderWriterEnabled, true),
           BOOL_PROP(kCharNToVarcharImplicitCast, false),
       };
 }
@@ -923,8 +923,8 @@ uint64_t SystemConfig::maxLocalExchangePartitionBufferSize() const {
       .value();
 }
 
-bool SystemConfig::textWriterEnabled() const {
-  return optionalProperty<bool>(kTextWriterEnabled).value();
+bool SystemConfig::textReaderWriterEnabled() const {
+  return optionalProperty<bool>(kTextReaderWriterEnabled).value();
 }
 
 bool SystemConfig::charNToVarcharImplicitCast() const {
diff --git a/presto-native-execution/presto_cpp/main/common/Configs.h b/presto-native-execution/presto_cpp/main/common/Configs.h
@@ -759,9 +759,9 @@ 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"};
+  // Add to temporarily help with gradual rollout for text reader and writer
+  // TODO: remove once text reader and writer are fully rolled out
+  static constexpr std::string_view kTextReaderWriterEnabled{"text-reader-writer-enabled"};
 
   /// Enable the type char(n) with the same behavior as unbounded varchar.
   /// char(n) type is not supported by parser when set to false.
@@ -1057,7 +1057,7 @@ class SystemConfig : public ConfigBase {
 
   uint64_t maxLocalExchangePartitionBufferSize() const;
 
-  bool textWriterEnabled() const;
+  bool textReaderWriterEnabled() const;
 
   bool charNToVarcharImplicitCast() const;
 };
diff --git a/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeGeneralQueries.java b/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeGeneralQueries.java
@@ -403,6 +403,70 @@ public void testDateFilter()
         }
     }
 
+    @Test(groups = {"textfile"})
+    public void testTableReader()
+    {
+        assertQuery("SELECT * FROM nation_text");
+    }
+
+    @Test(groups = {"textfile"})
+    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', partitioned_by = ARRAY['ds'])", 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 +1323,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 +1723,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 +1882,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();
diff --git a/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeTpcdsQueries.java b/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeTpcdsQueries.java
@@ -90,6 +90,7 @@ private static void createTpcdsCallCenter(QueryRunner queryRunner, Session sessi
             switch (storageFormat) {
                 case "PARQUET":
                 case "ORC":
+                case "TEXTFILE":
                     queryRunner.execute(session, "CREATE TABLE call_center AS " +
                             "SELECT * FROM tpcds.tiny.call_center");
                     break;
@@ -158,6 +159,7 @@ private static void createTpcdsDateDim(QueryRunner queryRunner, Session session,
             switch (storageFormat) {
                 case "PARQUET":
                 case "ORC":
+                case "TEXTFILE":
                     queryRunner.execute(session, "CREATE TABLE date_dim AS " +
                             "SELECT * FROM tpcds.tiny.date_dim");
                     break;
@@ -202,6 +204,7 @@ private static void createTpcdsItem(QueryRunner queryRunner, Session session, St
             switch (storageFormat) {
                 case "PARQUET":
                 case "ORC":
+                case "TEXTFILE":
                     queryRunner.execute(session, "CREATE TABLE item AS " +
                             "SELECT * FROM tpcds.tiny.item");
                     break;
@@ -246,6 +249,7 @@ private static void createTpcdsStore(QueryRunner queryRunner, Session session, S
             switch (storageFormat) {
                 case "PARQUET":
                 case "ORC":
+                case "TEXTFILE":
                     queryRunner.execute(session, "CREATE TABLE store AS " +
                             "SELECT * FROM tpcds.tiny.store");
                     break;
@@ -300,6 +304,7 @@ private static void createTpcdsWebPage(QueryRunner queryRunner, Session session,
             switch (storageFormat) {
                 case "PARQUET":
                 case "ORC":
+                case "TEXTFILE":
                     queryRunner.execute(session, "CREATE TABLE web_page AS " +
                             "SELECT * FROM tpcds.tiny.web_page");
                     break;
@@ -337,6 +342,7 @@ private static void createTpcdsWebSite(QueryRunner queryRunner, Session session,
             switch (storageFormat) {
                 case "PARQUET":
                 case "ORC":
+                case "TEXTFILE":
                     queryRunner.execute(session, "CREATE TABLE web_site AS " +
                             "SELECT * FROM tpcds.tiny.web_site");
                     break;
diff --git a/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/TestPrestoNativeTpcdsQueriesTextfileUsingThrift.java b/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/TestPrestoNativeTpcdsQueriesTextfileUsingThrift.java
@@ -0,0 +1,45 @@
+/*
+ * 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.
+ */
+package com.facebook.presto.nativeworker;
+
+import com.facebook.presto.testing.ExpectedQueryRunner;
+import com.facebook.presto.testing.QueryRunner;
+import org.testng.annotations.Test;
+
+@Test(groups = {"textfile"})
+public class TestPrestoNativeTpcdsQueriesTextfileUsingThrift
+        extends AbstractTestNativeTpcdsQueries
+{
+    @Override
+    protected QueryRunner createQueryRunner()
+            throws Exception
+    {
+        return PrestoNativeQueryRunnerUtils.nativeHiveQueryRunnerBuilder()
+                .setStorageFormat("TEXTFILE")
+                .setAddStorageFormatToPath(true)
+                .setUseThrift(true)
+                .build();
+    }
+
+    @Override
+    protected ExpectedQueryRunner createExpectedQueryRunner()
+            throws Exception
+    {
+        this.storageFormat = "TEXTFILE";
+        return PrestoNativeQueryRunnerUtils.javaHiveQueryRunnerBuilder()
+                .setStorageFormat(this.storageFormat)
+                .setAddStorageFormatToPath(true)
+                .build();
+    }
+}
diff --git a/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/TestPrestoNativeTpchQueriesTextfileUsingJSON.java b/presto-native-execution/src/test/java/com/facebook/presto/nativeworker/TestPrestoNativeTpchQueriesTextfileUsingJSON.java
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+package com.facebook.presto.nativeworker;
+
+import com.facebook.presto.testing.ExpectedQueryRunner;
+import com.facebook.presto.testing.QueryRunner;
+import org.testng.annotations.Test;
+
+@Test(groups = {"textfile"})
+public class TestPrestoNativeTpchQueriesTextfileUsingJSON
+        extends AbstractTestNativeTpchQueries
+{
+    private final String storageFormat = "TEXTFILE";
+
+    @Override
+    protected QueryRunner createQueryRunner() throws Exception
+    {
+        return PrestoNativeQueryRunnerUtils.nativeHiveQueryRunnerBuilder()
+                .setStorageFormat(storageFormat)
+                .setAddStorageFormatToPath(true)
+                .build();
+    }
+
+    @Override
+    protected ExpectedQueryRunner createExpectedQueryRunner() throws Exception
+    {
+        return PrestoNativeQueryRunnerUtils.javaHiveQueryRunnerBuilder()
+                .setStorageFormat(storageFormat)
+                .setAddStorageFormatToPath(true)
+                .build();
+    }
+}
diff --git a/presto-native-execution/src/test/java/com/facebook/presto/spark/TestPrestoSparkNativeGeneralQueries.java b/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
diff --git a/presto-native-tests/pom.xml b/presto-native-tests/pom.xml
@@ -234,7 +234,7 @@
                 <configuration>
                     <forkCount>1</forkCount>
                     <reuseForks>false</reuseForks>
-                    <excludedGroups>remote-function,textfile_reader</excludedGroups>
+                    <excludedGroups>remote-function,textfile</excludedGroups>
                     <systemPropertyVariables>
                         <PRESTO_SERVER>/root/project/build/debug/presto_cpp/main/presto_server</PRESTO_SERVER>
                     </systemPropertyVariables>

Original file line number	Diff line number	Diff line change
`@@ -260,7 +260,7 @@ SystemConfig::SystemConfig() {`
`260`	`260`	`NUM_PROP(kExchangeIoEvbViolationThresholdMs, 1000),`
`261`	`261`	`NUM_PROP(kHttpSrvIoEvbViolationThresholdMs, 1000),`
`262`	`262`	`NUM_PROP(kMaxLocalExchangePartitionBufferSize, 65536),`
`263`		`- BOOL_PROP(kTextWriterEnabled, true),`
	`263`	`+ BOOL_PROP(kTextReaderWriterEnabled, true),`
`264`	`264`	`BOOL_PROP(kCharNToVarcharImplicitCast, false),`
`265`	`265`	`};`
`266`	`266`	`}`
`@@ -923,8 +923,8 @@ uint64_t SystemConfig::maxLocalExchangePartitionBufferSize() const {`
`923`	`923`	`.value();`
`924`	`924`	`}`
`925`	`925`
`926`		`-bool SystemConfig::textWriterEnabled() const {`
`927`		`- return optionalProperty<bool>(kTextWriterEnabled).value();`
	`926`	`+bool SystemConfig::textReaderWriterEnabled() const {`
	`927`	`+ return optionalProperty<bool>(kTextReaderWriterEnabled).value();`
`928`	`928`	`}`
`929`	`929`
`930`	`930`	`bool SystemConfig::charNToVarcharImplicitCast() const {`