viirya commented on code in PR #7123:
URL: https://github.com/apache/datafusion-comet/pull/7123#discussion_r4328243129
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -1269,16 +1281,34 @@ 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 and CSV scans are converted only by their own
" +
+ "`spark.comet.convert` configs. A Text scan is converted only when " +
+ "`spark.comet.convert.broadcastBuildSide.text.enabled` is also
enabled, because Text " +
+ "files can hold bytes that are not valid UTF-8. Set this to false to
keep such build " +
+ "sides on Spark.")
.booleanConf
.createWithDefault(true)
+ val COMET_CONVERT_BROADCAST_BUILD_SIDE_TEXT_ENABLED: ConfigEntry[Boolean] =
Review Comment:
Do we need both new Text configs? `spark.comet.convert.text.enabled` already
covers the build side, so this one only adds the option of converting a Text
build side while every other Text scan stays on Spark.
If Text followed the CSV, JSON and Parquet model, where a build side is
converted only by the format's own switch, the datasources paragraph could say
"Parquet, JSON, CSV and Text build sides are converted only by their own
switches". The fallback reason could point at
`spark.comet.convert.text.enabled`, and `shouldConvertTextScan` would lose its
middle branch. Most of the Latin-1 explanation that is now repeated across the
two config docs and the scaladocs on `BROADCAST_BUILD_SIDE_TAG`,
`tagBroadcastBuildSideLeaves` and `shouldConvertTextScan` would go away too.
And we would not start a pattern of one `broadcastBuildSide.<format>.enabled`
for each format we later need to exclude.
The #6008 allowlist case would set `spark.comet.convert.text.enabled=true`
instead. Is there a workload where converting the build side but not other Text
scans matters? @parthchandra's answer on the #6008 use case may settle this.
##########
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] =
Review Comment:
If this config stays, two existing docs still describe `FileSourceScan` in
`spark.comet.sparkToColumnar.supportedOperatorList` as the way to convert "file
formats other than Parquet, JSON and CSV": the `supportedOperatorList` doc
further down in this file, and the "To convert any other leaf operator"
paragraph in `datasources.md`. Could they mention Text?
Related to that, the other leaves that gained their own
`spark.comet.convert` config (Range, InMemoryTableScan, RDDScan,
OneRowRelation) keep the `sparkToColumnar` path through `isConversionEnabled`,
which logs a deprecation warning. `shouldConvertTextScan` keeps the path for
Text without one. Is Text meant to stay supported there long term, or should it
be deprecated the same way? The docs should say which.
##########
docs/source/user-guide/latest/compatibility/index.md:
##########
@@ -97,9 +97,38 @@ 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`, when the scan is on a broadcast
build side and
+ `spark.comet.convert.broadcastBuildSide.text.enabled=true`, or when
+ `spark.comet.sparkToColumnar.enabled=true` and
`spark.comet.sparkToColumnar.supportedOperatorList`
+ names the scan (see [Data sources](../datasources.md#text)). For example,
with a build-side 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, or set `spark.comet.expression.Min.enabled=false`,
`spark.comet.expression.Max.enabled=false`
Review Comment:
The three-config workaround seems to depend on the cast being the
aggregate's own argument. If the cast is in a projection below the aggregate,
for example `SELECT min(s) FROM (SELECT CAST(b AS STRING) AS s FROM t)` over a
native Parquet scan, Spark keeps that `Project`, because `CollapseProject` does
not fold a `Project` into an `Aggregate` above it. Comet then runs the
projection natively and decodes `b` before the Spark aggregate sees it. Did the
probe cover that shape? If the workaround does not hold there, could the
sentence say the three configs apply when the cast is directly inside `min` or
`max`, and keep disabling Comet as the general answer?
--
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]
