Struct TableSpec 

pub struct TableSpec<'a> { /* private fields */ }

The description of a queue table: its name, the column that identifies a row, one slot per role, the message’s data columns, the form its rows are claimed in, and what its transactions open at.

#[derive(Inbox)] builds one as a constant, and a dialect reads it to build statements. The id column is part of the constructor and every role has one slot, which its setter fills once (a second call panics), so a table without an id or with a role played twice cannot be described. Column names are strings, so the types do not catch a name used twice or a lock key reading a column the table lacks: #[derive(Inbox)] refuses both at compile time, and a description written by hand meets the database’s own checks when the statements are prepared.

Examples

A dialect of the service’s own reads the description to build a statement:

use ruststream_sqlx_dialect::{
    Dialect, Form, Param, Role, Statement, StatementError, TableSpec,
};

/// SQL Server, a database without a built-in dialect, keeping the finished rows of a table with
/// groups in its `done` group.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        // The service's tables take their rows by row lock.
        if spec.form() != Form::RowLock {
            return Err(StatementError::UnsupportedForm {
                dialect: self.name(),
                form: spec.form().name(),
            });
        }
        // A table that names no schema lives in SQL Server's default one, `dbo`.
        let mut table = String::new();
        self.quote_into(spec.schema().unwrap_or("dbo"), &mut table);
        table.push('.');
        self.quote_into(spec.table(), &mut table);
        let mut id = String::new();
        self.quote_into(spec.id().name(), &mut id);
        let Some(group) = spec.column(Role::Group) else {
            return Ok(Statement::new(
                format!("DELETE FROM {table} WHERE {id} = @p1"),
                [Param::Id],
            ));
        };
        let mut done = String::new();
        self.quote_into(group.name(), &mut done);
        Ok(Statement::new(
            format!("UPDATE {table} SET {done} = 'done' WHERE {id} = @p1"),
            [Param::Id],
        ))
    }

    fn quote_into(&self, ident: &str, out: &mut String) {
        out.push('[');
        out.push_str(&ident.replace(']', "]]"));
        out.push(']');
    }
}

Implementations

impl<'a> TableSpec<'a>

