fix: reject casts involving non-default collated strings

[stantheman0128]

Which issue does this PR close?

Closes #4489.

Rationale for this change

Spark 4.0 carries collation metadata on StringType, but serializeDataType maps every StringType to a single proto type id (QueryPlanSerde.scala:565), so the collation is dropped on the way into the native plan with no warning. Nothing in CometCast stopped that from happening.

CometCast.isSupported matches string casts through case (DataTypes.StringType, _) and case (_, DataTypes.StringType). Those only fail to match a collated StringType because DataTypes.StringType is the default-collation singleton and Scala pattern equality compares the whole instance. The right answer fell out of an accident of pattern matching rather than a check anyone wrote, which is what the issue title means by "implicit". The fromType == toType shortcut at CometCast.scala:189 ran ahead of all of it, so identity casts on collated types were reported Compatible() regardless. CAST(ARRAY<STRING COLLATE UTF8_LCASE> AS ARRAY<STRING COLLATE UTF8_LCASE>) is the clearest example. Results are right today because the cast is a byte-level no-op, but the plan reached the native side with the collation stripped and nothing recording that.

Seven other places already use CometTypeShim.hasNonDefaultStringCollation for this exact purpose (arrays.scala, collectionOperations.scala, datetime.scala, maps.scala, predicates.scala, strings.scala, CometExprShim4x.scala). CometCast mixes in CometTypeShim already and simply never called it.

What changes are included in this PR?

Adds the guard to CometCast.isSupported, above the fromType == toType shortcut so identity casts are checked too. hasNonDefaultStringCollation walks nested element, key, value, and field types, and is stubbed to false on Spark 3.x where collation does not exist, so the guard compiles away to a constant on the 3.x profiles.

Adds CometCastCollatedStringSuite under spark/src/test/spark-4.x, which every 4.x profile compiles. It covers the scalar matrix in both directions and between two collations, the nested cases (array element, struct field, map key, map value, and the array-to-string recursion), and it checks that default-collation casts are still Compatible so the guard cannot quietly over-block.

Three pairs that the guard newly blocks are worth calling out, because they were Compatible before and are not identity casts. A struct whose collated field is unchanged while a sibling field is cast came out Compatible, because the field zip answered per field and the collated field hit the identity shortcut. MapType(lcase, IntegerType) -> MapType(lcase, LongType) did the same through the key. ArrayType(NullType) -> ArrayType(lcase) was Compatible through the elementType == NullType branch, which runs ahead of everything else. Each of those let a collated type reach the native plan on a sibling's cast, and each now has a test.

Three end-to-end tests run a query and assert what the planner does with the answer. Two use CAST(_1 COLLATE utf8_lcase AS INT) under both settings of spark.comet.exec.scalaUDF.codegen.enabled. The third casts a struct carrying a collated field, which is the shape that actually exercises the new guard end to end. A scalar identity cast cannot be reached from SQL, because Spark's SimplifyCasts drops a cast whose child already has the target type, and there is a comment in the suite recording that.

One correction to an earlier revision of this PR. Unsupported does not mean the query falls back to Spark. CometCast mixes in CodegenDispatchFallback, so exprToProtoInternal offers the expression to the JVM codegen dispatcher before recording any fallback reason (QueryPlanSerde.scala:869-886). spark.comet.exec.scalaUDF.codegen.enabled defaults to true and CometBatchKernelCodegen admits ResolvedCollation, so under default config a collated cast usually stays inside the Comet pipeline running Spark's own doGenCode. The test names and the suite Scaladoc now say the cast has no native path instead.

Registers the suite in pr_build_linux.yml and pr_build_macos.yml.

How are these changes tested?

Run on Linux against a debug libcomet.so built from this branch. On the spark-4.1 profile (Spark 4.1.3, Scala 2.13.17, JDK 17):

$ ./mvnw -B -Pspark-4.1 test -Dsuites="org.apache.comet.CometCastCollatedStringSuite" -Dtest=none -pl spark

