use rainier_orm::sea_query::Value;
use rainier_orm::{Dialect, Entity, ExecOutcome, ShardRoute};
use rainier_support::{Error, Result};
use crate::connection::Database;
use crate::row::{ColumnRequest, OwnedRow};
use crate::statement::{routing_key, Prepared};
#[derive(Debug)]
pub struct RawQuery<'a> {
database: &'a Database,
sql: String,
params: Vec<Value>,
route: ShardRoute,
rewrite_placeholders: bool,
}
impl<'a> RawQuery<'a> {
pub fn new(database: &'a Database, sql: impl Into<String>) -> Self {
Self {
database,
sql: sql.into(),
params: Vec::new(),
route: ShardRoute::Global,
rewrite_placeholders: true,
}
}
#[must_use = "this returns the query rather than modifying it in place"]
pub fn bind(mut self, value: impl Into<Value>) -> Self {
self.params.push(value.into());
self
}
#[must_use = "this returns the query rather than modifying it in place"]
pub fn bind_all(mut self, values: impl IntoIterator<Item = impl Into<Value>>) -> Self {
self.params.extend(values.into_iter().map(Into::into));
self
}
#[must_use = "this returns the query rather than modifying it in place"]
pub fn route_by(mut self, key: impl Into<Value>) -> Self {
let key = key.into();
self.route = match routing_key(&key) {
Some(key) => ShardRoute::Key(key),
None => ShardRoute::Global,
};
self
}
#[must_use = "this returns the query rather than modifying it in place"]
pub fn on_shard_key(mut self, key: u64) -> Self {
self.route = ShardRoute::Key(key);
self
}
#[must_use = "this returns the query rather than modifying it in place"]
pub fn raw_placeholders(mut self) -> Self {
self.rewrite_placeholders = false;
self
}
pub fn prepared(self) -> Prepared {
let sql = if self.rewrite_placeholders && self.database.dialect() == Dialect::Postgres {
to_numbered_placeholders(&self.sql)
} else {
self.sql
};
Prepared { sql, params: self.params, route: self.route }
}
pub async fn execute(self) -> Result<ExecOutcome> {
let database = self.database;
database.execute(self.prepared()).await
}
pub async fn fetch(self, columns: Vec<ColumnRequest>) -> Result<Vec<OwnedRow>> {
let database = self.database;
database.fetch(self.prepared(), columns).await
}
pub async fn fetch_all<E: Entity>(self) -> Result<Vec<E>> {
let database = self.database;
database.fetch_all::<E>(self.prepared()).await
}
pub async fn fetch_one<E: Entity>(self) -> Result<Option<E>> {
let database = self.database;
database.fetch_one::<E>(self.prepared()).await
}
pub async fn scalar_i64(self, column: &str) -> Result<Option<i64>> {
let requested = vec![ColumnRequest::new(column, rainier_orm::ColumnType::BigInt)];
let rows = self.fetch(requested).await?;
match rows.first() {
Some(row) => Ok(rainier_orm::Row::get_i64(row, column).map_err(Error::from)?),
None => Ok(None),
}
}
pub async fn scalar_string(self, column: &str) -> Result<Option<String>> {
let requested = vec![ColumnRequest::new(column, rainier_orm::ColumnType::Text)];
let rows = self.fetch(requested).await?;
match rows.first() {
Some(row) => Ok(rainier_orm::Row::get_string(row, column).map_err(Error::from)?),
None => Ok(None),
}
}
pub async fn column(self, column: &str) -> Result<Vec<String>> {
let requested = vec![ColumnRequest::new(column, rainier_orm::ColumnType::Text)];
let rows = self.fetch(requested).await?;
rows.iter()
.map(|row| rainier_orm::Row::get_string(row, column).map_err(Error::from))
.filter_map(Result::transpose)
.collect()
}
}
fn to_numbered_placeholders(sql: &str) -> String {
let mut out = String::with_capacity(sql.len() + 8);
let mut next = 1;
let mut quote: Option<char> = None;
for c in sql.chars() {
match quote {
Some(open) => {
out.push(c);
if c == open {
quote = None;
}
}
None => match c {
'\'' | '"' | '`' => {
quote = Some(c);
out.push(c);
}
'?' => {
out.push('$');
out.push_str(&next.to_string());
next += 1;
}
_ => out.push(c),
},
}
}
out
}
impl Database {
pub fn query(&self, sql: impl Into<String>) -> RawQuery<'_> {
RawQuery::new(self, sql)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::testing::MemoryConnection;
fn database(dialect: Dialect) -> Database {
Database::new(MemoryConnection::new(dialect))
}
#[test]
fn sqlite_and_mysql_keep_question_marks() {
for dialect in [Dialect::Sqlite, Dialect::MySql] {
let prepared = database(dialect)
.query("SELECT * FROM users WHERE team_id = ? AND name = ?")
.bind(7)
.bind("Ada")
.prepared();
assert_eq!(prepared.sql, "SELECT * FROM users WHERE team_id = ? AND name = ?");
assert_eq!(prepared.params.len(), 2);
}
}
#[test]
fn postgres_gets_numbered_placeholders() {
let prepared = database(Dialect::Postgres)
.query("SELECT * FROM users WHERE team_id = ? AND name = ?")
.bind(7)
.bind("Ada")
.prepared();
assert_eq!(prepared.sql, "SELECT * FROM users WHERE team_id = $1 AND name = $2");
}
#[test]
fn a_question_mark_inside_a_literal_is_not_a_placeholder() {
let prepared = database(Dialect::Postgres)
.query("SELECT * FROM posts WHERE title = 'why?' AND author_id = ?")
.bind(1)
.prepared();
assert_eq!(prepared.sql, "SELECT * FROM posts WHERE title = 'why?' AND author_id = $1");
}
#[test]
fn a_question_mark_inside_a_quoted_identifier_is_left_alone() {
let prepared = database(Dialect::Postgres)
.query(r#"SELECT "odd?column" FROM t WHERE id = ?"#)
.prepared();
assert_eq!(prepared.sql, r#"SELECT "odd?column" FROM t WHERE id = $1"#);
}
#[test]
fn raw_placeholders_leaves_postgres_alone() {
let sql = "SELECT * FROM events WHERE payload ? 'user_id' AND id = $1";
let prepared = database(Dialect::Postgres).query(sql).bind(1).raw_placeholders().prepared();
assert_eq!(prepared.sql, sql);
}
#[test]
fn a_query_is_global_until_it_is_routed() {
let prepared = database(Dialect::Sqlite).query("SELECT 1").prepared();
assert_eq!(prepared.route, ShardRoute::Global);
let routed = database(Dialect::Sqlite).query("SELECT 1").route_by(4096u64).prepared();
assert_eq!(routed.route, ShardRoute::Key(4096));
}
#[test]
fn a_string_key_routes_the_same_way_the_orm_routes_it() {
let routed = database(Dialect::Sqlite).query("SELECT 1").route_by("tenant-a").prepared();
assert_eq!(routed.route, ShardRoute::Key(rainier_orm::stable_hash(b"tenant-a")));
}
#[test]
fn a_key_with_no_routing_meaning_stays_global() {
let routed = database(Dialect::Sqlite).query("SELECT 1").route_by(1.5f64).prepared();
assert_eq!(routed.route, ShardRoute::Global);
}
#[test]
fn bind_all_keeps_its_order() {
let prepared = database(Dialect::Sqlite)
.query("SELECT * FROM users WHERE id IN (?, ?, ?)")
.bind_all([1, 2, 3])
.prepared();
assert_eq!(prepared.params.len(), 3);
assert_eq!(prepared.params[0], Value::Int(Some(1)));
assert_eq!(prepared.params[2], Value::Int(Some(3)));
}
}