[andygrove]

Which issue does this PR close?

Closes #7118.

Rationale for this change

#6786 made Comet convert a Text scan on a broadcast join's build side to Arrow by default, where 1.1.0 left it on Spark. Spark's Text reader keeps the bytes of each line as they are, and Arrow strings must be valid UTF-8, so Comet replaces each ill-formed sequence with U+FFFD. For a Latin-1 file, hex, length, hashing, equality and ordering can then disagree with Spark, so 1.2.0 could return different results from 1.1.0 for the same query.

The other formats that reach the bridge don't have this problem: ORC, Avro and Iceberg define strings as UTF-8, and Parquet, CSV and JSON have their own conversion configs that the bridge never overrides. So instead of turning the bridge off for every format, this keeps it on by default and makes Text follow the same model as CSV, JSON and Parquet: a Text scan, on a build side or anywhere else, is converted only by its own config. The compatibility guide also said that invalid UTF-8 "does not occur for text read from Parquet", and it left out some of the paths through which such strings reach Comet.

What changes are included in this PR?

• New spark.comet.convert.text.enabled (default false) converts every Text scan, V1 and V2, like spark.comet.convert.csv.enabled, json and parquet. spark.comet.convert.broadcastBuildSide.enabled stays on by default, as on main, but a Text build side is bridged only when spark.comet.convert.text.enabled is set. When the bridge skips a Text build side, the scan records a fallback reason that names that config.

• Text goes through isConversionEnabled, like Range, InMemoryTableScan, RDDScan and OneRowRelation. Naming FileSourceScan or BatchScan in spark.comet.sparkToColumnar.supportedOperatorList still converts Text, as in 1.1.0, but is deprecated and logs the same warning. The config docs, datasources.md and the migration guide say so.

• compatibility/index.md: the "Strings with non-UTF-8 bytes" section lists the ways such strings reach Comet:

  • Text conversion. CSV and JSON are not on the list, because Spark decodes CSV and treats JSON lines with such bytes as malformed.
  • Comet columnar shuffle over rows from Spark.
  • CAST(binary AS STRING). For min/max over such a cast, the guide gives the configs that keep the result in Spark.

  The Parquet sentence now says that Comet's native Parquet scan never produces such strings.

• datasources.md: a Text section for the new config, and the build-side paragraph says Parquet, JSON, CSV and Text build sides are converted only by their own switches.

• The Text-format source on a broadcast build side blocks native BroadcastHashJoin #6008 build-side tests move from CometExecSuite into CometBroadcastBuildSideSuite, in their own commit with unchanged bodies. A later commit switches them from Text to ORC, because with a per-format Text switch a Text probe would be converted too and the "leaves no bridge" assertions would stop testing anything. ORC is still an unsupported format that the bridge converts.

How are these changes tested?

• A CometBroadcastBuildSideSuite test for Text broadcast build sides and string min/max apply the invalid UTF-8 decode where 1.1.0 stayed on Spark #7118 reads a Latin-1 Text build side through V1 and V2, with AQE on and off:
  • By default the join stays on Spark, hex and length match Spark, and the fallback reason is reported.
  • With spark.comet.convert.text.enabled, the build side is converted and returns the decoded values that the compatibility guide quotes.
• The Text-format source on a broadcast build side blocks native BroadcastHashJoin #6008 tests in CometBroadcastBuildSideSuite run against an ORC build side.
• A CometExecSuite test, "read Text file", checks that a Text scan is not converted by default and that spark.comet.convert.text.enabled and the sparkToColumnar operator list convert it, for V1 and V2.
• A throwaway probe on Spark 4.1 checked each new statement in the compatibility guide against Spark:
  • With default configs, repartitioning a Latin-1 Text table returns EFBFBD where Spark keeps E9.
  • With spark.comet.expression.Min.enabled=false and spark.comet.expression.Max.enabled=false, min/max over CAST(binary AS STRING) still swaps across two input partitions, because the columnar shuffle decodes the partial results. Adding spark.comet.shuffle.mode=native gives Spark's answer.
  • Spark's CSV reader returns 636166EFBFBD for a Latin-1 caf\xE9. Its JSON reader returns null for the field and keeps the raw bytes only in _corrupt_record.
