andygrove commented on code in PR #6784:
URL: https://github.com/apache/datafusion-comet/pull/6784#discussion_r4542257260
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -190,6 +190,27 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
+ val COMET_TEXT_V2_NATIVE_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.scan.text.v2.enabled")
+ .category(CATEGORY_TESTING)
+ .doc(
+ "Whether to use the native Comet V2 reader for the text file format (a
single " +
+ "'value: string' column). Requires the V2 text source (remove 'text'
from " +
+ "spark.sql.sources.useV1SourceList). Default: false. " +
+ "Experimental: Performance benefits are workload-dependent.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val COMET_SCAN_TEXT_MAX_FILE_SIZE: ConfigEntry[Long] =
+ conf("spark.comet.scan.text.maxFileSize")
Review Comment:
This key is in the `testing` category, but `datasources.md` points users at
it, and someone with larger lookup files has a legitimate reason to raise it.
`config_conventions.md` puts a key like that in a non-`testing` category,
because `testing` keys are exempt from the versioning policy and are listed
under Development & Testing. Could it move to `CATEGORY_SCAN` before this
merges? The `v2.enabled` flag can stay in `testing` like the CSV one.
##########
spark/src/test/scala/org/apache/comet/text/CometTextNativeReadSuite.scala:
##########
@@ -0,0 +1,710 @@
+/*
+ * 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.comet.text
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.util.concurrent.atomic.AtomicLong
+import java.util.zip.GZIPOutputStream
+
+import org.scalatest.time.{Millis, Seconds, Span}
+
+import org.apache.hadoop.fs.{FileSystem, Path}
+import org.apache.hadoop.io.compress.GzipCodec
+import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
+import org.apache.spark.sql.{CometTestBase, DataFrame}
+import org.apache.spark.sql.comet.CometTextNativeScanExec
+import org.apache.spark.sql.functions.input_file_name
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.{CometConf, ExtendedExplainInfo}
+import org.apache.comet.CometSparkSessionExtensions.isSpark41Plus
+
+/** A gzip codec whose file extension holds a space, to test codec lookup on
decoded paths. */
+class SpacedGzipCodec extends GzipCodec {
+ override def getDefaultExtension: String = ".packed gz"
+}
+
+class CometTextNativeReadSuite extends CometTestBase {
+ private val TEST_TEXT_PATH = "src/test/resources/test-data/text-test-1.txt"
+
+ private def withTextConf(f: => Unit): Unit = {
+ withSQLConf(
+ CometConf.COMET_TEXT_V2_NATIVE_ENABLED.key -> "true",
+ // Route only `text` through the V2 path; keep every other format on V1
so a Parquet probe
+ // side (used by the broadcast-join test) is not pushed through the
unsupported V2 scan.
+ SQLConf.USE_V1_SOURCE_LIST.key -> "avro,csv,json,kafka,orc,parquet")(f)
+ }
+
+ test("native text read - lines to value column") {
+ withTextConf {
+ val df = spark.read.text(TEST_TEXT_PATH)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("native text read - count") {
+ withTextConf {
+ val df = spark.read.text(TEST_TEXT_PATH).selectExpr("count(*) as c")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("native text read - filter and projection") {
+ withTextConf {
+ val df = spark.read.text(TEST_TEXT_PATH).where("value like '%bar%'")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("native text read - preserves line order within a file") {
+ withTempDir { dir =>
+ val file = new File(dir, "ordered.txt")
+ val lines = (1 to 500).map(i => s"row-$i")
+ Files.write(file.toPath,
lines.mkString("\n").getBytes(StandardCharsets.UTF_8))
+ withTextConf {
+ val df = spark.read.text(file.getAbsolutePath)
+ // Assert the scan actually ran natively (a Spark fallback would also
preserve order).
+ checkSparkAnswerAndOperator(df, Seq(classOf[CometTextNativeScanExec]))
+ // checkSparkAnswer sorts both sides, so assert order explicitly here.
+ val got = df.collect().map(_.getString(0)).toSeq
+ assert(got == lines, "native text scan must preserve within-file line
order")
+ }
+ }
+ }
+
+ test("native text read - fallback for input_file_name") {
+ withTextConf {
+ val df = spark.read.text(TEST_TEXT_PATH).withColumn("f",
input_file_name())
+ // The blanket V2 input_file_name gate (shared by all native V2 scans)
emits this.
+ checkSparkAnswerAndFallbackReason(df, "is not compatible with
input_file_name")
+ }
+ }
+
+ test("native text read - fallback when native exec is disabled") {
+ withSQLConf(
+ CometConf.COMET_TEXT_V2_NATIVE_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "false",
+ SQLConf.USE_V1_SOURCE_LIST.key -> "") {
+ checkSparkAnswerAndFallbackReason(
+ spark.read.text(TEST_TEXT_PATH),
+ s"Native Text scan requires ${CometConf.COMET_EXEC_ENABLED.key} to be
enabled")
+ }
+ }
+
+ test("native text read - wholetext") {
+ withTextConf {
+ val df = spark.read.option("wholetext", "true").text(TEST_TEXT_PATH)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("native text read - custom lineSep") {
+ withTempDir { dir =>
+ val file = new File(dir, "custom-linesep.txt")
+ Files.write(file.toPath, "a|b|café|日本".getBytes(StandardCharsets.UTF_8))
+ withTextConf {
+ val df = spark.read.option("lineSep", "|").text(file.getAbsolutePath)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
+
+ test("native text read - custom lineSep with trailing separator") {
+ withTempDir { dir =>
+ val file = new File(dir, "trailing-linesep.txt")
+ // A trailing separator must not produce an extra empty row (matches
Spark, SPARK-23577).
+ Files.write(file.toPath, "a|b|c|".getBytes(StandardCharsets.UTF_8))
+ withTextConf {
+ val df = spark.read.option("lineSep", "|").text(file.getAbsolutePath)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
+
+ test("native text read - non-ASCII multibyte custom lineSep") {
+ withTempDir { dir =>
+ val file = new File(dir, "jp-linesep.txt")
+ // Japanese ideographic comma (U+3002, 3 UTF-8 bytes) as the record
separator.
+ Files.write(file.toPath, "a。café。日本".getBytes(StandardCharsets.UTF_8))
+ withTextConf {
+ val df = spark.read.option("lineSep", "。").text(file.getAbsolutePath)
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+ }
+
+ test("native text read - CRLF and CR line endings") {
+ withTempDir { dir =>
+ val file = new File(dir, "crlf.txt")
+ // Mix of \r\n, \r, and \n; the default reader splits on all three.
+ Files.write(file.toPath, "a\r\nb\rc\nd".getBytes(StandardCharsets.UTF_8))
+ withTextConf {
+ checkSparkAnswerAndOperator(spark.read.text(file.getAbsolutePath))
+ }
+ }
+ }
+
+ test("native text read - invalid UTF-8 bytes match Spark on collect") {
+ withTempDir { dir =>
+ val file = new File(dir, "invalid-utf8.txt")
+ // 0xFF is not valid UTF-8; both readers render it as the replacement
char on collect.
+ Files.write(file.toPath, Array[Byte]('a', 0xff.toByte, 'b', '\n', 'c'))
+ withTextConf {
+ checkSparkAnswerAndOperator(spark.read.text(file.getAbsolutePath))
+ }
+ }
+ }
+
+ test("native text read - wholetext count") {
+ withTextConf {
+ val df = spark.read
+ .option("wholetext", "true")
+ .text(TEST_TEXT_PATH)
+ .selectExpr("count(*) as c")
+ checkSparkAnswerAndOperator(df)
+ }
+ }
+
+ test("native text read - falls back for files split into byte ranges") {
+ withTempDir { dir =>
+ val file = new File(dir, "many-lines.txt")
+ val content = (1 to 2000).map(i => s"line-$i-日本テスト").mkString("\n")
+ Files.write(file.toPath, content.getBytes(StandardCharsets.UTF_8))
+ withSQLConf(
+ CometConf.COMET_TEXT_V2_NATIVE_ENABLED.key -> "true",
+ SQLConf.USE_V1_SOURCE_LIST.key -> "",
+ // Force the file to be split into several byte-range partitions.
+ SQLConf.FILES_MAX_PARTITION_BYTES.key -> "1024") {
+ val df = spark.read.text(file.getAbsolutePath)
+ assert(df.rdd.getNumPartitions > 1, "test requires the file to
actually split")
+ checkSparkAnswerAndFallbackReason(
+ df,
+ "Comet native Text scan does not support files split into byte
ranges")
+ }
+ }
+ }
+
+ test("native text read - empty file") {
Review Comment:
Nit, optional: could we also cover `wholetext=true` on an empty file here? I
expect one empty row from Spark and the native reader returns one, but neither
this test nor the `wholetext` ones pin it.
--
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]
