feat: add native spark_sequence kernel for integral element types

docs/source/contributor-guide/expression-audits/array_funcs.md

@@ -176,6 +176,16 @@
 - Spark 4.0.1 (audited 2026-05-27): semantics unchanged; ANSI default flips to `true`.
 - Spark 4.1.1 (audited 2026-05-27): `inputTypes` tightened to `Seq(ArrayType, IntegralType)` (analysis-time only); runtime unchanged.
 
+## sequence
+
+- Spark 3.4.3 (audited 2026-08-29): `Sequence(start, stop, stepOpt, timeZoneId)`; `Sequence.impl` selects the implementation from `dataType.elementType`, so the integral/temporal split is knowable at plan time. Codegen for the integral path checks boundaries with a plain `IllegalArgumentException("Illegal sequence boundaries: ...")`, then calls the static `Sequence.sequenceLength`, which raises `SparkRuntimeException(_LEGACY_ERROR_TEMP_2161)` past `MAX_ROUNDED_ARRAY_LENGTH` and `internalError("Unreachable code reached.")` when `stop - start` overflows Long but the exact length is within the limit. Default step is per-row `start <= stop ? 1 : -1`.
+- Spark 3.5.8 (audited 2026-08-29): internal refactors only (`DataTypeUtils.sameType`, `PhysicalIntegralType.integral`); runtime semantics identical to 3.4.3.
+- Spark 4.0.1 (audited 2026-08-29): boundary error becomes `SparkIllegalArgumentException(_LEGACY_ERROR_TEMP_3243)` and the length error becomes `COLLECTION_SIZE_LIMIT_EXCEEDED.PARAMETER` (now carrying the function name); adds `throwable` optimizer hint. `sequenceLength` itself is unchanged.
+- Spark 4.1.1 (audited 2026-08-29): byte-identical `Sequence` class body to 4.0.1.
+- Comet routes integral element types (`ByteType`/`ShortType`/`IntegerType`/`LongType`) via `CometSequence` to the native `spark_sequence` kernel ([#5349](https://github.com/apache/datafusion-comet/issues/5349)): one pass over the generated elements, child buffer reserved once per batch, no per-row allocation. The two-argument form is evaluated with Spark's per-row default step inside the kernel. Both error conditions and the internal-error edge are reproduced through `SparkError` and mapped per Spark version by `ShimSparkErrorConverter`. Date/timestamp/timestamp_ntz sequences return `Unsupported` and run on the JVM codegen dispatcher (`CodegenDispatchFallback`), pending the timezone/DST/legacy-calendar work.
+- Per-batch capacity ceiling: the native kernel writes every row's generated elements into one Arrow child buffer whose offsets are `i32`, so the sum of every row's length in a single Arrow batch must fit in `i32::MAX`. Spark itself has no equivalent limit because it stores each row as its own `long[]`. If the total is exceeded, or if the allocator refuses the reservation, the query fails with a `SparkError::SequenceBatchTooLarge` message that names `spark.comet.batchSize` as the actionable knob (lower it to group fewer rows per batch). The `try_reserve` path guarantees the failure surfaces as a query error rather than an allocator abort.
+- Argument-shape restriction: `CometSequence` reports `Unsupported` for any `Sequence` whose `start`, `stop`, or `step` is not a leaf expression, and routes those through the JVM codegen dispatcher (`CodegenDispatchFallback`). DataFusion evaluates each scalar-UDF argument over the whole batch before calling the outer kernel, so a non-leaf argument would run on rows that Spark's per-row null short-circuit (or a `CASE` branch) would have discarded, and could raise where Spark would have returned `NULL`.
+
 ## shuffle
 
 - Spark 3.4.3 (audited 2026-07-02): `Shuffle(child, randomSeed: Option[Long])`; `inputTypes = Seq(ArrayType)`, `dataType = child.dataType`, non-deterministic and stateful. Seeds a Commons Math3 `MersenneTwister` with `randomSeed + partitionIndex` and applies the "inside-out" Fisher-Yates from `RandomIndicesGenerator`. Only the one-argument `shuffle(array)` form exists in SQL. NULL input returns NULL without advancing the RNG.

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 |
 | `shuffle` | ✅ | Native | Binary/struct/map elements fall back |
 | `slice` | ✅ | Native | Native ([#4149](https://github.com/apache/datafusion-comet/pull/4149)) |
 | `sort_array` | ✅ | Hybrid | Nested struct/null arrays fall back |

native/common/src/error.rs

@@ -135,12 +135,38 @@ pub enum SparkError {
     #[error("[EXCEED_LIMIT_LENGTH] Cannot create a map with {size} elements which exceeds the limit {max_size}.")]
     ExceedMapSizeLimit { size: i32, max_size: i32 },
 
-    #[error("[COLLECTION_SIZE_LIMIT_EXCEEDED] Cannot create array with {num_elements} elements which exceeds the limit {max_elements}.")]
+    /// `num_elements` is a decimal string because Spark reports the unclamped length, which can
+    /// exceed i64 (e.g. sequence(Long.MinValue, Long.MaxValue, 1)). The JVM shim maps this to the
+    /// version-appropriate `createArrayWithElementsExceedLimitError`, passing `function_name`
+    /// through on Spark 4.x (which includes it in the message) and ignoring it on 3.x.
+    #[error("[COLLECTION_SIZE_LIMIT_EXCEEDED] Can't create array with {num_elements} elements which exceeding the array size limit {max_elements}.")]
     CollectionSizeLimitExceeded {
-        num_elements: i64,
+        num_elements: String,
         max_elements: i64,
+        function_name: String,
+    },
+
+    /// Step direction does not match the start/stop bounds in `sequence`. The JVM shim maps this
+    /// to a plain IllegalArgumentException on Spark 3.x and SparkIllegalArgumentException
+    /// (_LEGACY_ERROR_TEMP_3243) on 4.x, matching what Spark's codegen throws.
+    #[error("[_LEGACY_ERROR_TEMP_3243] Illegal sequence boundaries: {start} to {stop} by {step}")]
+    SequenceIllegalBoundaries {
+        start: String,
+        stop: String,
+        step: String,
     },
 
+    /// Sum of every row's sequence length in one Arrow batch exceeds Comet's per-batch offset
+    /// ceiling (i32::MAX) or the allocator could not satisfy the reservation. Spark itself has
+    /// no equivalent limit because it stores each row as its own `long[]`. Reported with a
+    /// message that names `spark.comet.batchSize` as the actionable knob.
+    #[error(
+        "Comet's native `sequence` kernel cannot materialize a batch with {total_elements} \
+        total elements: it exceeds the per-batch limit or the allocator refused the reservation. \
+        Lower `spark.comet.batchSize` so fewer rows are grouped per batch."
+    )]
+    SequenceBatchTooLarge { total_elements: String },
+
     #[error("[NOT_NULL_ASSERT_VIOLATION] The field `{field_name}` cannot be null.")]
     NotNullAssertViolation { field_name: String },
 
@@ -306,6 +332,8 @@ impl SparkError {
             SparkError::MapKeyValueDiffSizes => "MapKeyValueDiffSizes",
             SparkError::ExceedMapSizeLimit { .. } => "ExceedMapSizeLimit",
             SparkError::CollectionSizeLimitExceeded { .. } => "CollectionSizeLimitExceeded",
+            SparkError::SequenceIllegalBoundaries { .. } => "SequenceIllegalBoundaries",
+            SparkError::SequenceBatchTooLarge { .. } => "SequenceBatchTooLarge",
             SparkError::NotNullAssertViolation { .. } => "NotNullAssertViolation",
             SparkError::ValueIsNull { .. } => "ValueIsNull",
             SparkError::CannotParseTimestamp { .. } => "CannotParseTimestamp",
@@ -450,10 +478,24 @@ impl SparkError {
             SparkError::CollectionSizeLimitExceeded {
                 num_elements,
                 max_elements,
+                function_name,
             } => {
                 serde_json::json!({
                     "numElements": num_elements,
                     "maxElements": max_elements,
+                    "functionName": function_name,
+                })
+            }
+            SparkError::SequenceIllegalBoundaries { start, stop, step } => {
+                serde_json::json!({
+                    "start": start,
+                    "stop": stop,
+                    "step": step,
+                })
+            }
+            SparkError::SequenceBatchTooLarge { total_elements } => {
+                serde_json::json!({
+                    "totalElements": total_elements,
                 })
             }
             SparkError::NotNullAssertViolation { field_name } => {
@@ -626,6 +668,7 @@ impl SparkError {
             | SparkError::MapKeyValueDiffSizes
             | SparkError::ExceedMapSizeLimit { .. }
             | SparkError::CollectionSizeLimitExceeded { .. }
+            | SparkError::SequenceBatchTooLarge { .. } // Comet-specific extension
             | SparkError::NotNullAssertViolation { .. }
             | SparkError::ValueIsNull { .. } // Comet-specific extension
             | SparkError::UnexpectedPositiveValue { .. }
@@ -644,7 +687,8 @@ impl SparkError {
             // IllegalArgumentException
             SparkError::DatatypeCannotOrder { .. }
             | SparkError::InvalidUtf8String { .. }
-            | SparkError::IllegalDayOfWeek { .. } => {
+            | SparkError::IllegalDayOfWeek { .. }
+            | SparkError::SequenceIllegalBoundaries { .. } => {
                 "org/apache/spark/SparkIllegalArgumentException"
             }
 
@@ -722,6 +766,10 @@ impl SparkError {
             SparkError::CollectionSizeLimitExceeded { .. } => {
                 Some("COLLECTION_SIZE_LIMIT_EXCEEDED")
             }
+            SparkError::SequenceIllegalBoundaries { .. } => Some("_LEGACY_ERROR_TEMP_3243"),
+
+            // Comet-specific: no Spark error class, the shim builds the message itself.
+            SparkError::SequenceBatchTooLarge { .. } => None,
 
             // Null validation errors
             SparkError::NotNullAssertViolation { .. } => Some("NOT_NULL_ASSERT_VIOLATION"),

native/spark-expr/Cargo.toml

@@ -287,6 +287,10 @@ harness = false
 name = "contains"
 harness = false
 
+[[bench]]
+name = "sequence"
+harness = false
+
 [[bench]]
 name = "timestamp_trunc"
 harness = false

native/spark-expr/benches/sequence.rs

@@ -0,0 +1,133 @@
+// 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 arrow::array::Int64Array;
+use arrow::datatypes::{DataType, Field};
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::physical_plan::ColumnarValue;
+use datafusion_comet_spark_expr::spark_sequence;
+use std::hint::black_box;
+use std::sync::Arc;
+
+const NUM_ROWS: usize = 8192;
+
+fn list_of_i64() -> DataType {
+    DataType::List(Arc::new(Field::new_list_field(DataType::Int64, false)))
+}
+
+/// start/stop columns generating `elems_per_row` elements per row (ascending, step 1).
+/// When `null_every` is Some(n), every nth row of `start` is null.
+fn args_with_len(elems_per_row: i64, null_every: Option<usize>) -> Vec<ColumnarValue> {
+    let start = Int64Array::from(
+        (0..NUM_ROWS)
+            .map(|i| match null_every {
+                Some(n) if i % n == 0 => None,
+                _ => Some(i as i64),
+            })
+            .collect::<Vec<_>>(),
+    );
+    let stop = Int64Array::from(
+        (0..NUM_ROWS)
+            .map(|i| Some(i as i64 + elems_per_row - 1))
+            .collect::<Vec<_>>(),
+    );
+    vec![
+        ColumnarValue::Array(Arc::new(start)),
+        ColumnarValue::Array(Arc::new(stop)),
+    ]
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let return_type = list_of_i64();
+
+    let mut group = c.benchmark_group("sequence");
+
+    // Short sequences: per-row overhead dominates.
+    for elems in [2i64, 5] {
+        let args = args_with_len(elems, None);
+        group.bench_function(format!("short_{elems}_elems"), |b| {
+            b.iter(|| black_box(spark_sequence(&args, &return_type).unwrap()))
+        });
+    }
+
+    // Long sequences: element throughput dominates. 365 is the date-spine shape from the
+    // issue; 10k stresses the child buffer reservation.
+    for elems in [365i64, 10_000] {
+        let args = args_with_len(elems, None);
+        group.bench_function(format!("long_{elems}_elems"), |b| {
+            b.iter(|| black_box(spark_sequence(&args, &return_type).unwrap()))
+        });
+    }
+
+    // Descending with explicit negative step.
+    {
+        let start = Int64Array::from((0..NUM_ROWS).map(|i| i as i64 + 364).collect::<Vec<_>>());
+        let stop = Int64Array::from((0..NUM_ROWS).map(|i| i as i64).collect::<Vec<_>>());
+        let step = Int64Array::from(vec![-1i64; NUM_ROWS]);
+        let args = vec![
+            ColumnarValue::Array(Arc::new(start)),
+            ColumnarValue::Array(Arc::new(stop)),
+            ColumnarValue::Array(Arc::new(step)),
+        ];
+        group.bench_function("descending_365_elems", |b| {
+            b.iter(|| black_box(spark_sequence(&args, &return_type).unwrap()))
+        });
+    }
+
+    // Zero step with start == stop: single-element rows through the step==0 path.
+    {
+        let start = Int64Array::from((0..NUM_ROWS).map(|i| i as i64).collect::<Vec<_>>());
+        let stop = Int64Array::from((0..NUM_ROWS).map(|i| i as i64).collect::<Vec<_>>());
+        let step = Int64Array::from(vec![0i64; NUM_ROWS]);
+        let args = vec![
+            ColumnarValue::Array(Arc::new(start)),
+            ColumnarValue::Array(Arc::new(stop)),
+            ColumnarValue::Array(Arc::new(step)),
+        ];
+        group.bench_function("zero_step_start_eq_stop", |b| {
+            b.iter(|| black_box(spark_sequence(&args, &return_type).unwrap()))
+        });
+    }
+
+    // Sparse (every 10th row) and dense (every 2nd row) nulls over the date-spine shape.
+    for (label, every) in [("sparse_nulls", 10usize), ("dense_nulls", 2)] {
+        let args = args_with_len(365, Some(every));
+        group.bench_function(format!("{label}_365_elems"), |b| {
+            b.iter(|| black_box(spark_sequence(&args, &return_type).unwrap()))
+        });
+    }
+
+    // Error path: the boundary check rejects the first row.
+    {
+        let start = Int64Array::from(vec![0i64; NUM_ROWS]);
+        let stop = Int64Array::from(vec![100i64; NUM_ROWS]);
+        let step = Int64Array::from(vec![-1i64; NUM_ROWS]);
+        let args = vec![
+            ColumnarValue::Array(Arc::new(start)),
+            ColumnarValue::Array(Arc::new(stop)),
+            ColumnarValue::Array(Arc::new(step)),
+        ];
+        group.bench_function("error_illegal_boundaries", |b| {
+            b.iter(|| black_box(spark_sequence(&args, &return_type).unwrap_err()))
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);

native/spark-expr/src/array_funcs/mod.rs

@@ -23,6 +23,7 @@ mod arrays_zip;
 mod flatten;
 mod get_array_struct_fields;
 mod list_extract;
+mod sequence;
 mod size;
 
 pub use array_insert::ArrayInsert;
@@ -33,4 +34,5 @@ pub use arrays_zip::SparkArraysZipFunc;
 pub use flatten::SparkFlatten;
 pub use get_array_struct_fields::GetArrayStructFields;
 pub use list_extract::ListExtract;
+pub use sequence::spark_sequence;
 pub use size::{spark_size, SparkSizeFunc};