• npx prettier --check passes on the docs files.

[sunchao]

Reviewed the entire five-file diff from 3355602e268819750884aa0ce0dd47002c6ea580 to 18707d2dcc6bed4d5470e156f1fc5d565852c5a3. The PR is not a draft. Followed AGENTS.md and review-comet-pr. No sibling skill applies to these configuration, documentation and test changes. The snapshot and live discussion contained no existing review concerns.

Summary

• Prior state and problem: Automatic broadcast build-side conversion was enabled by default. Converting Text containing invalid UTF-8 could replace original bytes and change query results.
• Design approach: Default the existing bridge configuration to false, explain the compatibility boundary, and explicitly enable conversion in bridge tests.
• Correctness: The planner checks the configuration before tagging and converting scans, including scans retaining old tags. One P2 issue is reproduced: two existing tests still require bridging without enabling it.
• Compatibility analysis: Checked Spark sources for 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Text readers preserve bytes, binary-to-string casts retain them, and binary string ordering explains the documented examples. Compared with branch-1.1, this restores the earlier default Text build-side fallback. The string min/max divergence already exists at the supplied base.
• Key design decisions: Preserve explicit opt-in and document the unchanged binary-cast aggregation behavior. Other conversion settings retain their existing behavior.
• Implementation sketch: Change one configuration default and its description, update two guide pages, enable the bridge in CometBroadcastBuildSideSuite, and introduce scoped bridgeTest registrations in CometExecSuite.
• Performance: No new per-row work or allocation is introduced. Default fallback avoids conversion but forgoes native-join acceleration for affected sources. No additional performance defect was established.
• Design: Reusing the existing gate makes the conservative default straightforward to understand. The documentation explains the correctness tradeoff.
• Abstraction & complexity: The small by-name bridgeTest helper appropriately scopes configuration to test execution. Its adoption is incomplete, causing the reported failures.
• Behavioral changes worth calling out: Automatic conversion becomes opt-in for unsupported file and V2 build scans, including Text, ORC and non-native Iceberg scans. Explicitly enabled conversion continues to decode invalid UTF-8.
• Suggested improvements: Apply bridgeTest to the two remaining tests that require native bridging, while retaining the default-off assertion.

Exact-head CI: run 39358471685 had 19 successful checks, six running checks and 17 skipped checks at final inspection. No completed check reported failure. The four Comet test shards and TPC-H/TPC-DS checks were still running. Spark SQL, Iceberg and macOS checks were skipped.

Validation: Reactor test-compile passed. On Spark 4.1.3/JDK 21, the two focused tests failed with the new default and both passed with the bridge explicitly enabled. Tests used exact-head JVM sources and the base CI native artifact, whose native/common source trees are unchanged. Bounded Spark UTF8String and Comet decoder runs confirmed the documented byte, length and ordering examples. The full Comet matrix and upstream Spark SQL/Iceberg suites were not run locally.

[sunchao]

[P2] Enable the remaining bridge-dependent tests explicitly. The Text nested-loop join test and nested native-join test in CometExecSuite (registrations at lines 616 and 806) still use plain test. With this new default, their Text scans correctly remain on Spark, but their assertions require a bridge and native join. Both fail on Spark 4.1.3/JDK 21 at lines 633 and 835. Explicitly enabling the configuration makes both pass. This breaks the CI exec suite. Please register these tests through bridgeTest while keeping the production default off.