pub const fn new(table: &'a str, id: Column<'a>, form: Form<'a>) -> Self

A table in the connection’s default schema, with the column that identifies a row and the form its rows are claimed in.

#[derive(Inbox)] builds the description from #[inbox(table = "..")], the #[field(id)] field and the form the struct declares.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// The service's email queue, its rows taken by row lock.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .payload(Column::new("payload"));

// A row the handler acknowledged leaves the table.
let ack = Postgres.ack(&EMAILS)?;
assert_eq!(ack.sql(), r#"DELETE FROM "email_jobs" WHERE "job_id" = $1"#);

pub const fn within(self, schema: &'a str) -> Self

The same table, inside schema.

#[inbox(schema = "app")] sets it.

Panics

When the description already names a schema. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// The queue lives in the `app` schema, and every statement names it.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .within("app")
        .payload(Column::new("payload"));

let ack = Postgres.ack(&EMAILS)?;
assert_eq!(ack.sql(), r#"DELETE FROM "app"."email_jobs" WHERE "job_id" = $1"#);

pub const fn group(self, column: Column<'a>) -> Self

The same table, split into groups by column; a subscription reads one group.

#[field(group)] on a field sets it.

Panics

When the description already splits into groups, by this setter or by fifo_group. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{
    ClaimShape, Column, Form, Param, Postgres, RowLock, TableSpec,
};

// One table serves several queues; a subscription binds its queue's name.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .group(Column::new("queue"))
        .payload(Column::new("payload"));

let claim = Postgres.lock_claim(&EMAILS, ClaimShape::Ids)?;
assert_eq!(
    claim.sql(),
    r#"SELECT "job_id" FROM "email_jobs" WHERE "queue" = $1 ORDER BY "job_id" LIMIT $2 FOR UPDATE SKIP LOCKED"#
);
assert_eq!(claim.params(), [Param::Group, Param::Limit]);

pub const fn fifo_group(self, column: Column<'a>) -> Self

The same table, split into groups by column, with each group in order: at most one row of a group is in work, taken in claim order.

A claim takes the group’s head, its first unfinished row in claim order, and takes nothing while a row of the group is in work or the head is not yet due: the claim’s transaction first takes the group (Dialect::fifo_guard), and a lease claim waits while a row of the group holds a lease. A delayed retry gives its row a later retry_after, so the row moves behind the rows of its group due earlier.

The advisory lock form keeps a group in order through its lock key instead, so its claim and its take refuse a FIFO group (StatementError::AdvisoryFifo).

#[field(group, fifo = true)] on a field sets it.

Panics

When the description already splits into groups, by this setter or by group. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Param, Postgres, TableSpec};

// The emails of one customer go out in order.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .fifo_group(Column::new("customer"))
        .payload(Column::new("payload"));

// The claim's transaction first takes the customer's group.
let guard = Postgres.fifo_guard(&EMAILS)?;
assert_eq!(guard.map(|guard| guard.params().to_vec()), Some(vec![Param::Group]));

pub const fn partition_key(self, column: Column<'a>) -> Self

The same table, with the delivery’s partition key read from column.

#[field(partition_key)] on a field sets it.

Panics

When the description already has a partition_key column: a role belongs to one column. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{ClaimShape, Column, Form, Postgres, RowLock, TableSpec};

// The emails of one tenant run in one lane of `workers(n, by_key)`.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .partition_key(Column::new("tenant"))
        .payload(Column::new("payload"));

let claim = Postgres.lock_claim(&EMAILS, ClaimShape::Roles)?;
assert_eq!(
    claim.sql(),
    r#"SELECT "job_id" AS "id", "tenant" AS "partition_key", "payload" AS "payload" FROM "email_jobs" ORDER BY "job_id" LIMIT $1 FOR UPDATE SKIP LOCKED"#
);

pub const fn priority(self, column: Column<'a>) -> Self

The same table, with rows claimed in the order of column, a smaller value first.

#[field(priority)] on a field sets it.

Panics

When the description already has a priority column: a role belongs to one column. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{ClaimShape, Column, Form, Postgres, RowLock, TableSpec};

// An email with a smaller `priority` goes out first.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .priority(Column::new("priority"))
        .payload(Column::new("payload"));

let claim = Postgres.lock_claim(&EMAILS, ClaimShape::Ids)?;
assert_eq!(
    claim.sql(),
    r#"SELECT "job_id" FROM "email_jobs" ORDER BY "priority", "job_id" LIMIT $1 FOR UPDATE SKIP LOCKED"#
);

pub const fn retry_after(self, column: Column<'a>) -> Self

The same table, where column holds the time before which a row is not claimed.

#[field(retry_after)] on a field sets it.

Panics

When the description already has a retry_after column: a role belongs to one column. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{
    ClaimShape, Column, Form, Param, Postgres, RowLock, TableSpec,
};

// An email waits in the table until its `run_at`.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .retry_after(Column::new("run_at"))
        .payload(Column::new("payload"));

let claim = Postgres.lock_claim(&EMAILS, ClaimShape::Ids)?;
assert_eq!(
    claim.sql(),
    r#"SELECT "job_id" FROM "email_jobs" WHERE "run_at" <= $1 ORDER BY "run_at", "job_id" LIMIT $2 FOR UPDATE SKIP LOCKED"#
);
assert_eq!(claim.params(), [Param::Now, Param::Limit]);

pub const fn attempt(self, column: Column<'a>) -> Self

The same table, where column counts the attempts.

#[field(attempt)] on a field sets it.

Panics

When the description already has a attempt column: a role belongs to one column. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// A retried email counts one more attempt.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .attempt(Column::new("attempt"))
        .payload(Column::new("payload"));

let retry = Postgres.retry(&EMAILS)?.map(|retry| retry.sql().to_owned());
assert_eq!(
    retry.as_deref(),
    Some(r#"UPDATE "email_jobs" SET "attempt" = "attempt" + 1 WHERE "job_id" = $1"#),
);

pub const fn processed_at(self, column: Column<'a>) -> Self

The same table, where acknowledgement sets column instead of deleting the row.

#[field(processed_at)] on a field sets it.

Panics

When the description already has a processed_at column: a role belongs to one column. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Param, Postgres, TableSpec};

// A sent email stays in the table, stamped with the time it was sent.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .processed_at(Column::new("sent_at"))
        .payload(Column::new("payload"));

let ack = Postgres.ack(&EMAILS)?;
assert_eq!(ack.sql(), r#"UPDATE "email_jobs" SET "sent_at" = $1 WHERE "job_id" = $2"#);
assert_eq!(ack.params(), [Param::Now, Param::Id]);

pub const fn headers(self, column: Column<'a>) -> Self

The same table, where column holds the delivery’s headers.

#[field(headers)] on a field sets it.

Panics

When the description already has a headers column: a role belongs to one column. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// An email carries its headers in one column, which the insert writes.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .headers(Column::new("headers"))
        .payload(Column::new("payload"));

let insert = Postgres.insert(&EMAILS)?;
assert_eq!(
    insert.sql(),
    r#"INSERT INTO "email_jobs" ("job_id", "headers", "payload") VALUES ($1, $2, $3)"#
);

pub const fn payload(self, column: Column<'a>) -> Self

The same table, where column holds the message bytes a handler decodes.

#[field(payload)] on a field sets it.

Panics

When the description already has a payload column: a role belongs to one column. In a const the panic is a build error.

use ruststream_sqlx_dialect::{Column, Form, TableSpec};

// Two columns for one role do not build.
const JOBS: TableSpec<'static> = TableSpec::new("jobs", Column::new("id"), Form::RowLock)
    .payload(Column::new("payload"))
    .payload(Column::new("body"));

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// The handler decodes the email from `payload`, which the fetch reads.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .payload(Column::new("payload"));

let fetch = Postgres.fetch(&EMAILS)?;
assert_eq!(
    fetch.sql(),
    r#"SELECT "job_id", "payload" FROM "email_jobs" WHERE "job_id" = ANY($1)"#
);

pub const fn data(self, columns: &'a [Column<'a>]) -> Self

The same table, with the columns of the message’s own data: the columns without a role.

#[derive(Inbox)] passes every field without a role here.

Panics

When the description already lists data columns: the list is given once. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// The email's recipient and subject are columns of their own, after the roles.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .payload(Column::new("payload"))
        .data(&[Column::new("recipient"), Column::new("subject")]);

let insert = Postgres.insert(&EMAILS)?;
assert_eq!(
    insert.sql(),
    r#"INSERT INTO "email_jobs" ("job_id", "payload", "recipient", "subject") VALUES ($1, $2, $3, $4)"#
);

pub const fn fetching(self, columns: &'a [Column<'a>]) -> Self

The same table, with the columns a message assembled from it reads beside the headers struct’s: listed after the data, wherever the statements list the table’s columns.

#[derive(Inbox)] passes the message struct’s own fields here when the struct flattens a headers struct into its #[field(headers)] field, which describes the rest of the table. The default fetch then names every column it reads, so a column the table lacks stops the subscription when its statements are prepared.

Panics

When the description already lists fetched columns: the list is given once. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// The message flattens a headers struct into `headers` and reads `recipient` itself.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .headers(Column::new("headers"))
        .payload(Column::new("payload"))
        .fetching(&[Column::new("recipient")]);

let fetch = Postgres.fetch(&EMAILS)?;
assert_eq!(
    fetch.sql(),
    r#"SELECT "job_id", "headers", "payload", "recipient" FROM "email_jobs" WHERE "job_id" = ANY($1)"#
);

pub const fn selecting_all(self) -> Self

The same table, read with *: the struct flattens another, so the columns are not all known.

A dead-letter move to another table then copies the row by position, so that table has the same columns in the same order.

A struct with a #[sqlx(flatten)] field sets it.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Postgres, TableSpec};

// The email struct flattens another, so the fetch reads every column.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .payload(Column::new("payload"))
        .selecting_all();

let fetch = Postgres.fetch(&EMAILS)?;
assert_eq!(fetch.sql(), r#"SELECT * FROM "email_jobs" WHERE "job_id" = ANY($1)"#);

pub const fn database_clock(self) -> Self

Makes the statements read the database’s own clock instead of a time the service binds.

#[inbox(clock = DatabaseClock)] sets it.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Param, Postgres, TableSpec};

// A sent email is stamped by the database's clock, so the service binds no time.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .processed_at(Column::new("sent_at"))
        .payload(Column::new("payload"))
        .database_clock();

let ack = Postgres.ack(&EMAILS)?;
assert_eq!(
    ack.sql(),
    r#"UPDATE "email_jobs" SET "sent_at" = statement_timestamp() WHERE "job_id" = $1"#
);
assert_eq!(ack.params(), [Param::Id]);

pub const fn isolation(self, isolation: Isolation) -> Self

The same table, with its transactions opened at the isolation level isolation.

The row lock claim’s transaction opens at it, and so does the transaction a broker opens for a handler’s writes. A table opens at one level or in one mode.

#[inbox(isolation = repeatable_read)] sets it.

Panics

When the description already opens at a level or in a mode, by this setter or by mode. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Isolation, Postgres, TableSpec};

// The handler's reads of other tables see one snapshot.
const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .payload(Column::new("payload"))
        .isolation(Isolation::RepeatableRead);

let begin = Postgres.begin(EMAILS.opening())?;
assert_eq!(begin, Some("BEGIN ISOLATION LEVEL REPEATABLE READ"));

pub const fn mode(self, mode: Mode) -> Self

The same table, with its SQLite transactions opened in mode.

A table opens in one mode or at one level.

#[inbox(mode = immediate)] sets it.

Panics

When the description already opens at a level or in a mode, by this setter or by isolation. In a const the panic is a build error.

Examples

use ruststream_sqlx_dialect::{Column, Dialect, Form, Mode, Sqlite, TableSpec};

// The handler's transaction takes the write lock before its first read.
const EMAILS: TableSpec<'static> = TableSpec::new(
    "email_jobs",
    Column::new("job_id"),
    Form::Lease(Column::new("locked_until")),
)
.payload(Column::new("payload"))
.mode(Mode::Immediate);

let begin = Sqlite.begin(EMAILS.opening())?;
assert_eq!(begin, Some("BEGIN IMMEDIATE"));

impl<'a> TableSpec<'a>

pub const fn table(&self) -> &'a str

The table’s name, without its schema.

Examples

use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    // The service keeps its finished emails for an audit, in the `sent` group; every other
    // table deletes its finished rows.
    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        if spec.table() == "email_jobs" {
            return Ok(Statement::new(
                "UPDATE [email_jobs] SET [name] = 'sent' WHERE [job_id] = @p1",
                [Param::Id],
            ));
        }
        let mut sql = String::from("DELETE FROM ");
        self.quote_into(spec.table(), &mut sql);
        sql.push_str(" WHERE ");
        self.quote_into(spec.id().name(), &mut sql);
        sql.push_str(" = @p1");
        Ok(Statement::new(sql, [Param::Id]))
    }
}

pub const fn schema(&self) -> Option<&'a str>

The schema the table lives in, or None for the connection’s default.

Examples

use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        // A table that names no schema lives in SQL Server's default one, `dbo`.
        let mut sql = String::from("DELETE FROM ");
        self.quote_into(spec.schema().unwrap_or("dbo"), &mut sql);
        sql.push('.');
        self.quote_into(spec.table(), &mut sql);
        sql.push_str(" WHERE ");
        self.quote_into(spec.id().name(), &mut sql);
        sql.push_str(" = @p1");
        Ok(Statement::new(sql, [Param::Id]))
    }
}

pub const fn id(&self) -> Column<'a>

The column that identifies a row.

Examples

use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        // Every settlement names its row by the id column.
        let mut sql = String::from("DELETE FROM ");
        self.quote_into(spec.table(), &mut sql);
        sql.push_str(" WHERE ");
        self.quote_into(spec.id().name(), &mut sql);
        sql.push_str(" = @p1");
        Ok(Statement::new(sql, [Param::Id]))
    }
}

pub const fn form(&self) -> Form<'a>

The form the table’s rows are claimed in.

Examples

use ruststream_sqlx_dialect::{Dialect, Form, Param, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        let mut sql = String::from("DELETE FROM ");
        self.quote_into(spec.table(), &mut sql);
        sql.push_str(" WHERE ");
        self.quote_into(spec.id().name(), &mut sql);
        sql.push_str(" = @p1");
        match spec.form() {
            // In the lease form the row goes only while it holds the delivery's lease.
            Form::Lease(expiry) => {
                sql.push_str(" AND ");
                self.quote_into(expiry.name(), &mut sql);
                sql.push_str(" = @p2");
                Ok(Statement::new(sql, [Param::Id, Param::Held]))
            }
            _ => Ok(Statement::new(sql, [Param::Id])),
        }
    }
}

pub const fn opening(&self) -> Opening

What the table’s transactions open at: Opening::Default unless the table names an isolation level or a mode.

Examples

use ruststream_sqlx_dialect::{
    ClaimShape, Dialect, Isolation, Opening, Param, RowLock, Statement, StatementError,
    TableSpec,
};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl RowLock for Mssql {
    // The claim of the service's one queue table, which it reads whole.
    fn lock_claim(
        &self,
        spec: &TableSpec<'_>,
        _shape: ClaimShape,
    ) -> Result<Statement, StatementError> {
        // READPAST, which skips the rows another claim holds, refuses SERIALIZABLE.
        let opening = spec.opening();
        if opening == Opening::Isolation(Isolation::Serializable) {
            return Err(StatementError::UnsupportedOpening {
                dialect: self.name(),
                opening: opening.name(),
            });
        }
        Ok(Statement::new(
            "SELECT TOP (@p1) [job_id], [payload] FROM [email_jobs] \
             WITH (UPDLOCK, READPAST) ORDER BY [job_id]",
            [Param::Limit],
        ))
    }
}

pub const fn is_fifo(&self) -> bool

Whether each group of the table keeps its order: at most one row of a group is in work.

Examples

use ruststream_sqlx_dialect::{
    ClaimShape, Dialect, Param, RowLock, Statement, StatementError, TableSpec,
};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl RowLock for Mssql {
    // The claim of the service's one queue table, which it reads whole.
    fn lock_claim(
        &self,
        spec: &TableSpec<'_>,
        _shape: ClaimShape,
    ) -> Result<Statement, StatementError> {
        // This claim keeps no group in order.
        if spec.is_fifo() {
            return Err(StatementError::UnsupportedFifo { dialect: self.name() });
        }
        Ok(Statement::new(
            "SELECT TOP (@p1) [job_id], [payload] FROM [email_jobs] \
             WITH (UPDLOCK, READPAST) ORDER BY [job_id]",
            [Param::Limit],
        ))
    }
}

pub const fn selects_all(&self) -> bool

Whether statements select * instead of listing the columns.

Examples

use ruststream_sqlx_dialect::{
    Dialect, Param, Statement, StatementError, TableName, TableSpec,
};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn dead_letter_table(
        &self,
        spec: &TableSpec<'_>,
        target: TableName<'_>,
    ) -> Result<Vec<Statement>, StatementError> {
        // A struct that flattens another hides columns, so its row moves by position.
        let mut columns = String::new();
        if spec.selects_all() {
            columns.push('*');
        } else {
            for column in spec.columns() {
                if !columns.is_empty() {
                    columns.push_str(", ");
                }
                self.quote_into(column.name(), &mut columns);
            }
        }
        // A target that names no schema lives in SQL Server's default one, `dbo`.
        let mut into = String::new();
        self.quote_into(target.schema().unwrap_or("dbo"), &mut into);
        into.push('.');
        self.quote_into(target.table(), &mut into);
        let mut from = String::new();
        self.quote_into(spec.table(), &mut from);
        let mut id = String::new();
        self.quote_into(spec.id().name(), &mut id);
        let into_columns = if spec.selects_all() {
            String::new()
        } else {
            format!(" ({columns})")
        };
        Ok(vec![
            Statement::new(
                format!(
                    "INSERT INTO {into}{into_columns} SELECT {columns} FROM {from} \
                     WHERE {id} = @p1"
                ),
                [Param::Id],
            ),
            Statement::new(format!("DELETE FROM {from} WHERE {id} = @p1"), [Param::Id]),
        ])
    }
}

pub const fn uses_database_clock(&self) -> bool

Whether the statements read the database’s own clock.

Examples

use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        // A table on the database's clock reads the server's time; otherwise the service
        // binds it.
        if spec.uses_database_clock() {
            return Ok(Statement::new(
                "UPDATE [email_jobs] SET [processed_at] = SYSUTCDATETIME() \
                 WHERE [job_id] = @p1",
                [Param::Id],
            ));
        }
        Ok(Statement::new(
            "UPDATE [email_jobs] SET [processed_at] = @p1 WHERE [job_id] = @p2",
            [Param::Now, Param::Id],
        ))
    }
}

