ruststream_sqlx_dialect/dialect.rs
1//! The trait every dialect implements: names, placeholders, and the statements every form runs.
2
3use std::fmt::Debug;
4use std::num::NonZeroUsize;
5
6use crate::opening::Opening;
7use crate::spec::TableSpec;
8use crate::statement::{Statement, StatementError};
9use crate::table_name::TableName;
10
11#[doc = include_str!("dialect/README.md")]
12pub trait Dialect: Debug + Send + Sync {
13 /// The dialect's name, for messages.
14 ///
15 /// # Examples
16 ///
17 /// ```
18 /// # use std::num::NonZeroUsize;
19 /// use ruststream_sqlx_dialect::{Dialect, Form, Param, Statement, StatementError, TableSpec};
20 /// # use ruststream_sqlx_dialect::TableName;
21 ///
22 /// /// SQL Server, a database without a built-in dialect.
23 /// #[derive(Debug)]
24 /// pub struct Mssql;
25 ///
26 /// impl Dialect for Mssql {
27 /// // The name a refusal of this dialect carries.
28 /// fn name(&self) -> &'static str {
29 /// "mssql"
30 /// }
31 ///
32 /// // A table in another form stops with "the mssql dialect has no statements for the lease
33 /// // form".
34 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
35 /// if spec.form() != Form::RowLock {
36 /// return Err(StatementError::UnsupportedForm {
37 /// dialect: self.name(),
38 /// form: spec.form().name(),
39 /// });
40 /// }
41 /// Ok(Statement::new(
42 /// "DELETE FROM [email_jobs] WHERE [job_id] = @p1",
43 /// [Param::Id],
44 /// ))
45 /// }
46 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
47 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
48 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
49 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
50 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
51 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
52 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
53 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
54 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
55 /// }
56 /// ```
57 fn name(&self) -> &'static str;
58
59 /// Appends `ident` to `out`, quoted as a name of this database.
60 ///
61 /// # Examples
62 ///
63 /// ```
64 /// # use std::num::NonZeroUsize;
65 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
66 /// # use ruststream_sqlx_dialect::TableName;
67 ///
68 /// /// SQL Server, a database without a built-in dialect.
69 /// #[derive(Debug)]
70 /// pub struct Mssql;
71 ///
72 /// impl Dialect for Mssql {
73 /// fn name(&self) -> &'static str {
74 /// "mssql"
75 /// }
76 ///
77 /// // A name in brackets keeps its case and its spaces; a closing bracket doubles.
78 /// fn quote_into(&self, ident: &str, out: &mut String) {
79 /// out.push('[');
80 /// out.push_str(&ident.replace(']', "]]"));
81 /// out.push(']');
82 /// }
83 ///
84 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
85 /// let mut sql = String::from("DELETE FROM ");
86 /// self.quote_into(spec.table(), &mut sql);
87 /// sql.push_str(" WHERE ");
88 /// self.quote_into(spec.id().name(), &mut sql);
89 /// sql.push_str(" = @p1");
90 /// Ok(Statement::new(sql, [Param::Id]))
91 /// }
92 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
93 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
94 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
95 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
96 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
97 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
98 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
99 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
100 /// }
101 /// ```
102 fn quote_into(&self, ident: &str, out: &mut String);
103
104 /// Appends the placeholder of parameter number `index` (counted from 1) to `out`.
105 ///
106 /// # Examples
107 ///
108 /// ```
109 /// use std::num::NonZeroUsize;
110 ///
111 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
112 /// # use ruststream_sqlx_dialect::TableName;
113 ///
114 /// /// SQL Server, a database without a built-in dialect.
115 /// #[derive(Debug)]
116 /// pub struct Mssql;
117 ///
118 /// impl Dialect for Mssql {
119 /// fn name(&self) -> &'static str {
120 /// "mssql"
121 /// }
122 ///
123 /// // SQL Server numbers its parameters `@p1`, `@p2` and on.
124 /// fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) {
125 /// out.push_str("@p");
126 /// out.push_str(&index.to_string());
127 /// }
128 ///
129 /// fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
130 /// let mut sql = String::from("UPDATE [email_jobs] SET [retry_after] = ");
131 /// self.placeholder_into(NonZeroUsize::MIN, &mut sql);
132 /// sql.push_str(" WHERE [job_id] = ");
133 /// self.placeholder_into(NonZeroUsize::MIN.saturating_add(1), &mut sql);
134 /// Ok(Statement::new(sql, [Param::RetryAfter, Param::Id]))
135 /// }
136 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
137 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
138 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
139 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
140 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
141 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
142 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
143 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
144 /// }
145 /// ```
146 fn placeholder_into(&self, index: NonZeroUsize, out: &mut String);
147
148 /// The statement that reads the rows of claimed ids, bound as one list.
149 ///
150 /// # Errors
151 ///
152 /// [`StatementError::UnsupportedFetch`] when the dialect cannot read rows by a list of ids, so
153 /// a claim of the service's own needs a fetch of its own too. Postgres builds it for every
154 /// table; MySQL and SQLite refuse it.
155 ///
156 /// # Examples
157 ///
158 /// ```
159 /// # use std::num::NonZeroUsize;
160 /// use ruststream_sqlx_dialect::{Dialect, Statement, StatementError, TableSpec};
161 /// # use ruststream_sqlx_dialect::TableName;
162 ///
163 /// /// SQL Server, a database without a built-in dialect.
164 /// #[derive(Debug)]
165 /// pub struct Mssql;
166 ///
167 /// impl Dialect for Mssql {
168 /// fn name(&self) -> &'static str {
169 /// "mssql"
170 /// }
171 ///
172 /// // SQL Server binds no list of ids: a table whose claim is the service's own lists
173 /// // `fetch`
174 /// // in `custom(..)` beside `claim`.
175 /// fn fetch(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
176 /// Err(StatementError::UnsupportedFetch { dialect: self.name() })
177 /// }
178 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
179 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
180 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
181 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
182 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
183 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
184 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
185 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
186 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
187 /// }
188 /// ```
189 fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
190
191 /// The statement that acknowledges a row: it deletes the row, or sets `processed_at` when the
192 /// table has that column.
193 ///
194 /// # Errors
195 ///
196 /// [`StatementError::UnsupportedForm`] when the dialect has no statements for the table's
197 /// form; [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock.
198 ///
199 /// # Examples
200 ///
201 /// ```
202 /// # use std::num::NonZeroUsize;
203 /// use ruststream_sqlx_dialect::{
204 /// Dialect, Form, Param, Role, Statement, StatementError, TableSpec,
205 /// };
206 /// # use ruststream_sqlx_dialect::TableName;
207 ///
208 /// /// SQL Server, a database without a built-in dialect; the service's tables take their rows
209 /// /// by row lock.
210 /// #[derive(Debug)]
211 /// pub struct Mssql;
212 ///
213 /// impl Dialect for Mssql {
214 /// fn name(&self) -> &'static str {
215 /// "mssql"
216 /// }
217 ///
218 /// // A finished row goes, or stays with the moment it finished where the table keeps one.
219 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
220 /// if spec.form() != Form::RowLock {
221 /// return Err(StatementError::UnsupportedForm {
222 /// dialect: self.name(),
223 /// form: spec.form().name(),
224 /// });
225 /// }
226 /// let mut table = String::new();
227 /// self.quote_into(spec.table(), &mut table);
228 /// let mut id = String::new();
229 /// self.quote_into(spec.id().name(), &mut id);
230 /// let Some(processed_at) = spec.column(Role::ProcessedAt) else {
231 /// return Ok(Statement::new(
232 /// format!("DELETE FROM {table} WHERE {id} = @p1"),
233 /// [Param::Id],
234 /// ));
235 /// };
236 /// let mut finished = String::new();
237 /// self.quote_into(processed_at.name(), &mut finished);
238 /// Ok(Statement::new(
239 /// format!("UPDATE {table} SET {finished} = @p1 WHERE {id} = @p2"),
240 /// [Param::Now, Param::Id],
241 /// ))
242 /// }
243 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
244 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
245 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
246 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
247 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
248 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
249 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
250 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
251 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
252 /// }
253 /// ```
254 fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
255
256 /// The statement that releases a row for another attempt at once, or `None` when releasing
257 /// the row needs no statement.
258 ///
259 /// In the lease form it clears the lease, and the attempt stays as the claim counted it. In the
260 /// advisory lock form it is `None`: the take counted the attempt, and the unlock frees the row.
261 ///
262 /// # Errors
263 ///
264 /// [`StatementError::UnsupportedForm`] when the dialect has no statements for the table's
265 /// form; [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock.
266 ///
267 /// # Examples
268 ///
269 /// ```
270 /// # use std::num::NonZeroUsize;
271 /// use ruststream_sqlx_dialect::{Dialect, Param, Role, Statement, StatementError, TableSpec};
272 /// # use ruststream_sqlx_dialect::TableName;
273 ///
274 /// /// SQL Server, a database without a built-in dialect; the service's tables take their rows
275 /// /// by row lock.
276 /// #[derive(Debug)]
277 /// pub struct Mssql;
278 ///
279 /// impl Dialect for Mssql {
280 /// fn name(&self) -> &'static str {
281 /// "mssql"
282 /// }
283 ///
284 /// // A retry counts the attempt where the table counts them. A table without the column
285 /// // needs no statement: the end of the claim's transaction frees the row.
286 /// fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
287 /// let Some(attempt) = spec.column(Role::Attempt) else {
288 /// return Ok(None);
289 /// };
290 /// let mut table = String::new();
291 /// self.quote_into(spec.table(), &mut table);
292 /// let mut count = String::new();
293 /// self.quote_into(attempt.name(), &mut count);
294 /// let mut id = String::new();
295 /// self.quote_into(spec.id().name(), &mut id);
296 /// Ok(Some(Statement::new(
297 /// format!("UPDATE {table} SET {count} = {count} + 1 WHERE {id} = @p1"),
298 /// [Param::Id],
299 /// )))
300 /// }
301 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
302 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
303 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
304 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
305 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
306 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
307 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
308 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
309 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
310 /// }
311 /// ```
312 fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError>;
313
314 /// The statement that releases a row for another attempt after a delay, bound as
315 /// [`Param::RetryAfter`](crate::Param::RetryAfter).
316 ///
317 /// # Errors
318 ///
319 /// [`StatementError::MissingRole`] when the table has no `retry_after` column;
320 /// [`StatementError::UnsupportedForm`] when the dialect has no statements for the table's
321 /// form; [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock.
322 ///
323 /// # Examples
324 ///
325 /// ```
326 /// # use std::num::NonZeroUsize;
327 /// use ruststream_sqlx_dialect::{Dialect, Param, Role, Statement, StatementError, TableSpec};
328 /// # use ruststream_sqlx_dialect::TableName;
329 ///
330 /// /// SQL Server, a database without a built-in dialect; the service's tables take their rows
331 /// /// by row lock.
332 /// #[derive(Debug)]
333 /// pub struct Mssql;
334 ///
335 /// impl Dialect for Mssql {
336 /// fn name(&self) -> &'static str {
337 /// "mssql"
338 /// }
339 ///
340 /// // The row waits in its table until the delay ends.
341 /// fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
342 /// let Some(retry_after) = spec.column(Role::RetryAfter) else {
343 /// return Err(StatementError::MissingRole {
344 /// statement: "retry_after",
345 /// role: Role::RetryAfter,
346 /// });
347 /// };
348 /// let mut table = String::new();
349 /// self.quote_into(spec.table(), &mut table);
350 /// let mut until = String::new();
351 /// self.quote_into(retry_after.name(), &mut until);
352 /// let mut id = String::new();
353 /// self.quote_into(spec.id().name(), &mut id);
354 /// Ok(Statement::new(
355 /// format!("UPDATE {table} SET {until} = @p1 WHERE {id} = @p2"),
356 /// [Param::RetryAfter, Param::Id],
357 /// ))
358 /// }
359 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
360 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
361 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
362 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
363 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
364 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
365 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
366 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
367 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
368 /// }
369 /// ```
370 fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
371
372 /// The statement that drops a row: it deletes the row, or sets `processed_at` when the table
373 /// has that column.
374 ///
375 /// # Errors
376 ///
377 /// [`StatementError::UnsupportedForm`] when the dialect has no statements for the table's
378 /// form; [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock.
379 ///
380 /// # Examples
381 ///
382 /// ```
383 /// # use std::num::NonZeroUsize;
384 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
385 /// # use ruststream_sqlx_dialect::TableName;
386 ///
387 /// /// SQL Server, a database without a built-in dialect.
388 /// #[derive(Debug)]
389 /// pub struct Mssql;
390 ///
391 /// impl Dialect for Mssql {
392 /// fn name(&self) -> &'static str {
393 /// "mssql"
394 /// }
395 ///
396 /// // A dropped email stays in the service's one queue table, marked with the time it was
397 /// // dropped.
398 /// fn discard(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
399 /// Ok(Statement::new(
400 /// "UPDATE [email_jobs] SET [dropped_at] = @p1 WHERE [job_id] = @p2",
401 /// [Param::Now, Param::Id],
402 /// ))
403 /// }
404 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
405 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
406 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
407 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
408 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
409 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
410 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
411 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
412 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
413 /// }
414 /// ```
415 fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
416
417 /// The statement that moves a row whose attempts are spent to another group, bound as
418 /// [`Param::Destination`](crate::Param::Destination).
419 ///
420 /// # Errors
421 ///
422 /// [`StatementError::MissingRole`] when the table has no `group` column;
423 /// [`StatementError::UnsupportedForm`] when the dialect has no statements for the table's
424 /// form; [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock.
425 ///
426 /// # Examples
427 ///
428 /// ```
429 /// # use std::num::NonZeroUsize;
430 /// use ruststream_sqlx_dialect::{Dialect, Param, Role, Statement, StatementError, TableSpec};
431 /// # use ruststream_sqlx_dialect::TableName;
432 ///
433 /// /// SQL Server, a database without a built-in dialect; the service's tables take their rows
434 /// /// by row lock.
435 /// #[derive(Debug)]
436 /// pub struct Mssql;
437 ///
438 /// impl Dialect for Mssql {
439 /// fn name(&self) -> &'static str {
440 /// "mssql"
441 /// }
442 ///
443 /// // A dead letter starts its new group with the attempt count of a new row.
444 /// fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
445 /// let Some(group) = spec.column(Role::Group) else {
446 /// return Err(StatementError::MissingRole {
447 /// statement: "dead_letter_group",
448 /// role: Role::Group,
449 /// });
450 /// };
451 /// let mut sql = String::from("UPDATE ");
452 /// self.quote_into(spec.table(), &mut sql);
453 /// sql.push_str(" SET ");
454 /// self.quote_into(group.name(), &mut sql);
455 /// sql.push_str(" = @p1");
456 /// if let Some(attempt) = spec.column(Role::Attempt) {
457 /// sql.push_str(", ");
458 /// self.quote_into(attempt.name(), &mut sql);
459 /// sql.push_str(" = DEFAULT");
460 /// }
461 /// sql.push_str(" WHERE ");
462 /// self.quote_into(spec.id().name(), &mut sql);
463 /// sql.push_str(" = @p2");
464 /// Ok(Statement::new(sql, [Param::Destination, Param::Id]))
465 /// }
466 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
467 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
468 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
469 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
470 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
471 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
472 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
473 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
474 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
475 /// }
476 /// ```
477 fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
478
479 /// The statements that move a row whose attempts are spent to `target`, a table with the same
480 /// columns; they run in one transaction.
481 ///
482 /// A table read with `*` ([`TableSpec::selects_all`]) has columns the description does not
483 /// name, so its row moves by position: `target` has the same columns in the same order. In
484 /// the lease form the row arrives without a lease, so whatever reads `target` can claim it at
485 /// once.
486 ///
487 /// # Errors
488 ///
489 /// [`StatementError::UnsupportedForm`] when the dialect has no statements for the table's
490 /// form; [`StatementError::LeaseOnDatabaseClock`] for a lease table on the database's clock;
491 /// [`StatementError::Flattened`] for a lease table read with `*`, whose lease column the move
492 /// cannot name.
493 ///
494 /// # Examples
495 ///
496 /// ```
497 /// # use std::num::NonZeroUsize;
498 /// use ruststream_sqlx_dialect::{
499 /// Dialect, Param, Statement, StatementError, TableName, TableSpec,
500 /// };
501 ///
502 /// /// SQL Server, a database without a built-in dialect.
503 /// #[derive(Debug)]
504 /// pub struct Mssql;
505 ///
506 /// impl Dialect for Mssql {
507 /// fn name(&self) -> &'static str {
508 /// "mssql"
509 /// }
510 ///
511 /// // A dead email of the service's one queue table moves with the time it died.
512 /// fn dead_letter_table(
513 /// &self,
514 /// _spec: &TableSpec<'_>,
515 /// target: TableName<'_>,
516 /// ) -> Result<Vec<Statement>, StatementError> {
517 /// let mut into = String::new();
518 /// self.quote_into(target.schema().unwrap_or("dbo"), &mut into);
519 /// into.push('.');
520 /// self.quote_into(target.table(), &mut into);
521 /// Ok(vec![
522 /// Statement::new(
523 /// format!(
524 /// "INSERT INTO {into} ([job_id], [payload], [died_at]) \
525 /// SELECT [job_id], [payload], @p1 FROM [email_jobs] WHERE [job_id] = @p2"
526 /// ),
527 /// [Param::Now, Param::Id],
528 /// ),
529 /// Statement::new("DELETE FROM [email_jobs] WHERE [job_id] = @p1", [Param::Id]),
530 /// ])
531 /// }
532 ///
533 /// fn quote_into(&self, ident: &str, out: &mut String) {
534 /// out.push('[');
535 /// out.push_str(&ident.replace(']', "]]"));
536 /// out.push(']');
537 /// }
538 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
539 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
540 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
541 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
542 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
543 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
544 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
545 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
546 /// }
547 /// ```
548 fn dead_letter_table(
549 &self,
550 spec: &TableSpec<'_>,
551 target: TableName<'_>,
552 ) -> Result<Vec<Statement>, StatementError>;
553
554 /// The statement that inserts a row: every column the database does not fill, in the order of
555 /// [`TableSpec::columns`], each bound as [`Param::Column`](crate::Param::Column).
556 ///
557 /// # Errors
558 ///
559 /// [`StatementError::Flattened`] when the table is read with `*`
560 /// ([`TableSpec::selects_all`]), whose columns the description cannot see.
561 ///
562 /// # Examples
563 ///
564 /// ```
565 /// # use std::num::NonZeroUsize;
566 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
567 /// # use ruststream_sqlx_dialect::TableName;
568 ///
569 /// /// SQL Server, a database without a built-in dialect.
570 /// #[derive(Debug)]
571 /// pub struct Mssql;
572 ///
573 /// impl Dialect for Mssql {
574 /// fn name(&self) -> &'static str {
575 /// "mssql"
576 /// }
577 ///
578 /// // Publishing a job twice keeps one row: the insert skips a row whose id is already
579 /// // there. The service's one queue table lists its id, then its payload, and the lock
580 /// // held to the end of the statement keeps two inserts of one id apart.
581 /// fn insert(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
582 /// Ok(Statement::new(
583 /// "INSERT INTO [email_jobs] ([job_id], [payload]) SELECT @p1, @p2 \
584 /// WHERE NOT EXISTS (SELECT 1 FROM [email_jobs] WITH (UPDLOCK, HOLDLOCK) \
585 /// WHERE [job_id] = @p1)",
586 /// [Param::Column(0), Param::Column(1)],
587 /// ))
588 /// }
589 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
590 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
591 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
592 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
593 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
594 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
595 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
596 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
597 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
598 /// }
599 /// ```
600 fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
601
602 /// The query that reads the server's version as one text column, or `None` when the
603 /// dialect's statements run on every version of its server.
604 ///
605 /// The provided method returns `None`: the broker asks the server nothing, and every statement
606 /// runs on whatever version the server is.
607 ///
608 /// # Examples
609 ///
610 /// ```
611 /// # use std::num::NonZeroUsize;
612 /// use ruststream_sqlx_dialect::{Dialect, Statement, StatementError, TableSpec};
613 /// # use ruststream_sqlx_dialect::TableName;
614 ///
615 /// /// SQL Server, a database without a built-in dialect.
616 /// #[derive(Debug)]
617 /// pub struct Mssql;
618 ///
619 /// impl Dialect for Mssql {
620 /// fn name(&self) -> &'static str {
621 /// "mssql"
622 /// }
623 ///
624 /// // The server reports its major version as a number: 14 is SQL Server 2017.
625 /// fn server_version(&self) -> Option<&'static str> {
626 /// Some("SELECT CAST(SERVERPROPERTY('ProductMajorVersion') AS nvarchar(128))")
627 /// }
628 ///
629 /// fn check_server(
630 /// &self,
631 /// _spec: &TableSpec<'_>,
632 /// version: &str,
633 /// ) -> Result<(), StatementError> {
634 /// match version.trim().parse::<u32>() {
635 /// Ok(major) if major >= 14 => Ok(()),
636 /// _ => Err(StatementError::ServerTooOld {
637 /// dialect: self.name(),
638 /// server: version.to_owned(),
639 /// required: "SQL Server 2017",
640 /// }),
641 /// }
642 /// }
643 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
644 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
645 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
646 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
647 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
648 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
649 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
650 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
651 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
652 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
653 /// }
654 /// ```
655 fn server_version(&self) -> Option<&'static str> {
656 None
657 }
658
659 /// Refuses a server older than the statements of `spec` need; `version` is what the
660 /// [`server_version`](Self::server_version) query returned.
661 ///
662 /// The provided method accepts every version, as a dialect whose statements run on every
663 /// version of its server does.
664 ///
665 /// # Errors
666 ///
667 /// [`StatementError::ServerTooOld`] when the server predates the dialect's statements for the
668 /// table's form.
669 ///
670 /// # Examples
671 ///
672 /// ```
673 /// # use std::num::NonZeroUsize;
674 /// use ruststream_sqlx_dialect::{Dialect, Statement, StatementError, TableSpec};
675 /// # use ruststream_sqlx_dialect::TableName;
676 ///
677 /// /// SQL Server, a database without a built-in dialect.
678 /// #[derive(Debug)]
679 /// pub struct Mssql;
680 ///
681 /// impl Dialect for Mssql {
682 /// fn name(&self) -> &'static str {
683 /// "mssql"
684 /// }
685 ///
686 /// fn server_version(&self) -> Option<&'static str> {
687 /// Some("SELECT CAST(SERVERPROPERTY('ProductMajorVersion') AS nvarchar(128))")
688 /// }
689 ///
690 /// // The service's statements need SQL Server 2017, so an older server stops the
691 /// // subscription.
692 /// fn check_server(
693 /// &self,
694 /// _spec: &TableSpec<'_>,
695 /// version: &str,
696 /// ) -> Result<(), StatementError> {
697 /// match version.trim().parse::<u32>() {
698 /// Ok(major) if major >= 14 => Ok(()),
699 /// _ => Err(StatementError::ServerTooOld {
700 /// dialect: self.name(),
701 /// server: version.to_owned(),
702 /// required: "SQL Server 2017",
703 /// }),
704 /// }
705 /// }
706 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
707 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
708 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
709 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
710 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
711 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
712 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
713 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
714 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
715 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
716 /// }
717 /// ```
718 fn check_server(&self, spec: &TableSpec<'_>, version: &str) -> Result<(), StatementError> {
719 let _ = (spec, version);
720 Ok(())
721 }
722
723 /// The statement that opens a transaction at `opening` in place of `BEGIN`, or `None` when
724 /// `BEGIN` opens it.
725 ///
726 /// A broker opens the row lock claim's transaction with it, at the table's opening
727 /// ([`TableSpec::opening`]). The statement leaves the connection inside a transaction, as
728 /// `BEGIN` does. A dialect accepts the openings it implements [`Opens`](crate::Opens) for. The
729 /// provided method opens [`Opening::Default`] with `BEGIN` and refuses every level and mode: on
730 /// a dialect that keeps it, a table that declares `isolation` or `mode` stops when its
731 /// subscription starts.
732 ///
733 /// # Errors
734 ///
735 /// [`StatementError::UnsupportedOpening`] for an opening the dialect does not open: an
736 /// isolation level on SQLite, a SQLite mode elsewhere, READ UNCOMMITTED on Postgres.
737 ///
738 /// # Examples
739 ///
740 /// A dialect of the service's own opens the levels its database keeps, and says so with
741 /// `Opens`:
742 ///
743 /// ```
744 /// # use std::num::NonZeroUsize;
745 /// use ruststream_sqlx_dialect::{
746 /// Dialect, Isolation, Opening, Opens, Statement, StatementError, TableSpec, level,
747 /// };
748 /// # use ruststream_sqlx_dialect::TableName;
749 ///
750 /// /// SQL Server, a database without a built-in dialect.
751 /// #[derive(Debug)]
752 /// pub struct Mssql;
753 ///
754 /// impl Dialect for Mssql {
755 /// fn name(&self) -> &'static str {
756 /// "mssql"
757 /// }
758 ///
759 /// // SQL Server starts a transaction with `BEGIN TRANSACTION` and keeps a level it set for
760 /// // the rest of the session, so every opening names its level.
761 /// fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> {
762 /// match opening {
763 /// Opening::Default | Opening::Isolation(Isolation::ReadCommitted) => Ok(Some(
764 /// "SET TRANSACTION ISOLATION LEVEL READ COMMITTED; BEGIN TRANSACTION",
765 /// )),
766 /// Opening::Isolation(Isolation::RepeatableRead) => Ok(Some(
767 /// "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ; BEGIN TRANSACTION",
768 /// )),
769 /// other => Err(StatementError::UnsupportedOpening {
770 /// dialect: self.name(),
771 /// opening: other.name(),
772 /// }),
773 /// }
774 /// }
775 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
776 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
777 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
778 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
779 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
780 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
781 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
782 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
783 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
784 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
785 /// }
786 ///
787 /// // A table at one of these levels mounts on `Mssql`.
788 /// impl Opens<level::ReadCommitted> for Mssql {}
789 /// impl Opens<level::RepeatableRead> for Mssql {}
790 /// ```
791 fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> {
792 match opening {
793 Opening::Default => Ok(None),
794 Opening::Isolation(_) | Opening::Mode(_) => Err(opening.refused(self.name())),
795 }
796 }
797
798 /// The statement that sets the savepoint after a claim, inside the claim's transaction: what
799 /// a handler writes after it can be discarded while the claim's own work stays.
800 ///
801 /// The provided method returns `SAVEPOINT ruststream_claim`, the SQL standard's form, which
802 /// Postgres, MySQL and SQLite take; a database that sets a savepoint another way returns its
803 /// own statement here and in [`rollback_to_savepoint`](Self::rollback_to_savepoint).
804 ///
805 /// # Examples
806 ///
807 /// ```
808 /// # use std::num::NonZeroUsize;
809 /// use ruststream_sqlx_dialect::Dialect;
810 /// # use ruststream_sqlx_dialect::{Statement, StatementError, TableName, TableSpec};
811 ///
812 /// /// SQL Server, a database without a built-in dialect.
813 /// #[derive(Debug)]
814 /// pub struct Mssql;
815 ///
816 /// impl Dialect for Mssql {
817 /// fn name(&self) -> &'static str {
818 /// "mssql"
819 /// }
820 ///
821 /// // SQL Server names a savepoint with `SAVE TRANSACTION`.
822 /// fn savepoint(&self) -> &'static str {
823 /// "SAVE TRANSACTION ruststream_claim"
824 /// }
825 ///
826 /// fn rollback_to_savepoint(&self) -> &'static str {
827 /// "ROLLBACK TRANSACTION ruststream_claim"
828 /// }
829 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
830 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
831 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
832 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
833 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
834 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
835 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
836 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
837 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
838 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
839 /// }
840 /// ```
841 fn savepoint(&self) -> &'static str {
842 "SAVEPOINT ruststream_claim"
843 }
844
845 /// The statement that discards what the transaction did after [`savepoint`](Self::savepoint),
846 /// and keeps the transaction open.
847 ///
848 /// The provided method returns `ROLLBACK TO SAVEPOINT ruststream_claim`, the SQL standard's
849 /// form, which goes back to the savepoint the provided [`savepoint`](Self::savepoint) sets; a
850 /// dialect that sets its savepoint another way returns the statement that goes back to it.
851 ///
852 /// # Examples
853 ///
854 /// ```
855 /// # use std::num::NonZeroUsize;
856 /// use ruststream_sqlx_dialect::Dialect;
857 /// # use ruststream_sqlx_dialect::{Statement, StatementError, TableName, TableSpec};
858 ///
859 /// /// SQL Server, a database without a built-in dialect.
860 /// #[derive(Debug)]
861 /// pub struct Mssql;
862 ///
863 /// impl Dialect for Mssql {
864 /// fn name(&self) -> &'static str {
865 /// "mssql"
866 /// }
867 ///
868 /// fn savepoint(&self) -> &'static str {
869 /// "SAVE TRANSACTION ruststream_claim"
870 /// }
871 ///
872 /// // SQL Server goes back to a savepoint with `ROLLBACK TRANSACTION`, and the
873 /// // transaction stays open.
874 /// fn rollback_to_savepoint(&self) -> &'static str {
875 /// "ROLLBACK TRANSACTION ruststream_claim"
876 /// }
877 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
878 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
879 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
880 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
881 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
882 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
883 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
884 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
885 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
886 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
887 /// }
888 /// ```
889 fn rollback_to_savepoint(&self) -> &'static str {
890 "ROLLBACK TO SAVEPOINT ruststream_claim"
891 }
892
893 /// The statement a claim of a table with FIFO groups runs first in its transaction, to take
894 /// the subscription's group for it; `None` for a table without FIFO groups, or a dialect that
895 /// takes no group.
896 ///
897 /// It binds [`Param::Group`](crate::Param::Group) and returns one row whose first column is a
898 /// 64-bit integer: nonzero when the claim's transaction now holds the group, zero when another
899 /// transaction holds it. It does not wait. A claim that reads zero ends empty, so a row that
900 /// enters the group ahead of a head in work waits for that head to settle. The transaction
901 /// holds the group until it ends: in the row lock form until the delivery settles, in the
902 /// lease form until the claim commits, after which the lease claim takes nothing while a row
903 /// of the group holds a lease.
904 ///
905 /// A dialect of the service's own returns such a statement where its database can hold the
906 /// group for one transaction: a lock the transaction owns, or a locking read of the group's
907 /// rows. `None`, which the provided method returns, leaves the order to the claim alone. That
908 /// suffices where claims run one at a time and see every lease, as on SQLite. Where two claims
909 /// run side by side, it gives the order up: a row that enters the group ahead of a head in
910 /// work (a smaller `priority`, an earlier `retry_after`, a dead letter moved into the group)
911 /// becomes a second head, and the other claim takes it while the first is still in work.
912 ///
913 /// # Errors
914 ///
915 /// For a table with FIFO groups, the refusals of its claim: [`StatementError::AdvisoryFifo`]
916 /// in the advisory lock form, [`StatementError::LeaseOnDatabaseClock`] for a lease table on
917 /// the database's clock, [`StatementError::IdentifierTooLong`] for a name the database does
918 /// not keep. A guard may also refuse the isolation level its claim opens at: MySQL refuses a
919 /// row lock table at SERIALIZABLE with [`StatementError::FifoAtSerializable`].
920 ///
921 /// # Examples
922 ///
923 /// A dialect of the service's own takes the group with a lock its transaction owns, so a FIFO
924 /// group keeps one row in work:
925 ///
926 /// ```
927 /// # use std::num::NonZeroUsize;
928 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
929 /// # use ruststream_sqlx_dialect::TableName;
930 ///
931 /// /// SQL Server, a database without a built-in dialect.
932 /// #[derive(Debug)]
933 /// pub struct Mssql;
934 ///
935 /// impl Dialect for Mssql {
936 /// fn name(&self) -> &'static str {
937 /// "mssql"
938 /// }
939 ///
940 /// // An application lock on the table's name and the group, owned by the claim's
941 /// // transaction. `@LockTimeout = 0` answers at once, and a negative result means another
942 /// // transaction holds the group.
943 /// fn fifo_guard(
944 /// &self,
945 /// spec: &TableSpec<'_>,
946 /// ) -> Result<Option<Statement>, StatementError> {
947 /// if !spec.is_fifo() {
948 /// return Ok(None);
949 /// }
950 /// let table = spec.table().replace('\'', "''");
951 /// Ok(Some(Statement::new(
952 /// format!(
953 /// "DECLARE @resource nvarchar(255) = CONCAT(N'{table}:', @p1), @result int; \
954 /// EXEC @result = sp_getapplock @Resource = @resource, \
955 /// @LockMode = 'Exclusive', @LockOwner = 'Transaction', @LockTimeout = 0; \
956 /// SELECT CAST(CASE WHEN @result >= 0 THEN 1 ELSE 0 END AS bigint)"
957 /// ),
958 /// [Param::Group],
959 /// )))
960 /// }
961 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
962 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
963 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
964 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
965 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
966 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
967 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
968 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
969 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
970 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
971 /// }
972 /// ```
973 fn fifo_guard(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
974 let _ = spec;
975 Ok(None)
976 }
977}