Add opt-in write-audit-publish for Snowflake table models

.changes/unreleased/Features-20260918-002011.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: Add opt-in write-audit-publish builds for native Snowflake SQL table models using same-schema working tables and a required passing data-test gate.
+time: 2026-09-18T00:20:11.745413-04:00
+custom:
+    author: ""
+    issue: ""
+    project: dbt-core

.github/workflows/ci.yml

@@ -122,6 +122,17 @@ jobs:
       - name: Run tests
         run: cargo nextest run --workspace
 
+      - name: Build dbt for offline WAP checks
+        run: cargo build -p dbt-sa-cli --bin dbt
+
+      - name: Run offline WAP Python checks
+        env:
+          DBT_WAP_TEST_BIN: ${{ github.workspace }}/target/debug/dbt
+          PYTHONDONTWRITEBYTECODE: "1"
+        run: |
+          test -x "$DBT_WAP_TEST_BIN"
+          python3 -m unittest discover -s crates/dbt-sa-cli/tests -p 'test_snowflake_wap_*.py' -v
+
       - name: sccache stats
         if: always()
         run: sccache --show-stats || true

crates/dbt-adapter-sql/src/ident.rs

@@ -263,11 +263,13 @@ pub fn quote_identifier(id: &str, backend: AdapterType) -> String {
 /// value can never terminate the literal early. The literal-value member of
 /// the family next to [`sanitize_identifier`] (strip) and [`quote_identifier`]
 /// (quote), which handle identifiers.
-pub fn escape_string_literal(s: &str, _backend: AdapterType) -> String {
-    // ANSI '' doubling — correct for every currently supported backend.
-    // Dialects with different literal-escape rules (e.g. backslash-escaped
-    // strings) grow a match arm on `_backend` when they arrive.
-    s.replace('\'', "''")
+pub fn escape_string_literal(s: &str, backend: AdapterType) -> String {
+    match backend {
+        // Snowflake interprets backslash escapes inside single-quoted literals.
+        // https://docs.snowflake.com/en/sql-reference/data-types-text#escape-sequences-in-single-quoted-string-constants
+        AdapterType::Snowflake => s.replace('\\', "\\\\").replace('\'', "''"),
+        _ => s.replace('\'', "''"),
+    }
 }
 
 #[cfg(test)]
@@ -301,6 +303,16 @@ mod sanitize_tests {
         assert_eq!(e("plain"), "plain");
     }
 
+    #[test]
+    fn snowflake_string_literals_preserve_backslashes_and_quotes() {
+        let escape = |s: &str| escape_string_literal(s, AdapterType::Snowflake);
+        assert_eq!(escape(r"path\name"), r"path\\name");
+        assert_eq!(escape(r"\'; DROP TABLE x; --"), r"\\''; DROP TABLE x; --");
+        assert_eq!(escape("trailing\\"), "trailing\\\\");
+        assert_eq!(escape("o'brien"), "o''brien");
+        assert_eq!(escape("plain"), "plain");
+    }
+
     #[test]
     fn duckdb_sanitize_strips_sql_metacharacters() {
         let s = |n: &str| sanitize_identifier(n, AdapterType::DuckDB);

crates/dbt-adapter/src/metadata/snowflake/mod.rs

@@ -25,6 +25,7 @@ use dbt_adapter_engine::{ConnectionFactory, MapReduce};
 use dbt_adbc::{Connection, QueryCtx};
 use dbt_common::AsyncAdapterResult;
 use dbt_common::ErrorCode;
+use dbt_common::FsResult;
 use dbt_common::cancellation::Cancellable;
 use dbt_common::cancellation::CancellationToken;
 use dbt_common::tracing::dbt_emit::{emit_debug_log_message, emit_warn_log_message};
@@ -44,6 +45,45 @@ use std::sync::Arc;
 
 const SNOWFLAKE_METADATA_NODE_ID: &str = "snowflake-metadata";
 
+/// Inspect an exact relation without depending on schema-list pagination limits.
+///
+/// Callers must filter the returned names for exact equality: `STARTS WITH` may
+/// also return longer names. Two rows let callers detect temporary-table
+/// shadowing. SHOW orders names lexicographically, so an exact match comes first.
+/// https://docs.snowflake.com/en/sql-reference/sql/show-objects
+pub fn relation_lookup_sql(relation: &dyn BaseRelation) -> FsResult<String> {
+    lookup_sql(relation, "OBJECTS")
+}
+
+/// Inspect the lifecycle and native-table flags not exposed by SHOW OBJECTS.
+/// https://docs.snowflake.com/en/sql-reference/sql/show-tables
+pub fn table_lookup_sql(relation: &dyn BaseRelation) -> FsResult<String> {
+    lookup_sql(relation, "TABLES")
+}
+
+fn lookup_sql(relation: &dyn BaseRelation, object_type: &str) -> FsResult<String> {
+    let metadata_error = |error| dbt_common::FsError::from_jinja_err(error, "WAP relation lookup");
+    let database = quote_identifier(
+        &relation
+            .database_as_resolved_str()
+            .map_err(metadata_error)?,
+        AdapterType::Snowflake,
+    );
+    let schema = quote_identifier(
+        &relation.schema_as_resolved_str().map_err(metadata_error)?,
+        AdapterType::Snowflake,
+    );
+    let identifier = escape_string_literal(
+        &relation
+            .identifier_as_resolved_str()
+            .map_err(metadata_error)?,
+        AdapterType::Snowflake,
+    );
+    Ok(format!(
+        "SHOW {object_type} IN {database}.{schema} STARTS WITH '{identifier}' LIMIT 2"
+    ))
+}
+
 fn metadata_warehouse_error(err: impl Display) -> AdapterError {
     AdapterError::new(AdapterErrorKind::Configuration, err.to_string())
 }
@@ -2100,6 +2140,44 @@ mod tests {
     use arrow_schema::{DataType, Field};
     use std::sync::Mutex;
 
+    #[test]
+    fn relation_lookup_resolves_unquoted_names_and_bounds_results() {
+        let relation = Relation::new(
+            AdapterType::Snowflake,
+            "database".to_string(),
+            "schema".to_string(),
+            "orders".to_string(),
+        )
+        .with_quoting(ResolvedQuoting::falses());
+        assert_eq!(
+            relation_lookup_sql(&relation).unwrap(),
+            "SHOW OBJECTS IN \"DATABASE\".\"SCHEMA\" STARTS WITH 'ORDERS' LIMIT 2"
+        );
+        assert_eq!(
+            table_lookup_sql(&relation).unwrap(),
+            "SHOW TABLES IN \"DATABASE\".\"SCHEMA\" STARTS WITH 'ORDERS' LIMIT 2"
+        );
+    }
+
+    #[test]
+    fn relation_lookup_preserves_quoted_case_and_escapes_literal_characters() {
+        let relation = Relation::new(
+            AdapterType::Snowflake,
+            "My\"Database".to_string(),
+            "My.Schema".to_string(),
+            r"O\'Brien".to_string(),
+        )
+        .with_quoting(ResolvedQuoting::trues());
+        assert_eq!(
+            relation_lookup_sql(&relation).unwrap(),
+            r#"SHOW OBJECTS IN "My""Database"."My.Schema" STARTS WITH 'O\\''Brien' LIMIT 2"#
+        );
+        assert_eq!(
+            table_lookup_sql(&relation).unwrap(),
+            r#"SHOW TABLES IN "My""Database"."My.Schema" STARTS WITH 'O\\''Brien' LIMIT 2"#
+        );
+    }
+
     struct FakeConnectionFactory {
         recycled: Arc<Mutex<u32>>,
     }

crates/dbt-adapter/src/relation/relation_impl.rs

@@ -14,7 +14,7 @@ use crate::relation::{RelationObject, StaticBaseRelation};
 use crate::value::none_value;
 
 use dbt_adapter_core::AdapterType;
-use dbt_adapter_sql::ident::max_identifier_length;
+use dbt_adapter_sql::ident::{max_identifier_length, quote_identifier};
 use dbt_common::{ErrorCode, FsResult, constants::DBT_CTE_PREFIX, fs_err};
 use dbt_frontend_common::ident::Identifier;
 use dbt_schema_store::CanonicalFqn;
@@ -285,7 +285,7 @@ impl BaseRelationProperties for Relation {
             }
         };
 
-        let schema = if self.quote_policy().database {
+        let schema = if self.quote_policy().schema {
             schema_str
         } else {
             match self.adapter_type {
@@ -295,7 +295,7 @@ impl BaseRelationProperties for Relation {
             }
         };
 
-        let ident = if self.quote_policy().database {
+        let ident = if self.quote_policy().identifier {
             ident_str
         } else {
             match self.adapter_type {
@@ -1176,6 +1176,18 @@ impl BaseRelation for Relation {
         }
     }
 
+    fn quoted(&self, component: &str) -> String {
+        match self.adapter_type {
+            AdapterType::Snowflake => quote_identifier(component, self.adapter_type),
+            _ => format!(
+                "{}{}{}",
+                self.quote_character(),
+                component,
+                self.quote_character()
+            ),
+        }
+    }
+
     fn render_self_as_str(&self) -> String {
         if self.adapter_type == AdapterType::DuckDB
             && let Some(external) = &self.external
@@ -1582,6 +1594,70 @@ mod tests {
             assert_eq!(relation.semantic_fqn(), "\"MyDB\".\"myschema\".\"MyTable\"");
         }
 
+        #[test]
+        fn relation_rendering_escapes_embedded_quotes_in_each_component() {
+            let relation = Relation::new(
+                AdapterType::Snowflake,
+                "db\"name".to_owned(),
+                "schema.with.dot".to_owned(),
+                "table\"name".to_owned(),
+            )
+            .with_quoting(Policy::enabled());
+            let expected = "\"db\"\"name\".\"schema.with.dot\".\"table\"\"name\"";
+            assert_eq!(relation.render_self_as_str(), expected);
+            assert_eq!(relation.semantic_fqn(), expected);
+
+            let mixed = relation.with_quoting(Policy {
+                database: true,
+                schema: false,
+                identifier: true,
+            });
+            assert_eq!(
+                mixed.render_self_as_str(),
+                "\"db\"\"name\".schema.with.dot.\"table\"\"name\""
+            );
+        }
+
+        #[test]
+        fn test_canonical_fqn_uses_each_component_quote_policy() {
+            for (adapter, normalized) in [
+                (AdapterType::Snowflake, ["MYDB", "MYSCHEMA", "MYTABLE"]),
+                (AdapterType::Postgres, ["mydb", "myschema", "mytable"]),
+            ] {
+                for (quoting, expected) in [
+                    (
+                        Policy {
+                            database: true,
+                            schema: false,
+                            identifier: false,
+                        },
+                        ["MyDB", normalized[1], normalized[2]],
+                    ),
+                    (
+                        Policy {
+                            database: false,
+                            schema: true,
+                            identifier: true,
+                        },
+                        [normalized[0], "MySchema", "MyTable"],
+                    ),
+                ] {
+                    let fqn = Relation::new(
+                        adapter,
+                        "MyDB".to_string(),
+                        "MySchema".to_string(),
+                        "MyTable".to_string(),
+                    )
+                    .with_quoting(quoting)
+                    .get_canonical_fqn()
+                    .unwrap();
+                    assert_eq!(fqn.catalog().as_str(), expected[0]);
+                    assert_eq!(fqn.schema().as_str(), expected[1]);
+                    assert_eq!(fqn.table().as_str(), expected[2]);
+                }
+            }
+        }
+
         fn filter_relation() -> Relation {
             Relation::new(
                 AdapterType::Postgres,

crates/dbt-compilation/src/schedule.rs

@@ -6,7 +6,7 @@ use dbt_common::node_selector::IndirectSelection;
 use dbt_common::path::DbtPath;
 use dbt_schemas::state::CacheState;
 
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct DbtCustomScheduleDescription {
     pub unique_ids: Vec<String>,
     pub include_parents: bool,
@@ -17,6 +17,8 @@ pub struct DbtCustomScheduleDescription {
     /// to reflect the semantics they want (e.g. `Empty` when the originating
     /// command does not execute tests).
     pub indirect_selection: IndirectSelection,
+    /// Retry rebuilds WAP models and reruns every audit of their fresh candidates.
+    pub is_retry: bool,
 }
 
 pub enum DbtScheduleDescription<'a> {