parthchandra commented on code in PR #6784:
URL: https://github.com/apache/datafusion-comet/pull/6784#discussion_r4379955988
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -477,6 +419,155 @@ 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, built from the case-preserving
option map like
+ // Spark's TextScan.createReaderFactory (Hadoop keys are case
sensitive, but the scan's
+ // option map lowercases them), 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.asCaseSensitiveMap.asScala.toMap)
+ val hasCompressedFile = scan.fileIndex.inputFiles.exists { path =>
+ ShimFileFormat.isCompressedFile(hadoopConf, new
org.apache.hadoop.fs.Path(path))
+ }
+ if (hasCompressedFile) {
+ fallbackReasons += "Comet does not support compressed text files"
+ }
+ // Spark skips unreadable/missing files when these are set; the native
reader's DataFusion
+ // FileStream fails the whole scan instead (OnError::Fail). Fall back
to match Spark.
+ if (SQLConf.get.ignoreCorruptFiles || scan.options.getBoolean(
+ "ignoreCorruptFiles",
+ false)) {
+ fallbackReasons += "Comet native Text scan does not support
ignoreCorruptFiles"
+ }
+ if (SQLConf.get.ignoreMissingFiles || scan.options.getBoolean(
+ "ignoreMissingFiles",
+ false)) {
+ fallbackReasons += "Comet native Text scan does not support
ignoreMissingFiles"
+ }
+ // Spark's line reader skips any line at or over
+ // mapreduce.input.linerecordreader.line.maxlength (default
unlimited); the key can come
+ // from spark.hadoop.*, core-site.xml, or a read option. The native
reader does not
+ // truncate, so fall back when a finite limit is set.
+ if
(hadoopConf.getInt("mapreduce.input.linerecordreader.line.maxlength",
Int.MaxValue)
+ != Int.MaxValue) {
+ fallbackReasons += "Comet native Text scan does not support " +
+ "mapreduce.input.linerecordreader.line.maxlength"
+ }
+ // A wholetext file is never split (TextScan.isSplitable is false for
wholeText), so the
+ // split-file guard above does not bound it. Reading a very large file
whole is unbounded
+ // memory, and the whole file becomes one Arrow string value. Because
a lossy UTF-8 decode
+ // can expand ill-formed bytes up to 3x, cap the raw file size at
Int.MaxValue/3 so the
+ // decoded value stays under Arrow's 2GB 32-bit string-offset limit;
fall back above that.
+ // (The native reader also fails cleanly rather than panicking if a
value still overflows.)
+ val textOptions =
+ new org.apache.spark.sql.execution.datasources.text.TextOptions(
+ scan.options.asScala.toMap)
+ // Spark reads a custom line separator with Hadoop's
LineReader.readCustomLine, which on a
+ // mismatch backtracks only one byte instead of following the
separator's own prefix
+ // overlap. For a separator that overlaps itself (e.g. "aab", whose
"aa" prefix can restart
+ // a match) that reader can split differently from the native reader's
exact substring
+ // scan, and the difference depends on where the separator falls
across Hadoop's input
+ // buffer boundaries. wholetext ignores the separator, so only guard
line mode. Fall back
+ // so the two always agree.
+ if (!textOptions.wholeText) {
+ textOptions.lineSeparatorInRead.foreach { sep =>
+ if (CometScanRule.separatorSelfOverlaps(sep)) {
+ fallbackReasons += "Comet native Text scan does not support a
self-overlapping " +
+ "line separator (Spark's line reader can split it differently)"
+ }
+ }
+ }
+ if (textOptions.wholeText) {
+ val maxWholeTextBytes =
+ math.min(session.sessionState.conf.filesMaxPartitionBytes,
Int.MaxValue.toLong / 3)
+ val hasLargeWholeTextFile = scanExec.inputPartitions.exists {
+ case fp: FilePartition => fp.files.exists(_.length >
maxWholeTextBytes)
+ case _ => false
+ }
+ if (hasLargeWholeTextFile) {
+ fallbackReasons += "Comet native Text scan does not support large
wholetext files " +
+ s"(over ${SQLConf.FILES_MAX_PARTITION_BYTES.key}, capped below
2GB)"
+ }
+ }
+ // Comet's native reader opens files through object_store, which only
understands a fixed
+ // set of URL schemes. A text file on a custom Hadoop scheme
(viewfs://, oss://, ...) or a
+ // path object_store rejects would be claimed here then hard-fail at
execution, whereas
+ // Spark reads it via the Hadoop FS API. Use the same shared gate as
the native Parquet
+ // path. A scan that spans several object stores is split per store in
the serde.
+ CometScanRule
+ .rootPathFallbackReason(
+ "Native Text scan",
+ scan.fileIndex.rootPaths.map(_.toUri),
Review Comment:
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.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -477,6 +419,155 @@ 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, built from the case-preserving
option map like
+ // Spark's TextScan.createReaderFactory (Hadoop keys are case
sensitive, but the scan's
+ // option map lowercases them), 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.asCaseSensitiveMap.asScala.toMap)
+ val hasCompressedFile = scan.fileIndex.inputFiles.exists { path =>
+ ShimFileFormat.isCompressedFile(hadoopConf, new
org.apache.hadoop.fs.Path(path))
+ }
+ if (hasCompressedFile) {
+ fallbackReasons += "Comet does not support compressed text files"
+ }
+ // Spark skips unreadable/missing files when these are set; the native
reader's DataFusion
+ // FileStream fails the whole scan instead (OnError::Fail). Fall back
to match Spark.
+ if (SQLConf.get.ignoreCorruptFiles || scan.options.getBoolean(
+ "ignoreCorruptFiles",
+ false)) {
+ fallbackReasons += "Comet native Text scan does not support
ignoreCorruptFiles"
+ }
+ if (SQLConf.get.ignoreMissingFiles || scan.options.getBoolean(
+ "ignoreMissingFiles",
+ false)) {
+ fallbackReasons += "Comet native Text scan does not support
ignoreMissingFiles"
+ }
+ // Spark's line reader skips any line at or over
+ // mapreduce.input.linerecordreader.line.maxlength (default
unlimited); the key can come
+ // from spark.hadoop.*, core-site.xml, or a read option. The native
reader does not
+ // truncate, so fall back when a finite limit is set.
+ if
(hadoopConf.getInt("mapreduce.input.linerecordreader.line.maxlength",
Int.MaxValue)
+ != Int.MaxValue) {
+ fallbackReasons += "Comet native Text scan does not support " +
+ "mapreduce.input.linerecordreader.line.maxlength"
+ }
+ // A wholetext file is never split (TextScan.isSplitable is false for
wholeText), so the
+ // split-file guard above does not bound it. Reading a very large file
whole is unbounded
+ // memory, and the whole file becomes one Arrow string value. Because
a lossy UTF-8 decode
+ // can expand ill-formed bytes up to 3x, cap the raw file size at
Int.MaxValue/3 so the
+ // decoded value stays under Arrow's 2GB 32-bit string-offset limit;
fall back above that.
+ // (The native reader also fails cleanly rather than panicking if a
value still overflows.)
+ val textOptions =
+ new org.apache.spark.sql.execution.datasources.text.TextOptions(
+ scan.options.asScala.toMap)
+ // Spark reads a custom line separator with Hadoop's
LineReader.readCustomLine, which on a
+ // mismatch backtracks only one byte instead of following the
separator's own prefix
+ // overlap. For a separator that overlaps itself (e.g. "aab", whose
"aa" prefix can restart
+ // a match) that reader can split differently from the native reader's
exact substring
+ // scan, and the difference depends on where the separator falls
across Hadoop's input
+ // buffer boundaries. wholetext ignores the separator, so only guard
line mode. Fall back
+ // so the two always agree.
+ if (!textOptions.wholeText) {
+ textOptions.lineSeparatorInRead.foreach { sep =>
+ if (CometScanRule.separatorSelfOverlaps(sep)) {
+ fallbackReasons += "Comet native Text scan does not support a
self-overlapping " +
+ "line separator (Spark's line reader can split it differently)"
+ }
+ }
+ }
+ if (textOptions.wholeText) {
+ val maxWholeTextBytes =
+ math.min(session.sessionState.conf.filesMaxPartitionBytes,
Int.MaxValue.toLong / 3)
+ val hasLargeWholeTextFile = scanExec.inputPartitions.exists {
+ case fp: FilePartition => fp.files.exists(_.length >
maxWholeTextBytes)
+ case _ => false
+ }
+ if (hasLargeWholeTextFile) {
+ fallbackReasons += "Comet native Text scan does not support large
wholetext files " +
+ s"(over ${SQLConf.FILES_MAX_PARTITION_BYTES.key}, capped below
2GB)"
+ }
+ }
+ // Comet's native reader opens files through object_store, which only
understands a fixed
+ // set of URL schemes. A text file on a custom Hadoop scheme
(viewfs://, oss://, ...) or a
+ // path object_store rejects would be claimed here then hard-fail at
execution, whereas
+ // Spark reads it via the Hadoop FS API. Use the same shared gate as
the native Parquet
+ // path. A scan that spans several object stores is split per store in
the serde.
+ CometScanRule
+ .rootPathFallbackReason(
+ "Native Text scan",
+ scan.fileIndex.rootPaths.map(_.toUri),
+
NativeConfig.resolveLibhdfsSchemes(session.sessionState.newHadoopConf()),
+ hadoopConf,
+ isBucketedScan = false)
+ .foreach(fallbackReasons += _)
+ if (fallbackReasons.isEmpty) {
Review Comment:
Fixed. In line mode the scan now falls back when any
mapreduce.job.input.file.must.* key is set, because the native reader cannot
enforce those options. Added a test that checks the plan and does not run the
query.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -477,6 +419,155 @@ 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, built from the case-preserving
option map like
+ // Spark's TextScan.createReaderFactory (Hadoop keys are case
sensitive, but the scan's
+ // option map lowercases them), 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.asCaseSensitiveMap.asScala.toMap)
+ val hasCompressedFile = scan.fileIndex.inputFiles.exists { path =>
+ ShimFileFormat.isCompressedFile(hadoopConf, new
org.apache.hadoop.fs.Path(path))
+ }
+ if (hasCompressedFile) {
+ fallbackReasons += "Comet does not support compressed text files"
+ }
+ // Spark skips unreadable/missing files when these are set; the native
reader's DataFusion
+ // FileStream fails the whole scan instead (OnError::Fail). Fall back
to match Spark.
+ if (SQLConf.get.ignoreCorruptFiles || scan.options.getBoolean(
+ "ignoreCorruptFiles",
+ false)) {
+ fallbackReasons += "Comet native Text scan does not support
ignoreCorruptFiles"
+ }
+ if (SQLConf.get.ignoreMissingFiles || scan.options.getBoolean(
+ "ignoreMissingFiles",
+ false)) {
+ fallbackReasons += "Comet native Text scan does not support
ignoreMissingFiles"
+ }
+ // Spark's line reader skips any line at or over
+ // mapreduce.input.linerecordreader.line.maxlength (default
unlimited); the key can come
+ // from spark.hadoop.*, core-site.xml, or a read option. The native
reader does not
+ // truncate, so fall back when a finite limit is set.
+ if
(hadoopConf.getInt("mapreduce.input.linerecordreader.line.maxlength",
Int.MaxValue)
Review Comment:
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.
##########
native/core/src/execution/operators/text_scan.rs:
##########
@@ -0,0 +1,682 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+//! Native Text file scan, mirroring Spark's `text` data source.
+//!
+//! Spark's text reader produces a single `value: string` column. With the
default options each
+//! line of the file is one row (split on `\n`, `\r\n`, or `\r`, with the
terminator stripped).
+//! With `wholetext=true` the whole file is a single row. A custom `lineSep`
splits on exactly that
+//! byte sequence instead of the universal-newline set. See the
`TextOptions`/`HadoopFileLinesReader`
+//! logic in Spark's `org.apache.spark.sql.execution.datasources.text` package.
+
+use crate::execution::operators::ExecutionError;
+use arrow::array::StringBuilder;
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::{RecordBatch, RecordBatchOptions};
+use datafusion::common::Result;
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::physical_expr::projection::ProjectionExprs;
+use datafusion::physical_plan::metrics::{BaselineMetrics,
ExecutionPlanMetricsSet};
+use datafusion::physical_plan::DisplayFormatType;
+use datafusion_comet_common::decode_utf8_spark_lossy;
+use datafusion_comet_proto::spark_operator::TextOptions;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_compression_type::FileCompressionType;
+use datafusion_datasource::file_groups::FileGroup;
+use datafusion_datasource::file_scan_config::{FileScanConfig,
FileScanConfigBuilder};
+use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
+use datafusion_datasource::projection::{ProjectionOpener, SplitProjection};
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_datasource::{as_file_source, PartitionedFile, TableSchema};
+use futures::StreamExt;
+use object_store::{ObjectStore, ObjectStoreExt};
+use std::borrow::Cow;
+use std::fmt;
+use std::io::Read;
+use std::sync::Arc;
+
+/// Batch size used when the execution framework does not set one before
opening the file.
+const DEFAULT_TEXT_BATCH_SIZE: usize = 8192;
+
+/// UTF-8 byte-order mark. Spark's line reader strips a leading one; wholetext
keeps it.
+const UTF8_BOM: &[u8] = &[0xEF, 0xBB, 0xBF];
+
+/// Cap on a batch's decoded `value` buffer bytes, kept well below Arrow's 2GB
32-bit offset limit
+/// so a batch of many rows (or a few long lines) cannot overflow. Measured
against the decoded
+/// string length, since a lossy UTF-8 decode can expand ill-formed bytes up
to 3x.
+const MAX_BATCH_VALUE_BYTES: usize = 1 << 30;
+
+pub fn init_text_datasource_exec(
+ object_store_url: ObjectStoreUrl,
+ file_groups: Vec<Vec<PartitionedFile>>,
+ data_schema: SchemaRef,
+ _partition_schema: SchemaRef,
+ projection_vector: Vec<usize>,
+ text_options: &TextOptions,
+) -> Result<Arc<DataSourceExec>, ExecutionError> {
+ let text_source = TextSource::new(data_schema, text_options);
+
+ let file_groups = file_groups.into_iter().map(FileGroup::new).collect();
+
+ let file_scan_config = FileScanConfigBuilder::new(object_store_url,
Arc::new(text_source))
+ .with_file_groups(file_groups)
+ .with_projection_indices(Some(projection_vector))?
+ .build();
+
+ Ok(DataSourceExec::from_data_source(file_scan_config))
+}
+
+/// A [`FileSource`] for Spark's text format. The file schema is always the
single `value` column;
+/// projection (and partition columns, if any) is handled by
[`ProjectionOpener`], exactly as the
+/// CSV source does.
+#[derive(Debug, Clone)]
+struct TextSource {
+ table_schema: TableSchema,
+ projection: SplitProjection,
+ batch_size: Option<usize>,
+ metrics: ExecutionPlanMetricsSet,
+ whole_text: bool,
+ // The raw separator bytes when a custom `lineSep` is set; None means
universal newline
+ // (`\n`, `\r\n`, `\r`).
+ line_sep: Option<Vec<u8>>,
+}
+
+impl TextSource {
+ fn new(file_schema: SchemaRef, options: &TextOptions) -> Self {
+ let table_schema: TableSchema = file_schema.into();
+ // `line_sep` arrives as the exact separator bytes Spark computed
(`lineSeparatorInRead`,
+ // already encoded with the file's charset), so we split on them
directly.
+ let line_sep = options.line_sep.clone().filter(|sep| !sep.is_empty());
+ Self {
+ projection: SplitProjection::unprojected(&table_schema),
+ table_schema,
+ batch_size: None,
+ metrics: ExecutionPlanMetricsSet::new(),
+ whole_text: options.whole_text,
+ line_sep,
+ }
+ }
+}
+
+impl From<TextSource> for Arc<dyn FileSource> {
+ fn from(source: TextSource) -> Self {
+ as_file_source(source)
+ }
+}
+
+impl FileSource for TextSource {
+ fn create_file_opener(
+ &self,
+ object_store: Arc<dyn ObjectStore>,
+ base_config: &FileScanConfig,
+ partition_index: usize,
+ ) -> Result<Arc<dyn FileOpener>> {
+ // The inner opener must emit batches whose schema is the file schema
projected by
+ // `file_indices` (that is what ProjectionOpener feeds its projector).
For text that is
+ // either the single `value` column or, for a projection-less count,
an empty schema.
+ let file_schema = self.table_schema.file_schema();
+ let projected_file_schema =
Arc::new(file_schema.project(&self.projection.file_indices)?);
+ let opener = Arc::new(TextOpener {
+ projected_file_schema,
+ file_compression_type: base_config.file_compression_type,
+ object_store,
+ partition_index,
+ batch_size: self.batch_size.unwrap_or(DEFAULT_TEXT_BATCH_SIZE),
+ metrics: self.metrics.clone(),
+ whole_text: self.whole_text,
+ line_sep: self.line_sep.clone(),
+ }) as Arc<dyn FileOpener>;
+ ProjectionOpener::try_new(self.projection.clone(), opener, file_schema)
+ }
+
+ fn table_schema(&self) -> &TableSchema {
+ &self.table_schema
+ }
+
+ fn with_batch_size(&self, batch_size: usize) -> Arc<dyn FileSource> {
+ let mut conf = self.clone();
+ conf.batch_size = Some(batch_size);
+ Arc::new(conf)
+ }
+
+ fn try_pushdown_projection(
+ &self,
+ projection: &ProjectionExprs,
+ ) -> Result<Option<Arc<dyn FileSource>>> {
+ let mut source = self.clone();
+ let new_projection = self.projection.source.try_merge(projection)?;
+ source.projection =
SplitProjection::new(self.table_schema.file_schema(), &new_projection);
+ Ok(Some(Arc::new(source)))
+ }
+
+ fn projection(&self) -> Option<&ProjectionExprs> {
+ Some(&self.projection.source)
+ }
+
+ fn metrics(&self) -> &ExecutionPlanMetricsSet {
+ &self.metrics
+ }
+
+ fn file_type(&self) -> &str {
+ "text"
+ }
+
+ // A custom line separator can straddle a byte-range boundary, and we read
whole files rather
+ // than boundary-aligned ranges, so this source must not let DataFusion
repartition files by
+ // byte range.
+ fn supports_repartitioning(&self) -> bool {
+ false
+ }
+
+ fn fmt_extra(&self, t: DisplayFormatType, f: &mut fmt::Formatter) ->
fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ write!(f, ", whole_text={}", self.whole_text)
+ }
+ DisplayFormatType::TreeRender => Ok(()),
+ }
+ }
+
+ fn apply_expressions(
+ &self,
+ f: &mut dyn FnMut(
+ &Arc<dyn datafusion::physical_plan::PhysicalExpr>,
+ ) -> Result<datafusion::common::tree_node::TreeNodeRecursion>,
+ ) -> Result<datafusion::common::tree_node::TreeNodeRecursion> {
+
datafusion::physical_plan::apply_expression_roots(self.projection.source.iter(),
f)
+ }
+}
+
+/// A [`FileOpener`] that reads a whole text file and yields the projected
file schema in batches.
+struct TextOpener {
+ // The file schema projected by `file_indices`: the single `value` column,
or an empty schema
+ // for a projection-less count.
+ projected_file_schema: SchemaRef,
+ file_compression_type: FileCompressionType,
+ object_store: Arc<dyn ObjectStore>,
+ partition_index: usize,
+ batch_size: usize,
+ metrics: ExecutionPlanMetricsSet,
+ whole_text: bool,
+ line_sep: Option<Vec<u8>>,
+}
+
+impl FileOpener for TextOpener {
+ fn open(&self, partitioned_file: PartitionedFile) ->
Result<FileOpenFuture> {
+ let store = Arc::clone(&self.object_store);
+ let projected_file_schema = Arc::clone(&self.projected_file_schema);
+ let file_compression_type = self.file_compression_type;
+ let batch_size = self.batch_size;
+ let whole_text = self.whole_text;
+ let line_sep = self.line_sep.clone();
+ let baseline_metrics = BaselineMetrics::new(&self.metrics,
self.partition_index);
Review Comment:
Fixed. The opener now records the fetched byte length as bytes_scanned, and
the JVM leaf exposes bytes_scanned next to output_rows, so bytes and records
reach the task input metrics. Added a test that reads four 6-byte files and
checks the task listener sees 24 bytes and 12 records from the native pass
only.
##########
native/core/src/execution/operators/text_scan.rs:
##########
@@ -0,0 +1,682 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+//! Native Text file scan, mirroring Spark's `text` data source.
+//!
+//! Spark's text reader produces a single `value: string` column. With the
default options each
+//! line of the file is one row (split on `\n`, `\r\n`, or `\r`, with the
terminator stripped).
+//! With `wholetext=true` the whole file is a single row. A custom `lineSep`
splits on exactly that
+//! byte sequence instead of the universal-newline set. See the
`TextOptions`/`HadoopFileLinesReader`
+//! logic in Spark's `org.apache.spark.sql.execution.datasources.text` package.
+
+use crate::execution::operators::ExecutionError;
+use arrow::array::StringBuilder;
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::{RecordBatch, RecordBatchOptions};
+use datafusion::common::Result;
+use datafusion::datasource::object_store::ObjectStoreUrl;
+use datafusion::physical_expr::projection::ProjectionExprs;
+use datafusion::physical_plan::metrics::{BaselineMetrics,
ExecutionPlanMetricsSet};
+use datafusion::physical_plan::DisplayFormatType;
+use datafusion_comet_common::decode_utf8_spark_lossy;
+use datafusion_comet_proto::spark_operator::TextOptions;
+use datafusion_datasource::file::FileSource;
+use datafusion_datasource::file_compression_type::FileCompressionType;
+use datafusion_datasource::file_groups::FileGroup;
+use datafusion_datasource::file_scan_config::{FileScanConfig,
FileScanConfigBuilder};
+use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
+use datafusion_datasource::projection::{ProjectionOpener, SplitProjection};
+use datafusion_datasource::source::DataSourceExec;
+use datafusion_datasource::{as_file_source, PartitionedFile, TableSchema};
+use futures::StreamExt;
+use object_store::{ObjectStore, ObjectStoreExt};
+use std::borrow::Cow;
+use std::fmt;
+use std::io::Read;
+use std::sync::Arc;
+
+/// Batch size used when the execution framework does not set one before
opening the file.
+const DEFAULT_TEXT_BATCH_SIZE: usize = 8192;
+
+/// UTF-8 byte-order mark. Spark's line reader strips a leading one; wholetext
keeps it.
+const UTF8_BOM: &[u8] = &[0xEF, 0xBB, 0xBF];
+
+/// Cap on a batch's decoded `value` buffer bytes, kept well below Arrow's 2GB
32-bit offset limit
+/// so a batch of many rows (or a few long lines) cannot overflow. Measured
against the decoded
+/// string length, since a lossy UTF-8 decode can expand ill-formed bytes up
to 3x.
+const MAX_BATCH_VALUE_BYTES: usize = 1 << 30;
+
+pub fn init_text_datasource_exec(
+ object_store_url: ObjectStoreUrl,
+ file_groups: Vec<Vec<PartitionedFile>>,
+ data_schema: SchemaRef,
+ _partition_schema: SchemaRef,
+ projection_vector: Vec<usize>,
+ text_options: &TextOptions,
+) -> Result<Arc<DataSourceExec>, ExecutionError> {
+ let text_source = TextSource::new(data_schema, text_options);
+
+ let file_groups = file_groups.into_iter().map(FileGroup::new).collect();
+
+ let file_scan_config = FileScanConfigBuilder::new(object_store_url,
Arc::new(text_source))
+ .with_file_groups(file_groups)
+ .with_projection_indices(Some(projection_vector))?
+ .build();
+
+ Ok(DataSourceExec::from_data_source(file_scan_config))
+}
+
+/// A [`FileSource`] for Spark's text format. The file schema is always the
single `value` column;
+/// projection (and partition columns, if any) is handled by
[`ProjectionOpener`], exactly as the
+/// CSV source does.
+#[derive(Debug, Clone)]
+struct TextSource {
+ table_schema: TableSchema,
+ projection: SplitProjection,
+ batch_size: Option<usize>,
+ metrics: ExecutionPlanMetricsSet,
+ whole_text: bool,
+ // The raw separator bytes when a custom `lineSep` is set; None means
universal newline
+ // (`\n`, `\r\n`, `\r`).
+ line_sep: Option<Vec<u8>>,
+}
+
+impl TextSource {
+ fn new(file_schema: SchemaRef, options: &TextOptions) -> Self {
+ let table_schema: TableSchema = file_schema.into();
+ // `line_sep` arrives as the exact separator bytes Spark computed
(`lineSeparatorInRead`,
+ // already encoded with the file's charset), so we split on them
directly.
+ let line_sep = options.line_sep.clone().filter(|sep| !sep.is_empty());
+ Self {
+ projection: SplitProjection::unprojected(&table_schema),
+ table_schema,
+ batch_size: None,
+ metrics: ExecutionPlanMetricsSet::new(),
+ whole_text: options.whole_text,
+ line_sep,
+ }
+ }
+}
+
+impl From<TextSource> for Arc<dyn FileSource> {
+ fn from(source: TextSource) -> Self {
+ as_file_source(source)
+ }
+}
+
+impl FileSource for TextSource {
+ fn create_file_opener(
+ &self,
+ object_store: Arc<dyn ObjectStore>,
+ base_config: &FileScanConfig,
+ partition_index: usize,
+ ) -> Result<Arc<dyn FileOpener>> {
+ // The inner opener must emit batches whose schema is the file schema
projected by
+ // `file_indices` (that is what ProjectionOpener feeds its projector).
For text that is
+ // either the single `value` column or, for a projection-less count,
an empty schema.
+ let file_schema = self.table_schema.file_schema();
+ let projected_file_schema =
Arc::new(file_schema.project(&self.projection.file_indices)?);
+ let opener = Arc::new(TextOpener {
+ projected_file_schema,
+ file_compression_type: base_config.file_compression_type,
+ object_store,
+ partition_index,
+ batch_size: self.batch_size.unwrap_or(DEFAULT_TEXT_BATCH_SIZE),
+ metrics: self.metrics.clone(),
+ whole_text: self.whole_text,
+ line_sep: self.line_sep.clone(),
+ }) as Arc<dyn FileOpener>;
+ ProjectionOpener::try_new(self.projection.clone(), opener, file_schema)
+ }
+
+ fn table_schema(&self) -> &TableSchema {
+ &self.table_schema
+ }
+
+ fn with_batch_size(&self, batch_size: usize) -> Arc<dyn FileSource> {
+ let mut conf = self.clone();
+ conf.batch_size = Some(batch_size);
+ Arc::new(conf)
+ }
+
+ fn try_pushdown_projection(
+ &self,
+ projection: &ProjectionExprs,
+ ) -> Result<Option<Arc<dyn FileSource>>> {
+ let mut source = self.clone();
+ let new_projection = self.projection.source.try_merge(projection)?;
+ source.projection =
SplitProjection::new(self.table_schema.file_schema(), &new_projection);
+ Ok(Some(Arc::new(source)))
+ }
+
+ fn projection(&self) -> Option<&ProjectionExprs> {
+ Some(&self.projection.source)
+ }
+
+ fn metrics(&self) -> &ExecutionPlanMetricsSet {
+ &self.metrics
+ }
+
+ fn file_type(&self) -> &str {
+ "text"
+ }
+
+ // A custom line separator can straddle a byte-range boundary, and we read
whole files rather
+ // than boundary-aligned ranges, so this source must not let DataFusion
repartition files by
+ // byte range.
+ fn supports_repartitioning(&self) -> bool {
+ false
+ }
+
+ fn fmt_extra(&self, t: DisplayFormatType, f: &mut fmt::Formatter) ->
fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ write!(f, ", whole_text={}", self.whole_text)
+ }
+ DisplayFormatType::TreeRender => Ok(()),
+ }
+ }
+
+ fn apply_expressions(
+ &self,
+ f: &mut dyn FnMut(
+ &Arc<dyn datafusion::physical_plan::PhysicalExpr>,
+ ) -> Result<datafusion::common::tree_node::TreeNodeRecursion>,
+ ) -> Result<datafusion::common::tree_node::TreeNodeRecursion> {
+
datafusion::physical_plan::apply_expression_roots(self.projection.source.iter(),
f)
+ }
+}
+
+/// A [`FileOpener`] that reads a whole text file and yields the projected
file schema in batches.
+struct TextOpener {
+ // The file schema projected by `file_indices`: the single `value` column,
or an empty schema
+ // for a projection-less count.
+ projected_file_schema: SchemaRef,
+ file_compression_type: FileCompressionType,
+ object_store: Arc<dyn ObjectStore>,
+ partition_index: usize,
+ batch_size: usize,
+ metrics: ExecutionPlanMetricsSet,
+ whole_text: bool,
+ line_sep: Option<Vec<u8>>,
+}
+
+impl FileOpener for TextOpener {
+ fn open(&self, partitioned_file: PartitionedFile) ->
Result<FileOpenFuture> {
+ let store = Arc::clone(&self.object_store);
+ let projected_file_schema = Arc::clone(&self.projected_file_schema);
+ let file_compression_type = self.file_compression_type;
+ let batch_size = self.batch_size;
+ let whole_text = self.whole_text;
+ let line_sep = self.line_sep.clone();
+ let baseline_metrics = BaselineMetrics::new(&self.metrics,
self.partition_index);
+
+ Ok(Box::pin(async move {
+ let range = partitioned_file.range.clone();
+ let location = partitioned_file.object_meta.location;
+ let raw = store.get(&location).await?.bytes().await?;
Review Comment:
Fixed. The scan now falls back for a local file that has a Hadoop .crc file
next to it, since Hadoop verifies that checksum on read and the native reader
does not. This also means local data Spark wrote itself stays on Spark. I noted
that in the data sources guide. Added a test that writes a file through
Hadoop's local filesystem, so the .crc exists, and checks the scan falls back.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -477,6 +419,155 @@ 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, built from the case-preserving
option map like
+ // Spark's TextScan.createReaderFactory (Hadoop keys are case
sensitive, but the scan's
+ // option map lowercases them), 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.asCaseSensitiveMap.asScala.toMap)
+ val hasCompressedFile = scan.fileIndex.inputFiles.exists { path =>
+ ShimFileFormat.isCompressedFile(hadoopConf, new
org.apache.hadoop.fs.Path(path))
Review Comment:
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.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometTextNativeScanExec.scala:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.spark.sql.comet
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
+import org.apache.spark.sql.catalyst.plans.physical.{Partitioning,
UnknownPartitioning}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.text.TextOptions
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.text.TextScan
+
+import com.google.common.base.Objects
+
+import org.apache.comet.{CometConf, ConfigEntry}
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+import org.apache.comet.objectstore.NativeConfig
+import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
+import org.apache.comet.serde.OperatorOuterClass.Operator
+import org.apache.comet.serde.operator.{partition2Proto, schema2Proto}
+
+/*
+ * Native Text scan operator that delegates file reading to datafusion.
Mirrors the native CSV
+ * scan path; produces a single `value: string` column.
+ */
+case class CometTextNativeScanExec(
+ override val nativeOp: Operator,
+ override val output: Seq[Attribute],
+ @transient override val originalPlan: BatchScanExec,
+ override val serializedPlanOpt: SerializedPlan)
+ extends CometLeafExec {
+ override val supportsColumnar: Boolean = true
+
+ override val nodeName: String = "CometTextNativeScan"
+
+ // Show only the semantic output in EXPLAIN / the SQL tab / the event log.
The default would print
+ // the whole `nativeOp` protobuf, which carries every input file path (large
for a lookup
+ // directory of many files). Mirrors CometNativeScanExec.
+ override def stringArgs: Iterator[Any] = Iterator(output)
+
+ // The serialized partitions, which split any Spark partition that mixes
object stores.
+ override def outputPartitioning: Partitioning = UnknownPartitioning(
+ nativeOp.getTextScan.getFilePartitionsCount)
+
+ override def outputOrdering: Seq[SortOrder] = Nil
+
+ override protected def doCanonicalize(): SparkPlan = {
+ CometTextNativeScanExec(nativeOp, output, originalPlan, serializedPlanOpt)
+ }
+
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: CometTextNativeScanExec =>
+ this.output == other.output &&
+ this.serializedPlanOpt == other.serializedPlanOpt &&
+ this.originalPlan == other.originalPlan
+ case _ =>
+ false
+ }
+ }
+
+ override def hashCode(): Int = {
+ Objects.hashCode(output, serializedPlanOpt, originalPlan)
+ }
+}
+
+object CometTextNativeScanExec extends CometOperatorSerde[CometBatchScanExec] {
+
+ override def enabledConfig: Option[ConfigEntry[Boolean]] = Some(
+ CometConf.COMET_TEXT_V2_NATIVE_ENABLED)
+
+ override def convert(
+ op: CometBatchScanExec,
+ builder: Operator.Builder,
+ childOp: Operator*): Option[Operator] = {
+ val textScanBuilder = OperatorOuterClass.TextScan.newBuilder()
+ val textScan = op.wrapped.scan.asInstanceOf[TextScan]
+ val sessionState = op.session.sessionState
+ val options = new TextOptions(textScan.options.asScala.toMap)
+ // Build the Hadoop conf from the scan's read options (case preserved),
matching Spark's
+ // TextScan.createReaderFactory, so per-read settings like fs.s3a.endpoint
reach native.
+ val hadoopConf =
+
sessionState.newHadoopConfWithOptions(textScan.options.asCaseSensitiveMap.asScala.toMap)
+ val s3CompliantSchemes = NativeConfig.resolveS3CompliantSchemes(hadoopConf)
+ val libhdfsSchemes = NativeConfig.resolveLibhdfsSchemes(hadoopConf)
+ val inputPartitions = op.inputPartitions.map(_.asInstanceOf[FilePartition])
+ val multiStoreReason = CometScanUtils.multiStoreFallbackReason(
+ "Native Text scan",
+ inputPartitions.view.flatMap(_.files.view.map(_.pathUri)),
+ s3CompliantSchemes,
+ libhdfsSchemes,
+ isBucketedScan = false)
+ if (multiStoreReason.nonEmpty) {
+ // CometExecRule falls back to the wrapped scan, so tag it too for the
explain output.
+ withFallbackReason(op, multiStoreReason.get)
+ withFallbackReason(op.wrapped, multiStoreReason.get)
+ return None
+ }
+ // The options of every scheme the scan reads, or a decline for the ABFS
layouts one
+ // resolution cannot serve. Checked before the split below so that its
reason is reported.
+ val fileUris = inputPartitions.view.flatMap(_.files).map(_.pathUri)
+ val objectStoreOptions =
+ NativeConfig.extractScanObjectStoreOptions(hadoopConf, fileUris) match {
+ case Right(options) => options
+ case Left(reason) =>
+ withFallbackReason(op, reason)
+ withFallbackReason(op.wrapped, reason)
+ return None
+ }
+ // Native planning reads each partition through one object store.
+ val filePartitions = CometScanUtils.splitPartitionsByStore(
+ inputPartitions,
+ file => NativeConfig.objectStoreKey(file.pathUri, s3CompliantSchemes,
libhdfsSchemes))
+ // Spark 3.4 reports a one-partition scan as SinglePartition and plans no
exchange above it,
+ // so the operators above rely on the scan's partition count.
+ if (filePartitions.length != inputPartitions.length &&
+ !op.wrapped.outputPartitioning.isInstanceOf[UnknownPartitioning]) {
+ val reason = "Native Text scan would split a partition that mixes object
stores, but " +
+ "Spark planned the operators above it for the scan's own partitioning"
+ withFallbackReason(op, reason)
+ withFallbackReason(op.wrapped, reason)
+ return None
+ }
+ val textOptionsProto = textOptions2Proto(options)
+ val dataSchemaProto = schema2Proto(textScan.dataSchema)
+ val readSchemaFieldNames = textScan.readDataSchema.fieldNames
+ val projectionVector = textScan.dataSchema.zipWithIndex
+ .filter { case (field, _) =>
+ readSchemaFieldNames.contains(field.name)
+ }
+ .map(_._2.asInstanceOf[Integer])
+ val partitionSchemaProto = schema2Proto(textScan.readPartitionSchema)
+ val partitionsProto = filePartitions.map(partition2Proto(_,
textScan.readPartitionSchema))
+
+ textScanBuilder.putAllObjectStoreOptions(objectStoreOptions.asJava)
+ textScanBuilder.setTextOptions(textOptionsProto)
+ textScanBuilder.addAllFilePartitions(partitionsProto.asJava)
Review Comment:
Agreed this costs more as file count grows. I'd like to do it as a follow-up
PR. It means splitting TextScan into common and per-partition data with a
plan-data injector, the way the native Parquet scan does it, plus a native
planner change. The native text scan targets small lookup tables, and the
native CSV scan has the same cost today, so I'd rather fix both together.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometTextNativeScanExec.scala:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.spark.sql.comet
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
+import org.apache.spark.sql.catalyst.plans.physical.{Partitioning,
UnknownPartitioning}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.text.TextOptions
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.text.TextScan
+
+import com.google.common.base.Objects
+
+import org.apache.comet.{CometConf, ConfigEntry}
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+import org.apache.comet.objectstore.NativeConfig
+import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
+import org.apache.comet.serde.OperatorOuterClass.Operator
+import org.apache.comet.serde.operator.{partition2Proto, schema2Proto}
+
+/*
+ * Native Text scan operator that delegates file reading to datafusion.
Mirrors the native CSV
+ * scan path; produces a single `value: string` column.
+ */
+case class CometTextNativeScanExec(
+ override val nativeOp: Operator,
+ override val output: Seq[Attribute],
+ @transient override val originalPlan: BatchScanExec,
+ override val serializedPlanOpt: SerializedPlan)
+ extends CometLeafExec {
+ override val supportsColumnar: Boolean = true
+
+ override val nodeName: String = "CometTextNativeScan"
+
+ // Show only the semantic output in EXPLAIN / the SQL tab / the event log.
The default would print
+ // the whole `nativeOp` protobuf, which carries every input file path (large
for a lookup
+ // directory of many files). Mirrors CometNativeScanExec.
+ override def stringArgs: Iterator[Any] = Iterator(output)
+
+ // The serialized partitions, which split any Spark partition that mixes
object stores.
+ override def outputPartitioning: Partitioning = UnknownPartitioning(
+ nativeOp.getTextScan.getFilePartitionsCount)
+
+ override def outputOrdering: Seq[SortOrder] = Nil
+
+ override protected def doCanonicalize(): SparkPlan = {
+ CometTextNativeScanExec(nativeOp, output, originalPlan, serializedPlanOpt)
Review Comment:
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.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -477,6 +419,155 @@ 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, built from the case-preserving
option map like
+ // Spark's TextScan.createReaderFactory (Hadoop keys are case
sensitive, but the scan's
+ // option map lowercases them), 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.asCaseSensitiveMap.asScala.toMap)
+ val hasCompressedFile = scan.fileIndex.inputFiles.exists { path =>
Review Comment:
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.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometCsvNativeScanExec.scala:
##########
@@ -97,8 +97,10 @@ object CometCsvNativeScanExec extends
CometOperatorSerde[CometBatchScanExec] {
val timeZone = sessionState.conf.sessionLocalTimeZone
new CSVOptions(csvScan.options.asScala.toMap, columnPruning, timeZone)
}
+ // Build the Hadoop conf from the scan's read options (case preserved),
matching Spark's
+ // CSVScan.createReaderFactory, so per-read settings like fs.s3a.endpoint
reach native.
val hadoopConf =
-
sessionState.newHadoopConfWithOptions(op.session.sparkContext.conf.getAll.toMap)
+
sessionState.newHadoopConfWithOptions(csvScan.options.asCaseSensitiveMap.asScala.toMap)
Review Comment:
Fixed. The CSV serde now uses `resolveScanLibhdfsSchemes` and declines files
whose scheme is routed through `libhdfs` only by a read option. The
case-preserved options stay. I made `lowerScheme`, `ScanLibhdfsSchemes` and
`resolveScanLibhdfsSchemes` public because this class is in a different
package. Added a CSV test with `fs.comet.libhdfs.schemes=file,hdfs` as a read
option.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometTextNativeScanExec.scala:
##########
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.spark.sql.comet
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
+import org.apache.spark.sql.catalyst.plans.physical.{Partitioning,
UnknownPartitioning}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.FilePartition
+import org.apache.spark.sql.execution.datasources.text.TextOptions
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.text.TextScan
+
+import com.google.common.base.Objects
+
+import org.apache.comet.{CometConf, ConfigEntry}
+import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
+import org.apache.comet.objectstore.NativeConfig
+import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
+import org.apache.comet.serde.OperatorOuterClass.Operator
+import org.apache.comet.serde.operator.{partition2Proto, schema2Proto}
+
+/*
+ * Native Text scan operator that delegates file reading to datafusion.
Mirrors the native CSV
+ * scan path; produces a single `value: string` column.
+ */
+case class CometTextNativeScanExec(
+ override val nativeOp: Operator,
+ override val output: Seq[Attribute],
+ @transient override val originalPlan: BatchScanExec,
+ override val serializedPlanOpt: SerializedPlan)
+ extends CometLeafExec {
+ override val supportsColumnar: Boolean = true
+
+ override val nodeName: String = "CometTextNativeScan"
+
+ // Show only the semantic output in EXPLAIN / the SQL tab / the event log.
The default would print
+ // the whole `nativeOp` protobuf, which carries every input file path (large
for a lookup
+ // directory of many files). Mirrors CometNativeScanExec.
+ override def stringArgs: Iterator[Any] = Iterator(output)
+
+ // The serialized partitions, which split any Spark partition that mixes
object stores.
+ override def outputPartitioning: Partitioning = UnknownPartitioning(
+ nativeOp.getTextScan.getFilePartitionsCount)
+
+ override def outputOrdering: Seq[SortOrder] = Nil
+
+ override protected def doCanonicalize(): SparkPlan = {
+ CometTextNativeScanExec(nativeOp, output, originalPlan, serializedPlanOpt)
+ }
+
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: CometTextNativeScanExec =>
+ this.output == other.output &&
+ this.serializedPlanOpt == other.serializedPlanOpt &&
+ this.originalPlan == other.originalPlan
+ case _ =>
+ false
+ }
+ }
+
+ override def hashCode(): Int = {
+ Objects.hashCode(output, serializedPlanOpt, originalPlan)
+ }
+}
+
+object CometTextNativeScanExec extends CometOperatorSerde[CometBatchScanExec] {
+
+ override def enabledConfig: Option[ConfigEntry[Boolean]] = Some(
+ CometConf.COMET_TEXT_V2_NATIVE_ENABLED)
+
+ override def convert(
+ op: CometBatchScanExec,
+ builder: Operator.Builder,
+ childOp: Operator*): Option[Operator] = {
+ val textScanBuilder = OperatorOuterClass.TextScan.newBuilder()
+ val textScan = op.wrapped.scan.asInstanceOf[TextScan]
+ val sessionState = op.session.sessionState
+ val options = new TextOptions(textScan.options.asScala.toMap)
+ // Build the Hadoop conf from the scan's read options (case preserved),
matching Spark's
+ // TextScan.createReaderFactory, so per-read settings like fs.s3a.endpoint
reach native.
+ val hadoopConf =
+
sessionState.newHadoopConfWithOptions(textScan.options.asCaseSensitiveMap.asScala.toMap)
+ val s3CompliantSchemes = NativeConfig.resolveS3CompliantSchemes(hadoopConf)
+ val libhdfsSchemes = NativeConfig.resolveLibhdfsSchemes(hadoopConf)
+ val inputPartitions = op.inputPartitions.map(_.asInstanceOf[FilePartition])
+ val multiStoreReason = CometScanUtils.multiStoreFallbackReason(
+ "Native Text scan",
+ inputPartitions.view.flatMap(_.files.view.map(_.pathUri)),
+ s3CompliantSchemes,
+ libhdfsSchemes,
+ isBucketedScan = false)
+ if (multiStoreReason.nonEmpty) {
+ // CometExecRule falls back to the wrapped scan, so tag it too for the
explain output.
+ withFallbackReason(op, multiStoreReason.get)
+ withFallbackReason(op.wrapped, multiStoreReason.get)
+ return None
+ }
+ // The options of every scheme the scan reads, or a decline for the ABFS
layouts one
+ // resolution cannot serve. Checked before the split below so that its
reason is reported.
+ val fileUris = inputPartitions.view.flatMap(_.files).map(_.pathUri)
+ val objectStoreOptions =
+ NativeConfig.extractScanObjectStoreOptions(hadoopConf, fileUris) match {
+ case Right(options) => options
+ case Left(reason) =>
+ withFallbackReason(op, reason)
+ withFallbackReason(op.wrapped, reason)
+ return None
+ }
+ // Native planning reads each partition through one object store.
+ val filePartitions = CometScanUtils.splitPartitionsByStore(
+ inputPartitions,
+ file => NativeConfig.objectStoreKey(file.pathUri, s3CompliantSchemes,
libhdfsSchemes))
+ // Spark 3.4 reports a one-partition scan as SinglePartition and plans no
exchange above it,
+ // so the operators above rely on the scan's partition count.
+ if (filePartitions.length != inputPartitions.length &&
+ !op.wrapped.outputPartitioning.isInstanceOf[UnknownPartitioning]) {
+ val reason = "Native Text scan would split a partition that mixes object
stores, but " +
+ "Spark planned the operators above it for the scan's own partitioning"
+ withFallbackReason(op, reason)
+ withFallbackReason(op.wrapped, reason)
+ return None
+ }
+ val textOptionsProto = textOptions2Proto(options)
+ val dataSchemaProto = schema2Proto(textScan.dataSchema)
+ val readSchemaFieldNames = textScan.readDataSchema.fieldNames
+ val projectionVector = textScan.dataSchema.zipWithIndex
+ .filter { case (field, _) =>
+ readSchemaFieldNames.contains(field.name)
+ }
+ .map(_._2.asInstanceOf[Integer])
+ val partitionSchemaProto = schema2Proto(textScan.readPartitionSchema)
+ val partitionsProto = filePartitions.map(partition2Proto(_,
textScan.readPartitionSchema))
+
+ textScanBuilder.putAllObjectStoreOptions(objectStoreOptions.asJava)
+ textScanBuilder.setTextOptions(textOptionsProto)
+ textScanBuilder.addAllFilePartitions(partitionsProto.asJava)
Review Comment:
https://github.com/apache/datafusion-comet/issues/7243
--
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]