Evidence: Reproduced with ./mvnw test -Dtest=none '-Dsuites=org.apache.comet.exec.CometExecSuite @broadcast nested loop join build side with a text source goes native (#6008),org.apache.comet.exec.CometExecSuite @a native broadcast join nested in a Spark-staying join keeps its bridge (#6008)' using the local build settings described above. Default configuration: 0 passed, 2 failed, with expected the text build side to be bridged and the inner join should be a CometBroadcastHashJoinExec. Repeating the same selection with JAVA_TOOL_OPTIONS=-Dspark.comet.convert.broadcastBuildSide.enabled=true produced 2 passed, 0 failed.

[viirya]

Thanks for picking this up before the 1.2.0 branch cut. Defaulting the build-side bridge to off is the simplest way to get the 1.1.0 results back for Text build sides. The gate is checked both when tagging and when bridging, so a leftover tag can't bring the conversion back. The compatibility guide also needed the more honest wording about where invalid UTF-8 comes from.

On the tests, sunchao already pointed out the two CometExecSuite tests that still need the bridge, and the [exec] shard fails on exactly those two. Beyond that, five more #6008 tests keep passing but stop testing anything now that the default is off. Details are inline on bridgeTest.

The new compatibility section needs a few corrections, also inline. The Parquet sentence contradicts the #4121 paragraph right below it. CSV and JSON don't carry invalid UTF-8 into Comet, because Spark has already decoded them. The list leaves out Comet columnar shuffle, which is the default way a Text table's bytes reach the decode. And there is a per-expression config that looks like a narrower workaround for the min/max case than turning Comet off.

[viirya]

Building on sunchao's comment: besides the two tests that now fail, five more #6008 tests keep passing but no longer test anything.

• broadcast join with an unsupported probe stays on Spark
• broadcast join that stays on Spark leaves no build-side bridge
• Spark-staying broadcast join with an aggregate build side leaves no bridge
• Spark-staying broadcast join with a union build side leaves no bridge
• Spark 3.4 AQE DPP with a V1 Parquet fact and a text dimension

Each one asserts that nothing is bridged or tagged. With the default off, tagBroadcastBuildSideLeaves returns early at CometExecRule.scala:1577, so those assertions hold before hasOnlyNativeScans or clearWastedBuildSideBridges ever runs. The filter/aggregate/union revert tests came out of the #6786 review rounds, so it would be a shame to lose them without anyone noticing. Could these use bridgeTest too?

Another option is moving the #6008 tests into CometBroadcastBuildSideSuite, which already turns the bridge on for the whole suite. Then this helper can go away, and a test added later can't pick up the wrong default by accident.

[viirya]

