risingwavelabs/risingwave · error · SinkError::Iceberg

error converting StreamChunk to Arrow RecordBatch: {err}

Error message

error converting StreamChunk to Arrow RecordBatch: {err}

What it means

RisingWave failed to convert an append-path StreamChunk into an Arrow RecordBatch using the writer's `arrow_schema` via `IcebergArrowConvert.to_record_batch`. This conversion checks that each chunk column's data type and nullability match the schema derived from the Iceberg table, so it fails on any type/shape mismatch. The error is wrapped in `SinkError::Iceberg`.

Source

Thrown at src/connector/src/sink/iceberg/writer.rs:729

                self.project_idx_vec = ProjectIdxVec::Done(project_idx_vec);
            }
            ProjectIdxVec::Done(idx_vec) => {
                chunk = chunk.project(idx_vec);
            }
        }
        if ops.is_empty() {
            return Ok(None);
        }
        let write_batch_size = chunk.estimated_heap_size();
        let batch = match &self.writer {
            IcebergWriterDispatch::Append { .. } => {
                // separate out insert chunk
                let filters =
                    chunk.visibility() & ops.iter().map(|op| *op == Op::Insert).collect::<Bitmap>();
                chunk.set_visibility(filters);
                IcebergArrowConvert
                    .to_record_batch(self.arrow_schema.clone(), &chunk.compact_vis())
                    .map_err(|err| SinkError::Iceberg(anyhow!(err)))?
            }
            IcebergWriterDispatch::Upsert {
                arrow_schema_with_op_column,
                ..
            } => {
                let chunk = IcebergArrowConvert
                    .to_record_batch(self.arrow_schema.clone(), &chunk)
                    .map_err(|err| SinkError::Iceberg(anyhow!(err)))?;
                let ops = Arc::new(Int32Array::from(
                    ops.iter()
                        .map(|op| match op {
                            Op::UpdateInsert | Op::Insert => INSERT_OP,
                            Op::UpdateDelete | Op::Delete => DELETE_OP,
                        })
                        .collect_vec(),
                ));
                let mut columns = chunk.columns().to_vec();
                columns.push(ops);

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the wrapped `err` to see which column/type mismatched.
2. Align MV column types with the Iceberg table (cast in the MV query or alter the table).
3. Recreate the sink after any MV schema change so the Arrow schema is rebuilt.
4. Ensure columns declared non-nullable in Iceberg never receive NULLs upstream.

Example fix

// before: MV emits INT32 but table column is LONG (Int64)
CREATE SINK s FROM mv WITH (connector='iceberg', table='t');
// after: cast to match the table schema
CREATE SINK s FROM (SELECT id::BIGINT AS id, ... FROM mv) WITH (connector='iceberg', table='t');

// Check chunk column types against the Arrow schema before conversion
fn chunk_matches_schema(chunk: &StreamChunk, schema: &arrow_schema::SchemaRef) -> bool {
    chunk.columns().len() == schema.fields().len()
        && chunk.columns().iter().zip(schema.fields()).all(|(col, f)| {
            col.array_ref().data_type() == f.data_type()
        })
}

fn column_type_matches(col: &arrow_array::ArrayRef, field: &arrow_schema::FieldRef) -> bool {
    col.data_type() == field.data_type()
}

let batch = IcebergArrowConvert
    .to_record_batch(self.arrow_schema.clone(), &chunk.compact_vis())
    .map_err(|e| SinkError::Iceberg(anyhow!("chunk/schema mismatch: {e:#}; verify MV column types match the Iceberg table")))?;

When it happens

Trigger: `write_batch`/`write_batch_with_position` on the append dispatch where the StreamChunk column types do not match the Iceberg-derived Arrow schema — e.g. MV column types changed, or the Iceberg table schema and the MV schema diverged (cast/rounding differences at sink creation).

Common situations: Altering the MV after sink creation; sink created against a table whose column types differ (e.g. INT vs BIGINT, timestamp precision mismatch); non-nullable Iceberg columns receiving null data from the MV.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• schema_id should be a u64
• error converting Arrow schema to Iceberg schema: {err}
• RecordBatch::try_new failed: {err}
• support negative scale for arrow decimal
• Convert from arrow error: {0}

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