ruststream_sqlx_dialect/spec/reading.rs
1//! What a description holds, read back: its names, its form, its switches, the column of each
2//! role and every column in order.
3
4use super::{Grouping, TableSpec};
5use crate::column::Column;
6use crate::form::Form;
7use crate::opening::Opening;
8use crate::role::Role;
9
10impl<'a> TableSpec<'a> {
11 /// The table's name, without its schema.
12 ///
13 /// # Examples
14 ///
15 /// ```
16 /// # use std::num::NonZeroUsize;
17 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
18 /// # use ruststream_sqlx_dialect::TableName;
19 ///
20 /// /// SQL Server, a database without a built-in dialect.
21 /// #[derive(Debug)]
22 /// pub struct Mssql;
23 ///
24 /// impl Dialect for Mssql {
25 /// fn name(&self) -> &'static str {
26 /// "mssql"
27 /// }
28 ///
29 /// // The service keeps its finished emails for an audit, in the `sent` group; every other
30 /// // table deletes its finished rows.
31 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
32 /// if spec.table() == "email_jobs" {
33 /// return Ok(Statement::new(
34 /// "UPDATE [email_jobs] SET [name] = 'sent' WHERE [job_id] = @p1",
35 /// [Param::Id],
36 /// ));
37 /// }
38 /// let mut sql = String::from("DELETE FROM ");
39 /// self.quote_into(spec.table(), &mut sql);
40 /// sql.push_str(" WHERE ");
41 /// self.quote_into(spec.id().name(), &mut sql);
42 /// sql.push_str(" = @p1");
43 /// Ok(Statement::new(sql, [Param::Id]))
44 /// }
45 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
46 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
47 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
48 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
49 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
50 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
51 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
52 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
53 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
54 /// }
55 /// ```
56 #[must_use]
57 pub const fn table(&self) -> &'a str {
58 self.table
59 }
60
61 /// The schema the table lives in, or `None` for the connection's default.
62 ///
63 /// # Examples
64 ///
65 /// ```
66 /// # use std::num::NonZeroUsize;
67 /// # use ruststream_sqlx_dialect::TableName;
68 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
69 ///
70 /// /// SQL Server, a database without a built-in dialect.
71 /// #[derive(Debug)]
72 /// pub struct Mssql;
73 ///
74 /// impl Dialect for Mssql {
75 /// fn name(&self) -> &'static str {
76 /// "mssql"
77 /// }
78 ///
79 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
80 /// // A table that names no schema lives in SQL Server's default one, `dbo`.
81 /// let mut sql = String::from("DELETE FROM ");
82 /// self.quote_into(spec.schema().unwrap_or("dbo"), &mut sql);
83 /// sql.push('.');
84 /// self.quote_into(spec.table(), &mut sql);
85 /// sql.push_str(" WHERE ");
86 /// self.quote_into(spec.id().name(), &mut sql);
87 /// sql.push_str(" = @p1");
88 /// Ok(Statement::new(sql, [Param::Id]))
89 /// }
90 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
91 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
92 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
93 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
94 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
95 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
96 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
97 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
98 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
99 /// }
100 /// ```
101 #[must_use]
102 pub const fn schema(&self) -> Option<&'a str> {
103 self.schema
104 }
105
106 /// The column that identifies a row.
107 ///
108 /// # Examples
109 ///
110 /// ```
111 /// # use std::num::NonZeroUsize;
112 /// # use ruststream_sqlx_dialect::TableName;
113 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
114 ///
115 /// /// SQL Server, a database without a built-in dialect.
116 /// #[derive(Debug)]
117 /// pub struct Mssql;
118 ///
119 /// impl Dialect for Mssql {
120 /// fn name(&self) -> &'static str {
121 /// "mssql"
122 /// }
123 ///
124 /// fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
125 /// // Every settlement names its row by the id column.
126 /// let mut sql = String::from("DELETE FROM ");
127 /// self.quote_into(spec.table(), &mut sql);
128 /// sql.push_str(" WHERE ");
129 /// self.quote_into(spec.id().name(), &mut sql);
130 /// sql.push_str(" = @p1");
131 /// Ok(Statement::new(sql, [Param::Id]))
132 /// }
133 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
134 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
135 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
136 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
137 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
138 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
139 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
140 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
141 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
142 /// }
143 /// ```
144 #[must_use]
145 pub const fn id(&self) -> Column<'a> {
146 self.id
147 }
148
149 /// The form the table's rows are claimed in.
150 ///
151 /// # Examples
152 ///
153 /// ```
154 /// # use std::num::NonZeroUsize;
155 /// # use ruststream_sqlx_dialect::TableName;
156 /// use ruststream_sqlx_dialect::{Dialect, Form, Param, Statement, StatementError, TableSpec};
157 ///
158 /// /// SQL Server, a database without a built-in dialect.
159 /// #[derive(Debug)]
160 /// pub struct Mssql;
161 ///
162 /// impl Dialect for Mssql {
163 /// fn name(&self) -> &'static str {
164 /// "mssql"
165 /// }
166 ///
167 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
168 /// let mut sql = String::from("DELETE FROM ");
169 /// self.quote_into(spec.table(), &mut sql);
170 /// sql.push_str(" WHERE ");
171 /// self.quote_into(spec.id().name(), &mut sql);
172 /// sql.push_str(" = @p1");
173 /// match spec.form() {
174 /// // In the lease form the row goes only while it holds the delivery's lease.
175 /// Form::Lease(expiry) => {
176 /// sql.push_str(" AND ");
177 /// self.quote_into(expiry.name(), &mut sql);
178 /// sql.push_str(" = @p2");
179 /// Ok(Statement::new(sql, [Param::Id, Param::Held]))
180 /// }
181 /// _ => Ok(Statement::new(sql, [Param::Id])),
182 /// }
183 /// }
184 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
185 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
186 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
187 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
188 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
189 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
190 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
191 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
192 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
193 /// }
194 /// ```
195 #[must_use]
196 pub const fn form(&self) -> Form<'a> {
197 self.form
198 }
199
200 /// What the table's transactions open at: [`Opening::Default`] unless the table names an
201 /// isolation level or a mode.
202 ///
203 /// # Examples
204 ///
205 /// ```
206 /// # use std::num::NonZeroUsize;
207 /// use ruststream_sqlx_dialect::{
208 /// ClaimShape, Dialect, Isolation, Opening, Param, RowLock, Statement, StatementError,
209 /// TableSpec,
210 /// };
211 /// # use ruststream_sqlx_dialect::TableName;
212 ///
213 /// /// SQL Server, a database without a built-in dialect.
214 /// #[derive(Debug)]
215 /// pub struct Mssql;
216 ///
217 /// impl RowLock for Mssql {
218 /// // The claim of the service's one queue table, which it reads whole.
219 /// fn lock_claim(
220 /// &self,
221 /// spec: &TableSpec<'_>,
222 /// _shape: ClaimShape,
223 /// ) -> Result<Statement, StatementError> {
224 /// // READPAST, which skips the rows another claim holds, refuses SERIALIZABLE.
225 /// let opening = spec.opening();
226 /// if opening == Opening::Isolation(Isolation::Serializable) {
227 /// return Err(StatementError::UnsupportedOpening {
228 /// dialect: self.name(),
229 /// opening: opening.name(),
230 /// });
231 /// }
232 /// Ok(Statement::new(
233 /// "SELECT TOP (@p1) [job_id], [payload] FROM [email_jobs] \
234 /// WITH (UPDLOCK, READPAST) ORDER BY [job_id]",
235 /// [Param::Limit],
236 /// ))
237 /// }
238 /// }
239 /// # impl Dialect for Mssql {
240 /// # fn name(&self) -> &'static str { "mssql" }
241 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
242 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
243 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
244 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
245 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
246 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
247 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
248 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
249 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
250 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
251 /// # }
252 /// ```
253 #[must_use]
254 pub const fn opening(&self) -> Opening {
255 self.opening
256 }
257
258 /// Whether each group of the table keeps its order: at most one row of a group is in work.
259 ///
260 /// # Examples
261 ///
262 /// ```
263 /// # use std::num::NonZeroUsize;
264 /// use ruststream_sqlx_dialect::{
265 /// ClaimShape, Dialect, Param, RowLock, Statement, StatementError, TableSpec,
266 /// };
267 /// # use ruststream_sqlx_dialect::TableName;
268 ///
269 /// /// SQL Server, a database without a built-in dialect.
270 /// #[derive(Debug)]
271 /// pub struct Mssql;
272 ///
273 /// impl RowLock for Mssql {
274 /// // The claim of the service's one queue table, which it reads whole.
275 /// fn lock_claim(
276 /// &self,
277 /// spec: &TableSpec<'_>,
278 /// _shape: ClaimShape,
279 /// ) -> Result<Statement, StatementError> {
280 /// // This claim keeps no group in order.
281 /// if spec.is_fifo() {
282 /// return Err(StatementError::UnsupportedFifo { dialect: self.name() });
283 /// }
284 /// Ok(Statement::new(
285 /// "SELECT TOP (@p1) [job_id], [payload] FROM [email_jobs] \
286 /// WITH (UPDLOCK, READPAST) ORDER BY [job_id]",
287 /// [Param::Limit],
288 /// ))
289 /// }
290 /// }
291 /// # impl Dialect for Mssql {
292 /// # fn name(&self) -> &'static str { "mssql" }
293 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
294 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
295 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
296 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
297 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
298 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
299 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
300 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
301 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
302 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
303 /// # }
304 /// ```
305 #[must_use]
306 pub const fn is_fifo(&self) -> bool {
307 matches!(self.grouping, Grouping::Fifo(_))
308 }
309
310 /// Whether statements select `*` instead of listing the columns.
311 ///
312 /// # Examples
313 ///
314 /// ```
315 /// # use std::num::NonZeroUsize;
316 /// use ruststream_sqlx_dialect::{
317 /// Dialect, Param, Statement, StatementError, TableName, TableSpec,
318 /// };
319 ///
320 /// /// SQL Server, a database without a built-in dialect.
321 /// #[derive(Debug)]
322 /// pub struct Mssql;
323 ///
324 /// impl Dialect for Mssql {
325 /// fn name(&self) -> &'static str {
326 /// "mssql"
327 /// }
328 ///
329 /// fn dead_letter_table(
330 /// &self,
331 /// spec: &TableSpec<'_>,
332 /// target: TableName<'_>,
333 /// ) -> Result<Vec<Statement>, StatementError> {
334 /// // A struct that flattens another hides columns, so its row moves by position.
335 /// let mut columns = String::new();
336 /// if spec.selects_all() {
337 /// columns.push('*');
338 /// } else {
339 /// for column in spec.columns() {
340 /// if !columns.is_empty() {
341 /// columns.push_str(", ");
342 /// }
343 /// self.quote_into(column.name(), &mut columns);
344 /// }
345 /// }
346 /// // A target that names no schema lives in SQL Server's default one, `dbo`.
347 /// let mut into = String::new();
348 /// self.quote_into(target.schema().unwrap_or("dbo"), &mut into);
349 /// into.push('.');
350 /// self.quote_into(target.table(), &mut into);
351 /// let mut from = String::new();
352 /// self.quote_into(spec.table(), &mut from);
353 /// let mut id = String::new();
354 /// self.quote_into(spec.id().name(), &mut id);
355 /// let into_columns = if spec.selects_all() {
356 /// String::new()
357 /// } else {
358 /// format!(" ({columns})")
359 /// };
360 /// Ok(vec![
361 /// Statement::new(
362 /// format!(
363 /// "INSERT INTO {into}{into_columns} SELECT {columns} FROM {from} \
364 /// WHERE {id} = @p1"
365 /// ),
366 /// [Param::Id],
367 /// ),
368 /// Statement::new(format!("DELETE FROM {from} WHERE {id} = @p1"), [Param::Id]),
369 /// ])
370 /// }
371 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
372 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
373 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
374 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
375 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
376 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
377 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
378 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
379 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
380 /// }
381 /// ```
382 #[must_use]
383 pub const fn selects_all(&self) -> bool {
384 self.select_all
385 }
386
387 /// Whether the statements read the database's own clock.
388 ///
389 /// # Examples
390 ///
391 /// ```
392 /// # use std::num::NonZeroUsize;
393 /// # use ruststream_sqlx_dialect::TableName;
394 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
395 ///
396 /// /// SQL Server, a database without a built-in dialect.
397 /// #[derive(Debug)]
398 /// pub struct Mssql;
399 ///
400 /// impl Dialect for Mssql {
401 /// fn name(&self) -> &'static str {
402 /// "mssql"
403 /// }
404 ///
405 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
406 /// // A table on the database's clock reads the server's time; otherwise the service
407 /// // binds it.
408 /// if spec.uses_database_clock() {
409 /// return Ok(Statement::new(
410 /// "UPDATE [email_jobs] SET [processed_at] = SYSUTCDATETIME() \
411 /// WHERE [job_id] = @p1",
412 /// [Param::Id],
413 /// ));
414 /// }
415 /// Ok(Statement::new(
416 /// "UPDATE [email_jobs] SET [processed_at] = @p1 WHERE [job_id] = @p2",
417 /// [Param::Now, Param::Id],
418 /// ))
419 /// }
420 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
421 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
422 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
423 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
424 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
425 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
426 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
427 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
428 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
429 /// }
430 /// ```
431 #[must_use]
432 pub const fn uses_database_clock(&self) -> bool {
433 self.database_clock
434 }
435
436 /// The message's data columns, in the order the description lists them.
437 ///
438 /// # Examples
439 ///
440 /// ```
441 /// use ruststream_sqlx_dialect::{Column, Form, TableSpec};
442 ///
443 /// const EMAILS: TableSpec<'static> =
444 /// TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
445 /// .data(&[Column::new("recipient"), Column::new("subject")]);
446 ///
447 /// let names: Vec<&str> = EMAILS.data_columns().iter().map(|column| column.name()).collect();
448 /// assert_eq!(names, ["recipient", "subject"]);
449 /// ```
450 #[must_use]
451 pub const fn data_columns(&self) -> &'a [Column<'a>] {
452 self.data
453 }
454
455 /// The columns only a message assembled from the table reads, in the order the description
456 /// lists them ([`fetching`](Self::fetching)).
457 ///
458 /// # Examples
459 ///
460 /// ```
461 /// use ruststream_sqlx_dialect::{Column, Form, TableSpec};
462 ///
463 /// const ORDERS: TableSpec<'static> =
464 /// TableSpec::new("order_jobs", Column::new("job_id"), Form::RowLock)
465 /// .data(&[Column::new("tenant")])
466 /// .fetching(&[Column::new("note")]);
467 ///
468 /// let names: Vec<&str> = ORDERS.fetched_columns().iter().map(|column| column.name()).collect();
469 /// assert_eq!(names, ["note"]);
470 /// ```
471 #[must_use]
472 pub const fn fetched_columns(&self) -> &'a [Column<'a>] {
473 self.fetched
474 }
475
476 /// The column that plays `role`, or `None` when the table has none.
477 ///
478 /// The `locked_until` column is the one the lease form carries.
479 ///
480 /// # Examples
481 ///
482 /// ```
483 /// # use std::num::NonZeroUsize;
484 /// # use ruststream_sqlx_dialect::TableName;
485 /// use ruststream_sqlx_dialect::{Dialect, Param, Role, Statement, StatementError, TableSpec};
486 ///
487 /// /// SQL Server, a database without a built-in dialect.
488 /// #[derive(Debug)]
489 /// pub struct Mssql;
490 ///
491 /// impl Dialect for Mssql {
492 /// fn name(&self) -> &'static str {
493 /// "mssql"
494 /// }
495 ///
496 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
497 /// let mut table = String::new();
498 /// self.quote_into(spec.table(), &mut table);
499 /// let mut id = String::new();
500 /// self.quote_into(spec.id().name(), &mut id);
501 /// // A table with a `processed_at` column keeps its finished rows and marks them.
502 /// let Some(done) = spec.column(Role::ProcessedAt) else {
503 /// return Ok(Statement::new(
504 /// format!("DELETE FROM {table} WHERE {id} = @p1"),
505 /// [Param::Id],
506 /// ));
507 /// };
508 /// let mut done_at = String::new();
509 /// self.quote_into(done.name(), &mut done_at);
510 /// Ok(Statement::new(
511 /// format!("UPDATE {table} SET {done_at} = @p1 WHERE {id} = @p2"),
512 /// [Param::Now, Param::Id],
513 /// ))
514 /// }
515 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
516 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
517 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
518 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
519 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
520 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
521 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
522 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
523 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
524 /// }
525 /// ```
526 #[must_use]
527 pub const fn column(&self, role: Role) -> Option<Column<'a>> {
528 match role {
529 Role::Id => Some(self.id),
530 Role::Group => match self.grouping {
531 Grouping::None => None,
532 Grouping::Groups(column) | Grouping::Fifo(column) => Some(column),
533 },
534 Role::PartitionKey => self.partition_key,
535 Role::Priority => self.priority,
536 Role::RetryAfter => self.retry_after,
537 Role::Attempt => self.attempt,
538 Role::LockedUntil => match self.form {
539 Form::Lease(column) => Some(column),
540 Form::RowLock | Form::Advisory(_) => None,
541 },
542 Role::ProcessedAt => self.processed_at,
543 Role::Headers => self.headers,
544 Role::Payload => self.payload,
545 }
546 }
547
548 /// Every column: the one of each role in [`Role::ALL`] order, then the message's data, then
549 /// the columns a message assembled from the table reads ([`fetching`](Self::fetching)).
550 ///
551 /// # Examples
552 ///
553 /// ```
554 /// # use ruststream_sqlx_dialect::TableName;
555 /// use std::num::NonZeroUsize;
556 ///
557 /// use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};
558 ///
559 /// /// SQL Server, a database without a built-in dialect.
560 /// #[derive(Debug)]
561 /// pub struct Mssql;
562 ///
563 /// impl Dialect for Mssql {
564 /// fn name(&self) -> &'static str {
565 /// "mssql"
566 /// }
567 ///
568 /// fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
569 /// if spec.selects_all() {
570 /// return Err(StatementError::Flattened { statement: "insert" });
571 /// }
572 /// let mut names = String::new();
573 /// let mut values = String::new();
574 /// let mut params = Vec::new();
575 /// // Every column the database does not fill, bound by its position in the
576 /// // description.
577 /// for (index, column) in spec.columns().enumerate() {
578 /// if column.is_generated() {
579 /// continue;
580 /// }
581 /// if !params.is_empty() {
582 /// names.push_str(", ");
583 /// values.push_str(", ");
584 /// }
585 /// self.quote_into(column.name(), &mut names);
586 /// let position = NonZeroUsize::MIN.saturating_add(params.len());
587 /// self.placeholder_into(position, &mut values);
588 /// params.push(Param::Column(index));
589 /// }
590 /// let mut table = String::new();
591 /// self.quote_into(spec.table(), &mut table);
592 /// let sql = format!("INSERT INTO {table} ({names}) VALUES ({values})");
593 /// Ok(Statement::new(sql, params))
594 /// }
595 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
596 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
597 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
598 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
599 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
600 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
601 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
602 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
603 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
604 /// }
605 /// ```
606 pub fn columns(&self) -> impl Iterator<Item = Column<'a>> {
607 Columns::new(self)
608 }
609}
610
611/// Every column of a description in [`TableSpec::columns`] order, walked by a `const fn` as well
612/// as by an iterator, so a `const` insert and a statement built at run time list the same columns.
613#[derive(Debug, Clone)]
614pub(crate) struct Columns<'s, 'a> {
615 spec: &'s TableSpec<'a>,
616 role: usize,
617 data: usize,
618 fetched: usize,
619}
620
621impl<'s, 'a> Columns<'s, 'a> {
622 pub(crate) const fn new(spec: &'s TableSpec<'a>) -> Self {
623 Self {
624 spec,
625 role: 0,
626 data: 0,
627 fetched: 0,
628 }
629 }
630
631 /// The next column, or `None` past the last.
632 pub(crate) const fn next_column(&mut self) -> Option<Column<'a>> {
633 while self.role < Role::ALL.len() {
634 let role = Role::ALL[self.role];
635 self.role += 1;
636 if let Some(column) = self.spec.column(role) {
637 return Some(column);
638 }
639 }
640 if self.data < self.spec.data.len() {
641 self.data += 1;
642 return Some(self.spec.data[self.data - 1]);
643 }
644 if self.fetched < self.spec.fetched.len() {
645 self.fetched += 1;
646 return Some(self.spec.fetched[self.fetched - 1]);
647 }
648 None
649 }
650}
651
652impl<'a> Iterator for Columns<'_, 'a> {
653 type Item = Column<'a>;
654
655 fn next(&mut self) -> Option<Column<'a>> {
656 self.next_column()
657 }
658}