You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2026/09/24 04:55:12 UTC
[PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
sunchao opened a new pull request, #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170
## Which issue does this PR close?
Part of #4496. Depends on #5403; this draft is stacked on that PR and should be rebased after it lands. Review the follow-up separately in [commit 821dbf1be](https://github.com/sunchao/arrow-datafusion-comet/commit/821dbf1bee38e76b50515f4c0c7a4a2a3e9b7e0c).
## Rationale for this change
#5403 uses Spark's JVM dispatcher for collated `array_min`/`array_max`. This follow-up executes the UTF8 collation family in Rust, including comparisons inside arrays and structs. For example, `array_min` of `['a', 'B']` under `UTF8_LCASE` returns the original `'a'` without invoking Spark's comparator.
## What changes are included in this PR?
- Carry string-leaf collations into the existing native extrema comparator. Support `UTF8_BINARY`, `UTF8_BINARY_RTRIM`, `UTF8_LCASE`, and `UTF8_LCASE_RTRIM`, retaining the first original value on ties.
- Match Spark's Unicode 16/17 lowercase rules, including dotted I and final sigma, using a pinned, generated mapping table. The generator records source commits and checksums; no ICU dependency is added to native builds.
- Allow byte-preserving string casts that assign collations to run natively, while preserving Spark's CHAR/VARCHAR length checks. Keep locale-based ICU collations on the existing dispatcher path.
- Update the existing focused JVM/SQL coverage and add two native tests covering string layouts, original winners, and version-specific comparison rules.
The repository's `implement-comet-expression` skill guided the implementation and source audit.
## How are these changes tested?
- Native extrema tests: 11 passed.
- Spark 4.1: 5 tests passed across the focused `CometArrayExpressionSuite` extrema cases, `CometSqlFileTestSuite array_extrema_collation`, and a CHAR-padding regression, with a JNI library built from this branch. Native-path tests disable the JVM dispatcher and check nested mixed collations and output schema.
- Exhaustive comparison against the pinned Unicode sources: 2,224,128 versioned scalar checks, with zero differences. Generator output is reproducible.
- `cargo clippy --workspace --all-targets -- -D warnings`, rustfmt, Spotless, ScalaStyle, Apache RAT, Markdown formatting, and `git diff --check` pass.
All Spark profiles and the Spark 4.1 SQL suite are requested through CI labels. Their hosted verdicts are pending. No end-to-end performance result is claimed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#issuecomment-5982764581
Rebased onto current main and retained Spark evaluation for UTF8_BINARY_RTRIM until its native slowdown is addressed, including nested strings and allowIncompatible=true. The routing/result regression covers both dispatcher settings. Also fixed the benchmark Long argument; matching JNI build, Spark 4.1 compilation/style, generated docs and six focused JVM tests passed. The PR description now reflects this scope and current validation; no benchmark speedup is claimed. Strict Spark 3.5 compilation was dependency-blocked locally, and broader hosted checks remain pending.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#issuecomment-5916171413
cc @andygrove @viirya @comphead could you take another look? thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4126147168
##########
native/spark-expr/src/array_funcs/array_extrema.rs:
##########
@@ -0,0 +1,440 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::cmp::Ordering;
+use std::sync::Arc;
+
+use arrow::array::{
+ make_array, make_comparator, new_empty_array, Array, ArrayAccessor, ArrayRef, AsArray,
+ DynComparator, ListArray, MutableArrayData, PrimitiveArray, PrimitiveBuilder, StringArrayType,
+ StructArray, UInt32Array,
+};
+use arrow::buffer::NullBuffer;
+use arrow::compute::{take, SortOptions};
+use arrow::datatypes::{ArrowPrimitiveType, DataType, Float32Type, Float64Type};
+use datafusion::common::{exec_err, Result, ScalarValue};
+use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf};
+use datafusion::logical_expr::{
+ ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
+};
+use num::Float;
+
+#[derive(Debug, Clone, Copy, Hash, Eq, PartialEq)]
+enum Utf8Collation {
+ Binary,
+ BinaryRtrim,
+ Lcase,
+ LcaseRtrim,
+}
+
+impl Utf8Collation {
+ fn compare(self, mut left: &str, mut right: &str, unicode_version: u32) -> Ordering {
+ if matches!(self, Self::BinaryRtrim | Self::LcaseRtrim) {
+ // Spark RTRIM ignores trailing U+0020, not arbitrary Unicode whitespace.
+ left = left.trim_end_matches(' ');
+ right = right.trim_end_matches(' ');
+ }
+ if matches!(self, Self::Binary | Self::BinaryRtrim) {
+ return left.cmp(right);
+ }
+ fn lower(value: &str, unicode_version: u32) -> impl Iterator<Item = u32> + '_ {
+ value.chars().flat_map(move |c| {
+ let cp = c as u32;
+ let [first, second] = match cp {
+ // Spark treats final sigma as ordinary sigma.
+ 0x3c2 => [0x3c3, 0],
+ // Unicode 17 adds these mappings to the library's Unicode 16 data.
+ 0xa7ce | 0xa7d2 | 0xa7d4 if unicode_version == 17 => [cp + 1, 0],
+ 0x16ea0..=0x16eb8 if unicode_version == 17 => [cp + 0x1b, 0],
+ _ => unicode_case_mapping::to_lowercase(c),
+ };
+ // The library uses zero for an unchanged code point or absent second value.
+ std::iter::once(if first == 0 { cp } else { first })
+ .chain((second != 0).then_some(second))
+ })
+ }
+ // Compare ASCII prefixes lazily so a retained long winner is not rescanned.
+ for (offset, (l, r)) in left.bytes().zip(right.bytes()).enumerate() {
+ if !l.is_ascii() || !r.is_ascii() {
+ return lower(&left[offset..], unicode_version)
+ .cmp(lower(&right[offset..], unicode_version));
+ }
+ let ordering = l.to_ascii_lowercase().cmp(&r.to_ascii_lowercase());
+ if ordering != Ordering::Equal {
+ return ordering;
+ }
+ }
+ left.len().cmp(&right.len())
+ }
+}
+
+/// Spark's array_min/array_max retain the first non-null value on an ordering tie.
+/// In particular, signed zeros compare equal and all NaNs compare equal and greater
+/// than non-NaNs. Nested arrays and structs use the same ordering, with nulls first.
+#[derive(Debug, Hash, Eq, PartialEq)]
+pub struct SparkArrayExtrema {
+ is_min: bool,
+ datafusion_udf: Arc<ScalarUDF>,
+ string_collations: Vec<Utf8Collation>,
+ unicode_version: u32,
+}
+
+impl SparkArrayExtrema {
+ pub fn new(is_min: bool) -> Self {
+ Self {
+ is_min,
+ // Capture the original implementation, not a registry lookup: these UDFs
+ // replace the DataFusion names in Comet's function registry.
+ datafusion_udf: if is_min {
+ array_min_udf()
+ } else {
+ array_max_udf()
+ },
+ string_collations: Vec::new(),
+ unicode_version: 0,
+ }
+ }
+
+ /// Scala validates the Unicode version and lists string-leaf collations depth-first.
+ pub fn with_collations(
+ is_min: bool,
+ collations: &[String],
+ unicode_version: u32,
+ ) -> Result<Self> {
+ let string_collations = collations
+ .iter()
+ .map(|name| match name.as_str() {
+ "UTF8_BINARY" => Ok(Utf8Collation::Binary),
+ "UTF8_BINARY_RTRIM" => Ok(Utf8Collation::BinaryRtrim),
+ "UTF8_LCASE" => Ok(Utf8Collation::Lcase),
+ "UTF8_LCASE_RTRIM" => Ok(Utf8Collation::LcaseRtrim),
+ _ => exec_err!("Unsupported array extrema collation: {name}"),
+ })
+ .collect::<Result<Vec<_>>>()?;
+ Ok(Self {
+ string_collations,
+ unicode_version,
+ ..Self::new(is_min)
+ })
+ }
+}
+
+impl ScalarUDFImpl for SparkArrayExtrema {
+ fn name(&self) -> &str {
+ if self.is_min {
+ "array_min"
+ } else {
+ "array_max"
+ }
+ }
+
+ fn signature(&self) -> &Signature {
+ self.datafusion_udf.signature()
+ }
+
+ fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
+ self.datafusion_udf.return_type(arg_types)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let [input] = args.args.as_slice() else {
+ return exec_err!("{} takes exactly one argument", self.name());
+ };
+ let element_type = self.return_type(&[input.data_type()])?;
+
+ // DataFusion's non-primitive path reconstructs an array from scalars, which
+ // cannot infer a type from an empty iterator. Keep the declared element type.
+ if matches!(input, ColumnarValue::Array(array) if array.is_empty()) {
+ return Ok(ColumnarValue::Array(new_empty_array(&element_type)));
+ }
+ if self.string_collations.is_empty()
+ && !matches!(
+ element_type,
+ DataType::Float32 | DataType::Float64 | DataType::List(_) | DataType::Struct(_)
+ )
+ {
+ return self.datafusion_udf.invoke_with_args(args);
Review Comment:
Agreed. In 341174464, plain `Utf8`, `LargeUtf8`, and `Utf8View` now use the existing comparator/take path. I kept delegation unchanged for unrelated element types to keep this scoped. Added default-binary coverage for the three string layouts, null/empty/sliced inputs and scalar results. Selected StringView results also compact their backing buffers, with a regression guard against retaining a large losing value.
The new release Criterion comparison covers eight shapes, both min/max, and 64 arrays of eight strings. All outputs matched DataFusion; DataFusion/Comet mean-time ratios were 1.38-4.61x across the 16 comparisons (32 timings). The full Spark benchmark's plain-binary controls improved 1.06-1.45x against #5403. The PR description now separates these kernel and end-to-end measurements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4188935638
##########
spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala:
##########
@@ -1751,6 +1751,22 @@ class CometNativeCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ test("collation casts retain Spark fallback") {
+ assume(CometSparkSessionExtensions.isSpark40Plus)
+ withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") {
+ withParquetTable(Seq(("a", "A"), ("x ", "x")), "collation_cast") {
+ for (collation <- Seq("UTF8_LCASE", "UTF8_BINARY_RTRIM")) {
+ val a = s"CAST(_1 AS STRING COLLATE $collation)"
+ val b = s"CAST(_2 AS STRING COLLATE $collation)"
+ checkSparkAnswerAndFallbackReason(
+ s"SELECT array_contains(array($a), $b), " +
+ s"arrays_overlap(array($a), array($b)) FROM collation_cast",
+ "Cast from StringType")
Review Comment:
Fixed in 7d80ca40cf. The fallback assertion now uses `CometCast.nonDefaultCollationReason`, matching the serializer’s explicit collation guard. The focused Spark 4.1 collation-cast test and strict Spark 3.5 test compilation pass. Prior-head hosted Spark 4.0/4.1/4.2 expression logs confirmed this stale assertion; broader CI is rerunning on the new head.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4209880058
##########
native/spark-expr/src/array_funcs/array_extrema.rs:
##########
@@ -53,8 +105,33 @@ impl SparkArrayExtrema {
} else {
array_max_udf()
},
+ string_collations: Vec::new(),
+ unicode_version: 0,
}
}
+
+ /// Scala validates the Unicode version and lists string-leaf collations depth-first.
+ pub fn with_collations(
Review Comment:
The doc comment says Scala validates the Unicode version, but this constructor accepts any value. With `UTF8_LCASE` and a version other than 16 or 17, it quietly uses the Unicode 16 table. Could it return an error in that case, so a mismatch between the serde and the native side fails loudly instead of returning a different winner than Spark?
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -256,16 +256,57 @@ object CometArrayIntersect
private object ArrayExtremaSupport extends CometTypeShim {
val incompatReason: String =
- "Array extrema use binary string ordering for non-UTF8_BINARY collations " +
+ "Array extrema support UTF8_BINARY, UTF8_LCASE, and UTF8_LCASE_RTRIM natively; " +
+ "UTF8_LCASE requires Unicode 16 or 17. Other collations use binary ordering when opted in " +
"(https://github.com/apache/datafusion-comet/issues/4496)."
- def getSupportLevel(elementType: DataType): SupportLevel = {
- if (hasNonDefaultStringCollation(elementType)) {
- // The dispatcher runs Spark's own comparison with the original collation IDs, including
- // strings nested in arrays or structs. Keep the native bytewise comparison opt-in only.
- Incompatible(Some(incompatReason))
- } else {
+ val binaryRtrimReason: String =
+ "UTF8_BINARY_RTRIM array extrema use Spark's comparator."
+
+ private def stringCollations(dt: DataType): Seq[String] = dt match {
+ case stringType: StringType => Seq(stringCollationName(stringType))
+ case ArrayType(elementType, _) => stringCollations(elementType)
+ case StructType(fields) => fields.toSeq.flatMap(f => stringCollations(f.dataType))
+ case _ => Seq.empty
+ }
+
+ private def supportsCollations(collations: Seq[String]): Boolean = collations.forall {
+ case "UTF8_BINARY" | "UTF8_BINARY_RTRIM" => true
+ case "UTF8_LCASE" | "UTF8_LCASE_RTRIM" =>
+ collationUnicodeVersion == 16 || collationUnicodeVersion == 17
+ case _ => false
+ }
+
+ def getSupportLevel(elementType: DataType): SupportLevel =
+ // Keep this path on the dispatcher until its native performance regression is resolved.
+ // Use Unsupported so allowIncompatible cannot silently change its comparison semantics.
+ if (stringCollations(elementType).contains("UTF8_BINARY_RTRIM")) {
+ Unsupported(Some(binaryRtrimReason))
+ } else if (supportsCollations(stringCollations(elementType))) {
Compatible()
+ } else {
+ Incompatible(Some(incompatReason))
+ }
+
+ def convert(
+ expr: Expression,
+ inputs: Seq[Attribute],
+ binding: Boolean,
+ isMin: Boolean): Option[ExprOuterClass.Expr] = {
+ val collations = stringCollations(expr.dataType)
+ exprToProtoInternal(expr.children.head, inputs, binding).map { child =>
+ val builder = ExprOuterClass.ArrayExtrema
+ .newBuilder()
+ .setChild(child)
+ .setIsMin(isMin)
+ // Preserve the existing binary comparison when unsupported collations are explicitly
+ // opted in. By default those expressions use the JVM dispatcher.
+ if (!collations.forall(_ == "UTF8_BINARY") && supportsCollations(collations)) {
Review Comment:
When the element type mixes a supported and an unsupported collation, for example a struct with a `UTF8_LCASE` field and a `UNICODE_CI` field, and the user opts in, `supportsCollations` is false for the whole list. No collations are sent, so the `UTF8_LCASE` field is also compared bytewise. The incompatible reason says only the other collations use binary ordering. Would it work to send the list with each unsupported entry mapped to `UTF8_BINARY`? That keeps the reason accurate and keeps the supported leaves correct under opt-in.
##########
native/spark-expr/src/array_funcs/array_extrema.rs:
##########
@@ -293,5 +389,100 @@ fn take_extrema_values(values: &ArrayRef, indices: &UInt32Array) -> Result<Array
}
}
+/// Build one comparator per child array, adding string collation to Spark's ordering.
+/// Noncollated values use the shared Spark comparator, including its floating-point rules.
+fn collated_comparator(
+ array: &ArrayRef,
+ collations: &mut std::slice::Iter<Utf8Collation>,
+ unicode_version: u32,
+) -> Result<DynComparator> {
+ if collations.as_slice().is_empty() {
+ return spark_comparator(array.as_ref(), array.as_ref());
+ }
+ match array.data_type() {
+ DataType::List(_) => {
+ let array = array.as_list::<i32>();
+ let compare = collated_comparator(array.values(), collations, unicode_version)?;
+ let offsets = array.offsets().clone();
+ Ok(nulls_first(array.nulls().cloned(), move |left, right| {
+ let left_start = offsets[left] as usize;
+ let right_start = offsets[right] as usize;
+ let left_len = offsets[left + 1] as usize - left_start;
+ let right_len = offsets[right + 1] as usize - right_start;
+ for offset in 0..left_len.min(right_len) {
+ let ordering = compare(left_start + offset, right_start + offset);
+ if ordering != Ordering::Equal {
+ return ordering;
+ }
+ }
+ left_len.cmp(&right_len)
+ }))
+ }
+ DataType::Struct(_) => {
+ let array = array.as_struct();
+ let fields = array
+ .columns()
+ .iter()
+ .map(|field| collated_comparator(field, collations, unicode_version))
+ .collect::<Result<Vec<_>>>()?;
+ Ok(nulls_first(array.nulls().cloned(), move |left, right| {
+ fields
+ .iter()
+ .map(|compare| compare(left, right))
+ .find(|&ordering| ordering != Ordering::Equal)
+ .unwrap_or(Ordering::Equal)
+ }))
+ }
+ DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View
+ if !collations.as_slice().is_empty() =>
+ {
+ let collation = collations.next().unwrap();
+ Ok(match array.data_type() {
+ DataType::Utf8 => string_comparator(
+ array.as_string::<i32>().clone(),
+ *collation,
+ unicode_version,
+ ),
+ DataType::LargeUtf8 => string_comparator(
+ array.as_string::<i64>().clone(),
+ *collation,
+ unicode_version,
+ ),
+ _ => string_comparator(array.as_string_view().clone(), *collation, unicode_version),
+ })
+ }
+ _ => spark_comparator(array.as_ref(), array.as_ref()),
Review Comment:
The Scala side and this function each walk the element type depth-first, and a string leaf picks its collation by position. Nothing checks that the two walks agree. This arm does not consume an entry. So a string-bearing subtree it does not recognize, such as a `LargeList`, a `FixedSizeList` or a dictionary-encoded string (all of which `spark_comparator` accepts), would compare bytewise and shift every later leaf onto the wrong collation without an error. Comet does not produce those types here today, but a later change could introduce that drift with no failing test. Could this arm return an error when the type contains a string while collations remain?
##########
spark/src/test/resources/sql-tests/expressions/array/array_extrema_collation.sql:
##########
@@ -41,17 +48,39 @@ SELECT id, array_min(array(a, b)), array_max(array(a, b)),
FROM test_array_extrema_collation
-- Case-insensitive ordering differs from binary ordering for 'a' and 'B'.
-query expect_dispatch(array_min,array_max)
+-- Non-ASCII ties run in both input orders and must retain the first original value.
+query expect_native(array_min,array_max)
SELECT id, array_min(array(CAST(a AS STRING COLLATE UTF8_LCASE), CAST(b AS STRING COLLATE UTF8_LCASE))),
array_max(array(CAST(a AS STRING COLLATE UTF8_LCASE), CAST(b AS STRING COLLATE UTF8_LCASE)))
FROM test_array_extrema_collation
--- Collation detection and Spark comparison also recurse through struct fields and arrays.
-query expect_dispatch(array_min,array_max)
+-- Collation comparison also recurses through struct fields and arrays. The second
+-- string field uses binary ordering, independently of the first field's collation.
+query expect_native(array_min,array_max)
SELECT id, array_min(array(
- named_struct('s', array(CAST(a AS STRING COLLATE UTF8_LCASE)), 'f', x),
- named_struct('s', array(CAST(b AS STRING COLLATE UTF8_LCASE)), 'f', y))),
+ named_struct('s', array(CAST(a AS STRING COLLATE UTF8_LCASE)), 'binary', a, 'f', x),
Review Comment:
This query puts the collated field ahead of the binary one, so it would not catch a `UTF8_BINARY` leaf that fails to consume its entry. Could you add a variant with the binary field first, for example `named_struct('binary', a, 's', array(CAST(a AS STRING COLLATE UTF8_LCASE)), 'f', x)`? Then the positional mapping is guarded from both directions.
##########
native/spark-expr/src/array_funcs/array_extrema.rs:
##########
@@ -156,9 +243,14 @@ where
result.finish()
}
-fn nested_extrema(array: &ListArray, is_min: bool) -> Result<ArrayRef> {
+fn nested_extrema(
+ array: &ListArray,
+ is_min: bool,
+ mut collations: std::slice::Iter<Utf8Collation>,
+ unicode_version: u32,
+) -> Result<ArrayRef> {
let values = array.values();
- let compare = spark_comparator(values.as_ref(), values.as_ref())?;
+ let compare = collated_comparator(values, &mut collations, unicode_version)?;
Review Comment:
Related to the fallback arm below, could this check that `collations` is empty once the comparator is built and return an error otherwise? Leftover entries mean the native walk and `stringCollations` on the Scala side disagreed, and today that would go unnoticed.
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -256,16 +256,57 @@ object CometArrayIntersect
private object ArrayExtremaSupport extends CometTypeShim {
val incompatReason: String =
- "Array extrema use binary string ordering for non-UTF8_BINARY collations " +
+ "Array extrema support UTF8_BINARY, UTF8_LCASE, and UTF8_LCASE_RTRIM natively; " +
+ "UTF8_LCASE requires Unicode 16 or 17. Other collations use binary ordering when opted in " +
"(https://github.com/apache/datafusion-comet/issues/4496)."
- def getSupportLevel(elementType: DataType): SupportLevel = {
- if (hasNonDefaultStringCollation(elementType)) {
- // The dispatcher runs Spark's own comparison with the original collation IDs, including
- // strings nested in arrays or structs. Keep the native bytewise comparison opt-in only.
- Incompatible(Some(incompatReason))
- } else {
+ val binaryRtrimReason: String =
+ "UTF8_BINARY_RTRIM array extrema use Spark's comparator."
+
+ private def stringCollations(dt: DataType): Seq[String] = dt match {
+ case stringType: StringType => Seq(stringCollationName(stringType))
+ case ArrayType(elementType, _) => stringCollations(elementType)
+ case StructType(fields) => fields.toSeq.flatMap(f => stringCollations(f.dataType))
+ case _ => Seq.empty
+ }
+
+ private def supportsCollations(collations: Seq[String]): Boolean = collations.forall {
+ case "UTF8_BINARY" | "UTF8_BINARY_RTRIM" => true
+ case "UTF8_LCASE" | "UTF8_LCASE_RTRIM" =>
+ collationUnicodeVersion == 16 || collationUnicodeVersion == 17
+ case _ => false
+ }
+
+ def getSupportLevel(elementType: DataType): SupportLevel =
+ // Keep this path on the dispatcher until its native performance regression is resolved.
+ // Use Unsupported so allowIncompatible cannot silently change its comparison semantics.
+ if (stringCollations(elementType).contains("UTF8_BINARY_RTRIM")) {
+ Unsupported(Some(binaryRtrimReason))
Review Comment:
The comment above says `Unsupported` keeps `allowIncompatible` from silently changing the comparison semantics. With `Incompatible` and opt-in, though, `convert` would send `UTF8_BINARY_RTRIM`, and the native comparator handles it correctly, so the semantics would not change. The real reason is the performance regression, and the comment should say that. This branch also makes the native RTRIM support unreachable. That covers the `UTF8_BINARY_RTRIM` case in `supportsCollations`, `Utf8Collation::BinaryRtrim` and its mapping in `with_collations`, and the unit tests that exercise them. Could you either remove those pieces, or keep them and open a tracking issue for the regression? Then this branch, the comment and `binaryRtrimReason` could link to it. `binaryRtrimReason` also ends up in the generated compatibility guide as an unsupported reason, and "use Spark's comparator" alone does not tell a reader why.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4112441437
##########
native/spark-expr/src/array_funcs/array_extrema.rs:
##########
@@ -0,0 +1,440 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::cmp::Ordering;
+use std::sync::Arc;
+
+use arrow::array::{
+ make_array, make_comparator, new_empty_array, Array, ArrayAccessor, ArrayRef, AsArray,
+ DynComparator, ListArray, MutableArrayData, PrimitiveArray, PrimitiveBuilder, StringArrayType,
+ StructArray, UInt32Array,
+};
+use arrow::buffer::NullBuffer;
+use arrow::compute::{take, SortOptions};
+use arrow::datatypes::{ArrowPrimitiveType, DataType, Float32Type, Float64Type};
+use datafusion::common::{exec_err, Result, ScalarValue};
+use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf};
+use datafusion::logical_expr::{
+ ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
+};
+use num::Float;
+
+#[derive(Debug, Clone, Copy, Hash, Eq, PartialEq)]
+enum Utf8Collation {
+ Binary,
+ BinaryRtrim,
+ Lcase,
+ LcaseRtrim,
+}
+
+impl Utf8Collation {
+ fn compare(self, mut left: &str, mut right: &str, unicode_version: u32) -> Ordering {
+ if matches!(self, Self::BinaryRtrim | Self::LcaseRtrim) {
+ // Spark RTRIM ignores trailing U+0020, not arbitrary Unicode whitespace.
+ left = left.trim_end_matches(' ');
+ right = right.trim_end_matches(' ');
+ }
+ if matches!(self, Self::Binary | Self::BinaryRtrim) {
+ return left.cmp(right);
+ }
+ fn lower(value: &str, unicode_version: u32) -> impl Iterator<Item = u32> + '_ {
+ value.chars().flat_map(move |c| {
+ let cp = c as u32;
+ let [first, second] = match cp {
+ // Spark treats final sigma as ordinary sigma.
+ 0x3c2 => [0x3c3, 0],
+ // Unicode 17 adds these mappings to the library's Unicode 16 data.
+ 0xa7ce | 0xa7d2 | 0xa7d4 if unicode_version == 17 => [cp + 1, 0],
+ 0x16ea0..=0x16eb8 if unicode_version == 17 => [cp + 0x1b, 0],
+ _ => unicode_case_mapping::to_lowercase(c),
+ };
+ // The library uses zero for an unchanged code point or absent second value.
+ std::iter::once(if first == 0 { cp } else { first })
+ .chain((second != 0).then_some(second))
+ })
+ }
+ // Compare ASCII prefixes lazily so a retained long winner is not rescanned.
+ for (offset, (l, r)) in left.bytes().zip(right.bytes()).enumerate() {
+ if !l.is_ascii() || !r.is_ascii() {
+ return lower(&left[offset..], unicode_version)
+ .cmp(lower(&right[offset..], unicode_version));
+ }
+ let ordering = l.to_ascii_lowercase().cmp(&r.to_ascii_lowercase());
+ if ordering != Ordering::Equal {
+ return ordering;
+ }
+ }
+ left.len().cmp(&right.len())
+ }
+}
+
+/// Spark's array_min/array_max retain the first non-null value on an ordering tie.
+/// In particular, signed zeros compare equal and all NaNs compare equal and greater
+/// than non-NaNs. Nested arrays and structs use the same ordering, with nulls first.
+#[derive(Debug, Hash, Eq, PartialEq)]
+pub struct SparkArrayExtrema {
+ is_min: bool,
+ datafusion_udf: Arc<ScalarUDF>,
+ string_collations: Vec<Utf8Collation>,
+ unicode_version: u32,
+}
+
+impl SparkArrayExtrema {
+ pub fn new(is_min: bool) -> Self {
+ Self {
+ is_min,
+ // Capture the original implementation, not a registry lookup: these UDFs
+ // replace the DataFusion names in Comet's function registry.
+ datafusion_udf: if is_min {
+ array_min_udf()
+ } else {
+ array_max_udf()
+ },
+ string_collations: Vec::new(),
+ unicode_version: 0,
+ }
+ }
+
+ /// Scala validates the Unicode version and lists string-leaf collations depth-first.
+ pub fn with_collations(
+ is_min: bool,
+ collations: &[String],
+ unicode_version: u32,
+ ) -> Result<Self> {
+ let string_collations = collations
+ .iter()
+ .map(|name| match name.as_str() {
+ "UTF8_BINARY" => Ok(Utf8Collation::Binary),
+ "UTF8_BINARY_RTRIM" => Ok(Utf8Collation::BinaryRtrim),
+ "UTF8_LCASE" => Ok(Utf8Collation::Lcase),
+ "UTF8_LCASE_RTRIM" => Ok(Utf8Collation::LcaseRtrim),
+ _ => exec_err!("Unsupported array extrema collation: {name}"),
+ })
+ .collect::<Result<Vec<_>>>()?;
+ Ok(Self {
+ string_collations,
+ unicode_version,
+ ..Self::new(is_min)
+ })
+ }
+}
+
+impl ScalarUDFImpl for SparkArrayExtrema {
+ fn name(&self) -> &str {
+ if self.is_min {
+ "array_min"
+ } else {
+ "array_max"
+ }
+ }
+
+ fn signature(&self) -> &Signature {
+ self.datafusion_udf.signature()
+ }
+
+ fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
+ self.datafusion_udf.return_type(arg_types)
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ let [input] = args.args.as_slice() else {
+ return exec_err!("{} takes exactly one argument", self.name());
+ };
+ let element_type = self.return_type(&[input.data_type()])?;
+
+ // DataFusion's non-primitive path reconstructs an array from scalars, which
+ // cannot infer a type from an empty iterator. Keep the declared element type.
+ if matches!(input, ColumnarValue::Array(array) if array.is_empty()) {
+ return Ok(ColumnarValue::Array(new_empty_array(&element_type)));
+ }
+ if self.string_collations.is_empty()
+ && !matches!(
+ element_type,
+ DataType::Float32 | DataType::Float64 | DataType::List(_) | DataType::Struct(_)
+ )
+ {
+ return self.datafusion_udf.invoke_with_args(args);
Review Comment:
Plain `UTF8_BINARY` strings still take this branch. DataFusion's non-primitive path builds a `ScalarValue` per row and then rebuilds the array from them, which is also what main does today. I copied this file into a release microbench over the same shapes as `arrayExtremaCollationBenchmark`, 262,144 rows of 8 strings. Sending plain strings to `nested_extrema` instead, where they fall through to the existing `make_comparator` arm, took 11, 20 and 7 ms against DataFusion's 24, 40 and 32 ms for the short ASCII, long ASCII and Unicode shapes, with identical output in every case. Would you consider limiting this delegation to primitive element types?
##########
native/proto/src/proto/expr.proto:
##########
@@ -540,6 +540,10 @@ message ScalarFunc {
repeated Expr args = 2;
DataType return_type = 3;
bool fail_on_error = 4;
+ // array_min/max only: collations of every string leaf in the element type,
+ // visiting array elements and struct fields depth-first. Empty means binary ordering.
+ repeated string string_collations = 5;
Review Comment:
These two fields only mean something for `array_min` and `array_max`, but they sit on the generic `ScalarFunc` that every scalar function uses, and the planner then picks the UDF by name in `create_scalar_function_expr`. Array functions with extra parameters, such as `ArrayInsert`, `ArrayJoin` and `ArraysZip`, get their own message instead. Would a dedicated message for the extrema be a better fit?
##########
spark/src/test/resources/sql-tests/expressions/array/array_extrema_collation.sql:
##########
@@ -0,0 +1,74 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- MinSparkVersion: 4.0
+-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false
+-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_array_extrema_collation(
+ id int, a string, b string, x double, y double) USING parquet
+
+statement
+INSERT INTO test_array_extrema_collation VALUES
+ (1, 'a', 'B', double('0.0'), double('-0.0')),
+ (2, 'B', 'a', double('-0.0'), double('0.0')),
+ (3, 'A', 'a', double('-0.0'), double('0.0')),
+ (4, NULL, 'B', NULL, double('0.0')),
+ (5, NULL, NULL, NULL, NULL),
+ (6, 'x ', 'x', 1.0, 2.0)
Review Comment:
Every string in this fixture is ASCII, so the non-ASCII branch of `Utf8Collation::compare` only runs in the Rust unit test. Could you add a few rows like `İ` against `i̇`, `ς` against `Σ`, and the Kelvin sign against `k`, in both orders so the first-winner rule shows? I ran those through `checkSparkAnswerAndImpl` on 4.1 and 4.2 and they match Spark, so they would go in as regression guards.
##########
native/spark-expr/src/array_funcs/array_extrema.rs:
##########
@@ -0,0 +1,440 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::cmp::Ordering;
+use std::sync::Arc;
+
+use arrow::array::{
+ make_array, make_comparator, new_empty_array, Array, ArrayAccessor, ArrayRef, AsArray,
+ DynComparator, ListArray, MutableArrayData, PrimitiveArray, PrimitiveBuilder, StringArrayType,
+ StructArray, UInt32Array,
+};
+use arrow::buffer::NullBuffer;
+use arrow::compute::{take, SortOptions};
+use arrow::datatypes::{ArrowPrimitiveType, DataType, Float32Type, Float64Type};
+use datafusion::common::{exec_err, Result, ScalarValue};
+use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf};
+use datafusion::logical_expr::{
+ ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
+};
+use num::Float;
+
+#[derive(Debug, Clone, Copy, Hash, Eq, PartialEq)]
+enum Utf8Collation {
+ Binary,
+ BinaryRtrim,
+ Lcase,
+ LcaseRtrim,
+}
+
+impl Utf8Collation {
+ fn compare(self, mut left: &str, mut right: &str, unicode_version: u32) -> Ordering {
+ if matches!(self, Self::BinaryRtrim | Self::LcaseRtrim) {
+ // Spark RTRIM ignores trailing U+0020, not arbitrary Unicode whitespace.
+ left = left.trim_end_matches(' ');
+ right = right.trim_end_matches(' ');
+ }
+ if matches!(self, Self::Binary | Self::BinaryRtrim) {
+ return left.cmp(right);
+ }
+ fn lower(value: &str, unicode_version: u32) -> impl Iterator<Item = u32> + '_ {
+ value.chars().flat_map(move |c| {
+ let cp = c as u32;
+ let [first, second] = match cp {
+ // Spark treats final sigma as ordinary sigma.
+ 0x3c2 => [0x3c3, 0],
+ // Unicode 17 adds these mappings to the library's Unicode 16 data.
+ 0xa7ce | 0xa7d2 | 0xa7d4 if unicode_version == 17 => [cp + 1, 0],
+ 0x16ea0..=0x16eb8 if unicode_version == 17 => [cp + 0x1b, 0],
Review Comment:
Only U+A7CE of these 28 Unicode 17 mappings has a test. I dropped U+A7D2 from the arm above, and separately ended this range at U+16EB7, and all 11 native tests stayed green both times. The table is right today, since I checked every code point against Spark 4.2's `lowerCaseCodePoints` with ICU 78.3. Could the unit test loop over all 28 pairs and assert they compare equal under 17 and distinct under 16?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4178686419
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometArrayExpressionBenchmark.scala:
##########
@@ -146,7 +149,66 @@ object CometArrayExpressionBenchmark extends CometBenchmarkBase {
}
}
+ def arrayExtremaCollationBenchmark(values: Int, expectNative: Boolean = true): Unit = {
+ // Repeated values exercise dictionaries; long prefixes and Unicode exercise comparison work.
+ val shapes = Seq(
+ ("short ASCII, no nulls", "AbC", 1, 0),
+ ("long ASCII, sparse nulls", "AbC", 32, 8),
+ ("Unicode, dense nulls", "ΣİÄ", 4, 50))
+ for ((shape, prefix, repeats, nullPercent) <- shapes) {
+ withTempPath { dir =>
+ withTempTable("parquetV1Table") {
+ val columns = (0 until 8).map { i =>
+ val text = if (i % 2 == 0) prefix else prefix.toLowerCase(java.util.Locale.ROOT)
+ s"""CASE WHEN PMOD(value + ${i * 13}, 100) < $nullPercent THEN NULL
+ |ELSE CONCAT(REPEAT('$text', $repeats),
+ | CAST(PMOD(value + ${i / 2}, 64) AS STRING), '${" " * (i % 3)}')
+ |END AS c$i""".stripMargin
+ }
+ prepareTable(dir, spark.sql(s"SELECT ${columns.mkString(", ")} FROM $tbl"))
+ for (collation <- Seq(
+ "UTF8_BINARY",
+ "UTF8_BINARY_RTRIM",
+ "UTF8_LCASE",
+ "UTF8_LCASE_RTRIM");
+ function <- Seq("array_min", "array_max")) {
+ val input = (0 until 8)
+ .map(i => s"CAST(c$i AS STRING COLLATE $collation)")
+ .mkString("array(", ", ", ")")
+ val query = s"SELECT $function($input) FROM parquetV1Table"
+ // The dispatcher baseline uses the same inputs and verifies its execution path too.
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true") {
+ val plan = stripAQEPlan(spark.sql(query).queryExecution.executedPlan)
+ val explain = new ExtendedExplainInfo()
+ if (expectNative || collation == "UTF8_BINARY") {
+ require(
+ explain.getNativeExpressions(plan).contains(function) &&
+ !explain.getCodegenDispatchExpressions(plan).contains(function),
+ s"$function did not run natively: $plan")
+ } else {
+ require(
+ explain.getCodegenDispatchExpressions(plan).contains(function),
+ s"$function did not use the dispatcher: $plan")
+ }
+ }
+ runExpressionBenchmark(s"$function $collation - $shape", values, query)
Review Comment:
Changed the benchmark cardinality argument to values.toLong. Spark 4.1 compilation and six focused tests pass. The strict Spark 3.5 attempt was blocked by unavailable AWS SDK dependencies locally, so the existing all-profile CI label remains the verification for that profile.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4178984852
##########
spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala:
##########
@@ -1751,6 +1751,22 @@ class CometNativeCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
}
}
+ test("collation casts retain Spark fallback") {
+ assume(CometSparkSessionExtensions.isSpark40Plus)
+ withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") {
+ withParquetTable(Seq(("a", "A"), ("x ", "x")), "collation_cast") {
+ for (collation <- Seq("UTF8_LCASE", "UTF8_BINARY_RTRIM")) {
+ val a = s"CAST(_1 AS STRING COLLATE $collation)"
+ val b = s"CAST(_2 AS STRING COLLATE $collation)"
+ checkSparkAnswerAndFallbackReason(
+ s"SELECT array_contains(array($a), $b), " +
+ s"arrays_overlap(array($a), array($b)) FROM collation_cast",
+ "Cast from StringType")
Review Comment:
[P2] Use `CometCast.nonDefaultCollationReason` for this assertion. On Spark 4.x with dispatch disabled, these collated casts correctly fall back, but the current guard reports `Cast involving a non-default string collation is not supported ...`, which does not contain `Cast from StringType`. Consequently, the newly added `collation casts retain Spark fallback` test fails deterministically despite correct fallback behavior, failing all three Spark 4.x expression jobs and Required Checks.
Evidence: Exact-head CI reproduces this assertion failure on Spark 4.0, 4.1, and 4.2. Spark 4.1 log: https://github.com/apache/datafusion-comet/actions/runs/37222748605/job/111502757158. It reports `Expected fallback reason 'Cast from StringType' not found in [Cast involving a non-default string collation is not supported ...]`. The pinned head defines that actual message in `CometCast.nonDefaultCollationReason`, and `checkSparkAnswerAndFallbackReasons` requires a substring match. Both files and this test are unchanged between the pinned head and CI’s tested merge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4151914759
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometArrayExpressionBenchmark.scala:
##########
@@ -146,7 +149,66 @@ object CometArrayExpressionBenchmark extends CometBenchmarkBase {
}
}
+ def arrayExtremaCollationBenchmark(values: Int, expectNative: Boolean = true): Unit = {
+ // Repeated values exercise dictionaries; long prefixes and Unicode exercise comparison work.
+ val shapes = Seq(
+ ("short ASCII, no nulls", "AbC", 1, 0),
+ ("long ASCII, sparse nulls", "AbC", 32, 8),
+ ("Unicode, dense nulls", "ΣİÄ", 4, 50))
+ for ((shape, prefix, repeats, nullPercent) <- shapes) {
+ withTempPath { dir =>
+ withTempTable("parquetV1Table") {
+ val columns = (0 until 8).map { i =>
+ val text = if (i % 2 == 0) prefix else prefix.toLowerCase(java.util.Locale.ROOT)
+ s"""CASE WHEN PMOD(value + ${i * 13}, 100) < $nullPercent THEN NULL
+ |ELSE CONCAT(REPEAT('$text', $repeats),
+ | CAST(PMOD(value + ${i / 2}, 64) AS STRING), '${" " * (i % 3)}')
+ |END AS c$i""".stripMargin
+ }
+ prepareTable(dir, spark.sql(s"SELECT ${columns.mkString(", ")} FROM $tbl"))
+ for (collation <- Seq(
+ "UTF8_BINARY",
+ "UTF8_BINARY_RTRIM",
+ "UTF8_LCASE",
+ "UTF8_LCASE_RTRIM");
+ function <- Seq("array_min", "array_max")) {
+ val input = (0 until 8)
+ .map(i => s"CAST(c$i AS STRING COLLATE $collation)")
+ .mkString("array(", ", ", ")")
+ val query = s"SELECT $function($input) FROM parquetV1Table"
+ // The dispatcher baseline uses the same inputs and verifies its execution path too.
+ withSQLConf(
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true") {
+ val plan = stripAQEPlan(spark.sql(query).queryExecution.executedPlan)
+ val explain = new ExtendedExplainInfo()
+ if (expectNative || collation == "UTF8_BINARY") {
+ require(
+ explain.getNativeExpressions(plan).contains(function) &&
+ !explain.getCodegenDispatchExpressions(plan).contains(function),
+ s"$function did not run natively: $plan")
+ } else {
+ require(
+ explain.getCodegenDispatchExpressions(plan).contains(function),
+ s"$function did not use the dispatcher: $plan")
+ }
+ }
+ runExpressionBenchmark(s"$function $collation - $shape", values, query)
Review Comment:
[P2] Make the cardinality conversion explicit here. `values` is an `Int`, while `runExpressionBenchmark` takes a `Long`. Running `./mvnw -B test-compile -Pspark-3.5 -Pstrict-warnings -DskipTests` therefore fails with `implicit numeric widening`, because this profile treats that warning as an error. This prevents the required compilation check from passing even though the benchmark is not executed on Spark 3.5. Pass `values.toLong` at this call.
Evidence: Exact-head CI job https://github.com/apache/datafusion-comet/actions/runs/36747428085/job/110016920898 ran the stated command and reported `CometArrayExpressionBenchmark.scala:196: implicit numeric widening`, followed by `BUILD FAILURE`. Its tested merge tree is identical to the requested head. The receiving parameter is declared `cardinality: Long` in `CometBenchmarkBase.scala:137`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#issuecomment-5876853209
@andygrove Thanks, these are valid points. The four inline requests are addressed in 341174464, with replies in each thread.
On the main scope/performance question: agreed that this does not enable native scans of collated-schema columns. The new kernel serves query-produced/cast collated values, and those input casts can still dispatch. I made that limitation explicit in the description.
I ran the identical `arrayExtremaCollationBenchmark` fixture on #5403 head `ae755c459e0acd95f72a53f538d959e2adec572c` and this PR head `3411744640822ba46c36649b7171fe6df129fd80`, with release JNI libraries, Spark 4.1.3, JDK 17, local[1], and the same 262,144-row inputs. All 24 cases completed on each head, with assertions distinguishing dispatcher extrema on #5403 from native extrema here. The full Comet-versus-Comet mean-time table and methodology are now in the PR description.
The comparison confirms an important tradeoff: Unicode LCASE/LCASE_RTRIM is about 1.9x faster, but short-ASCII collated cases mostly regress, long-ASCII LCASE is roughly level, and BINARY_RTRIM is slower on all three shapes (3.5-3.7x slower for long ASCII). Plain-binary controls improve 1.06-1.45x. So the previous Spark-versus-Comet table was not sufficient to justify a general speedup claim; the new results do not support one either. In particular, the RTRIM result remains a performance concern for the native-enablement decision, not something these test/proto changes fix.
This is one local paired run and includes intervening upstream changes between the old head and rebased PR, not a strict single-patch ablation. Final-source validation passed 14 focused native checks and five Spark 4.1 JVM tests; broader CI is still pending.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: evaluate UTF8 array extrema collations natively [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6170:
URL: https://github.com/apache/datafusion-comet/pull/6170#discussion_r4126091403
##########
native/spark-expr/src/array_funcs/array_extrema.rs:
##########
@@ -0,0 +1,440 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::cmp::Ordering;
+use std::sync::Arc;
+
+use arrow::array::{
+ make_array, make_comparator, new_empty_array, Array, ArrayAccessor, ArrayRef, AsArray,
+ DynComparator, ListArray, MutableArrayData, PrimitiveArray, PrimitiveBuilder, StringArrayType,
+ StructArray, UInt32Array,
+};
+use arrow::buffer::NullBuffer;
+use arrow::compute::{take, SortOptions};
+use arrow::datatypes::{ArrowPrimitiveType, DataType, Float32Type, Float64Type};
+use datafusion::common::{exec_err, Result, ScalarValue};
+use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf};
+use datafusion::logical_expr::{
+ ColumnarValue, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
+};
+use num::Float;
+
+#[derive(Debug, Clone, Copy, Hash, Eq, PartialEq)]
+enum Utf8Collation {
+ Binary,
+ BinaryRtrim,
+ Lcase,
+ LcaseRtrim,
+}
+
+impl Utf8Collation {
+ fn compare(self, mut left: &str, mut right: &str, unicode_version: u32) -> Ordering {
+ if matches!(self, Self::BinaryRtrim | Self::LcaseRtrim) {
+ // Spark RTRIM ignores trailing U+0020, not arbitrary Unicode whitespace.
+ left = left.trim_end_matches(' ');
+ right = right.trim_end_matches(' ');
+ }
+ if matches!(self, Self::Binary | Self::BinaryRtrim) {
+ return left.cmp(right);
+ }
+ fn lower(value: &str, unicode_version: u32) -> impl Iterator<Item = u32> + '_ {
+ value.chars().flat_map(move |c| {
+ let cp = c as u32;
+ let [first, second] = match cp {
+ // Spark treats final sigma as ordinary sigma.
+ 0x3c2 => [0x3c3, 0],
+ // Unicode 17 adds these mappings to the library's Unicode 16 data.
+ 0xa7ce | 0xa7d2 | 0xa7d4 if unicode_version == 17 => [cp + 1, 0],
+ 0x16ea0..=0x16eb8 if unicode_version == 17 => [cp + 0x1b, 0],
Review Comment:
Agreed. Added an explicit table of all 28 Unicode 17 upper/lower pairs in `utf8_lcase_unicode_17_mappings_are_version_gated`. Each pair is checked in both input orders: equal under Unicode 17, and distinct (with opposite ordering) under Unicode 16. The test data is independent of the implementation's range/match arms, so both omissions you described are covered. This passed with the final source in 341174464.
##########
native/proto/src/proto/expr.proto:
##########
@@ -540,6 +540,10 @@ message ScalarFunc {
repeated Expr args = 2;
DataType return_type = 3;
bool fail_on_error = 4;
+ // array_min/max only: collations of every string leaf in the element type,
+ // visiting array elements and struct fields depth-first. Empty means binary ordering.
+ repeated string string_collations = 5;
Review Comment:
Agreed, a dedicated message fits the existing array serializers better. 341174464 adds `ArrayExtrema` with the child, min/max mode, string-leaf collations, and Unicode version. All native extrema now use it; the extrema-only fields on `ScalarFunc` and the planner's function-name special case are removed. Added Scala serialization coverage and a native planner regression for min/max and binary/LCASE behavior. The focused Spark 4.1 JNI/serialization tests passed with the final library.
##########
spark/src/test/resources/sql-tests/expressions/array/array_extrema_collation.sql:
##########
@@ -0,0 +1,74 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- MinSparkVersion: 4.0
+-- Config: spark.comet.expression.ArrayMin.allowIncompatible=false
+-- Config: spark.comet.expression.ArrayMax.allowIncompatible=false
+-- Config: spark.comet.exec.scalaUDF.codegen.enabled=true
+
+statement
+CREATE TABLE test_array_extrema_collation(
+ id int, a string, b string, x double, y double) USING parquet
+
+statement
+INSERT INTO test_array_extrema_collation VALUES
+ (1, 'a', 'B', double('0.0'), double('-0.0')),
+ (2, 'B', 'a', double('-0.0'), double('0.0')),
+ (3, 'A', 'a', double('-0.0'), double('0.0')),
+ (4, NULL, 'B', NULL, double('0.0')),
+ (5, NULL, NULL, NULL, NULL),
+ (6, 'x ', 'x', 1.0, 2.0)
Review Comment:
Added all three pairs in both input orders to the SQL fixture in 341174464: U+0130 versus i + U+0307, final sigma versus capital sigma, and Kelvin sign versus k. They run through the existing native-extrema assertions, including the LCASE first-winner cases. The updated fixture passed against Spark 4.1.3 with the final release JNI library; I have not rerun Spark 4.2 locally, so that remains for the requested CI matrix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org