fix: defer expensive fallbacks and filter null VRL results (#14559)

src/api/grpc/src/handler/grpc/request/ingest.rs

@@ -118,17 +118,7 @@ impl Ingest for Ingester {
                 }
             }
             StreamType::EnrichmentTables => {
-                let json_records: Vec<json::Map<String, json::Value>> =
-                    json::from_slice(&in_data.data).unwrap_or({
-                        let vec_value: Vec<json::Value> = json::from_slice(&in_data.data).unwrap();
-                        vec_value
-                            .into_iter()
-                            .filter_map(|v| match v {
-                                json::Value::Object(map) => Some(map),
-                                _ => None,
-                            })
-                            .collect()
-                    });
+                let json_records = parse_enrichment_records(&in_data.data);
                 let append_data = match req.metadata {
                     Some(metadata) => metadata
                         .data
@@ -217,6 +207,19 @@ fn ok_reply() -> IngestionResponse {
     }
 }
 
+fn parse_enrichment_records(data: &[u8]) -> Vec<json::Map<String, json::Value>> {
+    json::from_slice(data).unwrap_or_else(|_| {
+        let vec_value: Vec<json::Value> = json::from_slice(data).unwrap();
+        vec_value
+            .into_iter()
+            .filter_map(|v| match v {
+                json::Value::Object(map) => Some(map),
+                _ => None,
+            })
+            .collect()
+    })
+}
+
 /// The proto has only `status_code` + `message`, so `207` carries the partial-failure JSON.
 fn encode_metrics_reply(resp: &ingestion_common::IngestionResponse) -> IngestionResponse {
     if resp.code != 200 {
@@ -264,6 +267,36 @@ mod tests {
         resp
     }
 
+    #[test]
+    fn test_parse_enrichment_records_object_array() {
+        let records =
+            parse_enrichment_records(br#"[{"id": 1}, {"id": 2, "nested": {"ok": true}}]"#);
+        assert_eq!(
+            json::to_value(records).unwrap(),
+            json::json!([
+                {"id": 1}, {"id": 2, "nested": {"ok": true}}
+            ])
+        );
+        assert!(parse_enrichment_records(b"[]").is_empty());
+    }
+
+    #[test]
+    fn test_parse_enrichment_records_mixed_array() {
+        let records =
+            parse_enrichment_records(br#"[null, {"id": 1}, 2, "text", false, [], {"id": 2}]"#);
+        assert_eq!(
+            json::to_value(records).unwrap(),
+            json::json!([{"id": 1}, {"id": 2}])
+        );
+        assert!(parse_enrichment_records(b"[null, 1, false]").is_empty());
+    }
+
+    #[test]
+    #[should_panic]
+    fn test_parse_enrichment_records_rejects_non_array() {
+        parse_enrichment_records(br#"{"id": 1}"#);
+    }
+
     #[test]
     fn test_encode_metrics_reply_rejection_keeps_code() {
         let reply = encode_metrics_reply(&metrics_resp(503, 0, Some("blocked")));

src/api/management/src/request/ai/chat.rs

@@ -623,7 +623,8 @@ pub async fn chat_stream(Path(org_id): Path<String>, in_req: axum::extract::Requ
         user_id,
         forward_headers
             .get(X_O2_ASSISTANT_SESSION_ID.as_str())
-            .unwrap_or(&"none".to_string())
+            .map(String::as_str)
+            .unwrap_or("none")
     );
 
     // Parse JSON body
@@ -1123,7 +1124,7 @@ pub async fn confirm_action(
                         StatusCode::OK,
                         axum::Json(
                             serde_json::from_str::<serde_json::Value>(&response_body)
-                                .unwrap_or(serde_json::json!({"ok": true})),
+                                .unwrap_or_else(|_| serde_json::json!({"ok": true})),
                         ),
                     )
                         .into_response()

src/api/management/src/request/ai/toolsets.rs

@@ -245,7 +245,7 @@ pub async fn update(
         .data
         .as_ref()
         .map(|v| serde_json::to_string(v).unwrap())
-        .or(existing.data.clone());
+        .or_else(|| existing.data.clone());
 
     let updated = OrgToolset {
         id: existing.id.clone(),

src/api/management/src/request/organization/org.rs

@@ -92,7 +92,8 @@ pub async fn organizations(
 
     let limit = query
         .get("page_size")
-        .unwrap_or(&"100".to_string())
+        .map(String::as_str)
+        .unwrap_or("100")
         .parse::<i64>()
         .ok();
     let is_root_user = is_root_user(user_id);
@@ -190,7 +191,8 @@ pub async fn all_organizations(
     let mut org_names = HashSet::new();
     let limit = query
         .get("page_size")
-        .unwrap_or(&"100".to_string())
+        .map(String::as_str)
+        .unwrap_or("100")
         .parse::<i64>()
         .ok();
 

src/api/search/src/search/around.rs

@@ -146,8 +146,8 @@ pub(crate) async fn around(
             .unwrap();
 
     // search forward
-    let fw_sql =
-        check_or_add_order_by_timestamp(&around_sql, false).unwrap_or(around_sql.to_string());
+    let fw_sql = check_or_add_order_by_timestamp(&around_sql, false)
+        .unwrap_or_else(|_| around_sql.to_string());
     let fw_sql =
         config::utils::query_select_utils::replace_o2_custom_patterns(&fw_sql).unwrap_or(fw_sql);
 
@@ -187,8 +187,8 @@ pub(crate) async fn around(
         .await?;
 
     // search backward
-    let bw_sql =
-        check_or_add_order_by_timestamp(&around_sql, true).unwrap_or(around_sql.to_string());
+    let bw_sql = check_or_add_order_by_timestamp(&around_sql, true)
+        .unwrap_or_else(|_| around_sql.to_string());
     let bw_sql =
         config::utils::query_select_utils::replace_o2_custom_patterns(&bw_sql).unwrap_or(bw_sql);
     let req = config::meta::search::Request {

src/api/search/src/search/mod.rs

@@ -849,7 +849,7 @@ pub async fn build_search_request_per_field(
 
     let schema = infra::schema::get(org_id, stream_name, stream_type)
         .await
-        .unwrap_or(Schema::empty());
+        .unwrap_or_else(|_| Schema::empty());
     let fields = req
         .fields
         .iter()
@@ -1161,7 +1161,7 @@ async fn values_inner(
     // skip fields which aren't part of the schema
     let schema = infra::schema::get(org_id, stream_name, stream_type)
         .await
-        .unwrap_or(Schema::empty());
+        .unwrap_or_else(|_| Schema::empty());
 
     let mut query_results = Vec::with_capacity(fields.len());
     let sql_where = if where_str.is_empty() {

src/config/src/meta/alerts/mod.rs

@@ -147,9 +147,8 @@ impl TriggerCondition {
             _ => 0,
         };
         let start_utc = start_from.map_or(Ok(Utc::now()), |from| {
-            chrono::DateTime::<Utc>::from_timestamp_micros(from).ok_or(anyhow::anyhow!(
-                "Error converting start_from value to timestamp"
-            ))
+            chrono::DateTime::<Utc>::from_timestamp_micros(from)
+                .ok_or_else(|| anyhow::anyhow!("Error converting start_from value to timestamp"))
         })?;
 
         if self.frequency_type == FrequencyType::Cron {

src/config/src/meta/sql.rs

@@ -101,7 +101,7 @@ pub fn resolve_stream_names(sql: &str) -> Result<Vec<String>, anyhow::Error> {
     let dialect = &PostgreSqlDialect {};
     let statement = DFParser::parse_sql_with_dialect(sql, dialect)?
         .pop_back()
-        .ok_or(anyhow::anyhow!("Failed to parse sql"))?;
+        .ok_or_else(|| anyhow::anyhow!("Failed to parse sql"))?;
     let (table_refs, _) = resolve_table_references(&statement, true)?;
     let mut tables = Vec::new();
     for table in table_refs {
@@ -114,7 +114,7 @@ pub fn resolve_stream_names_with_type(sql: &str) -> Result<Vec<TableReference>,
     let dialect = &PostgreSqlDialect {};
     let statement = DFParser::parse_sql_with_dialect(sql, dialect)?
         .pop_back()
-        .ok_or(anyhow::anyhow!("Failed to parse sql"))?;
+        .ok_or_else(|| anyhow::anyhow!("Failed to parse sql"))?;
     let (table_refs, _) = resolve_table_references(&statement, true)?;
     let mut tables = Vec::new();
     for table in table_refs {

src/config/src/utils/download_utils.rs

@@ -26,23 +26,23 @@ pub async fn download_file(client: &Client, url: &str, path: &str) -> Result<(),
         .get(url)
         .send()
         .await
-        .or(Err(format!("Failed to GET from '{url}'")))?;
+        .map_err(|_| format!("Failed to GET from '{url}'"))?;
     let total_size = res
         .content_length()
-        .ok_or(format!("Failed to get content length from '{url}'"))?;
+        .ok_or_else(|| format!("Failed to get content length from '{url}'"))?;
 
     // download chunks
     let mut file = File::create(path)
         .await
-        .or(Err(format!("Failed to create file '{path}'")))?;
+        .map_err(|_| format!("Failed to create file '{path}'"))?;
     let mut downloaded: u64 = 0;
     let mut stream = res.bytes_stream();
 
     while let Some(item) = stream.next().await {
-        let chunk = item.or(Err("Error while downloading file".to_string()))?;
+        let chunk = item.map_err(|_| "Error while downloading file".to_string())?;
         file.write_all(&chunk)
             .await
-            .or(Err("Error while writing to file".to_string()))?;
+            .map_err(|_| "Error while writing to file".to_string())?;
         let new = min(downloaded + (chunk.len() as u64), total_size);
         downloaded = new;
     }

src/config/src/utils/sysinfo/os.rs

@@ -14,15 +14,15 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 pub fn get_os_name() -> String {
-    sysinfo::System::name().unwrap_or("unknown".to_string())
+    sysinfo::System::name().unwrap_or_else(|| "unknown".to_string())
 }
 
 pub fn get_os_version() -> String {
-    sysinfo::System::os_version().unwrap_or("unknown".to_string())
+    sysinfo::System::os_version().unwrap_or_else(|| "unknown".to_string())
 }
 
 pub fn get_hostname() -> String {
-    sysinfo::System::host_name().unwrap_or("unknown".to_string())
+    sysinfo::System::host_name().unwrap_or_else(|| "unknown".to_string())
 }
 
 #[cfg(target_os = "linux")]

src/core/src/alerts/alert.rs

@@ -2479,10 +2479,10 @@ impl AlertExt for Alert {
         if !self.workflows.is_empty() {
             let data: Vec<_> = rows.iter().map(|v| Value::Object(v.clone())).collect();
 
-            let source_id = self
-                .id
-                .as_ref()
-                .map_or(format!("{}/{}", self.org_id, self.name), |v| v.to_string());
+            let source_id = self.id.as_ref().map_or_else(
+                || format!("{}/{}", self.org_id, self.name),
+                |v| v.to_string(),
+            );
 
             let metadata: HashMap<String, Value> = vec![
                 ("org_id", self.org_id.clone().into()),

src/core/src/functions.rs

@@ -460,7 +460,8 @@ pub async fn delete_function(org_id: &str, fn_name: &str) -> Result<(), Function
     }
     let pipeline_dep = get_dependencies(org_id, fn_name).await;
     if !pipeline_dep.is_empty() {
-        let pipeline_data = serde_json::to_string(&pipeline_dep).unwrap_or("[]".to_string());
+        let pipeline_data =
+            serde_json::to_string(&pipeline_dep).unwrap_or_else(|_| "[]".to_string());
         return Err(FunctionDeleteError::PipelineDependencies(format!(
             "Warning: Function '{}' has {} pipeline dependencies. Please remove these pipelines first: {}",
             fn_name,

src/core/src/incidents.rs

@@ -29,7 +29,7 @@ async fn get_event_metadata(
 ) -> Result<HashMap<String, Value>, anyhow::Error> {
     let incident = get_incident_with_alerts(org_id, incident_id)
         .await?
-        .ok_or(anyhow::anyhow!("incident with given id not found"))?;
+        .ok_or_else(|| anyhow::anyhow!("incident with given id not found"))?;
 
     let mut metadata: Vec<(&str, Value)> = Vec::new();
 
@@ -97,7 +97,10 @@ async fn get_event_metadata(
 
         IncidentEventType::Resolved { user_id } => {
             metadata.push(("event_type", "resolved".into()));
-            metadata.push(("user_id", user_id.unwrap_or("system".to_string()).into()));
+            metadata.push((
+                "user_id",
+                user_id.unwrap_or_else(|| "system".to_string()).into(),
+            ));
         }
 
         IncidentEventType::Reopened { user_id, reason } => {
@@ -120,8 +123,14 @@ async fn get_event_metadata(
         }
         IncidentEventType::AssignmentChanged { from, to } => {
             metadata.push(("event_type", "assignment_changed".into()));
-            metadata.push(("from_user", from.unwrap_or("unknown".to_string()).into()));
-            metadata.push(("to_user", to.unwrap_or("unknown".to_string()).into()));
+            metadata.push((
+                "from_user",
+                from.unwrap_or_else(|| "unknown".to_string()).into(),
+            ));
+            metadata.push((
+                "to_user",
+                to.unwrap_or_else(|| "unknown".to_string()).into(),
+            ));
         }
 
         IncidentEventType::Comment { user_id, comment } => {

src/core/src/logs/ingest.rs

@@ -1045,7 +1045,7 @@ fn deserialize_aws_record_from_vec(data: Vec<u8>, request_id: &str) -> Result<Ve
                     value = json::to_value(event)?;
                     let local_val = value
                         .as_object_mut()
-                        .ok_or(anyhow::anyhow!("Error to convert Value to object"))?;
+                        .ok_or_else(|| anyhow::anyhow!("Error to convert Value to object"))?;
 
                     local_val.insert("requestId".to_owned(), request_id.into());
                     local_val.insert(
@@ -1093,22 +1093,21 @@ fn deserialize_aws_record_from_vec(data: Vec<u8>, request_id: &str) -> Result<Ve
                 let timestamp = kfh_metric_data.timestamp;
 
                 let mut parsed_metric_value = json::to_value(kfh_metric_data)?;
-                let local_parsed_metric_value = parsed_metric_value.as_object_mut().ok_or(
-                    anyhow::anyhow!("CloudWatch metrics failed to parse Metric Object"),
-                )?;
+                let local_parsed_metric_value =
+                    parsed_metric_value.as_object_mut().ok_or_else(|| {
+                        anyhow::anyhow!("CloudWatch metrics failed to parse Metric Object")
+                    })?;
 
-                for (value_name, value_val) in values.as_object().ok_or(anyhow::anyhow!(
-                    "CloudWatch metrics failed to Metric Value Object"
-                ))? {
+                for (value_name, value_val) in values.as_object().ok_or_else(|| {
+                    anyhow::anyhow!("CloudWatch metrics failed to Metric Value Object")
+                })? {
                     local_parsed_metric_value.insert(value_name.to_owned(), value_val.to_owned());
                 }
                 local_parsed_metric_value.remove("value");
 
                 let metric_dimensions = dimensions
                     .as_object()
-                    .ok_or(anyhow::anyhow!(
-                        "CloudWatch metrics dimensions parsing failed"
-                    ))?
+                    .ok_or_else(|| anyhow::anyhow!("CloudWatch metrics dimensions parsing failed"))?
                     .iter()
                     .map(|(k, v)| format!("{k}=[{v}]"))
                     .collect::<Vec<_>>()

src/core/src/logs/otlp.rs

@@ -148,7 +148,7 @@ pub async fn handle_request(
     // check stream
     let stream_name = in_stream_name
         .map(|name| format_stream_name(name.to_string()))
-        .unwrap_or("default".to_string());
+        .unwrap_or_else(|| "default".to_string());
     check_ingestion_allowed(org_id, StreamType::Logs, Some(&stream_name)).await?;
 
     let cfg = get_config();

src/core/src/metrics/json.rs

@@ -112,7 +112,10 @@ async fn buffer_record(
     };
     let stream_name = match stream_name {
         Some(name) => name.to_string(),
-        None => match record.get(NAME_LABEL).ok_or(anyhow!("missing __name__"))? {
+        None => match record
+            .get(NAME_LABEL)
+            .ok_or_else(|| anyhow!("missing __name__"))?
+        {
             json::Value::String(s) => format_stream_name(s.to_string()),
             _ => {
                 return Err(anyhow::anyhow!("invalid __name__, need to be string"));
@@ -241,7 +244,11 @@ fn finish_rows(
     let first_type = records.first().map(|(_, metric_type)| *metric_type);
     let mut rows = Vec::with_capacity(records.len());
     for (mut record, _) in records {
-        let value = parse_metric_value(record.get(VALUE_LABEL).ok_or(anyhow!("missing value"))?)?;
+        let value = parse_metric_value(
+            record
+                .get(VALUE_LABEL)
+                .ok_or_else(|| anyhow!("missing value"))?,
+        )?;
         if let Some(existing) = record.get_mut(VALUE_LABEL) {
             *existing = value;
         }

src/core/src/metrics/prom.rs

@@ -148,7 +148,7 @@ pub async fn remote_write(
         let metric_name = format_stream_name(item.metric_family_name.to_string());
         let schema = infra::schema::get(org_id, &metric_name, StreamType::Metrics)
             .await
-            .unwrap_or(Schema::empty());
+            .unwrap_or_else(|_| Schema::empty());
         if schema.metadata().contains_key(METADATA_LABEL) {
             // already has metadata, skip
             continue;

src/core/src/pipeline/batch_execution.rs

@@ -1126,7 +1126,7 @@ impl ExecutablePipeline {
             .filter_map(|exec_node| {
                 if exec_node.children.is_empty() {
                     if let NodeData::Stream(stream_params) = &exec_node.node_data {
-                        (!stream_params.stream_name.contains("{")).then_some(stream_params.clone())
+                        (!stream_params.stream_name.contains("{")).then(|| stream_params.clone())
                     } else {
                         None
                     }
@@ -1966,7 +1966,7 @@ async fn process_function_node(
     let stream_name = metadata
         .stream_name
         .clone()
-        .unwrap_or("pipeline".to_string());
+        .unwrap_or_else(|| "pipeline".to_string());
     let mut result_array_records = Vec::new();
     while let Some(pipeline_item) = channels.receiver.recv().await {
         channels

src/core/src/self_reporting/search.rs

@@ -104,7 +104,7 @@ pub async fn get_usage(
 pub async fn get_license_usage_data_from_node(node_id: String) -> Result<GetLicenseUsageResponse> {
     let node = get_node_by_uuid(&node_id)
         .await
-        .ok_or(anyhow::anyhow!("node {node_id} not found"))?;
+        .ok_or_else(|| anyhow::anyhow!("node {node_id} not found"))?;
     let node = std::sync::Arc::new(node) as std::sync::Arc<dyn NodeInfo>;
 
     let task = tokio::task::spawn(async move {

src/core/src/slo/job.rs

@@ -416,7 +416,7 @@ pub fn promql_rows(
                 let slice_start = t_micros / 1_000_000 - slice_interval_secs;
                 let e = acc
                     .entry((slice_start, key.clone()))
-                    .or_insert((0.0, labels.clone()));
+                    .or_insert_with(|| (0.0, labels.clone()));
                 e.0 += value;
             }
         }

src/core/src/traces/inferred.rs

@@ -186,18 +186,20 @@ where
             .filter(|host| !host.is_empty() && !is_ip_address(host))
     };
 
-    let fallback_host = host_attr("server.address")
-        .or_else(|| host_attr("net.peer.name"))
-        .or_else(|| host_attr("http.host"))
-        .or_else(|| {
-            attr("url.full")
-                .or_else(|| attr("http.url"))
-                .and_then(|u| url::Url::parse(&u).ok())
-                .and_then(|u| match u.host() {
-                    Some(url::Host::Domain(d)) => Some(d.to_string()),
-                    _ => None, // IP hosts are redacted, same as host_attr
-                })
-        });
+    let fallback_host = || {
+        host_attr("server.address")
+            .or_else(|| host_attr("net.peer.name"))
+            .or_else(|| host_attr("http.host"))
+            .or_else(|| {
+                attr("url.full")
+                    .or_else(|| attr("http.url"))
+                    .and_then(|u| url::Url::parse(&u).ok())
+                    .and_then(|u| match u.host() {
+                        Some(url::Host::Domain(d)) => Some(d.to_string()),
+                        _ => None, // IP hosts are redacted, same as host_attr
+                    })
+            })
+    };
 
     // Explicit peer.service overrides every naming rule.
     let peer_service = attr("peer.service");
@@ -208,7 +210,7 @@ where
     if db_system.is_some() || db_name.is_some() {
         let name = peer_service
             .or(db_name)
-            .or(fallback_host)
+            .or_else(fallback_host)
             .or_else(|| db_system.clone())?;
         return Some(InferredService {
             name,
@@ -223,7 +225,7 @@ where
     if messaging_system.is_some() || destination.is_some() {
         let name = peer_service
             .or(destination)
-            .or(fallback_host)
+            .or_else(fallback_host)
             .or_else(|| messaging_system.clone())?;
         return Some(InferredService {
             name,
@@ -238,7 +240,7 @@ where
     if rpc_system.is_some() || rpc_service.is_some() {
         let name = peer_service
             .or(rpc_service)
-            .or(fallback_host)
+            .or_else(fallback_host)
             .or_else(|| rpc_system.clone())?;
         return Some(InferredService {
             name,
@@ -248,7 +250,7 @@ where
     }
 
     // external (http or generic network peer)
-    let name = peer_service.or(fallback_host)?;
+    let name = peer_service.or_else(fallback_host)?;
     let is_http = attr("http.request.method")
         .or_else(|| attr("http.method"))
         .or_else(|| attr("url.full"))
@@ -450,6 +452,116 @@ mod tests {
         peer_keys(SPAN_KIND_CLIENT, attrs).and_then(|keys| keys.key)
     }
 
+    #[test]
+    fn test_explicit_names_skip_host_lookup_and_preserve_classification() {
+        for (system_key, system, name_key, service_type) in [
+            (
+                "db.system",
+                "postgresql",
+                "db.namespace",
+                INFER_TYPE_DATABASE,
+            ),
+            (
+                "messaging.system",
+                "kafka",
+                "messaging.destination.name",
+                INFER_TYPE_QUEUE,
+            ),
+            ("rpc.system", "grpc", "rpc.service", INFER_TYPE_RPC),
+        ] {
+            for explicit_peer in [false, true] {
+                let result = derive_inferred_service(SPAN_KIND_CLIENT, |key| {
+                    assert!(
+                        !matches!(
+                            key,
+                            "server.address"
+                                | "net.peer.name"
+                                | "http.host"
+                                | "url.full"
+                                | "http.url"
+                        ),
+                        "unexpected fallback lookup: {key}"
+                    );
+                    if key == system_key {
+                        Some(system.to_string())
+                    } else if key == name_key {
+                        Some("named-dependency".to_string())
+                    } else if explicit_peer && key == "peer.service" {
+                        Some("explicit-peer".to_string())
+                    } else {
+                        None
+                    }
+                })
+                .unwrap();
+                assert_eq!(
+                    result.name,
+                    if explicit_peer {
+                        "explicit-peer"
+                    } else {
+                        "named-dependency"
+                    }
+                );
+                assert_eq!(result.service_type, service_type);
+                assert_eq!(result.system.as_deref(), Some(system));
+            }
+        }
+    }
+
+    #[test]
+    fn test_external_peer_skips_host_but_keeps_http_classification() {
+        let result = derive_inferred_service(SPAN_KIND_CLIENT, |key| {
+            assert!(
+                !matches!(key, "server.address" | "net.peer.name" | "http.host"),
+                "unexpected host lookup: {key}"
+            );
+            match key {
+                "peer.service" => Some("payments".to_string()),
+                "url.full" => Some("https://example.com/payments".to_string()),
+                _ => None,
+            }
+        })
+        .unwrap();
+        assert_eq!(result.name, "payments");
+        assert_eq!(result.service_type, INFER_TYPE_EXTERNAL);
+        assert_eq!(result.system.as_deref(), Some("http"));
+    }
+
+    #[test]
+    fn test_host_fallback_preserves_precedence() {
+        for (attrs, name) in [
+            (
+                vec![
+                    ("server.address", "primary:443"),
+                    ("net.peer.name", "secondary"),
+                    ("url.full", "https://url-host/path"),
+                ],
+                "primary",
+            ),
+            (
+                vec![
+                    ("server.address", "127.0.0.1"),
+                    ("net.peer.name", "secondary:80"),
+                    ("url.full", "https://url-host/path"),
+                ],
+                "secondary",
+            ),
+            (
+                vec![
+                    ("server.address", "127.0.0.1"),
+                    ("url.full", "https://url-host/path"),
+                ],
+                "url-host",
+            ),
+        ] {
+            let mut attrs = attrs;
+            attrs.push(("db.system", "postgresql"));
+            let result = derive(SPAN_KIND_CLIENT, &attrs).unwrap();
+            assert_eq!(result.name, name);
+            assert_eq!(result.service_type, INFER_TYPE_DATABASE);
+            assert_eq!(result.system.as_deref(), Some("postgresql"));
+        }
+    }
+
     #[test]
     fn test_database_with_db_name() {
         let result = derive(

src/core/src/users.rs

@@ -1222,7 +1222,7 @@ pub async fn list_user_invites(user_id: &str, only_pending: bool) -> Result<Resp
                     org_name: db::organization::get_org(&invite.org_id)
                         .await
                         .map(|org| org.name)
-                        .unwrap_or("default".to_string()),
+                        .unwrap_or_else(|_| "default".to_string()),
                     role: invite.role,
                     org_id: invite.org_id,
                     token: invite.token,

src/core/src/workflows/mod.rs

@@ -895,7 +895,7 @@ async fn execute_workflow(
 ) -> Result<WorkflowExecutionStatus, anyhow::Error> {
     let workflow = get_workflow_by_id(org_id, id)
         .await?
-        .ok_or(anyhow::anyhow!("workflow with given id not found"))?;
+        .ok_or_else(|| anyhow::anyhow!("workflow with given id not found"))?;
 
     if !workflow.enabled {
         return Ok(WorkflowExecutionStatus::Success);
@@ -942,7 +942,7 @@ pub async fn retry_run(
 ) -> Result<WorkflowResult, anyhow::Error> {
     let workflow = workflows::get_by_org_wid(org_id, wid)
         .await?
-        .ok_or(anyhow::anyhow!("workflow with given id not found"))?;
+        .ok_or_else(|| anyhow::anyhow!("workflow with given id not found"))?;
 
     let mut start_id = None;
     for node in &workflow.nodes {
@@ -992,9 +992,11 @@ pub async fn retry_run(
 
     let node_id = from_node.as_ref().unwrap_or(&start_id);
 
-    let inputs = ip_map.input_map.remove(node_id).ok_or(anyhow::anyhow!(
-        "node id {node_id} does not have any associated input data in the stored inputs"
-    ))?;
+    let inputs = ip_map.input_map.remove(node_id).ok_or_else(|| {
+        anyhow::anyhow!(
+            "node id {node_id} does not have any associated input data in the stored inputs"
+        )
+    })?;
 
     let start_time = chrono::Utc::now().timestamp_micros();
     let res = executable

src/db/src/alerts/alert.rs

@@ -582,7 +582,7 @@ pub fn cache_alert_key(org: &str, alert_id: &str) -> String {
 
 /// Returns the key used to schedule a trigger for the alert.
 pub fn scheduler_key(alert_id: Option<Ksuid>) -> String {
-    alert_id.map_or(DEFAULT_FOLDER.to_string(), |id| id.to_string())
+    alert_id.map_or_else(|| DEFAULT_FOLDER.to_string(), |id| id.to_string())
 }
 
 #[cfg(test)]

src/db/src/alerts/destinations.rs

@@ -234,7 +234,7 @@ pub(super) fn parse_event_key<'a>(
 ) -> Result<(&'a str, &'a str), anyhow::Error> {
     let item_key = event_key
         .strip_prefix(prefix)
-        .ok_or(anyhow::anyhow!("event key missing prefix"))?;
+        .ok_or_else(|| anyhow::anyhow!("event key missing prefix"))?;
     let mut keys = item_key.split('/');
     let org_id = keys
         .next()

src/enrichment_data/src/enrichment_table/mod.rs

@@ -177,7 +177,7 @@ pub async fn save_enrichment_data(
     let db_schema = stream_schema_map
         .get(&stream_name)
         .map(|s| s.schema().as_ref().clone())
-        .unwrap_or(Schema::empty());
+        .unwrap_or_else(Schema::empty);
     if !db_schema.fields().is_empty() && db_schema.fields().ne(inferred_schema.fields()) {
         log::error!("Schema mismatch for enrichment table {org_id}/{stream_name}");
         return Ok(MetaHttpResponse::error_with_header(

src/enrichment_data/src/enrichment_table/url_processor.rs

@@ -1665,7 +1665,7 @@ async fn process_enrichment_table_url(
     let schema = stream_schema_map
         .get(&stream_name)
         .map(|s| s.schema().as_ref().clone())
-        .unwrap_or(arrow_schema::Schema::empty());
+        .unwrap_or_else(arrow_schema::Schema::empty);
 
     // Calculate total expected size
     let current_size_in_bytes = if append_data {
@@ -1844,7 +1844,7 @@ async fn save_enrichment_batch(
         let db_schema = stream_schema_map
             .get(&stream_name)
             .map(|s| s.schema().as_ref().clone())
-            .unwrap_or(arrow_schema::Schema::empty());
+            .unwrap_or_else(arrow_schema::Schema::empty);
 
         if !db_schema.fields().is_empty() && db_schema.fields().ne(inferred_schema.fields()) {
             return Err(anyhow!(

src/infra/src/coordinator/alerts.rs

@@ -35,7 +35,7 @@ pub async fn emit_put_event(
     cluster_coordinator
         .put(
             &key,
-            bytes::Bytes::from(folder_id.unwrap_or("default".to_string())),
+            bytes::Bytes::from(folder_id.unwrap_or_else(|| "default".to_string())),
             true,
             None,
         )

src/infra/src/table/ratelimit.rs

@@ -165,7 +165,7 @@ async fn add_batch(rules: Vec<RatelimitRule>) -> Result<(), anyhow::Error> {
             rule_id: Set(rule.rule_id.unwrap()),
             rule_type: Set(rule
                 .rule_type
-                .unwrap_or(RatelimitRuleType::Exact.to_string())),
+                .unwrap_or_else(|| RatelimitRuleType::Exact.to_string())),
             user_role: Set(rule.user_role.unwrap_or_default()),
             user_id: Set(rule.user_id.unwrap_or_default()),
             api_group_name: Set(rule.api_group_name.unwrap_or_default()),
@@ -245,7 +245,7 @@ async fn add_upsert_batch(rules: Vec<RatelimitRule>) -> Result<(), anyhow::Error
                 active_model.rule_id = Set(existing.rule_id.clone());
                 active_model.rule_type = Set(rule
                     .rule_type
-                    .unwrap_or(RatelimitRuleType::Exact.to_string()));
+                    .unwrap_or_else(|| RatelimitRuleType::Exact.to_string()));
                 active_model.user_role = Set(rule.user_role.unwrap_or_default());
                 active_model.user_id = Set(rule.user_id.unwrap_or_default());
                 active_model.api_group_name = Set(rule.api_group_name.unwrap_or_default());
@@ -278,7 +278,7 @@ async fn add_upsert_batch(rules: Vec<RatelimitRule>) -> Result<(), anyhow::Error
                         rule_id: Set(rule.rule_id.unwrap()),
                         rule_type: Set(rule
                             .rule_type
-                            .unwrap_or(RatelimitRuleType::Exact.to_string())),
+                            .unwrap_or_else(|| RatelimitRuleType::Exact.to_string())),
                         user_role: Set(rule.user_role.unwrap_or_default()),
                         user_id: Set(rule.user_id.unwrap_or_default()),
                         api_group_name: Set(rule.api_group_name.unwrap_or_default()),
@@ -322,7 +322,7 @@ async fn add_single(rule: RatelimitRule) -> Result<(), anyhow::Error> {
                 rule_id: Set(rule.rule_id.unwrap()),
                 rule_type: Set(rule
                     .rule_type
-                    .unwrap_or(RatelimitRuleType::Exact.to_string())),
+                    .unwrap_or_else(|| RatelimitRuleType::Exact.to_string())),
                 user_role: Set(rule.user_role.unwrap_or("".to_string())),
                 user_id: Set(rule.user_id.unwrap_or_default()),
                 api_group_name: Set(rule.api_group_name.unwrap_or_default()),
@@ -372,7 +372,7 @@ async fn update_single(rule: RatelimitRule) -> Result<(), anyhow::Error> {
                     Column::RuleType,
                     Expr::value(
                         rule.rule_type
-                            .unwrap_or(RatelimitRuleType::Exact.to_string()),
+                            .unwrap_or_else(|| RatelimitRuleType::Exact.to_string()),
                     ),
                 )
                 .col_expr(Column::Threshold, Expr::value(rule.threshold))
@@ -460,7 +460,7 @@ async fn update_batch(rules: Vec<RatelimitRule>) -> Result<(), anyhow::Error> {
                 Column::RuleType,
                 Expr::value(
                     rule.rule_type
-                        .unwrap_or(RatelimitRuleType::Exact.to_string()),
+                        .unwrap_or_else(|| RatelimitRuleType::Exact.to_string()),
                 ),
             )
             .col_expr(Column::Threshold, Expr::value(rule.threshold))

src/metrics_index/src/pruner.rs

@@ -75,7 +75,9 @@ pub async fn search(
     let mut index_files = BTreeMap::new();
     for file in files.iter() {
         // only indexed metrics files own a sidecar; other layouts stay as they are
-        if MetricsFileLayout::of(&file.key) != Some(MetricsFileLayout::Indexed) {
+        if index_files.contains_key(&file.key)
+            || MetricsFileLayout::of(&file.key) != Some(MetricsFileLayout::Indexed)
+        {
             continue;
         }
         let Some(sidecar_path) = MetricsFileLayout::metrics_index_path(&file.key) else {
@@ -104,12 +106,10 @@ pub async fn search(
         } else {
             String::new()
         };
-        index_files.entry(file.key.clone()).or_insert((
-            file.account.clone(),
-            sidecar_path,
-            cache_key,
-            expected_rows,
-        ));
+        index_files.insert(
+            file.key.clone(),
+            (file.account.clone(), sidecar_path, cache_key, expected_rows),
+        );
     }
     if index_files.is_empty() {
         return Ok(None);

src/search/src/tantivy/mod.rs

@@ -470,7 +470,7 @@ async fn search_tantivy_index(
 
     // generate the tantivy query
     let condition: IndexCondition =
-        index_condition.ok_or(anyhow::anyhow!("IndexCondition not found"))?;
+        index_condition.ok_or_else(|| anyhow::anyhow!("IndexCondition not found"))?;
     let (mut query, has_skipped_conditions) =
         condition.to_tantivy_query(trace_id, tantivy_schema.clone(), fts_field)?;
 

src/search_service/src/cache/mod.rs

@@ -1131,10 +1131,7 @@ pub fn apply_vrl_to_response(
                         .as_array()
                         .unwrap()
                         .iter()
-                        .filter_map(|v| {
-                            (!v.is_null())
-                                .then_some(config::utils::flatten::flatten(v.clone()).unwrap())
-                        })
+                        .filter_map(|v| super::flatten_vrl_result(v.clone()))
                         .collect()
                 } else {
                     let mut error = "".to_string();
@@ -1155,8 +1152,7 @@ pub fn apply_vrl_to_response(
                             if let Some(e) = err {
                                 error = e;
                             }
-                            (!ret_val.is_null())
-                                .then_some(config::utils::flatten::flatten(ret_val).unwrap())
+                            super::flatten_vrl_result(ret_val)
                         })
                         .collect();
                     if !error.is_empty() {
@@ -1267,6 +1263,61 @@ pub async fn apply_regex_to_response(
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_apply_vrl_to_response_preserves_hits_on_compile_failure() {
+        let mut response = config::meta::search::Response {
+            hits: vec![json::json!({"nested": {"value": 42}})],
+            ..Default::default()
+        };
+        let expected = response.hits.clone();
+        let hits = apply_vrl_to_response(
+            Some(". = [".to_string()),
+            &mut response,
+            "default",
+            "test",
+            "test",
+        );
+        assert_eq!(hits, expected);
+    }
+
+    #[test]
+    fn test_apply_vrl_to_response_filters_null_per_record() {
+        let mut response = config::meta::search::Response {
+            hits: vec![
+                json::json!({"drop": true}),
+                json::json!({"drop": false, "nested": {"value": 42}}),
+            ],
+            ..Default::default()
+        };
+        let hits = apply_vrl_to_response(
+            Some("if .drop == true { . = null } else { del(.drop) }; .".to_string()),
+            &mut response,
+            "default",
+            "test",
+            "test",
+        );
+        assert_eq!(hits, vec![json::json!({"nested_value": 42})]);
+    }
+
+    #[test]
+    fn test_apply_vrl_to_response_filters_null_array_items() {
+        let mut response = config::meta::search::Response {
+            hits: vec![json::json!({"input": true})],
+            ..Default::default()
+        };
+        let hits = apply_vrl_to_response(
+            Some(
+                "#ResultArray#SkipVRL#\n. = [null, {\"nested\": {\"value\": 42}}, null]"
+                    .to_string(),
+            ),
+            &mut response,
+            "default",
+            "test",
+            "test",
+        );
+        assert_eq!(hits, vec![json::json!({"nested_value": 42})]);
+    }
+
     #[test]
     fn test_is_result_array_skip_vrl() {
         let query_fn = r#"#ResultArray#SkipVRL#

src/search_service/src/cluster/http.rs

@@ -144,7 +144,7 @@ pub async fn search(
 
                         ret_val
                             .as_array()
-                            .ok_or(Error::Message("Expected array".to_string()))?
+                            .ok_or_else(|| Error::Message("Expected array".to_string()))?
                             .iter()
                             .filter_map(|v| {
                                 if !v.is_null() && v.is_object() {

src/search_service/src/file_list.rs

@@ -180,7 +180,8 @@ pub async fn query_by_ids(
         db_files.extend(
             dumped_files
                 .iter()
-                .filter_map(|file| ids_set.contains(&file.id).then_some(file.into())),
+                .filter(|file| ids_set.contains(&file.id))
+                .map(|file| file.into()),
         );
     }
 

src/search_service/src/grpc/flight.rs

@@ -150,7 +150,7 @@ pub async fn search(
 
     let db_schema = infra::schema::get(&org_id, &stream_name, stream_type)
         .await
-        .unwrap_or(arrow_schema::Schema::empty());
+        .unwrap_or_else(|_| arrow_schema::Schema::empty());
     let stream_settings = unwrap_stream_settings(&db_schema);
     let stream_created_at = unwrap_stream_created_at(&db_schema);
     let fst_fields = get_stream_setting_fts_fields(&stream_settings)
@@ -178,11 +178,8 @@ pub async fn search(
         .index_info
         .equal_keys
         .iter()
-        .filter_map(|v| {
-            latest_schema_map
-                .contains_key(&v.key)
-                .then_some((v.key.to_string(), v.value.to_string()))
-        })
+        .filter(|v| latest_schema_map.contains_key(&v.key))
+        .map(|v| (v.key.to_string(), v.value.to_string()))
         .collect::<Vec<_>>();
 
     // get all tables

src/search_service/src/lib.rs

@@ -511,8 +511,7 @@ pub async fn search_multi(
                                     .collect::<Vec<_>>();
                                 Some(serde_json::Value::Array(flattened_array))
                             } else {
-                                (!v.is_null())
-                                    .then_some(config::utils::flatten::flatten(v.clone()).unwrap())
+                                flatten_vrl_result(v.clone())
                             }
                         })
                         .collect()
@@ -535,8 +534,7 @@ pub async fn search_multi(
                             if let Some(e) = err {
                                 error = e;
                             }
-                            (!ret_val.is_null())
-                                .then_some(config::utils::flatten::flatten(ret_val).unwrap())
+                            flatten_vrl_result(ret_val)
                         })
                         .collect();
                     if !error.is_empty() {
@@ -1089,6 +1087,10 @@ pub fn check_search_allowed(_org_id: &str, _stream: Option<&str>) -> Result<(),
     Ok(())
 }
 
+fn flatten_vrl_result(value: json::Value) -> Option<json::Value> {
+    (!value.is_null()).then(|| config::utils::flatten::flatten(value).unwrap())
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -1098,6 +1100,27 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_flatten_vrl_result_filters_null_and_preserves_nested_records() {
+        assert_eq!(flatten_vrl_result(json::Value::Null), None);
+        assert_eq!(
+            flatten_vrl_result(json::json!({"nested": {"value": 42}})),
+            Some(json::json!({"nested_value": 42}))
+        );
+        let values = vec![
+            json::Value::Null,
+            json::json!({"nested": {"value": 42}}),
+            json::Value::Null,
+        ];
+        assert_eq!(
+            values
+                .into_iter()
+                .filter_map(flatten_vrl_result)
+                .collect::<Vec<_>>(),
+            vec![json::json!({"nested_value": 42})]
+        );
+    }
+
     #[test]
     fn test_generate_filter_from_equal_items_empty() {
         let result = generate_filter_from_equal_items(&[]);

src/search_service/src/partition/sql_context.rs

@@ -64,7 +64,8 @@ impl PartitionSqlContext {
         let ts_column = get_ts_col_order_by(&sql, TIMESTAMP_COL_NAME, is_complex).map(|(v, _)| v);
         let is_streaming_agg = is_streaming_aggregate(&req.sql, ts_column.as_deref());
         let apply_over_hits = req.query_fn.as_ref().is_some_and(|v| {
-            !v.is_empty() && RESULT_ARRAY.is_match(&base64::decode_url(v).unwrap_or(v.to_string()))
+            !v.is_empty()
+                && RESULT_ARRAY.is_match(&base64::decode_url(v).unwrap_or_else(|_| v.to_string()))
         });
 
         let use_single_partition = is_explain

src/stream/src/lib.rs

@@ -176,7 +176,7 @@ pub fn stream_res(
     stats.created_at = unwrap_stream_created_at(&schema).unwrap_or_default();
 
     let metrics_meta = if stream_type == StreamType::Metrics {
-        let mut meta = get_prom_metadata_from_schema(&schema).unwrap_or(promql::Metadata {
+        let mut meta = get_prom_metadata_from_schema(&schema).unwrap_or_else(|| promql::Metadata {
             metric_type: promql::MetricType::Empty,
             metric_family_name: stream_name.to_string(),
             help: stream_name.to_string(),

src/super_cluster_queue/src/pipelines.rs

@@ -25,7 +25,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> {
         MessageType::PipelinePut => {
             let bytes = msg
                 .value
-                .ok_or(Error::Message("Message missing value".to_string()))?;
+                .ok_or_else(|| Error::Message("Message missing value".to_string()))?;
             let pipeline: Pipeline = json::from_slice(&bytes).inspect_err(|e| {
                 log::error!(
                     "[SUPER_CLUSTER:PIPELINE] Failed to deserialize message value to pipeline: {e}"
@@ -42,7 +42,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> {
         MessageType::EnrichmentUrlPut => {
             let bytes = msg
                 .value
-                .ok_or(Error::Message("Message missing value".to_string()))?;
+                .ok_or_else(|| Error::Message("Message missing value".to_string()))?;
             let mut enrichment_url_job: enrichment_table_urls::EnrichmentTableUrlRecord = json::from_slice(&bytes).inspect_err(|e| {
                 log::error!(
                     "[SUPER_CLUSTER:PIPELINE] Failed to deserialize message value to pipeline: {e}"
@@ -64,7 +64,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> {
             // format is /{BACKFILL_JOBS_KEY}/org/job_id
             let bytes = msg
                 .value
-                .ok_or(Error::Message("Message missing value".to_string()))?;
+                .ok_or_else(|| Error::Message("Message missing value".to_string()))?;
             let backfill_job: backfill_jobs::BackfillJob = json::from_slice(&bytes).inspect_err(|e| {
                 log::error!(
                     "[SUPER_CLUSTER:BACKFILL] Failed to deserialize message value to backfill job: {e}"
@@ -76,7 +76,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> {
             // format is /{BACKFILL_JOBS_KEY}/org/job_id
             let bytes = msg
                 .value
-                .ok_or(Error::Message("Message missing value".to_string()))?;
+                .ok_or_else(|| Error::Message("Message missing value".to_string()))?;
             let backfill_job: backfill_jobs::BackfillJob = json::from_slice(&bytes).inspect_err(|e| {
                 log::error!(
                     "[SUPER_CLUSTER:BACKFILL] Failed to deserialize message value to backfill job: {e}"

src/super_cluster_queue/src/ratelimit.rs

@@ -28,7 +28,7 @@ pub async fn process(msg: Message) -> Result<()> {
 
     let bytes = msg
         .value
-        .ok_or(Error::Message("Message missing value".to_string()))?;
+        .ok_or_else(|| Error::Message("Message missing value".to_string()))?;
     let rule = RuleEntry::try_from(&bytes).map_err(|e| Error::Message(e.to_string()))?;
 
     match msg.message_type {

src/synthetics/src/status_pages.rs

@@ -318,7 +318,7 @@ async fn apply_engine_actions(
             // Latest resolution wins for the merge window.
             let entry = recently_resolved
                 .entry(check_id)
-                .or_insert((n.id.clone(), resolved_at));
+                .or_insert_with(|| (n.id.clone(), resolved_at));
             if resolved_at > entry.1 {
                 *entry = (n.id.clone(), resolved_at);
             }

src/usage_reporting/src/lib.rs

@@ -242,7 +242,7 @@ pub async fn report_request_usage_stats(
     return;
 
     let now = DateTime::from_timestamp_micros(timestamp).unwrap();
-    let request_body = stats.request_body.unwrap_or(usage_type.to_string());
+    let request_body = stats.request_body.unwrap_or_else(|| usage_type.to_string());
     let user_email = stats.user_email.unwrap_or_default();
     let mut usages = Vec::with_capacity(if num_functions > 0 { 2 } else { 1 });