Sequence invocation ceiling

docs/source/changelog/1.2.0.md

@@ -0,0 +1,29 @@
+<!--
+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.
+-->
+
+# DataFusion Comet 1.2.0 Changelog (Unreleased)
+
+## Behaviour Changes
+
+Native integral `sequence` now has a per-invocation output byte ceiling, derived from the configured
+memory pool and task slots by default in off-heap mode. Valid queries that exceed it can fail before
+output allocation. Set `spark.comet.exec.sequence.maxBytesPerInvocation=-1` to restore the previous
+absence of this byte ceiling. Existing per-row and batch element limits remain in effect. See the
+[Upgrade Guide](../user-guide/latest/migration-guide.md#native-integral-sequence-byte-ceiling) for
+scope and tuning options.

docs/source/changelog/index.md

@@ -24,6 +24,7 @@ Per-release change logs for Apache DataFusion Comet.
 ```{toctree}
 :maxdepth: 1
 
+1.2.0 (Unreleased) <1.2.0>
 1.1.0 <1.1.0>
 1.0.0 <1.0.0>
 0.17.1 <0.17.1>

docs/source/contributor-guide/memory_management.md

@@ -80,7 +80,8 @@ page:
 | --------------------------------------- | ----------- | ------------------------------------------------------------- | ----------------- |
 | Spark execution + storage (on-heap)     | JVM heap    | `spark.executor.memory` and the unified memory manager        | Yes               |
 | Spark Tungsten (off-heap)               | Off-heap    | `spark.memory.offHeap.size` via `TaskMemoryManager`           | Yes               |
-| Comet native (Rust global allocator)    | Native heap | `memory_limit` (see below), enforced only via the memory pool | Reservations only |
+| Comet native (Rust global allocator)    | Native heap | Memory pools for declared reservations; no allocator-wide limit | Reservations only |
+| Comet integral `sequence` output        | Native heap | Logical per-invocation ceiling, without a pool reservation    | No                |
 | Comet JVM Arrow (`CometArrowAllocator`) | Off-heap    | **Nothing**: a `RootAllocator(Long.MaxValue)`                 | No                |
 | Comet JVM shuffle pages                 | Off-heap    | `spark.memory.offHeap.size` via `TaskMemoryManager`           | Yes               |
 
@@ -97,8 +98,10 @@ allocates the bytes and in who is able to see them.
 - **Native heap** is allocated by Rust through its global allocator, in the same process. No JVM
   code makes the call, so no JVM-side allocator or metric ever measures these bytes. The only layer
   that sees any of them is Comet's own memory pool, and then only the portion that operators
-  explicitly reserve. That portion is still charged to Spark, as the next paragraph describes; what
-  is never reserved is measured by nothing and budgeted by nobody.
+  explicitly reserve. That portion is still charged to Spark, as the next paragraph describes.
+  Unreserved allocations have no pool accounting. Individual guards, such as integral `sequence`'s
+  per-invocation ceiling, can refuse a logical output without making a reservation; they do not
+  bound total native usage.
 
 **The budget is shared even though the memory is not.** `memory_limit` is derived from
 `spark.memory.offHeap.size` (see [Where Comet's budget comes from](#where-comets-budget-comes-from)),
@@ -257,6 +260,29 @@ back a slice of the off-heap pool for the memory Comet does not reserve, but it
 
 The only room Spark leaves for memory outside the pool is `spark.executor.memoryOverhead`.
 
+### Integral sequence invocation ceiling
+
+`getMemoryConfig` also resolves `spark.comet.exec.sequence.maxBytesPerInvocation`. Unless explicitly
+overridden, each native integral `sequence` invocation is limited to `memory_limit` divided by the
+configured task slots: `max(1, floor(local or executor cores / CPUs allocated to the task))` when
+the core count is known. Local-cluster uses its configured worker cores if `spark.executor.cores`
+is unset. Standalone can assign cores without exposing their count in the executor configuration;
+in that case the fallback is the whole `memory_limit`, not an inferred task share. Set
+`spark.executor.cores` or an explicit sequence ceiling to choose a smaller limit. The actual task
+CPU allocation and executor configuration respect resource profiles. This fixed ceiling does not
+depend on live tasks, registered consumers or pool usage. On-heap mode has no default ceiling;
+`-1` explicitly disables it and `0` is a real zero-byte ceiling. The resolved value crosses JNI in the existing
+config map, without changing the memory pool or reserving anything from it.
+
+The kernel checks logical output values, list offsets and validity bytes after validating all
+rows, before allocating its values and offsets. Row-length scratch storage and validity preparation
+can precede that check. Allocator padding and shared input backing outside a validity slice do not
+count. Earlier outputs, concurrent invocations and downstream sort, shuffle or explode do not
+consume the ceiling. A scalar invocation still produces one list; generic scalar broadcasts and
+copies of folded literals happen outside it. This bounds one expansion, and does not close the
+accounting or lifetime gaps described below. See
+[Limiting Integral Sequence Output](../user-guide/latest/tuning/memory.md#limiting-integral-sequence-output).
+
 ### Resolving the pool type
 
 `parse_memory_pool_config` (`native/core/src/execution/memory_pools/config.rs`) turns the pool-type
@@ -610,10 +636,11 @@ somewhat less abrupt.
 
 ## Open problems
 
-Nothing described above prevents an executor from being OOM-killed. Every enforcement point Comet
-has bounds _declared reservations_, and the sections above describe several structural reasons why
-declared reservations are a lower bound on physical usage. The known gaps, roughly in order of how
-much they matter:
+Nothing described above guarantees that an executor avoids being OOM-killed. Memory pools bound
+_declared reservations_, while the integral `sequence` ceiling bounds one logical output without
+reserving it. Neither bounds total resident memory, and the sections above describe several reasons
+why declared reservations are a lower bound on physical usage. The known gaps, roughly in order of
+how much they matter:
 
 - **Real native usage is observed but not acted on.** Comet counts the bytes Rust's allocator has
   handed out, and each executor logs that count next to the pools' reservations. Nothing reads it

docs/source/user-guide/latest/migration-guide.md

@@ -59,8 +59,22 @@ key is removed.
 
 ## Upgrading to Comet 1.2.0
 
-Comet `1.2.0` makes no behavior changes that need a `spark.comet.legacy.*` key. The changes below
-need none either, but check whether any of them applies to your deployment.
+Review the following changes when upgrading to Comet `1.2.0`.
+
+### Native Integral Sequence Byte Ceiling
+
+Native integral `sequence` now checks each invocation's logical Arrow output size before allocating
+its values and offsets. In off-heap mode the default ceiling comes from the configured memory pool
+limit and task slots, so a valid query that previously completed can now fail when one invocation
+exceeds that ceiling. On-heap test mode has no default ceiling.
+
+Set `spark.comet.exec.sequence.maxBytesPerInvocation=-1` to restore the pre-1.2.0 absence of this
+byte ceiling. Existing Spark per-row and Arrow batch element limits still apply.
+
+To retain the new check, reduce `spark.comet.batchSize`, shorten the sequences or choose a larger
+`spark.comet.exec.sequence.maxBytesPerInvocation`. Retained results, concurrent calls and later
+scalar or folded-literal expansion do not consume an invocation's ceiling. See
+[Limiting Integral Sequence Output](tuning/memory.md#limiting-integral-sequence-output).
 
 ### In-Memory Cache Enabled by Default
 

docs/source/user-guide/latest/tuning/memory.md

@@ -272,6 +272,43 @@ narrow tables when memory is plentiful.
 `spark.comet.shuffle.jvm.batchSize` controls the batch size used when the JVM columnar shuffle writer
 flushes sorted spill files. A value larger than `spark.comet.batchSize` is capped at `spark.comet.batchSize`.
 
+## Limiting Integral Sequence Output
+
+`spark.comet.exec.sequence.maxBytesPerInvocation` limits the logical Arrow output of each native
+integral `sequence` invocation. The kernel validates all rows and checks this ceiling before
+allocating the output values and offsets. A refusal fails the task with the required byte count and
+the limit. Lower `spark.comet.batchSize` to fit fewer sequences in an invocation, shorten a sequence
+that is too large on its own, or explicitly raise this setting. Byte units such as `64m` are accepted;
+`-1` restores the pre-1.2.0 absence of this byte ceiling, and `0` allows no output bytes. Existing
+Spark per-row and Arrow batch element limits still apply. See the
+[Upgrade Guide](../migration-guide.md).
+
+When the setting is absent in off-heap mode and the core count is known, the default is:
+
+```text
+task slots = max(1, floor(local or executor cores / CPUs allocated to the task))
+sequence ceiling = Comet's configured memory pool limit / task slots
+```
+
+The pool limit is resolved by `CometExecIterator.getMemoryConfig` from
+`spark.memory.offHeap.size * spark.comet.exec.memoryPool.fraction`. Local mode uses the thread count
+in `spark.master`; executors use `spark.executor.cores`. For `local-cluster[workers,cores,memory]`,
+the configured worker cores are used when executor cores are unset. Standalone may assign cores
+without exposing their count in the executor configuration. In that case the fallback is the whole
+configured pool limit, not an inferred task share; set `spark.executor.cores` or an explicit
+sequence ceiling to choose a smaller limit.
+
+The task's CPU allocation normally comes from `spark.task.cpus`, and may be overridden by a resource
+profile; the executor configuration also reflects non-default profile allocations. The ceiling
+does not change with live task counts or pool usage. On-heap test mode has no default ceiling; an
+explicit setting still applies there.
+
+Logical bytes include the child values at their integral width, four bytes per list offset
+(including the terminal offset), and one validity bit per row when null lists are present, rounded
+up to whole bytes. Even an empty output has a four-byte terminal offset. Allocator padding, scratch
+space and shared input backing outside a validity slice are excluded. Validity preparation and the
+row-length scratch buffer may be allocated before the final check.
+
 ## Limiting Spill Disk Usage
 
 Native operators that spill to disk (aggregate, sort, shuffle) are bounded by

native/core/src/execution/jni_api.rs

@@ -132,7 +132,7 @@ use crate::execution::tracing::{
 
 use crate::execution::memory_pools::logging_pool::LoggingMemoryPool;
 use crate::execution::spark_config::{
-    SparkConfig, COMET_DEBUG_ENABLED, COMET_DEBUG_MEMORY,
+    sequence_max_bytes, SparkConfig, COMET_DEBUG_ENABLED, COMET_DEBUG_MEMORY,
     COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED, COMET_EXEC_SPILL_COMPRESSION_CODEC,
     COMET_EXPLAIN_NATIVE_ENABLED, COMET_MAX_TEMP_DIRECTORY_SIZE,
     COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED, COMET_STRICT_TIMESTAMP_LABELS,
@@ -558,6 +558,8 @@ struct ExecutionContext {
     pub debug_native: bool,
     /// Whether boundaries fail on a relabelled timestamp instead of casting it (test-only)
     pub strict_timestamp_labels: bool,
+    /// Resolved logical output ceiling for each integral sequence invocation.
+    sequence_max_bytes: Option<usize>,
     /// Whether to write native plans with metrics to stdout
     pub explain_native: bool,
     /// Whether to log memory usage on each call to execute_plan
@@ -710,6 +712,7 @@ struct PlanSettings {
     spark_config: HashMap<String, String>,
     debug_native: bool,
     strict_timestamp_labels: bool,
+    sequence_max_bytes: Option<usize>,
     explain_native: bool,
     tracing_enabled: bool,
     max_temp_directory_size: u64,
@@ -731,6 +734,7 @@ fn prepare_plan_settings(serialized_spark_configs: &[u8]) -> CometResult<PlanSet
     Ok(PlanSettings {
         debug_native: spark_config.get_bool(COMET_DEBUG_ENABLED),
         strict_timestamp_labels: spark_config.get_bool(COMET_STRICT_TIMESTAMP_LABELS),
+        sequence_max_bytes: sequence_max_bytes(&spark_config)?,
         explain_native: spark_config.get_bool(COMET_EXPLAIN_NATIVE_ENABLED),
         tracing_enabled: spark_config.get_bool(COMET_TRACING_ENABLED),
         max_temp_directory_size: spark_config
@@ -883,6 +887,7 @@ fn create_execution_context(
         session_ctx: session,
         debug_native: settings.debug_native,
         strict_timestamp_labels: settings.strict_timestamp_labels,
+        sequence_max_bytes: settings.sequence_max_bytes,
         explain_native: settings.explain_native,
         tracing_enabled: settings.tracing_enabled,
         rust_thread_id,
@@ -1464,6 +1469,7 @@ fn execute_plan(
         let planner = PhysicalPlanner::new(Arc::clone(&exec_context.session_ctx), partition)
             .with_exec_id(exec_context_id)
             .with_strict_timestamp_labels(exec_context.strict_timestamp_labels)
+            .with_sequence_max_bytes(exec_context.sequence_max_bytes)
             .with_sql_text_pool(&exec_context.spark_plan)
             .with_task_context(exec_context.task_context.clone())
             .with_class_loader(exec_context.class_loader.clone())

native/core/src/execution/planner.rs

@@ -94,7 +94,7 @@ use datafusion_comet_spark_expr::{
     create_comet_physical_fun, create_comet_physical_fun_with_eval_mode, BinaryOutputStyle,
     BloomFilterAgg, BloomFilterMightContain, CheckedBinaryExpr, CometCollectList, CometCollectSet,
     CsvWriteOptions, EvalMode, ListPositionsExpr, SparkArraysZipFunc, SparkBloomFilterVersion,
-    SparkListAgg, SparkPercentile, Subquery, SumInteger, ToCsv,
+    SparkListAgg, SparkPercentile, SparkSequence, Subquery, SumInteger, ToCsv,
 };
 use datafusion_datasource::TableSchema;
 use iceberg::expr::Bind;
@@ -358,6 +358,7 @@ pub struct PhysicalPlanner {
     /// Whether a boundary that would cast a timestamp's timezone label to the declared one fails
     /// instead. Test-only; see `spark.comet.testing.strictTimestampLabels.enabled`.
     strict_timestamp_labels: bool,
+    sequence_max_bytes: Option<usize>,
 }
 
 impl Default for PhysicalPlanner {
@@ -379,6 +380,7 @@ impl PhysicalPlanner {
             shuffle_partition_pusher: None,
             float_operands: FloatOperands::Normalize,
             strict_timestamp_labels: false,
+            sequence_max_bytes: None,
         }
     }
 
@@ -462,6 +464,11 @@ impl PhysicalPlanner {
         self
     }
 
+    pub fn with_sequence_max_bytes(mut self, max_bytes: Option<usize>) -> Self {
+        self.sequence_max_bytes = max_bytes;
+        self
+    }
+
     pub(crate) fn strict_timestamp_labels(&self) -> bool {
         self.strict_timestamp_labels
     }
@@ -4085,12 +4092,19 @@ impl PhysicalPlanner {
                 }
             };
 
-        let fun_expr = create_comet_physical_fun(
-            fun_name,
-            data_type.clone(),
-            &self.session_ctx.state(),
-            Some(expr.fail_on_error),
-        )?;
+        let fun_expr = if fun_name == "spark_sequence" {
+            Arc::new(ScalarUDF::new_from_impl(SparkSequence::new(
+                data_type.clone(),
+                self.sequence_max_bytes,
+            )))
+        } else {
+            create_comet_physical_fun(
+                fun_name,
+                data_type.clone(),
+                &self.session_ctx.state(),
+                Some(expr.fail_on_error),
+            )?
+        };
 
         let args = args
             .into_iter()

native/core/src/execution/spark_config.rs

@@ -17,6 +17,10 @@
 
 use std::collections::HashMap;
 
+use crate::errors::{CometError, CometResult};
+
+pub(crate) const COMET_EXEC_SEQUENCE_MAX_BYTES_PER_INVOCATION: &str =
+    "spark.comet.exec.sequence.maxBytesPerInvocation";
 pub(crate) const COMET_TRACING_ENABLED: &str = "spark.comet.tracing.enabled";
 pub(crate) const COMET_DEBUG_ENABLED: &str = "spark.comet.debug.enabled";
 pub(crate) const COMET_EXPLAIN_NATIVE_ENABLED: &str = "spark.comet.explain.native.enabled";
@@ -32,6 +36,21 @@ pub(crate) const COMET_STRICT_TIMESTAMP_LABELS: &str =
     "spark.comet.testing.strictTimestampLabels.enabled";
 pub(crate) const SPARK_EXECUTOR_CORES: &str = "spark.executor.cores";
 
+/// The JVM resolves the task's default and byte units before sending this value.
+pub(crate) fn sequence_max_bytes(config: &HashMap<String, String>) -> CometResult<Option<usize>> {
+    match config
+        .get(COMET_EXEC_SEQUENCE_MAX_BYTES_PER_INVOCATION)
+        .map(String::as_str)
+    {
+        None | Some("-1") => Ok(None),
+        Some(value) => value.parse::<usize>().map(Some).map_err(|_| {
+            CometError::Config(format!(
+                "Invalid resolved {COMET_EXEC_SEQUENCE_MAX_BYTES_PER_INVOCATION}: {value}"
+            ))
+        }),
+    }
+}
+
 /// Comet configs read through this trait must be resolved by the JVM first:
 /// `CometExecIterator.serializeCometSQLConfs` sends booleans as `true` or `false` and sizes as a
 /// bare byte count. A config missing from its list arrives exactly as the user wrote it, and a
@@ -61,3 +80,28 @@ impl SparkConfig for HashMap<String, String> {
             .unwrap_or(default_value)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn sequence_ceiling_requires_resolved_bytes() {
+        let mut config = HashMap::new();
+        assert_eq!(sequence_max_bytes(&config).unwrap(), None);
+        for (value, expected) in [("-1", None), ("0", Some(0)), ("1024", Some(1024))] {
+            config.insert(
+                COMET_EXEC_SEQUENCE_MAX_BYTES_PER_INVOCATION.into(),
+                value.into(),
+            );
+            assert_eq!(sequence_max_bytes(&config).unwrap(), expected);
+        }
+        for value in ["-2", "1k", "invalid", "18446744073709551616"] {
+            config.insert(
+                COMET_EXEC_SEQUENCE_MAX_BYTES_PER_INVOCATION.into(),
+                value.into(),
+            );
+            assert!(sequence_max_bytes(&config).is_err(), "{value}");
+        }
+    }
+}