hyperdb_mcp/schema.rs
1// Copyright (c) 2026, Salesforce, Inc. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0 OR MIT
3
4//! Schema inference and type mapping.
5//!
6//! # Three-Tier Inference
7//!
8//! The tier is selected automatically based on the data source:
9//!
10//! | Tier | Source | Strategy |
11//! |------|--------|----------|
12//! | **Exact** | Arrow IPC, Parquet | Types read from file metadata — zero guessing. |
13//! | **Structural** | JSON | Full scan of all objects. Per-column type widening (see below). |
14//! | **Heuristic** | CSV | Header row for names, first 1 000 rows sampled for types. Ambiguous → TEXT. |
15//!
16//! All tiers can be bypassed with an explicit `schema` override from the caller.
17//!
18//! # Type Widening Rules (Structural / Heuristic Tiers)
19//!
20//! When a JSON or CSV column contains mixed types across rows, the widening
21//! chain determines the final type:
22//!
23//! ```text
24//! Null < Bool < Int < BigInt < Double < Date < Timestamp < Text
25//! ```
26//!
27//! Specific rules (implemented in `resolve_type`):
28//! - **All null** → TEXT (safe catch-all).
29//! - **Uniform non-null** → that type, unchanged.
30//! - **Mixed numeric** (Int / `BigInt` / Double) → widest numeric type seen.
31//! - **Any other mix** (e.g. Bool + Int, Date + Text) → TEXT.
32//!
33//! # Arrow / Parquet Type Mapping
34//!
35//! The exact-tier mapping from Arrow types to Hyper SQL types is handled by
36//! `crate::ingest_arrow::arrow_type_to_hyper`. Key mappings:
37//!
38//! | Arrow | Hyper |
39//! |-------|-------|
40//! | Int16/Int32/Int64 | SMALLINT/INT/BIGINT |
41//! | UInt16/UInt32/UInt64 | INT/BIGINT/BIGINT (promoted to signed) |
42//! | Float32/Float64 | DOUBLE PRECISION |
43//! | Utf8/LargeUtf8 | TEXT |
44//! | Date32/Date64 | DATE |
45//! | Timestamp(_, None/Some) | TIMESTAMP / TIMESTAMPTZ |
46//! | Decimal128/Decimal256 | NUMERIC |
47
48use crate::error::{ErrorCode, McpError};
49use hyperdb_api::{SqlType, TableDefinition};
50use serde_json::Value;
51use std::collections::{BTreeMap, BTreeSet};
52
53/// A column's name, Hyper SQL type (as a string like `"INT"` or `"DOUBLE PRECISION"`),
54/// and nullability. This is the internal schema representation shared across all
55/// ingest and table-creation paths.
56#[derive(Debug, Clone)]
57pub struct ColumnSchema {
58 pub name: String,
59 /// Hyper type name (e.g. `"TEXT"`, `"BIGINT"`, `"NUMERIC(12,2)"`).
60 pub hyper_type: String,
61 pub nullable: bool,
62}
63
64/// Build a hyperdb-api `TableDefinition` from a list of `ColumnSchema`.
65///
66/// Uses the consuming builder pattern required by `TableDefinition`.
67///
68/// # Errors
69///
70/// - Returns [`ErrorCode::EmptyData`] if `columns` is empty.
71/// - Returns [`ErrorCode::SchemaMismatch`] if any column's `hyper_type`
72/// cannot be resolved by [`map_hyper_type`].
73pub fn build_table_def(
74 table_name: &str,
75 columns: &[ColumnSchema],
76) -> Result<TableDefinition, McpError> {
77 if columns.is_empty() {
78 return Err(McpError::new(
79 ErrorCode::EmptyData,
80 "No columns to create table from",
81 ));
82 }
83 let mut def = TableDefinition::new(table_name);
84 for col in columns {
85 let sql_type = map_hyper_type(&col.hyper_type).ok_or_else(|| {
86 McpError::new(
87 ErrorCode::SchemaMismatch,
88 format!("Unknown type: {}", col.hyper_type),
89 )
90 })?;
91 if col.nullable {
92 def = def.add_nullable_column(&col.name, sql_type);
93 } else {
94 def = def.add_required_column(&col.name, sql_type);
95 }
96 }
97 Ok(def)
98}
99
100/// Map a user-facing type name string (e.g. `"INT"`, `"NUMERIC(12,2)"`) to a
101/// [`hyperdb_api::SqlType`]. Accepts `PostgreSQL` aliases (`INT4`, `FLOAT8`, `VARCHAR`)
102/// so schema overrides from diverse sources work without normalization.
103#[must_use]
104pub fn map_hyper_type(type_name: &str) -> Option<SqlType> {
105 let upper = type_name.trim().to_uppercase();
106 match upper.as_str() {
107 "SMALLINT" | "INT2" => Some(SqlType::small_int()),
108 "INT" | "INTEGER" | "INT4" => Some(SqlType::int()),
109 "BIGINT" | "INT8" => Some(SqlType::big_int()),
110 "FLOAT" | "FLOAT4" | "REAL" => Some(SqlType::double()),
111 "DOUBLE" | "DOUBLE PRECISION" | "FLOAT8" => Some(SqlType::double()),
112 "TEXT" | "VARCHAR" | "STRING" => Some(SqlType::text()),
113 "BOOL" | "BOOLEAN" => Some(SqlType::bool()),
114 "DATE" => Some(SqlType::date()),
115 "TIME" => Some(SqlType::time()),
116 "TIMESTAMP" => Some(SqlType::timestamp()),
117 "TIMESTAMPTZ" | "TIMESTAMP WITH TIME ZONE" => Some(SqlType::timestamp_tz()),
118 "BYTEA" | "BYTES" => Some(SqlType::bytes()),
119 _ if upper.starts_with("NUMERIC") => {
120 // Parse NUMERIC(p,s) or default to NUMERIC(38,0)
121 if let Some(inner) = upper
122 .strip_prefix("NUMERIC(")
123 .and_then(|s| s.strip_suffix(')'))
124 {
125 let parts: Vec<&str> = inner.split(',').collect();
126 let precision = parts
127 .first()
128 .and_then(|p| p.trim().parse().ok())
129 .unwrap_or(38);
130 let scale = parts
131 .get(1)
132 .and_then(|s| s.trim().parse().ok())
133 .unwrap_or(0);
134 Some(SqlType::numeric(precision, scale))
135 } else {
136 Some(SqlType::numeric(38, 0))
137 }
138 }
139 _ => None,
140 }
141}
142
143// --- Tier 2: JSON Schema Inference ---
144
145/// Intermediate type tag used during schema inference. The widening order is:
146/// Null < Bool < Int < `BigInt` < Double < Date < Timestamp < Text.
147/// When a column has mixed non-numeric types, it collapses to Text.
148#[derive(Debug, Clone, PartialEq)]
149enum InferredType {
150 Null,
151 Bool,
152 Int,
153 BigInt,
154 Double,
155 Date,
156 Timestamp,
157 Text,
158}
159
160/// Infer a [`ColumnSchema`] for each key in a JSON array of objects (Tier 2).
161///
162/// Every object is scanned so that keys appearing in only some rows are detected
163/// as nullable. Per-column types are resolved with numeric widening
164/// (Int → `BigInt` → Double) and fall back to TEXT for mixed types.
165///
166/// Uses `BTreeSet`/`BTreeMap` for deterministic column ordering across runs.
167///
168/// # Errors
169///
170/// - Returns [`ErrorCode::SchemaMismatch`] if `json_str` is not a valid
171/// top-level JSON array of objects.
172/// - Returns [`ErrorCode::EmptyData`] if the array is empty.
173///
174/// # Panics
175///
176/// Does not panic in practice. The `col_types.get_mut(key).unwrap()` and
177/// `col_present.get_mut(key).unwrap()` calls are guarded by a preceding
178/// initialization loop that inserts every `key` from `all_keys` into
179/// both maps.
180pub fn infer_json_schema(json_str: &str) -> Result<Vec<ColumnSchema>, McpError> {
181 let array: Vec<serde_json::Map<String, Value>> =
182 serde_json::from_str(json_str).map_err(|e| {
183 McpError::new(
184 ErrorCode::SchemaMismatch,
185 format!("Invalid JSON array: {e}"),
186 )
187 })?;
188
189 if array.is_empty() {
190 return Err(McpError::new(ErrorCode::EmptyData, "JSON array is empty"));
191 }
192
193 // Collect all keys (BTreeSet for deterministic ordering)
194 let mut all_keys = BTreeSet::new();
195 for obj in &array {
196 for key in obj.keys() {
197 all_keys.insert(key.clone());
198 }
199 }
200
201 let total_rows = array.len();
202 let mut col_types: BTreeMap<String, Vec<InferredType>> = BTreeMap::new();
203 let mut col_present: BTreeMap<String, usize> = BTreeMap::new();
204
205 for key in &all_keys {
206 col_types.insert(key.clone(), Vec::new());
207 col_present.insert(key.clone(), 0);
208 }
209
210 for obj in &array {
211 for key in &all_keys {
212 match obj.get(key.as_str()) {
213 None => {}
214 Some(Value::Null) => {
215 col_types.get_mut(key).unwrap().push(InferredType::Null);
216 *col_present.get_mut(key).unwrap() += 1;
217 }
218 Some(val) => {
219 col_types
220 .get_mut(key)
221 .unwrap()
222 .push(infer_json_value_type(val));
223 *col_present.get_mut(key).unwrap() += 1;
224 }
225 }
226 }
227 }
228
229 let mut columns = Vec::new();
230 for key in &all_keys {
231 let types = &col_types[key];
232 let present_count = col_present[key];
233 let nullable = present_count < total_rows || types.contains(&InferredType::Null);
234 let resolved = resolve_type(types);
235 columns.push(ColumnSchema {
236 name: key.clone(),
237 hyper_type: inferred_to_hyper_name(&resolved),
238 nullable,
239 });
240 }
241
242 Ok(columns)
243}
244
245/// Classify a single JSON value. Numbers that fit in i32 are `Int`, larger
246/// integers are `BigInt`, anything with a fractional part is `Double`.
247/// Strings are further inspected for ISO 8601 date/timestamp patterns.
248fn infer_json_value_type(val: &Value) -> InferredType {
249 match val {
250 Value::Null => InferredType::Null,
251 Value::Bool(_) => InferredType::Bool,
252 Value::Number(n) => {
253 if let Some(i) = n.as_i64() {
254 if i32::try_from(i).is_ok() {
255 InferredType::Int
256 } else {
257 InferredType::BigInt
258 }
259 } else {
260 InferredType::Double
261 }
262 }
263 Value::String(s) => infer_string_type(s),
264 _ => InferredType::Text,
265 }
266}
267
268/// Attempt to recognize ISO 8601 date (`YYYY-MM-DD`) or timestamp
269/// (`YYYY-MM-DDThh:mm:ss`) patterns in a string value. Returns `Text` for
270/// anything that doesn't match.
271///
272/// # Safety of the string slices
273///
274/// Each indexing like `s[0..4]` requires `s.len()` to be at least the
275/// upper bound (otherwise it panics). The `s.len() == 10` / `s.len() >=
276/// 19` guards are the leftmost clauses in each `if` so Rust's
277/// short-circuit `&&` evaluation proves the length invariant before the
278/// slice operations run.
279fn infer_string_type(s: &str) -> InferredType {
280 // Try ISO 8601 date: YYYY-MM-DD
281 if s.len() == 10
282 && s.chars().nth(4) == Some('-')
283 && s.chars().nth(7) == Some('-')
284 && s[0..4].parse::<u16>().is_ok()
285 && s[5..7].parse::<u8>().is_ok()
286 && s[8..10].parse::<u8>().is_ok()
287 {
288 return InferredType::Date;
289 }
290 // Try ISO 8601 timestamp: YYYY-MM-DDThh:mm:ss
291 if s.len() >= 19
292 && s.chars().nth(10) == Some('T')
293 && s[0..10].contains('-')
294 && s[11..].contains(':')
295 {
296 return InferredType::Timestamp;
297 }
298 InferredType::Text
299}
300
301/// Resolve a column's final type from the per-row type observations.
302///
303/// Rules:
304/// 1. All-null columns become TEXT (safe catch-all).
305/// 2. Uniform non-null types pass through unchanged.
306/// 3. Mixed numeric types widen: Int → `BigInt` → Double.
307/// 4. Any other mix (e.g. Bool + Int, Date + Text) collapses to TEXT.
308fn resolve_type(types: &[InferredType]) -> InferredType {
309 let non_null: Vec<&InferredType> = types.iter().filter(|t| **t != InferredType::Null).collect();
310 if non_null.is_empty() {
311 return InferredType::Text; // All null → TEXT
312 }
313
314 let first = non_null[0];
315 let all_same = non_null.iter().all(|t| *t == first);
316 if all_same {
317 return first.clone();
318 }
319
320 // Numeric widening: Int -> BigInt -> Double
321 let all_numeric = non_null.iter().all(|t| {
322 matches!(
323 t,
324 InferredType::Int | InferredType::BigInt | InferredType::Double
325 )
326 });
327 if all_numeric {
328 if non_null.iter().any(|t| **t == InferredType::Double) {
329 return InferredType::Double;
330 }
331 if non_null.iter().any(|t| **t == InferredType::BigInt) {
332 return InferredType::BigInt;
333 }
334 return InferredType::Int;
335 }
336
337 // Mixed types → TEXT
338 InferredType::Text
339}
340
341/// Convert an [`InferredType`] to the Hyper SQL type name used in DDL.
342fn inferred_to_hyper_name(t: &InferredType) -> String {
343 match t {
344 InferredType::Null | InferredType::Text => "TEXT".into(),
345 InferredType::Bool => "BOOL".into(),
346 InferredType::Int => "INT".into(),
347 InferredType::BigInt => "BIGINT".into(),
348 InferredType::Double => "DOUBLE PRECISION".into(),
349 InferredType::Date => "DATE".into(),
350 InferredType::Timestamp => "TIMESTAMP".into(),
351 }
352}
353
354// --- Tier 3: CSV Schema Inference ---
355
356/// Infer a [`ColumnSchema`] for each CSV column (Tier 3).
357///
358/// When `has_header` is true, the first row provides column names; otherwise
359/// columns are named `col_0`, `col_1`, etc. Up to 1 000 data rows are sampled
360/// to determine types. All CSV columns are marked nullable because CSV has no
361/// way to express a NOT NULL constraint.
362///
363/// # Errors
364///
365/// - Returns [`ErrorCode::SchemaMismatch`] when the CSV header line or
366/// any sampled record cannot be parsed.
367/// - Returns [`ErrorCode::EmptyData`] when there are no data rows (to
368/// infer column count in the headerless case) or when the header row
369/// is empty.
370pub fn infer_csv_schema(csv_text: &str, has_header: bool) -> Result<Vec<ColumnSchema>, McpError> {
371 let mut reader = csv::ReaderBuilder::new()
372 .has_headers(has_header)
373 .from_reader(csv_text.as_bytes());
374
375 let headers: Vec<String> = if has_header {
376 reader
377 .headers()
378 .map_err(|e| {
379 McpError::new(ErrorCode::SchemaMismatch, format!("CSV header error: {e}"))
380 })?
381 .iter()
382 .map(std::string::ToString::to_string)
383 .collect()
384 } else {
385 // Peek at first record to get column count
386 let first = reader.records().next();
387 match first {
388 Some(Ok(ref rec)) => (0..rec.len()).map(|i| format!("col_{i}")).collect(),
389 _ => return Err(McpError::new(ErrorCode::EmptyData, "CSV has no data rows")),
390 }
391 };
392
393 if headers.is_empty() {
394 return Err(McpError::new(ErrorCode::EmptyData, "CSV has no columns"));
395 }
396
397 let num_cols = headers.len();
398 let mut col_types: Vec<Vec<InferredType>> = vec![Vec::new(); num_cols];
399
400 // Re-read from start for sampling (up to 1000 rows)
401 let mut sample_reader = csv::ReaderBuilder::new()
402 .has_headers(has_header)
403 .from_reader(csv_text.as_bytes());
404
405 for (row_idx, result) in sample_reader.records().enumerate() {
406 if row_idx >= 1000 {
407 break;
408 }
409 let record = result.map_err(|e| {
410 McpError::new(
411 ErrorCode::SchemaMismatch,
412 format!("CSV parse error at row {}: {e}", row_idx + 1),
413 )
414 })?;
415 for (col_idx, field) in record.iter().enumerate() {
416 if col_idx < num_cols {
417 col_types[col_idx].push(infer_csv_field_type(field));
418 }
419 }
420 }
421
422 let columns: Vec<ColumnSchema> = headers
423 .into_iter()
424 .enumerate()
425 .map(|(i, name)| {
426 let resolved = resolve_type(&col_types[i]);
427 ColumnSchema {
428 name,
429 hyper_type: inferred_to_hyper_name(&resolved),
430 nullable: true, // CSV columns are always nullable
431 }
432 })
433 .collect();
434
435 Ok(columns)
436}
437
438/// Second-pass streaming widen: re-read the given CSV source and, for columns
439/// the first-pass inference classified as `INT`, `BIGINT`, or `DOUBLE PRECISION`,
440/// promote the type if a value outside its current range appears anywhere in
441/// the file (not just the first 1 000 rows).
442///
443/// The first pass handles column naming and ambiguous-type resolution; this
444/// pass exists specifically to catch "big value hidden near the end of a CSV"
445/// — the exact bug that prompted this code path, where OWID keeps world-
446/// aggregate populations (~8 billion) in the last rows of a file whose first
447/// thousand rows only contain country-sized numbers.
448///
449/// Promotion rules per numeric column:
450/// * `INT` → `BIGINT` if any value exceeds `i32` range.
451/// * `INT` / `BIGINT` → `NUMERIC(38,0)` if any value exceeds `i64` range.
452/// * `INT` / `BIGINT` → `DOUBLE PRECISION` if any value contains a decimal point
453/// or exponent (mixed integer/float column).
454///
455/// Columns with non-numeric inferred types are left untouched. Nullability is
456/// preserved. Empty fields are ignored.
457///
458/// # Errors
459///
460/// Returns [`ErrorCode::SchemaMismatch`] when the CSV parser fails on
461/// any row (typically unbalanced quotes, mismatched delimiters, or
462/// non-UTF-8 bytes reported by the `csv` crate).
463///
464/// # Panics
465///
466/// Does not panic in practice. The `stats.get_mut(&col_idx).expect("preallocated")`
467/// invariant holds because `stats` is preallocated with one entry per
468/// `candidate_idxs` value, and the loop only uses indices from that
469/// same slice.
470pub fn widen_csv_numeric_columns<R: std::io::Read>(
471 reader: R,
472 has_header: bool,
473 columns: &mut [ColumnSchema],
474) -> Result<(), McpError> {
475 // Only bother if at least one column is a candidate for widening.
476 let candidate_idxs: Vec<usize> = columns
477 .iter()
478 .enumerate()
479 .filter(|(_, c)| {
480 matches!(
481 c.hyper_type.as_str(),
482 "INT" | "INTEGER" | "BIGINT" | "DOUBLE PRECISION"
483 )
484 })
485 .map(|(i, _)| i)
486 .collect();
487 if candidate_idxs.is_empty() {
488 return Ok(());
489 }
490
491 // Per-column observed-value state. `min`/`max` track integer extrema as
492 // `i128` to cover everything up to `i64::MIN`..`i64::MAX` plus headroom for
493 // overflow detection. `has_decimal` flips true when any field looks like a
494 // float (contains `.`, `e`, or `E`) so we can promote to DOUBLE.
495 #[derive(Default)]
496 struct ColStats {
497 min: Option<i128>,
498 max: Option<i128>,
499 has_decimal: bool,
500 overflow_i128: bool,
501 }
502 let mut stats: std::collections::HashMap<usize, ColStats> = candidate_idxs
503 .iter()
504 .map(|i| (*i, ColStats::default()))
505 .collect();
506
507 let mut rdr = csv::ReaderBuilder::new()
508 .has_headers(has_header)
509 .from_reader(reader);
510
511 for (row_idx, result) in rdr.records().enumerate() {
512 let record = result.map_err(|e| {
513 McpError::new(
514 ErrorCode::SchemaMismatch,
515 format!("CSV parse error at row {}: {e}", row_idx + 1),
516 )
517 })?;
518 for &col_idx in &candidate_idxs {
519 let Some(field) = record.get(col_idx) else {
520 continue;
521 };
522 let trimmed = field.trim();
523 if trimmed.is_empty()
524 || trimmed.eq_ignore_ascii_case("null")
525 || trimmed.eq_ignore_ascii_case("na")
526 {
527 continue;
528 }
529 let s = stats.get_mut(&col_idx).expect("preallocated");
530 if trimmed.contains('.') || trimmed.contains('e') || trimmed.contains('E') {
531 s.has_decimal = true;
532 continue;
533 }
534 match trimmed.parse::<i128>() {
535 Ok(n) => {
536 s.min = Some(s.min.map_or(n, |m| m.min(n)));
537 s.max = Some(s.max.map_or(n, |m| m.max(n)));
538 }
539 Err(_) => {
540 // Non-decimal, non-integer value in a numeric column. Leave
541 // widening to the first-pass classifier (it will already
542 // have picked TEXT if this happened inside the sample) and
543 // just skip. Any truly ambiguous column collapses there.
544 s.overflow_i128 = true;
545 }
546 }
547 }
548 }
549
550 for (&col_idx, s) in &stats {
551 let col = &mut columns[col_idx];
552 let i32_range = i128::from(i32::MIN)..=i128::from(i32::MAX);
553 let i64_range = i128::from(i64::MIN)..=i128::from(i64::MAX);
554 match col.hyper_type.as_str() {
555 "INT" | "INTEGER" => {
556 if s.has_decimal {
557 col.hyper_type = "DOUBLE PRECISION".into();
558 } else if s.overflow_i128
559 || !s.min.map_or(true, |m| i64_range.contains(&m))
560 || !s.max.map_or(true, |m| i64_range.contains(&m))
561 {
562 col.hyper_type = "NUMERIC(38,0)".into();
563 } else if !s.min.map_or(true, |m| i32_range.contains(&m))
564 || !s.max.map_or(true, |m| i32_range.contains(&m))
565 {
566 col.hyper_type = "BIGINT".into();
567 }
568 }
569 "BIGINT" => {
570 if s.has_decimal {
571 col.hyper_type = "DOUBLE PRECISION".into();
572 } else if s.overflow_i128
573 || !s.min.map_or(true, |m| i64_range.contains(&m))
574 || !s.max.map_or(true, |m| i64_range.contains(&m))
575 {
576 col.hyper_type = "NUMERIC(38,0)".into();
577 }
578 }
579 _ => {}
580 }
581 }
582 Ok(())
583}
584
585/// Classify a single CSV field value. Empty strings, `"null"`, and `"NA"`
586/// (case-insensitive) are treated as null. Boolean literals, integers, floats,
587/// and ISO date/timestamp patterns are recognized before falling back to TEXT.
588fn infer_csv_field_type(field: &str) -> InferredType {
589 let trimmed = field.trim();
590 if trimmed.is_empty()
591 || trimmed.eq_ignore_ascii_case("null")
592 || trimmed.eq_ignore_ascii_case("na")
593 {
594 return InferredType::Null;
595 }
596 if trimmed.eq_ignore_ascii_case("true") || trimmed.eq_ignore_ascii_case("false") {
597 return InferredType::Bool;
598 }
599 if let Ok(i) = trimmed.parse::<i64>() {
600 if i32::try_from(i).is_ok() {
601 return InferredType::Int;
602 }
603 return InferredType::BigInt;
604 }
605 if trimmed.parse::<f64>().is_ok() {
606 return InferredType::Double;
607 }
608 infer_string_type(trimmed)
609}
610
611/// Parse a user-provided schema override (`{"column_name": "TYPE", ...}`) into
612/// a `Vec<ColumnSchema>`. Validates each type name against [`map_hyper_type`] and
613/// rejects unknown types early. All override columns are marked nullable.
614///
615/// This variant treats the override as a complete schema: it becomes the full
616/// `Vec<ColumnSchema>` in whatever order the JSON object iterates. Used mostly
617/// by tests and code paths without an inferred schema to merge onto; prefer
618/// [`apply_schema_override`] for ingest paths so columns stay aligned with the
619/// source file's header order.
620///
621/// # Errors
622///
623/// - Returns [`ErrorCode::SchemaMismatch`] if any value in `schema` is
624/// not a string.
625/// - Returns [`ErrorCode::SchemaMismatch`] if any type name does not
626/// resolve via [`map_hyper_type`].
627pub fn parse_schema_override(
628 schema: &serde_json::Map<String, Value>,
629) -> Result<Vec<ColumnSchema>, McpError> {
630 let mut columns = Vec::new();
631 for (name, type_val) in schema {
632 let type_name = type_val.as_str().ok_or_else(|| {
633 McpError::new(
634 ErrorCode::SchemaMismatch,
635 format!("Schema type for '{name}' must be a string"),
636 )
637 })?;
638 if map_hyper_type(type_name).is_none() {
639 return Err(McpError::new(
640 ErrorCode::SchemaMismatch,
641 format!("Unknown type '{type_name}' for column '{name}'"),
642 ));
643 }
644 columns.push(ColumnSchema {
645 name: name.clone(),
646 hyper_type: type_name.to_uppercase(),
647 nullable: true,
648 });
649 }
650 Ok(columns)
651}
652
653/// Normalize a raw MCP `schema` parameter value into the column-name → type
654/// map expected by [`apply_schema_override`] and [`parse_schema_override`].
655///
656/// The MCP tool parameter is declared as `Option<serde_json::Value>` so the
657/// `rmcp` / `schemars` pipeline emits a permissive `true` JSON Schema. In
658/// practice some MCP clients forward this field as a **JSON-encoded string**
659/// rather than a raw JSON object — e.g. Windsurf/Cascade serializes
660/// `{"postal_code": "TEXT"}` as `"\"{\\\"postal_code\\\": \\\"TEXT\\\"}\""`.
661/// If we only accepted `Value::Object` (the old `v.as_object().cloned()`
662/// pattern) the override was silently dropped and ingest would fail with a
663/// confusing `22P02 invalid input syntax` error from hyperd when a column that
664/// the user explicitly wanted TEXT stayed INT.
665///
666/// Accepted shapes:
667///
668/// * `None` / `Some(Value::Null)` — no override.
669/// * `Some(Value::Object(m))` — used directly.
670/// * `Some(Value::String(s))` — `s` is parsed as JSON; must decode to an
671/// object. A non-object payload (array, number, etc.) is rejected with
672/// `SchemaMismatch` so the caller gets a clear error rather than a silent
673/// no-op.
674///
675/// Any other shape is rejected with `SchemaMismatch` for the same reason.
676///
677/// # Errors
678///
679/// - Returns [`ErrorCode::SchemaMismatch`] if a `Value::String` payload
680/// is non-empty but not valid JSON, or if it decodes to anything
681/// other than an object or null.
682/// - Returns [`ErrorCode::SchemaMismatch`] for any other `Value` shape
683/// (boolean, number, array, etc.).
684pub fn normalize_schema_param(
685 schema: Option<&Value>,
686) -> Result<Option<serde_json::Map<String, Value>>, McpError> {
687 let Some(v) = schema else {
688 return Ok(None);
689 };
690 match v {
691 Value::Null => Ok(None),
692 Value::Object(m) => Ok(Some(m.clone())),
693 Value::String(s) => {
694 let trimmed = s.trim();
695 if trimmed.is_empty() {
696 return Ok(None);
697 }
698 let parsed: Value = serde_json::from_str(trimmed).map_err(|e| {
699 McpError::new(
700 ErrorCode::SchemaMismatch,
701 format!(
702 "`schema` parameter is a string but not valid JSON: {e}. \
703 Expected an object like {{\"col\": \"TEXT\"}}."
704 ),
705 )
706 })?;
707 match parsed {
708 Value::Object(m) => Ok(Some(m)),
709 Value::Null => Ok(None),
710 other => Err(McpError::new(
711 ErrorCode::SchemaMismatch,
712 format!(
713 "`schema` parameter must be a JSON object mapping column names \
714 to type strings, got {}.",
715 json_type_name(&other)
716 ),
717 )),
718 }
719 }
720 other => Err(McpError::new(
721 ErrorCode::SchemaMismatch,
722 format!(
723 "`schema` parameter must be a JSON object mapping column names to type \
724 strings, got {}.",
725 json_type_name(other)
726 ),
727 )),
728 }
729}
730
731/// Short human-readable name of a JSON value's kind for error messages.
732pub(crate) fn json_type_name(v: &Value) -> &'static str {
733 match v {
734 Value::Null => "null",
735 Value::Bool(_) => "boolean",
736 Value::Number(_) => "number",
737 Value::String(_) => "string",
738 Value::Array(_) => "array",
739 Value::Object(_) => "object",
740 }
741}
742
743/// Overlay a user-provided override (`{"column_name": "TYPE", ...}`) on top of
744/// an already-inferred column list, preserving the inferred column **order** and
745/// replacing only the types listed in the override.
746///
747/// This is the semantics used by all ingest paths (`load_file`, `load_data`,
748/// `query_file`, `query_data`): the source file's header determines column order
749/// and the set of columns; the override is a partial name→type dictionary that
750/// lets callers force a wider type (e.g. `{"Population": "BIGINT"}`) without
751/// enumerating every column.
752///
753/// Unknown override keys are rejected with a `SchemaMismatch` error that lists
754/// the real column names so LLMs can self-correct without a round-trip.
755///
756/// # Errors
757///
758/// Returns [`ErrorCode::SchemaMismatch`] when:
759/// - Any override value is not a string.
760/// - Any override type name fails [`map_hyper_type`] resolution.
761/// - An override key names a column that is not present in `inferred`
762/// (the error lists the real column names).
763pub fn apply_schema_override(
764 mut inferred: Vec<ColumnSchema>,
765 override_map: &serde_json::Map<String, Value>,
766) -> Result<Vec<ColumnSchema>, McpError> {
767 // Validate: every override key must match a real inferred column and every
768 // override value must be a known type string.
769 let known: std::collections::HashSet<&str> = inferred.iter().map(|c| c.name.as_str()).collect();
770 for (name, type_val) in override_map {
771 if !known.contains(name.as_str()) {
772 let real: Vec<&str> = inferred.iter().map(|c| c.name.as_str()).collect();
773 return Err(McpError::new(
774 ErrorCode::SchemaMismatch,
775 format!("Override key '{name}' does not match any column. Known columns: {real:?}"),
776 ));
777 }
778 let type_name = type_val.as_str().ok_or_else(|| {
779 McpError::new(
780 ErrorCode::SchemaMismatch,
781 format!("Schema type for '{name}' must be a string"),
782 )
783 })?;
784 if map_hyper_type(type_name).is_none() {
785 return Err(McpError::new(
786 ErrorCode::SchemaMismatch,
787 format!("Unknown type '{type_name}' for column '{name}'"),
788 ));
789 }
790 }
791
792 // Apply: overlay types by column name. Column order is the inferred order.
793 for col in &mut inferred {
794 if let Some(v) = override_map.get(&col.name).and_then(|v| v.as_str()) {
795 col.hyper_type = v.trim().to_uppercase();
796 }
797 }
798 Ok(inferred)
799}