Repository navigation
Text broadcast build sides and string min/max apply the invalid UTF-8 decode where 1.1.0 stayed on Spark #7118
Description
Describe the bug
Two changes on main send strings that used to stay in Spark through Comet's invalid UTF-8 decode, so 1.2.0 can return different results from 1.1.0 for the same query. Comet replaces each ill-formed UTF-8 sequence with U+FFFD, as the compatibility guide documents under "Strings with non-UTF-8 bytes". Spark keeps the original bytes, so hex, length, octet_length, hashing, equality and ordering can disagree. In 1.1.0 both of these query shapes ran entirely in Spark.
1. Text source on a broadcast build side (#6786). spark.comet.convert.broadcastBuildSide.enabled defaults to true, so a Text scan feeding a broadcast join's build side is now converted to Arrow. Text is the source most likely to hold invalid UTF-8, for example a Latin-1 file. With a build-side file holding the bytes 1 caf\xE9:
hex(value) |
|
|---|---|
| Spark, Comet 1.1.0 | 3120636166E9 |
main |
3120636166EFBFBD |
length changes too: for the value 2 na\xEFve, Spark returns 5 and main returns 7.
-- dim.txt contains the line "1 caf\xE9" (Latin-1)
SELECT /*+ BROADCAST(lt) */ f.v, hex(lt.value), length(lt.value)
FROM fact f JOIN text.`/path/dim.txt` lt ON f.s = substring(lt.value, 1, 1)Setting spark.comet.convert.broadcastBuildSide.enabled=false restores the 1.1.0 result.
2. min/max over a string cast from binary (#6927). min and max on strings fell back in 1.1.0, which kept the CAST below them in Spark too. They are native now, so the cast decodes the bytes before the comparison, and min and max can swap:
CREATE TABLE t(grp string, b binary) USING parquet;
INSERT INTO t VALUES ('a', X'80'), ('a', X'C3A9');
SELECT grp, min(cast(b AS string)), max(cast(b AS string)) FROM t GROUP BY grp;Spark returns min = the 0x80 string and max = é. main returns min = é and max = U+FFFD.
Expected behavior
Either these paths keep the 1.1.0 results by default, or the compatibility guide names them. The guide currently says invalid UTF-8 "does not occur for text read from Parquet or produced by string expressions", which leaves a reader to conclude that a query over Parquet and Text files is unaffected.
Options for 1.2.0
- Keep both as they are and document them. Update the compatibility guide to name Text sources on a broadcast build side and the config that turns the conversion off.
- Default
spark.comet.convert.broadcastBuildSide.enabledtofalsefor 1.2.0, since Text is the source Text-format source on a broadcast build side blocks nativeBroadcastHashJoin#6008 targets and the one most likely to hold these bytes.
The min/max case falls under the "ordering" difference the guide already lists, so documentation covers it unless we want the aggregate to fall back when its input is a cast from binary.
Additional context
Found while auditing the changes since 1.1.0 for regressions before cutting branch-1.2. Part of #4764.
Activity
fix: keep Text build sides on Spark by default and add Text conversio…
fix: keep Text build sides on Spark by default and add Text conversio…
fix: keep Text build sides on Spark by default and add Text conversio…