1use std::collections::HashMap;
142use std::sync::Arc;
143
144use serde_json::Value;
145use tokio::io::{AsyncReadExt, AsyncWriteExt};
146use tokio::net::{TcpListener, TcpStream};
147
148use crate::db::Db;
149
150const OID_BOOL: i32 = 16;
152const OID_INT8: i32 = 20;
153const OID_FLOAT8: i32 = 701;
154const OID_TEXT: i32 = 25;
155
156const PROTO_V3: i32 = 196_608; const SSL_REQUEST: i32 = 80_877_103;
158const GSS_REQUEST: i32 = 80_877_104;
159const CANCEL_REQUEST: i32 = 80_877_102;
160
161pub trait DbResolver: Send + Sync + 'static {
167 fn resolve(&self, name: &str) -> Option<Arc<Db>>;
175 fn token(&self) -> Option<String> {
177 None
178 }
179}
180
181struct Out(Vec<u8>);
184
185impl Out {
186 fn msg(tag: u8) -> Self {
187 Out(vec![tag, 0, 0, 0, 0])
189 }
190 fn i16(&mut self, v: i16) { self.0.extend_from_slice(&v.to_be_bytes()); }
191 fn i32(&mut self, v: i32) { self.0.extend_from_slice(&v.to_be_bytes()); }
192 fn cstr(&mut self, s: &str) {
193 self.0.extend_from_slice(s.replace('\0', "").as_bytes());
196 self.0.push(0);
197 }
198 fn bytes(&mut self, b: &[u8]) { self.0.extend_from_slice(b); }
199 fn finish(mut self) -> Vec<u8> {
201 let len = (self.0.len() - 1) as i32;
202 self.0[1..5].copy_from_slice(&len.to_be_bytes());
203 self.0
204 }
205}
206
207fn err_msg(code: &str, message: &str) -> Vec<u8> {
208 let mut m = Out::msg(b'E');
209 m.bytes(b"S"); m.cstr("ERROR");
210 m.bytes(b"C"); m.cstr(code);
211 m.bytes(b"M"); m.cstr(message);
212 m.0.push(0);
213 m.finish()
214}
215
216fn ready() -> Vec<u8> {
217 let mut m = Out::msg(b'Z');
218 m.bytes(b"I"); m.finish()
220}
221
222fn command_complete(tag: &str) -> Vec<u8> {
223 let mut m = Out::msg(b'C');
224 m.cstr(tag);
225 m.finish()
226}
227
228#[derive(Debug, PartialEq, Clone)]
238pub struct Col {
239 pub src: String,
240 pub out: String,
241}
242
243impl Col {
244 fn same(name: &str) -> Self {
245 Col { src: name.to_string(), out: name.to_string() }
246 }
247 fn renamed(src: &str, out: &str) -> Self {
248 Col { src: src.to_string(), out: out.to_string() }
249 }
250}
251
252#[derive(Debug, PartialEq)]
270pub enum Stmt {
271 Query { nql: String, project: Vec<Col> },
273 Insert { coll: String, rows: Vec<InsertRow>, returning: Vec<Col> },
275 Update { coll: String, set: Vec<(String, Value)>, nql: String, returning: Vec<Col> },
277 Delete { coll: String, nql: String, returning: Vec<Col> },
279 Canned { cols: Vec<String>, row: Vec<String> },
281 Ok(&'static str),
283}
284
285#[derive(Debug, PartialEq, Clone)]
288pub struct InsertRow {
289 pub id: Option<String>,
291 pub doc: serde_json::Map<String, Value>,
292 pub caused_by: Vec<String>,
295 pub valid_from: Option<String>,
296 pub valid_to: Option<String>,
297}
298
299fn normalise(sql: &str) -> String {
302 let mut out = String::with_capacity(sql.len());
303 let mut chars = sql.chars().peekable();
304 let mut in_s = false;
305 while let Some(c) = chars.next() {
306 if in_s {
307 out.push(c);
308 if c == '\'' { in_s = false; }
309 continue;
310 }
311 match c {
312 '\'' => { in_s = true; out.push(c); }
313 '-' if chars.peek() == Some(&'-') => {
314 for n in chars.by_ref() { if n == '\n' { break; } }
316 out.push(' ');
317 }
318 '/' if chars.peek() == Some(&'*') => {
319 chars.next();
320 let mut prev = ' ';
321 while let Some(n) = chars.next() {
322 if prev == '*' && n == '/' { break; }
323 prev = n;
324 }
325 out.push(' ');
326 }
327 _ => out.push(c),
328 }
329 }
330 out.split_whitespace().collect::<Vec<_>>().join(" ")
331}
332
333fn strip_column_qualifiers(
370 tail: &str,
371 coll: &str,
372 alias: Option<&str>,
373) -> Result<String, String> {
374 let bare = coll.rsplit('.').next().unwrap_or(coll);
375 let b: Vec<char> = tail.chars().collect();
376 let mut out = String::with_capacity(tail.len());
377 let mut i = 0usize;
378 let ident_start = |c: char| c.is_alphabetic() || c == '_';
379 let ident_char = |c: char| c.is_alphanumeric() || c == '_';
380
381 while i < b.len() {
382 if b[i] == '\'' {
384 out.push(b[i]);
385 i += 1;
386 while i < b.len() {
387 out.push(b[i]);
388 if b[i] == '\'' {
389 if b.get(i + 1) == Some(&'\'') {
391 out.push('\'');
392 i += 2;
393 continue;
394 }
395 i += 1;
396 break;
397 }
398 i += 1;
399 }
400 continue;
401 }
402 if b[i] == '"' {
408 out.push(b[i]);
409 i += 1;
410 while i < b.len() {
411 out.push(b[i]);
412 if b[i] == '"' { i += 1; break; }
413 i += 1;
414 }
415 continue;
416 }
417 if !ident_start(b[i]) {
418 out.push(b[i]);
421 i += 1;
422 continue;
423 }
424
425 let start = i;
426 while i < b.len() && ident_char(b[i]) {
427 i += 1;
428 }
429 let word: String = b[start..i].iter().collect();
430
431 if b.get(i) == Some(&'.') && b.get(i + 1).is_some_and(|c| ident_start(*c)) {
433 let fstart = i + 1;
434 let mut j = fstart;
435 while j < b.len() && ident_char(b[j]) {
436 j += 1;
437 }
438 let field: String = b[fstart..j].iter().collect();
439 let is_call = b[j..].iter().find(|c| !c.is_whitespace()) == Some(&'(');
444 if is_call {
445 out.push_str(&word);
446 out.push('.');
447 out.push_str(&field);
448 i = j;
449 continue;
450 }
451 let matches_alias = alias.is_some_and(|a| word.eq_ignore_ascii_case(a));
452 if matches_alias || word.eq_ignore_ascii_case(bare) || word.eq_ignore_ascii_case(coll) {
453 out.push_str(&field);
454 i = j;
455 continue;
456 }
457 return Err(format!(
458 "no table or alias named {:?} in this query — this statement reads \
459 {:?}{}, and a qualifier naming anything else would have to be \
460 answered from a relation that is not in its FROM clause",
461 word,
462 bare,
463 alias.map(|a| format!(" (aliased {:?})", a)).unwrap_or_default()
464 ));
465 }
466 out.push_str(&word);
467 }
468 Ok(out)
469}
470
471fn flatten_count_of_subquery(projection: &str, rest: &str) -> Option<String> {
479 let (outer_expr, outer_alias) = split_output_alias(projection.trim());
483 let ou = outer_expr.to_uppercase().replace(' ', "");
484 if ou != "COUNT(*)" {
485 return None;
486 }
487
488 let b: Vec<char> = rest.chars().collect();
491 let mut depth = 0i32;
492 let mut in_s = false;
493 let mut end = None;
494 for (i, &c) in b.iter().enumerate() {
495 match c {
496 '\'' => in_s = !in_s,
497 '(' if !in_s => depth += 1,
498 ')' if !in_s => {
499 depth -= 1;
500 if depth == 0 {
501 end = Some(i);
502 break;
503 }
504 }
505 _ => {}
506 }
507 }
508 let end = end?;
509 let inner = b[1..end].iter().collect::<String>().trim().to_string();
510
511 let trailing = b[end + 1..].iter().collect::<String>();
514 let (_alias, after) = split_table_alias(trailing.trim());
515 if !after.trim().is_empty() {
516 return None;
517 }
518
519 let iu = inner.to_uppercase();
520 if !iu.starts_with("SELECT") {
521 return None;
522 }
523 for kw in ["LIMIT", "OFFSET", "GROUP BY", "HAVING", "UNION", "INTERSECT", "EXCEPT", "JOIN"] {
525 if find_kw(&iu, kw).is_some() {
526 return None;
527 }
528 }
529 if find_kw(&iu, "DISTINCT").is_some() {
530 return None;
531 }
532 let inner_from = find_kw(&iu, "FROM")?;
534 let inner_list = inner[..inner_from].to_uppercase();
535 for agg in ["COUNT(", "SUM(", "AVG(", "MIN(", "MAX(", "ARRAY_AGG(", "STRING_AGG("] {
536 if inner_list.contains(agg) {
537 return None;
538 }
539 }
540 let inner_rest = inner[inner_from + 4..].trim();
542 if inner_rest.starts_with('(') {
543 return None;
544 }
545
546 let mut tail = inner_rest.to_string();
548 let tu = tail.to_uppercase();
549 if let Some(ob) = find_kw(&tu, "ORDER BY") {
550 tail = tail[..ob].trim_end().to_string();
551 }
552 Some(format!(
553 "SELECT count(*){} FROM {}",
554 outer_alias.map(|a| format!(" AS {}", a)).unwrap_or_default(),
555 tail
556 ))
557}
558
559const CLAUSE_WORDS: &[&str] = &[
564 "WHERE", "GROUP", "ORDER", "LIMIT", "OFFSET", "HAVING", "FOR", "VALID",
565 "TRACE", "TRAVERSE", "SEARCH", "RETURNING", "UNION", "INTERSECT", "EXCEPT",
566 "JOIN", "LEFT", "RIGHT", "INNER", "FULL", "CROSS", "ON", "USING", "SET",
567];
568
569fn split_table_alias(tail: &str) -> (Option<String>, &str) {
580 let t = tail.trim_start();
581 let first_end = t.find(char::is_whitespace).unwrap_or(t.len());
582 let first = &t[..first_end];
583 let fu = first.to_uppercase();
584
585 if fu == "AS" {
586 let rest = t[first_end..].trim_start();
587 let end = rest.find(char::is_whitespace).unwrap_or(rest.len());
588 let word = &rest[..end];
589 if word.eq_ignore_ascii_case("OF") {
590 return (None, t); }
592 if word.is_empty() {
593 return (None, t);
594 }
595 return (Some(word.trim_matches('"').to_string()), rest[end..].trim_start());
596 }
597 if first.is_empty() || CLAUSE_WORDS.contains(&fu.as_str()) {
598 return (None, t);
599 }
600 if first.chars().next().is_some_and(|c| c.is_alphabetic() || c == '_' || c == '"') {
603 return (Some(first.trim_matches('"').to_string()), t[first_end..].trim_start());
604 }
605 (None, t)
606}
607
608fn split_top_level(s: &str, delim: char) -> Vec<String> {
613 let mut out = vec![];
614 let mut cur = String::new();
615 let mut depth = 0i32;
616 let mut in_s = false;
617 let mut in_d = false;
618 for c in s.chars() {
619 match c {
620 '\'' if !in_d => { in_s = !in_s; cur.push(c); }
621 '"' if !in_s => { in_d = !in_d; cur.push(c); }
622 '(' if !in_s && !in_d => { depth += 1; cur.push(c); }
623 ')' if !in_s && !in_d => { depth -= 1; cur.push(c); }
624 c if c == delim && depth == 0 && !in_s && !in_d => {
625 out.push(std::mem::take(&mut cur));
626 }
627 _ => cur.push(c),
628 }
629 }
630 out.push(cur);
631 out
632}
633
634fn split_output_alias(p: &str) -> (&str, Option<&str>) {
640 let pu = p.to_uppercase();
641 if let Some(at) = find_kw(&pu, "AS") {
642 let alias = p[at + 2..].trim().trim_matches('"');
643 if !alias.is_empty() {
644 return (p[..at].trim(), Some(alias));
645 }
646 }
647 let b: Vec<char> = p.chars().collect();
651 let mut depth = 0i32;
652 let mut in_s = false;
653 let mut cut = None;
654 for (i, &c) in b.iter().enumerate() {
655 match c {
656 '\'' => in_s = !in_s,
657 '(' if !in_s => depth += 1,
658 ')' if !in_s => depth -= 1,
659 c if c.is_whitespace() && depth == 0 && !in_s => cut = Some(i),
660 _ => {}
661 }
662 }
663 match cut {
664 Some(i) => {
665 let alias = p[i..].trim().trim_matches('"');
666 if alias.is_empty() { (p, None) } else { (p[..i].trim(), Some(alias)) }
667 }
668 None => (p, None),
669 }
670}
671
672fn nql_string(s: &str) -> String {
679 format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
680}
681
682fn sql_literals_to_nql(s: &str) -> String {
683 let mut out = String::with_capacity(s.len());
684 let mut it = s.chars().peekable();
685 while let Some(c) = it.next() {
686 match c {
687 '\'' => {
688 out.push('"');
689 while let Some(ch) = it.next() {
690 if ch == '\'' {
691 if it.peek() == Some(&'\'') {
692 it.next();
693 out.push('\''); } else {
695 break;
696 }
697 } else if ch == '"' {
698 out.push('\\');
701 out.push('"');
702 } else {
703 out.push(ch);
704 }
705 }
706 out.push('"');
707 }
708 '<' if it.peek() == Some(&'>') => { it.next(); out.push_str("!="); }
709 _ => out.push(c),
710 }
711 }
712 out
713}
714
715fn strip_prefix_ci(s: &str, prefix: &str) -> Option<String> {
716 if s.len() >= prefix.len() && s[..prefix.len()].eq_ignore_ascii_case(prefix) {
717 Some(s[prefix.len()..].trim_start().to_string())
718 } else {
719 None
720 }
721}
722
723fn find_kw(s: &str, kw: &str) -> Option<usize> {
726 let bytes = s.as_bytes();
727 let k = kw.as_bytes();
728 let mut depth = 0i32;
729 let mut in_s = false;
730 let mut in_d = false;
731 let mut i = 0usize;
732 while i < bytes.len() {
733 let c = bytes[i];
734 if in_s { if c == b'\'' { in_s = false; } i += 1; continue; }
735 if in_d { if c == b'"' { in_d = false; } i += 1; continue; }
736 match c {
737 b'\'' => { in_s = true; i += 1; continue; }
738 b'"' => { in_d = true; i += 1; continue; }
739 b'(' => { depth += 1; i += 1; continue; }
740 b')' => { depth -= 1; i += 1; continue; }
741 _ => {}
742 }
743 if depth == 0 && i + k.len() <= bytes.len()
744 && bytes[i..i + k.len()].eq_ignore_ascii_case(k)
745 {
746 let before_ok = i == 0 || !(bytes[i - 1] as char).is_alphanumeric() && bytes[i - 1] != b'_';
747 let after = i + k.len();
748 let after_ok = after >= bytes.len()
749 || !(bytes[after] as char).is_alphanumeric() && bytes[after] != b'_';
750 if before_ok && after_ok {
751 return Some(i);
752 }
753 }
754 i += 1;
755 }
756 None
757}
758
759fn split_top(s: &str, sep: char) -> Vec<String> {
763 let mut out = vec![];
764 let mut cur = String::new();
765 let mut depth = 0i32;
766 let mut in_s = false;
767 let mut it = s.chars().peekable();
768 while let Some(c) = it.next() {
769 if in_s {
770 cur.push(c);
771 if c == '\'' {
772 if it.peek() == Some(&'\'') { cur.push(it.next().unwrap()); } else { in_s = false; }
774 }
775 continue;
776 }
777 match c {
778 '\'' => { in_s = true; cur.push(c); }
779 '(' => { depth += 1; cur.push(c); }
780 ')' => { depth -= 1; cur.push(c); }
781 x if x == sep && depth == 0 => { out.push(cur.trim().to_string()); cur.clear(); }
782 _ => cur.push(c),
783 }
784 }
785 if !cur.trim().is_empty() { out.push(cur.trim().to_string()); }
786 out
787}
788
789fn sql_value(raw: &str) -> Result<Value, String> {
795 let t = raw.trim();
796 if t.is_empty() {
797 return Err("empty value".into());
798 }
799 let up = t.to_uppercase();
800 if up == "NULL" { return Ok(Value::Null); }
801 if up == "TRUE" { return Ok(Value::Bool(true)); }
802 if up == "FALSE" { return Ok(Value::Bool(false)); }
803 if t.starts_with('\'') && t.ends_with('\'') && t.len() >= 2 {
804 let inner = &t[1..t.len() - 1];
806 return Ok(Value::String(inner.replace("''", "'")));
807 }
808 if let Ok(i) = t.parse::<i64>() { return Ok(Value::from(i)); }
809 if let Ok(f) = t.parse::<f64>() { return Ok(Value::from(f)); }
810 Err(format!(
811 "cannot use {:?} as a value — this endpoint accepts string literals, \
812 numbers, TRUE/FALSE and NULL. Expressions, casts and function calls \
813 are not evaluated, because storing an unevaluated expression as text \
814 would be worse than refusing it", t))
815}
816
817fn split_returning(tail: &str) -> (String, Vec<Col>) {
819 let tu = tail.to_uppercase();
820 match find_kw(&tu, "RETURNING") {
821 None => (tail.to_string(), vec![]),
822 Some(at) => {
823 let head = tail[..at].trim().to_string();
824 let list = tail[at + "RETURNING".len()..].trim();
825 if list == "*" {
826 return (head, vec![]); }
828 let cols = split_top(list, ',')
829 .into_iter()
830 .map(|p| {
831 let raw = p.split_whitespace().next().unwrap_or(&p).to_string();
832 let name = raw.rsplit('.').next().unwrap_or(&raw).trim_matches('"').to_string();
833 Col::same(&name)
834 })
835 .collect();
836 (head, cols)
837 }
838 }
839}
840
841fn take_reserved(doc: &mut serde_json::Map<String, Value>) -> (Option<String>, Vec<String>, Option<String>, Option<String>) {
843 let id = doc.remove("_id").or_else(|| doc.remove("id"))
844 .and_then(|v| match v {
845 Value::String(s) => Some(s),
846 Value::Null => None,
847 other => Some(other.to_string()), });
849 let caused_by = match doc.remove("_caused_by") {
850 Some(Value::String(s)) => vec![s],
851 Some(Value::Array(a)) => a.into_iter()
852 .filter_map(|v| v.as_str().map(str::to_string)).collect(),
853 _ => vec![],
854 };
855 let vf = doc.remove("_valid_from").and_then(|v| v.as_str().map(str::to_string));
856 let vt = doc.remove("_valid_to").and_then(|v| v.as_str().map(str::to_string));
857 (id, caused_by, vf, vt)
858}
859
860fn translate_insert(sql: &str) -> Result<Stmt, String> {
862 let rest = strip_prefix_ci(sql, "INSERT")
863 .and_then(|r| strip_prefix_ci(&r, "INTO"))
864 .ok_or("expected INSERT INTO")?;
865 let ru = rest.to_uppercase();
869 let values_at = find_kw(&ru, "VALUES").ok_or(
870 "expected VALUES — `INSERT … SELECT` is not supported on this endpoint")?;
871 let head = rest[..values_at].trim().to_string();
872 let open = head.find('(').ok_or(
873 "INSERT needs an explicit column list — `INSERT INTO t (a, b) VALUES (…)`. \
874 NEDB is schemaless, so there is no declared column order to infer from")?;
875 let coll = head[..open].trim().trim_matches('"');
876 let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
877 if coll.is_empty() {
878 return Err("expected a collection name after INSERT INTO".into());
879 }
880 let close = head.rfind(')').ok_or("unterminated column list")?;
881 if close < open {
882 return Err("malformed column list".into());
883 }
884 let tail_from_values = rest[values_at..].to_string();
885 let cols: Vec<String> = split_top(&head[open + 1..close], ',')
886 .into_iter()
887 .map(|c| c.trim().trim_matches('"').to_string())
888 .collect();
889 if cols.is_empty() {
890 return Err("the column list is empty".into());
891 }
892
893 let after = strip_prefix_ci(&tail_from_values, "VALUES")
894 .ok_or("expected VALUES after the column list")?;
895 let (values_part, returning) = split_returning(&after);
896
897 let mut rows = vec![];
898 for group in split_top(&values_part, ',') {
899 let g = group.trim();
900 if !(g.starts_with('(') && g.ends_with(')')) {
901 return Err(format!("expected a parenthesised row of values, got {:?}", g));
902 }
903 let vals = split_top(&g[1..g.len() - 1], ',');
904 if vals.len() != cols.len() {
905 return Err(format!(
906 "{} values for {} columns — every row must match the column list",
907 vals.len(), cols.len()));
908 }
909 let mut doc = serde_json::Map::new();
910 for (c, v) in cols.iter().zip(vals.iter()) {
911 doc.insert(c.clone(), sql_value(v)?);
912 }
913 let (id, caused_by, valid_from, valid_to) = take_reserved(&mut doc);
914 rows.push(InsertRow { id, doc, caused_by, valid_from, valid_to });
915 }
916 if rows.is_empty() {
917 return Err("INSERT with no rows".into());
918 }
919 Ok(Stmt::Insert { coll, rows, returning })
920}
921
922fn translate_update(sql: &str) -> Result<Stmt, String> {
924 let rest = strip_prefix_ci(sql, "UPDATE").ok_or("expected UPDATE")?;
925 let ru = rest.to_uppercase();
926 let set_at = find_kw(&ru, "SET").ok_or("expected SET in UPDATE")?;
927 let target = rest[..set_at].trim();
930 let mut parts = target.split_whitespace();
931 let coll = parts.next().unwrap_or("").trim_matches('"');
932 let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
933 let upd_alias: Option<String> = match parts.next() {
934 Some(w) if w.eq_ignore_ascii_case("AS") => {
935 parts.next().map(|a| a.trim_matches('"').to_string())
936 }
937 Some(w) => Some(w.trim_matches('"').to_string()),
938 None => None,
939 };
940 if coll.is_empty() {
941 return Err("expected a collection name after UPDATE".into());
942 }
943 let after_set = rest[set_at + 3..].trim().to_string();
944 let (after_set, returning) = split_returning(&after_set);
945
946 let au = after_set.to_uppercase();
948 let (assigns_raw, where_raw) = match find_kw(&au, "WHERE") {
949 Some(at) => (after_set[..at].to_string(), after_set[at..].to_string()),
950 None => (after_set.clone(), String::new()),
951 };
952
953 let mut set = vec![];
954 for a in split_top(&assigns_raw, ',') {
955 let eq = a.find('=').ok_or(format!("expected `col = value` in SET, got {:?}", a))?;
956 let col = a[..eq].trim().trim_matches('"').to_string();
957 if col.is_empty() {
958 return Err("empty column name in SET".into());
959 }
960 set.push((col, sql_value(&a[eq + 1..])?));
961 }
962 if set.is_empty() {
963 return Err("UPDATE with no assignments".into());
964 }
965 let where_raw = strip_column_qualifiers(where_raw.trim(), &coll, upd_alias.as_deref())?;
968 let nql = format!("FROM {} {}", coll, sql_literals_to_nql(&where_raw))
969 .trim().to_string();
970 Ok(Stmt::Update { coll, set, nql, returning })
971}
972
973fn translate_delete(sql: &str) -> Result<Stmt, String> {
975 let rest = strip_prefix_ci(sql, "DELETE")
976 .and_then(|r| strip_prefix_ci(&r, "FROM"))
977 .ok_or("expected DELETE FROM")?;
978 let (rest, returning) = split_returning(&rest);
979 let end = rest.find(' ').unwrap_or(rest.len());
980 let coll = rest[..end].trim().trim_matches('"');
981 let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
982 if coll.is_empty() {
983 return Err("expected a collection name after DELETE FROM".into());
984 }
985 let (del_alias, where_raw) = split_table_alias(rest[end..].trim());
986 let where_raw = strip_column_qualifiers(where_raw, &coll, del_alias.as_deref())?;
987 let nql = format!("FROM {} {}", coll, sql_literals_to_nql(&where_raw))
988 .trim().to_string();
989 Ok(Stmt::Delete { coll, nql, returning })
990}
991
992pub fn translate(sql_raw: &str) -> Result<Stmt, String> {
994 let sql = normalise(sql_raw);
995 let sql = sql.trim().trim_end_matches(';').trim();
996 if sql.is_empty() {
997 return Ok(Stmt::Ok(""));
998 }
999 let upper = sql.to_uppercase();
1000
1001 if upper.starts_with("SET ") || upper.starts_with("BEGIN") || upper.starts_with("COMMIT")
1006 || upper.starts_with("ROLLBACK") || upper.starts_with("DISCARD")
1007 || upper.starts_with("LISTEN ") || upper.starts_with("UNLISTEN ")
1008 {
1009 return Ok(Stmt::Ok(if upper.starts_with("SET") { "SET" } else { "OK" }));
1011 }
1012 if upper.starts_with("SHOW ") {
1013 let name = sql[5..].trim().to_lowercase();
1014 let val = match name.as_str() {
1015 "transaction_isolation" | "default_transaction_isolation" => "read committed",
1016 "server_version" => SERVER_VERSION,
1017 "server_encoding" | "client_encoding" => "UTF8",
1018 "standard_conforming_strings" => "on",
1019 "is_superuser" => "off",
1020 _ => "",
1021 };
1022 return Ok(Stmt::Canned { cols: vec![name], row: vec![val.to_string()] });
1023 }
1024 if upper == "SELECT VERSION()" {
1025 return Ok(Stmt::Canned {
1026 cols: vec!["version".into()],
1027 row: vec![full_version_string()],
1028 });
1029 }
1030 if upper == "SELECT 1" || upper == "SELECT 1;" {
1031 return Ok(Stmt::Canned { cols: vec!["?column?".into()], row: vec!["1".into()] });
1032 }
1033 if upper.starts_with("SELECT CURRENT_SCHEMA") {
1034 return Ok(Stmt::Canned { cols: vec!["current_schema".into()], row: vec!["public".into()] });
1035 }
1036 if upper.starts_with("SELECT CURRENT_DATABASE") {
1037 return Ok(Stmt::Canned { cols: vec!["current_database".into()], row: vec!["nedb".into()] });
1038 }
1039 if upper.starts_with("SELECT CURRENT_USER") || upper.starts_with("SELECT USER") {
1040 return Ok(Stmt::Canned { cols: vec!["current_user".into()], row: vec!["nedb".into()] });
1041 }
1042
1043 if upper.starts_with("INSERT") { return translate_insert(sql); }
1047 if upper.starts_with("UPDATE") { return translate_update(sql); }
1048 if upper.starts_with("DELETE") { return translate_delete(sql); }
1049
1050 for (kw, why) in [
1052 ("CREATE", "DDL is not supported — collections are created implicitly by the first write to them, because NEDB is schemaless"),
1053 ("ALTER", "DDL is not supported — there is no schema to alter"),
1054 ("DROP", "DDL is not supported; drop a database with DELETE /v1/databases/<db>"),
1055 ("TRUNCATE", "not supported, and not an oversight: NEDB is append-only so that history cannot be discarded. That is the product"),
1056 ("COPY", "not supported; use GET /v1/databases/<db>/since for bulk export"),
1057 ("GRANT", "there is no SQL-level privilege system; auth is the bearer token"),
1058 ("REVOKE", "there is no SQL-level privilege system; auth is the bearer token"),
1059 ] {
1060 if upper.starts_with(kw) {
1061 return Err(format!("{} is not supported — {}", kw, why));
1062 }
1063 }
1064 if !upper.starts_with("SELECT") {
1065 return Err(format!(
1066 "only SELECT, INSERT, UPDATE and DELETE are supported on the Postgres \
1067 endpoint (got {:?})",
1068 sql.split_whitespace().next().unwrap_or("")
1069 ));
1070 }
1071 for (kw, why) in [
1072 (" JOIN ", "JOIN is not supported — NQL is single-collection; join in your client or model the relation with LINK/TRAVERSE"),
1073 (" UNION ", "UNION is not supported"),
1074 (" INTERSECT ", "INTERSECT is not supported"),
1075 (" EXCEPT ", "EXCEPT is not supported"),
1076 (" OVER (", "window functions are not supported"),
1077 ("DISTINCT ", "DISTINCT is not supported — GROUP BY <col> gives the distinct values with counts"),
1078 ] {
1079 if upper.contains(kw) {
1080 return Err(why.to_string());
1081 }
1082 }
1083 if find_kw(&upper, "FROM").is_none() {
1084 return Err("SELECT without FROM is not supported on this endpoint".into());
1085 }
1086
1087 let after_select = strip_prefix_ci(sql, "SELECT").ok_or("expected SELECT")?;
1089 let from_at = find_kw(&after_select.to_uppercase(), "FROM")
1090 .ok_or("expected FROM after the select list")?;
1091 let projection = after_select[..from_at].trim().to_string();
1092 let rest = after_select[from_at + 4..].trim().to_string();
1093 if rest.is_empty() {
1094 return Err("expected a collection name after FROM".into());
1095 }
1096 if rest.starts_with('(') {
1115 if let Some(flat) = flatten_count_of_subquery(&projection, &rest) {
1116 return translate(&flat);
1120 }
1121 return Err("subqueries in FROM are not supported — except \
1122 `SELECT count(*) FROM (…)`, which is rewritten to a flat \
1123 count when the inner query has no LIMIT, OFFSET, DISTINCT, \
1124 GROUP BY or aggregate of its own (any of those would make the \
1125 two counts different numbers)".into());
1126 }
1127 let coll_end = rest.find(' ').unwrap_or(rest.len());
1128 let coll = &rest[..coll_end];
1129 if coll.contains(',') {
1130 return Err("selecting from more than one collection is not supported (no JOIN)".into());
1131 }
1132 let bare = coll.rsplit('.').next().unwrap_or(coll).trim_matches('"');
1139 let qualified = coll
1140 .split('.')
1141 .map(|p| p.trim_matches('"'))
1142 .collect::<Vec<_>>()
1143 .join(".");
1144 let coll = if qualified.starts_with("information_schema.") {
1145 qualified.as_str()
1146 } else {
1147 bare
1148 };
1149 let tail = rest[coll_end..].trim();
1150
1151 let mut agg_clause = String::new();
1166 let mut agg_srcs: Vec<String> = vec![];
1167 let mut project: Vec<Col> = vec![];
1168
1169 if projection == "*" {
1170 } else {
1172 for part in split_top_level(&projection, ',') {
1173 let p = part.trim();
1174 if p.is_empty() {
1175 return Err("empty column in the select list".into());
1176 }
1177 let (expr, alias) = split_output_alias(p);
1178 let eu = expr.to_uppercase();
1179
1180 if eu.starts_with("COUNT(") {
1183 if agg_clause.is_empty() {
1184 agg_clause = " COUNT".to_string();
1185 }
1186 agg_srcs.push("count".to_string());
1187 project.push(Col::renamed("count", alias.unwrap_or("count")));
1188 continue;
1189 }
1190 if let Some(agg) = ["SUM", "AVG", "MIN", "MAX"]
1191 .iter()
1192 .find(|a| eu.starts_with(&format!("{}(", a)))
1193 {
1194 let inner = expr[agg.len() + 1..].trim_end_matches(')').trim();
1195 if inner.is_empty() || inner == "*" {
1196 return Err(format!("{}() needs a column", agg));
1197 }
1198 let inner = inner.rsplit('.').next().unwrap_or(inner).trim_matches('"');
1199 let named = format!("{} {}", agg, inner);
1200 if !agg_clause.is_empty() && agg_clause.trim() != "COUNT" && agg_clause.trim() != named {
1201 return Err(format!(
1202 "only one of SUM/AVG/MIN/MAX is supported per statement \
1203 (already have {:?}, then {:?}) — NQL's grouped row carries \
1204 the group key, `count`, and ONE named aggregate",
1205 agg_clause.trim(), named));
1206 }
1207 agg_clause = format!(" {}", named);
1208 let src = format!("{}_{}", agg.to_lowercase(), inner);
1211 project.push(Col::renamed(&src, alias.unwrap_or(&agg.to_lowercase())));
1212 agg_srcs.push(src);
1213 continue;
1214 }
1215 let bare = expr.rsplit('.').next().unwrap_or(expr).trim_matches('"');
1223 let is_column = !bare.is_empty()
1224 && !bare.starts_with(|c: char| c.is_ascii_digit())
1225 && bare.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '$');
1226 if !is_column {
1227 return Err(format!(
1228 "expressions in the select list are not supported ({:?}) — \
1229 supported: *, a column list, COUNT(*), or SUM/AVG/MIN/MAX(col). \
1230 Compute it in your client, or read the column and map it there",
1231 p));
1232 }
1233 let name = bare;
1234 project.push(Col::renamed(name, alias.unwrap_or(name)));
1235 }
1236 }
1237
1238 let (alias, tail) = split_table_alias(tail);
1247 let mut tail = strip_column_qualifiers(tail, coll, alias.as_deref())?;
1248 let tu = tail.to_uppercase();
1249 if let Some(at) = find_kw(&tu, "AS OF SYSTEM TIME") {
1250 let before = tail[..at].to_string();
1251 let after = tail[at + "AS OF SYSTEM TIME".len()..].trim_start().to_string();
1252 let end = after.find(' ').unwrap_or(after.len());
1254 let seq = after[..end].trim().trim_matches('\'').trim_matches('"').to_string();
1255 if seq.parse::<u64>().is_err() {
1256 return Err(format!(
1257 "AS OF SYSTEM TIME takes a NEDB sequence number here, not a timestamp (got {:?}). \
1258 NEDB's history is sequence-addressed and never garbage-collected, so a seq is \
1259 exact where a wall-clock time would be approximate", seq));
1260 }
1261 tail = format!("{} AS OF {} {}", before.trim(), seq, after[end..].trim())
1262 .trim()
1263 .to_string();
1264 }
1265
1266 let tu_ord = tail.to_uppercase();
1279 if let Some(ob_at) = find_kw(&tu_ord, "ORDER BY") {
1280 let start = ob_at + "ORDER BY".len();
1281 let end = ["LIMIT", "OFFSET", "GROUP BY", "TRACE", "TRAVERSE", "SEARCH"]
1283 .iter()
1284 .filter_map(|k| find_kw(&tu_ord[start..], k).map(|at| start + at))
1285 .min()
1286 .unwrap_or(tail.len());
1287 let mut keys = vec![];
1288 for item in split_top_level(&tail[start..end], ',') {
1289 let item = item.trim();
1290 if item.is_empty() {
1291 continue;
1292 }
1293 let mut parts = item.split_whitespace();
1294 let first = parts.next().unwrap_or("");
1295 let rest: Vec<&str> = parts.collect();
1296 match first.parse::<usize>() {
1297 Ok(n) if n >= 1 => {
1298 let col = project.get(n - 1).ok_or_else(|| {
1299 if project.is_empty() {
1300 format!(
1301 "ORDER BY {} is a select-list POSITION, and `SELECT *` \
1302 has no list to index — name the column instead", n)
1303 } else {
1304 format!(
1305 "ORDER BY {} is out of range: the select list has {} \
1306 column(s)", n, project.len())
1307 }
1308 })?;
1309 keys.push(
1310 std::iter::once(col.src.as_str())
1311 .chain(rest.iter().copied())
1312 .collect::<Vec<_>>()
1313 .join(" "),
1314 );
1315 }
1316 _ => keys.push(item.to_string()),
1319 }
1320 }
1321 tail = format!("{} ORDER BY {} {}", &tail[..ob_at], keys.join(", "), &tail[end..])
1322 .split_whitespace()
1323 .collect::<Vec<_>>()
1324 .join(" ");
1325 }
1326
1327 let mut gkey: Option<String> = None;
1335 let tu_all = tail.to_uppercase();
1336 if let Some(gb_at) = find_kw(&tu_all, "GROUP BY") {
1337 let head = tail[..gb_at].trim_end().to_string();
1338 let after = tail[gb_at + "GROUP BY".len()..].trim_start();
1339 let key_end = after.find(|c: char| c == ' ' || c == ',').unwrap_or(after.len());
1340 let group_key = after[..key_end].trim().trim_matches('"').to_string();
1341 let after_key = after[key_end..].trim_start();
1342 gkey = Some(group_key.clone());
1343
1344 if after_key.starts_with(',') {
1348 return Err(format!(
1349 "GROUP BY takes one key here (got {:?} and more) — NQL groups by a \
1350 single field, and grouping by only the first would aggregate over \
1351 rows the query meant to keep apart",
1352 group_key));
1353 }
1354
1355 for c in &project {
1356 let ok = c.src == group_key
1357 || c.src == "count"
1358 || agg_srcs.contains(&c.src);
1359 if !ok {
1360 return Err(format!(
1361 "column {:?} must appear in the GROUP BY clause or be used in an \
1362 aggregate function — a grouped row carries the group key, `count`, \
1363 and the aggregate, nothing else",
1364 c.src));
1365 }
1366 }
1367
1368 tail = format!("{} GROUP BY {}{} {}", head, group_key, agg_clause, after_key)
1379 .split_whitespace()
1380 .collect::<Vec<_>>()
1381 .join(" ");
1382 agg_clause.clear();
1383 }
1384
1385 let tu_hav = tail.to_uppercase();
1400 if let Some(h_at) = find_kw(&tu_hav, "HAVING") {
1401 let start = h_at + "HAVING".len();
1402 let end = ["ORDER BY", "LIMIT", "OFFSET"]
1403 .iter()
1404 .filter_map(|k| find_kw(&tu_hav[start..], k).map(|at| start + at))
1405 .min()
1406 .unwrap_or(tail.len());
1407 let clause = tail[start..end].to_string();
1408 let lhs_end = clause
1410 .find(|c: char| "<>=!".contains(c))
1411 .unwrap_or(clause.len());
1412 let lhs = clause[..lhs_end].trim();
1413 if !lhs.is_empty() {
1414 let lu = lhs.to_uppercase();
1415 let is_count = lu == "COUNT" || lu.replace(' ', "") == "COUNT(*)"
1422 || project.iter().any(|c| c.out.eq_ignore_ascii_case(lhs) && c.src == "count");
1423 let mapped = if is_count {
1424 Some("count".to_string())
1425 } else {
1426 agg_srcs.iter().find(|s| s.eq_ignore_ascii_case(lhs)).cloned().or_else(|| {
1428 project.iter()
1429 .find(|c| c.out.eq_ignore_ascii_case(lhs) && agg_srcs.contains(&c.src))
1430 .map(|c| c.src.clone())
1431 })
1432 };
1433 match mapped {
1434 Some(m) => {
1435 let rewritten = format!("{} {}", m, clause[lhs_end..].trim());
1439 tail = format!("{} HAVING {} {}",
1440 tail[..h_at].trim(), rewritten.trim(), tail[end..].trim())
1441 .trim().to_string();
1442 }
1443 None if gkey.as_deref().map(|g| g.eq_ignore_ascii_case(lhs)) == Some(true) => {}
1444 None => {
1445 return Err(format!(
1446 "HAVING names {:?}, which this grouped row does not carry. \
1447 It has the group key{}{}. Filtering on anything else would \
1448 answer zero rows rather than report a mistake",
1449 lhs,
1450 gkey.as_deref().map(|g| format!(" ({:?})", g)).unwrap_or_default(),
1451 if agg_srcs.is_empty() { String::new() }
1452 else { format!(", plus {}", agg_srcs.join(", ")) }));
1453 }
1454 }
1455 }
1456 }
1457
1458
1459 let tail = sql_literals_to_nql(&tail);
1460 let nql = format!("FROM {}{}{}", coll,
1461 if agg_clause.is_empty() { String::new() } else { agg_clause },
1462 if tail.is_empty() { String::new() } else { format!(" {}", tail) });
1463
1464 Ok(Stmt::Query { nql: nql.trim().to_string(), project })
1465}
1466
1467const SERVER_VERSION: &str = "15.0";
1468
1469pub fn version_string() -> String {
1471 full_version_string()
1472}
1473
1474fn full_version_string() -> String {
1475 format!(
1476 "PostgreSQL {} (NEDB {}) — tamper-evident, append-only, permanent \
1477 history. SELECT + INSERT/UPDATE/DELETE; an UPDATE is a new version, \
1478 so prior values stay readable with AS OF SYSTEM TIME.",
1479 SERVER_VERSION,
1480 env!("CARGO_PKG_VERSION")
1481 )
1482}
1483
1484fn columns_for(rows: &[Value], project: &[Col]) -> Vec<Col> {
1492 if !project.is_empty() {
1493 return project.to_vec();
1494 }
1495 let mut plain: Vec<String> = vec![];
1496 let mut meta: Vec<String> = vec![];
1497 for r in rows {
1498 if let Value::Object(m) = r {
1499 for k in m.keys() {
1500 let target = if k.starts_with('_') { &mut meta } else { &mut plain };
1501 if !target.contains(k) {
1502 target.push(k.clone());
1503 }
1504 }
1505 }
1506 }
1507 meta.sort();
1514 plain.extend(meta);
1515 plain.into_iter().map(|k| Col::same(&k)).collect()
1516}
1517
1518fn oid_of_value(v: &Value) -> Option<i32> {
1520 match v {
1521 Value::Null => None,
1522 Value::Bool(_) => Some(OID_BOOL),
1523 Value::Number(n) => Some(if n.is_i64() || n.is_u64() { OID_INT8 } else { OID_FLOAT8 }),
1524 Value::String(_) => Some(OID_TEXT),
1525 _ => Some(OID_TEXT),
1527 }
1528}
1529
1530fn unify_oid(a: i32, b: i32) -> i32 {
1537 if a == b {
1538 return a;
1539 }
1540 match (a, b) {
1541 (OID_INT8, OID_FLOAT8) | (OID_FLOAT8, OID_INT8) => OID_FLOAT8,
1542 _ => OID_TEXT,
1543 }
1544}
1545
1546pub fn oid_for_column(rows: &[Value], col: &str) -> i32 {
1558 oid_for(rows, col)
1559}
1560
1561fn has_evidence(rows: &[Value], col: &str) -> bool {
1567 rows.iter().any(|r| matches!(r.get(col), Some(v) if !v.is_null()))
1568}
1569
1570fn oid_for(rows: &[Value], col: &str) -> i32 {
1571 let mut acc: Option<i32> = None;
1572 for r in rows {
1573 if let Some(o) = r.get(col).and_then(oid_of_value) {
1574 acc = Some(match acc {
1575 None => o,
1576 Some(prev) => unify_oid(prev, o),
1577 });
1578 if acc == Some(OID_TEXT) {
1579 break; }
1581 }
1582 }
1583 acc.unwrap_or(OID_TEXT)
1584}
1585
1586fn cell(v: Option<&Value>) -> Option<String> {
1588 match v {
1589 None | Some(Value::Null) => None, Some(Value::String(s)) => Some(s.clone()),
1591 Some(Value::Bool(b)) => Some(if *b { "t".into() } else { "f".into() }),
1592 Some(other) => Some(other.to_string()),
1593 }
1594}
1595
1596fn cell_binary(v: Option<&Value>, oid: i32) -> Result<Option<Vec<u8>>, String> {
1608 let v = match v {
1609 None | Some(Value::Null) => return Ok(None),
1610 Some(v) => v,
1611 };
1612 let as_f64 = |n: &serde_json::Number| n.as_f64()
1613 .ok_or_else(|| "a number too large to send as float8".to_string());
1614 Ok(Some(match (oid, v) {
1615 (OID_BOOL, Value::Bool(b)) => vec![u8::from(*b)],
1616 (OID_INT2, Value::Number(n)) => {
1617 let i = n.as_i64().ok_or("not an integer")?;
1618 i16::try_from(i).map_err(|_| format!("{} does not fit in int2", i))?
1619 .to_be_bytes().to_vec()
1620 }
1621 (OID_INT4, Value::Number(n)) => {
1622 let i = n.as_i64().ok_or("not an integer")?;
1623 i32::try_from(i).map_err(|_| format!("{} does not fit in int4", i))?
1624 .to_be_bytes().to_vec()
1625 }
1626 (OID_INT8, Value::Number(n)) => {
1627 n.as_i64().ok_or("not an integer")?.to_be_bytes().to_vec()
1628 }
1629 (OID_FLOAT4, Value::Number(n)) => (as_f64(n)? as f32).to_be_bytes().to_vec(),
1630 (OID_FLOAT8, Value::Number(n)) => as_f64(n)?.to_be_bytes().to_vec(),
1631 (OID_TEXT | OID_VARCHAR | OID_NAME | OID_UNKNOWN | OID_JSON, _) => {
1633 cell(Some(v)).unwrap_or_default().into_bytes()
1634 }
1635 (OID_JSONB, _) => {
1637 let mut b = vec![1u8];
1638 b.extend_from_slice(cell(Some(v)).unwrap_or_default().as_bytes());
1639 b
1640 }
1641 (oid, val) => {
1642 let kind = match val {
1643 Value::Bool(_) => "a boolean",
1644 Value::Number(_) => "a number",
1645 Value::String(_) => "a string",
1646 Value::Array(_) => "an array",
1647 _ => "an object",
1648 };
1649 return Err(format!(
1650 "cannot send {} in binary format as type OID {} — the field holds \
1651 more than one type across documents, so it cannot be described \
1652 by a single Postgres type. Select it with a text cast, or use a \
1653 text-format client",
1654 kind, oid
1655 ));
1656 }
1657 }))
1658}
1659
1660fn row_description_fmt(cols: &[Col], oids: &[i32], fmts: &[i16]) -> Vec<u8> {
1662 let mut m = Out::msg(b'T');
1663 m.i16(cols.len() as i16);
1664 for (i, c) in cols.iter().enumerate() {
1665 m.cstr(&c.out);
1666 m.i32(0); m.i16((i + 1) as i16); m.i32(oids.get(i).copied().unwrap_or(OID_TEXT));
1669 m.i16(-1); m.i32(-1); m.i16(fmts.get(i).copied().unwrap_or(0));
1672 }
1673 m.finish()
1674}
1675
1676fn row_description(cols: &[Col], oids: &[i32]) -> Vec<u8> {
1677 row_description_fmt(cols, oids, &[])
1678}
1679
1680fn data_row_bytes(vals: &[Option<Vec<u8>>]) -> Vec<u8> {
1681 let mut m = Out::msg(b'D');
1682 m.i16(vals.len() as i16);
1683 for v in vals {
1684 match v {
1685 None => m.i32(-1),
1686 Some(b) => {
1687 m.i32(b.len() as i32);
1688 m.bytes(b);
1689 }
1690 }
1691 }
1692 m.finish()
1693}
1694
1695fn data_row(vals: &[Option<String>]) -> Vec<u8> {
1696 let owned: Vec<Option<Vec<u8>>> =
1697 vals.iter().map(|v| v.as_ref().map(|s| s.as_bytes().to_vec())).collect();
1698 data_row_bytes(&owned)
1699}
1700
1701pub fn encode_rows(rows: &[Value], project: &[Col]) -> Vec<u8> {
1711 let cols = columns_for(rows, project);
1712 let oids: Vec<i32> = cols.iter().map(|c| oid_for(rows, &c.src)).collect();
1713 let mut out = row_description(&cols, &oids);
1714 for r in rows {
1715 let vals: Vec<Option<String>> = cols.iter().map(|c| cell(r.get(&c.src))).collect();
1716 out.extend_from_slice(&data_row(&vals));
1717 }
1718 out
1719}
1720
1721pub fn encode_result(rows: &[Value], project: &[Col]) -> Vec<u8> {
1723 let mut out = encode_rows(rows, project);
1724 out.extend_from_slice(&command_complete(&format!("SELECT {}", rows.len())));
1725 out
1726}
1727
1728const OID_INT2: i32 = 21;
1755const OID_INT4: i32 = 23;
1756const OID_OID: i32 = 26;
1757const OID_FLOAT4: i32 = 700;
1758const OID_VARCHAR: i32 = 1043;
1759const OID_NAME: i32 = 19;
1760const OID_UNKNOWN: i32 = 705;
1761const OID_JSON: i32 = 114;
1762const OID_JSONB: i32 = 3802;
1763
1764fn param_count(sql: &str) -> usize {
1770 let b = sql.as_bytes();
1771 let mut i = 0usize;
1772 let mut in_s = false;
1773 let mut max = 0usize;
1774 while i < b.len() {
1775 let c = b[i];
1776 if in_s {
1777 if c == b'\'' {
1778 in_s = false;
1779 }
1780 i += 1;
1781 continue;
1782 }
1783 if c == b'\'' {
1784 in_s = true;
1785 i += 1;
1786 continue;
1787 }
1788 if c == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
1789 let mut j = i + 1;
1790 let mut n = 0usize;
1791 while j < b.len() && b[j].is_ascii_digit() {
1792 n = n * 10 + (b[j] - b'0') as usize;
1793 j += 1;
1794 }
1795 max = max.max(n);
1796 i = j;
1797 continue;
1798 }
1799 i += 1;
1800 }
1801 max
1802}
1803
1804fn infer_field_oid(db: Option<&Arc<Db>>, coll: &str, field: &str) -> i32 {
1813 match field {
1817 "_seq" => return OID_INT8,
1818 "_id" | "_hash" | "_prev" | "_collection" | "_valid_from" | "_valid_to" => return OID_TEXT,
1819 _ => {}
1820 }
1821 if !field.is_empty() && crate::pgcatalog::is_catalog(coll) {
1827 if let Some(rows) = crate::pgcatalog::rows(coll, db) {
1828 return oid_for(&rows, field);
1829 }
1830 }
1831 let db = match db {
1832 Some(db) => db,
1833 None => return OID_TEXT,
1834 };
1835 if coll.is_empty() || field.is_empty() {
1836 return OID_TEXT;
1837 }
1838 let rows = match crate::nql::query(db, &format!("FROM {} LIMIT {}", coll, TYPE_SAMPLE)) {
1839 Ok((rows, _)) => rows,
1840 Err(_) => return OID_TEXT,
1841 };
1842 oid_for(&rows, field)
1846}
1847
1848fn evaluator_shape(
1883 sql: &str,
1884 db: Option<&Arc<Db>>,
1885 coll: &str,
1886) -> Option<(Vec<Col>, Vec<i32>)> {
1887 let sel = crate::sqlselect::parse(sql).ok()?;
1888 if sel.items.iter().any(|i| matches!(i.expr, crate::sqlselect::Expr::Star
1891 | crate::sqlselect::Expr::QualifiedStar(_)))
1892 {
1893 return None;
1894 }
1895
1896 let mut cols: Vec<Col> = Vec::new();
1897 let mut oids: Vec<i32> = Vec::new();
1898 for item in &sel.items {
1899 let name = match &item.alias {
1900 Some(a) => a.clone(),
1901 None => match &item.expr {
1902 crate::sqlselect::Expr::Column { name, .. } => name.clone(),
1903 crate::sqlselect::Expr::Agg { name, .. } => name.to_ascii_lowercase(),
1904 crate::sqlselect::Expr::Func { name, .. } => name.to_ascii_lowercase(),
1905 _ => return None,
1909 },
1910 };
1911 oids.push(expr_oid(&item.expr, db, coll)?);
1912 cols.push(Col::renamed(&name, &name));
1913 }
1914 if cols.is_empty() {
1915 return None;
1916 }
1917 Some((cols, oids))
1918}
1919
1920fn expr_oid(e: &crate::sqlselect::Expr, db: Option<&Arc<Db>>, coll: &str) -> Option<i32> {
1922 use crate::sqlselect::Expr;
1923 match e {
1924 Expr::Column { name, .. } => Some(infer_field_oid(db, coll, name)),
1925 Expr::Literal(v) => Some(oid_of_value(v).unwrap_or(OID_TEXT)),
1926 Expr::Agg { name, args, .. } | Expr::Func { name, args } => {
1933 let f = name.to_ascii_lowercase();
1934 match f.as_str() {
1935 "count" => Some(OID_INT8),
1937 "avg" => Some(OID_FLOAT8),
1940 "sum" | "min" | "max" => match args.first() {
1943 Some(Expr::Column { name, .. }) => match infer_field_oid(db, coll, name) {
1944 OID_INT8 => Some(OID_INT8),
1945 OID_FLOAT8 => Some(OID_FLOAT8),
1946 other => Some(other),
1947 },
1948 _ => None,
1949 },
1950 _ => None,
1951 }
1952 }
1953 _ => None,
1954 }
1955}
1956
1957fn aggregate_oid(src: &str, db: Option<&Arc<Db>>, coll: &str) -> Option<i32> {
1958 if src == "count" {
1959 return Some(OID_INT8);
1960 }
1961 for (prefix, fixed) in [
1962 ("count_", Some(OID_INT8)),
1963 ("avg_", Some(OID_FLOAT8)),
1964 ("sum_", None),
1965 ("min_", None),
1966 ("max_", None),
1967 ] {
1968 if let Some(field) = src.strip_prefix(prefix) {
1969 return Some(match fixed {
1970 Some(oid) => oid,
1971 None => match infer_field_oid(db, coll, field) {
1974 OID_INT8 => OID_INT8,
1975 OID_FLOAT8 => OID_FLOAT8,
1976 other => other,
1979 },
1980 });
1981 }
1982 }
1983 None
1984}
1985
1986const TYPE_SAMPLE: usize = 200;
1992
1993fn stmt_collection(sql: &str) -> String {
1995 let s = normalise(sql);
1996 let up = s.to_uppercase();
1997 let after = if let Some(at) = find_kw(&up, "FROM") {
1998 &s[at + 4..]
1999 } else if let Some(rest) = strip_prefix_ci(&s, "UPDATE") {
2000 return rest
2001 .split_whitespace()
2002 .next()
2003 .unwrap_or("")
2004 .rsplit('.')
2005 .next()
2006 .unwrap_or("")
2007 .trim_matches('"')
2008 .to_string();
2009 } else if let Some(rest) = strip_prefix_ci(&s, "INSERT INTO") {
2010 return rest
2011 .split(|c: char| c.is_whitespace() || c == '(')
2012 .find(|t| !t.is_empty())
2013 .unwrap_or("")
2014 .rsplit('.')
2015 .next()
2016 .unwrap_or("")
2017 .trim_matches('"')
2018 .to_string();
2019 } else {
2020 return String::new();
2021 };
2022 after
2023 .trim()
2024 .split(|c: char| c.is_whitespace())
2025 .find(|t| !t.is_empty())
2026 .unwrap_or("")
2027 .rsplit('.')
2028 .next()
2029 .unwrap_or("")
2030 .trim_matches('"')
2031 .to_string()
2032}
2033
2034fn param_fields(sql: &str, n_params: usize) -> Vec<Option<String>> {
2045 let s = normalise(sql);
2046 let mut out = vec![None; n_params];
2047
2048 let up = s.to_uppercase();
2051 if up.starts_with("INSERT") {
2052 if let (Some(open), Some(vals_at)) = (s.find('('), find_kw(&up, "VALUES")) {
2053 if open < vals_at {
2054 if let Some(close) = s[open..vals_at].rfind(')') {
2055 let cols: Vec<String> = split_top(&s[open + 1..open + close], ',')
2056 .into_iter()
2057 .map(|c| c.trim().trim_matches('"').to_string())
2058 .collect();
2059 let tail = &s[vals_at..];
2061 let mut seen = 0usize;
2062 let b = tail.as_bytes();
2063 let mut i = 0usize;
2064 let mut in_s = false;
2065 while i < b.len() {
2066 if in_s {
2067 if b[i] == b'\'' { in_s = false; }
2068 i += 1;
2069 continue;
2070 }
2071 if b[i] == b'\'' { in_s = true; i += 1; continue; }
2072 if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2073 let mut j = i + 1;
2074 let mut num = 0usize;
2075 while j < b.len() && b[j].is_ascii_digit() {
2076 num = num * 10 + (b[j] - b'0') as usize;
2077 j += 1;
2078 }
2079 if num >= 1 && num <= n_params {
2080 if let Some(c) = cols.get(seen % cols.len().max(1)) {
2081 out[num - 1] = Some(c.clone());
2082 }
2083 }
2084 seen += 1;
2085 i = j;
2086 continue;
2087 }
2088 i += 1;
2089 }
2090 return out;
2091 }
2092 }
2093 }
2094 }
2095
2096 let b = s.as_bytes();
2098 let mut i = 0usize;
2099 let mut in_s = false;
2100 while i < b.len() {
2101 if in_s {
2102 if b[i] == b'\'' { in_s = false; }
2103 i += 1;
2104 continue;
2105 }
2106 if b[i] == b'\'' { in_s = true; i += 1; continue; }
2107 if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2108 let mut j = i + 1;
2109 let mut num = 0usize;
2110 while j < b.len() && b[j].is_ascii_digit() {
2111 num = num * 10 + (b[j] - b'0') as usize;
2112 j += 1;
2113 }
2114 if num >= 1 && num <= n_params {
2115 let left = &s[..i];
2116 let trimmed = left.trim_end_matches(|c: char| {
2119 c.is_whitespace() || "=<>!+-*/%(,".contains(c)
2120 });
2121 let mut tok = trimmed
2124 .rsplit(|c: char| c.is_whitespace() || c == '(' || c == ',')
2125 .find(|t| !t.is_empty())
2126 .unwrap_or("")
2127 .trim_matches('"');
2128 let mut before = trimmed;
2129 for _ in 0..4 {
2130 let upper_tok = tok.to_uppercase();
2131 if upper_tok.starts_with('$')
2137 || matches!(upper_tok.as_str(),
2138 "LIKE" | "ILIKE" | "IN" | "BETWEEN" | "AND" | "OR" | "NOT" | "IS") {
2139 before = before[..before.len() - tok.len()].trim_end_matches(|c: char| {
2140 c.is_whitespace() || "=<>!(,".contains(c)
2141 });
2142 tok = before
2143 .rsplit(|c: char| c.is_whitespace() || c == '(' || c == ',')
2144 .find(|t| !t.is_empty())
2145 .unwrap_or("")
2146 .trim_matches('"');
2147 } else {
2148 break;
2149 }
2150 }
2151 if !tok.is_empty()
2152 && tok.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '.')
2153 && !tok.chars().next().map(|c| c.is_ascii_digit()).unwrap_or(true)
2154 {
2155 out[num - 1] = Some(tok.rsplit('.').next().unwrap_or(tok).to_string());
2156 }
2157 }
2158 i = j;
2159 continue;
2160 }
2161 i += 1;
2162 }
2163 out
2164}
2165
2166fn clause_param_oids(sql: &str, n_params: usize) -> Vec<Option<i32>> {
2176 let s = normalise(sql);
2177 let mut out = vec![None; n_params];
2178 let b = s.as_bytes();
2179 let mut i = 0usize;
2180 let mut in_s = false;
2181 while i < b.len() {
2182 if in_s {
2183 if b[i] == b'\'' { in_s = false; }
2184 i += 1;
2185 continue;
2186 }
2187 if b[i] == b'\'' { in_s = true; i += 1; continue; }
2188 if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2189 let mut j = i + 1;
2190 let mut num = 0usize;
2191 while j < b.len() && b[j].is_ascii_digit() {
2192 num = num * 10 + (b[j] - b'0') as usize;
2193 j += 1;
2194 }
2195 if num >= 1 && num <= n_params {
2196 let left = s[..i].trim_end().to_uppercase();
2197 out[num - 1] = if left.ends_with("VALID AS OF") {
2200 Some(OID_TEXT)
2201 } else if left.ends_with("AS OF SYSTEM TIME")
2202 || left.ends_with("FOR SYSTEM_TIME AS OF")
2203 || left.ends_with("AS OF")
2204 || left.ends_with("LIMIT")
2205 || left.ends_with("OFFSET")
2206 {
2207 Some(OID_INT8)
2208 } else {
2209 None
2210 };
2211 }
2212 i = j;
2213 continue;
2214 }
2215 i += 1;
2216 }
2217 out
2218}
2219
2220fn infer_param_oids(sql: &str, declared: &[i32], db: Option<&Arc<Db>>) -> Vec<i32> {
2226 let n = param_count(sql).max(declared.len());
2227 if n == 0 {
2228 return vec![];
2229 }
2230 let coll = stmt_collection(sql);
2231 let fields = param_fields(sql, n);
2232 let clauses = clause_param_oids(sql, n);
2233 (0..n)
2234 .map(|i| match declared.get(i) {
2235 Some(&oid) if oid != 0 => oid,
2236 _ => match clauses[i] {
2239 Some(oid) => oid,
2240 None => match &fields[i] {
2241 Some(f) => infer_field_oid(db, &coll, f),
2242 None => OID_TEXT,
2243 },
2244 },
2245 })
2246 .collect()
2247}
2248
2249fn decode_param(raw: Option<&[u8]>, oid: i32, format: i16) -> Result<Option<String>, String> {
2255 let bytes = match raw {
2256 None => return Ok(None),
2257 Some(b) => b,
2258 };
2259 let quote = |s: &str| format!("'{}'", s.replace('\'', "''"));
2260
2261 if format == 0 {
2262 let s = String::from_utf8_lossy(bytes).to_string();
2263 return Ok(Some(match oid {
2264 OID_BOOL => {
2265 let t = matches!(s.as_str(), "t" | "true" | "TRUE" | "1" | "yes" | "on");
2266 if t { "TRUE".into() } else { "FALSE".into() }
2267 }
2268 OID_INT2 | OID_INT4 | OID_INT8 | OID_OID | OID_FLOAT4 | OID_FLOAT8 => {
2269 if s.parse::<f64>().is_ok() { s } else { quote(&s) }
2273 }
2274 _ => quote(&s),
2279 }));
2280 }
2281 if format != 1 {
2282 return Err(format!("unsupported parameter format code {}", format));
2283 }
2284
2285 let need = |n: usize| -> Result<(), String> {
2287 if bytes.len() == n {
2288 Ok(())
2289 } else {
2290 Err(format!(
2291 "binary parameter of type OID {} should be {} bytes, got {}",
2292 oid, n, bytes.len()
2293 ))
2294 }
2295 };
2296 Ok(Some(match oid {
2297 OID_BOOL => {
2298 need(1)?;
2299 if bytes[0] != 0 { "TRUE".into() } else { "FALSE".into() }
2300 }
2301 OID_INT2 => {
2302 need(2)?;
2303 i16::from_be_bytes([bytes[0], bytes[1]]).to_string()
2304 }
2305 OID_INT4 => {
2306 need(4)?;
2307 i32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]).to_string()
2308 }
2309 OID_OID => {
2310 need(4)?;
2311 u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]).to_string()
2312 }
2313 OID_INT8 => {
2314 need(8)?;
2315 i64::from_be_bytes(bytes[..8].try_into().unwrap()).to_string()
2316 }
2317 OID_FLOAT4 => {
2318 need(4)?;
2319 let f = f32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
2320 fmt_float(f as f64)
2321 }
2322 OID_FLOAT8 => {
2323 need(8)?;
2324 fmt_float(f64::from_be_bytes(bytes[..8].try_into().unwrap()))
2325 }
2326 OID_TEXT | OID_VARCHAR | OID_NAME | OID_UNKNOWN | OID_JSON | 0 => {
2327 quote(&String::from_utf8_lossy(bytes))
2328 }
2329 OID_JSONB => {
2330 let body = if bytes.first() == Some(&1) { &bytes[1..] } else { bytes };
2332 quote(&String::from_utf8_lossy(body))
2333 }
2334 other => {
2335 return Err(format!(
2336 "parameter type OID {} is not supported in binary format — \
2337 the supported set is bool, int2/int4/int8, float4/float8, \
2338 text/varchar/json/jsonb. Send it as text, or cast it in the \
2339 statement",
2340 other
2341 ))
2342 }
2343 }))
2344}
2345
2346fn fmt_float(f: f64) -> String {
2348 if f.is_nan() {
2349 "'NaN'".into()
2350 } else if f.is_infinite() {
2351 if f > 0.0 { "'Infinity'".into() } else { "'-Infinity'".into() }
2352 } else if f.fract() == 0.0 && f.abs() < 1e15 {
2353 format!("{:.0}", f)
2354 } else {
2355 f.to_string()
2356 }
2357}
2358
2359fn substitute_params(sql: &str, params: &[Option<String>]) -> Result<String, String> {
2367 let b = sql.as_bytes();
2368 let mut out = String::with_capacity(sql.len() + 16);
2369 let mut i = 0usize;
2370 let mut in_s = false;
2371 while i < b.len() {
2372 let c = b[i];
2373 if in_s {
2374 out.push(c as char);
2375 if c == b'\'' { in_s = false; }
2376 i += 1;
2377 continue;
2378 }
2379 if c == b'\'' {
2380 in_s = true;
2381 out.push('\'');
2382 i += 1;
2383 continue;
2384 }
2385 if c == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2386 let mut j = i + 1;
2387 let mut n = 0usize;
2388 while j < b.len() && b[j].is_ascii_digit() {
2389 n = n * 10 + (b[j] - b'0') as usize;
2390 j += 1;
2391 }
2392 match params.get(n.wrapping_sub(1)) {
2393 Some(Some(lit)) => out.push_str(lit),
2394 Some(None) => out.push_str("NULL"),
2395 None => {
2396 return Err(format!(
2397 "bind message supplies {} parameter(s) but the statement uses ${}",
2398 params.len(), n
2399 ))
2400 }
2401 }
2402 i = j;
2403 continue;
2404 }
2405 out.push(c as char);
2406 i += 1;
2407 }
2408 Ok(out)
2409}
2410
2411struct Prepared {
2413 sql: String,
2414 param_oids: Vec<i32>,
2417 out_shape: Option<Option<(Vec<Col>, Vec<i32>)>>,
2425}
2426
2427fn prepared_shape<'a>(
2429 p: &'a mut Prepared,
2430 db: Option<&Arc<Db>>,
2431) -> &'a Option<(Vec<Col>, Vec<i32>)> {
2432 if p.out_shape.is_none() {
2433 p.out_shape = Some(describe_shape(&p.sql, db, p.param_oids.len()));
2434 }
2435 p.out_shape.as_ref().expect("just filled")
2436}
2437
2438struct Portal {
2440 sql: String,
2441 result: Option<PortalResult>,
2447 frozen: Option<Vec<Col>>,
2457 formats: Vec<i16>,
2459 declared: Option<(Vec<Col>, Vec<i32>)>,
2466}
2467
2468impl Portal {
2469 fn format_of(&self, i: usize) -> i16 {
2472 match self.formats.len() {
2473 0 => 0,
2474 1 => self.formats[0],
2475 _ => self.formats.get(i).copied().unwrap_or(0),
2476 }
2477 }
2478 fn shape(&self, r: &PortalResult) -> (Vec<Col>, Vec<i32>) {
2480 match &self.declared {
2481 Some((cols, oids)) if self.formats.iter().any(|f| *f == 1) => {
2482 (cols.clone(), oids.clone())
2483 }
2484 _ => {
2485 let cols = columns_for(&r.rows, &r.project);
2486 let oids = cols.iter().map(|c| oid_for(&r.rows, &c.src)).collect();
2487 (cols, oids)
2488 }
2489 }
2490 }
2491}
2492
2493struct PortalResult {
2494 rows: Vec<Value>,
2495 project: Vec<Col>,
2496 has_rows: bool,
2497 tag: String,
2498 tag_counts_rows: bool,
2499 sent: usize,
2501}
2502
2503fn parse_complete() -> Vec<u8> { Out::msg(b'1').finish() }
2504fn bind_complete() -> Vec<u8> { Out::msg(b'2').finish() }
2505fn close_complete() -> Vec<u8> { Out::msg(b'3').finish() }
2506fn no_data() -> Vec<u8> { Out::msg(b'n').finish() }
2507fn portal_suspended() -> Vec<u8> { Out::msg(b's').finish() }
2508
2509fn parameter_description(oids: &[i32]) -> Vec<u8> {
2510 let mut m = Out::msg(b't');
2511 m.i16(oids.len() as i16);
2512 for o in oids {
2513 m.i32(*o);
2514 }
2515 m.finish()
2516}
2517
2518fn take_cstr(body: &[u8], at: &mut usize) -> String {
2520 let start = *at;
2521 while *at < body.len() && body[*at] != 0 {
2522 *at += 1;
2523 }
2524 let s = String::from_utf8_lossy(&body[start..*at]).to_string();
2525 if *at < body.len() {
2526 *at += 1; }
2528 s
2529}
2530
2531fn take_i16(body: &[u8], at: &mut usize) -> Result<i16, String> {
2532 if *at + 2 > body.len() {
2533 return Err("truncated message".into());
2534 }
2535 let v = i16::from_be_bytes([body[*at], body[*at + 1]]);
2536 *at += 2;
2537 Ok(v)
2538}
2539
2540fn take_i32(body: &[u8], at: &mut usize) -> Result<i32, String> {
2541 if *at + 4 > body.len() {
2542 return Err("truncated message".into());
2543 }
2544 let v = i32::from_be_bytes([body[*at], body[*at + 1], body[*at + 2], body[*at + 3]]);
2545 *at += 4;
2546 Ok(v)
2547}
2548
2549fn sample_columns(db: Option<&Arc<Db>>, coll: &str) -> Vec<Col> {
2555 let db = match db {
2556 Some(db) => db,
2557 None => return vec![],
2558 };
2559 let rows = match crate::nql::query(db, &format!("FROM {} LIMIT 25", coll)) {
2560 Ok((rows, _)) => rows,
2561 Err(_) => return vec![],
2562 };
2563 let mut names: Vec<String> = vec![];
2564 for r in &rows {
2565 if let Value::Object(m) = r {
2566 for k in m.keys() {
2567 if !names.iter().any(|n| n == k) {
2568 names.push(k.clone());
2569 }
2570 }
2571 }
2572 }
2573 names.sort();
2574 names.iter().map(|n| Col::same(n)).collect()
2575}
2576
2577fn describe_shape(
2585 sql: &str,
2586 db: Option<&Arc<Db>>,
2587 n_params: usize,
2588) -> Option<(Vec<Col>, Vec<i32>)> {
2589 let probe = probe_sql(sql, n_params);
2590
2591 let coll = stmt_collection(sql);
2602
2603 if sql_engine_owns(&probe) {
2604 if let Ok(Some((done, _))) = try_catalog_select(&probe, db) {
2605 if done.project.is_empty() {
2606 return None;
2607 }
2608 let fallback = evaluator_shape(&probe, db, &coll);
2627 let oids: Vec<i32> = done
2628 .project
2629 .iter()
2630 .enumerate()
2631 .map(|(i, c)| {
2632 let seen = has_evidence(&done.rows, &c.src);
2633 if seen {
2634 oid_for(&done.rows, &c.src)
2635 } else {
2636 fallback
2637 .as_ref()
2638 .and_then(|(_, o)| o.get(i).copied())
2639 .unwrap_or(OID_TEXT)
2640 }
2641 })
2642 .collect();
2643 return Some((done.project, oids));
2644 }
2645 }
2646
2647
2648 if sql_engine_owns(&probe) {
2653 if let Some(shape) = evaluator_shape(&probe, db, &coll) {
2654 return Some(shape);
2655 }
2656 }
2657
2658 let stmt = translate(&probe).ok()?;
2659
2660 let cols = match stmt {
2661 Stmt::Ok(_) => return None,
2662 Stmt::Canned { cols, .. } => cols.iter().map(|c| Col::same(c)).collect(),
2663 Stmt::Query { project, .. } => {
2664 if project.is_empty() { sample_columns(db, &coll) } else { project }
2665 }
2666 Stmt::Insert { returning, .. } | Stmt::Update { returning, .. } | Stmt::Delete { returning, .. } => {
2667 if !wants_returning(sql) {
2668 return None;
2669 }
2670 if returning.is_empty() { sample_columns(db, &coll) } else { returning }
2671 }
2672 };
2673 if cols.is_empty() {
2674 return None;
2678 }
2679 let oids = cols
2680 .iter()
2681 .map(|c| {
2682 aggregate_oid(&c.src, db, &coll)
2683 .unwrap_or_else(|| infer_field_oid(db, &coll, &c.src))
2684 })
2685 .collect();
2686 Some((cols, oids))
2687}
2688
2689fn probe_sql(sql: &str, n_params: usize) -> String {
2697 let stub: Vec<Option<String>> = vec![Some("0".to_string()); n_params];
2698 substitute_params(sql, &stub).unwrap_or_else(|_| sql.to_string())
2699}
2700
2701fn ensure_executed(
2703 portal: &mut Portal,
2704 db_name: &str,
2705 db: Option<&Arc<Db>>,
2706 read_only: bool,
2707) -> Result<(), Vec<u8>> {
2708 if portal.result.is_some() {
2709 return Ok(());
2710 }
2711 let ex = execute_stmt(&portal.sql, db_name, db, read_only)?;
2712 let project = if let Some(f) = &portal.frozen {
2715 f.clone()
2716 } else {
2717 let p = if ex.project.is_empty() {
2718 columns_for(&ex.rows, &[])
2719 } else {
2720 ex.project.clone()
2721 };
2722 portal.frozen = Some(p.clone());
2723 p
2724 };
2725 portal.result = Some(PortalResult {
2726 rows: ex.rows,
2727 project,
2728 has_rows: ex.has_rows,
2729 tag: ex.tag,
2730 tag_counts_rows: ex.tag_counts_rows,
2731 sent: 0,
2732 });
2733 Ok(())
2734}
2735
2736async fn read_exact(sock: &mut TcpStream, n: usize) -> std::io::Result<Vec<u8>> {
2739 let mut buf = vec![0u8; n];
2740 sock.read_exact(&mut buf).await?;
2741 Ok(buf)
2742}
2743
2744async fn read_i32(sock: &mut TcpStream) -> std::io::Result<i32> {
2745 let b = read_exact(sock, 4).await?;
2746 Ok(i32::from_be_bytes([b[0], b[1], b[2], b[3]]))
2747}
2748
2749fn parse_startup_params(body: &[u8]) -> HashMap<String, String> {
2750 let mut out = HashMap::new();
2751 let mut parts = body.split(|b| *b == 0).map(|s| String::from_utf8_lossy(s).to_string());
2752 while let (Some(k), Some(v)) = (parts.next(), parts.next()) {
2753 if k.is_empty() {
2754 break;
2755 }
2756 out.insert(k, v);
2757 }
2758 out
2759}
2760
2761async fn handle(mut sock: TcpStream, resolver: Arc<dyn DbResolver>, read_only: bool) -> std::io::Result<()> {
2763 let params = loop {
2765 let len = read_i32(&mut sock).await?;
2766 if len < 8 || len > 1 << 20 {
2767 return Ok(()); }
2769 let code = read_i32(&mut sock).await?;
2770 let body = read_exact(&mut sock, (len - 8) as usize).await?;
2771 match code {
2772 SSL_REQUEST | GSS_REQUEST => {
2773 sock.write_all(b"N").await?;
2775 continue;
2776 }
2777 CANCEL_REQUEST => return Ok(()), PROTO_V3 => break parse_startup_params(&body),
2779 other => {
2780 let major = other >> 16;
2781 sock.write_all(&err_msg(
2782 "0A000",
2783 &format!("unsupported frontend protocol {}.{} — this endpoint speaks 3.0",
2784 major, other & 0xffff),
2785 )).await?;
2786 return Ok(());
2787 }
2788 }
2789 };
2790
2791 let db_name = params.get("database").cloned().unwrap_or_default();
2792
2793 let resolved: Option<Arc<Db>> = {
2799 let r = Arc::clone(&resolver);
2800 let name = db_name.clone();
2801 tokio::task::spawn_blocking(move || r.resolve(&name))
2802 .await
2803 .unwrap_or(None)
2804 };
2805
2806 if let Some(expected) = resolver.token() {
2808 let mut m = Out::msg(b'R');
2810 m.i32(3);
2811 sock.write_all(&m.finish()).await?;
2812
2813 let tag = read_exact(&mut sock, 1).await?;
2814 if tag[0] != b'p' {
2815 sock.write_all(&err_msg("28000", "expected a password message")).await?;
2816 return Ok(());
2817 }
2818 let len = read_i32(&mut sock).await?;
2819 if len < 4 || len > 1 << 16 {
2820 return Ok(());
2821 }
2822 let body = read_exact(&mut sock, (len - 4) as usize).await?;
2823 let supplied = String::from_utf8_lossy(&body).trim_end_matches('\0').to_string();
2824 let ok = supplied.len() == expected.len()
2826 && supplied.bytes().zip(expected.bytes()).fold(0u8, |a, (x, y)| a | (x ^ y)) == 0;
2827 if !ok {
2828 sock.write_all(&err_msg("28P01", "password authentication failed")).await?;
2829 return Ok(());
2830 }
2831 }
2832
2833 let mut m = Out::msg(b'R');
2834 m.i32(0); sock.write_all(&m.finish()).await?;
2836
2837 for (k, v) in [
2838 ("server_version", SERVER_VERSION),
2839 ("server_encoding", "UTF8"),
2840 ("client_encoding", "UTF8"),
2841 ("DateStyle", "ISO, MDY"),
2842 ("integer_datetimes", "on"),
2843 ("standard_conforming_strings", "on"),
2844 ("application_name", "nedbd"),
2845 ] {
2846 let mut p = Out::msg(b'S');
2847 p.cstr(k);
2848 p.cstr(v);
2849 sock.write_all(&p.finish()).await?;
2850 }
2851 let mut k = Out::msg(b'K');
2852 k.i32(std::process::id() as i32);
2853 k.i32(0);
2854 sock.write_all(&k.finish()).await?;
2855 sock.write_all(&ready()).await?;
2856
2857 let mut prepared: HashMap<String, Prepared> = HashMap::new();
2863 let mut portals: HashMap<String, Portal> = HashMap::new();
2864 let mut failed = false;
2868
2869 loop {
2870 let mut tag = [0u8; 1];
2871 if sock.read_exact(&mut tag).await.is_err() {
2872 return Ok(()); }
2874 let len = read_i32(&mut sock).await?;
2875 if len < 4 || len > 64 << 20 {
2876 return Ok(());
2877 }
2878 let body = read_exact(&mut sock, (len - 4) as usize).await?;
2879
2880 if failed && tag[0] != b'S' && tag[0] != b'X' {
2882 continue;
2883 }
2884
2885 match tag[0] {
2886 b'X' => return Ok(()), b'Q' => {
2889 let sql = String::from_utf8_lossy(&body).trim_end_matches('\0').to_string();
2890 let out = run_simple_query(&sql, &db_name, resolved.as_ref(), read_only);
2891 sock.write_all(&out).await?;
2892 sock.write_all(&ready()).await?;
2893 portals.remove("");
2895 }
2896
2897 b'P' => {
2899 let mut at = 0usize;
2900 let name = take_cstr(&body, &mut at);
2901 let sql = take_cstr(&body, &mut at);
2902 let n = take_i16(&body, &mut at).unwrap_or(0).max(0) as usize;
2903 let mut declared = Vec::with_capacity(n);
2904 let mut bad = false;
2905 for _ in 0..n {
2906 match take_i32(&body, &mut at) {
2907 Ok(o) => declared.push(o),
2908 Err(_) => { bad = true; break; }
2909 }
2910 }
2911 if bad {
2912 sock.write_all(&err_msg("08P01", "malformed Parse message")).await?;
2913 failed = true;
2914 continue;
2915 }
2916 let probe = probe_sql(&sql, param_count(&sql));
2926 if !sql_engine_owns(&probe) {
2927 if let Err(why) = translate(&probe) {
2928 sock.write_all(&err_msg("0A000", &why)).await?;
2929 failed = true;
2930 continue;
2931 }
2932 }
2933 let param_oids = infer_param_oids(&sql, &declared, resolved.as_ref());
2934 prepared.insert(name, Prepared { sql, param_oids, out_shape: None });
2935 sock.write_all(&parse_complete()).await?;
2936 }
2937
2938 b'B' => {
2940 let mut at = 0usize;
2941 let portal_name = take_cstr(&body, &mut at);
2942 let stmt_name = take_cstr(&body, &mut at);
2943 if !prepared.contains_key(&stmt_name) {
2944 sock.write_all(&err_msg("26000", &format!(
2945 "prepared statement {:?} does not exist", stmt_name))).await?;
2946 failed = true;
2947 continue;
2948 }
2949 let p = &prepared[&stmt_name];
2950 let mut want_formats: Vec<i16> = vec![];
2951 let res: Result<String, String> = (|| {
2952 let nfmt = take_i16(&body, &mut at)? .max(0) as usize;
2953 let mut fmts = Vec::with_capacity(nfmt);
2954 for _ in 0..nfmt {
2955 fmts.push(take_i16(&body, &mut at)?);
2956 }
2957 let nparam = take_i16(&body, &mut at)?.max(0) as usize;
2958 let mut vals: Vec<Option<String>> = Vec::with_capacity(nparam);
2959 for i in 0..nparam {
2960 let l = take_i32(&body, &mut at)?;
2961 let raw: Option<Vec<u8>> = if l < 0 {
2962 None
2963 } else {
2964 let l = l as usize;
2965 if at + l > body.len() {
2966 return Err("truncated Bind parameter".into());
2967 }
2968 let v = body[at..at + l].to_vec();
2969 at += l;
2970 Some(v)
2971 };
2972 let f = match fmts.len() {
2975 0 => 0,
2976 1 => fmts[0],
2977 _ => *fmts.get(i).unwrap_or(&0),
2978 };
2979 let oid = *p.param_oids.get(i).unwrap_or(&OID_TEXT);
2980 vals.push(decode_param(raw.as_deref(), oid, f)?);
2981 }
2982 let nres = take_i16(&body, &mut at)?.max(0) as usize;
2987 for _ in 0..nres {
2988 let f = take_i16(&body, &mut at)?;
2989 if f != 0 && f != 1 {
2990 return Err(format!("unknown result format code {}", f));
2991 }
2992 want_formats.push(f);
2993 }
2994 substitute_params(&p.sql, &vals)
2995 })();
2996 match res {
2997 Ok(sql) => {
2998 let declared = if want_formats.iter().any(|f| *f == 1) {
3001 let p = prepared.get_mut(&stmt_name).expect("checked above");
3002 prepared_shape(p, resolved.as_ref()).clone()
3003 } else {
3004 None
3005 };
3006 portals.insert(portal_name, Portal {
3007 sql, result: None, frozen: None,
3008 formats: want_formats, declared,
3009 });
3010 sock.write_all(&bind_complete()).await?;
3011 }
3012 Err(why) => {
3013 sock.write_all(&err_msg("08P01", &why)).await?;
3014 failed = true;
3015 }
3016 }
3017 }
3018
3019 b'D' => {
3021 let kind = body.first().copied().unwrap_or(b'S');
3022 let mut at = 1usize;
3023 let name = take_cstr(&body, &mut at);
3024 if kind == b'S' {
3025 if !prepared.contains_key(&name) {
3026 sock.write_all(&err_msg("26000", &format!(
3027 "prepared statement {:?} does not exist", name))).await?;
3028 failed = true;
3029 continue;
3030 }
3031 let p = prepared.get_mut(&name).expect("checked above");
3032 let oids = p.param_oids.clone();
3033 sock.write_all(¶meter_description(&oids)).await?;
3036 let out = match prepared_shape(p, resolved.as_ref()) {
3040 Some((cols, col_oids)) => row_description(cols, col_oids),
3041 None => no_data(),
3042 };
3043 sock.write_all(&out).await?;
3044 } else {
3045 let portal = match portals.get_mut(&name) {
3046 Some(p) => p,
3047 None => {
3048 sock.write_all(&err_msg("34000", &format!(
3049 "portal {:?} does not exist", name))).await?;
3050 failed = true;
3051 continue;
3052 }
3053 };
3054 match ensure_executed(portal, &db_name, resolved.as_ref(), read_only) {
3059 Err(encoded) => {
3060 sock.write_all(&encoded).await?;
3061 failed = true;
3062 }
3063 Ok(()) => {
3064 let r = portal.result.as_ref().expect("just executed");
3065 if !r.has_rows {
3066 sock.write_all(&no_data()).await?;
3067 } else {
3068 let (cols, oids) = portal.shape(r);
3069 let fmts: Vec<i16> =
3070 (0..cols.len()).map(|i| portal.format_of(i)).collect();
3071 sock.write_all(&row_description_fmt(&cols, &oids, &fmts)).await?;
3072 }
3073 }
3074 }
3075 }
3076 }
3077
3078 b'E' => {
3080 let mut at = 0usize;
3081 let name = take_cstr(&body, &mut at);
3082 let max_rows = take_i32(&body, &mut at).unwrap_or(0);
3083 let portal = match portals.get_mut(&name) {
3084 Some(p) => p,
3085 None => {
3086 sock.write_all(&err_msg("34000", &format!(
3087 "portal {:?} does not exist", name))).await?;
3088 failed = true;
3089 continue;
3090 }
3091 };
3092 if let Err(encoded) = ensure_executed(portal, &db_name, resolved.as_ref(), read_only) {
3093 sock.write_all(&encoded).await?;
3094 failed = true;
3095 continue;
3096 }
3097 let r = portal.result.as_ref().expect("just executed");
3098 if !r.has_rows {
3099 let tag = r.tag.clone();
3100 sock.write_all(&command_complete(&tag)).await?;
3101 continue;
3102 }
3103 let (cols, oids) = portal.shape(r);
3104 let limit = if max_rows > 0 {
3105 (r.sent + max_rows as usize).min(r.rows.len())
3106 } else {
3107 r.rows.len()
3108 };
3109 let mut encoded: Vec<Vec<u8>> = Vec::with_capacity(limit - r.sent);
3114 let mut fail: Option<String> = None;
3115 for row in &r.rows[r.sent..limit] {
3116 let mut vals: Vec<Option<Vec<u8>>> = Vec::with_capacity(cols.len());
3117 for (i, c) in cols.iter().enumerate() {
3118 let v = row.get(&c.src);
3119 let got = if portal.format_of(i) == 1 {
3120 cell_binary(v, oids.get(i).copied().unwrap_or(OID_TEXT))
3121 .map_err(|e| format!("column {:?}: {}", c.out, e))
3122 } else {
3123 Ok(cell(v).map(|s| s.into_bytes()))
3124 };
3125 match got {
3126 Ok(b) => vals.push(b),
3127 Err(e) => { fail = Some(e); break; }
3128 }
3129 }
3130 if fail.is_some() {
3131 break;
3132 }
3133 encoded.push(data_row_bytes(&vals));
3134 }
3135 if let Some(why) = fail {
3136 sock.write_all(&err_msg("22P03", &why)).await?;
3137 failed = true;
3138 continue;
3139 }
3140 let mut out = vec![];
3141 for e in &encoded {
3142 out.extend_from_slice(e);
3143 }
3144 let r = portal.result.as_mut().expect("just executed");
3145 r.sent = limit;
3146 if max_rows > 0 && r.sent < r.rows.len() {
3150 out.extend_from_slice(&portal_suspended());
3151 } else {
3152 let tag = if r.tag_counts_rows {
3153 format!("{} {}", r.tag, r.sent)
3154 } else {
3155 r.tag.clone()
3156 };
3157 out.extend_from_slice(&command_complete(&tag));
3158 }
3159 sock.write_all(&out).await?;
3160 }
3161
3162 b'C' => {
3164 let kind = body.first().copied().unwrap_or(b'S');
3165 let mut at = 1usize;
3166 let name = take_cstr(&body, &mut at);
3167 if kind == b'S' {
3168 prepared.remove(&name);
3169 } else {
3170 portals.remove(&name);
3171 }
3172 sock.write_all(&close_complete()).await?;
3175 }
3176
3177 b'H' => {}
3181
3182 b'S' => {
3183 failed = false;
3184 sock.write_all(&ready()).await?;
3185 }
3186
3187 other => {
3188 sock.write_all(&err_msg(
3189 "08P01",
3190 &format!("unexpected frontend message {:?}", other as char),
3191 )).await?;
3192 failed = true;
3193 }
3194 }
3195 }
3196}
3197
3198const READ_ONLY_MSG: &str =
3199 "this endpoint is running read-only (NEDBD_PG_READ_ONLY=1). Writes are \
3200 implemented but disabled on this server — unset the flag to allow them.";
3201
3202fn no_db(db_name: &str) -> Vec<u8> {
3203 err_msg("3D000", &format!(
3204 "database {:?} is not open on this server — create it first \
3205 (POST /v1/databases), or connect with -d <name>", db_name))
3206}
3207
3208fn catalog_name(n: &str) -> String {
3212 let joined: Vec<&str> = n.split('.').collect();
3213 if joined.len() >= 2 && joined[joined.len() - 2] == "information_schema" {
3214 format!("information_schema.{}", joined[joined.len() - 1])
3215 } else {
3216 joined[joined.len() - 1].to_string()
3217 }
3218}
3219
3220#[derive(Default, Clone)]
3265struct TableVerbs {
3266 valid_as_of: Option<String>,
3267 search: Option<String>,
3268 trace: Option<String>,
3270 trace_reverse: bool,
3272 traverse: Option<String>,
3274}
3275
3276impl TableVerbs {
3277 fn first_unsupported_on_catalogue(&self) -> Option<&'static str> {
3279 if self.valid_as_of.is_some() {
3280 Some("VALID AS OF")
3281 } else if self.search.is_some() {
3282 Some("SEARCH")
3283 } else if self.trace.is_some() {
3284 Some("TRACE")
3285 } else if self.traverse.is_some() {
3286 Some("TRAVERSE")
3287 } else {
3288 None
3289 }
3290 }
3291
3292}
3293
3294fn stale_sql_engine_flag() -> bool {
3302 use std::sync::OnceLock;
3303 static ON: OnceLock<bool> = OnceLock::new();
3304 *ON.get_or_init(|| {
3305 let set = std::env::var("NEDBD_SQL_ENGINE").is_ok();
3306 if set {
3307 eprintln!(
3308 "[nedbd] NEDBD_SQL_ENGINE is set but no longer does anything. The SQL \
3309 evaluator now answers every SELECT it can parse; statements it cannot \
3310 parse still fall through to the translator. You can remove the variable."
3311 );
3312 }
3313 set
3314 })
3315}
3316
3317fn compose_prefiltered(cname: &str, scan: &crate::relation::Scan, pre: &str) -> String {
3331 let mut q = format!("FROM {}", cname);
3332 if let Some(seq) = scan.as_of {
3333 q.push_str(&format!(" AS OF {}", seq));
3334 }
3335 if let Some(d) = &scan.valid_as_of {
3336 q.push_str(&format!(" VALID AS OF {}", nql_string(d)));
3337 }
3338 q.push_str(&format!(" WHERE {}", pre));
3339 if let Some(t) = &scan.search {
3340 q.push_str(&format!(" SEARCH {}", nql_string(t)));
3341 }
3342 if let Some(edge) = &scan.trace {
3343 q.push_str(&format!(" TRACE {}", edge));
3344 if scan.trace_reverse {
3345 q.push_str(" REVERSE");
3346 }
3347 }
3348 if let Some(rel) = &scan.traverse {
3349 q.push_str(&format!(" TRAVERSE {}", rel));
3350 }
3351 q
3352}
3353
3354fn sql_engine_owns(sql: &str) -> bool {
3355 let Ok(sel) = crate::sqlselect::parse(sql) else { return false };
3356 let touched = sel.base_relations();
3357 if touched.is_empty() {
3358 return translate(sql).is_err();
3359 }
3360 if touched.iter().any(|t| crate::pgcatalog::is_catalog(&catalog_name(t))) {
3361 return true;
3362 }
3363 let _ = stale_sql_engine_flag();
3393
3394 !force_translator_for_parity()
3410}
3411
3412fn force_translator_for_parity() -> bool {
3419 use std::sync::OnceLock;
3420 static ON: OnceLock<bool> = OnceLock::new();
3421 *ON.get_or_init(|| {
3422 let on = matches!(
3423 std::env::var("NEDB_PARITY_FORCE_TRANSLATOR").as_deref(),
3424 Ok("1") | Ok("true") | Ok("on")
3425 );
3426 if on {
3427 eprintln!(
3428 "[nedbd] NEDB_PARITY_FORCE_TRANSLATOR is set — user collections are being \
3429 answered by the TRANSLATOR. This is a test lever for the parity harness, \
3430 not a supported configuration: projections are dropped on this path."
3431 );
3432 }
3433 on
3434 })
3435}
3436
3437fn try_catalog_select(
3450 sql: &str,
3451 db: Option<&Arc<Db>>,
3452) -> Result<Option<(Executed, crate::sqlplan::Plan)>, Vec<u8>> {
3453 let sel = match crate::sqlselect::parse(sql) {
3454 Ok(sel) => sel,
3455 Err(why) => {
3456 if mentions_catalog(sql) {
3465 return Err(err_msg("0A000", &format!(
3466 "this catalogue query uses SQL this endpoint does not \
3467 implement: {}", why)));
3468 }
3469 return Ok(None);
3470 }
3471 };
3472
3473 if !sql_engine_owns(sql) {
3479 return Ok(None);
3480 }
3481
3482 let temporal: std::collections::HashMap<String, u64> = {
3509 let mut seen: std::collections::HashMap<String, Vec<Option<u64>>> =
3514 std::collections::HashMap::new();
3515 for t in sel.from.iter().chain(sel.joins.iter().map(|j| &j.table)) {
3516 seen.entry(catalog_name(&t.name).to_ascii_lowercase())
3517 .or_default()
3518 .push(t.as_of);
3519 }
3520 let mut out: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
3521 for (key, ats) in &seen {
3522 let mut distinct: Vec<Option<u64>> = ats.clone();
3523 distinct.sort();
3524 distinct.dedup();
3525 match distinct.as_slice() {
3526 [Some(seq)] => {
3528 out.insert(key.clone(), *seq);
3529 }
3530 [None] => {}
3531 _ => {
3534 let mixed_tip = distinct.contains(&None);
3535 let seqs: Vec<String> =
3536 distinct.iter().flatten().map(|s| s.to_string()).collect();
3537 let detail = if mixed_tip {
3538 format!(
3539 "at the tip and AS OF {}",
3540 seqs.join(" and "))
3541 } else {
3542 format!("AS OF {}", seqs.join(" and "))
3543 };
3544 return Err(err_msg("0A000", &format!(
3545 "{:?} is read {} in one statement. This endpoint reads each \
3546 collection once per statement, so it cannot serve both — and \
3547 answering from either one would silently return the same rows for \
3548 both arms, which is the comparison failing to be a comparison. Ask \
3549 the two questions separately.",
3550 key, detail)));
3551 }
3552 }
3553 }
3554 out
3555 };
3556
3557 let nql_verbs: std::collections::HashMap<String, TableVerbs> = {
3564 let mut out: std::collections::HashMap<String, TableVerbs> =
3565 std::collections::HashMap::new();
3566 for t in sel.from.iter().chain(sel.joins.iter().map(|j| &j.table)) {
3567 let k = catalog_name(&t.name).to_ascii_lowercase();
3568 let e = out.entry(k.clone()).or_default();
3569 if t.trace.is_some() {
3574 e.trace_reverse = t.trace_reverse;
3575 }
3576 for (slot, incoming, verb) in [
3577 (&mut e.valid_as_of, &t.valid_as_of, "VALID AS OF"),
3578 (&mut e.search, &t.search, "SEARCH"),
3579 (&mut e.trace, &t.trace, "TRACE"),
3580 (&mut e.traverse, &t.traverse, "TRAVERSE"),
3581 ] {
3582 match (slot.as_deref(), incoming.as_deref()) {
3583 (Some(a), Some(b)) if a != b => {
3584 return Err(err_msg("0A000", &format!(
3585 "{:?} is read with two different {} arguments in one statement \
3586 ({:?} and {:?}). This endpoint reads each collection once, so \
3587 it cannot serve both. Ask the two questions separately.",
3588 k, verb, a, b)));
3589 }
3590 (None, Some(b)) => *slot = Some(b.to_string()),
3591 _ => {}
3592 }
3593 }
3594 }
3595 out
3596 };
3597
3598 let pushdown_prefilters: std::collections::HashMap<String, String> = {
3599 let refs: Vec<&crate::sqlselect::TableRef> = sel
3600 .from
3601 .iter()
3602 .chain(sel.joins.iter().map(|j| &j.table))
3603 .collect();
3604 let bindings: Vec<String> = refs.iter().map(|t| t.binding()).collect();
3605 let nullable = crate::sqlpush::nullable_bindings(&sel);
3606 let mut out = std::collections::HashMap::new();
3607 let mut ambiguous: Vec<String> = vec![];
3608 for t in &refs {
3609 let key = catalog_name(&t.name).to_ascii_lowercase();
3610 if out.contains_key(&key) || ambiguous.contains(&key) {
3611 out.remove(&key);
3612 ambiguous.push(key);
3613 continue;
3614 }
3615 if let Some(p) = crate::sqlpush::nql_prefilter(
3616 sel.where_.as_ref(), &t.binding(), &bindings, &nullable) {
3617 out.insert(key, p);
3618 }
3619 }
3620 out
3621 };
3622
3623 let resolve = |name: &str| -> anyhow::Result<Option<Box<dyn crate::sqlselect::Relation>>> {
3624 let cname = catalog_name(name);
3625 {
3638 let k = cname.to_ascii_lowercase();
3639 let bad = if temporal.contains_key(&k) {
3640 Some("AS OF SYSTEM TIME")
3641 } else {
3642 nql_verbs.get(&k).and_then(|v| v.first_unsupported_on_catalogue())
3643 };
3644 if let Some(clause) = bad {
3645 if crate::pgcatalog::is_catalog(&cname) {
3646 anyhow::bail!(
3647 "{} is not supported on the catalogue relation {:?} — a catalogue is \
3648 synthesised from the store's current shape rather than read from the \
3649 log, so it has no history to reach and no document text to search. \
3650 Ignoring the clause would answer your question with present-day rows \
3651 and look like it worked",
3652 clause, cname);
3653 }
3654 }
3655 }
3656 if let Some(rows) = crate::pgcatalog::rows(&cname, db) {
3657 return Ok(Some(crate::sqlselect::from_vec(rows)));
3661 }
3662 let key = cname.to_ascii_lowercase();
3682 let pre = pushdown_prefilters.get(&key);
3683 let Some(db) = db else { return Ok(None) };
3706
3707 let verbs = nql_verbs.get(&key);
3718 let scan = crate::relation::Scan {
3719 coll: cname.to_string(),
3720 as_of: temporal.get(&key).copied(),
3721 valid_as_of: verbs.and_then(|v| v.valid_as_of.clone()),
3722 search: verbs.and_then(|v| v.search.clone()),
3723 trace: verbs.and_then(|v| v.trace.clone()),
3724 trace_reverse: verbs.map(|v| v.trace_reverse).unwrap_or(false),
3725 traverse: verbs.and_then(|v| v.traverse.clone()),
3726 trace_limit: crate::relation::DEFAULT_TRACE_LIMIT,
3727 };
3728
3729 if let Some(p) = pre {
3737 let filtered = compose_prefiltered(&cname, &scan, p);
3738 if let Ok((rows, _)) = crate::nql::query(db, &filtered) {
3739 return Ok(Some(crate::sqlselect::from_vec(rows)));
3740 }
3741 }
3742 let catalogue = crate::pgcatalog::is_catalog(&cname)
3776 || cname.starts_with("pg_")
3777 || cname.starts_with("information_schema.");
3778 let known = catalogue
3779 || db.collections().iter().any(|c| c == &cname)
3780 || !db.list_ids_including_deleted(&cname).is_empty();
3781
3782 if !known && !sel.joins.is_empty() {
3802 return Ok(None);
3803 }
3804 Ok(Some(crate::sqlselect::from_vec(crate::relation::read_json(db, &scan))))
3805 };
3806
3807 let (cols, rows, plan) = crate::sqlselect::execute_explain(
3808 &sel,
3809 &resolve,
3810 crate::sqljoin::JoinExec::Auto,
3811 )
3812 .map_err(|e| err_msg("42601", &e.to_string()))?;
3813
3814 Ok(Some((
3815 Executed {
3816 rows,
3817 project: cols
3821 .iter()
3822 .map(|c| Col::renamed(&c.key, &c.name))
3823 .collect(),
3824 has_rows: true,
3825 tag: "SELECT".into(),
3826 tag_counts_rows: true,
3827 },
3828 plan,
3829 )))
3830}
3831
3832fn strip_explain(sql: &str) -> Option<&str> {
3839 let t = sql.trim().trim_end_matches(';').trim();
3840 let mut rest = t.strip_prefix("EXPLAIN").or_else(|| t.strip_prefix("explain"))?;
3841 if !rest.starts_with(char::is_whitespace) {
3843 return None;
3844 }
3845 rest = rest.trim_start();
3846 loop {
3847 let low = rest.to_lowercase();
3848 if let Some(r) = low.strip_prefix("analyze").or_else(|| low.strip_prefix("analyse")) {
3849 if r.starts_with(char::is_whitespace) || r.is_empty() {
3850 rest = rest[rest.len() - r.len()..].trim_start();
3851 continue;
3852 }
3853 }
3854 if let Some(r) = low.strip_prefix("verbose") {
3855 if r.starts_with(char::is_whitespace) || r.is_empty() {
3856 rest = rest[rest.len() - r.len()..].trim_start();
3857 continue;
3858 }
3859 }
3860 break;
3861 }
3862 Some(rest)
3863}
3864
3865fn plan_result(lines: Vec<String>) -> Executed {
3868 Executed {
3869 rows: lines
3870 .into_iter()
3871 .map(|l| serde_json::json!({ "QUERY PLAN": l }))
3872 .collect(),
3873 project: vec![Col::same("QUERY PLAN")],
3874 has_rows: true,
3875 tag: "EXPLAIN".into(),
3876 tag_counts_rows: false,
3877 }
3878}
3879
3880fn mentions_catalog(sql: &str) -> bool {
3887 let low = sql.to_lowercase();
3888 low.contains("pg_catalog.")
3889 || low.contains("information_schema.")
3890 || low.contains("from pg_")
3891 || low.contains("join pg_")
3892}
3893
3894fn catalog_target(nql: &str) -> Option<String> {
3899 let coll = crate::nql::parse(nql).ok()?.coll;
3900 if crate::pgcatalog::is_catalog(&coll) {
3901 Some(coll)
3902 } else {
3903 None
3904 }
3905}
3906
3907fn wants_returning(sql: &str) -> bool {
3911 find_kw(&sql.to_uppercase(), "RETURNING").is_some()
3912}
3913
3914fn next_row_id() -> String {
3916 use std::sync::atomic::{AtomicU64, Ordering};
3917 static N: AtomicU64 = AtomicU64::new(0);
3918 let n = N.fetch_add(1, Ordering::Relaxed);
3919 let ts = std::time::SystemTime::now()
3920 .duration_since(std::time::UNIX_EPOCH)
3921 .map(|d| d.as_micros())
3922 .unwrap_or(0);
3923 format!("r{}{}", ts, n)
3924}
3925
3926pub struct Executed {
3934 pub rows: Vec<Value>,
3936 pub project: Vec<Col>,
3938 pub has_rows: bool,
3942 pub tag: String,
3945 pub tag_counts_rows: bool,
3947}
3948
3949impl Executed {
3950 fn nothing(tag: &str) -> Self {
3951 Executed { rows: vec![], project: vec![], has_rows: false, tag: tag.to_string(), tag_counts_rows: false }
3952 }
3953 fn tag_for(&self, sent: usize) -> String {
3955 if self.tag_counts_rows { format!("{} {}", self.tag, sent) } else { self.tag.clone() }
3956 }
3957}
3958
3959pub fn execute_sql(db: &Arc<Db>, sql: &str, read_only: bool)
3985 -> std::result::Result<Executed, String>
3986{
3987 execute_stmt(sql, "", Some(db), read_only).map_err(|wire| decode_wire_error(&wire))
3988}
3989
3990fn decode_wire_error(buf: &[u8]) -> String {
3997 let mut code: Option<String> = None;
3998 let mut msg: Option<String> = None;
3999 let body = if buf.len() > 5 { &buf[5..] } else { buf };
4001 let mut i = 0usize;
4002 while i < body.len() && body[i] != 0 {
4003 let field = body[i];
4004 i += 1;
4005 let start = i;
4006 while i < body.len() && body[i] != 0 { i += 1; }
4007 let text = String::from_utf8_lossy(&body[start..i]).into_owned();
4008 i += 1; match field {
4010 b'C' => code = Some(text),
4011 b'M' => msg = Some(text),
4012 _ => {}
4013 }
4014 }
4015 match (code, msg) {
4016 (Some(c), Some(m)) => format!("{} ({})", m, c),
4017 (None, Some(m)) => m,
4018 _ => format!(
4021 "the engine refused the statement and the error could not be decoded ({} bytes of wire response)", buf.len()
4022 ),
4023 }
4024}
4025
4026fn execute_stmt(
4027 stmt_sql: &str,
4028 db_name: &str,
4029 db: Option<&Arc<Db>>,
4030 read_only: bool,
4031) -> Result<Executed, Vec<u8>> {
4032 if let Some(inner) = strip_explain(stmt_sql) {
4041 if let Some((_, plan)) = try_catalog_select(inner, db)? {
4042 return Ok(plan_result(plan.render()));
4043 }
4044 let mut lines = vec![];
4045 match translate(inner) {
4046 Ok(_) => {
4047 lines.push(
4048 "NQL path — this statement is translated to NQL and \
4049 executed by the storage engine, not by the SQL evaluator."
4050 .to_string(),
4051 );
4052 lines.push(
4053 "No plan is reported, because the SQL evaluator is not \
4054 what runs it. Reporting one would describe a pipeline \
4055 that never executed."
4056 .to_string(),
4057 );
4058 lines.push(
4059 "The SQL evaluator (joins, CASE, scalar functions, a \
4060 hash-join planner) currently serves catalogue queries."
4061 .to_string(),
4062 );
4063 }
4064 Err(why) => lines.push(format!("cannot be executed: {why}")),
4065 }
4066 return Ok(plan_result(lines));
4067 }
4068
4069 if let Some((done, _plan)) = try_catalog_select(stmt_sql, db)? {
4070 return Ok(done);
4071 }
4072
4073 let stmt = translate(stmt_sql).map_err(|why| err_msg("0A000", &why))?;
4074
4075 macro_rules! need_db {
4078 () => {
4079 match db {
4080 Some(db) => db,
4081 None => return Err(no_db(db_name)),
4082 }
4083 };
4084 }
4085 macro_rules! need_write {
4086 () => {
4087 if read_only {
4088 return Err(err_msg("25006", READ_ONLY_MSG));
4089 }
4090 };
4091 }
4092
4093 match stmt {
4094 Stmt::Ok(tag) => Ok(Executed::nothing(if tag.is_empty() { "SELECT 0" } else { tag })),
4095
4096 Stmt::Canned { cols, row } => {
4097 let mut obj = serde_json::Map::new();
4100 for (c, v) in cols.iter().zip(row.iter()) {
4101 obj.insert(c.clone(), Value::String(v.clone()));
4102 }
4103 Ok(Executed {
4104 rows: vec![Value::Object(obj)],
4105 project: cols.iter().map(|c| Col::same(c)).collect(),
4106 has_rows: true,
4107 tag: "SELECT".into(),
4108 tag_counts_rows: true,
4109 })
4110 }
4111
4112 Stmt::Query { nql, project } => {
4113 if let Some(coll) = catalog_target(&nql) {
4123 let rows = crate::pgcatalog::rows(&coll, db)
4124 .expect("catalog_target only returns names pgcatalog serves");
4125 let rows = crate::nql::query_rows(rows, &nql)
4126 .map_err(|e| err_msg("42601", &e.to_string()))?;
4127 return Ok(Executed {
4128 rows, project, has_rows: true,
4129 tag: "SELECT".into(), tag_counts_rows: true,
4130 });
4131 }
4132 let db = need_db!();
4133 let (rows, _) = crate::nql::query(db, &nql).map_err(|e| {
4134 err_msg("42601", &format!("{} (translated to NQL: {})", e, nql))
4135 })?;
4136 Ok(Executed { rows, project, has_rows: true, tag: "SELECT".into(), tag_counts_rows: true })
4137 }
4138
4139 Stmt::Insert { coll, rows, returning } => {
4140 let db = need_db!();
4141 need_write!();
4142 let mut written: Vec<Value> = vec![];
4143 for (i, r) in rows.iter().enumerate() {
4144 let id = match &r.id {
4148 Some(id) => id.clone(),
4149 None => format!("{}-{}", next_row_id(), i),
4150 };
4151 let node = db
4152 .put(&coll, &id, Value::Object(r.doc.clone()),
4153 r.caused_by.clone(), r.valid_from.clone(), r.valid_to.clone())
4154 .map_err(|e| err_msg("XX000", &format!("INSERT failed: {}", e)))?;
4155 written.push(crate::nql::node_to_json(&node));
4156 }
4157 let n = written.len();
4158 let has_rows = wants_returning(stmt_sql);
4159 Ok(Executed {
4160 rows: if has_rows { written } else { vec![] },
4161 project: returning,
4162 has_rows,
4163 tag: format!("INSERT 0 {}", n),
4165 tag_counts_rows: false,
4166 })
4167 }
4168
4169 Stmt::Update { coll, set, nql, returning } => {
4170 let db = need_db!();
4171 need_write!();
4172 let (matched, _) = crate::nql::query(db, &nql).map_err(|e| {
4175 err_msg("42601", &format!("{} (translated to NQL: {})", e, nql))
4176 })?;
4177 let mut written: Vec<Value> = vec![];
4178 for row in &matched {
4179 let id = match row.get("_id").and_then(|v| v.as_str()) {
4180 Some(id) => id.to_string(),
4181 None => continue,
4182 };
4183 let mut doc = match db.get(&coll, &id) {
4187 Some(n) => match n.data {
4188 Value::Object(m) => m,
4189 _ => serde_json::Map::new(),
4190 },
4191 None => continue,
4192 };
4193 for (k, v) in &set {
4194 doc.insert(k.clone(), v.clone());
4195 }
4196 let node = db
4199 .put(&coll, &id, Value::Object(doc), vec![], None, None)
4200 .map_err(|e| err_msg("XX000", &format!("UPDATE failed: {}", e)))?;
4201 written.push(crate::nql::node_to_json(&node));
4202 }
4203 let n = written.len();
4204 let has_rows = wants_returning(stmt_sql);
4205 Ok(Executed {
4206 rows: if has_rows { written } else { vec![] },
4207 project: returning,
4208 has_rows,
4209 tag: format!("UPDATE {}", n),
4210 tag_counts_rows: false,
4211 })
4212 }
4213
4214 Stmt::Delete { coll, nql, returning } => {
4215 let db = need_db!();
4216 need_write!();
4217 let (matched, _) = crate::nql::query(db, &nql).map_err(|e| {
4218 err_msg("42601", &format!("{} (translated to NQL: {})", e, nql))
4219 })?;
4220 let returned = matched.clone();
4223 let mut n = 0usize;
4224 for row in &matched {
4225 if let Some(id) = row.get("_id").and_then(|v| v.as_str()) {
4226 match db.delete(&coll, id) {
4227 Ok(true) => n += 1,
4228 Ok(false) => {}
4229 Err(e) => return Err(err_msg("XX000", &format!("DELETE failed: {}", e))),
4230 }
4231 }
4232 }
4233 let has_rows = wants_returning(stmt_sql);
4234 Ok(Executed {
4235 rows: if has_rows { returned } else { vec![] },
4236 project: returning,
4237 has_rows,
4238 tag: format!("DELETE {}", n),
4239 tag_counts_rows: false,
4240 })
4241 }
4242 }
4243}
4244
4245fn run_simple_query(sql: &str, db_name: &str, db: Option<&Arc<Db>>, read_only: bool) -> Vec<u8> {
4247 let mut out = vec![];
4248 let statements = split_statements(sql);
4249 if statements.is_empty() {
4250 return Out::msg(b'I').finish();
4252 }
4253 for stmt_sql in statements {
4254 match execute_stmt(&stmt_sql, db_name, db, read_only) {
4255 Err(encoded) => {
4257 out.extend_from_slice(&encoded);
4258 return out;
4259 }
4260 Ok(ex) => {
4261 if ex.has_rows {
4262 out.extend_from_slice(&encode_rows(&ex.rows, &ex.project));
4263 }
4264 out.extend_from_slice(&command_complete(&ex.tag_for(ex.rows.len())));
4265 }
4266 }
4267 }
4268 out
4269}
4270
4271fn split_statements(sql: &str) -> Vec<String> {
4273 let mut out = vec![];
4274 let mut cur = String::new();
4275 let mut in_s = false;
4276 for c in sql.chars() {
4277 match c {
4278 '\'' => { in_s = !in_s; cur.push(c); }
4279 ';' if !in_s => {
4280 if !cur.trim().is_empty() { out.push(cur.clone()); }
4281 cur.clear();
4282 }
4283 _ => cur.push(c),
4284 }
4285 }
4286 if !cur.trim().is_empty() {
4287 out.push(cur);
4288 }
4289 out
4290}
4291
4292pub async fn run(host: &str, port: u16, resolver: Arc<dyn DbResolver>) -> anyhow::Result<()> {
4294 let read_only = std::env::var("NEDBD_PG_READ_ONLY")
4298 .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
4299 .unwrap_or(false);
4300 let listener = TcpListener::bind((host, port)).await?;
4301 println!(" pgwire postgres endpoint on {}:{} — psql / DBeaver / psycopg ({})",
4302 host, port,
4303 if read_only { "SELECT only — read-only mode" } else { "SELECT + INSERT/UPDATE/DELETE" });
4304 loop {
4305 let (sock, _peer) = match listener.accept().await {
4306 Ok(v) => v,
4307 Err(e) => {
4308 eprintln!(" [pgwire] accept failed: {}", e);
4309 continue;
4310 }
4311 };
4312 let r = Arc::clone(&resolver);
4313 tokio::spawn(async move {
4314 let _ = sock.set_nodelay(true);
4315 if let Err(e) = handle(sock, r, read_only).await {
4316 if e.kind() != std::io::ErrorKind::UnexpectedEof
4318 && e.kind() != std::io::ErrorKind::ConnectionReset
4319 {
4320 eprintln!(" [pgwire] connection error: {}", e);
4321 }
4322 }
4323 });
4324 }
4325}
4326
4327#[cfg(test)]
4330mod explain_tests {
4331 use super::*;
4332
4333 #[test]
4334 fn a_bare_explain_is_stripped() {
4335 assert_eq!(strip_explain("EXPLAIN SELECT 1"), Some("SELECT 1"));
4336 assert_eq!(strip_explain("explain select 1"), Some("select 1"));
4337 assert_eq!(strip_explain(" EXPLAIN SELECT 1 ; "), Some("SELECT 1"));
4338 }
4339
4340 #[test]
4341 fn analyze_and_verbose_are_accepted_and_ignored() {
4342 assert_eq!(strip_explain("EXPLAIN ANALYZE SELECT 1"), Some("SELECT 1"));
4346 assert_eq!(strip_explain("EXPLAIN ANALYSE SELECT 1"), Some("SELECT 1"));
4347 assert_eq!(strip_explain("EXPLAIN VERBOSE SELECT 1"), Some("SELECT 1"));
4348 assert_eq!(strip_explain("EXPLAIN ANALYZE VERBOSE SELECT 1"), Some("SELECT 1"));
4349 assert_eq!(strip_explain("explain analyze verbose select 1"), Some("select 1"));
4350 }
4351
4352 #[test]
4353 fn a_word_merely_starting_with_explain_is_not_a_keyword() {
4354 assert_eq!(strip_explain("EXPLAINED SELECT 1"), None);
4355 assert_eq!(strip_explain("SELECT 1"), None);
4356 assert_eq!(strip_explain("SELECT explain FROM t"), None);
4357 }
4358
4359 #[test]
4360 fn a_column_named_analyze_is_not_eaten() {
4361 assert_eq!(strip_explain("EXPLAIN analyzed_view"), Some("analyzed_view"));
4364 }
4365
4366 #[test]
4367 fn the_plan_result_has_postgres_shape() {
4368 let e = plan_result(vec!["Seq Scan on t".into(), "note".into()]);
4369 assert_eq!(e.project.len(), 1);
4370 assert_eq!(e.project[0].out, "QUERY PLAN");
4371 assert_eq!(e.rows.len(), 2);
4372 assert_eq!(e.rows[0]["QUERY PLAN"], "Seq Scan on t");
4373 assert_eq!(e.tag, "EXPLAIN");
4374 assert!(!e.tag_counts_rows);
4376 }
4377}
4378
4379#[cfg(test)]
4380mod tests {
4381 use super::*;
4382 use serde_json::json;
4383
4384 fn q(sql: &str) -> String {
4385 match translate(sql) {
4386 Ok(Stmt::Query { nql, .. }) => nql,
4387 other => panic!("expected a query for {:?}, got {:?}", sql, other),
4388 }
4389 }
4390 fn proj(sql: &str) -> Vec<String> {
4392 match translate(sql) {
4393 Ok(Stmt::Query { project, .. }) => project.iter().map(|c| c.out.clone()).collect(),
4394 other => panic!("expected a query for {:?}, got {:?}", sql, other),
4395 }
4396 }
4397 fn proj_pairs(sql: &str) -> Vec<(String, String)> {
4399 match translate(sql) {
4400 Ok(Stmt::Query { project, .. }) =>
4401 project.iter().map(|c| (c.src.clone(), c.out.clone())).collect(),
4402 other => panic!("expected a query for {:?}, got {:?}", sql, other),
4403 }
4404 }
4405 fn names(cols: &[Col]) -> Vec<String> { cols.iter().map(|c| c.out.clone()).collect() }
4406
4407 fn cols_of(sql: &str) -> Vec<Col> {
4411 match translate(sql).unwrap() {
4412 Stmt::Query { project, .. } => project,
4413 other => panic!("{:?}", other),
4414 }
4415 }
4416
4417 #[test]
4418 fn select_star_becomes_bare_from() {
4419 assert_eq!(q("SELECT * FROM orders"), "FROM orders");
4420 assert_eq!(q("select * from orders;"), "FROM orders");
4421 assert_eq!(proj("SELECT * FROM orders"), Vec::<String>::new());
4422 }
4423
4424 #[test]
4425 fn a_column_list_becomes_a_projection_not_a_clause() {
4426 assert_eq!(q("SELECT status, total FROM orders"), "FROM orders");
4429 assert_eq!(proj("SELECT status, total FROM orders"), vec!["status", "total"]);
4430 }
4431
4432 #[test]
4433 fn a_qualifier_reduces_to_the_field_while_an_ALIAS_is_the_name_the_client_sees() {
4434 let cols = cols_of("SELECT o.status AS s, o.total total, o.region FROM orders o");
4442 assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
4443 vec!["status", "total", "region"]);
4444 assert_eq!(cols.iter().map(|c| c.out.clone()).collect::<Vec<_>>(),
4445 vec!["s", "total", "region"]);
4446 assert_eq!(q("SELECT * FROM public.orders"), "FROM orders");
4447 assert_eq!(q("SELECT * FROM \"orders\""), "FROM orders");
4448 }
4449
4450 #[test]
4451 fn a_select_list_may_MIX_columns_with_an_aggregate() {
4452 assert_eq!(q("SELECT status, count(*) AS count_1 FROM orders GROUP BY status"),
4460 "FROM orders GROUP BY status COUNT");
4461 assert_eq!(q("SELECT status, count(*) FROM orders WHERE total > 1 GROUP BY status ORDER BY status LIMIT 5"),
4464 "FROM orders WHERE total > 1 GROUP BY status COUNT ORDER BY status LIMIT 5");
4465 assert_eq!(q("SELECT count(*) FROM orders"), "FROM orders COUNT");
4467 assert_eq!(q("SELECT sum(total) FROM orders"), "FROM orders SUM total");
4468 let e = translate("SELECT status, count(*) FROM orders GROUP BY status, region").unwrap_err();
4472 assert!(e.contains("GROUP BY takes one key"), "{}", e);
4473 let cols = cols_of("SELECT status, count(*) AS count_1 FROM orders GROUP BY status");
4474 assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
4475 vec!["status", "count"]);
4476 assert_eq!(cols.iter().map(|c| c.out.clone()).collect::<Vec<_>>(),
4477 vec!["status", "count_1"]);
4478
4479 let cols = cols_of("SELECT status, count(*), sum(total) FROM orders GROUP BY status");
4482 assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
4483 vec!["status", "count", "sum_total"]);
4484 assert_eq!(q("SELECT status, count(*), sum(total) FROM orders GROUP BY status"),
4485 "FROM orders GROUP BY status SUM total");
4486
4487 assert_eq!(q("SELECT o.status, sum(o.total) FROM orders o GROUP BY o.status"),
4489 "FROM orders GROUP BY status SUM total");
4490
4491 let e = translate("SELECT status, sum(total), avg(total) FROM orders GROUP BY status")
4494 .unwrap_err();
4495 assert!(e.contains("only one of SUM/AVG/MIN/MAX"), "{}", e);
4496
4497 let e = translate("SELECT status, total, count(*) FROM orders GROUP BY status")
4499 .unwrap_err();
4500 assert!(e.contains("must appear in the GROUP BY clause"), "{}", e);
4501 }
4502
4503 #[test]
4504 fn ORDER_BY_an_ordinal_resolves_to_that_select_list_column() {
4505 assert_eq!(q("SELECT status, total FROM orders ORDER BY 1"),
4510 "FROM orders ORDER BY status");
4511 assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 DESC"),
4512 "FROM orders ORDER BY total DESC");
4513 assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 DESC, 1"),
4515 "FROM orders ORDER BY total DESC, status");
4516 assert_eq!(q("SELECT status, total FROM orders ORDER BY 1, total DESC"),
4517 "FROM orders ORDER BY status, total DESC");
4518 assert_eq!(q("SELECT status, count(*) AS n FROM orders GROUP BY status ORDER BY 1"),
4522 "FROM orders GROUP BY status COUNT ORDER BY status");
4523 assert_eq!(q("SELECT status, count(*) AS n FROM orders GROUP BY status ORDER BY 2 DESC"),
4525 "FROM orders GROUP BY status COUNT ORDER BY count DESC");
4526 assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 LIMIT 1"),
4529 "FROM orders ORDER BY total LIMIT 1");
4530 assert_eq!(q("SELECT status FROM orders WHERE total > 1 ORDER BY 1"),
4532 "FROM orders WHERE total > 1 ORDER BY status");
4533
4534 let e = translate("SELECT status FROM orders ORDER BY 4").unwrap_err();
4538 assert!(e.contains("out of range") && e.contains("1 column"), "{}", e);
4539 let e = translate("SELECT * FROM orders ORDER BY 1").unwrap_err();
4540 assert!(e.contains("no list to index"), "{}", e);
4541 }
4542
4543 #[test]
4544 fn count_of_a_subquery_flattens_only_when_the_two_counts_MUST_agree() {
4545 assert_eq!(
4549 q("SELECT count(*) AS count_1 FROM (SELECT orders._id AS a, orders.status AS b \
4550 FROM orders WHERE orders.status = 'paid') AS anon_1"),
4551 r#"FROM orders COUNT WHERE status = "paid""#);
4554 assert_eq!(q("SELECT count(*) FROM (SELECT orders._id FROM orders) AS anon_1"),
4556 "FROM orders COUNT");
4557 assert_eq!(q("SELECT count(*) FROM (SELECT _id FROM orders ORDER BY total DESC) AS a"),
4559 "FROM orders COUNT");
4560 let cols = cols_of("SELECT count(*) AS count_1 FROM (SELECT _id FROM orders) AS a");
4562 assert_eq!(cols[0].src, "count");
4563 assert_eq!(cols[0].out, "count_1");
4564
4565 for sql in [
4568 "SELECT count(*) FROM (SELECT _id FROM orders LIMIT 1) AS a",
4570 "SELECT count(*) FROM (SELECT _id FROM orders OFFSET 1) AS a",
4571 "SELECT count(*) FROM (SELECT status FROM orders GROUP BY status) AS a",
4573 "SELECT count(*) FROM (SELECT count(*) FROM orders) AS a",
4575 "SELECT count(*) FROM (SELECT sum(total) FROM orders) AS a",
4576 "SELECT count(*), status FROM (SELECT status FROM orders) AS a",
4578 "SELECT status FROM (SELECT status FROM orders) AS a",
4579 "SELECT count(*) FROM (SELECT x FROM (SELECT _id AS x FROM orders) AS b) AS a",
4581 ] {
4582 let e = translate(sql).unwrap_err();
4583 assert!(e.contains("subqueries in FROM"), "{} -> {}", sql, e);
4584 }
4585
4586 for (sql, needle) in [
4591 ("SELECT count(*) FROM (SELECT DISTINCT status FROM orders) AS a", "DISTINCT"),
4592 ("SELECT count(*) FROM (SELECT a FROM t UNION SELECT b FROM u) AS x", "UNION"),
4593 ] {
4594 let e = translate(sql).unwrap_err();
4595 assert!(e.contains(needle), "{} -> {}", sql, e);
4596 }
4597 }
4598
4599 #[test]
4600 fn a_QUALIFIED_column_in_WHERE_finds_its_field_instead_of_ZERO_ROWS() {
4601 assert_eq!(q("SELECT _id FROM orders WHERE orders.status = 'paid'"),
4609 r#"FROM orders WHERE status = "paid""#);
4610 assert_eq!(q("SELECT _id FROM orders WHERE orders.total > 50"),
4611 "FROM orders WHERE total > 50");
4612 assert_eq!(q("SELECT _id FROM orders ORDER BY orders.total DESC LIMIT 2"),
4614 "FROM orders ORDER BY total DESC LIMIT 2");
4615 assert_eq!(q("SELECT status, count(*) FROM orders GROUP BY orders.status"),
4616 "FROM orders GROUP BY status COUNT");
4617
4618 assert_eq!(q("SELECT o.status FROM orders o WHERE o.status = 'paid'"),
4622 r#"FROM orders WHERE status = "paid""#);
4623 assert_eq!(q("SELECT o.status FROM orders AS o WHERE o.total > 1"),
4624 "FROM orders WHERE total > 1");
4625
4626 let e = translate("SELECT _id FROM orders WHERE nosuch.status = 'paid'").unwrap_err();
4631 assert!(e.contains("no table or alias named \"nosuch\""), "{}", e);
4632 let e = translate("SELECT _id FROM orders o WHERE p.status = 'paid'").unwrap_err();
4633 assert!(e.contains("aliased \"o\""), "the message names the alias in scope: {}", e);
4634
4635 assert_eq!(q("SELECT _id FROM orders WHERE status = 'pa.id'"),
4637 r#"FROM orders WHERE status = "pa.id""#);
4638 assert_eq!(q("SELECT _id FROM orders WHERE total > 1.5"),
4640 "FROM orders WHERE total > 1.5");
4641
4642 match translate("UPDATE orders o SET status = 'x' WHERE o.total > 5").unwrap() {
4644 Stmt::Update { coll, nql, .. } => {
4645 assert_eq!(coll, "orders", "the alias is not part of the collection name");
4646 assert_eq!(nql, "FROM orders WHERE total > 5");
4647 }
4648 other => panic!("{:?}", other),
4649 }
4650 match translate("DELETE FROM orders o WHERE o.status = 'paid'").unwrap() {
4651 Stmt::Delete { coll, nql, .. } => {
4652 assert_eq!(coll, "orders");
4653 assert_eq!(nql, r#"FROM orders WHERE status = "paid""#);
4654 }
4655 other => panic!("{:?}", other),
4656 }
4657
4658 assert_eq!(q("SELECT _id FROM orders AS OF SYSTEM TIME 3 WHERE orders.total > 1"),
4660 "FROM orders AS OF 3 WHERE total > 1");
4661 }
4662
4663 #[test]
4664 fn where_clauses_pass_through_with_sql_literals_rewritten() {
4665 assert_eq!(q("SELECT * FROM orders WHERE status = 'paid'"),
4666 r#"FROM orders WHERE status = "paid""#);
4667 assert_eq!(q("SELECT * FROM orders WHERE status <> 'paid'"),
4668 r#"FROM orders WHERE status != "paid""#);
4669 assert_eq!(q("SELECT * FROM orders WHERE status IN ('paid','open')"),
4670 r#"FROM orders WHERE status IN ("paid","open")"#);
4671 }
4672
4673 #[test]
4676 fn a_doubled_sql_quote_is_one_literal_character() {
4677 assert_eq!(q("SELECT * FROM t WHERE name = 'it''s'"),
4678 r#"FROM t WHERE name = "it's""#);
4679 }
4680
4681 #[test]
4684 fn a_double_quote_inside_a_sql_literal_is_escaped_for_nql() {
4685 assert_eq!(q(r#"SELECT * FROM t WHERE name = 'say "hi"'"#),
4686 r#"FROM t WHERE name = "say \"hi\"""#);
4687 }
4688
4689 #[test]
4690 fn the_shared_clauses_are_handed_to_nql_unchanged() {
4691 assert_eq!(q("SELECT * FROM orders ORDER BY total DESC LIMIT 10 OFFSET 5"),
4692 "FROM orders ORDER BY total DESC LIMIT 10 OFFSET 5");
4693 assert_eq!(q("SELECT * FROM orders GROUP BY region"), "FROM orders GROUP BY region");
4694 assert_eq!(q("SELECT * FROM o WHERE total BETWEEN 1 AND 9 ORDER BY a, b DESC"),
4695 "FROM o WHERE total BETWEEN 1 AND 9 ORDER BY a, b DESC");
4696 }
4697
4698 #[test]
4705 fn an_aggregate_is_one_column_named_as_sql_names_it() {
4706 assert_eq!(proj_pairs("SELECT COUNT(*) FROM orders"),
4707 vec![("count".to_string(), "count".to_string())]);
4708 assert_eq!(proj_pairs("SELECT SUM(total) FROM orders"),
4709 vec![("sum_total".to_string(), "sum".to_string())]);
4710 assert_eq!(proj_pairs("SELECT avg(total) FROM orders"),
4711 vec![("avg_total".to_string(), "avg".to_string())]);
4712 assert_eq!(proj_pairs("SELECT MIN(total) FROM orders"),
4713 vec![("min_total".to_string(), "min".to_string())]);
4714 let rows = vec![json!({"count": 4, "sum_total": 420, "value": 420})];
4716 let p = vec![Col::renamed("sum_total", "sum")];
4717 let cols = columns_for(&rows, &p);
4718 assert_eq!(names(&cols), vec!["sum"], "one column, SQL's name");
4719 assert_eq!(cell(rows[0].get(&cols[0].src)), Some("420".to_string()));
4720 }
4721
4722 #[test]
4727 fn a_bare_column_with_group_by_is_refused_not_nulled() {
4728 let e = translate("SELECT region, total FROM orders GROUP BY region").unwrap_err();
4729 assert!(e.contains("must appear in the GROUP BY clause"), "{}", e);
4730 assert!(e.contains("total"), "the message names the offending column: {}", e);
4731
4732 assert!(translate("SELECT region FROM orders GROUP BY region").is_ok());
4734 assert!(translate("SELECT region, count FROM orders GROUP BY region").is_ok());
4735 assert!(translate("SELECT SUM(total) FROM orders GROUP BY region").is_ok());
4737 assert!(translate("SELECT * FROM orders GROUP BY region").is_ok());
4739 }
4740
4741 #[test]
4742 fn count_star_becomes_nql_count() {
4743 assert_eq!(q("SELECT COUNT(*) FROM orders"), "FROM orders COUNT");
4744 assert_eq!(q("SELECT count(*) FROM orders WHERE total > 5"),
4745 "FROM orders COUNT WHERE total > 5");
4746 }
4747
4748 #[test]
4749 fn aggregates_carry_their_target_column() {
4750 assert_eq!(q("SELECT SUM(total) FROM orders"), "FROM orders SUM total");
4751 assert_eq!(q("SELECT avg(total) FROM orders WHERE region = 'eu'"),
4752 r#"FROM orders AVG total WHERE region = "eu""#);
4753 assert!(translate("SELECT SUM(*) FROM orders").is_err());
4754 }
4755
4756 #[test]
4759 fn as_of_system_time_bridges_to_nql_as_of() {
4760 assert_eq!(q("SELECT * FROM orders AS OF SYSTEM TIME 42"),
4761 "FROM orders AS OF 42");
4762 assert_eq!(q("SELECT * FROM orders AS OF SYSTEM TIME 42 WHERE total > 1"),
4763 "FROM orders AS OF 42 WHERE total > 1");
4764 let e = translate("SELECT * FROM orders AS OF SYSTEM TIME '2026-01-01'").unwrap_err();
4766 assert!(e.contains("sequence number"), "{}", e);
4767 }
4768
4769 #[test]
4778 fn a_select_list_expression_is_refused_rather_than_answered_blank() {
4779 for sql in [
4780 "SELECT total * 2 FROM orders",
4781 "SELECT total, total*2 AS doubled FROM orders",
4782 "SELECT total + 1 FROM orders",
4783 "SELECT status || 'x' FROM orders",
4784 "SELECT -total FROM orders",
4785 "SELECT lower(status) FROM orders",
4786 ] {
4787 let e = translate(sql).unwrap_err();
4788 assert!(e.contains("expressions in the select list"), "{} -> {}", sql, e);
4789 }
4790 assert_eq!(q("SELECT _id, status FROM orders"), "FROM orders");
4793 assert_eq!(q("SELECT \"status\" FROM orders"), "FROM orders");
4794 assert_eq!(q("SELECT orders.status FROM orders"), "FROM orders");
4795 assert_eq!(q("SELECT o.status FROM orders o"), "FROM orders");
4796 assert_eq!(q("SELECT total AS t FROM orders"), "FROM orders");
4797 assert!(translate("SELECT count(*) FROM orders").is_ok());
4798 assert!(translate("SELECT sum(total) FROM orders").is_ok());
4799 }
4800
4801 #[test]
4809 fn having_is_translated_to_nqls_spelling_and_refuses_an_unknown_key() {
4810 for sql in [
4812 "SELECT status, count(*) AS n FROM orders GROUP BY status HAVING count(*) > 1",
4813 "SELECT status, count(*) AS n FROM orders GROUP BY status HAVING n > 1",
4814 "SELECT status, count(*) FROM orders GROUP BY status HAVING COUNT > 1",
4815 "SELECT status, count(*) FROM orders GROUP BY status HAVING count > 1",
4816 ] {
4817 let got = q(sql);
4818 assert_eq!(got, "FROM orders GROUP BY status COUNT HAVING count > 1",
4819 "{} -> {}", sql, got);
4820 }
4821 assert_eq!(q("SELECT status, sum(total) AS s FROM orders GROUP BY status HAVING s > 100"),
4823 "FROM orders GROUP BY status SUM total HAVING sum_total > 100");
4824 assert_eq!(q("SELECT status, sum(total) FROM orders GROUP BY status HAVING sum_total > 100"),
4826 "FROM orders GROUP BY status SUM total HAVING sum_total > 100");
4827 assert_eq!(q("SELECT status, count(*) FROM orders GROUP BY status HAVING status > 'a'"),
4830 "FROM orders GROUP BY status COUNT HAVING status > \"a\"");
4831 let e = translate(
4833 "SELECT status, count(*) FROM orders GROUP BY status HAVING nosuch > 1").unwrap_err();
4834 assert!(e.contains("HAVING names") && e.contains("nosuch"), "{}", e);
4835 assert!(e.contains("zero rows"), "the message must say what it prevented: {}", e);
4836 }
4837
4838 #[test]
4839 fn handshake_queries_are_answered_so_clients_can_connect() {
4840 assert!(matches!(translate("SELECT version()"), Ok(Stmt::Canned { .. })));
4841 assert!(matches!(translate("SHOW transaction_isolation"), Ok(Stmt::Canned { .. })));
4842 assert!(matches!(translate("SELECT current_schema()"), Ok(Stmt::Canned { .. })));
4843 assert!(matches!(translate("SET extra_float_digits = 3"), Ok(Stmt::Ok(_))));
4844 assert!(matches!(translate("BEGIN"), Ok(Stmt::Ok(_))));
4845 assert!(matches!(translate(""), Ok(Stmt::Ok(_))));
4846 }
4847
4848 #[test]
4851 fn unsupported_sql_is_refused_with_a_reason() {
4852 for (sql, expect) in [
4853 ("INSERT INTO t VALUES (1)", "explicit column list"),
4854 ("CREATE TABLE t (a int)", "DDL"),
4855 ("TRUNCATE t", "append-only"),
4856 ("GRANT ALL ON t TO x", "privilege system"),
4857 ("SELECT * FROM a JOIN b ON a.x = b.x", "JOIN is not supported"),
4858 ("SELECT * FROM a UNION SELECT * FROM b", "UNION"),
4859 ("SELECT DISTINCT region FROM orders", "GROUP BY"),
4860 ("SELECT * FROM (SELECT 1) x", "subqueries in FROM"),
4861 ("SELECT * FROM a, b", "more than one collection"),
4862 ("SELECT lower(status) FROM orders", "expressions in the select list"),
4863 ("VACUUM", "only SELECT"),
4864 ] {
4865 let e = translate(sql).unwrap_err();
4866 assert!(e.contains(expect), "for {:?} expected {:?} in {:?}", sql, expect, e);
4867 }
4868 }
4869
4870 fn ins(sql: &str) -> (String, Vec<InsertRow>, Vec<Col>) {
4878 match translate(sql) {
4879 Ok(Stmt::Insert { coll, rows, returning }) => (coll, rows, returning),
4880 other => panic!("expected INSERT for {:?}, got {:?}", sql, other),
4881 }
4882 }
4883
4884 #[test]
4885 fn insert_becomes_a_put_per_row() {
4886 let (coll, rows, ret) = ins("INSERT INTO orders (_id, status, total) VALUES ('o1', 'paid', 120)");
4887 assert_eq!(coll, "orders");
4888 assert_eq!(rows.len(), 1);
4889 assert_eq!(rows[0].id.as_deref(), Some("o1"));
4890 assert_eq!(rows[0].doc.get("status"), Some(&json!("paid")));
4891 assert_eq!(rows[0].doc.get("total"), Some(&json!(120)));
4892 assert!(!rows[0].doc.contains_key("_id"));
4894 assert!(ret.is_empty());
4895 }
4896
4897 #[test]
4898 fn a_multi_row_insert_yields_one_row_each() {
4899 let (_, rows, _) = ins(
4900 "INSERT INTO t (id, n) VALUES ('a', 1), ('b', 2), ('c', 3)");
4901 assert_eq!(rows.len(), 3);
4902 assert_eq!(rows[1].id.as_deref(), Some("b"));
4903 assert_eq!(rows[2].doc.get("n"), Some(&json!(3)));
4904 }
4905
4906 #[test]
4907 fn an_insert_without_an_id_column_lets_the_server_assign_one() {
4908 let (_, rows, _) = ins("INSERT INTO t (n) VALUES (1)");
4909 assert_eq!(rows[0].id, None, "the executor mints a unique key");
4910 assert_eq!(rows[0].doc.get("n"), Some(&json!(1)));
4911 }
4912
4913 #[test]
4916 fn insert_lifts_provenance_out_of_reserved_columns() {
4917 let (_, rows, _) = ins(
4918 "INSERT INTO audit (_id, _caused_by, _valid_from, kind) \
4919 VALUES ('e1', 'abc123', '2026-01-01', 'reprice')");
4920 assert_eq!(rows[0].caused_by, vec!["abc123".to_string()]);
4921 assert_eq!(rows[0].valid_from.as_deref(), Some("2026-01-01"));
4922 assert_eq!(rows[0].doc.get("kind"), Some(&json!("reprice")));
4923 for k in ["_id", "_caused_by", "_valid_from"] {
4925 assert!(!rows[0].doc.contains_key(k), "{} leaked into the doc", k);
4926 }
4927 }
4928
4929 #[test]
4930 fn insert_values_cover_the_scalar_types() {
4931 let (_, rows, _) = ins(
4932 "INSERT INTO t (s, i, f, b, n) VALUES ('x', 42, 1.5, TRUE, NULL)");
4933 assert_eq!(rows[0].doc.get("s"), Some(&json!("x")));
4934 assert_eq!(rows[0].doc.get("i"), Some(&json!(42)));
4935 assert_eq!(rows[0].doc.get("f"), Some(&json!(1.5)));
4936 assert_eq!(rows[0].doc.get("b"), Some(&json!(true)));
4937 assert_eq!(rows[0].doc.get("n"), Some(&Value::Null));
4938 }
4939
4940 #[test]
4943 fn insert_literals_survive_quotes_and_commas() {
4944 let (_, rows, _) = ins("INSERT INTO t (a, b) VALUES ('it''s', 'x,y')");
4945 assert_eq!(rows[0].doc.get("a"), Some(&json!("it's")));
4946 assert_eq!(rows[0].doc.get("b"), Some(&json!("x,y")));
4947 }
4948
4949 #[test]
4950 fn insert_refuses_what_it_cannot_store_faithfully() {
4951 assert!(translate("INSERT INTO t (a) VALUES (1 + 1)").is_err());
4953 assert!(translate("INSERT INTO t (a) VALUES (now())").is_err());
4954 let e = translate("INSERT INTO t (a, b) VALUES (1)").unwrap_err();
4956 assert!(e.contains("values for"), "{}", e);
4957 let e2 = translate("INSERT INTO t VALUES (1)").unwrap_err();
4959 assert!(e2.contains("explicit column list"), "{}", e2);
4960 }
4961
4962 #[test]
4963 fn update_finds_rows_with_the_full_predicate_surface() {
4964 match translate("UPDATE orders SET status = 'void' WHERE total < 50 AND region IN ('eu')") {
4965 Ok(Stmt::Update { coll, set, nql, .. }) => {
4966 assert_eq!(coll, "orders");
4967 assert_eq!(set, vec![("status".to_string(), json!("void"))]);
4968 assert_eq!(nql, r#"FROM orders WHERE total < 50 AND region IN ("eu")"#);
4970 }
4971 other => panic!("expected UPDATE, got {:?}", other),
4972 }
4973 }
4974
4975 #[test]
4976 fn update_without_where_targets_the_whole_collection() {
4977 match translate("UPDATE t SET a = 1") {
4979 Ok(Stmt::Update { nql, .. }) => assert_eq!(nql, "FROM t"),
4980 other => panic!("expected UPDATE, got {:?}", other),
4981 }
4982 }
4983
4984 #[test]
4985 fn update_handles_several_assignments() {
4986 match translate("UPDATE t SET a = 1, b = 'x,y', c = NULL WHERE id = 'k'") {
4987 Ok(Stmt::Update { set, .. }) => {
4988 assert_eq!(set.len(), 3);
4989 assert_eq!(set[1], ("b".to_string(), json!("x,y")));
4990 assert_eq!(set[2], ("c".to_string(), Value::Null));
4991 }
4992 other => panic!("expected UPDATE, got {:?}", other),
4993 }
4994 assert!(translate("UPDATE t SET").is_err());
4995 assert!(translate("UPDATE t SET a").is_err());
4996 }
4997
4998 #[test]
4999 fn delete_becomes_a_predicate_over_the_collection() {
5000 match translate("DELETE FROM orders WHERE status = 'void'") {
5001 Ok(Stmt::Delete { coll, nql, .. }) => {
5002 assert_eq!(coll, "orders");
5003 assert_eq!(nql, r#"FROM orders WHERE status = "void""#);
5004 }
5005 other => panic!("expected DELETE, got {:?}", other),
5006 }
5007 match translate("DELETE FROM t") {
5008 Ok(Stmt::Delete { nql, .. }) => assert_eq!(nql, "FROM t"),
5009 other => panic!("expected DELETE, got {:?}", other),
5010 }
5011 }
5012
5013 #[test]
5014 fn returning_is_parsed_off_every_write() {
5015 let (_, _, ret) = ins("INSERT INTO t (a) VALUES (1) RETURNING a, _id");
5016 assert_eq!(ret.iter().map(|c| c.out.clone()).collect::<Vec<_>>(), vec!["a", "_id"]);
5017 let (_, _, star) = ins("INSERT INTO t (a) VALUES (1) RETURNING *");
5020 assert!(star.is_empty());
5021 assert!(wants_returning("INSERT INTO t (a) VALUES (1) RETURNING *"));
5022 assert!(!wants_returning("INSERT INTO t (a) VALUES (1)"));
5023
5024 match translate("UPDATE t SET a = 1 WHERE id = 'k' RETURNING a") {
5025 Ok(Stmt::Update { nql, returning, .. }) => {
5026 assert_eq!(returning.len(), 1);
5027 assert!(!nql.to_uppercase().contains("RETURNING"), "{}", nql);
5029 }
5030 other => panic!("expected UPDATE, got {:?}", other),
5031 }
5032 match translate("DELETE FROM t WHERE id = 'k' RETURNING *") {
5033 Ok(Stmt::Delete { nql, .. }) =>
5034 assert!(!nql.to_uppercase().contains("RETURNING"), "{}", nql),
5035 other => panic!("expected DELETE, got {:?}", other),
5036 }
5037 }
5038
5039 #[test]
5040 fn a_keyword_inside_a_value_is_not_a_clause() {
5041 match translate("UPDATE t SET note = 'where returning from' WHERE id = 'k'") {
5042 Ok(Stmt::Update { set, nql, .. }) => {
5043 assert_eq!(set[0].1, json!("where returning from"));
5044 assert_eq!(nql, r#"FROM t WHERE id = "k""#);
5045 }
5046 other => panic!("expected UPDATE, got {:?}", other),
5047 }
5048 }
5049
5050 #[test]
5051 fn split_top_respects_quotes_and_nesting() {
5052 assert_eq!(split_top("a, b, c", ',').len(), 3);
5053 assert_eq!(split_top("(1, 2), (3, 4)", ',').len(), 2);
5054 assert_eq!(split_top("'a,b', c", ',').len(), 2);
5055 assert_eq!(split_top("'it''s, fine', c", ',').len(), 2);
5056 }
5057
5058 #[test]
5059 fn comments_and_whitespace_do_not_confuse_the_translator() {
5060 assert_eq!(q("SELECT *\n FROM orders -- trailing note\n"), "FROM orders");
5061 assert_eq!(q("SELECT /* inline */ * FROM orders"), "FROM orders");
5062 assert_eq!(q("SELECT * FROM t WHERE note = 'from here to JOIN'"),
5064 r#"FROM t WHERE note = "from here to JOIN""#);
5065 }
5066
5067 #[test]
5068 fn find_kw_ignores_quotes_parens_and_substrings() {
5069 assert_eq!(find_kw("SELECT A FROM B", "FROM"), Some(9));
5070 assert_eq!(find_kw("SELECT 'FROM' FROM B", "FROM"), Some(14));
5071 assert_eq!(find_kw("SELECT F(x FROM y) FROM B", "FROM"), Some(19));
5072 assert_eq!(find_kw("SELECT FROMAGE", "FROM"), None);
5073 assert_eq!(find_kw("SELECT X_FROM", "FROM"), None);
5074 }
5075
5076 #[test]
5079 fn provenance_columns_sort_after_the_users_own_fields() {
5080 let rows = vec![json!({"_id":"1","_hash":"ab","status":"paid","total":9})];
5081 assert_eq!(names(&columns_for(&rows, &[])),
5082 vec!["status", "total", "_hash", "_id"]);
5083 }
5084
5085 #[test]
5086 fn an_explicit_projection_sets_the_column_order() {
5087 let rows = vec![json!({"a":1,"b":2})];
5088 let p = vec![Col::same("b"), Col::same("a")];
5089 assert_eq!(names(&columns_for(&rows, &p)), vec!["b", "a"]);
5090 }
5091
5092 #[test]
5093 fn columns_are_the_union_across_sparse_rows() {
5094 let rows = vec![json!({"a":1}), json!({"b":2})];
5096 assert_eq!(names(&columns_for(&rows, &[])), vec!["a", "b"]);
5097 }
5098
5099 #[test]
5100 fn type_oids_follow_the_first_non_null_value() {
5101 let rows = vec![json!({"i":1,"f":1.5,"b":true,"s":"x","n":null})];
5102 assert_eq!(oid_for(&rows, "i"), OID_INT8);
5103 assert_eq!(oid_for(&rows, "f"), OID_FLOAT8);
5104 assert_eq!(oid_for(&rows, "b"), OID_BOOL);
5105 assert_eq!(oid_for(&rows, "s"), OID_TEXT);
5106 assert_eq!(oid_for(&rows, "n"), OID_TEXT);
5108 assert_eq!(oid_for(&rows, "absent"), OID_TEXT);
5109 }
5110
5111 #[test]
5112 fn a_column_that_is_null_in_the_first_row_still_gets_its_type() {
5113 let rows = vec![json!({"v": null}), json!({"v": 7})];
5114 assert_eq!(oid_for(&rows, "v"), OID_INT8);
5115 }
5116
5117 #[test]
5118 fn cells_render_in_postgres_text_format() {
5119 assert_eq!(cell(Some(&json!("x"))), Some("x".to_string()));
5120 assert_eq!(cell(Some(&json!(true))), Some("t".to_string()));
5121 assert_eq!(cell(Some(&json!(false))), Some("f".to_string()));
5122 assert_eq!(cell(Some(&json!(42))), Some("42".to_string()));
5123 assert_eq!(cell(Some(&json!(null))), None);
5124 assert_eq!(cell(None), None);
5125 assert_eq!(cell(Some(&json!({"a":1}))), Some("{\"a\":1}".to_string()));
5127 }
5128
5129 #[test]
5132 fn message_framing_length_excludes_the_tag() {
5133 let mut m = Out::msg(b'Z');
5134 m.bytes(b"I");
5135 let bytes = m.finish();
5136 assert_eq!(bytes[0], b'Z');
5137 assert_eq!(i32::from_be_bytes([bytes[1], bytes[2], bytes[3], bytes[4]]), 5);
5138 assert_eq!(bytes.len(), 6);
5139 }
5140
5141 #[test]
5142 fn a_result_set_encodes_as_description_then_rows_then_complete() {
5143 let rows = vec![json!({"a": 1}), json!({"a": 2})];
5144 let out = encode_result(&rows, &[]);
5145 assert_eq!(out[0], b'T');
5146 let tags: Vec<u8> = {
5147 let mut t = vec![];
5149 let mut i = 0usize;
5150 while i < out.len() {
5151 t.push(out[i]);
5152 let len = i32::from_be_bytes([out[i+1], out[i+2], out[i+3], out[i+4]]) as usize;
5153 i += 1 + len;
5154 }
5155 t
5156 };
5157 assert_eq!(tags, vec![b'T', b'D', b'D', b'C'],
5158 "one description, one row each, one completion");
5159 }
5160
5161 #[test]
5166 fn a_write_with_returning_emits_exactly_one_command_complete() {
5167 let rows = vec![json!({"_id": "o1", "total": 9})];
5168 let mut out = encode_rows(&rows, &[Col::same("_id")]);
5169 out.extend_from_slice(&command_complete("INSERT 0 1"));
5170 let mut tags = vec![];
5171 let mut i = 0usize;
5172 while i < out.len() {
5173 tags.push(out[i]);
5174 let len = i32::from_be_bytes([out[i+1], out[i+2], out[i+3], out[i+4]]) as usize;
5175 i += 1 + len;
5176 }
5177 assert_eq!(tags, vec![b'T', b'D', b'C'], "one description, one row, ONE tag");
5178 assert_eq!(tags.iter().filter(|t| **t == b'C').count(), 1);
5179 assert!(!encode_rows(&rows, &[]).contains(&b'C')
5181 || encode_rows(&rows, &[]).iter().filter(|b| **b == b'C').count() > 0);
5182 let bare = encode_rows(&rows, &[Col::same("_id")]);
5183 let mut bare_tags = vec![];
5184 let mut j = 0usize;
5185 while j < bare.len() {
5186 bare_tags.push(bare[j]);
5187 let len = i32::from_be_bytes([bare[j+1], bare[j+2], bare[j+3], bare[j+4]]) as usize;
5188 j += 1 + len;
5189 }
5190 assert_eq!(bare_tags, vec![b'T', b'D'], "encode_rows never appends a tag");
5191 }
5192
5193 #[test]
5194 fn an_empty_result_still_sends_a_description() {
5195 let out = encode_result(&[], &[Col::same("a")]);
5196 assert_eq!(out[0], b'T', "clients need the shape even with no rows");
5197 }
5198
5199 #[test]
5200 fn statements_split_on_top_level_semicolons_only() {
5201 assert_eq!(split_statements("SELECT 1; SELECT 2").len(), 2);
5202 assert_eq!(split_statements("SELECT ';'").len(), 1);
5203 assert_eq!(split_statements("SELECT 1;").len(), 1);
5204 assert_eq!(split_statements(" ").len(), 0);
5205 }
5206
5207 #[test]
5208 fn an_error_names_its_sqlstate() {
5209 let e = String::from_utf8_lossy(&err_msg("0A000", "x")).to_string();
5210 assert!(e.contains("ERROR"));
5211 assert!(e.contains("0A000"));
5212 }
5213
5214 #[test]
5217 fn placeholders_are_counted_outside_string_literals() {
5218 assert_eq!(param_count("SELECT a FROM t WHERE b = $1 AND c = $2"), 2);
5219 assert_eq!(param_count("SELECT a FROM t"), 0);
5220 assert_eq!(param_count("WHERE a = $2 OR b = $2 OR c = $1"), 2);
5222 assert_eq!(param_count("SELECT a FROM t WHERE b = '$1'"), 0,
5223 "a placeholder inside a literal is data, not a parameter");
5224 assert_eq!(param_count("WHERE a = $10 AND b = $1"), 10,
5225 "two-digit indexes must not be read as $1 followed by 0");
5226 }
5227
5228 #[test]
5229 fn parameters_are_spliced_as_literals() {
5230 let out = substitute_params("WHERE a = $1 AND b = $2 AND c = $3",
5231 &[Some("'x'".into()), Some("42".into()), None]).unwrap();
5232 assert_eq!(out, "WHERE a = 'x' AND b = 42 AND c = NULL");
5233 }
5234
5235 #[test]
5236 fn substitution_leaves_string_literals_alone() {
5237 let out = substitute_params("WHERE a = '$1' AND b = $1", &[Some("9".into())]).unwrap();
5238 assert_eq!(out, "WHERE a = '$1' AND b = 9");
5239 }
5240
5241 #[test]
5242 fn too_few_parameters_is_an_error_not_a_silent_null() {
5243 let e = substitute_params("WHERE a = $2", &[Some("1".into())]).unwrap_err();
5246 assert!(e.contains("$2"), "{}", e);
5247 }
5248
5249 #[test]
5250 fn a_quote_in_a_parameter_cannot_escape_its_literal() {
5251 let lit = decode_param(Some(b"it's"), OID_TEXT, 0).unwrap().unwrap();
5252 assert_eq!(lit, "'it''s'");
5253 let out = substitute_params("WHERE a = $1", &[Some(lit)]).unwrap();
5255 assert_eq!(out, "WHERE a = 'it''s'");
5256 }
5257
5258 #[test]
5259 fn binary_parameters_decode_in_every_width_psycopg_sends() {
5260 assert_eq!(decode_param(Some(&[0x00, 0x2a]), OID_INT2, 1).unwrap().unwrap(), "42");
5263 assert_eq!(decode_param(Some(&[0, 0, 0, 7]), OID_INT4, 1).unwrap().unwrap(), "7");
5264 assert_eq!(
5265 decode_param(Some(&[0, 0, 0, 0, 0, 0, 0, 9]), OID_INT8, 1).unwrap().unwrap(), "9");
5266 assert_eq!(
5267 decode_param(Some(&0x400c_0000_0000_0000u64.to_be_bytes()), OID_FLOAT8, 1)
5268 .unwrap().unwrap(), "3.5");
5269 assert_eq!(decode_param(Some(&[1]), OID_BOOL, 1).unwrap().unwrap(), "TRUE");
5270 assert_eq!(decode_param(Some(&[0]), OID_BOOL, 1).unwrap().unwrap(), "FALSE");
5271 }
5272
5273 #[test]
5274 fn a_negative_binary_integer_keeps_its_sign() {
5275 assert_eq!(decode_param(Some(&(-5i32).to_be_bytes()), OID_INT4, 1).unwrap().unwrap(), "-5");
5276 assert_eq!(decode_param(Some(&(-5i16).to_be_bytes()), OID_INT2, 1).unwrap().unwrap(), "-5");
5277 }
5278
5279 #[test]
5280 fn a_binary_parameter_of_the_wrong_width_is_refused() {
5281 let e = decode_param(Some(&[0x2a]), OID_INT4, 1).unwrap_err();
5284 assert!(e.contains("4 bytes"), "{}", e);
5285 }
5286
5287 #[test]
5288 fn an_unspecified_text_parameter_is_treated_as_a_string() {
5289 assert_eq!(decode_param(Some(b"hello"), 0, 0).unwrap().unwrap(), "'hello'");
5292 }
5293
5294 #[test]
5295 fn a_null_parameter_decodes_to_none_in_every_format() {
5296 assert_eq!(decode_param(None, OID_TEXT, 0).unwrap(), None);
5297 assert_eq!(decode_param(None, OID_INT8, 1).unwrap(), None);
5298 }
5299
5300 #[test]
5301 fn an_unsupported_binary_type_says_so_by_name() {
5302 let e = decode_param(Some(&[0u8; 8]), 1114, 1).unwrap_err();
5303 assert!(e.contains("1114"), "{}", e);
5304 assert!(e.contains("text"), "the error should point at the way out: {}", e);
5305 }
5306
5307 #[test]
5308 fn a_text_number_that_is_not_a_number_gets_quoted() {
5309 assert_eq!(decode_param(Some(b"oops"), OID_INT8, 0).unwrap().unwrap(), "'oops'");
5312 }
5313
5314 #[test]
5315 fn a_client_declared_type_is_believed_over_inference() {
5316 let oids = infer_param_oids("SELECT a FROM t WHERE b = $1 AND c = $2", &[OID_INT4, 0], None);
5319 assert_eq!(oids, vec![OID_INT4, OID_TEXT]);
5320 }
5321
5322 #[test]
5323 fn parameter_arity_is_taken_from_the_sql_when_the_client_declares_none() {
5324 let oids = infer_param_oids("SELECT a FROM t WHERE b = $1 AND c = $2", &[], None);
5327 assert_eq!(oids.len(), 2);
5328 }
5329
5330 #[test]
5331 fn the_field_behind_each_placeholder_is_identified() {
5332 assert_eq!(
5333 param_fields("SELECT a FROM t WHERE qty > $1 AND status = $2", 2),
5334 vec![Some("qty".to_string()), Some("status".to_string())]);
5335 }
5336
5337 #[test]
5338 fn word_operators_do_not_hide_the_field() {
5339 assert_eq!(param_fields("SELECT a FROM t WHERE name LIKE $1", 1),
5340 vec![Some("name".to_string())]);
5341 assert_eq!(param_fields("SELECT a FROM t WHERE qty BETWEEN $1 AND $2", 2),
5342 vec![Some("qty".to_string()), Some("qty".to_string())]);
5343 assert_eq!(param_fields("SELECT a FROM t WHERE region IN ($1, $2)", 2),
5344 vec![Some("region".to_string()), Some("region".to_string())]);
5345 }
5346
5347 #[test]
5348 fn a_clause_position_types_from_the_grammar_not_from_a_column() {
5349 assert_eq!(
5353 infer_param_oids("SELECT a FROM t AS OF SYSTEM TIME $1 WHERE b = $2", &[], None),
5354 vec![OID_INT8, OID_TEXT]);
5355 assert_eq!(infer_param_oids("SELECT a FROM t AS OF $1", &[], None), vec![OID_INT8]);
5356 assert_eq!(
5359 infer_param_oids("SELECT a FROM t VALID AS OF $1", &[], None), vec![OID_TEXT]);
5360 assert_eq!(
5361 infer_param_oids("SELECT a FROM t LIMIT $1 OFFSET $2", &[], None),
5362 vec![OID_INT8, OID_INT8]);
5363 }
5364
5365 #[test]
5366 fn an_aggregate_column_types_from_what_the_aggregate_means() {
5367 assert_eq!(aggregate_oid("count", None, "t"), Some(OID_INT8));
5371 assert_eq!(aggregate_oid("avg_fee", None, "t"), Some(OID_FLOAT8),
5372 "an average is fractional even over integers");
5373 assert_eq!(aggregate_oid("max__seq", None, "t"), Some(OID_INT8));
5376 assert_eq!(aggregate_oid("total", None, "t"), None, "not an aggregate");
5377 }
5378
5379 #[test]
5380 fn the_parse_probe_uses_a_literal_that_every_clause_accepts() {
5381 let probe = probe_sql("SELECT a FROM t AS OF SYSTEM TIME $1 WHERE b = $2", 2);
5385 assert!(!probe.contains("NULL"), "{}", probe);
5386 assert!(translate(&probe).is_ok(), "the probe must parse: {}", probe);
5387 }
5388
5389 #[test]
5390 fn a_column_with_mixed_types_across_documents_is_advertised_as_text() {
5391 let rows = vec![json!({"x": 3}), json!({"x": "n/a"})];
5395 assert_eq!(oid_for(&rows, "x"), OID_TEXT);
5396 let rows = vec![json!({"x": 3}), json!({"x": 1.5})];
5398 assert_eq!(oid_for(&rows, "x"), OID_FLOAT8);
5399 let rows = vec![json!({"x": Value::Null}), json!({"x": 7})];
5401 assert_eq!(oid_for(&rows, "x"), OID_INT8);
5402 }
5403
5404 #[test]
5405 fn binary_output_encodes_each_advertised_type() {
5406 assert_eq!(cell_binary(Some(&json!(true)), OID_BOOL).unwrap().unwrap(), vec![1]);
5407 assert_eq!(cell_binary(Some(&json!(42)), OID_INT8).unwrap().unwrap(),
5408 42i64.to_be_bytes().to_vec());
5409 assert_eq!(cell_binary(Some(&json!(3.5)), OID_FLOAT8).unwrap().unwrap(),
5410 3.5f64.to_be_bytes().to_vec());
5411 assert_eq!(cell_binary(Some(&json!("hi")), OID_TEXT).unwrap().unwrap(), b"hi".to_vec());
5413 assert_eq!(cell_binary(Some(&Value::Null), OID_INT8).unwrap(), None);
5414 assert_eq!(cell(Some(&json!(true))).unwrap(), "t");
5416 }
5417
5418 #[test]
5419 fn a_value_that_does_not_fit_its_advertised_binary_type_is_refused() {
5420 let e = cell_binary(Some(&json!("nope")), OID_INT8).unwrap_err();
5425 assert!(e.contains("a string"), "{}", e);
5426 assert!(e.contains("more than one type"), "the error should explain WHY: {}", e);
5427 }
5428
5429 #[test]
5430 fn a_row_description_carries_the_requested_format_per_column() {
5431 let cols = [Col::same("a"), Col::same("b")];
5432 let m = row_description_fmt(&cols, &[OID_INT8, OID_TEXT], &[1, 0]);
5433 assert_eq!(m[0], b'T');
5434 assert_eq!(m[m.len() - 1], 0, "the last column was requested as text");
5436 }
5437
5438 #[test]
5439 fn a_qualified_column_resolves_to_its_bare_name() {
5440 assert_eq!(param_fields("SELECT a FROM t WHERE t.qty = $1", 1),
5441 vec![Some("qty".to_string())]);
5442 }
5443
5444 #[test]
5445 fn insert_placeholders_map_positionally_to_the_column_list() {
5446 assert_eq!(
5447 param_fields("INSERT INTO t (_id, qty, status) VALUES ($1, $2, $3)", 3),
5448 vec![Some("_id".to_string()), Some("qty".to_string()), Some("status".to_string())]);
5449 }
5450
5451 #[test]
5452 fn a_set_clause_placeholder_finds_its_column() {
5453 assert_eq!(param_fields("UPDATE t SET status = $1 WHERE _id = $2", 2),
5454 vec![Some("status".to_string()), Some("_id".to_string())]);
5455 }
5456
5457 #[test]
5458 fn the_target_collection_is_found_for_every_statement_kind() {
5459 assert_eq!(stmt_collection("SELECT a FROM inv WHERE b = $1"), "inv");
5460 assert_eq!(stmt_collection("UPDATE inv SET a = $1"), "inv");
5461 assert_eq!(stmt_collection("DELETE FROM inv WHERE a = $1"), "inv");
5462 assert_eq!(stmt_collection("INSERT INTO inv (a) VALUES ($1)"), "inv");
5463 assert_eq!(stmt_collection("SELECT a FROM public.inv"), "inv");
5465 assert_eq!(stmt_collection("INSERT INTO inv(a) VALUES ($1)"), "inv");
5466 }
5467
5468 #[test]
5469 fn engine_metadata_fields_type_without_touching_storage() {
5470 assert_eq!(infer_field_oid(None, "t", "_seq"), OID_INT8);
5471 assert_eq!(infer_field_oid(None, "t", "_id"), OID_TEXT);
5472 }
5473
5474 #[test]
5475 fn the_protocol_acknowledgements_are_single_empty_messages() {
5476 for (m, tag) in [
5478 (parse_complete(), b'1'), (bind_complete(), b'2'),
5479 (close_complete(), b'3'), (no_data(), b'n'), (portal_suspended(), b's'),
5480 ] {
5481 assert_eq!(m.len(), 5, "{:?}", tag as char);
5482 assert_eq!(m[0], tag);
5483 assert_eq!(i32::from_be_bytes([m[1], m[2], m[3], m[4]]), 4);
5484 }
5485 }
5486
5487 #[test]
5488 fn parameter_description_reports_its_arity_and_types() {
5489 let m = parameter_description(&[OID_TEXT, OID_INT8]);
5490 assert_eq!(m[0], b't');
5491 assert_eq!(i16::from_be_bytes([m[5], m[6]]), 2);
5492 assert_eq!(i32::from_be_bytes([m[7], m[8], m[9], m[10]]), OID_TEXT);
5493 assert_eq!(i32::from_be_bytes([m[11], m[12], m[13], m[14]]), OID_INT8);
5494 }
5495
5496 #[test]
5497 fn a_cstring_is_taken_without_its_terminator() {
5498 let body = b"one\0two\0".to_vec();
5499 let mut at = 0usize;
5500 assert_eq!(take_cstr(&body, &mut at), "one");
5501 assert_eq!(take_cstr(&body, &mut at), "two");
5502 assert_eq!(at, body.len());
5503 }
5504
5505 #[test]
5506 fn truncated_integers_are_reported_rather_than_read_past_the_end() {
5507 let body = vec![0u8, 1];
5508 let mut at = 0usize;
5509 assert!(take_i32(&body, &mut at).is_err());
5510 let mut at = 0usize;
5511 assert!(take_i16(&body, &mut at).is_ok());
5512 }
5513
5514 #[test]
5515 fn a_binary_result_format_request_is_refused_rather_than_faked() {
5516 let out = encode_rows(&[], &[Col::same("a")]);
5519 let desc_format = &out[out.len() - 2..];
5520 assert_eq!(i16::from_be_bytes([desc_format[0], desc_format[1]]), 0,
5521 "every column is advertised as text format");
5522 }
5523
5524 #[test]
5525 fn a_float_parameter_does_not_render_as_rust_infinity() {
5526 assert_eq!(fmt_float(f64::INFINITY), "'Infinity'");
5527 assert_eq!(fmt_float(f64::NEG_INFINITY), "'-Infinity'");
5528 assert_eq!(fmt_float(f64::NAN), "'NaN'");
5529 assert_eq!(fmt_float(3.0), "3", "a whole float should not gain a .0 tail");
5530 assert_eq!(fmt_float(3.5), "3.5");
5531 }
5532}
5533