Support Spark collations in comparison, hashing, grouping and string functions

Start with stage 1 of the staged plan: shared collation name parsing/normalization and the collation_key UDF, carrying collation as Arrow field metadata (spark.collation) so DataFusion operators stay untouched. Read CollationFactory.java for id decoding and names, then port literal cases into a .slt file under crates/sqllogictest/test_files/ (the SQL parser does not handle COLLATE, so use function forms). Done: collation_key yields order-compatible bytes for UTF8_LCASE with matching test vectors; ICU collations are a later stage behind a flag.

Part of apache/datafusion-comet#7300 (Tier 5: collate and collation-aware comparison, GROUP BY and joins).

What the function does

Spark 4.0 added collations to STRING. The collation is part of the type: StringType(collationId), where
collationId is a 32-bit id decoded by CollationFactory (bit 29: UTF8 family vs ICU; bit 18: RTRIM; for ICU,
bit 17 case-insensitive, bit 16 accent-insensitive, bits 0-11 the locale). Families:

• UTF8_BINARY (id 0, the default): byte comparison. The only collation with supportsBinaryEquality.
• UTF8_LCASE (id 1): compares lowerCaseCodePoints(s) bytes. This is per-code-point ICU simple lowercasing, plus
  İ (U+0130) becoming i + U+0307 and final ς becoming σ. It is not Unicode case folding.
• ICU collations: UNICODE (root locale) or a locale (en, de, sv, tr, en_USA, ...), with optional _CI
  and/or _AI. They map to ICU strengths: CS+AS = ks=level3, CI+AS = level2, CS+AI = level1 with case level
  on, CI+AI = level1. Equality is collator.compare(a, b) == 0. The sort key is ICU CollationKey bytes.
• _RTRIM on any of the above: trailing U+0020 spaces only are removed before comparing, hashing and searching.

