feat: txt file support

.github/workflows/pr_build_linux.yml

@@ -517,6 +517,7 @@ jobs:
               org.apache.comet.iceberg.IcebergReflectionSuite
               org.apache.comet.serde.operator.IcebergWriteProtoTranslationSuite
               org.apache.comet.csv.CometCsvNativeReadSuite
+              org.apache.comet.text.CometTextNativeReadSuite
               org.apache.comet.CometFuzzTestSuite
               org.apache.comet.CometFuzzIcebergSuite
               org.apache.comet.DataGeneratorSuite

.github/workflows/pr_build_macos.yml

@@ -183,6 +183,7 @@ jobs:
               org.apache.comet.iceberg.IcebergReflectionSuite
               org.apache.comet.serde.operator.IcebergWriteProtoTranslationSuite
               org.apache.comet.csv.CometCsvNativeReadSuite
+              org.apache.comet.text.CometTextNativeReadSuite
               org.apache.comet.CometFuzzTestSuite
               org.apache.comet.CometFuzzIcebergSuite
               org.apache.comet.DataGeneratorSuite

docs/source/user-guide/latest/datasources.md

@@ -51,11 +51,27 @@ converted into Arrow format, allowing the Comet pipeline to take over after that
 
 ### Text
 
-Comet does not provide a Rust-based Text scan, but when `spark.comet.convert.text.enabled` is enabled, data from Spark's
-Text reader is immediately converted into Arrow format, allowing the Comet pipeline to take over after that. Spark's Text
-reader keeps the bytes of each line as they are, so a file in an encoding other than UTF-8, such as Latin-1, holds strings
-that are not valid UTF-8. Arrow strings must be valid UTF-8, so the conversion replaces those bytes with `U+FFFD`, and
-results that depend on the original bytes can differ from Spark. Enable it only for Text files that are UTF-8. See
+Comet provides experimental Rust-based text scan support (a single `value: string` column). When
+`spark.comet.scan.text.v2.enabled` is enabled, text files are read in Rust. This feature is experimental and
+performance benefits are workload-dependent. Only Spark's DataSource V2 text scan is accelerated, and Spark reads
+text through the V1 API by default, so also remove `text` from `spark.sql.sources.useV1SourceList`. The native scan
+falls back to Spark for cases it does not handle -- including compressed files, files larger than
+`spark.comet.scan.text.maxFileSize` (64MB by default, since the reader loads each file whole), files Spark splits
+into byte ranges, unsupported filesystem schemes, and `input_file_name()` or
+`ignoreCorruptFiles`/`ignoreMissingFiles` usage. It also falls back for local files that have a Hadoop checksum
+(`.crc`) file, such as data Spark itself wrote to the local filesystem, because Hadoop verifies the checksum on read and
+the native reader does not. It targets small lookup tables such as a broadcast join's build side.
+Reading a partition column from a partitioned directory falls back to Spark, but selecting only `value` stays native.
+
+Text files are a common source of ill-formed UTF-8. The native reader replaces invalid byte sequences with the
+Unicode replacement character (U+FFFD), so byte-level comparisons and functions can differ from Spark on those rows;
+see [Strings with non-UTF-8 bytes](compatibility/index.md#strings-with-non-utf-8-bytes) in the compatibility guide.
+
+Separately, `spark.comet.convert.text.enabled` converts data from Spark's own Text reader into Arrow format
+immediately, allowing the Comet pipeline to take over after that. Spark's Text reader keeps the bytes of each line as
+they are, so a file in an encoding other than UTF-8, such as Latin-1, holds strings that are not valid UTF-8. Arrow
+strings must be valid UTF-8, so the conversion replaces those bytes with `U+FFFD`, and results that depend on the
+original bytes can differ from Spark. Enable it only for Text files that are UTF-8. See
 [Strings with non-UTF-8 bytes](compatibility/index.md#strings-with-non-utf-8-bytes).
 
 ### Other Spark inputs
@@ -130,13 +146,13 @@ Comet supports most standard storage systems, such as local file system and obje
 
 ### Scans over more than one object store
 
-A native Parquet or CSV scan reads each of its partitions through one object store, such as one S3
+A native Parquet, CSV, or Text scan reads each of its partitions through one object store, such as one S3
 bucket, one GCS bucket, or one HDFS name node. When a scan's files live in more than one store,
 Comet puts the files of each store into separate partitions, so such a scan can run with a different
 number of partitions than Spark would use. A native Parquet scan packs each store's files on their
-own, so `spark.sql.files.maxPartitionNum` applies to each store's files separately. A native CSV
-scan that would split a partition falls back to Spark when Spark planned the operators above it for
-the scan's own partitioning, as Spark 3.4 does for a scan of one partition.
+own, so `spark.sql.files.maxPartitionNum` applies to each store's files separately. A native CSV or
+Text scan that would split a partition falls back to Spark when Spark planned the operators above
+it for the scan's own partitioning, as Spark 3.4 does for a scan of one partition.
 
 Such a scan forwards the object store settings of every scheme it reads, so its files can mix
 schemes such as `s3a://`, `gs://` and `hdfs://`. It falls back to Spark when it reads one bucket

docs/source/user-guide/latest/parquet.md

@@ -98,8 +98,9 @@ reads that fail where Spark succeeds are listed in
   query, or cast the column to `BINARY` before persisting, if you need to preserve non-UTF-8 bytes.
   By contrast, Comet decodes non-UTF-8 bytes at the JVM-to-native Arrow FFI boundaries using
   JVM-compatible replacement semantics. This covers native query input from the JVM-exported Arrow
-  stream (including JVM scans, shuffle reads, and `mapInArrow`), columnar-to-row conversion, and JVM
-  UDF results. It does not change the native Parquet reader: rejecting invalid UTF-8 there remains
+  stream (including JVM scans, shuffle reads, and `mapInArrow`), columnar-to-row conversion, JVM
+  UDF results, and the native text scan (which replaces invalid bytes in each line's `value` with
+  U+FFFD). It does not change the native Parquet reader: rejecting invalid UTF-8 there remains
   Gap A of [#4764](https://github.com/apache/datafusion-comet/issues/4764). See
   [Strings with non-UTF-8 bytes](compatibility/index.md#strings-with-non-utf-8-bytes),
   [#4121](https://github.com/apache/datafusion-comet/issues/4121), and the tracking issue above.

docs/source/user-guide/latest/understanding-comet-plans.md

@@ -298,6 +298,7 @@ by role. Names match what is shown in the plan output.
 | `CometNativeScan`        | Parquet scan that runs entirely in Rust via DataFusion.                                                            |
 | `CometIcebergNativeScan` | Iceberg Parquet scan that runs entirely in Rust via DataFusion.                                                    |
 | `CometCsvNativeScan`     | CSV scan that runs entirely in Rust via DataFusion (experimental).                                                 |
+| `CometTextNativeScan`    | Text scan (single `value` string column) that runs entirely in Rust via DataFusion (experimental).                 |
 | `CometInMemoryTableScan` | JVM-side scan of a table cached in Comet's Arrow format (experimental, disabled by default).                       |
 | `CometLocalTableScan`    | JVM-side scan of a `LocalTableScanExec` (disabled by default).                                                     |
 | `CometEmptyRelation`     | Empty native input that replaces `EmptyRelationExec` (Spark 4.0 and later).                                        |

native/core/src/execution/jni_api.rs

@@ -491,6 +491,7 @@ fn op_name(op: &OpStruct) -> &'static str {
         OpStruct::ParquetWriter(_) => "ParquetWriter",
         OpStruct::Explode(_) => "Explode",
         OpStruct::CsvScan(_) => "CsvScan",
+        OpStruct::TextScan(_) => "TextScan",
         OpStruct::ShuffleScan(_) => "ShuffleScan",
         OpStruct::BroadcastNestedLoopJoin(_) => "BroadcastNestedLoopJoin",
         OpStruct::Sample(_) => "Sample",

native/core/src/execution/operators/mod.rs

@@ -49,6 +49,8 @@ pub mod projection;
 mod scan;
 mod shuffle_scan;
 mod sorted_input;
+mod text_scan;
 pub use csv_scan::init_csv_datasource_exec;
 pub use shuffle_scan::ShuffleScanExec;
 pub use sorted_input::SortedInputExec;
+pub use text_scan::init_text_datasource_exec;