diff --git a/src/service/pipeline/batch_execution.rs b/src/service/pipeline/batch_execution.rs index c20d02c9a9..723a672841 100644 --- a/src/service/pipeline/batch_execution.rs +++ b/src/service/pipeline/batch_execution.rs @@ -792,6 +792,14 @@ async fn process_node( }; } + // A null / non-object record means an upstream function filtered it + // out (e.g. VRL `. = null`). Drop it instead of forwarding to the + // destination, where it would fail at handle_timestamp with + // "Value is not an object" and inflate the failed count. + if !record.is_object() { + continue; + } + let mut destination_stream = stream_params.clone(); if destination_stream.stream_name.contains("{") { match resolve_stream_name(&destination_stream.stream_name, &record) { @@ -943,6 +951,13 @@ async fn process_node( flattened = true; } + // A null / non-object record means an upstream function filtered it out + // (e.g. VRL `. = null`). Skip it: it can't satisfy any condition and + // `as_object().unwrap()` below would panic on a null. + if !record.is_object() { + continue; + } + // Evaluate based on condition version let eval_timer = Instant::now(); let passes = match condition_params {