feat: add agent_options.output_format (csv/md_table) to _search response (#13343)

Design at: #13343

## What

`_search` gains an optional `agent_options.output_format` parameter
(`json` | `csv` | `md_table`, default `json`). When set to
`csv`/`md_table`, the response's `hits` array is rendered as a single
compact string block:

```json
{ "query": { "sql": "...", "start_time": ..., "end_time": ... },
  "agent_options": { "output_format": "csv" } }
```

Response then carries three new optional fields (never serialized
otherwise): `format` (the format actually used), `data` (the formatted
block), `advisory` (note about server-side formatting decisions), with
`columns` reporting the column order and `hits` emptied.

## Why

Agent/MCP consumers pay per token, and a JSON hits array repeats every
key on every row — the repeated keys, not whitespace, are the dominant
cost (minify only gets to ~78%). csv removes them: ~40% fewer tokens
than minified JSON on typical log results, letting the same context
budget hold ~1.7x more rows. `md_table` (~8% more than csv) aligns
better for small result sets. This is the first piece of the Agent
query-optimization (AX) track; `agent_options` is the extension point
for the follow-ups (token-budget pagination, query budget/continuation).

## Design

- **Serialization edge only**: formatting runs in the `_search` handler
after `SearchService::cache::search` returns. Engine, partition
scheduling and result cache always see original JSON hits; cached
entries never contain formatted strings, and the cache key is unchanged,
so a UI (json) query and an agent (csv) query share the same cache
entry.
- **Model-friendly escaping**: newlines inside cells become a literal
`\n` so no record ever spans physical lines (multi-line RFC 4180 quoted
cells are a known LLM parsing hazard); commas/quotes still get standard
csv quoting so rows stay machine-parsable. `md_table` escapes `|`.
Nested objects/arrays are minified JSON in the cell.
- **Column order**: server-provided `columns` when present, otherwise
first-seen key order with `_timestamp` promoted to the front.
- **Sparse fallback**: results with >= 20 columns and > 50% empty cells
(typical `select *` on a wide stream) fall back to ndjson — per-row keys
are cheaper and less error-prone there than a mostly-empty grid. The
fallback is announced via `advisory`, never silent. Non-object rows fall
back the same way.
- **Zero impact when unset**: requests without `agent_options` produce
byte-for-byte identical responses. The remaining 9 files changed are
mechanical `agent_options: None` additions to internal `Request` struct
literals.
- The `_search` `x-o2-mcp` tool description now mentions the option, and
`SearchAgentOptions` flows into the generated OpenAPI/MCP tool schema
automatically.

## Testing

- 9 unit tests in `agent_format.rs`: json no-op, csv basics +
`_timestamp` promotion, newline/comma/quote/nested escaping,
missing/null cells, server column order, md_table pipe escaping, sparse
fallback, non-object fallback, empty hits.
- Verified against a running local instance: default path unchanged (no
new keys), csv/md_table output with all escaping cases, sparse stream
(32 cols, 90% empty) falls back to ndjson with advisory, invalid
`output_format` returns 422 listing valid variants, `SearchAgentOptions`
present in `/api-doc/openapi.json`.
- `cargo check --workspace`, `cargo clippy -p openobserve-api-query`,
`cargo fmt` clean.
This commit is contained in:
Hengfei Yang
2026-07-22 04:48:19 +00:00
committed by GitHub
parent be31b47172
commit 39b2b3ca77
11 changed files with 390 additions and 1 deletions
+320
View File
@@ -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 <http://www.gnu.org/licenses/>.
//! 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<String> {
let derive_order = res.columns.is_empty();
let mut columns = res.columns.clone();
let mut seen: hashbrown::HashSet<String> = 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::<Vec<_>>()
.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::<Vec<_>>()
.join(","),
);
for hit in hits {
out.push('\n');
out.push_str(
&columns
.iter()
.map(|c| csv_escape(&cell_value(hit, c)))
.collect::<Vec<_>>()
.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::<Vec<_>>();
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::<Vec<_>>();
out.push_str(&format!("| {} |", row.join(" | ")));
}
out
}
#[cfg(test)]
mod tests {
use config::utils::json::json;
use super::*;
fn response_with_hits(hits: Vec<json::Value>) -> 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<json::Value> = (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(""));
}
}
+2
View File
@@ -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)
+8 -1
View File
@@ -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);
+1
View File
@@ -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;
+2
View File
@@ -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
+1
View File
@@ -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 {
+42
View File
@@ -74,12 +74,39 @@ pub struct Request {
pub clear_cache: bool,
#[serde(skip_serializing_if = "Option::is_none", default)]
pub local_mode: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none", default)]
pub agent_options: Option<AgentOptions>,
}
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<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pub peak_memory_usage: Option<f64>,
/// 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<String>,
/// Formatted result block when `format` is set; `hits` is emptied.
#[serde(skip_serializing_if = "Option::is_none", default)]
pub data: Option<String>,
/// Human/agent-readable note about server-side formatting decisions.
#[serde(skip_serializing_if = "Option::is_none", default)]
pub advisory: Option<String>,
}
/// 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
+1
View File
@@ -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:?}"
+2
View File
@@ -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
+10
View File
@@ -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();
+1
View File
@@ -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();