fix: keep Text build sides on Spark by default and add Text conversion configs

docs/source/user-guide/latest/compatibility/index.md

@@ -97,9 +97,41 @@ underlying bytes rather than on the rendered text:
   bytes) but `true` in Comet. Equality, joins, grouping, ordering, and byte-based string functions
   such as `contains` can therefore disagree with Spark when non-UTF-8 bytes are involved.
 
-Both differences require string data that is not valid UTF-8, which does not occur for text read from
-Parquet or produced by string expressions. Consistent handling of invalid UTF-8 across all native
-string paths is tracked by
+Both differences require string data that is not valid UTF-8. Comet's native Parquet scan never
+produces such strings (it rejects them, see below), and neither do most string expressions. They can
+reach Comet in these ways, which are the ones to check before relying on a result:
+
+- **Text files.** Spark's Text reader keeps the bytes of each line as they are, so a Text file in an
+  encoding other than UTF-8, such as Latin-1, holds strings that are not valid UTF-8. Comet decodes
+  them when it converts a Text scan to Arrow, which it does only when
+  `spark.comet.convert.text.enabled=true` (see [Data sources](../datasources.md#text)), or, as a
+  deprecated alternative, when `spark.comet.sparkToColumnar.enabled=true` and
+  `spark.comet.sparkToColumnar.supportedOperatorList` names the scan. For example, with a Text file
+  holding the bytes `1 caf\xE9`, `hex(value)` returns `3120636166EFBFBD` instead of Spark's
+  `3120636166E9`, and `length` of `2 na\xEFve` is 7 instead of 5. Spark decodes CSV files as it reads
+  them, so they never hold such strings, and its JSON reader treats a line with such bytes as
+  malformed, so only the corrupt record column keeps them. Spark's ORC and Avro readers keep stored
+  bytes without checking them, but those formats define strings as UTF-8, so only a malformed file is
+  affected.
+- **Comet columnar shuffle.** When Comet shuffles rows that come from a Spark operator, such as a
+  Text scan or a Spark partial aggregate, it converts them to Arrow and decodes their strings in the
+  same way. It does this with the default `spark.comet.shuffle.mode=auto`, so, for example,
+  repartitioning a Latin-1 Text table can return `U+FFFD` where Spark keeps the original bytes.
+  Setting `spark.comet.shuffle.mode=native` keeps such shuffles on Spark.
+- **`CAST(binary AS STRING)`.** Comet decodes the bytes where Spark keeps them. Because `min` and `max`
+  over strings run natively, a cast below them is decoded before the comparison, and the two can swap.
+  With `b` holding `X'80'` and `X'C3A9'`, Spark returns `min(CAST(b AS STRING))` = the `0x80` string
+  and `max` = `é`, whereas Comet returns `min` = `é` and `max` = `U+FFFD`. This is the ordering
+  difference above. If the cast input can hold bytes that are not valid UTF-8, disable Comet for the
+  query. When the cast is the direct argument of `min` or `max`, a narrower alternative is to set
+  `spark.comet.expression.Min.enabled=false`, `spark.comet.expression.Max.enabled=false` and
+  `spark.comet.shuffle.mode=native`. All three are needed: with only the first two, the aggregate
+  runs in Spark, but Comet's columnar shuffle still decodes its partial results. These settings are
+  not enough when the cast is in a projection below the aggregate, for example
+  `SELECT min(s) FROM (SELECT CAST(b AS STRING) AS s FROM t)`, because Comet runs that projection
+  natively and decodes the bytes before the aggregate sees them.
+
+Consistent handling of invalid UTF-8 across all native string paths is tracked by
 [#4764](https://github.com/apache/datafusion-comet/issues/4764).
 
 Separately, Comet's native Parquet scan currently rejects string columns whose stored bytes are not

docs/source/user-guide/latest/datasources.md

@@ -49,6 +49,15 @@ 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 does not provide a Rust-based Text scan, but when `spark.comet.convert.text.enabled` is enabled, data from Spark's
+Text reader is immediately converted into Arrow format, allowing the Comet pipeline to take over after that. Spark's Text
+reader keeps the bytes of each line as they are, so a file in an encoding other than UTF-8, such as Latin-1, holds strings
+that are not valid UTF-8. Arrow strings must be valid UTF-8, so the conversion replaces those bytes with `U+FFFD`, and
+results that depend on the original bytes can differ from Spark. Enable it only for Text files that are UTF-8. See
+[Strings with non-UTF-8 bytes](compatibility/index.md#strings-with-non-utf-8-bytes).
+
 ### Other Spark inputs
 
 Comet can also convert the output of these Spark inputs to Arrow format, so that the operators
@@ -67,12 +76,15 @@ because the row it converts has no columns.
   such as JDBC tables, which Spark scans with `RowDataSourceScanExec`.
 
 To convert any other leaf operator, such as the scan of a Data Source V2 connector or of a file
-format other than Parquet, JSON and CSV, set `spark.comet.sparkToColumnar.enabled=true` and name the
-operator in `spark.comet.sparkToColumnar.supportedOperatorList` by its Spark class name without the
-`Exec` suffix, such as `BatchScan` or `FileSourceScan`.
+format other than Parquet, JSON, CSV and Text, set `spark.comet.sparkToColumnar.enabled=true` and name
+the operator in `spark.comet.sparkToColumnar.supportedOperatorList` by its Spark class name without the
+`Exec` suffix, such as `BatchScan` or `FileSourceScan`. Naming either to convert a Text scan still
+works but is deprecated, like naming `Range`, `InMemoryTableScan`, `RDDScan`, `OneRowRelation` or
+`RowDataSourceScan`, and will stop working in a future major release. Use
+`spark.comet.convert.text.enabled` instead.
 
 One case is converted automatically: when the build (broadcast) side of a broadcast join is a file
-or V2 scan Comet cannot read natively - for example a small Text lookup table - and every scan on
+or V2 scan Comet cannot read natively - for example a small ORC lookup table - and every scan on
 the probe side is already native, Comet converts that build-side scan to Arrow so the whole join
 runs natively over the large probe. This is controlled by
 `spark.comet.convert.broadcastBuildSide.enabled`, which is **on by default**, unlike the switches
@@ -81,6 +93,10 @@ the scan can make the copied volume larger, and with dynamic partition pruning t
 a second time for the pruning subquery. Set `spark.comet.convert.broadcastBuildSide.enabled=false` to
 keep such build sides on Spark.
 
+Parquet, JSON, CSV and Text build sides are converted only by their own switches above. A Text file
+can hold bytes that are not valid UTF-8 (see [Text](#text)), so a Text lookup table stays on Spark
+unless `spark.comet.convert.text.enabled` is set.
+
 ### Spark-to-Comet conversion types
 
 Spark-to-Comet conversion supports `ARRAY<STRING>` and `MAP<STRING,STRING>` with binary

docs/source/user-guide/latest/migration-guide.md

@@ -86,6 +86,9 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+Converting a Text scan with `FileSourceScan` or `BatchScan` in the list is deprecated in the same
+way, and logs the same warning. Use `spark.comet.convert.text.enabled` instead.
+
 ### Iceberg Writes
 
 Comet's Iceberg write path is now switched by one setting, `spark.comet.write.iceberg.enabled`,

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -227,6 +227,18 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(false)
 
+  val COMET_CONVERT_FROM_TEXT_ENABLED: ConfigEntry[Boolean] =
+    conf("spark.comet.convert.text.enabled")
+      .category(CATEGORY_EXEC)
+      .doc(
+        "When enabled, data from Spark (non-native) Text v1 and v2 scans will be converted to " +
+          "Arrow format. Arrow strings must be valid UTF-8, so Comet replaces bytes that are " +
+          "not valid UTF-8, for example in a Latin-1 file, with U+FFFD, and results that " +
+          "depend on the original bytes can differ from Spark. Enable this only for Text " +
+          "sources that are UTF-8. See the compatibility guide.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_CONVERT_FROM_RANGE_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.convert.range.enabled")
       .category(CATEGORY_EXEC)
@@ -1257,10 +1269,11 @@ object CometConf extends ShimCometConf {
           s"`${COMET_SPARK_TO_ARROW_ENABLED.key}` is true, each named by its Spark class name " +
           "without the `Exec` suffix. It is for operators that have no `spark.comet.convert` " +
           "config of their own, such as `BatchScan` for a Data Source V2 connector, or " +
-          "`FileSourceScan` for file formats other than Parquet, JSON and CSV. Naming `Range`, " +
-          "`InMemoryTableScan`, `RDDScan`, `OneRowRelation` or `RowDataSourceScan` is " +
-          "deprecated and will stop working in a future major release. Use their " +
-          "`spark.comet.convert` configs instead.")
+          "`FileSourceScan` for file formats other than Parquet, JSON, CSV and Text. Naming " +
+          "`Range`, `InMemoryTableScan`, `RDDScan`, `OneRowRelation` or `RowDataSourceScan`, " +
+          "or naming `FileSourceScan` or `BatchScan` to convert a Text scan, is deprecated " +
+          "and will stop working in a future major release. Use their `spark.comet.convert` " +
+          "configs instead.")
       .stringConf
       .toSequence
       .createWithDefault(Nil)
@@ -1269,13 +1282,14 @@ object CometConf extends ShimCometConf {
     conf("spark.comet.convert.broadcastBuildSide.enabled")
       .category(CATEGORY_EXEC)
       .doc("When enabled, Comet converts a file or V2 scan that it cannot scan natively (for " +
-        "example a Text source) to Arrow when that scan feeds a broadcast join's build side and " +
+        "example an ORC source) to Arrow when that scan feeds a broadcast join's build side and " +
         "every scan on the probe side is already native, so the broadcast join runs natively " +
         "over the large probe. The build side is usually small (auto-broadcasts are capped by " +
         "the broadcast threshold), so the row to Arrow copy is usually cheap - though an " +
         "explicit BROADCAST hint or a selective filter above the scan can make it larger. With " +
         "dynamic partition pruning the dimension is also read a second time for the pruning " +
-        "subquery. Set this to false to keep such build sides on Spark.")
+        "subquery. Parquet, JSON, CSV and Text scans are converted only by their own " +
+        "`spark.comet.convert` configs. Set this to false to keep such build sides on Spark.")
       .booleanConf
       .createWithDefault(true)
 

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -45,10 +45,12 @@ import org.apache.spark.sql.execution.datasources.{InsertIntoHadoopFsRelationCom
 import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat
 import org.apache.spark.sql.execution.datasources.json.JsonFileFormat
 import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.execution.datasources.text.TextFileFormat
 import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, V2CommandExec}
 import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
 import org.apache.spark.sql.execution.datasources.v2.json.JsonScan
 import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
+import org.apache.spark.sql.execution.datasources.v2.text.TextScan
 import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, BroadcastExchangeLike, Exchange, ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike}
 import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
 import org.apache.spark.sql.execution.window.WindowExec
@@ -192,6 +194,15 @@ object CometExecRule {
     "Spark to Arrow conversion is not compatible with input_file_name, " +
       "input_file_block_start, or input_file_block_length"
 
+  /**
+   * Why Comet does not bridge a Text scan on a broadcast build side. A Text file can hold bytes
+   * that are not valid UTF-8, and the conversion replaces them with U+FFFD. See issue #7118.
+   */
+  private val TEXT_BUILD_SIDE_FALLBACK_REASON: String =
+    "Comet converts a Text scan on a broadcast build side to Arrow only when " +
+      s"${CometConf.COMET_CONVERT_FROM_TEXT_ENABLED.key}=true, because a Text " +
+      "file can hold bytes that are not valid UTF-8"
+
   /**
    * Tag set on a scan leaf that sits on a broadcast join's build side. When
    * `spark.comet.convert.broadcastBuildSide.enabled` is on, such a scan is bridged to Arrow with
@@ -989,7 +1000,7 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
       tagUnsafePartialAggregates(planWithJoinRewritten)
 
       // Tag leaves on broadcast join build sides so shouldApplySparkToColumnar can bridge an
-      // unsupported build-side scan (e.g. Text) to Arrow and let the join run natively (#6008).
+      // unsupported build-side scan (e.g. ORC) to Arrow and let the join run natively (#6008).
       tagBroadcastBuildSideLeaves(planWithJoinRewritten)
 
       var transformed = transform(planWithJoinRewritten)
@@ -1459,8 +1470,9 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
             case _: CSVFileFormat => CometConf.COMET_CONVERT_FROM_CSV_ENABLED.get(conf)
             case _: JsonFileFormat => CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf)
             case _: ParquetFileFormat => CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf)
-            // Unsupported format (e.g. Text): bridge it if the general opt-in is on or it feeds a
-            // broadcast build side. The per-format arms above keep honoring their own opt-outs.
+            case _: TextFileFormat => shouldConvertTextScan(conf, op)
+            // Unsupported format (e.g. ORC): bridge it if the general opt-in is on or it feeds a
+            // broadcast build side. The per-format arms above keep honoring their own switches.
             case _ => shouldBridgeUnsupportedScan(conf, op)
           }
         // Convert Spark DS v2 scan to Arrow format
@@ -1469,6 +1481,7 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
             case _: CSVScan => CometConf.COMET_CONVERT_FROM_CSV_ENABLED.get(conf)
             case _: JsonScan => CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf)
             case _: ParquetScan => CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf)
+            case _: TextScan => shouldConvertTextScan(conf, op)
             case _ => shouldBridgeUnsupportedScan(conf, op)
           }
         case _: RangeExec =>
@@ -1516,12 +1529,28 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
   /**
    * An unsupported file-source scan is bridged to Arrow when the general sparkToColumnar opt-in
    * covers it, or when it feeds a broadcast join's build side (see issue #6008). This is only
-   * reached from the unsupported-format arms, so the per-format `COMET_CONVERT_FROM_*` opt-outs
-   * for CSV/JSON/Parquet still take precedence.
+   * reached from the unsupported-format arms, so the per-format `COMET_CONVERT_FROM_*` switches
+   * for CSV/JSON/Parquet/Text still take precedence.
    */
   private def shouldBridgeUnsupportedScan(conf: SQLConf, op: SparkPlan): Boolean =
     isSparkToArrowEnabled(conf, op) || isBroadcastBuildSideLeaf(op)
 
+  /**
+   * A Text scan is converted only by `spark.comet.convert.text.enabled`, like CSV, JSON and
+   * Parquet, and so is a Text scan on a broadcast build side: a Text file can hold bytes that are
+   * not valid UTF-8, for example a Latin-1 file, and the conversion replaces them with U+FFFD, so
+   * results that depend on the original bytes would differ from Spark. A Text build side that the
+   * bridge would otherwise convert records why it did not, so the explain output names the
+   * config. See issue #7118.
+   */
+  private def shouldConvertTextScan(conf: SQLConf, op: SparkPlan): Boolean =
+    isConversionEnabled(conf, op, CometConf.COMET_CONVERT_FROM_TEXT_ENABLED) || {
+      if (isBroadcastBuildSideLeaf(op)) {
+        withFallbackReason(op, CometExecRule.TEXT_BUILD_SIDE_FALLBACK_REASON)
+      }
+      false
+    }
+
   /**
    * True when the build-side auto-bridge should run: its own config is on and Comet broadcast
    * exchange conversion is enabled. Without the latter the `BroadcastExchangeExec` cannot become
@@ -1550,7 +1579,7 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
 
   /**
    * Tag the file-source scan leaves on a broadcast join's build side so
-   * shouldApplySparkToColumnar can bridge an unsupported build-side scan (e.g. a Text scan) to
+   * shouldApplySparkToColumnar can bridge an unsupported build-side scan (e.g. an ORC scan) to
    * Arrow, letting the whole join run natively. The build side is usually small (auto-broadcasts
    * are capped by the broadcast threshold), so the row to Arrow copy is usually cheap - but that
    * is a heuristic, not a bound: an explicit BROADCAST hint can force a larger build side, and a
@@ -1680,10 +1709,11 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
    * Two guards keep the walk honest. It stops at a `CometBroadcastExchangeExec` - a nested
    * broadcast join that went native keeps its bridge. And it only clears a tag (and so asks for a
    * second transform) when the tagged scan sits under a `CometSparkToColumnarExec`, i.e. it was
-   * actually bridged. A CSV/JSON/Parquet scan whose per-format arm left it on Spark still carries
-   * the tag but was never bridged, so touching it would force a pointless second transform. The
-   * descent enters `CometScanWrapper`'s wrapped `originalPlan` by hand because it is a leaf, so a
-   * plain `foreach` would never reach the bridged scan inside it. See issue #6008.
+   * actually bridged. A CSV/JSON/Parquet/Text scan whose per-format arm left it on Spark still
+   * carries the tag but was never bridged, so touching it would force a pointless second
+   * transform. The descent enters `CometScanWrapper`'s wrapped `originalPlan` by hand because it
+   * is a leaf, so a plain `foreach` would never reach the bridged scan inside it. See issue
+   * #6008.
    */
   private def clearWastedBuildSideBridges(plan: SparkPlan): Boolean = {
     var cleared = false