#![allow(clippy::upper_case_acronyms)]
#[cfg(feature = "clickhouse")]
use super::SingleSqlStatement;
use super::connection_pool;
use super::{SqlStatement, common_helpers};
use super::super::capabilities::Capabilities;
use crate::db_first;
use crate::model::{
Model, NoInclude, Relation, RelationHandle, RelationInfo, RelationPathInfo, RelationQuery,
RelationSelection, TableRouteValue, ThroughRelation, Tracked, Value, WritableModel,
routed_model_table_name_for_db,
};
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
use crate::query::builder::Select;
#[cfg(feature = "clickhouse")]
use crate::query::builder::ProjectionSelect;
use crate::query::builder::{
ContextFilter, DerivedSelect, DerivedTableSelect, FilterQuery, NamedFilterQuery, WhereExpr,
WithoutFilterQuery,
};
use crate::query::filter::FilterExpr;
use crate::query::insert::{IntoInsertAssignment, IntoInsertDefaultColumn};
use crate::raw_sql::{IntoRawSql, RawSql};
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
pub type TransactionFuture<'a, R> = Pin<Box<dyn Future<Output = crate::Result<R>> + Send + 'a>>;
pub type BatchQueryFuture<'a, T> = Pin<Box<dyn Future<Output = crate::Result<T>> + Send + 'a>>;
static SAVEPOINT_COUNTER: AtomicUsize = AtomicUsize::new(0);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IsolationLevel {
ReadUncommitted,
ReadCommitted,
RepeatableRead,
Serializable,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct TransactionOptions {
pub isolation: Option<IsolationLevel>,
pub read_only: bool,
}
impl TransactionOptions {
pub fn new() -> Self {
Self::default()
}
pub fn isolation(mut self, isolation: IsolationLevel) -> Self {
self.isolation = Some(isolation);
self
}
pub fn read_only(mut self) -> Self {
self.read_only = true;
self
}
pub fn serializable() -> Self {
Self::new().isolation(IsolationLevel::Serializable)
}
}
pub struct BatchFuture<'a, B> {
batch: B,
_marker: std::marker::PhantomData<&'a ()>,
}
impl<'a, B> BatchFuture<'a, B> {
pub(crate) fn new(batch: B) -> Self {
Self {
batch,
_marker: std::marker::PhantomData,
}
}
}
impl<'a, B> std::future::IntoFuture for BatchFuture<'a, B>
where
B: BatchQueries<'a> + Send + 'a,
{
type Output = crate::Result<B::Output>;
type IntoFuture = BatchQueryFuture<'a, B::Output>;
fn into_future(self) -> Self::IntoFuture {
self.batch.into_batch_future()
}
}
pub struct BatchManyFuture<'a, Q> {
queries: Vec<Q>,
_marker: std::marker::PhantomData<&'a ()>,
}
impl<'a, Q> BatchManyFuture<'a, Q> {
pub(crate) fn new<I>(queries: I) -> Self
where
I: IntoIterator<Item = Q>,
{
Self {
queries: queries.into_iter().collect(),
_marker: std::marker::PhantomData,
}
}
}
impl<'a, Q> std::future::IntoFuture for BatchManyFuture<'a, Q>
where
Q: BatchQuery<'a> + Send + 'a,
{
type Output = crate::Result<Vec<Q::Output>>;
type IntoFuture = BatchQueryFuture<'a, Vec<Q::Output>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move {
let mut results = Vec::with_capacity(self.queries.len());
for query in self.queries {
results.push(query.into_batch_future().await?);
}
Ok(results)
})
}
}
pub trait BatchQuery<'a>: Sized {
type Output: Send + 'a;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output>;
}
pub trait BatchQueries<'a>: Sized {
type Output: Send + 'a;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output>;
}
macro_rules! impl_batch_tuple {
($($name:ident => $var:ident),+ $(,)?) => {
impl<'a, $($name,)+> BatchQueries<'a> for ($($name,)+)
where
$($name: BatchQuery<'a> + Send + 'a,)+
{
type Output = ($($name::Output,)+);
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
let ($($var,)+) = self;
Box::pin(async move {
$(
let $var = $var.into_batch_future().await?;
)+
Ok(($($var,)+))
})
}
}
};
}
impl<'a> BatchQueries<'a> for () {
type Output = ();
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async { Ok(()) })
}
}
impl_batch_tuple!(A => a);
impl_batch_tuple!(A => a, B => b);
impl_batch_tuple!(A => a, B => b, C => c);
impl_batch_tuple!(A => a, B => b, C => c, D => d);
impl_batch_tuple!(A => a, B => b, C => c, D => d, E => e);
impl_batch_tuple!(A => a, B => b, C => c, D => d, E => e, F => f);
impl_batch_tuple!(A => a, B => b, C => c, D => d, E => e, F => f, G => g);
impl_batch_tuple!(A => a, B => b, C => c, D => d, E => e, F => f, G => g, H => h);
#[cfg(feature = "sqlite")]
use super::super::sqlite_backend;
#[cfg(feature = "postgresql")]
use super::super::postgresql_backend;
#[cfg(feature = "mysql")]
use super::super::mysql_backend;
#[cfg(feature = "mssql")]
use super::super::mssql_backend;
#[cfg(feature = "duckdb")]
use super::super::duckdb_backend;
#[cfg(feature = "clickhouse")]
use super::super::clickhouse_backend;
#[cfg(feature = "influxdb")]
use super::super::influxdb_backend;
fn relation_filter_values(values: Vec<Value>) -> Vec<crate::query::filter::Value> {
let mut seen = std::collections::HashSet::new();
values
.into_iter()
.filter(|value| !matches!(value, Value::Null))
.filter(|value| seen.insert(common_helpers::model_value_key(value)))
.collect()
}
#[allow(dead_code)]
fn unsupported_feature(backend: super::super::DbType, feature: &'static str) -> crate::OrmerError {
crate::OrmerError::UnsupportedFeature { backend, feature }
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
pub(crate) fn compute_pending_migrations<M: crate::migration::Migration>(
applied: Vec<crate::migration::MigrationInfo>,
migrations: &[M],
) -> crate::Result<Vec<crate::migration::MigrationInfo>> {
let applied = applied
.into_iter()
.map(|migration| (migration.version, migration.checksum))
.collect::<std::collections::BTreeMap<_, _>>();
let mut sorted = migrations.iter().collect::<Vec<_>>();
sorted.sort_by_key(|migration| migration.version());
let mut seen = std::collections::BTreeSet::new();
let mut pending = Vec::new();
for migration in sorted {
if !seen.insert(migration.version()) {
return Err(crate::ormer_error!(
"Duplicate migration version {}",
migration.version()
));
}
if let Some(checksum) = applied.get(&migration.version()) {
if *checksum != migration.checksum() {
return Err(crate::ormer_error!(
"Migration {} checksum changed after it was applied",
migration.version()
));
}
continue;
}
pending.push(crate::migration::MigrationInfo {
version: migration.version(),
name: migration.name().to_string(),
checksum: migration.checksum(),
});
}
Ok(pending)
}
pub(crate) fn primary_key_filter<T: Model>(
key: impl crate::model::PrimaryKey,
) -> crate::Result<WhereExpr> {
let pk_columns = T::primary_key_columns();
let pk_values = key.into_values();
if pk_columns.is_empty() {
return Err(crate::ormer_error!(
"Model {} does not have a primary key",
T::TABLE_NAME
));
}
if pk_columns.len() != pk_values.len() {
return Err(crate::ormer_error!(
"Primary key column count ({}) does not match value count ({})",
pk_columns.len(),
pk_values.len()
));
}
let filters = common_helpers::primary_key_filter_exprs(pk_columns, pk_values);
let Some(filter) = common_helpers::and_filter_exprs(filters) else {
return Err(crate::ormer_error!(
"Model {} does not have a primary key filter",
T::TABLE_NAME
));
};
Ok(WhereExpr::from_filter(filter))
}
pub(crate) fn relation_owner_key(path: RelationPathInfo) -> &'static RelationInfo {
match path {
RelationPathInfo::Direct { relation } => relation,
RelationPathInfo::Through { via_relation, .. } => via_relation,
}
}
pub(crate) async fn find_by_id_with_executor<T>(
exec: SelectExecutor<'_, T>,
key: impl crate::model::PrimaryKey,
) -> crate::Result<Option<T>>
where
T: Model + 'static + Send + Sync,
{
let where_expr = primary_key_filter::<T>(key)?;
let results = exec
.filter(|_| where_expr)
.range(..1)
.collect::<Vec<T>>()
.await?;
Ok(results.into_iter().next())
}
pub(crate) async fn find_related_with_executor<'a, T, S>(
exec: &SelectExecutor<'a, T>,
owner: &T,
relation: &S,
) -> crate::Result<Vec<S::Target>>
where
T: Model + 'static + Send + Sync,
S: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync,
S::Target: Send + Sync,
S::Via: Send + Sync,
{
let path = relation.path_info()?;
let key = owner.relation_key_value(relation_owner_key(path))?;
exec.select_related_with_selection(vec![key], relation).await
}
pub(crate) async fn preload_with_executor<'a, T, S>(
exec: &SelectExecutor<'a, T>,
owners: &mut [T],
relation: S,
) -> crate::Result<()>
where
T: Model + 'static + Send + Sync,
S: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync,
S::Target: Send + Sync,
S::Via: Send + Sync,
{
exec.preload_models_with_selection(owners, relation).await
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
pub(crate) fn clickhouse_projection_gate(
db_type: super::super::DbType,
) -> Option<&'static str> {
if Capabilities::of(db_type).advanced_grouping {
return None;
}
#[cfg(feature = "influxdb")]
if db_type == super::super::DbType::InfluxDB {
return Some(
"GROUP BY aggregation on InfluxDB (InfluxQL GROUP BY only supports time buckets and tags)",
);
}
Some("GROUP BY aggregation")
}
pub trait NestedInclude<'a, Owner: Model>: Clone {
fn load_nested_include<'b>(
self,
executor: &'b SelectExecutor<'a, Owner>,
owners: &'b mut [Owner],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
'a: 'b;
}
impl<'a, Owner: Model> NestedInclude<'a, Owner> for NoInclude {
fn load_nested_include<'b>(
self,
_executor: &'b SelectExecutor<'a, Owner>,
_owners: &'b mut [Owner],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
'a: 'b,
{
Box::pin(async { Ok(()) })
}
}
impl<'a, Owner, Target> NestedInclude<'a, Owner> for Relation<Owner, Target>
where
Owner: Model + 'static + Send + Sync,
Target: Model + Clone + 'static + Send + Sync,
{
fn load_nested_include<'b>(
self,
executor: &'b SelectExecutor<'a, Owner>,
owners: &'b mut [Owner],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
'a: 'b,
{
Box::pin(async move { executor.preload_models_with_selection(owners, self).await })
}
}
impl<'a, Owner, Via, Target> NestedInclude<'a, Owner> for ThroughRelation<Owner, Via, Target>
where
Owner: Model + 'static + Send + Sync,
Via: Model + Clone + 'static + Send + Sync,
Target: Model + Clone + 'static + Send + Sync,
{
fn load_nested_include<'b>(
self,
executor: &'b SelectExecutor<'a, Owner>,
owners: &'b mut [Owner],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
'a: 'b,
{
Box::pin(async move { executor.preload_models_with_selection(owners, self).await })
}
}
impl<'a, Owner, Target, Handle, Nested> NestedInclude<'a, Owner>
for RelationQuery<Owner, Target, Handle, Nested>
where
Owner: Model + 'static + Send + Sync,
Target: Model + Clone + 'static + Send + Sync,
Handle: RelationHandle<Owner, Target> + Clone + Send + Sync + 'static,
Handle::Via: Send + Sync,
Nested: NestedInclude<'a, Target> + Clone + Send + Sync + 'static,
{
fn load_nested_include<'b>(
self,
executor: &'b SelectExecutor<'a, Owner>,
owners: &'b mut [Owner],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
'a: 'b,
{
Box::pin(async move { executor.preload_models_with_selection(owners, self).await })
}
}
pub trait RelationNestedLoader<'a, Owner: Model>: RelationSelection<Owner> {
fn load_nested<'b>(
&'b self,
executor: &'b SelectExecutor<'a, Self::Target>,
related: &'b mut [Self::Target],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
Self::Target: Send + Sync,
'a: 'b;
}
impl<'a, Owner, Target> RelationNestedLoader<'a, Owner> for Relation<Owner, Target>
where
Owner: Model + 'static + Send + Sync,
Target: Model + Clone + 'static + Send + Sync,
{
fn load_nested<'b>(
&'b self,
_executor: &'b SelectExecutor<'a, Self::Target>,
_related: &'b mut [Self::Target],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
Self::Target: Send + Sync,
'a: 'b,
{
Box::pin(async { Ok(()) })
}
}
impl<'a, Owner, Via, Target> RelationNestedLoader<'a, Owner> for ThroughRelation<Owner, Via, Target>
where
Owner: Model + 'static + Send + Sync,
Via: Model + Clone + 'static + Send + Sync,
Target: Model + Clone + 'static + Send + Sync,
{
fn load_nested<'b>(
&'b self,
_executor: &'b SelectExecutor<'a, Self::Target>,
_related: &'b mut [Self::Target],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
Self::Target: Send + Sync,
'a: 'b,
{
Box::pin(async { Ok(()) })
}
}
impl<'a, Owner, Target, Handle, Nested> RelationNestedLoader<'a, Owner>
for RelationQuery<Owner, Target, Handle, Nested>
where
Owner: Model + 'static + Send + Sync,
Target: Model + Clone + 'static + Send + Sync,
Handle: RelationHandle<Owner, Target> + Clone + Send + Sync + 'static,
Handle::Via: Send + Sync,
Nested: NestedInclude<'a, Target> + Clone + Send + Sync + 'static,
{
fn load_nested<'b>(
&'b self,
executor: &'b SelectExecutor<'a, Self::Target>,
related: &'b mut [Self::Target],
) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send + 'b>>
where
Owner: 'static + Send + Sync,
Self::Target: Send + Sync,
'a: 'b,
{
let nested = self.nested().clone();
Box::pin(async move { nested.load_nested_include(executor, related).await })
}
}
pub enum Database {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::Database),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::Database),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::Database),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::Database),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::Database),
#[cfg(feature = "clickhouse")]
ClickHouse(super::super::clickhouse_backend::Database),
#[cfg(feature = "influxdb")]
InfluxDB(super::super::influxdb_backend::Database),
}
pub struct ReplicatedDatabaseBuilder {
db_type: super::super::DbType,
write_connection: Option<String>,
read_connections: Vec<String>,
}
pub struct ReplicatedDatabase {
db_type: super::super::DbType,
write: Database,
reads: Vec<Database>,
next_read: AtomicUsize,
}
impl ReplicatedDatabaseBuilder {
pub(crate) fn new(db_type: super::super::DbType) -> Self {
Self {
db_type,
write_connection: None,
read_connections: Vec::new(),
}
}
pub fn write(mut self, connection_string: impl Into<String>) -> Self {
self.write_connection = Some(connection_string.into());
self
}
pub fn read(mut self, connection_string: impl Into<String>) -> Self {
self.read_connections.push(connection_string.into());
self
}
pub async fn connect(self) -> crate::Result<ReplicatedDatabase> {
let Some(write_connection) = self.write_connection else {
return Err(crate::ormer_error!(
"replicated database requires a write connection"
));
};
let write = Database::connect(self.db_type, &write_connection).await?;
let mut reads = Vec::with_capacity(self.read_connections.len());
for connection in self.read_connections {
reads.push(Database::connect(self.db_type, &connection).await?);
}
Ok(ReplicatedDatabase {
db_type: self.db_type,
write,
reads,
next_read: AtomicUsize::new(0),
})
}
}
impl ReplicatedDatabase {
pub fn db_type(&self) -> super::super::DbType {
self.db_type
}
pub fn sql_trace(&self) -> crate::SqlTraceBuilder {
crate::global_sql_trace().builder()
}
pub fn write(&self) -> &Database {
&self.write
}
pub fn read(&self) -> &Database {
if self.reads.is_empty() {
return &self.write;
}
let index = self.next_read.fetch_add(1, Ordering::Relaxed) % self.reads.len();
&self.reads[index]
}
pub fn scope(&self) -> DatabaseScope<'_> {
self.write().scope()
}
pub async fn transaction<R, F>(&self, f: F) -> crate::Result<R>
where
F: for<'tx> FnOnce(&'tx mut Transaction<'_>) -> TransactionFuture<'tx, R>,
{
self.write.transaction(f).await
}
pub async fn transaction_opts<R, F>(
&self,
options: TransactionOptions,
f: F,
) -> crate::Result<R>
where
F: for<'tx> FnOnce(&'tx mut Transaction<'_>) -> TransactionFuture<'tx, R>,
{
self.write.transaction_opts(options, f).await
}
}
pub struct DerivedTableSelectExecutor<'a, R: Model> {
db: &'a Database,
select: DerivedTableSelect<R>,
}
#[derive(Clone)]
pub struct DatabaseScope<'a> {
db: &'a Database,
context_filters: Vec<ContextFilter>,
}
impl<'a> DatabaseScope<'a> {
pub fn select<T: Model>(&self) -> SelectExecutor<'a, T> {
self.db
.select::<T>()
.with_context_filters(self.context_filters.clone())
}
pub async fn find_by_id<T: Model + 'static + Send + Sync>(
&self,
key: impl crate::model::PrimaryKey,
) -> crate::Result<Option<T>> {
find_by_id_with_executor(self.select::<T>(), key).await
}
pub async fn find_related<T, S>(
&self,
owner: &T,
relation: S,
) -> crate::Result<Vec<S::Target>>
where
T: Model + 'static + Send + Sync,
S: RelationSelection<T>,
for<'b> S: RelationNestedLoader<'b, T> + Send + Sync,
S::Target: Send + Sync,
S::Via: Send + Sync,
{
find_related_with_executor(&self.select::<T>(), owner, &relation).await
}
pub async fn preload<T, S>(
&self,
owners: &mut [T],
relation: S,
) -> crate::Result<()>
where
T: Model + 'static + Send + Sync,
S: RelationSelection<T>,
for<'b> S: RelationNestedLoader<'b, T> + Send + Sync,
S::Target: Send + Sync,
S::Via: Send + Sync,
{
preload_with_executor(&self.select::<T>(), owners, relation).await
}
pub fn delete<T: WritableModel>(&self) -> ScopedDeleteExecutor<'a, T> {
ScopedDeleteExecutor {
inner: self.db.delete::<T>(),
context_filters: self.context_filters.clone(),
disabled_filters: Vec::new(),
}
}
pub fn update<T: WritableModel>(&self) -> ScopedUpdateExecutor<'a, T> {
ScopedUpdateExecutor {
inner: self.db.update::<T>(),
context_filters: self.context_filters.clone(),
disabled_filters: Vec::new(),
}
}
}
#[cfg(feature = "clickhouse")]
async fn clickhouse_select_models<T, C>(
db: &super::super::clickhouse_backend::Database,
select: Select<T>,
) -> crate::Result<C>
where
T: Model,
C: FromIterator<T>,
{
let (sql, params) = select.try_to_sql_with_params(super::super::DbType::ClickHouse)?;
let columns = T::columns();
let rows = db
.select_values(RawSql::new(sql).with_params(params), Some(&columns))
.await?;
rows.iter()
.map(|values| T::from_row_values(values))
.collect()
}
#[cfg(feature = "influxdb")]
async fn influx_select_models<T, C>(
db: &influxdb_backend::Database,
select: Select<T>,
) -> crate::Result<C>
where
T: Model,
C: FromIterator<T>,
{
let (sql, params) = select.try_to_sql_with_params(super::super::DbType::InfluxDB)?;
let columns = T::columns();
let rows = db
.select_values(RawSql::new(sql).with_params(params), Some(&columns))
.await?;
rows.iter()
.map(|values| T::from_row_values(values))
.collect()
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
async fn clickhouse_aggregate_on_backend<T, R>(
db: ClickHouseSelectBackend<'_>,
aggregate: crate::query::builder::AggregateSelect<T, R>,
) -> crate::Result<R>
where
T: Model,
R: crate::model::FromValue,
{
let backend = clickhouse_select_backend_db_type(db);
#[cfg_attr(not(feature = "influxdb"), allow(unused_mut))]
let (mut sql, params) = aggregate.try_to_sql_with_params(backend)?;
#[cfg(feature = "influxdb")]
if backend == super::super::DbType::InfluxDB && sql.starts_with("SELECT AVG(") {
sql = sql.replacen("SELECT AVG(", "SELECT MEAN(", 1);
}
let rows = match db {
#[cfg(feature = "clickhouse")]
ClickHouseSelectBackend::ClickHouse(db) => {
db.select_values(RawSql::new(sql).with_params(params), None).await?
}
#[cfg(feature = "influxdb")]
ClickHouseSelectBackend::Influx(db) => {
db.raw_select_values(RawSql::new(sql).with_params(params), None).await?
}
};
let value = rows
.first()
.and_then(|row| row.first().cloned())
.unwrap_or(Value::Null);
R::from_value(&value)
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
async fn clickhouse_select_models_on_backend<T, C>(
db: ClickHouseSelectBackend<'_>,
select: Select<T>,
) -> crate::Result<C>
where
T: Model,
C: FromIterator<T>,
{
match db {
#[cfg(feature = "clickhouse")]
ClickHouseSelectBackend::ClickHouse(db) => {
clickhouse_select_models::<T, C>(db, select).await
}
#[cfg(feature = "influxdb")]
ClickHouseSelectBackend::Influx(db) => influx_select_models::<T, C>(db, select).await,
}
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
async fn clickhouse_select_first_on_backend<T>(
db: ClickHouseSelectBackend<'_>,
select: Select<T>,
) -> crate::Result<Option<T>>
where
T: Model + Send + Sync,
{
Ok(
clickhouse_select_models_on_backend::<T, Vec<T>>(db, select)
.await?
.into_iter()
.next(),
)
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
async fn clickhouse_fetch_page_on_backend<T>(
db: ClickHouseSelectBackend<'_>,
select: Select<T>,
) -> crate::Result<crate::query::builder::CursorPage<T>>
where
T: Model + Send + Sync,
{
let (select, cursor_columns) = select.prepare_cursor_page()?;
let items =
clickhouse_select_models_on_backend::<T, Vec<T>>(db, select.clone()).await?;
select.finish_cursor_page(items, &cursor_columns)
}
impl<'a, T: Model> NamedFilterQuery<T> for DatabaseScope<'a> {
fn apply_named_filter(mut self, name: &'static str, expr: WhereExpr) -> Self {
self.context_filters
.push(ContextFilter::new::<T>(name, expr));
self
}
}
pub enum CreateTableExecutor<'a, T: crate::model::WritableModel> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::CreateTableExecutor<'a, T>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::CreateTableExecutor<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::CreateTableExecutor<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::CreateTableExecutor<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::CreateTableExecutor<'a, T>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a T>,
},
#[cfg(feature = "influxdb")]
InfluxDB(
&'a influxdb_backend::Database,
std::marker::PhantomData<&'a T>,
),
}
impl<'a, T: crate::model::WritableModel> CreateTableExecutor<'a, T> {
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn with_table_name(self, table_name: &str) -> Self {
match self {
#[cfg(feature = "sqlite")]
CreateTableExecutor::Sqlite(exec) => {
CreateTableExecutor::Sqlite(exec.with_table_name(table_name))
}
#[cfg(feature = "postgresql")]
CreateTableExecutor::PostgreSQL(exec) => {
CreateTableExecutor::PostgreSQL(exec.with_table_name(table_name))
}
#[cfg(feature = "mysql")]
CreateTableExecutor::MySQL(exec) => {
CreateTableExecutor::MySQL(exec.with_table_name(table_name))
}
#[cfg(feature = "mssql")]
CreateTableExecutor::MSSQL(exec) => {
CreateTableExecutor::MSSQL(exec.with_table_name(table_name))
}
#[cfg(feature = "duckdb")]
CreateTableExecutor::DuckDB(exec) => {
CreateTableExecutor::DuckDB(exec.with_table_name(table_name))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ CreateTableExecutor::Unsupported { .. } => unsupported,
#[cfg(feature = "influxdb")]
unsupported @ CreateTableExecutor::InfluxDB(..) => unsupported,
}
}
pub fn with_route_columnstore(self) -> Self {
match self {
#[cfg(feature = "postgresql")]
CreateTableExecutor::PostgreSQL(exec) => {
CreateTableExecutor::PostgreSQL(exec.with_route_columnstore())
}
#[allow(unreachable_patterns)]
other => other,
}
}
pub fn route_table(self, key: impl Into<String>, value: impl TableRouteValue) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
let db_type = match &self {
#[cfg(feature = "sqlite")]
CreateTableExecutor::Sqlite(_) => crate::abstract_layer::DbType::Sqlite,
#[cfg(feature = "postgresql")]
CreateTableExecutor::PostgreSQL(_) => crate::abstract_layer::DbType::PostgreSQL,
#[cfg(feature = "mysql")]
CreateTableExecutor::MySQL(_) => crate::abstract_layer::DbType::MySQL,
#[cfg(feature = "mssql")]
CreateTableExecutor::MSSQL(_) => crate::abstract_layer::DbType::MSSQL,
#[cfg(feature = "duckdb")]
CreateTableExecutor::DuckDB(_) => crate::abstract_layer::DbType::DuckDB,
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
CreateTableExecutor::Unsupported { backend, .. } => *backend,
#[cfg(feature = "influxdb")]
CreateTableExecutor::InfluxDB(..) => super::super::DbType::InfluxDB,
};
let table_name = routed_model_table_name_for_db::<T>(db_type, &route)
.unwrap_or_else(|err| panic!("Failed to render table route: {}", err));
self.with_table_name(&table_name)
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "sqlite")]
CreateTableExecutor::Sqlite(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
CreateTableExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
CreateTableExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
CreateTableExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
CreateTableExecutor::DuckDB(exec) => exec.to_sql(),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
CreateTableExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(*backend, feature)),
#[cfg(feature = "influxdb")]
CreateTableExecutor::InfluxDB(..) => Err(unsupported_feature(
super::super::DbType::InfluxDB,
"create_table to_sql (retention policy is applied over HTTP)",
)),
}
}
pub async fn execute(self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
CreateTableExecutor::Sqlite(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
CreateTableExecutor::PostgreSQL(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
CreateTableExecutor::MySQL(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
CreateTableExecutor::MSSQL(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
CreateTableExecutor::DuckDB(exec) => exec.execute().await,
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
CreateTableExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
#[cfg(feature = "influxdb")]
CreateTableExecutor::InfluxDB(db, _) => db.create_table::<T>().await,
}?;
crate::model::clear_version_snapshots::<T>();
Ok(())
}
}
pub enum DropTableExecutor<'a, T: crate::model::WritableModel> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::DropTableExecutor<'a, T>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::DropTableExecutor<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::DropTableExecutor<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::DropTableExecutor<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::DropTableExecutor<'a, T>),
#[cfg(feature = "clickhouse")]
ClickHouse(
&'a super::super::clickhouse_backend::Database,
std::marker::PhantomData<T>,
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a T>,
},
#[cfg(feature = "influxdb")]
InfluxDB(
&'a influxdb_backend::Database,
std::marker::PhantomData<&'a T>,
),
}
impl<'a, T: crate::model::WritableModel> DropTableExecutor<'a, T> {
#[allow(unused_variables)]
pub fn route_table(
self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
#[allow(unused_variables)]
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
match self {
#[cfg(feature = "sqlite")]
this @ DropTableExecutor::Sqlite(..) => this,
#[cfg(feature = "postgresql")]
DropTableExecutor::PostgreSQL(exec) => {
DropTableExecutor::PostgreSQL(exec.with_table_route(route))
}
#[cfg(feature = "mysql")]
this @ DropTableExecutor::MySQL(..) => this,
#[cfg(feature = "mssql")]
this @ DropTableExecutor::MSSQL(..) => this,
#[cfg(feature = "duckdb")]
this @ DropTableExecutor::DuckDB(..) => this,
#[cfg(feature = "clickhouse")]
this @ DropTableExecutor::ClickHouse(..) => this,
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ DropTableExecutor::Unsupported { .. } => unsupported,
#[cfg(feature = "influxdb")]
this @ DropTableExecutor::InfluxDB(..) => this,
}
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "sqlite")]
DropTableExecutor::Sqlite(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
DropTableExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
DropTableExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
DropTableExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
DropTableExecutor::DuckDB(exec) => exec.to_sql(),
#[cfg(feature = "clickhouse")]
DropTableExecutor::ClickHouse(_, _) => {
let table = crate::model::quote_qualified_identifier(
super::super::DbType::ClickHouse,
T::table_name_for_db(super::super::DbType::ClickHouse),
);
Ok(SqlStatement::single(
super::super::DbType::ClickHouse,
format!("DROP TABLE IF EXISTS {table}"),
Vec::new(),
))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
DropTableExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(*backend, feature)),
#[cfg(feature = "influxdb")]
DropTableExecutor::InfluxDB(..) => {
let table = crate::model::quote_qualified_identifier(
super::super::DbType::InfluxDB,
T::table_name_for_db(super::super::DbType::InfluxDB),
);
Ok(SqlStatement::single(
super::super::DbType::InfluxDB,
format!("DROP MEASUREMENT {table}"),
Vec::new(),
))
}
}
}
pub async fn execute(self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
DropTableExecutor::Sqlite(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
DropTableExecutor::PostgreSQL(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
DropTableExecutor::MySQL(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
DropTableExecutor::MSSQL(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
DropTableExecutor::DuckDB(exec) => exec.execute().await,
#[cfg(feature = "clickhouse")]
DropTableExecutor::ClickHouse(db, _) => db.drop_table::<T>().await,
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
DropTableExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
#[cfg(feature = "influxdb")]
DropTableExecutor::InfluxDB(db, _) => db.drop_table::<T>().await,
}?;
crate::model::clear_version_snapshots::<T>();
Ok(())
}
}
pub enum TruncateTableExecutor<'a, T: crate::model::WritableModel> {
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::TruncateTableExecutor<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::TruncateTableExecutor<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::TruncateTableExecutor<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::TruncateTableExecutor<'a, T>),
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a T>,
},
}
impl<'a, T: crate::model::WritableModel> TruncateTableExecutor<'a, T> {
#[allow(unused_variables)]
pub fn route_table(
self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
#[allow(unused_variables)]
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
match self {
#[cfg(feature = "postgresql")]
TruncateTableExecutor::PostgreSQL(exec) => {
TruncateTableExecutor::PostgreSQL(exec.with_table_route(route))
}
#[cfg(feature = "mysql")]
this @ TruncateTableExecutor::MySQL(..) => this,
#[cfg(feature = "mssql")]
this @ TruncateTableExecutor::MSSQL(..) => this,
#[cfg(feature = "duckdb")]
this @ TruncateTableExecutor::DuckDB(..) => this,
unsupported @ TruncateTableExecutor::Unsupported { .. } => unsupported,
}
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "postgresql")]
TruncateTableExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
TruncateTableExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
TruncateTableExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
TruncateTableExecutor::DuckDB(exec) => exec.to_sql(),
TruncateTableExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(*backend, feature)),
}
}
pub async fn execute(self) -> crate::Result<()> {
match self {
#[cfg(feature = "postgresql")]
TruncateTableExecutor::PostgreSQL(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
TruncateTableExecutor::MySQL(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
TruncateTableExecutor::MSSQL(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
TruncateTableExecutor::DuckDB(exec) => exec.execute().await,
TruncateTableExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
}?;
crate::model::clear_version_snapshots::<T>();
Ok(())
}
}
pub enum InsertExecutor<'a, I: crate::model::Insertable> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::InsertExecutor<'a, I>),
#[cfg(feature = "sqlite")]
SqliteTxn(sqlite_backend::TransactionInsertExecutor<'a, I>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::InsertExecutor<'a, I>),
#[cfg(feature = "postgresql")]
PostgreSQLTxn(postgresql_backend::TransactionInsertExecutor<'a, I>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::InsertExecutor<'a, I>),
#[cfg(feature = "mysql")]
MySQLTxn(mysql_backend::TransactionInsertExecutor<'a, I>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::InsertExecutor<'a, I>),
#[cfg(feature = "mssql")]
MSSQLTxn(mssql_backend::TransactionInsertExecutor<'a, I>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::InsertExecutor<'a, I>),
#[cfg(feature = "duckdb")]
DuckDBTxn(duckdb_backend::TransactionInsertExecutor<'a, I>),
#[cfg(feature = "clickhouse")]
ClickHouse(
&'a clickhouse_backend::Database,
I,
Option<crate::query::insert::InsertConflict>,
std::marker::PhantomData<I::Model>,
),
#[cfg(feature = "influxdb")]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
models: I,
_marker: std::marker::PhantomData<I::Model>,
_lifetime: std::marker::PhantomData<&'a ()>,
},
#[cfg(feature = "influxdb")]
InfluxDB(
&'a influxdb_backend::Database,
I,
Option<crate::query::insert::InsertConflict>,
std::marker::PhantomData<I::Model>,
),
}
#[deprecated(
since = "0.2.12",
note = "TransactionInsertExecutor 已合并为 InsertExecutor,请改用 InsertExecutor"
)]
pub type TransactionInsertExecutor<'a, I> = InsertExecutor<'a, I>;
pub enum InsertPartialExecutor<'a, T: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::InsertPartialExecutor<'a, T>,
std::marker::PhantomData<&'a T>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::InsertPartialExecutor<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::InsertPartialExecutor<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::InsertPartialExecutor<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::InsertPartialExecutor<'a, T>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a T>,
},
}
impl<'a, T: Model> InsertPartialExecutor<'a, T> {
#[allow(unused_variables)]
pub fn route_table(
self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
#[allow(unused_variables)]
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
match self {
#[cfg(feature = "sqlite")]
InsertPartialExecutor::Sqlite(exec, phantom) => {
InsertPartialExecutor::Sqlite(exec.with_table_route(route), phantom)
}
#[cfg(feature = "postgresql")]
InsertPartialExecutor::PostgreSQL(exec) => {
InsertPartialExecutor::PostgreSQL(exec.with_table_route(route))
}
#[cfg(feature = "mysql")]
InsertPartialExecutor::MySQL(exec) => {
InsertPartialExecutor::MySQL(exec.with_table_route(route))
}
#[cfg(feature = "mssql")]
InsertPartialExecutor::MSSQL(exec) => {
InsertPartialExecutor::MSSQL(exec.with_table_route(route))
}
#[cfg(feature = "duckdb")]
InsertPartialExecutor::DuckDB(exec) => {
InsertPartialExecutor::DuckDB(exec.with_table_route(route))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ InsertPartialExecutor::Unsupported { .. } => unsupported,
}
}
}
impl<'a, T: Model + Send + Sync> InsertPartialExecutor<'a, T> {
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn set<F, A>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> A,
A: IntoInsertAssignment<T>,
{
match self {
#[cfg(feature = "sqlite")]
InsertPartialExecutor::Sqlite(exec, phantom) => {
InsertPartialExecutor::Sqlite(exec.set(f), phantom)
}
#[cfg(feature = "postgresql")]
InsertPartialExecutor::PostgreSQL(exec) => {
InsertPartialExecutor::PostgreSQL(exec.set(f))
}
#[cfg(feature = "mysql")]
InsertPartialExecutor::MySQL(exec) => InsertPartialExecutor::MySQL(exec.set(f)),
#[cfg(feature = "mssql")]
InsertPartialExecutor::MSSQL(exec) => InsertPartialExecutor::MSSQL(exec.set(f)),
#[cfg(feature = "duckdb")]
InsertPartialExecutor::DuckDB(exec) => InsertPartialExecutor::DuckDB(exec.set(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ InsertPartialExecutor::Unsupported { .. } => unsupported,
}
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn default<F, C>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> C,
C: IntoInsertDefaultColumn<T>,
{
match self {
#[cfg(feature = "sqlite")]
InsertPartialExecutor::Sqlite(exec, phantom) => {
InsertPartialExecutor::Sqlite(exec.default(f), phantom)
}
#[cfg(feature = "postgresql")]
InsertPartialExecutor::PostgreSQL(exec) => {
InsertPartialExecutor::PostgreSQL(exec.default(f))
}
#[cfg(feature = "mysql")]
InsertPartialExecutor::MySQL(exec) => InsertPartialExecutor::MySQL(exec.default(f)),
#[cfg(feature = "mssql")]
InsertPartialExecutor::MSSQL(exec) => InsertPartialExecutor::MSSQL(exec.default(f)),
#[cfg(feature = "duckdb")]
InsertPartialExecutor::DuckDB(exec) => InsertPartialExecutor::DuckDB(exec.default(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ InsertPartialExecutor::Unsupported { .. } => unsupported,
}
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "sqlite")]
InsertPartialExecutor::Sqlite(exec, _) => exec.to_sql(),
#[cfg(feature = "postgresql")]
InsertPartialExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
InsertPartialExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
InsertPartialExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
InsertPartialExecutor::DuckDB(exec) => exec.to_sql(),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
InsertPartialExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(*backend, feature)),
}
}
pub async fn execute(self) -> crate::Result<<T as Model>::AutoIncrementKeyType> {
match self {
#[cfg(feature = "sqlite")]
InsertPartialExecutor::Sqlite(exec, _) => exec.execute().await,
#[cfg(feature = "postgresql")]
InsertPartialExecutor::PostgreSQL(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
InsertPartialExecutor::MySQL(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
InsertPartialExecutor::MSSQL(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
InsertPartialExecutor::DuckDB(exec) => exec.execute().await,
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
InsertPartialExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
}
}
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
fn insert_conflict_or_default(
conflict: &mut Option<crate::query::insert::InsertConflict>,
) -> &mut crate::query::insert::InsertConflict {
conflict.get_or_insert_with(crate::query::insert::InsertConflict::default)
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
fn insert_conflict_is_configured(
conflict: Option<&crate::query::insert::InsertConflict>,
) -> bool {
conflict.is_some_and(|conflict| conflict.is_configured())
}
impl<'a, I: crate::model::Insertable + Send + Sync> InsertExecutor<'a, I> {
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
fn with_insert_conflict<F>(self, apply: F) -> Self
where
F: FnOnce(&mut crate::query::insert::InsertConflict),
{
match self {
#[cfg(feature = "clickhouse")]
InsertExecutor::ClickHouse(db, models, mut conflict, marker) => {
apply(insert_conflict_or_default(&mut conflict));
InsertExecutor::ClickHouse(db, models, conflict, marker)
}
#[cfg(feature = "influxdb")]
InsertExecutor::InfluxDB(db, models, mut conflict, marker) => {
apply(insert_conflict_or_default(&mut conflict));
InsertExecutor::InfluxDB(db, models, conflict, marker)
}
other => other,
}
}
pub fn on_conflict<F, C>(self, f: F) -> Self
where
F: FnOnce(<I::Model as Model>::Where) -> C,
C: crate::query::insert::ConflictColumns,
{
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => InsertExecutor::Sqlite(exec.on_conflict(f)),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => InsertExecutor::SqliteTxn(exec.on_conflict(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => InsertExecutor::PostgreSQL(exec.on_conflict(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => {
InsertExecutor::PostgreSQLTxn(exec.on_conflict(f))
}
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => InsertExecutor::MySQL(exec.on_conflict(f)),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => InsertExecutor::MySQLTxn(exec.on_conflict(f)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => InsertExecutor::MSSQL(exec.on_conflict(f)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => InsertExecutor::MSSQLTxn(exec.on_conflict(f)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => InsertExecutor::DuckDB(exec.on_conflict(f)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => InsertExecutor::DuckDBTxn(exec.on_conflict(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
exec => exec.with_insert_conflict(|conflict| {
conflict.target = Some(crate::query::insert::InsertConflictTarget::Columns(
f(<I::Model as Model>::Where::default()).conflict_columns(),
));
}),
}
}
pub fn on_constraint<Target>(self, target: Target) -> Self
where
Target: crate::query::insert::IntoInsertConflictTarget<I::Model>,
{
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => InsertExecutor::Sqlite(exec.on_constraint(target)),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => {
InsertExecutor::SqliteTxn(exec.on_constraint(target))
}
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => {
InsertExecutor::PostgreSQL(exec.on_constraint(target))
}
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => {
InsertExecutor::PostgreSQLTxn(exec.on_constraint(target))
}
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => InsertExecutor::MySQL(exec.on_constraint(target)),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => {
InsertExecutor::MySQLTxn(exec.on_constraint(target))
}
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => InsertExecutor::MSSQL(exec.on_constraint(target)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => {
InsertExecutor::MSSQLTxn(exec.on_constraint(target))
}
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => InsertExecutor::DuckDB(exec.on_constraint(target)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => {
InsertExecutor::DuckDBTxn(exec.on_constraint(target))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
exec => exec.with_insert_conflict(|conflict| {
conflict.target = Some(target.into_insert_conflict_target());
}),
}
}
pub fn conflict_where<F, W>(self, f: F) -> Self
where
F: FnOnce(<I::Model as Model>::Where) -> W,
W: Into<WhereExpr>,
{
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => InsertExecutor::Sqlite(exec.conflict_where(f)),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => InsertExecutor::SqliteTxn(exec.conflict_where(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => InsertExecutor::PostgreSQL(exec.conflict_where(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => {
InsertExecutor::PostgreSQLTxn(exec.conflict_where(f))
}
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => InsertExecutor::MySQL(exec.conflict_where(f)),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => {
InsertExecutor::MySQLTxn(exec.conflict_where(f))
}
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => InsertExecutor::MSSQL(exec.conflict_where(f)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => {
InsertExecutor::MSSQLTxn(exec.conflict_where(f))
}
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => InsertExecutor::DuckDB(exec.conflict_where(f)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => {
InsertExecutor::DuckDBTxn(exec.conflict_where(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
exec => exec.with_insert_conflict(|conflict| {
conflict.target_filter = Some(crate::query::insert::where_expr_to_filter(f(
<I::Model as Model>::Where::default(),
)));
}),
}
}
pub fn do_nothing(self) -> Self {
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => InsertExecutor::Sqlite(exec.do_nothing()),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => InsertExecutor::SqliteTxn(exec.do_nothing()),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => InsertExecutor::PostgreSQL(exec.do_nothing()),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => {
InsertExecutor::PostgreSQLTxn(exec.do_nothing())
}
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => InsertExecutor::MySQL(exec.do_nothing()),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => InsertExecutor::MySQLTxn(exec.do_nothing()),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => InsertExecutor::MSSQL(exec.do_nothing()),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => InsertExecutor::MSSQLTxn(exec.do_nothing()),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => InsertExecutor::DuckDB(exec.do_nothing()),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => InsertExecutor::DuckDBTxn(exec.do_nothing()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
exec => exec.with_insert_conflict(|conflict| {
conflict.action = Some(crate::query::insert::InsertConflictAction::DoNothing);
}),
}
}
pub fn do_update(self) -> Self {
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => InsertExecutor::Sqlite(exec.do_update()),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => InsertExecutor::SqliteTxn(exec.do_update()),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => InsertExecutor::PostgreSQL(exec.do_update()),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => {
InsertExecutor::PostgreSQLTxn(exec.do_update())
}
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => InsertExecutor::MySQL(exec.do_update()),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => InsertExecutor::MySQLTxn(exec.do_update()),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => InsertExecutor::MSSQL(exec.do_update()),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => InsertExecutor::MSSQLTxn(exec.do_update()),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => InsertExecutor::DuckDB(exec.do_update()),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => InsertExecutor::DuckDBTxn(exec.do_update()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
exec => exec.with_insert_conflict(|conflict| {
conflict.action = Some(crate::query::insert::InsertConflictAction::DoUpdate);
}),
}
}
pub fn do_update_if<F, W>(self, f: F) -> Self
where
F: FnOnce(<I::Model as Model>::Where) -> W,
W: Into<WhereExpr>,
{
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => InsertExecutor::Sqlite(exec.do_update_if(f)),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => InsertExecutor::SqliteTxn(exec.do_update_if(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => InsertExecutor::PostgreSQL(exec.do_update_if(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => {
InsertExecutor::PostgreSQLTxn(exec.do_update_if(f))
}
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => InsertExecutor::MySQL(exec.do_update_if(f)),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => InsertExecutor::MySQLTxn(exec.do_update_if(f)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => InsertExecutor::MSSQL(exec.do_update_if(f)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => InsertExecutor::MSSQLTxn(exec.do_update_if(f)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => InsertExecutor::DuckDB(exec.do_update_if(f)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => InsertExecutor::DuckDBTxn(exec.do_update_if(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
exec => exec.with_insert_conflict(|conflict| {
let update_filter = crate::query::insert::where_expr_to_filter(f(
<I::Model as Model>::Where::default(),
));
conflict.action = Some(crate::query::insert::InsertConflictAction::DoUpdate);
conflict.update_filter = Some(update_filter);
}),
}
}
pub fn set<F>(self, f: F) -> Self
where
F: FnOnce(&mut <I::Model as Model>::Update),
{
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => InsertExecutor::Sqlite(exec.set(f)),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => InsertExecutor::SqliteTxn(exec.set(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => InsertExecutor::PostgreSQL(exec.set(f)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => {
InsertExecutor::PostgreSQLTxn(exec.set(f))
}
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => InsertExecutor::MySQL(exec.set(f)),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => InsertExecutor::MySQLTxn(exec.set(f)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => InsertExecutor::MSSQL(exec.set(f)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => InsertExecutor::MSSQLTxn(exec.set(f)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => InsertExecutor::DuckDB(exec.set(f)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => InsertExecutor::DuckDBTxn(exec.set(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
exec => exec.with_insert_conflict(|conflict| {
let mut update = <I::Model as Model>::Update::default();
f(&mut update);
conflict
.action
.get_or_insert(crate::query::insert::InsertConflictAction::DoUpdate);
conflict.assignments.extend(
<<I::Model as Model>::Update as crate::query::update::UpdateFields>::assignments(
&update,
),
);
}),
}
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => exec.to_sql(),
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => exec.to_sql(),
#[cfg(feature = "clickhouse")]
InsertExecutor::ClickHouse(_, models, conflict, _) => {
let refs = models.as_refs();
let statements = common_helpers::build_insert_statements_with_conflict::<I::Model>(
super::super::DbType::ClickHouse,
&refs,
conflict.as_ref(),
)?;
Ok(SqlStatement::batch(
super::super::DbType::ClickHouse,
statements
.into_iter()
.map(|statement| SingleSqlStatement::new(statement.sql, statement.params))
.collect(),
))
}
#[cfg(feature = "influxdb")]
InsertExecutor::Unsupported {
backend,
feature,
..
} => Err(unsupported_feature(*backend, feature)),
#[cfg(feature = "influxdb")]
InsertExecutor::InfluxDB(..) => Err(unsupported_feature(
super::super::DbType::InfluxDB,
"insert to_sql (InfluxDB writes use Line Protocol over HTTP)",
)),
}
}
pub async fn execute(
self,
) -> crate::Result<<I::Model as crate::model::Model>::AutoIncrementKeyType> {
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => exec.execute().await,
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(exec) => exec.execute().await,
#[cfg(feature = "clickhouse")]
InsertExecutor::ClickHouse(db, mut models, conflict, _) => {
if insert_conflict_is_configured(conflict.as_ref()) {
return Err(unsupported_feature(
super::super::DbType::ClickHouse,
"insert conflict handling",
));
}
if models.as_refs().is_empty() {
return Ok(<I::Model as Model>::AutoIncrementKeyType::default());
}
let ctx = crate::HookContext::new(crate::HookOperation::Insert);
models.run_before_insert(ctx).await?;
{
let refs = models.as_refs();
db.insert_model_rows::<I::Model>(&refs).await?;
}
models.run_after_insert(ctx).await?;
Ok(<I::Model as Model>::AutoIncrementKeyType::default())
}
#[cfg(feature = "influxdb")]
InsertExecutor::Unsupported {
backend,
feature,
models,
..
} => {
drop(models);
Err(unsupported_feature(backend, feature))
}
#[cfg(feature = "influxdb")]
InsertExecutor::InfluxDB(db, mut models, conflict, _) => {
if insert_conflict_is_configured(conflict.as_ref()) {
return Err(unsupported_feature(
super::super::DbType::InfluxDB,
"insert conflict handling",
));
}
if models.as_refs().is_empty() {
return Ok(<I::Model as Model>::AutoIncrementKeyType::default());
}
crate::abstract_layer::influxdb_backend::validate_influx_model::<I::Model>(
super::super::DbType::InfluxDB,
)?;
let ctx = crate::HookContext::new(crate::HookOperation::Insert);
models.run_before_insert(ctx).await?;
{
let refs = models.as_refs();
let lines = crate::abstract_layer::influxdb_backend::render_line_protocol::<
I::Model,
>(&refs)?;
db.write_lines_with_policy(
&lines,
crate::abstract_layer::influxdb_backend::model_retention_policy_name::<
I::Model,
>()
.as_deref(),
)
.await?;
}
models.run_after_insert(ctx).await?;
Ok(<I::Model as Model>::AutoIncrementKeyType::default())
}
}
}
pub fn without_hooks(self) -> crate::WithoutHooksExecutor<Self> {
crate::WithoutHooksExecutor(self)
}
pub async fn returning(self) -> crate::Result<Vec<I::Model>> {
let backend = match &self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(_) => super::super::DbType::Sqlite,
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(_) => super::super::DbType::Sqlite,
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => exec.db_type(),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(_) => super::super::DbType::PostgreSQL,
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(_) => super::super::DbType::MySQL,
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(_) => super::super::DbType::MySQL,
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(_) => super::super::DbType::MSSQL,
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(_) => super::super::DbType::MSSQL,
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(_) => super::super::DbType::DuckDB,
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(_) => super::super::DbType::DuckDB,
#[cfg(feature = "clickhouse")]
InsertExecutor::ClickHouse(..) => super::super::DbType::ClickHouse,
#[cfg(feature = "influxdb")]
InsertExecutor::Unsupported { backend, .. } => *backend,
#[cfg(feature = "influxdb")]
InsertExecutor::InfluxDB(..) => super::super::DbType::InfluxDB,
};
Capabilities::ensure(backend, |caps| caps.dml_returning, "DML RETURNING")?;
match self {
#[cfg(feature = "sqlite")]
InsertExecutor::Sqlite(exec) => exec.returning().await,
#[cfg(feature = "sqlite")]
InsertExecutor::SqliteTxn(_) => Err(unsupported_feature(
super::super::DbType::Sqlite,
"DML RETURNING inside transactions",
)),
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQL(exec) => exec.returning().await,
#[cfg(feature = "postgresql")]
InsertExecutor::PostgreSQLTxn(_) => Err(unsupported_feature(
super::super::DbType::PostgreSQL,
"DML RETURNING inside transactions",
)),
#[cfg(feature = "mysql")]
InsertExecutor::MySQL(exec) => exec.returning().await,
#[cfg(feature = "mysql")]
InsertExecutor::MySQLTxn(_) => Err(unsupported_feature(
super::super::DbType::MySQL,
"DML RETURNING inside transactions",
)),
#[cfg(feature = "mssql")]
InsertExecutor::MSSQL(exec) => exec.returning().await,
#[cfg(feature = "mssql")]
InsertExecutor::MSSQLTxn(_) => Err(unsupported_feature(
super::super::DbType::MSSQL,
"DML RETURNING inside transactions",
)),
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDB(exec) => exec.returning().await,
#[cfg(feature = "duckdb")]
InsertExecutor::DuckDBTxn(_) => Err(unsupported_feature(
super::super::DbType::DuckDB,
"DML RETURNING inside transactions",
)),
#[cfg(feature = "clickhouse")]
InsertExecutor::ClickHouse(..) => Err(unsupported_feature(
super::super::DbType::ClickHouse,
"DML RETURNING",
)),
#[cfg(feature = "influxdb")]
InsertExecutor::Unsupported {
backend,
feature,
..
} => Err(unsupported_feature(backend, feature)),
#[cfg(feature = "influxdb")]
InsertExecutor::InfluxDB(..) => Err(unsupported_feature(
super::super::DbType::InfluxDB,
"DML RETURNING",
)),
}
}
}
impl<'a, I> std::future::IntoFuture for InsertExecutor<'a, I>
where
I: crate::model::Insertable + Send + Sync,
<I::Model as crate::model::Model>::AutoIncrementKeyType: Send,
<I as crate::model::Insertable>::Model: Send + Sync,
Self: 'a,
{
type Output = crate::Result<<I::Model as crate::model::Model>::AutoIncrementKeyType>;
type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(self.execute())
}
}
pub enum InsertOrUpdateExecutor<'a, I: crate::model::Insertable> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::InsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "sqlite")]
SqliteTxn(sqlite_backend::TransactionInsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::InsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "postgresql")]
PostgreSQLTxn(postgresql_backend::TransactionInsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::InsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "mysql")]
MySQLTxn(mysql_backend::TransactionInsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::InsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "mssql")]
MSSQLTxn(mssql_backend::TransactionInsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::InsertOrUpdateExecutor<'a, I>),
#[cfg(feature = "duckdb")]
DuckDBTxn(duckdb_backend::TransactionInsertOrUpdateExecutor<'a, I>),
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a I>,
},
}
#[deprecated(
since = "0.2.12",
note = "TransactionInsertOrUpdateExecutor 已合并为 InsertOrUpdateExecutor,请改用 InsertOrUpdateExecutor"
)]
pub type TransactionInsertOrUpdateExecutor<'a, I> = InsertOrUpdateExecutor<'a, I>;
pub struct InsertGraphExecutor<'a, T: crate::model::GraphWritable> {
db: &'a Database,
model: &'a mut T,
}
pub struct UpdateGraphExecutor<'a, T: crate::model::GraphWritable> {
db: &'a Database,
model: &'a mut T,
}
impl<'a, I: crate::model::Insertable + Send + Sync> InsertOrUpdateExecutor<'a, I> {
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "sqlite")]
InsertOrUpdateExecutor::Sqlite(exec) => exec.to_sql(),
#[cfg(feature = "sqlite")]
InsertOrUpdateExecutor::SqliteTxn(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
InsertOrUpdateExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
InsertOrUpdateExecutor::PostgreSQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
InsertOrUpdateExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
InsertOrUpdateExecutor::MySQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
InsertOrUpdateExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
InsertOrUpdateExecutor::MSSQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
InsertOrUpdateExecutor::DuckDB(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
InsertOrUpdateExecutor::DuckDBTxn(exec) => exec.to_sql(),
InsertOrUpdateExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(*backend, feature)),
}
}
pub async fn execute(self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
InsertOrUpdateExecutor::Sqlite(exec) => exec.execute().await,
#[cfg(feature = "sqlite")]
InsertOrUpdateExecutor::SqliteTxn(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
InsertOrUpdateExecutor::PostgreSQL(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
InsertOrUpdateExecutor::PostgreSQLTxn(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
InsertOrUpdateExecutor::MySQL(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
InsertOrUpdateExecutor::MySQLTxn(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
InsertOrUpdateExecutor::MSSQL(exec) => exec.execute().await.map(|_| ()),
#[cfg(feature = "mssql")]
InsertOrUpdateExecutor::MSSQLTxn(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
InsertOrUpdateExecutor::DuckDB(exec) => exec.execute().await.map(|_| ()),
#[cfg(feature = "duckdb")]
InsertOrUpdateExecutor::DuckDBTxn(exec) => exec.execute().await,
InsertOrUpdateExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
}
}
pub fn without_hooks(self) -> crate::WithoutHooksExecutor<Self> {
crate::WithoutHooksExecutor(self)
}
}
pub(crate) fn log_rollback_failure<T>(result: crate::Result<T>) {
if let Err(err) = result {
eprintln!("[ormer] rollback failed during error handling: {err}");
}
}
pub(crate) async fn run_txn_closure<R, F>(txn: Transaction<'_>, f: F) -> crate::Result<R>
where
F: for<'tx> FnOnce(&'tx mut Transaction<'_>) -> TransactionFuture<'tx, R>,
{
let mut txn = txn;
match f(&mut txn).await {
Ok(value) => {
txn.commit().await?;
Ok(value)
}
Err(err) => {
log_rollback_failure(txn.rollback().await);
Err(err)
}
}
}
pub(crate) async fn apply_transaction_options_or_rollback(
txn: Transaction<'_>,
options: TransactionOptions,
) -> crate::Result<Transaction<'_>> {
let mut txn = txn;
if let Err(err) = apply_transaction_options(&mut txn, options).await {
log_rollback_failure(txn.rollback().await);
return Err(err);
}
Ok(txn)
}
impl<'a, T> InsertGraphExecutor<'a, T>
where
T: crate::model::GraphWritable + Send + Sync + 'a,
<T as Model>::AutoIncrementKeyType: Into<crate::model::Value>,
{
pub async fn execute(self) -> crate::Result<()> {
let mut tx = self.db.begin().await?;
if let Err(err) = async {
let key = tx.insert(&*self.model).execute().await?;
let key_value = crate::model::graph_auto_increment_key_value(key);
if !crate::model::graph_is_no_auto_increment_key(&key_value) {
self.model
.assign_column_value(<T as Model>::primary_key_columns()[0], key_value)?;
}
<T as crate::model::GraphWritable>::insert_graph_relations(&mut tx, self.model).await
}
.await
{
log_rollback_failure(tx.rollback().await);
return Err(err);
}
tx.commit().await
}
}
impl<'a, T> UpdateGraphExecutor<'a, T>
where
T: crate::model::GraphWritable + Send + Sync + 'a,
{
pub async fn execute(self) -> crate::Result<u64> {
let mut tx = self.db.begin().await?;
let affected = match async {
#[cfg(feature = "duckdb")]
let affected = if tx.db_type() == super::super::DbType::DuckDB {
match common_helpers::model_update_plan(&*self.model, None) {
Some(plan) => {
let statement =
common_helpers::build_duckdb_graph_update_sql::<T>(&*self.model, &plan)?;
match &mut tx {
Transaction::DuckDB(txn) => {
let executor = txn.update::<T>();
<duckdb_backend::UpdateExecutor<T> as super::SqlExecutor>::execute_with_sql(
executor,
SqlStatement::single(
super::super::DbType::DuckDB,
statement.sql,
statement.params,
),
)
.await?
}
_ => {
return Err(crate::OrmerError::invalid_operation(
"DuckDB graph update dispatched to a non-DuckDB transaction",
));
}
}
}
None => 0,
}
} else {
tx.update::<T>().set_model(&*self.model).execute().await?
};
#[cfg(not(feature = "duckdb"))]
let affected = tx.update::<T>().set_model(&*self.model).execute().await?;
<T as crate::model::GraphWritable>::update_graph_relations(&mut tx, self.model).await?;
Ok::<u64, crate::OrmerError>(affected)
}
.await
{
Ok(affected) => affected,
Err(err) => {
log_rollback_failure(tx.rollback().await);
return Err(err);
}
};
tx.commit().await?;
Ok(affected)
}
}
pub struct SaveExecutor<'a, T: WritableModel + crate::model::GraphWritable> {
conn: SaveConn<'a, T>,
model: &'a mut Tracked<T>,
}
enum SaveConn<'a, T: WritableModel + crate::model::GraphWritable> {
Db(&'a Database),
Pooled {
sql: crate::Result<SqlStatement>,
run: SaveTxnRun<'a, T>,
},
Txn {
sql: crate::Result<SqlStatement>,
run: SaveTxnRun<'a, T>,
},
}
type SaveTxnFuture<'a, T> =
Pin<Box<dyn Future<Output = (crate::Result<u64>, &'a mut Tracked<T>)> + 'a>>;
type SaveTxnRun<'a, T> = Box<dyn FnOnce(&'a mut Tracked<T>) -> SaveTxnFuture<'a, T> + Send + 'a>;
fn finish_save<T: crate::model::Model>(affected: u64, model: &mut Tracked<T>) -> crate::Result<u64> {
if affected > 0 {
model.accept_changes();
}
Ok(affected)
}
impl<'a, T: WritableModel + crate::model::Model + crate::model::GraphWritable>
SaveExecutor<'a, T>
{
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match &self.conn {
SaveConn::Db(db) => {
let fields = self.model.dirty_columns();
if fields.is_empty() {
return Ok(SqlStatement::batch(db.db_type(), Vec::new()));
}
db.update::<T>()
.set_model_columns(self.model.as_model(), &fields)
.to_sql()
}
SaveConn::Txn { sql, .. } | SaveConn::Pooled { sql, .. } => sql.clone(),
}
}
pub async fn execute(self) -> crate::Result<u64> {
let SaveExecutor { conn, mut model } = self;
let affected = match conn {
SaveConn::Db(db) => {
let mut tx = db.begin().await?;
match save_dirty_columns_and_relations(&mut tx, model).await {
Ok(affected) => {
tx.commit().await?;
affected
}
Err(err) => {
log_rollback_failure(tx.rollback().await);
return Err(err);
}
}
}
SaveConn::Txn { run, .. } | SaveConn::Pooled { run, .. } => {
let (result, m) = run(model).await;
model = m;
result?
}
};
finish_save(affected, model)
}
#[deprecated(since = "0.2.11", note = "use `execute()` instead")]
pub async fn exec(self) -> crate::Result<u64> {
self.execute().await
}
pub async fn execute_with_hooks(self) -> crate::Result<u64>
where
T: crate::BeforeUpdate + crate::AfterUpdate + Send + Sync,
{
let SaveExecutor { conn, mut model } = self;
let mut ctx = crate::HookContext::new(crate::HookOperation::Update);
if matches!(conn, SaveConn::Txn { .. }) {
ctx = ctx.transaction();
}
if ctx.hooks_enabled() {
crate::BeforeUpdate::before_update(model.as_model_mut(), &mut ctx).await?;
}
let affected = match conn {
SaveConn::Db(db) => {
let mut tx = db.begin().await?;
let result = async {
let affected =
save_dirty_columns_and_relations(&mut tx, model).await?;
if affected > 0 && ctx.hooks_enabled() {
crate::AfterUpdate::after_update(model.as_model(), &mut ctx).await?;
}
Ok::<u64, crate::OrmerError>(affected)
}
.await;
match result {
Ok(affected) => {
tx.commit().await?;
affected
}
Err(err) => {
log_rollback_failure(tx.rollback().await);
return Err(err);
}
}
}
SaveConn::Txn { run, .. } | SaveConn::Pooled { run, .. } => {
let (result, m) = run(model).await;
model = m;
let affected = result?;
if affected > 0 && ctx.hooks_enabled() {
crate::AfterUpdate::after_update(model.as_model(), &mut ctx).await?;
}
affected
}
};
finish_save(affected, model)
}
pub fn without_hooks(self) -> crate::WithoutHooksExecutor<Self> {
crate::WithoutHooksExecutor(self)
}
}
pub enum InsertOrIgnoreExecutor<'a, I: crate::model::Insertable> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::InsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "sqlite")]
SqliteTxn(sqlite_backend::TransactionInsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::InsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "postgresql")]
PostgreSQLTxn(postgresql_backend::TransactionInsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::InsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "mysql")]
MySQLTxn(mysql_backend::TransactionInsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::InsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "mssql")]
MSSQLTxn(mssql_backend::TransactionInsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::InsertOrIgnoreExecutor<'a, I>),
#[cfg(feature = "duckdb")]
DuckDBTxn(duckdb_backend::TransactionInsertOrIgnoreExecutor<'a, I>),
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a I>,
},
}
#[deprecated(
since = "0.2.12",
note = "TransactionInsertOrIgnoreExecutor 已合并为 InsertOrIgnoreExecutor,请改用 InsertOrIgnoreExecutor"
)]
pub type TransactionInsertOrIgnoreExecutor<'a, I> = InsertOrIgnoreExecutor<'a, I>;
impl<'a, I: crate::model::Insertable + Send + Sync> InsertOrIgnoreExecutor<'a, I> {
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "sqlite")]
InsertOrIgnoreExecutor::Sqlite(exec) => exec.to_sql(),
#[cfg(feature = "sqlite")]
InsertOrIgnoreExecutor::SqliteTxn(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
InsertOrIgnoreExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
InsertOrIgnoreExecutor::PostgreSQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
InsertOrIgnoreExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
InsertOrIgnoreExecutor::MySQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
InsertOrIgnoreExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
InsertOrIgnoreExecutor::MSSQLTxn(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
InsertOrIgnoreExecutor::DuckDB(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
InsertOrIgnoreExecutor::DuckDBTxn(exec) => exec.to_sql(),
InsertOrIgnoreExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(*backend, feature)),
}
}
pub async fn execute(self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
InsertOrIgnoreExecutor::Sqlite(exec) => exec.execute().await,
#[cfg(feature = "sqlite")]
InsertOrIgnoreExecutor::SqliteTxn(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
InsertOrIgnoreExecutor::PostgreSQL(exec) => exec.execute().await,
#[cfg(feature = "postgresql")]
InsertOrIgnoreExecutor::PostgreSQLTxn(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
InsertOrIgnoreExecutor::MySQL(exec) => exec.execute().await,
#[cfg(feature = "mysql")]
InsertOrIgnoreExecutor::MySQLTxn(exec) => exec.execute().await,
#[cfg(feature = "mssql")]
InsertOrIgnoreExecutor::MSSQL(exec) => exec.execute().await.map(|_| ()),
#[cfg(feature = "mssql")]
InsertOrIgnoreExecutor::MSSQLTxn(exec) => exec.execute().await,
#[cfg(feature = "duckdb")]
InsertOrIgnoreExecutor::DuckDB(exec) => exec.execute().await.map(|_| ()),
#[cfg(feature = "duckdb")]
InsertOrIgnoreExecutor::DuckDBTxn(exec) => exec.execute().await,
InsertOrIgnoreExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
}
}
pub fn without_hooks(self) -> crate::WithoutHooksExecutor<Self> {
crate::WithoutHooksExecutor(self)
}
}
impl Database {
pub fn replicated(db_type: super::super::DbType) -> ReplicatedDatabaseBuilder {
ReplicatedDatabaseBuilder::new(db_type)
}
pub fn sql_trace(&self) -> crate::SqlTraceBuilder {
crate::global_sql_trace().builder()
}
pub async fn connect(
db_type: super::super::DbType,
connection_string: &str,
) -> crate::Result<Self> {
match db_type {
#[cfg(feature = "sqlite")]
super::super::DbType::Sqlite => {
let db = sqlite_backend::Database::connect(db_type, connection_string).await?;
Ok(Database::Sqlite(db))
}
#[cfg(feature = "postgresql")]
super::super::DbType::PostgreSQL => {
let db = postgresql_backend::Database::connect(db_type, connection_string).await?;
Ok(Database::PostgreSQL(db))
}
#[cfg(feature = "questdb")]
super::super::DbType::QuestDB => {
let db = postgresql_backend::Database::connect(db_type, connection_string).await?;
Ok(Database::PostgreSQL(db))
}
#[cfg(feature = "mysql")]
super::super::DbType::MySQL => {
let db = mysql_backend::Database::connect(db_type, connection_string).await?;
Ok(Database::MySQL(db))
}
#[cfg(feature = "mssql")]
super::super::DbType::MSSQL => {
let db = mssql_backend::Database::connect(db_type, connection_string).await?;
Ok(Database::MSSQL(db))
}
#[cfg(feature = "duckdb")]
super::super::DbType::DuckDB => {
let db = duckdb_backend::Database::connect(db_type, connection_string).await?;
Ok(Database::DuckDB(db))
}
#[cfg(feature = "clickhouse")]
super::super::DbType::ClickHouse => {
let db = super::super::clickhouse_backend::Database::connect(connection_string)?;
Ok(Database::ClickHouse(db))
}
#[cfg(feature = "influxdb")]
super::super::DbType::InfluxDB => {
let db = super::super::influxdb_backend::Database::connect(connection_string)?;
Ok(Database::InfluxDB(db))
}
}
}
pub fn create_table<T: WritableModel>(&self) -> CreateTableExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => CreateTableExecutor::Sqlite(db.create_table::<T>()),
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => CreateTableExecutor::PostgreSQL(db.create_table::<T>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => CreateTableExecutor::MySQL(db.create_table::<T>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => CreateTableExecutor::MSSQL(db.create_table::<T>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => CreateTableExecutor::DuckDB(db.create_table::<T>()),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(_) => CreateTableExecutor::Unsupported {
backend: super::super::DbType::ClickHouse,
feature: "CREATE TABLE without explicit ClickHouse engine settings",
_marker: std::marker::PhantomData,
},
#[cfg(feature = "influxdb")]
Database::InfluxDB(db) => {
CreateTableExecutor::InfluxDB(db, std::marker::PhantomData)
}
}
}
pub async fn validate_table<T: WritableModel>(&self) -> crate::Result<()> {
if !Capabilities::of(self.db_type()).schema_introspection {
return Err(unsupported_feature(self.db_type(), "validate_table"));
}
match self {
#[cfg(feature = "questdb")]
Database::PostgreSQL(db) if db.db_type().is_questdb() => {
let _ = db;
self.validate_table_questdb::<T>().await
}
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => db.validate_table::<T>().await,
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => db.validate_table::<T>().await,
#[cfg(feature = "mysql")]
Database::MySQL(db) => db.validate_table::<T>().await,
#[cfg(feature = "mssql")]
Database::MSSQL(db) => db.validate_table::<T>().await,
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => db.validate_table::<T>().await,
#[allow(unreachable_patterns)]
_ => Err(unsupported_feature(self.db_type(), "validate_table")),
}
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb",
feature = "clickhouse"
)),
allow(unused_variables)
)]
pub async fn generate_entities(&self, schema: Option<&str>) -> crate::Result<String> {
#[cfg(feature = "influxdb")]
#[allow(irrefutable_let_patterns)]
if let Database::InfluxDB(_) = self {
return Err(unsupported_feature(
super::super::DbType::InfluxDB,
"generate_entities",
));
}
let tables = match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => db.db_first_tables(schema).await?,
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => db.db_first_tables(schema).await?,
#[cfg(feature = "mysql")]
Database::MySQL(db) => db.db_first_tables(schema).await?,
#[cfg(feature = "mssql")]
Database::MSSQL(db) => db.db_first_tables(schema).await?,
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => db.db_first_tables(schema).await?,
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => db.db_first_tables(schema).await?,
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => Vec::new(),
};
db_first::generate_entities(self.db_type(), &tables)
}
pub fn insert<I: crate::model::Insertable>(&self, models: I) -> InsertExecutor<'_, I> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => InsertExecutor::Sqlite(db.insert::<I>(models)),
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => InsertExecutor::PostgreSQL(db.insert::<I>(models)),
#[cfg(feature = "mysql")]
Database::MySQL(db) => InsertExecutor::MySQL(db.insert::<I>(models)),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => InsertExecutor::MSSQL(db.insert::<I>(models)),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => InsertExecutor::DuckDB(db.insert::<I>(models)),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => {
InsertExecutor::ClickHouse(db, models, None, std::marker::PhantomData)
}
#[cfg(feature = "influxdb")]
Database::InfluxDB(db) => InsertExecutor::InfluxDB(
db,
models,
None,
std::marker::PhantomData,
),
}
}
pub fn insert_partial<T: WritableModel + Send + Sync>(&self) -> InsertPartialExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
InsertPartialExecutor::Sqlite(db.insert_partial::<T>(), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => InsertPartialExecutor::PostgreSQL(db.insert_partial::<T>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => InsertPartialExecutor::MySQL(db.insert_partial::<T>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => InsertPartialExecutor::MSSQL(db.insert_partial::<T>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => InsertPartialExecutor::DuckDB(db.insert_partial::<T>()),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(_) => InsertPartialExecutor::Unsupported {
backend: super::super::DbType::ClickHouse,
feature: "partial Model insert on ClickHouse",
_marker: std::marker::PhantomData,
},
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => InsertPartialExecutor::Unsupported {
backend: super::super::DbType::InfluxDB,
feature: "insert_partial",
_marker: std::marker::PhantomData,
},
}
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn insert_model<T>(
&self,
model: impl crate::model::InsertModel<T>,
) -> InsertPartialExecutor<'_, T>
where
T: WritableModel + Send + Sync,
{
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
InsertPartialExecutor::Sqlite(db.insert_model::<T>(model), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => {
InsertPartialExecutor::PostgreSQL(db.insert_model::<T>(model))
}
#[cfg(feature = "mysql")]
Database::MySQL(db) => InsertPartialExecutor::MySQL(db.insert_model::<T>(model)),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => InsertPartialExecutor::MSSQL(db.insert_model::<T>(model)),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => InsertPartialExecutor::DuckDB(db.insert_model::<T>(model)),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(_) => InsertPartialExecutor::Unsupported {
backend: super::super::DbType::ClickHouse,
feature: "partial Model insert on ClickHouse",
_marker: std::marker::PhantomData,
},
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => InsertPartialExecutor::Unsupported {
backend: super::super::DbType::InfluxDB,
feature: "insert_model",
_marker: std::marker::PhantomData,
},
}
}
pub fn insert_graph<'a, T>(&'a self, model: &'a mut T) -> InsertGraphExecutor<'a, T>
where
T: crate::model::GraphWritable,
{
InsertGraphExecutor { db: self, model }
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn insert_or_update<I: crate::model::Insertable>(
&self,
models: I,
) -> InsertOrUpdateExecutor<'_, I> {
if !Capabilities::of(self.db_type()).insert_conflict {
return InsertOrUpdateExecutor::Unsupported {
backend: self.db_type(),
feature: "insert conflict handling",
_marker: std::marker::PhantomData,
};
}
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
InsertOrUpdateExecutor::Sqlite(db.insert_or_update::<I>(models))
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => {
InsertOrUpdateExecutor::PostgreSQL(db.insert_or_update::<I>(models))
}
#[cfg(feature = "mysql")]
Database::MySQL(db) => InsertOrUpdateExecutor::MySQL(db.insert_or_update::<I>(models)),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => InsertOrUpdateExecutor::MSSQL(db.insert_or_update::<I>(models)),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => {
InsertOrUpdateExecutor::DuckDB(db.insert_or_update::<I>(models))
}
#[allow(unreachable_patterns)]
_ => InsertOrUpdateExecutor::Unsupported {
backend: self.db_type(),
feature: "insert conflict handling",
_marker: std::marker::PhantomData,
},
}
}
pub fn upsert<I: crate::model::Insertable>(&self, models: I) -> InsertOrUpdateExecutor<'_, I> {
self.insert_or_update(models)
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn insert_or_ignore<I: crate::model::Insertable>(
&self,
models: I,
) -> InsertOrIgnoreExecutor<'_, I> {
if !Capabilities::of(self.db_type()).insert_ignore {
return InsertOrIgnoreExecutor::Unsupported {
backend: self.db_type(),
feature: "insert ignore",
_marker: std::marker::PhantomData,
};
}
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
InsertOrIgnoreExecutor::Sqlite(db.insert_or_ignore::<I>(models))
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => {
InsertOrIgnoreExecutor::PostgreSQL(db.insert_or_ignore::<I>(models))
}
#[cfg(feature = "mysql")]
Database::MySQL(db) => InsertOrIgnoreExecutor::MySQL(db.insert_or_ignore::<I>(models)),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => InsertOrIgnoreExecutor::MSSQL(db.insert_or_ignore::<I>(models)),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => {
InsertOrIgnoreExecutor::DuckDB(db.insert_or_ignore::<I>(models))
}
#[allow(unreachable_patterns)]
_ => InsertOrIgnoreExecutor::Unsupported {
backend: self.db_type(),
feature: "insert ignore",
_marker: std::marker::PhantomData,
},
}
}
pub fn batch<'a, B>(&'a self, batch: B) -> BatchFuture<'a, B>
where
B: BatchQueries<'a>,
{
BatchFuture::new(batch)
}
pub fn batch_many<'a, I, Q>(&'a self, queries: I) -> BatchManyFuture<'a, Q>
where
I: IntoIterator<Item = Q>,
Q: BatchQuery<'a>,
{
BatchManyFuture::new(queries)
}
pub async fn find_by_id<T: Model + 'static + std::marker::Send + std::marker::Sync>(
&self,
key: impl crate::model::PrimaryKey,
) -> crate::Result<Option<T>> {
find_by_id_with_executor(self.select::<T>(), key).await
}
pub async fn find_related<T, S>(
&self,
owner: &T,
relation: S,
) -> crate::Result<Vec<S::Target>>
where
T: Model + 'static + std::marker::Send + std::marker::Sync,
S: RelationSelection<T>,
for<'b> S: RelationNestedLoader<'b, T> + std::marker::Send + std::marker::Sync,
S::Target: std::marker::Send + std::marker::Sync,
S::Via: std::marker::Send + std::marker::Sync,
{
find_related_with_executor(&self.select::<T>(), owner, &relation).await
}
pub async fn preload<T, S>(
&self,
owners: &mut [T],
relation: S,
) -> crate::Result<()>
where
T: Model + 'static + std::marker::Send + std::marker::Sync,
S: RelationSelection<T>,
for<'b> S: RelationNestedLoader<'b, T> + std::marker::Send + std::marker::Sync,
S::Target: std::marker::Send + std::marker::Sync,
S::Via: std::marker::Send + std::marker::Sync,
{
preload_with_executor(&self.select::<T>(), owners, relation).await
}
pub fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => SelectExecutor::Sqlite(db.select::<T>()),
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => SelectExecutor::PostgreSQL(db.select::<T>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => SelectExecutor::MySQL(db.select::<T>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => SelectExecutor::MSSQL(db.select::<T>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => SelectExecutor::DuckDB(db.select::<T>()),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => {
SelectExecutor::ClickHouse(
ClickHouseSelectBackend::ClickHouse(db),
crate::query::builder::Select::default(),
)
}
#[cfg(feature = "influxdb")]
Database::InfluxDB(db) => SelectExecutor::ClickHouse(
ClickHouseSelectBackend::Influx(db),
crate::query::builder::Select::default(),
),
}
}
pub fn scope(&self) -> DatabaseScope<'_> {
DatabaseScope {
db: self,
context_filters: Vec::new(),
}
}
pub fn from_derived<R: Model>(
&self,
derived: DerivedSelect<R>,
) -> DerivedTableSelectExecutor<'_, R> {
DerivedTableSelectExecutor {
db: self,
select: crate::query::builder::from_derived(derived),
}
}
pub fn select_union<T: Model>(
&self,
union: crate::query::builder::UnionSelect<T>,
) -> UnionSelectExecutor<'_, T> {
UnionSelectExecutor { db: self, select: union }
}
pub fn select_column<T: Model, V>(&self) -> ProjectionSelectExecutor<'_, T, V> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
ProjectionSelectExecutor::Sqlite(db.select_column::<T, V>())
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => {
ProjectionSelectExecutor::PostgreSQL(db.select_column::<T, V>())
}
#[cfg(feature = "mysql")]
Database::MySQL(db) => ProjectionSelectExecutor::MySQL(db.select_column::<T, V>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => ProjectionSelectExecutor::MSSQL(db.select_column::<T, V>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => ProjectionSelectExecutor::DuckDB(db.select_column::<T, V>()),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => match clickhouse_projection_gate(
super::super::DbType::ClickHouse,
) {
None => ProjectionSelectExecutor::ClickHouse(db, ProjectionSelect::new()),
Some(feature) => ProjectionSelectExecutor::Unsupported {
backend: super::super::DbType::ClickHouse,
feature,
_marker: std::marker::PhantomData,
},
},
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => ProjectionSelectExecutor::Unsupported {
backend: super::super::DbType::InfluxDB,
feature: clickhouse_projection_gate(super::super::DbType::InfluxDB)
.unwrap_or("GROUP BY aggregation"),
_marker: std::marker::PhantomData,
},
}
}
pub fn delete<T: WritableModel>(&self) -> DeleteExecutor<'_, T> {
if !Capabilities::of(self.db_type()).row_delete {
return DeleteExecutor::Unsupported {
backend: self.db_type(),
feature: "row delete",
_marker: std::marker::PhantomData,
};
}
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
DeleteExecutor::Sqlite(db.delete::<T>(), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => DeleteExecutor::PostgreSQL(db.delete::<T>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => DeleteExecutor::MySQL(db.delete::<T>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => DeleteExecutor::MSSQL(db.delete::<T>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => DeleteExecutor::DuckDB(db.delete::<T>()),
#[allow(unreachable_patterns)]
_ => DeleteExecutor::Unsupported {
backend: self.db_type(),
feature: "row delete",
_marker: std::marker::PhantomData,
},
}
}
pub fn delete_blocks<T: WritableModel>(&self) -> BlockDeleteExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => BlockDeleteExecutor::fallback(
super::super::DbType::Sqlite,
DeleteExecutor::Sqlite(db.delete::<T>(), std::marker::PhantomData),
),
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => BlockDeleteExecutor::PostgreSQL(db.delete_blocks::<T>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => BlockDeleteExecutor::fallback(
super::super::DbType::MySQL,
DeleteExecutor::MySQL(db.delete::<T>()),
),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => BlockDeleteExecutor::fallback(
super::super::DbType::MSSQL,
DeleteExecutor::MSSQL(db.delete::<T>()),
),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => BlockDeleteExecutor::fallback(
super::super::DbType::DuckDB,
DeleteExecutor::DuckDB(db.delete::<T>()),
),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => {
BlockDeleteExecutor::ClickHouse(clickhouse_backend::BlockDeleteExecutor::new(db))
}
#[cfg(feature = "influxdb")]
Database::InfluxDB(db) => {
BlockDeleteExecutor::InfluxDB(influxdb_backend::BlockDeleteExecutor::new(db))
}
}
}
pub fn update<T: WritableModel>(&self) -> UpdateExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
UpdateExecutor::Sqlite(db.update::<T>(), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => UpdateExecutor::PostgreSQL(db.update::<T>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => UpdateExecutor::MySQL(db.update::<T>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => UpdateExecutor::MSSQL(db.update::<T>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => UpdateExecutor::DuckDB(db.update::<T>()),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(_) => UpdateExecutor::Unsupported {
backend: super::super::DbType::ClickHouse,
feature: "row update on ClickHouse; use execute_sql",
_marker: std::marker::PhantomData,
},
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => UpdateExecutor::Unsupported {
backend: super::super::DbType::InfluxDB,
feature: "row update",
_marker: std::marker::PhantomData,
},
}
}
pub fn save<'a, T: WritableModel + crate::model::GraphWritable>(
&'a self,
model: &'a mut Tracked<T>,
) -> SaveExecutor<'a, T> {
SaveExecutor {
conn: SaveConn::Db(self),
model,
}
}
pub fn update_graph<'a, T>(&'a self, model: &'a mut T) -> UpdateGraphExecutor<'a, T>
where
T: crate::model::GraphWritable,
{
UpdateGraphExecutor { db: self, model }
}
pub fn from<T: Model + 'static, R: Model>(&self) -> RelatedSelectExecutor<'_, T, R> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
RelatedSelectExecutor::Sqlite(db.related::<T, R>(), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => RelatedSelectExecutor::PostgreSQL(db.related::<T, R>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => RelatedSelectExecutor::MySQL(db.related::<T, R>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => RelatedSelectExecutor::MSSQL(db.related::<T, R>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => RelatedSelectExecutor::DuckDB(db.related::<T, R>()),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(_) => RelatedSelectExecutor::Unsupported {
backend: super::super::DbType::ClickHouse,
feature: "related (multi-table) select",
_marker: std::marker::PhantomData,
},
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => RelatedSelectExecutor::Unsupported {
backend: super::super::DbType::InfluxDB,
feature: "relation select",
_marker: std::marker::PhantomData,
},
}
}
pub async fn begin(&self) -> crate::Result<Transaction<'_>> {
if !Capabilities::of(self.db_type()).transactions {
return Err(unsupported_feature(self.db_type(), "transactions"));
}
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
let txn = db.begin().await?;
Ok(Transaction::Sqlite(txn))
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => {
let txn = db.begin().await?;
Ok(Transaction::PostgreSQL(txn))
}
#[cfg(feature = "mysql")]
Database::MySQL(db) => {
let txn = db.begin().await?;
Ok(Transaction::MySQL(txn))
}
#[cfg(feature = "mssql")]
Database::MSSQL(db) => {
let txn = db.begin().await?;
Ok(Transaction::MSSQL(txn))
}
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => {
let txn = db.begin().await?;
Ok(Transaction::DuckDB(txn))
}
#[allow(unreachable_patterns)]
_ => Err(unsupported_feature(self.db_type(), "transactions")),
}
}
pub async fn transaction<R, F>(&self, f: F) -> crate::Result<R>
where
F: for<'tx> FnOnce(&'tx mut Transaction<'_>) -> TransactionFuture<'tx, R>,
{
self.transaction_opts(TransactionOptions::new(), f).await
}
pub async fn transaction_opts<R, F>(
&self,
options: TransactionOptions,
f: F,
) -> crate::Result<R>
where
F: for<'tx> FnOnce(&'tx mut Transaction<'_>) -> TransactionFuture<'tx, R>,
{
let txn = self.begin_opts(options).await?;
run_txn_closure(txn, f).await
}
async fn begin_opts(&self, options: TransactionOptions) -> crate::Result<Transaction<'_>> {
#[cfg(feature = "mysql")]
#[allow(irrefutable_let_patterns)]
if let Database::MySQL(db) = self {
let txn = db.begin_with_opts(options).await?;
return Ok(Transaction::MySQL(txn));
}
apply_transaction_options_or_rollback(self.begin().await?, options).await
}
pub fn drop_table<T: WritableModel>(&self) -> DropTableExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => DropTableExecutor::Sqlite(db.drop_table::<T>()),
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => DropTableExecutor::PostgreSQL(db.drop_table::<T>()),
#[cfg(feature = "mysql")]
Database::MySQL(db) => DropTableExecutor::MySQL(db.drop_table::<T>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => DropTableExecutor::MSSQL(db.drop_table::<T>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => DropTableExecutor::DuckDB(db.drop_table::<T>()),
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => DropTableExecutor::ClickHouse(db, std::marker::PhantomData),
#[cfg(feature = "influxdb")]
Database::InfluxDB(db) => {
DropTableExecutor::InfluxDB(db, std::marker::PhantomData)
}
}
}
pub fn truncate_table<T: WritableModel>(&self) -> TruncateTableExecutor<'_, T> {
if !Capabilities::of(self.db_type()).truncate {
return TruncateTableExecutor::Unsupported {
backend: self.db_type(),
feature: "truncate_table",
_marker: std::marker::PhantomData,
};
}
match self {
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => {
TruncateTableExecutor::PostgreSQL(db.truncate_table::<T>())
}
#[cfg(feature = "mysql")]
Database::MySQL(db) => TruncateTableExecutor::MySQL(db.truncate_table::<T>()),
#[cfg(feature = "mssql")]
Database::MSSQL(db) => TruncateTableExecutor::MSSQL(db.truncate_table::<T>()),
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => TruncateTableExecutor::DuckDB(db.truncate_table::<T>()),
#[allow(unreachable_patterns)]
_ => TruncateTableExecutor::Unsupported {
backend: self.db_type(),
feature: "truncate_table",
_marker: std::marker::PhantomData,
},
}
}
pub fn select_sql<T>(&self, sql: impl IntoRawSql) -> RawSelectExecutor<'_, T> {
RawSelectExecutor::from_conn(ConnRef::Db(self), sql.into_raw_sql())
}
pub async fn execute_sql(&self, sql: impl IntoRawSql) -> crate::Result<u64> {
exec_raw_sql_on(ConnRefMut::Db(self), sql.into_raw_sql()).await
}
pub async fn table_row_count(&self, table_name: &str) -> crate::Result<u64> {
let sql = format!(
"SELECT COUNT(*) FROM {}",
common_helpers::quote_table_name_with_schema(self.db_type(), table_name)
);
let rows = self.select_sql::<i64>(sql).collect::<Vec<i64>>().await?;
Ok(rows.into_iter().next().unwrap_or(0).max(0) as u64)
}
#[cfg(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb",
feature = "clickhouse"
))]
pub fn create_pool(
db_type: super::super::DbType,
connection_string: &str,
) -> super::connection_pool::PoolBuilder {
super::connection_pool::PoolBuilder::new(db_type, connection_string)
}
}
impl super::DbExecutor for Database {
fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
Database::select::<T>(self)
}
fn select_column<T: Model, V>(&self) -> ProjectionSelectExecutor<'_, T, V> {
Database::select_column::<T, V>(self)
}
}
impl<'a, R: Model> DerivedTableSelectExecutor<'a, R> {
pub fn filter<F, W>(mut self, f: F) -> Self
where
F: FnOnce(R::Where) -> W,
W: Into<WhereExpr>,
{
self.select = self.select.filter(f);
self
}
pub fn order_by<F, O>(mut self, f: F) -> Self
where
F: FnOnce(R::Where) -> O,
O: Into<crate::OrderBy>,
{
self.select = self.select.order_by(f);
self
}
pub fn order_by_desc<F, O>(mut self, f: F) -> Self
where
F: FnOnce(R::Where) -> O,
O: Into<crate::OrderBy>,
{
self.select = self.select.order_by_desc(f);
self
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(mut self, range: RR) -> Self {
self.select = self.select.range(range);
self
}
pub fn collect<C>(self) -> DerivedTableCollectFuture<'a, R, C>
where
R: crate::model::FromRowValues + 'static,
C: FromIterator<R> + 'static,
{
DerivedTableCollectFuture {
db: self.db,
select: self.select,
_marker: std::marker::PhantomData,
}
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
let db_type = self.db.db_type();
#[cfg(feature = "postgresql")]
if matches!(db_type, crate::DbType::PostgreSQL) {
let (sql, params, rust_types) =
self.select.try_to_sql_with_params_and_types(db_type)?;
return Ok(SqlStatement::batch(
db_type,
vec![super::SingleSqlStatement::new(sql, params).with_param_rust_types(rust_types)],
));
}
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
Ok(SqlStatement::single(db_type, sql, params))
}
}
pub struct DerivedTableCollectFuture<'a, R: Model, C> {
db: &'a Database,
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb",
feature = "clickhouse"
)),
allow(dead_code)
)]
select: DerivedTableSelect<R>,
_marker: std::marker::PhantomData<C>,
}
impl<'a, R, C> std::future::IntoFuture for DerivedTableCollectFuture<'a, R, C>
where
R: Model + crate::model::FromRowValues + 'static + std::marker::Send,
C: FromIterator<R> + 'static,
{
type Output = crate::Result<C>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb",
feature = "clickhouse"
)),
allow(unused_variables)
)]
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move {
let db_type = self.db.db_type();
match self.db {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => {
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
db.select_raw::<R, C>(&sql, params).await
}
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => {
let (sql, params, rust_types) =
self.select.try_to_sql_with_params_and_types(db_type)?;
db.select_raw_with_types::<R, C>(&sql, params, rust_types)
.await
}
#[cfg(feature = "mysql")]
Database::MySQL(db) => {
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
db.select_raw::<R, C>(&sql, params).await
}
#[cfg(feature = "mssql")]
Database::MSSQL(db) => {
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
db.select_raw::<R, C>(&sql, params).await
}
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => {
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
db.select_raw::<R, C>(&sql, params).await
}
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => {
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
let rows = db
.select_values(RawSql::new(sql).with_params(params), R::row_columns())
.await?;
rows.into_iter()
.map(|values| <R as crate::model::FromRowValues>::from_row_values(&values))
.collect::<crate::Result<C>>()
}
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => Err(unsupported_feature(
super::super::DbType::InfluxDB,
"derived table select",
)),
}
})
}
}
pub struct UnionSelectExecutor<'a, T: Model> {
db: &'a Database,
select: crate::query::builder::UnionSelect<T>,
}
impl<'a, T: Model> UnionSelectExecutor<'a, T> {
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
let db_type = self.db.db_type();
#[cfg(feature = "mssql")]
let (sql, params) = if db_type == super::super::DbType::MSSQL {
self.select.to_sql_with_params_unparenthesized(db_type)
} else {
self.select.try_to_sql_with_params(db_type)?
};
#[cfg(not(feature = "mssql"))]
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
Ok(SqlStatement::single(db_type, sql, params))
}
pub fn collect<C>(self) -> UnionCollectFuture<'a, T, C>
where
T: crate::model::FromRowValues + 'static,
C: FromIterator<T> + 'static,
{
UnionCollectFuture {
db: self.db,
select: self.select,
_marker: std::marker::PhantomData,
}
}
pub async fn first(self) -> crate::Result<Option<T>>
where
T: crate::model::FromRowValues + 'static + Send,
{
Ok(self
.collect::<Vec<T>>()
.await?
.into_iter()
.next())
}
}
pub struct UnionCollectFuture<'a, T: Model, C> {
db: &'a Database,
select: crate::query::builder::UnionSelect<T>,
_marker: std::marker::PhantomData<C>,
}
impl<'a, T, C> std::future::IntoFuture for UnionCollectFuture<'a, T, C>
where
T: Model + crate::model::FromRowValues + 'static + std::marker::Send,
C: FromIterator<T> + 'static,
{
type Output = crate::Result<C>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb",
feature = "clickhouse"
)),
allow(unused_variables)
)]
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move {
let db_type = self.db.db_type();
#[cfg(feature = "mssql")]
let (sql, params) = if db_type == super::super::DbType::MSSQL {
self.select.to_sql_with_params_unparenthesized(db_type)
} else {
self.select.try_to_sql_with_params(db_type)?
};
#[cfg(not(feature = "mssql"))]
let (sql, params) = self.select.try_to_sql_with_params(db_type)?;
match self.db {
#[cfg(feature = "sqlite")]
Database::Sqlite(db) => db.select_raw::<T, C>(&sql, params).await,
#[cfg(feature = "postgresql")]
Database::PostgreSQL(db) => db.select_raw::<T, C>(&sql, params).await,
#[cfg(feature = "mysql")]
Database::MySQL(db) => db.select_raw::<T, C>(&sql, params).await,
#[cfg(feature = "duckdb")]
Database::DuckDB(db) => db.select_raw::<T, C>(&sql, params).await,
#[cfg(feature = "mssql")]
Database::MSSQL(db) => db.select_raw::<T, C>(&sql, params).await,
#[cfg(feature = "clickhouse")]
Database::ClickHouse(db) => {
let rows = db
.select_values(RawSql::new(sql).with_params(params), T::row_columns())
.await?;
rows.into_iter()
.map(|values| <T as crate::model::FromRowValues>::from_row_values(&values))
.collect::<crate::Result<C>>()
}
#[cfg(feature = "influxdb")]
Database::InfluxDB(_) => Err(unsupported_feature(
super::super::DbType::InfluxDB,
"union select",
)),
}
})
}
}
#[derive(Clone, Copy)]
pub(crate) enum ConnRef<'a> {
Db(&'a Database),
Txn(&'a Transaction<'a>),
Pooled(&'a connection_pool::ConnectionWrapper),
}
pub(crate) enum ConnRefMut<'a, 'tx> {
Db(&'a Database),
Txn(&'a mut Transaction<'tx>),
Pooled(&'a connection_pool::ConnectionWrapper),
}
pub(crate) async fn exec_raw_sql_on(
conn: ConnRefMut<'_, '_>,
sql: RawSql,
) -> crate::Result<u64> {
match conn {
#[cfg(feature = "sqlite")]
ConnRefMut::Db(Database::Sqlite(db)) => {
let (sql, params) = sql.render(super::super::DbType::Sqlite)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "sqlite")]
ConnRefMut::Txn(Transaction::Sqlite(txn)) => {
let (sql, params) = sql.render(super::super::DbType::Sqlite)?;
txn.exec_raw(&sql, params).await
}
#[cfg(feature = "postgresql")]
ConnRefMut::Db(Database::PostgreSQL(db)) => {
let (sql, params) = sql.render(super::super::DbType::PostgreSQL)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "postgresql")]
ConnRefMut::Txn(Transaction::PostgreSQL(txn)) => {
let (sql, params) = sql.render(super::super::DbType::PostgreSQL)?;
txn.exec_raw(&sql, params).await
}
#[cfg(feature = "mysql")]
ConnRefMut::Db(Database::MySQL(db)) => {
let (sql, params) = sql.render(super::super::DbType::MySQL)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "mysql")]
ConnRefMut::Txn(Transaction::MySQL(txn)) => {
let (sql, params) = sql.render(super::super::DbType::MySQL)?;
txn.exec_raw(&sql, params).await
}
#[cfg(feature = "mssql")]
ConnRefMut::Db(Database::MSSQL(db)) => {
let (sql, params) = sql.render(super::super::DbType::MSSQL)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "mssql")]
ConnRefMut::Txn(Transaction::MSSQL(txn)) => {
let (sql, params) = sql.render(super::super::DbType::MSSQL)?;
txn.exec_raw(&sql, params).await
}
#[cfg(feature = "duckdb")]
ConnRefMut::Db(Database::DuckDB(db)) => {
let (sql, params) = sql.render(super::super::DbType::DuckDB)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "duckdb")]
ConnRefMut::Txn(Transaction::DuckDB(txn)) => {
let (sql, params) = sql.render(super::super::DbType::DuckDB)?;
txn.exec_raw(&sql, params).await
}
#[cfg(feature = "postgresql")]
ConnRefMut::Pooled(connection_pool::ConnectionWrapper::PostgreSQL(db)) => {
let (sql, params) = sql.render(super::super::DbType::PostgreSQL)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "mysql")]
ConnRefMut::Pooled(connection_pool::ConnectionWrapper::MySQL(db)) => {
let (sql, params) = sql.render(super::super::DbType::MySQL)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "mssql")]
ConnRefMut::Pooled(connection_pool::ConnectionWrapper::MSSQL(db)) => {
let (sql, params) = sql.render(super::super::DbType::MSSQL)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "duckdb")]
ConnRefMut::Pooled(connection_pool::ConnectionWrapper::DuckDB(db)) => {
let (sql, params) = sql.render(super::super::DbType::DuckDB)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "sqlite")]
ConnRefMut::Pooled(connection_pool::ConnectionWrapper::Sqlite(db)) => {
let (sql, params) = sql.render(super::super::DbType::Sqlite)?;
db.exec_raw(&sql, params).await
}
#[cfg(feature = "clickhouse")]
ConnRefMut::Db(Database::ClickHouse(db)) => {
db.execute_sql(sql).await?;
Ok(0)
}
#[cfg(feature = "clickhouse")]
ConnRefMut::Pooled(connection_pool::ConnectionWrapper::ClickHouse(db)) => {
db.execute_sql(sql).await?;
Ok(0)
}
#[cfg(feature = "influxdb")]
ConnRefMut::Db(Database::InfluxDB(db)) => {
db.execute_sql(sql).await?;
Ok(0)
}
#[cfg(feature = "influxdb")]
ConnRefMut::Pooled(connection_pool::ConnectionWrapper::InfluxDB(db)) => {
db.execute_sql(sql).await?;
Ok(0)
}
ConnRefMut::Txn(Transaction::_Phantom(infallible, _)) => match *infallible {},
}
}
pub struct RawSelectExecutor<'a, T> {
conn: ConnRef<'a>,
sql: RawSql,
_marker: std::marker::PhantomData<T>,
}
impl<'a, T> RawSelectExecutor<'a, T> {
pub(crate) fn from_conn(conn: ConnRef<'a>, sql: RawSql) -> Self {
Self {
conn,
sql,
_marker: std::marker::PhantomData,
}
}
pub fn collect<C>(self) -> RawCollectFuture<'a, T, C>
where
T: crate::model::FromRowValues + 'static,
C: FromIterator<T> + 'static,
{
RawCollectFuture {
conn: self.conn,
sql: self.sql,
_marker: std::marker::PhantomData,
}
}
}
#[deprecated(
since = "0.2.12",
note = "TransactionRawSelectExecutor 已合并为 RawSelectExecutor,请改用 RawSelectExecutor"
)]
pub type TransactionRawSelectExecutor<'a, 'tx, T> = RawSelectExecutor<'a, T>;
#[deprecated(
since = "0.2.12",
note = "PooledRawSelectExecutor 已合并为 RawSelectExecutor,请改用 RawSelectExecutor"
)]
pub type PooledRawSelectExecutor<'conn, 'pool, T> = RawSelectExecutor<'conn, T>;
pub struct RawCollectFuture<'a, T, C> {
conn: ConnRef<'a>,
sql: RawSql,
_marker: std::marker::PhantomData<(T, C)>,
}
#[deprecated(
since = "0.2.12",
note = "TransactionRawCollectFuture 已合并为 RawCollectFuture,请改用 RawCollectFuture"
)]
pub type TransactionRawCollectFuture<'a, 'tx, T, C> = RawCollectFuture<'a, T, C>;
impl<'a, T, C> std::future::IntoFuture for RawCollectFuture<'a, T, C>
where
T: crate::model::FromRowValues + 'static + std::marker::Send,
C: FromIterator<T> + 'static,
{
type Output = crate::Result<C>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move {
match self.conn {
#[cfg(feature = "sqlite")]
ConnRef::Db(Database::Sqlite(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::Sqlite)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "sqlite")]
ConnRef::Txn(Transaction::Sqlite(txn)) => {
let (sql, params) = self.sql.render(super::super::DbType::Sqlite)?;
txn.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "postgresql")]
ConnRef::Db(Database::PostgreSQL(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::PostgreSQL)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "postgresql")]
ConnRef::Txn(Transaction::PostgreSQL(txn)) => {
let (sql, params) = self.sql.render(super::super::DbType::PostgreSQL)?;
txn.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "mysql")]
ConnRef::Db(Database::MySQL(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::MySQL)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "mysql")]
ConnRef::Txn(Transaction::MySQL(txn)) => {
let (sql, params) = self.sql.render(super::super::DbType::MySQL)?;
txn.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "mssql")]
ConnRef::Db(Database::MSSQL(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::MSSQL)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "mssql")]
ConnRef::Txn(Transaction::MSSQL(txn)) => {
let (sql, params) = self.sql.render(super::super::DbType::MSSQL)?;
txn.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "duckdb")]
ConnRef::Db(Database::DuckDB(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::DuckDB)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "duckdb")]
ConnRef::Txn(Transaction::DuckDB(txn)) => {
let (sql, params) = self.sql.render(super::super::DbType::DuckDB)?;
txn.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "postgresql")]
ConnRef::Pooled(connection_pool::ConnectionWrapper::PostgreSQL(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::PostgreSQL)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "mysql")]
ConnRef::Pooled(connection_pool::ConnectionWrapper::MySQL(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::MySQL)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "mssql")]
ConnRef::Pooled(connection_pool::ConnectionWrapper::MSSQL(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::MSSQL)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "duckdb")]
ConnRef::Pooled(connection_pool::ConnectionWrapper::DuckDB(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::DuckDB)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "sqlite")]
ConnRef::Pooled(connection_pool::ConnectionWrapper::Sqlite(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::Sqlite)?;
db.select_raw::<T, C>(&sql, params).await
}
#[cfg(feature = "clickhouse")]
ConnRef::Db(Database::ClickHouse(db)) => {
let (sql, params) = self.sql.render(super::super::DbType::ClickHouse)?;
let rows = db
.select_values(
crate::raw_sql::RawSql::new(sql).with_params(params),
<T as crate::model::FromRowValues>::row_columns(),
)
.await?;
rows.into_iter()
.map(|values| T::from_row_values(&values))
.collect::<crate::Result<C>>()
}
#[cfg(feature = "clickhouse")]
ConnRef::Pooled(connection_pool::ConnectionWrapper::ClickHouse(db)) => {
let rows = db
.select_values(self.sql, <T as crate::model::FromRowValues>::row_columns())
.await?;
rows.into_iter()
.map(|values| T::from_row_values(&values))
.collect::<crate::Result<C>>()
}
#[cfg(feature = "influxdb")]
ConnRef::Db(Database::InfluxDB(db)) => {
let rows = db
.select_values(
self.sql,
<T as crate::model::FromRowValues>::row_columns(),
)
.await?;
rows.into_iter()
.map(|values| T::from_row_values(&values))
.collect::<crate::Result<C>>()
}
#[cfg(feature = "influxdb")]
ConnRef::Pooled(connection_pool::ConnectionWrapper::InfluxDB(db)) => {
let rows = db
.select_values(self.sql, <T as crate::model::FromRowValues>::row_columns())
.await?;
rows.into_iter()
.map(|values| T::from_row_values(&values))
.collect::<crate::Result<C>>()
}
ConnRef::Txn(Transaction::_Phantom(infallible, _)) => match *infallible {},
}
})
}
}
pub enum SelectExecutor<'a, T: Model> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::SelectExecutor<'a, T>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::SelectExecutor<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::SelectExecutor<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::SelectExecutor<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::SelectExecutor<'a, T>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ClickHouse(crate::abstract_layer::common::unified::ClickHouseSelectBackend<'a>, crate::query::builder::Select<T>),
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[derive(Clone, Copy)]
pub enum ClickHouseSelectBackend<'a> {
#[cfg(feature = "clickhouse")]
ClickHouse(&'a clickhouse_backend::Database),
#[cfg(feature = "influxdb")]
#[doc(hidden)]
Influx(&'a influxdb_backend::Database),
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
fn clickhouse_select_backend_db_type(
db: ClickHouseSelectBackend<'_>,
) -> super::super::DbType {
match db {
#[cfg(feature = "clickhouse")]
ClickHouseSelectBackend::ClickHouse(_) => super::super::DbType::ClickHouse,
#[cfg(feature = "influxdb")]
ClickHouseSelectBackend::Influx(_) => super::super::DbType::InfluxDB,
}
}
crate::impl_unified_select_executor_methods!(SelectExecutor);
impl<'a, T: Model> SelectExecutor<'a, T> {
pub fn fields<F, G>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> G,
G: crate::query::builder::GroupByColumns,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectExecutor::Sqlite(exec.fields(f)),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => SelectExecutor::PostgreSQL(exec.fields(f)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectExecutor::MySQL(exec.fields(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectExecutor::MSSQL(exec.fields(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectExecutor::DuckDB(exec.fields(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
SelectExecutor::ClickHouse(db, select.fields(f))
}
}
}
pub fn query(self, query: impl Into<String>) -> Self {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectExecutor::Sqlite(exec.query(query)),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => SelectExecutor::PostgreSQL(exec.query(query)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectExecutor::MySQL(exec.query(query)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectExecutor::MSSQL(exec.query(query)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectExecutor::DuckDB(exec.query(query)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
SelectExecutor::ClickHouse(db, select.query(query))
}
}
}
pub fn mode(self, mode: crate::query::filter::FullTextMode) -> Self {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectExecutor::Sqlite(exec.mode(mode)),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => SelectExecutor::PostgreSQL(exec.mode(mode)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectExecutor::MySQL(exec.mode(mode)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectExecutor::MSSQL(exec.mode(mode)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectExecutor::DuckDB(exec.mode(mode)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
SelectExecutor::ClickHouse(db, select.mode(mode))
}
}
}
pub fn language(self, language: impl Into<String>) -> Self {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectExecutor::Sqlite(exec.language(language)),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => SelectExecutor::PostgreSQL(exec.language(language)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectExecutor::MySQL(exec.language(language)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectExecutor::MSSQL(exec.language(language)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectExecutor::DuckDB(exec.language(language)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
SelectExecutor::ClickHouse(db, select.language(language))
}
}
}
pub fn rank(self, rank: crate::query::filter::FullTextRank) -> Self {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectExecutor::Sqlite(exec.rank(rank)),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => SelectExecutor::PostgreSQL(exec.rank(rank)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectExecutor::MySQL(exec.rank(rank)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectExecutor::MSSQL(exec.rank(rank)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectExecutor::DuckDB(exec.rank(rank)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
SelectExecutor::ClickHouse(db, select.rank(rank))
}
}
}
}
impl<'a, T: Model> FilterQuery<T> for SelectExecutor<'a, T> {
fn append_filter_expr(self, expr: WhereExpr) -> Self {
SelectExecutor::append_filter_expr(self, expr)
}
}
impl<'a, T: Model> NamedFilterQuery<T> for SelectExecutor<'a, T> {
fn apply_named_filter(self, name: &'static str, expr: WhereExpr) -> Self {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectExecutor::Sqlite(
exec.with_context_filters(vec![ContextFilter::new::<T>(name, expr)]),
),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => SelectExecutor::PostgreSQL(
exec.with_context_filters(vec![ContextFilter::new::<T>(name, expr)]),
),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectExecutor::MySQL(
exec.with_context_filters(vec![ContextFilter::new::<T>(name, expr)]),
),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectExecutor::MSSQL(
exec.with_context_filters(vec![ContextFilter::new::<T>(name, expr)]),
),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectExecutor::DuckDB(
exec.with_context_filters(vec![ContextFilter::new::<T>(name, expr)]),
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => SelectExecutor::ClickHouse(
db,
NamedFilterQuery::<T>::apply_named_filter(select, name, expr),
),
}
}
}
impl<'a, T: Model> WithoutFilterQuery<T> for SelectExecutor<'a, T> {
fn without_filter(self, name: &'static str) -> Self {
SelectExecutor::without_filter(self, name)
}
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
impl<'a, T: Model> SelectExecutor<'a, T> {
fn select_model<R: Model>(&self) -> SelectExecutor<'a, R> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectExecutor::Sqlite(exec.select_model::<R>()),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
SelectExecutor::PostgreSQL(exec.select_model::<R>())
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectExecutor::MySQL(exec.select_model::<R>()),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectExecutor::MSSQL(exec.select_model::<R>()),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectExecutor::DuckDB(exec.select_model::<R>()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => SelectExecutor::ClickHouse(
*db,
Select::default().with_context_filters(select.context_filters()),
),
}
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => exec.to_sql(),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => exec.to_sql(),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => exec.to_sql(),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => exec.to_sql(),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => exec.to_sql(),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
let backend = clickhouse_select_backend_db_type(*db);
let (sql, params) = select.try_to_sql_with_params(backend)?;
Ok(SqlStatement::single(backend, sql, params))
}
}
}
pub fn include<F, S>(self, f: F) -> IncludedSelectExecutor<'a, T, S>
where
F: FnOnce(T::Where) -> S,
S: RelationSelection<T>,
{
let where_obj = T::Where::default();
IncludedSelectExecutor {
select: self,
selection: f(where_obj),
_marker: std::marker::PhantomData,
}
}
pub(crate) async fn select_related_with_selection<S>(
&self,
keys: Vec<Value>,
selection: &S,
) -> crate::Result<Vec<S::Target>>
where
S: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync,
S::Target: Send + Sync,
S::Via: Send + Sync,
T: 'static + Send + Sync,
{
match selection.path_info()? {
RelationPathInfo::Direct { relation } => {
self.select_target_models::<S>(relation.target_key, keys, selection)
.await
}
RelationPathInfo::Through {
via_relation,
target_relation,
..
} => {
let via_items = self.select_via_models::<S>(via_relation, keys).await?;
let target_keys = via_items
.iter()
.filter_map(|item| item.column_value(target_relation.local_key))
.collect();
self.select_target_models::<S>(target_relation.target_key, target_keys, selection)
.await
}
}
}
async fn select_target_models<S>(
&self,
target_key: &str,
keys: Vec<Value>,
selection: &S,
) -> crate::Result<Vec<S::Target>>
where
S: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync,
S::Target: Send + Sync,
S::Via: Send + Sync,
T: 'static + Send + Sync,
{
let values = relation_filter_values(keys);
if values.is_empty() {
return Ok(Vec::new());
}
let mut exec = self.select_model::<S::Target>().filter(|_| {
WhereExpr::from_filter(FilterExpr::In {
column: target_key.to_string(),
values,
})
});
for filter in selection.filters().iter().cloned() {
exec = exec.filter(|_| WhereExpr::from_filter(filter));
}
for order in selection.order_by().iter().cloned() {
exec = exec.order_by(|_| order);
}
if selection.range_start().is_some() || selection.range_end().is_some() {
exec = exec.range(crate::query::builder::RangeBounds {
start: selection.range_start(),
end: selection.range_end(),
});
}
let mut related = exec.collect::<Vec<S::Target>>().await?;
let target_select = self.select_model::<S::Target>();
selection.load_nested(&target_select, &mut related).await?;
Ok(related)
}
async fn select_via_models<S>(
&self,
via_relation: &RelationInfo,
keys: Vec<Value>,
) -> crate::Result<Vec<S::Via>>
where
S: RelationSelection<T>,
S::Via: Send + Sync,
{
let values = relation_filter_values(keys);
if values.is_empty() {
return Ok(Vec::new());
}
self.select_model::<S::Via>()
.filter(|_| {
WhereExpr::from_filter(FilterExpr::In {
column: via_relation.target_key.to_string(),
values,
})
})
.collect::<Vec<S::Via>>()
.await
}
pub(crate) async fn preload_models_with_selection<S>(
&self,
owners: &mut [T],
selection: S,
) -> crate::Result<()>
where
S: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync,
S::Target: Send + Sync,
S::Via: Send + Sync,
T: 'static + Send + Sync,
{
let path = selection.path_info()?;
let owner_relation = relation_owner_key(path);
let owner_keys = owners
.iter()
.map(|owner| owner.relation_key_value(owner_relation))
.collect::<crate::Result<Vec<_>>>()?;
match path {
RelationPathInfo::Direct { relation } => {
let related = self
.select_target_models::<S>(relation.target_key, owner_keys, &selection)
.await?;
let mut grouped: std::collections::HashMap<String, Vec<S::Target>> =
std::collections::HashMap::new();
for item in related {
if let Some(key) = item.column_value(relation.target_key) {
grouped
.entry(common_helpers::model_value_key(&key))
.or_default()
.push(item);
}
}
for owner in owners {
let key = owner.relation_key_value(relation)?;
let values = grouped
.get(&common_helpers::model_value_key(&key))
.cloned()
.unwrap_or_default();
owner.assign_relation(relation.name, values)?;
}
}
RelationPathInfo::Through {
relation,
via_relation,
target_relation,
} => {
let via_items = self
.select_via_models::<S>(via_relation, owner_keys)
.await?;
let mut target_keys_by_owner: std::collections::HashMap<String, Vec<String>> =
std::collections::HashMap::new();
for item in &via_items {
if let (Some(owner_key), Some(target_key)) = (
item.column_value(via_relation.target_key),
item.column_value(target_relation.local_key),
) {
let target_key = common_helpers::model_value_key(&target_key);
target_keys_by_owner
.entry(common_helpers::model_value_key(&owner_key))
.or_default()
.push(target_key.clone());
}
}
let target_key_values = via_items
.iter()
.filter_map(|item| item.column_value(target_relation.local_key))
.collect();
let related = self
.select_target_models::<S>(
target_relation.target_key,
target_key_values,
&selection,
)
.await?;
let mut targets_by_key: std::collections::HashMap<String, Vec<S::Target>> =
std::collections::HashMap::new();
for item in &related {
if let Some(key) = item.column_value(target_relation.target_key) {
targets_by_key
.entry(common_helpers::model_value_key(&key))
.or_default()
.push(item.clone());
}
}
for owner in owners {
let key = owner.relation_key_value(via_relation)?;
let key = common_helpers::model_value_key(&key);
let values = target_keys_by_owner
.get(&key)
.map(|target_keys| {
target_keys
.iter()
.flat_map(|target_key| {
targets_by_key.get(target_key).into_iter().flatten()
})
.cloned()
.collect()
})
.unwrap_or_default();
owner.assign_relation(relation.name, values)?;
}
}
}
Ok(())
}
pub fn from<R: Model>(self) -> RelatedSelectExecutor<'a, T, R>
where
T: Model + 'static,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
RelatedSelectExecutor::Sqlite(exec.from::<R>(), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
RelatedSelectExecutor::PostgreSQL(exec.from::<R>())
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => RelatedSelectExecutor::MySQL(exec.from::<R>()),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => RelatedSelectExecutor::MSSQL(exec.from::<R>()),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => RelatedSelectExecutor::DuckDB(exec.from::<R>()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => RelatedSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "related (multi-table) select on ClickHouse",
_marker: std::marker::PhantomData,
},
}
}
pub fn from3<R1: Model, R2: Model>(self) -> MultiTableSelectExecutor<'a, T, R1, R2>
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => MultiTableSelectExecutor::Sqlite(
exec.from3::<R1, R2>(),
std::marker::PhantomData,
),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
MultiTableSelectExecutor::PostgreSQL(exec.from3::<R1, R2>())
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
MultiTableSelectExecutor::MySQL(exec.from3::<R1, R2>())
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
MultiTableSelectExecutor::MSSQL(exec.from3::<R1, R2>())
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => MultiTableSelectExecutor::DuckDB(
exec.from3::<R1, R2>(),
std::marker::PhantomData,
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => MultiTableSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "related (multi-table) select",
_marker: std::marker::PhantomData,
},
}
}
pub fn from4<R1: Model, R2: Model, R3: Model>(
self,
) -> FourTableSelectExecutor<'a, T, R1, R2, R3>
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => FourTableSelectExecutor::Sqlite(
exec.from4::<R1, R2, R3>(),
std::marker::PhantomData,
),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
FourTableSelectExecutor::PostgreSQL(exec.from4::<R1, R2, R3>())
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
FourTableSelectExecutor::MySQL(exec.from4::<R1, R2, R3>())
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
FourTableSelectExecutor::MSSQL(exec.from4::<R1, R2, R3>())
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => FourTableSelectExecutor::DuckDB(
exec.from4::<R1, R2, R3>(),
std::marker::PhantomData,
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => FourTableSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "related (multi-table) select",
_marker: std::marker::PhantomData,
},
}
}
pub fn left_join<J: Model>(
self,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> LeftJoinedSelectExecutor<'a, T, J> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
LeftJoinedSelectExecutor::Sqlite(exec.left_join::<J>(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
LeftJoinedSelectExecutor::PostgreSQL(exec.left_join::<J>(f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => LeftJoinedSelectExecutor::MySQL(exec.left_join::<J>(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => LeftJoinedSelectExecutor::MSSQL(exec.left_join::<J>(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
LeftJoinedSelectExecutor::DuckDB(exec.left_join::<J>(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => LeftJoinedSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "LEFT JOIN select",
_marker: std::marker::PhantomData,
},
}
}
pub fn inner_join<J: Model>(
self,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> InnerJoinedSelectExecutor<'a, T, J> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
InnerJoinedSelectExecutor::Sqlite(exec.inner_join::<J>(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
InnerJoinedSelectExecutor::PostgreSQL(exec.inner_join::<J>(f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
InnerJoinedSelectExecutor::MySQL(exec.inner_join::<J>(f))
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
InnerJoinedSelectExecutor::MSSQL(exec.inner_join::<J>(f))
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
InnerJoinedSelectExecutor::DuckDB(exec.inner_join::<J>(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => InnerJoinedSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "INNER JOIN select",
_marker: std::marker::PhantomData,
},
}
}
pub fn right_join<J: Model>(
self,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> RightJoinedSelectExecutor<'a, T, J> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
RightJoinedSelectExecutor::Sqlite(exec.right_join::<J>(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
RightJoinedSelectExecutor::PostgreSQL(exec.right_join::<J>(f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
RightJoinedSelectExecutor::MySQL(exec.right_join::<J>(f))
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
RightJoinedSelectExecutor::MSSQL(exec.right_join::<J>(f))
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
RightJoinedSelectExecutor::DuckDB(exec.right_join::<J>(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => RightJoinedSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "RIGHT JOIN select",
_marker: std::marker::PhantomData,
},
}
}
pub fn left_join_derived<J: Model>(
self,
derived: DerivedSelect<J>,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> LeftJoinedSelectExecutor<'a, T, J> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => LeftJoinedSelectExecutor::Sqlite(
exec.left_join_derived::<J>(derived, f),
std::marker::PhantomData,
),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
LeftJoinedSelectExecutor::PostgreSQL(exec.left_join_derived::<J>(derived, f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
LeftJoinedSelectExecutor::MySQL(exec.left_join_derived::<J>(derived, f))
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
LeftJoinedSelectExecutor::MSSQL(exec.left_join_derived::<J>(derived, f))
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
LeftJoinedSelectExecutor::DuckDB(exec.left_join_derived::<J>(derived, f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => LeftJoinedSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "JOIN select with a derived table",
_marker: std::marker::PhantomData,
},
}
}
pub fn inner_join_derived<J: Model>(
self,
derived: DerivedSelect<J>,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> InnerJoinedSelectExecutor<'a, T, J> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => InnerJoinedSelectExecutor::Sqlite(
exec.inner_join_derived::<J>(derived, f),
std::marker::PhantomData,
),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
InnerJoinedSelectExecutor::PostgreSQL(exec.inner_join_derived::<J>(derived, f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
InnerJoinedSelectExecutor::MySQL(exec.inner_join_derived::<J>(derived, f))
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
InnerJoinedSelectExecutor::MSSQL(exec.inner_join_derived::<J>(derived, f))
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
InnerJoinedSelectExecutor::DuckDB(exec.inner_join_derived::<J>(derived, f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => InnerJoinedSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "JOIN select with a derived table",
_marker: std::marker::PhantomData,
},
}
}
pub fn right_join_derived<J: Model>(
self,
derived: DerivedSelect<J>,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> RightJoinedSelectExecutor<'a, T, J> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => RightJoinedSelectExecutor::Sqlite(
exec.right_join_derived::<J>(derived, f),
std::marker::PhantomData,
),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
RightJoinedSelectExecutor::PostgreSQL(exec.right_join_derived::<J>(derived, f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
RightJoinedSelectExecutor::MySQL(exec.right_join_derived::<J>(derived, f))
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
RightJoinedSelectExecutor::MSSQL(exec.right_join_derived::<J>(derived, f))
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
RightJoinedSelectExecutor::DuckDB(exec.right_join_derived::<J>(derived, f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => RightJoinedSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "JOIN select with a derived table",
_marker: std::marker::PhantomData,
},
}
}
pub fn collect<C: FromIterator<T> + 'static>(&self) -> CollectFuture<'a, T, C>
where
T: 'static,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => CollectFuture::Sqlite(exec.clone().collect::<C>()),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
CollectFuture::PostgreSQL(exec.clone_with_client().collect::<C>())
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
CollectFuture::MySQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
CollectFuture::MSSQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => CollectFuture::DuckDB(exec.clone().collect::<C>()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
CollectFuture::ClickHouse(
*db,
select.clone(),
std::marker::PhantomData,
)
}
}
}
pub fn first(self) -> FirstFuture<'a, T>
where
T: 'static,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => FirstFuture::Sqlite(exec.first()),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => FirstFuture::PostgreSQL(exec.first()),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => FirstFuture::MySQL(exec.first()),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => FirstFuture::MSSQL(exec.first()),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => FirstFuture::DuckDB(exec.first()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => FirstFuture::ClickHouse(db, select),
}
}
pub fn count<F, C>(self, f: F) -> AggregateFuture<'a, T, usize>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
AggregateFuture::Sqlite(exec.count(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => AggregateFuture::PostgreSQL(exec.count(f)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => AggregateFuture::MySQL(exec.count(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => AggregateFuture::MSSQL(exec.count(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => AggregateFuture::DuckDB(exec.count(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
AggregateFuture::ClickHouse(db, select.count(f), std::marker::PhantomData)
}
}
}
pub fn sum<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
C: crate::query::builder::AggregateResultType + 'static,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
AggregateFuture::Sqlite(exec.sum(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => AggregateFuture::PostgreSQL(exec.sum(f)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => AggregateFuture::MySQL(exec.sum(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => AggregateFuture::MSSQL(exec.sum(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => AggregateFuture::DuckDB(exec.sum(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
AggregateFuture::ClickHouse(db, select.sum(f), std::marker::PhantomData)
}
}
}
pub fn avg<F, C>(self, f: F) -> AggregateFuture<'a, T, Option<f64>>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
C: crate::query::builder::AggregateResultType + 'static,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
AggregateFuture::Sqlite(exec.avg(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => AggregateFuture::PostgreSQL(exec.avg(f)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => AggregateFuture::MySQL(exec.avg(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => AggregateFuture::MSSQL(exec.avg(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => AggregateFuture::DuckDB(exec.avg(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
AggregateFuture::ClickHouse(db, select.avg(f), std::marker::PhantomData)
}
}
}
pub fn max<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
C: crate::query::builder::AggregateResultType + 'static,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
AggregateFuture::Sqlite(exec.max(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => AggregateFuture::PostgreSQL(exec.max(f)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => AggregateFuture::MySQL(exec.max(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => AggregateFuture::MSSQL(exec.max(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => AggregateFuture::DuckDB(exec.max(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
AggregateFuture::ClickHouse(db, select.max(f), std::marker::PhantomData)
}
}
}
pub fn min<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
C: crate::query::builder::AggregateResultType + 'static,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
AggregateFuture::Sqlite(exec.min(f), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => AggregateFuture::PostgreSQL(exec.min(f)),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => AggregateFuture::MySQL(exec.min(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => AggregateFuture::MSSQL(exec.min(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => AggregateFuture::DuckDB(exec.min(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
AggregateFuture::ClickHouse(db, select.min(f), std::marker::PhantomData)
}
}
}
}
pub enum DeleteExecutor<'a, T: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::DeleteExecutor<T>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::DeleteExecutor<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::DeleteExecutor<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::DeleteExecutor<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::DeleteExecutor<T>),
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a T>,
},
}
crate::impl_unified_delete_executor!(DeleteExecutor);
impl<'a, T: Model> DeleteExecutor<'a, T> {
#[allow(unused_variables)]
pub fn route_table(
self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
#[allow(unused_variables)]
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
match self {
#[cfg(feature = "sqlite")]
DeleteExecutor::Sqlite(exec, phantom) => {
DeleteExecutor::Sqlite(exec.with_table_route(route), phantom)
}
#[cfg(feature = "postgresql")]
DeleteExecutor::PostgreSQL(exec) => {
DeleteExecutor::PostgreSQL(exec.with_table_route(route))
}
#[cfg(feature = "mysql")]
DeleteExecutor::MySQL(exec) => DeleteExecutor::MySQL(exec.with_table_route(route)),
#[cfg(feature = "mssql")]
DeleteExecutor::MSSQL(exec) => DeleteExecutor::MSSQL(exec.with_table_route(route)),
#[cfg(feature = "duckdb")]
DeleteExecutor::DuckDB(exec) => DeleteExecutor::DuckDB(exec.with_table_route(route)),
unsupported @ DeleteExecutor::Unsupported { .. } => unsupported,
}
}
}
impl<'a, T: Model> NamedFilterQuery<T> for DeleteExecutor<'a, T> {
fn apply_named_filter(self, _name: &'static str, expr: WhereExpr) -> Self {
self.filter(|_| expr)
}
}
impl<'a, T: Model> super::SqlExecutor for DeleteExecutor<'a, T> {
type Output = u64;
fn to_sql(&self) -> crate::Result<SqlStatement> {
DeleteExecutor::to_sql(self)
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
match self {
#[cfg(feature = "sqlite")]
DeleteExecutor::Sqlite(exec, _) => exec.execute_with_sql(sql).await,
#[cfg(feature = "postgresql")]
DeleteExecutor::PostgreSQL(exec) => exec.execute_with_sql(sql).await,
#[cfg(feature = "mysql")]
DeleteExecutor::MySQL(exec) => exec.execute_with_sql(sql).await,
#[cfg(feature = "mssql")]
DeleteExecutor::MSSQL(exec) => exec.execute_with_sql(sql).await,
#[cfg(feature = "duckdb")]
DeleteExecutor::DuckDB(exec) => exec.execute_with_sql(sql).await,
DeleteExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct BlockDeleteResult {
pub blocks_dropped: u64,
pub rows_deleted: Option<u64>,
}
impl BlockDeleteResult {
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(dead_code)
)]
pub(crate) fn from_row_count(rows: u64) -> Self {
Self {
blocks_dropped: 0,
rows_deleted: Some(rows),
}
}
}
pub enum BlockDeleteExecutor<'a, T: Model> {
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::BlockDeleteExecutor<'a, T>),
#[cfg(feature = "clickhouse")]
ClickHouse(clickhouse_backend::BlockDeleteExecutor<'a, T>),
#[cfg(feature = "influxdb")]
InfluxDB(influxdb_backend::BlockDeleteExecutor<'a, T>),
#[cfg(any(
feature = "sqlite",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
))]
Fallback {
db_type: super::super::DbType,
key: Option<crate::abstract_layer::common::common_helpers::BlockKey>,
range: Option<crate::abstract_layer::common::common_helpers::BlockRange>,
delete: DeleteExecutor<'a, T>,
},
}
crate::impl_unified_block_delete_executor!(BlockDeleteExecutor);
impl<'a, T: Model> BlockDeleteExecutor<'a, T> {
#[allow(unused_variables)]
pub fn route_table(
self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
#[allow(unused_variables)]
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
match self {
#[cfg(feature = "postgresql")]
BlockDeleteExecutor::PostgreSQL(exec) => {
BlockDeleteExecutor::PostgreSQL(exec.with_table_route(route))
}
#[cfg(feature = "clickhouse")]
this @ BlockDeleteExecutor::ClickHouse(..) => this,
#[cfg(feature = "influxdb")]
this @ BlockDeleteExecutor::InfluxDB(..) => this,
#[cfg(any(
feature = "sqlite",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
))]
BlockDeleteExecutor::Fallback {
db_type,
key,
range,
delete,
} => BlockDeleteExecutor::Fallback {
db_type,
key,
range,
delete: delete.with_table_route(route),
},
}
}
}
#[cfg(any(
feature = "sqlite",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
))]
impl<'a, T: Model> BlockDeleteExecutor<'a, T> {
pub(crate) fn fallback(
db_type: super::super::DbType,
delete: DeleteExecutor<'a, T>,
) -> Self {
BlockDeleteExecutor::Fallback {
db_type,
key: crate::abstract_layer::common::common_helpers::resolve_block_key::<T>(db_type)
.ok(),
range: None,
delete,
}
}
}
impl<'a, T: Model> super::SqlExecutor for BlockDeleteExecutor<'a, T> {
type Output = BlockDeleteResult;
fn to_sql(&self) -> crate::Result<SqlStatement> {
BlockDeleteExecutor::to_sql(self)
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb",
feature = "clickhouse"
)),
allow(unused_variables)
)]
async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
match self {
#[cfg(feature = "postgresql")]
BlockDeleteExecutor::PostgreSQL(exec) => exec
.execute_with_sql(sql)
.await
.map(BlockDeleteResult::from_row_count),
#[cfg(feature = "clickhouse")]
BlockDeleteExecutor::ClickHouse(exec) => exec.execute_with_sql(sql).await,
#[cfg(feature = "influxdb")]
BlockDeleteExecutor::InfluxDB(_) => Err(unsupported_feature(
super::super::DbType::InfluxDB,
"block delete execute_with_sql (the native backend uses the HTTP delete API)",
)),
#[cfg(any(
feature = "sqlite",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
))]
BlockDeleteExecutor::Fallback { delete, .. } => delete
.execute_with_sql(sql)
.await
.map(BlockDeleteResult::from_row_count),
}
}
async fn execute(self) -> crate::Result<Self::Output> {
BlockDeleteExecutor::execute(self).await
}
}
pub enum UpdateExecutor<'a, T: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::UpdateExecutor<T>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::UpdateExecutor<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::UpdateExecutor<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::UpdateExecutor<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::UpdateExecutor<T>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a T>,
},
}
crate::impl_unified_update_executor!(UpdateExecutor);
impl<'a, T: Model> UpdateExecutor<'a, T> {
#[allow(unused_variables)]
pub fn route_table(
self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
#[allow(unused_variables)]
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
match self {
#[cfg(feature = "sqlite")]
UpdateExecutor::Sqlite(exec, phantom) => {
UpdateExecutor::Sqlite(exec.with_table_route(route), phantom)
}
#[cfg(feature = "postgresql")]
UpdateExecutor::PostgreSQL(exec) => {
UpdateExecutor::PostgreSQL(exec.with_table_route(route))
}
#[cfg(feature = "mysql")]
UpdateExecutor::MySQL(exec) => UpdateExecutor::MySQL(exec.with_table_route(route)),
#[cfg(feature = "mssql")]
UpdateExecutor::MSSQL(exec) => UpdateExecutor::MSSQL(exec.with_table_route(route)),
#[cfg(feature = "duckdb")]
UpdateExecutor::DuckDB(exec) => UpdateExecutor::DuckDB(exec.with_table_route(route)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ UpdateExecutor::Unsupported { .. } => unsupported,
}
}
}
impl<'a, T: Model> UpdateExecutor<'a, T> {
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub(crate) fn set_model_columns(self, model: &T, fields: &[String]) -> Self {
match self {
#[cfg(feature = "sqlite")]
UpdateExecutor::Sqlite(exec, phantom) => {
UpdateExecutor::Sqlite(exec.set_model_fields(model, fields), phantom)
}
#[cfg(feature = "postgresql")]
UpdateExecutor::PostgreSQL(exec) => {
UpdateExecutor::PostgreSQL(exec.set_model_fields(model, fields))
}
#[cfg(feature = "mysql")]
UpdateExecutor::MySQL(exec) => {
UpdateExecutor::MySQL(exec.set_model_fields(model, fields))
}
#[cfg(feature = "mssql")]
UpdateExecutor::MSSQL(exec) => {
UpdateExecutor::MSSQL(exec.set_model_fields(model, fields))
}
#[cfg(feature = "duckdb")]
UpdateExecutor::DuckDB(exec) => {
UpdateExecutor::DuckDB(exec.set_model_fields(model, fields))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ UpdateExecutor::Unsupported { .. } => unsupported,
}
}
}
impl<'a, T: Model> NamedFilterQuery<T> for UpdateExecutor<'a, T> {
fn apply_named_filter(self, _name: &'static str, expr: WhereExpr) -> Self {
self.filter(|_| expr)
}
}
impl<'a, T: Model> super::SqlExecutor for UpdateExecutor<'a, T> {
type Output = u64;
fn to_sql(&self) -> crate::Result<SqlStatement> {
UpdateExecutor::to_sql(self)
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
match self {
#[cfg(feature = "sqlite")]
UpdateExecutor::Sqlite(exec, _) => exec.execute_with_sql(sql).await,
#[cfg(feature = "postgresql")]
UpdateExecutor::PostgreSQL(exec) => exec.execute_with_sql(sql).await,
#[cfg(feature = "mysql")]
UpdateExecutor::MySQL(exec) => exec.execute_with_sql(sql).await,
#[cfg(feature = "mssql")]
UpdateExecutor::MSSQL(exec) => exec.execute_with_sql(sql).await,
#[cfg(feature = "duckdb")]
UpdateExecutor::DuckDB(exec) => exec.execute_with_sql(sql).await,
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
UpdateExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
}
}
}
pub struct ScopedDeleteExecutor<'a, T: Model> {
pub(crate) inner: DeleteExecutor<'a, T>,
pub(crate) context_filters: Vec<ContextFilter>,
pub(crate) disabled_filters: Vec<&'static str>,
}
pub struct ScopedUpdateExecutor<'a, T: Model> {
pub(crate) inner: UpdateExecutor<'a, T>,
pub(crate) context_filters: Vec<ContextFilter>,
pub(crate) disabled_filters: Vec<&'static str>,
}
fn scoped_filter_exprs<T: Model>(
context_filters: &[ContextFilter],
disabled_filters: &[&'static str],
) -> Vec<FilterExpr> {
context_filters
.iter()
.filter(|filter| !disabled_filters.iter().any(|name| *name == filter.name()))
.filter_map(ContextFilter::filter_for::<T>)
.collect()
}
fn append_scoped_filters<T: Model>(
statement: &mut SqlStatement,
context_filters: &[ContextFilter],
disabled_filters: &[&'static str],
) -> crate::Result<()> {
let filters = scoped_filter_exprs::<T>(context_filters, disabled_filters);
if filters.is_empty() {
return Ok(());
}
for single in &mut statement.statements {
let mut scope_sql = String::new();
let mut param_idx = single.params.len() + 1;
for (index, filter) in filters.iter().enumerate() {
if index > 0 {
scope_sql.push_str(" AND ");
}
common_helpers::format_filter_with_params(
filter,
&mut scope_sql,
&mut param_idx,
&mut single.params,
statement.db_type,
)?;
}
match find_toplevel_where(&single.sql) {
Some(insert_at) => {
single.sql.insert(insert_at, '(');
single.sql.push_str(") AND ");
single.sql.push_str(&scope_sql);
}
None => {
single.sql.push_str(" WHERE ");
single.sql.push_str(&scope_sql);
}
}
}
Ok(())
}
fn find_toplevel_where(sql: &str) -> Option<usize> {
let bytes = sql.as_bytes();
let mut depth = 0i32;
let mut in_string = false;
let mut i = 0;
while i < bytes.len() {
let b = bytes[i];
if in_string {
if b == b'\'' {
if i + 1 < bytes.len() && bytes[i + 1] == b'\'' {
i += 2;
continue;
}
in_string = false;
}
} else {
match b {
b'\'' => in_string = true,
b'(' => depth += 1,
b')' => depth -= 1,
b'W' if depth == 0 => {
let is_word_start = i == 0 || bytes[i - 1] == b' ' || bytes[i - 1] == b'\n';
if is_word_start && sql[i..].starts_with("WHERE") {
let mut j = i + "WHERE".len();
while j < bytes.len() && bytes[j] == b' ' {
j += 1;
}
return Some(j);
}
}
_ => {}
}
}
i += 1;
}
None
}
impl<'a, T: Model> ScopedDeleteExecutor<'a, T> {
pub fn route_table(
mut self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
self.inner = self.inner.route_table(key, value);
self
}
pub fn with_table_route(mut self, route: crate::model::TableRoute) -> Self {
self.inner = self.inner.with_table_route(route);
self
}
pub fn filter<F, W>(mut self, f: F) -> Self
where
F: FnOnce(T::Where) -> W,
W: Into<WhereExpr>,
{
self.inner = self.inner.filter(f);
self
}
pub fn model(mut self, model: &T) -> Self {
self.inner = self.inner.model(model);
self
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
let mut statement = self.inner.to_sql()?;
append_scoped_filters::<T>(
&mut statement,
&self.context_filters,
&self.disabled_filters,
)?;
Ok(statement)
}
pub async fn execute(self) -> crate::Result<u64> {
<Self as super::SqlExecutor>::execute(self).await
}
pub fn without_hooks(self) -> crate::WithoutHooksExecutor<Self> {
crate::WithoutHooksExecutor(self)
}
#[deprecated(since = "0.2.11", note = "use `execute()` instead")]
pub async fn exec(self) -> crate::Result<u64> {
self.execute().await
}
}
impl<'a, T: Model> NamedFilterQuery<T> for ScopedDeleteExecutor<'a, T> {
fn apply_named_filter(self, _name: &'static str, expr: WhereExpr) -> Self {
self.filter(|_| expr)
}
}
impl<'a, T: Model> WithoutFilterQuery<T> for ScopedDeleteExecutor<'a, T> {
fn without_filter(mut self, name: &'static str) -> Self {
if !self.disabled_filters.contains(&name) {
self.disabled_filters.push(name);
}
self
}
}
impl<'a, T: Model> super::SqlExecutor for ScopedDeleteExecutor<'a, T> {
type Output = u64;
fn to_sql(&self) -> crate::Result<SqlStatement> {
ScopedDeleteExecutor::to_sql(self)
}
async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
self.inner.execute_with_sql(sql).await
}
}
impl<'a, T: Model> ScopedUpdateExecutor<'a, T> {
pub fn route_table(
mut self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
self.inner = self.inner.route_table(key, value);
self
}
pub fn with_table_route(mut self, route: crate::model::TableRoute) -> Self {
self.inner = self.inner.with_table_route(route);
self
}
pub fn filter<F, W>(mut self, f: F) -> Self
where
F: FnOnce(T::Where) -> W,
W: Into<WhereExpr>,
{
self.inner = self.inner.filter(f);
self
}
pub fn set<F>(mut self, f: F) -> Self
where
F: FnOnce(&mut T::Update),
{
self.inner = self.inner.set(f);
self
}
pub fn set_model<I: crate::model::Insertable<Model = T>>(mut self, models: I) -> Self {
self.inner = self.inner.set_model(models);
self
}
pub fn set_model_fields<I, F, M>(mut self, models: I, fields_fn: F) -> Self
where
I: crate::model::Insertable<Model = T>,
F: FnOnce(T::Where) -> M,
M: crate::query::builder::MapToResult,
{
self.inner = self.inner.set_model_fields(models, fields_fn);
self
}
pub fn to_sql(&self) -> crate::Result<SqlStatement> {
let mut statement = self.inner.to_sql()?;
append_scoped_filters::<T>(
&mut statement,
&self.context_filters,
&self.disabled_filters,
)?;
Ok(statement)
}
pub async fn execute(self) -> crate::Result<u64> {
<Self as super::SqlExecutor>::execute(self).await
}
pub fn without_hooks(self) -> crate::WithoutHooksExecutor<Self> {
crate::WithoutHooksExecutor(self)
}
}
impl<'a, T: Model> NamedFilterQuery<T> for ScopedUpdateExecutor<'a, T> {
fn apply_named_filter(self, _name: &'static str, expr: WhereExpr) -> Self {
self.filter(|_| expr)
}
}
impl<'a, T: Model> WithoutFilterQuery<T> for ScopedUpdateExecutor<'a, T> {
fn without_filter(mut self, name: &'static str) -> Self {
if !self.disabled_filters.contains(&name) {
self.disabled_filters.push(name);
}
self
}
}
impl<'a, T: Model> super::SqlExecutor for ScopedUpdateExecutor<'a, T> {
type Output = u64;
fn to_sql(&self) -> crate::Result<SqlStatement> {
ScopedUpdateExecutor::to_sql(self)
}
async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
self.inner.execute_with_sql(sql).await
}
}
pub enum CollectFuture<'a, T: Model, C: FromIterator<T>> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::CollectFuture<'a, T, C>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::CollectFuture<'a, T, C>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::CollectFuture<'a, T, C>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::CollectFuture<'a, T, C>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::CollectFuture<'a, T, C>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ClickHouse(
ClickHouseSelectBackend<'a>,
crate::query::builder::Select<T>,
std::marker::PhantomData<C>,
),
}
pub enum FirstFuture<'a, T: Model> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::FirstFuture<'a, T>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::FirstFuture<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::FirstFuture<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::FirstFuture<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::FirstFuture<'a, T>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ClickHouse(
ClickHouseSelectBackend<'a>,
crate::query::builder::Select<T>,
),
}
pub enum AggregateFuture<'a, T: Model, R> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::AggregateFuture<T, R>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::AggregateFuture<'a, T, R>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::AggregateFuture<'a, T, R>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::AggregateFuture<'a, T, R>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::AggregateFuture<T, R>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ClickHouse(
ClickHouseSelectBackend<'a>,
crate::query::builder::AggregateSelect<T, R>,
std::marker::PhantomData<&'a R>,
),
}
crate::impl_unified_aggregate_future!(AggregateFuture);
macro_rules! impl_unified_multi_table_select_family {
(
$exec:ident, $future:ident,
$exec_doc:literal, $future_doc:literal, $count_doc:literal,
($($r:ident),+)
$(, duckdb_extra { $($duck_extra:tt)* })?
) => {
#[doc = $exec_doc]
pub enum $exec<'a, T: Model, $($r: Model),+> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::$exec<T, $($r),+>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "duckdb")]
DuckDB(
duckdb_backend::$exec<T, $($r),+>
$(, $($duck_extra)*)?
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, $($r),+)>,
},
}
#[doc = $future_doc]
pub enum $future<'a, T: Model, $($r: Model),+> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::$exec<T, $($r),+>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "duckdb")]
DuckDB(
duckdb_backend::$exec<T, $($r),+>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, $($r),+)>,
},
}
crate::impl_unified_related_count_future!(
$future,
count,
[
'a,
T: crate::Model + 'static + std::marker::Send + std::marker::Sync,
$($r: crate::Model + 'static + std::marker::Send + std::marker::Sync),+
],
['a, T, $($r),+]
);
impl<'a, T: Model + 'static, $($r: Model + 'static),+> $exec<'a, T, $($r),+> {
#[doc = $count_doc]
pub fn count(self) -> $future<'a, T, $($r),+> {
match self {
#[cfg(feature = "sqlite")]
$exec::Sqlite(exec, phantom) => $future::Sqlite(exec, phantom),
#[cfg(feature = "postgresql")]
$exec::PostgreSQL(exec) => $future::PostgreSQL(exec),
#[cfg(feature = "mysql")]
$exec::MySQL(exec) => $future::MySQL(exec),
#[cfg(feature = "mssql")]
$exec::MSSQL(exec) => $future::MSSQL(exec),
#[cfg(feature = "duckdb")]
$exec::DuckDB(exec, ..) => {
$future::DuckDB(exec, std::marker::PhantomData)
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
$exec::Unsupported {
backend,
feature,
..
} => $future::Unsupported {
backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
}
};
}
macro_rules! impl_unified_multi_table_collect {
(
$exec:ident, $collect_future:ident,
$method_doc:literal, $future_doc:literal,
($($r:ident),+)
) => {
impl<'a, T: Model + 'static, $($r: Model + 'static),+> $exec<'a, T, $($r),+> {
#[doc = $method_doc]
pub fn collect(self) -> $collect_future<'a, T, $($r),+> {
match self {
#[cfg(feature = "sqlite")]
$exec::Sqlite(exec, phantom) => {
$collect_future::Sqlite(exec, phantom)
}
#[cfg(feature = "postgresql")]
$exec::PostgreSQL(exec) => $collect_future::PostgreSQL(exec),
#[cfg(feature = "mysql")]
$exec::MySQL(exec) => $collect_future::MySQL(exec),
#[cfg(feature = "mssql")]
$exec::MSSQL(exec) => $collect_future::MSSQL(exec),
#[cfg(feature = "duckdb")]
$exec::DuckDB(exec, ..) => {
$collect_future::DuckDB(exec, std::marker::PhantomData)
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
$exec::Unsupported {
backend,
feature,
..
} => $collect_future::Unsupported {
backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
}
#[doc = $future_doc]
pub enum $collect_future<'a, T: Model, $($r: Model),+> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::$exec<T, $($r),+>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::$exec<'a, T, $($r),+>),
#[cfg(feature = "duckdb")]
DuckDB(
duckdb_backend::$exec<T, $($r),+>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, $($r),+)>,
},
}
impl<
'a,
T: crate::Model + 'static + std::marker::Send + std::marker::Sync,
$($r: crate::Model + 'static + std::marker::Send + std::marker::Sync),+
> std::future::IntoFuture for $collect_future<'a, T, $($r),+>
where
Self: 'a,
{
type Output = crate::Result<Vec<T>>;
type IntoFuture = std::pin::Pin<
Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>,
>;
fn into_future(self) -> Self::IntoFuture {
match self {
#[cfg(feature = "sqlite")]
$collect_future::Sqlite(exec, _) => {
Box::pin(async move { exec.collect_rows().await })
}
#[cfg(feature = "postgresql")]
$collect_future::PostgreSQL(exec) => {
Box::pin(async move { exec.collect_rows().await })
}
#[cfg(feature = "mysql")]
$collect_future::MySQL(exec) => {
Box::pin(async move { exec.collect_rows().await })
}
#[cfg(feature = "mssql")]
$collect_future::MSSQL(exec) => {
Box::pin(async move { exec.collect_rows().await })
}
#[cfg(feature = "duckdb")]
$collect_future::DuckDB(exec, _) => {
Box::pin(async move { exec.collect_rows().await })
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
$collect_future::Unsupported {
backend,
feature,
..
} => Box::pin(async move {
Err(unsupported_feature(backend, feature))
}),
}
}
}
};
}
impl_unified_multi_table_select_family!(
RelatedSelectExecutor,
RelatedCountFuture,
"统一的 RelatedSelectExecutor 枚举",
"统一的关联查询同谓词行数统计 Future(page.md 缺失一:分页 total_count 场景)。\n\n结果为与原子查询同谓词的 `SELECT COUNT(*)`,不受 range()/order_by() 影响。",
"统计同谓词总行数(列表分页 total_count 用):\n生成 `SELECT COUNT(*) FROM (<原子查询>)`,不受 range()/order_by() 影响。\n\n返回 `usize`(统一层把后端的 `i64` 行数强转),与\n[`SelectExecutor::count`] 的返回类型一致。",
(R)
);
macro_rules! impl_unified_multi_table_route {
(
$exec:ident, ($($r:ident),+)
$(, duckdb_extra { $($duck_pat:tt)* } { $($duck_expr:tt)* })?
) => {
impl<'a, T: Model, $($r: Model),+> $exec<'a, T, $($r),+> {
#[allow(unused_variables)]
pub fn route_table(
self,
key: impl Into<String>,
value: impl crate::model::TableRouteValue,
) -> Self {
let mut route = crate::model::TableRoute::new();
route.insert(key, value);
self.with_table_route(route)
}
#[allow(unused_variables)]
pub fn with_table_route(self, route: crate::model::TableRoute) -> Self {
match self {
#[cfg(feature = "sqlite")]
$exec::Sqlite(exec, phantom) => {
$exec::Sqlite(exec.with_table_route(route), phantom)
}
#[cfg(feature = "postgresql")]
$exec::PostgreSQL(exec) => $exec::PostgreSQL(exec.with_table_route(route)),
#[cfg(feature = "mysql")]
$exec::MySQL(exec) => $exec::MySQL(exec.with_table_route(route)),
#[cfg(feature = "mssql")]
$exec::MSSQL(exec) => $exec::MSSQL(exec.with_table_route(route)),
#[cfg(feature = "duckdb")]
$exec::DuckDB(exec $(, $($duck_pat)*)?) => $exec::DuckDB(
exec.with_table_route(route)
$(, $($duck_expr)*)?
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ $exec::Unsupported { .. } => unsupported,
}
}
}
};
}
impl_unified_multi_table_route!(RelatedSelectExecutor, (R));
impl_unified_multi_table_select_family!(
MultiTableSelectExecutor,
MultiTableCountFuture,
"统一的 MultiTableSelectExecutor 枚举",
"统一的三表关联查询同谓词行数统计 Future。",
"统计同谓词总行数(列表分页 total_count 用)。\n\n返回 `usize`(统一层把后端的 `i64` 行数强转),与\n[`SelectExecutor::count`] 的返回类型一致。",
(R1, R2),
duckdb_extra { std::marker::PhantomData<&'a ()> }
);
impl_unified_multi_table_route!(
MultiTableSelectExecutor,
(R1, R2),
duckdb_extra { .. } { std::marker::PhantomData }
);
impl_unified_multi_table_collect!(
MultiTableSelectExecutor,
MultiTableCollectFuture,
"执行查询并收集主表行(L18):多表关联 SQL 只选择主表列,\n返回 `Vec<T>`;关联表仅用于过滤。",
"统一的三表关联查询主表行收集 Future(L18:from3 此前只能 count)。\n\nSQL 只选择主表列,返回 `Vec<T>`。",
(R1, R2)
);
impl_unified_multi_table_select_family!(
FourTableSelectExecutor,
FourTableCountFuture,
"统一的 FourTableSelectExecutor 枚举",
"统一的四表关联查询同谓词行数统计 Future。",
"统计同谓词总行数(列表分页 total_count 用)。\n\n返回 `usize`(统一层把后端的 `i64` 行数强转),与\n[`SelectExecutor::count`] 的返回类型一致。",
(R1, R2, R3),
duckdb_extra { std::marker::PhantomData<&'a ()> }
);
impl_unified_multi_table_route!(
FourTableSelectExecutor,
(R1, R2, R3),
duckdb_extra { .. } { std::marker::PhantomData }
);
impl_unified_multi_table_collect!(
FourTableSelectExecutor,
FourTableCollectFuture,
"执行查询并收集主表行(L18):多表关联 SQL 只选择主表列,\n返回 `Vec<T>`;关联表仅用于过滤。",
"统一的四表关联查询主表行收集 Future(L18:from4 此前只能 count)。\n\nSQL 只选择主表列,返回 `Vec<T>`。",
(R1, R2, R3)
);
pub enum InnerJoinedSelectExecutor<'a, T: Model, J: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::InnerJoinedSelectExecutor<T, J>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::InnerJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::InnerJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::InnerJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::InnerJoinedSelectExecutor<T, J>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, J)>,
},
}
pub enum RightJoinedSelectExecutor<'a, T: Model, J: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::RightJoinedSelectExecutor<T, J>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::RightJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::RightJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::RightJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::RightJoinedSelectExecutor<T, J>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, J)>,
},
}
pub enum LeftJoinedSelectExecutor<'a, T: Model, J: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::LeftJoinedSelectExecutor<T, J>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::LeftJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::LeftJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::LeftJoinedSelectExecutor<'a, T, J>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::LeftJoinedSelectExecutor<T, J>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, J)>,
},
}
pub enum LeftJoinCollectFuture<'a, T: Model, J: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::LeftJoinCollectFuture<T, J>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::LeftJoinCollectFuture<'a, T, J>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::LeftJoinCollectFuture<'a, T, J>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::LeftJoinCollectFuture<'a, T, J>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::LeftJoinCollectFuture<T, J>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, J)>,
},
}
pub enum InnerJoinCollectFuture<'a, T: Model, J: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::InnerJoinCollectFuture<T, J>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::InnerJoinCollectFuture<'a, T, J>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::InnerJoinCollectFuture<'a, T, J>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::InnerJoinCollectFuture<'a, T, J>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::InnerJoinCollectFuture<T, J>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, J)>,
},
}
pub enum RightJoinCollectFuture<'a, T: Model, J: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::RightJoinCollectFuture<T, J>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::RightJoinCollectFuture<'a, T, J>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::RightJoinCollectFuture<'a, T, J>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::RightJoinCollectFuture<'a, T, J>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::RightJoinCollectFuture<T, J>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, J)>,
},
}
crate::impl_unified_collect_future!(CollectFuture);
impl<'a, T: Model + 'static + std::marker::Send + std::marker::Sync> std::future::IntoFuture
for FirstFuture<'a, T>
{
type Output = crate::Result<Option<T>>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
fn into_future(self) -> Self::IntoFuture {
match self {
#[cfg(feature = "sqlite")]
FirstFuture::Sqlite(future) => Box::pin(future.into_future()),
#[cfg(feature = "postgresql")]
FirstFuture::PostgreSQL(future) => Box::pin(future.into_future()),
#[cfg(feature = "mysql")]
FirstFuture::MySQL(future) => Box::pin(future.into_future()),
#[cfg(feature = "mssql")]
FirstFuture::MSSQL(future) => Box::pin(future.into_future()),
#[cfg(feature = "duckdb")]
FirstFuture::DuckDB(future) => Box::pin(future.into_future()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
FirstFuture::ClickHouse(db, select) => {
Box::pin(async move { clickhouse_select_first_on_backend(db, select).await })
}
}
}
}
crate::impl_unified_related_select_executor!(RelatedSelectExecutor);
pub enum RelatedCollectFuture<'a, T: Model, R: Model> {
#[cfg(feature = "sqlite")]
Sqlite(
sqlite_backend::RelatedCollectFuture<T, R>,
std::marker::PhantomData<&'a ()>,
),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::RelatedCollectFuture<'a, T, R>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::RelatedCollectFuture<'a, T, R>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::RelatedCollectFuture<'a, T, R>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::RelatedCollectFuture<T, R>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, R)>,
},
}
crate::impl_unified_related_collect_future!(RelatedCollectFuture);
pub enum Transaction<'a> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::Transaction),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::Transaction<'a>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::Transaction<'a>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::Transaction<'a>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::Transaction),
#[doc(hidden)]
_Phantom(std::convert::Infallible, std::marker::PhantomData<&'a ()>),
}
#[deprecated(
since = "0.2.12",
note = "TransactionSaveExecutor 已合并为 SaveExecutor,请改用 SaveExecutor"
)]
pub type TransactionSaveExecutor<'a, 'tx, T> = SaveExecutor<'a, T>;
async fn save_dirty_columns_and_relations<T>(
txn: &mut Transaction<'_>,
model: &mut Tracked<T>,
) -> crate::Result<u64>
where
T: WritableModel + crate::model::Model + crate::model::GraphWritable,
{
let fields = model.dirty_columns();
let mut affected = 0u64;
if !fields.is_empty() {
affected += txn
.update::<T>()
.set_model_columns(model.as_model(), &fields)
.execute()
.await?;
}
affected += model.sync_graph_relations(txn).await?;
Ok(affected)
}
async fn save_on_pooled_conn<T>(
conn: &connection_pool::PooledConnection<'_>,
model: &mut Tracked<T>,
) -> crate::Result<u64>
where
T: WritableModel + crate::model::Model + crate::model::GraphWritable,
{
let mut tx = conn.begin().await?;
match save_dirty_columns_and_relations(&mut tx, model).await {
Ok(affected) => {
tx.commit().await?;
Ok(affected)
}
Err(err) => {
log_rollback_failure(tx.rollback().await);
Err(err)
}
}
}
impl<'a, T: WritableModel + crate::model::GraphWritable> SaveExecutor<'a, T> {
pub(crate) fn from_pooled<'pool>(
conn: &'a connection_pool::PooledConnection<'pool>,
model: &'a mut Tracked<T>,
) -> Self
where
'pool: 'a,
{
let sql = {
let fields = model.dirty_columns();
if fields.is_empty() {
Ok(SqlStatement::batch(conn.db_type(), Vec::new()))
} else {
conn.update::<T>()
.set_model_columns(model.as_model(), &fields)
.to_sql()
}
};
SaveExecutor {
conn: SaveConn::Pooled {
sql,
run: Box::new(move |model: &'a mut Tracked<T>| -> SaveTxnFuture<'a, T> {
Box::pin(async move {
let result = save_on_pooled_conn(conn, model).await;
(result, model)
})
}),
},
model,
}
}
}
#[cfg(any(feature = "postgresql", feature = "mysql", feature = "mssql"))]
pub(crate) fn isolation_level_sql(isolation: IsolationLevel) -> &'static str {
match isolation {
IsolationLevel::ReadUncommitted => "READ UNCOMMITTED",
IsolationLevel::ReadCommitted => "READ COMMITTED",
IsolationLevel::RepeatableRead => "REPEATABLE READ",
IsolationLevel::Serializable => "SERIALIZABLE",
}
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub(crate) async fn apply_transaction_options(
txn: &mut Transaction<'_>,
options: TransactionOptions,
) -> crate::Result<()> {
match txn.db_type() {
#[cfg(feature = "sqlite")]
super::super::DbType::Sqlite => {
if options.isolation.is_some() || options.read_only {
return Err(unsupported_feature(
super::super::DbType::Sqlite,
"transaction options on SQLite",
));
}
Ok(())
}
#[cfg(feature = "postgresql")]
super::super::DbType::PostgreSQL => {
if let Some(isolation) = options.isolation {
txn.execute_sql(format!(
"SET TRANSACTION ISOLATION LEVEL {}",
isolation_level_sql(isolation)
))
.await?;
}
if options.read_only {
txn.execute_sql("SET TRANSACTION READ ONLY").await?;
}
Ok(())
}
#[cfg(feature = "mssql")]
super::super::DbType::MSSQL => {
if let Some(isolation) = options.isolation {
txn.execute_sql(format!(
"SET TRANSACTION ISOLATION LEVEL {}",
isolation_level_sql(isolation)
))
.await?;
}
if options.read_only {
return Err(unsupported_feature(
super::super::DbType::MSSQL,
"read-only transactions",
));
}
Ok(())
}
#[cfg(feature = "duckdb")]
super::super::DbType::DuckDB => {
if options.isolation.is_some() || options.read_only {
return Err(unsupported_feature(
super::super::DbType::DuckDB,
"transaction options on DuckDB",
));
}
Ok(())
}
#[allow(unreachable_patterns)]
_ => Err(unsupported_feature(txn.db_type(), "transaction options")),
}
}
pub struct TransactionScope<'a, 'tx> {
txn: &'tx Transaction<'a>,
context_filters: Vec<ContextFilter>,
}
impl<'a, 'tx> TransactionScope<'a, 'tx> {
pub fn with_context_filter<T: Model>(mut self, name: &'static str, expr: WhereExpr) -> Self {
self.context_filters
.push(ContextFilter::new::<T>(name, expr));
self
}
pub fn without_filter(mut self, name: &'static str) -> Self {
self.context_filters
.retain(|filter| filter.name() != name);
self
}
pub fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
self.txn
.select::<T>()
.with_context_filters(self.context_filters.clone())
}
pub fn delete<T: WritableModel>(&self) -> ScopedDeleteExecutor<'_, T> {
ScopedDeleteExecutor {
inner: self.txn.delete::<T>(),
context_filters: self.context_filters.clone(),
disabled_filters: Vec::new(),
}
}
pub fn update<T: WritableModel>(&self) -> ScopedUpdateExecutor<'_, T> {
ScopedUpdateExecutor {
inner: self.txn.update::<T>(),
context_filters: self.context_filters.clone(),
disabled_filters: Vec::new(),
}
}
pub async fn find_by_id<T: Model + 'static + Send + Sync>(
&self,
key: impl crate::model::PrimaryKey,
) -> crate::Result<Option<T>> {
find_by_id_with_executor(self.select::<T>(), key).await
}
}
impl<'a> Transaction<'a> {
pub fn scope<'tx>(&'tx self) -> TransactionScope<'a, 'tx> {
TransactionScope {
txn: self,
context_filters: Vec::new(),
}
}
pub fn db_type(&self) -> super::super::DbType {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(_) => super::super::DbType::Sqlite,
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(_) => super::super::DbType::PostgreSQL,
#[cfg(feature = "mysql")]
Transaction::MySQL(_) => super::super::DbType::MySQL,
#[cfg(feature = "mssql")]
Transaction::MSSQL(_) => super::super::DbType::MSSQL,
#[cfg(feature = "duckdb")]
Transaction::DuckDB(_) => super::super::DbType::DuckDB,
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
pub fn select_sql<T>(&mut self, sql: impl IntoRawSql) -> RawSelectExecutor<'_, T> {
RawSelectExecutor::from_conn(ConnRef::Txn(self), sql.into_raw_sql())
}
pub async fn execute_sql(&mut self, sql: impl IntoRawSql) -> crate::Result<u64> {
exec_raw_sql_on(ConnRefMut::Txn(self), sql.into_raw_sql()).await
}
pub async fn savepoint<R, F>(&mut self, f: F) -> crate::Result<R>
where
F: for<'tx> FnOnce(&'tx mut Transaction<'a>) -> TransactionFuture<'tx, R>,
{
Capabilities::ensure(self.db_type(), |caps| caps.savepoints, "savepoints")?;
let name = format!(
"__ormer_savepoint_{}",
SAVEPOINT_COUNTER.fetch_add(1, Ordering::Relaxed)
);
#[cfg(feature = "mssql")]
let is_mssql = matches!(self.db_type(), super::super::DbType::MSSQL);
#[cfg(not(feature = "mssql"))]
let is_mssql = false;
if is_mssql {
self.execute_sql(format!("SAVE TRANSACTION {name}")).await?;
} else {
self.execute_sql(format!("SAVEPOINT {name}")).await?;
}
match f(self).await {
Ok(value) => {
if !is_mssql {
self.execute_sql(format!("RELEASE SAVEPOINT {name}"))
.await?;
}
Ok(value)
}
Err(err) => {
if is_mssql {
log_rollback_failure(
self.execute_sql(format!("ROLLBACK TRANSACTION {name}")).await,
);
} else {
log_rollback_failure(
self.execute_sql(format!("ROLLBACK TO SAVEPOINT {name}")).await,
);
log_rollback_failure(
self.execute_sql(format!("RELEASE SAVEPOINT {name}")).await,
);
}
Err(err)
}
}
}
pub async fn commit(self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => txn.commit().await,
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => txn.commit().await,
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => txn.commit().await,
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => txn.commit().await,
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => txn.commit().await,
Transaction::_Phantom(infallible, _) => match infallible {},
}
}
pub async fn rollback(self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => txn.rollback().await,
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => txn.rollback().await,
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => txn.rollback().await,
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => txn.rollback().await,
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => txn.rollback().await,
Transaction::_Phantom(infallible, _) => match infallible {},
}
}
pub async fn close(self) -> crate::Result<()> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => txn.close().await,
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => txn.close().await,
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => txn.close().await,
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => txn.close().await,
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => txn.close().await,
Transaction::_Phantom(infallible, _) => match infallible {},
}
}
pub async fn find_by_id<T: Model + 'static + std::marker::Send + std::marker::Sync>(
&self,
key: impl crate::model::PrimaryKey,
) -> crate::Result<Option<T>> {
find_by_id_with_executor(self.select::<T>(), key).await
}
pub fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => SelectExecutor::Sqlite(txn.select::<T>()),
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => SelectExecutor::PostgreSQL(txn.select::<T>()),
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => SelectExecutor::MySQL(txn.select::<T>()),
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => SelectExecutor::MSSQL(txn.select::<T>()),
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => SelectExecutor::DuckDB(txn.select::<T>()),
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
pub fn batch<'b, B>(&'b self, batch: B) -> BatchFuture<'b, B>
where
B: BatchQueries<'b>,
{
BatchFuture::new(batch)
}
pub fn batch_many<'b, I, Q>(&'b self, queries: I) -> BatchManyFuture<'b, Q>
where
I: IntoIterator<Item = Q>,
Q: BatchQuery<'b>,
{
BatchManyFuture::new(queries)
}
pub fn select_column<T: Model, V>(&self) -> ProjectionSelectExecutor<'_, T, V> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => {
ProjectionSelectExecutor::Sqlite(txn.select_column::<T, V>())
}
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => {
ProjectionSelectExecutor::PostgreSQL(txn.select_column::<T, V>())
}
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => ProjectionSelectExecutor::MySQL(txn.select_column::<T, V>()),
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => ProjectionSelectExecutor::MSSQL(txn.select_column::<T, V>()),
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => {
ProjectionSelectExecutor::DuckDB(txn.select_column::<T, V>())
}
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
pub fn delete<T: WritableModel>(&self) -> DeleteExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => {
DeleteExecutor::Sqlite(txn.delete::<T>(), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => DeleteExecutor::PostgreSQL(txn.delete::<T>()),
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => DeleteExecutor::MySQL(txn.delete::<T>()),
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => DeleteExecutor::MSSQL(txn.delete::<T>()),
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => DeleteExecutor::DuckDB(txn.delete::<T>()),
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
pub fn update<T: WritableModel>(&self) -> UpdateExecutor<'_, T> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => {
UpdateExecutor::Sqlite(txn.update::<T>(), std::marker::PhantomData)
}
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => UpdateExecutor::PostgreSQL(txn.update::<T>()),
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => UpdateExecutor::MySQL(txn.update::<T>()),
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => UpdateExecutor::MSSQL(txn.update::<T>()),
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => UpdateExecutor::DuckDB(txn.update::<T>()),
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
pub fn save<'op, T: WritableModel + crate::model::GraphWritable>(
&'op mut self,
model: &'op mut Tracked<T>,
) -> SaveExecutor<'op, T> {
let sql = {
let fields = model.dirty_columns();
if fields.is_empty() {
Ok(SqlStatement::batch(self.db_type(), Vec::new()))
} else {
self.update::<T>()
.set_model_columns(model.as_model(), &fields)
.to_sql()
}
};
let txn = self;
SaveExecutor {
conn: SaveConn::Txn {
sql,
run: Box::new(move |model: &'op mut Tracked<T>| -> SaveTxnFuture<'op, T> {
Box::pin(async move {
let result = save_dirty_columns_and_relations(txn, model).await;
(result, model)
})
}),
},
model,
}
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn insert<I: crate::model::Insertable>(&mut self, models: I) -> InsertExecutor<'_, I> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => InsertExecutor::SqliteTxn(txn.insert::<I>(models)),
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => {
InsertExecutor::PostgreSQLTxn(txn.insert::<I>(models))
}
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => InsertExecutor::MySQLTxn(txn.insert::<I>(models)),
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => InsertExecutor::MSSQLTxn(txn.insert::<I>(models)),
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => InsertExecutor::DuckDBTxn(txn.insert::<I>(models)),
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn insert_or_update<I: crate::model::Insertable>(
&mut self,
models: I,
) -> InsertOrUpdateExecutor<'_, I> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => {
InsertOrUpdateExecutor::SqliteTxn(txn.insert_or_update::<I>(models))
}
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => {
InsertOrUpdateExecutor::PostgreSQLTxn(txn.insert_or_update::<I>(models))
}
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => {
InsertOrUpdateExecutor::MySQLTxn(txn.insert_or_update::<I>(models))
}
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => {
InsertOrUpdateExecutor::MSSQLTxn(txn.insert_or_update::<I>(models))
}
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => {
InsertOrUpdateExecutor::DuckDBTxn(txn.insert_or_update::<I>(models))
}
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
pub fn upsert<I: crate::model::Insertable>(
&mut self,
models: I,
) -> InsertOrUpdateExecutor<'_, I> {
self.insert_or_update(models)
}
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn insert_or_ignore<I: crate::model::Insertable>(
&mut self,
models: I,
) -> InsertOrIgnoreExecutor<'_, I> {
match self {
#[cfg(feature = "sqlite")]
Transaction::Sqlite(txn) => {
InsertOrIgnoreExecutor::SqliteTxn(txn.insert_or_ignore::<I>(models))
}
#[cfg(feature = "postgresql")]
Transaction::PostgreSQL(txn) => {
InsertOrIgnoreExecutor::PostgreSQLTxn(txn.insert_or_ignore::<I>(models))
}
#[cfg(feature = "mysql")]
Transaction::MySQL(txn) => {
InsertOrIgnoreExecutor::MySQLTxn(txn.insert_or_ignore::<I>(models))
}
#[cfg(feature = "mssql")]
Transaction::MSSQL(txn) => {
InsertOrIgnoreExecutor::MSSQLTxn(txn.insert_or_ignore::<I>(models))
}
#[cfg(feature = "duckdb")]
Transaction::DuckDB(txn) => {
InsertOrIgnoreExecutor::DuckDBTxn(txn.insert_or_ignore::<I>(models))
}
Transaction::_Phantom(infallible, _) => match *infallible {},
}
}
}
impl<'a> super::DbExecutor for Transaction<'a> {
fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
Transaction::select::<T>(self)
}
fn select_column<T: Model, V>(&self) -> ProjectionSelectExecutor<'_, T, V> {
Transaction::select_column::<T, V>(self)
}
}
crate::impl_unified_join_executor!(LeftJoinedSelectExecutor);
impl<'a, T: Model, J: Model> LeftJoinedSelectExecutor<'a, T, J> {
pub fn collect<C: FromIterator<(T, Option<J>)> + 'static>(
&self,
) -> LeftJoinCollectFuture<'a, T, J>
where
T: 'static,
J: 'static,
{
match self {
#[cfg(feature = "sqlite")]
LeftJoinedSelectExecutor::Sqlite(exec, phantom) => {
LeftJoinCollectFuture::Sqlite(exec.clone().collect::<C>(), *phantom)
}
#[cfg(feature = "postgresql")]
LeftJoinedSelectExecutor::PostgreSQL(exec) => {
LeftJoinCollectFuture::PostgreSQL(exec.clone_with_client().collect::<C>())
}
#[cfg(feature = "mysql")]
LeftJoinedSelectExecutor::MySQL(exec) => {
LeftJoinCollectFuture::MySQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "mssql")]
LeftJoinedSelectExecutor::MSSQL(exec) => {
LeftJoinCollectFuture::MSSQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "duckdb")]
LeftJoinedSelectExecutor::DuckDB(exec) => {
LeftJoinCollectFuture::DuckDB(exec.collect::<C>())
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
LeftJoinedSelectExecutor::Unsupported {
backend, feature, ..
} => LeftJoinCollectFuture::Unsupported {
backend: *backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
}
crate::impl_unified_join_executor!(InnerJoinedSelectExecutor);
impl<'a, T: Model, J: Model> InnerJoinedSelectExecutor<'a, T, J> {
pub fn collect<C: FromIterator<(T, J)> + 'static>(&self) -> InnerJoinCollectFuture<'a, T, J>
where
T: 'static,
J: 'static,
{
match self {
#[cfg(feature = "sqlite")]
InnerJoinedSelectExecutor::Sqlite(exec, phantom) => {
InnerJoinCollectFuture::Sqlite(exec.clone().collect::<C>(), *phantom)
}
#[cfg(feature = "postgresql")]
InnerJoinedSelectExecutor::PostgreSQL(exec) => {
InnerJoinCollectFuture::PostgreSQL(exec.clone_with_client().collect::<C>())
}
#[cfg(feature = "mysql")]
InnerJoinedSelectExecutor::MySQL(exec) => {
InnerJoinCollectFuture::MySQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "mssql")]
InnerJoinedSelectExecutor::MSSQL(exec) => {
InnerJoinCollectFuture::MSSQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "duckdb")]
InnerJoinedSelectExecutor::DuckDB(exec) => {
InnerJoinCollectFuture::DuckDB(exec.collect::<C>())
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
InnerJoinedSelectExecutor::Unsupported {
backend, feature, ..
} => InnerJoinCollectFuture::Unsupported {
backend: *backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
}
crate::impl_unified_join_executor!(RightJoinedSelectExecutor);
impl<'a, T: Model, J: Model> RightJoinedSelectExecutor<'a, T, J> {
pub fn collect<C: FromIterator<(Option<T>, J)> + 'static>(
&self,
) -> RightJoinCollectFuture<'a, T, J>
where
T: 'static,
J: 'static,
{
match self {
#[cfg(feature = "sqlite")]
RightJoinedSelectExecutor::Sqlite(exec, phantom) => {
RightJoinCollectFuture::Sqlite(exec.clone().collect::<C>(), *phantom)
}
#[cfg(feature = "postgresql")]
RightJoinedSelectExecutor::PostgreSQL(exec) => {
RightJoinCollectFuture::PostgreSQL(exec.clone_with_client().collect::<C>())
}
#[cfg(feature = "mysql")]
RightJoinedSelectExecutor::MySQL(exec) => {
RightJoinCollectFuture::MySQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "mssql")]
RightJoinedSelectExecutor::MSSQL(exec) => {
RightJoinCollectFuture::MSSQL(exec.clone_with_pool().collect::<C>())
}
#[cfg(feature = "duckdb")]
RightJoinedSelectExecutor::DuckDB(exec) => {
RightJoinCollectFuture::DuckDB(exec.collect::<C>())
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
RightJoinedSelectExecutor::Unsupported {
backend, feature, ..
} => RightJoinCollectFuture::Unsupported {
backend: *backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
}
crate::impl_unified_join_collect_future!(LeftJoinCollectFuture, crate::Result<Vec<(T, Option<J>)>>);
crate::impl_unified_join_collect_future!(InnerJoinCollectFuture, crate::Result<Vec<(T, J)>>);
crate::impl_unified_join_collect_future!(
RightJoinCollectFuture,
crate::Result<Vec<(Option<T>, J)>>
);
pub enum ProjectionSelectExecutor<'a, T: Model, V> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::ProjectionSelectExecutor<'a, T, V>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::ProjectionSelectExecutor<'a, T, V>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::ProjectionSelectExecutor<'a, T, V>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::ProjectionSelectExecutor<'a, T, V>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::ProjectionSelectExecutor<'a, T, V>),
#[cfg(feature = "clickhouse")]
ClickHouse(&'a clickhouse_backend::Database, ProjectionSelect<T, V>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, V)>,
},
}
#[deprecated(
since = "0.2.12",
note = "MappedSelectExecutor 已合并为 ProjectionSelectExecutor,请改用 ProjectionSelectExecutor"
)]
pub type MappedSelectExecutor<'a, T, V> = ProjectionSelectExecutor<'a, T, V>;
#[deprecated(
since = "0.2.12",
note = "GroupedSelectExecutor 已合并为 ProjectionSelectExecutor,请改用 ProjectionSelectExecutor"
)]
pub type GroupedSelectExecutor<'a, T, V> = ProjectionSelectExecutor<'a, T, V>;
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
impl<'a, T: Model, V> ProjectionSelectExecutor<'a, T, V> {
pub fn group_by<F, G>(self, f: F) -> Self
where
F: FnOnce(<T as Model>::Where) -> G,
G: crate::query::builder::GroupByColumns,
{
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => {
ProjectionSelectExecutor::Sqlite(exec.group_by(f))
}
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => {
ProjectionSelectExecutor::PostgreSQL(exec.group_by(f))
}
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => ProjectionSelectExecutor::MySQL(exec.group_by(f)),
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => ProjectionSelectExecutor::MSSQL(exec.group_by(f)),
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => {
ProjectionSelectExecutor::DuckDB(exec.group_by(f))
}
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(db, select) => {
ProjectionSelectExecutor::ClickHouse(db, select.group_by(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ ProjectionSelectExecutor::Unsupported { .. } => unsupported,
}
}
pub fn having<F, W>(self, f: F) -> Self
where
F: FnOnce(<T as Model>::Where) -> W,
W: Into<crate::query::builder::WhereExpr>,
{
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => {
ProjectionSelectExecutor::Sqlite(exec.having(f))
}
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => {
ProjectionSelectExecutor::PostgreSQL(exec.having(f))
}
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => ProjectionSelectExecutor::MySQL(exec.having(f)),
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => ProjectionSelectExecutor::MSSQL(exec.having(f)),
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => {
ProjectionSelectExecutor::DuckDB(exec.having(f))
}
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(db, select) => {
ProjectionSelectExecutor::ClickHouse(db, select.having(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ ProjectionSelectExecutor::Unsupported { .. } => unsupported,
}
}
pub fn filter<F, W>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> W,
W: Into<crate::query::builder::WhereExpr>,
{
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => {
ProjectionSelectExecutor::Sqlite(exec.filter(f))
}
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => {
ProjectionSelectExecutor::PostgreSQL(exec.filter(f))
}
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => ProjectionSelectExecutor::MySQL(exec.filter(f)),
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => ProjectionSelectExecutor::MSSQL(exec.filter(f)),
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => {
ProjectionSelectExecutor::DuckDB(exec.filter(f))
}
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(db, select) => {
ProjectionSelectExecutor::ClickHouse(db, select.filter(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
unsupported @ ProjectionSelectExecutor::Unsupported { .. } => unsupported,
}
}
pub fn collect<C>(&self) -> ProjectionCollectFuture<'a, T, V, C>
where
T: 'static,
V: crate::model::FromRowValues + 'static,
C: FromIterator<V> + 'static,
{
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => {
ProjectionCollectFuture::Sqlite(exec.collect::<C>())
}
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => {
ProjectionCollectFuture::PostgreSQL(exec.collect::<C>())
}
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => {
ProjectionCollectFuture::MySQL(exec.collect::<C>())
}
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => {
ProjectionCollectFuture::MSSQL(exec.collect::<C>())
}
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => {
ProjectionCollectFuture::DuckDB(exec.collect::<C>())
}
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(db, select) => ProjectionCollectFuture::ClickHouse(
db,
select.clone(),
std::marker::PhantomData,
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ProjectionSelectExecutor::Unsupported {
backend, feature, ..
} => ProjectionCollectFuture::Unsupported {
backend: *backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
pub fn as_model<R: Model>(self) -> crate::Result<DerivedSelect<R>>
where
T: Send + Sync + 'static,
V: Send + Sync + 'static,
{
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => Ok(exec.as_model::<R>()),
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => Ok(exec.as_model::<R>()),
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => Ok(exec.as_model::<R>()),
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => Ok(exec.as_model::<R>()),
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => Ok(exec.as_model::<R>()),
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(_, _) => Err(unsupported_feature(
super::super::DbType::ClickHouse,
"Model select_column on ClickHouse; use select_sql",
)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ProjectionSelectExecutor::Unsupported {
backend, feature, ..
} => Err(unsupported_feature(backend, feature)),
}
}
}
impl<'a, T: Model, V> Clone for ProjectionSelectExecutor<'a, T, V> {
fn clone(&self) -> Self {
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => ProjectionSelectExecutor::Sqlite(exec.clone()),
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => {
ProjectionSelectExecutor::PostgreSQL(exec.clone_with_client())
}
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => {
ProjectionSelectExecutor::MySQL(exec.clone_with_pool())
}
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => {
ProjectionSelectExecutor::MSSQL(exec.clone_with_pool())
}
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => ProjectionSelectExecutor::DuckDB(exec.clone()),
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(db, select) => {
ProjectionSelectExecutor::ClickHouse(db, select.clone())
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ProjectionSelectExecutor::Unsupported {
backend, feature, ..
} => ProjectionSelectExecutor::Unsupported {
backend: *backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
}
pub enum ProjectionCollectFuture<'a, T: Model, V, C: FromIterator<V>> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::ProjectionCollectFuture<'a, T, V, C>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::ProjectionCollectFuture<'a, T, V, C>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::ProjectionCollectFuture<'a, T, V, C>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::ProjectionCollectFuture<'a, T, V, C>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::ProjectionCollectFuture<'a, T, V, C>),
#[cfg(feature = "clickhouse")]
ClickHouse(
&'a clickhouse_backend::Database,
ProjectionSelect<T, V>,
std::marker::PhantomData<C>,
),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, V, C)>,
},
}
#[deprecated(
since = "0.2.12",
note = "MappedCollectFuture 已合并为 ProjectionCollectFuture,请改用 ProjectionCollectFuture"
)]
pub type MappedCollectFuture<'a, T, V, C> = ProjectionCollectFuture<'a, T, V, C>;
#[deprecated(
since = "0.2.12",
note = "GroupedCollectFuture 已合并为 ProjectionCollectFuture,请改用 ProjectionCollectFuture"
)]
pub type GroupedCollectFuture<'a, T, V, C> = ProjectionCollectFuture<'a, T, V, C>;
impl<
'a,
T: Model + 'static + std::marker::Send + std::marker::Sync,
V: crate::model::FromRowValues + 'static + std::marker::Send + std::marker::Sync,
C: FromIterator<V> + 'static,
> std::future::IntoFuture for ProjectionCollectFuture<'a, T, V, C>
{
type Output = crate::Result<C>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
fn into_future(self) -> Self::IntoFuture {
match self {
#[cfg(feature = "sqlite")]
ProjectionCollectFuture::Sqlite(future) => Box::pin(future.into_future()),
#[cfg(feature = "postgresql")]
ProjectionCollectFuture::PostgreSQL(future) => Box::pin(future.into_future()),
#[cfg(feature = "mysql")]
ProjectionCollectFuture::MySQL(future) => Box::pin(future.into_future()),
#[cfg(feature = "mssql")]
ProjectionCollectFuture::MSSQL(future) => Box::pin(future.into_future()),
#[cfg(feature = "duckdb")]
ProjectionCollectFuture::DuckDB(future) => Box::pin(future.into_future()),
#[cfg(feature = "clickhouse")]
ProjectionCollectFuture::ClickHouse(db, select, _) => Box::pin(async move {
let (sql, params) = select.try_to_sql_with_params(super::super::DbType::ClickHouse)?;
let (_, rows) = db
.select_named_values(RawSql::new(sql).with_params(params))
.await?;
rows.iter()
.map(|values| V::from_row_values(values))
.collect()
}),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ProjectionCollectFuture::Unsupported {
backend, feature, ..
} => Box::pin(async move { Err(unsupported_feature(backend, feature)) }),
}
}
}
pub enum ModelCollectWithFuture<'a, T: Model + 'static, V: 'static, C, M, F> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::ModelCollectWithFuture<'a, T, V, C, M, F>),
#[cfg(feature = "postgresql")]
PostgreSQLCollect(
postgresql_backend::ProjectionCollectFuture<'a, T, V, Vec<V>>,
F,
std::marker::PhantomData<&'a (T, C, M)>,
),
#[cfg(feature = "mysql")]
MySQLCollect(
mysql_backend::ProjectionCollectFuture<'a, T, V, Vec<V>>,
F,
std::marker::PhantomData<&'a (T, C, M)>,
),
#[cfg(feature = "mssql")]
MSSQLCollect(
mssql_backend::ProjectionCollectFuture<'a, T, V, Vec<V>>,
F,
std::marker::PhantomData<&'a (T, C, M)>,
),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::ModelCollectWithFuture<'a, T, V, C, M, F>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
#[doc(hidden)]
Unsupported {
backend: super::super::DbType,
feature: &'static str,
_marker: std::marker::PhantomData<&'a (T, V, C, M, F)>,
},
}
impl<'a, T: Model> SelectExecutor<'a, T> {
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn map_to<F, M>(self, f: F) -> ProjectionSelectExecutor<'a, T, M::Output>
where
F: FnOnce(<T as Model>::Where) -> M,
M: crate::query::builder::MapToResult,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => ProjectionSelectExecutor::Sqlite(exec.map_to(f)),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
ProjectionSelectExecutor::PostgreSQL(exec.map_to(f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => ProjectionSelectExecutor::MySQL(exec.map_to(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => ProjectionSelectExecutor::MSSQL(exec.map_to(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => ProjectionSelectExecutor::DuckDB(exec.map_to(f)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => ProjectionSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "select capability on ClickHouse",
_marker: std::marker::PhantomData,
},
}
}
pub fn select_column<F, V>(self, f: F) -> ProjectionSelectExecutor<'a, T, V>
where
F: FnOnce(<T as Model>::Where) -> V,
V: crate::query::builder::SelectColumnResult,
{
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
ProjectionSelectExecutor::Sqlite(exec.select_column(f))
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
ProjectionSelectExecutor::PostgreSQL(exec.select_column(f))
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => ProjectionSelectExecutor::MySQL(exec.select_column(f)),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => ProjectionSelectExecutor::MSSQL(exec.select_column(f)),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
ProjectionSelectExecutor::DuckDB(exec.select_column(f))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => {
let backend = clickhouse_select_backend_db_type(db);
if let Some(feature) = clickhouse_projection_gate(backend) {
return ProjectionSelectExecutor::Unsupported {
backend,
feature,
_marker: std::marker::PhantomData,
};
}
#[cfg(feature = "clickhouse")]
#[cfg_attr(not(feature = "influxdb"), allow(irrefutable_let_patterns))]
if let ClickHouseSelectBackend::ClickHouse(db) = db {
return ProjectionSelectExecutor::ClickHouse(db, select.select_column(f));
}
#[cfg(not(feature = "clickhouse"))]
{
let _ = (select, f);
}
ProjectionSelectExecutor::Unsupported {
backend,
feature: "GROUP BY aggregation",
_marker: std::marker::PhantomData,
}
}
}
}
pub fn map_to_view<V: crate::model::ViewModel>(
self,
) -> ProjectionSelectExecutor<'a, T, V> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => {
ProjectionSelectExecutor::Sqlite(exec.map_to_view::<V>())
}
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => {
ProjectionSelectExecutor::PostgreSQL(exec.map_to_view::<V>())
}
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => {
ProjectionSelectExecutor::MySQL(exec.map_to_view::<V>())
}
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => {
ProjectionSelectExecutor::MSSQL(exec.map_to_view::<V>())
}
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => {
ProjectionSelectExecutor::DuckDB(exec.map_to_view::<V>())
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, _) => ProjectionSelectExecutor::Unsupported {
backend: clickhouse_select_backend_db_type(db),
feature: "view-model projection",
_marker: std::marker::PhantomData,
},
}
}
}
pub struct IncludedSelectExecutor<'a, T: Model, S: RelationSelection<T>> {
select: SelectExecutor<'a, T>,
selection: S,
_marker: std::marker::PhantomData<S>,
}
impl<'a, T: Model, S: RelationSelection<T>> IncludedSelectExecutor<'a, T, S> {
pub fn include<F, S2>(self, f: F) -> DoubleIncludedSelectExecutor<'a, T, S, S2>
where
F: FnOnce(T::Where) -> S2,
S2: RelationSelection<T>,
{
DoubleIncludedSelectExecutor {
select: self.select,
first: self.selection,
second: f(T::Where::default()),
_marker: std::marker::PhantomData,
}
}
pub fn collect<C>(self) -> IncludedCollectFuture<'a, T, S, C>
where
T: 'static,
S: 'static,
S::Target: Clone + 'static,
C: FromIterator<T> + 'static,
{
IncludedCollectFuture {
select: self.select,
selection: self.selection,
_marker: std::marker::PhantomData,
}
}
}
pub struct DoubleIncludedSelectExecutor<
'a,
T: Model,
S1: RelationSelection<T>,
S2: RelationSelection<T>,
> {
select: SelectExecutor<'a, T>,
first: S1,
second: S2,
_marker: std::marker::PhantomData<(S1, S2)>,
}
impl<'a, T, S1, S2> DoubleIncludedSelectExecutor<'a, T, S1, S2>
where
T: Model,
S1: RelationSelection<T>,
S2: RelationSelection<T>,
{
pub fn collect<C>(self) -> DoubleIncludedCollectFuture<'a, T, S1, S2, C>
where
T: 'static,
S1: 'static,
S2: 'static,
S1::Target: Clone + 'static,
S2::Target: Clone + 'static,
C: FromIterator<T> + 'static,
{
DoubleIncludedCollectFuture {
select: self.select,
first: self.first,
second: self.second,
_marker: std::marker::PhantomData,
}
}
}
pub struct IncludedCollectFuture<'a, T: Model, S: RelationSelection<T>, C> {
select: SelectExecutor<'a, T>,
selection: S,
_marker: std::marker::PhantomData<C>,
}
pub struct DoubleIncludedCollectFuture<
'a,
T: Model,
S1: RelationSelection<T>,
S2: RelationSelection<T>,
C,
> {
select: SelectExecutor<'a, T>,
first: S1,
second: S2,
_marker: std::marker::PhantomData<C>,
}
impl<
'a,
T: Model + 'static + std::marker::Send + std::marker::Sync,
S: RelationSelection<T> + RelationNestedLoader<'a, T> + std::marker::Send + std::marker::Sync + 'a,
C: FromIterator<T> + 'static,
> std::future::IntoFuture for IncludedCollectFuture<'a, T, S, C>
where
S::Target: std::marker::Send + std::marker::Sync,
S::Via: std::marker::Send + std::marker::Sync,
{
type Output = crate::Result<C>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move {
let mut owners = self.select.collect::<Vec<T>>().await?;
self.select
.preload_models_with_selection(&mut owners, self.selection)
.await?;
Ok(owners.into_iter().collect())
})
}
}
impl<
'a,
T: Model + 'static + std::marker::Send + std::marker::Sync,
S1: RelationSelection<T> + RelationNestedLoader<'a, T> + std::marker::Send + std::marker::Sync + 'a,
S2: RelationSelection<T> + RelationNestedLoader<'a, T> + std::marker::Send + std::marker::Sync + 'a,
C: FromIterator<T> + 'static,
> std::future::IntoFuture for DoubleIncludedCollectFuture<'a, T, S1, S2, C>
where
S1::Target: std::marker::Send + std::marker::Sync,
S1::Via: std::marker::Send + std::marker::Sync,
S2::Target: std::marker::Send + std::marker::Sync,
S2::Via: std::marker::Send + std::marker::Sync,
{
type Output = crate::Result<C>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move {
let mut owners = self.select.collect::<Vec<T>>().await?;
self.select
.preload_models_with_selection(&mut owners, self.first)
.await?;
self.select
.preload_models_with_selection(&mut owners, self.second)
.await?;
Ok(owners.into_iter().collect())
})
}
}
#[cfg(any(feature = "postgresql", feature = "mysql", feature = "mssql"))]
macro_rules! collect_with_clone_branch {
($exec:ident, $f:ident, [$($variant:ident)::+], $clone:ident) => {{
let future = $exec.$clone().collect::<Vec<_>>();
$($variant)::+(future, $f, std::marker::PhantomData)
}};
}
impl<'a, T: Model, V> ProjectionSelectExecutor<'a, T, V> {
#[cfg_attr(
not(any(
feature = "sqlite",
feature = "postgresql",
feature = "mysql",
feature = "mssql",
feature = "duckdb"
)),
allow(unused_variables)
)]
pub fn collect_with<C, F, M>(self, f: F) -> ModelCollectWithFuture<'a, T, V, C, M, F>
where
T: 'static,
V: crate::model::FromRowValues + 'static,
C: FromIterator<M> + 'static,
F: Fn(V) -> M + Clone + 'static,
M: 'static,
{
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => {
ModelCollectWithFuture::Sqlite(exec.collect_with::<C, F, M>(f))
}
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => collect_with_clone_branch!(
exec,
f,
[ModelCollectWithFuture::PostgreSQLCollect],
clone_with_client
),
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => collect_with_clone_branch!(
exec,
f,
[ModelCollectWithFuture::MySQLCollect],
clone_with_pool
),
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => collect_with_clone_branch!(
exec,
f,
[ModelCollectWithFuture::MSSQLCollect],
clone_with_pool
),
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => {
ModelCollectWithFuture::DuckDB(exec.collect_with::<C, F, M>(f))
}
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(_, _) => ModelCollectWithFuture::Unsupported {
backend: super::super::DbType::ClickHouse,
feature: "collect_with on ClickHouse; use select_sql",
_marker: std::marker::PhantomData,
},
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ProjectionSelectExecutor::Unsupported {
backend, feature, ..
} => ModelCollectWithFuture::Unsupported {
backend,
feature,
_marker: std::marker::PhantomData,
},
}
}
}
impl<'a, T: Model, V> ProjectionSelectExecutor<'a, T, V> {
fn into_in_filter(self, column: String) -> crate::query::filter::FilterExpr {
let in_subquery = |subquery: crate::Result<(String, Vec<crate::model::Value>)>| {
crate::query::filter::FilterExpr::InSubqueryDynamic {
column,
subquery: crate::query::filter::DynamicSubquery::new(move |_| subquery.clone()),
}
};
match self {
#[cfg(feature = "sqlite")]
ProjectionSelectExecutor::Sqlite(exec) => in_subquery(exec.to_subquery_sql()),
#[cfg(feature = "postgresql")]
ProjectionSelectExecutor::PostgreSQL(exec) => in_subquery(exec.to_subquery_sql()),
#[cfg(feature = "mysql")]
ProjectionSelectExecutor::MySQL(exec) => in_subquery(exec.to_subquery_sql()),
#[cfg(feature = "mssql")]
ProjectionSelectExecutor::MSSQL(exec) => in_subquery(exec.to_subquery_sql()),
#[cfg(feature = "duckdb")]
ProjectionSelectExecutor::DuckDB(exec) => in_subquery(exec.to_subquery_sql()),
#[cfg(feature = "clickhouse")]
ProjectionSelectExecutor::ClickHouse(_, _) => in_subquery(Err(
crate::OrmerError::UnsupportedFeature {
backend: super::super::DbType::ClickHouse,
feature: "subquery on ClickHouse; use select_sql",
},
)),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ProjectionSelectExecutor::Unsupported {
backend, feature, ..
} => in_subquery(Err(crate::OrmerError::UnsupportedFeature {
backend,
feature,
})),
}
}
}
impl<'a, T: Model, V: crate::query::builder::ColumnValueType> crate::query::builder::IsInValues<V>
for ProjectionSelectExecutor<'a, T, V>
{
fn to_in_expr(self, column: String) -> crate::query::builder::WhereExpr {
crate::query::builder::WhereExpr::from_filter(self.into_in_filter(column))
}
}
impl<'a, 'b, T: Model, V: crate::query::builder::ColumnValueType>
crate::query::builder::IsInValues<V> for &'b ProjectionSelectExecutor<'a, T, V>
{
fn to_in_expr(self, column: String) -> crate::query::builder::WhereExpr {
crate::query::builder::WhereExpr::from_filter(self.clone().into_in_filter(column))
}
}
#[cfg(any(feature = "postgresql", feature = "mysql", feature = "mssql"))]
async fn collect_with_map<V, M, C, F, Fut>(future: Fut, mapper: F) -> crate::Result<C>
where
Fut: Future<Output = crate::Result<Vec<V>>>,
F: Fn(V) -> M,
C: FromIterator<M>,
{
Ok(future.await?.into_iter().map(mapper).collect())
}
impl<'a, T, V, C, M, F> std::future::IntoFuture for ModelCollectWithFuture<'a, T, V, C, M, F>
where
T: Model + 'static + std::marker::Send + std::marker::Sync,
V: crate::model::FromRowValues + 'static + std::marker::Send + std::marker::Sync,
C: FromIterator<M> + 'static,
M: 'static + std::marker::Send,
F: Fn(V) -> M + Clone + Send + 'static,
{
type Output = crate::Result<C>;
type IntoFuture =
std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;
fn into_future(self) -> Self::IntoFuture {
match self {
#[cfg(feature = "sqlite")]
ModelCollectWithFuture::Sqlite(future) => Box::pin(future.into_future()),
#[cfg(feature = "postgresql")]
ModelCollectWithFuture::PostgreSQLCollect(future, mapper, _) => {
Box::pin(collect_with_map(future.into_future(), mapper))
}
#[cfg(feature = "mysql")]
ModelCollectWithFuture::MySQLCollect(future, mapper, _) => {
Box::pin(collect_with_map(future.into_future(), mapper))
}
#[cfg(feature = "mssql")]
ModelCollectWithFuture::MSSQLCollect(future, mapper, _) => {
Box::pin(collect_with_map(future.into_future(), mapper))
}
#[cfg(feature = "duckdb")]
ModelCollectWithFuture::DuckDB(future) => Box::pin(future.into_future()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ModelCollectWithFuture::Unsupported {
backend, feature, ..
} => Box::pin(async move { Err(unsupported_feature(backend, feature)) }),
}
}
}
impl<'a, T> BatchQuery<'a> for SelectExecutor<'a, T>
where
T: Model + 'static + Send + Sync,
{
type Output = Vec<T>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<T>>().await })
}
}
impl<'a, T> BatchQuery<'a> for FirstFuture<'a, T>
where
T: Model + 'static + Send + Sync,
{
type Output = Option<T>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
std::future::IntoFuture::into_future(self)
}
}
impl<'a, T, R> BatchQuery<'a> for AggregateFuture<'a, T, R>
where
T: Model + 'static + Send,
R: crate::model::FromValue + 'static + Send,
{
type Output = R;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
std::future::IntoFuture::into_future(self)
}
}
impl<'a, R> BatchQuery<'a> for DerivedTableSelectExecutor<'a, R>
where
R: Model + crate::model::FromRowValues + 'static + Send,
{
type Output = Vec<R>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<R>>().await })
}
}
impl<'a, T> BatchQuery<'a> for RawSelectExecutor<'a, T>
where
T: crate::model::FromRowValues + 'static + Send,
{
type Output = Vec<T>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<T>>().await })
}
}
impl<'a, T, R> BatchQuery<'a> for RelatedSelectExecutor<'a, T, R>
where
T: Model + 'static + Send + Sync,
R: Model + 'static + Send + Sync,
{
type Output = Vec<T>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<T>>().await })
}
}
impl<'a, T, J> BatchQuery<'a> for LeftJoinedSelectExecutor<'a, T, J>
where
T: Model + 'static + Send + Sync,
J: Model + 'static + Send + Sync,
{
type Output = Vec<(T, Option<J>)>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<(T, Option<J>)>>().await })
}
}
impl<'a, T, J> BatchQuery<'a> for InnerJoinedSelectExecutor<'a, T, J>
where
T: Model + 'static + Send + Sync,
J: Model + 'static + Send + Sync,
{
type Output = Vec<(T, J)>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<(T, J)>>().await })
}
}
impl<'a, T, J> BatchQuery<'a> for RightJoinedSelectExecutor<'a, T, J>
where
T: Model + 'static + Send + Sync,
J: Model + 'static + Send + Sync,
{
type Output = Vec<(Option<T>, J)>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<(Option<T>, J)>>().await })
}
}
impl<'a, T, V> BatchQuery<'a> for ProjectionSelectExecutor<'a, T, V>
where
T: Model + 'static + Send + Sync,
V: crate::model::FromRowValues + 'static + Send + Sync,
{
type Output = Vec<V>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<V>>().await })
}
}
impl<'a, T, S> BatchQuery<'a> for IncludedSelectExecutor<'a, T, S>
where
T: Model + 'static + Send + Sync,
S: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync + 'a + 'static,
S::Target: Clone + 'static + Send + Sync,
S::Via: Send + Sync,
{
type Output = Vec<T>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<T>>().await })
}
}
impl<'a, T, S1, S2> BatchQuery<'a> for DoubleIncludedSelectExecutor<'a, T, S1, S2>
where
T: Model + 'static + Send + Sync,
S1: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync + 'a + 'static,
S2: RelationSelection<T> + RelationNestedLoader<'a, T> + Send + Sync + 'a + 'static,
S1::Target: Clone + 'static + Send + Sync,
S1::Via: Send + Sync,
S2::Target: Clone + 'static + Send + Sync,
S2::Via: Send + Sync,
{
type Output = Vec<T>;
fn into_batch_future(self) -> BatchQueryFuture<'a, Self::Output> {
Box::pin(async move { self.collect::<Vec<T>>().await })
}
}
pub enum SelectStream<'a, T: Model> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::SelectStream<'a, T>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::SelectStream<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::SelectStream<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::SelectStream<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::SelectStream<'a, T>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ClickHouse(
ClickHouseSelectBackend<'a>,
crate::query::builder::Select<T>,
),
}
impl<'a, T: Model> SelectExecutor<'a, T> {
pub fn stream(self) -> SelectStream<'a, T> {
match self {
#[cfg(feature = "sqlite")]
SelectExecutor::Sqlite(exec) => SelectStream::Sqlite(exec.stream()),
#[cfg(feature = "postgresql")]
SelectExecutor::PostgreSQL(exec) => SelectStream::PostgreSQL(exec.stream()),
#[cfg(feature = "mysql")]
SelectExecutor::MySQL(exec) => SelectStream::MySQL(exec.stream()),
#[cfg(feature = "mssql")]
SelectExecutor::MSSQL(exec) => SelectStream::MSSQL(exec.stream()),
#[cfg(feature = "duckdb")]
SelectExecutor::DuckDB(exec) => SelectStream::DuckDB(exec.stream()),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectExecutor::ClickHouse(db, select) => SelectStream::ClickHouse(db, select),
}
}
}
pub enum SelectStreamIterator<'a, T: Model> {
#[cfg(feature = "sqlite")]
Sqlite(sqlite_backend::SelectStreamIterator<'a, T>),
#[cfg(feature = "postgresql")]
PostgreSQL(postgresql_backend::SelectStreamIterator<'a, T>),
#[cfg(feature = "mysql")]
MySQL(mysql_backend::SelectStreamIterator<'a, T>),
#[cfg(feature = "mssql")]
MSSQL(mssql_backend::SelectStreamIterator<'a, T>),
#[cfg(feature = "duckdb")]
DuckDB(duckdb_backend::SelectStreamIterator<'a, T>),
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
ClickHouse(ClickHouseStreamState<T>, std::marker::PhantomData<&'a ()>),
}
impl<'a, T: Model + 'static> SelectStream<'a, T> {
pub async fn into_iter(self) -> crate::Result<SelectStreamIterator<'a, T>> {
match self {
#[cfg(feature = "sqlite")]
SelectStream::Sqlite(stream) => {
let iter = stream.into_iter().await?;
Ok(SelectStreamIterator::Sqlite(iter))
}
#[cfg(feature = "postgresql")]
SelectStream::PostgreSQL(stream) => {
let iter = stream.into_iter().await?;
Ok(SelectStreamIterator::PostgreSQL(iter))
}
#[cfg(feature = "mysql")]
SelectStream::MySQL(stream) => {
let iter = stream.into_iter().await?;
Ok(SelectStreamIterator::MySQL(iter))
}
#[cfg(feature = "mssql")]
SelectStream::MSSQL(stream) => {
let iter = stream.into_iter().await?;
Ok(SelectStreamIterator::MSSQL(iter))
}
#[cfg(feature = "duckdb")]
SelectStream::DuckDB(stream) => {
let iter = stream.into_iter().await?;
Ok(SelectStreamIterator::DuckDB(iter))
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectStream::ClickHouse(db, select) => {
#[cfg(feature = "clickhouse")]
#[cfg_attr(not(feature = "influxdb"), allow(irrefutable_let_patterns))]
if let ClickHouseSelectBackend::ClickHouse(db) = db {
let (sql, params) =
select.try_to_sql_with_params(super::super::DbType::ClickHouse)?;
let cursor = db.select_json_stream(RawSql::new(sql).with_params(params))?;
return Ok(SelectStreamIterator::ClickHouse(
ClickHouseStreamState::Streaming(ClickHouseJsonStreamDecoder {
cursor,
columns: T::columns(),
buffer: Vec::new(),
finished: false,
_marker: std::marker::PhantomData,
}),
std::marker::PhantomData,
));
}
#[cfg(feature = "influxdb")]
#[allow(irrefutable_let_patterns)]
if let ClickHouseSelectBackend::Influx(db) = db {
let rows = influx_select_models::<T, Vec<T>>(db, select).await?;
return Ok(SelectStreamIterator::ClickHouse(
ClickHouseStreamState::Buffered(rows.into_iter()),
std::marker::PhantomData,
));
}
Err(crate::OrmerError::invalid_operation(
"SelectStream::ClickHouse backend matched neither arm",
))
}
}
}
}
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
pub enum ClickHouseStreamState<T: Model> {
Buffered(std::vec::IntoIter<T>),
#[cfg(feature = "clickhouse")]
Streaming(ClickHouseJsonStreamDecoder<T>),
}
#[cfg(feature = "clickhouse")]
pub struct ClickHouseJsonStreamDecoder<T: Model> {
cursor: clickhouse::query::BytesCursor,
columns: Vec<&'static str>,
buffer: Vec<u8>,
finished: bool,
_marker: std::marker::PhantomData<T>,
}
#[cfg(feature = "clickhouse")]
impl<T: Model> ClickHouseJsonStreamDecoder<T> {
async fn next_row(&mut self) -> Option<crate::Result<T>> {
loop {
if let Some(pos) = self.buffer.iter().position(|&byte| byte == b'\n') {
let line: Vec<u8> = self.buffer.drain(..=pos).collect();
let line = String::from_utf8_lossy(&line[..line.len() - 1]).into_owned();
if line.trim().is_empty() {
continue;
}
let row: serde_json::Value = match serde_json::from_str(&line) {
Ok(row) => row,
Err(error) => {
return Some(Err(crate::ormer_error!(
"Invalid ClickHouse JSONEachRow row: {error}"
)))
}
};
let values = match super::super::clickhouse_backend::named_json_row_values(
&row,
&self.columns,
) {
Ok(values) => values,
Err(error) => return Some(Err(error)),
};
return Some(T::from_row_values(&values));
}
if self.finished {
return None;
}
match self.cursor.next().await {
Ok(Some(chunk)) => self.buffer.extend_from_slice(&chunk),
Ok(None) => {
self.finished = true;
}
Err(error) => {
self.finished = true;
return Some(Err(crate::OrmerError::from_external(
"clickhouse::BytesCursor::next",
error,
)));
}
}
}
}
}
impl<'a, T: Model + 'static> SelectStreamIterator<'a, T> {
pub async fn next(&mut self) -> Option<crate::Result<T>> {
match self {
#[cfg(feature = "sqlite")]
SelectStreamIterator::Sqlite(iter) => iter.next().await,
#[cfg(feature = "postgresql")]
SelectStreamIterator::PostgreSQL(iter) => iter.next().await,
#[cfg(feature = "mysql")]
SelectStreamIterator::MySQL(iter) => iter.next().await,
#[cfg(feature = "mssql")]
SelectStreamIterator::MSSQL(iter) => iter.next().await,
#[cfg(feature = "duckdb")]
SelectStreamIterator::DuckDB(iter) => iter.next().await,
#[cfg(any(feature = "clickhouse", feature = "influxdb"))]
SelectStreamIterator::ClickHouse(state, _) => match state {
ClickHouseStreamState::Buffered(rows) => rows.next().map(Ok),
#[cfg(feature = "clickhouse")]
ClickHouseStreamState::Streaming(decoder) => decoder.next_row().await,
},
}
}
}