CometCastCollatedStringSuite:
- cast collated string to IntegerType has no native path (22 milliseconds)
- cast IntegerType to collated string has no native path (1 millisecond)
- cast collated string to default-collation StringType has no native path (2 milliseconds)
- cast default-collation StringType to collated string has no native path (1 millisecond)
- cast between two different collations has no native path (82 milliseconds)
- cast collated string to the same collation has no native path (1 millisecond)
- cast array of collated strings to another collation has no native path (2 milliseconds)
- cast array of collated strings to the same collation has no native path (1 millisecond)
- cast array of collated strings to StringType has no native path (0 milliseconds)
- cast struct with a collated field has no native path (7 milliseconds)
- cast map with a collated key has no native path (1 millisecond)
- cast map with a collated value has no native path (1 millisecond)
- cast struct whose collated field is unchanged while a sibling field is cast (1 millisecond)
- cast map whose collated key is unchanged while the value type is cast (1 millisecond)
- cast array of nulls to array of collated strings has no native path (1 millisecond)
- default-collation string casts are untouched by the collation guard (3 milliseconds)
- nested default-collation string casts are untouched by the collation guard (1 millisecond)
- cast from a collated string falls back to Spark when codegen dispatch is off (7 seconds, 295 milliseconds)
- cast from a collated string routes through the codegen dispatcher when it is on (630 milliseconds)
- cast of a struct carrying a collated field has no native path end to end (555 milliseconds)
Run completed in 13 seconds, 115 milliseconds.
Tests: succeeded 20, failed 0, canceled 0, ignored 0, pending 0

The guard is production code, so the suites in its blast radius were run as well. On spark-4.1:

$ ./mvnw -B -Pspark-4.1 test -Dtest=none -pl spark \
    -Dsuites="org.apache.comet.CometNativeCastSuite,org.apache.spark.sql.CometCollationSuite,org.apache.comet.CometSqlFileTestSuite"

Run completed in 18 minutes, 6 seconds.
Total number of tests run: 631
Tests: succeeded 631, failed 0, canceled 0, ignored 7, pending 0

Per-suite from the surefire XML: CometNativeCastSuite 175 tests, CometCollationSuite 19 tests, CometSqlFileTestSuite the remainder, all with failures="0" errors="0".

And on spark-4.0 (Spark 4.0.4, JDK 17), which is where the full CometCollationSuite including the #4051 join tests lives:

$ ./mvnw -B -Pspark-4.0 test -Dtest=none -pl spark \
    -Dsuites="org.apache.comet.CometCastCollatedStringSuite,org.apache.comet.CometNativeCastSuite,org.apache.spark.sql.CometCollationSuite"

Run completed in 6 minutes, 6 seconds.
Tests: succeeded 212, failed 0, canceled 0, ignored 7, pending 0

CometCastCollatedStringSuite 20 tests, CometNativeCastSuite 175 tests, CometCollationSuite 24 tests, all failures="0" errors="0".

mvn spotless:check -Pspark-4.1 -pl spark,common passes with the spotless index removed first, so the result is a real check rather than a cache hit: 407 Scala files and 54 Java files, 0 needing changes.

Scope of what was run locally: the spark-4.0 and spark-4.1 profiles. spark-3.4, spark-3.5, and spark-4.2 were not run here, so CI is the first place those execute. On the 3.x profiles the guard is inert by construction, because CometTypeShim.hasNonDefaultStringCollation is a false literal there (spark/src/main/spark-3.x/org/apache/comet/shims/CometTypeShim.scala:31). CometSqlFileTestSuite was run on 4.1 only, not on 4.0.

spark/src/test/resources/sql-tests/expressions/string/collation.sql is the largest existing test in the blast radius, since it casts a default-collation column to a collated target on nearly every query. Those pairs keep the same answer and the same reason string under the guard, because the source type there is the default StringType and unsupported(fromType, toType) formats to the same text either way.

[andygrove]

Thanks for picking this up. The analysis of why StringType equality produces the fallback today is accurate and clearly written, and the writeup in the PR description made this easy to follow. A few things I'd like to work through before merge.

Unsupported on CometCast does not mean "falls back to Spark"

