feat: route translate and to_csv through codegen dispatch by default - #5032
feat: route translate and to_csv through codegen dispatch by default#5032andygrove wants to merge 2 commits into
Conversation
`translate` (StringTranslate) and `to_csv` (StructsToCsv) have native implementations that are incompatible with Spark by default, so today the whole enclosing operator falls back to Spark unless allowIncompatible is set. Route them through the JVM codegen dispatcher by default instead, using the same NativeOptInAvailable pattern as `to_json` and `split`: the dispatcher runs Spark's own generated code (bit-exact) while keeping the operator on the native pipeline, and the faster native path stays available via allowIncompatible. Update the translate SQL test (previously asserting fallback) to verify native dispatch, and add a to_csv test.
7cd0e22 to
3abb614
Compare
andygrove
changed the title
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
First pass, thanks @andygrove!
| val childExprs = expr.children.map(exprToProtoInternal(_, inputs, binding)) | ||
| val optExpr = scalarFunctionExprToProto("translate", childExprs: _*) | ||
| optExprWithFallbackReason(optExpr, expr, expr.children: _*) |
There was a problem hiding this comment.
First, lines 134-136 are a verbatim copy of CometScalarFunction.convert (spark/src/main/scala/org/apache/comet/serde/CometScalarFunction.scala:29-33), so the function name "translate" now lives in two places. Dropping the CometScalarFunction[StringTranslate]("translate") base to hand-inline its body is also a step away from the established idiom for this exact case, which keeps the base and overrides convert to choose a branch: CometInitCap (strings.scala:162-174), CometStringReplace (:192-199), CometCaseConversionBase (:67-78).
Second, and more to the point: the whole getSupportLevel plus convert branching structure is what CodegenDispatchFallback exists to express declaratively. QueryPlanSerde.exprToProtoInternal already routes a non-opted-in Incompatible result from such a serde through emitJvmCodegenDispatch and falls back to Spark only when the dispatcher refuses (QueryPlanSerde.scala:838-870, dispatchIfFallback at :1109-1117), and CodegenDispatchFallback extends NativeOptInAvailable so GenerateDocs renders the same compatible-by-default row (GenerateDocs.scala:193). So the entire diff to this object should be one mixin on the original:
object CometStringTranslate
extends CometScalarFunction[StringTranslate]("translate")
with CodegenDispatchFallback {
private val incompatReason = ...
override def getIncompatibleReasons(): Seq[String] = Seq(incompatReason)
override def getSupportLevel(expr: StringTranslate): SupportLevel = Incompatible(Some(incompatReason))
}That keeps the existing Incompatible declaration honest, keeps the function name in one place, and gets the logWarning on the opted-in path (QueryPlanSerde.scala:841-846) that the hand-rolled version drops.
There was a problem hiding this comment.
You are right on both counts, and the mixin does the whole job. Done in 26994c9 — the entire change to this object is now with CodegenDispatchFallback on the original declaration, exactly as you wrote it. The CometScalarFunction[StringTranslate]("translate") base is restored, so "translate" lives in one place again, and the opted-in logWarning at QueryPlanSerde.scala:841-846 comes back with it. The pre-existing Incompatible(Some(incompatReason)) was already the right classification, so it stays untouched.
| object CometStructsToCsv extends CometCodegenDispatch[StructsToCsv] with NativeOptInAvailable { | ||
|
|
||
| private val incompatibleDataTypes = Seq(DateType, TimestampType, TimestampNTZType, BinaryType) | ||
|
|
||
| override def getIncompatibleReasons(): Seq[String] = Seq( | ||
| "Date, Timestamp, TimestampNTZ, and Binary data types may produce different results" + | ||
| " (https://github.com/apache/datafusion-comet/issues/3232)") | ||
|
|
||
| override def getUnsupportedReasons(): Seq[String] = Seq( | ||
| "Complex types (arrays, maps, structs) in the schema are not supported") | ||
|
|
||
| override def getSupportLevel(expr: StructsToCsv): SupportLevel = { | ||
| // The native ToCsv path only supports non-complex, compatible field types. Everything else | ||
| // (and the default, unless opted in) runs through the codegen dispatcher, which is bit-exact. | ||
| private def nativeSupported(expr: StructsToCsv): Boolean = { | ||
| val dataTypes = expr.inputSchema.fields.map(_.dataType) | ||
| val containsComplexType = dataTypes.exists(DataTypeSupport.isComplexType) | ||
| if (containsComplexType) { | ||
| return Unsupported( | ||
| Some( | ||
| s"The schema ${expr.inputSchema} is not supported because it includes a complex type")) | ||
| } | ||
| val containsIncompatibleDataTypes = dataTypes.exists(incompatibleDataTypes.contains) | ||
| if (containsIncompatibleDataTypes) { | ||
| return Incompatible( | ||
| Some( | ||
| s"The schema ${expr.inputSchema} is not supported because " + | ||
| s"it includes a incompatible data types: $incompatibleDataTypes")) | ||
| } | ||
| // https://github.com/apache/datafusion-comet/issues/3232 | ||
| Incompatible() | ||
| !dataTypes.exists(DataTypeSupport.isComplexType) && | ||
| !dataTypes.exists(incompatibleDataTypes.contains) | ||
| } | ||
|
|
||
| override def getSupportLevel(expr: StructsToCsv): SupportLevel = | ||
| if (!CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeSupported(expr)) { | ||
| Compatible(nativeOptIn = | ||
| Some(NativeOptIn(CometConf.getExprAllowIncompatConfigKey(getExprConfigName(expr))))) | ||
| } else { | ||
| Compatible() | ||
| } | ||
|
|
||
| override def convert( | ||
| expr: StructsToCsv, | ||
| inputs: Seq[Attribute], | ||
| binding: Boolean): Option[ExprOuterClass.Expr] = { | ||
| for { | ||
| childProto <- exprToProtoInternal(expr.child, inputs, binding) | ||
| } yield { | ||
| val optionsProto = options2Proto(expr.options, expr.timeZoneId) | ||
| val toCsv = ExprOuterClass.ToCsv | ||
| .newBuilder() | ||
| .setChild(childProto) | ||
| .setOptions(optionsProto) | ||
| .build() | ||
| ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build() | ||
| if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeSupported(expr)) { | ||
| for { | ||
| childProto <- exprToProtoInternal(expr.child, inputs, binding) | ||
| } yield { | ||
| val optionsProto = options2Proto(expr.options, expr.timeZoneId) | ||
| val toCsv = ExprOuterClass.ToCsv | ||
| .newBuilder() | ||
| .setChild(childProto) | ||
| .setOptions(optionsProto) | ||
| .build() | ||
| ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build() | ||
| } | ||
| } else { | ||
| // Default: route through the codegen dispatcher so Spark's own doGenCode runs inside the |
There was a problem hiding this comment.
The same argument applies here, and more strongly because the pre-existing getSupportLevel already returned exactly the right classifications: Unsupported for complex field types, Incompatible for the #3232 types, Incompatible() otherwise. With with CodegenDispatchFallback on line 262 those three outcomes route as intended with no other change: Unsupported to the dispatcher plus a logDebug (QueryPlanSerde.scala:827-833), non-opted-in Incompatible to the dispatcher plus the [COMET-INFO] opt-in hint (:854-860), opted-in Incompatible to the native proto (:847). That turns a +30/-30 rewrite into a one-line change and removes the duplicated nativeSupported(expr) test that currently has to be kept in sync between :279 and :290.
Deleting getUnsupportedReasons (was structs.scala:270-271) also loses real information. The native path still cannot do complex types, and under NativeOptInAvailable the documented surface for native-path limitations is getIncompatibleReasons (spark/src/main/scala/org/apache/comet/serde/CometExpressionSerde.scala:52-63). Keeping CodegenDispatchFallback preserves both accessors as-is; if you keep the rewrite instead, the complex-type limitation needs folding into getIncompatibleReasons at :266-268, otherwise a user who sets allowIncompatible has no way to find out why they did not get the native path.
There was a problem hiding this comment.
Agreed — reverted to the original three-way getSupportLevel plus the mixin in 26994c9. The +30/-30 rewrite is now one mixin and a comment.
Confirming your reading of the routing, since it is the whole basis for the change: Unsupported (complex field types) reaches the dispatcher plus logDebug at QueryPlanSerde.scala:819-831, non-opted-in Incompatible (#3232 types, and the Incompatible() catch-all) reaches the dispatcher plus the [COMET-INFO] hint at :848-860, and opted-in Incompatible reaches handler.convert — the native proto — at :846. So all three original outcomes route as intended with no other change.
getUnsupportedReasons is restored, and the duplicated nativeSupported predicate is gone along with the sync hazard between the two call sites. I also dropped CometCodegenDispatch as the base: it hardcodes getSupportLevel = Compatible() and convert = emitJvmCodegenDispatch for expressions with no native path at all, so overriding convert to sometimes go native was working against it.
| .setOptions(optionsProto) | ||
| .build() | ||
| ExprOuterClass.Expr.newBuilder().setToCsv(toCsv).build() | ||
| if (CometConf.isExprAllowIncompat(getExprConfigName(expr)) && nativeSupported(expr)) { |
There was a problem hiding this comment.
When isExprAllowIncompat is true and nativeSupported is false, the user asked for the native path and silently gets the dispatcher with nothing recorded. CometStructsToJson has the same hole so this is not new, but it deserves a logDebug naming the disqualifying field type. Using CodegenDispatchFallback gets this for free through the Unsupported arm.
There was a problem hiding this comment.
Resolved by the redesign rather than by adding a logDebug: with CodegenDispatchFallback this case lands in the framework's Unsupported arm, which already logs at debug naming the disqualifying reason (QueryPlanSerde.scala:819-831). So it is covered for free, as you noted.
| CREATE TABLE test_translate(s string, from_str string, to_str string) USING parquet | ||
|
|
||
| statement | ||
| INSERT INTO test_translate VALUES ('hello', 'el', 'ip'), ('hello', 'aeiou', '12345'), ('', 'a', 'b'), (NULL, 'a', 'b'), ('hello', '', ''), ('abc', 'abc', 'x') |
There was a problem hiding this comment.
The four query blocks do assert native execution, so flipping them off expect_fallback is real new coverage of the routing. What is missing is the payload: the header comment at :18-22 names two specific ways the native path diverges, and the six rows on line 28 are ASCII-only and hit neither. The file is now identical to string_translate_enabled.sql in CREATE TABLE, INSERT rows and queries, differing only in the config header, so nothing distinguishes the dispatcher from the native path here. Please add the two rows that are only assertable now that the default is bit-exact: a combining-mark input for the grapheme versus code-point difference, and a to argument containing U+0000 for the deletion-sentinel difference. To keep the fixture ASCII, build them with decode(X'65CC81', 'UTF-8') and decode(X'00', 'UTF-8') rather than embedding the characters. Those two rows are the regression test for this PR.
There was a problem hiding this comment.
Added in 26994c9, built with decode() as you suggested so the fixture stays ASCII:
(concat('caf', decode(X'65CC81', 'UTF-8')), 'e', 'E'), -- "e" + U+0301: one grapheme, two code points
('hello', 'l', decode(X'00', 'UTF-8')) -- U+0000 as the `to` argumentI also checked that they actually discriminate rather than passing vacuously, since that is the point of them. Adding the same two rows to string_translate_enabled.sql, which opts into the native path, fails:
- sql-file: expressions/string/string_translate_enabled.sql *** FAILED ***
Error executing SQL 'SELECT translate(s, from_str, to_str) FROM test_translate_enabled'
Results do not match for query
So the rows pass under the dispatcher and fail under the native path, which makes them a real regression test for the routing. I left string_translate_enabled.sql as it was — its header explicitly says it covers inputs where the two implementations agree, so the divergent rows belong in the default file only.
| private def nativeSupported(expr: StructsToCsv): Boolean = { | ||
| val dataTypes = expr.inputSchema.fields.map(_.dataType) | ||
| val containsComplexType = dataTypes.exists(DataTypeSupport.isComplexType) | ||
| if (containsComplexType) { | ||
| return Unsupported( | ||
| Some( | ||
| s"The schema ${expr.inputSchema} is not supported because it includes a complex type")) | ||
| } | ||
| val containsIncompatibleDataTypes = dataTypes.exists(incompatibleDataTypes.contains) | ||
| if (containsIncompatibleDataTypes) { | ||
| return Incompatible( | ||
| Some( | ||
| s"The schema ${expr.inputSchema} is not supported because " + | ||
| s"it includes a incompatible data types: $incompatibleDataTypes")) | ||
| } | ||
| // https://github.com/apache/datafusion-comet/issues/3232 | ||
| Incompatible() | ||
| !dataTypes.exists(DataTypeSupport.isComplexType) && | ||
| !dataTypes.exists(incompatibleDataTypes.contains) | ||
| } |
There was a problem hiding this comment.
nativeSupported is the new branch predicate driving both getSupportLevel (:279) and convert (:290), and no test takes it false. The two cases it excludes are the reason the change exists:
Date, Timestamp, TimestampNTZ and Binary fields were Incompatible before this PR and fell the operator back to Spark. That is #3232.
Complex field types were Unsupported before and also fell back. They are legal to_csv input: StructsToCsv.checkInputDataTypes accepts arrays, maps and nested structs and rejects only Variant (Spark csvExpressions.scala:240-261 on master and branch-4.0; branch-3.5 has no gate at all). Nothing tests that the kernel can hand converter() a nested InternalRow, which is the new and untried part of this path.
The cheapest place to cover the first group is the existing suite rather than a new fixture. CometCsvExpressionSuite's test named "to_csv - default options" (spark/src/test/scala/org/apache/comet/CometCsvExpressionSuite.scala:36) is actually run under allowIncompatible=true (:50), and it selects c0-c5, c7, c8, c9 and c12 out of the 14-column fuzz schema, skipping exactly c6 Double, c10 Timestamp, c11 TimestampNTZ and c13 Binary (FuzzDataGenerator.scala:303-317). Those are the columns the native path cannot match Spark on. After this PR the dispatcher matches them by construction, so that test can drop the allowIncompatible wrapper and add those four columns, which is what would actually justify the name it already has.
There was a problem hiding this comment.
Took your suggestion for the first group and it works: in 26994c9 CometCsvExpressionSuite's "to_csv - default options" drops the allowIncompatible=true wrapper and now selects the whole fuzz schema rather than a hand-picked subset, so it covers c6 Double, c10 Timestamp and c11 TimestampNTZ. It finally tests what its name says.
c13 Binary I had to leave out, and not because of a divergence. Spark's CSV converter renders BinaryType with Java's default Object.toString(), so a row comes out as:
...,3128558622871195782,[B@731af74
That trailing token is an identity hash of the byte-array instance, so it differs between any two evaluations. With c13 included, every row mismatches and the only differing token is that hash — the double, timestamp and TimestampNTZ columns all match exactly:
![false,-1,-28684,...,3332-12-27T13:01:58.882-08:00,3332-12-15T18:12:14.824,...,[B@731af74]
[false,-1,-28684,...,3332-12-27T13:01:58.882-08:00,3332-12-15T18:12:14.824,...,[B@72f0225a]
So the value is not assertable by any engine, including Spark against itself. I noted that in the test rather than silently dropping the column.
On the complex-type group, the news is worse and I could not write the test. You are right that checkInputDataTypes accepts arrays, maps and nested structs, so they reach the converter — but Spark's output for them is not a comparable value. Probed against Spark 3.5 with Comet disabled entirely (spark.comet.enabled=false), so this is pure Spark:
- non-null values render as Java identity strings:
org.apache.spark.sql.vectorized.ColumnarArray@1ada50f0,ColumnarRow@4dd6f621,ColumnarMap@71e38661 - any null complex value throws
NullPointerExceptioninsideUnsafeWriter.write(UnsafeWriter.java:110), with no Comet frames in the stack
So to_csv over complex types produces either an identity hash or a crash in Spark itself. There is no value-parity assertion to make, and I did not want to add a fixture that pins garbage. I documented the finding in to_csv.sql with the reasoning and the fact that it is Spark behavior.
Worth noting this is not a regression from the PR: before the change these schemas were Unsupported and fell the projection back to Spark, and now they reach the dispatcher, which runs the same Spark doGenCode. The NPE and the identity strings are identical either way — the change only decides whether the rest of the projection stays native. Happy to file a separate issue against Spark's to_csv for the complex-type behavior if you think it is worth tracking on our side.
|
|
||
| -- column struct: values with delimiters and quotes exercise Spark's CSV quoting rules | ||
| query | ||
| SELECT to_csv(named_struct('a', a, 'b', b, 'c', c)) FROM test_to_csv |
There was a problem hiding this comment.
options is the main axis of behavior difference for this expression and nothing in the new fixture varies it. CometCsvExpressionSuite:71-161 covers delimiter, escape, quoteAll and nullValue but only under allowIncompatible=true, so the dispatcher path has no options coverage anywhere. to_csv(s, map('sep', ';')) plus a timestampFormat over a timestamp field would cover both this and the #3232 types in one query.
There was a problem hiding this comment.
Added in 26994c9. to_csv.sql now covers map('sep', ';') over the existing rows (which also changes which values need quoting), plus a timestampFormat/dateFormat query over a new date+timestamp table — so the same query exercises both the options axis and the #3232 types that previously fell back. The dispatcher path had no options coverage anywhere before this.
Replace the hand-rolled getSupportLevel/convert branching with the `CodegenDispatchFallback` mixin, which is what the framework already provides for exactly this shape. `exprToProtoInternal` routes an `Unsupported` result to the dispatcher plus a logDebug, a non-opted-in `Incompatible` to the dispatcher plus the [COMET-INFO] opt-in hint, and an opted-in `Incompatible` to the native proto with a logWarning. For `CometStringTranslate` that makes the whole diff one mixin on the original declaration: the pre-existing `Incompatible(Some(incompatReason))` was already the right classification. This also restores the `CometScalarFunction` base, so the function name "translate" stops being duplicated and the opted-in logWarning comes back. For `CometStructsToCsv` the original three-way `getSupportLevel` was likewise already correct -- `Unsupported` for complex field types, `Incompatible` for the apache#3232 types, `Incompatible()` otherwise -- so the +30/-30 rewrite collapses to the mixin. That restores `getUnsupportedReasons`, drops the duplicated `nativeSupported` predicate that had to stay in sync between two call sites, and picks up the logDebug on the opted-in-but-unsupported path for free. Also drops `CometCodegenDispatch` as the base, which is for expressions with no native path at all. Tests: - string_translate.sql gains the two rows that actually distinguish the paths: "e" + U+0301 (one grapheme, two code points) and U+0000 as the `to` argument. Built with decode(X'65CC81') / decode(X'00') to keep the fixture ASCII. Verified these discriminate: adding them to string_translate_enabled.sql, which opts into the native path, fails. - to_csv.sql gains options coverage (`sep`, and `timestampFormat`/`dateFormat` over date and timestamp fields), which the dispatcher path had nowhere. - CometCsvExpressionSuite "to_csv - default options" drops the allowIncompatible=true wrapper and covers c6 Double, c10 Timestamp and c11 TimestampNTZ, which it previously had to skip. It now tests what its name says. Two field types are deliberately left unasserted, both Spark behavior rather than Comet's, confirmed against Spark 3.5 with Comet disabled: - Binary: Spark's CSV converter renders it with Java's default toString, e.g. "[B@731af74", an identity hash that differs between any two evaluations, so it is not comparable by any engine including Spark against itself. - Complex types: a non-null array/map/struct renders as "org.apache.spark.sql.vectorized.ColumnarArray@1ada50f0" and any null complex value throws NullPointerException inside Spark's UnsafeWriter.write. Before this change these schemas fell back to Spark; now they reach the dispatcher, which runs the same Spark code, so the observable result is unchanged.
|
@mbutrovich all six threads addressed in 26994c9, details inline. You were right about the central point, and taking it shrank the PR considerably. The whole serde diff is now two mixins.
Tests. Two things I could not assert, both Spark behavior — verified with
Neither is a regression here: those schemas previously fell back to Spark and now reach the dispatcher, which runs the same Verified on spark-3.5: One unrelated note: this branch is behind |
2 participants
Which issue does this PR close?
Closes #.
Rationale for this change
translate(StringTranslate) andto_csv(StructsToCsv) have native implementations that are incompatible with Spark by default, so today the whole enclosing operator falls back to Spark unless allowIncompatible is set. Routing them through the JVM codegen dispatcher instead keeps the operator on the native pipeline while producing bit-exact Spark results, and preserves the faster native path as an opt-in.What changes are included in this PR?
translate(StringTranslate) andto_csv(StructsToCsv) through the JVM codegen dispatcher by default, using the sameNativeOptInAvailablepattern asto_jsonandsplit. The dispatcher runs Spark's own generated code (bit-exact) while keeping the operator on the native pipeline; the native path stays available viaallowIncompatible.to_csv.How are these changes tested?
string_translateSQL test (previously asserting fallback) to verify native codegen dispatch.to_csvSQL file test.