fix: resolve qualified table names against the schema

crates/sqlquack-duckdb/src/blocks/constants.rs

@@ -100,6 +100,8 @@ pub const EPOCH_TEXT_PATTERN: &str = "[0-9]{1,18}";
 /// Epoch magnitudes below which integer text counts seconds, milliseconds and microseconds.
 pub const EPOCH_SCALE_LIMITS: [i64; 3] =
     [31_536_000_000, 31_536_000_000_000, 31_536_000_000_000_000];
+/// What separates the catalog, schema and table parts of a relation name.
+pub const NAME_SEPARATOR: &str = ".";
 /// Quoted identifiers: `duckdb.quoted_identifier`.
 pub const QUOTED_IDENTIFIER: &str = "duckdb.quoted_identifier";
 /// String literals: `duckdb.string_literal`.

crates/sqlquack-duckdb/src/checking/_helpers/engine.rs

@@ -8,6 +8,7 @@ use duckdb::Connection;
 use sqlquack_core::provenance::models::ProvenanceLog;
 
 use crate::blocks::main::quoted_identifier::quoted_identifier;
+use crate::checking::main::quoted_relation::quoted_relation;
 
 use crate::checking::constants::{DISABLE_OPTIMIZER, MACRO_PREFIX};
 use crate::checking::errors::{CheckStage, PrintedSqlFailure};
