[parthchandra]

Which issue does this PR close?

Part of #6008.

Rationale for this change

Comet cannot natively scan Spark's text file format, reporting [COMET: Unsupported file format Text].
This cascades on the build side of a broadcast join. When a small lookup table is read from a text file,
the build branch (text scan -> ... -> BroadcastExchange) stays on Spark, the exchange never becomes a
CometBroadcastExchange, and the BroadcastHashJoin cannot run natively even when the large probe side
is fully native. A tiny text lookup can disqualify native acceleration of the whole join.

This PR implements option B from #6008: a native Text reader (single value: string column,
line-delimited), mirroring the existing native CSV V2 scan path, so a text leaf no longer blocks native
execution. Option A (a generic SparkToColumnar bridge for any unsupported build side) is in #6786 and
is complementary.

What changes are included in this PR?

New config spark.comet.scan.text.v2.enabled (default false, experimental). As with the native CSV
scan, only Spark's DataSource V2 text path is accelerated, so text must also be removed from
spark.sql.sources.useV1SourceList. A second config, spark.comet.scan.text.maxFileSize (default 64MB),
caps the size of each file the native reader will take.

Native (Rust):

• New TextSource/TextOpener (native/core/src/execution/operators/text_scan.rs): a DataFusion
  FileSource/FileOpener producing a single nullable value: Utf8 column. Line splitting matches
  Spark/Hadoop LineRecordReader: default universal newline (\n, \r\n, \r), a custom lineSep
  (the exact charset-encoded lineSeparatorInRead bytes), or wholetext=true (one row per file).
• BOM handling follows Hadoop's skipUtfByteOrderMark. In line mode the reader splits first and then
  strips a leading UTF-8 BOM from the first record only, for any separator. A file that is only a BOM has
  no rows, and a BOM followed by a separator leaves an empty first row. lineSep equal to the BOM is
  unaffected, because the BOM is then the separator. wholetext keeps the BOM.
• Bytes are decoded the same lossy way Spark renders them on collect (decode_utf8_spark_lossy). Each
  batch's value buffer is bounded below Arrow's 2GB offset limit, returning a clean error on a single
  oversized value rather than panicking.
• Native text is restricted to unsplit files (CometScanRule falls back for split files). A defensive
  byte-range filter keeps results correct even if a split reached the reader.
• New TextScan/TextOptions protobuf messages, planner wiring (OpStruct::TextScan), and
  operator-registry / jni_api entries. Like the CSV arm, the planner builds the object store from the
  current partition's first file, because partitions can differ in store.

JVM (Scala):

• CometTextNativeScanExec (mirrors CometCsvNativeScanExec), the configs, and dispatch in
  CometExecRule. A declined Text conversion falls back to the wrapped Spark scan, as CSV does.
• Multiple object stores: the serde declines unsafe store combinations through
  CometScanUtils.multiStoreFallbackReason and splits a Spark partition that mixes stores through
  splitPartitionsByStore, so each native partition reads one store. This uses the same machinery as the
  native CSV scan.
