risingwavelabs/risingwave · error · SinkError::Iceberg

RecordBatch::try_new failed: {err}

Error message

RecordBatch::try_new failed: {err}

What it means

Arrow's `RecordBatch::try_new` rejected the assembled columns for the upsert path: after converting the chunk and appending the synthetic Int32 `op` column, the columns must exactly match `arrow_schema_with_op_column` (same count, order, types, lengths). Any mismatch — wrong `op` column type, wrong column order/count, or row-length inconsistency — makes `try_new` fail. RisingWave wraps the Arrow error in `SinkError::Iceberg`.

Source

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

            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);
                RecordBatch::try_new(arrow_schema_with_op_column.clone(), columns)
                    .map_err(|err| SinkError::Iceberg(anyhow!(err)))?
            }
        };
        Ok(Some((batch, write_batch_size)))
    }

    pub async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {
        self.prepare_writer()?;
        let Some((batch, write_batch_size)) = self.process_chunk(chunk)? else {
            return Ok(());
        };

        let writer = self.writer.get_writer().unwrap();
        let batch_rows = batch.num_rows();
        let batch_columns = batch.num_columns();
        writer
            .write(batch)
            .instrument_await("iceberg_write")
            .await

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the wrapped Arrow error — it names the exact field/index that mismatched.
2. Verify chunk columns (plus the appended op column) match `arrow_schema_with_op_column` in count, order, and type.
3. Recreate the sink so both the schema and the op-column construction are rebuilt consistently.
4. If reproducible on a fresh sink, report as a RisingWave internal bug with the schema and column layout.

// Verify the op column matches the schema's expected op field before try_new
let op_field = arrow_schema_with_op_column.field_with_name("op").unwrap();
anyhow::ensure!(op_field.data_type() == &ArrowDataType::Int32, "op column must be Int32");
anyhow::ensure!(columns.len() == arrow_schema_with_op_column.fields().len(), "column count mismatch");

fn batch_columns_match(schema: &arrow_schema::SchemaRef, columns: &[arrow_array::ArrayRef]) -> bool {
    schema.fields().len() == columns.len()
        && schema.fields().iter().zip(columns).all(|(f, c)| f.data_type() == c.data_type() && (f.is_nullable() || c.null_count() == 0))
}

RecordBatch::try_new(arrow_schema_with_op_column.clone(), columns)
    .map_err(|e| SinkError::Iceberg(anyhow!("RecordBatch::try_new failed: {e:#}; schema/column layout diverged — recreate the sink")))?

When it happens

Trigger: `write_batch`/`write_batch_with_position` on the upsert dispatch where the schema-with-op-column doesn't match the constructed column vector — e.g. the schema was built with the `op` field in a different position/type, or the chunk column types diverged from the schema.

Common situations: Internal schema construction drift (e.g. `op` column appended at the wrong index relative to the schema used at build time); table schema evolved so `arrow_schema_with_op_column` no longer matches converted chunk columns; version changes in the Arrow dependency altering validation strictness.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• error converting Arrow schema to Iceberg schema: {err}
• error converting StreamChunk to Arrow RecordBatch: {err}
• support negative scale for arrow decimal
• Can't create iceberg sink write result from empty data!
• iceberg sink metadata should be an object

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