diff --git a/docs/configuration.md b/docs/configuration.md
index cb8fe058..e04b7301 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -175,6 +175,11 @@ VALUES
 `text[]`, and `lsn`, `xid`, `commit_ts`, `is_deleted` for metadata column names.
 See [Query destination data](destination-tables.md)
 
+`config_namespace` carries namespace defaults: `target_database`, `auto_create`,
+`auto_create_name` (the derived-name template, see
+[Name auto-created tables](destination-tables.md#name-auto-created-tables)) and
+`drop_table_strategy`
+
 Use `config_column` to configure individual columns. Set `namespace` to schema
 name, `relname` to table name, and `attname` to column name. Use `match` for
 name patterns and `target_type` to choose a ClickHouse type. Set `nan`,
diff --git a/docs/destination-tables.md b/docs/destination-tables.md
index a6675ec6..d559c0a8 100644
--- a/docs/destination-tables.md
+++ b/docs/destination-tables.md
@@ -2,10 +2,48 @@
 
 walshadow creates ClickHouse tables in configured `[ch] database` unless a
 namespace or table rule chooses another database. Destination table name
-defaults to source relation name
+defaults to `<source namespace>_<source relation>`, so `public.orders` lands as
+`public_orders` and `audit.orders` as `audit_orders` — same-named tables from
+different PostgreSQL schemas never collide
 
-Map same-named tables from different PostgreSQL schemas to distinct ClickHouse
-tables or databases to avoid collisions
+## Name auto-created tables
+
+`auto_create_name` sets the derived name per namespace as a template over
+`$database$` (source database), `$schema$` (source namespace) and `$table$`
+(source relation). Default is `$schema$_$table$`
+
+```toml
+[namespace.audit]
+auto_create = true
+auto_create_name = "$database$_$schema$_$table$"   # audit.orders -> app_audit_orders
+
+[namespace.shop]
+auto_create = true
+auto_create_name = "$table$_v2"                    # shop.orders -> orders_v2
+
+[namespace.public]
+auto_create = true
+auto_create_name = "$table$"                       # public.orders -> orders
+```
+
+A template naming no `$table$` is read as a prefix, so `wh_` means
+`wh_$table$` and `""` means `$table$`. Write `$$` for a literal `$`. An unknown
+placeholder or an unterminated `$` is a config error, and in the
+`config_namespace` overlay it is rejected with a warning, leaving the previous
+value in place
+
+| `auto_create_name` | `app`, `audit.orders` becomes |
+|---|---|
+| unset | `audit_orders` |
+| `$schema$_` | `audit_orders` |
+| `$database$_$schema$_$table$` | `app_audit_orders` |
+| `$table$_v2` | `orders_v2` |
+| `wh_` | `wh_orders` |
+| `""` or `$table$` | `orders` |
+
+`target_table` names a destination outright and ignores the template. The
+template applies when the table is created, so changing it leaves existing
+destination tables where they are and starts writing to the new name
 
 ## Generated shape
 
diff --git a/docs/getting-started.md b/docs/getting-started.md
index d816bedc..42de8b76 100644
--- a/docs/getting-started.md
+++ b/docs/getting-started.md
@@ -97,7 +97,7 @@ Insert or update one selected source row, then query matching ClickHouse table
 ```bash
 clickhouse-client --host ch.internal --database cdc \
     --user default --password secret --query \
-    "SELECT * FROM users FINAL WHERE _is_deleted = 0 ORDER BY id"
+    "SELECT * FROM public_users FINAL WHERE _is_deleted = 0 ORDER BY id"
 ```
 
 `FINAL` resolves versions written during updates or replay. `_is_deleted = 0`
diff --git a/docs/limitations.md b/docs/limitations.md
index 0e0257b3..9134dc97 100644
--- a/docs/limitations.md
+++ b/docs/limitations.md
@@ -40,11 +40,12 @@ and workload limits before attaching
 - date and timestamp values are written without checking ClickHouse calendar
   limits. Values outside supported ranges can return incorrect dates or times
   without stopping replication
-- same-named tables from different PostgreSQL schemas or databases need
-  explicit destination mapping. Config parsing rejects conflicts between
-  explicit entries; `replicate_all` names destinations after source tables, so
-  the second table to claim one destination is refused with an error and left
-  unreplicated until it names its own `target_database` or `target_table`
+- same-named tables from different PostgreSQL databases share one destination
+  and merge their rows into it, silently. Derived names carry `$schema$`, so
+  schemas do not collide, but databases do. Separate them with
+  `target_database`, `target_table`, or an `auto_create_name` carrying
+  `$database$`, see
+  [Choose shared or separate destinations](multi-database.md)
 - `base_backup` and `object_store` table loads publish with staging-table swap, database must support `EXCHANGE TABLES`
 - backup rows inserted into staging do not fire destination materialized views, live rows copied back after swap can fire twice
 
diff --git a/docs/multi-database.md b/docs/multi-database.md
index 078e33de..0ed0c041 100644
--- a/docs/multi-database.md
+++ b/docs/multi-database.md
@@ -34,27 +34,33 @@ the key level decides which one a name means
 Startup checks reject database names missing from `pg_database`, so a typo
 causes an error instead of replicating a database nobody asked for
 
-## Use separate destinations
+## Choose shared or separate destinations
 
-Tables with matching names in different source databases use the same
-ClickHouse table by default. Parsing rejects this conflict:
+Tables with matching names in different source databases land in the same
+ClickHouse table, merging their rows there. Derived names carry `$schema$`, so
+same-named tables in different *schemas* already differ
 
-```text
-`database.billing.table.public.orders` and `table.public.orders` both write to
-cdc.orders, set a different target_database or target_table
-```
+Merging is allowed, not an error. It suits shards of one logical table, where
+every source has the same shape. Rows from different sources interleave by
+`_lsn`, and because `ReplacingMergeTree` keys on the sort key, two sources
+writing the same key values collapse onto each other — give them disjoint keys
+or separate destinations
 
-Set separate destinations for each database, schema, or table:
+Separate per database, schema, or table:
 
 ```toml
 [database.billing.namespace.public]
 target_database = "billing_cdc"
 ```
 
-Tables with matching names in different schemas need the same treatment, see
-[limitations](limitations.md). Config checks cover explicit entries;
-destinations `replicate_all` derives are claimed as each table is first seen,
-and a second claim on one destination is refused
+Or keep one ClickHouse database and separate the names, since `auto_create_name`
+reads `$database$`:
+
+```toml
+[namespace.public]
+auto_create = true
+auto_create_name = "$database$_$schema$_$table$"
+```
 
 ## Run `ctl` against one database
 
diff --git a/docs/quickstart.md b/docs/quickstart.md
index c0adb57e..4a4cf1be 100644
--- a/docs/quickstart.md
+++ b/docs/quickstart.md
@@ -50,7 +50,7 @@ docker compose -f docker/docker-compose.yml \
 docker compose -f docker/docker-compose.yml \
     -f docker/docker-compose.quickstart.yml exec clickhouse \
     clickhouse-client --query \
-    "SELECT id, name, email FROM walshadow.users FINAL ORDER BY id"
+    "SELECT id, name, email FROM walshadow.public_users FINAL ORDER BY id"
 ```
 
 Expected result includes updated address:
diff --git a/sql/runtime_config_install.sql b/sql/runtime_config_install.sql
index 3bf10c52..e3a7c956 100644
--- a/sql/runtime_config_install.sql
+++ b/sql/runtime_config_install.sql
@@ -34,6 +34,9 @@ CREATE TABLE IF NOT EXISTS :"walshadow_schema".config_namespace (
     namespace           text PRIMARY KEY,
     target_database     text,
     auto_create         boolean,
+    auto_create_name    text,        -- derived table name template over
+                                     -- $database$/$schema$/$table$; NULL
+                                     -- means '$schema$_$table$'
     drop_table_strategy text         -- overrides config_global for this namespace
 );
 
@@ -79,6 +82,7 @@ CREATE TABLE IF NOT EXISTS :"walshadow_schema".config_column (
 -- Additive upgrade: columns introduced after the initial config_table shape.
 -- CREATE TABLE IF NOT EXISTS above no-ops on an existing install, so an
 -- upgrading deployment re-runs this to gain the columns the newer daemon reads.
+ALTER TABLE :"walshadow_schema".config_namespace ADD COLUMN IF NOT EXISTS auto_create_name text;
 ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS replicate       boolean;
 ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS initial_load    text;
 ALTER TABLE :"walshadow_schema".config_table ADD COLUMN IF NOT EXISTS target_database text;
diff --git a/src/bin/stream/bootstrap.rs b/src/bin/stream/bootstrap.rs
index 32cf3d6e..4c9fa38c 100644
--- a/src/bin/stream/bootstrap.rs
+++ b/src/bin/stream/bootstrap.rs
@@ -1233,6 +1233,7 @@ pub(crate) async fn bootstrap_build_mapping(
             walshadow::ch_ddl::DdlConfig::from_resolved(
                 &snap,
                 emitter_cfg.database.clone(),
+                emitter_cfg.source.dbname.clone(),
                 emitter_cfg.soft_delete,
                 emitter_cfg.system_columns.clone(),
                 emitter_cfg.replicate_all,
diff --git a/src/bin/stream/runtime_cfg.rs b/src/bin/stream/runtime_cfg.rs
index 8f1d5faa..c4876527 100644
--- a/src/bin/stream/runtime_cfg.rs
+++ b/src/bin/stream/runtime_cfg.rs
@@ -122,8 +122,8 @@ pub(crate) async fn load_runtime_config(
     for row in client
         .query(
             &format!(
-                "SELECT namespace, target_database, auto_create, drop_table_strategy \
-                 FROM {s}.config_namespace"
+                "SELECT namespace, target_database, auto_create, auto_create_name, \
+                 drop_table_strategy FROM {s}.config_namespace"
             ),
             &[],
         )
@@ -136,6 +136,7 @@ pub(crate) async fn load_runtime_config(
             NamespaceRow {
                 target_database: row.get("target_database"),
                 auto_create: row.get("auto_create"),
+                auto_create_name: row.get("auto_create_name"),
                 drop_table_strategy: row.get("drop_table_strategy"),
             },
         );
diff --git a/src/bin/stream/session.rs b/src/bin/stream/session.rs
index e4e4d15c..b5ec5bd4 100644
--- a/src/bin/stream/session.rs
+++ b/src/bin/stream/session.rs
@@ -854,15 +854,11 @@ pub(crate) async fn run_session(
         let mut applicators: HashMap<Oid, walshadow::ch_ddl::DdlApplicator> = HashMap::new();
         let mut backfillers: HashMap<Oid, Arc<dyn walshadow::opt_in::Backfiller>> = HashMap::new();
         let mut resolvers: Vec<Arc<ConfigResolver>> = Vec::with_capacity(db_conns.len());
-        // One claim per destination across every database, so `replicate_all`
-        // cannot name one ClickHouse table from two source tables
-        let targets = Arc::new(walshadow::mapping::TargetOwners::default());
         for (i, conn) in db_conns.iter().enumerate() {
             let built = build_source_db(SourceDbInputs {
                 args,
                 conn,
                 primary: i == primary_index,
-                targets: &targets,
                 desc_log: &desc_logs[i],
                 spill_dir: if i == primary_index {
                     args.spill_dir.clone()
diff --git a/src/bin/stream/source_db.rs b/src/bin/stream/source_db.rs
index 8c02700e..9378dae3 100644
--- a/src/bin/stream/source_db.rs
+++ b/src/bin/stream/source_db.rs
@@ -26,7 +26,6 @@ pub(crate) struct SourceDbInputs<'a> {
     /// `[source] dbname`: the one database a cluster backup can load
     pub(crate) primary: bool,
     /// Destinations every followed database has claimed
-    pub(crate) targets: &'a Arc<walshadow::mapping::TargetOwners>,
     pub(crate) desc_log: &'a Arc<walshadow::desc_log::DescriptorLog>,
     /// Backfill ledger lives here, beside this database's descriptor log
     pub(crate) spill_dir: PathBuf,
@@ -138,6 +137,7 @@ pub(crate) async fn build_source_db(input: SourceDbInputs<'_>) -> anyhow::Result
     let ddl_cfg = walshadow::ch_ddl::DdlConfig::from_resolved(
         &config_rx.borrow(),
         cfg.database.clone(),
+        cfg.source.dbname.clone(),
         cfg.soft_delete,
         cfg.system_columns.clone(),
         cfg.replicate_all,
@@ -153,8 +153,7 @@ pub(crate) async fn build_source_db(input: SourceDbInputs<'_>) -> anyhow::Result
     .await
     .context("init DDL applicator")?
     .with_resolver(resolver.clone())
-    .with_oracle(input.oracle.clone())
-    .with_target_owners(conn.oid, input.targets.clone());
+    .with_oracle(input.oracle.clone());
     // Backfiller for `initial_load` opt-ins (COPY / backup-sourced):
     // own source session + CH tail per backfill or pass, spill-dir
     // ledger dedups restarts. Wired whenever the emitter runs, since an
diff --git a/src/config.rs b/src/config.rs
index 3e29a860..c71c9234 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -36,8 +36,8 @@ use crate::column_rules::{ColumnRule, ColumnRules, ColumnRulesBuilder, Substitut
 use crate::emit::ch_emitter::EmitterConfig;
 use crate::filter::shadow_relations::ShadowHeld;
 use crate::mapping::{
-    DropTableStrategy, MappingHandle, NamespaceMapping, Retype, TableMapping, TableTarget,
-    derive_columns_for_mapping, fold_diff_into_mapping,
+    DropTableStrategy, MappingHandle, NameTemplate, NamespaceMapping, Retype, TableMapping,
+    TableTarget, derive_columns_for_mapping, fold_diff_into_mapping,
 };
 use crate::runtime_config::{ConfigEvent, ConfigOverlay, TableRow};
 use crate::schema::{RelDescriptor, RelName, SchemaDiff};
@@ -460,7 +460,7 @@ impl ConfigResolver {
                 .unwrap_or_else(|| Self::target_db_for(&inner, &rel.namespace)),
             table: table_override
                 .or(settings.target_table)
-                .unwrap_or_else(|| rel.name.to_string()),
+                .unwrap_or_else(|| Self::target_table_for(&inner, &rel)),
         };
         let column_rules = self.tx.borrow().column_rules.clone();
         let columns = derive_columns_for_mapping(desc, &column_rules);
@@ -586,6 +586,25 @@ impl ConfigResolver {
     /// CH target database for a namespace: per-namespace override (overlay then
     /// TOML) else the global `[ch] database`. Mirrors
     /// [`crate::emit::ch_ddl::DdlConfig::target_database_for`].
+    fn target_table_for(inner: &MergeInputs, rel: &RelName) -> String {
+        let ns = &*rel.namespace;
+        inner
+            .overlay
+            .namespaces
+            .get(ns)
+            .and_then(|r| r.auto_create_name.as_deref())
+            .and_then(|raw| NameTemplate::parse(raw).ok())
+            .or_else(|| {
+                inner
+                    .base
+                    .namespaces
+                    .get(ns)
+                    .and_then(|m| m.auto_create_name.clone())
+            })
+            .unwrap_or_default()
+            .render(&inner.base.source.dbname, ns, &rel.name)
+    }
+
     fn target_db_for(inner: &MergeInputs, namespace: &str) -> String {
         inner
             .overlay
@@ -807,6 +826,15 @@ impl ConfigResolver {
             if let Some(v) = row.auto_create {
                 entry.auto_create = v;
             }
+            if let Some(v) = &row.auto_create_name {
+                match NameTemplate::parse(v) {
+                    Ok(t) => entry.auto_create_name = Some(t),
+                    Err(e) => {
+                        rejections += 1;
+                        tracing::warn!(target: "walshadow::config", namespace = %ns, error = %e, "config_namespace.auto_create_name rejected");
+                    }
+                }
+            }
             if let Some(v) = &row.drop_table_strategy {
                 if let Ok(s) = v.parse::<DropTableStrategy>() {
                     entry.drop_table_strategy = Some(s);
@@ -1036,6 +1064,7 @@ mod tests {
             "public".into(),
             NamespaceRow {
                 auto_create: Some(true),
+                auto_create_name: Some("$table$".into()),
                 target_database: Some("default".into()),
                 drop_table_strategy: Some("drop".into()),
             },
@@ -1066,6 +1095,11 @@ mod tests {
         let ns = r.namespaces.get("public").unwrap();
         assert!(ns.auto_create);
         assert_eq!(ns.target_database.as_deref(), Some("default"));
+        assert_eq!(
+            ns.auto_create_name,
+            Some(NameTemplate::parse("$table$").unwrap()),
+            "an overlay template reaches the resolved namespace"
+        );
         assert_eq!(ns.drop_table_strategy, Some(DropTableStrategy::Drop));
         let t = r.tables.get(&RelName::new("public", "events")).unwrap();
         assert_eq!(t.target, TableTarget::new("default", "events_v2"));
@@ -1247,6 +1281,7 @@ mod tests {
         let ddl = crate::emit::ch_ddl::DdlConfig::from_resolved(
             &r,
             "db".into(),
+            "app".into(),
             false,
             Arc::default(),
             false,
@@ -1296,6 +1331,7 @@ mod tests {
         let ddl = crate::emit::ch_ddl::DdlConfig::from_resolved(
             &r,
             "db".into(),
+            "app".into(),
             false,
             Arc::default(),
             false,
@@ -1360,8 +1396,16 @@ mod tests {
             &OptInState::default(),
             &ColumnRules::default(),
         );
-        DdlConfig::from_resolved(&r, "db".into(), false, Arc::default(), false, None)
-            .auto_create_namespaces
+        DdlConfig::from_resolved(
+            &r,
+            "db".into(),
+            "app".into(),
+            false,
+            Arc::default(),
+            false,
+            None,
+        )
+        .auto_create_namespaces
     }
 
     #[test]
@@ -1523,13 +1567,54 @@ mod tests {
         let snap = rx.borrow_and_update();
         let rel = RelName::new("public", "events");
         let t = snap.tables.get(&rel).expect("mapping present");
-        assert_eq!(t.target.table, "events", "target derived from descriptor");
+        assert_eq!(
+            t.target.table, "public_events",
+            "routing target must match the name ensure_ch_table created",
+        );
         assert!(!t.columns.is_empty(), "columns derived from descriptor");
         // Fenced routing map written for the decode pool.
         assert!(mapping.with(|m| m.contains_key(&rel)).await);
         assert_eq!(resolver.opt_in_total(), 1);
     }
 
+    #[tokio::test]
+    async fn opt_in_routing_follows_the_namespace_template() {
+        let base = EmitterConfig::from_toml_str(
+            "[ch]\n\
+             [namespace.audit]\nauto_create = true\nauto_create_name = \"$table$\"\n\
+             [namespace.shop]\nauto_create = true\nauto_create_name = \"wh_\"\n",
+        )
+        .unwrap();
+        let (resolver, mut rx) = ConfigResolver::new(
+            &base,
+            CliOverrides::default(),
+            None,
+            toml::Table::new(),
+            dummy_handles(),
+            ResolverBoot::default(),
+        );
+        for (ns, rel) in [("audit", "events"), ("shop", "items"), ("app", "logs")] {
+            resolver
+                .materialize_opt_in(&rel_desc(ns, rel), None, None)
+                .await;
+            assert!(rx.changed().await.is_ok());
+        }
+        let snap = rx.borrow_and_update();
+        let table_of = |ns: &str, rel: &str| {
+            snap.tables
+                .get(&RelName::new(ns, rel))
+                .map(|t| t.target.table.clone())
+                .unwrap_or_else(|| panic!("{ns}.{rel} unmapped"))
+        };
+        assert_eq!(table_of("audit", "events"), "events", "$table$ is bare");
+        assert_eq!(table_of("shop", "items"), "wh_items");
+        assert_eq!(
+            table_of("app", "logs"),
+            "app_logs",
+            "a namespace with no entry derives from its own name",
+        );
+    }
+
     #[tokio::test]
     async fn exclude_table_removes_mapping() {
         let base = EmitterConfig::from_toml_str(
diff --git a/src/emit/ch_ddl.rs b/src/emit/ch_ddl.rs
index d74e8bcb..5e974f14 100644
--- a/src/emit/ch_ddl.rs
+++ b/src/emit/ch_ddl.rs
@@ -36,7 +36,7 @@ use crate::decode::heap_decoder::{self, ColumnValue};
 use crate::emit::ch_emitter::EmitterConfig;
 use crate::mapping::{
     ColumnMapping, DropTableStrategy, MappingHandle, MappingSnapshot, NamespaceMapping, Retype,
-    RetypeKind, SystemColumns, TableMapping, TableTarget, TargetOwners, apply_column_rule,
+    RetypeKind, SystemColumns, TableMapping, TableTarget, apply_column_rule,
     derive_columns_for_mapping, fold_diff_into_mapping, retyped_target,
 };
 use crate::ops::oracle::{Oracle, OracleCell};
@@ -64,6 +64,9 @@ pub struct DdlConfig {
     /// CH database DDL targets when neither per-table mapping nor source
     /// namespace overrides the destination
     pub target_database: String,
+    /// Source database name, for `$database$` in a namespace's
+    /// `auto_create_name`
+    pub source_database: String,
     /// Per-namespace overrides, fallback to the global fields above when
     /// a namespace has none
     pub namespaces: HashMap<String, NamespaceMapping>,
@@ -86,6 +89,7 @@ impl DdlConfig {
     pub fn from_resolved(
         resolved: &ResolvedConfig,
         target_database: String,
+        source_database: String,
         soft_delete: bool,
         system: Arc<SystemColumns>,
         replicate_all: bool,
@@ -103,6 +107,7 @@ impl DdlConfig {
             replicate_all,
             runtime_config_schema,
             target_database,
+            source_database,
             namespaces: resolved.namespaces.clone(),
             soft_delete,
             system,
@@ -118,6 +123,14 @@ impl DdlConfig {
             .unwrap_or(&self.target_database)
     }
 
+    fn derived_table_for(&self, rel: &RelName) -> String {
+        self.namespaces
+            .get(&*rel.namespace)
+            .and_then(|n| n.auto_create_name.clone())
+            .unwrap_or_default()
+            .render(&self.source_database, &rel.namespace, &rel.name)
+    }
+
     fn drop_strategy_for(&self, namespace: &str) -> DropTableStrategy {
         self.namespaces
             .get(namespace)
@@ -139,7 +152,7 @@ impl DdlConfig {
             table: settings
                 .target_table
                 .clone()
-                .unwrap_or_else(|| rel.name.to_string()),
+                .unwrap_or_else(|| self.derived_table_for(rel)),
         }
     }
 
@@ -214,10 +227,6 @@ pub struct DdlApplicator {
     /// Shadow PG renderer for tier-3 fast defaults
     oracle: Option<Arc<Oracle>>,
     ensured_databases: HashSet<String>,
-    /// This applicator's source database and the destinations every followed
-    /// database has claimed. Unset for a single database, which owns every
-    /// destination it names
-    owner: Option<(u32, Arc<TargetOwners>)>,
     pub stats: DdlStats,
 }
 
@@ -246,7 +255,6 @@ impl DdlApplicator {
             resolver: None,
             oracle: None,
             ensured_databases: HashSet::new(),
-            owner: None,
             stats: DdlStats::default(),
         })
     }
@@ -263,34 +271,6 @@ impl DdlApplicator {
         self
     }
 
-    /// Refuse destinations another followed database already writes to
-    pub fn with_target_owners(mut self, db_oid: u32, owners: Arc<TargetOwners>) -> Self {
-        self.owner = Some((db_oid, owners));
-        self
-    }
-
-    /// `false` when another database owns `target`, ie nothing may be created
-    /// or mapped there
-    async fn claims_target(&mut self, target: &TableTarget, rel: &RelName) -> bool {
-        let Some((db_oid, owners)) = &self.owner else {
-            return true;
-        };
-        let Err((owner_db, owner_rel)) = owners.claim(target, *db_oid, rel).await else {
-            return true;
-        };
-        tracing::error!(
-            target: "walshadow::ch_ddl",
-            qname = %rel,
-            db_oid = *db_oid,
-            owner_qname = %owner_rel,
-            owner_db_oid = owner_db,
-            destination = %target.sql(),
-            "another source database already writes this destination; \
-             set target_database or target_table to separate them",
-        );
-        false
-    }
-
     pub fn config(&self) -> &DdlConfig {
         &self.config
     }
@@ -308,6 +288,7 @@ impl DdlApplicator {
         self.config = DdlConfig::from_resolved(
             &snap,
             self.config.target_database.clone(),
+            self.config.source_database.clone(),
             self.config.soft_delete,
             self.config.system.clone(),
             self.config.replicate_all,
@@ -347,10 +328,6 @@ impl DdlApplicator {
         // Mapped dest created from the mapping when missing; IF NOT EXISTS
         // no-ops an operator-managed table and re-creates after strategy=drop.
         if let Some(m) = self.mapping_for(&desc.rel_name).await {
-            if !self.claims_target(&m.target, &desc.rel_name).await {
-                self.stats.skipped += 1;
-                return Ok(());
-            }
             self.ensure_database(&m.target.database).await?;
             let settings = cfg.rules.settings(&desc.rel_name);
             let sql = render_create_table_from_mapping(desc, &m, &cfg.create_shape(&settings));
@@ -373,10 +350,6 @@ impl DdlApplicator {
             self.stats.skipped += 1;
             return Ok(());
         };
-        if !self.claims_target(&mapping.target, &desc.rel_name).await {
-            self.stats.skipped += 1;
-            return Ok(());
-        }
         self.ensure_database(&mapping.target.database).await?;
         self.execute(&sql).await?;
         self.stats.creates_applied += 1;
@@ -406,10 +379,6 @@ impl DdlApplicator {
             self.stats.skipped += 1;
             return Ok(false);
         };
-        if !self.claims_target(&target, &desc.rel_name).await {
-            self.stats.skipped += 1;
-            return Ok(false);
-        }
         self.ensure_database(&target.database).await?;
         self.execute(&sql).await?;
         self.stats.creates_applied += 1;
@@ -614,15 +583,7 @@ impl DdlApplicator {
             }
             _ => false,
         };
-        let effect = predict_route_effect(&self.plan_config(config), mapping, event, excluded)?;
-        // Trailing rows of the applying commit route through this prediction,
-        // so it may not name a destination another database owns
-        if let Some((rel, Some(m))) = &effect
-            && !self.claims_target(&m.target, rel).await
-        {
-            return Ok(None);
-        }
-        Ok(effect)
+        predict_route_effect(&self.plan_config(config), mapping, event, excluded)
     }
 
     fn plan_config(&self, frozen: Option<&ResolvedConfig>) -> DdlConfig {
@@ -631,6 +592,7 @@ impl DdlApplicator {
                 DdlConfig::from_resolved(
                     rc,
                     self.config.target_database.clone(),
+                    self.config.source_database.clone(),
                     self.config.soft_delete,
                     self.config.system.clone(),
                     self.config.replicate_all,
@@ -655,9 +617,6 @@ impl DdlApplicator {
     }
 
     async fn forget_mapping(&mut self, rel: &RelName) {
-        if let Some((db_oid, owners)) = &self.owner {
-            owners.release(*db_oid, rel).await;
-        }
         if let Some(r) = &self.resolver {
             r.forget_derived_mapping(rel).await;
         } else {
@@ -1167,7 +1126,7 @@ pub fn render_create_table_from_mapping(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::mapping::{ColumnMapping, TableMapping};
+    use crate::mapping::{ColumnMapping, NameTemplate, TableMapping};
     use std::sync::LazyLock;
 
     static SYS: LazyLock<Arc<SystemColumns>> = LazyLock::new(Arc::default);
@@ -1182,6 +1141,7 @@ mod tests {
             let ddl = DdlConfig::from_resolved(
                 &resolved,
                 config.database.clone(),
+                "app".into(),
                 false,
                 SYS.clone(),
                 false,
@@ -1229,6 +1189,78 @@ mod tests {
         assert!(!is_system_namespace("public", Some("walshadow")));
     }
 
+    fn namespaces_with_bare_prefix(cfg: &mut DdlConfig) {
+        cfg.namespaces.insert(
+            "audit".to_string(),
+            crate::mapping::NamespaceMapping {
+                target_database: None,
+                auto_create: true,
+                auto_create_name: Some(NameTemplate::parse("$table$").unwrap()),
+                drop_table_strategy: None,
+                initial_load: None,
+            },
+        );
+    }
+
+    #[test]
+    fn auto_create_name_names_derived_destinations_only() {
+        use crate::mapping::NamespaceMapping;
+        use ahash::{HashMap, HashMapExt};
+        let mut namespaces = HashMap::new();
+        namespaces.insert(
+            "public".to_string(),
+            NamespaceMapping {
+                target_database: None,
+                auto_create: true,
+                auto_create_name: Some(NameTemplate::parse("wh_").unwrap()),
+                drop_table_strategy: None,
+                initial_load: None,
+            },
+        );
+        let mut cfg = DdlConfig {
+            drop_table_strategy: DropTableStrategy::Retain,
+            auto_create_namespaces: HashSet::new(),
+            replicate_all: false,
+            runtime_config_schema: None,
+            target_database: "default".into(),
+            source_database: "app".into(),
+            namespaces,
+            soft_delete: false,
+            system: Arc::default(),
+            rules: Arc::default(),
+            column_rules: Arc::default(),
+        };
+        let derived = cfg.create_target(&TableRule::default(), &RelName::new("public", "orders"));
+        assert_eq!(derived.database, "default");
+        assert_eq!(
+            derived.table, "wh_orders",
+            "an explicit namespace prefix beats the namespace name",
+        );
+
+        let pinned = cfg.create_target(
+            &TableRule {
+                target_table: Some("orders_v2".into()),
+                ..TableRule::default()
+            },
+            &RelName::new("public", "orders"),
+        );
+        assert_eq!(
+            pinned.table, "orders_v2",
+            "an explicit target_table names the destination outright",
+        );
+
+        let unconfigured =
+            cfg.create_target(&TableRule::default(), &RelName::new("audit", "orders"));
+        assert_eq!(
+            unconfigured.table, "audit_orders",
+            "a namespace that sets no prefix derives from its own name",
+        );
+
+        namespaces_with_bare_prefix(&mut cfg);
+        let bare = cfg.create_target(&TableRule::default(), &RelName::new("audit", "orders"));
+        assert_eq!(bare.table, "orders", "an empty prefix opts out");
+    }
+
     #[test]
     fn per_namespace_target_and_drop_override_global() {
         use crate::mapping::NamespaceMapping;
@@ -1239,6 +1271,7 @@ mod tests {
             NamespaceMapping {
                 target_database: Some("warehouse".into()),
                 auto_create: true,
+                auto_create_name: None,
                 drop_table_strategy: Some(DropTableStrategy::Drop),
                 initial_load: None,
             },
@@ -1248,6 +1281,7 @@ mod tests {
             NamespaceMapping {
                 target_database: None,
                 auto_create: true,
+                auto_create_name: None,
                 drop_table_strategy: None,
                 initial_load: None,
             },
@@ -1258,6 +1292,7 @@ mod tests {
             replicate_all: false,
             runtime_config_schema: None,
             target_database: "default".into(),
+            source_database: "app".into(),
             namespaces,
             soft_delete: false,
             system: Arc::default(),
@@ -1284,6 +1319,7 @@ mod tests {
             replicate_all: false,
             runtime_config_schema: None,
             target_database: "default".into(),
+            source_database: "app".into(),
             namespaces: HashMap::new(),
             soft_delete: false,
             system: Arc::default(),
@@ -1377,6 +1413,7 @@ mod tests {
             replicate_all: false,
             runtime_config_schema: Some("walshadow".into()),
             target_database: "default".into(),
+            source_database: "app".into(),
             namespaces: ahash::HashMap::default(),
             soft_delete: false,
             system: Arc::default(),
@@ -1387,7 +1424,7 @@ mod tests {
         assert!(cfg.auto_creates(&events));
         assert_eq!(
             cfg.create_target(&cfg.rules.settings(&events), &events),
-            TableTarget::new("warehouse", "events_1")
+            TableTarget::new("warehouse", "app_events_1")
         );
         assert_eq!(
             cfg.declared_scope(&RelName::new("app", "events_audit")),
@@ -1405,7 +1442,7 @@ mod tests {
         let other = RelName::new("other", "t");
         assert_eq!(
             cfg.create_target(&cfg.rules.settings(&other), &other),
-            TableTarget::new("default", "t")
+            TableTarget::new("default", "other_t")
         );
         let settings = cfg.rules.settings(&other);
         assert_eq!(
@@ -2194,6 +2231,7 @@ mod tests {
             replicate_all: false,
             runtime_config_schema: None,
             target_database: "default".into(),
+            source_database: "app".into(),
             namespaces: ahash::HashMap::default(),
             soft_delete: false,
             system: Arc::default(),
@@ -2228,6 +2266,7 @@ mod tests {
             replicate_all: true,
             runtime_config_schema: None,
             target_database: "default".into(),
+            source_database: "app".into(),
             namespaces: ahash::HashMap::default(),
             soft_delete: false,
             system: Arc::default(),
@@ -2244,7 +2283,7 @@ mod tests {
         let empty = MappingSnapshot::default();
         let predicted = predict_route_effect(&cfg, &empty, &added, false).unwrap();
         assert!(
-            matches!(&predicted, Some((r, Some(m))) if &*r.name == "fresh" && m.target.table == "fresh"),
+            matches!(&predicted, Some((r, Some(m))) if &*r.name == "fresh" && m.target.table == "public_fresh"),
             "{predicted:?}"
         );
         assert!(
diff --git a/src/emit/ch_emitter.rs b/src/emit/ch_emitter.rs
index 8a495ccc..d4af70e9 100644
--- a/src/emit/ch_emitter.rs
+++ b/src/emit/ch_emitter.rs
@@ -735,30 +735,6 @@ fn reject_backup_loads(db: &str, entries: &DbEntries) -> Result<(), EmitterError
     Ok(())
 }
 
-/// Resolve ClickHouse destination to check for conflicting mappings
-/// Skip patterns and entries with replication disabled
-fn declared_target(
-    ch_database: &str,
-    schema_database: Option<&str>,
-    rel: &str,
-    t: &TablePatch,
-) -> Option<(String, String)> {
-    if t.replicate == Some(false)
-        || matches!(t.match_kind, Some(MatchKind::Glob | MatchKind::Regex))
-    {
-        return None;
-    }
-    let database = t
-        .target_database
-        .clone()
-        .or_else(|| schema_database.map(str::to_owned))
-        .unwrap_or_else(|| ch_database.to_owned());
-    Some((
-        database,
-        t.target_table.clone().unwrap_or_else(|| rel.to_owned()),
-    ))
-}
-
 #[derive(serde::Deserialize)]
 #[serde(default)]
 struct ChSection {
@@ -869,6 +845,7 @@ impl Default for StreamSection {
 struct NamespacePatch {
     target_database: Option<String>,
     auto_create: Option<bool>,
+    auto_create_name: Option<String>,
     #[serde(default, deserialize_with = "crate::toml_de::de_from_str")]
     drop_table_strategy: Option<DropTableStrategy>,
     #[serde(default, deserialize_with = "crate::toml_de::de_from_str")]
@@ -1068,62 +1045,32 @@ impl EmitterConfig {
             reject_backup_loads(db, entries)?;
         }
         let local = buckets.remove(&scope).unwrap_or_default();
-        // Entries of the scoped database read back as the operator wrote them
-        let local_ctx = |ns: &str, name: &str| {
-            if scope == default_db {
-                format!("table.{ns}.{name}")
-            } else {
-                format!("database.{scope}.table.{ns}.{name}")
-            }
-        };
         for (ns, n) in local.namespace {
+            let auto_create_name = n
+                .auto_create_name
+                .as_deref()
+                .map(crate::mapping::NameTemplate::parse)
+                .transpose()
+                .map_err(|e| {
+                    EmitterError::Config(format!("`namespace.{ns}.auto_create_name`: {e}"))
+                })?;
             out.namespaces.insert(
                 ns,
                 NamespaceMapping {
                     target_database: n.target_database,
                     auto_create: n.auto_create.unwrap_or(false),
+                    auto_create_name,
                     drop_table_strategy: n.drop_table_strategy,
                     initial_load: n.initial_load,
                 },
             );
         }
-        // Prevent processes from writing different source tables to one destination
-        let mut owners: HashMap<(String, String), String> = HashMap::new();
         for (ns, rels) in local.table {
             for (name, t) in rels {
-                let schema_db = out
-                    .namespaces
-                    .get(&ns)
-                    .and_then(|n| n.target_database.clone());
-                if let Some(target) =
-                    declared_target(&out.database, schema_db.as_deref(), &name, &t)
-                {
-                    owners.insert(target, local_ctx(&ns, &name));
-                }
                 out.fold_table_entry(&ns, name, t)?;
             }
         }
-        for (db, entries) in buckets {
-            for (ns, rels) in &entries.table {
-                let schema_db = entries
-                    .namespace
-                    .get(ns)
-                    .and_then(|n| n.target_database.as_deref());
-                for (name, t) in rels {
-                    let Some(target) = declared_target(&out.database, schema_db, name, t) else {
-                        continue;
-                    };
-                    let ctx = format!("database.{db}.table.{ns}.{name}");
-                    if let Some(owner) = owners.get(&target) {
-                        return Err(EmitterError::Config(format!(
-                            "`{ctx}` and `{owner}` both write to {}.{}, set a different \
-                             target_database or target_table",
-                            target.0, target.1
-                        )));
-                    }
-                    owners.insert(target, ctx);
-                }
-            }
+        for db in buckets.into_keys() {
             out.databases.push(db);
         }
         out.databases
@@ -1263,7 +1210,13 @@ impl EmitterConfig {
                     .and_then(|n| n.target_database.clone())
             })
             .unwrap_or_else(|| self.database.clone());
-        let table = rule.target_table.unwrap_or(name);
+        let table = rule.target_table.unwrap_or_else(|| {
+            self.namespaces
+                .get(ns)
+                .and_then(|n| n.auto_create_name.clone())
+                .unwrap_or_default()
+                .render(&self.source.dbname, ns, &name)
+        });
         self.tables.insert(
             rel.clone(),
             TableMapping {
@@ -2805,6 +2758,53 @@ mod tests {
         }
     }
 
+    #[test]
+    fn namespace_auto_create_name_parses() {
+        let parse = |toml: &str| {
+            EmitterConfig::from_toml_str(toml)
+                .expect("parses")
+                .namespaces
+                .get("public")
+                .expect("namespace parsed")
+                .auto_create_name
+                .clone()
+        };
+        let rendered = |toml: &str| {
+            parse(toml)
+                .unwrap_or_default()
+                .render("app", "public", "orders")
+        };
+        assert_eq!(
+            parse("[ch]\n[namespace.public]\nauto_create = true\n"),
+            None,
+            "an absent template derives from the namespace name",
+        );
+        for (template, want) in [
+            ("", "orders"),
+            ("wh_", "wh_orders"),
+            ("$schema$_", "public_orders"),
+            ("$table$_v2", "orders_v2"),
+            ("$database$_$schema$_$table$", "app_public_orders"),
+            ("$$lit$$_$table$", "$lit$_orders"),
+        ] {
+            let toml = format!(
+                "[ch]\n[namespace.public]\nauto_create = true\nauto_create_name = \"{template}\"\n"
+            );
+            assert_eq!(rendered(&toml), want, "template {template:?}");
+        }
+        for bad in ["$nope$_$table$", "$table$_$unclosed"] {
+            let toml = format!(
+                "[ch]\n[namespace.public]\nauto_create = true\nauto_create_name = \"{bad}\"\n"
+            );
+            let err = EmitterConfig::from_toml_str(&toml).expect_err("rejects");
+            let msg = err.to_string();
+            assert!(
+                msg.contains("namespace.public.auto_create_name"),
+                "{bad:?} names the key: {msg}"
+            );
+        }
+    }
+
     #[test]
     fn defaults_match_high_throughput_profile() {
         let c = EmitterConfig::from_toml_str("[ch]\nhost = \"h\"\n").expect("parses");
@@ -4097,8 +4097,8 @@ mod tests {
         );
         let rel = RelName::new("public", "foo");
         let t = c.tables.get(&rel).expect("mapping present");
-        // target_database/target_table omitted: [ch] database + source relname
-        assert_eq!(t.target, TableTarget::new("default", "foo"));
+        // target_database/target_table omitted: [ch] database + derived name
+        assert_eq!(t.target, TableTarget::new("default", "public_foo"));
         assert_eq!(t.columns.len(), 2);
         assert_eq!(t.columns[0].src_attnum, 1);
         assert_eq!(t.columns[1].target_type, "Nullable(String)");
@@ -4430,10 +4430,10 @@ mod tests {
         .unwrap();
         let dotted_ns = c.tables.get(&RelName::new("a.b", "c")).expect("a.b / c");
         let dotted_rel = c.tables.get(&RelName::new("a", "b.c")).expect("a / b.c");
-        assert_eq!(dotted_ns.target, TableTarget::new("default", "c"));
-        assert_eq!(dotted_rel.target, TableTarget::new("default", "b.c"));
+        assert_eq!(dotted_ns.target, TableTarget::new("default", "a.b_c"));
+        assert_eq!(dotted_rel.target, TableTarget::new("default", "a_b.c"));
         // Interpolation quotes the dot inside the identifier
-        assert_eq!(dotted_rel.target.sql(), "`default`.`b.c`");
+        assert_eq!(dotted_rel.target.sql(), "`default`.`a_b.c`");
     }
 
     /// Database prefixes preserve table identity within source database
@@ -4568,27 +4568,24 @@ mod tests {
         assert_eq!(c.databases, vec!["app".to_string(), "billing".to_string()]);
     }
 
-    /// Reject mappings that mix source databases in one ClickHouse table
+    /// Two source databases may share one ClickHouse table: rows merge there
     #[test]
-    fn cross_database_destination_collision_is_rejected() {
-        let msg = EmitterConfig::from_toml_str(
-            "[ch]\n\
+    fn cross_database_destination_collision_is_allowed() {
+        let shared = "[ch]\n\
              database = \"cdc\"\n\
              [source]\n\
              dbname = \"app\"\n\
              [table.public.orders]\n\
              replicate = true\n\
              [database.billing.table.public.orders]\n\
-             replicate = true\n",
-        )
-        .expect_err("collides")
-        .to_string();
-        assert!(
-            msg.contains("`database.billing.table.public.orders`"),
-            "{msg}"
+             replicate = true\n";
+        let c = EmitterConfig::from_toml_str(shared).expect("merging destinations is allowed");
+        assert_eq!(c.databases, vec!["app".to_string(), "billing".to_string()]);
+        assert_eq!(
+            c.table_entries.iter().map(|(r, ..)| r).collect::<Vec<_>>(),
+            vec![&RelName::new("public", "orders")],
+            "this process still scopes to its own database",
         );
-        assert!(msg.contains("`table.public.orders`"), "{msg}");
-        assert!(msg.contains("cdc.orders"), "{msg}");
         for distinct in [
             "target_database = \"billing_cdc\"",
             "target_table = \"billing_orders\"",
diff --git a/src/mapping.rs b/src/mapping.rs
index 4b8f2611..36bf4d9b 100644
--- a/src/mapping.rs
+++ b/src/mapping.rs
@@ -40,44 +40,6 @@ impl TableTarget {
     }
 }
 
-/// ClickHouse tables claimed by one source database. `replicate_all` names a
-/// destination after its source table, so two databases holding one table name
-/// would otherwise merge their rows into one destination
-#[derive(Debug, Default)]
-pub struct TargetOwners {
-    owned: RwLock<HashMap<TableTarget, (u32, RelName)>>,
-}
-
-impl TargetOwners {
-    /// Claim `target` for one database's relation, idempotent for the owner
-    /// `Err` names the database oid and relation already writing there
-    pub async fn claim(
-        &self,
-        target: &TableTarget,
-        db_oid: u32,
-        rel: &RelName,
-    ) -> Result<(), (u32, RelName)> {
-        let mut owned = self.owned.write().await;
-        let claimant = (db_oid, rel.clone());
-        match owned.get(target) {
-            Some(owner) if *owner == claimant => Ok(()),
-            Some(owner) => Err(owner.clone()),
-            None => {
-                owned.insert(target.clone(), claimant);
-                Ok(())
-            }
-        }
-    }
-
-    /// Release what one relation claimed, so a re-created table claims again
-    pub async fn release(&self, db_oid: u32, rel: &RelName) {
-        self.owned
-            .write()
-            .await
-            .retain(|_, (oid, owner)| (*oid, &*owner) != (db_oid, rel));
-    }
-}
-
 impl std::fmt::Display for TableTarget {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}.{}", self.database, self.table)
@@ -88,10 +50,94 @@ impl std::fmt::Display for TableTarget {
 pub struct NamespaceMapping {
     pub target_database: Option<String>,
     pub auto_create: bool,
+    pub auto_create_name: Option<NameTemplate>,
     pub drop_table_strategy: Option<DropTableStrategy>,
     pub initial_load: Option<InitialLoadMode>,
 }
 
+/// Destination name for a table no `target_table` names, as `$database$` /
+/// `$schema$` / `$table$` placeholders around literal text (`$$` is a literal
+/// `$`). A template naming no `$table$` is read as a prefix, so `wh_` means
+/// `wh_$table$` and `""` means `$table$`
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct NameTemplate(Vec<NamePart>);
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum NamePart {
+    Literal(String),
+    Database,
+    Schema,
+    Table,
+}
+
+impl Default for NameTemplate {
+    fn default() -> Self {
+        Self(vec![
+            NamePart::Schema,
+            NamePart::Literal("_".into()),
+            NamePart::Table,
+        ])
+    }
+}
+
+impl NameTemplate {
+    pub fn parse(src: &str) -> Result<Self, String> {
+        let mut parts = Vec::new();
+        let mut literal = String::new();
+        let mut rest = src;
+        while let Some(at) = rest.find('$') {
+            literal.push_str(&rest[..at]);
+            rest = &rest[at + 1..];
+            if let Some(tail) = rest.strip_prefix('$') {
+                literal.push('$');
+                rest = tail;
+                continue;
+            }
+            let Some(end) = rest.find('$') else {
+                return Err(format!(
+                    "`{src}`: unterminated `$`; write `$$` for a literal `$`"
+                ));
+            };
+            let part = match &rest[..end] {
+                "database" => NamePart::Database,
+                "schema" => NamePart::Schema,
+                "table" => NamePart::Table,
+                unknown => {
+                    return Err(format!(
+                        "`{src}`: unknown placeholder `${unknown}$`; expected `$database$`, `$schema$` or `$table$`"
+                    ));
+                }
+            };
+            if !literal.is_empty() {
+                parts.push(NamePart::Literal(std::mem::take(&mut literal)));
+            }
+            parts.push(part);
+            rest = &rest[end + 1..];
+        }
+        literal.push_str(rest);
+        if !literal.is_empty() {
+            parts.push(NamePart::Literal(literal));
+        }
+        if !parts.contains(&NamePart::Table) {
+            parts.push(NamePart::Table);
+        }
+        Ok(Self(parts))
+    }
+
+    pub fn render(&self, database: &str, schema: &str, table: &str) -> String {
+        let mut out = String::new();
+        for part in &self.0 {
+            match part {
+                NamePart::Literal(s) => out.push_str(s),
+                NamePart::Database => out.push_str(database),
+                NamePart::Schema => out.push_str(schema),
+                NamePart::Table => out.push_str(table),
+            }
+        }
+        out
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub enum DropTableStrategy {
     #[default]
@@ -476,38 +522,48 @@ fn quote_ident(name: &str) -> String {
 mod tests {
     use super::*;
 
-    /// One destination belongs to one source relation of one database
-    #[tokio::test]
-    async fn target_owners_refuse_a_second_claimant() {
-        let owners = TargetOwners::default();
-        let target = TableTarget::new("cdc", "orders");
-        let app = RelName::new("public", "orders");
-        let billing = RelName::new("billing", "orders");
-        owners.claim(&target, 16384, &app).await.expect("first");
-        owners
-            .claim(&target, 16384, &app)
-            .await
-            .expect("same claimant re-claims");
-        assert_eq!(
-            owners.claim(&target, 5, &app).await,
-            Err((16384, app.clone())),
-            "another database writing the same table"
-        );
+    #[test]
+    fn name_template_renders_placeholders_literals_and_escapes() {
+        let render = |t: &str| {
+            NameTemplate::parse(t)
+                .expect(t)
+                .render("app", "audit", "orders")
+        };
         assert_eq!(
-            owners.claim(&target, 16384, &billing).await,
-            Err((16384, app.clone())),
-            "another schema of one database writing the same table"
+            NameTemplate::default().render("app", "audit", "orders"),
+            "audit_orders"
         );
-        // A different destination is free, and a dropped one frees its own
-        owners
-            .claim(&TableTarget::new("cdc", "orders_v2"), 5, &app)
-            .await
-            .expect("distinct target");
-        owners.release(16384, &app).await;
-        owners
-            .claim(&target, 5, &billing)
-            .await
-            .expect("released target is claimable");
+        assert_eq!(render("$schema$_$table$"), "audit_orders");
+        assert_eq!(render("$database$_$schema$_$table$"), "app_audit_orders");
+        assert_eq!(render("$table$_suffix"), "orders_suffix");
+        assert_eq!(render("$table$"), "orders");
+        assert_eq!(render("$table$$table$"), "ordersorders");
+        assert_eq!(render("$$"), "$orders", "`$$` is a literal `$`");
+        assert_eq!(render("a$$b_$table$"), "a$b_orders");
+        assert_eq!(render("$$table$$"), "$table$orders");
+    }
+
+    #[test]
+    fn name_template_without_table_reads_as_a_prefix() {
+        let render = |t: &str| {
+            NameTemplate::parse(t)
+                .expect(t)
+                .render("app", "audit", "orders")
+        };
+        assert_eq!(render(""), "orders");
+        assert_eq!(render("wh_"), "wh_orders");
+        assert_eq!(render("$schema$_"), "audit_orders");
+        assert_eq!(render("$database$_"), "app_orders");
+    }
+
+    #[test]
+    fn name_template_rejects_unknown_and_unterminated_placeholders() {
+        let err = NameTemplate::parse("$relname$_$table$").expect_err("unknown");
+        assert!(err.contains("`$relname$`"), "{err}");
+        assert!(err.contains("$database$"), "names the alternatives: {err}");
+        let err = NameTemplate::parse("$table$_$half").expect_err("unterminated");
+        assert!(err.contains("unterminated"), "{err}");
+        assert!(err.contains("`$$`"), "points at the escape: {err}");
     }
 
     /// Parse a `[system_columns]` body the way `ConfigDocument` does
diff --git a/src/ops/control.rs b/src/ops/control.rs
index b1c63f7b..10d216ad 100644
--- a/src/ops/control.rs
+++ b/src/ops/control.rs
@@ -432,7 +432,12 @@ fn status_tables(root: &Table, db: &str, ch_database: &str) -> Vec<Value> {
             let database = block_str("target_database")
                 .or_else(|| namespace_str(root, db, ns, "target_database"))
                 .unwrap_or_else(|| ch_database.to_owned());
-            let table = block_str("target_table").unwrap_or_else(|| rel.to_owned());
+            let table = block_str("target_table").unwrap_or_else(|| {
+                namespace_str(root, db, ns, "auto_create_name")
+                    .and_then(|raw| crate::mapping::NameTemplate::parse(&raw).ok())
+                    .unwrap_or_default()
+                    .render(db, ns, rel)
+            });
             let initial_load = block_str("initial_load")
                 .or_else(|| namespace_str(root, db, ns, "initial_load"))
                 .unwrap_or_else(|| "none".into());
@@ -840,9 +845,12 @@ mod tests {
                 "[source]\ndbname = \"app\"\n\
                  [ch]\ndatabase = \"cdc\"\n\
                  [namespace.shop]\ntarget_database = \"warehouse\"\n\
+                 [namespace.audit]\nauto_create_name = \"$table$\"\n\
                  [table.public.orders]\nreplicate = true\ninitial_load = \"copy\"\n\
                  [table.public.audit]\nreplicate = false\n\
-                 [table.shop.items]\nreplicate = true\ntarget_table = \"line_items\"\n"
+                 [table.shop.items]\nreplicate = true\ntarget_table = \"line_items\"\n\
+                 [table.shop.prices]\nreplicate = true\n\
+                 [table.audit.events]\nreplicate = true\n"
             )
             .await
             .starts_with("OK")
@@ -863,12 +871,27 @@ mod tests {
                 .cloned()
                 .unwrap_or_else(|| panic!("{source} missing from {status}"))
         };
-        assert_eq!(tables.len(), 2, "{status}");
+        assert_eq!(tables.len(), 4, "{status}");
 
         let orders = entry("app.public.orders");
         assert_eq!(
             orders.get("destination_table").and_then(Value::as_str),
-            Some("cdc.orders")
+            Some("cdc.public_orders"),
+            "a derived name carries the source namespace as its prefix",
+        );
+        assert_eq!(
+            entry("app.shop.prices")
+                .get("destination_table")
+                .and_then(Value::as_str),
+            Some("warehouse.shop_prices"),
+            "the prefix follows the source namespace, not `public`",
+        );
+        assert_eq!(
+            entry("app.audit.events")
+                .get("destination_table")
+                .and_then(Value::as_str),
+            Some("cdc.events"),
+            "a `$table$` auto_create_name derives a bare name",
         );
         assert_eq!(
             orders.get("initial_load").and_then(Value::as_str),
diff --git a/src/runtime_config.rs b/src/runtime_config.rs
index 6e572080..726b1198 100644
--- a/src/runtime_config.rs
+++ b/src/runtime_config.rs
@@ -66,6 +66,9 @@ pub struct GlobalRow {
 pub struct NamespaceRow {
     pub target_database: Option<String>,
     pub auto_create: Option<bool>,
+    /// Raw `$database$`/`$schema$`/`$table$` template for derived destination
+    /// names, parsed at resolver merge
+    pub auto_create_name: Option<String>,
     pub drop_table_strategy: Option<String>,
 }
 
@@ -345,6 +348,7 @@ pub fn interpret(
                 row: NamespaceRow {
                     target_database: field_string(rel, &cols, "target_database"),
                     auto_create: field_bool(rel, &cols, "auto_create"),
+                    auto_create_name: field_string(rel, &cols, "auto_create_name"),
                     drop_table_strategy: field_string(rel, &cols, "drop_table_strategy"),
                 },
             })
diff --git a/tests/bootstrap_types_e2e.rs b/tests/bootstrap_types_e2e.rs
index 3c77d212..15d6b8bb 100644
--- a/tests/bootstrap_types_e2e.rs
+++ b/tests/bootstrap_types_e2e.rs
@@ -176,7 +176,7 @@ async fn direct_bootstrap_all_types_end_to_end() {
         let deadline = std::time::Instant::now() + Duration::from_secs(90);
         loop {
             let n = ch
-                .query("SELECT count() FROM default.all_types FINAL WHERE _is_deleted = 0")
+                .query("SELECT count() FROM default.public_all_types FINAL WHERE _is_deleted = 0")
                 .unwrap_or_default();
             if n == "1" {
                 break;
@@ -219,7 +219,7 @@ async fn direct_bootstrap_all_types_end_to_end() {
         for (expr, want) in &checks {
             let got = ch
                 .query(&format!(
-                    "SELECT toString({expr}) FROM default.all_types FINAL WHERE id = 1"
+                    "SELECT toString({expr}) FROM default.public_all_types FINAL WHERE id = 1"
                 ))
                 .unwrap_or_default();
             if &got != want {
diff --git a/tests/bootstrap_unavailable_extension_e2e.rs b/tests/bootstrap_unavailable_extension_e2e.rs
index 922cfd42..0a4131cd 100644
--- a/tests/bootstrap_unavailable_extension_e2e.rs
+++ b/tests/bootstrap_unavailable_extension_e2e.rs
@@ -203,7 +203,7 @@ async fn bootstrap_survives_extension_absent_from_oracle() {
         let deadline = Instant::now() + Duration::from_secs(90);
         loop {
             let n = ch
-                .query("SELECT count() FROM default.t FINAL WHERE _is_deleted = 0")
+                .query("SELECT count() FROM default.public_t FINAL WHERE _is_deleted = 0")
                 .unwrap_or_default();
             if n == "1" {
                 break;
@@ -221,7 +221,7 @@ async fn bootstrap_survives_extension_absent_from_oracle() {
         }
         // The oracle-routed enum column must still resolve.
         let feel = ch
-            .query("SELECT feel FROM default.t FINAL WHERE id = 1")
+            .query("SELECT feel FROM default.public_t FINAL WHERE id = 1")
             .context("read enum column from CH")?;
         if feel != "happy" {
             anyhow::bail!("enum column not decoded: feel = {feel:?}");
diff --git a/tests/bootstrap_window_wal_not_landed_raw_e2e.rs b/tests/bootstrap_window_wal_not_landed_raw_e2e.rs
index 1804e598..4811e554 100644
--- a/tests/bootstrap_window_wal_not_landed_raw_e2e.rs
+++ b/tests/bootstrap_window_wal_not_landed_raw_e2e.rs
@@ -194,7 +194,7 @@ async fn backup_window_wal_never_materialises_the_shadow() {
         // byte assertions below are about a bootstrap that actually finished
         fx::wait_for_ch_value(
             &ch,
-            "SELECT count() FROM default.big FINAL WHERE id = 2000001",
+            "SELECT count() FROM default.public_big FINAL WHERE id = 2000001",
             "1",
             Duration::from_secs(180),
         )
@@ -204,7 +204,7 @@ async fn backup_window_wal_never_materialises_the_shadow() {
             .context("source count")?;
         fx::wait_for_ch_value(
             &ch,
-            "SELECT count() FROM default.big FINAL WHERE _is_deleted = 0",
+            "SELECT count() FROM default.public_big FINAL WHERE _is_deleted = 0",
             src_count.trim(),
             Duration::from_secs(180),
         )
diff --git a/tests/catalog_ael_boundary.rs b/tests/catalog_ael_boundary.rs
index a6e38f6b..224a15cc 100644
--- a/tests/catalog_ael_boundary.rs
+++ b/tests/catalog_ael_boundary.rs
@@ -153,6 +153,7 @@ async fn run_boundary_ddl(held: Held<'_>, d_ddl: &str) -> Result<(), String> {
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/catalog_rewrite_e2e.rs b/tests/catalog_rewrite_e2e.rs
index 21da3936..dd5e6fe7 100644
--- a/tests/catalog_rewrite_e2e.rs
+++ b/tests/catalog_rewrite_e2e.rs
@@ -55,6 +55,7 @@ async fn catalog_rewrites_keep_shadow_replaying() {
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/common/inproc_harness.rs b/tests/common/inproc_harness.rs
index 6e333e49..18a14a6e 100644
--- a/tests/common/inproc_harness.rs
+++ b/tests/common/inproc_harness.rs
@@ -863,6 +863,7 @@ async fn build_pipeline_inner(
     let ddl_cfg = DdlConfig::from_resolved(
         &config_rx.borrow(),
         emitter_cfg.database.clone(),
+        emitter_cfg.source.dbname.clone(),
         emitter_cfg.soft_delete,
         emitter_cfg.system_columns.clone(),
         emitter_cfg.replicate_all,
diff --git a/tests/control_plane_e2e.rs b/tests/control_plane_e2e.rs
index be11b613..c7dcdb32 100644
--- a/tests/control_plane_e2e.rs
+++ b/tests/control_plane_e2e.rs
@@ -120,7 +120,7 @@ impl Harness {
         let ch = fx::ChServer::spawn(ch_tmp, ports.ch_tcp, ports.ch_http).context("spawn ch")?;
         ch.query("CREATE DATABASE IF NOT EXISTS demo")?;
         ch.query(
-            "CREATE OR REPLACE TABLE demo.users (\
+            "CREATE OR REPLACE TABLE demo.demo_users (\
                 id Int64, name String, email String,\
                 _lsn UInt64, _xid UInt32,\
                 _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool\
@@ -263,7 +263,7 @@ impl Harness {
             .context("daemon metrics endpoint never came up")?;
         // Seed row must be on CH before any drill runs.
         h.wait_ch(
-            "SELECT email FROM demo.users FINAL WHERE _is_deleted = 0 AND id = 1",
+            "SELECT email FROM demo.demo_users FINAL WHERE _is_deleted = 0 AND id = 1",
             "alice@seed",
             Duration::from_secs(30),
         )
@@ -658,10 +658,10 @@ fn gated() -> bool {
 }
 
 const USER_EMAIL: &str =
-    "SELECT argMax(email, _lsn) FROM demo.users WHERE _is_deleted = 0 AND id = 1";
+    "SELECT argMax(email, _lsn) FROM demo.demo_users WHERE _is_deleted = 0 AND id = 1";
 
 const SECOND_EMAIL: &str =
-    "SELECT argMax(email, _lsn) FROM demo.users WHERE _is_deleted = 0 AND id = 2";
+    "SELECT argMax(email, _lsn) FROM demo.demo_users WHERE _is_deleted = 0 AND id = 2";
 
 /// Socket-only, so one fixed number per role is enough (see `common/ports.rs`).
 /// Distinct from source and shadow because a drill can put all three sockets
@@ -787,7 +787,7 @@ async fn live_table_opt_in_auto_creates_on_reload() {
         )?;
         h.psql("INSERT INTO demo.gizmos VALUES (1, 'alpha')")?;
         assert_eq!(
-            h.ch_get("EXISTS TABLE demo.gizmos")?,
+            h.ch_get("EXISTS TABLE demo.demo_gizmos")?,
             "0",
             "gizmos must not exist on CH before opt-in",
         );
@@ -802,19 +802,22 @@ async fn live_table_opt_in_auto_creates_on_reload() {
         let deadline = Instant::now() + Duration::from_secs(45);
         while Instant::now() < deadline {
             h.psql("UPDATE demo.users SET email = 'tick@x' WHERE id = 1")?;
-            if h.ch_get("EXISTS TABLE demo.gizmos").unwrap_or_default() == "1" {
+            if h.ch_get("EXISTS TABLE demo.demo_gizmos")
+                .unwrap_or_default()
+                == "1"
+            {
                 created = true;
                 break;
             }
             tokio::time::sleep(Duration::from_millis(500)).await;
         }
         if !created {
-            bail!("opt-in never auto-created the CH table demo.gizmos");
+            bail!("opt-in never auto-created the CH table demo.demo_gizmos");
         }
 
         // Pre-opt-in row proves copy ran
         h.wait_ch(
-            "SELECT argMax(label, _lsn) FROM demo.gizmos WHERE _is_deleted = 0 AND id = 1",
+            "SELECT argMax(label, _lsn) FROM demo.demo_gizmos WHERE _is_deleted = 0 AND id = 1",
             "alpha",
             Duration::from_secs(20),
         )
@@ -823,7 +826,7 @@ async fn live_table_opt_in_auto_creates_on_reload() {
 
         h.psql("INSERT INTO demo.gizmos VALUES (2, 'beta')")?;
         h.wait_ch(
-            "SELECT argMax(label, _lsn) FROM demo.gizmos WHERE _is_deleted = 0 AND id = 2",
+            "SELECT argMax(label, _lsn) FROM demo.demo_gizmos WHERE _is_deleted = 0 AND id = 2",
             "beta",
             Duration::from_secs(15),
         )
@@ -880,7 +883,7 @@ async fn ch_password_rotation_via_ctl_keeps_streaming() {
         // of its own and has to carry the rotated credential too
         h.psql("UPDATE demo.users SET name = repeat('t', 200000) WHERE id = 1")?;
         h.wait_ch(
-            "SELECT length(argMax(name, _lsn)) FROM demo.users WHERE _is_deleted = 0 AND id = 1",
+            "SELECT length(argMax(name, _lsn)) FROM demo.demo_users WHERE _is_deleted = 0 AND id = 1",
             "200000",
             Duration::from_secs(30),
         )
@@ -1178,7 +1181,7 @@ async fn pause_apply_resume_reroutes_backlog_whole() {
         h.wait_ch(USER_EMAIL, "marker1@x", Duration::from_secs(15))
             .await?;
         assert_eq!(
-            h.ch_get("EXISTS TABLE demo.widgets")?,
+            h.ch_get("EXISTS TABLE demo.demo_widgets")?,
             "0",
             "widgets must not exist on CH before opt-in",
         );
@@ -1199,7 +1202,7 @@ async fn pause_apply_resume_reroutes_backlog_whole() {
 
         // Backlog xact plans after resume → routes under the new config.
         h.wait_ch(
-            "SELECT argMax(label, _lsn) FROM demo.widgets WHERE _is_deleted = 0 AND id = 2",
+            "SELECT argMax(label, _lsn) FROM demo.demo_widgets WHERE _is_deleted = 0 AND id = 2",
             "while-paused",
             Duration::from_secs(30),
         )
@@ -1209,7 +1212,7 @@ async fn pause_apply_resume_reroutes_backlog_whole() {
         // Post-resume stream continues under the same config.
         h.psql("INSERT INTO demo.widgets VALUES (3, 'post-resume')")?;
         h.wait_ch(
-            "SELECT argMax(label, _lsn) FROM demo.widgets WHERE _is_deleted = 0 AND id = 3",
+            "SELECT argMax(label, _lsn) FROM demo.demo_widgets WHERE _is_deleted = 0 AND id = 3",
             "post-resume",
             Duration::from_secs(15),
         )
@@ -1217,7 +1220,7 @@ async fn pause_apply_resume_reroutes_backlog_whole() {
 
         // id=1 was planned pre-pause under the old config: discarded stays
         // discarded, no partial re-plan.
-        let n = h.ch_get("SELECT count() FROM demo.widgets FINAL WHERE _is_deleted = 0")?;
+        let n = h.ch_get("SELECT count() FROM demo.demo_widgets FINAL WHERE _is_deleted = 0")?;
         assert_eq!(n, "2", "exactly the backlog + post-resume rows land");
 
         assert!(h.alive(), "daemon exited during pause/apply/resume");
@@ -1306,7 +1309,7 @@ async fn pause_publishes_frozen_frontier_lsns() {
 /// Newest version of one row, without FINAL so a merge cannot hide a stale
 /// duplicate.
 fn email_of(id: u32) -> String {
-    format!("SELECT argMax(email, _lsn) FROM demo.users WHERE _is_deleted = 0 AND id = {id}")
+    format!("SELECT argMax(email, _lsn) FROM demo.demo_users WHERE _is_deleted = 0 AND id = {id}")
 }
 
 /// Steps 2 to 7 of architecture/recovery.md against standby
diff --git a/tests/ddl_replicates.rs b/tests/ddl_replicates.rs
index b3a6375a..a55c4efb 100644
--- a/tests/ddl_replicates.rs
+++ b/tests/ddl_replicates.rs
@@ -215,6 +215,7 @@ async fn create_table_auto_replicates_in_namespace() {
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
@@ -313,6 +314,7 @@ async fn drop_table_strategy_drop_removes_dest() {
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
@@ -552,6 +554,7 @@ async fn auto_create_honors_per_namespace_target_database() {
         NamespaceMapping {
             target_database: Some("warehouse".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
@@ -695,17 +698,18 @@ async fn create_table_auto_replicates_from_toml_namespace() {
 
     let tbls = ch
         .query(
-            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'new_t'",
+            "SELECT name FROM system.tables \
+             WHERE database = 'walshadow_test' AND name = 's15toml_new_t'",
         )
         .expect("ch table existence");
     assert_eq!(
-        tbls, "new_t",
-        "TOML [namespace] auto_create must create the CH table"
+        tbls, "s15toml_new_t",
+        "TOML [namespace] auto_create must create the CH table, named from the namespace"
     );
 
     let body = ch
         .query(
-            "SELECT argMax(body, _lsn) FROM walshadow_test.new_t \
+            "SELECT argMax(body, _lsn) FROM walshadow_test.s15toml_new_t \
              WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch body");
diff --git a/tests/desc_log_e2e.rs b/tests/desc_log_e2e.rs
index ef74d94d..3bde430d 100644
--- a/tests/desc_log_e2e.rs
+++ b/tests/desc_log_e2e.rs
@@ -56,6 +56,7 @@ async fn prepared_ddl_drains_at_commit_prepared() {
         walshadow::mapping::NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/dirty_admission_e2e.rs b/tests/dirty_admission_e2e.rs
index 175cb282..fb04869a 100644
--- a/tests/dirty_admission_e2e.rs
+++ b/tests/dirty_admission_e2e.rs
@@ -67,6 +67,7 @@ async fn build_drill_with(
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/foreign_database_e2e.rs b/tests/foreign_database_e2e.rs
index ee628445..12a4b30f 100644
--- a/tests/foreign_database_e2e.rs
+++ b/tests/foreign_database_e2e.rs
@@ -52,6 +52,7 @@ async fn foreign_database_ddl_and_dml_never_reach_the_followed_output() {
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/optin_initial_load_none_e2e.rs b/tests/optin_initial_load_none_e2e.rs
index 739248b7..ff654800 100644
--- a/tests/optin_initial_load_none_e2e.rs
+++ b/tests/optin_initial_load_none_e2e.rs
@@ -145,11 +145,15 @@ async fn optin_initial_load_none_skips_snapshot_but_streams_cdc() {
         // knows it, then assert the snapshot rows never landed.
         let deadline = Instant::now() + Duration::from_secs(60);
         loop {
-            if ch.query("EXISTS TABLE default.t").unwrap_or_default() == "1" {
+            if ch
+                .query("EXISTS TABLE default.public_t")
+                .unwrap_or_default()
+                == "1"
+            {
                 break;
             }
             if Instant::now() >= deadline {
-                anyhow::bail!("CH table default.t never created");
+                anyhow::bail!("CH table default.public_t never created");
             }
             std::thread::sleep(Duration::from_millis(250));
         }
@@ -160,12 +164,12 @@ async fn optin_initial_load_none_skips_snapshot_but_streams_cdc() {
         )?;
         fx::wait_for_ch_value(
             &ch,
-            "SELECT name FROM default.t FINAL WHERE id = 100000",
+            "SELECT name FROM default.public_t FINAL WHERE id = 100000",
             "[1,2]",
             Duration::from_secs(60),
         )?;
         anyhow::ensure!(
-            ch.query("SELECT count() FROM default.t FINAL WHERE id <= 500")? == "0",
+            ch.query("SELECT count() FROM default.public_t FINAL WHERE id <= 500")? == "0",
             "initial_load=none loaded snapshot rows"
         );
 
@@ -178,7 +182,7 @@ async fn optin_initial_load_none_skips_snapshot_but_streams_cdc() {
         let deadline = Instant::now() + Duration::from_secs(60);
         loop {
             let n = ch
-                .query("SELECT count() FROM default.t FINAL WHERE id = 100001")
+                .query("SELECT count() FROM default.public_t FINAL WHERE id = 100001")
                 .unwrap_or_default();
             if n == "1" {
                 break;
diff --git a/tests/pending_capture_e2e.rs b/tests/pending_capture_e2e.rs
index ff02afb5..d04f8cd2 100644
--- a/tests/pending_capture_e2e.rs
+++ b/tests/pending_capture_e2e.rs
@@ -76,6 +76,7 @@ async fn build_drill(slot: fx::Ports, schema_sql: &str, app_name: &str) -> Drill
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/pgbench_acceptance.rs b/tests/pgbench_acceptance.rs
index 315f5031..1061b4ca 100644
--- a/tests/pgbench_acceptance.rs
+++ b/tests/pgbench_acceptance.rs
@@ -121,7 +121,7 @@ fn create_pgbench_ch_tables(ch: &fx::ChServer, database: &str) -> Result<()> {
     // catalog has no attnum=5 pre-ALTER); post-ALTER UPDATE WAL records
     // arrive with c=7 (item 1 attmissingval substitution).
     ch.query(&format!(
-        "CREATE OR REPLACE TABLE {database}.pgbench_accounts (\
+        "CREATE OR REPLACE TABLE {database}.public_pgbench_accounts (\
             aid Int32, bid Int32, abalance Int32, filler String, c Nullable(Int32),\
             _lsn UInt64, _xid UInt32,\
             _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool\
@@ -132,14 +132,14 @@ fn create_pgbench_ch_tables(ch: &fx::ChServer, database: &str) -> Result<()> {
     // comment: only pgbench_accounts.filler gets blank-padded; the
     // other three default to NULL).
     ch.query(&format!(
-        "CREATE OR REPLACE TABLE {database}.pgbench_branches (\
+        "CREATE OR REPLACE TABLE {database}.public_pgbench_branches (\
             bid Int32, bbalance Int32, filler Nullable(String),\
             _lsn UInt64, _xid UInt32,\
             _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool\
          ) ENGINE = ReplacingMergeTree(_lsn, _is_deleted) ORDER BY bid"
     ))?;
     ch.query(&format!(
-        "CREATE OR REPLACE TABLE {database}.pgbench_tellers (\
+        "CREATE OR REPLACE TABLE {database}.public_pgbench_tellers (\
             tid Int32, bid Int32, tbalance Int32, filler Nullable(String),\
             _lsn UInt64, _xid UInt32,\
             _commit_ts DateTime64(6, 'UTC'), _is_deleted Bool\
@@ -148,7 +148,7 @@ fn create_pgbench_ch_tables(ch: &fx::ChServer, database: &str) -> Result<()> {
     // pgbench_history has no PK; order by (tid, mtime, aid) for FINAL
     // merges, lean on `_lsn` for ReplacingMergeTree's version semantics.
     ch.query(&format!(
-        "CREATE OR REPLACE TABLE {database}.pgbench_history (\
+        "CREATE OR REPLACE TABLE {database}.public_pgbench_history (\
             tid Int32, bid Int32, aid Int32, delta Int32,\
             mtime DateTime64(6), filler Nullable(String),\
             _lsn UInt64, _xid UInt32,\
@@ -386,10 +386,14 @@ async fn run_ddl_intermix(
         //    0 history rows. The bootstrap drains asynchronously, so poll CH
         //    until it matches rather than racing an immediate assert.
         for (src_table, ch_table, expected) in [
-            ("pgbench_accounts", "default.pgbench_accounts", 100_000_u64),
-            ("pgbench_branches", "default.pgbench_branches", 1),
-            ("pgbench_tellers", "default.pgbench_tellers", 10),
-            ("pgbench_history", "default.pgbench_history", 0),
+            (
+                "pgbench_accounts",
+                "default.public_pgbench_accounts",
+                100_000_u64,
+            ),
+            ("pgbench_branches", "default.public_pgbench_branches", 1),
+            ("pgbench_tellers", "default.public_pgbench_tellers", 10),
+            ("pgbench_history", "default.public_pgbench_history", 0),
         ] {
             let src = psql_source(&source, &format!("SELECT count(*) FROM {src_table}"))?;
             anyhow::ensure!(
@@ -521,15 +525,27 @@ async fn run_ddl_intermix(
             "pgbench_tellers",
             "pgbench_history",
         ] {
-            let _ = ch.query(&format!("OPTIMIZE TABLE default.{t} FINAL"));
+            let _ = ch.query(&format!("OPTIMIZE TABLE default.public_{t} FINAL"));
         }
 
         // 16. Parity assertions per table.
         for (src_table, ch_table, sum_col) in [
-            ("pgbench_accounts", "default.pgbench_accounts", "abalance"),
-            ("pgbench_branches", "default.pgbench_branches", "bbalance"),
-            ("pgbench_tellers", "default.pgbench_tellers", "tbalance"),
-            ("pgbench_history", "default.pgbench_history", "delta"),
+            (
+                "pgbench_accounts",
+                "default.public_pgbench_accounts",
+                "abalance",
+            ),
+            (
+                "pgbench_branches",
+                "default.public_pgbench_branches",
+                "bbalance",
+            ),
+            (
+                "pgbench_tellers",
+                "default.public_pgbench_tellers",
+                "tbalance",
+            ),
+            ("pgbench_history", "default.public_pgbench_history", "delta"),
         ] {
             let src_count = psql_source(&source, &format!("SELECT count(*) FROM {src_table}"))?;
             let ch_count = ch.query(&format!(
@@ -571,7 +587,7 @@ async fn run_ddl_intermix(
         //     substitution path fired end-to-end), (b) no row has c set
         //     to anything other than 7 or NULL.
         let updated_count = ch.query(
-            "SELECT count() FROM default.pgbench_accounts FINAL \
+            "SELECT count() FROM default.public_pgbench_accounts FINAL \
              WHERE _is_deleted = 0 AND c IS NOT NULL",
         )?;
         anyhow::ensure!(
@@ -580,7 +596,7 @@ async fn run_ddl_intermix(
              default substitution did not fire (updated_count={updated_count})"
         );
         let off_value = ch.query(
-            "SELECT count() FROM default.pgbench_accounts FINAL \
+            "SELECT count() FROM default.public_pgbench_accounts FINAL \
              WHERE _is_deleted = 0 AND c IS NOT NULL AND c != 7",
         )?;
         anyhow::ensure!(
diff --git a/tests/runtime_config_e2e.rs b/tests/runtime_config_e2e.rs
index d525eada..0f612367 100644
--- a/tests/runtime_config_e2e.rs
+++ b/tests/runtime_config_e2e.rs
@@ -47,8 +47,8 @@
 //!    * Operator inserts `config_namespace (auto_create=true)`, no
 //!      `config_table` row and no TOML mapping.
 //!    * Source `CREATE TABLE` in the namespace + INSERT.
-//!    * Expect: the namespace flag alone auto-creates the CH table and the
-//!      row lands.
+//!    * Expect: the namespace flag alone auto-creates the CH table, named
+//!      `app_thing` off the source namespace, and the row lands.
 //!
 //! 8. `pre_opt_in_xact_discards_post_opt_in_routes`
 //!    * No TOML mapping, no `initial_load`: a row committed before the
@@ -175,19 +175,19 @@ async fn opt_in_via_config_table_replicates_new_table() {
 
     let tbls = ch
         .query(
-            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'events'",
+            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'app_events'",
         )
         .expect("ch table existence");
-    assert_eq!(tbls, "events", "opt-in must auto-create the CH table");
+    assert_eq!(tbls, "app_events", "opt-in must auto-create the CH table");
 
     let n = ch
-        .query("SELECT count() FROM walshadow_test.events FINAL WHERE _is_deleted = 0")
+        .query("SELECT count() FROM walshadow_test.app_events FINAL WHERE _is_deleted = 0")
         .expect("ch count");
     assert_eq!(n, "1", "post-opt-in insert must reach CH");
 
     let body = ch
         .query(
-            "SELECT argMax(body, _lsn) FROM walshadow_test.events \
+            "SELECT argMax(body, _lsn) FROM walshadow_test.app_events \
              WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch body");
@@ -377,7 +377,7 @@ async fn forward_decl_materializes_on_create_table() {
     // Parked: nothing materialised yet.
     let pre = ch
         .query(
-            "SELECT count() FROM system.tables WHERE database = 'walshadow_test' AND name = 'later'",
+            "SELECT count() FROM system.tables WHERE database = 'walshadow_test' AND name = 'app_later'",
         )
         .expect("ch system.tables");
     assert_eq!(pre, "0", "forward-decl must not create a CH table yet");
@@ -405,17 +405,17 @@ async fn forward_decl_materializes_on_create_table() {
 
     let tbls = ch
         .query(
-            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'later'",
+            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'app_later'",
         )
         .expect("ch table existence");
     assert_eq!(
-        tbls, "later",
+        tbls, "app_later",
         "CREATE TABLE must materialise the parked opt-in"
     );
 
     let body = ch
         .query(
-            "SELECT argMax(body, _lsn) FROM walshadow_test.later \
+            "SELECT argMax(body, _lsn) FROM walshadow_test.app_later \
              WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch body");
@@ -524,7 +524,7 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() {
     let mut n = String::new();
     while std::time::Instant::now() < deadline {
         n = ch
-            .query("SELECT count(DISTINCT id) FROM walshadow_test.inventory FINAL WHERE _is_deleted = 0")
+            .query("SELECT count(DISTINCT id) FROM walshadow_test.app_inventory FINAL WHERE _is_deleted = 0")
             .unwrap_or_default();
         if n == "5" {
             break;
@@ -538,7 +538,7 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() {
         .query(
             "SELECT argMax(name, _lsn), argMax(price, _lsn), argMax(added_at, _lsn), \
                     argMax(meta, _lsn) \
-             FROM walshadow_test.inventory WHERE _is_deleted = 0 AND id = 2",
+             FROM walshadow_test.app_inventory WHERE _is_deleted = 0 AND id = 2",
         )
         .expect("ch backfilled row");
     assert_eq!(bolt, "bolt\t12.5\t2024-01-02 03:04:06.000000\t{\"b\": 2}");
@@ -547,7 +547,7 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() {
     let big = ch
         .query(
             "SELECT length(argMax(blob, _lsn)), length(argMax(note, _lsn)) \
-             FROM walshadow_test.inventory WHERE _is_deleted = 0 AND id = 2",
+             FROM walshadow_test.app_inventory WHERE _is_deleted = 0 AND id = 2",
         )
         .expect("ch large values");
     assert_eq!(big, "30000\t120000");
@@ -557,7 +557,7 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() {
         .query(
             "SELECT argMaxIf(weight, _lsn, id = 2), argMaxIf(weight, _lsn, id = 3), \
                     argMaxIf(weight, _lsn, id = 4) \
-             FROM walshadow_test.inventory WHERE _is_deleted = 0",
+             FROM walshadow_test.app_inventory WHERE _is_deleted = 0",
         )
         .expect("ch numeric specials");
     assert_eq!(weights, "NaN\tInfinity\t-Infinity");
@@ -567,7 +567,7 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() {
         .query(
             "SELECT argMax(name, _lsn), isNull(argMax(added_at, _lsn)), \
                     isNull(argMax(meta, _lsn)) \
-             FROM walshadow_test.inventory WHERE _is_deleted = 0 AND id = 3",
+             FROM walshadow_test.app_inventory WHERE _is_deleted = 0 AND id = 3",
         )
         .expect("ch null row");
     assert_eq!(crate_row, "crate\t1\t1");
@@ -575,7 +575,7 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() {
     // Post-opt-in UPDATE (commit_lsn > S) beats the COPY baseline.
     let anvil = ch
         .query(
-            "SELECT argMax(name, _lsn) FROM walshadow_test.inventory \
+            "SELECT argMax(name, _lsn) FROM walshadow_test.app_inventory \
              WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch mutated row");
@@ -587,7 +587,7 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() {
     // Post-opt-in INSERT streams via WAL, no COPY involvement.
     let dowel = ch
         .query(
-            "SELECT argMax(name, _lsn) FROM walshadow_test.inventory \
+            "SELECT argMax(name, _lsn) FROM walshadow_test.app_inventory \
              WHERE _is_deleted = 0 AND id = 100",
         )
         .expect("ch streamed row");
@@ -670,7 +670,7 @@ async fn opt_in_then_alter_add_column_reaches_ch() {
     let qty_col = ch
         .query(
             "SELECT count() FROM system.columns \
-             WHERE database = 'walshadow_test' AND table = 'gadgets' AND name = 'qty'",
+             WHERE database = 'walshadow_test' AND table = 'app_gadgets' AND name = 'qty'",
         )
         .expect("ch column existence");
     assert_eq!(qty_col, "1", "post-opt-in ALTER must ADD COLUMN on CH");
@@ -679,7 +679,7 @@ async fn opt_in_then_alter_add_column_reaches_ch() {
     let post = ch
         .query(
             "SELECT argMax(name, _lsn), argMax(qty, _lsn) \
-             FROM walshadow_test.gadgets WHERE _is_deleted = 0 AND id = 2",
+             FROM walshadow_test.app_gadgets WHERE _is_deleted = 0 AND id = 2",
         )
         .expect("ch post-alter row");
     assert_eq!(post, "post-alter\t7");
@@ -688,7 +688,7 @@ async fn opt_in_then_alter_add_column_reaches_ch() {
     let pre = ch
         .query(
             "SELECT argMax(name, _lsn), isNull(argMax(qty, _lsn)) \
-             FROM walshadow_test.gadgets WHERE _is_deleted = 0 AND id = 1",
+             FROM walshadow_test.app_gadgets WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch pre-alter row");
     assert_eq!(pre, "pre-alter\t1");
@@ -766,17 +766,18 @@ async fn auto_create_namespace_via_config_namespace() {
 
     let tbls = ch
         .query(
-            "SELECT name FROM system.tables WHERE database = 'walshadow_test' AND name = 'thing'",
+            "SELECT name FROM system.tables \
+             WHERE database = 'walshadow_test' AND name = 'app_thing'",
         )
         .expect("ch table existence");
     assert_eq!(
-        tbls, "thing",
-        "config_namespace.auto_create must create the CH table"
+        tbls, "app_thing",
+        "config_namespace.auto_create must create the CH table, named from the namespace"
     );
 
     let body = ch
         .query(
-            "SELECT argMax(body, _lsn) FROM walshadow_test.thing \
+            "SELECT argMax(body, _lsn) FROM walshadow_test.app_thing \
              WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch body");
@@ -982,18 +983,18 @@ async fn pre_opt_in_xact_discards_post_opt_in_routes() {
     assert!(discarded >= 1, "pre-opt-in xact must be a counted discard");
 
     let n = ch
-        .query("SELECT count() FROM walshadow_test.metrics FINAL WHERE _is_deleted = 0")
+        .query("SELECT count() FROM walshadow_test.app_metrics FINAL WHERE _is_deleted = 0")
         .expect("ch count");
     assert_eq!(n, "1", "exactly the post-opt-in row lands");
 
     let gone = ch
-        .query("SELECT count() FROM walshadow_test.metrics WHERE id = 1")
+        .query("SELECT count() FROM walshadow_test.app_metrics WHERE id = 1")
         .expect("ch pre-opt-in row");
     assert_eq!(gone, "0", "pre-opt-in row must never reach CH");
 
     let v = ch
         .query(
-            "SELECT argMax(v, _lsn) FROM walshadow_test.metrics \
+            "SELECT argMax(v, _lsn) FROM walshadow_test.app_metrics \
              WHERE _is_deleted = 0 AND id = 2",
         )
         .expect("ch v");
@@ -1072,7 +1073,7 @@ async fn pattern_row_scopes_tables_by_glob() {
 
     let body = ch
         .query(
-            "SELECT argMax(body, _lsn) FROM walshadow_test.events_2026 \
+            "SELECT argMax(body, _lsn) FROM walshadow_test.app_events_2026 \
              WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch body");
@@ -1166,13 +1167,13 @@ async fn opt_in_row_pins_order_by_and_primary_key() {
     pipeline.shutdown().await.expect("pipeline drains clean");
 
     let ddl = ch
-        .query("SHOW CREATE TABLE walshadow_test.keyed")
+        .query("SHOW CREATE TABLE walshadow_test.app_keyed")
         .expect("show create");
     assert!(ddl.contains("ORDER BY (tenant, id)"), "{ddl}");
     assert!(ddl.contains("PRIMARY KEY (tenant)"), "{ddl}");
 
     let n = ch
-        .query("SELECT count() FROM walshadow_test.keyed FINAL WHERE _is_deleted = 0")
+        .query("SELECT count() FROM walshadow_test.app_keyed FINAL WHERE _is_deleted = 0")
         .expect("ch count");
     assert_eq!(n, "1", "post-opt-in insert must reach CH");
 }
@@ -1251,7 +1252,7 @@ async fn pattern_row_shapes_auto_created_tables() {
     pipeline.shutdown().await.expect("pipeline drains clean");
 
     let ddl = ch
-        .query("SHOW CREATE TABLE walshadow_test.events_2026")
+        .query("SHOW CREATE TABLE walshadow_test.app_events_2026")
         .expect("show create");
     assert!(ddl.contains("`_peerdb_version` UInt64"), "{ddl}");
     assert!(!ddl.contains("_is_deleted"), "marker dropped: {ddl}");
@@ -1259,13 +1260,15 @@ async fn pattern_row_shapes_auto_created_tables() {
 
     // A relation the pattern misses keeps the cluster-wide names
     let other = ch
-        .query("SHOW CREATE TABLE walshadow_test.orders")
+        .query("SHOW CREATE TABLE walshadow_test.app_orders")
         .expect("show create");
     assert!(other.contains("`_lsn` UInt64"), "{other}");
     assert!(other.contains("_is_deleted"), "{other}");
 
     let body = ch
-        .query("SELECT argMax(body, _peerdb_version) FROM walshadow_test.events_2026 WHERE id = 1")
+        .query(
+            "SELECT argMax(body, _peerdb_version) FROM walshadow_test.app_events_2026 WHERE id = 1",
+        )
         .expect("ch body");
     assert_eq!(body, "shaped", "rows INSERT under the renamed columns");
 }
@@ -1372,7 +1375,7 @@ async fn copy_load_resolves_unchanged_toast_after_update() {
             "SELECT pg_switch_wal()".into(),
         ],
         "walshadow-copy-toast",
-        "SELECT count() FROM walshadow_test.docs FINAL WHERE _is_deleted = 0",
+        "SELECT count() FROM walshadow_test.app_docs FINAL WHERE _is_deleted = 0",
         "2",
     )
     .await;
@@ -1381,7 +1384,7 @@ async fn copy_load_resolves_unchanged_toast_after_update() {
     let doc = ch
         .query(
             "SELECT tag, length(big), substring(big, 1, 32), substring(big, 6369, 32) \
-             FROM walshadow_test.docs FINAL WHERE _is_deleted = 0 AND id = 1",
+             FROM walshadow_test.app_docs FINAL WHERE _is_deleted = 0 AND id = 1",
         )
         .expect("ch updated row");
     assert_eq!(
@@ -1416,7 +1419,7 @@ async fn copy_load_pins_output_settings() {
         ],
         "walshadow-copy-settings",
         "SELECT groupArray(span) FROM \
-         (SELECT span FROM walshadow_test.spans FINAL WHERE _is_deleted = 0 ORDER BY id)",
+         (SELECT span FROM walshadow_test.app_spans FINAL WHERE _is_deleted = 0 ORDER BY id)",
         "['1 day 02:03:04','1 day 02:03:04']",
     )
     .await;
diff --git a/tests/schema_evolution_cdc.rs b/tests/schema_evolution_cdc.rs
index aab9427f..1272b399 100644
--- a/tests/schema_evolution_cdc.rs
+++ b/tests/schema_evolution_cdc.rs
@@ -50,6 +50,7 @@ async fn run(
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/shadow_stays_catalog_scale_e2e.rs b/tests/shadow_stays_catalog_scale_e2e.rs
index ae8959f9..a76e5a0d 100644
--- a/tests/shadow_stays_catalog_scale_e2e.rs
+++ b/tests/shadow_stays_catalog_scale_e2e.rs
@@ -223,7 +223,7 @@ async fn user_relation_never_materialises_on_the_shadow() {
         let deadline = Instant::now() + Duration::from_secs(180);
         loop {
             let n = ch
-                .query("SELECT count() FROM default.big")
+                .query("SELECT count() FROM default.public_big")
                 .unwrap_or_default();
             if n == ROWS.to_string() {
                 break;
@@ -272,7 +272,7 @@ async fn user_relation_never_materialises_on_the_shadow() {
         let deadline = Instant::now() + Duration::from_secs(180);
         loop {
             let n = ch
-                .query("SELECT count() FROM default.big FINAL WHERE id = 2000001")
+                .query("SELECT count() FROM default.public_big FINAL WHERE id = 2000001")
                 .unwrap_or_default();
             if n == "1" {
                 break;
diff --git a/tests/source_dbname_resolution_e2e.rs b/tests/source_dbname_resolution_e2e.rs
index 8bdbed51..50ac5a84 100644
--- a/tests/source_dbname_resolution_e2e.rs
+++ b/tests/source_dbname_resolution_e2e.rs
@@ -204,9 +204,12 @@ async fn shadow_catalog_follows_applied_source_dbname() {
 
         let deadline = Instant::now() + Duration::from_secs(90);
         loop {
-            if ch.query("EXISTS TABLE default.dup").unwrap_or_default() == "1"
+            if ch
+                .query("EXISTS TABLE default.public_dup")
+                .unwrap_or_default()
+                == "1"
                 && ch
-                    .query("SELECT b FROM default.dup FINAL WHERE a = 1")
+                    .query("SELECT b FROM default.public_dup FINAL WHERE a = 1")
                     .unwrap_or_default()
                     == "from-duptest"
             {
@@ -216,7 +219,8 @@ async fn shadow_catalog_follows_applied_source_dbname() {
                 Instant::now() < deadline,
                 "bootstrap did not load duptest's row (saw {:?}); the shadow \
                  catalog is resolved against the wrong database",
-                ch.query("SELECT b FROM default.dup FINAL WHERE a = 1").ok(),
+                ch.query("SELECT b FROM default.public_dup FINAL WHERE a = 1")
+                    .ok(),
             );
             std::thread::sleep(Duration::from_millis(200));
         }
@@ -238,7 +242,7 @@ async fn shadow_catalog_follows_applied_source_dbname() {
         let deadline = Instant::now() + Duration::from_secs(90);
         loop {
             let mine = ch
-                .query("SELECT b FROM default.dup FINAL WHERE a = 2")
+                .query("SELECT b FROM default.public_dup FINAL WHERE a = 2")
                 .unwrap_or_default();
             anyhow::ensure!(
                 mine != "cdc-postgres",
@@ -272,7 +276,7 @@ async fn shadow_catalog_follows_applied_source_dbname() {
              database's pre-boot row must not appear"
         );
         anyhow::ensure!(
-            ch.query("SELECT count() FROM default.dup FINAL")
+            ch.query("SELECT count() FROM default.public_dup FINAL")
                 .unwrap_or_default()
                 == "2",
             "duptest's destination holds its own two rows and no others"
diff --git a/tests/subxact.rs b/tests/subxact.rs
index 312f3c92..5756622d 100644
--- a/tests/subxact.rs
+++ b/tests/subxact.rs
@@ -412,6 +412,7 @@ async fn savepoint_after_ddl_rollback_discards_column_and_rows() {
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },
diff --git a/tests/system_columns_cdc.rs b/tests/system_columns_cdc.rs
index 510607c2..7954b46d 100644
--- a/tests/system_columns_cdc.rs
+++ b/tests/system_columns_cdc.rs
@@ -55,6 +55,7 @@ async fn renamed_system_columns_and_operator_keys() {
         NamespaceMapping {
             target_database: Some("walshadow_test".into()),
             auto_create: true,
+            auto_create_name: Some(walshadow::mapping::NameTemplate::parse("$table$").unwrap()),
             drop_table_strategy: None,
             initial_load: None,
         },