viaduck/docs/architecture.d2 at main · PostHog/viaduck · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
direction: right

source: Source DuckLake {
  style.fill: "#e8f4f8"
  catalog: Catalog (Postgres + S3) {
    shape: cylinder
  }
  events: events table {
    shape: sql_table
    company: VARCHAR
    event_id: BIGINT
    payload: VARCHAR
  }
}

pg: Postgres {
  style.fill: "#f3e8f8"
  state: viaduck.viaduck_state {
    shape: sql_table
    destination_id: TEXT
    instance_id: TEXT
    last_snapshot_id: BIGINT
    rows_replicated: BIGINT
    last_error: TEXT
  }
}

viaduck: Viaduck {
  style.fill: "#fff3e0"

  pollthread: Poll thread {
    cdc: "CDC Read\n(table_changes,\nafter_snapshot exclusive)" {
      shape: step
    }
    resolve: "Phase 1\nPreimage Resolution" {
      shape: step
    }
    route: "Router\n(Arrow split)" {
      shape: step
    }
    buffer: "Per-destination\nbuffers" {
      shape: queue
    }
    cdc -> resolve: "inserts + deletes\n+ updates"
    resolve -> route
    route -> buffer: "advance in-memory\nposition"
  }

  workers: "Flush workers (delivery.workers)" {
    conflict: "Phase 2\nConflict Resolution\n(rowid, unioned reads)" {
      shape: step
    }
    apply: "Phase 3\nWinner(k) dedup,\ndelete → upsert (txn)" {
      shape: step
    }
    conflict -> apply
  }

  pollthread.buffer -> workers.conflict: "flush triggers:\ninterval / rows / bytes\n/ memory / shutdown"
}

dest: Destination DuckLakes {
  style.fill: "#e8f5e9"
  quacksworth: Quacksworth Catalog {
    shape: cylinder
    events: events
  }
  mallardine: Mallardine Catalog {
    shape: cylinder
    events: events
  }
  tealford: Tealford Catalog {
    shape: cylinder
    events: events
  }
  "...": "N catalogs" {
    shape: text
    style.font-size: 14
  }
}

source.events -> viaduck.pollthread.cdc: "CDC changes\n(position, current]"
viaduck.workers.apply -> dest.quacksworth: "delete + upsert"
viaduck.workers.apply -> dest.mallardine: "delete + upsert"
viaduck.workers.apply -> dest.tealford: "delete + upsert"
viaduck.workers.apply -> pg.state: "advance cursor\n(after dest commit,\nmonotonicity-guarded)"