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.

Solutions

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

// 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()))
    }
}

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

• unable to send response
• {batch write failure, with context()}
• bulk write respond with write errors
• bulk write write errors
• Can't create deltalake sink write result from empty data!

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

Appendix: 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)