CometCast mixes in CodegenDispatchFallback (CometCast.scala:38). In QueryPlanSerde.scala:869-886, an Unsupported support level first goes to dispatchIfFallback, and the fallback reason is only recorded if the dispatcher declines. spark.comet.exec.scalaUDF.codegen.enabled defaults to true (CometConf.scala:364), and CometBatchKernelCodegen deliberately admits ResolvedCollation (CometBatchKernelCodegen.scala:156-157), so on default config a collated cast most likely stays inside the Comet pipeline running Spark's own doGenCode rather than falling back.

That outcome is still result-correct, so this is not a Comet bug. But every test name in the suite says "falls back", and the PR description says Comet "correctly falls back to Spark", and neither is quite right. Could those be reworded to say the cast has no native path?

Could these live in CometCollationSuite?

CometCollationSuite (spark/src/test/spark-4.0/org/apache/spark/sql/CometCollationSuite.scala) is already the home for collation fallback tests across #1947, #4051, and #4646. Reusing it would also avoid a new registration in two workflow files.

If the reason for a separate suite is that CometCollationSuite lives in spark-4.0 and so does not run on the 4.1 or 4.2 profiles, that is a good catch. In that case, would moving CometCollationSuite itself to spark-4.x be the better change? That gets the whole existing collation suite running on 4.1 and 4.2 as well.

End-to-end coverage is what the issue asked for, and it looks reachable

Issue #4489 asks for tests asserting that the cast falls back and does not run native, and this suite only exercises isSupported in isolation. The datetime tests in CometCollationSuite show that expression-level collation is reachable end to end from a plain-string Parquet column, and that the serde's reason surfaces because getSupportLevel runs before children are serialized. Would something like this work?

withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") {
  checkSparkAnswerAndFallbackReason(
    "SELECT CAST(_1 COLLATE utf8_lcase AS INT) FROM tbl",
    "Cast from StringType(UTF8_LCASE) to IntegerType is not supported")
}

That verifies the runtime outcome and the reason string together, which is a lot stronger than asserting on the matrix alone. It would also be worth adding the COMET_SCALA_UDF_CODEGEN_ENABLED=true counterpart with checkSparkAnswerAndOperator so the dispatcher path is pinned down too. If the cast turns out not to be reachable end to end because something upstream short-circuits, could you document why in the same style as the join tests in CometCollationSuite?

The identity-cast baselines conflict with the guard the issue prefers

CometCast already mixes in CometTypeShim (CometCast.scala:37), which gives you hasNonDefaultStringCollation. That helper already walks nested element, key, value, and field types (spark/src/main/spark-4.x/org/apache/comet/shims/CometTypeShim.scala:41-48). So option 1 from the issue is about three lines at the top of isSupported, above the fromType == toType shortcut:

if (hasNonDefaultStringCollation(fromType) || hasNonDefaultStringCollation(toType)) {
  return unsupported(fromType, toType)
}

Would you consider adding that here rather than leaving it for a follow-up? Two reasons. The issue title is "implicit and untested", and this PR only addresses the second half. And assert(CometCast.isSupported(lcase, lcase, None, evalMode) == Compatible()) would fail once that guard exists, so whoever adds it later has to choose between weakening the guard and deleting your test. Adding the guard now and dropping that baseline avoids that.

Nested collated types are the part that is only safe by accident

Because the fromType == toType shortcut at CometCast.scala:189 runs before any pattern matching, CAST(ARRAY<STRING COLLATE UTF8_LCASE> AS ARRAY<STRING COLLATE UTF8_LCASE>) returns Compatible(), and the same holds for a struct with a collated field or a map with a collated key. serializeDataType maps every StringType to proto type id 7 (QueryPlanSerde.scala:565), so the collation is silently dropped from the proto with no warning. Identity is byte-safe so results are right today, but this is the same implicit behavior the issue describes and it is the case hasNonDefaultStringCollation was written for.

Since the suite is specifically about pinning the matrix down, it would be worth covering the nested paths as well: ArrayType(lcase) -> ArrayType(unicode), a StructType with a collated field, a MapType with a collated key, and the array-to-string recursion at CometCast.scala:203. Those go through different code than the scalar catch-all.

Small correction in the suite Scaladoc

