[native] Add TextReader

Author: xin-zhang2

presto-native-execution/pom.xml

@@ -413,7 +413,7 @@
                 <configuration>
                     <forkCount>1</forkCount>
                     <reuseForks>false</reuseForks>
-                    <excludedGroups>remote-function,textfile_reader</excludedGroups>
+                    <excludedGroups>remote-function</excludedGroups>
                     <systemPropertyVariables>
                         <PRESTO_SERVER>/root/project/build/debug/presto_cpp/main/presto_server</PRESTO_SERVER>
                         <DATA_DIR>/tmp/velox</DATA_DIR>
@@ -465,7 +465,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>

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

@@ -70,6 +70,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

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"
@@ -1430,7 +1431,8 @@ void PrestoServer::registerFileReadersAndWriters() {
   velox::orc::registerOrcReaderFactory();
   velox::parquet::registerParquetReaderFactory();
   velox::parquet::registerParquetWriterFactory();
-  if (SystemConfig::instance()->textWriterEnabled()) {
+  if (SystemConfig::instance()->textReaderWriterEnabled()) {
+    velox::text::registerTextReaderFactory();
     velox::text::registerTextWriterFactory();
   }
 }
@@ -1440,7 +1442,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();
   }
 }

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

@@ -261,7 +261,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),
           BOOL_PROP(kEnumTypesEnabled, true),
       };
@@ -929,8 +929,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 {

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

@@ -761,9 +761,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.
@@ -1067,7 +1067,7 @@ class SystemConfig : public ConfigBase {
 
   uint64_t maxLocalExchangePartitionBufferSize() const;
 
-  bool textWriterEnabled() const;
+  bool textReaderWriterEnabled() const;
 
   bool charNToVarcharImplicitCast() const;
 

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

@@ -413,6 +413,76 @@ public void testDateFilter()
         }
     }
 
+    @Test(groups = {"textfile"})
+    public void testReadTableWithTextfileFormat()
+    {
+        assertQuery("SELECT * FROM nation_text");
+
+        String tmpTableName = generateRandomTableName();
+        try {
+            getExpectedQueryRunner().execute(getSession(), 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>)," +
+                            "price DECIMAL(15,2)," +
+                            "amount DECIMAL(21,6)," +
+                            "event_date DATE," +
+                            "ds VARCHAR" +
+                            ") WITH (format = 'TEXTFILE', partitioned_by = ARRAY['ds'])", tmpTableName), ImmutableList.of());
+            getExpectedQueryRunner().execute(getSession(), format(
+                    "INSERT INTO %s (" +
+                            "id," +
+                            "name," +
+                            "is_active," +
+                            "score," +
+                            "created_at," +
+                            "tags," +
+                            "metrics," +
+                            "properties," +
+                            "flags," +
+                            "nested_struct," +
+                            "price," +
+                            "amount," +
+                            "event_date," +
+                            "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'])" +
+                            ")," +
+                            "DECIMAL '12.34'," +
+                            "CAST('-123456789012345.123456' as DECIMAL(21,6))," +
+                            "DATE '2024-02-29'," +
+                            "'2025-07-01'" +
+                            ")", tmpTableName), ImmutableList.of());
+            // created_at is skipped because of the inconsistency in TIMESTAMP columns between Presto and Velox.
+            // https://github.com/facebookincubator/velox/issues/8127
+            assertQuery(format("SELECT id, name, is_active, score, tags, metrics, properties, flags, nested_struct, price, amount, event_date, ds FROM %s", tmpTableName));
+        }
+        finally {
+            dropTableIfExists(tmpTableName);
+        }
+    }
+
     @Test
     public void testOrderBy()
     {
@@ -1269,18 +1339,6 @@ public void testReadTableWithUnsupportedJsonFormat()
         assertQueryFails("SELECT * FROM nation_json", "(?s).*ReaderFactory is not registered for format json.*");
     }
 
-    @Test(groups = {"no_textfile_reader"})
-    public void testReadTableWithUnsupportedTextfileFormat()
-    {
-        assertQueryFails("SELECT * FROM nation_text", "(?s).*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()

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;

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();
+    }
+}

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();
+    }
+}

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

@@ -82,10 +82,6 @@ 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".
-    @Override
-    public void testReadTableWithTextfileFormat() {}
-
     // Disable: Not supporte by POS
     @Override
     @Ignore