{"record":{"id":"f252d546d9be7e98","repo":"risingwavelabs/risingwave","slug":"only-support-insert-op-in-batch-source-executor","errorCode":null,"errorMessage":"Only support insert op in batch source executor","messagePattern":"Only support insert op in batch source executor","errorType":"validation","errorClass":"BatchError","httpStatus":null,"severity":"error","filePath":"src/batch/executors/src/executor/source.rs","lineNumber":184,"sourceCode":"        for event in stream {\n            let chunk = match event.map_err(BatchError::connector)? {\n                SourceReaderEvent::DataChunk(chunk) => chunk,\n                SourceReaderEvent::SplitProgress(_) => continue,\n            };\n            let data_chunk = convert_stream_chunk_to_batch_chunk(chunk)?;\n            if data_chunk.capacity() > 0 {\n                yield data_chunk;\n            }\n        }\n    }\n}\n\nfn convert_stream_chunk_to_batch_chunk(chunk: StreamChunk) -> Result<DataChunk> {\n    // chunk read from source must be compact\n    assert!(chunk.data_chunk().is_vis_compacted());\n\n    if chunk.ops().iter().any(|op| *op != Op::Insert) {\n        bail!(\"Only support insert op in batch source executor\");\n    }\n\n    Ok(chunk.data_chunk().clone())\n}\n","sourceCodeStart":166,"sourceCodeEnd":189,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/executors/src/executor/source.rs#L166-L189","documentation":"convert_stream_chunk_to_batch_chunk converts chunks read by the batch source executor into batch DataChunks, and only INSERT operations are valid there. A batch query scanning a source expects append-only data; any Delete, UpdateInsert, or UpdateDelete op means the source produced non-append-only stream data and the executor bails out.","triggerScenarios":"Batch source executor reads a StreamChunk whose ops contain anything other than Op::Insert — e.g. the underlying connector/table emits deletes or updates, or the chunk was not visibility-compacted (the preceding assert fails).","commonSituations":"Pointing a batch query at a source or table that can emit mutation records (CDC with deletes/updates) when the batch path only supports append-only reads; connector or stream-fragment bugs producing non-insert ops; version changes where batch source semantics changed.","solutions":["Ensure the queried source/table is append-only for batch reads, or filter out mutation records upstream.","Use a materialized view over the source (which handles updates/deletes) and query the MV instead of the raw source in batch mode.","If the connector is expected to be append-only, investigate why it emitted Delete/Update ops (e.g. CDC configuration emitting full row changes).","Check that the stream chunk is compacted before conversion; call chunk.compact() upstream if needed."],"exampleFix":"// before: batch query directly against CDC source emitting updates/deletes\nSELECT * FROM cdc_source;\n// after: query a materialized view that handles mutations\nCREATE MATERIALIZED VIEW mv AS SELECT * FROM cdc_source;\nSELECT * FROM mv;","handlingStrategy":"validation","validationCode":"// Rust: check ops before converting\nfn is_append_only(chunk: &StreamChunk) -> bool {\n    chunk.ops().iter().all(|op| *op == Op::Insert)\n}","typeGuard":null,"tryCatchPattern":"match convert_stream_chunk_to_batch_chunk(chunk) {\n    Ok(c) => /* consume */,\n    Err(e) if e.to_string().contains(\"Only support insert op\") => {\n        // fallback: query a materialized view instead of the raw source\n    }\n}","preventionTips":["Only batch-query append-only sources; use materialized views for CDC/mutation data.","Compact stream chunks (chunk.compact()) before conversion.","Verify connector configuration does not emit deletes/updates on sources read in batch mode."],"tags":["batch-executor","source-connector","stream-chunk","append-only"],"backgroundTag":"unsupported-operation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}