• CometScanRule claims a V2 TextScan only when safe, otherwise falling back to Spark:
  • native exec disabled, or input_file_name/input_file_block_start/input_file_block_length (the
    shared V2 gate)
  • a multi-column read schema (matching Spark's TextScan.verifyReadSchema), partition columns, or
    metadata columns
  • compressed files, through a new version-aware ShimFileFormat.isCompressedFile. Spark 4.1+ also
    treats .gzip and .zstd as compressed (HadoopCodecStreams), earlier versions use
    CompressionCodecFactory.
  • files Spark split into byte ranges, files over spark.comet.scan.text.maxFileSize, and oversized
    wholetext files
  • ignoreCorruptFiles/ignoreMissingFiles, and a finite mapreduce.input.linerecordreader.line.maxlength
  • a self-overlapping custom lineSep (for example aab). Hadoop's readCustomLine backs up one byte
    on a mismatch instead of following the separator's own overlap, so it can split differently from an
    exact scan, depending on where the separator falls in its input buffer. Separators with no
    self-overlap (\r\n, multi-byte CJK, and so on) stay native.
  • unsupported filesystem schemes and object_store-rejected paths
• The scheme, multi-store and rejected-path checks are now one shared helper,
  CometScanRule.rootPathFallbackReason, used by both the native Parquet V1 gate and the Text gate.
  Parquet's fallback messages are unchanged.

Changes to existing code:

• CometCsvNativeScanExec builds its Hadoop conf from the scan's read options (case preserved), matching
  Spark's CSVScan.createReaderFactory, so per-read settings such as fs.s3a.endpoint reach native
  execution. It also overrides stringArgs so EXPLAIN, the SQL tab and the event log no longer print the
  serialized plan with every input path. The Text node does both too.
• The Parquet V1 gate calls the shared rootPathFallbackReason helper described above instead of
  carrying its own copy of the checks.

Docs: a Text subsection in the data sources guide, a CometTextNativeScan row in
understanding-comet-plans.md, and the native text scan added to the non-UTF-8 handling note in
parquet.md. The multi-store section of the data sources guide now names Text.

How are these changes tested?

• Rust unit tests in text_scan.rs: line splitting (universal newline, trailing terminator,
  empty/interior-empty lines, custom single- and multi-byte separators, wholetext), byte-range split
  partitioning, first-record BOM stripping (default and custom separators, BOM-only file, BOM followed by
  a separator, lineSep equal to the BOM), invalid UTF-8 -> U+FFFD, and the count/empty-projection path.
• CometTextNativeReadSuite compares Comet to Spark (checkSparkAnswerAndOperator) for basic read,
  count, filter, wholetext (+ count, directory), custom lineSep (including a trailing separator and a
  non-ASCII multibyte separator), CRLF/CR line endings, invalid UTF-8, BOMs (default separator, custom
  separator, lineSep equal to the BOM, BOM-only files), multi-file directories, files in separate
  partitions, and empty files. It also covers a broadcast join with a text build side running fully
  native.
• The same suite asserts the correct fallback (checkSparkAnswerAndFallbackReason) for split files,
  compressed files (.gz, and .gzip/.zstd on Spark 4.1+), ignoreCorruptFiles/ignoreMissingFiles,
  line.maxlength, files over the size ceiling, input_file_name, large wholetext, a self-overlapping
  lineSep, partition columns, and native-exec-disabled.
• AbfsScanFallbackSuite covers a Text scan over two ABFS accounts and over two containers of one
  account. The serializer declines both layouts, and the query must fall back to Spark and return the
  right rows rather than fail with a ClassCastException.
• New suites are registered in pr_build_linux.yml and pr_build_macos.yml.
• Compiles against the Spark 4.0 and 4.1 profiles. The five-profile CI matrix exercises the
  version-specific compression shims.

[andygrove]

Thanks for taking this on, Parth. Lining the reader up with the CSV path and covering BOMs, custom lineSep bytes and wholetext is a lot of careful work. I checked the line splitting against Spark with a differential fuzz of 150 random byte files across 8 option variants. That is 1200 comparisons with BOMs, bare \r, \r\n, invalid UTF-8 and several lineSep values, and I found no differences on Spark 4.1 and 3.5. A Parquet fact table joined to a text lookup also goes fully native, which is the point of the PR. The optimized native library is faster than Spark's reader as well.

I am requesting changes for two cases where the native scan returns different rows from Spark without raising an error. Both are in inline comments.

1. On Spark 4.1 and later, .gzip and .zstd files are decompressed by Spark but are not caught by the compressed-file check, so the native scan returns the raw compressed bytes.
2. mapreduce.input.linerecordreader.line.maxlength makes Spark skip long lines, and the native scan ignores it.

I would also like the memory use addressed in the same round. The reader holds about 3.3 times the file size outside the memory pool, and the inline comment on text_scan.rs has the numbers and some options. The remaining inline comments cover the duplicated checks in CometScanRule, stringArgs, and test coverage.

Two doc updates fall outside this diff, so I am noting them here. The Scans table in docs/source/user-guide/latest/understanding-comet-plans.md lists CometCsvNativeScan and needs a CometTextNativeScan row. The list of paths that decode non-UTF-8 bytes in docs/source/user-guide/latest/compatibility/scans.md should include the native text reader.

The compression handling depends on the Spark version and the PR CI only runs the default profile, so the run-all-spark-profiles label is worth running once the fix is in.

[andygrove]

This test only calls collect(), so it would still pass if the scan fell back to Spark. Could it also assert that the executed plan contains CometTextNativeScanExec?

[parthchandra]

Done. preserves line order within a file now calls
checkSparkAnswerAndOperator(df, Seq(classOf[CometTextNativeScanExec]))
first, so it fails if the scan falls back. The collect() is kept only to check order.

[andygrove]

Two details here don't match the code. The wholetext cap in CometScanRule is min(spark.sql.files.maxPartitionBytes, Int.MaxValue / 3), which is 128 MB by default, so "roughly 2GB" is misleading. And only reads of partition columns fall back. Selecting just value from a partitioned directory stays native.

Text files are also the most likely way for ill-formed UTF-8 to reach Comet. The reader replaces it with U+FFFD, so byte-level functions and equality can differ from Spark on those rows. For the bytes 61 FF 62, value = 'a\uFFFDb' is true in Comet and false in Spark. Could this section link to "Strings with non-UTF-8 bytes" in the compatibility guide?

[parthchandra]

Fixed all three: the size-limit wording, the partition-column note (selecting value stays native), and a link to the non-UTF-8 section.

[parthchandra]

Thanks for the thorough review and the differential fuzz — really appreciate it. I've addressed all the inline comments plus the two out-of-diff doc notes.

[parthchandra]

Fixed all three: the size-limit wording, the partition-column note (selecting value stays native), and a link to the non-UTF-8 section.

[sunchao]

Reviewed 23bba3ea1824d3b40db0b2126be162576bbe1097 against base 9876862df98ee10fbded5365a4abeaa4a2b022b9. The PR is not a draft. Reviewed all 23 files across all three PR commits using merge base 7a1c042f37c0a4e0cad97ba16aaf558ad30dcb69. Base-only join and SQL-test changes are not PR regressions.

Routed skills: review-comet-pr, review-comet-ffi-pr, and review-comet-memory-pr. Also consulted review-comet-expression-pr during scope analysis. Read the existing reviews and all nine discussion threads. Two new P2 findings are detailed below.

Summary

• Prior state and problem: Text scans remained on Spark and could prevent a text lookup's broadcast join from running natively. This PR adds an opt-in native V2 text reader.
• Design approach: Follow the existing CSV scan integration, serialize Spark's text options and file partitions, and use a DataFusion FileSource with conservative Spark fallbacks.
• Correctness: Compared Spark's text options, scans, reader factories, and line-reader selection across 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. Reproduced a custom-separator/BOM mismatch that drops or changes records. A separate JVM probe confirms that the new serializer loses per-read Hadoop settings. Whole-file mode, empty projections, line termination, and Arrow offset bounds were also inspected.
• Compatibility analysis: The version-specific compression shims address the earlier .gzip/.zstd concern. The Hadoop line-length guard addresses the earlier skipped-record concern. Compared the affected paths with latest release branch branch-1.1 at e9efd9f764ee0a59b7898ff028d6985d4a7a28e1: native text support is new and disabled by default. Invalid UTF-8 replacement remains explicitly documented.
• Key design decisions: Require both V2 text routing and the native-text flag. Decline unsupported partition/metadata columns, split or compressed files, input-file expressions, and oversized files. The new file-size ceiling defaults to 64 MB.
• Implementation sketch: CometScanRule checks eligibility, CometTextNativeScanExec builds the protobuf, and the native planner creates TextSource/TextOpener. The opener reads, splits, decodes, and batches the file. Existing native-to-JVM Arrow export machinery is reused without changing its ownership rules.
• Performance: The reader still buffers the whole file, its line descriptors, and all output batches before yielding. The ceiling limits input size but does not reserve those allocations against the memory pool. No exact-head throughput benchmark was run. The CSV/text stringArgs changes avoid printing file-bearing protobufs in plans.
• Design: Reusing the CSV/DataFusion scan structure keeps the feature understandable. The shared object-store eligibility helper removes duplicated Parquet checks, but its alias-only bucket policy does not fully address the existing multi-bucket concern for the new text path.
• Abstraction & complexity: The source/opener split and projection wrapper fit DataFusion's interfaces. No separate abstraction defect met the P1/P2 bar. Configuration should flow consistently from the scan's options through eligibility checking and serialization.
• Behavioral changes worth calling out: Eligible text reads can now participate in native pipelines and broadcast joins. The two findings introduce unintended behavior when that opt-in is enabled. The existing multi-bucket concern remains for ordinary s3:// and s3a:// roots: aliasScanBuckets returns empty without an alias, while the text planner selects the first file's object store and strips the bucket from subsequent keys. Files from another bucket can therefore be read from the first bucket. This existing concern is not repeated as a new inline finding.
• Suggested improvements: Apply BOM handling after identifying the first record, preserving Hadoop's consumed-byte behavior. Serialize the text scan's case-preserved Hadoop options. Address the remaining ordinary multi-bucket case in the existing discussion, and add focused regression coverage for these triggers.

Exact-head CI: GitHub reports only the successful label check. No build or test verdict is available for this SHA, although run-all-spark-profiles is present.

Validation: Compiled the unchanged native splitting helpers into a disposable Rust probe and compared them with Hadoop's reader, Spark's reader, and Spark 4.1.3 V2 SQL execution. Ran a bounded Spark JVM configuration-propagation probe. git diff --check passed and the checkout remains unchanged. The focused Cargo test build failed in hdfs-sys because jni.h is unavailable. Full native/JVM suites, cloud reads, and Spark SQL CI suites were not run. The reproductions do not claim an integrated Comet execution test.

[sunchao]

[P2] Apply BOM handling after splitting the first record. With native V2 text enabled, a UTF-8 file containing \uFEFFa\uFEFFb and .option("lineSep", "\uFEFF") should return ["", "a", "b"], as Spark does. This code removes the initial separator before splitting and returns ["a", "b"]. With separator \uFEFFx, it also changes the first value from a to xa. These inputs pass the eligibility checks and silently change counts, filters, and join results. Hadoop reads the first record before checking its value for a BOM. Follow its first-record BOM/consumed-byte rules, or fall back for conflicting separators.

Evidence: Compiled strip_leading_bom, split_lines, and their split helpers directly from the reviewed head. The Rust probe returned two rows for both examples. Hadoop LineRecordReader, Spark HadoopLineRecordReader, and Spark 4.1.3 V2 SQL returned three rows: ["", "a", "b"]. A file containing only U+FEFF with that separator returned zero native-helper rows versus one empty Spark row. Ordinary BOM and BOM-only default-separator controls agreed.

[andygrove]

I reproduced this end to end on 3.4, 3.5, 4.0, 4.1 and 4.2. With lineSep set to U+FEFF, BOM a BOM b gives three rows in Spark and two natively, and a file holding only the BOM gives one row against none. Splitting first and then stripping the BOM from the first record would match Hadoop, as long as a file that is exactly the BOM still gives no rows with the default separator. Could you add tests for both cases?

[parthchandra]

Fixed. strip_leading_bom only strips with the default separator, so a custom lineSep equal to the BOM is left alone. Added tests: BOM a BOM b with lineSep=BOM → ["", "a", "b"], and a BOM-only file → no rows (default sep) / one empty row (lineSep=BOM).

[sunchao]

[P2] Preserve per-read Hadoop settings when serializing the text scan. For example, a text read using .option("fs.s3a.endpoint", "http://127.0.0.1:19000") expects that endpoint to be used. Spark's reader and the new eligibility gate incorporate read options, but this call rebuilds the configuration from SparkConf and discards them. extractObjectStoreOptions therefore sends the global/default endpoint to native execution, so enabling native text can break an otherwise working read against an S3-compatible service. Build this configuration from textScan.options.asCaseSensitiveMap.asScala.toMap, matching Spark's TextScan.createReaderFactory.

Evidence: A bounded Spark 4.1.3 JVM probe executed both configuration constructions with the endpoint supplied only in read options. Spark's options-aware configuration returned http://127.0.0.1:19000; the exact construction used at line 112 returned null. NativeConfig.extractObjectStoreOptions copies fs.s3a.* exclusively from that supplied configuration, and the native text planner passes the resulting map to object-store creation. No remote service or credentials were used in this reproduction.

[andygrove]

Agreed. The gate already builds its conf from the read options, so the serde should use the same one. CometCsvNativeScanExec has the same line. Could you fix both here, since this PR already touches the CSV node?

[parthchandra]

Fixed both nodes. Each builds its Hadoop conf from the scan's own options with newHadoopConfWithOptions(...asCaseSensitiveMap...), so settings like fs.s3a.endpoint reach native.

[andygrove]

Thanks for the quick turnaround, Parth. I reran my probes on 23bba3e against Spark 3.4, 3.5, 4.0, 4.1 and 4.2, and the suite passes on all five. Both cases I blocked on are fixed. The compression check is now version aware, and line.maxlength falls back from every place it can be set. The size ceiling also bounds the memory, which was my other main concern.

Three things still return different results from Spark, or fail where Spark succeeds, so I'm keeping this at request changes. A plain s3a:// read across two buckets still reads every file from the first bucket, because the new check only covers alias schemes. sunchao's two findings also hold up. I reproduced the U+FEFF lineSep case end to end on all five profiles, and the serde drops per-read Hadoop options such as fs.s3a.endpoint. The details are in the threads.

The branch now conflicts with main in understanding-comet-plans.md, where #5634 changed the CometInMemoryTableScan row. While it conflicts, GitHub can't build the merge ref, so Comet CI hasn't run on this head and the run-all-spark-profiles label hasn't produced anything. Merging main with the next push will sort that out.

[andygrove]

I reproduced this end to end on 3.4, 3.5, 4.0, 4.1 and 4.2. With lineSep set to U+FEFF, BOM a BOM b gives three rows in Spark and two natively, and a file holding only the BOM gives one row against none. Splitting first and then stripping the BOM from the first record would match Hadoop, as long as a file that is exactly the BOM still gives no rows with the default separator. Could you add tests for both cases?

[andygrove]

Agreed. The gate already builds its conf from the read options, so the serde should use the same one. CometCsvNativeScanExec has the same line. Could you fix both here, since this PR already touches the CSV node?

[andygrove]

The link text says Strings with non-UTF-8 bytes, but it points at compatibility/scans.md. That section is in compatibility/index.md, so could it link to compatibility/index.md#strings-with-non-utf-8-bytes?

[parthchandra]

Fixed. The link now points to compatibility/index.md#strings-with-non-utf-8-bytes.

[sunchao]

Reviewed commit 8dac218a9def3b4dcf0d6bbd3fd9ed954215b324. New P1/P2 findings are detailed inline.

[sunchao]

[P2] Update this assertion to the shared V2 fallback reason. transformV2Scan handles input_file_name before reaching the text-specific branch, so the actual reason starts with Native V2 scan is not compatible with input_file_name. The expected Native Text scan... string cannot match. This new test fails on all five Spark profiles and blocks Required Checks, although the fallback itself works. Assert the shared V2 message while retaining the answer and fallback checks.

Evidence: Exact-head run 38305916651 reports this as the sole failed test in every Spark 3.4, 3.5, 4.0, 4.1, and 4.2 scan job. The Spark 4.1 job, https://github.com/apache/datafusion-comet/actions/runs/38305916651/job/115002925730, reports 737 successful tests and one failure: expected Native Text scan is not compatible with input_file_name, actual Native V2 scan is not compatible with input_file_name, input_file_block_start, or input_file_block_length.

[parthchandra]

Fixed. Dropped the dead Text-specific guard; the test now asserts the shared V2 message, which the blanket gate at the top of transformV2Scan emits.

[mbutrovich]

If we think the common use case here is broadcasted tables (i.e., small input) then why not start with adding a config that enables SparkToColumnar for this data source, a la COMET_CONVERT_FROM_PARQUET_ENABLED? That avoids maintaining what looks to be complicated logic like what #6784 (comment) alludes to.

edit: I guess that's similar to what #6786 is?

[parthchandra]

If we think the common use case here is broadcasted tables (i.e., small input) then why not start with adding a config that enables SparkToColumnar for this data source, a la COMET_CONVERT_FROM_PARQUET_ENABLED? That avoids maintaining what looks to be complicated logic like what #6784 (comment) alludes to.

edit: I guess that's similar to what #6786 is?

Exactly. #6786 is essentially what you are suggesting. It is in fact more generic. Any unsupported feature (not just a txt file) that can cause the build side to fall back can be covered by inserting a SparkToColumnar before the join. #6786 tries to be conservative though ( to avoid breaking DPP)

[andygrove]

Heads up: #7090 merged the Parquet and Iceberg user docs into docs/source/user-guide/latest/parquet.md and iceberg.md, and deleted compatibility/scans.md, which this PR edits. Could you re-apply that change to parquet.md when you next rebase?

[parthchandra]

@sunchao, @andygrove any more comments?

[sunchao]

Reviewed commit 7e57fffa279f43685cf91c176f02ea869996d6e1. New P1/P2 findings are detailed inline.

[sunchao]

[P2] Restore the wrapped Spark scan when text conversion is declined. With native V2 text enabled, reading small uncompressed files from two ABFS containers passes the scan gate, but extractScanObjectStoreOptions rejects the layout and the text serializer returns None. This branch then retains CometBatchScanExec, which advertises columnar output although Spark’s text reader produces UnsafeRow. The query fails with ClassCastException instead of falling back and returning both files. Use getOrElse(scan.wrapped), as the adjacent CSV branch does, and cover a serializer-decline case.

Evidence: A bounded Spark 4.1.3 probe used the repository’s FakeAbfsFileSystem to read local files through two abfss container authorities. Spark returned alpha and beta. The unchanged exact-head extractScanObjectStoreOptions method returned Left for the two containers. Executing the exact-head CometBatchScanExec around that scan reproduced UnsafeRow cannot be cast to ColumnarBatch; executing its wrapped BatchScanExec returned alpha and beta. Source tracing connects that decline to this getOrElse branch. This was a component reproduction without Azure access or a full integrated Comet build.

[parthchandra]

Fixed. A declined Text conversion now returns scan.wrapped, like the CSV branch, so the query falls back to Spark's row-producing scan instead of failing with the ClassCastException. Added Text tests for the two ABFS layouts (different accounts, different containers). (CometExecRule.scala:462; AbfsScanFallbackSuite.scala)

[andygrove]

@sunchao, @andygrove any more comments?

I didn't see responses to all the threads from my previous review yet. Were they also addressed?

[andygrove]

Thanks for the quick work on the earlier rounds, Parth. I went back through every thread against 6f21c0dd39, and the compression shim, line.maxlength, the size ceiling, the per-read Hadoop conf, the ABFS decline and the new tests all hold up. CI is green on this head across all five Spark profiles. I also fuzzed the self-overlapping lineSep fallback against Hadoop's LineReader with 200k random inputs per separator and buffer sizes from 1 to 8 bytes. Separators without self-overlap matched an exact scan every time, and aab mismatched about 8,400 times, so the fallback covers the right class.

I'm keeping this at request changes for one case where the native scan returns different rows from Spark without an error.

Spark's HadoopLineRecordReader.skipUtfByteOrderMark strips a leading BOM from the first record whatever the separator is. The fix for the lineSep=BOM case now skips the strip for every custom separator (strip_leading_bom in text_scan.rs). I ran Hadoop 3.4.2's LineRecordReader on a few files to compare:

• BOM a|b with lineSep="|": Hadoop returns a, b. The native scan returns BOM+a, b.
• A file holding only the BOM with lineSep="|": Hadoop returns no rows. The native scan returns one row containing U+FEFF.
• BOM|x with lineSep="|": Hadoop returns an empty row and x. The native scan returns BOM and x.

These files pass every eligibility check, so counts, filters and joins over them would silently differ. I checked this against Hadoop's reader and Spark 4.1.3's source, not end to end through Comet.

Would it work to split first and then strip the BOM from the first record only, as Hadoop does? If that record is exactly the BOM with no terminator after it, it would produce no row. That also gives the right answer for lineSep=BOM and for the default separator. Could you add SQL tests for BOM a|b and for a BOM-only file, both with lineSep="|"? The existing BOM tests only cover the default separator and lineSep=BOM.

Two small notes on the description. The CSV node now builds its Hadoop conf from the read options and no longer prints the protobuf in stringArgs, which is the right change but isn't mentioned. And the multi-bucket reply mentions a crossLocations set that isn't in the diff. The behavior is still right because splitPartitionsByStore keys on the bucket, so I consider that thread resolved.

Once the BOM handling is fixed and tested I expect to approve.

[parthchandra]

@sunchao, @andygrove any more comments?

I didn't see responses to all the threads from my previous review yet. Were they also addressed?

Yes. I had missed replying on the "collect() only" thread and have answered it now.

[parthchandra]

Thanks for the quick work on the earlier rounds, Parth. I went back through every thread against 6f21c0dd39, and the compression shim, line.maxlength, the size ceiling, the per-read Hadoop conf, the ABFS decline and the new tests all hold up. CI is green on this head across all five Spark profiles. I also fuzzed the self-overlapping lineSep fallback against Hadoop's LineReader with 200k random inputs per separator and buffer sizes from 1 to 8 bytes. Separators without self-overlap matched an exact scan every time, and aab mismatched about 8,400 times, so the fallback covers the right class.

I'm keeping this at request changes for one case where the native scan returns different rows from Spark without an error.

Spark's HadoopLineRecordReader.skipUtfByteOrderMark strips a leading BOM from the first record whatever the separator is. The fix for the lineSep=BOM case now skips the strip for every custom separator (strip_leading_bom in text_scan.rs). I ran Hadoop 3.4.2's LineRecordReader on a few files to compare:

• BOM a|b with lineSep="|": Hadoop returns a, b. The native scan returns BOM+a, b.
• A file holding only the BOM with lineSep="|": Hadoop returns no rows. The native scan returns one row containing U+FEFF.
• BOM|x with lineSep="|": Hadoop returns an empty row and x. The native scan returns BOM and x.

These files pass every eligibility check, so counts, filters and joins over them would silently differ. I checked this against Hadoop's reader and Spark 4.1.3's source, not end to end through Comet.

Would it work to split first and then strip the BOM from the first record only, as Hadoop does? If that record is exactly the BOM with no terminator after it, it would produce no row. That also gives the right answer for lineSep=BOM and for the default separator. Could you add SQL tests for BOM a|b and for a BOM-only file, both with lineSep="|"? The existing BOM tests only cover the default separator and lineSep=BOM.

Two small notes on the description. The CSV node now builds its Hadoop conf from the read options and no longer prints the protobuf in stringArgs, which is the right change but isn't mentioned. And the multi-bucket reply mentions a crossLocations set that isn't in the diff. The behavior is still right because splitPartitionsByStore keys on the bucket, so I consider that thread resolved.

Once the BOM handling is fixed and tested I expect to approve.

Fixed. The reader now splits first and strips the BOM from the first record only, like Hadoop's skipUtfByteOrderMark. A BOM-only file gives no rows, BOM|x gives "" and x, and lineSep=BOM is unchanged. Added Rust and Scala tests for BOM a|b, BOM|x and a BOM-only file with lineSep="|".

Also, updated the description.

[sunchao]

Reviewed commit b04673de2177499c1a61c8d3ed130acba25ded22. New P1/P2 findings are detailed inline.

[sunchao]

[P2] Preserve Hadoop option casing in the eligibility configuration. With native V2 text enabled, set the Hadoop configuration key mapreduce.input.linerecordreader.line.maxlength to 5, then read ab\nabcdefghij\nxy with .option("mapreduce.input.linerecordreader.line.maxLength", "2147483647"). Spark preserves Hadoop key casing, so that differently cased option does not override the limit and Spark returns ab, xy. Here, scan.options.asScala lowercases the option name, making the gate see an unlimited line length and admit native execution. The native reader returns all three rows, silently changing counts, filters, and joins. Use scan.options.asCaseSensitiveMap.asScala.toMap, matching TextScan.createReaderFactory and the new serializer.

Evidence: A bounded Spark 4.1.3 V2 SQL probe reproduced Spark returning [[ab], [xy]]. Executing the two configuration constructions reported 2147483647 for this gate's lowercased map and 5 for Spark's case-preserving map. The unchanged splitting/BOM helpers compiled directly from this head returned [ab, abcdefghij, xy]. Source tracing confirms that the gate therefore skips its finite-line-length fallback and that native execution does not implement that limit. Spark's TextScan preserves option casing in all five inspected supported versions. This is component reproduction and source tracing, not a full integrated Comet run.

[parthchandra]

Fixed. The eligibility check now builds its Hadoop conf from scan.options.asCaseSensitiveMap.asScala.toMap, like TextScan.createReaderFactory and the serializer, so a differently cased option such as ...line.maxLength no longer hides a finite limit set in the Hadoop conf. Added a test with the limit set to 5 in the Hadoop conf and a differently cased option set to 2147483647. It falls back and matches Spark.

[andygrove]

Thanks @parthchandra. Splitting first and then stripping the BOM from the first record matches Hadoop for every file I tried, and the BOM a|b, BOM|x and BOM-only tests pin it.

sunchao's casing finding at CometScanRule.scala:483 holds up when I read it. The eligibility check builds its Hadoop conf from scan.options.asScala, which lowercases the keys, while TextScan.createReaderFactory and your serializer use asCaseSensitiveMap. A differently cased read option can then hide a finite line.maxlength from the check, and the native reader returns lines Spark skips. Could that call use scan.options.asCaseSensitiveMap.asScala.toMap, with a test for sunchao's case? That's the last thing I see, and I'll approve once it's in.

[parthchandra]

Thanks @parthchandra. Splitting first and then stripping the BOM from the first record matches Hadoop for every file I tried, and the BOM a|b, BOM|x and BOM-only tests pin it.

sunchao's casing finding at CometScanRule.scala:483 holds up when I read it. The eligibility check builds its Hadoop conf from scan.options.asScala, which lowercases the keys, while TextScan.createReaderFactory and your serializer use asCaseSensitiveMap. A differently cased read option can then hide a finite line.maxlength from the check, and the native reader returns lines Spark skips. Could that call use scan.options.asCaseSensitiveMap.asScala.toMap, with a test for sunchao's case? That's the last thing I see, and I'll approve once it's in.

Done. The check now builds its Hadoop conf from scan.options.asCaseSensitiveMap.asScala.toMap, like TextScan.createReaderFactory and the serializer. Added a test for sunchao's case: line.maxlength=5 in the Hadoop conf plus a differently cased ...line.maxLength=2147483647 read option now falls back and matches Spark.

[sunchao]

Reviewed commit 462bc44773b61520763adae4b9e6446c08810b3f. New P1/P2 findings are detailed inline.

[sunchao]

[P2] Resolve compression against the selected PartitionedFile.toPath. inputFiles returns URL-encoded strings. Constructing a Hadoop Path from that string can hide a registered codec suffix, admitting a compressed file whose payload is then returned as text.

Evidence: I registered a GzipCodec with extension .packed gz and wrote one\ntwo\n to small.packed gz. Spark 4.1.3 returned one and two. The gate checked .packed%20gz, and the executed native scan returned one row decoded from the gzip payload.

[parthchandra]

Fixed. The compression check now uses each selected file's decoded path (new Path(file.pathUri)) instead of the URL-encoded strings from inputFiles. A name like small.packed gz keeps its codec suffix, and the scan falls back. Added a test with a codec registered for .packed gz.

[sunchao]

[P2] Validate the selected file URIs before committing to native text. A directory root can pass this check while a child has a legal Hadoop filename that object_store rejects. The query then fails during native planning. Apply the backend’s path check to the listed files and fall back when one fails.

Evidence: I placed expected\n in a child whose name contained a literal tab and passed its parent directory to spark.read.text. Spark 4.1.3 returned expected. Enabling native text raised CometNativeException with Encountered illegal character sequence. A directly supplied filename containing a newline triggered the existing root fallback and read successfully.

[parthchandra]

Fixed. The scheme and object_store path checks now run on the selected files instead of only the root paths, so a child file with a tab in its name falls back instead of failing during native planning. Added a test with a file named a<TAB>b.txt inside a directory.

[sunchao]

[P2] Read line.maxlength only in line mode. Spark’s whole-file reader never consults this setting. The unconditional getInt can therefore fail an otherwise working wholetext=true query during planning. Move it under the !textOptions.wholeText guard.

Evidence: With Spark 4.1.3, wholetext=true, and mapreduce.input.linerecordreader.line.maxlength=64k, Spark returned the single value alpha\nbeta. Toggling only the native-text flag raised NumberFormatException. Both readers rejected 64k in line mode. With the valid integer 5, the whole-file query succeeded but unnecessarily fell back.

[parthchandra]

Fixed. line.maxlength is now read only in line mode, so wholetext queries no longer fail or fall back because of it. If the value is not an integer, the scan falls back so Spark reports it. Added a test for wholetext with the limit set to 5 and to 64k.

[sunchao]

[P2] Canonicalize the output attributes and wrapped BatchScanExec, and clear block serialization state in the canonical copy. Returning the original expression IDs makes equivalent text lookups compare unequal, so Spark repeats their scans and exchanges.

Evidence: With exchange reuse enabled and AQE disabled on Spark 4.1.3, joining one five-row text lookup twice executed two CometBroadcastExchange nodes, each emitting five rows. Spark and the native Parquet control each reused one broadcast. The native leaves reported sameResult=false while their wrapped scans reported true. A grouped UNION ALL likewise wrote 24 shuffle records in two map stages versus 12 in one.

[parthchandra]

Fixed. The canonical copy now normalizes the output attributes, uses the wrapped scan's canonical form, and clears the serialization state, so repeated lookups compare equal and Spark can reuse their exchanges. Added a test that two scans of the same file are sameResult.

[sunchao]

[P2] Check compression on Spark’s selected file partitions. fileIndex.inputFiles includes files eliminated by partition pruning, so a compressed file the query never reads can disable the native lookup. The driver also constructs codec factories for those excluded files. Use scanExec.inputPartitions and reuse the version-appropriate codec resolver.

Evidence: A Spark 4.1.3 query selecting only p='hot' used CometTextNativeScan and returned alpha. Changing only the cold partition’s file to .gz switched the same query to Spark’s BatchScan. In a separate fixture with one selected and 256 pruned plain files, the driver constructed the counting codec 258 times with native text enabled and once with it disabled.

[parthchandra]

Fixed along with the compression comment above. The check now uses inputPartitions, so a compressed file in a pruned partition no longer disables the native scan, and the driver no longer builds codec factories for files the query never reads. Added a test with a .gz file in a partition the filter excludes.

[parthchandra]

@andygrove @sunchao I believe all comments are addressed.

[sunchao]

I found two costs that grow with the input even for a count-only text scan. The Spark 3.4 scan job also failed the AQE DPP broadcast-stage-order test. I saw the same failure on #6107 and could not attribute it to the text reader. CI log

[sunchao]

A file consisting of newlines allocates one 16-byte slice descriptor per input byte before the first batch is returned. An isolated 8 MiB input used 128 MiB for the slice table alone. At the default 64 MiB file cap the table reaches 1 GiB, including for count(*). Could the reader split lines incrementally and retain only what the current output batch needs?

[sunchao]

On Spark 3.4, 3.5 and 4.0, the shim constructs a Hadoop CompressionCodecFactory for each path. For a directory of uncompressed files, this loop rebuilds the same codec registry for every file on the driver. Could this plan check create the factory once and reuse it across the selected paths?