nedb_engine/pgwire.rs
1// SPDX-FileCopyrightText: 2026 INTERCHAINED LLC
2// SPDX-License-Identifier: BUSL-1.1
3// NEDB · © 2026 INTERCHAINED LLC × Eth-Interchained × Vex (Claude Opus 5)
4
5//! A PostgreSQL wire-protocol endpoint for NEDB — reads **and** writes.
6//!
7//! # What this is
8//!
9//! A front door that speaks the PostgreSQL v3 wire protocol well enough that
10//! tools built for Postgres — `psql`, DBeaver, Metabase, Grafana, psycopg, any
11//! libpq client — can use a NEDB store with ordinary SQL. A documented subset
12//! of SQL is translated to NQL and to engine writes; everything else is
13//! refused with an error naming exactly what was not understood.
14//!
15//! It is **not** a claim of Postgres parity. It is a claim that the SQL people
16//! actually type works, and that the boundary is stated rather than discovered.
17//!
18//! # Why writes belong here
19//!
20//! The first cut of this module was read-only, on the reasoning that a NEDB
21//! write carries `caused_by`, valid-time bounds and idempotency, and none of
22//! that has a natural SQL spelling. That reasoning was wrong, and looking at
23//! the mapping is what made it obvious:
24//!
25//! | SQL | NEDB | and therefore |
26//! |---|---|---|
27//! | `INSERT` | a put | — |
28//! | `UPDATE … WHERE` | a NEW VERSION of each match | the prior value stays readable |
29//! | `DELETE … WHERE` | a tombstone | the deleted row stays in history |
30//!
31//! NEDB is append-only, so an `UPDATE` is *already* a versioned write and a
32//! `DELETE` is *already* a tombstone. Nothing is bent to fit. The consequence
33//! is the point of the whole endpoint:
34//!
35//! ```sql
36//! UPDATE orders SET total = 999 WHERE _id = 'o1';
37//! SELECT total FROM orders WHERE _id = 'o1'; -- 999
38//! SELECT total FROM orders AS OF SYSTEM TIME 0 WHERE _id = 'o1'; -- 120
39//! ```
40//!
41//! Run the SQL you would run against Postgres, and the tamper-evident history
42//! is free. No triggers, no audit table, no application code.
43//!
44//! Provenance is reachable too: `_caused_by`, `_valid_from` and `_valid_to` are
45//! reserved INSERT columns, lifted out of the payload into the write itself.
46//!
47//! Writes are ON by default — that is the parity position. Set
48//! `NEDBD_PG_READ_ONLY=1` for the deployment where this door must never mutate
49//! anything.
50//!
51//! # Supported SQL
52//!
53//! ```sql
54//! SELECT * | col [, col]* | COUNT(*) | <agg>(col)
55//! FROM <collection>
56//! [ AS OF SYSTEM TIME <seq> ] -- bridges to NQL's AS OF
57//! [ WHERE <predicate> ] -- the full NQL predicate surface
58//! [ GROUP BY <col> ] [ HAVING <predicate> ]
59//! [ ORDER BY <col> [ASC|DESC] (, ...) ] [ LIMIT <n> ] [ OFFSET <n> ]
60//!
61//! INSERT INTO <collection> (c1, c2) VALUES (v1, v2), (…) [RETURNING …]
62//! UPDATE <collection> SET c = v [, …] [WHERE <predicate>] [RETURNING …]
63//! DELETE FROM <collection> [WHERE <predicate>] [RETURNING …]
64//! ```
65//!
66//! Single-quoted SQL literals are rewritten to NQL's double-quoted form and
67//! `<>` to `!=`. Column projection is applied here, after NQL returns whole
68//! documents, because NQL is FROM-first and has no projection clause.
69//!
70//! That translation serves user collections. Statements that read the
71//! catalogue (`pg_catalog.*`, `information_schema.*`) go instead to the real
72//! SQL evaluator in `sqlselect` — joins, subqueries, `EXISTS`, `ARRAY(...)`,
73//! `ANY`/`ALL`, `UNION`, derived tables, `LATERAL`, aggregates, `CASE`, scalar
74//! functions — because that is what psql's `\d` family is written in. Every
75//! psql 17 backslash command that can succeed against an empty-of-features
76//! Postgres exits 0 here, verified by driving the real binary
77//! (`tests/test_psql_introspection.py`).
78//!
79//! Not supported on the user-collection path, each refused by name: JOIN,
80//! subqueries, CTEs, window functions, DDL, `TRUNCATE`, `GRANT`/`REVOKE`.
81//! `INSERT` requires an explicit column list, because NEDB is schemaless and
82//! there is no declared column order to infer.
83//!
84//! # Protocol coverage
85//!
86//! **Both** protocols are implemented:
87//!
88//! * the **simple query protocol** (`Q`) — what `psql` and libpq's `PQexec`
89//! use, and therefore psycopg2, which interpolates parameters client-side;
90//! * the **extended query protocol** (`Parse`/`Bind`/`Describe`/`Execute`/
91//! `Close`/`Sync`/`Flush`) — what psycopg3, asyncpg and the JDBC driver use
92//! for every parameterised statement. Without it those three could not run a
93//! single query, so "psql works" was a long way from "your framework works".
94//!
95//! Parameters arrive in text *and* binary format, prepared statements and
96//! portals are per-connection, and a row-capped `Execute` suspends its portal
97//! (`PortalSuspended`) so a JDBC `setFetchSize` pages instead of stalling.
98//!
99//! ## Parameter typing in a store with no schema
100//!
101//! The extended protocol needs types for `$1..$n`, which a relational server
102//! reads out of its catalogue. NEDB has none — so the types are sampled from
103//! the documents already stored, and the stored data *is* the schema. Where a
104//! placeholder sits in a clause rather than beside a column
105//! (`AS OF SYSTEM TIME $1`, `LIMIT $1`) the grammar supplies the type instead,
106//! and an aggregate column is typed from what the aggregate means: a `COUNT` is
107//! an integer, an `AVG` fractional.
108//!
109//! This is not polish. A client that declares its own parameter types
110//! (psycopg3, JDBC) is believed and only its unspecified slots are inferred —
111//! but asyncpg declares none, asks, and then **refuses the call client-side**
112//! if the answer is wrong. Advertising "text" for everything does not degrade
113//! gracefully there; it fails with `expected str, got int` before a query is
114//! ever sent.
115//!
116//! SSL is declined (`N`), so connections are cleartext — hence the loopback
117//! default.
118//!
119//! Authentication mirrors the HTTP surface: with `NEDBD_TOKEN` set the password
120//! must equal it; otherwise any connection is accepted.
121//!
122//! Still outside the boundary, and refused by name: SQL-level cursors
123//! (`DECLARE`/`FETCH`), window functions, set operations on the
124//! user-collection path, and binary *result* format for a column whose stored
125//! values disagree about their type across documents.
126//!
127//! # What an ORM needs, and what it cost to learn
128//!
129//! Speaking psql is not speaking to a framework, and the difference was three
130//! defects deep. SQLAlchemy could not CONNECT (its dialect opens with
131//! `select pg_catalog.version()`, which a table of exact spellings missed); its
132//! reflection needed `GROUP BY` and `array_agg(x ORDER BY y)`; and a QUALIFIED
133//! column in a `WHERE` clause returned ZERO ROWS — silently — because NQL
134//! looks a field up flat and no document has a field named `orders.status`.
135//! Every ORM qualifies its predicates, so every filtered query lied.
136//!
137//! None of that was visible to psql, which is why
138//! `tests/pgwire_suite.py` drives asyncpg, SQLAlchemy and node-postgres
139//! against a live daemon on every push.
140
141use 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
150// ── Postgres type OIDs we hand out ──────────────────────────────────────────
151const 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; // 3.0 << 16
157const SSL_REQUEST: i32 = 80_877_103;
158const GSS_REQUEST: i32 = 80_877_104;
159const CANCEL_REQUEST: i32 = 80_877_102;
160
161/// How a caller resolves a database name to an open `Db`.
162///
163/// A trait object rather than a concrete handle so this module does not depend
164/// on `server::Manager` — which keeps the protocol code unit-testable against a
165/// plain `Db` with no HTTP stack in the way.
166pub trait DbResolver: Send + Sync + 'static {
167 /// Look up an open database by the name the client connected with.
168 ///
169 /// MAY BLOCK. The implementation is allowed to take a lock, so this is
170 /// always called from `spawn_blocking` — never on an async worker. Taking
171 /// a tokio `RwLock::blocking_read()` on a runtime thread panics outright
172 /// ("Cannot block the current thread from within a runtime"), which is
173 /// exactly how the first cut of this failed.
174 fn resolve(&self, name: &str) -> Option<Arc<Db>>;
175 /// The bearer token, when one is configured. `None` = open access.
176 fn token(&self) -> Option<String> {
177 None
178 }
179}
180
181// ── wire encoding helpers ───────────────────────────────────────────────────
182
183struct Out(Vec<u8>);
184
185impl Out {
186 fn msg(tag: u8) -> Self {
187 // Tag, then a 4-byte length placeholder patched in `finish`.
188 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 // A NUL inside an identifier would truncate the field and desynchronise
194 // the stream, so strip rather than trust.
195 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 /// Patch the length prefix (which covers the length field itself, not the tag).
200 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"); // idle, not in a transaction
219 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// ── SQL → NQL translation ───────────────────────────────────────────────────
229
230/// One output column: the key to read from the row, and the name to show.
231///
232/// The two differ for aggregates. NQL answers `SUM(total)` with a row holding
233/// `sum_total` (plus `count` and a legacy `value` alias), while SQL callers
234/// expect a single column called `sum`. Carrying both halves keeps NEDB's
235/// internal key names off the wire — the first cut leaked `['count','value']`
236/// out of a `SELECT COUNT(*)`, which is two columns where SQL promises one.
237#[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/// What a translated statement asks for.
253///
254/// The write variants exist because SQL's write semantics and NEDB's storage
255/// model line up almost exactly, which was not obvious until it was written
256/// down:
257///
258/// | SQL | NEDB |
259/// |---|---|
260/// | `INSERT` | a put |
261/// | `UPDATE … WHERE` | a NEW VERSION of each matching document |
262/// | `DELETE … WHERE` | a tombstone |
263///
264/// NEDB is append-only, so an `UPDATE` is *already* a versioned write and a
265/// `DELETE` is *already* a tombstone. Nothing is being bent to fit. The
266/// consequence is the thing worth selling: run the SQL you would run against
267/// Postgres, and the tamper-evident history falls out for free — the prior
268/// value is still readable with `AS OF SYSTEM TIME`.
269#[derive(Debug, PartialEq)]
270pub enum Stmt {
271 /// Run this NQL, then project these columns (empty = all).
272 Query { nql: String, project: Vec<Col> },
273 /// `INSERT INTO coll (cols) VALUES (…), (…) [RETURNING …]`
274 Insert { coll: String, rows: Vec<InsertRow>, returning: Vec<Col> },
275 /// `UPDATE coll SET … [WHERE …] [RETURNING …]` — a new version per match.
276 Update {
277 coll: String,
278 set: Vec<(String, Value)>,
279 /// The SQL `WHERE …` as written (column qualifiers stripped), which is
280 /// what actually selects the rows. See `rows_for_write`.
281 where_sql: String,
282 /// The same predicate rendered as NQL. No longer used to SELECT
283 /// anything — kept because it is the translation the `translate_*`
284 /// tests pin, and because an operator reading a 42601 wants to see it.
285 nql: String,
286 returning: Vec<Col>,
287 },
288 /// `DELETE FROM coll [WHERE …] [RETURNING …]` — a tombstone per match.
289 Delete { coll: String, where_sql: String, nql: String, returning: Vec<Col> },
290 /// Answer from a fixed table — the handshake queries clients send on connect.
291 Canned { cols: Vec<String>, row: Vec<String> },
292 /// Nothing to do (empty statement, or a SET the client does not need honoured).
293 Ok(&'static str),
294}
295
296/// One row of an `INSERT`: an explicit id when the statement supplied one, the
297/// document body, and optional provenance lifted out of reserved columns.
298#[derive(Debug, PartialEq, Clone)]
299pub struct InsertRow {
300 /// From an `_id` or `id` column. `None` means the server assigns one.
301 pub id: Option<String>,
302 pub doc: serde_json::Map<String, Value>,
303 /// From a `_caused_by` column — the causal parents, so provenance is
304 /// reachable from SQL rather than only from the HTTP API.
305 pub caused_by: Vec<String>,
306 pub valid_from: Option<String>,
307 pub valid_to: Option<String>,
308}
309
310/// Strip SQL comments and collapse whitespace, so the matchers below can be
311/// simple without being fragile about formatting.
312fn normalise(sql: &str) -> String {
313 let mut out = String::with_capacity(sql.len());
314 let mut chars = sql.chars().peekable();
315 let mut in_s = false;
316 while let Some(c) = chars.next() {
317 if in_s {
318 out.push(c);
319 if c == '\'' { in_s = false; }
320 continue;
321 }
322 match c {
323 '\'' => { in_s = true; out.push(c); }
324 '-' if chars.peek() == Some(&'-') => {
325 // line comment
326 for n in chars.by_ref() { if n == '\n' { break; } }
327 out.push(' ');
328 }
329 '/' if chars.peek() == Some(&'*') => {
330 chars.next();
331 let mut prev = ' ';
332 while let Some(n) = chars.next() {
333 if prev == '*' && n == '/' { break; }
334 prev = n;
335 }
336 out.push(' ');
337 }
338 _ => out.push(c),
339 }
340 }
341 out.split_whitespace().collect::<Vec<_>>().join(" ")
342}
343
344/// Rewrite SQL literal/operator spellings into NQL's.
345///
346/// Only `'…'` → `"…"` and `<>` → `!=`. Done with an explicit scan rather than a
347/// regex so a quote inside a string cannot be mistaken for a delimiter: SQL
348/// escapes an embedded quote by doubling it (`'it''s'`), and that has to become
349/// a single character inside the NQL string rather than terminating it.
350/// Drop the table qualifier from every column reference in a clause tail.
351///
352/// # The silent wrong answer this removes
353///
354/// NQL has no notion of a qualifier: `field_value` looks a field up FLAT, in
355/// one map. So `WHERE orders.status = 'paid'` asked for a field literally
356/// named `orders.status`, no document had one, and the query returned ZERO
357/// ROWS — with no error and no warning, an empty result that reads exactly
358/// like "you have no paid orders".
359///
360/// Every ORM qualifies its predicates. SQLAlchemy emits
361/// `SELECT orders._id FROM orders WHERE orders.status = 'paid'` for the most
362/// ordinary filter there is, so EVERY filtered query answered empty, `.get(pk)`
363/// answered `None`, and `filter_by` answered nothing. The select list had
364/// always stripped qualifiers; the tail was "handed to the NQL parser
365/// unchanged", which is right for the clause GRAMMAR and wrong for a name NQL
366/// cannot interpret.
367///
368/// # Why a mismatched qualifier is an ERROR, not a strip
369///
370/// A qualifier naming something other than this statement's own collection
371/// means the query referenced a relation that is not in its FROM clause.
372/// Stripping it would answer with rows from the one relation that IS there —
373/// a different wrong answer wearing the same empty-looking clothes. Aliases
374/// are refused on this path already, so the collection's own name is the only
375/// qualifier that can be correct.
376///
377/// Runs BEFORE `sql_literals_to_nql`, so only SQL's single-quoted strings have
378/// to be skipped — the rewrite to NQL's double-quoted form has not happened
379/// yet, and a qualifier can never appear inside a literal.
380fn strip_column_qualifiers(
381 tail: &str,
382 coll: &str,
383 alias: Option<&str>,
384) -> Result<String, String> {
385 let bare = coll.rsplit('.').next().unwrap_or(coll);
386 let b: Vec<char> = tail.chars().collect();
387 let mut out = String::with_capacity(tail.len());
388 let mut i = 0usize;
389 let ident_start = |c: char| c.is_alphabetic() || c == '_';
390 let ident_char = |c: char| c.is_alphanumeric() || c == '_';
391
392 while i < b.len() {
393 // A single-quoted literal is copied through untouched.
394 if b[i] == '\'' {
395 out.push(b[i]);
396 i += 1;
397 while i < b.len() {
398 out.push(b[i]);
399 if b[i] == '\'' {
400 // A doubled '' is one literal quote, not a close.
401 if b.get(i + 1) == Some(&'\'') {
402 out.push('\'');
403 i += 2;
404 continue;
405 }
406 i += 1;
407 break;
408 }
409 i += 1;
410 }
411 continue;
412 }
413 // A double-quoted run is copied through too. NQL reads double quotes
414 // as a STRING delimiter rather than an identifier one, so a SQL
415 // delimited identifier is a genuine divergence — but it already fails
416 // LOUDLY in the NQL parser ("expected field name, got Str"), and a
417 // loud failure is not this function's problem to solve quietly.
418 if b[i] == '"' {
419 out.push(b[i]);
420 i += 1;
421 while i < b.len() {
422 out.push(b[i]);
423 if b[i] == '"' { i += 1; break; }
424 i += 1;
425 }
426 continue;
427 }
428 if !ident_start(b[i]) {
429 // A number like `1.5` starts with a digit, so it never enters the
430 // identifier branch and its dot is never touched.
431 out.push(b[i]);
432 i += 1;
433 continue;
434 }
435
436 let start = i;
437 while i < b.len() && ident_char(b[i]) {
438 i += 1;
439 }
440 let word: String = b[start..i].iter().collect();
441
442 // `qual.field` — a dot followed immediately by another identifier.
443 if b.get(i) == Some(&'.') && b.get(i + 1).is_some_and(|c| ident_start(*c)) {
444 let fstart = i + 1;
445 let mut j = fstart;
446 while j < b.len() && ident_char(b[j]) {
447 j += 1;
448 }
449 let field: String = b[fstart..j].iter().collect();
450 // A qualified FUNCTION call (`pg_catalog.something(`) is left
451 // exactly as written: this path does not implement functions at
452 // all, and NQL's own refusal names the function, which is more use
453 // to the reader than a claim about relations.
454 let is_call = b[j..].iter().find(|c| !c.is_whitespace()) == Some(&'(');
455 if is_call {
456 out.push_str(&word);
457 out.push('.');
458 out.push_str(&field);
459 i = j;
460 continue;
461 }
462 let matches_alias = alias.is_some_and(|a| word.eq_ignore_ascii_case(a));
463 if matches_alias || word.eq_ignore_ascii_case(bare) || word.eq_ignore_ascii_case(coll) {
464 out.push_str(&field);
465 i = j;
466 continue;
467 }
468 return Err(format!(
469 "no table or alias named {:?} in this query — this statement reads \
470 {:?}{}, and a qualifier naming anything else would have to be \
471 answered from a relation that is not in its FROM clause",
472 word,
473 bare,
474 alias.map(|a| format!(" (aliased {:?})", a)).unwrap_or_default()
475 ));
476 }
477 out.push_str(&word);
478 }
479 Ok(out)
480}
481
482/// Rewrite `SELECT count(*) FROM (<inner>) [AS] alias` into a flat count over
483/// the inner query's own collection and predicate — or `None` when the shapes
484/// do not permit it.
485///
486/// `None` is a REFUSAL, never a fallback: every caller reports the boundary
487/// rather than trying something else, because the alternative to an exact
488/// count is a wrong one.
489fn flatten_count_of_subquery(projection: &str, rest: &str) -> Option<String> {
490 // The outer select list must be nothing but `count(*)`, optionally
491 // aliased. Any other column would have to come from the derived table's
492 // output, which a flat count does not produce.
493 let (outer_expr, outer_alias) = split_output_alias(projection.trim());
494 let ou = outer_expr.to_uppercase().replace(' ', "");
495 if ou != "COUNT(*)" {
496 return None;
497 }
498
499 // Take the balanced parenthesised span, honouring literals so a `)` inside
500 // a string cannot close it early.
501 let b: Vec<char> = rest.chars().collect();
502 let mut depth = 0i32;
503 let mut in_s = false;
504 let mut end = None;
505 for (i, &c) in b.iter().enumerate() {
506 match c {
507 '\'' => in_s = !in_s,
508 '(' if !in_s => depth += 1,
509 ')' if !in_s => {
510 depth -= 1;
511 if depth == 0 {
512 end = Some(i);
513 break;
514 }
515 }
516 _ => {}
517 }
518 }
519 let end = end?;
520 let inner = b[1..end].iter().collect::<String>().trim().to_string();
521
522 // Nothing may follow the derived table but its alias — a join or a second
523 // FROM item changes what is being counted.
524 let trailing = b[end + 1..].iter().collect::<String>();
525 let (_alias, after) = split_table_alias(trailing.trim());
526 if !after.trim().is_empty() {
527 return None;
528 }
529
530 let iu = inner.to_uppercase();
531 if !iu.starts_with("SELECT") {
532 return None;
533 }
534 // Each of these would make the inner row count differ from the flat one.
535 for kw in ["LIMIT", "OFFSET", "GROUP BY", "HAVING", "UNION", "INTERSECT", "EXCEPT", "JOIN"] {
536 if find_kw(&iu, kw).is_some() {
537 return None;
538 }
539 }
540 if find_kw(&iu, "DISTINCT").is_some() {
541 return None;
542 }
543 // An inner aggregate already reduced the rows to one.
544 let inner_from = find_kw(&iu, "FROM")?;
545 let inner_list = inner[..inner_from].to_uppercase();
546 for agg in ["COUNT(", "SUM(", "AVG(", "MIN(", "MAX(", "ARRAY_AGG(", "STRING_AGG("] {
547 if inner_list.contains(agg) {
548 return None;
549 }
550 }
551 // A nested derived table is not walked — one level is the claim.
552 let inner_rest = inner[inner_from + 4..].trim();
553 if inner_rest.starts_with('(') {
554 return None;
555 }
556
557 // `ORDER BY` cannot change a count, so it is dropped rather than refused.
558 let mut tail = inner_rest.to_string();
559 let tu = tail.to_uppercase();
560 if let Some(ob) = find_kw(&tu, "ORDER BY") {
561 tail = tail[..ob].trim_end().to_string();
562 }
563 Some(format!(
564 "SELECT count(*){} FROM {}",
565 outer_alias.map(|a| format!(" AS {}", a)).unwrap_or_default(),
566 tail
567 ))
568}
569
570/// Words that begin a clause and can therefore never be a bare table alias.
571///
572/// `AS` is absent on purpose: it introduces an alias, and `AS OF` is
573/// disambiguated by looking at the word after it.
574const CLAUSE_WORDS: &[&str] = &[
575 "WHERE", "GROUP", "ORDER", "LIMIT", "OFFSET", "HAVING", "FOR", "VALID",
576 "TRACE", "TRAVERSE", "SEARCH", "RETURNING", "UNION", "INTERSECT", "EXCEPT",
577 "JOIN", "LEFT", "RIGHT", "INNER", "FULL", "CROSS", "ON", "USING", "SET",
578];
579
580/// Take a table alias off the front of a clause tail: `FROM orders o WHERE …`.
581///
582/// Returns the alias and the rest of the tail. The alias is REMOVED because
583/// NQL has no table-alias syntax and would report an "unexpected token" on it
584/// — which is how `FROM orders o` used to fail. Removing it here and teaching
585/// `strip_column_qualifiers` to accept it is what makes `SELECT o.status FROM
586/// orders o` work at all.
587///
588/// `AS OF SYSTEM TIME` also starts with `AS`, so the word AFTER `AS` decides:
589/// `AS OF` is a time-travel clause, anything else is an alias.
590fn split_table_alias(tail: &str) -> (Option<String>, &str) {
591 let t = tail.trim_start();
592 let first_end = t.find(char::is_whitespace).unwrap_or(t.len());
593 let first = &t[..first_end];
594 let fu = first.to_uppercase();
595
596 if fu == "AS" {
597 let rest = t[first_end..].trim_start();
598 let end = rest.find(char::is_whitespace).unwrap_or(rest.len());
599 let word = &rest[..end];
600 if word.eq_ignore_ascii_case("OF") {
601 return (None, t); // `AS OF …`, not an alias
602 }
603 if word.is_empty() {
604 return (None, t);
605 }
606 return (Some(word.trim_matches('"').to_string()), rest[end..].trim_start());
607 }
608 if first.is_empty() || CLAUSE_WORDS.contains(&fu.as_str()) {
609 return (None, t);
610 }
611 // A bare identifier here can only be an alias — the collection name was
612 // already consumed by the caller.
613 if first.chars().next().is_some_and(|c| c.is_alphabetic() || c == '_' || c == '"') {
614 return (Some(first.trim_matches('"').to_string()), t[first_end..].trim_start());
615 }
616 (None, t)
617}
618
619/// Split on a delimiter that is at PAREN DEPTH ZERO and outside a literal.
620///
621/// `projection.split(',')` cuts `SUM(a, b)` in half; a select list is not a
622/// flat comma list once it can contain calls.
623fn split_top_level(s: &str, delim: char) -> Vec<String> {
624 let mut out = vec![];
625 let mut cur = String::new();
626 let mut depth = 0i32;
627 let mut in_s = false;
628 let mut in_d = false;
629 for c in s.chars() {
630 match c {
631 '\'' if !in_d => { in_s = !in_s; cur.push(c); }
632 '"' if !in_s => { in_d = !in_d; cur.push(c); }
633 '(' if !in_s && !in_d => { depth += 1; cur.push(c); }
634 ')' if !in_s && !in_d => { depth -= 1; cur.push(c); }
635 c if c == delim && depth == 0 && !in_s && !in_d => {
636 out.push(std::mem::take(&mut cur));
637 }
638 _ => cur.push(c),
639 }
640 }
641 out.push(cur);
642 out
643}
644
645/// Split `expr AS name` / `expr name` into the expression and its output name.
646///
647/// The alias is the name the CLIENT will look the column up by — SQLAlchemy
648/// reads `count(*) AS count_1` back as `count_1`, so dropping the alias and
649/// returning a column called `count` hands it a result it cannot find.
650fn split_output_alias(p: &str) -> (&str, Option<&str>) {
651 let pu = p.to_uppercase();
652 if let Some(at) = find_kw(&pu, "AS") {
653 let alias = p[at + 2..].trim().trim_matches('"');
654 if !alias.is_empty() {
655 return (p[..at].trim(), Some(alias));
656 }
657 }
658 // A bare alias: `count(*) count_1`. Only after a closing paren or a plain
659 // identifier, and never when the tail is itself part of the expression —
660 // so the split point is the LAST whitespace outside any parenthesis.
661 let b: Vec<char> = p.chars().collect();
662 let mut depth = 0i32;
663 let mut in_s = false;
664 let mut cut = None;
665 for (i, &c) in b.iter().enumerate() {
666 match c {
667 '\'' => in_s = !in_s,
668 '(' if !in_s => depth += 1,
669 ')' if !in_s => depth -= 1,
670 c if c.is_whitespace() && depth == 0 && !in_s => cut = Some(i),
671 _ => {}
672 }
673 }
674 match cut {
675 Some(i) => {
676 let alias = p[i..].trim().trim_matches('"');
677 if alias.is_empty() { (p, None) } else { (p[..i].trim(), Some(alias)) }
678 }
679 None => (p, None),
680 }
681}
682
683/// One string, in NQL's spelling — double-quoted, inner quotes escaped.
684///
685/// These values arrive already UNQUOTED from the SQL parser, so they cannot be
686/// pasted into an NQL query as-is: a value containing `"` would close the
687/// literal early and the rest of it would be parsed as grammar. Which is the
688/// shape of an injection, not merely a syntax error.
689fn nql_string(s: &str) -> String {
690 format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
691}
692
693fn sql_literals_to_nql(s: &str) -> String {
694 let mut out = String::with_capacity(s.len());
695 let mut it = s.chars().peekable();
696 while let Some(c) = it.next() {
697 match c {
698 '\'' => {
699 out.push('"');
700 while let Some(ch) = it.next() {
701 if ch == '\'' {
702 if it.peek() == Some(&'\'') {
703 it.next();
704 out.push('\''); // doubled '' is one literal quote
705 } else {
706 break;
707 }
708 } else if ch == '"' {
709 // A double quote inside a SQL literal must be escaped
710 // for NQL, whose lexer collapses \" to a literal quote.
711 out.push('\\');
712 out.push('"');
713 } else {
714 out.push(ch);
715 }
716 }
717 out.push('"');
718 }
719 '<' if it.peek() == Some(&'>') => { it.next(); out.push_str("!="); }
720 _ => out.push(c),
721 }
722 }
723 out
724}
725
726fn strip_prefix_ci(s: &str, prefix: &str) -> Option<String> {
727 if s.len() >= prefix.len() && s[..prefix.len()].eq_ignore_ascii_case(prefix) {
728 Some(s[prefix.len()..].trim_start().to_string())
729 } else {
730 None
731 }
732}
733
734/// Find a top-level keyword (not inside quotes or parentheses), returning its
735/// byte offset. Case-insensitive, and only matches on word boundaries.
736fn find_kw(s: &str, kw: &str) -> Option<usize> {
737 let bytes = s.as_bytes();
738 let k = kw.as_bytes();
739 let mut depth = 0i32;
740 let mut in_s = false;
741 let mut in_d = false;
742 let mut i = 0usize;
743 while i < bytes.len() {
744 let c = bytes[i];
745 if in_s { if c == b'\'' { in_s = false; } i += 1; continue; }
746 if in_d { if c == b'"' { in_d = false; } i += 1; continue; }
747 match c {
748 b'\'' => { in_s = true; i += 1; continue; }
749 b'"' => { in_d = true; i += 1; continue; }
750 b'(' => { depth += 1; i += 1; continue; }
751 b')' => { depth -= 1; i += 1; continue; }
752 _ => {}
753 }
754 if depth == 0 && i + k.len() <= bytes.len()
755 && bytes[i..i + k.len()].eq_ignore_ascii_case(k)
756 {
757 let before_ok = i == 0 || !(bytes[i - 1] as char).is_alphanumeric() && bytes[i - 1] != b'_';
758 let after = i + k.len();
759 let after_ok = after >= bytes.len()
760 || !(bytes[after] as char).is_alphanumeric() && bytes[after] != b'_';
761 if before_ok && after_ok {
762 return Some(i);
763 }
764 }
765 i += 1;
766 }
767 None
768}
769
770/// Split a comma-separated list at the TOP level, ignoring commas inside
771/// quotes or parentheses — so `VALUES (1, 'a,b'), (2, 'c')` splits into two
772/// groups and not four.
773fn split_top(s: &str, sep: char) -> Vec<String> {
774 let mut out = vec![];
775 let mut cur = String::new();
776 let mut depth = 0i32;
777 let mut in_s = false;
778 let mut it = s.chars().peekable();
779 while let Some(c) = it.next() {
780 if in_s {
781 cur.push(c);
782 if c == '\'' {
783 // A doubled '' is an escaped quote, not the end of the literal.
784 if it.peek() == Some(&'\'') { cur.push(it.next().unwrap()); } else { in_s = false; }
785 }
786 continue;
787 }
788 match c {
789 '\'' => { in_s = true; cur.push(c); }
790 '(' => { depth += 1; cur.push(c); }
791 ')' => { depth -= 1; cur.push(c); }
792 x if x == sep && depth == 0 => { out.push(cur.trim().to_string()); cur.clear(); }
793 _ => cur.push(c),
794 }
795 }
796 if !cur.trim().is_empty() { out.push(cur.trim().to_string()); }
797 out
798}
799
800/// Parse one SQL scalar literal into JSON.
801///
802/// Deliberately narrow: a string, a number, a boolean, or NULL. Anything else
803/// — a function call, an expression, a cast — is refused by name rather than
804/// coerced into a string that would silently store the wrong value.
805fn sql_value(raw: &str) -> Result<Value, String> {
806 let t = raw.trim();
807 if t.is_empty() {
808 return Err("empty value".into());
809 }
810 let up = t.to_uppercase();
811 if up == "NULL" { return Ok(Value::Null); }
812 if up == "TRUE" { return Ok(Value::Bool(true)); }
813 if up == "FALSE" { return Ok(Value::Bool(false)); }
814 if t.starts_with('\'') && t.ends_with('\'') && t.len() >= 2 {
815 // Unwrap, collapsing the SQL '' escape to one quote.
816 let inner = &t[1..t.len() - 1];
817 return Ok(Value::String(inner.replace("''", "'")));
818 }
819 if let Ok(i) = t.parse::<i64>() { return Ok(Value::from(i)); }
820 if let Ok(f) = t.parse::<f64>() { return Ok(Value::from(f)); }
821 Err(format!(
822 "cannot use {:?} as a value — this endpoint accepts string literals, \
823 numbers, TRUE/FALSE and NULL. Expressions, casts and function calls \
824 are not evaluated, because storing an unevaluated expression as text \
825 would be worse than refusing it", t))
826}
827
828/// Pull a trailing `RETURNING …` off a statement, returning (head, columns).
829fn split_returning(tail: &str) -> (String, Vec<Col>) {
830 let tu = tail.to_uppercase();
831 match find_kw(&tu, "RETURNING") {
832 None => (tail.to_string(), vec![]),
833 Some(at) => {
834 let head = tail[..at].trim().to_string();
835 let list = tail[at + "RETURNING".len()..].trim();
836 if list == "*" {
837 return (head, vec![]); // empty projection = every column
838 }
839 let cols = split_top(list, ',')
840 .into_iter()
841 .map(|p| {
842 let raw = p.split_whitespace().next().unwrap_or(&p).to_string();
843 let name = raw.rsplit('.').next().unwrap_or(&raw).trim_matches('"').to_string();
844 Col::same(&name)
845 })
846 .collect();
847 (head, cols)
848 }
849 }
850}
851
852/// Columns whose names are reserved: they carry provenance rather than data.
853fn take_reserved(doc: &mut serde_json::Map<String, Value>) -> (Option<String>, Vec<String>, Option<String>, Option<String>) {
854 let id = doc.remove("_id").or_else(|| doc.remove("id"))
855 .and_then(|v| match v {
856 Value::String(s) => Some(s),
857 Value::Null => None,
858 other => Some(other.to_string()), // a numeric key is a fine id
859 });
860 let caused_by = match doc.remove("_caused_by") {
861 Some(Value::String(s)) => vec![s],
862 Some(Value::Array(a)) => a.into_iter()
863 .filter_map(|v| v.as_str().map(str::to_string)).collect(),
864 _ => vec![],
865 };
866 let vf = doc.remove("_valid_from").and_then(|v| v.as_str().map(str::to_string));
867 let vt = doc.remove("_valid_to").and_then(|v| v.as_str().map(str::to_string));
868 (id, caused_by, vf, vt)
869}
870
871/// `INSERT INTO coll (c1, c2) VALUES (v1, v2), (…) [RETURNING …]`
872fn translate_insert(sql: &str) -> Result<Stmt, String> {
873 let rest = strip_prefix_ci(sql, "INSERT")
874 .and_then(|r| strip_prefix_ci(&r, "INTO"))
875 .ok_or("expected INSERT INTO")?;
876 // Locate VALUES first. Everything before it is `coll (col, …)`; searching
877 // for `(` without that bound finds the VALUES parenthesis instead and
878 // swallows the keyword into the collection name.
879 let ru = rest.to_uppercase();
880 let values_at = find_kw(&ru, "VALUES").ok_or(
881 "expected VALUES — `INSERT … SELECT` is not supported on this endpoint")?;
882 let head = rest[..values_at].trim().to_string();
883 let open = head.find('(').ok_or(
884 "INSERT needs an explicit column list — `INSERT INTO t (a, b) VALUES (…)`. \
885 NEDB is schemaless, so there is no declared column order to infer from")?;
886 let coll = head[..open].trim().trim_matches('"');
887 let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
888 if coll.is_empty() {
889 return Err("expected a collection name after INSERT INTO".into());
890 }
891 let close = head.rfind(')').ok_or("unterminated column list")?;
892 if close < open {
893 return Err("malformed column list".into());
894 }
895 let tail_from_values = rest[values_at..].to_string();
896 let cols: Vec<String> = split_top(&head[open + 1..close], ',')
897 .into_iter()
898 .map(|c| c.trim().trim_matches('"').to_string())
899 .collect();
900 if cols.is_empty() {
901 return Err("the column list is empty".into());
902 }
903
904 let after = strip_prefix_ci(&tail_from_values, "VALUES")
905 .ok_or("expected VALUES after the column list")?;
906 let (values_part, returning) = split_returning(&after);
907
908 let mut rows = vec![];
909 for group in split_top(&values_part, ',') {
910 let g = group.trim();
911 if !(g.starts_with('(') && g.ends_with(')')) {
912 return Err(format!("expected a parenthesised row of values, got {:?}", g));
913 }
914 let vals = split_top(&g[1..g.len() - 1], ',');
915 if vals.len() != cols.len() {
916 return Err(format!(
917 "{} values for {} columns — every row must match the column list",
918 vals.len(), cols.len()));
919 }
920 let mut doc = serde_json::Map::new();
921 for (c, v) in cols.iter().zip(vals.iter()) {
922 doc.insert(c.clone(), sql_value(v)?);
923 }
924 let (id, caused_by, valid_from, valid_to) = take_reserved(&mut doc);
925 rows.push(InsertRow { id, doc, caused_by, valid_from, valid_to });
926 }
927 if rows.is_empty() {
928 return Err("INSERT with no rows".into());
929 }
930 Ok(Stmt::Insert { coll, rows, returning })
931}
932
933/// `UPDATE coll SET a = 1, b = 'x' [WHERE …] [RETURNING …]`
934fn translate_update(sql: &str) -> Result<Stmt, String> {
935 let rest = strip_prefix_ci(sql, "UPDATE").ok_or("expected UPDATE")?;
936 let ru = rest.to_uppercase();
937 let set_at = find_kw(&ru, "SET").ok_or("expected SET in UPDATE")?;
938 // `UPDATE orders o SET …` — Postgres allows an alias here, and taking the
939 // whole span as the collection name made it part of the name ("orders o").
940 let target = rest[..set_at].trim();
941 let mut parts = target.split_whitespace();
942 let coll = parts.next().unwrap_or("").trim_matches('"');
943 let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
944 let upd_alias: Option<String> = match parts.next() {
945 Some(w) if w.eq_ignore_ascii_case("AS") => {
946 parts.next().map(|a| a.trim_matches('"').to_string())
947 }
948 Some(w) => Some(w.trim_matches('"').to_string()),
949 None => None,
950 };
951 if coll.is_empty() {
952 return Err("expected a collection name after UPDATE".into());
953 }
954 let after_set = rest[set_at + 3..].trim().to_string();
955 let (after_set, returning) = split_returning(&after_set);
956
957 // WHERE ends the assignment list; everything after it is a NQL predicate.
958 let au = after_set.to_uppercase();
959 let (assigns_raw, where_raw) = match find_kw(&au, "WHERE") {
960 Some(at) => (after_set[..at].to_string(), after_set[at..].to_string()),
961 None => (after_set.clone(), String::new()),
962 };
963
964 let mut set = vec![];
965 for a in split_top(&assigns_raw, ',') {
966 let eq = a.find('=').ok_or(format!("expected `col = value` in SET, got {:?}", a))?;
967 let col = a[..eq].trim().trim_matches('"').to_string();
968 if col.is_empty() {
969 return Err("empty column name in SET".into());
970 }
971 set.push((col, sql_value(&a[eq + 1..])?));
972 }
973 if set.is_empty() {
974 return Err("UPDATE with no assignments".into());
975 }
976 // The matching rows are found with an ordinary NQL read, so the whole
977 // predicate surface (IN, BETWEEN, LIKE, OR, …) works in an UPDATE too.
978 let where_raw = strip_column_qualifiers(where_raw.trim(), &coll, upd_alias.as_deref())?;
979 let nql = format!("FROM {} {}", coll, sql_literals_to_nql(&where_raw))
980 .trim().to_string();
981 Ok(Stmt::Update { coll, set, where_sql: where_raw, nql, returning })
982}
983
984/// `DELETE FROM coll [WHERE …] [RETURNING …]`
985fn translate_delete(sql: &str) -> Result<Stmt, String> {
986 let rest = strip_prefix_ci(sql, "DELETE")
987 .and_then(|r| strip_prefix_ci(&r, "FROM"))
988 .ok_or("expected DELETE FROM")?;
989 let (rest, returning) = split_returning(&rest);
990 let end = rest.find(' ').unwrap_or(rest.len());
991 let coll = rest[..end].trim().trim_matches('"');
992 let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
993 if coll.is_empty() {
994 return Err("expected a collection name after DELETE FROM".into());
995 }
996 let (del_alias, where_raw) = split_table_alias(rest[end..].trim());
997 let where_raw = strip_column_qualifiers(where_raw, &coll, del_alias.as_deref())?;
998 let nql = format!("FROM {} {}", coll, sql_literals_to_nql(&where_raw))
999 .trim().to_string();
1000 Ok(Stmt::Delete { coll, where_sql: where_raw, nql, returning })
1001}
1002
1003/// Translate one SQL statement into something executable, or explain why not.
1004pub fn translate(sql_raw: &str) -> Result<Stmt, String> {
1005 let sql = normalise(sql_raw);
1006 let sql = sql.trim().trim_end_matches(';').trim();
1007 if sql.is_empty() {
1008 return Ok(Stmt::Ok(""));
1009 }
1010 let upper = sql.to_uppercase();
1011
1012 // ── the handshake. Clients issue these before anything useful; answering
1013 // them with plausible values is the difference between "connects" and
1014 // "hangs on startup". They are canned on purpose — NEDB has no pg_catalog
1015 // and pretending otherwise would be worse than a clear boundary.
1016 if upper.starts_with("SET ") || upper.starts_with("BEGIN") || upper.starts_with("COMMIT")
1017 || upper.starts_with("ROLLBACK") || upper.starts_with("DISCARD")
1018 || upper.starts_with("LISTEN ") || upper.starts_with("UNLISTEN ")
1019 {
1020 // Accepted and ignored: there is one implicit read-only transaction.
1021 return Ok(Stmt::Ok(if upper.starts_with("SET") { "SET" } else { "OK" }));
1022 }
1023 if upper.starts_with("SHOW ") {
1024 let name = sql[5..].trim().to_lowercase();
1025 let val = match name.as_str() {
1026 "transaction_isolation" | "default_transaction_isolation" => "read committed",
1027 "server_version" => SERVER_VERSION,
1028 "server_encoding" | "client_encoding" => "UTF8",
1029 "standard_conforming_strings" => "on",
1030 "is_superuser" => "off",
1031 _ => "",
1032 };
1033 return Ok(Stmt::Canned { cols: vec![name], row: vec![val.to_string()] });
1034 }
1035 if upper == "SELECT VERSION()" {
1036 return Ok(Stmt::Canned {
1037 cols: vec!["version".into()],
1038 row: vec![full_version_string()],
1039 });
1040 }
1041 if upper == "SELECT 1" || upper == "SELECT 1;" {
1042 return Ok(Stmt::Canned { cols: vec!["?column?".into()], row: vec!["1".into()] });
1043 }
1044 if upper.starts_with("SELECT CURRENT_SCHEMA") {
1045 return Ok(Stmt::Canned { cols: vec!["current_schema".into()], row: vec!["public".into()] });
1046 }
1047 if upper.starts_with("SELECT CURRENT_DATABASE") {
1048 return Ok(Stmt::Canned { cols: vec!["current_database".into()], row: vec!["nedb".into()] });
1049 }
1050 if upper.starts_with("SELECT CURRENT_USER") || upper.starts_with("SELECT USER") {
1051 return Ok(Stmt::Canned { cols: vec!["current_user".into()], row: vec!["nedb".into()] });
1052 }
1053
1054 // ── writes ───────────────────────────────────────────────────────────────
1055 // SQL's write semantics and NEDB's append-only model line up, so these are
1056 // first-class rather than refused. See the `Stmt` doc comment.
1057 if upper.starts_with("INSERT") { return translate_insert(sql); }
1058 if upper.starts_with("UPDATE") { return translate_update(sql); }
1059 if upper.starts_with("DELETE") { return translate_delete(sql); }
1060
1061 // ── the refusals that remain, each naming the boundary ──────────────────
1062 for (kw, why) in [
1063 ("CREATE", "DDL is not supported — collections are created implicitly by the first write to them, because NEDB is schemaless"),
1064 ("ALTER", "DDL is not supported — there is no schema to alter"),
1065 ("DROP", "DDL is not supported; drop a database with DELETE /v1/databases/<db>"),
1066 ("TRUNCATE", "not supported, and not an oversight: NEDB is append-only so that history cannot be discarded. That is the product"),
1067 ("COPY", "not supported; use GET /v1/databases/<db>/since for bulk export"),
1068 ("GRANT", "there is no SQL-level privilege system; auth is the bearer token"),
1069 ("REVOKE", "there is no SQL-level privilege system; auth is the bearer token"),
1070 ] {
1071 if upper.starts_with(kw) {
1072 return Err(format!("{} is not supported — {}", kw, why));
1073 }
1074 }
1075 if !upper.starts_with("SELECT") {
1076 return Err(format!(
1077 "only SELECT, INSERT, UPDATE and DELETE are supported on the Postgres \
1078 endpoint (got {:?})",
1079 sql.split_whitespace().next().unwrap_or("")
1080 ));
1081 }
1082 for (kw, why) in [
1083 (" JOIN ", "JOIN is not supported — NQL is single-collection; join in your client or model the relation with LINK/TRAVERSE"),
1084 (" UNION ", "UNION is not supported"),
1085 (" INTERSECT ", "INTERSECT is not supported"),
1086 (" EXCEPT ", "EXCEPT is not supported"),
1087 (" OVER (", "window functions are not supported"),
1088 ("DISTINCT ", "DISTINCT is not supported — GROUP BY <col> gives the distinct values with counts"),
1089 ] {
1090 if upper.contains(kw) {
1091 return Err(why.to_string());
1092 }
1093 }
1094 if find_kw(&upper, "FROM").is_none() {
1095 return Err("SELECT without FROM is not supported on this endpoint".into());
1096 }
1097
1098 // ── SELECT <projection> FROM <rest> ──────────────────────────────────────
1099 let after_select = strip_prefix_ci(sql, "SELECT").ok_or("expected SELECT")?;
1100 let from_at = find_kw(&after_select.to_uppercase(), "FROM")
1101 .ok_or("expected FROM after the select list")?;
1102 let projection = after_select[..from_at].trim().to_string();
1103 let rest = after_select[from_at + 4..].trim().to_string();
1104 if rest.is_empty() {
1105 return Err("expected a collection name after FROM".into());
1106 }
1107 // ── the one derived table with a provable flat equivalent ───────────────
1108 //
1109 // `SELECT count(*) FROM (SELECT … FROM coll WHERE …) AS anon` is what
1110 // EVERY ORM emits for `.count()` — SQLAlchemy's `Query.count()` wraps the
1111 // whole query in a subquery unconditionally. Refusing it means "SQLAlchemy
1112 // works, except counting", which is not a boundary anyone would accept.
1113 //
1114 // Counting a derived table whose rows are exactly the inner query's rows
1115 // is counting the inner query, so the rewrite is an IDENTITY rather than
1116 // an approximation. Each guard below names a construct that would break
1117 // that identity, and anything carrying one is still refused:
1118 //
1119 // * `LIMIT` / `OFFSET` — caps the row count before it is counted
1120 // * `DISTINCT` — collapses duplicates, so the counts differ
1121 // * `GROUP BY` — the inner rows ARE the groups
1122 // * an inner aggregate — already one row, counting it answers 1
1123 // * anything but `count(*)` outside — the outer list would need the
1124 // inner columns, which a flat count cannot supply
1125 if rest.starts_with('(') {
1126 if let Some(flat) = flatten_count_of_subquery(&projection, &rest) {
1127 // Recurses ONCE at most: the rewrite is only produced when the
1128 // inner FROM names a real collection, so the flat statement can
1129 // never re-enter this branch.
1130 return translate(&flat);
1131 }
1132 return Err("subqueries in FROM are not supported — except \
1133 `SELECT count(*) FROM (…)`, which is rewritten to a flat \
1134 count when the inner query has no LIMIT, OFFSET, DISTINCT, \
1135 GROUP BY or aggregate of its own (any of those would make the \
1136 two counts different numbers)".into());
1137 }
1138 let coll_end = rest.find(' ').unwrap_or(rest.len());
1139 let coll = &rest[..coll_end];
1140 if coll.contains(',') {
1141 return Err("selecting from more than one collection is not supported (no JOIN)".into());
1142 }
1143 // Postgres clients often qualify as schema.table; NEDB has one namespace,
1144 // so the schema is dropped — EXCEPT for `information_schema`, whose table
1145 // names (`tables`, `columns`) are words a user could plausibly name a
1146 // collection. Keeping the qualifier there is what stops
1147 // `SELECT * FROM information_schema.tables` and a real collection called
1148 // `tables` from resolving to the same thing.
1149 let bare = coll.rsplit('.').next().unwrap_or(coll).trim_matches('"');
1150 let qualified = coll
1151 .split('.')
1152 .map(|p| p.trim_matches('"'))
1153 .collect::<Vec<_>>()
1154 .join(".");
1155 let coll = if qualified.starts_with("information_schema.") {
1156 qualified.as_str()
1157 } else {
1158 bare
1159 };
1160 let tail = rest[coll_end..].trim();
1161
1162 // ── the select list ──────────────────────────────────────────────────────
1163 //
1164 // Parsed ITEM BY ITEM, which is what lets a list MIX plain columns with an
1165 // aggregate — and that mixture is exactly what a `GROUP BY` query is.
1166 // SQLAlchemy writes `SELECT orders.status, count(*) AS count_1 FROM orders
1167 // GROUP BY orders.status` for the most ordinary grouped query there is,
1168 // and the previous check refused any list containing a parenthesis at all,
1169 // so the whole shape was unreachable even though NQL expresses it
1170 // natively.
1171 //
1172 // NQL's grouped row carries the group key, `count`, and at most one NAMED
1173 // aggregate — so `count(*)` is always available and one of SUM/AVG/MIN/MAX
1174 // may join it. A second named aggregate is refused by name rather than
1175 // silently dropped.
1176 let mut agg_clause = String::new();
1177 let mut agg_srcs: Vec<String> = vec![];
1178 let mut project: Vec<Col> = vec![];
1179
1180 if projection == "*" {
1181 // everything
1182 } else {
1183 for part in split_top_level(&projection, ',') {
1184 let p = part.trim();
1185 if p.is_empty() {
1186 return Err("empty column in the select list".into());
1187 }
1188 let (expr, alias) = split_output_alias(p);
1189 let eu = expr.to_uppercase();
1190
1191 // COUNT(*) and COUNT(col) both become NQL's bare COUNT: NQL counts
1192 // the group, and a per-column non-null count is not expressible.
1193 if eu.starts_with("COUNT(") {
1194 if agg_clause.is_empty() {
1195 agg_clause = " COUNT".to_string();
1196 }
1197 agg_srcs.push("count".to_string());
1198 project.push(Col::renamed("count", alias.unwrap_or("count")));
1199 continue;
1200 }
1201 if let Some(agg) = ["SUM", "AVG", "MIN", "MAX"]
1202 .iter()
1203 .find(|a| eu.starts_with(&format!("{}(", a)))
1204 {
1205 let inner = expr[agg.len() + 1..].trim_end_matches(')').trim();
1206 if inner.is_empty() || inner == "*" {
1207 return Err(format!("{}() needs a column", agg));
1208 }
1209 let inner = inner.rsplit('.').next().unwrap_or(inner).trim_matches('"');
1210 let named = format!("{} {}", agg, inner);
1211 if !agg_clause.is_empty() && agg_clause.trim() != "COUNT" && agg_clause.trim() != named {
1212 return Err(format!(
1213 "only one of SUM/AVG/MIN/MAX is supported per statement \
1214 (already have {:?}, then {:?}) — NQL's grouped row carries \
1215 the group key, `count`, and ONE named aggregate",
1216 agg_clause.trim(), named));
1217 }
1218 agg_clause = format!(" {}", named);
1219 // NQL emits `<agg>_<field>`; SQL names the column after the
1220 // function unless the query aliased it.
1221 let src = format!("{}_{}", agg.to_lowercase(), inner);
1222 project.push(Col::renamed(&src, alias.unwrap_or(&agg.to_lowercase())));
1223 agg_srcs.push(src);
1224 continue;
1225 }
1226 // A paren used to be the whole test for "is this an expression",
1227 // and it let every paren-free one through: `total * 2` became a
1228 // FIELD NAME, no document had a field called "total * 2", and the
1229 // column came back blank for every row with no error. Same silent
1230 // class as the qualified-WHERE bug -- a wrong answer that looks
1231 // like data. So the test is now the positive one: what survives
1232 // has to BE a column reference.
1233 let bare = expr.rsplit('.').next().unwrap_or(expr).trim_matches('"');
1234 let is_column = !bare.is_empty()
1235 && !bare.starts_with(|c: char| c.is_ascii_digit())
1236 && bare.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '$');
1237 if !is_column {
1238 return Err(format!(
1239 "expressions in the select list are not supported ({:?}) — \
1240 supported: *, a column list, COUNT(*), or SUM/AVG/MIN/MAX(col). \
1241 Compute it in your client, or read the column and map it there",
1242 p));
1243 }
1244 let name = bare;
1245 project.push(Col::renamed(name, alias.unwrap_or(name)));
1246 }
1247 }
1248
1249 // ── clause tail: AS OF SYSTEM TIME → AS OF, then pass the rest through ──
1250 //
1251 // The clause keywords NQL shares with SQL (WHERE, GROUP BY, HAVING,
1252 // ORDER BY, LIMIT, OFFSET) are deliberately handed to the NQL parser
1253 // unchanged rather than re-parsed here. NQL is the authority on what is
1254 // valid; re-implementing its grammar would give two parsers to disagree.
1255 // `FROM orders o WHERE …` — the alias is taken off the tail (NQL has no
1256 // alias syntax) and then ACCEPTED as a qualifier on the columns.
1257 let (alias, tail) = split_table_alias(tail);
1258 let mut tail = strip_column_qualifiers(tail, coll, alias.as_deref())?;
1259 let tu = tail.to_uppercase();
1260 if let Some(at) = find_kw(&tu, "AS OF SYSTEM TIME") {
1261 let before = tail[..at].to_string();
1262 let after = tail[at + "AS OF SYSTEM TIME".len()..].trim_start().to_string();
1263 // Take the argument token. A QUOTED datetime may contain spaces
1264 // ('2026-01-01 12:00:00') — cut at the CLOSING quote, not the first
1265 // space, or the second half of the datetime leaks into the tail and
1266 // parses as garbage. A bare token still cuts at whitespace.
1267 let (arg, rest) = if let Some(stripped) = after.strip_prefix('\'') {
1268 match stripped.find('\'') {
1269 Some(close) => (after[..close + 2].to_string(), after[close + 2..].trim_start()),
1270 None => return Err(format!(
1271 "AS OF SYSTEM TIME: an unterminated string literal in the datetime position: {after:?}")),
1272 }
1273 } else if let Some(stripped) = after.strip_prefix('"') {
1274 match stripped.find('"') {
1275 Some(close) => (after[..close + 2].to_string(), after[close + 2..].trim_start()),
1276 None => return Err(format!(
1277 "AS OF SYSTEM TIME: an unterminated string literal in the datetime position: {after:?}")),
1278 }
1279 } else {
1280 let end = after.find(' ').unwrap_or(after.len());
1281 (after[..end].to_string(), after[end..].trim_start())
1282 };
1283 let arg = arg.trim().trim_matches('\'').trim_matches('"').to_string();
1284 // Resolved by TYPE: a bare integer stays a sequence (the original
1285 // contract, byte-for-byte); a quoted string is a wall-clock moment.
1286 // The marker carries the high bit so a datetime can never collide
1287 // with a real seq; the temporal map below resolves it against the
1288 // store's ts index where the Db is in hand. The sqlselect parser
1289 // applies the same rule (one grammar, two front-ends).
1290 if arg.parse::<u64>().is_ok() {
1291 tail = format!("{} AS OF {} {}", before.trim(), arg, rest)
1292 .trim()
1293 .to_string();
1294 } else {
1295 let marker = crate::wallclock::WallClock::parse(&arg)
1296 .map_err(|e| format!(
1297 "AS OF SYSTEM TIME: {} — accepted forms: an ISO 8601 datetime \
1298 or date, or unix seconds/millis with an explicit s/ms unit; \
1299 a bare integer stays a sequence number",
1300 e))?
1301 .as_marker();
1302 tail = format!("{} AS OF {} {}", before.trim(), marker, rest)
1303 .trim()
1304 .to_string();
1305 }
1306 }
1307
1308 // ── ORDER BY <ordinal> → ORDER BY <that select-list column> ─────────────
1309 //
1310 // SQL lets a sort key be a POSITION in the select list, and clients write
1311 // it constantly — `ORDER BY 1, 2` is how psql's own catalogue queries sort,
1312 // and node-postgres sent `GROUP BY status ORDER BY 1` in the very first
1313 // run of the driver harness. NQL has no ordinals: it read the `1` as a
1314 // literal and refused with "expected field name, got Num(1.0)".
1315 //
1316 // The projection is already parsed here, so the position resolves to a
1317 // real field name. An ordinal past the end of the select list, or one used
1318 // with `SELECT *` where there is no list to index, is refused with the
1319 // reason — guessing a column would sort by something the query never named.
1320 let tu_ord = tail.to_uppercase();
1321 if let Some(ob_at) = find_kw(&tu_ord, "ORDER BY") {
1322 let start = ob_at + "ORDER BY".len();
1323 // The clause runs to the next one, or to the end of the tail.
1324 let end = ["LIMIT", "OFFSET", "GROUP BY", "TRACE", "TRAVERSE", "SEARCH"]
1325 .iter()
1326 .filter_map(|k| find_kw(&tu_ord[start..], k).map(|at| start + at))
1327 .min()
1328 .unwrap_or(tail.len());
1329 let mut keys = vec![];
1330 for item in split_top_level(&tail[start..end], ',') {
1331 let item = item.trim();
1332 if item.is_empty() {
1333 continue;
1334 }
1335 let mut parts = item.split_whitespace();
1336 let first = parts.next().unwrap_or("");
1337 let rest: Vec<&str> = parts.collect();
1338 match first.parse::<usize>() {
1339 Ok(n) if n >= 1 => {
1340 let col = project.get(n - 1).ok_or_else(|| {
1341 if project.is_empty() {
1342 format!(
1343 "ORDER BY {} is a select-list POSITION, and `SELECT *` \
1344 has no list to index — name the column instead", n)
1345 } else {
1346 format!(
1347 "ORDER BY {} is out of range: the select list has {} \
1348 column(s)", n, project.len())
1349 }
1350 })?;
1351 keys.push(
1352 std::iter::once(col.src.as_str())
1353 .chain(rest.iter().copied())
1354 .collect::<Vec<_>>()
1355 .join(" "),
1356 );
1357 }
1358 // Not an ordinal — a named column, or `1 + 1`, which NQL will
1359 // judge for itself.
1360 _ => keys.push(item.to_string()),
1361 }
1362 }
1363 tail = format!("{} ORDER BY {} {}", &tail[..ob_at], keys.join(", "), &tail[end..])
1364 .split_whitespace()
1365 .collect::<Vec<_>>()
1366 .join(" ");
1367 }
1368
1369 // ── GROUP BY: refuse a bare column that SQL would refuse ─────────────────
1370 //
1371 // A grouped NQL row holds only the group key, `count` and the aggregate —
1372 // so projecting `total` from `GROUP BY region` found nothing and rendered
1373 // NULL. Silently answering NULL for a column the query cannot produce is
1374 // the exact failure shape this engine keeps getting bitten by, so it is an
1375 // error, using Postgres's own wording so the message is already familiar.
1376 let mut gkey: Option<String> = None;
1377 let tu_all = tail.to_uppercase();
1378 if let Some(gb_at) = find_kw(&tu_all, "GROUP BY") {
1379 let head = tail[..gb_at].trim_end().to_string();
1380 let after = tail[gb_at + "GROUP BY".len()..].trim_start();
1381 let key_end = after.find(|c: char| c == ' ' || c == ',').unwrap_or(after.len());
1382 let group_key = after[..key_end].trim().trim_matches('"').to_string();
1383 let after_key = after[key_end..].trim_start();
1384 gkey = Some(group_key.clone());
1385
1386 // NQL groups by ONE field. Taking the first key and leaving the rest
1387 // in the tail would group by something narrower than the query asked
1388 // for — more rows than Postgres returns, each aggregating too much.
1389 if after_key.starts_with(',') {
1390 return Err(format!(
1391 "GROUP BY takes one key here (got {:?} and more) — NQL groups by a \
1392 single field, and grouping by only the first would aggregate over \
1393 rows the query meant to keep apart",
1394 group_key));
1395 }
1396
1397 for c in &project {
1398 let ok = c.src == group_key
1399 || c.src == "count"
1400 || agg_srcs.contains(&c.src);
1401 if !ok {
1402 return Err(format!(
1403 "column {:?} must appear in the GROUP BY clause or be used in an \
1404 aggregate function — a grouped row carries the group key, `count`, \
1405 and the aggregate, nothing else",
1406 c.src));
1407 }
1408 }
1409
1410 // NQL's aggregate belongs IMMEDIATELY AFTER the group key
1411 // (`GROUP BY status COUNT`), not after the collection name. Emitting
1412 // `FROM orders COUNT GROUP BY status` is refused by the NQL parser
1413 // with "only one aggregate per query" — which is how the most
1414 // ordinary grouped query an ORM writes still failed even once its
1415 // select list parsed.
1416 //
1417 // `count` rides along free with a named aggregate — an NQL grouped row
1418 // carries the key, `count` AND the aggregate — so only the named one
1419 // is emitted when both were asked for.
1420 tail = format!("{} GROUP BY {}{} {}", head, group_key, agg_clause, after_key)
1421 .split_whitespace()
1422 .collect::<Vec<_>>()
1423 .join(" ");
1424 agg_clause.clear();
1425 }
1426
1427 // ── HAVING <agg> → the spelling NQL's grouped row actually carries ──────
1428 //
1429 // NQL's grouped row has fields named `count` and `<agg>_<field>`, and its
1430 // HAVING matches on those. Every SQL client writes something else:
1431 //
1432 // HAVING count(*) > 1 -> NQL parse error (loud, fine)
1433 // HAVING COUNT > 1 -> ZERO ROWS, no error
1434 // HAVING n > 1 -> ZERO ROWS, no error (`n` being the SQL alias)
1435 //
1436 // The last two are the dangerous ones: HAVING is advertised as supported,
1437 // and a filter that silently matches nothing reads as "no groups qualified"
1438 // rather than "your predicate named a field that does not exist". So the
1439 // aggregate spellings are translated, and anything left that is not a
1440 // group-key or aggregate field is refused BY NAME.
1441 let tu_hav = tail.to_uppercase();
1442 if let Some(h_at) = find_kw(&tu_hav, "HAVING") {
1443 let start = h_at + "HAVING".len();
1444 let end = ["ORDER BY", "LIMIT", "OFFSET"]
1445 .iter()
1446 .filter_map(|k| find_kw(&tu_hav[start..], k).map(|at| start + at))
1447 .min()
1448 .unwrap_or(tail.len());
1449 let clause = tail[start..end].to_string();
1450 // The left-hand side of the first comparison is the key being filtered.
1451 let lhs_end = clause
1452 .find(|c: char| "<>=!".contains(c))
1453 .unwrap_or(clause.len());
1454 let lhs = clause[..lhs_end].trim();
1455 if !lhs.is_empty() {
1456 let lu = lhs.to_uppercase();
1457 // `count(*)`, `COUNT(*)`, `count`, or the alias the query gave the
1458 // count -- all mean NQL's `count`.
1459 // The alias test has to tie THIS column to the count. Asking only
1460 // "is there a count anywhere in the projection" matched the GROUP
1461 // BY key too, so `HAVING status > 'a'` -- a perfectly legitimate
1462 // filter on the group key -- was rewritten into `count > 'a'`.
1463 let is_count = lu == "COUNT" || lu.replace(' ', "") == "COUNT(*)"
1464 || project.iter().any(|c| c.out.eq_ignore_ascii_case(lhs) && c.src == "count");
1465 let mapped = if is_count {
1466 Some("count".to_string())
1467 } else {
1468 // A named aggregate, by its NQL source name or by its alias.
1469 agg_srcs.iter().find(|s| s.eq_ignore_ascii_case(lhs)).cloned().or_else(|| {
1470 project.iter()
1471 .find(|c| c.out.eq_ignore_ascii_case(lhs) && agg_srcs.contains(&c.src))
1472 .map(|c| c.src.clone())
1473 })
1474 };
1475 match mapped {
1476 Some(m) => {
1477 // The space matters: `count> 1` happens to parse today, but
1478 // relying on the tokenizer being forgiving is how a rewrite
1479 // breaks the next time the grammar tightens.
1480 let rewritten = format!("{} {}", m, clause[lhs_end..].trim());
1481 tail = format!("{} HAVING {} {}",
1482 tail[..h_at].trim(), rewritten.trim(), tail[end..].trim())
1483 .trim().to_string();
1484 }
1485 None if gkey.as_deref().map(|g| g.eq_ignore_ascii_case(lhs)) == Some(true) => {}
1486 None => {
1487 return Err(format!(
1488 "HAVING names {:?}, which this grouped row does not carry. \
1489 It has the group key{}{}. Filtering on anything else would \
1490 answer zero rows rather than report a mistake",
1491 lhs,
1492 gkey.as_deref().map(|g| format!(" ({:?})", g)).unwrap_or_default(),
1493 if agg_srcs.is_empty() { String::new() }
1494 else { format!(", plus {}", agg_srcs.join(", ")) }));
1495 }
1496 }
1497 }
1498 }
1499
1500
1501 let tail = sql_literals_to_nql(&tail);
1502 let nql = format!("FROM {}{}{}", coll,
1503 if agg_clause.is_empty() { String::new() } else { agg_clause },
1504 if tail.is_empty() { String::new() } else { format!(" {}", tail) });
1505
1506 Ok(Stmt::Query { nql: nql.trim().to_string(), project })
1507}
1508
1509const SERVER_VERSION: &str = "15.0";
1510
1511/// The `version()` string, for the SQL engine's `version()` function.
1512pub fn version_string() -> String {
1513 full_version_string()
1514}
1515
1516fn full_version_string() -> String {
1517 format!(
1518 "PostgreSQL {} (NEDB {}) — tamper-evident, append-only, permanent \
1519 history. SELECT + INSERT/UPDATE/DELETE; an UPDATE is a new version, \
1520 so prior values stay readable with AS OF SYSTEM TIME.",
1521 SERVER_VERSION,
1522 env!("CARGO_PKG_VERSION")
1523 )
1524}
1525
1526// ── result shaping ──────────────────────────────────────────────────────────
1527
1528/// Pick the column order for a result set.
1529///
1530/// With an explicit projection, that order. Otherwise the union of keys across
1531/// the returned rows — `_`-prefixed provenance columns last, so `psql` shows
1532/// the user's own fields first and `_hash` does not push `status` off screen.
1533fn columns_for(rows: &[Value], project: &[Col]) -> Vec<Col> {
1534 if !project.is_empty() {
1535 return project.to_vec();
1536 }
1537 let mut plain: Vec<String> = vec![];
1538 let mut meta: Vec<String> = vec![];
1539 for r in rows {
1540 if let Value::Object(m) = r {
1541 for k in m.keys() {
1542 let target = if k.starts_with('_') { &mut meta } else { &mut plain };
1543 if !target.contains(k) {
1544 target.push(k.clone());
1545 }
1546 }
1547 }
1548 }
1549 // The user's own fields keep the DOCUMENT'S order -- `serde_json`'s
1550 // `preserve_order` is on crate-wide precisely so they can, and Postgres
1551 // orders `*` by column definition rather than alphabetically. Sorting them
1552 // here made `SELECT *` answer in a different column order than the SQL
1553 // evaluator did, so a client reading by POSITION got different columns
1554 // depending on a deployment flag. Only the provenance block is sorted.
1555 meta.sort();
1556 plain.extend(meta);
1557 plain.into_iter().map(|k| Col::same(&k)).collect()
1558}
1559
1560/// The Postgres type of one JSON value.
1561fn oid_of_value(v: &Value) -> Option<i32> {
1562 match v {
1563 Value::Null => None,
1564 Value::Bool(_) => Some(OID_BOOL),
1565 Value::Number(n) => Some(if n.is_i64() || n.is_u64() { OID_INT8 } else { OID_FLOAT8 }),
1566 Value::String(_) => Some(OID_TEXT),
1567 // Arrays and objects render as their JSON text.
1568 _ => Some(OID_TEXT),
1569 }
1570}
1571
1572/// Reconcile two observed types for the same column.
1573///
1574/// A relational column has one type by construction. A NEDB collection does
1575/// not: document 1 may hold `qty: 3` and document 2 `qty: "three"`. Widening
1576/// to `text` on a conflict is the only answer that can carry both, and mixed
1577/// integers and floats widen to float8 for the same reason.
1578fn unify_oid(a: i32, b: i32) -> i32 {
1579 if a == b {
1580 return a;
1581 }
1582 match (a, b) {
1583 (OID_INT8, OID_FLOAT8) | (OID_FLOAT8, OID_INT8) => OID_FLOAT8,
1584 _ => OID_TEXT,
1585 }
1586}
1587
1588/// The type of `col` across EVERY row in the result, not just the first.
1589///
1590/// Taking the first non-null value's type was a latent wrong answer: a column
1591/// holding `3` in row one and `"n/a"` in row two was advertised as `int8`, and
1592/// a client that believes the description then fails parsing `"n/a"` as an
1593/// integer — or, on the binary path, cannot be sent the value at all.
1594/// Public alias so `pgcatalog` types a column EXACTLY as the wire does.
1595///
1596/// The catalogue reporting `bigint` for a column the protocol then sends as
1597/// text would be a self-contradiction a client is entitled to trust, so both
1598/// go through this one function rather than two that agree today.
1599pub fn oid_for_column(rows: &[Value], col: &str) -> i32 {
1600 oid_for(rows, col)
1601}
1602
1603/// Did any row actually carry a non-null value for this column?
1604///
1605/// `oid_for` cannot answer this: it folds "no evidence" and "evidence, all
1606/// text" into the same `OID_TEXT`. The difference matters, because one of
1607/// those is a measurement and the other is a default standing in for one.
1608fn has_evidence(rows: &[Value], col: &str) -> bool {
1609 rows.iter().any(|r| matches!(r.get(col), Some(v) if !v.is_null()))
1610}
1611
1612fn oid_for(rows: &[Value], col: &str) -> i32 {
1613 let mut acc: Option<i32> = None;
1614 for r in rows {
1615 if let Some(o) = r.get(col).and_then(oid_of_value) {
1616 acc = Some(match acc {
1617 None => o,
1618 Some(prev) => unify_oid(prev, o),
1619 });
1620 if acc == Some(OID_TEXT) {
1621 break; // text absorbs everything; no need to look further
1622 }
1623 }
1624 }
1625 acc.unwrap_or(OID_TEXT)
1626}
1627
1628/// Render one cell in the text format Postgres clients expect for format 0.
1629fn cell(v: Option<&Value>) -> Option<String> {
1630 match v {
1631 None | Some(Value::Null) => None, // NULL on the wire
1632 Some(Value::String(s)) => Some(s.clone()),
1633 Some(Value::Bool(b)) => Some(if *b { "t".into() } else { "f".into() }),
1634 Some(other) => Some(other.to_string()),
1635 }
1636}
1637
1638/// Render one cell in binary format for the type the column was advertised as.
1639///
1640/// Needed because asyncpg asks for binary results — it is not an optimisation
1641/// there, it is the only format it requests, so without this it cannot read a
1642/// single row. Text-format clients never reach this path.
1643///
1644/// A value that does not fit the advertised type is an error rather than a
1645/// coercion. The advertised type comes from sampling stored documents, so a
1646/// mismatch means the field is genuinely heterogeneous beyond the sample, and
1647/// quietly sending a zero (or the text bytes under a binary header) would
1648/// corrupt the value in a way the client cannot detect.
1649fn cell_binary(v: Option<&Value>, oid: i32) -> Result<Option<Vec<u8>>, String> {
1650 let v = match v {
1651 None | Some(Value::Null) => return Ok(None),
1652 Some(v) => v,
1653 };
1654 let as_f64 = |n: &serde_json::Number| n.as_f64()
1655 .ok_or_else(|| "a number too large to send as float8".to_string());
1656 Ok(Some(match (oid, v) {
1657 (OID_BOOL, Value::Bool(b)) => vec![u8::from(*b)],
1658 (OID_INT2, Value::Number(n)) => {
1659 let i = n.as_i64().ok_or("not an integer")?;
1660 i16::try_from(i).map_err(|_| format!("{} does not fit in int2", i))?
1661 .to_be_bytes().to_vec()
1662 }
1663 (OID_INT4, Value::Number(n)) => {
1664 let i = n.as_i64().ok_or("not an integer")?;
1665 i32::try_from(i).map_err(|_| format!("{} does not fit in int4", i))?
1666 .to_be_bytes().to_vec()
1667 }
1668 (OID_INT8, Value::Number(n)) => {
1669 n.as_i64().ok_or("not an integer")?.to_be_bytes().to_vec()
1670 }
1671 (OID_FLOAT4, Value::Number(n)) => (as_f64(n)? as f32).to_be_bytes().to_vec(),
1672 (OID_FLOAT8, Value::Number(n)) => as_f64(n)?.to_be_bytes().to_vec(),
1673 // For the text family, binary and text are the same bytes.
1674 (OID_TEXT | OID_VARCHAR | OID_NAME | OID_UNKNOWN | OID_JSON, _) => {
1675 cell(Some(v)).unwrap_or_default().into_bytes()
1676 }
1677 // jsonb is a one-byte version header then the JSON text.
1678 (OID_JSONB, _) => {
1679 let mut b = vec![1u8];
1680 b.extend_from_slice(cell(Some(v)).unwrap_or_default().as_bytes());
1681 b
1682 }
1683 (oid, val) => {
1684 let kind = match val {
1685 Value::Bool(_) => "a boolean",
1686 Value::Number(_) => "a number",
1687 Value::String(_) => "a string",
1688 Value::Array(_) => "an array",
1689 _ => "an object",
1690 };
1691 return Err(format!(
1692 "cannot send {} in binary format as type OID {} — the field holds \
1693 more than one type across documents, so it cannot be described \
1694 by a single Postgres type. Select it with a text cast, or use a \
1695 text-format client",
1696 kind, oid
1697 ));
1698 }
1699 }))
1700}
1701
1702/// A `RowDescription`, with a per-column wire format code.
1703fn row_description_fmt(cols: &[Col], oids: &[i32], fmts: &[i16]) -> Vec<u8> {
1704 let mut m = Out::msg(b'T');
1705 m.i16(cols.len() as i16);
1706 for (i, c) in cols.iter().enumerate() {
1707 m.cstr(&c.out);
1708 m.i32(0); // table OID — unknown
1709 m.i16((i + 1) as i16); // column attribute number
1710 m.i32(oids.get(i).copied().unwrap_or(OID_TEXT));
1711 m.i16(-1); // variable length
1712 m.i32(-1); // no type modifier
1713 m.i16(fmts.get(i).copied().unwrap_or(0));
1714 }
1715 m.finish()
1716}
1717
1718fn row_description(cols: &[Col], oids: &[i32]) -> Vec<u8> {
1719 row_description_fmt(cols, oids, &[])
1720}
1721
1722fn data_row_bytes(vals: &[Option<Vec<u8>>]) -> Vec<u8> {
1723 let mut m = Out::msg(b'D');
1724 m.i16(vals.len() as i16);
1725 for v in vals {
1726 match v {
1727 None => m.i32(-1),
1728 Some(b) => {
1729 m.i32(b.len() as i32);
1730 m.bytes(b);
1731 }
1732 }
1733 }
1734 m.finish()
1735}
1736
1737fn data_row(vals: &[Option<String>]) -> Vec<u8> {
1738 let owned: Vec<Option<Vec<u8>>> =
1739 vals.iter().map(|v| v.as_ref().map(|s| s.as_bytes().to_vec())).collect();
1740 data_row_bytes(&owned)
1741}
1742
1743/// Encode just the rows: `T` followed by one `D` per row, and NO
1744/// `CommandComplete`.
1745///
1746/// Split out because a write with `RETURNING` must emit `T`/`D`* and then its
1747/// OWN tag (`INSERT 0 3`, `UPDATE 1`). The first cut called `encode_result`
1748/// there, which appends `CommandComplete("SELECT n")` — so one statement sent
1749/// TWO CommandComplete messages. That is a protocol violation, and the visible
1750/// symptom was `RETURNING` silently yielding no rows at all: the client took
1751/// the first tag as the end of the statement and discarded the description.
1752pub fn encode_rows(rows: &[Value], project: &[Col]) -> Vec<u8> {
1753 let cols = columns_for(rows, project);
1754 let oids: Vec<i32> = cols.iter().map(|c| oid_for(rows, &c.src)).collect();
1755 let mut out = row_description(&cols, &oids);
1756 for r in rows {
1757 let vals: Vec<Option<String>> = cols.iter().map(|c| cell(r.get(&c.src))).collect();
1758 out.extend_from_slice(&data_row(&vals));
1759 }
1760 out
1761}
1762
1763/// A complete SELECT response: rows plus `CommandComplete("SELECT n")`.
1764pub fn encode_result(rows: &[Value], project: &[Col]) -> Vec<u8> {
1765 let mut out = encode_rows(rows, project);
1766 out.extend_from_slice(&command_complete(&format!("SELECT {}", rows.len())));
1767 out
1768}
1769
1770// ── the extended query protocol: Parse / Bind / Describe / Execute ──────────
1771//
1772// Why this exists at all: psycopg3, asyncpg and the JDBC driver do not speak
1773// the simple query protocol for parameterised statements. Without these six
1774// messages they cannot run a single query — psycopg3 hangs waiting for a
1775// `ParseComplete`, and asyncpg refuses before it ever sends a `Bind`. "psql
1776// works" is not the same as "the drivers your evaluators use work".
1777//
1778// Two facts about real drivers shaped everything below, and both were read off
1779// a wire transcript rather than assumed:
1780//
1781// 1. psycopg3 sends parameters in a MIXED format — a `str` as OID 0 in text
1782// format, but an `int` as int2/int4/int8 in BINARY, a float as float8
1783// binary, a bool as a single binary byte. A text-only decoder gets `\x00*`
1784// where it expected `42`.
1785//
1786// 2. asyncpg declares NO parameter types in `Parse` and then asks
1787// `Describe(statement)`, encoding its arguments from whatever OIDs come
1788// back. Answering "text" for all of them does not degrade gracefully — it
1789// makes asyncpg REFUSE the call client-side ("expected str, got int").
1790//
1791// (2) is the reason `infer_param_oids` exists. NEDB is schemaless, so there is
1792// no catalogue to read a column's type out of — the only honest source of truth
1793// is the data already stored, so the type is sampled from it.
1794
1795/// Parameter/result type OIDs handled on the binary path.
1796const OID_INT2: i32 = 21;
1797const OID_INT4: i32 = 23;
1798const OID_OID: i32 = 26;
1799const OID_FLOAT4: i32 = 700;
1800const OID_VARCHAR: i32 = 1043;
1801const OID_NAME: i32 = 19;
1802const OID_UNKNOWN: i32 = 705;
1803const OID_JSON: i32 = 114;
1804const OID_JSONB: i32 = 3802;
1805
1806/// How many `$n` placeholders a statement carries, and the highest index used.
1807///
1808/// Scans outside string literals so a `'$1'` inside a value is not mistaken for
1809/// a placeholder. Dollar-quoted bodies (`$tag$…$tag$`) are not recognised —
1810/// they need a procedural language NEDB does not have.
1811fn param_count(sql: &str) -> usize {
1812 let b = sql.as_bytes();
1813 let mut i = 0usize;
1814 let mut in_s = false;
1815 let mut max = 0usize;
1816 while i < b.len() {
1817 let c = b[i];
1818 if in_s {
1819 if c == b'\'' {
1820 in_s = false;
1821 }
1822 i += 1;
1823 continue;
1824 }
1825 if c == b'\'' {
1826 in_s = true;
1827 i += 1;
1828 continue;
1829 }
1830 if c == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
1831 let mut j = i + 1;
1832 let mut n = 0usize;
1833 while j < b.len() && b[j].is_ascii_digit() {
1834 n = n * 10 + (b[j] - b'0') as usize;
1835 j += 1;
1836 }
1837 max = max.max(n);
1838 i = j;
1839 continue;
1840 }
1841 i += 1;
1842 }
1843 max
1844}
1845
1846/// The JSON-shaped type of `field` as it is actually stored, sampled from the
1847/// collection, mapped onto the nearest Postgres OID.
1848///
1849/// This is the schemaless answer to "what type is this column?". A relational
1850/// server reads its catalogue; NEDB has none, so it reads the data. Sampling a
1851/// bounded number of rows keeps a `Describe` cheap, and the first row that
1852/// actually carries the field decides — a field missing from row one but
1853/// present in row nine still types correctly.
1854fn infer_field_oid(db: Option<&Arc<Db>>, coll: &str, field: &str) -> i32 {
1855 // `_`-prefixed names are engine metadata, not stored document fields, so
1856 // they type from the engine's own contract — no sampling, and no database
1857 // handle needed.
1858 match field {
1859 "_seq" => return OID_INT8,
1860 "_id" | "_hash" | "_prev" | "_collection" | "_valid_from" | "_valid_to" => return OID_TEXT,
1861 _ => {}
1862 }
1863 // A catalogue relation types its own columns. Sampling a USER collection
1864 // named `pg_type` finds nothing and falls back to text — and asyncpg,
1865 // which declares parameter types client-side and refuses the call when
1866 // the server's answer is wrong, then rejected `WHERE oid = $1` with
1867 // "expected str, got int" before a single byte was sent.
1868 if !field.is_empty() && crate::pgcatalog::is_catalog(coll) {
1869 if let Some(rows) = crate::pgcatalog::rows(coll, db) {
1870 return oid_for(&rows, field);
1871 }
1872 }
1873 let db = match db {
1874 Some(db) => db,
1875 None => return OID_TEXT,
1876 };
1877 if coll.is_empty() || field.is_empty() {
1878 return OID_TEXT;
1879 }
1880 let rows = match crate::nql::query(db, &format!("FROM {} LIMIT {}", coll, TYPE_SAMPLE)) {
1881 Ok((rows, _)) => rows,
1882 Err(_) => return OID_TEXT,
1883 };
1884 // Unified over the sample, not taken from the first hit: a field that is a
1885 // number in one document and a string in another has to be advertised as
1886 // text or a client cannot decode every row of it.
1887 oid_for(&rows, field)
1888}
1889
1890/// The type of an aggregate output column, which no document holds.
1891///
1892/// Sampling stored documents cannot type these: `COUNT(*)` produces a column
1893/// called `count` that exists in no document, so the sampler finds nothing and
1894/// falls back to text. A text-format client papers over that, but a binary
1895/// client is then handed the digits of a number under a text header and
1896/// `COUNT(*)` comes back as the string `"2"` instead of the integer `2`.
1897///
1898/// So aggregates are typed from what the aggregate MEANS: a count is always an
1899/// integer, an average is always fractional, and min/max/sum inherit the type
1900/// of the field they were computed over.
1901/// Column names and wire types for a statement the EVALUATOR will answer.
1902///
1903/// `describe_shape` derived both by calling `translate()`, which means it
1904/// described the TRANSLATOR's output. That was right while the translator
1905/// answered; once the evaluator did, the two disagreed about the one thing
1906/// `Describe` exists to report.
1907///
1908/// They disagree on naming. `SELECT sum(total)` is column `sum_total` to the
1909/// translator and `sum` to the evaluator, so `aggregate_oid("sum", ..)` found
1910/// no `sum_` prefix, fell through to `infer_field_oid(db, coll, "sum")`, found
1911/// no stored field called `sum`, and answered `OID_TEXT`.
1912///
1913/// A text OID is not a cosmetic defect in the BINARY protocol. `Describe`
1914/// happens before `Execute`, so the client is told the column is text and
1915/// decodes the bytes that way: asyncpg received the string `'420'` where
1916/// `420` was meant, and `AS OF SYSTEM TIME $1` came back `total='66'`. The
1917/// text protocol was unaffected — it re-derives types from the rows it
1918/// actually has — which is why psycopg2's suite stayed green while asyncpg's
1919/// did not.
1920///
1921/// Typed from the PARSED SELECT rather than from a sample of the output,
1922/// because `Describe` has no rows yet. That is also why this cannot simply
1923/// reuse the row-sniffing path.
1924fn evaluator_shape(
1925 sql: &str,
1926 db: Option<&Arc<Db>>,
1927 coll: &str,
1928) -> Option<(Vec<Col>, Vec<i32>)> {
1929 let sel = crate::sqlselect::parse(sql).ok()?;
1930 // `*` expands from the rows, which Describe does not have. Declining is
1931 // honest; the caller falls back and the text path types it from the rows.
1932 if sel.items.iter().any(|i| matches!(i.expr, crate::sqlselect::Expr::Star
1933 | crate::sqlselect::Expr::QualifiedStar(_)))
1934 {
1935 return None;
1936 }
1937
1938 let mut cols: Vec<Col> = Vec::new();
1939 let mut oids: Vec<i32> = Vec::new();
1940 for item in &sel.items {
1941 let name = match &item.alias {
1942 Some(a) => a.clone(),
1943 None => match &item.expr {
1944 crate::sqlselect::Expr::Column { name, .. } => name.clone(),
1945 crate::sqlselect::Expr::Agg { name, .. } => name.to_ascii_lowercase(),
1946 crate::sqlselect::Expr::Func { name, .. } => name.to_ascii_lowercase(),
1947 // Anything else is named by a rule this function should not
1948 // try to reproduce from memory. Declining beats guessing a
1949 // name the evaluator will not use.
1950 _ => return None,
1951 },
1952 };
1953 oids.push(expr_oid(&item.expr, db, coll)?);
1954 cols.push(Col::renamed(&name, &name));
1955 }
1956 if cols.is_empty() {
1957 return None;
1958 }
1959 Some((cols, oids))
1960}
1961
1962/// The wire type of one select-list expression.
1963fn expr_oid(e: &crate::sqlselect::Expr, db: Option<&Arc<Db>>, coll: &str) -> Option<i32> {
1964 use crate::sqlselect::Expr;
1965 match e {
1966 Expr::Column { name, .. } => Some(infer_field_oid(db, coll, name)),
1967 Expr::Literal(v) => Some(oid_of_value(v).unwrap_or(OID_TEXT)),
1968 // Aggregates are `Agg`, NOT `Func`. Matching only `Func` here is what
1969 // made this whole fallback inert: `expr_oid` answered None for every
1970 // aggregate, `evaluator_shape` propagated the None, and the caller's
1971 // `unwrap_or(OID_TEXT)` shipped `sum` as text. The unit tests did not
1972 // catch it because they exercised `aggregate_oid`, which types from a
1973 // NAME; nothing typed from a parsed expression until this existed.
1974 Expr::Agg { name, args, .. } | Expr::Func { name, args } => {
1975 let f = name.to_ascii_lowercase();
1976 match f.as_str() {
1977 // COUNT is a count whatever it counts.
1978 "count" => Some(OID_INT8),
1979 // An average is fractional even over integers — the case the
1980 // translator also special-cased.
1981 "avg" => Some(OID_FLOAT8),
1982 // SUM/MIN/MAX inherit the type they range over, so the
1983 // argument has to be resolved rather than assumed numeric.
1984 "sum" | "min" | "max" => match args.first() {
1985 Some(Expr::Column { name, .. }) => match infer_field_oid(db, coll, name) {
1986 OID_INT8 => Some(OID_INT8),
1987 OID_FLOAT8 => Some(OID_FLOAT8),
1988 other => Some(other),
1989 },
1990 _ => None,
1991 },
1992 _ => None,
1993 }
1994 }
1995 _ => None,
1996 }
1997}
1998
1999fn aggregate_oid(src: &str, db: Option<&Arc<Db>>, coll: &str) -> Option<i32> {
2000 if src == "count" {
2001 return Some(OID_INT8);
2002 }
2003 for (prefix, fixed) in [
2004 ("count_", Some(OID_INT8)),
2005 ("avg_", Some(OID_FLOAT8)),
2006 ("sum_", None),
2007 ("min_", None),
2008 ("max_", None),
2009 ] {
2010 if let Some(field) = src.strip_prefix(prefix) {
2011 return Some(match fixed {
2012 Some(oid) => oid,
2013 // SUM/MIN/MAX of an integer field is an integer; of a
2014 // fractional field, fractional.
2015 None => match infer_field_oid(db, coll, field) {
2016 OID_INT8 => OID_INT8,
2017 OID_FLOAT8 => OID_FLOAT8,
2018 // Summing or ordering a non-numeric field is not
2019 // meaningful; let the row-derived type answer.
2020 other => other,
2021 },
2022 });
2023 }
2024 }
2025 None
2026}
2027
2028/// How many documents to sample when typing a column.
2029///
2030/// Bounded so a `Describe` stays cheap. It is a sample, so a field that only
2031/// turns heterogeneous outside it can still surprise us — which is exactly why
2032/// `cell_binary` refuses a mismatch loudly instead of coercing.
2033const TYPE_SAMPLE: usize = 200;
2034
2035/// The collection a statement reads from or writes to, for type sampling.
2036fn stmt_collection(sql: &str) -> String {
2037 let s = normalise(sql);
2038 let up = s.to_uppercase();
2039 let after = if let Some(at) = find_kw(&up, "FROM") {
2040 &s[at + 4..]
2041 } else if let Some(rest) = strip_prefix_ci(&s, "UPDATE") {
2042 return rest
2043 .split_whitespace()
2044 .next()
2045 .unwrap_or("")
2046 .rsplit('.')
2047 .next()
2048 .unwrap_or("")
2049 .trim_matches('"')
2050 .to_string();
2051 } else if let Some(rest) = strip_prefix_ci(&s, "INSERT INTO") {
2052 return rest
2053 .split(|c: char| c.is_whitespace() || c == '(')
2054 .find(|t| !t.is_empty())
2055 .unwrap_or("")
2056 .rsplit('.')
2057 .next()
2058 .unwrap_or("")
2059 .trim_matches('"')
2060 .to_string();
2061 } else {
2062 return String::new();
2063 };
2064 after
2065 .trim()
2066 .split(|c: char| c.is_whitespace())
2067 .find(|t| !t.is_empty())
2068 .unwrap_or("")
2069 .rsplit('.')
2070 .next()
2071 .unwrap_or("")
2072 .trim_matches('"')
2073 .to_string()
2074}
2075
2076/// Which document field each `$n` is being compared against.
2077///
2078/// Three shapes cover essentially all driver-generated SQL:
2079/// `WHERE qty > $1` → the identifier immediately left of the operator
2080/// `SET status = $1` → same shape, inside the SET list
2081/// `INSERT INTO t (a,b) VALUES ($1,$2)` → positional against the column list
2082///
2083/// Anything it cannot read returns `None`, which types as `text`. Guessing
2084/// wrong here would make a driver encode a value the engine then fails to
2085/// match, so an unknown is left unknown on purpose.
2086fn param_fields(sql: &str, n_params: usize) -> Vec<Option<String>> {
2087 let s = normalise(sql);
2088 let mut out = vec![None; n_params];
2089
2090 // The INSERT column list maps positionally, which is more reliable than
2091 // scanning leftwards through a VALUES tuple.
2092 let up = s.to_uppercase();
2093 if up.starts_with("INSERT") {
2094 if let (Some(open), Some(vals_at)) = (s.find('('), find_kw(&up, "VALUES")) {
2095 if open < vals_at {
2096 if let Some(close) = s[open..vals_at].rfind(')') {
2097 let cols: Vec<String> = split_top(&s[open + 1..open + close], ',')
2098 .into_iter()
2099 .map(|c| c.trim().trim_matches('"').to_string())
2100 .collect();
2101 // `$1` is the first placeholder in the first tuple, and so on.
2102 let tail = &s[vals_at..];
2103 let mut seen = 0usize;
2104 let b = tail.as_bytes();
2105 let mut i = 0usize;
2106 let mut in_s = false;
2107 while i < b.len() {
2108 if in_s {
2109 if b[i] == b'\'' { in_s = false; }
2110 i += 1;
2111 continue;
2112 }
2113 if b[i] == b'\'' { in_s = true; i += 1; continue; }
2114 if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2115 let mut j = i + 1;
2116 let mut num = 0usize;
2117 while j < b.len() && b[j].is_ascii_digit() {
2118 num = num * 10 + (b[j] - b'0') as usize;
2119 j += 1;
2120 }
2121 if num >= 1 && num <= n_params {
2122 if let Some(c) = cols.get(seen % cols.len().max(1)) {
2123 out[num - 1] = Some(c.clone());
2124 }
2125 }
2126 seen += 1;
2127 i = j;
2128 continue;
2129 }
2130 i += 1;
2131 }
2132 return out;
2133 }
2134 }
2135 }
2136 }
2137
2138 // Otherwise: for each `$n`, walk left past the operator to the identifier.
2139 let b = s.as_bytes();
2140 let mut i = 0usize;
2141 let mut in_s = false;
2142 while i < b.len() {
2143 if in_s {
2144 if b[i] == b'\'' { in_s = false; }
2145 i += 1;
2146 continue;
2147 }
2148 if b[i] == b'\'' { in_s = true; i += 1; continue; }
2149 if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2150 let mut j = i + 1;
2151 let mut num = 0usize;
2152 while j < b.len() && b[j].is_ascii_digit() {
2153 num = num * 10 + (b[j] - b'0') as usize;
2154 j += 1;
2155 }
2156 if num >= 1 && num <= n_params {
2157 let left = &s[..i];
2158 // Skip the operator characters and whitespace sitting between
2159 // the identifier and the placeholder.
2160 let trimmed = left.trim_end_matches(|c: char| {
2161 c.is_whitespace() || "=<>!+-*/%(,".contains(c)
2162 });
2163 // A word operator (`LIKE`, `IN`, `BETWEEN`, `AND`) also sits
2164 // between them; step over it to reach the real identifier.
2165 let mut tok = trimmed
2166 .rsplit(|c: char| c.is_whitespace() || c == '(' || c == ',')
2167 .find(|t| !t.is_empty())
2168 .unwrap_or("")
2169 .trim_matches('"');
2170 let mut before = trimmed;
2171 for _ in 0..4 {
2172 let upper_tok = tok.to_uppercase();
2173 // `BETWEEN $1 AND $2` puts BOTH a word operator and an
2174 // earlier placeholder between `$2` and the column it
2175 // constrains, so a placeholder has to be stepped over too —
2176 // otherwise the upper bound of every range query types as
2177 // text while the lower bound types correctly.
2178 if upper_tok.starts_with('$')
2179 || matches!(upper_tok.as_str(),
2180 "LIKE" | "ILIKE" | "IN" | "BETWEEN" | "AND" | "OR" | "NOT" | "IS") {
2181 before = before[..before.len() - tok.len()].trim_end_matches(|c: char| {
2182 c.is_whitespace() || "=<>!(,".contains(c)
2183 });
2184 tok = before
2185 .rsplit(|c: char| c.is_whitespace() || c == '(' || c == ',')
2186 .find(|t| !t.is_empty())
2187 .unwrap_or("")
2188 .trim_matches('"');
2189 } else {
2190 break;
2191 }
2192 }
2193 if !tok.is_empty()
2194 && tok.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '.')
2195 && !tok.chars().next().map(|c| c.is_ascii_digit()).unwrap_or(true)
2196 {
2197 out[num - 1] = Some(tok.rsplit('.').next().unwrap_or(tok).to_string());
2198 }
2199 }
2200 i = j;
2201 continue;
2202 }
2203 i += 1;
2204 }
2205 out
2206}
2207
2208/// The type of a placeholder sitting in a CLAUSE position rather than beside a
2209/// column.
2210///
2211/// `AS OF SYSTEM TIME $1` has no column to sample — the token to its left is
2212/// the word `TIME`. Its type comes from the grammar instead, which is both
2213/// cheaper and more certain than any inference: a system-time bound is a
2214/// sequence number, a valid-time bound is a date string, and a page bound is an
2215/// integer. Without this, a parameterised time-travel query typed as text and
2216/// asyncpg refused to send the integer at all.
2217fn clause_param_oids(sql: &str, n_params: usize) -> Vec<Option<i32>> {
2218 let s = normalise(sql);
2219 let mut out = vec![None; n_params];
2220 let b = s.as_bytes();
2221 let mut i = 0usize;
2222 let mut in_s = false;
2223 while i < b.len() {
2224 if in_s {
2225 if b[i] == b'\'' { in_s = false; }
2226 i += 1;
2227 continue;
2228 }
2229 if b[i] == b'\'' { in_s = true; i += 1; continue; }
2230 if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2231 let mut j = i + 1;
2232 let mut num = 0usize;
2233 while j < b.len() && b[j].is_ascii_digit() {
2234 num = num * 10 + (b[j] - b'0') as usize;
2235 j += 1;
2236 }
2237 if num >= 1 && num <= n_params {
2238 let left = s[..i].trim_end().to_uppercase();
2239 // VALID AS OF is checked FIRST: it ends with "AS OF" too, and
2240 // its argument is a DATE STRING, not a sequence number.
2241 out[num - 1] = if left.ends_with("VALID AS OF") {
2242 Some(OID_TEXT)
2243 } else if left.ends_with("AS OF SYSTEM TIME")
2244 || left.ends_with("FOR SYSTEM_TIME AS OF")
2245 || left.ends_with("AS OF")
2246 || left.ends_with("LIMIT")
2247 || left.ends_with("OFFSET")
2248 {
2249 Some(OID_INT8)
2250 } else {
2251 None
2252 };
2253 }
2254 i = j;
2255 continue;
2256 }
2257 i += 1;
2258 }
2259 out
2260}
2261
2262/// The OIDs to advertise for `$1..$n`, sampled from stored data.
2263///
2264/// `declared` is what the client itself put in `Parse`. A client that states a
2265/// type is believed — it is about to encode its arguments that way, and second
2266///-guessing it would break the decode. Only the unspecified slots are inferred.
2267fn infer_param_oids(sql: &str, declared: &[i32], db: Option<&Arc<Db>>) -> Vec<i32> {
2268 let n = param_count(sql).max(declared.len());
2269 if n == 0 {
2270 return vec![];
2271 }
2272 let coll = stmt_collection(sql);
2273 let fields = param_fields(sql, n);
2274 let clauses = clause_param_oids(sql, n);
2275 (0..n)
2276 .map(|i| match declared.get(i) {
2277 Some(&oid) if oid != 0 => oid,
2278 // A clause position knows its own type from the grammar, so it
2279 // outranks sampling a column that is not even there.
2280 _ => match clauses[i] {
2281 Some(oid) => oid,
2282 None => match &fields[i] {
2283 Some(f) => infer_field_oid(db, &coll, f),
2284 None => OID_TEXT,
2285 },
2286 },
2287 })
2288 .collect()
2289}
2290
2291/// Decode one bound parameter into the SQL literal text to splice into the
2292/// statement.
2293///
2294/// `None` means SQL NULL. Format 1 is binary — see the module note on psycopg3
2295/// sending small integers as int2.
2296fn decode_param(raw: Option<&[u8]>, oid: i32, format: i16) -> Result<Option<String>, String> {
2297 let bytes = match raw {
2298 None => return Ok(None),
2299 Some(b) => b,
2300 };
2301 let quote = |s: &str| format!("'{}'", s.replace('\'', "''"));
2302
2303 if format == 0 {
2304 let s = String::from_utf8_lossy(bytes).to_string();
2305 return Ok(Some(match oid {
2306 OID_BOOL => {
2307 let t = matches!(s.as_str(), "t" | "true" | "TRUE" | "1" | "yes" | "on");
2308 if t { "TRUE".into() } else { "FALSE".into() }
2309 }
2310 OID_INT2 | OID_INT4 | OID_INT8 | OID_OID | OID_FLOAT4 | OID_FLOAT8 => {
2311 // Validate rather than trust: an unparseable "number" spliced
2312 // in bare would become a bare identifier in the NQL text and
2313 // produce a baffling error far from its cause.
2314 if s.parse::<f64>().is_ok() { s } else { quote(&s) }
2315 }
2316 // OID 0 with text format is psycopg3's `str`. Confirmed on the
2317 // wire: it declares a real numeric OID whenever the value is a
2318 // number, so an unspecified text parameter is genuinely a string
2319 // and quoting it is right rather than a guess.
2320 _ => quote(&s),
2321 }));
2322 }
2323 if format != 1 {
2324 return Err(format!("unsupported parameter format code {}", format));
2325 }
2326
2327 // ── binary ──────────────────────────────────────────────────────────────
2328 let need = |n: usize| -> Result<(), String> {
2329 if bytes.len() == n {
2330 Ok(())
2331 } else {
2332 Err(format!(
2333 "binary parameter of type OID {} should be {} bytes, got {}",
2334 oid, n, bytes.len()
2335 ))
2336 }
2337 };
2338 Ok(Some(match oid {
2339 OID_BOOL => {
2340 need(1)?;
2341 if bytes[0] != 0 { "TRUE".into() } else { "FALSE".into() }
2342 }
2343 OID_INT2 => {
2344 need(2)?;
2345 i16::from_be_bytes([bytes[0], bytes[1]]).to_string()
2346 }
2347 OID_INT4 => {
2348 need(4)?;
2349 i32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]).to_string()
2350 }
2351 OID_OID => {
2352 need(4)?;
2353 u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]).to_string()
2354 }
2355 OID_INT8 => {
2356 need(8)?;
2357 i64::from_be_bytes(bytes[..8].try_into().unwrap()).to_string()
2358 }
2359 OID_FLOAT4 => {
2360 need(4)?;
2361 let f = f32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
2362 fmt_float(f as f64)
2363 }
2364 OID_FLOAT8 => {
2365 need(8)?;
2366 fmt_float(f64::from_be_bytes(bytes[..8].try_into().unwrap()))
2367 }
2368 OID_TEXT | OID_VARCHAR | OID_NAME | OID_UNKNOWN | OID_JSON | 0 => {
2369 quote(&String::from_utf8_lossy(bytes))
2370 }
2371 OID_JSONB => {
2372 // jsonb binary is a 1-byte version header followed by the JSON text.
2373 let body = if bytes.first() == Some(&1) { &bytes[1..] } else { bytes };
2374 quote(&String::from_utf8_lossy(body))
2375 }
2376 other => {
2377 return Err(format!(
2378 "parameter type OID {} is not supported in binary format — \
2379 the supported set is bool, int2/int4/int8, float4/float8, \
2380 text/varchar/json/jsonb. Send it as text, or cast it in the \
2381 statement",
2382 other
2383 ))
2384 }
2385 }))
2386}
2387
2388/// Render a float without Rust's `inf`/`NaN` spellings leaking into SQL text.
2389fn fmt_float(f: f64) -> String {
2390 if f.is_nan() {
2391 "'NaN'".into()
2392 } else if f.is_infinite() {
2393 if f > 0.0 { "'Infinity'".into() } else { "'-Infinity'".into() }
2394 } else if f.fract() == 0.0 && f.abs() < 1e15 {
2395 format!("{:.0}", f)
2396 } else {
2397 f.to_string()
2398 }
2399}
2400
2401/// Splice decoded parameters into the statement text.
2402///
2403/// Textual substitution, deliberately: the whole SQL surface is already a text
2404/// translation into NQL, so one representation is simpler and cannot disagree
2405/// with itself. Every value arrives already rendered as a SQL literal by
2406/// `decode_param`, with embedded quotes doubled, so a parameter cannot break
2407/// out of its literal and alter the statement's shape.
2408fn substitute_params(sql: &str, params: &[Option<String>]) -> Result<String, String> {
2409 let b = sql.as_bytes();
2410 let mut out = String::with_capacity(sql.len() + 16);
2411 let mut i = 0usize;
2412 let mut in_s = false;
2413 while i < b.len() {
2414 let c = b[i];
2415 if in_s {
2416 out.push(c as char);
2417 if c == b'\'' { in_s = false; }
2418 i += 1;
2419 continue;
2420 }
2421 if c == b'\'' {
2422 in_s = true;
2423 out.push('\'');
2424 i += 1;
2425 continue;
2426 }
2427 if c == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
2428 let mut j = i + 1;
2429 let mut n = 0usize;
2430 while j < b.len() && b[j].is_ascii_digit() {
2431 n = n * 10 + (b[j] - b'0') as usize;
2432 j += 1;
2433 }
2434 match params.get(n.wrapping_sub(1)) {
2435 Some(Some(lit)) => out.push_str(lit),
2436 Some(None) => out.push_str("NULL"),
2437 None => {
2438 return Err(format!(
2439 "bind message supplies {} parameter(s) but the statement uses ${}",
2440 params.len(), n
2441 ))
2442 }
2443 }
2444 i = j;
2445 continue;
2446 }
2447 out.push(c as char);
2448 i += 1;
2449 }
2450 Ok(out)
2451}
2452
2453/// A parsed statement, held for the life of the connection (or until `Close`).
2454struct Prepared {
2455 sql: String,
2456 /// OIDs advertised for `$1..$n` — what `ParameterDescription` reports and
2457 /// what `Bind` values are decoded as.
2458 param_oids: Vec<i32>,
2459 /// The advertised output shape, computed on demand and then reused.
2460 ///
2461 /// Lazy because working it out samples stored documents, and a text-format
2462 /// client that never sends `Describe(statement)` should not pay for a scan
2463 /// on every `Parse` — psycopg3 parses once per query.
2464 ///
2465 /// `Some(None)` means "computed, and this statement returns no rows".
2466 out_shape: Option<Option<(Vec<Col>, Vec<i32>)>>,
2467}
2468
2469/// The output columns and types a statement advertises, computed once.
2470fn prepared_shape<'a>(
2471 p: &'a mut Prepared,
2472 db: Option<&Arc<Db>>,
2473) -> &'a Option<(Vec<Col>, Vec<i32>)> {
2474 if p.out_shape.is_none() {
2475 p.out_shape = Some(describe_shape(&p.sql, db, p.param_oids.len()));
2476 }
2477 p.out_shape.as_ref().expect("just filled")
2478}
2479
2480/// A bound statement: fully substituted SQL plus, once run, its result.
2481struct Portal {
2482 sql: String,
2483 /// Filled by the first `Describe` or `Execute` and reused afterwards.
2484 ///
2485 /// Executing once and streaming from the buffer is what makes a suspended
2486 /// portal safe: a second `Execute` on a partially-drained `INSERT` must
2487 /// continue the row stream, not perform the insert again.
2488 result: Option<PortalResult>,
2489 /// The output shape, frozen at the first `Describe`/`Execute`.
2490 ///
2491 /// A schemaless store derives `SELECT *`'s columns from the rows it found,
2492 /// which would let a `Describe` and a later `Execute` disagree about the
2493 /// column count — and a driver that was told three fields and handed two
2494 /// mis-decodes the row rather than failing loudly. Freezing the shape and
2495 /// projecting every row onto it makes the result set rectangular, as SQL
2496 /// promises. The simple protocol keeps the dynamic behaviour, where there
2497 /// is no `Describe` to contradict.
2498 frozen: Option<Vec<Col>>,
2499 /// Result-column format codes requested by `Bind`. Empty = all text.
2500 formats: Vec<i16>,
2501 /// The shape this portal's statement advertised, carried over from the
2502 /// prepared statement when any column is to be sent in BINARY.
2503 ///
2504 /// It has to be the ADVERTISED shape rather than one derived from the rows
2505 /// in hand: asyncpg built its decoders from `Describe`, so re-deriving a
2506 /// different type here would hand it bytes it cannot read.
2507 declared: Option<(Vec<Col>, Vec<i32>)>,
2508}
2509
2510impl Portal {
2511 /// The format code for column `i`, following the protocol's shorthands:
2512 /// no codes means all-text, one code applies to every column.
2513 fn format_of(&self, i: usize) -> i16 {
2514 match self.formats.len() {
2515 0 => 0,
2516 1 => self.formats[0],
2517 _ => self.formats.get(i).copied().unwrap_or(0),
2518 }
2519 }
2520 /// The columns and types to advertise and encode with.
2521 fn shape(&self, r: &PortalResult) -> (Vec<Col>, Vec<i32>) {
2522 match &self.declared {
2523 Some((cols, oids)) if self.formats.iter().any(|f| *f == 1) => {
2524 (cols.clone(), oids.clone())
2525 }
2526 _ => {
2527 let cols = columns_for(&r.rows, &r.project);
2528 let oids = cols.iter().map(|c| oid_for(&r.rows, &c.src)).collect();
2529 (cols, oids)
2530 }
2531 }
2532 }
2533}
2534
2535struct PortalResult {
2536 rows: Vec<Value>,
2537 project: Vec<Col>,
2538 has_rows: bool,
2539 tag: String,
2540 tag_counts_rows: bool,
2541 /// How many rows have gone out across all `Execute`s on this portal.
2542 sent: usize,
2543}
2544
2545fn parse_complete() -> Vec<u8> { Out::msg(b'1').finish() }
2546fn bind_complete() -> Vec<u8> { Out::msg(b'2').finish() }
2547fn close_complete() -> Vec<u8> { Out::msg(b'3').finish() }
2548fn no_data() -> Vec<u8> { Out::msg(b'n').finish() }
2549fn portal_suspended() -> Vec<u8> { Out::msg(b's').finish() }
2550
2551fn parameter_description(oids: &[i32]) -> Vec<u8> {
2552 let mut m = Out::msg(b't');
2553 m.i16(oids.len() as i16);
2554 for o in oids {
2555 m.i32(*o);
2556 }
2557 m.finish()
2558}
2559
2560/// Split a NUL-terminated string off the front of a message body.
2561fn take_cstr(body: &[u8], at: &mut usize) -> String {
2562 let start = *at;
2563 while *at < body.len() && body[*at] != 0 {
2564 *at += 1;
2565 }
2566 let s = String::from_utf8_lossy(&body[start..*at]).to_string();
2567 if *at < body.len() {
2568 *at += 1; // step over the NUL
2569 }
2570 s
2571}
2572
2573fn take_i16(body: &[u8], at: &mut usize) -> Result<i16, String> {
2574 if *at + 2 > body.len() {
2575 return Err("truncated message".into());
2576 }
2577 let v = i16::from_be_bytes([body[*at], body[*at + 1]]);
2578 *at += 2;
2579 Ok(v)
2580}
2581
2582fn take_i32(body: &[u8], at: &mut usize) -> Result<i32, String> {
2583 if *at + 4 > body.len() {
2584 return Err("truncated message".into());
2585 }
2586 let v = i32::from_be_bytes([body[*at], body[*at + 1], body[*at + 2], body[*at + 3]]);
2587 *at += 4;
2588 Ok(v)
2589}
2590
2591/// The field names a collection actually holds, sampled from stored documents.
2592///
2593/// The answer to `SELECT *` on a store with no schema. Sorted, because
2594/// `serde_json`'s map is ordered and both this and the row encoder must agree
2595/// on column order or the values land under the wrong headings.
2596fn sample_columns(db: Option<&Arc<Db>>, coll: &str) -> Vec<Col> {
2597 let db = match db {
2598 Some(db) => db,
2599 None => return vec![],
2600 };
2601 let rows = match crate::nql::query(db, &format!("FROM {} LIMIT 25", coll)) {
2602 Ok((rows, _)) => rows,
2603 Err(_) => return vec![],
2604 };
2605 let mut names: Vec<String> = vec![];
2606 for r in &rows {
2607 if let Value::Object(m) = r {
2608 for k in m.keys() {
2609 if !names.iter().any(|n| n == k) {
2610 names.push(k.clone());
2611 }
2612 }
2613 }
2614 }
2615 names.sort();
2616 names.iter().map(|n| Col::same(n)).collect()
2617}
2618
2619/// The result shape of a statement, worked out WITHOUT running it.
2620///
2621/// Needed for `Describe(statement)`, which arrives before any `Bind` — asyncpg
2622/// builds its row decoders from the answer. Only the select list is read off
2623/// the result; nothing touches storage except the type sampling.
2624///
2625/// Returns `None` when the statement returns no rows at all (`NoData`).
2626fn describe_shape(
2627 sql: &str,
2628 db: Option<&Arc<Db>>,
2629 n_params: usize,
2630) -> Option<(Vec<Col>, Vec<i32>)> {
2631 let probe = probe_sql(sql, n_params);
2632
2633 // The SQL evaluator describes its own output. It has to: `translate`
2634 // cannot parse a catalogue join at all, so without this a `Describe`
2635 // answered `NoData` — and a client told a SELECT has no output never
2636 // reads its rows.
2637 //
2638 // The probe is EXECUTED here, which is affordable precisely because this
2639 // path only serves catalogue relations and relation-free select lists.
2640 // Column types come from the values it actually produced, unified across
2641 // the rows by the same `oid_for` every other path uses — so a column
2642 // advertised `int8` is one the wire really encodes as int8.
2643 let coll = stmt_collection(sql);
2644
2645 if sql_engine_owns(&probe) {
2646 if let Ok(Some((done, _))) = try_catalog_select(&probe, db) {
2647 if done.project.is_empty() {
2648 return None;
2649 }
2650 // Sniffing the probe's OUTPUT is only sound when the probe
2651 // produced output. It frequently does not, and the reason is
2652 // structural rather than unlucky: `probe_sql` substitutes `0` for
2653 // every parameter, so `... WHERE region = $1` becomes
2654 // `... WHERE region = 0`, matches nothing, and hands this line an
2655 // empty `rows`. `oid_for` then finds no evidence and returns its
2656 // `unwrap_or(OID_TEXT)` default.
2657 //
2658 // In the BINARY protocol that default is not a shrug, it is a
2659 // wrong answer the client cannot recover from: `Describe`
2660 // precedes `Execute`, so asyncpg was told `sum` was text and
2661 // decoded 420 as the string "420". The text protocol re-derives
2662 // types from the rows it really got, which is why psycopg2's
2663 // suite stayed green throughout and only asyncpg's went red.
2664 //
2665 // So: evidence where there is evidence, and static inference from
2666 // the STORED data where there is none — which is what the
2667 // translator's `infer_field_oid` was doing all along.
2668 let fallback = evaluator_shape(&probe, db, &coll);
2669 let oids: Vec<i32> = done
2670 .project
2671 .iter()
2672 .enumerate()
2673 .map(|(i, c)| {
2674 let seen = has_evidence(&done.rows, &c.src);
2675 if seen {
2676 oid_for(&done.rows, &c.src)
2677 } else {
2678 fallback
2679 .as_ref()
2680 .and_then(|(_, o)| o.get(i).copied())
2681 .unwrap_or(OID_TEXT)
2682 }
2683 })
2684 .collect();
2685 return Some((done.project, oids));
2686 }
2687 }
2688
2689
2690 // Ask the engine that will actually answer. Falls through when the
2691 // evaluator declines to describe itself — `SELECT *` expands from rows
2692 // Describe has not read — and the translator's shape is then the better
2693 // of the two available answers rather than the right one.
2694 if sql_engine_owns(&probe) {
2695 if let Some(shape) = evaluator_shape(&probe, db, &coll) {
2696 return Some(shape);
2697 }
2698 }
2699
2700 let stmt = translate(&probe).ok()?;
2701
2702 let cols = match stmt {
2703 Stmt::Ok(_) => return None,
2704 Stmt::Canned { cols, .. } => cols.iter().map(|c| Col::same(c)).collect(),
2705 Stmt::Query { project, .. } => {
2706 if project.is_empty() { sample_columns(db, &coll) } else { project }
2707 }
2708 Stmt::Insert { returning, .. } | Stmt::Update { returning, .. } | Stmt::Delete { returning, .. } => {
2709 if !wants_returning(sql) {
2710 return None;
2711 }
2712 if returning.is_empty() { sample_columns(db, &coll) } else { returning }
2713 }
2714 };
2715 if cols.is_empty() {
2716 // Nothing could be determined. `NoData` is a lie for a SELECT, but a
2717 // RowDescription with zero columns is a worse one — it tells the client
2718 // the query definitively has no output.
2719 return None;
2720 }
2721 let oids = cols
2722 .iter()
2723 .map(|c| {
2724 aggregate_oid(&c.src, db, &coll)
2725 .unwrap_or_else(|| infer_field_oid(db, &coll, &c.src))
2726 })
2727 .collect();
2728 Some((cols, oids))
2729}
2730
2731/// A parse-only stand-in for a parameterised statement.
2732///
2733/// Substituting `NULL` was the obvious choice and the wrong one: a clause that
2734/// validates its argument rejects it, so `AS OF SYSTEM TIME $1` failed at
2735/// `Parse` — before the client ever bound a real sequence number. `0` parses
2736/// everywhere a literal can appear, and since only the SELECT list is read back
2737/// out, the stub's value never reaches an answer.
2738fn probe_sql(sql: &str, n_params: usize) -> String {
2739 let stub: Vec<Option<String>> = vec![Some("0".to_string()); n_params];
2740 substitute_params(sql, &stub).unwrap_or_else(|_| sql.to_string())
2741}
2742
2743/// Run a portal's statement if it has not run yet, then report its shape.
2744fn ensure_executed(
2745 portal: &mut Portal,
2746 db_name: &str,
2747 db: Option<&Arc<Db>>,
2748 read_only: bool,
2749) -> Result<(), Vec<u8>> {
2750 if portal.result.is_some() {
2751 return Ok(());
2752 }
2753 let ex = execute_stmt(&portal.sql, db_name, db, read_only)?;
2754 // Freeze the output shape on first sight so `Describe` and every later
2755 // `Execute` describe the same rectangle.
2756 let project = if let Some(f) = &portal.frozen {
2757 f.clone()
2758 } else {
2759 let p = if ex.project.is_empty() {
2760 columns_for(&ex.rows, &[])
2761 } else {
2762 ex.project.clone()
2763 };
2764 portal.frozen = Some(p.clone());
2765 p
2766 };
2767 portal.result = Some(PortalResult {
2768 rows: ex.rows,
2769 project,
2770 has_rows: ex.has_rows,
2771 tag: ex.tag,
2772 tag_counts_rows: ex.tag_counts_rows,
2773 sent: 0,
2774 });
2775 Ok(())
2776}
2777
2778// ── connection handling ─────────────────────────────────────────────────────
2779
2780async fn read_exact(sock: &mut TcpStream, n: usize) -> std::io::Result<Vec<u8>> {
2781 let mut buf = vec![0u8; n];
2782 sock.read_exact(&mut buf).await?;
2783 Ok(buf)
2784}
2785
2786async fn read_i32(sock: &mut TcpStream) -> std::io::Result<i32> {
2787 let b = read_exact(sock, 4).await?;
2788 Ok(i32::from_be_bytes([b[0], b[1], b[2], b[3]]))
2789}
2790
2791fn parse_startup_params(body: &[u8]) -> HashMap<String, String> {
2792 let mut out = HashMap::new();
2793 let mut parts = body.split(|b| *b == 0).map(|s| String::from_utf8_lossy(s).to_string());
2794 while let (Some(k), Some(v)) = (parts.next(), parts.next()) {
2795 if k.is_empty() {
2796 break;
2797 }
2798 out.insert(k, v);
2799 }
2800 out
2801}
2802
2803/// Serve one client connection to completion.
2804async fn handle(mut sock: TcpStream, resolver: Arc<dyn DbResolver>, read_only: bool) -> std::io::Result<()> {
2805 // ── startup, including the SSL negotiation clients try first ────────────
2806 let params = loop {
2807 let len = read_i32(&mut sock).await?;
2808 if len < 8 || len > 1 << 20 {
2809 return Ok(()); // nonsense framing — drop the connection
2810 }
2811 let code = read_i32(&mut sock).await?;
2812 let body = read_exact(&mut sock, (len - 8) as usize).await?;
2813 match code {
2814 SSL_REQUEST | GSS_REQUEST => {
2815 // Decline and let the client retry in the clear.
2816 sock.write_all(b"N").await?;
2817 continue;
2818 }
2819 CANCEL_REQUEST => return Ok(()), // nothing cancellable: reads are synchronous
2820 PROTO_V3 => break parse_startup_params(&body),
2821 other => {
2822 let major = other >> 16;
2823 sock.write_all(&err_msg(
2824 "0A000",
2825 &format!("unsupported frontend protocol {}.{} — this endpoint speaks 3.0",
2826 major, other & 0xffff),
2827 )).await?;
2828 return Ok(());
2829 }
2830 }
2831 };
2832
2833 let db_name = params.get("database").cloned().unwrap_or_default();
2834
2835 // Resolve the database ONCE, here, on a blocking thread.
2836 //
2837 // A Postgres connection is bound to one database for its whole life, so
2838 // per-connection resolution is both correct and simpler than resolving per
2839 // statement — and it keeps the lock acquisition off the async worker.
2840 let resolved: Option<Arc<Db>> = {
2841 let r = Arc::clone(&resolver);
2842 let name = db_name.clone();
2843 tokio::task::spawn_blocking(move || r.resolve(&name))
2844 .await
2845 .unwrap_or(None)
2846 };
2847
2848 // ── auth: mirror the HTTP surface ───────────────────────────────────────
2849 if let Some(expected) = resolver.token() {
2850 // AuthenticationCleartextPassword (3)
2851 let mut m = Out::msg(b'R');
2852 m.i32(3);
2853 sock.write_all(&m.finish()).await?;
2854
2855 let tag = read_exact(&mut sock, 1).await?;
2856 if tag[0] != b'p' {
2857 sock.write_all(&err_msg("28000", "expected a password message")).await?;
2858 return Ok(());
2859 }
2860 let len = read_i32(&mut sock).await?;
2861 if len < 4 || len > 1 << 16 {
2862 return Ok(());
2863 }
2864 let body = read_exact(&mut sock, (len - 4) as usize).await?;
2865 let supplied = String::from_utf8_lossy(&body).trim_end_matches('\0').to_string();
2866 // Constant-time-ish: compare lengths and bytes without early return.
2867 let ok = supplied.len() == expected.len()
2868 && supplied.bytes().zip(expected.bytes()).fold(0u8, |a, (x, y)| a | (x ^ y)) == 0;
2869 if !ok {
2870 sock.write_all(&err_msg("28P01", "password authentication failed")).await?;
2871 return Ok(());
2872 }
2873 }
2874
2875 let mut m = Out::msg(b'R');
2876 m.i32(0); // AuthenticationOk
2877 sock.write_all(&m.finish()).await?;
2878
2879 for (k, v) in [
2880 ("server_version", SERVER_VERSION),
2881 ("server_encoding", "UTF8"),
2882 ("client_encoding", "UTF8"),
2883 ("DateStyle", "ISO, MDY"),
2884 ("integer_datetimes", "on"),
2885 ("standard_conforming_strings", "on"),
2886 ("application_name", "nedbd"),
2887 ] {
2888 let mut p = Out::msg(b'S');
2889 p.cstr(k);
2890 p.cstr(v);
2891 sock.write_all(&p.finish()).await?;
2892 }
2893 let mut k = Out::msg(b'K');
2894 k.i32(std::process::id() as i32);
2895 k.i32(0);
2896 sock.write_all(&k.finish()).await?;
2897 sock.write_all(&ready()).await?;
2898
2899 // ── message loop ────────────────────────────────────────────────────────
2900 //
2901 // Prepared statements and portals live for the connection. `""` is the
2902 // unnamed statement/portal, which every driver reuses constantly — it is an
2903 // ordinary entry in the map rather than a special case.
2904 let mut prepared: HashMap<String, Prepared> = HashMap::new();
2905 let mut portals: HashMap<String, Portal> = HashMap::new();
2906 // After an error inside an extended-protocol sequence, everything up to the
2907 // next `Sync` is discarded. Skipping this is how a server ends up answering
2908 // a Bind the client has already abandoned, and the stream desynchronises.
2909 let mut failed = false;
2910
2911 loop {
2912 let mut tag = [0u8; 1];
2913 if sock.read_exact(&mut tag).await.is_err() {
2914 return Ok(()); // client hung up
2915 }
2916 let len = read_i32(&mut sock).await?;
2917 if len < 4 || len > 64 << 20 {
2918 return Ok(());
2919 }
2920 let body = read_exact(&mut sock, (len - 4) as usize).await?;
2921
2922 // `Sync` always clears the error state; `Terminate` always applies.
2923 if failed && tag[0] != b'S' && tag[0] != b'X' {
2924 continue;
2925 }
2926
2927 match tag[0] {
2928 b'X' => return Ok(()), // Terminate
2929
2930 b'Q' => {
2931 let sql = String::from_utf8_lossy(&body).trim_end_matches('\0').to_string();
2932 let out = run_simple_query(&sql, &db_name, resolved.as_ref(), read_only);
2933 sock.write_all(&out).await?;
2934 sock.write_all(&ready()).await?;
2935 // A simple query closes the unnamed portal, per the protocol.
2936 portals.remove("");
2937 }
2938
2939 // ── Parse: name, SQL, declared parameter type OIDs ─────────────
2940 b'P' => {
2941 let mut at = 0usize;
2942 let name = take_cstr(&body, &mut at);
2943 let sql = take_cstr(&body, &mut at);
2944 let n = take_i16(&body, &mut at).unwrap_or(0).max(0) as usize;
2945 let mut declared = Vec::with_capacity(n);
2946 let mut bad = false;
2947 for _ in 0..n {
2948 match take_i32(&body, &mut at) {
2949 Ok(o) => declared.push(o),
2950 Err(_) => { bad = true; break; }
2951 }
2952 }
2953 if bad {
2954 sock.write_all(&err_msg("08P01", "malformed Parse message")).await?;
2955 failed = true;
2956 continue;
2957 }
2958 // Reject unsupported SQL here rather than at Execute, so the
2959 // client learns at the point it asked — which is also where
2960 // Postgres reports it.
2961 //
2962 // The SQL evaluator gets asked first, or a catalogue query
2963 // would be refused at `Parse` by the NQL path that was never
2964 // going to run it — and the extended protocol is where every
2965 // ORM and async driver lives, so refusing here refuses them
2966 // all.
2967 let probe = probe_sql(&sql, param_count(&sql));
2968 if !sql_engine_owns(&probe) {
2969 if let Err(why) = translate(&probe) {
2970 sock.write_all(&err_msg("0A000", &why)).await?;
2971 failed = true;
2972 continue;
2973 }
2974 }
2975 let param_oids = infer_param_oids(&sql, &declared, resolved.as_ref());
2976 prepared.insert(name, Prepared { sql, param_oids, out_shape: None });
2977 sock.write_all(&parse_complete()).await?;
2978 }
2979
2980 // ── Bind: portal, statement, formats, values, result formats ───
2981 b'B' => {
2982 let mut at = 0usize;
2983 let portal_name = take_cstr(&body, &mut at);
2984 let stmt_name = take_cstr(&body, &mut at);
2985 if !prepared.contains_key(&stmt_name) {
2986 sock.write_all(&err_msg("26000", &format!(
2987 "prepared statement {:?} does not exist", stmt_name))).await?;
2988 failed = true;
2989 continue;
2990 }
2991 let p = &prepared[&stmt_name];
2992 let mut want_formats: Vec<i16> = vec![];
2993 let res: Result<String, String> = (|| {
2994 let nfmt = take_i16(&body, &mut at)? .max(0) as usize;
2995 let mut fmts = Vec::with_capacity(nfmt);
2996 for _ in 0..nfmt {
2997 fmts.push(take_i16(&body, &mut at)?);
2998 }
2999 let nparam = take_i16(&body, &mut at)?.max(0) as usize;
3000 let mut vals: Vec<Option<String>> = Vec::with_capacity(nparam);
3001 for i in 0..nparam {
3002 let l = take_i32(&body, &mut at)?;
3003 let raw: Option<Vec<u8>> = if l < 0 {
3004 None
3005 } else {
3006 let l = l as usize;
3007 if at + l > body.len() {
3008 return Err("truncated Bind parameter".into());
3009 }
3010 let v = body[at..at + l].to_vec();
3011 at += l;
3012 Some(v)
3013 };
3014 // Zero format codes means "all text"; one means "this
3015 // format for every parameter"; otherwise one per value.
3016 let f = match fmts.len() {
3017 0 => 0,
3018 1 => fmts[0],
3019 _ => *fmts.get(i).unwrap_or(&0),
3020 };
3021 let oid = *p.param_oids.get(i).unwrap_or(&OID_TEXT);
3022 vals.push(decode_param(raw.as_deref(), oid, f)?);
3023 }
3024 // Result format codes. asyncpg asks for binary on every
3025 // column, so honouring these is not an optimisation — it
3026 // is the difference between asyncpg reading rows and
3027 // refusing the result outright.
3028 let nres = take_i16(&body, &mut at)?.max(0) as usize;
3029 for _ in 0..nres {
3030 let f = take_i16(&body, &mut at)?;
3031 if f != 0 && f != 1 {
3032 return Err(format!("unknown result format code {}", f));
3033 }
3034 want_formats.push(f);
3035 }
3036 substitute_params(&p.sql, &vals)
3037 })();
3038 match res {
3039 Ok(sql) => {
3040 // Binary encoding must use the types the client was
3041 // TOLD about, so pull the advertised shape across.
3042 let declared = if want_formats.iter().any(|f| *f == 1) {
3043 let p = prepared.get_mut(&stmt_name).expect("checked above");
3044 prepared_shape(p, resolved.as_ref()).clone()
3045 } else {
3046 None
3047 };
3048 portals.insert(portal_name, Portal {
3049 sql, result: None, frozen: None,
3050 formats: want_formats, declared,
3051 });
3052 sock.write_all(&bind_complete()).await?;
3053 }
3054 Err(why) => {
3055 sock.write_all(&err_msg("08P01", &why)).await?;
3056 failed = true;
3057 }
3058 }
3059 }
3060
3061 // ── Describe: 'S' statement, or 'P' portal ─────────────────────
3062 b'D' => {
3063 let kind = body.first().copied().unwrap_or(b'S');
3064 let mut at = 1usize;
3065 let name = take_cstr(&body, &mut at);
3066 if kind == b'S' {
3067 if !prepared.contains_key(&name) {
3068 sock.write_all(&err_msg("26000", &format!(
3069 "prepared statement {:?} does not exist", name))).await?;
3070 failed = true;
3071 continue;
3072 }
3073 let p = prepared.get_mut(&name).expect("checked above");
3074 let oids = p.param_oids.clone();
3075 // asyncpg encodes its arguments from this, so the count has
3076 // to be right or it refuses the call before sending a Bind.
3077 sock.write_all(¶meter_description(&oids)).await?;
3078 // Describe(statement) happens before Bind, so the requested
3079 // result format is not known yet; Postgres reports text
3080 // here too and the client's own Bind decides the encoding.
3081 let out = match prepared_shape(p, resolved.as_ref()) {
3082 Some((cols, col_oids)) => row_description(cols, col_oids),
3083 None => no_data(),
3084 };
3085 sock.write_all(&out).await?;
3086 } else {
3087 let portal = match portals.get_mut(&name) {
3088 Some(p) => p,
3089 None => {
3090 sock.write_all(&err_msg("34000", &format!(
3091 "portal {:?} does not exist", name))).await?;
3092 failed = true;
3093 continue;
3094 }
3095 };
3096 // A bound portal can be run: doing it here means the
3097 // RowDescription reports the columns and types actually
3098 // present, which is strictly better than a guess. psycopg3
3099 // takes this path on every query.
3100 match ensure_executed(portal, &db_name, resolved.as_ref(), read_only) {
3101 Err(encoded) => {
3102 sock.write_all(&encoded).await?;
3103 failed = true;
3104 }
3105 Ok(()) => {
3106 let r = portal.result.as_ref().expect("just executed");
3107 if !r.has_rows {
3108 sock.write_all(&no_data()).await?;
3109 } else {
3110 let (cols, oids) = portal.shape(r);
3111 let fmts: Vec<i16> =
3112 (0..cols.len()).map(|i| portal.format_of(i)).collect();
3113 sock.write_all(&row_description_fmt(&cols, &oids, &fmts)).await?;
3114 }
3115 }
3116 }
3117 }
3118 }
3119
3120 // ── Execute: portal, maximum rows (0 = all) ────────────────────
3121 b'E' => {
3122 let mut at = 0usize;
3123 let name = take_cstr(&body, &mut at);
3124 let max_rows = take_i32(&body, &mut at).unwrap_or(0);
3125 let portal = match portals.get_mut(&name) {
3126 Some(p) => p,
3127 None => {
3128 sock.write_all(&err_msg("34000", &format!(
3129 "portal {:?} does not exist", name))).await?;
3130 failed = true;
3131 continue;
3132 }
3133 };
3134 if let Err(encoded) = ensure_executed(portal, &db_name, resolved.as_ref(), read_only) {
3135 sock.write_all(&encoded).await?;
3136 failed = true;
3137 continue;
3138 }
3139 let r = portal.result.as_ref().expect("just executed");
3140 if !r.has_rows {
3141 let tag = r.tag.clone();
3142 sock.write_all(&command_complete(&tag)).await?;
3143 continue;
3144 }
3145 let (cols, oids) = portal.shape(r);
3146 let limit = if max_rows > 0 {
3147 (r.sent + max_rows as usize).min(r.rows.len())
3148 } else {
3149 r.rows.len()
3150 };
3151 // Encode the whole batch BEFORE writing any of it. A value that
3152 // cannot be sent in the advertised binary type has to become an
3153 // error instead of a truncated row stream — half a result set
3154 // followed by an error is far harder to diagnose than an error.
3155 let mut encoded: Vec<Vec<u8>> = Vec::with_capacity(limit - r.sent);
3156 let mut fail: Option<String> = None;
3157 for row in &r.rows[r.sent..limit] {
3158 let mut vals: Vec<Option<Vec<u8>>> = Vec::with_capacity(cols.len());
3159 for (i, c) in cols.iter().enumerate() {
3160 let v = row.get(&c.src);
3161 let got = if portal.format_of(i) == 1 {
3162 cell_binary(v, oids.get(i).copied().unwrap_or(OID_TEXT))
3163 .map_err(|e| format!("column {:?}: {}", c.out, e))
3164 } else {
3165 Ok(cell(v).map(|s| s.into_bytes()))
3166 };
3167 match got {
3168 Ok(b) => vals.push(b),
3169 Err(e) => { fail = Some(e); break; }
3170 }
3171 }
3172 if fail.is_some() {
3173 break;
3174 }
3175 encoded.push(data_row_bytes(&vals));
3176 }
3177 if let Some(why) = fail {
3178 sock.write_all(&err_msg("22P03", &why)).await?;
3179 failed = true;
3180 continue;
3181 }
3182 let mut out = vec![];
3183 for e in &encoded {
3184 out.extend_from_slice(e);
3185 }
3186 let r = portal.result.as_mut().expect("just executed");
3187 r.sent = limit;
3188 // More rows left and the client capped the batch: suspend the
3189 // portal instead of completing it. This is what a JDBC
3190 // `setFetchSize` and a psycopg3 server-side cursor rely on.
3191 if max_rows > 0 && r.sent < r.rows.len() {
3192 out.extend_from_slice(&portal_suspended());
3193 } else {
3194 let tag = if r.tag_counts_rows {
3195 format!("{} {}", r.tag, r.sent)
3196 } else {
3197 r.tag.clone()
3198 };
3199 out.extend_from_slice(&command_complete(&tag));
3200 }
3201 sock.write_all(&out).await?;
3202 }
3203
3204 // ── Close: 'S' statement, or 'P' portal ───────────────────────
3205 b'C' => {
3206 let kind = body.first().copied().unwrap_or(b'S');
3207 let mut at = 1usize;
3208 let name = take_cstr(&body, &mut at);
3209 if kind == b'S' {
3210 prepared.remove(&name);
3211 } else {
3212 portals.remove(&name);
3213 }
3214 // Closing something that was never open is explicitly not an
3215 // error in the protocol.
3216 sock.write_all(&close_complete()).await?;
3217 }
3218
3219 // Flush: everything is written unbuffered already, so this is a
3220 // no-op — but it must NOT produce a ReadyForQuery, or a client that
3221 // flushes mid-sequence (asyncpg does, after Describe) loses sync.
3222 b'H' => {}
3223
3224 b'S' => {
3225 failed = false;
3226 sock.write_all(&ready()).await?;
3227 }
3228
3229 other => {
3230 sock.write_all(&err_msg(
3231 "08P01",
3232 &format!("unexpected frontend message {:?}", other as char),
3233 )).await?;
3234 failed = true;
3235 }
3236 }
3237 }
3238}
3239
3240const READ_ONLY_MSG: &str =
3241 "this endpoint is running read-only (NEDBD_PG_READ_ONLY=1). Writes are \
3242 implemented but disabled on this server — unset the flag to allow them.";
3243
3244fn no_db(db_name: &str) -> Vec<u8> {
3245 err_msg("3D000", &format!(
3246 "database {:?} is not open on this server — create it first \
3247 (POST /v1/databases), or connect with -d <name>", db_name))
3248}
3249
3250/// `pg_catalog.pg_class` → `pg_class`, but `information_schema.tables` keeps
3251/// its qualifier, because `tables` is a plausible collection name and the
3252/// catalogue must never shadow a user's own data.
3253fn catalog_name(n: &str) -> String {
3254 let joined: Vec<&str> = n.split('.').collect();
3255 if joined.len() >= 2 && joined[joined.len() - 2] == "information_schema" {
3256 format!("information_schema.{}", joined[joined.len() - 1])
3257 } else {
3258 joined[joined.len() - 1].to_string()
3259 }
3260}
3261
3262/// Does the SQL evaluator own this statement?
3263///
3264/// Two ways in. The first is obvious: it reads a catalogue relation.
3265///
3266/// The second is a statement with NO relation at all — a select list of
3267/// literals and scalar function calls, which is exactly what this evaluator
3268/// does and which the SQL→NQL path cannot express (NQL is FROM-first). That
3269/// path answers a handful of EXACT spellings from a canned table
3270/// (`SELECT 1`, `SELECT VERSION()`, `SELECT CURRENT_SCHEMA`), and those
3271/// answers are what existing clients already see — so this predicate rescues
3272/// only what it REFUSES, leaving every spelling it does handle alone.
3273///
3274/// That gap was not hypothetical. SQLAlchemy's PostgreSQL dialect opens every
3275/// connection with `select pg_catalog.version()`, which is one character of
3276/// qualification away from the canned `SELECT VERSION()` and therefore missed
3277/// it — so the engine refused the first statement of dialect initialisation
3278/// and NO SQLAlchemy application could connect at all. A canned list of
3279/// spellings is the same brittleness `pgcatalog` exists to avoid; the fix is
3280/// to let the evaluator answer, because it has `version()`,
3281/// `current_setting()` and the rest as real functions.
3282///
3283/// Cheap: one parse, no execution, no storage access.
3284/// Opt-in: route USER-collection `SELECT`s through the SQL evaluator too.
3285///
3286/// `NEDBD_SQL_ENGINE=1`. Default OFF, and the default is the point — this
3287/// changes which engine answers ordinary queries, and the two engines have to
3288/// be shown to agree before anyone's production reads move. Flipping it is a
3289/// deployment decision, not a build one, so it is read from the environment
3290/// once rather than compiled in.
3291///
3292/// What it unlocks is everything the translator refuses because NQL cannot
3293/// express it: joins, subqueries, `EXISTS`, `UNION`/`INTERSECT`/`EXCEPT`,
3294/// several named aggregates in one grouped row, `array_agg(x ORDER BY y)`.
3295/// What it must not lose is what only the translator has — and a statement the
3296/// evaluator's grammar cannot parse (`TRACE`, `SEARCH`, `VALID AS OF`,
3297/// `TRAVERSE`, every write) still falls through to the translator on its own,
3298/// because `parse` fails and this function is never consulted.
3299/// NQL's table-level verbs, gathered per relation name.
3300///
3301/// A struct rather than the tuple this started as. It held
3302/// `(valid_as_of, search)`; adding `TRACE` and `TRAVERSE` would have made it a
3303/// four-tuple indexed by `.0` through `.3`, and the resolver reads these in a
3304/// different order than it builds them — which is precisely how a positional
3305/// tuple turns into `SEARCH` being rendered where `VALID AS OF` was meant.
3306#[derive(Default, Clone)]
3307struct TableVerbs {
3308 valid_as_of: Option<String>,
3309 search: Option<String>,
3310 /// The edge type for `TRACE <edge>`.
3311 trace: Option<String>,
3312 /// `REVERSE` — walk effects rather than causes.
3313 trace_reverse: bool,
3314 /// The relation name for `TRAVERSE <rel>`.
3315 traverse: Option<String>,
3316}
3317
3318impl TableVerbs {
3319 /// Does this relation carry any verb the catalogue cannot answer?
3320 fn first_unsupported_on_catalogue(&self) -> Option<&'static str> {
3321 if self.valid_as_of.is_some() {
3322 Some("VALID AS OF")
3323 } else if self.search.is_some() {
3324 Some("SEARCH")
3325 } else if self.trace.is_some() {
3326 Some("TRACE")
3327 } else if self.traverse.is_some() {
3328 Some("TRAVERSE")
3329 } else {
3330 None
3331 }
3332 }
3333
3334}
3335
3336/// Whether `NEDBD_SQL_ENGINE` is still set in someone's environment.
3337///
3338/// The flag no longer selects anything — the evaluator answers every SELECT it
3339/// can parse. It is read only so a deployment that still exports it is TOLD
3340/// the variable is now inert, rather than left believing it is holding a
3341/// switch that no longer exists. Silence here is how an operator ends up
3342/// certain their reads are on the old path.
3343fn stale_sql_engine_flag() -> bool {
3344 use std::sync::OnceLock;
3345 static ON: OnceLock<bool> = OnceLock::new();
3346 *ON.get_or_init(|| {
3347 let set = std::env::var("NEDBD_SQL_ENGINE").is_ok();
3348 if set {
3349 eprintln!(
3350 "[nedbd] NEDBD_SQL_ENGINE is set but no longer does anything. The SQL \
3351 evaluator now answers every SELECT it can parse; statements it cannot \
3352 parse still fall through to the translator. You can remove the variable."
3353 );
3354 }
3355 set
3356 })
3357}
3358
3359/// The pre-filtered scan, still spelled in NQL.
3360///
3361/// The LAST place a relation is expressed as text, and it survives for a
3362/// reason that does not apply to the others: the pre-filter is an
3363/// OPTIMISATION. `sqlpush` renders the part of the `WHERE` that NQL evaluates
3364/// identically, so pushing it saves reading rows — and the evaluator's real
3365/// `WHERE` runs above regardless, so getting it wrong costs a wasted row and
3366/// never an answer. Everything else about the scan is a MEANING, and meanings
3367/// now travel as a `relation::Scan` that cannot drop a field.
3368///
3369/// Derived FROM that same struct rather than from the original clauses, so the
3370/// two cannot disagree about what is being read. When the index scan learns to
3371/// take a predicate directly, this function and NQL's parser go together.
3372fn compose_prefiltered(cname: &str, scan: &crate::relation::Scan, pre: &str) -> String {
3373 let mut q = format!("FROM {}", cname);
3374 if let Some(seq) = scan.as_of {
3375 q.push_str(&format!(" AS OF {}", seq));
3376 }
3377 if let Some(d) = &scan.valid_as_of {
3378 q.push_str(&format!(" VALID AS OF {}", nql_string(d)));
3379 }
3380 q.push_str(&format!(" WHERE {}", pre));
3381 if let Some(t) = &scan.search {
3382 q.push_str(&format!(" SEARCH {}", nql_string(t)));
3383 }
3384 if let Some(edge) = &scan.trace {
3385 q.push_str(&format!(" TRACE {}", edge));
3386 if scan.trace_reverse {
3387 q.push_str(" REVERSE");
3388 }
3389 }
3390 if let Some(rel) = &scan.traverse {
3391 q.push_str(&format!(" TRAVERSE {}", rel));
3392 }
3393 q
3394}
3395
3396fn sql_engine_owns(sql: &str) -> bool {
3397 let Ok(sel) = crate::sqlselect::parse(sql) else { return false };
3398 let touched = sel.base_relations();
3399 if touched.is_empty() {
3400 return translate(sql).is_err();
3401 }
3402 if touched.iter().any(|t| crate::pgcatalog::is_catalog(&catalog_name(t))) {
3403 return true;
3404 }
3405 // A user collection reaches the evaluator too, unconditionally. There is
3406 // ONE evaluator now.
3407 //
3408 // This used to return `sql_engine_for_collections()` — an env flag,
3409 // default OFF, on the argument that a collection "has a working answer on
3410 // both paths, so the choice between them is a judgement about parity".
3411 // That argument stopped being true. The translator's answer is not a
3412 // second correct answer, it is a worse one:
3413 //
3414 // SELECT who FROM orders translator -> who, total, _id, _hash,
3415 // _seq, _coll
3416 // evaluator -> who
3417 //
3418 // The projection list was ignored entirely, because NQL has no projection
3419 // to translate it into. `sum(total), avg(total)` in one grouped row is not
3420 // slow on the translator, it is unrepresentable. A flag whose two
3421 // positions give different answers to the same correct SQL is not a
3422 // parity switch, it is a bug with a toggle.
3423 //
3424 // What made this safe to flip is that the fallthrough was never the flag.
3425 // A statement this evaluator cannot PARSE never reaches here — `parse`
3426 // fails at the top of this function and the translator takes it, which is
3427 // still how every write, and anything outside the SELECT grammar, is
3428 // served. Removing the flag narrows nothing; it stops answering parseable
3429 // SQL with a translation of it.
3430 //
3431 // Called here only for its one-shot warning: this is the first point at
3432 // which a deployment still exporting the variable is demonstrably running
3433 // the evaluator, which is exactly when saying so is useful.
3434 let _ = stale_sql_engine_flag();
3435
3436 // The translator has not gone anywhere. It still answers every write and
3437 // every statement this evaluator cannot parse, so the two paths still
3438 // coexist and still have to agree where both can answer. That agreement is
3439 // proven by tests/test_pgwire_parity.py, which spawns two daemons and
3440 // compares them — and which became a TAUTOLOGY the moment the flag it used
3441 // to tell them apart stopped selecting anything. Its own header warned
3442 // about exactly this failure, from the environment side; this is the same
3443 // failure from the code side.
3444 //
3445 // So the lever survives for the harness, under a name no one will mistake
3446 // for a product switch, and pointed the other way: it forces the
3447 // TRANSLATOR rather than enabling the evaluator. Nothing in the product
3448 // reads it, the default path has no flag in it at all, and a parity run
3449 // that forgets to set it compares the evaluator with itself and is
3450 // supposed to look wrong.
3451 !force_translator_for_parity()
3452}
3453
3454/// TEST-ONLY. Forces user collections back onto the translator.
3455///
3456/// Not a supported configuration and not a fallback: it exists so
3457/// `test_pgwire_parity.py` can still put a translator daemon next to an
3458/// evaluator daemon now that `NEDBD_SQL_ENGINE` selects nothing. Setting it in
3459/// production gives you the projection-dropping answers this change removed.
3460fn force_translator_for_parity() -> bool {
3461 use std::sync::OnceLock;
3462 static ON: OnceLock<bool> = OnceLock::new();
3463 *ON.get_or_init(|| {
3464 let on = matches!(
3465 std::env::var("NEDB_PARITY_FORCE_TRANSLATOR").as_deref(),
3466 Ok("1") | Ok("true") | Ok("on")
3467 );
3468 if on {
3469 eprintln!(
3470 "[nedbd] NEDB_PARITY_FORCE_TRANSLATOR is set — user collections are being \
3471 answered by the TRANSLATOR. This is a test lever for the parity harness, \
3472 not a supported configuration: projections are dropped on this path."
3473 );
3474 }
3475 on
3476 })
3477}
3478
3479/// Run a `SELECT` through the full SQL engine when it touches the catalogue.
3480///
3481/// The gate is deliberately narrow: a statement goes to `sqlselect` only when
3482/// one of its tables is a catalogue relation. Everything else keeps the
3483/// SQL→NQL path, which has the index pushdown, `AS OF`, `TRACE` and the
3484/// bounded scans — and whose join story is a real planning question rather
3485/// than a nested loop. Routing a large collection through a nested-loop join
3486/// would be a promise this engine cannot keep.
3487///
3488/// `None` means "not mine": the caller falls through to the ordinary path, so
3489/// the error the client sees is the ordinary path's error rather than a
3490/// confusing one from a parser that was never meant to handle the statement.
3491fn try_catalog_select(
3492 sql: &str,
3493 db: Option<&Arc<Db>>,
3494) -> Result<Option<(Executed, crate::sqlplan::Plan)>, Vec<u8>> {
3495 let sel = match crate::sqlselect::parse(sql) {
3496 Ok(sel) => sel,
3497 Err(why) => {
3498 // A statement that plainly reads the catalogue but that this
3499 // engine cannot parse gets the PARSE error, not the NQL path's.
3500 //
3501 // Falling through unconditionally produced an actively false
3502 // message: `\d` and `\dp` were told "JOIN is not supported",
3503 // which stopped being true the moment joins started working — and
3504 // a wrong explanation is worse than a blunt one, because it sends
3505 // the reader to fix the wrong thing.
3506 if mentions_catalog(sql) {
3507 return Err(err_msg("0A000", &format!(
3508 "this catalogue query uses SQL this endpoint does not \
3509 implement: {}", why)));
3510 }
3511 return Ok(None);
3512 }
3513 };
3514
3515 // Which relations does it read — at ANY depth? `\dd` names its catalogue
3516 // relations only inside a derived table, and `\dT` only inside two
3517 // subqueries; a walk over the top-level FROM list alone would route both
3518 // to the NQL path, which cannot parse them and would report an error that
3519 // sends the reader to fix the wrong thing.
3520 if !sql_engine_owns(sql) {
3521 return Ok(None);
3522 }
3523
3524 // The storage pre-filter, resolved per relation NAME and computed once.
3525 //
3526 // The resolver is handed a name (`orders`) but the WHERE clause qualifies
3527 // by BINDING (`o.status` for `FROM orders o`), so the predicate has to be
3528 // looked up by name and rendered against that relation's binding. Getting
3529 // this wrong is silent: the pre-filter simply never matches and the scan
3530 // quietly reads the whole collection, which is exactly what EXPLAIN caught
3531 // the first time round — `Seq Scan on orders o (actual rows=3)` when the
3532 // query wanted two.
3533 //
3534 // A name appearing TWICE (a self-join, `FROM t a JOIN t b`) maps to two
3535 // different bindings with different predicates, and one scan cannot serve
3536 // both. Those are dropped rather than guessed at.
3537 // `AS OF SYSTEM TIME <seq>`, per relation name.
3538 //
3539 // The resolver is keyed by NAME, so one collection named twice gets ONE
3540 // scan. `FROM orders AS OF 1 o JOIN orders n` asks for that collection at
3541 // two different sequences at once, and a single scan cannot serve both.
3542 //
3543 // This is REFUSED rather than resolved to one of them, and the reason is
3544 // worth keeping: the first version dropped the qualifier when a name was
3545 // ambiguous — the same "don't guess" instinct that is right for a
3546 // pre-filter. It is wrong here. Dropping a pre-filter costs a wasted row;
3547 // dropping an AS OF answers a question about the past with data from the
3548 // present, and it does it silently. The query `... orders AS OF 1 o JOIN
3549 // orders n ...` returned the CURRENT value for both sides and looked fine.
3550 let temporal: std::collections::HashMap<String, u64> = {
3551 // Wall-clock markers (a quoted datetime in `AS OF SYSTEM TIME ''`)
3552 // resolve to real sequences HERE, before the one-scan-per-name
3553 // reconciliation, so two datetime spellings naming the same moment
3554 // are equal — and a wall-clock + tip mix reports "at the tip and AS
3555 // OF <seq>" honestly rather than leaking a tagged marker into the
3556 // executor. Resolution is `Db::seq_at`: the last seq whose write-time
3557 // is at or before the moment, over the ts index the cold scan fills.
3558 let resolve_marker = |m: u64| -> Result<u64, Vec<u8>> {
3559 if (m & crate::wallclock::WALL_CLOCK_FLAG) == 0 {
3560 return Ok(m); // bare integer — a seq, untouched, backcompat
3561 }
3562 let moment = crate::wallclock::WallClock::from_marker(m)
3563 .ok_or_else(|| err_msg("0A000", "invalid wall-clock marker"))?;
3564 let db = db.ok_or_else(|| err_msg("0A000",
3565 "AS OF SYSTEM TIME by datetime names a database; connect with one"))?;
3566 if !db.ts_index_ready() {
3567 return Err(err_msg("0A000",
3568 "the write-time index is not ready on this boot (warm start defers it). \
3569 Run `nedb-cli repair` or a cold scan, or AS OF a bare sequence number"));
3570 }
3571 match db.seq_at(moment.epoch_secs()) {
3572 Some(seq) => Ok(seq),
3573 None => {
3574 let floor = db.history_floor();
3575 // Distinguish "before anything" from "pruned" — the two
3576 // read differently to an operator (one is routine, the
3577 // other is the compaction tradeoff answering).
3578 if floor > 0 {
3579 Err(err_msg("0A000", &format!(
3580 "history at or before that moment is no longer available — \
3581 the store was compacted past it (history floor {}). \
3582 AS OF a bare sequence at or after the floor instead", floor)))
3583 } else {
3584 Err(err_msg("0A000", &format!(
3585 "no writes at or before that moment in this database — \
3586 nothing existed yet (the first write is at seq {}). \
3587 A timestamp answers about the past; there is no past here yet",
3588 db.seq.load(std::sync::atomic::Ordering::SeqCst))))
3589 }
3590 }
3591 }
3592 };
3593 // Gather every sequence each name is read at first, INCLUDING the
3594 // absent one, then judge. Deciding as we walk got this wrong: the
3595 // first arm of a self-join was judged before it had been recorded, so
3596 // a legitimate pair reported the wrong reason.
3597 let mut seen: std::collections::HashMap<String, Vec<Option<u64>>> =
3598 std::collections::HashMap::new();
3599 for t in sel.from.iter().chain(sel.joins.iter().map(|j| &j.table)) {
3600 let resolved = match t.as_of {
3601 Some(m) => Some(resolve_marker(m).map_err(|e| e)?),
3602 None => None,
3603 };
3604 seen.entry(catalog_name(&t.name).to_ascii_lowercase())
3605 .or_default()
3606 .push(resolved);
3607 }
3608 let mut out: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
3609 for (key, ats) in &seen {
3610 let mut distinct: Vec<Option<u64>> = ats.clone();
3611 distinct.sort();
3612 distinct.dedup();
3613 match distinct.as_slice() {
3614 // One sequence for this name, however many times it appears.
3615 [Some(seq)] => {
3616 out.insert(key.clone(), *seq);
3617 }
3618 [None] => {}
3619 // More than one. Say WHICH disagreement it is, because the two
3620 // read very differently to whoever wrote the query.
3621 _ => {
3622 let mixed_tip = distinct.contains(&None);
3623 let seqs: Vec<String> =
3624 distinct.iter().flatten().map(|s| s.to_string()).collect();
3625 let detail = if mixed_tip {
3626 format!(
3627 "at the tip and AS OF {}",
3628 seqs.join(" and "))
3629 } else {
3630 format!("AS OF {}", seqs.join(" and "))
3631 };
3632 return Err(err_msg("0A000", &format!(
3633 "{:?} is read {} in one statement. This endpoint reads each \
3634 collection once per statement, so it cannot serve both — and \
3635 answering from either one would silently return the same rows for \
3636 both arms, which is the comparison failing to be a comparison. Ask \
3637 the two questions separately.",
3638 key, detail)));
3639 }
3640 }
3641 }
3642 out
3643 };
3644
3645 // NQL's own verbs, per relation name: `(VALID AS OF, SEARCH)`.
3646 //
3647 // Same one-scan-per-name constraint as the temporal map, and the same
3648 // verdict for the same reason: two different values for one scan is
3649 // REFUSED, because silently picking one would answer a different question
3650 // than the one asked and look like it worked.
3651 let nql_verbs: std::collections::HashMap<String, TableVerbs> = {
3652 let mut out: std::collections::HashMap<String, TableVerbs> =
3653 std::collections::HashMap::new();
3654 for t in sel.from.iter().chain(sel.joins.iter().map(|j| &j.table)) {
3655 let k = catalog_name(&t.name).to_ascii_lowercase();
3656 let e = out.entry(k.clone()).or_default();
3657 // REVERSE rides with the edge type rather than being reconciled on
3658 // its own: `TRACE caused_by` and `TRACE caused_by REVERSE` are two
3659 // different questions about the same edge, and reconciling the
3660 // direction separately would let them merge into one scan.
3661 if t.trace.is_some() {
3662 e.trace_reverse = t.trace_reverse;
3663 }
3664 for (slot, incoming, verb) in [
3665 (&mut e.valid_as_of, &t.valid_as_of, "VALID AS OF"),
3666 (&mut e.search, &t.search, "SEARCH"),
3667 (&mut e.trace, &t.trace, "TRACE"),
3668 (&mut e.traverse, &t.traverse, "TRAVERSE"),
3669 ] {
3670 match (slot.as_deref(), incoming.as_deref()) {
3671 (Some(a), Some(b)) if a != b => {
3672 return Err(err_msg("0A000", &format!(
3673 "{:?} is read with two different {} arguments in one statement \
3674 ({:?} and {:?}). This endpoint reads each collection once, so \
3675 it cannot serve both. Ask the two questions separately.",
3676 k, verb, a, b)));
3677 }
3678 (None, Some(b)) => *slot = Some(b.to_string()),
3679 _ => {}
3680 }
3681 }
3682 }
3683 out
3684 };
3685
3686 let pushdown_prefilters: std::collections::HashMap<String, String> = {
3687 let refs: Vec<&crate::sqlselect::TableRef> = sel
3688 .from
3689 .iter()
3690 .chain(sel.joins.iter().map(|j| &j.table))
3691 .collect();
3692 let bindings: Vec<String> = refs.iter().map(|t| t.binding()).collect();
3693 let nullable = crate::sqlpush::nullable_bindings(&sel);
3694 let mut out = std::collections::HashMap::new();
3695 let mut ambiguous: Vec<String> = vec![];
3696 for t in &refs {
3697 let key = catalog_name(&t.name).to_ascii_lowercase();
3698 if out.contains_key(&key) || ambiguous.contains(&key) {
3699 out.remove(&key);
3700 ambiguous.push(key);
3701 continue;
3702 }
3703 if let Some(p) = crate::sqlpush::nql_prefilter(
3704 sel.where_.as_ref(), &t.binding(), &bindings, &nullable) {
3705 out.insert(key, p);
3706 }
3707 }
3708 out
3709 };
3710
3711 let resolve = |name: &str| -> anyhow::Result<Option<Box<dyn crate::sqlselect::Relation>>> {
3712 let cname = catalog_name(name);
3713 // A catalogue relation is SYNTHESISED from the current shape of the
3714 // store: it has no log, so it has no history, and there is nothing for
3715 // a temporal or full-text qualifier to mean.
3716 //
3717 // Refused rather than ignored, and the difference is the entire point.
3718 // Ignoring `AS OF SYSTEM TIME 0` answers a question about the past with
3719 // present-day rows and looks like it worked — and that is exactly what
3720 // started happening here the moment the SQL parser learned `AS OF`:
3721 // before, the statement failed to parse and fell through to the
3722 // translator, which refused it properly. Teaching one layer a clause
3723 // silently un-taught another layer's refusal, and a test written long
3724 // before this change is what caught it.
3725 {
3726 let k = cname.to_ascii_lowercase();
3727 let bad = if temporal.contains_key(&k) {
3728 Some("AS OF SYSTEM TIME")
3729 } else {
3730 nql_verbs.get(&k).and_then(|v| v.first_unsupported_on_catalogue())
3731 };
3732 if let Some(clause) = bad {
3733 if crate::pgcatalog::is_catalog(&cname) {
3734 anyhow::bail!(
3735 "{} is not supported on the catalogue relation {:?} — a catalogue is \
3736 synthesised from the store's current shape rather than read from the \
3737 log, so it has no history to reach and no document text to search. \
3738 Ignoring the clause would answer your question with present-day rows \
3739 and look like it worked",
3740 clause, cname);
3741 }
3742 }
3743 }
3744 if let Some(rows) = crate::pgcatalog::rows(&cname, db) {
3745 // A synthesised catalogue relation is small and built eagerly;
3746 // wrapping it satisfies the streaming contract without pretending
3747 // it is lazy.
3748 return Ok(Some(crate::sqlselect::from_vec(rows)));
3749 }
3750 // A join between a catalogue relation and a real collection is
3751 // legitimate, so a user table still resolves.
3752 //
3753 // `nql::query` materialises whatever it is asked for, so what it is
3754 // ASKED for is the whole cost of this line. It used to be
3755 // `FROM <collection>` — every document, unconditionally, before a
3756 // single predicate ran. Free on a catalogue relation of a few dozen
3757 // synthesised rows; on a user collection it is the difference between
3758 // reading one document and reading all of them.
3759 //
3760 // `sqlpush::nql_prefilter` renders the part of the WHERE that NQL is
3761 // known to evaluate identically, and the full WHERE still runs above
3762 // this — so the pre-filter can only ever cost a wasted row, never an
3763 // answer. See the module note in `sqlpush` for why each refused
3764 // construct is refused.
3765 //
3766 // Still eager, and deliberately not claimed otherwise: this narrows
3767 // WHAT is materialised, not WHETHER it is. A lazy storage scan is the
3768 // other half and is tracked in HANDOFF.
3769 let key = cname.to_ascii_lowercase();
3770 let pre = pushdown_prefilters.get(&key);
3771 // Composed in NQL'S OWN CLAUSE ORDER, which its grammar fixes as
3772 //
3773 // FROM coll [AS OF seq] [VALID AS OF "date"] [WHERE p] [SEARCH "t"]
3774 //
3775 // and which is not negotiable: emit `AS OF` after `WHERE` and the NQL
3776 // parser reads it as part of the predicate expression. This is the
3777 // whole mechanism behind "NQL folded into neSQL" — the SQL side parses
3778 // the verbs and composes joins and subqueries around them, while the
3779 // NQL engine remains the one implementation that executes them.
3780 // Built once, parameterised by whether the pre-filter is included, so
3781 // the retry below cannot diverge from the real query by forgetting a
3782 // clause.
3783 //
3784 // It previously did. The retry was hand-rolled as
3785 // FROM <coll> [AS OF <seq>]
3786 // on the stated grounds that "the fallback drops the PRE-FILTER, which
3787 // is free". Dropping the pre-filter IS free -- the full WHERE runs
3788 // above. But that string also dropped VALID AS OF and SEARCH, which
3789 // are not free and have no equivalent up there: the retry answered
3790 // with rows nobody asked about and looked like it worked. The AS OF
3791 // case had already been found and special-cased; the other two were
3792 // the same bug standing next to it.
3793 let Some(db) = db else { return Ok(None) };
3794
3795 // The scan as DATA. No string is built and none is parsed: the
3796 // qualifiers go to the store as fields.
3797 //
3798 // This replaced `crate::nql::query(db, &compose(true))`, which
3799 // rendered `FROM coll AS OF n VALID AS OF '...' WHERE ... SEARCH '...'`
3800 // into text and handed it back to the NQL parser. That was a
3801 // translation living inside the thing built to stop translating, and
3802 // it failed the same way translations do: the retry path composed its
3803 // own shorter string and dropped two clauses, and `SEARCH 'o''brien'`
3804 // was a quoting question rather than a value.
3805 let verbs = nql_verbs.get(&key);
3806 let scan = crate::relation::Scan {
3807 coll: cname.to_string(),
3808 as_of: temporal.get(&key).copied(),
3809 valid_as_of: verbs.and_then(|v| v.valid_as_of.clone()),
3810 search: verbs.and_then(|v| v.search.clone()),
3811 trace: verbs.and_then(|v| v.trace.clone()),
3812 trace_reverse: verbs.map(|v| v.trace_reverse).unwrap_or(false),
3813 traverse: verbs.and_then(|v| v.traverse.clone()),
3814 trace_limit: crate::relation::DEFAULT_TRACE_LIMIT,
3815 };
3816
3817 // The pre-filter is the one part still expressed in NQL, because it is
3818 // the one part that is an OPTIMISATION rather than a meaning: the full
3819 // `WHERE` runs in the evaluator above regardless, so a pre-filter can
3820 // only ever save a row, never change an answer. When NQL declines it,
3821 // the scan simply happens unfiltered — which is what the query would
3822 // have done anyway, and no clause is lost with it because the scan is
3823 // a struct and the struct does not change.
3824 if let Some(p) = pre {
3825 let filtered = compose_prefiltered(&cname, &scan, p);
3826 if let Ok((rows, _)) = crate::nql::query(db, &filtered) {
3827 return Ok(Some(crate::sqlselect::from_vec(rows)));
3828 }
3829 }
3830 // A collection that does not exist is NOT an empty one.
3831 //
3832 // `nql::query` used to error on an unknown collection, and the `Err`
3833 // arm returned `Ok(None)` — which the evaluator reports as
3834 // `relation "x" does not exist`. Reading the store directly lost that
3835 // for free, because `relation::read` on a name nothing was ever
3836 // written under returns an empty Vec, indistinguishable from a
3837 // collection that exists and is empty.
3838 //
3839 // The cost of getting this wrong is a typo answering successfully:
3840 // `SELECT * FROM orders JOIN x ON true` returned `[]` rather than
3841 // naming `x`, and an empty join result looks exactly like a correct
3842 // answer about data that isn't there.
3843 //
3844 // `list_ids_including_deleted` rather than `collections`, so a
3845 // collection whose rows have all been deleted still EXISTS. Its
3846 // tombstones are the evidence it did.
3847 // A CATALOGUE relation is exempt, and the distinction is deliberate.
3848 // `pg_db_role_setting` and friends are things NEDB has nothing for;
3849 // the documented behaviour is that they are EMPTY rather than an
3850 // error, because a client introspecting the catalogue is asking "is
3851 // there anything here" and "no" is a valid answer. `psql \drds` walks
3852 // exactly such a relation, and my first version of this check broke
3853 // it. A user collection is the opposite case: nobody types a
3854 // collection name hoping it does not exist.
3855 // Membership is by SCHEMA, not by a list of names we happen to
3856 // implement. `is_catalog` alone was not enough: `pg_db_role_setting`
3857 // is in neither its match arm nor EMPTY_CATALOG, so `psql \drds`
3858 // started reporting `relation "pg_catalog.pg_db_role_setting" does
3859 // not exist` — a regression against the documented stance that what
3860 // NEDB has nothing for is EMPTY rather than an error. Enumerating
3861 // catalogue relations means the next introspection command psql
3862 // grows breaks the same way.
3863 let catalogue = crate::pgcatalog::is_catalog(&cname)
3864 || cname.starts_with("pg_")
3865 || cname.starts_with("information_schema.");
3866 let known = catalogue
3867 || db.collections().iter().any(|c| c == &cname)
3868 || !db.list_ids_including_deleted(&cname).is_empty();
3869
3870 // TWO CONTEXTS, TWO RIGHT ANSWERS — and they used to be distinguished
3871 // for free, because the evaluator only ever served catalogue
3872 // relations. Now that it serves user collections too, the distinction
3873 // has to be made on purpose or one of the two answers is lost.
3874 //
3875 // SINGLE RELATION -> EMPTY. NEDB is schemaless and a collection is
3876 // created by its first write, so "does not exist" and "is empty"
3877 // are the same observable state. Erroring makes it impossible to
3878 // read a collection before writing to it.
3879 //
3880 // A JOIN -> ERROR. Nobody joins against a relation they believe is
3881 // absent; there the name is a typo or a bug, and an empty join
3882 // result is indistinguishable from a correct answer about data that
3883 // is not there. `SELECT * FROM orders JOIN x ON true` returning []
3884 // is the failure this guards.
3885 //
3886 // I flattened both into "error" first, which broke `psql \drds` and
3887 // the documented schemaless read. The rule is the one the test for it
3888 // already spelled out.
3889 if !known && !sel.joins.is_empty() {
3890 return Ok(None);
3891 }
3892 Ok(Some(crate::sqlselect::from_vec(crate::relation::read_json(db, &scan))))
3893 };
3894
3895 let (cols, rows, plan) = crate::sqlselect::execute_explain(
3896 &sel,
3897 &resolve,
3898 crate::sqljoin::JoinExec::Auto,
3899 )
3900 .map_err(|e| err_msg("42601", &e.to_string()))?;
3901
3902 Ok(Some((
3903 Executed {
3904 rows,
3905 // The KEY is what the row is stored under; the NAME is what the
3906 // client sees. They differ when a select list has duplicate output
3907 // names, which PostgreSQL permits and generated SQL relies on.
3908 project: cols
3909 .iter()
3910 .map(|c| Col::renamed(&c.key, &c.name))
3911 .collect(),
3912 has_rows: true,
3913 tag: "SELECT".into(),
3914 tag_counts_rows: true,
3915 },
3916 plan,
3917 )))
3918}
3919
3920/// Strip a leading `EXPLAIN`, returning the statement it wraps.
3921///
3922/// `ANALYZE` and `VERBOSE` are accepted and ignored: this endpoint always
3923/// executes and always reports actual rows, so `EXPLAIN` and
3924/// `EXPLAIN ANALYZE` genuinely do the same thing here. Accepting the keyword
3925/// and silently doing the honest thing beats refusing a client's spelling.
3926fn strip_explain(sql: &str) -> Option<&str> {
3927 let t = sql.trim().trim_end_matches(';').trim();
3928 let mut rest = t.strip_prefix("EXPLAIN").or_else(|| t.strip_prefix("explain"))?;
3929 // Require a word boundary so `EXPLAINED` is not mistaken for a keyword.
3930 if !rest.starts_with(char::is_whitespace) {
3931 return None;
3932 }
3933 rest = rest.trim_start();
3934 loop {
3935 let low = rest.to_lowercase();
3936 if let Some(r) = low.strip_prefix("analyze").or_else(|| low.strip_prefix("analyse")) {
3937 if r.starts_with(char::is_whitespace) || r.is_empty() {
3938 rest = rest[rest.len() - r.len()..].trim_start();
3939 continue;
3940 }
3941 }
3942 if let Some(r) = low.strip_prefix("verbose") {
3943 if r.starts_with(char::is_whitespace) || r.is_empty() {
3944 rest = rest[rest.len() - r.len()..].trim_start();
3945 continue;
3946 }
3947 }
3948 break;
3949 }
3950 Some(rest)
3951}
3952
3953/// One text column named `QUERY PLAN`, which is exactly the shape PostgreSQL
3954/// returns — so `psql` prints it without special handling.
3955fn plan_result(lines: Vec<String>) -> Executed {
3956 Executed {
3957 rows: lines
3958 .into_iter()
3959 .map(|l| serde_json::json!({ "QUERY PLAN": l }))
3960 .collect(),
3961 project: vec![Col::same("QUERY PLAN")],
3962 has_rows: true,
3963 tag: "EXPLAIN".into(),
3964 tag_counts_rows: false,
3965 }
3966}
3967
3968/// Does the raw SQL plainly read a catalogue relation?
3969///
3970/// A cheap text check, used only to decide WHICH error to report when the
3971/// statement cannot be parsed — never to decide what a parsable statement
3972/// means. `pg_` is the giveaway: every catalogue relation is prefixed, and so
3973/// is the `pg_catalog` schema qualifier.
3974fn mentions_catalog(sql: &str) -> bool {
3975 let low = sql.to_lowercase();
3976 low.contains("pg_catalog.")
3977 || low.contains("information_schema.")
3978 || low.contains("from pg_")
3979 || low.contains("join pg_")
3980}
3981
3982/// The catalogue relation a translated query reads from, if any.
3983///
3984/// Reads the collection straight off the parsed NQL rather than re-parsing the
3985/// SQL, so it cannot disagree with what the executor is about to run.
3986fn catalog_target(nql: &str) -> Option<String> {
3987 let coll = crate::nql::parse(nql).ok()?.coll;
3988 if crate::pgcatalog::is_catalog(&coll) {
3989 Some(coll)
3990 } else {
3991 None
3992 }
3993}
3994
3995/// True when the statement carried a RETURNING clause. Checked against the raw
3996/// SQL because `RETURNING *` yields an EMPTY projection, which is otherwise
3997/// indistinguishable from "no RETURNING at all".
3998fn wants_returning(sql: &str) -> bool {
3999 find_kw(&sql.to_uppercase(), "RETURNING").is_some()
4000}
4001
4002/// A unique key for a server-assigned INSERT id.
4003fn next_row_id() -> String {
4004 use std::sync::atomic::{AtomicU64, Ordering};
4005 static N: AtomicU64 = AtomicU64::new(0);
4006 let n = N.fetch_add(1, Ordering::Relaxed);
4007 let ts = std::time::SystemTime::now()
4008 .duration_since(std::time::UNIX_EPOCH)
4009 .map(|d| d.as_micros())
4010 .unwrap_or(0);
4011 format!("r{}{}", ts, n)
4012}
4013
4014/// One executed statement, held apart from any wire encoding.
4015///
4016/// This type is why the simple and extended protocols share an execution path
4017/// rather than growing two copies of the SQL→NEDB semantics. The simple path
4018/// encodes it immediately; the extended path parks it in a portal and dribbles
4019/// the rows out across successive `Execute` messages. Both get identical
4020/// answers because both call `execute_stmt`.
4021pub struct Executed {
4022 /// The rows the client gets — a SELECT's result, or a write's `RETURNING`.
4023 pub rows: Vec<Value>,
4024 /// How to project them (empty = every key in the row).
4025 pub project: Vec<Col>,
4026 /// Whether the client asked for rows at all. Distinct from `rows.is_empty()`:
4027 /// a `SELECT` matching nothing still owes a `RowDescription`, while an
4028 /// `UPDATE` without `RETURNING` owes `NoData`.
4029 pub has_rows: bool,
4030 /// The command tag, already rendered — except for a SELECT, where the row
4031 /// count is only known once the rows have actually been sent.
4032 pub tag: String,
4033 /// True when `tag` is a SELECT-shaped tag whose count is the rows sent.
4034 pub tag_counts_rows: bool,
4035}
4036
4037impl Executed {
4038 fn nothing(tag: &str) -> Self {
4039 Executed { rows: vec![], project: vec![], has_rows: false, tag: tag.to_string(), tag_counts_rows: false }
4040 }
4041 /// Render the final `CommandComplete` given how many rows went out.
4042 fn tag_for(&self, sent: usize) -> String {
4043 if self.tag_counts_rows { format!("{} {}", self.tag, sent) } else { self.tag.clone() }
4044 }
4045}
4046
4047/// Run ONE statement. `Err` carries an already-encoded `ErrorResponse`.
4048///
4049/// Every SQL→NEDB decision lives here, which is the point: the extended query
4050/// protocol added below is then purely a matter of message framing, and cannot
4051/// drift from the simple path's semantics.
4052/// Run one neSQL statement against a database, in process.
4053///
4054/// # Why this exists
4055///
4056/// Until this, the engine had exactly one SQL execution path and it was welded
4057/// to the wire protocol: `execute_stmt` is private, takes the connection's
4058/// read-only flag, and reports failure as ALREADY-ENCODED Postgres error bytes.
4059/// Nothing outside a pgwire session could run SQL against a `Db`.
4060///
4061/// That was survivable while the only SQL client was a socket. It stopped being
4062/// survivable when neSQL — which owns the language — needed to run the language
4063/// from a CLI, because the alternatives were a CLI that opens a TCP connection
4064/// to its own process, or a second SQL front end living in the CLI. The second
4065/// one is worse than it sounds: it makes the CLI a quieter second authority on
4066/// what the language accepts, and the first divergence between them would be
4067/// discovered by a user, not by us.
4068///
4069/// So the path the wire already takes is exposed, with the error decoded into
4070/// text. Same parser, same translator, same evaluator, same decision about
4071/// which engine runs a statement — one authority.
4072/// The rows an `UPDATE` or `DELETE` will act on — chosen by the SQL evaluator.
4073///
4074/// This used to render the predicate as NQL and run `nql::query`, which meant
4075/// a write could only match what the NQL parser understood, even though the
4076/// statement arrived as SQL and the read path had long since stopped needing a
4077/// translation. `UPDATE … WHERE _id IN (SELECT …)` was unreachable for exactly
4078/// that reason: the subquery translated into NQL text the NQL parser cannot
4079/// parse. Selecting with a real `SELECT` closes that gap by not having a second
4080/// predicate implementation to fall short of the first.
4081///
4082/// Whole rows, not just `_id`: `DELETE … RETURNING` has to capture the row
4083/// BEFORE the tombstone, so the selection is what it returns.
4084///
4085/// # The unknown-collection guard is not incidental
4086///
4087/// `nql::query` ERRORS on a collection that does not exist; the evaluator's
4088/// scan returns no rows, because a schemaless read of an absent collection is
4089/// legitimately empty. Swapping one for the other without this check would
4090/// turn `UPDATE nowhere SET x = 1` from a loud 42P01 into a silent
4091/// `UPDATE 0` — a write that reports success having done nothing, which is the
4092/// worst available outcome and the reason this function refuses first.
4093fn rows_for_write(db: &Arc<Db>, coll: &str, where_sql: &str, nql: &str)
4094 -> std::result::Result<Vec<Value>, Vec<u8>>
4095{
4096 let known = db.collections().iter().any(|c| c == coll)
4097 || !db.list_ids_including_deleted(coll).is_empty();
4098 if !known {
4099 return Err(err_msg("42P01", &format!("relation \"{}\" does not exist", coll)));
4100 }
4101 let sel = format!("SELECT * FROM {} {}", coll, where_sql).trim().to_string();
4102 // read_only: this is the SELECT half of the write, and nothing it does
4103 // should be able to write. The caller already passed `need_write!()`.
4104 execute_sql(db, &sel, true)
4105 .map(|done| done.rows)
4106 .map_err(|e| err_msg("42601", &format!(
4107 "{} (selecting rows with: {}; the NQL rendering of this predicate \
4108 would have been: {})", e, sel, nql)))
4109}
4110
4111pub fn execute_sql(db: &Arc<Db>, sql: &str, read_only: bool)
4112 -> std::result::Result<Executed, String>
4113{
4114 execute_stmt(sql, "", Some(db), read_only).map_err(|wire| decode_wire_error(&wire))
4115}
4116
4117/// Pull the human-readable message out of an encoded ErrorResponse.
4118///
4119/// The wire format is a sequence of NUL-terminated `field-code || text` runs
4120/// terminated by an empty field. `M` is the primary message and `C` the
4121/// SQLSTATE; both are reported, because a caller who loses the SQLSTATE loses
4122/// the only machine-stable part of the error.
4123fn decode_wire_error(buf: &[u8]) -> String {
4124 let mut code: Option<String> = None;
4125 let mut msg: Option<String> = None;
4126 // Skip the 1-byte tag and 4-byte length when they are present.
4127 let body = if buf.len() > 5 { &buf[5..] } else { buf };
4128 let mut i = 0usize;
4129 while i < body.len() && body[i] != 0 {
4130 let field = body[i];
4131 i += 1;
4132 let start = i;
4133 while i < body.len() && body[i] != 0 { i += 1; }
4134 let text = String::from_utf8_lossy(&body[start..i]).into_owned();
4135 i += 1; // the NUL
4136 match field {
4137 b'C' => code = Some(text),
4138 b'M' => msg = Some(text),
4139 _ => {}
4140 }
4141 }
4142 match (code, msg) {
4143 (Some(c), Some(m)) => format!("{} ({})", m, c),
4144 (None, Some(m)) => m,
4145 // Never silently produce an empty error. A failure we cannot read is
4146 // still a failure, and saying so beats returning "".
4147 _ => format!(
4148 "the engine refused the statement and the error could not be decoded ({} bytes of wire response)", buf.len()
4149 ),
4150 }
4151}
4152
4153fn execute_stmt(
4154 stmt_sql: &str,
4155 db_name: &str,
4156 db: Option<&Arc<Db>>,
4157 read_only: bool,
4158) -> Result<Executed, Vec<u8>> {
4159 // The full SQL engine gets first refusal, but ONLY for statements that
4160 // touch the catalogue — see `try_catalog_select`. It has to run before
4161 // `translate`, because `translate` targets NQL and NQL cannot express a
4162 // join, a CASE or a scalar function at all.
4163 // EXPLAIN reports which engine would run the statement, and a plan only
4164 // when the SQL evaluator is the engine that actually runs it. Describing a
4165 // pipeline the statement would not take is the one thing an EXPLAIN must
4166 // never do.
4167 if let Some(inner) = strip_explain(stmt_sql) {
4168 if let Some((_, plan)) = try_catalog_select(inner, db)? {
4169 return Ok(plan_result(plan.render()));
4170 }
4171 let mut lines = vec![];
4172 match translate(inner) {
4173 Ok(_) => {
4174 lines.push(
4175 "NQL path — this statement is translated to NQL and \
4176 executed by the storage engine, not by the SQL evaluator."
4177 .to_string(),
4178 );
4179 lines.push(
4180 "No plan is reported, because the SQL evaluator is not \
4181 what runs it. Reporting one would describe a pipeline \
4182 that never executed."
4183 .to_string(),
4184 );
4185 lines.push(
4186 "The SQL evaluator (joins, CASE, scalar functions, a \
4187 hash-join planner) currently serves catalogue queries."
4188 .to_string(),
4189 );
4190 }
4191 Err(why) => lines.push(format!("cannot be executed: {why}")),
4192 }
4193 return Ok(plan_result(lines));
4194 }
4195
4196 if let Some((done, _plan)) = try_catalog_select(stmt_sql, db)? {
4197 return Ok(done);
4198 }
4199
4200 let stmt = translate(stmt_sql).map_err(|why| err_msg("0A000", &why))?;
4201
4202 // Every arm below that touches storage needs a database; resolve the
4203 // "no such database" answer once instead of at each use.
4204 macro_rules! need_db {
4205 () => {
4206 match db {
4207 Some(db) => db,
4208 None => return Err(no_db(db_name)),
4209 }
4210 };
4211 }
4212 macro_rules! need_write {
4213 () => {
4214 if read_only {
4215 return Err(err_msg("25006", READ_ONLY_MSG));
4216 }
4217 };
4218 }
4219
4220 match stmt {
4221 Stmt::Ok(tag) => Ok(Executed::nothing(if tag.is_empty() { "SELECT 0" } else { tag })),
4222
4223 Stmt::Canned { cols, row } => {
4224 // Fold the canned answer into an ordinary row so the encoders,
4225 // the portal machinery and `Describe` all see one shape.
4226 let mut obj = serde_json::Map::new();
4227 for (c, v) in cols.iter().zip(row.iter()) {
4228 obj.insert(c.clone(), Value::String(v.clone()));
4229 }
4230 Ok(Executed {
4231 rows: vec![Value::Object(obj)],
4232 project: cols.iter().map(|c| Col::same(c)).collect(),
4233 has_rows: true,
4234 tag: "SELECT".into(),
4235 tag_counts_rows: true,
4236 })
4237 }
4238
4239 Stmt::Query { nql, project } => {
4240 // A catalogue relation is synthesised from the live database
4241 // rather than read from it — but it is still queried with the
4242 // ORDINARY predicate path, so WHERE / ORDER BY / LIMIT and the
4243 // `~` operators work on it because they are the same operators.
4244 //
4245 // Checked BEFORE `need_db!()`: `SELECT * FROM pg_namespace` has to
4246 // answer even when the client connected without naming a database,
4247 // which is exactly what psql does on startup. Refusing there is
4248 // how "psql cannot connect" starts.
4249 if let Some(coll) = catalog_target(&nql) {
4250 let rows = crate::pgcatalog::rows(&coll, db)
4251 .expect("catalog_target only returns names pgcatalog serves");
4252 let rows = crate::nql::query_rows(rows, &nql)
4253 .map_err(|e| err_msg("42601", &e.to_string()))?;
4254 return Ok(Executed {
4255 rows, project, has_rows: true,
4256 tag: "SELECT".into(), tag_counts_rows: true,
4257 });
4258 }
4259 let db = need_db!();
4260 let (rows, _) = crate::nql::query(db, &nql).map_err(|e| {
4261 err_msg("42601", &format!("{} (translated to NQL: {})", e, nql))
4262 })?;
4263 Ok(Executed { rows, project, has_rows: true, tag: "SELECT".into(), tag_counts_rows: true })
4264 }
4265
4266 Stmt::Insert { coll, rows, returning } => {
4267 let db = need_db!();
4268 need_write!();
4269 let mut written: Vec<Value> = vec![];
4270 for (i, r) in rows.iter().enumerate() {
4271 // The engine requires an id. When the statement did not supply
4272 // one, mint a unique key rather than silently overwriting a
4273 // shared default.
4274 let id = match &r.id {
4275 Some(id) => id.clone(),
4276 None => format!("{}-{}", next_row_id(), i),
4277 };
4278 let node = db
4279 .put(&coll, &id, Value::Object(r.doc.clone()),
4280 r.caused_by.clone(), r.valid_from.clone(), r.valid_to.clone())
4281 .map_err(|e| err_msg("XX000", &format!("INSERT failed: {}", e)))?;
4282 written.push(crate::nql::node_to_json(&node));
4283 }
4284 let n = written.len();
4285 let has_rows = wants_returning(stmt_sql);
4286 Ok(Executed {
4287 rows: if has_rows { written } else { vec![] },
4288 project: returning,
4289 has_rows,
4290 // Postgres reports `INSERT <oid> <rows>`; the oid is always 0.
4291 tag: format!("INSERT 0 {}", n),
4292 tag_counts_rows: false,
4293 })
4294 }
4295
4296 Stmt::Update { coll, set, where_sql, nql, returning } => {
4297 let db = need_db!();
4298 need_write!();
4299 // Rows come from the SQL evaluator, so an UPDATE matches exactly
4300 // what a SELECT with the same WHERE matches — one predicate
4301 // implementation, not two.
4302 let matched = rows_for_write(db, &coll, &where_sql, &nql)?;
4303 let mut written: Vec<Value> = vec![];
4304 for row in &matched {
4305 let id = match row.get("_id").and_then(|v| v.as_str()) {
4306 Some(id) => id.to_string(),
4307 None => continue,
4308 };
4309 // Merge onto the CURRENT stored document, not onto the query
4310 // row: a query row carries injected `_`-prefixed metadata that
4311 // must never be written back into the payload.
4312 let mut doc = match db.get(&coll, &id) {
4313 Some(n) => match n.data {
4314 Value::Object(m) => m,
4315 _ => serde_json::Map::new(),
4316 },
4317 None => continue,
4318 };
4319 for (k, v) in &set {
4320 doc.insert(k.clone(), v.clone());
4321 }
4322 // An UPDATE is a NEW VERSION — the prior value stays readable
4323 // with AS OF SYSTEM TIME. That is the whole point.
4324 let node = db
4325 .put(&coll, &id, Value::Object(doc), vec![], None, None)
4326 .map_err(|e| err_msg("XX000", &format!("UPDATE failed: {}", e)))?;
4327 written.push(crate::nql::node_to_json(&node));
4328 }
4329 let n = written.len();
4330 let has_rows = wants_returning(stmt_sql);
4331 Ok(Executed {
4332 rows: if has_rows { written } else { vec![] },
4333 project: returning,
4334 has_rows,
4335 tag: format!("UPDATE {}", n),
4336 tag_counts_rows: false,
4337 })
4338 }
4339
4340 Stmt::Delete { coll, where_sql, nql, returning } => {
4341 let db = need_db!();
4342 need_write!();
4343 let matched = rows_for_write(db, &coll, &where_sql, &nql)?;
4344 // RETURNING must be captured BEFORE the delete: after the tombstone
4345 // the row is no longer readable by id.
4346 let returned = matched.clone();
4347 let mut n = 0usize;
4348 for row in &matched {
4349 if let Some(id) = row.get("_id").and_then(|v| v.as_str()) {
4350 match db.delete(&coll, id) {
4351 Ok(true) => n += 1,
4352 Ok(false) => {}
4353 Err(e) => return Err(err_msg("XX000", &format!("DELETE failed: {}", e))),
4354 }
4355 }
4356 }
4357 let has_rows = wants_returning(stmt_sql);
4358 Ok(Executed {
4359 rows: if has_rows { returned } else { vec![] },
4360 project: returning,
4361 has_rows,
4362 tag: format!("DELETE {}", n),
4363 tag_counts_rows: false,
4364 })
4365 }
4366 }
4367}
4368
4369/// Execute a simple-query payload, which may hold several `;`-separated statements.
4370fn run_simple_query(sql: &str, db_name: &str, db: Option<&Arc<Db>>, read_only: bool) -> Vec<u8> {
4371 let mut out = vec![];
4372 let statements = split_statements(sql);
4373 if statements.is_empty() {
4374 // EmptyQueryResponse
4375 return Out::msg(b'I').finish();
4376 }
4377 for stmt_sql in statements {
4378 match execute_stmt(&stmt_sql, db_name, db, read_only) {
4379 // Abandon the rest of the batch on the first error, as Postgres does.
4380 Err(encoded) => {
4381 out.extend_from_slice(&encoded);
4382 return out;
4383 }
4384 Ok(ex) => {
4385 if ex.has_rows {
4386 out.extend_from_slice(&encode_rows(&ex.rows, &ex.project));
4387 }
4388 out.extend_from_slice(&command_complete(&ex.tag_for(ex.rows.len())));
4389 }
4390 }
4391 }
4392 out
4393}
4394
4395/// Split on `;` at the top level, ignoring separators inside string literals.
4396fn split_statements(sql: &str) -> Vec<String> {
4397 let mut out = vec![];
4398 let mut cur = String::new();
4399 let mut in_s = false;
4400 for c in sql.chars() {
4401 match c {
4402 '\'' => { in_s = !in_s; cur.push(c); }
4403 ';' if !in_s => {
4404 if !cur.trim().is_empty() { out.push(cur.clone()); }
4405 cur.clear();
4406 }
4407 _ => cur.push(c),
4408 }
4409 }
4410 if !cur.trim().is_empty() {
4411 out.push(cur);
4412 }
4413 out
4414}
4415
4416/// Bind and serve the Postgres read endpoint until the process exits.
4417pub async fn run(host: &str, port: u16, resolver: Arc<dyn DbResolver>) -> anyhow::Result<()> {
4418 // Writes are ON by default — that is the parity position. An operator who
4419 // wants the "system of proof beside your database" deployment, where this
4420 // door must never mutate anything, sets NEDBD_PG_READ_ONLY=1.
4421 let read_only = std::env::var("NEDBD_PG_READ_ONLY")
4422 .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
4423 .unwrap_or(false);
4424 let listener = TcpListener::bind((host, port)).await?;
4425 println!(" pgwire postgres endpoint on {}:{} — psql / DBeaver / psycopg ({})",
4426 host, port,
4427 if read_only { "SELECT only — read-only mode" } else { "SELECT + INSERT/UPDATE/DELETE" });
4428 loop {
4429 let (sock, _peer) = match listener.accept().await {
4430 Ok(v) => v,
4431 Err(e) => {
4432 eprintln!(" [pgwire] accept failed: {}", e);
4433 continue;
4434 }
4435 };
4436 let r = Arc::clone(&resolver);
4437 tokio::spawn(async move {
4438 let _ = sock.set_nodelay(true);
4439 if let Err(e) = handle(sock, r, read_only).await {
4440 // A client disconnecting mid-message is routine, not an incident.
4441 if e.kind() != std::io::ErrorKind::UnexpectedEof
4442 && e.kind() != std::io::ErrorKind::ConnectionReset
4443 {
4444 eprintln!(" [pgwire] connection error: {}", e);
4445 }
4446 }
4447 });
4448 }
4449}
4450
4451// ─────────────────────────────────────────────────────────────────────────────
4452
4453#[cfg(test)]
4454mod explain_tests {
4455 use super::*;
4456
4457 #[test]
4458 fn a_bare_explain_is_stripped() {
4459 assert_eq!(strip_explain("EXPLAIN SELECT 1"), Some("SELECT 1"));
4460 assert_eq!(strip_explain("explain select 1"), Some("select 1"));
4461 assert_eq!(strip_explain(" EXPLAIN SELECT 1 ; "), Some("SELECT 1"));
4462 }
4463
4464 #[test]
4465 fn analyze_and_verbose_are_accepted_and_ignored() {
4466 // This endpoint always executes and always reports actual rows, so
4467 // EXPLAIN and EXPLAIN ANALYZE genuinely do the same thing. Accepting
4468 // the client's spelling beats refusing it.
4469 assert_eq!(strip_explain("EXPLAIN ANALYZE SELECT 1"), Some("SELECT 1"));
4470 assert_eq!(strip_explain("EXPLAIN ANALYSE SELECT 1"), Some("SELECT 1"));
4471 assert_eq!(strip_explain("EXPLAIN VERBOSE SELECT 1"), Some("SELECT 1"));
4472 assert_eq!(strip_explain("EXPLAIN ANALYZE VERBOSE SELECT 1"), Some("SELECT 1"));
4473 assert_eq!(strip_explain("explain analyze verbose select 1"), Some("select 1"));
4474 }
4475
4476 #[test]
4477 fn a_word_merely_starting_with_explain_is_not_a_keyword() {
4478 assert_eq!(strip_explain("EXPLAINED SELECT 1"), None);
4479 assert_eq!(strip_explain("SELECT 1"), None);
4480 assert_eq!(strip_explain("SELECT explain FROM t"), None);
4481 }
4482
4483 #[test]
4484 fn a_column_named_analyze_is_not_eaten() {
4485 // `analyzed` merely starts with the keyword; the word boundary check
4486 // is what stops it being consumed as an option.
4487 assert_eq!(strip_explain("EXPLAIN analyzed_view"), Some("analyzed_view"));
4488 }
4489
4490 #[test]
4491 fn the_plan_result_has_postgres_shape() {
4492 let e = plan_result(vec!["Seq Scan on t".into(), "note".into()]);
4493 assert_eq!(e.project.len(), 1);
4494 assert_eq!(e.project[0].out, "QUERY PLAN");
4495 assert_eq!(e.rows.len(), 2);
4496 assert_eq!(e.rows[0]["QUERY PLAN"], "Seq Scan on t");
4497 assert_eq!(e.tag, "EXPLAIN");
4498 // EXPLAIN's tag carries no row count in PostgreSQL.
4499 assert!(!e.tag_counts_rows);
4500 }
4501}
4502
4503#[cfg(test)]
4504mod tests {
4505 use super::*;
4506 use serde_json::json;
4507
4508 fn q(sql: &str) -> String {
4509 match translate(sql) {
4510 Ok(Stmt::Query { nql, .. }) => nql,
4511 other => panic!("expected a query for {:?}, got {:?}", sql, other),
4512 }
4513 }
4514 /// Output column names, in order.
4515 fn proj(sql: &str) -> Vec<String> {
4516 match translate(sql) {
4517 Ok(Stmt::Query { project, .. }) => project.iter().map(|c| c.out.clone()).collect(),
4518 other => panic!("expected a query for {:?}, got {:?}", sql, other),
4519 }
4520 }
4521 /// (source key, output name) pairs, for the aggregate renaming.
4522 fn proj_pairs(sql: &str) -> Vec<(String, String)> {
4523 match translate(sql) {
4524 Ok(Stmt::Query { project, .. }) =>
4525 project.iter().map(|c| (c.src.clone(), c.out.clone())).collect(),
4526 other => panic!("expected a query for {:?}, got {:?}", sql, other),
4527 }
4528 }
4529 fn names(cols: &[Col]) -> Vec<String> { cols.iter().map(|c| c.out.clone()).collect() }
4530
4531 /// The full projection, so a test can assert the SRC and the OUT
4532 /// separately — they are different jobs and conflating them is how an
4533 /// alias got lost.
4534 fn cols_of(sql: &str) -> Vec<Col> {
4535 match translate(sql).unwrap() {
4536 Stmt::Query { project, .. } => project,
4537 other => panic!("{:?}", other),
4538 }
4539 }
4540
4541 #[test]
4542 fn select_star_becomes_bare_from() {
4543 assert_eq!(q("SELECT * FROM orders"), "FROM orders");
4544 assert_eq!(q("select * from orders;"), "FROM orders");
4545 assert_eq!(proj("SELECT * FROM orders"), Vec::<String>::new());
4546 }
4547
4548 #[test]
4549 fn a_column_list_becomes_a_projection_not_a_clause() {
4550 // NQL has no projection, so the column list is carried separately and
4551 // applied to the returned rows.
4552 assert_eq!(q("SELECT status, total FROM orders"), "FROM orders");
4553 assert_eq!(proj("SELECT status, total FROM orders"), vec!["status", "total"]);
4554 }
4555
4556 #[test]
4557 fn a_qualifier_reduces_to_the_field_while_an_ALIAS_is_the_name_the_client_sees() {
4558 // Two different jobs, and they used to be conflated. The SRC is what
4559 // NEDB reads out of the row, so a qualifier must be stripped from it.
4560 // The OUT is the name the CLIENT looks the column up by, so an alias
4561 // must be KEPT in it — `SELECT status AS s` returns a column called
4562 // `s`, and answering with one called `status` hands a client a result
4563 // it cannot find. SQLAlchemy writes `count(*) AS count_1` and then
4564 // reads `count_1`.
4565 let cols = cols_of("SELECT o.status AS s, o.total total, o.region FROM orders o");
4566 assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
4567 vec!["status", "total", "region"]);
4568 assert_eq!(cols.iter().map(|c| c.out.clone()).collect::<Vec<_>>(),
4569 vec!["s", "total", "region"]);
4570 assert_eq!(q("SELECT * FROM public.orders"), "FROM orders");
4571 assert_eq!(q("SELECT * FROM \"orders\""), "FROM orders");
4572 }
4573
4574 #[test]
4575 fn a_select_list_may_MIX_columns_with_an_aggregate() {
4576 // What a GROUP BY query actually looks like. The previous parser
4577 // refused any list containing a parenthesis, so this whole shape was
4578 // unreachable even though NQL expresses it natively — and it is the
4579 // single most common grouped query an ORM emits.
4580 // The aggregate sits IMMEDIATELY AFTER the group key — verified
4581 // against the running engine, which refuses the other order with
4582 // "only one aggregate per query".
4583 assert_eq!(q("SELECT status, count(*) AS count_1 FROM orders GROUP BY status"),
4584 "FROM orders GROUP BY status COUNT");
4585 // SQL puts GROUP BY before ORDER BY / LIMIT; the aggregate still lands
4586 // on the key, and the rest of the tail follows.
4587 assert_eq!(q("SELECT status, count(*) FROM orders WHERE total > 1 GROUP BY status ORDER BY status LIMIT 5"),
4588 "FROM orders WHERE total > 1 GROUP BY status COUNT ORDER BY status LIMIT 5");
4589 // A bare aggregate with NO grouping still goes after the collection.
4590 assert_eq!(q("SELECT count(*) FROM orders"), "FROM orders COUNT");
4591 assert_eq!(q("SELECT sum(total) FROM orders"), "FROM orders SUM total");
4592 // More than one group key is refused by name: NQL groups by a single
4593 // field, and using only the first would aggregate over rows the query
4594 // meant to keep apart.
4595 let e = translate("SELECT status, count(*) FROM orders GROUP BY status, region").unwrap_err();
4596 assert!(e.contains("GROUP BY takes one key"), "{}", e);
4597 let cols = cols_of("SELECT status, count(*) AS count_1 FROM orders GROUP BY status");
4598 assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
4599 vec!["status", "count"]);
4600 assert_eq!(cols.iter().map(|c| c.out.clone()).collect::<Vec<_>>(),
4601 vec!["status", "count_1"]);
4602
4603 // A named aggregate rides along with `count`, because an NQL grouped
4604 // row carries both.
4605 let cols = cols_of("SELECT status, count(*), sum(total) FROM orders GROUP BY status");
4606 assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
4607 vec!["status", "count", "sum_total"]);
4608 assert_eq!(q("SELECT status, count(*), sum(total) FROM orders GROUP BY status"),
4609 "FROM orders GROUP BY status SUM total");
4610
4611 // A qualifier on the aggregate's column is stripped like any other.
4612 assert_eq!(q("SELECT o.status, sum(o.total) FROM orders o GROUP BY o.status"),
4613 "FROM orders GROUP BY status SUM total");
4614
4615 // Two NAMED aggregates cannot both be carried, and that is refused by
4616 // name rather than silently dropping one.
4617 let e = translate("SELECT status, sum(total), avg(total) FROM orders GROUP BY status")
4618 .unwrap_err();
4619 assert!(e.contains("only one of SUM/AVG/MIN/MAX"), "{}", e);
4620
4621 // A column that is neither a key nor an aggregate is still refused.
4622 let e = translate("SELECT status, total, count(*) FROM orders GROUP BY status")
4623 .unwrap_err();
4624 assert!(e.contains("must appear in the GROUP BY clause"), "{}", e);
4625 }
4626
4627 #[test]
4628 fn ORDER_BY_an_ordinal_resolves_to_that_select_list_column() {
4629 // SQL lets a sort key be a POSITION, and clients write it constantly.
4630 // NQL has no ordinals — it read the `1` as a literal and refused with
4631 // "expected field name, got Num(1.0)". node-postgres sent
4632 // `GROUP BY status ORDER BY 1` in the harness's first run.
4633 assert_eq!(q("SELECT status, total FROM orders ORDER BY 1"),
4634 "FROM orders ORDER BY status");
4635 assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 DESC"),
4636 "FROM orders ORDER BY total DESC");
4637 // Several keys, mixing ordinals with names, and a direction on each.
4638 assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 DESC, 1"),
4639 "FROM orders ORDER BY total DESC, status");
4640 assert_eq!(q("SELECT status, total FROM orders ORDER BY 1, total DESC"),
4641 "FROM orders ORDER BY status, total DESC");
4642 // An ordinal survives the GROUP BY splice, and resolves to the group
4643 // key rather than to the literal 1 — which is the exact shape that
4644 // failed in CI.
4645 assert_eq!(q("SELECT status, count(*) AS n FROM orders GROUP BY status ORDER BY 1"),
4646 "FROM orders GROUP BY status COUNT ORDER BY status");
4647 // An ordinal may name the AGGREGATE column too.
4648 assert_eq!(q("SELECT status, count(*) AS n FROM orders GROUP BY status ORDER BY 2 DESC"),
4649 "FROM orders GROUP BY status COUNT ORDER BY count DESC");
4650 // The clause boundary is respected: a following LIMIT is not swallowed
4651 // into the sort list, and `LIMIT 1` is not mistaken for an ordinal.
4652 assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 LIMIT 1"),
4653 "FROM orders ORDER BY total LIMIT 1");
4654 // A `1` anywhere else stays a literal.
4655 assert_eq!(q("SELECT status FROM orders WHERE total > 1 ORDER BY 1"),
4656 "FROM orders WHERE total > 1 ORDER BY status");
4657
4658 // Out of range, and `SELECT *` where there is no list to index, are
4659 // both refused with the reason — guessing a column would sort by
4660 // something the query never named.
4661 let e = translate("SELECT status FROM orders ORDER BY 4").unwrap_err();
4662 assert!(e.contains("out of range") && e.contains("1 column"), "{}", e);
4663 let e = translate("SELECT * FROM orders ORDER BY 1").unwrap_err();
4664 assert!(e.contains("no list to index"), "{}", e);
4665 }
4666
4667 #[test]
4668 fn count_of_a_subquery_flattens_only_when_the_two_counts_MUST_agree() {
4669 // `.count()` in every ORM wraps the whole query in a derived table.
4670 // Counting rows that ARE the inner query's rows is counting the inner
4671 // query, so this is an identity, not an approximation.
4672 assert_eq!(
4673 q("SELECT count(*) AS count_1 FROM (SELECT orders._id AS a, orders.status AS b \
4674 FROM orders WHERE orders.status = 'paid') AS anon_1"),
4675 // Verified against the running engine: with no GROUP BY the
4676 // aggregate may sit either side of WHERE and answers identically.
4677 r#"FROM orders COUNT WHERE status = "paid""#);
4678 // No predicate at all.
4679 assert_eq!(q("SELECT count(*) FROM (SELECT orders._id FROM orders) AS anon_1"),
4680 "FROM orders COUNT");
4681 // ORDER BY cannot change a count, so it is dropped rather than refused.
4682 assert_eq!(q("SELECT count(*) FROM (SELECT _id FROM orders ORDER BY total DESC) AS a"),
4683 "FROM orders COUNT");
4684 // The outer alias is the name the client reads the column back by.
4685 let cols = cols_of("SELECT count(*) AS count_1 FROM (SELECT _id FROM orders) AS a");
4686 assert_eq!(cols[0].src, "count");
4687 assert_eq!(cols[0].out, "count_1");
4688
4689 // Each guard is a construct that would make the two counts DIFFERENT
4690 // numbers, so each is refused rather than silently flattened.
4691 for sql in [
4692 // LIMIT / OFFSET cap the rows before they are counted
4693 "SELECT count(*) FROM (SELECT _id FROM orders LIMIT 1) AS a",
4694 "SELECT count(*) FROM (SELECT _id FROM orders OFFSET 1) AS a",
4695 // the inner rows ARE the groups
4696 "SELECT count(*) FROM (SELECT status FROM orders GROUP BY status) AS a",
4697 // an inner aggregate already reduced the rows to one
4698 "SELECT count(*) FROM (SELECT count(*) FROM orders) AS a",
4699 "SELECT count(*) FROM (SELECT sum(total) FROM orders) AS a",
4700 // the outer list would need the derived table's own columns
4701 "SELECT count(*), status FROM (SELECT status FROM orders) AS a",
4702 "SELECT status FROM (SELECT status FROM orders) AS a",
4703 // one level is the claim
4704 "SELECT count(*) FROM (SELECT x FROM (SELECT _id AS x FROM orders) AS b) AS a",
4705 ] {
4706 let e = translate(sql).unwrap_err();
4707 assert!(e.contains("subqueries in FROM"), "{} -> {}", sql, e);
4708 }
4709
4710 // DISTINCT and the set operators are caught EARLIER, by their own
4711 // rules, which scan the whole statement before the FROM list is even
4712 // read. Asserted separately so the test records which check owns each
4713 // refusal rather than implying one catch-all does.
4714 for (sql, needle) in [
4715 ("SELECT count(*) FROM (SELECT DISTINCT status FROM orders) AS a", "DISTINCT"),
4716 ("SELECT count(*) FROM (SELECT a FROM t UNION SELECT b FROM u) AS x", "UNION"),
4717 ] {
4718 let e = translate(sql).unwrap_err();
4719 assert!(e.contains(needle), "{} -> {}", sql, e);
4720 }
4721 }
4722
4723 #[test]
4724 fn a_QUALIFIED_column_in_WHERE_finds_its_field_instead_of_ZERO_ROWS() {
4725 // THE silent wrong answer. NQL looks a field up FLAT, so
4726 // `WHERE orders.status = 'paid'` asked for a field literally named
4727 // "orders.status", no document had one, and the query returned ZERO
4728 // ROWS with no error — an empty result that reads exactly like "you
4729 // have no paid orders". Every ORM qualifies its predicates, so every
4730 // filtered SQLAlchemy query answered empty and `.get(pk)` answered
4731 // None.
4732 assert_eq!(q("SELECT _id FROM orders WHERE orders.status = 'paid'"),
4733 r#"FROM orders WHERE status = "paid""#);
4734 assert_eq!(q("SELECT _id FROM orders WHERE orders.total > 50"),
4735 "FROM orders WHERE total > 50");
4736 // Every clause in the tail, not just WHERE.
4737 assert_eq!(q("SELECT _id FROM orders ORDER BY orders.total DESC LIMIT 2"),
4738 "FROM orders ORDER BY total DESC LIMIT 2");
4739 assert_eq!(q("SELECT status, count(*) FROM orders GROUP BY orders.status"),
4740 "FROM orders GROUP BY status COUNT");
4741
4742 // An alias is a legal qualifier and is accepted as one. It is also
4743 // REMOVED from the tail, because NQL has no alias syntax and reported
4744 // an "unexpected token" on it.
4745 assert_eq!(q("SELECT o.status FROM orders o WHERE o.status = 'paid'"),
4746 r#"FROM orders WHERE status = "paid""#);
4747 assert_eq!(q("SELECT o.status FROM orders AS o WHERE o.total > 1"),
4748 "FROM orders WHERE total > 1");
4749
4750 // A qualifier naming NEITHER the collection nor its alias is an
4751 // ERROR, not a strip. Stripping it would answer from the one relation
4752 // that IS present, which is a different wrong answer in the same
4753 // empty-looking clothes.
4754 let e = translate("SELECT _id FROM orders WHERE nosuch.status = 'paid'").unwrap_err();
4755 assert!(e.contains("no table or alias named \"nosuch\""), "{}", e);
4756 let e = translate("SELECT _id FROM orders o WHERE p.status = 'paid'").unwrap_err();
4757 assert!(e.contains("aliased \"o\""), "the message names the alias in scope: {}", e);
4758
4759 // A dot INSIDE a literal is data, not a qualifier.
4760 assert_eq!(q("SELECT _id FROM orders WHERE status = 'pa.id'"),
4761 r#"FROM orders WHERE status = "pa.id""#);
4762 // ...and a decimal point is not one either.
4763 assert_eq!(q("SELECT _id FROM orders WHERE total > 1.5"),
4764 "FROM orders WHERE total > 1.5");
4765
4766 // UPDATE and DELETE carry the same tail, and had the same bug.
4767 match translate("UPDATE orders o SET status = 'x' WHERE o.total > 5").unwrap() {
4768 Stmt::Update { coll, nql, .. } => {
4769 assert_eq!(coll, "orders", "the alias is not part of the collection name");
4770 assert_eq!(nql, "FROM orders WHERE total > 5");
4771 }
4772 other => panic!("{:?}", other),
4773 }
4774 match translate("DELETE FROM orders o WHERE o.status = 'paid'").unwrap() {
4775 Stmt::Delete { coll, nql, .. } => {
4776 assert_eq!(coll, "orders");
4777 assert_eq!(nql, r#"FROM orders WHERE status = "paid""#);
4778 }
4779 other => panic!("{:?}", other),
4780 }
4781
4782 // `AS OF SYSTEM TIME` also begins with AS and is NOT an alias.
4783 assert_eq!(q("SELECT _id FROM orders AS OF SYSTEM TIME 3 WHERE orders.total > 1"),
4784 "FROM orders AS OF 3 WHERE total > 1");
4785 }
4786
4787 #[test]
4788 fn where_clauses_pass_through_with_sql_literals_rewritten() {
4789 assert_eq!(q("SELECT * FROM orders WHERE status = 'paid'"),
4790 r#"FROM orders WHERE status = "paid""#);
4791 assert_eq!(q("SELECT * FROM orders WHERE status <> 'paid'"),
4792 r#"FROM orders WHERE status != "paid""#);
4793 assert_eq!(q("SELECT * FROM orders WHERE status IN ('paid','open')"),
4794 r#"FROM orders WHERE status IN ("paid","open")"#);
4795 }
4796
4797 /// SQL escapes an embedded quote by doubling it. That must become ONE
4798 /// character inside the NQL string, not terminate it.
4799 #[test]
4800 fn a_doubled_sql_quote_is_one_literal_character() {
4801 assert_eq!(q("SELECT * FROM t WHERE name = 'it''s'"),
4802 r#"FROM t WHERE name = "it's""#);
4803 }
4804
4805 /// A double quote inside a SQL literal has to be escaped for NQL, whose
4806 /// lexer collapses \" — otherwise it would close the string early.
4807 #[test]
4808 fn a_double_quote_inside_a_sql_literal_is_escaped_for_nql() {
4809 assert_eq!(q(r#"SELECT * FROM t WHERE name = 'say "hi"'"#),
4810 r#"FROM t WHERE name = "say \"hi\"""#);
4811 }
4812
4813 #[test]
4814 fn the_shared_clauses_are_handed_to_nql_unchanged() {
4815 assert_eq!(q("SELECT * FROM orders ORDER BY total DESC LIMIT 10 OFFSET 5"),
4816 "FROM orders ORDER BY total DESC LIMIT 10 OFFSET 5");
4817 assert_eq!(q("SELECT * FROM orders GROUP BY region"), "FROM orders GROUP BY region");
4818 assert_eq!(q("SELECT * FROM o WHERE total BETWEEN 1 AND 9 ORDER BY a, b DESC"),
4819 "FROM o WHERE total BETWEEN 1 AND 9 ORDER BY a, b DESC");
4820 }
4821
4822 /// An aggregate must surface as ONE column, named as SQL names it.
4823 ///
4824 /// NQL answers `SUM(total)` with `{count, sum_total, value}` — `value`
4825 /// being a back-compat alias. Passing that straight through gave
4826 /// `SELECT COUNT(*)` two columns (`count`, `value`) where SQL promises
4827 /// one, and leaked an internal key name onto the wire.
4828 #[test]
4829 fn an_aggregate_is_one_column_named_as_sql_names_it() {
4830 assert_eq!(proj_pairs("SELECT COUNT(*) FROM orders"),
4831 vec![("count".to_string(), "count".to_string())]);
4832 assert_eq!(proj_pairs("SELECT SUM(total) FROM orders"),
4833 vec![("sum_total".to_string(), "sum".to_string())]);
4834 assert_eq!(proj_pairs("SELECT avg(total) FROM orders"),
4835 vec![("avg_total".to_string(), "avg".to_string())]);
4836 assert_eq!(proj_pairs("SELECT MIN(total) FROM orders"),
4837 vec![("min_total".to_string(), "min".to_string())]);
4838 // And the encoded result really is one column with that name.
4839 let rows = vec![json!({"count": 4, "sum_total": 420, "value": 420})];
4840 let p = vec![Col::renamed("sum_total", "sum")];
4841 let cols = columns_for(&rows, &p);
4842 assert_eq!(names(&cols), vec!["sum"], "one column, SQL's name");
4843 assert_eq!(cell(rows[0].get(&cols[0].src)), Some("420".to_string()));
4844 }
4845
4846 /// A grouped NQL row holds the group key, `count` and the aggregate —
4847 /// nothing else. Projecting another column found nothing and rendered
4848 /// NULL, which is a silent wrong answer. Postgres errors; so do we, in
4849 /// Postgres's own words.
4850 #[test]
4851 fn a_bare_column_with_group_by_is_refused_not_nulled() {
4852 let e = translate("SELECT region, total FROM orders GROUP BY region").unwrap_err();
4853 assert!(e.contains("must appear in the GROUP BY clause"), "{}", e);
4854 assert!(e.contains("total"), "the message names the offending column: {}", e);
4855
4856 // The group key itself, and `count`, are both legitimate.
4857 assert!(translate("SELECT region FROM orders GROUP BY region").is_ok());
4858 assert!(translate("SELECT region, count FROM orders GROUP BY region").is_ok());
4859 // As is an aggregate over the grouped set.
4860 assert!(translate("SELECT SUM(total) FROM orders GROUP BY region").is_ok());
4861 // And `*` is unaffected — it returns whatever the grouped row holds.
4862 assert!(translate("SELECT * FROM orders GROUP BY region").is_ok());
4863 }
4864
4865 #[test]
4866 fn count_star_becomes_nql_count() {
4867 assert_eq!(q("SELECT COUNT(*) FROM orders"), "FROM orders COUNT");
4868 assert_eq!(q("SELECT count(*) FROM orders WHERE total > 5"),
4869 "FROM orders COUNT WHERE total > 5");
4870 }
4871
4872 #[test]
4873 fn aggregates_carry_their_target_column() {
4874 assert_eq!(q("SELECT SUM(total) FROM orders"), "FROM orders SUM total");
4875 assert_eq!(q("SELECT avg(total) FROM orders WHERE region = 'eu'"),
4876 r#"FROM orders AVG total WHERE region = "eu""#);
4877 assert!(translate("SELECT SUM(*) FROM orders").is_err());
4878 }
4879
4880 /// The bridge worth having: Postgres spells time travel
4881 /// `AS OF SYSTEM TIME`, and NEDB's is sequence-addressed and permanent.
4882 #[test]
4883 fn as_of_system_time_bridges_to_nql_as_of() {
4884 assert_eq!(q("SELECT * FROM orders AS OF SYSTEM TIME 42"),
4885 "FROM orders AS OF 42");
4886 assert_eq!(q("SELECT * FROM orders AS OF SYSTEM TIME 42 WHERE total > 1"),
4887 "FROM orders AS OF 42 WHERE total > 1");
4888 // A quoted datetime is a TAGGED marker (high bit — no real seq ever
4889 // sets it): the temporal map resolves it to a real seq where the Db
4890 // is in hand. Garbage in the quoted position still refuses, naming
4891 // the accepted forms.
4892 let translated = q("SELECT * FROM orders AS OF SYSTEM TIME '2026-01-01'");
4893 let marker: u64 = translated
4894 .split(" AS OF ").nth(1).and_then(|s| s.split_whitespace().next())
4895 .and_then(|s| s.parse().ok())
4896 .expect("the translated form carries the marker");
4897 assert_ne!(marker & crate::wallclock::WALL_CLOCK_FLAG, 0,
4898 "a datetime must arrive as a tagged marker, not a bare seq");
4899 let moment = crate::wallclock::WallClock::from_marker(marker).expect("decodes");
4900 assert_eq!(moment.epoch_secs(), 1_767_225_600.0); // 2026-01-01T00:00:00Z
4901 let e = translate("SELECT * FROM orders AS OF SYSTEM TIME 'not a time'").unwrap_err();
4902 assert!(e.contains("unrecognized datetime"), "{}", e);
4903 }
4904
4905 /// A select-list item that is not a column reference must be REFUSED, not
4906 /// turned into a field name.
4907 ///
4908 /// The guard used to be `expr.contains('(')`, which only catches expressions
4909 /// that happen to have a paren. `total * 2` sailed through, became the field
4910 /// name "total * 2", matched no document, and the column came back EMPTY for
4911 /// every row with no error. Same silent class as the qualified-WHERE bug: a
4912 /// wrong answer wearing the shape of data.
4913 #[test]
4914 fn a_select_list_expression_is_refused_rather_than_answered_blank() {
4915 for sql in [
4916 "SELECT total * 2 FROM orders",
4917 "SELECT total, total*2 AS doubled FROM orders",
4918 "SELECT total + 1 FROM orders",
4919 "SELECT status || 'x' FROM orders",
4920 "SELECT -total FROM orders",
4921 "SELECT lower(status) FROM orders",
4922 ] {
4923 let e = translate(sql).unwrap_err();
4924 assert!(e.contains("expressions in the select list"), "{} -> {}", sql, e);
4925 }
4926 // ...and the things that ARE column references still pass, or the fix
4927 // would have bought correctness by refusing everything.
4928 assert_eq!(q("SELECT _id, status FROM orders"), "FROM orders");
4929 assert_eq!(q("SELECT \"status\" FROM orders"), "FROM orders");
4930 assert_eq!(q("SELECT orders.status FROM orders"), "FROM orders");
4931 assert_eq!(q("SELECT o.status FROM orders o"), "FROM orders");
4932 assert_eq!(q("SELECT total AS t FROM orders"), "FROM orders");
4933 assert!(translate("SELECT count(*) FROM orders").is_ok());
4934 assert!(translate("SELECT sum(total) FROM orders").is_ok());
4935 }
4936
4937 /// HAVING has to reach NQL in the spelling NQL's grouped row actually uses.
4938 ///
4939 /// An NQL grouped row carries `count` and `<agg>_<field>`. SQL clients write
4940 /// `count(*)`, or the alias they gave it. `count(*)` failed LOUDLY (fine),
4941 /// but `COUNT` and an alias both passed through verbatim and answered ZERO
4942 /// ROWS — which reads as "no groups qualified" rather than "your predicate
4943 /// named a field that does not exist".
4944 #[test]
4945 fn having_is_translated_to_nqls_spelling_and_refuses_an_unknown_key() {
4946 // Every spelling a client might send for the count.
4947 for sql in [
4948 "SELECT status, count(*) AS n FROM orders GROUP BY status HAVING count(*) > 1",
4949 "SELECT status, count(*) AS n FROM orders GROUP BY status HAVING n > 1",
4950 "SELECT status, count(*) FROM orders GROUP BY status HAVING COUNT > 1",
4951 "SELECT status, count(*) FROM orders GROUP BY status HAVING count > 1",
4952 ] {
4953 let got = q(sql);
4954 assert_eq!(got, "FROM orders GROUP BY status COUNT HAVING count > 1",
4955 "{} -> {}", sql, got);
4956 }
4957 // A named aggregate, by its alias -- NQL calls the field `sum_total`.
4958 assert_eq!(q("SELECT status, sum(total) AS s FROM orders GROUP BY status HAVING s > 100"),
4959 "FROM orders GROUP BY status SUM total HAVING sum_total > 100");
4960 // ...and by NQL's own name for it, which must not be rewritten twice.
4961 assert_eq!(q("SELECT status, sum(total) FROM orders GROUP BY status HAVING sum_total > 100"),
4962 "FROM orders GROUP BY status SUM total HAVING sum_total > 100");
4963 // Filtering on the group key itself is legitimate and passes through
4964 // untouched -- the SQL literal becomes an NQL one, as everywhere else.
4965 assert_eq!(q("SELECT status, count(*) FROM orders GROUP BY status HAVING status > 'a'"),
4966 "FROM orders GROUP BY status COUNT HAVING status > \"a\"");
4967 // A key the grouped row cannot carry is an ERROR, not zero rows.
4968 let e = translate(
4969 "SELECT status, count(*) FROM orders GROUP BY status HAVING nosuch > 1").unwrap_err();
4970 assert!(e.contains("HAVING names") && e.contains("nosuch"), "{}", e);
4971 assert!(e.contains("zero rows"), "the message must say what it prevented: {}", e);
4972 }
4973
4974 #[test]
4975 fn handshake_queries_are_answered_so_clients_can_connect() {
4976 assert!(matches!(translate("SELECT version()"), Ok(Stmt::Canned { .. })));
4977 assert!(matches!(translate("SHOW transaction_isolation"), Ok(Stmt::Canned { .. })));
4978 assert!(matches!(translate("SELECT current_schema()"), Ok(Stmt::Canned { .. })));
4979 assert!(matches!(translate("SET extra_float_digits = 3"), Ok(Stmt::Ok(_))));
4980 assert!(matches!(translate("BEGIN"), Ok(Stmt::Ok(_))));
4981 assert!(matches!(translate(""), Ok(Stmt::Ok(_))));
4982 }
4983
4984 /// Every refusal has to name the boundary. "Syntax error" would send a
4985 /// developer hunting for a typo that is not there.
4986 #[test]
4987 fn unsupported_sql_is_refused_with_a_reason() {
4988 for (sql, expect) in [
4989 ("INSERT INTO t VALUES (1)", "explicit column list"),
4990 ("CREATE TABLE t (a int)", "DDL"),
4991 ("TRUNCATE t", "append-only"),
4992 ("GRANT ALL ON t TO x", "privilege system"),
4993 ("SELECT * FROM a JOIN b ON a.x = b.x", "JOIN is not supported"),
4994 ("SELECT * FROM a UNION SELECT * FROM b", "UNION"),
4995 ("SELECT DISTINCT region FROM orders", "GROUP BY"),
4996 ("SELECT * FROM (SELECT 1) x", "subqueries in FROM"),
4997 ("SELECT * FROM a, b", "more than one collection"),
4998 ("SELECT lower(status) FROM orders", "expressions in the select list"),
4999 ("VACUUM", "only SELECT"),
5000 ] {
5001 let e = translate(sql).unwrap_err();
5002 assert!(e.contains(expect), "for {:?} expected {:?} in {:?}", sql, expect, e);
5003 }
5004 }
5005
5006 // ── writes ───────────────────────────────────────────────────────────────
5007 //
5008 // SQL's write semantics and NEDB's append-only model line up: INSERT is a
5009 // put, UPDATE is a new version, DELETE is a tombstone. These tests pin the
5010 // parse; tests/test_pgwire.py proves the behaviour against a live server,
5011 // including that the PRIOR value is still readable afterwards.
5012
5013 fn ins(sql: &str) -> (String, Vec<InsertRow>, Vec<Col>) {
5014 match translate(sql) {
5015 Ok(Stmt::Insert { coll, rows, returning }) => (coll, rows, returning),
5016 other => panic!("expected INSERT for {:?}, got {:?}", sql, other),
5017 }
5018 }
5019
5020 #[test]
5021 fn insert_becomes_a_put_per_row() {
5022 let (coll, rows, ret) = ins("INSERT INTO orders (_id, status, total) VALUES ('o1', 'paid', 120)");
5023 assert_eq!(coll, "orders");
5024 assert_eq!(rows.len(), 1);
5025 assert_eq!(rows[0].id.as_deref(), Some("o1"));
5026 assert_eq!(rows[0].doc.get("status"), Some(&json!("paid")));
5027 assert_eq!(rows[0].doc.get("total"), Some(&json!(120)));
5028 // `_id` is the key, not a payload field.
5029 assert!(!rows[0].doc.contains_key("_id"));
5030 assert!(ret.is_empty());
5031 }
5032
5033 #[test]
5034 fn a_multi_row_insert_yields_one_row_each() {
5035 let (_, rows, _) = ins(
5036 "INSERT INTO t (id, n) VALUES ('a', 1), ('b', 2), ('c', 3)");
5037 assert_eq!(rows.len(), 3);
5038 assert_eq!(rows[1].id.as_deref(), Some("b"));
5039 assert_eq!(rows[2].doc.get("n"), Some(&json!(3)));
5040 }
5041
5042 #[test]
5043 fn an_insert_without_an_id_column_lets_the_server_assign_one() {
5044 let (_, rows, _) = ins("INSERT INTO t (n) VALUES (1)");
5045 assert_eq!(rows[0].id, None, "the executor mints a unique key");
5046 assert_eq!(rows[0].doc.get("n"), Some(&json!(1)));
5047 }
5048
5049 /// Provenance is reachable from SQL, not only from the HTTP API — which is
5050 /// the point of having writes here at all.
5051 #[test]
5052 fn insert_lifts_provenance_out_of_reserved_columns() {
5053 let (_, rows, _) = ins(
5054 "INSERT INTO audit (_id, _caused_by, _valid_from, kind) \
5055 VALUES ('e1', 'abc123', '2026-01-01', 'reprice')");
5056 assert_eq!(rows[0].caused_by, vec!["abc123".to_string()]);
5057 assert_eq!(rows[0].valid_from.as_deref(), Some("2026-01-01"));
5058 assert_eq!(rows[0].doc.get("kind"), Some(&json!("reprice")));
5059 // None of the reserved names leak into the stored payload.
5060 for k in ["_id", "_caused_by", "_valid_from"] {
5061 assert!(!rows[0].doc.contains_key(k), "{} leaked into the doc", k);
5062 }
5063 }
5064
5065 #[test]
5066 fn insert_values_cover_the_scalar_types() {
5067 let (_, rows, _) = ins(
5068 "INSERT INTO t (s, i, f, b, n) VALUES ('x', 42, 1.5, TRUE, NULL)");
5069 assert_eq!(rows[0].doc.get("s"), Some(&json!("x")));
5070 assert_eq!(rows[0].doc.get("i"), Some(&json!(42)));
5071 assert_eq!(rows[0].doc.get("f"), Some(&json!(1.5)));
5072 assert_eq!(rows[0].doc.get("b"), Some(&json!(true)));
5073 assert_eq!(rows[0].doc.get("n"), Some(&Value::Null));
5074 }
5075
5076 /// A doubled '' is one literal quote, and a comma inside a string is not a
5077 /// value separator.
5078 #[test]
5079 fn insert_literals_survive_quotes_and_commas() {
5080 let (_, rows, _) = ins("INSERT INTO t (a, b) VALUES ('it''s', 'x,y')");
5081 assert_eq!(rows[0].doc.get("a"), Some(&json!("it's")));
5082 assert_eq!(rows[0].doc.get("b"), Some(&json!("x,y")));
5083 }
5084
5085 #[test]
5086 fn insert_refuses_what_it_cannot_store_faithfully() {
5087 // An unevaluated expression stored as text would be a wrong value.
5088 assert!(translate("INSERT INTO t (a) VALUES (1 + 1)").is_err());
5089 assert!(translate("INSERT INTO t (a) VALUES (now())").is_err());
5090 // Column/value count mismatch.
5091 let e = translate("INSERT INTO t (a, b) VALUES (1)").unwrap_err();
5092 assert!(e.contains("values for"), "{}", e);
5093 // No column list at all.
5094 let e2 = translate("INSERT INTO t VALUES (1)").unwrap_err();
5095 assert!(e2.contains("explicit column list"), "{}", e2);
5096 }
5097
5098 #[test]
5099 fn update_finds_rows_with_the_full_predicate_surface() {
5100 match translate("UPDATE orders SET status = 'void' WHERE total < 50 AND region IN ('eu')") {
5101 Ok(Stmt::Update { coll, set, nql, .. }) => {
5102 assert_eq!(coll, "orders");
5103 assert_eq!(set, vec![("status".to_string(), json!("void"))]);
5104 // The WHERE became ordinary NQL, so IN/BETWEEN/LIKE all work.
5105 assert_eq!(nql, r#"FROM orders WHERE total < 50 AND region IN ("eu")"#);
5106 }
5107 other => panic!("expected UPDATE, got {:?}", other),
5108 }
5109 }
5110
5111 #[test]
5112 fn update_without_where_targets_the_whole_collection() {
5113 // Postgres allows it, so parity allows it.
5114 match translate("UPDATE t SET a = 1") {
5115 Ok(Stmt::Update { nql, .. }) => assert_eq!(nql, "FROM t"),
5116 other => panic!("expected UPDATE, got {:?}", other),
5117 }
5118 }
5119
5120 #[test]
5121 fn update_handles_several_assignments() {
5122 match translate("UPDATE t SET a = 1, b = 'x,y', c = NULL WHERE id = 'k'") {
5123 Ok(Stmt::Update { set, .. }) => {
5124 assert_eq!(set.len(), 3);
5125 assert_eq!(set[1], ("b".to_string(), json!("x,y")));
5126 assert_eq!(set[2], ("c".to_string(), Value::Null));
5127 }
5128 other => panic!("expected UPDATE, got {:?}", other),
5129 }
5130 assert!(translate("UPDATE t SET").is_err());
5131 assert!(translate("UPDATE t SET a").is_err());
5132 }
5133
5134 #[test]
5135 fn delete_becomes_a_predicate_over_the_collection() {
5136 match translate("DELETE FROM orders WHERE status = 'void'") {
5137 Ok(Stmt::Delete { coll, nql, .. }) => {
5138 assert_eq!(coll, "orders");
5139 assert_eq!(nql, r#"FROM orders WHERE status = "void""#);
5140 }
5141 other => panic!("expected DELETE, got {:?}", other),
5142 }
5143 match translate("DELETE FROM t") {
5144 Ok(Stmt::Delete { nql, .. }) => assert_eq!(nql, "FROM t"),
5145 other => panic!("expected DELETE, got {:?}", other),
5146 }
5147 }
5148
5149 #[test]
5150 fn returning_is_parsed_off_every_write() {
5151 let (_, _, ret) = ins("INSERT INTO t (a) VALUES (1) RETURNING a, _id");
5152 assert_eq!(ret.iter().map(|c| c.out.clone()).collect::<Vec<_>>(), vec!["a", "_id"]);
5153 // `RETURNING *` is an empty projection — every column — which is why
5154 // the executor checks the raw SQL for the keyword instead.
5155 let (_, _, star) = ins("INSERT INTO t (a) VALUES (1) RETURNING *");
5156 assert!(star.is_empty());
5157 assert!(wants_returning("INSERT INTO t (a) VALUES (1) RETURNING *"));
5158 assert!(!wants_returning("INSERT INTO t (a) VALUES (1)"));
5159
5160 match translate("UPDATE t SET a = 1 WHERE id = 'k' RETURNING a") {
5161 Ok(Stmt::Update { nql, returning, .. }) => {
5162 assert_eq!(returning.len(), 1);
5163 // RETURNING must NOT leak into the predicate.
5164 assert!(!nql.to_uppercase().contains("RETURNING"), "{}", nql);
5165 }
5166 other => panic!("expected UPDATE, got {:?}", other),
5167 }
5168 match translate("DELETE FROM t WHERE id = 'k' RETURNING *") {
5169 Ok(Stmt::Delete { nql, .. }) =>
5170 assert!(!nql.to_uppercase().contains("RETURNING"), "{}", nql),
5171 other => panic!("expected DELETE, got {:?}", other),
5172 }
5173 }
5174
5175 #[test]
5176 fn a_keyword_inside_a_value_is_not_a_clause() {
5177 match translate("UPDATE t SET note = 'where returning from' WHERE id = 'k'") {
5178 Ok(Stmt::Update { set, nql, .. }) => {
5179 assert_eq!(set[0].1, json!("where returning from"));
5180 assert_eq!(nql, r#"FROM t WHERE id = "k""#);
5181 }
5182 other => panic!("expected UPDATE, got {:?}", other),
5183 }
5184 }
5185
5186 #[test]
5187 fn split_top_respects_quotes_and_nesting() {
5188 assert_eq!(split_top("a, b, c", ',').len(), 3);
5189 assert_eq!(split_top("(1, 2), (3, 4)", ',').len(), 2);
5190 assert_eq!(split_top("'a,b', c", ',').len(), 2);
5191 assert_eq!(split_top("'it''s, fine', c", ',').len(), 2);
5192 }
5193
5194 #[test]
5195 fn comments_and_whitespace_do_not_confuse_the_translator() {
5196 assert_eq!(q("SELECT *\n FROM orders -- trailing note\n"), "FROM orders");
5197 assert_eq!(q("SELECT /* inline */ * FROM orders"), "FROM orders");
5198 // A keyword inside a string literal must not be treated as a clause.
5199 assert_eq!(q("SELECT * FROM t WHERE note = 'from here to JOIN'"),
5200 r#"FROM t WHERE note = "from here to JOIN""#);
5201 }
5202
5203 #[test]
5204 fn find_kw_ignores_quotes_parens_and_substrings() {
5205 assert_eq!(find_kw("SELECT A FROM B", "FROM"), Some(9));
5206 assert_eq!(find_kw("SELECT 'FROM' FROM B", "FROM"), Some(14));
5207 assert_eq!(find_kw("SELECT F(x FROM y) FROM B", "FROM"), Some(19));
5208 assert_eq!(find_kw("SELECT FROMAGE", "FROM"), None);
5209 assert_eq!(find_kw("SELECT X_FROM", "FROM"), None);
5210 }
5211
5212 // ── result encoding ──────────────────────────────────────────────────────
5213
5214 #[test]
5215 fn provenance_columns_sort_after_the_users_own_fields() {
5216 let rows = vec![json!({"_id":"1","_hash":"ab","status":"paid","total":9})];
5217 assert_eq!(names(&columns_for(&rows, &[])),
5218 vec!["status", "total", "_hash", "_id"]);
5219 }
5220
5221 #[test]
5222 fn an_explicit_projection_sets_the_column_order() {
5223 let rows = vec![json!({"a":1,"b":2})];
5224 let p = vec![Col::same("b"), Col::same("a")];
5225 assert_eq!(names(&columns_for(&rows, &p)), vec!["b", "a"]);
5226 }
5227
5228 #[test]
5229 fn columns_are_the_union_across_sparse_rows() {
5230 // A document store has no schema, so row 2 may carry a field row 1 lacks.
5231 let rows = vec![json!({"a":1}), json!({"b":2})];
5232 assert_eq!(names(&columns_for(&rows, &[])), vec!["a", "b"]);
5233 }
5234
5235 #[test]
5236 fn type_oids_follow_the_first_non_null_value() {
5237 let rows = vec![json!({"i":1,"f":1.5,"b":true,"s":"x","n":null})];
5238 assert_eq!(oid_for(&rows, "i"), OID_INT8);
5239 assert_eq!(oid_for(&rows, "f"), OID_FLOAT8);
5240 assert_eq!(oid_for(&rows, "b"), OID_BOOL);
5241 assert_eq!(oid_for(&rows, "s"), OID_TEXT);
5242 // All-null and absent columns fall back to text rather than guessing.
5243 assert_eq!(oid_for(&rows, "n"), OID_TEXT);
5244 assert_eq!(oid_for(&rows, "absent"), OID_TEXT);
5245 }
5246
5247 #[test]
5248 fn a_column_that_is_null_in_the_first_row_still_gets_its_type() {
5249 let rows = vec![json!({"v": null}), json!({"v": 7})];
5250 assert_eq!(oid_for(&rows, "v"), OID_INT8);
5251 }
5252
5253 #[test]
5254 fn cells_render_in_postgres_text_format() {
5255 assert_eq!(cell(Some(&json!("x"))), Some("x".to_string()));
5256 assert_eq!(cell(Some(&json!(true))), Some("t".to_string()));
5257 assert_eq!(cell(Some(&json!(false))), Some("f".to_string()));
5258 assert_eq!(cell(Some(&json!(42))), Some("42".to_string()));
5259 assert_eq!(cell(Some(&json!(null))), None);
5260 assert_eq!(cell(None), None);
5261 // Nested values render as JSON text rather than being dropped.
5262 assert_eq!(cell(Some(&json!({"a":1}))), Some("{\"a\":1}".to_string()));
5263 }
5264
5265 /// The framing has to be exact or the client desynchronises and hangs.
5266 /// Length covers the length field itself but not the tag byte.
5267 #[test]
5268 fn message_framing_length_excludes_the_tag() {
5269 let mut m = Out::msg(b'Z');
5270 m.bytes(b"I");
5271 let bytes = m.finish();
5272 assert_eq!(bytes[0], b'Z');
5273 assert_eq!(i32::from_be_bytes([bytes[1], bytes[2], bytes[3], bytes[4]]), 5);
5274 assert_eq!(bytes.len(), 6);
5275 }
5276
5277 #[test]
5278 fn a_result_set_encodes_as_description_then_rows_then_complete() {
5279 let rows = vec![json!({"a": 1}), json!({"a": 2})];
5280 let out = encode_result(&rows, &[]);
5281 assert_eq!(out[0], b'T');
5282 let tags: Vec<u8> = {
5283 // Walk the message stream by its own length prefixes.
5284 let mut t = vec![];
5285 let mut i = 0usize;
5286 while i < out.len() {
5287 t.push(out[i]);
5288 let len = i32::from_be_bytes([out[i+1], out[i+2], out[i+3], out[i+4]]) as usize;
5289 i += 1 + len;
5290 }
5291 t
5292 };
5293 assert_eq!(tags, vec![b'T', b'D', b'D', b'C'],
5294 "one description, one row each, one completion");
5295 }
5296
5297 /// A statement must emit EXACTLY ONE CommandComplete. A write with
5298 /// RETURNING that reused the SELECT encoder sent two, and the visible
5299 /// symptom was RETURNING yielding no rows: the client took the first tag
5300 /// as the end of the statement and threw the description away.
5301 #[test]
5302 fn a_write_with_returning_emits_exactly_one_command_complete() {
5303 let rows = vec![json!({"_id": "o1", "total": 9})];
5304 let mut out = encode_rows(&rows, &[Col::same("_id")]);
5305 out.extend_from_slice(&command_complete("INSERT 0 1"));
5306 let mut tags = vec![];
5307 let mut i = 0usize;
5308 while i < out.len() {
5309 tags.push(out[i]);
5310 let len = i32::from_be_bytes([out[i+1], out[i+2], out[i+3], out[i+4]]) as usize;
5311 i += 1 + len;
5312 }
5313 assert_eq!(tags, vec![b'T', b'D', b'C'], "one description, one row, ONE tag");
5314 assert_eq!(tags.iter().filter(|t| **t == b'C').count(), 1);
5315 // encode_rows alone must not carry a tag at all.
5316 assert!(!encode_rows(&rows, &[]).contains(&b'C')
5317 || encode_rows(&rows, &[]).iter().filter(|b| **b == b'C').count() > 0);
5318 let bare = encode_rows(&rows, &[Col::same("_id")]);
5319 let mut bare_tags = vec![];
5320 let mut j = 0usize;
5321 while j < bare.len() {
5322 bare_tags.push(bare[j]);
5323 let len = i32::from_be_bytes([bare[j+1], bare[j+2], bare[j+3], bare[j+4]]) as usize;
5324 j += 1 + len;
5325 }
5326 assert_eq!(bare_tags, vec![b'T', b'D'], "encode_rows never appends a tag");
5327 }
5328
5329 #[test]
5330 fn an_empty_result_still_sends_a_description() {
5331 let out = encode_result(&[], &[Col::same("a")]);
5332 assert_eq!(out[0], b'T', "clients need the shape even with no rows");
5333 }
5334
5335 #[test]
5336 fn statements_split_on_top_level_semicolons_only() {
5337 assert_eq!(split_statements("SELECT 1; SELECT 2").len(), 2);
5338 assert_eq!(split_statements("SELECT ';'").len(), 1);
5339 assert_eq!(split_statements("SELECT 1;").len(), 1);
5340 assert_eq!(split_statements(" ").len(), 0);
5341 }
5342
5343 #[test]
5344 fn an_error_names_its_sqlstate() {
5345 let e = String::from_utf8_lossy(&err_msg("0A000", "x")).to_string();
5346 assert!(e.contains("ERROR"));
5347 assert!(e.contains("0A000"));
5348 }
5349
5350 // ── the extended query protocol ─────────────────────────────────────────
5351
5352 #[test]
5353 fn placeholders_are_counted_outside_string_literals() {
5354 assert_eq!(param_count("SELECT a FROM t WHERE b = $1 AND c = $2"), 2);
5355 assert_eq!(param_count("SELECT a FROM t"), 0);
5356 // The highest index wins, because a parameter may be reused.
5357 assert_eq!(param_count("WHERE a = $2 OR b = $2 OR c = $1"), 2);
5358 assert_eq!(param_count("SELECT a FROM t WHERE b = '$1'"), 0,
5359 "a placeholder inside a literal is data, not a parameter");
5360 assert_eq!(param_count("WHERE a = $10 AND b = $1"), 10,
5361 "two-digit indexes must not be read as $1 followed by 0");
5362 }
5363
5364 #[test]
5365 fn parameters_are_spliced_as_literals() {
5366 let out = substitute_params("WHERE a = $1 AND b = $2 AND c = $3",
5367 &[Some("'x'".into()), Some("42".into()), None]).unwrap();
5368 assert_eq!(out, "WHERE a = 'x' AND b = 42 AND c = NULL");
5369 }
5370
5371 #[test]
5372 fn substitution_leaves_string_literals_alone() {
5373 let out = substitute_params("WHERE a = '$1' AND b = $1", &[Some("9".into())]).unwrap();
5374 assert_eq!(out, "WHERE a = '$1' AND b = 9");
5375 }
5376
5377 #[test]
5378 fn too_few_parameters_is_an_error_not_a_silent_null() {
5379 // The alternative — treating a missing parameter as NULL — turns a
5380 // client bug into a wrong answer with a 200-shaped response.
5381 let e = substitute_params("WHERE a = $2", &[Some("1".into())]).unwrap_err();
5382 assert!(e.contains("$2"), "{}", e);
5383 }
5384
5385 #[test]
5386 fn a_quote_in_a_parameter_cannot_escape_its_literal() {
5387 let lit = decode_param(Some(b"it's"), OID_TEXT, 0).unwrap().unwrap();
5388 assert_eq!(lit, "'it''s'");
5389 // And it survives a round trip through the splice unchanged.
5390 let out = substitute_params("WHERE a = $1", &[Some(lit)]).unwrap();
5391 assert_eq!(out, "WHERE a = 'it''s'");
5392 }
5393
5394 #[test]
5395 fn binary_parameters_decode_in_every_width_psycopg_sends() {
5396 // These are the exact encodings read off a psycopg3 wire transcript:
5397 // a small int arrives as int2, a float as float8, a bool as one byte.
5398 assert_eq!(decode_param(Some(&[0x00, 0x2a]), OID_INT2, 1).unwrap().unwrap(), "42");
5399 assert_eq!(decode_param(Some(&[0, 0, 0, 7]), OID_INT4, 1).unwrap().unwrap(), "7");
5400 assert_eq!(
5401 decode_param(Some(&[0, 0, 0, 0, 0, 0, 0, 9]), OID_INT8, 1).unwrap().unwrap(), "9");
5402 assert_eq!(
5403 decode_param(Some(&0x400c_0000_0000_0000u64.to_be_bytes()), OID_FLOAT8, 1)
5404 .unwrap().unwrap(), "3.5");
5405 assert_eq!(decode_param(Some(&[1]), OID_BOOL, 1).unwrap().unwrap(), "TRUE");
5406 assert_eq!(decode_param(Some(&[0]), OID_BOOL, 1).unwrap().unwrap(), "FALSE");
5407 }
5408
5409 #[test]
5410 fn a_negative_binary_integer_keeps_its_sign() {
5411 assert_eq!(decode_param(Some(&(-5i32).to_be_bytes()), OID_INT4, 1).unwrap().unwrap(), "-5");
5412 assert_eq!(decode_param(Some(&(-5i16).to_be_bytes()), OID_INT2, 1).unwrap().unwrap(), "-5");
5413 }
5414
5415 #[test]
5416 fn a_binary_parameter_of_the_wrong_width_is_refused() {
5417 // Truncating or zero-extending would produce a plausible wrong number,
5418 // which is the failure mode worth engineering against.
5419 let e = decode_param(Some(&[0x2a]), OID_INT4, 1).unwrap_err();
5420 assert!(e.contains("4 bytes"), "{}", e);
5421 }
5422
5423 #[test]
5424 fn an_unspecified_text_parameter_is_treated_as_a_string() {
5425 // psycopg3 declares OID 0 only for `str`; every number it sends carries
5426 // a real numeric OID. So quoting here is grounded, not a guess.
5427 assert_eq!(decode_param(Some(b"hello"), 0, 0).unwrap().unwrap(), "'hello'");
5428 }
5429
5430 #[test]
5431 fn a_null_parameter_decodes_to_none_in_every_format() {
5432 assert_eq!(decode_param(None, OID_TEXT, 0).unwrap(), None);
5433 assert_eq!(decode_param(None, OID_INT8, 1).unwrap(), None);
5434 }
5435
5436 #[test]
5437 fn an_unsupported_binary_type_says_so_by_name() {
5438 let e = decode_param(Some(&[0u8; 8]), 1114, 1).unwrap_err();
5439 assert!(e.contains("1114"), "{}", e);
5440 assert!(e.contains("text"), "the error should point at the way out: {}", e);
5441 }
5442
5443 #[test]
5444 fn a_text_number_that_is_not_a_number_gets_quoted() {
5445 // Splicing it in bare would emit a naked identifier into the NQL text
5446 // and fail somewhere far away from the cause.
5447 assert_eq!(decode_param(Some(b"oops"), OID_INT8, 0).unwrap().unwrap(), "'oops'");
5448 }
5449
5450 #[test]
5451 fn a_client_declared_type_is_believed_over_inference() {
5452 // The client is about to encode its argument that way; overriding it
5453 // would break the decode.
5454 let oids = infer_param_oids("SELECT a FROM t WHERE b = $1 AND c = $2", &[OID_INT4, 0], None);
5455 assert_eq!(oids, vec![OID_INT4, OID_TEXT]);
5456 }
5457
5458 #[test]
5459 fn parameter_arity_is_taken_from_the_sql_when_the_client_declares_none() {
5460 // asyncpg declares nothing and then refuses the call if the count that
5461 // comes back is wrong, so this is the load-bearing path for it.
5462 let oids = infer_param_oids("SELECT a FROM t WHERE b = $1 AND c = $2", &[], None);
5463 assert_eq!(oids.len(), 2);
5464 }
5465
5466 #[test]
5467 fn the_field_behind_each_placeholder_is_identified() {
5468 assert_eq!(
5469 param_fields("SELECT a FROM t WHERE qty > $1 AND status = $2", 2),
5470 vec![Some("qty".to_string()), Some("status".to_string())]);
5471 }
5472
5473 #[test]
5474 fn word_operators_do_not_hide_the_field() {
5475 assert_eq!(param_fields("SELECT a FROM t WHERE name LIKE $1", 1),
5476 vec![Some("name".to_string())]);
5477 assert_eq!(param_fields("SELECT a FROM t WHERE qty BETWEEN $1 AND $2", 2),
5478 vec![Some("qty".to_string()), Some("qty".to_string())]);
5479 assert_eq!(param_fields("SELECT a FROM t WHERE region IN ($1, $2)", 2),
5480 vec![Some("region".to_string()), Some("region".to_string())]);
5481 }
5482
5483 #[test]
5484 fn a_clause_position_types_from_the_grammar_not_from_a_column() {
5485 // `AS OF SYSTEM TIME $1` has no column beside it — the token to its
5486 // left is the word TIME. Typing it text made asyncpg refuse to send
5487 // the sequence number at all.
5488 assert_eq!(
5489 infer_param_oids("SELECT a FROM t AS OF SYSTEM TIME $1 WHERE b = $2", &[], None),
5490 vec![OID_INT8, OID_TEXT]);
5491 assert_eq!(infer_param_oids("SELECT a FROM t AS OF $1", &[], None), vec![OID_INT8]);
5492 // VALID AS OF also ends with "AS OF", but its argument is a DATE
5493 // STRING. Checking the longer clause first is load-bearing.
5494 assert_eq!(
5495 infer_param_oids("SELECT a FROM t VALID AS OF $1", &[], None), vec![OID_TEXT]);
5496 assert_eq!(
5497 infer_param_oids("SELECT a FROM t LIMIT $1 OFFSET $2", &[], None),
5498 vec![OID_INT8, OID_INT8]);
5499 }
5500
5501 #[test]
5502 fn an_aggregate_column_types_from_what_the_aggregate_means() {
5503 // No document holds a field called `count`, so sampling stored data
5504 // finds nothing and falls back to text — which hands a binary client
5505 // the string "2" for COUNT(*).
5506 assert_eq!(aggregate_oid("count", None, "t"), Some(OID_INT8));
5507 assert_eq!(aggregate_oid("avg_fee", None, "t"), Some(OID_FLOAT8),
5508 "an average is fractional even over integers");
5509 // SUM/MIN/MAX inherit the field's type; with no database to sample,
5510 // that resolves to text, and `_seq` is known from the engine contract.
5511 assert_eq!(aggregate_oid("max__seq", None, "t"), Some(OID_INT8));
5512 assert_eq!(aggregate_oid("total", None, "t"), None, "not an aggregate");
5513 }
5514
5515 #[test]
5516 fn the_parse_probe_uses_a_literal_that_every_clause_accepts() {
5517 // Stubbing with NULL was the obvious choice and the wrong one: clauses
5518 // that validate their argument rejected it, so `AS OF SYSTEM TIME $1`
5519 // failed at Parse before a real sequence was ever bound.
5520 let probe = probe_sql("SELECT a FROM t AS OF SYSTEM TIME $1 WHERE b = $2", 2);
5521 assert!(!probe.contains("NULL"), "{}", probe);
5522 assert!(translate(&probe).is_ok(), "the probe must parse: {}", probe);
5523 }
5524
5525 #[test]
5526 fn a_column_with_mixed_types_across_documents_is_advertised_as_text() {
5527 // Taking the first non-null value's type told the client `int8` and
5528 // then sent it "n/a" — which fails to parse client-side, and on the
5529 // binary path cannot be encoded at all.
5530 let rows = vec![json!({"x": 3}), json!({"x": "n/a"})];
5531 assert_eq!(oid_for(&rows, "x"), OID_TEXT);
5532 // Integers and floats in one column widen rather than conflict.
5533 let rows = vec![json!({"x": 3}), json!({"x": 1.5})];
5534 assert_eq!(oid_for(&rows, "x"), OID_FLOAT8);
5535 // A leading null must not decide the type.
5536 let rows = vec![json!({"x": Value::Null}), json!({"x": 7})];
5537 assert_eq!(oid_for(&rows, "x"), OID_INT8);
5538 }
5539
5540 #[test]
5541 fn binary_output_encodes_each_advertised_type() {
5542 assert_eq!(cell_binary(Some(&json!(true)), OID_BOOL).unwrap().unwrap(), vec![1]);
5543 assert_eq!(cell_binary(Some(&json!(42)), OID_INT8).unwrap().unwrap(),
5544 42i64.to_be_bytes().to_vec());
5545 assert_eq!(cell_binary(Some(&json!(3.5)), OID_FLOAT8).unwrap().unwrap(),
5546 3.5f64.to_be_bytes().to_vec());
5547 // For the text family, binary and text are the same bytes.
5548 assert_eq!(cell_binary(Some(&json!("hi")), OID_TEXT).unwrap().unwrap(), b"hi".to_vec());
5549 assert_eq!(cell_binary(Some(&Value::Null), OID_INT8).unwrap(), None);
5550 // A boolean renders as `t`/`f` in text but one byte in binary.
5551 assert_eq!(cell(Some(&json!(true))).unwrap(), "t");
5552 }
5553
5554 #[test]
5555 fn a_value_that_does_not_fit_its_advertised_binary_type_is_refused() {
5556 // Advertised types come from a bounded sample, so a field that only
5557 // turns heterogeneous outside it lands here. Sending a zero, or the
5558 // text bytes under a binary header, would corrupt the value in a way
5559 // the client cannot detect — so it is an error instead.
5560 let e = cell_binary(Some(&json!("nope")), OID_INT8).unwrap_err();
5561 assert!(e.contains("a string"), "{}", e);
5562 assert!(e.contains("more than one type"), "the error should explain WHY: {}", e);
5563 }
5564
5565 #[test]
5566 fn a_row_description_carries_the_requested_format_per_column() {
5567 let cols = [Col::same("a"), Col::same("b")];
5568 let m = row_description_fmt(&cols, &[OID_INT8, OID_TEXT], &[1, 0]);
5569 assert_eq!(m[0], b'T');
5570 // The trailing i16 of each field entry is its format code.
5571 assert_eq!(m[m.len() - 1], 0, "the last column was requested as text");
5572 }
5573
5574 #[test]
5575 fn a_qualified_column_resolves_to_its_bare_name() {
5576 assert_eq!(param_fields("SELECT a FROM t WHERE t.qty = $1", 1),
5577 vec![Some("qty".to_string())]);
5578 }
5579
5580 #[test]
5581 fn insert_placeholders_map_positionally_to_the_column_list() {
5582 assert_eq!(
5583 param_fields("INSERT INTO t (_id, qty, status) VALUES ($1, $2, $3)", 3),
5584 vec![Some("_id".to_string()), Some("qty".to_string()), Some("status".to_string())]);
5585 }
5586
5587 #[test]
5588 fn a_set_clause_placeholder_finds_its_column() {
5589 assert_eq!(param_fields("UPDATE t SET status = $1 WHERE _id = $2", 2),
5590 vec![Some("status".to_string()), Some("_id".to_string())]);
5591 }
5592
5593 #[test]
5594 fn the_target_collection_is_found_for_every_statement_kind() {
5595 assert_eq!(stmt_collection("SELECT a FROM inv WHERE b = $1"), "inv");
5596 assert_eq!(stmt_collection("UPDATE inv SET a = $1"), "inv");
5597 assert_eq!(stmt_collection("DELETE FROM inv WHERE a = $1"), "inv");
5598 assert_eq!(stmt_collection("INSERT INTO inv (a) VALUES ($1)"), "inv");
5599 // Clients qualify as schema.table; NEDB has one namespace.
5600 assert_eq!(stmt_collection("SELECT a FROM public.inv"), "inv");
5601 assert_eq!(stmt_collection("INSERT INTO inv(a) VALUES ($1)"), "inv");
5602 }
5603
5604 #[test]
5605 fn engine_metadata_fields_type_without_touching_storage() {
5606 assert_eq!(infer_field_oid(None, "t", "_seq"), OID_INT8);
5607 assert_eq!(infer_field_oid(None, "t", "_id"), OID_TEXT);
5608 }
5609
5610 #[test]
5611 fn the_protocol_acknowledgements_are_single_empty_messages() {
5612 // Each is a tag plus a 4-byte length of exactly 4.
5613 for (m, tag) in [
5614 (parse_complete(), b'1'), (bind_complete(), b'2'),
5615 (close_complete(), b'3'), (no_data(), b'n'), (portal_suspended(), b's'),
5616 ] {
5617 assert_eq!(m.len(), 5, "{:?}", tag as char);
5618 assert_eq!(m[0], tag);
5619 assert_eq!(i32::from_be_bytes([m[1], m[2], m[3], m[4]]), 4);
5620 }
5621 }
5622
5623 #[test]
5624 fn parameter_description_reports_its_arity_and_types() {
5625 let m = parameter_description(&[OID_TEXT, OID_INT8]);
5626 assert_eq!(m[0], b't');
5627 assert_eq!(i16::from_be_bytes([m[5], m[6]]), 2);
5628 assert_eq!(i32::from_be_bytes([m[7], m[8], m[9], m[10]]), OID_TEXT);
5629 assert_eq!(i32::from_be_bytes([m[11], m[12], m[13], m[14]]), OID_INT8);
5630 }
5631
5632 #[test]
5633 fn a_cstring_is_taken_without_its_terminator() {
5634 let body = b"one\0two\0".to_vec();
5635 let mut at = 0usize;
5636 assert_eq!(take_cstr(&body, &mut at), "one");
5637 assert_eq!(take_cstr(&body, &mut at), "two");
5638 assert_eq!(at, body.len());
5639 }
5640
5641 #[test]
5642 fn truncated_integers_are_reported_rather_than_read_past_the_end() {
5643 let body = vec![0u8, 1];
5644 let mut at = 0usize;
5645 assert!(take_i32(&body, &mut at).is_err());
5646 let mut at = 0usize;
5647 assert!(take_i16(&body, &mut at).is_ok());
5648 }
5649
5650 #[test]
5651 fn a_binary_result_format_request_is_refused_rather_than_faked() {
5652 // Sending text under a binary header corrupts every value silently,
5653 // which is far worse than an error naming the limitation.
5654 let out = encode_rows(&[], &[Col::same("a")]);
5655 let desc_format = &out[out.len() - 2..];
5656 assert_eq!(i16::from_be_bytes([desc_format[0], desc_format[1]]), 0,
5657 "every column is advertised as text format");
5658 }
5659
5660 #[test]
5661 fn a_float_parameter_does_not_render_as_rust_infinity() {
5662 assert_eq!(fmt_float(f64::INFINITY), "'Infinity'");
5663 assert_eq!(fmt_float(f64::NEG_INFINITY), "'-Infinity'");
5664 assert_eq!(fmt_float(f64::NAN), "'NaN'");
5665 assert_eq!(fmt_float(3.0), "3", "a whole float should not gain a .0 tail");
5666 assert_eq!(fmt_float(3.5), "3.5");
5667 }
5668}
5669