{"record":{"id":"c8c95fc9b1772139","repo":"risingwavelabs/risingwave","slug":"recordbatch-try-new-failed-err","errorCode":null,"errorMessage":"RecordBatch::try_new failed: {err}","messagePattern":"RecordBatch::try_new failed: (.+?)","errorType":"exception","errorClass":"SinkError::Iceberg","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/iceberg/writer.rs","lineNumber":749,"sourceCode":"            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);\n                RecordBatch::try_new(arrow_schema_with_op_column.clone(), columns)\n                    .map_err(|err| SinkError::Iceberg(anyhow!(err)))?\n            }\n        };\n        Ok(Some((batch, write_batch_size)))\n    }\n\n    pub async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {\n        self.prepare_writer()?;\n        let Some((batch, write_batch_size)) = self.process_chunk(chunk)? else {\n            return Ok(());\n        };\n\n        let writer = self.writer.get_writer().unwrap();\n        let batch_rows = batch.num_rows();\n        let batch_columns = batch.num_columns();\n        writer\n            .write(batch)\n            .instrument_await(\"iceberg_write\")\n            .await","sourceCodeStart":731,"sourceCodeEnd":767,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/iceberg/writer.rs#L731-L767","documentation":"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`.","triggerScenarios":"`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.","commonSituations":"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.","solutions":["Read the wrapped Arrow error — it names the exact field/index that mismatched.","Verify chunk columns (plus the appended op column) match `arrow_schema_with_op_column` in count, order, and type.","Recreate the sink so both the schema and the op-column construction are rebuilt consistently.","If reproducible on a fresh sink, report as a RisingWave internal bug with the schema and column layout."],"exampleFix":null,"handlingStrategy":"type-guard","validationCode":"// Verify the op column matches the schema's expected op field before try_new\nlet op_field = arrow_schema_with_op_column.field_with_name(\"op\").unwrap();\nanyhow::ensure!(op_field.data_type() == &ArrowDataType::Int32, \"op column must be Int32\");\nanyhow::ensure!(columns.len() == arrow_schema_with_op_column.fields().len(), \"column count mismatch\");","typeGuard":"fn batch_columns_match(schema: &arrow_schema::SchemaRef, columns: &[arrow_array::ArrayRef]) -> bool {\n    schema.fields().len() == columns.len()\n        && schema.fields().iter().zip(columns).all(|(f, c)| f.data_type() == c.data_type() && (f.is_nullable() || c.null_count() == 0))\n}","tryCatchPattern":"RecordBatch::try_new(arrow_schema_with_op_column.clone(), columns)\n    .map_err(|e| SinkError::Iceberg(anyhow!(\"RecordBatch::try_new failed: {e:#}; schema/column layout diverged — recreate the sink\")))?","preventionTips":["Build the schema-with-op-column and the op column from a single source of truth so index/type always agree","Add a debug assertion comparing schema fields with constructed columns in tests","Recreate the sink after any table schema evolution to rebuild both schemas consistently"],"tags":["iceberg","arrow","record-batch","schema-validation","sink"],"backgroundTag":"schema-validation-failed","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"}