parthchandra commented on code in PR #6784:
URL: https://github.com/apache/datafusion-comet/pull/6784#discussion_r4426908582
##########
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:
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.
--
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]
