panic in arrow_array RecordBatch.columns index out of bounds on single-pipeline multi-INSERT chained via a shared subject

[claud42]

Disclose: the content below was mostly generated by an AI agent. However, before creating this issue, I have manually verified the issue's existence, confirmed it could be reliably re-produced, eliminated other potential root causes, reviewed crash backtrace, and be responsible for accuracy of all content in this report.

Environment

• Arroyo version: 0.15.0 (image ghcr.io/arroyosystems/arroyo:0.15.0, single-container)
• Deployment: Docker Compose, embedded worker (not remote / kubernetes)
• Sources: NATS JetStream, format = 'json', watermark = event_time - INTERVAL '10 seconds'
• Parallelism: 1
• OS: Linux 6.8.0 (x86_64)

Summary

A single pipeline that contains two INSERT statements chained via a shared NATS subject (stage-1 sinks to subject X; stage-2's source reads subject X back; stage-2 sinks to a final subject) compiles cleanly and transitions Scheduling → Running, but then panics ~30–80 ms into the Running state with an arrow RecordBatch index-out-of-bounds. The controller restarts the worker; each restart panics identically within a few hundred ms. After ~5 restarts the job is marked Failed with Job has restarted too many times.

Panic (worker stdout, RUST_LOG=info)

Every worker restart panics with the identical 3-line chain, ~500 ms apart:

panicked at /usr/local/cargo/registry/src/index.crates.io-1949cf8c6b5b557f/arrow-array-55.2.0/src/record_batch.rs:607:22:
    index out of bounds: the len is 17 but the index is 17

panicked at crates/arroyo-operator/src/context.rs:596:22:
    called `Result::unwrap()` on an `Err` value: SendError { .. }

panicked at crates/arroyo-worker/src/lib.rs:779:14:
    called `Result::unwrap()` on an `Err` value: SendError { .. }

Controller log for the first restart (run_id=1):

state transition job_id="job_fkVk1YFBof" from="Scheduling" to="Running" duration_ms=412
ERROR arroyo_server_common: panicked at .../arrow-array-55.2.0/src/record_batch.rs:607:22:
    index out of bounds: the len is 17 but the index is 17
ERROR arroyo_server_common: panicked at crates/arroyo-operator/src/context.rs:596:22:
    called `Result::unwrap()` on an `Err` value: SendError { .. }
ERROR arroyo_controller::job_controller: task failed
    job_id="job_fkVk1YFBof" operator_id=22 subtask=0
    reason="task 319 panicked with message \"index out of bounds: the len is 17 but the index is 17\""
ERROR arroyo_controller::states::running: error while running
    error="worker failed" job_id="job_fkVk1YFBof"
state transition job_id="job_fkVk1YFBof" from="Running" to="Recovering" duration_ms=287
ERROR arroyo_server_common: panicked at crates/arroyo-worker/src/lib.rs:779:14:
    called `Result::unwrap()` on an `Err` value: SendError { .. }
WARN  arroyo_controller::states::recovering: failed to stop job
    error="status: Cancelled, message: \"h2 protocol error: http2 error\" ... stream no longer needed"
INFO  arroyo_controller::states::recovering: sending SIGKILL to workers
INFO  arroyo_controller::schedulers: Killing child worker_id=193 job_id="job_fkVk1YFBof"

After SIGKILL the state machine spins up a fresh worker (worker_id=194, run_id=2) which panics identically at the same operator ~30 ms into its own Running state. After run_id=5 the restart budget is exhausted and the job settles in Failed.

Operator topology in this pipeline

From the pipeline graph (GET /api/v1/pipelines/<id>), the 16-task DAG is:

node_id=  0  NatsSource<"EVENTS">                          // source_a read
node_id=  1  source_a -> watermark -> ArrowKey<_>
node_id=  4  sliding window -> ArrowKey<left>               // stage-1 source_a HOP agg
node_id=  6  NatsSource<"EVENTS">                          // source_b read
node_id=  7  source_b -> watermark -> ArrowKey<_>
node_id= 10  sliding window -> ArrowKey<right>              // stage-1 source_b HOP agg
node_id= 12  join -> sink projection                        // stage-1 join
node_id= 14  NatsSink<"events.chain.stage">                 // stage-1 sink
node_id= 15  NatsSource<"EVENTS">                          // intermediate_in read (same stream, stage-1 subject)
node_id= 16  intermediate_in -> watermark -> ArrowKey<_>
node_id= 19  sliding window -> ArrowKey<left>               // stage-2 intermediate HOP agg
node_id= 21  NatsSource<"EVENTS">                          // source_c read
node_id= 22  source_c -> watermark -> ArrowKey<_>           // STAGE-2 source_c KEY/WATERMARK CHAINED_OP
node_id= 25  sliding window -> ArrowKey<right>              // stage-2 source_c HOP agg
node_id= 27  join -> sink projection                        // stage-2 join
node_id= 29  NatsSink<"alerts.chain.final">

The originating panic site is operator_id=22, subtask=0, task 319 — the stage-2 source_c -> watermark -> ArrowKey<_> chained_op, i.e. the watermark/keying step that sits between the stage-2 source_c NATS source and its HOP aggregation. The SendError cascade is reported on operator_id=21, task 307, which is the immediately upstream NatsSource<"EVENTS"> for source_c — consistent with it trying to forward its next batch to op 22's already-dropped receiver.

Panic-cascade interpretation

1. Root: op 22 (stage-2 source_c watermark/keying chained_op) panics on arrow RecordBatch::project/column-style index lookup — something asks for column index 17 in a batch with len()==17 (valid indices 0..=16).
2. Cascade: upstream op 21 (stage-2 source_c NatsSource) panics on Result::unwrap() → SendError at arroyo-operator/src/context.rs:596:22 (downstream channel receiver gone).
3. Cascade: worker panics on Result::unwrap() → SendError at arroyo-worker/src/lib.rs:779:14 during shutdown propagation.

Steps 2 and 3 are artifacts of step 1, not independent bugs. The root cause is step 1.

Minimal reproducer

Three NATS source DDLs + one sink-that-is-also-a-source intermediate + one final sink + two INSERT statements in a single pipeline. All inside one POST /api/v1/pipelines call. Compiles cleanly, crashes on launch.

-- Source A — 12 user columns + watermark.
CREATE TABLE source_a (
    event_time TIMESTAMP NOT NULL,
    k_text      TEXT,          -- shared join key across all three sources
    c_text_01   TEXT,
    c_text_02   TEXT,
    c_text_03   TEXT,
    c_text_04   TEXT,
    c_text_05   TEXT,
    c_text_06   TEXT,
    c_text_07   TEXT,
    c_text_08   TEXT,
    c_text_09   TEXT,
    c_text_10   TEXT,
    watermark FOR event_time AS (event_time - INTERVAL '10 seconds')
) WITH (
    type = 'source', connector = 'nats', servers = 'nats:4222',
    stream = 'EVENTS',
    'consumer.filter_subjects' = 'events.src.a',
    'consumer.ack_policy' = 'Explicit', 'consumer.replay_policy' = 'Instant',
    format = 'json'
);

-- Source B — 13 user columns + watermark.
CREATE TABLE source_b (
    event_time TIMESTAMP NOT NULL,
    k_text      TEXT,
    c_text_01   TEXT,
    c_text_02   TEXT,
    c_text_03   TEXT,
    c_text_04   TEXT,
    c_text_05   TEXT,
    c_text_06   TEXT,
    c_int_01    INT,
    c_int_02    INT,
    c_int_03    INT,
    c_int_04    INT,
    c_int_05    INT,
    watermark FOR event_time AS (event_time - INTERVAL '10 seconds')
) WITH (
    type = 'source', connector = 'nats', servers = 'nats:4222',
    stream = 'EVENTS',
    'consumer.filter_subjects' = 'events.src.b',
    'consumer.ack_policy' = 'Explicit', 'consumer.replay_policy' = 'Instant',
    format = 'json'
);

-- Source C — 15 user columns + watermark. (Column count matters: see Root cause hypothesis.)
CREATE TABLE source_c (
    c_int_id    INT,
    event_time  TIMESTAMP NOT NULL,
    k_text      TEXT,
    c_text_01   TEXT,
    c_text_02   TEXT,
    c_text_03   TEXT,
    c_text_04   TEXT,
    c_text_05   TEXT,
    c_text_06   TEXT,
    c_text_07   TEXT,
    c_text_08   TEXT,
    c_text_09   TEXT,
    c_int_01    INT,
    c_text_10   TEXT,
    c_text_11   TEXT,
    watermark FOR event_time AS (event_time - INTERVAL '10 seconds')
) WITH (
    type = 'source', connector = 'nats', servers = 'nats:4222',
    stream = 'EVENTS',
    'consumer.filter_subjects' = 'events.src.c',
    'consumer.ack_policy' = 'Explicit', 'consumer.replay_policy' = 'Instant',
    format = 'json'
);

-- Intermediate sink — stage-1 output. Same JetStream stream, different subject.
CREATE TABLE intermediate_out (
    event_time TIMESTAMP, k_text TEXT,
    window_start TIMESTAMP, window_end TIMESTAMP,
    count_a BIGINT, count_b BIGINT
) WITH (
    type = 'sink', connector = 'nats', servers = 'nats:4222',
    subject = 'events.chain.stage',
    format = 'json'
);

-- Stage-2 source reading the intermediate subject back. Crucially declared
-- in the SAME pipeline as stage-1's sink — the trigger for this bug.
CREATE TABLE intermediate_in (
    event_time TIMESTAMP NOT NULL,
    k_text TEXT,
    window_start TIMESTAMP, window_end TIMESTAMP,
    count_a BIGINT, count_b BIGINT,
    watermark FOR event_time AS (event_time - INTERVAL '10 seconds')
) WITH (
    type = 'source', connector = 'nats', servers = 'nats:4222',
    stream = 'EVENTS',
    'consumer.filter_subjects' = 'events.chain.stage',
    'consumer.ack_policy' = 'Explicit', 'consumer.replay_policy' = 'Instant',
    format = 'json'
);

-- Final sink.
CREATE TABLE final_out (
    k_text TEXT,
    window_start TIMESTAMP, window_end TIMESTAMP,
    count_a BIGINT, count_b BIGINT,
    count_c BIGINT
) WITH (
    type = 'sink', connector = 'nats', servers = 'nats:4222',
    subject = 'alerts.chain.final',
    format = 'json'
);

-- Stage-1 INSERT: source_a × source_b HOP JOIN → intermediate subject.
INSERT INTO intermediate_out
WITH a_agg AS (
    SELECT k_text,
           HOP(INTERVAL '2 minutes', INTERVAL '10 minutes') AS window,
           COUNT(*) AS count_a
    FROM source_a
    WHERE c_text_01 = 'x' AND c_text_02 = 'y' AND k_text IS NOT NULL
    GROUP BY k_text, window
    HAVING COUNT(*) >= 1
),
b_agg AS (
    SELECT k_text,
           HOP(INTERVAL '2 minutes', INTERVAL '10 minutes') AS window,
           COUNT(*) AS count_b
    FROM source_b
    WHERE c_text_01 = 'allowed' AND k_text IS NOT NULL
    GROUP BY k_text, window
    HAVING COUNT(*) >= 1
)
SELECT
    a.window.start AS event_time,
    a.k_text,
    a.window.start AS window_start,
    a.window.end   AS window_end,
    a.count_a,
    b.count_b
FROM a_agg AS a
INNER JOIN b_agg AS b
       ON a.k_text = b.k_text
      AND a.window = b.window;

-- Stage-2 INSERT: reads intermediate subject, HOP-joins with source_c → final sink.
INSERT INTO final_out
WITH stage_agg AS (
    SELECT k_text,
           HOP(INTERVAL '2 minutes', INTERVAL '10 minutes') AS window,
           MIN(count_a) AS count_a,
           MIN(count_b) AS count_b
    FROM intermediate_in
    GROUP BY k_text, window
    HAVING COUNT(*) >= 1
),
c_agg AS (
    SELECT k_text,
           HOP(INTERVAL '2 minutes', INTERVAL '10 minutes') AS window,
           COUNT(*) AS count_c
    FROM source_c
    WHERE c_int_id IN (1, 3) AND k_text IS NOT NULL
    GROUP BY k_text, window
    HAVING COUNT(*) >= 1
)
SELECT
    x.k_text,
    x.window.start AS window_start,
    x.window.end   AS window_end,
    x.count_a,
    x.count_b,
    c.count_c
FROM stage_agg AS x
INNER JOIN c_agg AS c
       ON x.k_text = c.k_text
      AND x.window = c.window;

Reproduction steps

1. Start Arroyo 0.15.0 + a NATS JetStream container on a shared docker network. Create an EVENTS stream covering events.> (the intermediate stage-1 subject events.chain.stage is inside the same stream, per the DDL above).
2. Ensure at least some JSON messages are present on events.src.a, events.src.b, events.src.c (from a prior producer, or published fresh). The panic reproduces regardless of backlog size; it fires ~30–80 ms into Running, not after any specific message count.
3. POST /api/v1/pipelines with the SQL above and {"parallelism": 1, "udfs": []}. Returns 200 OK with a pipeline id — the planner accepts it.
4. Poll GET /api/v1/pipelines/{id}/jobs every ~5 s. Job transitions Scheduling → Running → Failed within ~15 seconds.
5. docker logs <arroyo-container> contains the 3-line panic chain above, repeated ~5 times as the controller restarts the worker.

Observed timing from a fresh run (wall clock, UTC):

T+0.000s   POST /api/v1/pipelines               → 200 OK (pipeline pl_ywhFULm9Vb, job_fkVk1YFBof)
T+0.307s   state: Created → Compiling → Scheduling
T+0.720s   state: Scheduling → Running
T+0.782s   PANIC arrow-array:607:22 index OOB len=17 idx=17 (operator_id=22, task 319)
T+1.008s   task failed → Running → Recovering
T+1.200s   SIGKILL worker 193; spin up worker 194 (run_id=2)
T+1.563s   worker 194 Scheduling → Running
T+1.606s   PANIC again (identical chain) on worker 194
...        (same cycle for run_id=3, 4, 5)
T+~13s     state: Failed, failure_message="Job has restarted too many times"

Root cause hypothesis

The panic line — arrow-array-55.2.0/src/record_batch.rs:607:22, index out of bounds: the len is 17 but the index is 17 — is almost certainly RecordBatch::column(i) or RecordBatch::project(&[…]) asking for column index 17 in a batch whose schema has exactly 17 columns (valid indices 0..=16). An off-by-one at a column-count / schema-width boundary in the operator's arrow-batch access path.

The batch width 17 is suggestive:

• source_c has 15 user-declared columns + 1 watermark column = 16. Plus one Arroyo-internal key/metadata column → 17 columns of physical arrow schema, last valid index 16.
• The panic location being operator_id=22 (the source_c -> watermark -> ArrowKey<_> chained_op) means the operator whose input schema has width 17 is being asked to project/fetch column 17 — one past the end.

The leading hypothesis is that when Arroyo compiles two INSERTs into a single job graph where the two INSERTs have independent logical plans connected only through a NATS subject (not an in-process edge), it is wiring some per-operator column-index metadata once for the whole job rather than per-subgraph. Specifically, a suspicion is that one of the INSERT's projections is computing "take column 17" based on a schema that has been augmented (e.g. by stage-2's own source metadata), but that augmentation is being applied to the wrong subgraph's operator — the stage-2 source_c watermark op is then handed column indices sized for a schema that includes extra columns which are actually only present in the other subgraph.

Corroborating evidence:

• The exact same SQL split into two pipelines (two POSTs → two independent job graphs, each with its own schema registry) has no panic.
• The panic fires at the first batch processed by op 22 — i.e. once real arrow data enters the column-access path, not at graph-install time. Plan-install doesn't catch it because plan-install validates logical types, not column indices against the physical schema registered for an individual operator.
• Op 22 is a chained_op (watermark -> ArrowKey<_>) — chained_op fusion is one place where column-index arithmetic is re-derived from the fused child operators' schemas. If the fused view of the child schemas is being computed against a shared-pipeline schema pool, single-pipeline-multi-INSERT is exactly the shape that would miscount.

This is a hypothesis from panic site + DAG structure + "split works / single fails" control; no pin-down of the specific code path that miscounts. Maintainers with access to the fused-schema / batch-projection paths in arroyo-planner / arroyo-operator would be better-placed to confirm. The file:line pointers below should localize it quickly.