use rudb_catalog::{Catalog, Entry, QualifiedName, duplicate_check, same_name};
use rudb_common::bounds::End;
use rudb_common::{
Bound as ColumnBound, Clustering, Error, Field, LogicalType, Result, Session, Stat, Value,
Width,
};
use rudb_parse::ast::{self, Ast};
use rudb_parse::{NONE, deparse, parse_ast};
use rudb_plan::{Arm, Expr, ExprRef, Node, Plan, SortKey};
use crate::binder::Binder;
use crate::parameters::Parameters;
#[derive(Debug)]
pub enum Bound {
Query(Plan),
CreateTable(CreateTable),
CreateView(CreateView),
DropTable(DropTable),
Schema(SchemaChange),
Sequence(SequenceChange),
Insert(Insert),
Setting(Setting),
Checkpoint,
Transaction(ast::Transaction),
Explain { plan: Plan, analyze: bool, statistics: bool },
}
#[derive(Debug)]
pub struct Setting {
pub name: String,
pub scope: ast::Scope,
pub value: Option<Value>,
pub pragma: bool,
}
#[derive(Debug)]
pub struct CreateTable {
pub name: QualifiedName,
pub columns: Vec<Field>,
pub source: Option<Plan>,
pub if_not_exists: bool,
pub or_replace: bool,
pub keys: Vec<rudb_catalog::Key>,
pub defaults: Vec<Option<String>>,
pub sequences: Vec<QualifiedName>,
pub checks: Vec<String>,
pub foreign: Vec<rudb_catalog::ForeignKey>,
}
#[derive(Debug)]
pub struct CreateView {
pub name: QualifiedName,
pub sql: String,
pub statement: String,
pub aliases: Vec<String>,
pub if_not_exists: bool,
pub or_replace: bool,
pub columns: Vec<Field>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SchemaChange {
pub catalog: String,
pub name: String,
pub drop: bool,
pub quiet: bool,
pub or_replace: bool,
pub cascade: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SequenceChange {
pub name: Option<QualifiedName>,
pub drop: bool,
pub if_not_exists: bool,
pub or_replace: bool,
pub cascade: bool,
pub options: rudb_common::sequence::Options,
}
#[derive(Debug)]
pub struct DropTable {
pub names: Vec<QualifiedName>,
pub kind: Entry,
}
#[derive(Debug)]
pub struct Insert {
pub name: QualifiedName,
pub source: Plan,
pub write: Write,
pub returning: Option<Box<Plan>>,
pub conflict: Option<Conflict>,
pub checks: Option<Checks>,
}
#[derive(Debug)]
pub struct Checks {
pub plan: Box<Plan>,
pub messages: Vec<String>,
}
#[derive(Debug)]
pub struct Conflict {
pub key: Option<usize>,
pub action: ConflictAction,
}
#[derive(Debug)]
pub enum ConflictAction {
Nothing,
Replace(Vec<usize>),
Update {
columns: Vec<usize>,
plan: Box<Plan>,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Write {
Append,
Update,
Delete,
}
fn returning(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
query: Option<ast::QueryRef>,
) -> Result<Option<Box<Plan>>> {
let Some(query) = query else { return Ok(None) };
let mut binder = Binder::with(catalog, parameters, session);
let (root, _) = binder.bind_query(ast, query)?;
Ok(Some(Box::new(finish(binder, root)?)))
}
pub fn bind_statement(ast: &Ast, catalog: &Catalog) -> Result<Bound> {
bind_statement_with(ast, catalog, &Parameters::new(), &Session::new())
}
pub fn bind_statement_with(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
) -> Result<Bound> {
bind_one(ast, catalog, parameters, session, false)
}
pub fn bind_statement_outlined(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
) -> Result<Bound> {
bind_one(ast, catalog, parameters, session, true)
}
fn bind_one(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
outlined: bool,
) -> Result<Bound> {
let statement = match ast.statements.as_slice() {
[statement] => *statement,
[] => return Err(Error::binder("no statement to bind")),
_ => return Err(Error::not_implemented("a script of more than one statement")),
};
match statement {
ast::Statement::Query(query) => {
let mut binder = Binder::with(catalog, parameters, session);
binder.outlined = outlined;
let (root, _) = binder.bind_query(ast, query)?;
Ok(Bound::Query(finish(binder, root)?))
}
ast::Statement::CreateTable(index) => {
create_table(ast, catalog, parameters, session, index)
}
ast::Statement::CreateView(index) => create_view(ast, catalog, parameters, session, index),
ast::Statement::DropTable(index) => drop_table(ast, catalog, index),
ast::Statement::Schema(index) => {
let written = ast.schema(index);
if written.temporary {
return Err(Error::binder("Temporary schemas are not supported"));
}
let parts: Vec<&str> = ast.name(written.name).collect();
let (catalog, name) = catalog.schema_name(&parts)?;
Ok(Bound::Schema(SchemaChange {
catalog,
name,
drop: written.drop,
quiet: written.quiet,
or_replace: written.or_replace,
cascade: written.cascade,
}))
}
ast::Statement::Sequence(index) => {
let written = ast.sequence(index);
let parts: Vec<&str> = ast.name(written.name).collect();
let name = if written.drop {
match catalog.resolve_sequence(&parts) {
Ok(name) => Some(name),
Err(_) if written.quiet => None,
Err(error) => return Err(error),
}
} else if written.temporary {
Some(catalog.resolve_for_create_temporary(&parts)?)
} else {
Some(catalog.resolve_for_create(&parts)?)
};
Ok(Bound::Sequence(SequenceChange {
name,
drop: written.drop,
if_not_exists: written.quiet,
or_replace: written.or_replace,
cascade: written.cascade,
options: written.options,
}))
}
ast::Statement::Insert(index) => insert(ast, catalog, parameters, session, index),
ast::Statement::Update(index) => change(ast, catalog, parameters, session, index, false),
ast::Statement::Delete(index) => change(ast, catalog, parameters, session, index, true),
ast::Statement::Set(index) | ast::Statement::Reset(index) => {
setting(ast, catalog, parameters, session, index)
}
ast::Statement::Checkpoint => Ok(Bound::Checkpoint),
ast::Statement::Transaction(kind) => Ok(Bound::Transaction(kind)),
ast::Statement::Explain { query, analyze, statistics } => {
let mut binder = Binder::with(catalog, parameters, session);
let (root, _) = binder.bind_query(ast, query)?;
Ok(Bound::Explain { plan: finish(binder, root)?, analyze, statistics })
}
}
}
pub fn bind_statement_sql(sql: &str, catalog: &Catalog) -> Result<Bound> {
let ast = parse_ast(sql)?;
bind_statement(&ast, catalog)
}
fn finish(binder: Binder<'_>, root: rudb_plan::NodeRef) -> Result<Plan> {
let mut plan = binder.into_plan();
plan.set_root(root);
plan.validate()?;
Ok(plan)
}
fn create_table(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
index: ast::CreateTableRef,
) -> Result<Bound> {
let written = ast.create_table(index);
let parts: Vec<&str> = ast.name(written.name).collect();
let name = if written.temporary {
catalog.resolve_for_create_temporary(&parts)?
} else {
catalog.resolve_for_create(&parts)?
};
let defs = ast.column_defs(written.columns);
let (mut columns, source) = if written.query == NONE {
let mut columns = Vec::with_capacity(defs.len());
for def in defs {
let text = ast.string(def.ty);
if text.is_empty() {
return Err(Error::binder(format!(
"Column \"{}\" was declared without a type",
ast.string(def.name)
)));
}
let ty = LogicalType::parse(text)?;
let column = ast.string(def.name);
columns.push(if def.not_null {
Field::required(column, ty)
} else {
Field::new(column, ty)
});
}
(columns, None)
} else {
let mut binder = Binder::with(catalog, parameters, session);
let (root, scope) = binder.bind_query(ast, written.query)?;
if defs.len() > scope.len() {
return Err(Error::binder("Target table has more colum names than query result."));
}
let mut columns = Vec::with_capacity(scope.columns.len());
for (at, column) in scope.columns.iter().enumerate() {
let named = match defs.get(at) {
Some(def) => ast.string(def.name).to_string(),
None => column.name.clone(),
};
columns.push(Field::new(named, column.ty.clone()));
}
if defs.is_empty() {
deduplicate(&mut columns);
}
(columns, Some(finish(binder, root)?))
};
duplicate_check(&columns)?;
let mut defaults = Vec::with_capacity(defs.len());
let mut sequences = Vec::new();
for def in defs {
defaults.push(if def.default == NONE {
None
} else {
let (text, used) = default_text(ast, def.default, catalog, parameters, session)?;
for name in used {
if !sequences.contains(&name) {
sequences.push(name);
}
}
Some(text)
});
}
let mut checks = Vec::new();
for &expr in ast.expr_list(written.checks) {
checks.push(check_text(ast, expr, &columns, catalog, parameters, session)?);
}
let mut keys = Vec::new();
for (at, &names) in ast.name_list(written.keys).iter().enumerate() {
let mut places = Vec::new();
for wanted in ast.name(names) {
let Some(place) = columns.iter().position(|field| same_name(&field.name, wanted))
else {
return Err(Error::catalog(format!(
"table \"{}\" does not have a column named \"{wanted}\"",
name.table
)));
};
places.push(place);
}
let primary = at as u32 == written.primary;
if primary {
for &place in &places {
columns[place].not_null = true;
}
}
keys.push(rudb_catalog::Key { columns: places, primary });
}
let mut foreign = Vec::new();
let lists = ast.name_list(written.foreign).iter();
let tables = ast.name_list(written.foreign_tables).iter();
let referenced = ast.name_list(written.foreign_referenced).iter();
for ((&names, &table), &wanted) in lists.zip(tables).zip(referenced) {
let names: Vec<&str> = ast.name(names).collect();
let parts: Vec<&str> = ast.name(table).collect();
let wanted: Vec<&str> = ast.name(wanted).collect();
let key = (names.as_slice(), parts.as_slice(), wanted.as_slice());
foreign.push(foreign_key(catalog, &name, (&columns, &keys), key)?);
}
Ok(Bound::CreateTable(CreateTable {
name,
columns,
source,
if_not_exists: written.if_not_exists,
or_replace: written.or_replace,
keys,
defaults,
checks,
foreign,
sequences,
}))
}
fn foreign_key(
catalog: &Catalog,
made: &QualifiedName,
(columns, keys): (&[Field], &[rudb_catalog::Key]),
(names, parts, wanted): (&[&str], &[&str], &[&str]),
) -> Result<rudb_catalog::ForeignKey> {
let mut places = Vec::with_capacity(names.len());
for &wanted in names {
let Some(place) = columns.iter().position(|field| same_name(&field.name, wanted)) else {
return Err(Error::binder(format!(
"Failed to create foreign key: referencing column \"{wanted}\" does not exist"
)));
};
places.push(place);
}
let own = parts.last().is_some_and(|last| same_name(last, &made.table))
&& catalog.resolve(parts).map_or(true, |resolved| resolved == *made);
let (table, fields, held): (QualifiedName, Vec<Field>, Vec<rudb_catalog::Key>) = if own {
(made.clone(), columns.to_vec(), keys.to_vec())
} else {
let resolved = catalog.resolve(parts)?;
if catalog.view(&resolved).is_ok() {
return Err(Error::binder("cannot reference a VIEW with a FOREIGN KEY"));
}
let table = catalog.table(&resolved)?;
(resolved, table.columns().to_vec(), table.keys().to_vec())
};
let referenced = if wanted.is_empty() {
let Some(primary) = held.iter().find(|key| key.primary) else {
return Err(Error::binder(format!(
"Failed to create foreign key: there is no primary key for referenced table \"{}\"",
table.table
)));
};
if primary.columns.len() != places.len() {
return Err(Error::parser(
"The number of referencing and referenced columns for foreign keys must be the same",
));
}
primary.columns.clone()
} else {
let mut referenced = Vec::with_capacity(wanted.len());
for &column in wanted {
let Some(place) = fields.iter().position(|field| same_name(&field.name, column)) else {
return Err(Error::binder(format!(
"Failed to create foreign key: referenced table \"{}\" does not have a column \
named \"{column}\"",
table.table
)));
};
referenced.push(place);
}
let mut sorted = referenced.clone();
sorted.sort_unstable();
let matched = held.iter().any(|key| {
let mut columns = key.columns.clone();
columns.sort_unstable();
columns == sorted
});
if !matched && held.is_empty() {
return Err(Error::binder(format!(
"Failed to create foreign key: there is no primary key or unique constraint for \
referenced table \"{}\"",
table.table
)));
}
if !matched {
return Err(Error::binder(format!(
"Failed to create foreign key: referenced table \"{}\" does not have a primary key \
or unique constraint on the columns {}",
table.table,
wanted.join(", ")
)));
}
referenced
};
for (&from, &to) in places.iter().zip(&referenced) {
if columns[from].ty != fields[to].ty {
return Err(Error::binder(format!(
"Failed to create foreign key: incompatible types between column \"{}\" (\"{}\") \
and column \"{}\" (\"{}\")",
fields[to].name, fields[to].ty, columns[from].name, columns[from].ty
)));
}
}
Ok(rudb_catalog::ForeignKey { columns: places, table, referenced })
}
fn check_text(
ast: &Ast,
expr: ast::ExprRef,
columns: &[Field],
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
) -> Result<String> {
if crate::expr::has_aggregate(ast, expr) {
return Err(Error::binder("aggregate functions are not allowed in check constraints"));
}
let mut binder = Binder::with(catalog, parameters, session);
let index = binder.fresh_index();
let mut scope = crate::scope::Scope::empty();
for (at, field) in columns.iter().enumerate() {
scope.push(crate::scope::Visible {
table: String::new(),
name: field.name.clone(),
binding: rudb_plan::ColumnBinding::new(index, at as u32),
ty: field.ty.clone(),
not_null: false,
key: None,
default: None,
qualified: false,
also: None,
});
}
match binder.bind_expr(ast, expr, &scope) {
Err(error) if error.message().starts_with("Referenced column \"") => {
let column = error.message().split('"').nth(1).unwrap_or_default();
Err(Error::binder(format!(
"Table does not contain column \"{column}\" referenced in check constraint!"
)))
}
Err(error) => Err(error),
Ok(_) if !binder.windows.is_empty() => {
Err(Error::binder("window functions are not allowed in check constraints"))
}
Ok(_) => Ok(deparse::expression(ast, expr)),
}
}
fn bind_checks(
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
name: &QualifiedName,
) -> Result<Option<Checks>> {
let table = catalog.table(name)?;
if table.checks().is_empty() {
return Ok(None);
}
let failed: Vec<String> =
table.checks().iter().map(|text| format!("NOT CAST(({text}) AS BOOLEAN)")).collect();
let ast = parse_ast(&format!("SELECT {}", failed.join(", ")))?;
let ast::Statement::Query(query) = ast.statements[0] else {
return Err(Error::internal("a check that is not an expression"));
};
let ast::QueryBody::Select(select) = ast.query(query).body else {
return Err(Error::internal("a check that is not an expression"));
};
let mut binder = Binder::with(catalog, parameters, session);
let (root, scope) =
binder.bind_catalog_table(&ast, name, name.table.clone(), ast::Slice::default())?;
let mut exprs = Vec::with_capacity(failed.len());
let mut names = Vec::with_capacity(failed.len());
for target in ast.target_list(ast.select(select).targets) {
exprs.push(binder.bind_expr(&ast, target.expr, &scope)?);
names.push(binder.plan_mut().intern("failed"));
}
let exprs = binder.plan_mut().add_expr_list(&exprs);
let names = binder.plan_mut().add_name_list(&names);
let index = binder.fresh_index();
let root = binder.plan_mut().add_node(Node::Project { input: root, index, exprs, names });
let messages = table
.checks()
.iter()
.map(|text| {
format!(
"CHECK constraint failed on table \"{}\" with expression CHECK({text})",
name.table
)
})
.collect();
Ok(Some(Checks { plan: Box::new(finish(binder, root)?), messages }))
}
fn default_text(
ast: &Ast,
expr: ast::ExprRef,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
) -> Result<(String, Vec<QualifiedName>)> {
if crate::expr::has_aggregate(ast, expr) {
return Err(Error::binder("DEFAULT value cannot contain aggregates!"));
}
let mut binder = Binder::with(catalog, parameters, session);
let before = binder.plan_mut().node_count();
match binder.bind_expr(ast, expr, &crate::scope::Scope::empty()) {
Err(error) if error.message().starts_with("Referenced ") => {
Err(Error::binder("DEFAULT value cannot contain column names"))
}
Err(error) => Err(error),
Ok(_) if binder.plan_mut().node_count() > before => {
Err(Error::binder("DEFAULT value cannot contain subqueries"))
}
Ok(_) if !binder.windows.is_empty() => {
Err(Error::binder("DEFAULT value cannot contain window functions!"))
}
Ok(_) => Ok((deparse::expression(ast, expr), binder.sequences)),
}
}
fn deduplicate(columns: &mut [Field]) {
for at in 0..columns.len() {
let taken = |name: &str, upto: usize, columns: &[Field]| {
columns[..upto].iter().any(|held| same_name(&held.name, name))
};
if !taken(&columns[at].name, at, columns) {
continue;
}
let mut suffix = 1;
let mut candidate = format!("{}_{suffix}", columns[at].name);
while taken(&candidate, at, columns) {
suffix += 1;
candidate = format!("{}_{suffix}", columns[at].name);
}
columns[at].name = candidate;
}
}
fn create_view(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
index: ast::CreateViewRef,
) -> Result<Bound> {
let written = ast.create_view(index);
let parts: Vec<&str> = ast.name(written.name).collect();
let name = if written.temporary {
catalog.resolve_for_create_temporary(&parts)?
} else {
catalog.resolve_for_create(&parts)?
};
let aliases: Vec<String> = ast.name(written.columns).map(str::to_string).collect();
let mut binder = Binder::with(catalog, parameters, session);
binder.outlined = true;
let (_, mut scope) = binder.bind_query(ast, written.query)?;
if aliases.len() > scope.len() {
return Err(Error::binder("More VIEW aliases than columns in query result"));
}
if !aliases.is_empty() {
let written: Vec<&str> = aliases.iter().map(String::as_str).collect();
scope.rename(&written, "unnamed_subquery")?;
}
Ok(Bound::CreateView(CreateView {
name,
sql: ast.string(written.sql).to_string(),
statement: deparse::create_view(ast, index),
aliases,
if_not_exists: written.if_not_exists,
or_replace: written.or_replace,
columns: scope.fields(),
}))
}
fn drop_table(ast: &Ast, catalog: &Catalog, index: ast::DropTableRef) -> Result<Bound> {
let written = ast.drop_table(index);
let kind = if written.view { Entry::View } else { Entry::Table };
let mut names = Vec::new();
for &name in ast.name_list(written.names) {
let parts: Vec<&str> = ast.name(name).collect();
match catalog.resolve_as(&parts, kind) {
Ok(resolved) => names.push(resolved),
Err(error) if written.if_exists => drop(error),
Err(error) => return Err(error),
}
}
Ok(Bound::DropTable(DropTable { names, kind }))
}
fn setting(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
index: ast::SettingRef,
) -> Result<Bound> {
let written = ast.setting(index);
let name = ast.string(written.name).to_string();
let value = if written.value == NONE {
None
} else {
let mut binder = Binder::with(catalog, parameters, session);
let bound = binder.bind_setting_value(ast, written.value)?;
let Expr::Constant(value) = *binder.plan().expr(bound) else {
return Err(Error::not_implemented(format!(
"a value for {name} that is not a constant"
)));
};
Some(binder.plan().value(value).clone())
};
Ok(Bound::Setting(Setting { name, scope: written.scope, value, pragma: written.pragma }))
}
fn clustered(
binder: &mut Binder<'_>,
input: rudb_plan::NodeRef,
scope: &crate::scope::Scope,
clustering: &Clustering,
targets: &[usize],
fields: &[Field],
) -> Result<rudb_plan::NodeRef> {
let mut keys: Vec<SortKey> = Vec::with_capacity(clustering.columns().len());
for (at, &column) in clustering.columns().iter().enumerate() {
let Some(from) = targets.iter().position(|&target| target == column as usize) else {
continue;
};
let source = &scope.columns[from];
let expr = binder.plan_mut().add_expr(Expr::Column(source.binding), source.ty.clone());
let expr = binder.checked_cast_to(expr, &fields[column as usize].ty, false)?;
let expr =
if at == 0 { bucketed(binder, expr, clustering.width(), fields, column) } else { expr };
keys.push(SortKey { expr, descending: false, nulls_first: false });
}
if keys.is_empty() {
return Ok(input);
}
let keys = binder.plan_mut().add_sort_keys(&keys);
Ok(binder.plan_mut().add_node(Node::Sort { input, keys }))
}
fn fitted(
binder: &Binder<'_>,
scope: &crate::scope::Scope,
clustering: &Clustering,
targets: &[usize],
) -> Clustering {
if clustering.width() != Width::Auto {
return clustering.clone();
}
let Some(from) = targets.iter().position(|&target| target == clustering.partition() as usize)
else {
return clustering.fitted(0, 0);
};
let source = &scope.columns[from];
let Some(zones) = binder.plan().sole_zones() else {
return clustering.fitted(0, 0);
};
let Some(at) = zones.column(&source.name) else {
return clustering.fitted(0, 0);
};
let rows = zones.surviving(&[]).unwrap_or(0);
let days = span(&zones.extreme(at, End::Low), &zones.extreme(at, End::High)).unwrap_or(0);
clustering.fitted(rows, days)
}
fn span(low: &Stat<ColumnBound>, high: &Stat<ColumnBound>) -> Option<u64> {
let (Stat::Known { value: low, .. }, Stat::Known { value: high, .. }) = (low, high) else {
return None;
};
let days = match (low, high) {
(ColumnBound::Int(low), ColumnBound::Int(high)) => high.checked_sub(*low)?,
(
ColumnBound::Scaled { unscaled: low, scale: at },
ColumnBound::Scaled { unscaled: high, scale: to },
) if at == to => {
let day = 86_400_i128.checked_mul(10_i128.checked_pow(u32::from(*at))?)?;
high.checked_sub(*low)? / day
}
_ => return None,
};
u64::try_from(days).ok()
}
fn bucketed(
binder: &mut Binder<'_>,
expr: ExprRef,
width: Width,
fields: &[Field],
column: u32,
) -> ExprRef {
if width == Width::Exact {
return expr;
}
let unit = binder.plan_mut().add_value(Value::Varchar(width.to_string().to_lowercase()));
let unit = binder.plan_mut().add_expr(Expr::Constant(unit), LogicalType::Varchar);
let args = binder.plan_mut().add_expr_list(&[unit, expr]);
let name = binder.plan_mut().intern("date_trunc");
let ty = fields[column as usize].ty.clone();
binder.plan_mut().add_expr(Expr::Function { name, args }, ty)
}
fn insert(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
index: ast::InsertRef,
) -> Result<Bound> {
let written = ast.insert(index);
let parts: Vec<&str> = ast.name(written.name).collect();
let name = catalog.resolve(&parts)?;
if catalog.entry(&name)? == Entry::View {
return Err(Error::catalog(format!("{} is not an table", name.table)));
}
let target = catalog.table(&name)?;
let fields: Vec<Field> = target.columns().to_vec();
let clustering = target.clustering().cloned();
let targets: Vec<usize> = if written.columns.is_empty() {
(0..fields.len()).collect()
} else {
let mut targets = Vec::new();
for column in ast.name(written.columns) {
let at = fields.iter().position(|field| same_name(&field.name, column)).ok_or_else(
|| {
Error::binder(format!(
"Table \"{}\" does not have a column named \"{column}\"",
name.table
))
},
)?;
if targets.contains(&at) {
return Err(Error::binder(format!(
"Column \"{column}\" is named twice in the same INSERT"
)));
}
targets.push(at);
}
targets
};
let defaults: Vec<(LogicalType, Option<String>)> = (0..fields.len())
.map(|at| (fields[at].ty.clone(), target.default(at).map(str::to_owned)))
.collect();
let mut binder = Binder::with(catalog, parameters, session);
let (root, scope) = if written.source == NONE {
(binder.plan_mut().add_node(Node::Dummy), crate::scope::Scope::empty())
} else {
if matches!(ast.query(written.source).body, ast::QueryBody::Values(_)) {
binder.insert_defaults = Some(targets.iter().map(|&at| defaults[at].clone()).collect());
}
binder.bind_query(ast, written.source)?
};
let targets = if written.source == NONE { Vec::new() } else { targets };
if scope.len() != targets.len() {
return Err(Error::binder(format!(
"Table \"{}\" has {} columns but {} values were supplied",
name.table,
targets.len(),
scope.len()
)));
}
let root = match &clustering {
None => root,
Some(clustering) => {
let fitted = fitted(&binder, &scope, clustering, &targets);
clustered(&mut binder, root, &scope, &fitted, &targets, &fields)?
}
};
let mut exprs: Vec<ExprRef> = Vec::with_capacity(fields.len());
let mut names = Vec::with_capacity(fields.len());
for (at, field) in fields.iter().enumerate() {
let expr = match targets.iter().position(|&target| target == at) {
Some(from) => {
let column = &scope.columns[from];
let expr =
binder.plan_mut().add_expr(Expr::Column(column.binding), column.ty.clone());
binder.checked_cast_to(expr, &field.ty, false)?
}
None => binder.bind_default(defaults[at].1.as_deref(), &field.ty)?,
};
exprs.push(expr);
let interned = binder.plan_mut().intern(&field.name);
names.push(interned);
}
let exprs = binder.plan_mut().add_expr_list(&exprs);
let names = binder.plan_mut().add_name_list(&names);
let index = binder.fresh_index();
let root = binder.plan_mut().add_node(Node::Project { input: root, index, exprs, names });
let source = finish(binder, root)?;
let returning = returning(ast, catalog, parameters, session, written.returning)?;
let conflict = match written.conflict {
Some(conflict) => {
Some(bind_conflict(ast, catalog, parameters, session, &name, &targets, conflict)?)
}
None => None,
};
let checks = bind_checks(catalog, parameters, session, &name)?;
Ok(Bound::Insert(Insert { name, source, write: Write::Append, returning, conflict, checks }))
}
fn bind_conflict(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
name: &QualifiedName,
targets: &[usize],
conflict: ast::Conflict,
) -> Result<Conflict> {
let table = catalog.table(name)?;
let fields = table.columns();
let keys = table.keys();
let key = if conflict.target.is_empty() {
if keys.is_empty() {
return Err(Error::binder(
"There are no UNIQUE/PRIMARY KEY constraints that refer to this table, specify ON \
CONFLICT columns manually",
));
}
match conflict.action {
ast::ConflictAction::Nothing => None,
_ if keys.len() > 1 => {
return Err(Error::binder(
"Conflict target has to be provided for a DO UPDATE operation when the table \
has multiple UNIQUE/PRIMARY KEY constraints",
));
}
_ => Some(0),
}
} else {
let mut wanted = Vec::new();
for column in ast.name(conflict.target) {
let Some(at) = fields.iter().position(|field| same_name(&field.name, column)) else {
return Err(Error::binder(format!(
"Table \"{}\" does not have a column with name \"{column}\"",
name.table
)));
};
wanted.push(at);
}
wanted.sort_unstable();
wanted.dedup();
let found = keys.iter().position(|key| {
let mut held = key.columns.clone();
held.sort_unstable();
held == wanted
});
let Some(found) = found else {
return Err(Error::binder(
"The specified columns as conflict target are not referenced by a UNIQUE/PRIMARY \
KEY CONSTRAINT or INDEX",
));
};
Some(found)
};
let action = match conflict.action {
ast::ConflictAction::Nothing => ConflictAction::Nothing,
ast::ConflictAction::Replace => ConflictAction::Replace(targets.to_vec()),
ast::ConflictAction::Update { columns: written, query } => {
let mut columns = Vec::new();
for column in ast.name(written) {
let Some(at) = fields.iter().position(|field| same_name(&field.name, column))
else {
return Err(Error::binder(format!(
"Referenced update column {column} not found in table!"
)));
};
if columns.contains(&at) {
return Err(Error::binder(format!(
"Multiple assignments to same column \"\"{column}\"\""
)));
}
columns.push(at);
}
let mut binder = Binder::with(catalog, parameters, session);
binder.upsert = true;
let (root, scope) = binder.bind_query(ast, query)?;
let mut exprs = Vec::with_capacity(scope.columns.len());
let mut names = Vec::with_capacity(scope.columns.len());
for (at, column) in scope.columns.iter().enumerate() {
let expr =
binder.plan_mut().add_expr(Expr::Column(column.binding), column.ty.clone());
let ty = columns.get(at).map_or(LogicalType::Boolean, |&to| fields[to].ty.clone());
exprs.push(binder.checked_cast_to(expr, &ty, false)?);
names.push(binder.plan_mut().intern(&column.name));
}
let exprs = binder.plan_mut().add_expr_list(&exprs);
let names = binder.plan_mut().add_name_list(&names);
let index = binder.fresh_index();
let root =
binder.plan_mut().add_node(Node::Project { input: root, index, exprs, names });
ConflictAction::Update { columns, plan: Box::new(finish(binder, root)?) }
}
};
Ok(Conflict { key, action })
}
fn change(
ast: &Ast,
catalog: &Catalog,
parameters: &Parameters,
session: &Session,
index: ast::InsertRef,
delete: bool,
) -> Result<Bound> {
let written = ast.insert(index);
let parts: Vec<&str> = ast.name(written.name).collect();
let name = catalog.resolve(&parts)?;
if catalog.entry(&name)? == Entry::View {
return Err(Error::binder(if delete {
"Can only delete from base table"
} else {
"Can only update base table"
}));
}
let fields: Vec<Field> = catalog.table(&name)?.columns().to_vec();
let mut targets: Vec<usize> = Vec::new();
for column in ast.name(written.columns) {
let at =
fields.iter().position(|field| same_name(&field.name, column)).ok_or_else(|| {
Error::binder(format!("Referenced update column {column} not found in table!"))
})?;
if targets.contains(&at) {
return Err(Error::binder(format!(
"Multiple assignments to same column \"\"{column}\"\""
)));
}
targets.push(at);
}
let mut defaulted = vec![false; targets.len()];
if let ast::QueryBody::Select(select) = ast.query(written.source).body {
let items = ast.target_list(ast.select(select).targets);
let first = items.len().saturating_sub(targets.len());
for (at, item) in items[first..].iter().enumerate() {
defaulted[at] = matches!(ast.expr(item.expr), ast::Expr::Default);
}
}
let table = catalog.table(&name)?;
let mut binder = Binder::with(catalog, parameters, session);
binder.default_as_null = defaulted.contains(&true);
let (root, scope) = binder.bind_query(ast, written.source)?;
binder.default_as_null = false;
let width = fields.len();
if scope.len() != width + 1 + targets.len() {
return Err(Error::internal(format!(
"an UPDATE source of {} columns over a table of {width}",
scope.len()
)));
}
let column = |binder: &mut Binder<'_>, at: usize| {
let column = &scope.columns[at];
binder.plan_mut().add_expr(Expr::Column(column.binding), column.ty.clone())
};
let hit = column(&mut binder, width);
let hit = binder.checked_cast_to(hit, &LogicalType::Boolean, false)?;
let mut exprs = Vec::with_capacity(width);
let mut names = Vec::with_capacity(width);
for (at, field) in fields.iter().enumerate() {
let old = column(&mut binder, at);
let expr = match targets.iter().position(|&target| target == at) {
Some(from) => {
let then = if defaulted[from] {
binder.bind_default(table.default(at), &field.ty)?
} else {
let new = column(&mut binder, width + 1 + from);
binder.checked_cast_to(new, &field.ty, false)?
};
let arms = binder.plan_mut().add_arms(&[Arm { when: hit, then }]);
binder
.plan_mut()
.add_expr(Expr::Case { arms, otherwise: Some(old) }, field.ty.clone())
}
None => old,
};
exprs.push(expr);
let interned = binder.plan_mut().intern(&field.name);
names.push(interned);
}
let yes = binder.add_constant(Value::Boolean(true));
let arms = binder.plan_mut().add_arms(&[Arm { when: hit, then: yes }]);
let otherwise = Some(binder.add_constant(Value::Boolean(false)));
exprs.push(binder.plan_mut().add_expr(Expr::Case { arms, otherwise }, LogicalType::Boolean));
let interned = binder.plan_mut().intern("changed");
names.push(interned);
let exprs = binder.plan_mut().add_expr_list(&exprs);
let names = binder.plan_mut().add_name_list(&names);
let index = binder.fresh_index();
let root = binder.plan_mut().add_node(Node::Project { input: root, index, exprs, names });
let source = finish(binder, root)?;
let returning = returning(ast, catalog, parameters, session, written.returning)?;
let write = if delete { Write::Delete } else { Write::Update };
let checks = if delete { None } else { bind_checks(catalog, parameters, session, &name)? };
Ok(Bound::Insert(Insert { name, source, write, returning, conflict: None, checks }))
}