#![allow(clippy::collapsible_if, clippy::never_loop)]
mod ddl;
use crate::bound_statement::*;
use akar_catalog::{Catalog, CatalogColumn, CatalogResult, IndexType};
use akar_common::error::BinderError;
use akar_common::types::LogicalTypeID;
use akar_parser::ast::{Clause, Expression, Statement, *};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
fn resolve_set_items(catalog: &Catalog, items: &[SetItem]) -> Result<Vec<BoundSetItem>, BinderError> {
let mut result = Vec::new();
for item in items {
match &item.property {
Expression::PropertyAccess(obj, prop_name) => {
let _var_name = match obj.as_ref() {
Expression::Variable(v) => v.clone(),
other => return Err(format!("Unsupported SET target: {:?}", other).into()),
};
let found = catalog.all_entries().find_map(|entry| {
entry.columns().iter().find(|c| c.name == *prop_name).map(|_| {
let is_node = entry.is_node_table();
(entry.name().to_string(), entry.table_id(), is_node)
})
});
match found {
Some((table_name, table_id, is_node)) => {
let col_idx = catalog
.get_entry_by_name(&table_name)
.and_then(|e| e.columns().iter().position(|c| c.name == *prop_name))
.unwrap_or(0);
result.push(BoundSetItem {
property: item.property.clone(),
value: item.value.clone(),
column_name: prop_name.clone(),
column_idx: col_idx,
table_name: table_name.to_string(),
table_id,
is_node,
});
}
None => {
return Err(format!("Property '{}' not found in any table", prop_name).into());
}
}
}
_ => return Err(format!("Expected property assignment in SET, got: {:?}", item.property).into()),
}
}
Ok(result)
}
pub struct Binder {
catalog: Arc<Mutex<Catalog>>,
}
impl Binder {
pub fn new(catalog: Arc<Mutex<Catalog>>) -> Self {
Self { catalog }
}
pub fn bind(&self, statement: Statement) -> Result<BoundStatement, BinderError> {
match statement {
Statement::Query(query) => self.bind_query(query),
Statement::CreateNodeTable(t) => self.bind_create_node_table(t),
Statement::CreateRelTable(t) => self.bind_create_rel_table(t),
Statement::DropTable(t) => self.bind_drop_table(t),
Statement::CopyFrom(c) => self.bind_copy_from(c),
Statement::CopyTo(c) => self.bind_copy_to(c),
Statement::AlterTable(a) => self.bind_alter_table(a),
Statement::CreateVectorIndex(v) => self.bind_create_vector_index(v),
Statement::CreateIndex(v) => self.bind_create_index(v),
Statement::DropIndex(v) => self.bind_drop_index(v),
Statement::Union(u) => self.bind_union(u),
Statement::Merge(m) => self.bind_merge(m),
Statement::StandaloneCall(c) => self.bind_standalone_call(c),
Statement::CreateDml(c) => self.bind_create_dml(c, &[]),
Statement::Explain(e) => self.bind_explain(e),
Statement::CreateSequence(s) => self.bind_create_sequence(s),
Statement::DropSequence(s) => self.bind_drop_sequence(s),
Statement::CreateMacro(m) => self.bind_create_macro(m),
Statement::ExportDatabase(e) => self.bind_export_database(e),
Statement::ImportDatabase(i) => self.bind_import_database(i),
Statement::Analyze(a) => self.bind_analyze(a),
Statement::CreateFtsIndex(f) => self.bind_create_fts_index(f),
Statement::Transaction(t) => self.bind_transaction(t),
Statement::Extension(e) => self.bind_extension(e),
Statement::AttachDatabase(a) => self.bind_attach_database(a),
Statement::DetachDatabase(d) => self.bind_detach_database(d),
Statement::UseDatabase(u) => self.bind_use_database(u),
Statement::LoadFrom(l) => self.bind_load_from(l),
Statement::CreateType(t) => self.bind_create_type(t),
Statement::CommentOnTable(c) => self.bind_comment_on_table(c),
Statement::CreateGraph(g) => self.bind_create_graph(g),
Statement::UseGraph(g) => self.bind_use_graph(g),
Statement::DropGraph(g) => self.bind_drop_graph(g),
}
}
pub fn parse_type(type_name: &str) -> Result<LogicalTypeID, BinderError> {
let upper = type_name.to_uppercase();
if upper.contains('[') && upper.ends_with(']') {
return Ok(LogicalTypeID::List);
}
if upper.starts_with("MAP(") {
return Ok(LogicalTypeID::Map);
}
if upper.starts_with("STRUCT(") {
return Ok(LogicalTypeID::Struct);
}
if upper.starts_with("UNION(") {
return Ok(LogicalTypeID::Union);
}
match upper.as_str() {
"BOOL" | "BOOLEAN" => Ok(LogicalTypeID::Bool),
"INT64" => Ok(LogicalTypeID::Int64),
"INT32" => Ok(LogicalTypeID::Int32),
"INT16" => Ok(LogicalTypeID::Int16),
"INT8" => Ok(LogicalTypeID::Int8),
"UINT64" => Ok(LogicalTypeID::UInt64),
"UINT32" => Ok(LogicalTypeID::UInt32),
"UINT16" => Ok(LogicalTypeID::UInt16),
"UINT8" => Ok(LogicalTypeID::UInt8),
"DOUBLE" => Ok(LogicalTypeID::Double),
"FLOAT" => Ok(LogicalTypeID::Float),
"STRING" => Ok(LogicalTypeID::String),
"BLOB" => Ok(LogicalTypeID::Blob),
"DATE" => Ok(LogicalTypeID::Date),
"TIMESTAMP" | "TIMESTAMP_MS" => Ok(LogicalTypeID::Timestamp),
"TIMESTAMP_SEC" => Ok(LogicalTypeID::TimestampSec),
"TIMESTAMP_NS" => Ok(LogicalTypeID::TimestampNs),
"TIMESTAMP_TZ" => Ok(LogicalTypeID::TimestampTz),
"INTERVAL" => Ok(LogicalTypeID::Interval),
"SERIAL" => Ok(LogicalTypeID::Serial),
"UINT128" => Ok(LogicalTypeID::UInt128),
"JSON" => Ok(LogicalTypeID::Json),
"TIME" | "DTIME" => Ok(LogicalTypeID::Time),
_ => Err(format!("Unknown type: {type_name}").into()),
}
}
pub fn parse_type_resolved(&self, type_name: &str) -> Result<LogicalTypeID, BinderError> {
if let Ok(t) = Self::parse_type(type_name) {
return Ok(t);
}
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let base = catalog
.resolve_type_alias(type_name)
.ok_or_else(|| format!("Unknown type: {type_name}"))?;
Self::parse_type(&base)
}
pub fn parse_compression(comp: Option<&str>) -> Result<akar_common::enums::CompressionType, BinderError> {
use akar_common::enums::CompressionType;
match comp {
None => Ok(CompressionType::Uncompressed), Some(s) => match s.to_uppercase().as_str() {
"UNCOMPRESSED" => Ok(CompressionType::Uncompressed),
"CONSTANT" => Ok(CompressionType::Constant),
"ONEVALUE" => Ok(CompressionType::OneValue),
"BOOLEAN" => Ok(CompressionType::Boolean),
"INTEGER_BITPACKING" => Ok(CompressionType::IntegerBitpacking),
"STRING_DICTIONARY" => Ok(CompressionType::StringDictionary),
"FLOAT" => Ok(CompressionType::Float),
"LIST_DELTA" => Ok(CompressionType::ListDelta),
_ => Err(format!("Unknown compression type: {s}").into()),
},
}
}
fn bind_query(&self, query: Query) -> Result<BoundStatement, BinderError> {
let mut clauses = Vec::new();
let mut variables: Vec<BoundVariable> = Vec::new();
for clause in query.clauses {
let (bound_clause, new_vars) = match clause {
Clause::Match(m) => {
let (bound, vars) = self.bind_match(&m, &variables)?;
(BoundClause::BoundMatch(bound), vars)
}
Clause::Return(r) => {
let bound = self.bind_return(&r, &variables)?;
(BoundClause::BoundReturn(bound), Vec::new())
}
Clause::With(r) => {
let bound = self.bind_return(&r, &variables)?;
let projected: Vec<BoundVariable> = bound
.expressions
.iter()
.filter_map(|be| {
if let Expression::Variable(v) = &be.expression {
if let Some(existing) = variables.iter().find(|x| &x.name == v) {
let alias = be.alias.clone().unwrap_or_else(|| v.clone());
return Some(BoundVariable {
name: alias,
table_id: existing.table_id,
label: existing.label.clone(),
is_node: existing.is_node,
});
}
}
be.alias.as_ref().map(|alias| BoundVariable {
name: alias.clone(),
table_id: 0,
label: None,
is_node: false,
})
})
.collect::<Vec<_>>();
(BoundClause::BoundWith(bound), projected)
}
Clause::Where(w) => {
let bound = self.bind_where(&w, &variables)?;
(BoundClause::BoundWhere(bound), Vec::new())
}
Clause::Create(c) => {
let (bound, vars) = self.bind_match_create(&c, &variables)?;
(BoundClause::BoundCreate(bound), vars)
}
Clause::Delete(d) => {
let bound = self.bind_delete(&d, &variables)?;
(BoundClause::BoundDelete(bound), Vec::new())
}
Clause::Set(s) => {
let bound = self.bind_set(&s, &variables)?;
(BoundClause::BoundSet(bound), Vec::new())
}
Clause::Unwind(u) => {
let bound = self.bind_unwind(&u)?;
let new_var = BoundVariable {
name: bound.variable.clone(),
table_id: 0,
label: None,
is_node: false,
};
(BoundClause::BoundUnwind(bound), vec![new_var])
}
Clause::Foreach(f) => {
let bound = self.bind_foreach(&f, &variables)?;
let new_var = BoundVariable {
name: bound.variable.clone(),
table_id: 0,
label: None,
is_node: false,
};
(BoundClause::BoundForeach(bound), vec![new_var])
}
Clause::OptionalMatch(m) => {
let (bound, vars) = self.bind_optional_match(&m, &variables)?;
(BoundClause::BoundOptionalMatch(bound), vars)
}
Clause::Merge(m) => {
let (bound, vars) = self.bind_merge_clause(&m, &variables)?;
(BoundClause::BoundMerge(bound), vars)
}
};
if matches!(bound_clause, BoundClause::BoundWith(_)) {
variables = new_vars;
} else {
variables.extend(new_vars);
}
clauses.push(bound_clause.clone());
if let BoundClause::BoundMatch(bound) = &bound_clause {
let mut inline_exprs = Vec::new();
for pattern in &bound.patterns {
if let Some(node_var) = &pattern.node_variable {
for (key, val_expr) in &pattern.properties {
let prop_access = akar_parser::ast::Expression::PropertyAccess(
Box::new(akar_parser::ast::Expression::Variable(node_var.clone())),
key.clone(),
);
let equals = akar_parser::ast::Expression::BinaryOp(
akar_parser::ast::BinaryOp::Equal,
Box::new(prop_access),
Box::new(val_expr.clone()),
);
inline_exprs.push(equals);
}
}
if let Some(edge) = &pattern.edge {
if let Some(edge_var) = &edge.variable {
for (key, val_expr) in &edge.properties {
let prop_access = akar_parser::ast::Expression::PropertyAccess(
Box::new(akar_parser::ast::Expression::Variable(edge_var.clone())),
key.clone(),
);
let equals = akar_parser::ast::Expression::BinaryOp(
akar_parser::ast::BinaryOp::Equal,
Box::new(prop_access),
Box::new(val_expr.clone()),
);
inline_exprs.push(equals);
}
}
}
}
if !inline_exprs.is_empty() {
let combined = inline_exprs
.into_iter()
.reduce(|acc, e| {
akar_parser::ast::Expression::BinaryOp(
akar_parser::ast::BinaryOp::And,
Box::new(acc),
Box::new(e),
)
})
.unwrap();
let bound_expr = self.resolve_expression(&combined, &variables)?;
clauses.push(BoundClause::BoundWhere(BoundWhereClause { expression: bound_expr }));
}
}
}
Ok(BoundStatement::BoundQuery(BoundQuery { clauses, variables }))
}
fn bind_match(
&self,
m: &MatchClause,
existing_vars: &[BoundVariable],
) -> Result<(BoundMatchClause, Vec<BoundVariable>), BinderError> {
let mut patterns = Vec::new();
let mut new_vars = Vec::new();
for pattern in &m.patterns {
let all_vars: Vec<BoundVariable> = existing_vars.iter().cloned().chain(new_vars.iter().cloned()).collect();
let (bound, nv) = self.bind_pattern(pattern, &all_vars, false)?;
patterns.push(bound);
new_vars.extend(nv);
}
let fts_query = match m.fts_query.as_ref().map(|fq| -> Result<BoundFtsQuery, String> {
let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
let (table_name, column_name) = catalog
.get_fts_index(&fq.index_name)
.map(|(t, c)| (t.to_string(), c.to_string()))
.unwrap_or_default();
Ok(BoundFtsQuery {
index_name: fq.index_name.clone(),
query_string: fq.query_string.clone(),
table_name,
column_name,
})
}) {
Some(r) => Some(r?),
None => None,
};
Ok((
BoundMatchClause {
patterns,
new_variables: new_vars.clone(),
fts_query,
},
new_vars,
))
}
fn bind_pattern(
&self,
pattern: &Pattern,
existing_vars: &[BoundVariable],
allow_existing: bool,
) -> Result<(BoundPattern, Vec<BoundVariable>), BinderError> {
let mut new_vars = Vec::new();
let mut node_table_id = None;
let mut bound_edge = None;
let (node_var, node_label) = if let Some(ref n) = pattern.node {
let var = n.variable.clone();
let label = n.labels.first().cloned();
if let Some(ref lbl) = label {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
match catalog.get_entry_by_name(lbl) {
Some(entry) if entry.is_node_table() => {
node_table_id = Some(entry.table_id());
}
Some(_entry) => {
return Err(format!("'{}' is not a node table", lbl).into());
}
None => {
return Err(format!("Table '{}' not found", lbl).into());
}
}
}
if let Some(ref v) = var {
if let Some(existing) = existing_vars.iter().find(|bv| bv.name == *v) {
let same_node = allow_existing
|| (existing.is_node
&& (label.is_none()
|| (node_table_id.is_some() && existing.table_id == node_table_id.unwrap_or(0))));
if same_node {
if node_table_id.is_none() {
node_table_id = Some(existing.table_id);
}
} else {
return Err(format!("Variable '{}' already defined", v).into());
}
} else {
new_vars.push(BoundVariable {
name: var.clone().unwrap_or_else(|| "_anon_".to_string()),
table_id: node_table_id.unwrap_or(0),
label: label.clone(),
is_node: true,
});
}
} else {
new_vars.push(BoundVariable {
name: "_anon_".to_string(),
table_id: node_table_id.unwrap_or(0),
label: label.clone(),
is_node: true,
});
}
(var, label)
} else {
(None, None)
};
if let Some(ref e) = pattern.edge {
let edge_var = e.variable.clone();
let edge_label = e.labels.first().cloned();
let mut rel_table_id = None;
if let Some(ref lbl) = edge_label {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
match catalog.get_entry_by_name(lbl) {
Some(entry) if entry.is_rel_table() => {
rel_table_id = Some(entry.table_id());
}
Some(_) => {
return Err(format!("'{}' is not a rel table", lbl).into());
}
None => {
return Err(format!("Rel table '{}' not found", lbl).into());
}
}
}
if let Some(ref v) = edge_var {
if existing_vars.iter().any(|bv| bv.name == *v) || new_vars.iter().any(|bv| bv.name == *v) {
return Err(format!("Variable '{}' already defined", v).into());
}
}
new_vars.push(BoundVariable {
name: edge_var.clone().unwrap_or_else(|| "_anon_edge_".to_string()),
table_id: rel_table_id.unwrap_or(0),
label: edge_label.clone(),
is_node: false,
});
bound_edge = Some(BoundEdgePattern {
variable: e.variable.clone(),
label: edge_label,
rel_table_id,
direction: e.direction.clone(),
properties: e.properties.clone(),
lower_bound: e.lower_bound,
upper_bound: e.upper_bound,
});
}
Ok((
BoundPattern {
node_variable: node_var,
node_label,
node_table_id,
properties: pattern.node.as_ref().map(|n| n.properties.clone()).unwrap_or_default(),
edge: bound_edge,
},
new_vars,
))
}
fn bind_return(&self, r: &ReturnClause, variables: &[BoundVariable]) -> Result<BoundReturnClause, BinderError> {
let mut expressions = Vec::new();
for item in &r.expressions {
match &item.expression {
Expression::Star => {
if variables.is_empty() {
return Err("RETURN or WITH * is not allowed when there are no variables in scope.".into());
}
for var in variables {
expressions.push(BoundExpression {
expression: Expression::Variable(var.name.clone()),
resolved_type: if var.is_node {
LogicalTypeID::Node
} else {
LogicalTypeID::Rel
},
is_constant: false,
alias: None,
});
}
}
_ => {
let mut resolved = self.resolve_expression(&item.expression, variables)?;
resolved.alias = item.alias.clone();
expressions.push(resolved);
}
}
}
let alias_types: Vec<(String, LogicalTypeID)> = expressions
.iter()
.filter_map(|be| be.alias.clone().map(|a| (a, be.resolved_type)))
.collect();
let order_by = r
.order_by
.as_ref()
.map(|items| {
items
.iter()
.map(|item| {
let resolved = match &item.expression {
Expression::Variable(name) => match alias_types.iter().find(|(alias, _)| alias == name) {
Some((alias, typ)) => BoundExpression {
expression: Expression::Variable(alias.clone()),
resolved_type: *typ,
is_constant: false,
alias: None,
},
None => self.resolve_expression(&item.expression, variables)?,
},
_ => self.resolve_expression(&item.expression, variables)?,
};
Ok(crate::bound_statement::BoundOrderByItem {
expression: resolved,
ascending: item.ascending,
})
})
.collect::<Result<Vec<_>, BinderError>>()
})
.transpose()?;
Ok(BoundReturnClause {
expressions,
distinct: r.distinct,
order_by,
limit: r.limit,
skip: r.skip,
limit_param: r.limit_param.clone(),
skip_param: r.skip_param.clone(),
})
}
fn bind_where(&self, w: &WhereClause, variables: &[BoundVariable]) -> Result<BoundWhereClause, BinderError> {
let resolved = self.resolve_expression(&w.expression, variables)?;
if resolved.resolved_type != LogicalTypeID::Bool && resolved.resolved_type != LogicalTypeID::Any {
return Err(format!("WHERE clause must be boolean, got {:?}", resolved.resolved_type).into());
}
Ok(BoundWhereClause { expression: resolved })
}
fn bind_match_create(
&self,
c: &CreateClause,
existing_vars: &[BoundVariable],
) -> Result<(BoundMatchClause, Vec<BoundVariable>), BinderError> {
let mut patterns = Vec::new();
let mut new_vars = Vec::new();
for pattern in &c.patterns {
let all_vars: Vec<BoundVariable> = existing_vars.iter().cloned().chain(new_vars.iter().cloned()).collect();
let (bound, nv) = self.bind_pattern(pattern, &all_vars, true)?;
patterns.push(bound);
new_vars.extend(nv);
}
Ok((
BoundMatchClause {
patterns,
new_variables: new_vars.clone(),
fts_query: None, },
new_vars,
))
}
fn resolve_expression(
&self,
expr: &Expression,
variables: &[BoundVariable],
) -> Result<BoundExpression, BinderError> {
match expr {
Expression::Constant(c) => {
let typ = match c {
Constant::Null => LogicalTypeID::Any,
Constant::Bool(_) => LogicalTypeID::Bool,
Constant::Integer(_) => LogicalTypeID::Int64,
Constant::Float(_) => LogicalTypeID::Double,
Constant::String(_) => LogicalTypeID::String,
};
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: typ,
is_constant: true,
alias: None,
})
}
Expression::Variable(name) => {
if let Some(var) = variables.iter().find(|v| v.name == *name) {
let typ = if var.is_node {
LogicalTypeID::Node
} else {
LogicalTypeID::Rel
};
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: typ,
is_constant: false,
alias: None,
})
} else if name.to_uppercase() == "COUNT" || name == "*" {
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::Int64,
is_constant: false,
alias: None,
})
} else {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
if let Some(entry) = catalog.get_entry_by_name(name) {
let typ = if entry.is_node_table() {
LogicalTypeID::Node
} else {
LogicalTypeID::Rel
};
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: typ,
is_constant: false,
alias: None,
})
} else {
Err(format!("Variable '{}' not in scope", name).into())
}
}
}
Expression::Parameter(_name) => {
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::Any,
is_constant: false,
alias: None,
})
}
Expression::PropertyAccess(obj, prop) => {
let bound_obj = self.resolve_expression(obj, variables)?;
let prop_type = match obj.as_ref() {
Expression::Variable(var_name) => {
if let Some(variable) = variables.iter().find(|v| v.name == *var_name) {
if let Some(ref table_label) = variable.label {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
match catalog.get_property_type(table_label, prop) {
Some(type_id) => type_id,
None => {
return Err(format!(
"Property '{}' not found on table '{}'",
prop, table_label
)
.into());
}
}
} else {
LogicalTypeID::Any
}
} else {
bound_obj.resolved_type
}
}
_ => {
LogicalTypeID::Any
}
};
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: prop_type,
is_constant: false,
alias: None,
})
}
Expression::FunctionCall(name, args) => {
let resolved_args: Result<Vec<BoundExpression>, BinderError> =
args.iter().map(|a| self.resolve_expression(a, variables)).collect();
let _args = resolved_args?;
let upper = name.to_uppercase();
let base = upper.strip_suffix("_DISTINCT").unwrap_or(&upper);
let return_type = match base {
"COUNT" | "SUM" | "MIN" | "MAX" | "AVG" => LogicalTypeID::Int64,
"NEXTVAL" | "CURRVAL" => LogicalTypeID::Int64,
"STARTS_WITH" | "ENDS_WITH" | "CONTAINS" => LogicalTypeID::Bool,
"TO_UPPER" | "TO_LOWER" | "UPPER" | "LOWER" | "UCASE" | "LCASE" | "TRIM" | "SUBSTRING"
| "REPLACE" => LogicalTypeID::String,
"ABS" | "CEIL" | "CEILING" | "FLOOR" | "ROUND" | "SQRT" | "LOG" | "EXP" | "SIN" | "COS" | "TAN" => {
LogicalTypeID::Double
}
"DATE" | "TIMESTAMP" => LogicalTypeID::Date,
"INT64" | "INT" => LogicalTypeID::Int64,
"FLOAT" | "DOUBLE" | "BOOL" | "BOOLEAN" | "STRING" | "BLOB" => LogicalTypeID::String,
_ => LogicalTypeID::Any,
};
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: return_type,
is_constant: false,
alias: None,
})
}
Expression::BinaryOp(op, left, right) => {
let left = self.resolve_expression(left, variables)?;
let right = self.resolve_expression(right, variables)?;
let result_type = match op {
BinaryOp::Equal
| BinaryOp::NotEqual
| BinaryOp::LessThan
| BinaryOp::LessThanOrEqual
| BinaryOp::GreaterThan
| BinaryOp::GreaterThanOrEqual
| BinaryOp::And
| BinaryOp::Or
| BinaryOp::Xor
| BinaryOp::In
| BinaryOp::NotIn
| BinaryOp::StartsWith
| BinaryOp::EndsWith
| BinaryOp::Contains
| BinaryOp::Like => LogicalTypeID::Bool,
BinaryOp::Add | BinaryOp::Subtract | BinaryOp::Multiply | BinaryOp::Divide | BinaryOp::Modulo => {
if left.resolved_type == LogicalTypeID::Double || right.resolved_type == LogicalTypeID::Double {
LogicalTypeID::Double
} else {
LogicalTypeID::Int64
}
}
BinaryOp::Concat => LogicalTypeID::String,
};
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: result_type,
is_constant: left.is_constant && right.is_constant,
alias: None,
})
}
Expression::UnaryOp(op, inner) => {
let inner = self.resolve_expression(inner, variables)?;
let result_type = match op {
UnaryOp::Not | UnaryOp::IsNull | UnaryOp::IsNotNull => LogicalTypeID::Bool,
UnaryOp::Negate => inner.resolved_type,
};
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: result_type,
is_constant: inner.is_constant,
alias: None,
})
}
Expression::List(items) => {
let resolved: Result<Vec<BoundExpression>, BinderError> =
items.iter().map(|i| self.resolve_expression(i, variables)).collect();
resolved?;
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::List,
is_constant: false,
alias: None,
})
}
Expression::Map(entries) => {
for (_, v) in entries {
self.resolve_expression(v, variables)?;
}
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::Map,
is_constant: false,
alias: None,
})
}
Expression::ExistsSubquery(query) => {
let _bound = self.bind_query(*query.clone())?;
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::Bool,
is_constant: false,
alias: None,
})
}
Expression::Case(case_expr) => {
if let Some(subj) = &case_expr.subject {
self.resolve_expression(subj, variables)?;
}
let mut result_type = LogicalTypeID::Any;
for alt in &case_expr.alternatives {
self.resolve_expression(&alt.when, variables)?;
let then_bound = self.resolve_expression(&alt.then, variables)?;
if result_type == LogicalTypeID::Any {
result_type = then_bound.resolved_type;
}
}
if let Some(else_e) = &case_expr.else_expr {
let else_bound = self.resolve_expression(else_e, variables)?;
if result_type == LogicalTypeID::Any {
result_type = else_bound.resolved_type;
}
}
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: result_type,
is_constant: false,
alias: None,
})
}
Expression::Star => {
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::Any,
is_constant: false,
alias: None,
})
}
Expression::ListPredicate {
quantifier: _,
list,
var_name,
predicate,
} => {
self.resolve_expression(list, variables)?;
let mut new_vars = variables.to_vec();
new_vars.push(crate::bound_statement::BoundVariable {
name: var_name.clone(),
table_id: 0,
label: None,
is_node: false,
});
self.resolve_expression(predicate, &new_vars)?;
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::Bool,
is_constant: false,
alias: None,
})
}
Expression::Lambda { var_name: _, body } => {
self.resolve_expression(body, variables)?;
Ok(BoundExpression {
expression: expr.clone(),
resolved_type: LogicalTypeID::Any,
is_constant: false,
alias: None,
})
}
}
}
fn bind_create_node_table(&self, t: CreateNodeTable) -> Result<BoundStatement, BinderError> {
if t.name.is_empty() {
return Err("Table name cannot be empty".into());
}
let mut columns = Vec::new();
for col in &t.columns {
let logical_type = self.parse_type_resolved(&col.type_name)?;
let compression = Self::parse_compression(col.compression.as_deref())?;
columns.push(CatalogColumn {
name: col.name.clone(),
logical_type,
is_primary_key: col.name == t.primary_key,
compression,
default_value: None,
});
}
if columns.is_empty() {
return Err("Table must have at least one column".into());
}
if !columns.iter().any(|c| c.is_primary_key) {
return Err(format!("CREATE NODE TABLE requires a PRIMARY KEY (missing '{}')", t.primary_key).into());
}
let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
match catalog.create_node_table(t.name.clone(), columns.clone()) {
CatalogResult::Created { .. } => {}
CatalogResult::AlreadyExists if t.if_not_exists => {
}
CatalogResult::AlreadyExists => {
return Err(format!("Table '{}' already exists", t.name).into());
}
_ => return Err("Failed to create table".into()),
}
Ok(BoundStatement::BoundCreateNodeTable(BoundCreateNodeTable {
name: t.name,
columns,
primary_key: t.primary_key,
if_not_exists: t.if_not_exists,
}))
}
fn bind_create_vector_index(&self, v: akar_parser::ast::CreateVectorIndex) -> Result<BoundStatement, BinderError> {
if v.index_name.is_empty() {
return Err("Index name cannot be empty".into());
}
if v.metric.is_empty() {
return Err("Metric must be specified (cosine, euclidean, l2, or dot)".into());
}
if v.dimensions == 0 {
return Err("Dimensions must be greater than 0".into());
}
let col_exists = {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(&v.table_name)
.ok_or_else(|| format!("Table '{}' not found", v.table_name))?;
entry.columns().iter().any(|c| c.name == v.column_name)
};
if !col_exists {
return Err(format!("Column '{}' not found in table '{}'", v.column_name, v.table_name).into());
}
match v.metric.to_lowercase().as_str() {
"cosine" | "euclidean" | "l2" | "dot" => {}
other => {
return Err(format!("Unknown metric '{other}'. Supported: cosine, euclidean, l2, dot").into());
}
}
let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
match catalog.create_vector_index(
v.index_name.clone(),
v.table_name.clone(),
v.column_name.clone(),
v.metric.clone(),
v.dimensions,
) {
CatalogResult::Created { .. } => {}
CatalogResult::AlreadyExists => {
return Err(format!("Vector index '{}' already exists", v.index_name).into());
}
CatalogResult::NotFound => {
return Err(format!("Table '{}' not found", v.table_name).into());
}
CatalogResult::Dropped { .. } => {
return Err("Unexpected: Dropped result from create_vector_index".into());
}
}
Ok(BoundStatement::BoundCreateVectorIndex(BoundCreateVectorIndex {
index_name: v.index_name,
table_name: v.table_name,
column_name: v.column_name,
metric: v.metric,
dimensions: v.dimensions,
}))
}
fn bind_create_index(&self, v: akar_parser::ast::CreateIndex) -> Result<BoundStatement, BinderError> {
if v.index_name.is_empty() {
return Err("Index name cannot be empty".into());
}
let index_type = IndexType::from_str(&v.index_type)
.ok_or_else(|| format!("Unknown index type '{}'. Use ART or HASH", v.index_type))?;
{
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(&v.table_name)
.ok_or_else(|| format!("Table '{}' not found", v.table_name))?;
let col_exists = entry.columns().iter().any(|c| c.name == v.property);
if !col_exists {
return Err(format!("Column '{}' not found in table '{}'", v.property, v.table_name).into());
}
let pk_col = entry.columns().iter().find(|c| c.is_primary_key);
if pk_col.map(|c| c.name.as_str()) != Some(v.property.as_str()) {
return Err(format!(
"Cannot create index on non-PK column '{}'. Only PK columns are supported.",
v.property
)
.into());
}
}
let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
catalog.create_index(&v.table_name, v.index_name.clone(), index_type, &v.property)?;
Ok(BoundStatement::BoundCreateIndex(BoundCreateIndex {
index_type,
index_name: v.index_name,
table_name: v.table_name,
column_name: v.property,
}))
}
fn bind_drop_index(&self, v: akar_parser::ast::DropIndex) -> Result<BoundStatement, BinderError> {
if v.index_name.is_empty() {
return Err("Index name cannot be empty".into());
}
let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
catalog.drop_index(&v.table_name, &v.index_name)?;
Ok(BoundStatement::BoundDropIndex(BoundDropIndex {
index_name: v.index_name,
table_name: v.table_name,
}))
}
fn bind_create_rel_table(&self, t: CreateRelTable) -> Result<BoundStatement, BinderError> {
if t.name.is_empty() {
return Err("Table name cannot be empty".into());
}
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let src_id = catalog
.get_table_id(&t.from)
.ok_or_else(|| format!("Source table '{}' not found", t.from))?;
let dst_id = catalog
.get_table_id(&t.to)
.ok_or_else(|| format!("Destination table '{}' not found", t.to))?;
drop(catalog);
let mut columns = Vec::new();
for col in &t.columns {
let logical_type = self.parse_type_resolved(&col.type_name)?;
let compression = Self::parse_compression(col.compression.as_deref())?;
columns.push(CatalogColumn {
name: col.name.clone(),
logical_type,
is_primary_key: false,
compression,
default_value: None,
});
}
let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
match catalog.create_rel_table(t.name.clone(), src_id, dst_id, columns.clone()) {
CatalogResult::Created { .. } => {}
CatalogResult::AlreadyExists if t.if_not_exists => {
}
CatalogResult::AlreadyExists => {
return Err(format!("Rel table '{}' already exists", t.name).into());
}
_ => return Err("Failed to create rel table".into()),
}
Ok(BoundStatement::BoundCreateRelTable(BoundCreateRelTable {
name: t.name,
from: t.from,
to: t.to,
src_table_id: src_id,
dst_table_id: dst_id,
columns,
if_not_exists: t.if_not_exists,
}))
}
fn bind_drop_table(&self, t: DropTable) -> Result<BoundStatement, BinderError> {
let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
match catalog.drop_table(&t.name) {
CatalogResult::Dropped { .. } => Ok(BoundStatement::BoundDropTable(BoundDropTable { name: t.name })),
CatalogResult::NotFound => Err(format!("Table '{}' not found", t.name).into()),
_ => Err("Failed to drop table".into()),
}
}
fn bind_unwind(&self, u: &akar_parser::ast::UnwindClause) -> Result<BoundUnwindClause, BinderError> {
match &u.expression {
akar_parser::ast::Expression::List(_) => {}
akar_parser::ast::Expression::Variable(_) => {}
akar_parser::ast::Expression::Parameter(_) => {}
_ => return Err(format!("UNWIND requires a list expression, got: {:?}", u.expression).into()),
}
if u.variable.is_empty() {
return Err("UNWIND requires a variable name".into());
}
Ok(BoundUnwindClause {
expression: u.expression.clone(),
variable: u.variable.clone(),
})
}
fn bind_foreach(
&self,
f: &akar_parser::ast::ForeachClause,
variables: &[BoundVariable],
) -> Result<BoundForeachClause, BinderError> {
match &f.expression {
akar_parser::ast::Expression::List(_) | akar_parser::ast::Expression::Variable(_) => {}
_ => return Err(format!("FOREACH requires a list expression, got: {:?}", f.expression).into()),
}
if f.variable.is_empty() {
return Err("FOREACH requires a variable name".into());
}
let mut local_vars = variables.to_vec();
local_vars.push(BoundVariable {
name: f.variable.clone(),
table_id: 0,
label: None,
is_node: false,
});
let mut sub_statements = Vec::new();
for clause in &f.clauses {
match clause {
akar_parser::ast::Clause::Create(cc) => {
let bound = self.bind_create_dml(cc.clone(), &local_vars)?;
sub_statements.push(bound);
}
akar_parser::ast::Clause::Set(sc) => {
let bound_set = self.bind_set(sc, &local_vars)?;
sub_statements.push(BoundStatement::BoundQuery(BoundQuery {
clauses: vec![BoundClause::BoundSet(bound_set)],
variables: local_vars.clone(),
}));
}
akar_parser::ast::Clause::Delete(dc) => {
let bound_delete = self.bind_delete(dc, &local_vars)?;
sub_statements.push(BoundStatement::BoundQuery(BoundQuery {
clauses: vec![BoundClause::BoundDelete(bound_delete)],
variables: local_vars.clone(),
}));
}
_ => {
return Err(format!("Unsupported FOREACH sub-clause: {:?}", clause).into());
}
}
}
Ok(BoundForeachClause {
variable: f.variable.clone(),
expression: f.expression.clone(),
sub_statements,
})
}
fn bind_optional_match(
&self,
m: &akar_parser::ast::OptionalMatchClause,
existing_vars: &[BoundVariable],
) -> Result<(BoundMatchClause, Vec<BoundVariable>), BinderError> {
let mut patterns = Vec::new();
let mut new_vars = Vec::new();
for pattern in &m.patterns {
let all_vars: Vec<BoundVariable> = existing_vars.iter().cloned().chain(new_vars.iter().cloned()).collect();
let (bound, nv) = self.bind_pattern(pattern, &all_vars, false)?;
patterns.push(bound);
new_vars.extend(nv);
}
Ok((
BoundMatchClause {
patterns,
new_variables: new_vars.clone(),
fts_query: None, },
new_vars,
))
}
fn bind_set(
&self,
s: &akar_parser::ast::SetClause,
variables: &[BoundVariable],
) -> Result<BoundSetClause, BinderError> {
let mut items = Vec::new();
for item in &s.items {
match &item.property {
akar_parser::ast::Expression::PropertyAccess(var_expr, prop_name) => {
match var_expr.as_ref() {
akar_parser::ast::Expression::Variable(var_name) => {
let bound_var = variables
.iter()
.find(|v| v.name == *var_name)
.ok_or_else(|| format!("Variable '{}' not in scope for SET", var_name))?;
items.push(BoundSetItem {
property: item.property.clone(),
value: item.value.clone(),
column_name: prop_name.clone(),
column_idx: 0, table_name: bound_var.label.clone().unwrap_or_default(),
table_id: bound_var.table_id,
is_node: bound_var.is_node,
});
}
_ => return Err("SET property must be on a variable".into()),
}
}
_ => return Err("SET requires property access expression (e.g., n.age)".into()),
}
}
Ok(BoundSetClause { items })
}
fn bind_union(&self, u: akar_parser::ast::UnionStatement) -> Result<BoundStatement, BinderError> {
let left = self.bind_query(u.left)?;
let right = self.bind_query(u.right)?;
Ok(BoundStatement::BoundUnion(BoundUnion {
left: Box::new(match left {
BoundStatement::BoundQuery(q) => q,
_ => unreachable!(),
}),
right: Box::new(match right {
BoundStatement::BoundQuery(q) => q,
_ => unreachable!(),
}),
all: u.all,
}))
}
fn bind_merge(&self, m: akar_parser::ast::MergeStatement) -> Result<BoundStatement, BinderError> {
let (bound, _) = self.bind_merge_clause(&m, &[])?;
Ok(BoundStatement::BoundMerge(bound))
}
fn bind_merge_clause(
&self,
m: &akar_parser::ast::MergeStatement,
variables: &[BoundVariable],
) -> Result<(BoundMerge, Vec<BoundVariable>), BinderError> {
let patterns = self.bind_merge_patterns(&m.patterns, variables)?;
let primary = patterns
.iter()
.find_map(|p| p.node.clone())
.ok_or("MERGE requires at least one node pattern")?;
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let on_create = resolve_set_items(&catalog, &m.on_create)?;
let on_match = resolve_set_items(&catalog, &m.on_match)?;
let mut new_vars = Vec::new();
for pat in &patterns {
if let Some(node) = &pat.node {
if let Some(name) = &node.variable {
let is_reference = variables.iter().any(|v| &v.name == name && v.is_node);
if !is_reference {
new_vars.push(BoundVariable {
name: name.clone(),
table_id: node.table_id,
label: Some(node.table_name.clone()),
is_node: true,
});
}
}
}
if let Some(edge) = &pat.edge {
if let Some(name) = &edge.variable {
new_vars.push(BoundVariable {
name: name.clone(),
table_id: edge.table_id,
label: Some(edge.table_name.clone()),
is_node: false,
});
}
}
}
Ok((
BoundMerge {
table_name: primary.table_name,
table_id: primary.table_id,
properties: primary.properties,
patterns,
on_create,
on_match,
},
new_vars,
))
}
fn bind_merge_patterns(
&self,
patterns: &[akar_parser::ast::Pattern],
variables: &[BoundVariable],
) -> Result<Vec<BoundCreatePattern>, BinderError> {
let mut bound = Vec::with_capacity(patterns.len());
for (i, pat) in patterns.iter().enumerate() {
let node = if let Some(ref n) = pat.node {
match n.labels.first() {
Some(label) => {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(label)
.ok_or_else(|| format!("Table '{label}' not found"))?;
if !entry.is_node_table() {
return Err(format!("'{label}' is not a node table").into());
}
Some(BoundNodeCreate {
variable: n.variable.clone(),
table_name: label.clone(),
table_id: entry.table_id(),
properties: n.properties.clone(),
})
}
None => {
let var_name = n.variable.as_ref().ok_or("MERGE node requires a label or a variable")?;
let existing =
variables
.iter()
.find(|v| &v.name == var_name && v.is_node)
.ok_or_else(|| {
format!(
"MERGE references unknown variable '{}' (no label to resolve table)",
var_name
)
})?;
let label = existing.label.clone().unwrap_or_default();
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(&label)
.ok_or_else(|| format!("Table '{label}' not found"))?;
if !entry.is_node_table() {
return Err(format!("'{label}' is not a node table").into());
}
Some(BoundNodeCreate {
variable: n.variable.clone(),
table_name: label.clone(),
table_id: existing.table_id,
properties: n.properties.clone(),
})
}
}
} else {
None
};
let edge = if let Some(ref e) = pat.edge {
let label = e.labels.first().ok_or("Edge requires a label (rel table name)")?;
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(label)
.ok_or_else(|| format!("Rel table '{label}' not found"))?;
if !entry.is_rel_table() {
return Err(format!("'{label}' is not a rel table").into());
}
let cur_var = pat
.node
.as_ref()
.and_then(|n| n.variable.clone())
.ok_or("Edge endpoints must be named node variables")?;
let nxt_var = patterns
.get(i + 1)
.and_then(|p| p.node.as_ref())
.and_then(|n| n.variable.clone())
.ok_or("Edge endpoints must be named node variables")?;
let (src_var, dst_var) = match e.direction {
akar_parser::ast::EdgeDirection::RightToLeft => (nxt_var, cur_var),
_ => (cur_var, nxt_var),
};
Some(BoundEdgeCreate {
variable: e.variable.clone(),
table_name: label.clone(),
table_id: entry.table_id(),
src_var,
dst_var,
properties: e.properties.clone(),
})
} else {
None
};
bound.push(BoundCreatePattern { node, edge });
}
Ok(bound)
}
fn bind_create_patterns(
&self,
patterns: &[akar_parser::ast::Pattern],
) -> Result<Vec<BoundCreatePattern>, BinderError> {
let mut bound = Vec::with_capacity(patterns.len());
for (i, pat) in patterns.iter().enumerate() {
let node = if let Some(ref n) = pat.node {
let label = n.labels.first().ok_or("CREATE/MERGE requires a label (table name)")?;
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(label)
.ok_or_else(|| format!("Table '{label}' not found"))?;
if !entry.is_node_table() {
return Err(format!("'{label}' is not a node table").into());
}
Some(BoundNodeCreate {
variable: n.variable.clone(),
table_name: label.clone(),
table_id: entry.table_id(),
properties: n.properties.clone(),
})
} else {
None
};
let edge = if let Some(ref e) = pat.edge {
let label = e.labels.first().ok_or("Edge requires a label (rel table name)")?;
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(label)
.ok_or_else(|| format!("Rel table '{label}' not found"))?;
if !entry.is_rel_table() {
return Err(format!("'{label}' is not a rel table").into());
}
let cur_var = pat
.node
.as_ref()
.and_then(|n| n.variable.clone())
.ok_or("Edge endpoints must be named node variables")?;
let nxt_var = patterns
.get(i + 1)
.and_then(|p| p.node.as_ref())
.and_then(|n| n.variable.clone())
.ok_or("Edge endpoints must be named node variables")?;
let (src_var, dst_var) = match e.direction {
akar_parser::ast::EdgeDirection::RightToLeft => (nxt_var, cur_var),
_ => (cur_var, nxt_var),
};
Some(BoundEdgeCreate {
variable: e.variable.clone(),
table_name: label.clone(),
table_id: entry.table_id(),
src_var,
dst_var,
properties: e.properties.clone(),
})
} else {
None
};
bound.push(BoundCreatePattern { node, edge });
}
Ok(bound)
}
fn bind_create_dml(
&self,
c: akar_parser::ast::CreateClause,
_variables: &[BoundVariable],
) -> Result<BoundStatement, BinderError> {
let patterns = self.bind_create_patterns(&c.patterns)?;
if patterns.iter().all(|p| p.node.is_none()) {
return Err("CREATE DML requires a node pattern".into());
}
Ok(BoundStatement::BoundCreateDml(BoundCreateDml { patterns }))
}
fn bind_standalone_call(&self, c: akar_parser::ast::StandaloneCall) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundStandaloneCall(BoundStandaloneCall {
function_name: c.function_name,
args: c.args,
}))
}
fn bind_explain(&self, e: akar_parser::ast::ExplainStatement) -> Result<BoundStatement, BinderError> {
let inner = self.bind(*e.statement)?;
Ok(BoundStatement::BoundExplain(BoundExplain {
inner: Box::new(inner),
explain_type: e.explain_type,
}))
}
fn bind_create_sequence(&self, s: akar_parser::ast::CreateSequence) -> Result<BoundStatement, BinderError> {
let increment = s.increment.unwrap_or(1);
if increment == 0 {
return Err("INCREMENT must not be zero".into());
}
let start_with = s.start_with.unwrap_or(if increment > 0 { 1 } else { -1 });
let min_value = s.min_value.unwrap_or(if increment > 0 { 1 } else { i64::MIN });
let max_value = s.max_value.unwrap_or(if increment > 0 { i64::MAX } else { -1 });
let cycle = s.cycle.unwrap_or(false);
if min_value > max_value {
return Err(format!(
"MINVALUE ({}) cannot be greater than MAXVALUE ({})",
min_value, max_value
)
.into());
}
if start_with < min_value || start_with > max_value {
return Err(format!(
"START WITH ({}) must be between MINVALUE ({}) and MAXVALUE ({})",
start_with, min_value, max_value
)
.into());
}
Ok(BoundStatement::BoundCreateSequence(BoundCreateSequence {
name: s.name,
if_not_exists: s.if_not_exists,
or_replace: s.or_replace,
start_with,
increment,
min_value,
max_value,
cycle,
}))
}
fn bind_drop_sequence(&self, s: akar_parser::ast::DropSequence) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundDropSequence(BoundDropSequence {
name: s.name,
if_exists: s.if_exists,
}))
}
fn bind_create_macro(&self, m: akar_parser::ast::CreateMacro) -> Result<BoundStatement, BinderError> {
let default_args: Vec<(String, String)> = m
.default_args
.iter()
.map(|(name, expr)| (name.clone(), expr_to_debug_string(expr)))
.collect();
let expression_str = expr_to_debug_string(&m.expression);
Ok(BoundStatement::BoundCreateMacro(BoundCreateMacro {
name: m.name,
positional_args: m.positional_args,
default_args,
expression: expression_str,
}))
}
fn bind_export_database(&self, e: akar_parser::ast::ExportDatabase) -> Result<BoundStatement, BinderError> {
let file_type = e
.options
.get("FORMAT")
.map(|s| s.to_lowercase())
.unwrap_or_else(|| "csv".to_string());
if file_type != "csv" && file_type != "parquet" {
return Err(format!("Unsupported export format '{file_type}'. Supported: csv, parquet").into());
}
let schema_only = e.options.get("SCHEMA_ONLY").map(|s| s == "true").unwrap_or(false);
Ok(BoundStatement::BoundExportDatabase(BoundExportDatabase {
file_path: e.file_path,
file_type,
schema_only,
options: e.options,
}))
}
fn bind_import_database(&self, i: akar_parser::ast::ImportDatabase) -> Result<BoundStatement, BinderError> {
let dir = Self::validate_import_database_dir(&i.file_path)?;
if !dir.join("schema.cypher").exists() {
return Err(format!("schema.cypher not found in '{}'", i.file_path).into());
}
let query = if dir.join("copy.cypher").exists() {
let schema = Self::read_file_within_dir(&dir, "schema.cypher")?;
let copy = Self::read_file_within_dir(&dir, "copy.cypher")?;
format!("{schema}\n{copy}")
} else {
Self::read_file_within_dir(&dir, "schema.cypher")?
};
let index_query = if dir.join("index.cypher").exists() {
Self::read_file_within_dir(&dir, "index.cypher")?
} else {
String::new()
};
Ok(BoundStatement::BoundImportDatabase(BoundImportDatabase {
file_path: i.file_path,
query,
index_query,
}))
}
fn validate_import_database_dir(raw_path: &str) -> Result<std::path::PathBuf, BinderError> {
if raw_path.is_empty() {
return Err("IMPORT DATABASE requires a non-empty directory path".into());
}
if raw_path.contains('\0') {
return Err("IMPORT DATABASE path contains a NUL byte".into());
}
let path = std::path::Path::new(raw_path);
for component in path.components() {
if matches!(component, std::path::Component::ParentDir) {
return Err(
format!("IMPORT DATABASE path '{raw_path}' contains '..' — path traversal is not allowed").into(),
);
}
}
if !path.exists() {
return Err(format!("Import directory '{raw_path}' not found").into());
}
if !path.is_dir() {
return Err(format!("'{raw_path}' is not a directory").into());
}
std::fs::canonicalize(path).map_err(|e| format!("Cannot resolve import directory '{raw_path}': {e}").into())
}
fn read_file_within_dir(dir: &std::path::Path, file_name: &str) -> Result<String, BinderError> {
let file_path = dir.join(file_name);
match std::fs::canonicalize(&file_path) {
Ok(resolved) if resolved.starts_with(dir) => {}
Ok(resolved) => {
return Err(format!(
"'{file_name}' resolves outside the import directory ({})",
resolved.display()
)
.into());
}
Err(e) => return Err(format!("Cannot read {file_name}: {e}").into()),
}
std::fs::read_to_string(&file_path).map_err(|e| format!("Cannot read {file_name}: {e}").into())
}
fn bind_analyze(&self, a: AnalyzeStatement) -> Result<BoundStatement, BinderError> {
let cat = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
let table_ids = if let Some(ref table_name) = a.table_name {
let id = cat
.get_table_id(table_name)
.ok_or_else(|| format!("Table '{table_name}' not found"))?;
vec![id]
} else {
cat.all_entries()
.filter(|e| e.is_node_table() || e.is_rel_table())
.map(|e| e.table_id())
.collect()
};
Ok(BoundStatement::BoundAnalyze(BoundAnalyze {
table_name: a.table_name,
table_ids,
}))
}
fn bind_transaction(&self, t: TransactionStatement) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundTransaction(BoundTransaction { action: t.action }))
}
fn bind_extension(&self, e: ExtensionStatement) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundExtension(BoundExtension {
action: e.action,
name: e.name,
}))
}
fn bind_attach_database(&self, a: AttachDatabase) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundAttachDatabase(BoundAttachDatabase {
path: a.path,
alias: a.alias,
options: a.options,
}))
}
fn bind_detach_database(&self, d: DetachDatabase) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundDetachDatabase(BoundDetachDatabase {
alias: d.alias,
}))
}
fn bind_use_database(&self, u: UseDatabase) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundUseDatabase(BoundUseDatabase { alias: u.alias }))
}
fn bind_load_from(&self, l: LoadFrom) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundLoadFrom(BoundLoadFrom {
path: l.path,
options: l.options,
}))
}
fn bind_create_type(&self, t: CreateType) -> Result<BoundStatement, BinderError> {
self.parse_type_resolved(&t.type_name)?;
{
let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
if catalog.get_type_alias(&t.name).is_some() {
return Err(format!("Type alias '{}' already exists", t.name).into());
}
}
Ok(BoundStatement::BoundCreateType(BoundCreateType {
name: t.name,
type_name: t.type_name,
}))
}
fn bind_comment_on_table(&self, c: CommentOnTable) -> Result<BoundStatement, BinderError> {
{
let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
catalog
.get_entry_by_name(&c.table_name)
.ok_or_else(|| format!("Table '{}' not found", c.table_name))?;
}
Ok(BoundStatement::BoundCommentOnTable(BoundCommentOnTable {
table_name: c.table_name,
comment: c.comment,
}))
}
fn bind_create_graph(&self, g: CreateGraph) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundCreateGraph(BoundCreateGraph {
name: g.name,
is_any: g.is_any,
}))
}
fn bind_use_graph(&self, g: UseGraph) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundUseGraph(BoundUseGraph { name: g.name }))
}
fn bind_drop_graph(&self, g: DropGraph) -> Result<BoundStatement, BinderError> {
Ok(BoundStatement::BoundDropGraph(BoundDropGraph { name: g.name }))
}
fn bind_create_fts_index(&self, f: CreateFtsIndex) -> Result<BoundStatement, BinderError> {
{
let catalog = self.catalog.lock().map_err(|e| format!("Lock error: {e}"))?;
let entry = catalog
.get_entry_by_name(&f.table_name)
.ok_or_else(|| format!("Table '{}' not found", f.table_name))?;
let has_column = entry.columns().iter().any(|c| c.name == f.column_name);
if !has_column {
return Err(format!("Column '{}' not found in table '{}'", f.column_name, f.table_name).into());
}
}
let index_name = f.index_name.clone();
{
let mut catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
catalog
.register_fts_index(index_name.clone(), f.table_name.clone(), f.column_name.clone())
.map_err(|e| format!("Failed to register FTS index: {e}"))?;
}
Ok(BoundStatement::BoundCreateFtsIndex(BoundCreateFtsIndex {
index_name: f.index_name,
table_name: f.table_name,
column_name: f.column_name,
tokenizer: f.tokenizer,
if_not_exists: f.if_not_exists,
}))
}
fn bind_alter_table(&self, a: akar_parser::ast::AlterTable) -> Result<BoundStatement, BinderError> {
let col_names: Vec<String> = {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(&a.table_name)
.ok_or_else(|| format!("Table '{}' not found", a.table_name))?;
entry.columns().iter().map(|c| c.name.clone()).collect()
};
fn has_name(col_names: &[String], name: &str) -> bool {
col_names.iter().any(|c| c.eq_ignore_ascii_case(name))
}
match &a.action {
akar_parser::ast::AlterAction::AddColumn { name: _, type_name } => {
self.parse_type_resolved(type_name)?;
}
akar_parser::ast::AlterAction::DropColumn { name } => {
if !has_name(&col_names, name) {
return Err(format!("Column '{name}' not found in table '{}'", a.table_name).into());
}
}
akar_parser::ast::AlterAction::RenameColumn { old_name, new_name } => {
if !has_name(&col_names, old_name) {
return Err(format!("Column '{old_name}' not found in table '{}'", a.table_name).into());
}
if has_name(&col_names, new_name) {
return Err(format!("Column '{new_name}' already exists in table '{}'", a.table_name).into());
}
}
akar_parser::ast::AlterAction::RenameTable { new_name: _ } => {
}
}
Ok(BoundStatement::BoundAlterTable(BoundAlterTable {
table_name: a.table_name,
action: a.action,
}))
}
fn bind_delete(
&self,
d: &akar_parser::ast::DeleteClause,
variables: &[BoundVariable],
) -> Result<BoundDeleteClause, BinderError> {
if d.expressions.is_empty() {
return Err("DELETE requires at least one expression".into());
}
let use_index = d.expressions.len() >= 8 && variables.len() >= 256;
let mut index: Option<HashMap<&str, &BoundVariable>> = None;
if use_index {
let mut map = HashMap::with_capacity(variables.len());
for v in variables {
map.entry(v.name.as_str()).or_insert(v);
}
index = Some(map);
}
let mut items = Vec::with_capacity(d.expressions.len());
for expr in &d.expressions {
match expr {
akar_parser::ast::Expression::Variable(var_name) => {
let var = match &index {
Some(map) => map.get(var_name.as_str()).copied(),
None => variables.iter().find(|v| v.name == *var_name),
}
.ok_or_else(|| format!("Variable '{}' not found in scope for DELETE", var_name))?;
items.push(BoundDeleteItem {
expression: expr.clone(),
table_name: var.label.clone().unwrap_or_default(),
table_id: var.table_id,
primary_key_column: String::new(),
is_node: var.is_node,
});
}
_ => return Err(format!("DELETE only supports variable references, got: {:?}", expr).into()),
}
}
Ok(BoundDeleteClause {
detach: d.detach,
items,
})
}
fn bind_copy_from(&self, c: akar_parser::ast::CopyFrom) -> Result<BoundStatement, BinderError> {
let catalog = self.catalog.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
let entry = catalog
.get_entry_by_name(&c.table_name)
.ok_or_else(|| format!("Table '{}' not found", c.table_name))?;
let table_id = entry.table_id();
let columns: Vec<akar_catalog::CatalogColumn> = entry.columns().to_vec();
let is_rel_table = entry.is_rel_table();
drop(catalog);
let path = std::path::Path::new(&c.file_path);
if !path.exists() {
return Err(format!("File '{}' not found", c.file_path).into());
}
if !path.is_file() {
return Err(format!("'{}' is not a file", c.file_path).into());
}
let header_val = c.options.get("HEADER").or_else(|| c.options.get("header"));
let delim_val = c.options.get("DELIM").or_else(|| c.options.get("delim"));
if let Some(hv) = header_val {
if hv.eq_ignore_ascii_case("true") && delim_val.is_some() {
let delimiter = delim_val.and_then(|d| d.chars().next()).unwrap_or(',');
let file = std::fs::File::open(&c.file_path)
.map_err(|e| format!("Cannot open file '{}': {}", c.file_path, e))?;
use std::io::{BufRead, BufReader};
let mut reader = BufReader::new(file);
let mut first_line = String::new();
reader
.read_line(&mut first_line)
.map_err(|e| format!("Cannot read file '{}': {}", c.file_path, e))?;
let trimmed = first_line.trim();
if trimmed.is_empty() {
return Err(format!("File '{}' is empty, cannot validate header", c.file_path).into());
}
let csv_col_count = trimmed.split(delimiter).count();
let expected_col_count = if is_rel_table { columns.len() + 2 } else { columns.len() };
if csv_col_count != expected_col_count {
return Err(format!(
"Column count mismatch: CSV header has {csv_col_count} columns \
but table '{}' has {expected_col_count} columns",
c.table_name,
)
.into());
}
}
}
Ok(BoundStatement::BoundCopyFrom(BoundCopyFrom {
table_name: c.table_name,
table_id,
file_path: c.file_path,
options: c.options,
columns,
}))
}
fn bind_copy_to(&self, c: akar_parser::ast::CopyTo) -> Result<BoundStatement, BinderError> {
let bound_query = match self.bind(Statement::Query(c.query))? {
BoundStatement::BoundQuery(q) => q,
_ => return Err("COPY TO inner statement must be a query".into()),
};
Ok(BoundStatement::BoundCopyTo(BoundCopyTo {
file_path: c.file_path,
format: c.format,
header: c.header,
query: bound_query,
}))
}
}
fn expr_to_debug_string(expr: &akar_parser::ast::Expression) -> String {
format!("{:?}", expr)
}
#[cfg(test)]
mod tests {
use super::*;
use akar_parser::ast::{Constant, DeleteClause, Expression};
fn binder() -> Binder {
Binder::new(Arc::new(Mutex::new(Catalog::new())))
}
fn var(name: &str, table_id: u64, label: Option<&str>, is_node: bool) -> BoundVariable {
BoundVariable {
name: name.to_string(),
table_id,
label: label.map(|s| s.to_string()),
is_node,
}
}
fn delete_clause(expressions: Vec<Expression>) -> DeleteClause {
DeleteClause {
detach: true,
expressions,
}
}
#[test]
fn delete_resolves_bound_variable_fields() {
let b = binder();
let variables = vec![var("a", 7, Some("Person"), true), var("r", 3, Some("Knows"), false)];
let d = delete_clause(vec![Expression::Variable("a".into()), Expression::Variable("r".into())]);
let BoundDeleteClause { detach, items } = b.bind_delete(&d, &variables).unwrap();
{
assert!(detach);
assert_eq!(items.len(), 2);
assert_eq!(items[0].table_name, "Person");
assert_eq!(items[0].table_id, 7);
assert!(items[0].is_node);
assert_eq!(items[1].table_name, "Knows");
assert_eq!(items[1].table_id, 3);
assert!(!items[1].is_node);
}
}
#[test]
fn delete_missing_variable_reports_original_error() {
let b = binder();
let variables = vec![var("a", 1, Some("Person"), true)];
let d = delete_clause(vec![Expression::Variable("ghost".into())]);
let err = b.bind_delete(&d, &variables).unwrap_err();
assert!(
err.to_string()
.contains("Variable 'ghost' not found in scope for DELETE"),
"unexpected error: {err}"
);
}
#[test]
fn delete_rejects_non_variable_expression() {
let b = binder();
let variables = vec![var("a", 1, Some("Person"), true)];
let d = delete_clause(vec![Expression::Constant(Constant::Integer(42))]);
let err = b.bind_delete(&d, &variables).unwrap_err();
assert!(
err.to_string().contains("DELETE only supports variable references"),
"unexpected error: {err}"
);
}
#[test]
fn delete_requires_at_least_one_expression() {
let b = binder();
let d = delete_clause(vec![]);
let err = b.bind_delete(&d, &[]).unwrap_err();
assert!(
err.to_string().contains("at least one expression"),
"unexpected error: {err}"
);
}
#[test]
fn delete_first_occurrence_wins_in_linear_path() {
let b = binder();
let variables = vec![var("dup", 1, Some("First"), true), var("dup", 2, Some("Second"), false)];
let d = delete_clause(vec![Expression::Variable("dup".into())]);
let bound = b.bind_delete(&d, &variables).unwrap();
assert_eq!(bound.items[0].table_id, 1, "first occurrence must win");
assert_eq!(bound.items[0].table_name, "First");
}
#[test]
fn delete_first_occurrence_wins_in_index_path() {
let b = binder();
let mut variables: Vec<BoundVariable> = (0..300).map(|i| var(&format!("v{i}"), i as u64, None, true)).collect();
variables[0].name = "dup".into();
variables[0].table_id = 999;
variables[150].name = "dup".into();
variables[150].table_id = 555;
let d = delete_clause(vec![Expression::Variable("dup".into()); 8]);
let bound = b.bind_delete(&d, &variables).unwrap();
assert_eq!(bound.items.len(), 8);
for item in &bound.items {
assert_eq!(item.table_id, 999, "first occurrence must win in the index path");
}
}
fn unique_temp_dir(tag: &str) -> std::path::PathBuf {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let dir = std::env::temp_dir().join(format!("akar_binder_sec_{}_{nanos}_{tag}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
dir
}
#[test]
fn import_validator_rejects_parent_dir_components() {
for evil in ["..", "../etc", "safe/../../../etc", "a/../b", "x/.."] {
let err = Binder::validate_import_database_dir(evil).unwrap_err();
assert!(
err.to_string().contains("path traversal"),
"'{evil}' must be rejected as traversal, got: {err}"
);
}
}
#[cfg(windows)]
#[test]
fn import_validator_rejects_backslash_traversal_on_windows() {
for evil in ["..\\etc", "safe\\..\\..\\etc"] {
let err = Binder::validate_import_database_dir(evil).unwrap_err();
assert!(
err.to_string().contains("path traversal"),
"'{evil}' must be rejected as traversal, got: {err}"
);
}
}
#[test]
fn import_validator_rejects_empty_and_nul_paths() {
assert!(Binder::validate_import_database_dir("").is_err());
assert!(Binder::validate_import_database_dir("dir\0x").is_err());
}
#[test]
fn import_bind_rejects_traversal_before_filesystem_access() {
let b = binder();
let stmt = akar_parser::parse("IMPORT DATABASE 'x/../../y'").unwrap();
let err = b.bind(stmt).unwrap_err();
assert!(
err.to_string().contains("path traversal"),
"bind must reject traversal, got: {err}"
);
}
#[test]
fn import_read_helper_accepts_files_inside_directory() {
let dir = unique_temp_dir("ok");
std::fs::write(
dir.join("schema.cypher"),
"CREATE NODE TABLE T(id INT64, PRIMARY KEY(id));",
)
.unwrap();
let canonical = std::fs::canonicalize(&dir).unwrap();
let content = Binder::read_file_within_dir(&canonical, "schema.cypher").unwrap();
assert!(content.contains("CREATE NODE TABLE T"));
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn import_read_helper_errors_for_missing_file() {
let dir = unique_temp_dir("missing");
let canonical = std::fs::canonicalize(&dir).unwrap();
assert!(Binder::read_file_within_dir(&canonical, "schema.cypher").is_err());
std::fs::remove_dir_all(&dir).ok();
}
#[cfg(unix)]
#[test]
fn import_read_helper_blocks_symlink_escape() {
let dir = unique_temp_dir("sym");
let outside = unique_temp_dir("outside");
let secret = outside.join("secret.txt");
std::fs::write(&secret, "top secret").unwrap();
std::os::unix::fs::symlink(&secret, dir.join("schema.cypher")).unwrap();
let canonical = std::fs::canonicalize(&dir).unwrap();
let err = Binder::read_file_within_dir(&canonical, "schema.cypher").unwrap_err();
assert!(
err.to_string().contains("outside the import directory"),
"symlink escape must be blocked, got: {err}"
);
std::fs::remove_dir_all(&dir).ok();
std::fs::remove_dir_all(&outside).ok();
}
#[test]
fn parse_type_array_with_dims_maps_to_list() {
assert_eq!(Binder::parse_type("FLOAT[384]").unwrap(), LogicalTypeID::List);
assert_eq!(Binder::parse_type("FLOAT[]").unwrap(), LogicalTypeID::List);
assert_eq!(Binder::parse_type("FLOAT[384][]").unwrap(), LogicalTypeID::List);
assert_eq!(Binder::parse_type("FLOAT").unwrap(), LogicalTypeID::Float);
assert_eq!(Binder::parse_type("INT64").unwrap(), LogicalTypeID::Int64);
}
}