{"record":{"id":"6c5ab5ddbce5f4cb","repo":"risingwavelabs/risingwave","slug":"filter-can-only-receive-bool-array","errorCode":null,"errorMessage":"Filter can only receive bool array","messagePattern":"Filter can only receive bool array","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/batch/executors/src/executor/filter.rs","lineNumber":68,"sourceCode":"    #[try_stream(boxed, ok = DataChunk, error = BatchError)]\n    async fn do_execute(self: Box<Self>) {\n        let mut data_chunk_builder =\n            DataChunkBuilder::new(self.child.schema().data_types(), self.chunk_size);\n\n        #[for_await]\n        for data_chunk in self.child.execute() {\n            let data_chunk = data_chunk?.compact_vis();\n            let vis_array = self.expr.eval(&data_chunk).await?;\n\n            if let Bool(vis) = vis_array.as_ref() {\n                // TODO: should we yield masked data chunk directly?\n                for data_chunk in\n                    data_chunk_builder.append_chunk(data_chunk.with_visibility(vis.to_bitmap()))\n                {\n                    yield data_chunk;\n                }\n            } else {\n                bail!(\"Filter can only receive bool array\");\n            }\n        }\n\n        if let Some(chunk) = data_chunk_builder.consume_all() {\n            yield chunk;\n        }\n    }\n}\n\nimpl BoxedExecutorBuilder for FilterExecutor {\n    async fn new_boxed_executor(\n        source: &ExecutorBuilder<'_>,\n        inputs: Vec<BoxedExecutor>,\n    ) -> Result<BoxedExecutor> {\n        let [input]: [_; 1] = inputs.try_into().unwrap();\n\n        let filter_node = try_match_expand!(\n            source.plan_node().get_node_body().unwrap(),","sourceCodeStart":50,"sourceCodeEnd":86,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/executors/src/executor/filter.rs#L50-L86","documentation":"The KV log store reader's `truncate` method was asked to truncate at an offset that is not newer than the offset of the most recent truncation already applied. Truncations must be monotonically increasing; re-truncating at or before the previous point would delete nothing new and indicates a bug in the caller's bookkeeping of progress (epoch/sequence offsets).","triggerScenarios":"Calling `KvLogStoreReader::truncate(offset)` where `offset <= self.truncate_offset`, while `offset.epoch() >= first_write_epoch` (i.e. truncating within the current buffered epoch range). Thrown at reader.rs:529.","commonSituations":"Stream executor recovery re-delivers an older barrier after a restart; Hummock version snapshots rewinding epoch progress; duplicated barrier handling in the upstream executor passing the same truncate offset twice; epoch/seq-id state not persisted across actor restarts.","solutions":["Fix the caller so it only issues truncate calls with strictly increasing offsets (skip if offset <= previous truncate offset).","Check how truncate_offset is persisted/recovered on actor restart; ensure recovery does not reset progress to an older barrier.","Verify barrier alignment logic upstream is not re-injecting an already-processed barrier epoch.","If seen after a version change, confirm the Hummock version/epoch watermark fed to the reader advances monotonically."],"exampleFix":"// before\nreader.truncate(offset); // called unconditionally on every barrier\n// after\nif reader.last_truncate_offset().map_or(true, |prev| offset > prev) {\n    reader.truncate(offset);\n}","handlingStrategy":"validation","validationCode":"// rust\nfn safe_truncate(reader: &mut KvLogStoreReader, offset: TruncateOffset) -> anyhow::Result<()> {\n    if let Some(prev) = reader.last_truncate_offset() {\n        anyhow::ensure!(offset > prev, \"skip stale truncate offset {:?} (prev {:?})\", offset, prev);\n    }\n    reader.truncate(offset)\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Track the last truncate offset in the caller and skip no-op/regressing truncations.","Persist truncate progress across restarts so recovery does not rewind.","Add an assertion/log when a barrier epoch regresses upstream."],"tags":["log-store","streaming","monotonic-offset","truncate"],"backgroundTag":"invalid-state-transition","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"}