{"record":{"id":"16a4a1b32bf5476c","repo":"risingwavelabs/risingwave","slug":"error-converting-streamchunk-to-arrow-recordbatch","errorCode":null,"errorMessage":"error converting StreamChunk to Arrow RecordBatch: {err}","messagePattern":"error converting StreamChunk to Arrow RecordBatch: (.+?)","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/writer.rs","lineNumber":729,"sourceCode":"                self.project_idx_vec = ProjectIdxVec::Done(project_idx_vec);\n            }\n            ProjectIdxVec::Done(idx_vec) => {\n                chunk = chunk.project(idx_vec);\n            }\n        }\n        if ops.is_empty() {\n            return Ok(None);\n        }\n        let write_batch_size = chunk.estimated_heap_size();\n        let batch = match &self.writer {\n            IcebergWriterDispatch::Append { .. } => {\n                // separate out insert chunk\n                let filters =\n                    chunk.visibility() & ops.iter().map(|op| *op == Op::Insert).collect::<Bitmap>();\n                chunk.set_visibility(filters);\n                IcebergArrowConvert\n                    .to_record_batch(self.arrow_schema.clone(), &chunk.compact_vis())\n                    .map_err(|err| SinkError::Iceberg(anyhow!(err)))?\n            }\n            IcebergWriterDispatch::Upsert {\n                arrow_schema_with_op_column,\n                ..\n            } => {\n                let chunk = IcebergArrowConvert\n                    .to_record_batch(self.arrow_schema.clone(), &chunk)\n                    .map_err(|err| SinkError::Iceberg(anyhow!(err)))?;\n                let ops = Arc::new(Int32Array::from(\n                    ops.iter()\n                        .map(|op| match op {\n                            Op::UpdateInsert | Op::Insert => INSERT_OP,\n                            Op::UpdateDelete | Op::Delete => DELETE_OP,\n                        })\n                        .collect_vec(),\n                ));\n                let mut columns = chunk.columns().to_vec();\n                columns.push(ops);","sourceCodeStart":711,"sourceCodeEnd":747,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/writer.rs#L711-L747","documentation":"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`.","triggerScenarios":"`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).","commonSituations":"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.","solutions":["Read the wrapped `err` to see which column/type mismatched.","Align MV column types with the Iceberg table (cast in the MV query or alter the table).","Recreate the sink after any MV schema change so the Arrow schema is rebuilt.","Ensure columns declared non-nullable in Iceberg never receive NULLs upstream."],"exampleFix":"// before: MV emits INT32 but table column is LONG (Int64)\nCREATE SINK s FROM mv WITH (connector='iceberg', table='t');\n// after: cast to match the table schema\nCREATE SINK s FROM (SELECT id::BIGINT AS id, ... FROM mv) WITH (connector='iceberg', table='t');","handlingStrategy":"validation","validationCode":"// Check chunk column types against the Arrow schema before conversion\nfn chunk_matches_schema(chunk: &StreamChunk, schema: &arrow_schema::SchemaRef) -> bool {\n    chunk.columns().len() == schema.fields().len()\n        && chunk.columns().iter().zip(schema.fields()).all(|(col, f)| {\n            col.array_ref().data_type() == f.data_type()\n        })\n}","typeGuard":"fn column_type_matches(col: &arrow_array::ArrayRef, field: &arrow_schema::FieldRef) -> bool {\n    col.data_type() == field.data_type()\n}","tryCatchPattern":"let batch = IcebergArrowConvert\n    .to_record_batch(self.arrow_schema.clone(), &chunk.compact_vis())\n    .map_err(|e| SinkError::Iceberg(anyhow!(\"chunk/schema mismatch: {e:#}; verify MV column types match the Iceberg table\")))?;","preventionTips":["Cast MV columns to exactly the Iceberg table types when creating the sink","Recreate the sink whenever the MV schema changes","Avoid NULLs in Iceberg columns declared as required (non-nullable)"],"tags":["iceberg","arrow","streamchunk","type-mismatch","sink"],"backgroundTag":"type-mismatch","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}