risingwavelabs/risingwave · error · SinkError::LanceDb

Lance fragment write task stopped before accepting a record

Error message

Lance fragment write task stopped before accepting a record batch

What it means

In `write_batch`, the writer sends each RecordBatch over a bounded mpsc channel to a spawned fragment-write task that calls `FileFragment::create_fragments`. A send error means the receiver side was dropped — the spawned task has already exited (with an error or panic) before accepting the batch. This is a wrapper error; the real cause is recorded in the fragment-write task's own error ("failed to write lance data files") or its JoinHandle.

Source

Thrown at src/connector/src/sink/lancedb.rs:472

    async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {
        let record_batch = LanceDbConvert
            .to_record_batch(self.arrow_schema.clone(), &chunk)
            .context("failed to convert DataChunk to RecordBatch for LanceDB")
            .map_err(SinkError::LanceDb)?;

        if self.fragment_write.is_none() {
            self.fragment_write = Some(self.start_fragment_write());
        }
        self.fragment_write
            .as_ref()
            .expect("fragment write should be initialized")
            .sender
            .as_ref()
            .expect("fragment write sender should be initialized")
            .send(record_batch)
            .await
            .map_err(|_| {
                SinkError::LanceDb(anyhow!(
                    "Lance fragment write task stopped before accepting a record batch"
                ))
            })?;
        Ok(())
    }

    async fn begin_epoch(&mut self, _epoch: u64) -> Result<()> {
        Ok(())
    }

    async fn abort(&mut self) -> Result<()> {
        let Some(mut fragment_write) = self.fragment_write.take() else {
            return Ok(());
        };

        drop(fragment_write.sender.take());
        match fragment_write
            .task

View on GitHub (pinned to 6469eb736d)

Solutions

Look for the preceding 'failed to write lance data files' or task panic error in the logs; that is the root cause to fix.
Verify the sink's RisingWave schema matches the target Lance table's Arrow schema (names, types, nullability).
Check object-store credentials, permissions, and rate limits for write operations on the dataset URI.
Retry/respawn the sink after transient storage issues; consider reducing batch pressure or WRITE_CHANNEL_CAPACITY-related memory usage.

Defensive patterns

Strategy: try-catch

Try / catch

// Detect the dead channel and surface the root cause from the write task
match sender.send(record_batch).await {
    Ok(()) => Ok(()),
    Err(_) => {
        // The fragment write task already failed; join it to recover the real error
        let root = task.await
            .context("Lance fragment write task failed")
            .and_then(|r| r.context("failed to write lance data files"));
        Err(SinkError::LanceDb(anyhow!("fragment writer stopped before accepting a batch").context(root.unwrap_err()))
    }
}

Prevention

Keep the RisingWave sink schema in sync with the target Lance table's Arrow schema
Ensure write permissions on the dataset URI's bucket/prefix before starting the sink
Watch for transient object-store throttling and add backoff in the storage layer
Always inspect the fragment-write task's JoinHandle result — the send error only signals the task died

When it happens

Trigger: The `FileFragment::create_fragments` task terminates early — e.g. object-store write failure, invalid arrow schema mismatch between RecordBatch and the dataset schema, storage auth error, task panic — and the channel's receiver is dropped while `send(record_batch).await` is pending or before the next send.

Common situations: Transient S3/GCS errors or throttling during heavy sink writes; RisingWave schema drift vs. the target Lance table; insufficient permissions to write to the dataset directory; process memory pressure causing the write task to fail.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/2d576a13c53bd14f. Report an issue: GitHub.