Collation-aware operations (Spark routes them through CollationSupport or the collation's comparator / sort key):
=, <=>, <, >, IN, nullif, greatest, least, ORDER BY, DISTINCT, GROUP BY, joins, set ops
(UNION, INTERSECT, EXCEPT), window partitioning, contains, startswith, endswith, instr, locate,
replace, split_part, split, substring_index, find_in_set, translate, trim / ltrim / rtrim / btrim,
upper, lower, initcap, str_to_map, and the array functions that compare elements (array_contains,
array_position, array_distinct, array_union, array_intersect, array_except, arrays_overlap,
array_min / array_max, sort_array). Regex functions (like, rlike, regexp_*) accept only UTF8_BINARY
and UTF8_LCASE (StringTypeBinaryLcase). Trim, translate and str_to_map reject case-sensitive
accent-insensitive collations (StringTypeNonCSAICollation). Collated map keys are rejected unless
spark.sql.collation.allowInMapKeys is set (UNSUPPORTED_FEATURE.COLLATIONS_IN_MAP_KEYS).

Which collation an expression uses comes from CollationTypeCoercion, by strength: Explicit (COLLATE) beats
Implicit (column, cast of a string), which beats Default (literal). Two different explicit collations fail with
COLLATION_MISMATCH.EXPLICIT (42P21). Two different implicit ones give an indeterminate collation. That fails with
INDETERMINATE_COLLATION / INDETERMINATE_COLLATION_IN_EXPRESSION (42P22) in operations that need a comparator,
and is allowed in pass-through ones such as concat.

Examples

From collations-basic.sql.out, collations-padding-trim.sql.out and CollationSuite:

-- t1(utf8_binary STRING COLLATE UTF8_BINARY, utf8_lcase STRING COLLATE UTF8_LCASE)
-- rows: ('aaa','aaa'), ('AAA','AAA'), ('bbb','bbb'), ('BBB','BBB')
SELECT count(*) FROM t1 GROUP BY utf8_lcase;                         -- 2 | 2
SELECT * FROM t1 WHERE utf8_lcase = 'aaa' COLLATE utf8_lcase;        -- AAA AAA | aaa aaa
SELECT * FROM t1 WHERE utf8_lcase < 'bbb' COLLATE utf8_lcase;        -- AAA AAA | aaa aaa
SELECT * FROM t1 WHERE utf8_binary < 'bbb';                          -- AAA AAA | BBB BBB | aaa aaa
SELECT 'aaa' COLLATE utf8_lcase = 'AAA' COLLATE utf8_lcase;          -- true
SELECT 'aaa' COLLATE utf8_binary_rtrim = 'aaa  ' COLLATE utf8_binary_rtrim;  -- true
SELECT 'aaa ' COLLATE unicode_ci_rtrim = 'AAA  ' COLLATE unicode_ci_rtrim;   -- true
SELECT 'a' COLLATE unicode < 'A';                                    -- true
SELECT 'aaa' COLLATE unicode < 'BBB' COLLATE unicode;                -- true
SELECT 'a' COLLATE unicode_ai = 'å';                                 -- true
SELECT 'a' COLLATE unicode_ci_ai = 'Å';                              -- true
SELECT 'Kypper' COLLATE sv < 'Köpfe';                                -- true
SELECT 'Kypper' COLLATE de > 'Köpfe';                                -- true
SELECT 'I' COLLATE tr_ci = 'ı';                                      -- true
SELECT least('aaa' COLLATE utf8_lcase, 'AAA' COLLATE utf8_lcase, 'a' COLLATE utf8_lcase);  -- a
SELECT nullif('aaa' COLLATE utf8_lcase, 'AAA' COLLATE utf8_lcase);   -- NULL
SELECT array_contains(ARRAY('aaa' COLLATE utf8_lcase), 'AAA' COLLATE utf8_lcase);  -- true
SELECT array_distinct(array('aaa' COLLATE utf8_lcase, 'AAA' COLLATE utf8_lcase));  -- ["aaa"]
SELECT array_except(array('aaa' COLLATE utf8_lcase), array('AAA' COLLATE utf8_lcase)); -- []
SELECT col1 COLLATE utf8_lcase FROM VALUES ('aaa'),('AAA'),('bbb'),('BBB'),('zzz'),('ZZZ')
  EXCEPT SELECT col1 COLLATE utf8_lcase FROM VALUES ('aaa'),('bbb');  -- zzz
SELECT ... UNION SELECT col1 COLLATE unicode_ci ...;                 -- error: INCOMPATIBLE_COLUMN_TYPE (UTF8_LCASE vs UNICODE_CI)
SELECT contains(collate('abc', 'UNICODE_CI'), collate('b', 'UNICODE'));  -- error: COLLATION_MISMATCH.EXPLICIT
SELECT TRIM(utf8_binary, utf8_lcase) FROM t1;                        -- error: INDETERMINATE_COLLATION_IN_EXPRESSION
SELECT TRIM(utf8_binary COLLATE unicode_ai, utf8_lcase COLLATE unicode_ai) FROM t1;  -- error: DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE

CollationSupportSuite cases for UTF8_LCASE: contains('The Kelvin.', 'Kelvin') is true (KELVIN SIGN
lowercases to k); contains('i̇', 'İ') is true; contains('İ', 'i') is false; σ, ς and Σ all match each
other. For UTF8_BINARY, UTF8_LCASE, UNICODE and UNICODE_CI, upper('ß fi ffi') gives SS FI FFI and
lower leaves it unchanged.

Edge cases to test

• Lowercasing vs case folding: UTF8_LCASE does not fold ß to ss, and ſ (U+017F, already lowercase) is not
  equal to s (code reading). İ expands to two code points, so lengths and match positions shift
  (instr, locate, substring_index report positions in the original string).
• Accent insensitivity (_AI): a = å, but trim, translate and str_to_map reject CS_AI collations.
• Ties in ordering: values equal under the collation (aaa, AAA) have no defined order between them. The
  representative a GROUP BY / DISTINCT / set op returns is whichever row comes first (aaa in the examples
  above). Tests should not depend on it.
• Hashing must agree with equality. Spark's shuffle partitioning uses CollationAwareMurmur3Hash (Murmur3 over the
  collation key bytes). Hash joins compare CollationKey(col), injected by RewriteCollationJoin. Hash aggregation
  is disabled for non-binary-stable keys (Aggregate.supportsHashAggregate), so Spark sort-aggregates with the
  collation ordering. The user-facing hash() / xxhash64() hash raw bytes since 4.0.1, unless
  spark.sql.legacy.collationAwareHashFunctions=true.
• Nested types: structs and arrays of collated strings in GROUP BY, joins and = (Spark recurses through
  CreateNamedStruct / ArrayTransform when building keys). Group by c1.utf8_lcase gives 1 row for aaa/AAA.