The header says the fallback happens "via canCastFromString/canCastToString's own catch-all when only one side is collated". That holds for lcase -> StringType and StringType -> lcase, but IntegerType -> lcase actually exits through canCastFromInt's catch-all at CometCast.scala:410-411, since (_, DataTypes.StringType) does not match a collated target. Worth correcting, since the explanation is the main value of this file.

CI

CI has not run yet on this branch. All three workflow runs are sitting in action_required waiting on approval, so I will get those going. The registration itself looks correct to me. java-test passes the list as -DwildcardSuites, so the suite is simply not matched on the 3.4 and 3.5 profiles rather than failing, which matches how CometWidthBucketSuite is already handled.

[stantheman0128]

Thanks, this was a genuinely useful review. I had the dispatch semantics backwards and that wording is now fixed everywhere it was copied.

Unsupported does not mean "falls back to Spark"

You are right, and I should have caught this from the repo itself. collation.sql already spells out the same mechanism for the predicate serdes, and predicates.scala carries the comment explaining it. I traced exprToProtoInternal again and the Unsupported branch really does reach dispatchIfFallback before any reason is recorded. Every test name, the suite Scaladoc, and the PR description now say the cast has no native path.

The guard

Added, three lines, above the fromType == toType shortcut so identity casts are covered. It calls hasNonDefaultStringCollation the same way the array, collection, datetime, map, predicate, and string serdes already do. The lcase -> lcase Compatible() baseline is gone, since the guard makes that pair Unsupported. The default-collation baseline stays and I added a nested version of it, so the guard cannot over-block without a test noticing.

Nested types

Covered: ArrayType(lcase) -> ArrayType(unicode), the identity array case, a struct with a collated field, a map with a collated key, a map with a collated value, and ArrayType(lcase) -> StringType for the recursion at CometCast.scala:203.

Working through which pairs the guard actually changes turned up three more that were Compatible on main, so I added tests for those too rather than leave the over-block surface undescribed. A struct whose collated field is unchanged while a sibling field is cast came out Compatible, because the field zip answered per field and the collated field hit the identity shortcut. MapType(lcase, IntegerType) -> MapType(lcase, LongType) did the same through the key. And ArrayType(NullType) -> ArrayType(lcase) was Compatible through the elementType == NullType branch at CometCast.scala:199, which runs before any of the rest. So the collated type could ride into the native plan on a sibling's cast in each case.

The remaining nested pairs already returned Unsupported. The guard only moved which type the reason string names.

Scaladoc correction

Fixed. The IntegerType -> lcase explanation now points at canCastFromInt's catch-all, and the test for that pair carries a comment saying why (_, DataTypes.StringType) never matches a collated target.

End-to-end coverage

Added both, your query and the =true counterpart, and the reason surfaces exactly as you predicted.

One thing I would rather flag myself than let read better than it is. Those two are guard-invariant. (StringType(UTF8_LCASE), IntegerType) already fell through to the case _ catch-all on main and produced the same reason string, so both pass with the guard reverted. They pin down the planner's treatment of an Unsupported cast, which is the thing your first point corrects, but they do not exercise the new check.

I tried to add one that does, using the identity pair the guard actually changed, and it turns out that hits the case you asked me to document. CAST(_1 COLLATE utf8_lcase AS STRING COLLATE UTF8_LCASE) never reaches Comet as a cast, because Spark's SimplifyCasts drops a cast whose child already has the target type. The query arrives at the planner as a bare Collate and the only fallback reason on the plan is collate is not supported, from a different serde:

Expected fallback reason 'Cast from StringType(UTF8_LCASE) to StringType(UTF8_LCASE) is not
supported' not found in [collate is not supported]

So the scalar identity pairs stay pinned at the isSupported level, with a comment in the suite explaining why, written in the style of the join tests in CometCollationSuite.

A struct turned out to be the way in. When a sibling field changes type the cast survives SimplifyCasts and the collated field rides along inside it, so there is now a third end-to-end test on CAST(struct(_2 AS a, _1 COLLATE utf8_lcase AS s) AS STRUCT<a: STRING, s: STRING COLLATE UTF8_LCASE>). That one does fail without the guard, because the old field zip answered Compatible and the struct went native with the collation dropped. The reason it produces is:

