sunchao commented on code in PR #6784:
URL: https://github.com/apache/datafusion-comet/pull/6784#discussion_r4367108555
##########
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:
[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.
##########
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:
[P2] Preserve Hadoop checksum verification for local text files. With
verification enabled and a stale `.crc` sidecar, Spark raises
`ChecksumException`. This direct object-store read returns the altered rows,
allowing a lookup to use corrupt keys or values. Use a checksum-aware read or
fall back when Hadoop would verify the file.
Evidence: On Spark 4.1.3, I wrote `one\n` through Hadoop’s local filesystem,
confirmed the sidecar existed, then replaced the data with `two\n` before
either query. Spark and Comet with native text disabled raised
`ChecksumException` at offset zero. `CometTextNativeScan` returned `two`.
##########
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:
[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.
##########
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:
[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.
##########
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:
[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.
##########
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:
[P2] Report native text input bytes through the existing Spark task-metrics
bridge. The opener records baseline metrics, and
`CometMetricNode.reportScanInputMetrics` selects leaves exposing
`bytes_scanned`. Consequently both bytes and records disappear from task events
and the stage UI. Record the fetched byte length and expose that metric
alongside `output_rows` on the JVM leaf.
Evidence: A Spark 4.1.3 task listener summed input metrics while collecting
four six-byte text files. Spark and Comet with native text disabled reported 24
bytes and 12 records. The native scan returned the same 12 rows across four
tasks and reported zero bytes and zero records.
##########
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:
[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.
##########
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:
[P2] Decline line-mode scans with Hadoop mandatory file-open options the
native opener cannot enforce. Spark propagates
`mapreduce.job.input.file.must.*` to the filesystem’s open builder. A failed
requirement must fail the read, but this branch admits native execution and
returns rows.
Evidence: On Spark 4.1.3, a local text read with
`.option("mapreduce.job.input.file.must.fs.s3a.readahead.range", "64K")` raised
`IllegalArgumentException` for an unknown mandatory key. `CometTextNativeScan`
returned `alpha` and `beta`. Changing the prefix from `must` to `option` let
both readers return the rows.
##########
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:
[P2] Apply the reader-only libhdfs scheme guard when forwarding CSV read
options. A read option can now route a previously working file through libhdfs
without a matching session route. Keep the case-preserved options and use
`resolveScanLibhdfsSchemes` to decline selected files whose scheme appears only
in the read configuration.
Evidence: A local V2 CSV read with `.option("fs.comet.libhdfs.schemes",
"file,hdfs")` returned `alpha` and `beta` through Spark and through the base
native CSV serializer. The head serializer failed to convert the `file:///`
path to a libhdfs URL. Removing the option made the head native scan succeed.
##########
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:
[P2] Give each native task its own file group. `addAllFilePartitions` embeds
every group in the plan that each task copies into JNI and decodes before
selecting `self.partition`. For many small files, that work grows with file
count multiplied by task count. Use Comet’s per-partition plan-data injection
and have the native planner consume the injected group.
Evidence: With 1,024 two-byte files and both `maxPartitionBytes` and
`openCostInBytes` set to `1m`, the actual `CometExecRDD` had 1,024 tasks, an
empty `commonByKey`, and an 85,039-byte plan containing all groups. Keeping the
first task’s group reduced its plan to 126 bytes. Across three isolated Rust
trials, 1,024 full-plan decodes took 159–177 ms and single-group decodes took
0.68–0.70 ms.
--
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]
