diff --git a/src/api_query/src/search/agent_format.rs b/src/api_query/src/search/agent_format.rs new file mode 100644 index 0000000000..8e4f123453 --- /dev/null +++ b/src/api_query/src/search/agent_format.rs @@ -0,0 +1,320 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +//! Reformat search `hits` into a compact string block for agent consumers +//! (`agent_options.output_format`). Runs at the serialization edge only: +//! result cache and the engine always see the original JSON hits. + +use config::{ + TIMESTAMP_COL_NAME, + meta::search::{OutputFormat, Response}, + utils::json, +}; + +/// A `select *` on a wide stream yields hits where most cells are absent; +/// per-row keys (ndjson) are then cheaper and less error-prone than a mostly +/// empty grid, so we fall back automatically (announced via `advisory`). +const SPARSE_FALLBACK_MIN_COLUMNS: usize = 20; +const SPARSE_FALLBACK_EMPTY_RATIO: f64 = 0.5; + +pub fn apply_output_format(res: &mut Response, format: OutputFormat) { + if format == OutputFormat::Json { + return; + } + + let format_name = match format { + OutputFormat::Csv => "csv", + OutputFormat::MdTable => "md_table", + OutputFormat::Json => unreachable!(), + }; + + if res.hits.is_empty() { + res.format = Some(format_name.to_string()); + res.data = Some(String::new()); + return; + } + + // Hits are expected to be objects; anything else (scalar rows from exotic + // queries) can't be laid out as a table, keep them as ndjson. + if !res.hits.iter().all(|hit| hit.is_object()) { + fallback_to_ndjson(res, "result rows are not flat objects"); + return; + } + + let columns = collect_columns(res); + + let total_cells = columns.len() * res.hits.len(); + let empty_cells: usize = res + .hits + .iter() + .map(|hit| { + let obj = hit.as_object().unwrap(); + columns + .iter() + .filter(|col| matches!(obj.get(*col), None | Some(json::Value::Null))) + .count() + }) + .sum(); + if columns.len() >= SPARSE_FALLBACK_MIN_COLUMNS + && (empty_cells as f64) > (total_cells as f64) * SPARSE_FALLBACK_EMPTY_RATIO + { + let reason = format!( + "result is sparse ({} columns, {}% empty cells)", + columns.len(), + empty_cells * 100 / total_cells + ); + fallback_to_ndjson(res, &reason); + return; + } + + let data = match format { + OutputFormat::Csv => render_csv(&columns, &res.hits), + OutputFormat::MdTable => render_md_table(&columns, &res.hits), + OutputFormat::Json => unreachable!(), + }; + + res.format = Some(format_name.to_string()); + res.data = Some(data); + res.columns = columns; + res.hits.clear(); +} + +/// Column order: server-provided `columns` first (if any), then remaining +/// keys in first-seen order, with `_timestamp` promoted to the front. +fn collect_columns(res: &Response) -> Vec { + let derive_order = res.columns.is_empty(); + let mut columns = res.columns.clone(); + let mut seen: hashbrown::HashSet = columns.iter().cloned().collect(); + for hit in &res.hits { + for key in hit.as_object().unwrap().keys() { + if seen.insert(key.clone()) { + columns.push(key.clone()); + } + } + } + if derive_order + && let Some(pos) = columns.iter().position(|c| c == TIMESTAMP_COL_NAME) + && pos > 0 + { + let ts = columns.remove(pos); + columns.insert(0, ts); + } + columns +} + +fn fallback_to_ndjson(res: &mut Response, reason: &str) { + let data = res + .hits + .iter() + .map(|hit| hit.to_string()) + .collect::>() + .join("\n"); + res.format = Some("ndjson".to_string()); + res.data = Some(data); + res.advisory = Some(format!("{reason}; returned as ndjson instead")); + res.hits.clear(); +} + +/// Stringify one cell. Nested objects/arrays become minified JSON. +fn cell_value(hit: &json::Value, column: &str) -> String { + match hit.get(column) { + None | Some(json::Value::Null) => String::new(), + Some(json::Value::String(s)) => s.clone(), + Some(v) => v.to_string(), + } +} + +/// Newlines become a literal `\n` (never RFC 4180 multi-line quoted cells: +/// models mis-parse records that span lines); commas/quotes still get the +/// standard quoting so the row stays machine-parsable. +fn csv_escape(value: &str) -> String { + let value = escape_newlines(value); + if value.contains(',') || value.contains('"') { + format!("\"{}\"", value.replace('"', "\"\"")) + } else { + value + } +} + +fn escape_newlines(value: &str) -> String { + if value.contains(['\n', '\r']) { + value.replace("\r\n", "\\n").replace(['\n', '\r'], "\\n") + } else { + value.to_string() + } +} + +fn render_csv(columns: &[String], hits: &[json::Value]) -> String { + let mut out = String::new(); + out.push_str( + &columns + .iter() + .map(|c| csv_escape(c)) + .collect::>() + .join(","), + ); + for hit in hits { + out.push('\n'); + out.push_str( + &columns + .iter() + .map(|c| csv_escape(&cell_value(hit, c))) + .collect::>() + .join(","), + ); + } + out +} + +fn md_escape(value: &str) -> String { + escape_newlines(value).replace('|', "\\|") +} + +fn render_md_table(columns: &[String], hits: &[json::Value]) -> String { + let mut out = String::new(); + let header = columns.iter().map(|c| md_escape(c)).collect::>(); + out.push_str(&format!("| {} |", header.join(" | "))); + out.push('\n'); + out.push_str(&format!("| {} |", vec!["---"; columns.len()].join(" | "))); + for hit in hits { + out.push('\n'); + let row = columns + .iter() + .map(|c| md_escape(&cell_value(hit, c))) + .collect::>(); + out.push_str(&format!("| {} |", row.join(" | "))); + } + out +} + +#[cfg(test)] +mod tests { + use config::utils::json::json; + + use super::*; + + fn response_with_hits(hits: Vec) -> Response { + Response { + hits, + ..Default::default() + } + } + + #[test] + fn json_format_is_noop() { + let mut res = response_with_hits(vec![json!({"a": 1})]); + apply_output_format(&mut res, OutputFormat::Json); + assert_eq!(res.hits.len(), 1); + assert!(res.format.is_none()); + assert!(res.data.is_none()); + } + + #[test] + fn csv_basic_with_timestamp_first() { + let mut res = response_with_hits(vec![ + json!({"_timestamp": 1000, "level": "info", "message": "hello"}), + json!({"_timestamp": 2000, "level": "error", "message": "boom"}), + ]); + apply_output_format(&mut res, OutputFormat::Csv); + assert_eq!(res.format.as_deref(), Some("csv")); + assert_eq!( + res.data.as_deref(), + Some("_timestamp,level,message\n1000,info,hello\n2000,error,boom") + ); + assert_eq!(res.columns, vec!["_timestamp", "level", "message"]); + assert!(res.hits.is_empty()); + } + + #[test] + fn csv_escapes_newline_comma_quote_and_nested() { + let mut res = response_with_hits(vec![json!({ + "msg": "line1\nline2", + "note": "a,b", + "quoted": "say \"hi\"", + "k8s": {"pod": "x"}, + })]); + apply_output_format(&mut res, OutputFormat::Csv); + let data = res.data.unwrap(); + let mut lines = data.lines(); + lines.next(); // header + let row = lines.next().unwrap(); + // no record spans multiple physical lines + assert!(lines.next().is_none()); + assert!(row.contains(r"line1\nline2")); + assert!(row.contains("\"a,b\"")); + assert!(row.contains(r#""say ""hi""""#)); + assert!(row.contains(r#""{""pod"":""x""}""#)); + } + + #[test] + fn csv_missing_and_null_cells_are_empty() { + let mut res = response_with_hits(vec![ + json!({"a": 1, "b": json::Value::Null}), + json!({"a": 2, "c": "x"}), + ]); + apply_output_format(&mut res, OutputFormat::Csv); + assert_eq!(res.data.as_deref(), Some("a,b,c\n1,,\n2,,x")); + } + + #[test] + fn csv_respects_server_columns_order() { + let mut res = response_with_hits(vec![json!({"a": 1, "z": 2})]); + res.columns = vec!["z".to_string(), "a".to_string()]; + apply_output_format(&mut res, OutputFormat::Csv); + assert_eq!(res.data.as_deref(), Some("z,a\n2,1")); + } + + #[test] + fn md_table_escapes_pipe() { + let mut res = response_with_hits(vec![json!({"msg": "a|b"})]); + apply_output_format(&mut res, OutputFormat::MdTable); + assert_eq!(res.format.as_deref(), Some("md_table")); + assert_eq!(res.data.as_deref(), Some("| msg |\n| --- |\n| a\\|b |")); + } + + #[test] + fn sparse_result_falls_back_to_ndjson() { + // 25 columns, each row fills only one of them -> ~96% empty + let hits: Vec = (0..10) + .map(|i| json!({format!("field_{}", i % 25): "v"})) + .collect(); + let mut res = response_with_hits(hits); + // widen the union past the min-columns threshold + for i in 10..25 { + res.hits.push(json!({format!("field_{i}"): "v"})); + } + apply_output_format(&mut res, OutputFormat::Csv); + assert_eq!(res.format.as_deref(), Some("ndjson")); + assert!(res.advisory.as_deref().unwrap().contains("sparse")); + assert_eq!(res.data.unwrap().lines().count(), 25); + assert!(res.hits.is_empty()); + } + + #[test] + fn non_object_hits_fall_back_to_ndjson() { + let mut res = response_with_hits(vec![json!(42), json!({"a": 1})]); + apply_output_format(&mut res, OutputFormat::Csv); + assert_eq!(res.format.as_deref(), Some("ndjson")); + assert_eq!(res.data.as_deref(), Some("42\n{\"a\":1}")); + } + + #[test] + fn empty_hits_yield_empty_data() { + let mut res = response_with_hits(vec![]); + apply_output_format(&mut res, OutputFormat::Csv); + assert_eq!(res.format.as_deref(), Some("csv")); + assert_eq!(res.data.as_deref(), Some("")); + } +} diff --git a/src/api_query/src/search/around.rs b/src/api_query/src/search/around.rs index c5d1f90343..b3a3df039d 100644 --- a/src/api_query/src/search/around.rs +++ b/src/api_query/src/search/around.rs @@ -187,6 +187,7 @@ pub(crate) async fn around( use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }; let resp_forward = SearchService::search(trace_id, org_id, stream_type, user_id.clone(), &req) .instrument(http_span.clone()) @@ -227,6 +228,7 @@ pub(crate) async fn around( use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }; let resp_backward = SearchService::search(trace_id, org_id, stream_type, user_id.clone(), &req) .instrument(http_span) diff --git a/src/api_query/src/search/mod.rs b/src/api_query/src/search/mod.rs index 07b9de443e..983f915469 100644 --- a/src/api_query/src/search/mod.rs +++ b/src/api_query/src/search/mod.rs @@ -74,6 +74,7 @@ use crate::{ }, }; +pub mod agent_format; pub(crate) mod around; pub mod error_utils; pub mod multi_streams; @@ -226,7 +227,7 @@ async fn can_use_distinct_stream( extensions( ("x-o2-ratelimit" = json!({"module": "Search", "operation": "get"})), ("x-o2-mcp" = json!({ - "description": "Search data with SQL, you can use `match_all('foo')` to search with full text search, also you can use `str_match(field, 'bar')` to search in a specific field; start_time, end_time can't be zero, need to be a valid micro timestamp. Note: in summary mode, response is capped at 100 hits, request detail='full' if you need more.", + "description": "Search data with SQL, you can use `match_all('foo')` to search with full text search, also you can use `str_match(field, 'bar')` to search in a specific field; start_time, end_time can't be zero, need to be a valid micro timestamp. Note: in summary mode, response is capped at 100 hits, request detail='full' if you need more. Tip: set agent_options.output_format='csv' to receive tabular hits as a compact csv block (~40% fewer tokens than json); 'md_table' for small result sets.", "category": "search", "pinned": true })) @@ -515,6 +516,10 @@ pub async fn search( res.is_partial = false; } + if let Some(opts) = req.agent_options.as_ref() { + agent_format::apply_output_format(&mut res, opts.output_format); + } + Json(res).into_response() } Err(err) => { @@ -1047,6 +1052,7 @@ pub async fn build_search_request_per_field( use_cache: req.use_cache, clear_cache: req.clear_cache, local_mode: None, + agent_options: None, }; let distinct_prefix = if can_use_distinct_stream { @@ -1284,6 +1290,7 @@ async fn values_v1( use_cache: default_use_cache(), clear_cache: get_clear_cache_from_request(query), local_mode: None, + agent_options: None, }; req.use_cache = get_use_cache_from_request(query); diff --git a/src/api_query/src/traces/dag.rs b/src/api_query/src/traces/dag.rs index da08f92d6e..50588ab055 100644 --- a/src/api_query/src/traces/dag.rs +++ b/src/api_query/src/traces/dag.rs @@ -237,6 +237,7 @@ pub async fn get_trace_dag( use_cache: get_use_cache_from_request(&query), clear_cache: false, local_mode: None, + agent_options: None, }; let stream_type = StreamType::Traces; diff --git a/src/api_query/src/traces/mod.rs b/src/api_query/src/traces/mod.rs index 3a297f77aa..d7da2c7658 100644 --- a/src/api_query/src/traces/mod.rs +++ b/src/api_query/src/traces/mod.rs @@ -490,6 +490,7 @@ pub async fn get_latest_traces( use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }; req.use_cache = get_use_cache_from_request(&query); @@ -1310,6 +1311,7 @@ async fn process_latest_traces_stream( use_cache, clear_cache: false, local_mode: None, + agent_options: None, }; // Get time partitions diff --git a/src/cli/data/export.rs b/src/cli/data/export.rs index 4db337fe43..6eefdc94d6 100644 --- a/src/cli/data/export.rs +++ b/src/cli/data/export.rs @@ -69,6 +69,7 @@ impl Context for Export { use_cache: false, clear_cache: false, local_mode: None, + agent_options: None, }; match SearchService::search("", &c.org, stream_type, None, &req).await { diff --git a/src/config/src/meta/search.rs b/src/config/src/meta/search.rs index 0bba4e0bce..327b5be921 100644 --- a/src/config/src/meta/search.rs +++ b/src/config/src/meta/search.rs @@ -74,12 +74,39 @@ pub struct Request { pub clear_cache: bool, #[serde(skip_serializing_if = "Option::is_none", default)] pub local_mode: Option, + #[serde(skip_serializing_if = "Option::is_none", default)] + pub agent_options: Option, } pub fn default_use_cache() -> bool { get_config().common.result_cache_enabled } +/// Agent-oriented response options. When absent the response is unchanged, so +/// existing clients (UI) are unaffected. +#[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)] +#[schema(as = SearchAgentOptions)] +pub struct AgentOptions { + /// Render `hits` as a compact string block in `data` instead of a JSON + /// array. Tabular results shrink to ~60% of their JSON token cost as csv. + #[serde(default)] + pub output_format: OutputFormat, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum OutputFormat { + /// Current behavior: `hits` is a JSON array of objects. + #[default] + Json, + /// `data` holds a CSV string; newlines inside cells are escaped to a + /// literal `\n` so every record stays on one line. + Csv, + /// `data` holds a markdown table; better column alignment for small + /// result sets at ~8% more tokens than csv. + MdTable, +} + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)] pub enum RequestEncoding { #[serde(rename = "base64")] @@ -257,6 +284,16 @@ pub struct Response { pub query_index: Option, #[serde(skip_serializing_if = "Option::is_none")] pub peak_memory_usage: Option, + /// Set when `agent_options.output_format` reformatted `hits` into `data`: + /// "csv", "md_table", or "ndjson" (automatic fallback for sparse results). + #[serde(skip_serializing_if = "Option::is_none", default)] + pub format: Option, + /// Formatted result block when `format` is set; `hits` is emptied. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub data: Option, + /// Human/agent-readable note about server-side formatting decisions. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub advisory: Option, } /// Paginated response used by list-style APIs (sessions, traces, users). @@ -449,6 +486,9 @@ impl Response { is_histogram_eligible: None, query_index: None, peak_memory_usage: None, + format: None, + data: None, + advisory: None, } } @@ -724,6 +764,7 @@ impl SearchHistoryRequest { use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }; Ok(search_req) } @@ -1342,6 +1383,7 @@ impl MultiStreamRequest { use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }); } res diff --git a/src/core/src/alerts/mod.rs b/src/core/src/alerts/mod.rs index cb2b3d1148..c3cfb7e997 100644 --- a/src/core/src/alerts/mod.rs +++ b/src/core/src/alerts/mod.rs @@ -433,6 +433,7 @@ impl QueryConditionExt for QueryCondition { use_cache: false, clear_cache: false, local_mode: None, + agent_options: None, }; log::debug!( "evaluate_scheduled trace_id: {trace_id}, begin to call SearchService::search, {req:?}" diff --git a/src/core/src/metrics/prom.rs b/src/core/src/metrics/prom.rs index 7fbe4b6b92..0f255a51a3 100644 --- a/src/core/src/metrics/prom.rs +++ b/src/core/src/metrics/prom.rs @@ -922,6 +922,7 @@ pub async fn get_series( use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }; let series = match search_service::search("", org_id, StreamType::Metrics, None, &req).await { Err(err) => { @@ -1132,6 +1133,7 @@ pub async fn get_label_values( use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }; let mut label_values = match search_service::search("", org_id, stream_type, None, &req).await { Ok(resp) => resp diff --git a/src/core/src/search/cache/cacher.rs b/src/core/src/search/cache/cacher.rs index 24cfac2cb6..b597fcff85 100644 --- a/src/core/src/search/cache/cacher.rs +++ b/src/core/src/search/cache/cacher.rs @@ -1085,6 +1085,9 @@ mod tests { query_index: None, peak_memory_usage: Some(1024000.0), histogram_breakdown_field: None, + format: None, + data: None, + advisory: None, }, deltas: vec![], has_cached_data: true, @@ -1204,6 +1207,9 @@ mod tests { query_index: None, peak_memory_usage: Some(1024000.0), histogram_breakdown_field: None, + format: None, + data: None, + advisory: None, }, deltas: vec![], has_cached_data: true, @@ -1244,6 +1250,9 @@ mod tests { query_index: None, peak_memory_usage: Some(1024000.0), histogram_breakdown_field: None, + format: None, + data: None, + advisory: None, }, deltas: vec![], has_cached_data: true, @@ -1313,6 +1322,7 @@ mod tests { use_cache: true, clear_cache: false, local_mode: None, + agent_options: None, }; let mut origin_sql = req.query.sql.clone(); let file_path = "test_org/logs/test_stream".to_string(); diff --git a/src/core/src/self_reporting/search.rs b/src/core/src/self_reporting/search.rs index 9a5bfe859d..3bb785b840 100644 --- a/src/core/src/self_reporting/search.rs +++ b/src/core/src/self_reporting/search.rs @@ -72,6 +72,7 @@ pub async fn get_usage( use_cache: default_use_cache(), clear_cache: false, local_mode: None, + agent_options: None, }; let trace_id = ider::uuid();