• RTRIM trims only U+0020. Tabs and NBSP are significant. Output values keep their trailing spaces.
• Invalid UTF-8 is replaced (toValidString / CODE_POINT_ITERATOR_MAKE_VALID) before ICU or lowercase comparison.
• Mixed strengths: column vs literal (implicit wins), explicit vs column (explicit wins), two columns with different
  collations (indeterminate), CAST(x AS STRING COLLATE ...).
• upper / lower with ICU locale collations use the collator's locale (UCharacter.toUpperCase(locale, s)),
  but an all-ASCII input takes an ASCII fast path first. So under tr, upper('i') gives I, while a string
  with any non-ASCII character gets Turkish rules (i becomes İ) (code reading of CollationAwareUTF8String).
  spark.sql.icu.caseMappings.enabled (default true) switches UTF8_BINARY case mapping to ICU.
• Empty strings, NULLs, and an empty search string in contains / instr / replace for ICU collations
  (contains returns true for an empty pattern).

Differences across Spark versions

• 3.4 / 3.5: no collations.
• 4.0: collations introduced. Trim collations behind spark.sql.collation.trim.enabled (default false outside
  tests). hash() / xxhash64() stopped being collation-aware in 4.0.1 (legacy config above).
• 4.1: trim collations on by default; spark.sql.collation.schemaLevel.enabled added. 4.2: trim config removed.
• icu4j is 76.1 in 4.0.4, 77.1 in 4.1.3 and 78.3 in 4.2.0. ICU-collation results and sort keys can change between
  these versions.

Implementation notes

• Spark: common/unsafe/src/main/java/org/apache/spark/sql/catalyst/util/CollationFactory.java (ids, names,
  comparators, getCollationKeyBytes), CollationSupport.java (per-function execBinary / execLowercase /
  execICU), CollationAwareUTF8String.java (lowercase search, lowerCaseCodePoints, case mapping),
  sql/catalyst/.../analysis/CollationTypeCoercion.scala, analysis/RewriteCollationJoin.scala,
  expressions/CollationKey.scala, expressions/hash.scala (CollationAwareMurmur3Hash),
  sql/api/.../internal/types/AbstractStringType.scala (StringTypeWithCollation, StringTypeBinaryLcase,
  StringTypeNonCSAICollation).
• DataFusion has no collation support. Arrow string types have no collation. The proposal is to carry it as
  Arrow field metadata (a key such as spark.collation = normalized name) on Utf8 / LargeUtf8 / Utf8View
  fields, including nested list element, struct child and map fields. UDFs read it via return_field_from_args /
  ScalarFunctionArgs::arg_fields. A missing key means UTF8_BINARY. Comet would need to put the collation into
  its serialized types.
• Operators (sort, hash join, aggregate, distinct) in DataFusion compare raw bytes. Following Spark, the planner
  (Comet or a DataFusion rewrite) can wrap keys in a collation_key(str) -> Binary UDF whose bytes order and compare
  like the collation, and keep the original column for output. This needs no changes to DataFusion operators.
• Staged plan:
  1. Shared infrastructure: collation name parsing and normalization, the metadata convention, collate /
     collation (separate issue), and collation_key.
  2. UTF8_LCASE (and _RTRIM variants) in pure Rust: char::to_lowercase (full mapping, so İ becomes i̇)
     plus ς to σ matches ICU simple lowercasing code point by code point, subject to the Unicode version.
     Then add comparisons, IN, greatest / least, the string search functions, upper / lower / initcap,
     and the array functions.
  3. ICU collations through the icu_collator / icu_casemap crates (ICU4X) or rust_icu (ICU4C bindings).
• Risk: bit-exact results for ICU collations depend on the ICU / CLDR version and implementation. ICU4X is not
  ICU4J, and Spark's version changes per release. Ordering and equality may differ on edge cases, and sort-key bytes
  will not match ICU4J CollationKey bytes. That matters wherever keys are shared with the JVM, for example
  Murmur3 partition ids when one side of a join is shuffled by Spark and the other natively. Gate ICU collations
  behind a compatibility flag and test against Spark-generated vectors per supported version.
• Tests: .slt files per function under crates/sqllogictest/test_files/ (function forms only, since the SQL parser
  does not handle COLLATE). Port the literal cases from CollationSupportSuite.java, CollationFactorySuite.scala
  and collations-*.sql.out. Document new UDFs with #[user_doc].

This issue was generated by an LLM (Claude) from the Apache Spark source code and tests. Examples come from Spark's ExpressionDescription examples, test suites and .sql.out golden files, but please verify them against Spark before relying on them.