@@ -21,7 +22,7 @@ pub(crate) struct Engine {
 }
 
 impl Engine {
-    /// Creates table `name` with `statement`, replacing a table created with other columns.
+    /// Creates table `name` with `statement`, dropping only a table it made earlier under that name.
     pub(crate) fn define_table(
         &mut self,
         name: &str,
@@ -30,10 +31,15 @@ impl Engine {
         if self.tables.get(name).map(String::as_str) == Some(statement) {
             return Ok(());
         }
-        let drop = format!(
-            "DROP TABLE IF EXISTS {}",
-            quoted_identifier(name, &ProvenanceLog::default())
-        );
+        let log = ProvenanceLog::default();
+        let drop = match (
+            statement.starts_with(MACRO_PREFIX),
+            self.tables.contains_key(name),
+        ) {
+            (true, _) => format!("DROP TABLE IF EXISTS {}", quoted_identifier(name, &log)),
+            (false, true) => format!("DROP TABLE IF EXISTS {}", quoted_relation(name, &log)),
+            (false, false) => String::new(),
+        };
         self.connection
             .execute_batch(&format!("{drop}; {statement}"))
             .map_err(|error| PrintedSqlFailure {
@@ -55,12 +61,13 @@ impl Engine {
             .map(|(name, statement)| (name.clone(), statement.clone()))
             .collect();
         for (name, statement) in stale {
-            let quoted = quoted_identifier(&name, &ProvenanceLog::default());
+            let log = ProvenanceLog::default();
             let drop = match statement.starts_with(MACRO_PREFIX) {
                 true => {
+                    let quoted = quoted_identifier(&name, &log);
                     format!("DROP MACRO TABLE IF EXISTS {quoted}; DROP MACRO IF EXISTS {quoted}")
                 }
-                false => format!("DROP TABLE IF EXISTS {quoted}"),
+                false => format!("DROP TABLE IF EXISTS {}", quoted_relation(&name, &log)),
             };
             self.connection
                 .execute_batch(&drop)

crates/sqlquack-duckdb/src/checking/_helpers/lexing.rs

@@ -6,7 +6,7 @@ use std::collections::BTreeSet;
 pub(crate) struct Lexed {
     /// Statements separated by semicolons outside strings, quoted names and comments.
     pub(crate) statements: usize,
-    /// Every quoted identifier and bare word, any of which may name a relation or function.
+    /// Every quoted identifier, bare word and dotted chain of them, any of which may name a relation.
     pub(crate) names: BTreeSet<String>,
 }
 
@@ -15,17 +15,20 @@ pub(crate) fn lexed(sql: &str) -> Lexed {
     let characters: Vec<char> = sql.chars().collect();
     let mut names: BTreeSet<String> = BTreeSet::new();
     let (mut statements, mut pending) = (0_usize, false);
+    let (mut chain, mut continuing) = (String::new(), false);
     let mut index = 0;
     while let Some(&character) = characters.get(index) {
         let next = characters.get(index + 1).copied();
         match (character, next) {
             ('\'', _) => {
+                chain.clear();
                 index = quoted_end(&characters, index, '\'').0;
                 pending = true;
             }
             ('"', _) => {
                 let (end, name) = quoted_end(&characters, index, '"');
-                names.insert(name);
+                chain = extended(&chain, std::mem::take(&mut continuing), &name);
+                names.extend([chain.clone(), name]);
                 index = end;
                 pending = true;
             }
@@ -59,11 +62,20 @@ pub(crate) fn lexed(sql: &str) -> Lexed {
                 {
                     index += 1;
                 }
-                names.insert(characters[start..index].iter().collect());
+                let name: String = characters[start..index].iter().collect();
+                chain = extended(&chain, std::mem::take(&mut continuing), &name);
+                names.extend([chain.clone(), name]);
                 pending = true;
             }
+            ('.', _) if !chain.is_empty() && !continuing => {
+                continuing = true;
+                pending = true;
+                index += 1;
+            }
             (character, _) => {
                 pending |= !character.is_whitespace();
+                chain.clear();
+                continuing = false;
                 index += 1;
             }
         }
@@ -74,6 +86,14 @@ pub(crate) fn lexed(sql: &str) -> Lexed {
     }
 }
 
+/// The dotted chain `name` ends: `chain.name` when a dot joined it to `chain`, else `name` alone.
+fn extended(chain: &str, continuing: bool, name: &str) -> String {
+    match continuing && !chain.is_empty() {
+        true => format!("{chain}.{name}"),
+        false => name.to_owned(),
+    }
+}
+
 /// The index past the quoted text opening at `start`, and its text with doubled quotes undone.
 fn quoted_end(characters: &[char], start: usize, quote: char) -> (usize, String) {
     let mut text = String::new();

crates/sqlquack-duckdb/src/checking/_helpers/tables.rs

@@ -3,10 +3,12 @@
 use sqlquack_core::provenance::models::ProvenanceLog;
 use sqlquack_snowflake::typing::models::SnowflakeType;
 
+use crate::blocks::constants::NAME_SEPARATOR;
 use crate::blocks::main::column_type::column_type;
 use crate::blocks::main::quoted_identifier::quoted_identifier;
 use crate::checking::constants::MACRO_PREFIX;
 use crate::checking::constants::SEMI_STRUCTURED_HOLDING_TYPE;
+use crate::checking::main::quoted_relation::quoted_relation;
 use crate::checking::models::{FunctionStub, StubReturns};
 use crate::checking::types::{FunctionLookup, RelationLookup};
 
@@ -24,7 +26,7 @@ pub(crate) fn table_definitions(
     definitions
 }
 
-/// `CREATE TABLE` for the columns with a DuckDB type, semi-structured ones held as `JSON`.
+/// `CREATE TABLE` for the typed columns, semi-structured as `JSON`, in its name's catalog and schema.
 fn create_table(name: &str, columns: &[(String, SnowflakeType)]) -> String {
     let log = ProvenanceLog::default();
     let definitions: Vec<String> = columns
@@ -38,9 +40,21 @@ fn create_table(name: &str, columns: &[(String, SnowflakeType)]) -> String {
             Some(format!("{} {duckdb}", quoted_identifier(column, &log)))
         })
         .collect();
+    let parts: Vec<&str> = name.split(NAME_SEPARATOR).collect();
+    let mut prelude = String::new();
+    if let [catalog, _, _] = parts.as_slice() {
+        let catalog = quoted_identifier(catalog, &log);
+        prelude.push_str(&format!("ATTACH IF NOT EXISTS ':memory:' AS {catalog}; "));
+    }
+    if let [qualifiers @ .., _] = parts.as_slice()
+        && !qualifiers.is_empty()
+    {
+        let schema = quoted_relation(&qualifiers.join(NAME_SEPARATOR), &log);
+        prelude.push_str(&format!("CREATE SCHEMA IF NOT EXISTS {schema}; "));
+    }
     format!(
-        "CREATE TABLE {} ({})",
-        quoted_identifier(name, &log),
+        "{prelude}CREATE TABLE {} ({})",
+        quoted_relation(name, &log),
         definitions.join(", ")
     )
 }

crates/sqlquack-duckdb/src/checking/main/mod.rs

@@ -2,3 +2,4 @@
 
 pub mod bound_outside_lambdas;
 pub mod check_printed;
+pub mod quoted_relation;

crates/sqlquack-duckdb/src/checking/main/quoted_relation.rs

@@ -0,0 +1,14 @@
+//! `quoted_relation`: a relation name as the check creates it and converted SQL reads it.
+
+use sqlquack_core::provenance::models::ProvenanceLog;
+
+use crate::blocks::constants::NAME_SEPARATOR;
+use crate::blocks::main::quoted_identifier::quoted_identifier;
+
+/// A relation named `TABLE`, `SCHEMA.TABLE` or `CATALOG.SCHEMA.TABLE`, each part double-quoted.
+pub fn quoted_relation(name: &str, log: &ProvenanceLog) -> String {
+    name.split(NAME_SEPARATOR)
+        .map(|part| quoted_identifier(part, log))
+        .collect::<Vec<_>>()
+        .join(NAME_SEPARATOR)
+}

crates/sqlquack-policy/src/policy/constants.rs

@@ -12,6 +12,12 @@ pub const TIMEZONE_PARAMETER: &str = "TIMEZONE";
 /// The Snowflake session parameter making quoted identifiers resolve to upper case too.
 pub const QUOTED_IDENTIFIERS_IGNORE_CASE: &str = "QUOTED_IDENTIFIERS_IGNORE_CASE";
 
+/// The session's current database, which completes a relation name written without one.
+pub const DATABASE_PARAMETER: &str = "DATABASE";
+
+/// The session's current schema, which completes a relation name written without one.
+pub const SCHEMA_PARAMETER: &str = "SCHEMA";
+
 /// Session parameters that cannot change a result, accepted and ignored, each with why.
 pub const RESULT_NEUTRAL_SESSION_PARAMETERS: &[(&str, &str)] = &[
     ("QUERY_TAG", "a label recorded in query history"),

crates/sqlquack-rules/src/catalogue/constants.rs

@@ -85,7 +85,7 @@ pub const RULES: &[RuleSpec] = &[
     RuleSpec {
         id: FROM_RELATION,
         constructs: &["node:table"],
-        conditions: "The relation is a CTE or has typed columns in the schema.",
+        conditions: "The relation is a CTE or has typed columns in the schema. A schema relation may be declared as TABLE, SCHEMA.TABLE or DATABASE.SCHEMA.TABLE, and a name reads the declared relation whose parts agree with the parts it writes, the longest agreement winning; it is written as the declared name. A name matching several declared relations equally, a qualified name a CTE of its declared name would capture, and two differently written names reading one declared relation are refused.",
         failures: &[],
     },
     RuleSpec {

crates/sqlquack-rules/src/conversion/_helpers/queries/sources.rs

@@ -9,6 +9,7 @@ use sqlquack_core::refusals::main::outside_conditions::outside_conditions;
 use sqlquack_core::refusals::main::unresolved::unresolved;
 use sqlquack_core::refusals::models::{Refusal, RefusalCategory};
 use sqlquack_duckdb::blocks::main::quoted_identifier::quoted_identifier;
+use sqlquack_duckdb::checking::main::quoted_relation::quoted_relation;
 use sqlquack_snowflake::typing::models::SnowflakeType;
 
 use crate::conversion::_helpers::expressions::project_functions::{exact_arguments, held_type};
@@ -35,6 +36,8 @@ use crate::conversion::constants::{
 };
 use crate::conversion::constants::{FULL_JOIN_KEYWORD, INNER_JOIN_KEYWORD};
 use sqlquack_semantics::lineage::main::stored_columns::stored_columns;
+use sqlquack_semantics::names::constants::RELATION_KEY_SEPARATOR;
+use sqlquack_semantics::names::main::relation_key::relation_key;
 use sqlquack_semantics::scopes::models::ScopeRelation;
 use sqlquack_semantics::types::models::{ColumnSpec, FunctionReturns};
 
@@ -549,15 +552,48 @@ fn table_source(
     let log = context.log;
     let _rule = log.rule(FROM_RELATION);
     let table = payload_object(payload, &construct)?;
-    require_understood(table, &["name", "alias"], &construct)?;
+    require_understood(table, &["name", "alias", "catalog", "schema"], &construct)?;
     let name = identifier_name(field(table, "name", &construct)?)
         .ok_or_else(|| missing(construct.clone(), "a table has no name".to_owned()))?;
-    let columns = context.relation(&name)?;
+    let mut parts: Vec<String> = ["catalog", "schema"]
+        .iter()
+        .filter_map(|part| table.get(*part).and_then(identifier_name))
+        .collect();
+    parts.push(name.clone());
+    let written = parts.join(RELATION_KEY_SEPARATOR);
+    let is_cte = parts.len() == 1 && context.ctes.iter().any(|(cte, _)| cte == &name);
+    let options = context.options;
+    if let (1, Some(schema)) = (parts.len(), &options.current_schema) {
+        parts.insert(0, schema.clone());
+    }
+    if let (2, Some(database)) = (parts.len(), &options.current_database) {
+        parts.insert(0, database.clone());
+    }
+    let completed = parts.join(RELATION_KEY_SEPARATOR);
+    let key = match is_cte {
+        true => name.clone(),
+        false => relation_key(&parts, context.schema)?.unwrap_or_else(|| written.clone()),
+    };
+    if !is_cte && context.ctes.iter().any(|(cte, _)| cte == &key) {
+        return Err(unresolved(format!(
+            "`{written}` reads the schema relation `{key}`, which a CTE of the same name hides"
+        )));
+    }
+    let columns = context.relation(&key)?;
     let alias = table
         .get("alias")
         .and_then(identifier_name)
         .unwrap_or_else(|| name.clone());
-    let is_cte = context.ctes.iter().any(|(cte, _)| cte == &name);
+    if !is_cte && let Some(earlier) = context.qualifiers.read(&key, &completed) {
+        return Err(unresolved(format!(
+            "`{earlier}` and `{completed}` both read the schema relation `{key}`; declare each \
+             under its qualified name, or set the session's DATABASE and SCHEMA, to tell them apart"
+        )));
+    }
+    let relation_name = match is_cte {
+        true => quoted_identifier(&key, log),
+        false => quoted_relation(&key, log),
+    };
     let mut guarded: Vec<String> = Vec::new();
     for column in columns
         .iter()
@@ -567,11 +603,10 @@ fn table_source(
         guarded.push(format!("{} AS {quoted}", guarded_column(&quoted, context)?));
     }
     let relation = match guarded.is_empty() {
-        true => quoted_identifier(&name, log),
+        true => relation_name,
         false => format!(
-            "(SELECT * REPLACE ({}) FROM {})",
-            guarded.join(", "),
-            quoted_identifier(&name, log)
+            "(SELECT * REPLACE ({}) FROM {relation_name})",
+            guarded.join(", ")
         ),
     };
     let columns = match is_cte {

crates/sqlquack-rules/src/conversion/_helpers/support/scope.rs

@@ -170,6 +170,8 @@ pub(crate) struct Context<'a> {
     pub(crate) order: &'a OrderTrack,
     /// The subquery values emitted so far, which DuckDB cannot bind inside a lambda body.
     pub(crate) subqueries: &'a SubqueryTrack,
+    /// The name each schema relation was written as so far.
+    pub(crate) qualifiers: &'a QualifierTrack,
     /// Whether query results may keep unordered arrays: CTE and FROM subquery bodies.
     pub(crate) free_projections: bool,
     /// Whether this converts a case's input fixture, whose literals stand for source data.
@@ -204,6 +206,24 @@ impl OrderTrack {
 pub(crate) struct StatementTracks {
     pub(crate) order: OrderTrack,
     pub(crate) subqueries: SubqueryTrack,
+    pub(crate) qualifiers: QualifierTrack,
+}
+
+/// The name each schema relation was written as, so two written relations never read one.
+#[derive(Debug, Default)]
+pub(crate) struct QualifierTrack {
+    read: std::cell::RefCell<std::collections::HashMap<String, String>>,
+}
+
+impl QualifierTrack {
+    /// Records the relation `key` read as `written`; the earlier name when it was written otherwise.
+    pub(crate) fn read(&self, key: &str, written: &str) -> Option<String> {
+        let mut read = self.read.borrow_mut();
+        let earlier = read
+            .entry(key.to_owned())
+            .or_insert_with(|| written.to_owned());
+        (earlier != written).then(|| earlier.clone())
+    }
 }
 
 /// A subquery value emitted in a statement's conversion.

crates/sqlquack-rules/src/conversion/main/collect_refusals.rs

@@ -86,6 +86,7 @@ fn collected_conversion(
         typed: &typed,
         order: &tracks.order,
         subqueries: &tracks.subqueries,
+        qualifiers: &tracks.qualifiers,
         free_projections: false,
         fixture_body: false,
     };

crates/sqlquack-rules/src/conversion/main/convert_function_body.rs

@@ -42,6 +42,7 @@ pub fn convert_function_body(
         typed: &typed,
         order: &tracks.order,
         subqueries: &tracks.subqueries,
+        qualifiers: &tracks.qualifiers,
         free_projections: false,
         fixture_body: false,
     };

crates/sqlquack-rules/src/conversion/main/convert_statement.rs

@@ -47,6 +47,7 @@ pub fn convert_statement(
         typed: &typed,
         order: &tracks.order,
         subqueries: &tracks.subqueries,
+        qualifiers: &tracks.qualifiers,
         free_projections: false,
         fixture_body: false,
     };

crates/sqlquack-rules/src/conversion/models.rs

@@ -38,6 +38,10 @@ pub struct ConversionOptions {
     pub quoted_exclusions: bool,
     /// Whether quoted identifiers resolve to upper case, as `QUOTED_IDENTIFIERS_IGNORE_CASE = true`.
     pub quoted_identifiers_ignore_case: bool,
+    /// The session's current database, resolved, which completes a relation name without one.
+    pub current_database: Option<String>,
+    /// The session's current schema, resolved, which completes a relation name without one.
+    pub current_schema: Option<String>,
     /// User overrides by the name of the function each replaces, upper case.
     pub overrides: BTreeMap<String, UserOverride>,
     /// Override names to the SHA-256 of the body recorded cases verified, which strict mode allows.

crates/sqlquack-rules/src/session/_helpers/session_profiles.rs

@@ -2,7 +2,9 @@
 
 use std::collections::{BTreeMap, BTreeSet};
 
-use sqlquack_policy::policy::constants::RESULT_NEUTRAL_SESSION_PARAMETERS;
+use sqlquack_policy::policy::constants::{
+    DATABASE_PARAMETER, RESULT_NEUTRAL_SESSION_PARAMETERS, SCHEMA_PARAMETER,
+};
 use sqlquack_policy::policy::models::{Policy, SessionValue};
 
 /// The first profile matching `given`, unset ones as `default` pins, `ignored` never compared.
@@ -20,6 +22,7 @@ pub(crate) fn matching_profile(
         .map(|(name, _)| *name)
         .collect();
     let mut known: BTreeSet<&str> = neutral.clone();
+    known.extend([DATABASE_PARAMETER, SCHEMA_PARAMETER]);
     for profile in policy.session_profiles.values() {
         known.extend(
             profile