andygrove commented on code in PR #6784: URL: https://github.com/apache/datafusion-comet/pull/6784#discussion_r4241903624
########## docs/source/user-guide/latest/datasources.md: ########## @@ -49,6 +49,22 @@ converted into Arrow format, allowing the Comet pipeline to take over after that Comet does not provide a Rust-based JSON scan, but when `spark.comet.convert.json.enabled` is enabled, data is immediately converted into Arrow format, allowing the Comet pipeline to take over after that. +### Text + +Comet provides experimental Rust-based text scan support (a single `value: string` column). When +`spark.comet.scan.text.v2.enabled` is enabled, text files are read in Rust. This feature is experimental and +performance benefits are workload-dependent. Only Spark's DataSource V2 text scan is accelerated, and Spark reads +text through the V1 API by default, so also remove `text` from `spark.sql.sources.useV1SourceList`. The native scan +falls back to Spark for cases it does not handle -- including compressed files, files larger than +`spark.comet.scan.text.maxFileSize` (64MB by default, since the reader loads each file whole), files Spark splits +into byte ranges, unsupported filesystem schemes, and `input_file_name()` or +`ignoreCorruptFiles`/`ignoreMissingFiles` usage. It targets small lookup tables such as a broadcast join's build side. +Reading a partition column from a partitioned directory falls back to Spark, but selecting only `value` stays native. + +Text files are a common source of ill-formed UTF-8. The native reader replaces invalid byte sequences with the +Unicode replacement character (U+FFFD), so byte-level comparisons and functions can differ from Spark on those rows; +see [Strings with non-UTF-8 bytes](compatibility/scans.md) in the compatibility guide. Review Comment: The link text says Strings with non-UTF-8 bytes, but it points at `compatibility/scans.md`. That section is in `compatibility/index.md`, so could it link to `compatibility/index.md#strings-with-non-utf-8-bytes`? ########## spark/src/main/scala/org/apache/spark/sql/comet/CometTextNativeScanExec.scala: ########## @@ -0,0 +1,140 @@ +/* + * 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.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) + + override def outputPartitioning: Partitioning = UnknownPartitioning( + originalPlan.inputPartitions.length) + + 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) + val filePartitions = op.inputPartitions.map(_.asInstanceOf[FilePartition]) + 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)) + + val objectStoreOptions = filePartitions.headOption + .flatMap { partitionFile => + val hadoopConf = sessionState + .newHadoopConfWithOptions(op.session.sparkContext.conf.getAll.toMap) Review Comment: Agreed. The gate already builds its conf from the read options, so the serde should use the same one. `CometCsvNativeScanExec` has the same line. Could you fix both here, since this PR already touches the CSV node? ########## native/core/src/execution/operators/text_scan.rs: ########## @@ -0,0 +1,614 @@ +// 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?; + + // Decompress only when the file is actually compressed. On Comet's plan config the file + // is always UNCOMPRESSED (CometScanRule falls back for compressed files), so we borrow + // the fetched bytes directly and avoid a full-file copy; the decompress arm is a + // defensive handler should compression ever be plumbed through. A byte range never + // coexists with decompression (compressed files are non-splittable). + // + // NOTE: the whole object is read into memory. CometScanRule restricts native text to + // unsplit files (it falls back when Spark splits a file into byte ranges), so in + // practice this reads each small file once; the range filter below is a correctness + // safety net for any split that still reaches here. + let buf: Cow<[u8]> = if file_compression_type.is_compressed() { + let mut decoder = file_compression_type.convert_read(std::io::Cursor::new(raw))?; + let mut decoded = Vec::new(); + decoder.read_to_end(&mut decoded)?; + Cow::Owned(decoded) + } else { + Cow::Borrowed(raw.as_ref()) + }; + + let mut timer = baseline_metrics.elapsed_compute().timer(); + // Spark's line reader (Hadoop LineRecordReader.skipUtfByteOrderMark) strips a leading + // UTF-8 BOM at the start of the file. Mirror that in line mode for the split that starts + // at file offset 0 (native only reads unsplit files, so start is always 0; a custom + // lineSep still strips, matching Hadoop). wholetext keeps the BOM, as Spark's + // WholeTextFileRecordReader does. + let at_file_start = range.as_ref().map(|r| r.start == 0).unwrap_or(true); + let content = strip_leading_bom(&buf, whole_text, at_file_start); Review Comment: I reproduced this end to end on 3.4, 3.5, 4.0, 4.1 and 4.2. With `lineSep` set to U+FEFF, `BOM a BOM b` gives three rows in Spark and two natively, and a file holding only the BOM gives one row against none. Splitting first and then stripping the BOM from the first record would match Hadoop, as long as a file that is exactly the BOM still gives no rows with the default separator. Could you add tests for both cases? ########## native/core/src/execution/operators/text_scan.rs: ########## @@ -0,0 +1,614 @@ +// 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: The ceiling works for me. A 63 MiB file stays native and added up to about 150 MB of RSS, and a 65 MiB file falls back with the new reason. Thanks. ########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## @@ -458,6 +459,147 @@ 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 read a thread-local that Spark's FileScanRDD + // sets; the native DataFusion scan does not, so they would return "" / -1. Same guard as + // the native Parquet path. + if (plan.exists(node => + node.expressions.exists(_.exists { + case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength => true + case _ => false + }))) { + return withFallbackReason( + scanExec, + "Native Text scan is not compatible with input_file_name, " + + "input_file_block_start, or input_file_block_length") + } + 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)" + } + // 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 + // (e.g. .gz, .bz2). Fall back when any input file is compressed. Use the options-aware + // Hadoop conf so a codec registered via a per-read option is honored (as Spark does). + val hadoopConf = session.sessionState.newHadoopConfWithOptions(scan.options.asScala.toMap) + val codecFactory = new org.apache.hadoop.io.compress.CompressionCodecFactory(hadoopConf) + val hasCompressedFile = scan.fileIndex.inputFiles.exists { path => + codecFactory.getCodec(new org.apache.hadoop.fs.Path(path)) != null + } + 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( Review Comment: Confirmed on all five profiles. It falls back when the limit comes from a read option, the session conf or the Hadoop conf, and the results match Spark. ########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## @@ -458,6 +459,147 @@ 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 read a thread-local that Spark's FileScanRDD + // sets; the native DataFusion scan does not, so they would return "" / -1. Same guard as + // the native Parquet path. + if (plan.exists(node => + node.expressions.exists(_.exists { + case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength => true + case _ => false + }))) { + return withFallbackReason( + scanExec, + "Native Text scan is not compatible with input_file_name, " + + "input_file_block_start, or input_file_block_length") + } + 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)" + } + // 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 + // (e.g. .gz, .bz2). Fall back when any input file is compressed. Use the options-aware + // Hadoop conf so a codec registered via a per-read option is honored (as Spark does). + val hadoopConf = session.sessionState.newHadoopConfWithOptions(scan.options.asScala.toMap) + val codecFactory = new org.apache.hadoop.io.compress.CompressionCodecFactory(hadoopConf) + val hasCompressedFile = scan.fileIndex.inputFiles.exists { path => + codecFactory.getCodec(new org.apache.hadoop.fs.Path(path)) != null + } + 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" + } + // 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) + 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://, ...) would + // be claimed here then hard-fail at execution, whereas Spark reads it via the Hadoop FS + // API. Decline such schemes (same gate as the native Parquet path); hdfs:// routes through + // libhdfs and is left to native. + val libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get() match { Review Comment: Thanks for pulling this into one helper. The new text check only fires for alias schemes, though, because `aliasScanBuckets` returns nothing when no root is an alias. So `spark.read.text("s3a://a/x.txt", "s3a://b/y.txt")` still goes native. The `TextScan` arm in `planner.rs` builds the object store from the first file of the first partition, so `y.txt` is read from bucket `a` even when it lands in a different partition. That is either a missing-key error or a silently wrong file. Could the text gate decline any scan whose root paths span more than one scheme or bucket? `RootPathInfo.bucket` already covers `s3`, `s3a` and `s3n`, and the URI authority would cover `gs://` and `abfs://`. ########## spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala: ########## @@ -458,6 +459,147 @@ 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 read a thread-local that Spark's FileScanRDD + // sets; the native DataFusion scan does not, so they would return "" / -1. Same guard as + // the native Parquet path. + if (plan.exists(node => + node.expressions.exists(_.exists { + case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength => true + case _ => false + }))) { + return withFallbackReason( + scanExec, + "Native Text scan is not compatible with input_file_name, " + + "input_file_block_start, or input_file_block_length") + } + 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)" + } + // 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 + // (e.g. .gz, .bz2). Fall back when any input file is compressed. Use the options-aware + // Hadoop conf so a codec registered via a per-read option is honored (as Spark does). + val hadoopConf = session.sessionState.newHadoopConfWithOptions(scan.options.asScala.toMap) + val codecFactory = new org.apache.hadoop.io.compress.CompressionCodecFactory(hadoopConf) + val hasCompressedFile = scan.fileIndex.inputFiles.exists { path => + codecFactory.getCodec(new org.apache.hadoop.fs.Path(path)) != null + } + if (hasCompressedFile) { + fallbackReasons += "Comet does not support compressed text files" Review Comment: This works now. On 4.1 and 4.2 the `.gz`, `.gzip`, `.zst` and `.zstd` files all fall back and match Spark, with and without `wholetext`. On 3.4, 3.5 and 4.0 the `.gzip` and `.zstd` files stay native and match Spark's raw read. The new test is named for `.gzip` and `.zstd` but only writes a `.gzip` file. Could it write a `.zstd` file too? `com.github.luben.zstd.ZstdOutputStream` is already on the test classpath. -- 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]