pub const fn data_columns(&self) -> &'a [Column<'a>]

The message’s data columns, in the order the description lists them.

Examples

use ruststream_sqlx_dialect::{Column, Form, TableSpec};

const EMAILS: TableSpec<'static> =
    TableSpec::new("email_jobs", Column::new("job_id"), Form::RowLock)
        .data(&[Column::new("recipient"), Column::new("subject")]);

let names: Vec<&str> = EMAILS.data_columns().iter().map(|column| column.name()).collect();
assert_eq!(names, ["recipient", "subject"]);

pub const fn fetched_columns(&self) -> &'a [Column<'a>]

The columns only a message assembled from the table reads, in the order the description lists them (fetching).

Examples

use ruststream_sqlx_dialect::{Column, Form, TableSpec};

const ORDERS: TableSpec<'static> =
    TableSpec::new("order_jobs", Column::new("job_id"), Form::RowLock)
        .data(&[Column::new("tenant")])
        .fetching(&[Column::new("note")]);

let names: Vec<&str> = ORDERS.fetched_columns().iter().map(|column| column.name()).collect();
assert_eq!(names, ["note"]);

pub const fn column(&self, role: Role) -> Option<Column<'a>>

The column that plays role, or None when the table has none.

The locked_until column is the one the lease form carries.