"which Spark requires to be valid UTF-8" conflicts with the last paragraph of this section. That paragraph says the native Parquet scan rejects such columns "rather than reading them like Spark" (#4121). Spark's Parquet reader wraps the stored bytes with UTF8String.fromBytes and doesn't validate them. Can we drop that clause, or say that Comet's native Parquet scan never produces such strings?

[viirya]

I don't think CSV and JSON belong here. Spark's CSV path decodes each line with new String(bytes, charset) and builds the value with UTF8String.fromString. JSON goes through an InputStreamReader, or through Jackson's byte parser, which rejects invalid UTF-8. So a Latin-1 CSV or JSON file has already been decoded to U+FFFD by Spark before the rows reach Comet, and Comet's decode doesn't change the result. Text keeps the raw bytes, which is why it's the case that matters. Could this bullet name Text, and other sources that wrap bytes without validating them, instead?

Separately, the list reads as the complete set of paths to check, but a Text scan's rows can also reach the decode through Comet columnar shuffle with the default settings. In auto mode, columnarShuffleFailureReasons accepts a row-based Spark child, and the row-to-Arrow writer decodes strings with decode_utf8_spark_lossy in get_string. That code has been there since 1.1.0 (#4524). So a GROUP BY or sort-merge join over a Latin-1 Text table would see U+FFFD on both releases. The opening paragraph of this section already mentions columnar shuffle. Could the list include it, so readers don't conclude that turning off the build-side switch is enough to keep Text bytes intact?

[viirya]

"is not controlled by a config; disable Comet for the query" seems broader than needed. The aggregate serde checks spark.comet.expression.<Name>.enabled (QueryPlanSerde.scala:830). Would setting spark.comet.expression.Min.enabled=false and spark.comet.expression.Max.enabled=false keep the aggregate, and the cast inside its argument, in Spark for this case? If so, that's a much narrower workaround to document.

[andygrove]

@sunchao @viirya thanks for the reviews. I've reworked this PR based on them, so it needs a fresh look.

Instead of turning the build-side bridge off for every format, the bridge stays on by default and Text is the exception. Of Spark's built-in sources that reach the bridge, Text is the only one whose strings carry no UTF-8 guarantee. ORC, Avro and Iceberg define strings as UTF-8, and Parquet, CSV and JSON never reach the bridge because they have their own spark.comet.convert.* configs. A Text build side is now bridged only with the new spark.comet.convert.broadcastBuildSide.text.enabled (off by default). Without it, the scan reports a fallback reason that names the config. For consistency with CSV, JSON and Parquet, there is also a new spark.comet.convert.text.enabled that converts every Text scan.

Tests:

• The Text-format source on a broadcast build side blocks native BroadcastHashJoin #6008 tests now live in CometBroadcastBuildSideSuite, which turns the Text opt-in on for the whole suite, as viirya suggested. That covers the two tests that failed and the five that had stopped testing anything, and the bridgeTest helper is gone. The move is its own commit (cd42aeb). The bodies are unchanged apart from bridgeTest( becoming test(.
• A new Text broadcast build sides and string min/max apply the invalid UTF-8 decode where 1.1.0 stayed on Spark #7118 test covers the default, the Text opt-in and spark.comet.convert.text.enabled with a Latin-1 file, through V1 and V2, with AQE on and off.

Compatibility guide: I took all four of viirya's points and checked each one with a probe against Spark 4.1.

• The Parquet sentence now says that Comet's native Parquet scan never produces such strings.
• The bullet names Text rather than CSV and JSON. Spark decodes CSV. For JSON, a line with such bytes is malformed, so the field comes back null and only _corrupt_record keeps the bytes. The guide now says that.
• Columnar shuffle is on the list now. With default configs, repartitioning a Latin-1 Text table returns EFBFBD where Spark keeps E9. The GROUP BY example matched Spark in my probe only because Comet left that shuffle on Spark, since its parent aggregate stayed on Spark.
• The per-expression configs alone don't fix min/max. With spark.comet.expression.Min.enabled=false and spark.comet.expression.Max.enabled=false, the aggregate runs in Spark, but the columnar shuffle between the partial and final aggregates still decodes the partial results. With two input partitions, min and max still swap. Adding spark.comet.shuffle.mode=native gives Spark's answer, so the guide lists all three.

I've updated the PR description and added run-spark-4.1-tests, since this changes the planner.

[sunchao]

Reviewed commit e8fe7583bc8d70b3a2646752ed38ed85839a0273. New P1/P2 findings are detailed inline.

[sunchao]

[P2] Remove the stale QueryStageExec import from CometExecSuite.scala:42 when moving this assertion. This relocation removes its final use in that suite, but leaves the import behind. Running the standard Spark 3.5 strict-warning build now fails during test compilation with Unused import, and all six Java-lint jobs fail Scalafix on the same import. Removing it from the original suite restores the build checks while retaining this assertion.

Evidence: Exact-head CI reproduced the failure with ./mvnw -B test-compile -Pspark-3.5,delta -Pstrict-warnings -DskipTests: CometExecSuite.scala:42: Unused import, followed by one error found and BUILD FAILURE. Strict compilation log: https://github.com/apache/datafusion-comet/actions/runs/39414444629/job/118601923006. The Spark 4.1 lint log explicitly proposes removing only QueryStageExec from that import: https://github.com/apache/datafusion-comet/actions/runs/39414444629/job/118601923427. All six lint-job logs show the same fix.

[andygrove]

Good catch, thanks. The move took its last real use, and the two remaining mentions are in a test name and a comment. Fixed in 67dc4c1 by dropping QueryStageExec from the import. Locally, the strict-warnings test-compile on Spark 3.5 (-Pspark-3.5,delta -Pstrict-warnings) and the Scalafix check on 4.1 now pass.

[andygrove]

@parthchandra pinging you as well, since this changes the default for the case in #6008.

The build-side bridge from #6786 stays on by default, but a Text build side is now bridged only when spark.comet.convert.broadcastBuildSide.text.enabled=true. Spark's Text reader keeps the raw bytes of each line, and the conversion replaces bytes that are not valid UTF-8 with U+FFFD, so a Latin-1 lookup file could return different results from 1.1.0 (#7118). That means an allowlist read from a Text file, like the one in #6008, now needs that setting before the join runs natively. A new spark.comet.convert.text.enabled converts every Text scan, if that suits you better. Other build sides that Comet can't read natively, such as ORC or an Iceberg table the native reader declines, are still bridged by default.

Does this work for your use case? I'd appreciate a review.

[viirya]

Thanks for the rework. Keeping the bridge on and carving out Text is a good fit for the problem in #7118. The gate goes through isBroadcastBuildSideLeaf, so a stale tag can't bring the conversion back, and the fallback reason makes the new opt-in easy to discover from the explain output. Moving the #6008 tests into CometBroadcastBuildSideSuite also fixes the two failing tests and the five that had stopped testing anything, without needing a helper. The compatibility guide changes cover everything raised in the last round, and the probe results behind each statement are appreciated.

I have three things inline.

• The PR adds two Text configs where one seems enough. If Text followed the per-format model that CSV, JSON and Parquet already use, the planner branch, the config docs and the guide would all get simpler.
• If spark.comet.convert.text.enabled stays, two existing docs still describe FileSourceScan in sparkToColumnar as the way to convert formats other than Parquet, JSON and CSV. Text also keeps that path without the deprecation warning that the other leaves with their own spark.comet.convert config log.
• The min/max workaround in the guide seems to hold only when the cast is the aggregate's own argument.

[viirya]

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.

[andygrove]

Yes, this makes sense. I think the extra config is confusing.

[andygrove]

No, I don't see a workload that needs it. Converting the build side while every other Text scan stays on Spark is only useful if the build side is the one place a Latin-1 file is safe, and nothing makes that so. I dropped spark.comet.convert.broadcastBuildSide.text.enabled. A Text scan, including one on a build side, is now converted only by spark.comet.convert.text.enabled, like CSV, JSON and Parquet. The fallback reason now names that config. The datasources paragraph says "Parquet, JSON, CSV and Text build sides are converted only by their own switches". The Latin-1 explanation now lives in the spark.comet.convert.text.enabled doc and the compatibility guide. The #6008 allowlist case sets spark.comet.convert.text.enabled=true. @parthchandra, please say if that doesn't work for you.

This also changed the #6008 tests. With a global Text switch they can't use Text any more, because a Text probe would be converted too and the "leaves no bridge" assertions would stop meaning anything. They now use ORC, which is still an unsupported format that gets bridged. The #7118 test keeps the Latin-1 Text file and covers the default and spark.comet.convert.text.enabled, for V1 and V2 with AQE on and off.

[parthchandra]

@andygrove sorry for the tardy response. Yes, this change would be fine.

[viirya]

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.

[andygrove]

Good catch on both. Text now goes through isConversionEnabled, like Range, InMemoryTableScan, RDDScan and OneRowRelation. Naming FileSourceScan or BatchScan for a Text scan still converts it, but logs the same deprecation warning. The supportedOperatorList doc and the "To convert any other leaf operator" paragraph in datasources.md now say "other than Parquet, JSON, CSV and Text". Both also say that naming those operators to convert Text is deprecated. Naming FileSourceScan for ORC and other formats without a config of their own is unchanged.

[viirya]

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?

[andygrove]

Right, the probe only covered the cast as the direct argument of min and max. With the cast in a projection below the aggregate, as in your example, Comet runs that projection natively and decodes b before the Spark aggregate sees it, so the three configs do not help. In 0e66bc0 the guide keeps disabling Comet as the general answer, gives the three configs only for the case where the cast is the direct argument of min or max, and names the projection shape as one they do not cover.

[sunchao]

Reviewed commit 510496429ab8f2634e4d2b91641cafccbbb5a56d. New P1/P2 findings are detailed inline.

[sunchao]

[P2] [P2] Remove the unused SparkConf import. This revision removes the suite’s sparkConf override, leaving this import without a use. All six exact-head Java-lint jobs now fail Scalafix, and compiling the suite with unused-import warnings promoted to errors also fails. The build should remain compilable after the override is removed. Please drop this import so the required checks can pass.

Evidence: Exact-head Spark 4.2 CI ran ./mvnw -B package -DskipTests scalafix:scalafix -Dscalafix.mode=CHECK -Psemanticdb -Pspark-4.2 and proposed removing precisely import org.apache.spark.SparkConf before failing: https://github.com/apache/datafusion-comet/actions/runs/39919227954/job/120301609004. The other five Java-lint logs show the same removal. Local Scala 2.13.17 compilation of the exact suite with -Ywarn-unused:imports -Xfatal-warnings exited 1 with CometBroadcastBuildSideSuite.scala:26: warning: Unused import. A disposable copy removing only that line compiled successfully.

[andygrove]

Fixed in 0e66bc0 by dropping the import. The same CI run also failed Spotless on a comment in CometExecRule.shouldConvertTextScan, which that commit reformats. A local compile test-compile scalafix:scalafix -Psemanticdb and spotless:apply now leave the tree unchanged.

[comphead]

The test move to CometBroadcastBuildSideSuite is faithful, since the bodies match the removed ones apart from swapping Text for ORC. A default-off spark.comet.convert.text.enabled named like the CSV, JSON and Parquet ones fits config_conventions.md. The title still says "add Text conversion configs" but after the rework there is one config, so the squash commit would describe one that does not exist. One note outside the diff. the deprecated sparkToColumnar settings warn once for each config they stand in for in CometExecRuleSuite covers the other conversion configs but not spark.comet.convert.text.enabled, and read Text file only checks that the deprecated path still converts, not that it warns. The migration guide promises that warning for Text, so a Text case there would pin it. The warning text uses Utils.getSimpleName(op.getClass), so for a Text scan it names FileSourceScanExec, the same name as a scan the docs say is not deprecated for ORC, and the case could also check that the message names the Text config.

[viirya]

Thanks for the quick turnaround. All three points from my last review are addressed. Text now follows the same model as CSV, JSON and Parquet, so the planner branch is a single check plus the fallback reason, and the config docs and the guide got simpler with it. Routing Text through isConversionEnabled gives it the same deprecation path as the other leaves, and the docs now say so in all three places. The min/max paragraph is scoped correctly now. Switching the #6008 tests to ORC makes sense too, since a global Text switch would also convert a Text probe and empty out the "leaves no bridge" assertions. The #7118 test also checks the decoded values with spark.comet.convert.text.enabled now, which ties the test to the numbers the guide quotes.

+1 to @comphead on the title. With one config left, it should be updated before merge so the squash commit matches the change. His suggestion to cover the Text deprecation warning in CometExecRuleSuite is worth adding as well. Approving with those in mind.

[parthchandra]

Thanks for this fix @andygrove !