{"record":{"id":"1d93130f34448026","repo":"risingwavelabs/risingwave","slug":"getchanneldeltastatsexecutor-should-have-no-child","errorCode":null,"errorMessage":"GetChannelDeltaStatsExecutor should have no child!","messagePattern":"GetChannelDeltaStatsExecutor should have no child!","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/batch/executors/src/executor/get_channel_delta_stats.rs","lineNumber":137,"sourceCode":"\n            let columns: Vec<_> = array_builders\n                .into_iter()\n                .map(|b| b.finish().into())\n                .collect();\n\n            let chunk = DataChunk::new(columns, rows.len());\n            yield chunk;\n        }\n    }\n}\n\nimpl BoxedExecutorBuilder for GetChannelDeltaStatsExecutor {\n    async fn new_boxed_executor(\n        source: &ExecutorBuilder<'_>,\n        inputs: Vec<BoxedExecutor>,\n    ) -> Result<BoxedExecutor> {\n        ensure!(\n            inputs.is_empty(),\n            \"GetChannelDeltaStatsExecutor should have no child!\"\n        );\n\n        let get_channel_delta_stats_node = try_match_expand!(\n            source.plan_node().get_node_body().unwrap(),\n            NodeBody::GetChannelDeltaStats\n        )?;\n\n        // Create a schema for channel stats\n        // This should match the expected schema from table_function.rs\n        let fields = vec![\n            Field::new(\"upstream_fragment_id\", DataType::Int32),\n            Field::new(\"downstream_fragment_id\", DataType::Int32),\n            Field::new(\"backpressure_rate\", DataType::Float64),\n            Field::new(\"recv_throughput\", DataType::Float64),\n            Field::new(\"send_throughput\", DataType::Float64),\n        ];\n","sourceCodeStart":119,"sourceCodeEnd":155,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/executors/src/executor/get_channel_delta_stats.rs#L119-L155","documentation":"During `deserialize_stream_chunk`, the number of decoded operations exceeded the `size_bound` the caller supplied for the chunk. `size_bound` caps how many rows a serialized chunk may contain so a maliciously or erroneously oversized payload cannot blow up memory; exceeding it aborts the decode.","triggerScenarios":"Decoding a `[start_seq_id, end_seq_id)` range whose stored `LogStoreOp::Row` entries total more than `size_bound` ops (checked after each `ops.push(op)`). Thrown at serde.rs:472.","commonSituations":"Read range spanning more rows than estimated (e.g. start/end seq ids computed from a bad row-count estimate); rows written without respecting the chunk size bound; replaying a range twice because seq-id tracking desynced.","solutions":["Ensure the write side never serializes more rows per chunk than the configured `size_bound`.","Fix the read range selection so `[start_seq_id, end_seq_id)` covers at most one chunk's rows.","Increase size_bound if the workload legitimately needs larger chunks (and align it with writer settings).","Inspect the seq-id range and stored rows to detect duplicated or miswritten entries."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// rust\n// estimate ops in the requested range before decoding\nlet est = reader.estimate_row_count(start_seq_id, end_seq_id);\nanyhow::ensure!(est <= size_bound, \"range [{}, {}) holds {} ops > size_bound {}\", start_seq_id, end_seq_id, est, size_bound);","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Enforce size_bound on the write side when serializing chunks.","Keep reader and writer size_bound settings in sync.","Never extend end_seq_id past one chunk's worth of rows."],"tags":["log-store","deserialization","size-bound","memory-safety"],"backgroundTag":"value-out-of-range","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"}