Examples

use ruststream_sqlx_dialect::{Dialect, Param, Role, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        let mut table = String::new();
        self.quote_into(spec.table(), &mut table);
        let mut id = String::new();
        self.quote_into(spec.id().name(), &mut id);
        // A table with a `processed_at` column keeps its finished rows and marks them.
        let Some(done) = spec.column(Role::ProcessedAt) else {
            return Ok(Statement::new(
                format!("DELETE FROM {table} WHERE {id} = @p1"),
                [Param::Id],
            ));
        };
        let mut done_at = String::new();
        self.quote_into(done.name(), &mut done_at);
        Ok(Statement::new(
            format!("UPDATE {table} SET {done_at} = @p1 WHERE {id} = @p2"),
            [Param::Now, Param::Id],
        ))
    }
}

pub fn columns(&self) -> impl Iterator<Item = Column<'a>>

Every column: the one of each role in Role::ALL order, then the message’s data, then the columns a message assembled from the table reads (fetching).

Examples

use std::num::NonZeroUsize;

use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Dialect for Mssql {
    fn name(&self) -> &'static str {
        "mssql"
    }

    fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        if spec.selects_all() {
            return Err(StatementError::Flattened { statement: "insert" });
        }
        let mut names = String::new();
        let mut values = String::new();
        let mut params = Vec::new();
        // Every column the database does not fill, bound by its position in the
        // description.
        for (index, column) in spec.columns().enumerate() {
            if column.is_generated() {
                continue;
            }
            if !params.is_empty() {
                names.push_str(", ");
                values.push_str(", ");
            }
            self.quote_into(column.name(), &mut names);
            let position = NonZeroUsize::MIN.saturating_add(params.len());
            self.placeholder_into(position, &mut values);
            params.push(Param::Column(index));
        }
        let mut table = String::new();
        self.quote_into(spec.table(), &mut table);
        let sql = format!("INSERT INTO {table} ({names}) VALUES ({values})");
        Ok(Statement::new(sql, params))
    }
}

Trait Implementations

impl<'a> Clone for TableSpec<'a>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<'a> Copy for TableSpec<'a>

impl<'a> Debug for TableSpec<'a>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'a> Eq for TableSpec<'a>

impl<'a> Hash for TableSpec<'a>

fn hash<__H: Hasher>(&self, state: &mut __H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl<'a> PartialEq for TableSpec<'a>

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl<'a> StructuralPartialEq for TableSpec<'a>