@@ -25,64 +25,20 @@ use clickhouse_c::{AsyncClient, Block, Event};
2525
2626use crate :: backup_backfill:: BackupRequest ;
2727use crate :: ch_emitter:: {
28- EmitterConfig , EmitterError , MappingHandle , RetryConfig , TableMapping , connect_client ,
29- drain_to_end_of_stream, is_retryable, quote_ident,
28+ EmitterConfig , EmitterError , MappingHandle , RetryConfig , TableMapping , TableTarget ,
29+ connect_client , drain_to_end_of_stream, is_retryable, quote_ident,
3030} ;
31+ use crate :: shadow_catalog:: RelName ;
3132
3233/// `orders` loads into `orders__wsstg`; deterministic so a retry or boot
3334/// recovery finds the prior attempt's table
3435pub const STAGING_SUFFIX : & str = "__wsstg" ;
3536
36- /// Split a mapping target into unquoted `(database, table)`. Accepts the
37- /// opt-in derived shape (`` `db`.`t` ``, doubled-backtick escapes) and bare
38- /// TOML shapes (`db.t`, `t`); a db-less target lands in `default_db`.
39- pub fn parse_target ( target : & str , default_db : & str ) -> Option < ( String , String ) > {
40- let mut parts: Vec < String > = Vec :: new ( ) ;
41- if target. contains ( '`' ) {
42- let mut it = target. chars ( ) . peekable ( ) ;
43- loop {
44- if it. next ( ) ? != '`' {
45- return None ;
46- }
47- let mut cur = String :: new ( ) ;
48- loop {
49- match it. next ( ) ? {
50- '`' if it. peek ( ) == Some ( & '`' ) => {
51- it. next ( ) ;
52- cur. push ( '`' ) ;
53- }
54- '`' => break ,
55- c => cur. push ( c) ,
56- }
57- }
58- parts. push ( cur) ;
59- match it. next ( ) {
60- None => break ,
61- Some ( '.' ) => continue ,
62- Some ( _) => return None ,
63- }
64- }
65- } else {
66- match target. rsplit_once ( '.' ) {
67- Some ( ( db, t) ) => parts. extend ( [ db. to_owned ( ) , t. to_owned ( ) ] ) ,
68- None => parts. push ( target. to_owned ( ) ) ,
69- }
70- }
71- match parts. len ( ) {
72- 1 => Some ( ( default_db. to_owned ( ) , parts. pop ( ) ?) ) ,
73- 2 => {
74- let t = parts. pop ( ) ?;
75- Some ( ( parts. pop ( ) ?, t) )
76- }
77- _ => None ,
78- }
79- }
80-
8137/// One rel's swap identities. `database`/`table` are the unquoted
8238/// destination parts; `s_lsn` drives the copy-back filter.
8339#[ derive( Debug , Clone ) ]
8440pub struct StagingRel {
85- pub qname : String ,
41+ pub rel : RelName ,
8642 pub database : String ,
8743 pub table : String ,
8844 pub s_lsn : u64 ,
@@ -127,37 +83,31 @@ pub async fn prepare(
12783) -> Result < StagingPlan > {
12884 let mut sess = StagingSession :: connect ( emitter) . await ?;
12985 let live_map = live. read ( ) . await . clone ( ) ;
130- let mut staged: HashMap < String , TableMapping > = HashMap :: new ( ) ;
86+ let mut staged: HashMap < RelName , TableMapping > = HashMap :: new ( ) ;
13187 let mut rels = Vec :: new ( ) ;
13288 for r in reqs {
133- let qname = r. desc . qualified_name . as_ref ( ) ;
134- let Some ( m) = live_map. get ( qname ) else {
89+ let name = & r. desc . rel_name ;
90+ let Some ( m) = live_map. get ( name ) else {
13591 tracing:: warn!(
13692 target: "walshadow::backfill_staging" ,
137- qname,
93+ qname = %name ,
13894 "no mapping at pass start; rows will skip" ,
13995 ) ;
14096 continue ;
14197 } ;
142- let Some ( ( database, table) ) = parse_target ( & m. target , & emitter. database ) else {
143- bail ! (
144- "backfill_staging: unparseable mapping target {:?} for {qname}" ,
145- m. target
146- ) ;
147- } ;
14898 let rel = StagingRel {
149- qname : qname . to_owned ( ) ,
150- database,
151- table,
99+ rel : name . clone ( ) ,
100+ database : m . target . database . clone ( ) ,
101+ table : m . target . table . clone ( ) ,
152102 s_lsn : r. s_lsn ,
153103 } ;
154104 sess. rebuild_staging ( & rel)
155105 . await
156- . with_context ( || format ! ( "backfill_staging: rebuild staging for {qname }" ) ) ?;
106+ . with_context ( || format ! ( "backfill_staging: rebuild staging for {name }" ) ) ?;
157107 staged. insert (
158- qname . to_owned ( ) ,
108+ name . clone ( ) ,
159109 TableMapping {
160- target : rel. staging_sql ( ) ,
110+ target : TableTarget :: new ( & rel. database , & rel . staging_table ( ) ) ,
161111 columns : m. columns . clone ( ) ,
162112 } ,
163113 ) ;
@@ -409,37 +359,10 @@ fn sql_str(s: &str) -> String {
409359mod tests {
410360 use super :: * ;
411361
412- #[ test]
413- fn parse_target_handles_quoted_and_bare_shapes ( ) {
414- assert_eq ! (
415- parse_target( "`db`.`orders`" , "default" ) ,
416- Some ( ( "db" . into( ) , "orders" . into( ) ) )
417- ) ;
418- assert_eq ! (
419- parse_target( "db.orders" , "default" ) ,
420- Some ( ( "db" . into( ) , "orders" . into( ) ) )
421- ) ;
422- assert_eq ! (
423- parse_target( "orders" , "default" ) ,
424- Some ( ( "default" . into( ) , "orders" . into( ) ) )
425- ) ;
426- assert_eq ! (
427- parse_target( "`orders`" , "default" ) ,
428- Some ( ( "default" . into( ) , "orders" . into( ) ) )
429- ) ;
430- // Embedded dot + doubled backtick stay inside the quoted ident
431- assert_eq ! (
432- parse_target( "`d.b`.`or``ders`" , "default" ) ,
433- Some ( ( "d.b" . into( ) , "or`ders" . into( ) ) )
434- ) ;
435- assert_eq ! ( parse_target( "`db`.`t`.`x`" , "default" ) , None ) ;
436- assert_eq ! ( parse_target( "`unterminated" , "default" ) , None ) ;
437- }
438-
439362 #[ test]
440363 fn staging_rel_renders_sql_names ( ) {
441364 let rel = StagingRel {
442- qname : "public. orders" . into ( ) ,
365+ rel : RelName :: new ( "public" , " orders") ,
443366 database : "db" . into ( ) ,
444367 table : "orders" . into ( ) ,
445368 s_lsn : 0x5000 ,
0 commit comments