hyperdb_mcp/
inspect.rs

1// Copyright (c) 2026, Salesforce, Inc. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0 OR MIT
3
4//! Dry-run file inspection for the `inspect_file` MCP tool.
5//!
6//! Given a path to a CSV, Parquet, or Arrow IPC file, [`inspect_source`]
7//! returns the schema we would use to create a table (using the exact same
8//! inference + full-file numeric widening as [`crate::ingest`]) along with
9//! per-column diagnostics that help an LLM pick a safer schema override
10//! *before* running `load_file`:
11//!
12//! * `null_count` — how many null / empty cells were seen in the full file.
13//! * `min` / `max` — smallest and largest values observed for numeric columns.
14//! * `sample_values` — a handful of raw string values from the first rows.
15//!
16//! Nothing is written to Hyper; the engine is not even touched. The result is
17//! intentionally cheap enough to call eagerly before every `load_file`, which
18//! is the workflow we recommend in the tool description and error suggestions.
19//!
20//! For non-CSV inputs we fall back to the file's embedded schema (Parquet /
21//! Arrow IPC carry exact types) and skip min/max because extracting those
22//! cheaply would mean decoding every batch.
23
24use crate::error::{ErrorCode, McpError};
25use crate::ingest::{detect_file_format, normalize_json_or_jsonl, InferredFileFormat};
26use crate::ingest_arrow::arrow_schema_to_columns;
27use crate::schema::{infer_csv_schema, infer_json_schema, widen_csv_numeric_columns, ColumnSchema};
28use arrow::datatypes::Schema as ArrowSchema;
29use serde_json::{json, Value};
30use std::path::Path;
31use std::sync::Arc;
32
33/// Default number of sample rows returned for each column if the caller does
34/// not request a specific count. Small enough to keep the response compact for
35/// LLM clients while still giving a human-readable preview.
36pub const DEFAULT_SAMPLE_ROWS: usize = 5;
37
38/// Per-column diagnostics returned alongside the inferred schema.
39#[derive(Debug, Default, Clone)]
40pub struct ColumnStats {
41    pub null_count: u64,
42    /// Minimum observed integer value for numeric columns (as `i128` to cover
43    /// anything the widen pass will accept). `None` for non-numeric columns
44    /// or columns that only ever contained floats / NULLs.
45    pub min_i128: Option<i128>,
46    pub max_i128: Option<i128>,
47    /// Minimum / maximum observed floating-point value. Populated for `DOUBLE
48    /// PRECISION` columns and integer columns that saw a decimal value.
49    pub min_f64: Option<f64>,
50    pub max_f64: Option<f64>,
51    /// Up to `sample_rows` raw string values from the beginning of the file.
52    pub sample_values: Vec<String>,
53}
54
55/// Full inspection result for a single source.
56#[derive(Debug, Clone)]
57pub struct InspectReport {
58    pub columns: Vec<ColumnSchema>,
59    pub stats: Vec<ColumnStats>,
60    pub row_count: u64,
61    pub file_format: String,
62    pub file_size_bytes: u64,
63    /// The first `sample_rows` rows as vectors of raw string values, aligned
64    /// to `columns`. For non-CSV inputs this is empty.
65    pub sample_rows: Vec<Vec<String>>,
66}
67
68impl InspectReport {
69    /// Serialize into the JSON shape returned by the MCP `inspect_file` tool.
70    #[must_use]
71    pub fn to_json(&self) -> Value {
72        let columns: Vec<Value> = self
73            .columns
74            .iter()
75            .zip(self.stats.iter())
76            .map(|(col, s)| {
77                let mut obj = json!({
78                    "name": col.name,
79                    "type": col.hyper_type,
80                    "nullable": col.nullable,
81                    "null_count": s.null_count,
82                    "sample_values": s.sample_values,
83                });
84                if let Some(m) = s.min_i128 {
85                    if let Ok(n) = i64::try_from(m) {
86                        obj["min"] = json!(n);
87                    } else {
88                        obj["min"] = json!(m.to_string());
89                    }
90                }
91                if let Some(m) = s.max_i128 {
92                    if let Ok(n) = i64::try_from(m) {
93                        obj["max"] = json!(n);
94                    } else {
95                        obj["max"] = json!(m.to_string());
96                    }
97                }
98                if let Some(m) = s.min_f64 {
99                    obj["min_f64"] = json!(m);
100                }
101                if let Some(m) = s.max_f64 {
102                    obj["max_f64"] = json!(m);
103                }
104                obj
105            })
106            .collect();
107        json!({
108            "columns": columns,
109            "row_count": self.row_count,
110            "file_format": self.file_format,
111            "file_size_bytes": self.file_size_bytes,
112            "sample_rows": self.sample_rows.iter().map(|row| {
113                self.columns.iter().zip(row.iter()).map(|(c, v)| (c.name.clone(), Value::String(v.clone()))).collect::<serde_json::Map<_, _>>()
114            }).collect::<Vec<_>>(),
115        })
116    }
117}
118
119/// Top-level dispatcher: pick the right inspector for `path` using the
120/// same extension + content-sniffing logic [`crate::ingest::detect_file_format`]
121/// drives for the ingest side. That shared dispatcher is what ensures
122/// `inspect_file` and `load_file` can never disagree on what a file is —
123/// if inspect says "this is JSONL", load will try it as JSONL too.
124///
125/// # Errors
126///
127/// - Returns [`ErrorCode::FileNotFound`] if `path` does not exist.
128/// - Propagates format-specific errors from the delegated inspector —
129///   `inspect_parquet`, `inspect_arrow_ipc`, `inspect_json`, or
130///   `inspect_csv` — each of which surfaces its own I/O, decoding,
131///   or schema-inference failures.
132pub fn inspect_source(path: &str, sample_rows: usize) -> Result<InspectReport, McpError> {
133    let file_path = Path::new(path);
134    if !file_path.exists() {
135        return Err(McpError::new(
136            ErrorCode::FileNotFound,
137            format!("File not found: {path}"),
138        ));
139    }
140    let file_size = std::fs::metadata(file_path).map_or(0, |m| m.len());
141    match detect_file_format(file_path) {
142        InferredFileFormat::Parquet => inspect_parquet(path, file_size),
143        InferredFileFormat::ArrowIpc => inspect_arrow_ipc(path, file_size),
144        InferredFileFormat::Json => inspect_json(path, file_size, sample_rows),
145        InferredFileFormat::Csv => inspect_csv(path, file_size, sample_rows),
146    }
147}
148
149/// JSON / JSONL inspector: reads the file, auto-detects between a
150/// top-level array and newline-delimited JSON via
151/// [`normalize_json_or_jsonl`], then runs the same
152/// [`infer_json_schema`] pass `ingest_json` uses so the reported schema
153/// matches what `load_file` would create.
154///
155/// Per-column stats collected:
156///
157/// * `null_count` — number of records where the key is missing or its
158///   value is `null`.
159/// * `sample_values` — up to `sample_rows` stringified non-null values
160///   from the first records.
161///
162/// `min`/`max` are intentionally skipped for JSON: unlike CSV, JSON
163/// types are self-describing so the schema already tells you what
164/// range to expect, and extracting numeric extrema would require
165/// iterating the full dataset in the inspector's hot path.
166fn inspect_json(path: &str, file_size: u64, sample_rows: usize) -> Result<InspectReport, McpError> {
167    let raw = std::fs::read_to_string(path)
168        .map_err(|e| McpError::new(ErrorCode::FileNotFound, format!("Cannot read file: {e}")))?;
169    inspect_json_from_text(&raw, file_size, sample_rows)
170}
171
172/// Inspect a JSON / JSONL text string (already in memory) without
173/// reading from a file. Used by the `inspect_file` handler when
174/// `json_extract_path` has already extracted the relevant slice.
175///
176/// Public so [`crate::server`] can call it after `extract_json_path`.
177///
178/// # Errors
179///
180/// - Propagates errors from [`normalize_json_or_jsonl`] (malformed
181///   JSONL) and [`infer_json_schema`] (schema inference failure).
182/// - Returns [`ErrorCode::SchemaMismatch`] if the normalized array
183///   cannot be re-parsed as JSON for stats collection.
184pub fn inspect_json_from_text(
185    raw: &str,
186    file_size: u64,
187    sample_rows: usize,
188) -> Result<InspectReport, McpError> {
189    // Remember the original shape for the `file_format` field in the
190    // report; `normalize_json_or_jsonl` always returns a top-level
191    // array so we'd otherwise lose that signal.
192    let is_array = raw.trim_start().starts_with('[');
193    let file_format = if is_array { "json" } else { "jsonl" };
194
195    let array_text = normalize_json_or_jsonl(raw)?;
196    let columns = infer_json_schema(&array_text)?;
197
198    // Parse the array a second time for stats — cheap, and keeps
199    // schema inference and stats independent.
200    let array: Vec<Value> = serde_json::from_str(&array_text)
201        .map_err(|e| McpError::new(ErrorCode::SchemaMismatch, format!("Invalid JSON: {e}")))?;
202
203    let sample_cap = sample_rows.max(1);
204    let mut stats: Vec<ColumnStats> = columns.iter().map(|_| ColumnStats::default()).collect();
205    let mut sample_rows_out: Vec<Vec<String>> = Vec::new();
206    let row_count = array.len() as u64;
207
208    for obj in &array {
209        let Some(object) = obj.as_object() else {
210            continue;
211        };
212        if sample_rows_out.len() < sample_cap {
213            let mut row_vals = Vec::with_capacity(columns.len());
214            for col in &columns {
215                row_vals.push(value_preview(object.get(&col.name)));
216            }
217            sample_rows_out.push(row_vals);
218        }
219        for (idx, col) in columns.iter().enumerate() {
220            let s = &mut stats[idx];
221            match object.get(&col.name) {
222                None | Some(Value::Null) => s.null_count += 1,
223                Some(v) => {
224                    if s.sample_values.len() < sample_cap {
225                        s.sample_values.push(value_preview(Some(v)));
226                    }
227                }
228            }
229        }
230    }
231
232    Ok(InspectReport {
233        columns,
234        stats,
235        row_count,
236        file_format: file_format.into(),
237        file_size_bytes: file_size,
238        sample_rows: sample_rows_out,
239    })
240}
241
242/// Render a JSON value into the short string form used for
243/// `sample_values` and `sample_rows` in the inspect report. Scalars
244/// stringify naturally; nested arrays/objects are rendered as compact
245/// JSON so the preview stays on one line and never grows unboundedly
246/// long.
247fn value_preview(v: Option<&Value>) -> String {
248    const MAX_LEN: usize = 120;
249    let raw = match v {
250        None | Some(Value::Null) => String::new(),
251        Some(Value::String(s)) => s.clone(),
252        Some(Value::Number(n)) => n.to_string(),
253        Some(Value::Bool(b)) => b.to_string(),
254        Some(other) => other.to_string(),
255    };
256    if raw.len() > MAX_LEN {
257        let mut s = raw[..MAX_LEN].to_string();
258        s.push('…');
259        s
260    } else {
261        raw
262    }
263}
264
265/// CSV inspector: reuses [`infer_csv_schema`] and [`widen_csv_numeric_columns`]
266/// so the schema returned here is *exactly* the one `load_file` would create,
267/// then adds a streaming pass that tallies nulls / min / max and captures
268/// sample values for each column.
269fn inspect_csv(path: &str, file_size: u64, sample_rows: usize) -> Result<InspectReport, McpError> {
270    let text = std::fs::read_to_string(path)
271        .map_err(|e| McpError::new(ErrorCode::FileNotFound, format!("Cannot read file: {e}")))?;
272
273    let mut columns = infer_csv_schema(&text, true)?;
274    widen_csv_numeric_columns(text.as_bytes(), true, &mut columns)?;
275
276    let mut stats: Vec<ColumnStats> = columns.iter().map(|_| ColumnStats::default()).collect();
277    let mut sample_rows_out: Vec<Vec<String>> = Vec::new();
278    let sample_cap = sample_rows.max(1);
279
280    let mut rdr = csv::ReaderBuilder::new()
281        .has_headers(true)
282        .from_reader(text.as_bytes());
283    let mut row_count: u64 = 0;
284    for result in rdr.records() {
285        let record = result.map_err(|e| {
286            McpError::new(
287                ErrorCode::SchemaMismatch,
288                format!("CSV parse error at row {}: {e}", row_count + 1),
289            )
290        })?;
291        row_count += 1;
292        // Capture sample rows from the top of the file.
293        if sample_rows_out.len() < sample_cap {
294            sample_rows_out.push(
295                record
296                    .iter()
297                    .map(std::string::ToString::to_string)
298                    .collect(),
299            );
300        }
301        // Per-column stats.
302        for (col_idx, col) in columns.iter().enumerate() {
303            let s = &mut stats[col_idx];
304            let raw = record.get(col_idx).unwrap_or("");
305            let trimmed = raw.trim();
306            if trimmed.is_empty()
307                || trimmed.eq_ignore_ascii_case("null")
308                || trimmed.eq_ignore_ascii_case("na")
309            {
310                s.null_count += 1;
311                continue;
312            }
313            if s.sample_values.len() < sample_cap {
314                s.sample_values.push(trimmed.to_string());
315            }
316            // Integer stats for integer-typed columns (post-widening).
317            if matches!(
318                col.hyper_type.as_str(),
319                "INT" | "INTEGER" | "BIGINT" | "NUMERIC(38,0)"
320            ) {
321                if let Ok(n) = trimmed.parse::<i128>() {
322                    s.min_i128 = Some(s.min_i128.map_or(n, |m| m.min(n)));
323                    s.max_i128 = Some(s.max_i128.map_or(n, |m| m.max(n)));
324                }
325            } else if col.hyper_type == "DOUBLE PRECISION" {
326                if let Ok(n) = trimmed.parse::<f64>() {
327                    s.min_f64 = Some(s.min_f64.map_or(n, |m| m.min(n)));
328                    s.max_f64 = Some(s.max_f64.map_or(n, |m| m.max(n)));
329                }
330            }
331        }
332    }
333
334    Ok(InspectReport {
335        columns,
336        stats,
337        row_count,
338        file_format: "csv".into(),
339        file_size_bytes: file_size,
340        sample_rows: sample_rows_out,
341    })
342}
343
344/// Parquet inspector: the file embeds exact types, so the schema is read from
345/// Parquet metadata and no widening pass is needed. Min/max extraction would
346/// require decoding batches, so we skip it and rely on the fact that Parquet
347/// readers already emit correctly-typed values.
348fn inspect_parquet(path: &str, file_size: u64) -> Result<InspectReport, McpError> {
349    let file = std::fs::File::open(path)
350        .map_err(|e| McpError::new(ErrorCode::FileNotFound, format!("Cannot open file: {e}")))?;
351    let builder = parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(file)
352        .map_err(|e| {
353            McpError::new(
354                ErrorCode::UnsupportedFormat,
355                format!("Invalid Parquet file: {e}"),
356            )
357        })?;
358    let arrow_schema: Arc<ArrowSchema> = Arc::clone(builder.schema());
359    // Parquet `num_rows` is an `i64`; a negative value means a corrupt footer.
360    // Report 0 rather than panic or silently wrap.
361    let row_count = u64::try_from(builder.metadata().file_metadata().num_rows()).unwrap_or(0);
362    let columns = arrow_schema_to_columns(&arrow_schema);
363    let stats = vec![ColumnStats::default(); columns.len()];
364    Ok(InspectReport {
365        columns,
366        stats,
367        row_count,
368        file_format: "parquet".into(),
369        file_size_bytes: file_size,
370        sample_rows: Vec::new(),
371    })
372}
373
374/// Arrow IPC inspector: schema is read from the file footer.
375fn inspect_arrow_ipc(path: &str, file_size: u64) -> Result<InspectReport, McpError> {
376    let file = std::fs::File::open(path)
377        .map_err(|e| McpError::new(ErrorCode::FileNotFound, format!("Cannot open file: {e}")))?;
378    let reader = arrow::ipc::reader::FileReader::try_new(file, None).map_err(|e| {
379        McpError::new(
380            ErrorCode::UnsupportedFormat,
381            format!("Invalid Arrow IPC file: {e}"),
382        )
383    })?;
384    let arrow_schema: Arc<ArrowSchema> = reader.schema();
385    let row_count: u64 = reader
386        .into_iter()
387        .map(|b| b.map_or(0, |rb| rb.num_rows() as u64))
388        .sum();
389    let columns = arrow_schema_to_columns(&arrow_schema);
390    let stats = vec![ColumnStats::default(); columns.len()];
391    Ok(InspectReport {
392        columns,
393        stats,
394        row_count,
395        file_format: "arrow_ipc".into(),
396        file_size_bytes: file_size,
397        sample_rows: Vec::new(),
398    })
399}
hyperdb_mcp/inspect.rs

hyperdb_mcp/
inspect.rs
