sunchao commented on code in PR #6784:
URL: https://github.com/apache/datafusion-comet/pull/6784#discussion_r4353456015
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -477,6 +419,152 @@ case class CometScanRule(session: SparkSession)
withFallbackReasons(scanExec, fallbackReasons.toSet)
}
+ case scan: TextScan if COMET_TEXT_V2_NATIVE_ENABLED.get() =>
+ if (scanExec.output.exists(_.isMetadataCol)) {
+ return withFallbackReason(
+ scanExec,
+ "Metadata columns are not supported for Text V2 scans")
+ }
+ // A native scan is useless without native execution: with exec
disabled, CometExecRule
+ // leaves this CometBatchScanExec unconverted and it fails at runtime.
Require native exec.
+ if (!COMET_EXEC_ENABLED.get()) {
+ return withFallbackReason(
+ scanExec,
+ s"Native Text scan requires ${COMET_EXEC_ENABLED.key} to be
enabled")
+ }
+ // input_file_name/_block_start/_block_length are already handled by
the blanket V2 gate
+ // near the top of transformV2Scan, so no Text-specific check is
needed here.
+ val fallbackReasons = new ListBuffer[String]()
+ val schemaSupported =
+ CometBatchScanExec.isSchemaSupported(scan.readDataSchema,
fallbackReasons)
+ if (!schemaSupported) {
+ fallbackReasons += s"Schema ${scan.readDataSchema} is not supported"
+ }
+ // Spark's text source allows only a single data column
(TextScan.verifyReadSchema). A
+ // user-forced multi-column read schema is invalid; fall back so Spark
raises its structured
+ // AnalysisException rather than letting the native reader fail with
an opaque error.
+ if (scan.readDataSchema.size > 1) {
+ fallbackReasons += "Text data source supports only a single column"
+ }
+ // Partition columns are not implemented for native Text scans (unlike
the native CSV path,
+ // which forwards them). Fall back to Spark for partitioned reads
rather than drop them.
+ if (scan.readPartitionSchema.nonEmpty) {
+ fallbackReasons += "Comet does not support partition columns in
native Text scans"
+ }
+ // The native reader loads each file whole. Spark splits a large
uncompressed text file
+ // into byte-range partitions (TextScan.isSplitable), and each split
would then re-read and
+ // buffer the whole file with no memory-pool reservation. Restrict
native execution to
+ // unsplit files (the small build-side lookups this targets); Spark
handles large files.
+ val hasSplitFile = scanExec.inputPartitions.exists {
+ case fp: FilePartition => fp.files.exists(_.start != 0)
+ case _ => false
+ }
+ if (hasSplitFile) {
+ fallbackReasons += "Comet native Text scan does not support files
split into byte " +
+ "ranges (large files)"
+ }
+ // The native reader loads each file whole and does not reserve the
buffer against the
+ // memory pool, so bound the per-file size; larger files fall back to
Spark. This keeps the
+ // feature to the small lookup tables it targets regardless of
maxPartitionBytes.
+ val maxTextFileSize = COMET_SCAN_TEXT_MAX_FILE_SIZE.get()
+ val hasOversizedFile = scanExec.inputPartitions.exists {
+ case fp: FilePartition => fp.files.exists(_.length > maxTextFileSize)
+ case _ => false
+ }
+ if (hasOversizedFile) {
+ fallbackReasons += "Comet native Text scan does not read files
larger than " +
+ s"${COMET_SCAN_TEXT_MAX_FILE_SIZE.key} ($maxTextFileSize bytes)"
+ }
+ // Comet's native plan does not carry a compression codec, so the
native reader would feed
+ // raw compressed bytes to the line splitter. Spark auto-detects a
codec by file extension.
+ // Use the options-aware Hadoop conf (so a per-read codec option is
honored) and the
+ // version-aware ShimFileFormat.isCompressedFile, which on Spark 4.1+
also catches the
+ // non-standard .gzip/.zstd extensions that CompressionCodecFactory
alone misses.
+ val hadoopConf =
session.sessionState.newHadoopConfWithOptions(scan.options.asScala.toMap)
Review Comment:
[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