Cast from StructType(StructField(a,IntegerType,true),StructField(s,StringType(UTF8_LCASE),true))
to StructType(StructField(a,StringType,true),StructField(s,StringType(UTF8_LCASE),true))
is not supported

Local runs are the spark-4.0 and spark-4.1 profiles, against a debug libcomet.so built from this branch. 3.4, 3.5, and 4.2 first execute in CI. On the 3.x profiles the guard is inert anyway, since hasNonDefaultStringCollation is a false literal in the 3.x shim.

Why these are Scala tests and not SQL fixtures

The review-comet-pr skill checked into this repo says expression tests should use CometSqlFileTestSuite where it can express them, so I should say why most of this suite does not.

The bulk of the file asserts on CometCast.isSupported over type pairs that SQL cannot construct. ArrayType(NullType) -> ArrayType(STRING COLLATE UTF8_LCASE), and a struct whose collated field is unchanged while a sibling field is cast, are only reachable by building the types in Scala. That is the carve-out the skill already allows for.

The three end-to-end tests were closer to workable as a fixture, but --Config and --ConfigMatrix are both file scoped, and ConfigMatrix reruns every query in the file under every combination. The query that has to assert a fallback reason with spark.comet.exec.scalaUDF.codegen.enabled=false and the query that has to assert native execution with it set to true therefore cannot share a file. Expressing them as fixtures means a file per query for what is two lines of SQL each. I am happy to split them out that way if you would rather have them there.

I did run the existing fixtures as part of the blast radius. CometSqlFileTestSuite passes on 4.1 with the guard in, including collation.sql, which is the fixture most exposed to this change since it casts a default-collation column to a collated target on nearly every query.

One gap I did not close

getSupportLevel returns Compatible() for any cast whose child is a Literal, before isSupported is consulted, so CAST('abc' AS STRING COLLATE UTF8_LCASE) still reaches the native side with the collation stripped. ConstantFolding normally removes that cast first, but CometSqlFileTestSuite excludes ConstantFolding for every fixture file it runs, so the path is reachable inside our own harness. It is the same class of problem as #4489 but it sits in a different method, and closing it means deciding what CometLiteral should do with a collated literal rather than adding a line to isSupported. Would you rather I pulled it into this PR or filed it separately?

Moving CometCollationSuite to spark-4.x

I looked at this and I do not think it fits inside this PR. Three things came up.

There is already a second copy at spark/src/test/spark-4.1/org/apache/spark/sql/CometCollationSuite.scala, added by #4097. On the 4.1 profile, src/test/spark-4.x and src/test/spark-4.1 are both test source roots (spark/pom.xml:538-540), so moving the 4.0 copy up without deleting that one gives two classes with the same fully qualified name.

The two copies differ by exactly the #4051 join block, and Spark 4.1 looks like the reason. BroadcastHashJoinExec and ShuffledHashJoinExec became case class ... private there, with an explicit companion apply that runs HashJoin.normalizeJoinKeys. That wraps collated keys in CollationKey, whose dataType is BinaryType. So on 4.1 and 4.2 the exec never receives a collated key, Comet's guard has nothing to reject, and the two result.isEmpty assertions would not hold. SortMergeJoinExec is still a plain public case class on both, so that one test would survive a move.

There is no spark/src/test/spark-4.2 directory at all, so 4.2 has no CometCollationSuite today. A move would run the shuffle and datetime tests there for the first time, which is the real payoff in your suggestion and also the part most likely to surface something new.

So the move is worth doing, but it means reconciling two divergent copies and deciding what happens to the #4051 join tests on 4.1 and later. Happy to file an issue and take it as a follow-up if you agree that is the right shape.

One thing that fell out of the above and may deserve its own issue. If Spark 4.1 normalizes collated join keys to binary before the exec is constructed, is Comet's collated-join guard from #4051 still reachable on 4.1 and later, and could Comet legitimately accept those joins natively there? I did not chase it far enough to be sure, but it did not look like something the current tests would tell us.

For this PR I kept the cast tests in their own spark-4.x suite. With two CometCollationSuite copies in the tree, folding them in would mean writing the same tests twice.