You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "0lai0 (via GitHub)" <gi...@apache.org> on 2026/09/01 15:26:42 UTC
[PR] feat: add native spark_sequence kernel for integral element types [datafusion-comet]
0lai0 opened a new pull request, #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614
## Which issue does this PR close?
Closes #5349.
## Rationale for this change
Integral `sequence(start, stop[, step])` currently runs through the JVM codegen dispatcher, which allocates two `long[]` per row. This PR adds a native kernel that reserves the Arrow child buffer once per batch. Date and timestamp sequences stay on the dispatcher (timezone / DST / legacy calendar).
## What changes are included in this PR?
- Native `spark_sequence` kernel (`native/spark-expr/src/array_funcs/sequence.rs`) for `Byte` / `Short` / `Int` / `Long`. Matches Spark `Sequence.sequenceLength`, including the overflow-report paths (`2^63`, `2^63+1`, and the internal-error edge).
- `CometSequence` with `CodegenDispatchFallback`: `Compatible()` for integral types, `Unsupported` for date/timestamp so they stay on the codegen dispatcher.
- Error mapping via `ShimSparkErrorConverter`: Spark 3.x throws `IllegalArgumentException("Illegal sequence boundaries: ...")`; Spark 4.x throws `SparkIllegalArgumentException("_LEGACY_ERROR_TEMP_3243")`.
- `CollectionSizeLimitExceeded` now carries a decimal `String` count (can exceed `i64`) and a `function_name` for Spark 4.x. This is the first producer of that error, and it also fixes a latent Spark 3.5 shim bug that passed a Scala tuple as `count` and rendered `(array,N)`.
- Docs: `sequence` marked `Hybrid` in `expressions.md`; audit notes added under `array_funcs.md`.
## How are these changes tested?
- Five unit tests in `sequence.rs`.
- `spark/src/test/resources/sql-tests/expressions/array/sequence.sql`: integral types, default/explicit step, nulls, explode, seven error cases, plus date/timestamp dispatcher coverage.
- `CometCodegenSuite`: integral sequences are native; date sequences still show "JVM codegen dispatcher".
**Criterion** (`cargo bench --bench sequence`, N=8192). Absolute numbers; the kernel is not on `main`.
| Shape | Time | Elems/batch | ns / elem |
|------------------------------|--------------|-------------|-----------|
| `short_2_elems` | 49.83 µs | 16,384 | 3.04 |
| `short_5_elems` | 55.35 µs | 40,960 | 1.35 |
| `long_365_elems` | 1.318 ms | 2,990,080 | 0.44 |
| `long_10000_elems` | 34.87 ms | 81,920,000 | 0.43 |
| `descending_365_elems` | 1.324 ms | 2,990,080 | 0.44 |
| `zero_step_start_eq_stop` | 35.70 µs | 8,192 | 4.36 |
| `sparse_nulls_365_elems` | 1.200 ms | 2,691,072 | 0.45 |
| `dense_nulls_365_elems` | 673.2 µs | 1,495,040 | 0.45 |
| `error_illegal_boundaries` | 673 ns | (errors) | - |
**Spark** (`CometSequenceBenchmark`, 8192 rows, Apple M5, Spark 4.1.3 / Scala 2.13)
| Shape | Spark best (ms) | Comet best (ms) | speedup |
|--------------------------------|-----------------|-----------------|---------|
| `seq_short_5_elems` | 15 | 5 | 2.9X |
| `seq_spine_365_elems` | 14 | 6 | 2.2X |
| `seq_long_10000_elems` | 59 | 57 | 1.0X |
| `seq_descending_default_step` | 13 | 6 | 2.3X |
| `seq_explicit_step_7` | 11 | 3 | 3.2X |
| `seq_sparse_nulls_365_elems` | 12 | 5 | 2.4X |
| `seq_date_spine_dispatcher` | 12 | 12 | 1.0X |
`seq_date_spine_dispatcher` is a control (date path unchanged). `seq_long_10000_elems` is memory-bandwidth-bound at 82M elements/batch. Typical speedup is 2X–3X on the shorter-list shapes the issue targets.
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#issuecomment-5552940259
Thanks @0lai0 and @sunchao for the review
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3914096130
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -924,4 +924,38 @@ object CometArraySort extends CometCodegenDispatch[ArraySort]
object CometZipWith extends CometCodegenDispatch[ZipWith]
-object CometSequence extends CometCodegenDispatch[Sequence]
+object CometSequence extends CometExpressionSerde[Sequence] with CodegenDispatchFallback {
+
+ private val temporalUnsupportedReason =
+ "date and timestamp element types run through the JVM codegen dispatcher"
+
+ override def getSupportLevel(expr: Sequence): SupportLevel = expr.start.dataType match {
+ case ByteType | ShortType | IntegerType | LongType => Compatible()
+ case DateType | TimestampType | TimestampNTZType =>
+ // Temporal sequences step through timezone/DST/legacy-calendar arithmetic
+ // (https://github.com/apache/datafusion-comet/issues/5349), so they stay on the JVM
+ // codegen dispatcher.
+ Unsupported(Some(temporalUnsupportedReason))
+ case other =>
+ Unsupported(Some(s"sequence with element type $other is not supported natively"))
+ }
+
+ override def getUnsupportedReasons(): Seq[String] = Seq(temporalUnsupportedReason)
+
+ override def convert(
+ expr: Sequence,
+ inputs: Seq[Attribute],
+ binding: Boolean): Option[ExprOuterClass.Expr] = {
+ val startExprProto = exprToProto(expr.start, inputs, binding)
+ val stopExprProto = exprToProto(expr.stop, inputs, binding)
+ // With no step argument the native kernel computes Spark's per-row default,
+ // `start <= stop ? 1 : -1`, which cannot be expressed as a plan-time literal.
+ val argProtos = Seq(startExprProto, stopExprProto) ++
+ expr.stepOpt.map(exprToProto(_, inputs, binding))
+ scalarFunctionExprToProtoWithReturnType(
Review Comment:
Agreed, the native UDF path can't short-circuit per row. Shapes with non-leaf arguments now fall back to the codegen dispatcher via hasLeafArgsOnly. I added that query to sequence.sql and a routing check in CometCodegenSuite. Thanks @sunchao for reiview.
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3914051112
##########
spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala:
##########
@@ -158,10 +158,21 @@ trait ShimSparkErrorConverter {
Some(QueryExecutionErrors.exceedMapSizeLimitError(params("size").toString.toInt))
case "CollectionSizeLimitExceeded" =>
- // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4
+ // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4; pass the
+ // decimal string through since the reported length can exceed Long range.
Some(
QueryExecutionErrors.createArrayWithElementsExceedLimitError(
- params("numElements").toString.toLong))
+ params("numElements").toString))
+
+ case "SequenceIllegalBoundaries" =>
+ // Spark 3.x codegen throws a plain IllegalArgumentException for sequence boundaries.
+ Some(
+ new IllegalArgumentException(
+ s"Illegal sequence boundaries: ${params("start")} to ${params("stop")} " +
+ s"by ${params("step")}"))
+
+ case "Internal" =>
Review Comment:
Oh thats a good point, this is indeed a user-visible change beyond sequence.
I have added a bullet to the PR description under What changes are included calling out that the new Internal arm affects ~20 existing native expressions and changes the message from SparkException(message, <text>) to [INTERNAL_ERROR] <text>. 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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#issuecomment-5553105326
This merged while I was still working through the review above, so none of it was blocking. I have moved the actionable items into #5712 so they do not get lost: the perf crossover above roughly a thousand elements per row and the missing dispatcher benchmark arm, the unbounded per-batch allocation outside the memory pool, the missing `SequenceBatchTooLarge` test, and the documentation items.
To be clear about what I did verify, since it is the reassuring part: all five of Spark's `sequenceLength` failure paths match byte for byte through the native kernel on both 3.5 and 4.1, the fuzz found no divergences, and fourteen downstream consumers of the produced list agree with Spark. Nice work on the error parity, it is exact.
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3913994254
##########
native/spark-expr/src/array_funcs/sequence.rs:
##########
@@ -0,0 +1,388 @@
+// 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.
+
+// Spark-compatible sequence(start, stop[, step]) for integral element types.
+//
+// Mirrors the code Spark's whole-stage codegen emits for `Sequence`
+// (`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala`,
+// identical from 3.4.3 through 4.1.1): the boundary check and `Sequence.sequenceLength` decide
+// per row how many elements to generate, then elements are `start + step * i`. Unlike the JVM
+// path, which allocates two `long[]` per row and copies every element three times, this kernel
+// reserves the Arrow child buffer once for the whole batch and writes each element exactly once.
+//
+// Date/timestamp sequences are not handled here; the Scala serde only routes IntegralType
+// sequences to this function.
+
+use std::sync::Arc;
+
+use arrow::array::{Array, ArrayRef, ListArray, NullBufferBuilder, PrimitiveArray};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::{
+ ArrowPrimitiveType, DataType, FieldRef, Int16Type, Int32Type, Int64Type, Int8Type,
+};
+use datafusion::common::cast::as_primitive_array;
+use datafusion::common::{exec_err, DataFusionError, Result, ScalarValue};
+use datafusion::logical_expr::ColumnarValue;
+
+use crate::SparkError;
+
+/// Spark's ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH (Integer.MAX_VALUE - 15).
+const MAX_ROUNDED_ARRAY_LENGTH: i128 = (i32::MAX - 15) as i128;
+
+pub fn spark_sequence(args: &[ColumnarValue], data_type: &DataType) -> Result<ColumnarValue> {
+ let child_field = match data_type {
+ DataType::List(field) => Arc::clone(field),
+ other => return exec_err!("spark_sequence expects a List return type, got {other:?}"),
+ };
+ if args.len() != 2 && args.len() != 3 {
+ return exec_err!(
+ "spark_sequence expects 2 or 3 arguments, got {}",
+ args.len()
+ );
+ }
+
+ let all_scalar = args
+ .iter()
+ .all(|arg| matches!(arg, ColumnarValue::Scalar(_)));
+ let arrays = ColumnarValue::values_to_arrays(args)?;
+ let step = arrays.get(2);
+
+ let result = match child_field.data_type() {
+ DataType::Int8 => {
+ sequence_integral::<Int8Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i8)
+ }
+ DataType::Int16 => {
+ sequence_integral::<Int16Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i16)
+ }
+ DataType::Int32 => {
+ sequence_integral::<Int32Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i32)
+ }
+ DataType::Int64 => {
+ sequence_integral::<Int64Type>(&arrays[0], &arrays[1], step, child_field, |v| v)
+ }
+ other => exec_err!("spark_sequence does not support element type {other:?}"),
+ }?;
+
+ if all_scalar {
+ Ok(ColumnarValue::Scalar(ScalarValue::try_from_array(
+ &result, 0,
+ )?))
+ } else {
+ Ok(ColumnarValue::Array(result))
+ }
+}
+
+fn sequence_integral<T: ArrowPrimitiveType>(
+ start: &ArrayRef,
+ stop: &ArrayRef,
+ step: Option<&ArrayRef>,
+ child_field: FieldRef,
+ from_i64: impl Fn(i64) -> T::Native,
+) -> Result<ArrayRef>
+where
+ T::Native: Into<i64>,
+{
+ let start = as_primitive_array::<T>(start)?;
+ let stop = as_primitive_array::<T>(stop)?;
+ let step = step.map(|arr| as_primitive_array::<T>(arr)).transpose()?;
+ let num_rows = start.len();
+
+ let row_is_null = |row: usize| {
+ start.is_null(row) || stop.is_null(row) || step.is_some_and(|arr| arr.is_null(row))
+ };
+ // With no explicit step, Spark uses `start <= stop ? 1 : -1` per row, so the direction
+ // always matches the bounds and the boundary check below cannot fail.
+ let row_step = |row: usize, start: i64, stop: i64| -> i64 {
+ match step {
+ Some(arr) => arr.value(row).into(),
+ None => {
+ if start <= stop {
+ 1
+ } else {
+ -1
+ }
+ }
+ }
+ };
+
+ // First pass: compute per-row lengths so the child buffer can be reserved once for the
+ // whole batch. Valid rows always produce at least one element, so length 0 marks a null row.
+ let mut lengths: Vec<usize> = Vec::with_capacity(num_rows);
+ let mut total: usize = 0;
+ for row in 0..num_rows {
+ if row_is_null(row) {
+ lengths.push(0);
+ continue;
+ }
+ let s: i64 = start.value(row).into();
+ let e: i64 = stop.value(row).into();
+ let len = sequence_length(s, e, row_step(row, s, e))?;
+ total += len;
+ lengths.push(len);
+ }
+ // Comet-specific ceiling: the sum of every row's length in one Arrow batch must fit in
+ // the i32 offset buffer. Spark has no equivalent guard because it stores each row as its
+ // own `long[]`. Report it with the same size-limit error class Spark would raise for a
+ // single overlong row so the user sees a familiar message.
+ if total > i32::MAX as usize {
+ return Err(DataFusionError::External(Box::new(
+ SparkError::CollectionSizeLimitExceeded {
+ num_elements: total.to_string(),
+ max_elements: i32::MAX as i64,
+ function_name: "sequence".to_string(),
+ },
+ )));
+ }
+
+ // Second pass: write elements straight into the child buffer and push offsets. The
+ // batch-total check above guarantees `values.len() <= i32::MAX` at every iteration, so
+ // the offset push cannot overflow.
+ let mut values: Vec<T::Native> = Vec::with_capacity(total);
Review Comment:
Thanks @andygrove for review.
I switched to `try_reserve_exact` and added SequenceBatchTooLarge pointing at spark.comet.batchSize. Documented the per-batch ceiling in `array_funcs.md` and a new Limitations section. Leaf-arg integral sequences stay Compatible().
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3910237757
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -924,4 +924,38 @@ object CometArraySort extends CometCodegenDispatch[ArraySort]
object CometZipWith extends CometCodegenDispatch[ZipWith]
-object CometSequence extends CometCodegenDispatch[Sequence]
+object CometSequence extends CometExpressionSerde[Sequence] with CodegenDispatchFallback {
+
+ private val temporalUnsupportedReason =
+ "date and timestamp element types run through the JVM codegen dispatcher"
+
+ override def getSupportLevel(expr: Sequence): SupportLevel = expr.start.dataType match {
+ case ByteType | ShortType | IntegerType | LongType => Compatible()
+ case DateType | TimestampType | TimestampNTZType =>
+ // Temporal sequences step through timezone/DST/legacy-calendar arithmetic
+ // (https://github.com/apache/datafusion-comet/issues/5349), so they stay on the JVM
+ // codegen dispatcher.
+ Unsupported(Some(temporalUnsupportedReason))
+ case other =>
+ Unsupported(Some(s"sequence with element type $other is not supported natively"))
+ }
+
+ override def getUnsupportedReasons(): Seq[String] = Seq(temporalUnsupportedReason)
+
+ override def convert(
+ expr: Sequence,
+ inputs: Seq[Attribute],
+ binding: Boolean): Option[ExprOuterClass.Expr] = {
+ val startExprProto = exprToProto(expr.start, inputs, binding)
+ val stopExprProto = exprToProto(expr.stop, inputs, binding)
+ // With no step argument the native kernel computes Spark's per-row default,
+ // `start <= stop ? 1 : -1`, which cannot be expressed as a plan-time literal.
+ val argProtos = Seq(startExprProto, stopExprProto) ++
+ expr.stepOpt.map(exprToProto(_, inputs, binding))
+ scalarFunctionExprToProtoWithReturnType(
Review Comment:
[P2] Preserve null short-circuiting before evaluating later arguments
Could this lowering preserve `Sequence`'s left-to-right null guards? For a Parquet table `t(s INT, k INT)` containing `(NULL, -1)` and `(1, 1)`, consider `SELECT sequence(s, size(sequence(1, 5, k))) FROM t`. Spark returns `NULL` for the first row without evaluating the inner sequence, and `[1,2,3,4,5]` for the second. Here both sequences become scalar UDFs, whose arguments DataFusion evaluates over the batch before calling the outer kernel. The inner `sequence(1, 5, -1)` therefore throws before the outer `row_is_null` check can discard that row. The previous dispatcher kept the whole expression tree inside Spark's guarded evaluation. Could we retain those guards or dispatch such shapes, and add a composed null/error regression case?
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3914045727
##########
spark/src/test/resources/sql-tests/expressions/array/sequence.sql:
##########
@@ -15,17 +15,131 @@
-- specific language governing permissions and limitations
-- under the License.
--- Routes sequence through the codegen dispatcher so behavior matches Spark exactly.
+-- sequence(start, stop[, step]) for integral element types runs on the native kernel
+-- (https://github.com/apache/datafusion-comet/issues/5349). Date and timestamp sequences
+-- stay on the JVM codegen dispatcher and are exercised at the bottom of this file.
statement
-CREATE TABLE test_sequence(a int, b int) USING parquet
+CREATE TABLE test_sequence(
+ b_start tinyint, b_stop tinyint, b_step tinyint,
+ s_start smallint, s_stop smallint, s_step smallint,
+ i_start int, i_stop int, i_step int,
+ l_start bigint, l_stop bigint, l_step bigint)
+USING parquet
+-- Row 2 descends, row 3 has start == stop, rows 4-6 carry NULLs in each argument position.
statement
-INSERT INTO test_sequence VALUES (1, 5), (5, 1), (3, 3), (NULL, 5)
+INSERT INTO test_sequence VALUES
+ (1Y, 5Y, 1Y, 1S, 5S, 1S, 1, 10, 3, 1L, 5L, 2L),
+ (-3Y, -1Y, 1Y, 100S, 90S, -2S, 20, 2, -6, 9223372036854775802L, 9223372036854775807L, 1L),
+ (0Y, 0Y, 0Y, -5S, -5S, 0S, 7, 7, 0, -9223372036854775808L, -9223372036854775800L, 3L),
+ (NULL, 5Y, 1Y, NULL, 5S, 1S, NULL, 10, 1, NULL, 5L, 1L),
+ (1Y, NULL, 1Y, 1S, NULL, 1S, 1, NULL, 1, 1L, NULL, 1L),
+ (1Y, 5Y, NULL, 1S, 5S, NULL, 1, 10, NULL, 1L, 5L, NULL)
+
+-- ============================================================================
+-- Explicit step, all four integral types
+-- ============================================================================
+
+query
+SELECT sequence(i_start, i_stop, i_step) FROM test_sequence
query
-SELECT a, b, sequence(a, b) FROM test_sequence
+SELECT sequence(l_start, l_stop, l_step) FROM test_sequence
+
+-- Column step for the narrow integral types exercises the Byte/Short monomorphizations
+-- of the native kernel, not just the literal-step shape.
+query
+SELECT sequence(b_start, b_stop, b_step) FROM test_sequence
+
+query
+SELECT sequence(s_start, s_stop, s_step) FROM test_sequence
+
+-- ============================================================================
+-- Default step: per-row start <= stop ? 1 : -1, both directions in one column
+-- ============================================================================
+
+query
+SELECT sequence(b_start, b_stop), sequence(s_start, s_stop) FROM test_sequence
+
+query
+SELECT sequence(i_start, i_stop), sequence(l_start, l_stop) FROM test_sequence
+
+-- ============================================================================
+-- Literal and mixed literal/column arguments
+-- ============================================================================
+
+query
+SELECT sequence(1, 10), sequence(10, 1), sequence(5, 5), sequence(5, 5, 0)
--- literal arguments with step
query
SELECT sequence(1, 5), sequence(5, 1, -1), sequence(1, 10, 2)
+
+query
+SELECT sequence(1L, 9L, 2L), sequence(-128Y, -120Y), sequence(32760S, 32767S)
+
+-- On row 2 the source row is (i_start=20, i_stop=2, i_step=-6), so the literal-step column
+-- asks for sequence(1, 2, 2) = [1] while the default-step column asks for sequence(20, 25)
+-- = [20, 21, 22, 23, 24, 25]. The two columns disagreeing in direction on the same row is
+-- intentional coverage, not an oversight.
+query
+SELECT sequence(1, i_stop, 2), sequence(i_start, 25) FROM test_sequence WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+query
+SELECT sequence(CAST(NULL AS int), 5), sequence(1, CAST(NULL AS int)), sequence(1, 5, CAST(NULL AS int))
+
+-- Integer.MIN_VALUE/MAX_VALUE bounds for int, and a sequence spanning zero
+query
+SELECT sequence(2147483642, 2147483647), sequence(-2147483648, -2147483643), sequence(-3, 3, 3)
+
+-- ============================================================================
+-- sequence feeding explode, the common date-spine shape (with integers)
+-- ============================================================================
+
+query
+SELECT i_start, x FROM test_sequence LATERAL VIEW explode(sequence(i_start, i_stop)) AS x WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+-- ============================================================================
+-- Error paths: step direction contradicts bounds, or zero step with start != stop
+-- ============================================================================
+
+query expect_error(Illegal sequence boundaries: 1 to 5 by -1)
Review Comment:
Thanks, Added all three to sequence.sql: full byte/short range, Int32 step overflow, and CASE WHEN guarded branch.
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3906787356
##########
spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala:
##########
@@ -158,10 +158,21 @@ trait ShimSparkErrorConverter {
Some(QueryExecutionErrors.exceedMapSizeLimitError(params("size").toString.toInt))
case "CollectionSizeLimitExceeded" =>
- // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4
+ // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4; pass the
+ // decimal string through since the reported length can exceed Long range.
Some(
QueryExecutionErrors.createArrayWithElementsExceedLimitError(
- params("numElements").toString.toLong))
+ params("numElements").toString))
+
+ case "SequenceIllegalBoundaries" =>
+ // Spark 3.x codegen throws a plain IllegalArgumentException for sequence boundaries.
+ Some(
+ new IllegalArgumentException(
+ s"Illegal sequence boundaries: ${params("start")} to ${params("stop")} " +
+ s"by ${params("step")}"))
+
+ case "Internal" =>
Review Comment:
The `case "Internal"` arm changes behavior well beyond `sequence`. There are around twenty `SparkError::Internal` producers today in `temporal.rs`, `numeric.rs`, `conversion_funcs/string.rs` and `rlike.rs`, and all of them previously fell through to the `None` branch in `SparkErrorConverter`, which renders as `new SparkException(msgParams.mkString(", "))`, so users saw `(message,<text>)`. After this they all become `[INTERNAL_ERROR] <text>`.
That is a clear improvement and I am not asking you to revert it. Could you call it out in the PR description though? It is a user-visible message change for a set of expressions that have nothing to do with `sequence`, and right now the "How are these changes tested?" section does not mention it. It would also be worth a pass over the Spark SQL suite diffs to confirm nothing was matching on the old shape.
The same arm is added to the 3.5 and 4.x shims, so this applies to all three.
##########
spark/src/test/resources/sql-tests/expressions/array/sequence.sql:
##########
@@ -15,17 +15,131 @@
-- specific language governing permissions and limitations
-- under the License.
--- Routes sequence through the codegen dispatcher so behavior matches Spark exactly.
+-- sequence(start, stop[, step]) for integral element types runs on the native kernel
+-- (https://github.com/apache/datafusion-comet/issues/5349). Date and timestamp sequences
+-- stay on the JVM codegen dispatcher and are exercised at the bottom of this file.
statement
-CREATE TABLE test_sequence(a int, b int) USING parquet
+CREATE TABLE test_sequence(
+ b_start tinyint, b_stop tinyint, b_step tinyint,
+ s_start smallint, s_stop smallint, s_step smallint,
+ i_start int, i_stop int, i_step int,
+ l_start bigint, l_stop bigint, l_step bigint)
+USING parquet
+-- Row 2 descends, row 3 has start == stop, rows 4-6 carry NULLs in each argument position.
statement
-INSERT INTO test_sequence VALUES (1, 5), (5, 1), (3, 3), (NULL, 5)
+INSERT INTO test_sequence VALUES
+ (1Y, 5Y, 1Y, 1S, 5S, 1S, 1, 10, 3, 1L, 5L, 2L),
+ (-3Y, -1Y, 1Y, 100S, 90S, -2S, 20, 2, -6, 9223372036854775802L, 9223372036854775807L, 1L),
+ (0Y, 0Y, 0Y, -5S, -5S, 0S, 7, 7, 0, -9223372036854775808L, -9223372036854775800L, 3L),
+ (NULL, 5Y, 1Y, NULL, 5S, 1S, NULL, 10, 1, NULL, 5L, 1L),
+ (1Y, NULL, 1Y, 1S, NULL, 1S, 1, NULL, 1, 1L, NULL, 1L),
+ (1Y, 5Y, NULL, 1S, 5S, NULL, 1, 10, NULL, 1L, 5L, NULL)
+
+-- ============================================================================
+-- Explicit step, all four integral types
+-- ============================================================================
+
+query
+SELECT sequence(i_start, i_stop, i_step) FROM test_sequence
query
-SELECT a, b, sequence(a, b) FROM test_sequence
+SELECT sequence(l_start, l_stop, l_step) FROM test_sequence
+
+-- Column step for the narrow integral types exercises the Byte/Short monomorphizations
+-- of the native kernel, not just the literal-step shape.
+query
+SELECT sequence(b_start, b_stop, b_step) FROM test_sequence
+
+query
+SELECT sequence(s_start, s_stop, s_step) FROM test_sequence
+
+-- ============================================================================
+-- Default step: per-row start <= stop ? 1 : -1, both directions in one column
+-- ============================================================================
+
+query
+SELECT sequence(b_start, b_stop), sequence(s_start, s_stop) FROM test_sequence
+
+query
+SELECT sequence(i_start, i_stop), sequence(l_start, l_stop) FROM test_sequence
+
+-- ============================================================================
+-- Literal and mixed literal/column arguments
+-- ============================================================================
+
+query
+SELECT sequence(1, 10), sequence(10, 1), sequence(5, 5), sequence(5, 5, 0)
--- literal arguments with step
query
SELECT sequence(1, 5), sequence(5, 1, -1), sequence(1, 10, 2)
+
+query
+SELECT sequence(1L, 9L, 2L), sequence(-128Y, -120Y), sequence(32760S, 32767S)
+
+-- On row 2 the source row is (i_start=20, i_stop=2, i_step=-6), so the literal-step column
+-- asks for sequence(1, 2, 2) = [1] while the default-step column asks for sequence(20, 25)
+-- = [20, 21, 22, 23, 24, 25]. The two columns disagreeing in direction on the same row is
+-- intentional coverage, not an oversight.
+query
+SELECT sequence(1, i_stop, 2), sequence(i_start, 25) FROM test_sequence WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+query
+SELECT sequence(CAST(NULL AS int), 5), sequence(1, CAST(NULL AS int)), sequence(1, 5, CAST(NULL AS int))
+
+-- Integer.MIN_VALUE/MAX_VALUE bounds for int, and a sequence spanning zero
+query
+SELECT sequence(2147483642, 2147483647), sequence(-2147483648, -2147483643), sequence(-3, 3, 3)
+
+-- ============================================================================
+-- sequence feeding explode, the common date-spine shape (with integers)
+-- ============================================================================
+
+query
+SELECT i_start, x FROM test_sequence LATERAL VIEW explode(sequence(i_start, i_stop)) AS x WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+-- ============================================================================
+-- Error paths: step direction contradicts bounds, or zero step with start != stop
+-- ============================================================================
+
+query expect_error(Illegal sequence boundaries: 1 to 5 by -1)
Review Comment:
The fixture is thorough on the error paths. There are three shapes I think are worth pinning down that it does not reach today.
The first is full narrow-type range, `sequence(-128Y, 127Y)` and `sequence(-32768S, 32767S)`. That is the one place where Spark's `arr(i) = start + step * num.fromInt(i)` genuinely wraps at 8 and 16 bits while your kernel accumulates in i64 and truncates on the way out. The two agree because the true value is always in range, but it is the case I would most want a regression test on, and `sequence(-128Y, -120Y)` does not get there.
The second is a step whose product with the index overflows int, something like `sequence(-2147483648, 2147483647, 1073741824)`, which exercises the `Int32` monomorphization at the boundary.
The third is `sequence` under a `CASE WHEN` where the throwing branch is not taken, for example `SELECT CASE WHEN step > 0 THEN sequence(1, 5, step) ELSE array(-1) END FROM t` with rows carrying negative and zero steps. DataFusion filters the batch before evaluating each `then` branch so this works today, but it is the one construct where an eagerly evaluated throwing expression would diverge from Spark, and it would be cheap insurance.
I ran all three locally against this branch and they pass on both 4.1 and 3.5, so this is about locking the behavior in rather than chasing a suspected bug.
##########
native/spark-expr/src/array_funcs/sequence.rs:
##########
@@ -0,0 +1,388 @@
+// 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.
+
+// Spark-compatible sequence(start, stop[, step]) for integral element types.
+//
+// Mirrors the code Spark's whole-stage codegen emits for `Sequence`
+// (`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala`,
+// identical from 3.4.3 through 4.1.1): the boundary check and `Sequence.sequenceLength` decide
+// per row how many elements to generate, then elements are `start + step * i`. Unlike the JVM
+// path, which allocates two `long[]` per row and copies every element three times, this kernel
+// reserves the Arrow child buffer once for the whole batch and writes each element exactly once.
+//
+// Date/timestamp sequences are not handled here; the Scala serde only routes IntegralType
+// sequences to this function.
+
+use std::sync::Arc;
+
+use arrow::array::{Array, ArrayRef, ListArray, NullBufferBuilder, PrimitiveArray};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::{
+ ArrowPrimitiveType, DataType, FieldRef, Int16Type, Int32Type, Int64Type, Int8Type,
+};
+use datafusion::common::cast::as_primitive_array;
+use datafusion::common::{exec_err, DataFusionError, Result, ScalarValue};
+use datafusion::logical_expr::ColumnarValue;
+
+use crate::SparkError;
+
+/// Spark's ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH (Integer.MAX_VALUE - 15).
+const MAX_ROUNDED_ARRAY_LENGTH: i128 = (i32::MAX - 15) as i128;
+
+pub fn spark_sequence(args: &[ColumnarValue], data_type: &DataType) -> Result<ColumnarValue> {
+ let child_field = match data_type {
+ DataType::List(field) => Arc::clone(field),
+ other => return exec_err!("spark_sequence expects a List return type, got {other:?}"),
+ };
+ if args.len() != 2 && args.len() != 3 {
+ return exec_err!(
+ "spark_sequence expects 2 or 3 arguments, got {}",
+ args.len()
+ );
+ }
+
+ let all_scalar = args
+ .iter()
+ .all(|arg| matches!(arg, ColumnarValue::Scalar(_)));
+ let arrays = ColumnarValue::values_to_arrays(args)?;
+ let step = arrays.get(2);
+
+ let result = match child_field.data_type() {
+ DataType::Int8 => {
+ sequence_integral::<Int8Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i8)
+ }
+ DataType::Int16 => {
+ sequence_integral::<Int16Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i16)
+ }
+ DataType::Int32 => {
+ sequence_integral::<Int32Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i32)
+ }
+ DataType::Int64 => {
+ sequence_integral::<Int64Type>(&arrays[0], &arrays[1], step, child_field, |v| v)
+ }
+ other => exec_err!("spark_sequence does not support element type {other:?}"),
+ }?;
+
+ if all_scalar {
+ Ok(ColumnarValue::Scalar(ScalarValue::try_from_array(
+ &result, 0,
+ )?))
+ } else {
+ Ok(ColumnarValue::Array(result))
+ }
+}
+
+fn sequence_integral<T: ArrowPrimitiveType>(
+ start: &ArrayRef,
+ stop: &ArrayRef,
+ step: Option<&ArrayRef>,
+ child_field: FieldRef,
+ from_i64: impl Fn(i64) -> T::Native,
+) -> Result<ArrayRef>
+where
+ T::Native: Into<i64>,
+{
+ let start = as_primitive_array::<T>(start)?;
+ let stop = as_primitive_array::<T>(stop)?;
+ let step = step.map(|arr| as_primitive_array::<T>(arr)).transpose()?;
+ let num_rows = start.len();
+
+ let row_is_null = |row: usize| {
+ start.is_null(row) || stop.is_null(row) || step.is_some_and(|arr| arr.is_null(row))
+ };
+ // With no explicit step, Spark uses `start <= stop ? 1 : -1` per row, so the direction
+ // always matches the bounds and the boundary check below cannot fail.
+ let row_step = |row: usize, start: i64, stop: i64| -> i64 {
+ match step {
+ Some(arr) => arr.value(row).into(),
+ None => {
+ if start <= stop {
+ 1
+ } else {
+ -1
+ }
+ }
+ }
+ };
+
+ // First pass: compute per-row lengths so the child buffer can be reserved once for the
+ // whole batch. Valid rows always produce at least one element, so length 0 marks a null row.
+ let mut lengths: Vec<usize> = Vec::with_capacity(num_rows);
+ let mut total: usize = 0;
+ for row in 0..num_rows {
+ if row_is_null(row) {
+ lengths.push(0);
+ continue;
+ }
+ let s: i64 = start.value(row).into();
+ let e: i64 = stop.value(row).into();
+ let len = sequence_length(s, e, row_step(row, s, e))?;
+ total += len;
+ lengths.push(len);
+ }
+ // Comet-specific ceiling: the sum of every row's length in one Arrow batch must fit in
+ // the i32 offset buffer. Spark has no equivalent guard because it stores each row as its
+ // own `long[]`. Report it with the same size-limit error class Spark would raise for a
+ // single overlong row so the user sees a familiar message.
+ if total > i32::MAX as usize {
+ return Err(DataFusionError::External(Box::new(
+ SparkError::CollectionSizeLimitExceeded {
+ num_elements: total.to_string(),
+ max_elements: i32::MAX as i64,
+ function_name: "sequence".to_string(),
+ },
+ )));
+ }
+
+ // Second pass: write elements straight into the child buffer and push offsets. The
+ // batch-total check above guarantees `values.len() <= i32::MAX` at every iteration, so
+ // the offset push cannot overflow.
+ let mut values: Vec<T::Native> = Vec::with_capacity(total);
Review Comment:
Nice work on the error-parity side of this, the `sequenceLength` port matches Spark on all three failure paths and I checked it against 3.4.3 through 4.1.1.
The thing I keep coming back to is `Vec::with_capacity(total)`. `sequence` is the first expression we have made native where the output size is unbounded relative to the input size. Every other `with_capacity` in `array_funcs/` is sized by `row_count` or `args.len()`, but here `total` is the sum of every row's generated length, so a single batch can ask for up to `i32::MAX` elements, which is 16 GiB for `bigint`. Your own benchmark shows the shape: `seq_long_10000_elems` materializes 8192 x 10000 x 8 bytes, so 655 MB in one allocation, where Spark holds one row's `long[]` at a time. That is also the one row in your table with no speedup, which makes me wonder whether the large-per-row case is paying for itself at all.
Two things I would like to see. Could the allocation go through `try_reserve` so an oversized batch surfaces as a query error rather than an allocator abort that takes the executor down? And could the batch ceiling be documented somewhere the user can find it?
On the ceiling specifically, the message a user gets today is misleading. `sequence(0, 262143)` over a full 8192-row batch lands on exactly `2147483648` total elements and trips the check, even though every individual array is well inside Spark's limit and Spark itself would run the query. The shim ignores the `max_elements` you pass, so the message reads "Can't create array with 2147483648 elements which exceeding the array size limit 2147483632", which points the user at a per-array limit that they have not actually exceeded and gives them nothing actionable. The real fix on their side is to lower `spark.comet.batchSize`. Given that, is `Compatible()` the right support level, or should this at least get a compatible note and a line in the audit entry?
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3921074910
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometSequenceBenchmark.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.benchmark
+
+/**
+ * Benchmark to measure performance of Comet's native `sequence` kernel against Spark's codegen
+ * (issue #5349). Integral shapes run natively under Comet; the date case stays on the JVM codegen
+ * dispatcher in both arms and is included to show that path is unchanged. To run this benchmark:
+ * {{{
+ * SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometSequenceBenchmark
+ * }}}
+ * Results will be written to "spark/benchmarks/CometSequenceBenchmark-**results.txt".
+ */
+object CometSequenceBenchmark extends CometBenchmarkBase {
+
+ private val sequenceQueries = List(
+ ("seq_short_5_elems", "SELECT sequence(c_start, c_start + 4) FROM parquetV1Table"),
+ ("seq_spine_365_elems", "SELECT sequence(c_start, c_start + 364) FROM parquetV1Table"),
Review Comment:
[P2] Make the integral benchmark exercise native sequence
Could you materialize the integral endpoints as columns in the prepared Parquet table, then verify that these cases use `spark_sequence` before timing them? Every integral query here passes an arithmetic expression such as `c_start + 4` or `c_null_start + 364`. `argsAreLiteralsOrRefs` rejects those arguments, so the complete `Sequence` goes through the JVM dispatcher, or falls back to Spark if dispatch is unavailable. `runExpressionBenchmark` only checks Comet operators and does not catch expression dispatch. These queries therefore cannot measure this native kernel's Spark-versus-Comet benefit. Could you refresh the comparison with leaf arguments and retain the date case as a dispatcher control?
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3914681406
##########
spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala:
##########
@@ -210,6 +210,63 @@ class CometCodegenSuite
}
}
+ private def withSequenceTable(f: => Unit): Unit = {
+ withTable("t") {
+ sql("CREATE TABLE t (a INT, b INT, d DATE) USING parquet")
+ sql("INSERT INTO t VALUES (1, 5, DATE'2024-01-01'), (9, 2, DATE'2024-03-01')")
+ withSQLConf(
+ CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true",
+ CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_PROJECT_ENABLED.key -> "true",
+ CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key ->
+ CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE)(f)
+ }
+ }
+
+ test("sequence with leaf integral args runs natively") {
+ // Integral sequence with column-reference/literal args lowers to the native spark_sequence
+ // kernel; no codegen-dispatch marker should appear.
+ withSequenceTable {
+ val df = sql("SELECT sequence(a, b), sequence(a, b, 2) FROM t")
Review Comment:
Thanks @sunchao, both P2 items addressed in the latest push.
1. Native gate is now Literal | Attribute | BoundReference only (argsAreLiteralsOrRefs). Nested calls, CASE WHEN, and zero-arg UDFs fall back to the dispatcher.
2. Fixture uses sign-correct stp so sequence(a, b, stp) is legal on both rows.
Added sequence with zero-arg UDF stop routes through the dispatcher (comet_seq_stopper()).
Null-short-circuit / CASE cases stay in sequence.sql.
Local validation:
CometCodegenSuite: 178/178 pass (4 sequence tests, incl. zero-arg UDF → dispatcher)
CometSqlFileTestSuite: expressions/array/sequence.sql + sequence_ansi.sql pass
cargo test -p datafusion-comet-spark-expr array_funcs::sequence: 5/5 pass
test-compile green on -Pspark-3.4, -Pspark-3.5, -Pspark-4.0 (shim SequenceBatchTooLarge)
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead merged PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3914120252
##########
spark/src/test/scala/org/apache/comet/CometCodegenSuite.scala:
##########
@@ -210,6 +210,63 @@ class CometCodegenSuite
}
}
+ private def withSequenceTable(f: => Unit): Unit = {
+ withTable("t") {
+ sql("CREATE TABLE t (a INT, b INT, d DATE) USING parquet")
+ sql("INSERT INTO t VALUES (1, 5, DATE'2024-01-01'), (9, 2, DATE'2024-03-01')")
+ withSQLConf(
+ CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true",
+ CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_PROJECT_ENABLED.key -> "true",
+ CometConf.COMET_EXTENDED_EXPLAIN_FORMAT.key ->
+ CometConf.COMET_EXTENDED_EXPLAIN_FORMAT_VERBOSE)(f)
+ }
+ }
+
+ test("sequence with leaf integral args runs natively") {
+ // Integral sequence with column-reference/literal args lowers to the native spark_sequence
+ // kernel; no codegen-dispatch marker should appear.
+ withSequenceTable {
+ val df = sql("SELECT sequence(a, b), sequence(a, b, 2) FROM t")
Review Comment:
[P2] Use legal bounds in the native-path fixture
`withSequenceTable` also inserts `(a, b) = (9, 2)`, so the second expression becomes `sequence(9, 2, 2)`. Spark rejects a positive step with descending bounds, and `checkSparkAnswerAndOperator` first collects the Spark reference result with Comet disabled. This test therefore raises before the output comparison or native-path assertion. Could the explicit-step case use a sign-correct step column or separate ascending/descending inputs, keeping its arguments as leaves so it still tests the native path? This conclusion is source-derived; I have not run the suite.
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3941143051
##########
spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala:
##########
@@ -158,10 +158,32 @@ trait ShimSparkErrorConverter {
Some(QueryExecutionErrors.exceedMapSizeLimitError(params("size").toString.toInt))
case "CollectionSizeLimitExceeded" =>
- // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4
+ // createArrayWithElementsExceedLimitError takes (count: Any) in Spark 3.4; pass the
+ // decimal string through since the reported length can exceed Long range.
Some(
QueryExecutionErrors.createArrayWithElementsExceedLimitError(
- params("numElements").toString.toLong))
+ params("numElements").toString))
+
+ case "SequenceIllegalBoundaries" =>
+ // Spark 3.x codegen throws a plain IllegalArgumentException for sequence boundaries.
+ Some(
+ new IllegalArgumentException(
+ s"Illegal sequence boundaries: ${params("start")} to ${params("stop")} " +
+ s"by ${params("step")}"))
+
+ case "SequenceBatchTooLarge" =>
Review Comment:
This message now exists in four places, `error.rs` plus all three shims, and unlike `SequenceIllegalBoundaries` it has no version-specific behaviour to justify living in the shims at all. Would a shared constant work, with each shim interpolating `params("totalElements")` into it? The same question applies to the `case "Internal"` arm just below, which is character-identical in all three files. Four copies of one English sentence is the kind of thing where a later wording fix lands in three places and misses the fourth.
##########
docs/source/user-guide/latest/expressions.md:
##########
@@ -168,7 +168,7 @@ The tables below list every Spark built-in expression with its current status.
| `element_at` | ✅ | Native | |
| `flatten` | ✅ | Native | Binary/struct/map elements fall back |
| `get` | ✅ | — | |
-| `sequence` | ✅ | Codegen dispatch | |
+| `sequence` | ✅ | Hybrid | Integral types run natively; date/timestamp sequences use codegen dispatch |
Review Comment:
This note describes the native versus dispatcher split, which a user cannot observe, and leaves out the per-batch ceiling, which is the one thing they can, since it is a query Spark runs that Comet fails. Could it mention that as well, something like "very large per-row sequences may exceed Comet's per-batch limit, lower `spark.comet.batchSize`"? The audit entry covers it well, but that is not where somebody who has just hit the error will be looking.
##########
spark/src/test/resources/sql-tests/expressions/array/sequence.sql:
##########
@@ -15,17 +15,175 @@
-- specific language governing permissions and limitations
-- under the License.
--- Routes sequence through the codegen dispatcher so behavior matches Spark exactly.
+-- sequence(start, stop[, step]) for integral element types runs on the native kernel
+-- (https://github.com/apache/datafusion-comet/issues/5349). Date and timestamp sequences
+-- stay on the JVM codegen dispatcher and are exercised at the bottom of this file.
statement
-CREATE TABLE test_sequence(a int, b int) USING parquet
+CREATE TABLE test_sequence(
+ b_start tinyint, b_stop tinyint, b_step tinyint,
+ s_start smallint, s_stop smallint, s_step smallint,
+ i_start int, i_stop int, i_step int,
+ l_start bigint, l_stop bigint, l_step bigint)
+USING parquet
+-- Row 2 descends, row 3 has start == stop, rows 4-6 carry NULLs in each argument position.
statement
-INSERT INTO test_sequence VALUES (1, 5), (5, 1), (3, 3), (NULL, 5)
+INSERT INTO test_sequence VALUES
+ (1Y, 5Y, 1Y, 1S, 5S, 1S, 1, 10, 3, 1L, 5L, 2L),
+ (-3Y, -1Y, 1Y, 100S, 90S, -2S, 20, 2, -6, 9223372036854775802L, 9223372036854775807L, 1L),
+ (0Y, 0Y, 0Y, -5S, -5S, 0S, 7, 7, 0, -9223372036854775808L, -9223372036854775800L, 3L),
+ (NULL, 5Y, 1Y, NULL, 5S, 1S, NULL, 10, 1, NULL, 5L, 1L),
+ (1Y, NULL, 1Y, 1S, NULL, 1S, 1, NULL, 1, 1L, NULL, 1L),
+ (1Y, 5Y, NULL, 1S, 5S, NULL, 1, 10, NULL, 1L, 5L, NULL)
+
+-- ============================================================================
+-- Explicit step, all four integral types
+-- ============================================================================
+
+query
+SELECT sequence(i_start, i_stop, i_step) FROM test_sequence
+
+query
+SELECT sequence(l_start, l_stop, l_step) FROM test_sequence
+
+-- Column step for the narrow integral types exercises the Byte/Short monomorphizations
+-- of the native kernel, not just the literal-step shape.
+query
+SELECT sequence(b_start, b_stop, b_step) FROM test_sequence
+
+query
+SELECT sequence(s_start, s_stop, s_step) FROM test_sequence
+
+-- ============================================================================
+-- Default step: per-row start <= stop ? 1 : -1, both directions in one column
+-- ============================================================================
+
+query
+SELECT sequence(b_start, b_stop), sequence(s_start, s_stop) FROM test_sequence
+
+query
+SELECT sequence(i_start, i_stop), sequence(l_start, l_stop) FROM test_sequence
+
+-- ============================================================================
+-- Literal and mixed literal/column arguments
+-- ============================================================================
query
-SELECT a, b, sequence(a, b) FROM test_sequence
+SELECT sequence(1, 10), sequence(10, 1), sequence(5, 5), sequence(5, 5, 0)
--- literal arguments with step
query
SELECT sequence(1, 5), sequence(5, 1, -1), sequence(1, 10, 2)
+
+query
+SELECT sequence(1L, 9L, 2L), sequence(-128Y, -120Y), sequence(32760S, 32767S)
+
+-- On row 2 the source row is (i_start=20, i_stop=2, i_step=-6), so the literal-step column
+-- asks for sequence(1, 2, 2) = [1] while the default-step column asks for sequence(20, 25)
+-- = [20, 21, 22, 23, 24, 25]. The two columns disagreeing in direction on the same row is
+-- intentional coverage, not an oversight.
+query
+SELECT sequence(1, i_stop, 2), sequence(i_start, 25) FROM test_sequence WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+query
+SELECT sequence(CAST(NULL AS int), 5), sequence(1, CAST(NULL AS int)), sequence(1, 5, CAST(NULL AS int))
+
+-- Integer.MIN_VALUE/MAX_VALUE bounds for int, and a sequence spanning zero
+query
+SELECT sequence(2147483642, 2147483647), sequence(-2147483648, -2147483643), sequence(-3, 3, 3)
+
+-- ============================================================================
+-- sequence feeding explode, the common date-spine shape (with integers)
+-- ============================================================================
+
+query
+SELECT i_start, x FROM test_sequence LATERAL VIEW explode(sequence(i_start, i_stop)) AS x WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+-- ============================================================================
+-- Error paths: step direction contradicts bounds, or zero step with start != stop
+-- ============================================================================
+
+query expect_error(Illegal sequence boundaries: 1 to 5 by -1)
+SELECT sequence(1, 5, -1)
+
+query expect_error(Illegal sequence boundaries: 10 to 2 by 3)
+SELECT sequence(10, 2, 3) FROM test_sequence LIMIT 1
+
+query expect_error(Illegal sequence boundaries: 1 to 5 by 0)
+SELECT sequence(1, 5, 0)
+
+-- ============================================================================
+-- Error paths: length exceeds MAX_ROUNDED_ARRAY_LENGTH
+-- ============================================================================
+
+query expect_error(the array size limit 2147483632)
+SELECT sequence(0L, 4294967296L, 1L)
+
+-- Math.addExact overflow inside Spark's sequenceLength: reported count is 2^63
+query expect_error(9223372036854775808)
+SELECT sequence(0L, 9223372036854775807L, 1L)
+
+-- Long.MinValue / -1 special case: reported count is 2^63 + 1
+query expect_error(9223372036854775809)
+SELECT sequence(0L, -9223372036854775808L, -1L)
+
+-- delta overflows long but the exact length is tiny: Spark reaches an internal error
+query expect_error(Unreachable code reached)
+SELECT sequence(-9223372036854775808L, 9223372036854775807L, 9223372036854775807L)
+
+-- ============================================================================
+-- Full narrow-type range: writes at the byte/short boundary. Spark's kernel
+-- accumulates with the element type's `Numeric`, wrapping at 8 and 16 bits;
+-- ours accumulates in i64 and truncates on the way out. The two agree because
+-- every element is inside range, but this locks in the boundary values.
+-- ============================================================================
+
+query
+SELECT sequence(-128Y, 127Y), sequence(-32768S, 32767S)
+
+-- ============================================================================
+-- Int32 boundary product: index * step overflows int, exercising the Int32
+-- monomorphization at the extreme.
+-- ============================================================================
+
+query
+SELECT sequence(-2147483648, 2147483647, 1073741824)
+
+-- ============================================================================
+-- Null short-circuit under a nested sequence: Spark's codegen returns NULL
+-- without evaluating the inner argument, so the inner `sequence(1, 5, -1)`
+-- must not fire on the NULL row. Non-leaf argument shapes stay on the JVM
+-- codegen dispatcher for this reason
+-- (https://github.com/apache/datafusion-comet/pull/5614#discussion_r3910237757).
Review Comment:
Could the `#discussion_r3910237757` link come out of this comment? The two sentences before it already say why non-leaf argument shapes stay on the dispatcher, which is the part a future reader needs. A pointer into a review thread records how the code came to be rather than what it does, and it will not survive the next change to this reasoning. The `#5349` link at the top of the file is the durable kind and is worth keeping.
##########
native/spark-expr/src/array_funcs/sequence.rs:
##########
@@ -0,0 +1,393 @@
+// 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.
+
+// Spark-compatible sequence(start, stop[, step]) for integral element types.
+//
+// Mirrors the code Spark's whole-stage codegen emits for `Sequence`
+// (`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala`,
+// identical from 3.4.3 through 4.1.1): the boundary check and `Sequence.sequenceLength` decide
+// per row how many elements to generate, then elements are `start + step * i`. Unlike the JVM
+// path, which allocates two `long[]` per row and copies every element three times, this kernel
+// reserves the Arrow child buffer once for the whole batch and writes each element exactly once.
+//
+// Date/timestamp sequences are not handled here; the Scala serde only routes IntegralType
+// sequences to this function.
+
+use std::sync::Arc;
+
+use arrow::array::{Array, ArrayRef, ListArray, NullBufferBuilder, PrimitiveArray};
+use arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use arrow::datatypes::{
+ ArrowPrimitiveType, DataType, FieldRef, Int16Type, Int32Type, Int64Type, Int8Type,
+};
+use datafusion::common::cast::as_primitive_array;
+use datafusion::common::{exec_err, DataFusionError, Result, ScalarValue};
+use datafusion::logical_expr::ColumnarValue;
+
+use crate::SparkError;
+
+/// Spark's ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH (Integer.MAX_VALUE - 15).
+const MAX_ROUNDED_ARRAY_LENGTH: i128 = (i32::MAX - 15) as i128;
+
+pub fn spark_sequence(args: &[ColumnarValue], data_type: &DataType) -> Result<ColumnarValue> {
+ let child_field = match data_type {
+ DataType::List(field) => Arc::clone(field),
+ other => return exec_err!("spark_sequence expects a List return type, got {other:?}"),
+ };
+ if args.len() != 2 && args.len() != 3 {
+ return exec_err!(
+ "spark_sequence expects 2 or 3 arguments, got {}",
+ args.len()
+ );
+ }
+
+ let all_scalar = args
+ .iter()
+ .all(|arg| matches!(arg, ColumnarValue::Scalar(_)));
+ let arrays = ColumnarValue::values_to_arrays(args)?;
+ let step = arrays.get(2);
+
+ let result = match child_field.data_type() {
+ DataType::Int8 => {
+ sequence_integral::<Int8Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i8)
+ }
+ DataType::Int16 => {
+ sequence_integral::<Int16Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i16)
+ }
+ DataType::Int32 => {
+ sequence_integral::<Int32Type>(&arrays[0], &arrays[1], step, child_field, |v| v as i32)
+ }
+ DataType::Int64 => {
+ sequence_integral::<Int64Type>(&arrays[0], &arrays[1], step, child_field, |v| v)
+ }
+ other => exec_err!("spark_sequence does not support element type {other:?}"),
+ }?;
+
+ if all_scalar {
+ Ok(ColumnarValue::Scalar(ScalarValue::try_from_array(
+ &result, 0,
+ )?))
+ } else {
+ Ok(ColumnarValue::Array(result))
+ }
+}
+
+fn sequence_integral<T: ArrowPrimitiveType>(
+ start: &ArrayRef,
+ stop: &ArrayRef,
+ step: Option<&ArrayRef>,
+ child_field: FieldRef,
+ from_i64: impl Fn(i64) -> T::Native,
+) -> Result<ArrayRef>
+where
+ T::Native: Into<i64>,
+{
+ let start = as_primitive_array::<T>(start)?;
+ let stop = as_primitive_array::<T>(stop)?;
+ let step = step.map(|arr| as_primitive_array::<T>(arr)).transpose()?;
+ let num_rows = start.len();
+
+ let row_is_null = |row: usize| {
+ start.is_null(row) || stop.is_null(row) || step.is_some_and(|arr| arr.is_null(row))
+ };
+ // With no explicit step, Spark uses `start <= stop ? 1 : -1` per row, so the direction
+ // always matches the bounds and the boundary check below cannot fail.
+ let row_step = |row: usize, start: i64, stop: i64| -> i64 {
+ match step {
+ Some(arr) => arr.value(row).into(),
+ None => {
+ if start <= stop {
+ 1
+ } else {
+ -1
+ }
+ }
+ }
+ };
+
+ // First pass: compute per-row lengths so the child buffer can be reserved once for the
+ // whole batch. Valid rows always produce at least one element, so length 0 marks a null row.
+ let mut lengths: Vec<usize> = Vec::with_capacity(num_rows);
+ let mut total: usize = 0;
+ for row in 0..num_rows {
+ if row_is_null(row) {
+ lengths.push(0);
+ continue;
+ }
+ let s: i64 = start.value(row).into();
+ let e: i64 = stop.value(row).into();
+ let len = sequence_length(s, e, row_step(row, s, e))?;
+ total += len;
+ lengths.push(len);
+ }
+ // Comet-specific ceiling: the sum of every row's length in one Arrow batch must fit in
+ // the i32 offset buffer. Spark has no equivalent guard because it stores each row as its
+ // own `long[]`, so the user may hit this on a query Spark itself would run. Report it via
+ // a dedicated error that names `spark.comet.batchSize` as the actionable knob rather than
+ // Spark's per-array size limit.
+ if total > i32::MAX as usize {
Review Comment:
The `try_reserve_exact` change does what I asked for, thank you. I think there is still a gap underneath it though.
This guard is on element count rather than bytes, so for `bigint` it only fires at around 17 GB, and the `Vec` comes from the global allocator rather than the DataFusion `MemoryPool`. That means the allocation is not counted against `spark.comet.memory*`, cannot be spilled, and applies no back-pressure.
At the default batch size I measured 50000 elements per row allocating 3.3 GB in a single reservation and completing fine, with peak process RSS 2852 MB above baseline against Spark's 1076 MB for the same query. On a Linux executor with overcommit the OOM killer arrives well before `try_reserve_exact` gets a chance to return `Err`, so the graceful path is the one a user is least likely to reach.
Could the ceiling be a byte budget as well as an element count, sized off the batch memory budget rather than `i32::MAX`? That would make `SequenceBatchTooLarge` fire while the executor is still healthy, which is the point at which its actionable message is worth something. This is also the peak-memory question from the earlier round, which I do not think has been answered with a measurement yet.
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -954,4 +954,61 @@ object CometArraySort extends CometCodegenDispatch[ArraySort]
object CometZipWith extends CometCodegenDispatch[ZipWith]
-object CometSequence extends CometCodegenDispatch[Sequence]
+object CometSequence extends CometExpressionSerde[Sequence] with CodegenDispatchFallback {
+
+ private val temporalUnsupportedReason =
+ "date and timestamp element types run through the JVM codegen dispatcher"
+
+ private val unsafeArgUnsupportedReason =
+ "sequence arguments must be literals or column references; other shapes run through the " +
+ "JVM codegen dispatcher to preserve Spark's per-row null short-circuit"
+
+ override def getSupportLevel(expr: Sequence): SupportLevel = expr.start.dataType match {
+ case ByteType | ShortType | IntegerType | LongType =>
+ // Spark's codegen for `Sequence` short-circuits per row: any null argument returns null
+ // without evaluating the rest. DataFusion evaluates each scalar-UDF argument over the
+ // whole batch before calling the outer kernel, so a sub-expression with side effects
+ // (a nested call, a `CASE WHEN`, even a zero-arg UDF like `boom()`) could fire on rows
+ // Spark's null check would have discarded. A tree-shape "no children" test is not
+ // enough — a zero-arg UDF has empty children but still executes. Only literals and
+ // column references are safe to lower natively; anything else falls back to the
+ // codegen dispatcher, which keeps the whole tree inside Spark's guarded evaluation.
+ if (argsAreLiteralsOrRefs(expr)) Compatible()
+ else Unsupported(Some(unsafeArgUnsupportedReason))
+ case DateType | TimestampType | TimestampNTZType =>
+ // Temporal sequences step through timezone/DST/legacy-calendar arithmetic
+ // (https://github.com/apache/datafusion-comet/issues/5349), so they stay on the JVM
+ // codegen dispatcher.
+ Unsupported(Some(temporalUnsupportedReason))
+ case other =>
+ Unsupported(Some(s"sequence with element type $other is not supported natively"))
+ }
+
+ private def argsAreLiteralsOrRefs(expr: Sequence): Boolean = {
Review Comment:
The gate itself is the right conservative call and I am not arguing with it. It is worth being explicit about how narrow the resulting fast path is though. The native kernel only engages when both endpoints already exist as columns or literals, so the idiomatic spine `sequence(x, x + n)` stays on the dispatcher, and so does anything behind a coercion `CAST`. Your own benchmark is the evidence, since it needed `c_stop_5` and friends materialised as stored columns before any integral case went native.
I also tried the workaround a user would reach for first, and it does not work. `FROM (SELECT c_start, c_start + 364 AS c_stop FROM p)` gets folded straight back by `CollapseProject`, and the explain still reports `JVM codegen dispatcher: sequence`. So there is no way to opt in short of rewriting the table.
Two things would help. Could the audit entry spell out which shapes reach the native path, since "leaf arguments only" is not something a user can map onto their own SQL? And separately, is a safe widening worth considering later, accepting an argument subtree that provably cannot throw and preserves nulls, which would cover `x + n` at least under non-ANSI? Happy for that to be a follow-up.
##########
spark/src/test/resources/sql-tests/expressions/array/sequence.sql:
##########
@@ -15,17 +15,175 @@
-- specific language governing permissions and limitations
-- under the License.
--- Routes sequence through the codegen dispatcher so behavior matches Spark exactly.
+-- sequence(start, stop[, step]) for integral element types runs on the native kernel
+-- (https://github.com/apache/datafusion-comet/issues/5349). Date and timestamp sequences
+-- stay on the JVM codegen dispatcher and are exercised at the bottom of this file.
statement
-CREATE TABLE test_sequence(a int, b int) USING parquet
+CREATE TABLE test_sequence(
+ b_start tinyint, b_stop tinyint, b_step tinyint,
+ s_start smallint, s_stop smallint, s_step smallint,
+ i_start int, i_stop int, i_step int,
+ l_start bigint, l_stop bigint, l_step bigint)
+USING parquet
+-- Row 2 descends, row 3 has start == stop, rows 4-6 carry NULLs in each argument position.
statement
-INSERT INTO test_sequence VALUES (1, 5), (5, 1), (3, 3), (NULL, 5)
+INSERT INTO test_sequence VALUES
+ (1Y, 5Y, 1Y, 1S, 5S, 1S, 1, 10, 3, 1L, 5L, 2L),
+ (-3Y, -1Y, 1Y, 100S, 90S, -2S, 20, 2, -6, 9223372036854775802L, 9223372036854775807L, 1L),
+ (0Y, 0Y, 0Y, -5S, -5S, 0S, 7, 7, 0, -9223372036854775808L, -9223372036854775800L, 3L),
+ (NULL, 5Y, 1Y, NULL, 5S, 1S, NULL, 10, 1, NULL, 5L, 1L),
+ (1Y, NULL, 1Y, 1S, NULL, 1S, 1, NULL, 1, 1L, NULL, 1L),
+ (1Y, 5Y, NULL, 1S, 5S, NULL, 1, 10, NULL, 1L, 5L, NULL)
+
+-- ============================================================================
+-- Explicit step, all four integral types
+-- ============================================================================
+
+query
+SELECT sequence(i_start, i_stop, i_step) FROM test_sequence
+
+query
+SELECT sequence(l_start, l_stop, l_step) FROM test_sequence
+
+-- Column step for the narrow integral types exercises the Byte/Short monomorphizations
+-- of the native kernel, not just the literal-step shape.
+query
+SELECT sequence(b_start, b_stop, b_step) FROM test_sequence
+
+query
+SELECT sequence(s_start, s_stop, s_step) FROM test_sequence
+
+-- ============================================================================
+-- Default step: per-row start <= stop ? 1 : -1, both directions in one column
+-- ============================================================================
+
+query
+SELECT sequence(b_start, b_stop), sequence(s_start, s_stop) FROM test_sequence
+
+query
+SELECT sequence(i_start, i_stop), sequence(l_start, l_stop) FROM test_sequence
+
+-- ============================================================================
+-- Literal and mixed literal/column arguments
+-- ============================================================================
query
-SELECT a, b, sequence(a, b) FROM test_sequence
+SELECT sequence(1, 10), sequence(10, 1), sequence(5, 5), sequence(5, 5, 0)
--- literal arguments with step
query
SELECT sequence(1, 5), sequence(5, 1, -1), sequence(1, 10, 2)
+
+query
+SELECT sequence(1L, 9L, 2L), sequence(-128Y, -120Y), sequence(32760S, 32767S)
+
+-- On row 2 the source row is (i_start=20, i_stop=2, i_step=-6), so the literal-step column
+-- asks for sequence(1, 2, 2) = [1] while the default-step column asks for sequence(20, 25)
+-- = [20, 21, 22, 23, 24, 25]. The two columns disagreeing in direction on the same row is
+-- intentional coverage, not an oversight.
+query
+SELECT sequence(1, i_stop, 2), sequence(i_start, 25) FROM test_sequence WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+query
+SELECT sequence(CAST(NULL AS int), 5), sequence(1, CAST(NULL AS int)), sequence(1, 5, CAST(NULL AS int))
+
+-- Integer.MIN_VALUE/MAX_VALUE bounds for int, and a sequence spanning zero
+query
+SELECT sequence(2147483642, 2147483647), sequence(-2147483648, -2147483643), sequence(-3, 3, 3)
+
+-- ============================================================================
+-- sequence feeding explode, the common date-spine shape (with integers)
+-- ============================================================================
+
+query
+SELECT i_start, x FROM test_sequence LATERAL VIEW explode(sequence(i_start, i_stop)) AS x WHERE i_start IS NOT NULL AND i_stop IS NOT NULL
+
+-- ============================================================================
+-- Error paths: step direction contradicts bounds, or zero step with start != stop
+-- ============================================================================
+
+query expect_error(Illegal sequence boundaries: 1 to 5 by -1)
+SELECT sequence(1, 5, -1)
+
+query expect_error(Illegal sequence boundaries: 10 to 2 by 3)
+SELECT sequence(10, 2, 3) FROM test_sequence LIMIT 1
+
+query expect_error(Illegal sequence boundaries: 1 to 5 by 0)
+SELECT sequence(1, 5, 0)
+
+-- ============================================================================
+-- Error paths: length exceeds MAX_ROUNDED_ARRAY_LENGTH
+-- ============================================================================
+
+query expect_error(the array size limit 2147483632)
Review Comment:
`SequenceBatchTooLarge` is the one behaviour in this PR that fails a query Spark completes, and I cannot find a test for it anywhere. It is also close to free to test, because the `total > i32::MAX` check runs in the first pass before anything is allocated.
Could a case go in around here? Something like this trips it at the default batch size and passes at half of it, so it also pins the remedy you documented:
```sql
statement
CREATE TABLE t_seq_ceiling(a INT, b INT) USING parquet
query expect_error(Lower `spark.comet.batchSize`)
SELECT sum(CAST(size(sequence(a, b)) AS BIGINT)) FROM t_seq_ceiling
```
with `a = 0, b = 262143` over 8192 rows in a single partition. I ran that against this branch and it produces exactly the message you intended, and lowering `spark.comet.batchSize` to 4096 makes it return Spark's answer instead. That is the `sequence(0, 262143)` case from my first pass, now confirmed end to end.
--
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: add native spark_sequence kernel for integral element types [datafusion-comet]
Posted by "0lai0 (via GitHub)" <gi...@apache.org>.
0lai0 commented on code in PR #5614:
URL: https://github.com/apache/datafusion-comet/pull/5614#discussion_r3931157120
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometSequenceBenchmark.scala:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.benchmark
+
+/**
+ * Benchmark to measure performance of Comet's native `sequence` kernel against Spark's codegen
+ * (issue #5349). Integral shapes run natively under Comet; the date case stays on the JVM codegen
+ * dispatcher in both arms and is included to show that path is unchanged. To run this benchmark:
+ * {{{
+ * SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometSequenceBenchmark
+ * }}}
+ * Results will be written to "spark/benchmarks/CometSequenceBenchmark-**results.txt".
+ */
+object CometSequenceBenchmark extends CometBenchmarkBase {
+
+ private val sequenceQueries = List(
+ ("seq_short_5_elems", "SELECT sequence(c_start, c_start + 4) FROM parquetV1Table"),
+ ("seq_spine_365_elems", "SELECT sequence(c_start, c_start + 364) FROM parquetV1Table"),
Review Comment:
Thanks for the feedback, I pushed the fix.
Stop endpoints are now materialized columns (`c_stop_5`, `c_stop_365`, `c_stop_10000`, `c_null_stop_365`), so every integral query passes only literals and column references and satisfies `CometSequence.argsAreLiteralsOrRefs` (`spark/src/main/scala/org/apache/comet/serde/arrays.scala:957`). The date case keeps the arithmetic form intentionally, because temporal `Sequence` unconditionally routes through the JVM dispatcher (`arrays.scala:948-952`); it stays in the list as the dispatcher control.
Benchmark on Apple M5 / OpenJDK 17.0.18, 8192 rows:
| case | Spark ns/row | Comet ns/row | Comet vs Spark |
| --- | ---: | ---: | ---: |
| seq_short_5_elems | 1795 | 645 | 2.8× |
| seq_spine_365_elems | 1796 | 798 | 2.3× |
| seq_long_10000_elems | 7922 | 7135 | 1.1× |
| seq_descending_default_step | 1746 | 736 | 2.4× |
| seq_explicit_step_7 | 1421 | 489 | 2.9× |
| seq_sparse_nulls_365_elems | 1651 | 663 | 2.5× |
| seq_date_spine_dispatcher (control) | 1528 | 1554 | 1.0× |
The integral vs date gap (2–3× vs ~1.0×) is the path evidence: leaf-arg queries now hit native; the date control stays on the dispatcher. The old `c_start + 4` shape would have been ~1.0× across the board because `argsAreLiteralsOrRefs` rejected it.
Dropped a plan-text `spark_sequence` check, that name only appears in the native proto, so it always reported `native=false`. Operator-level Comet coverage is already asserted by `findFirstNonCometOperator`; the timings cover the expression path.
--
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