fix(elasticsearch): preserve plain text REST responses

crates/dbx-core/src/db/elasticsearch_driver.rs

@@ -795,9 +795,9 @@ pub async fn execute_rest_query(client: &EsClient, input: &str) -> Result<crate:
     .map_err(|e| format!("Elasticsearch request failed: {e}"))?;
 
     let status = resp.status().as_u16();
-    let body: serde_json::Value = resp.json().await.unwrap_or_else(|_| serde_json::Value::Null);
+    let body = resp.text().await.map_err(|e| format!("Elasticsearch response read failed: {e}"))?;
 
-    parse_elasticsearch_response(status, body, start)
+    parse_elasticsearch_rest_response(status, &body, start)
 }
 
 // Size to use when `SELECT *` is run without an explicit LIMIT — large enough
@@ -955,6 +955,37 @@ fn parse_elasticsearch_response(
     }
 }
 
+fn parse_elasticsearch_rest_response(
+    status: u16,
+    body_text: &str,
+    start: std::time::Instant,
+) -> Result<crate::types::QueryResult, String> {
+    if body_text.trim().is_empty() {
+        return parse_elasticsearch_response(status, serde_json::Value::Null, start);
+    }
+
+    if let Ok(body) = serde_json::from_str::<serde_json::Value>(body_text) {
+        return parse_elasticsearch_response(status, body, start);
+    }
+
+    // CAT APIs default to text/plain for human-readable output. Keep those
+    // responses visible instead of dropping them when JSON parsing is not valid.
+    let rows: Vec<Vec<serde_json::Value>> =
+        body_text.lines().map(|line| vec![serde_json::Value::String(line.to_string())]).collect();
+    let affected_rows = rows.len() as u64;
+    Ok(crate::types::QueryResult {
+        columns: vec!["response".to_string()],
+        column_types: Vec::new(),
+        column_sortables: vec![],
+        rows,
+        affected_rows,
+        execution_time_ms: start.elapsed().as_millis(),
+        truncated: false,
+        session_id: None,
+        has_more: false,
+    })
+}
+
 fn parse_select_star_search_query(input: &str) -> Option<ElasticsearchSearchQuery> {
     let mut cursor = skip_sql_whitespace(input, 0);
     cursor = consume_sql_keyword(input, cursor, "select")?;
@@ -1690,6 +1721,50 @@ mod tests {
         assert_eq!(result.rows[0][routing_idx], json!("tenant-1"));
     }
 
+    #[test]
+    fn parses_plain_text_rest_response_without_dropping_body() {
+        let body =
+            "health status index           docs.count store.size\ngreen  open   app-log-2026-07 42         10mb\n";
+        let result = super::parse_elasticsearch_rest_response(200, body, std::time::Instant::now()).unwrap();
+
+        assert_eq!(result.columns, vec!["response"]);
+        assert_eq!(result.rows.len(), 2);
+        assert_eq!(result.rows[0][0], json!("health status index           docs.count store.size"));
+        assert_eq!(result.rows[1][0], json!("green  open   app-log-2026-07 42         10mb"));
+        assert_eq!(result.affected_rows, 2);
+    }
+
+    #[tokio::test]
+    async fn execute_rest_query_keeps_plain_text_response_body() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        let body =
+            "health status index           docs.count store.size\ngreen  open   app-log-2026-07 42         10mb\n";
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.unwrap();
+            let mut request = [0_u8; 1024];
+            let read = socket.read(&mut request).await.unwrap();
+            let request = String::from_utf8_lossy(&request[..read]);
+            assert!(request.starts_with("GET /_cat/indices "));
+            let response = format!(
+                "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
+                body.len(),
+                body
+            );
+            socket.write_all(response.as_bytes()).await.unwrap();
+        });
+
+        let client = EsClient::new(&format!("http://{addr}"), None, None, false, Duration::from_secs(1));
+        let result = super::execute_rest_query(&client, "GET /_cat/indices").await.unwrap();
+        server.await.unwrap();
+
+        assert_eq!(result.columns, vec!["response"]);
+        assert_eq!(result.rows.len(), 2);
+        assert_eq!(result.rows[1][0], json!("green  open   app-log-2026-07 42         10mb"));
+    }
+
     #[test]
     fn document_body_removes_elasticsearch_id_metadata() {
         let doc = super::elasticsearch_document_body_from_json(r#"{"_id":"abc","_routing":"tenant-1","name":"Alice"}"#)