/src/postgres/src/backend/utils/activity/pgstat_subscription.c
Line | Count | Source |
1 | | /* ------------------------------------------------------------------------- |
2 | | * |
3 | | * pgstat_subscription.c |
4 | | * Implementation of subscription statistics. |
5 | | * |
6 | | * This file contains the implementation of subscription statistics. It is kept |
7 | | * separate from pgstat.c to enforce the line between the statistics access / |
8 | | * storage implementation and the details about individual types of |
9 | | * statistics. |
10 | | * |
11 | | * Copyright (c) 2001-2026, PostgreSQL Global Development Group |
12 | | * |
13 | | * IDENTIFICATION |
14 | | * src/backend/utils/activity/pgstat_subscription.c |
15 | | * ------------------------------------------------------------------------- |
16 | | */ |
17 | | |
18 | | #include "postgres.h" |
19 | | |
20 | | #include "replication/worker_internal.h" |
21 | | #include "utils/pgstat_internal.h" |
22 | | |
23 | | |
24 | | /* |
25 | | * Report a subscription error. |
26 | | */ |
27 | | void |
28 | | pgstat_report_subscription_error(Oid subid) |
29 | 0 | { |
30 | 0 | PgStat_EntryRef *entry_ref; |
31 | 0 | PgStat_BackendSubEntry *pending; |
32 | 0 | LogicalRepWorkerType wtype = get_logical_worker_type(); |
33 | |
|
34 | 0 | entry_ref = pgstat_prep_pending_entry(PGSTAT_KIND_SUBSCRIPTION, |
35 | 0 | InvalidOid, subid, NULL); |
36 | 0 | pending = entry_ref->pending; |
37 | |
|
38 | 0 | switch (wtype) |
39 | 0 | { |
40 | 0 | case WORKERTYPE_APPLY: |
41 | 0 | pending->apply_error_count++; |
42 | 0 | break; |
43 | | |
44 | 0 | case WORKERTYPE_SEQUENCESYNC: |
45 | 0 | pending->sync_seq_error_count++; |
46 | 0 | break; |
47 | | |
48 | 0 | case WORKERTYPE_TABLESYNC: |
49 | 0 | pending->sync_table_error_count++; |
50 | 0 | break; |
51 | | |
52 | 0 | default: |
53 | | /* Should never happen. */ |
54 | 0 | Assert(0); |
55 | 0 | break; |
56 | 0 | } |
57 | 0 | } |
58 | | |
59 | | /* |
60 | | * Report a subscription conflict. |
61 | | */ |
62 | | void |
63 | | pgstat_report_subscription_conflict(Oid subid, ConflictType type) |
64 | 0 | { |
65 | 0 | PgStat_EntryRef *entry_ref; |
66 | 0 | PgStat_BackendSubEntry *pending; |
67 | |
|
68 | 0 | entry_ref = pgstat_prep_pending_entry(PGSTAT_KIND_SUBSCRIPTION, |
69 | 0 | InvalidOid, subid, NULL); |
70 | 0 | pending = entry_ref->pending; |
71 | 0 | pending->conflict_count[type]++; |
72 | 0 | } |
73 | | |
74 | | /* |
75 | | * Report creating the subscription. |
76 | | */ |
77 | | void |
78 | | pgstat_create_subscription(Oid subid) |
79 | 0 | { |
80 | | /* Ensures that stats are dropped if transaction rolls back */ |
81 | 0 | pgstat_create_transactional(PGSTAT_KIND_SUBSCRIPTION, |
82 | 0 | InvalidOid, subid); |
83 | | |
84 | | /* Create and initialize the subscription stats entry */ |
85 | 0 | pgstat_get_entry_ref(PGSTAT_KIND_SUBSCRIPTION, InvalidOid, subid, |
86 | 0 | true, NULL); |
87 | 0 | pgstat_reset_entry(PGSTAT_KIND_SUBSCRIPTION, InvalidOid, subid, 0); |
88 | 0 | } |
89 | | |
90 | | /* |
91 | | * Report dropping the subscription. |
92 | | * |
93 | | * Ensures that stats are dropped if transaction commits. |
94 | | */ |
95 | | void |
96 | | pgstat_drop_subscription(Oid subid) |
97 | 0 | { |
98 | 0 | pgstat_drop_transactional(PGSTAT_KIND_SUBSCRIPTION, |
99 | 0 | InvalidOid, subid); |
100 | 0 | } |
101 | | |
102 | | /* |
103 | | * Support function for the SQL-callable pgstat* functions. Returns |
104 | | * the collected statistics for one subscription or NULL. |
105 | | */ |
106 | | PgStat_StatSubEntry * |
107 | | pgstat_fetch_stat_subscription(Oid subid) |
108 | 0 | { |
109 | 0 | return (PgStat_StatSubEntry *) |
110 | 0 | pgstat_fetch_entry(PGSTAT_KIND_SUBSCRIPTION, InvalidOid, subid, NULL); |
111 | 0 | } |
112 | | |
113 | | /* |
114 | | * Flush out pending stats for the entry |
115 | | * |
116 | | * If nowait is true and the lock could not be immediately acquired, returns |
117 | | * false without flushing the entry. Otherwise returns true. |
118 | | */ |
119 | | bool |
120 | | pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait) |
121 | 0 | { |
122 | 0 | PgStat_BackendSubEntry *localent; |
123 | 0 | PgStatShared_Subscription *shsubent; |
124 | |
|
125 | 0 | localent = (PgStat_BackendSubEntry *) entry_ref->pending; |
126 | 0 | shsubent = (PgStatShared_Subscription *) entry_ref->shared_stats; |
127 | | |
128 | | /* localent always has non-zero content */ |
129 | |
|
130 | 0 | if (!pgstat_lock_entry(entry_ref, nowait)) |
131 | 0 | return false; |
132 | | |
133 | 0 | #define SUB_ACC(fld) shsubent->stats.fld += localent->fld |
134 | 0 | SUB_ACC(apply_error_count); |
135 | 0 | SUB_ACC(sync_seq_error_count); |
136 | 0 | SUB_ACC(sync_table_error_count); |
137 | 0 | for (int i = 0; i < CONFLICT_NUM_TYPES; i++) |
138 | 0 | SUB_ACC(conflict_count[i]); |
139 | 0 | #undef SUB_ACC |
140 | |
|
141 | 0 | pgstat_unlock_entry(entry_ref); |
142 | 0 | return true; |
143 | 0 | } |
144 | | |
145 | | void |
146 | | pgstat_subscription_reset_timestamp_cb(PgStatShared_Common *header, TimestampTz ts) |
147 | 0 | { |
148 | 0 | ((PgStatShared_Subscription *) header)->stats.stat_reset_timestamp = ts; |
149 | 0 | } |