use crate::backend::BackendKind;
use crate::generation::{CatalogManifest, ManifestTable};
use crate::ir::filter::ComparisonOp;
use crate::ir::operations::{
ConflictStrategy, LogicalAggregate, LogicalDelete, LogicalInclude, LogicalRead,
LogicalResourceOp, LogicalSearch, LogicalWrite, ResourceKind, ResourceOpKind,
};
use crate::ir::value::LogicalValue;
use super::sql_dialect::{SqlCompiler, SqlDialect};
use super::util::resolve_include_relation;
use super::{CompileContext, CompileError, CompiledRendering, Compiler};
struct Mssql;
impl SqlDialect for Mssql {
fn quote(ident: &str) -> String {
format!("[{ident}]")
}
fn placeholder(index: usize) -> String {
format!("@P{index}")
}
fn false_literal() -> &'static str {
"1=0"
}
fn having_true_literal() -> &'static str {
"1=1"
}
fn having_false_literal() -> &'static str {
"1=0"
}
fn sql_op_for(op: ComparisonOp) -> &'static str {
match op {
ComparisonOp::Eq => "=",
ComparisonOp::Ne => "<>",
ComparisonOp::Lt => "<",
ComparisonOp::Le => "<=",
ComparisonOp::Gt => ">",
ComparisonOp::Ge => ">=",
ComparisonOp::Like
| ComparisonOp::Contains
| ComparisonOp::StartsWith
| ComparisonOp::EndsWith
| ComparisonOp::ILike => "LIKE",
}
}
fn wrap_value_for_op(op: ComparisonOp, placeholder: &str) -> String {
let escaped = format!(
r"REPLACE(REPLACE(REPLACE(REPLACE({placeholder}, '\', '\\'), '%', '\%'), '_', '\_'), '[', '\[')"
);
match op {
ComparisonOp::Contains => format!(r"'%' + {escaped} + '%' ESCAPE '\'"),
ComparisonOp::StartsWith => format!(r"{escaped} + '%' ESCAPE '\'"),
ComparisonOp::EndsWith => format!(r"'%' + {escaped} ESCAPE '\'"),
_ => placeholder.to_string(),
}
}
}
type Ms = SqlCompiler<Mssql>;
#[derive(Debug, Default, Clone, Copy)]
pub struct MssqlCompiler;
impl Compiler for MssqlCompiler {
fn kind(&self) -> BackendKind {
BackendKind::Mssql
}
fn supports_read_include(&self) -> bool {
true
}
fn compile_read(
&self,
op: &LogicalRead,
ctx: &CompileContext<'_>,
) -> Result<CompiledRendering, CompileError> {
let table = Ms::resolve_table(&op.message_type, ctx.manifest)?;
let mut params: Vec<LogicalValue> = Vec::new();
let mut select_items = match &op.projection {
Some(p) if !p.is_select_all() => p
.fields
.iter()
.map(|f| {
let col = Ms::column_for(table, f, &op.message_type)?;
Ok(format!("[{col}]"))
})
.collect::<Result<Vec<_>, CompileError>>()?,
_ => vec!["*".to_string()],
};
for include in &op.include {
select_items.push(render_belongs_to_include(
table,
ctx.manifest,
include,
&op.message_type,
)?);
}
let select = select_items.join(", ");
let mut sql = format!(
"SELECT {select} FROM [{schema}].[{table}]",
schema = table.schema,
table = table.table,
);
if let Some(filter) = &op.filter
&& let Some(body) = Ms::render_where(filter, table, &op.message_type, &mut params)?
{
sql.push_str(&format!(" WHERE {body}"));
}
if !op.sort.is_empty() {
let parts = op
.sort
.iter()
.map(|s| {
let col = Ms::column_for(table, &s.field, &op.message_type)?;
let direction = s.direction.token().to_uppercase();
let null_prefix = match s.nulls {
crate::ir::projection::NullOrder::First => {
format!("(CASE WHEN [{col}] IS NULL THEN 0 ELSE 1 END), ")
}
crate::ir::projection::NullOrder::Last => {
format!("(CASE WHEN [{col}] IS NULL THEN 1 ELSE 0 END), ")
}
crate::ir::projection::NullOrder::Default => String::new(),
};
Ok(format!("{null_prefix}[{col}] {direction}"))
})
.collect::<Result<Vec<_>, CompileError>>()?;
sql.push_str(&format!(" ORDER BY {}", parts.join(", ")));
}
if let Some(pag) = &op.pagination {
if pag.uses_cursor() {
return Err(CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "keyset_cursor",
});
}
if (pag.limit.is_some() || pag.offset.is_some_and(|o| o > 0)) && op.sort.is_empty() {
return Err(CompileError::Malformed {
reason: "T-SQL OFFSET/FETCH NEXT requires ORDER BY; add a sort clause".into(),
});
}
let offset = pag.offset.unwrap_or(0);
if pag.limit.is_some() || offset > 0 {
sql.push_str(&format!(" OFFSET {offset} ROWS"));
if let Some(limit) = pag.limit {
sql.push_str(&format!(" FETCH NEXT {limit} ROWS ONLY"));
}
}
}
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params,
})
}
fn compile_write(
&self,
op: &LogicalWrite,
ctx: &CompileContext<'_>,
) -> Result<CompiledRendering, CompileError> {
if op.records.is_empty() {
return Err(CompileError::Malformed {
reason: "LogicalWrite::records must be non-empty".into(),
});
}
let table = Ms::resolve_table(&op.message_type, ctx.manifest)?;
let mut params: Vec<LogicalValue> = Vec::new();
let first = &op.records[0];
let columns: Vec<&str> = first
.keys()
.map(|k| Ms::column_for(table, k, &op.message_type))
.collect::<Result<Vec<_>, _>>()?;
let column_list = columns
.iter()
.map(|c| format!("[{c}]"))
.collect::<Vec<_>>()
.join(", ");
let mut value_rows = Vec::with_capacity(op.records.len());
for (idx, record) in op.records.iter().enumerate() {
if record.len() != first.len() || !first.keys().all(|k| record.contains_key(k)) {
return Err(CompileError::Malformed {
reason: format!(
"record {idx} has different field set than record 0; \
all records in one LogicalWrite must share the same fields"
),
});
}
let row = first
.keys()
.map(|k| Ms::push_param(&mut params, record[k].clone()))
.collect::<Vec<_>>()
.join(", ");
value_rows.push(format!("({row})"));
}
match &op.conflict {
ConflictStrategy::Error => {
let sql = format!(
"INSERT INTO [{schema}].[{table}] ({column_list}) VALUES {values}",
schema = table.schema,
table = table.table,
values = value_rows.join(", "),
);
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params,
})
}
ConflictStrategy::Ignore => {
if table.primary_key.is_empty() {
return Err(CompileError::Malformed {
reason: format!(
"INSERT IGNORE on '{}' requires a primary key in the manifest",
op.message_type
),
});
}
if op.records.len() != 1 {
return Err(CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "batch_insert_ignore",
});
}
let on_clause = build_merge_on_clause(table, &op.message_type, &table.primary_key)?;
let sql = format!(
"MERGE INTO [{schema}].[{table}] AS target \
USING (VALUES {values}) AS source({column_list}) \
ON {on_clause} \
WHEN NOT MATCHED THEN \
INSERT ({column_list}) VALUES ({source_values});",
schema = table.schema,
table = table.table,
values = value_rows.join(", "),
source_values = columns
.iter()
.map(|c| format!("source.[{c}]"))
.collect::<Vec<_>>()
.join(", "),
);
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params,
})
}
ConflictStrategy::Replace | ConflictStrategy::Update { .. } => {
let conflict_fields: Vec<String> = match op.conflict.conflict_target() {
Some(cols) => cols.to_vec(),
None => {
if table.primary_key.is_empty() {
return Err(CompileError::Malformed {
reason: format!(
"upsert on '{}' requires a primary key in the manifest or \
an explicit conflict_on target",
op.message_type
),
});
}
table.primary_key.clone()
}
};
if op.records.len() != 1 {
return Err(CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "batch_upsert",
});
}
let on_clause = build_merge_on_clause(table, &op.message_type, &conflict_fields)?;
let target_cols: Vec<&str> = match &op.conflict {
ConflictStrategy::Update { fields, .. } => fields
.iter()
.map(|f| Ms::column_for(table, f, &op.message_type))
.collect::<Result<Vec<_>, _>>()?,
ConflictStrategy::Replace => columns.clone(),
_ => unreachable!(),
};
let set_clause = target_cols
.iter()
.map(|c| format!("target.[{c}] = source.[{c}]"))
.collect::<Vec<_>>()
.join(", ");
let sql = format!(
"MERGE INTO [{schema}].[{table}] AS target \
USING (VALUES {values}) AS source({column_list}) \
ON {on_clause} \
WHEN MATCHED THEN UPDATE SET {set_clause} \
WHEN NOT MATCHED THEN \
INSERT ({column_list}) VALUES ({source_values});",
schema = table.schema,
table = table.table,
values = value_rows.join(", "),
source_values = columns
.iter()
.map(|c| format!("source.[{c}]"))
.collect::<Vec<_>>()
.join(", "),
);
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params,
})
}
}
}
fn compile_delete(
&self,
op: &LogicalDelete,
ctx: &CompileContext<'_>,
) -> Result<CompiledRendering, CompileError> {
let table = Ms::resolve_table(&op.message_type, ctx.manifest)?;
let mut params: Vec<LogicalValue> = Vec::new();
let body = Ms::render_where(&op.filter, table, &op.message_type, &mut params)?.ok_or_else(
|| CompileError::Malformed {
reason: "LogicalDelete::filter cannot be empty; use Drop resource to truncate"
.into(),
},
)?;
if body == "1=0" {
return Err(CompileError::Malformed {
reason: "LogicalDelete::filter resolves to FALSE; refusing no-op delete".into(),
});
}
let sql = format!(
"DELETE FROM [{schema}].[{table}] WHERE {body}",
schema = table.schema,
table = table.table,
);
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params,
})
}
fn compile_aggregate(
&self,
op: &LogicalAggregate,
ctx: &CompileContext<'_>,
) -> Result<CompiledRendering, CompileError> {
if op.aggregates.is_empty() {
return Err(CompileError::Malformed {
reason: "LogicalAggregate::aggregates must be non-empty".into(),
});
}
super::util::validate_aggregate_aliases(&op.aggregates)?;
let table = Ms::resolve_table(&op.message_type, ctx.manifest)?;
let group_names: Vec<&str> = op
.group_by
.iter()
.map(|f| Ms::column_for(table, f, &op.message_type))
.collect::<Result<Vec<_>, _>>()?;
super::util::validate_no_groupby_alias_collision(&group_names, &op.aggregates)?;
let mut params: Vec<LogicalValue> = Vec::new();
let mut select_parts: Vec<String> = Vec::new();
let group_columns: Vec<String> = op
.group_by
.iter()
.map(|f| {
let col = Ms::column_for(table, f, &op.message_type)?;
Ok::<_, CompileError>(format!("[{col}]"))
})
.collect::<Result<Vec<_>, _>>()?;
for col in &group_columns {
select_parts.push(col.clone());
}
for agg in &op.aggregates {
select_parts.push(Ms::render_aggregate(agg, table, &op.message_type)?);
}
let mut sql = format!(
"SELECT {sel} FROM [{schema}].[{table}]",
sel = select_parts.join(", "),
schema = table.schema,
table = table.table,
);
let user_body = match &op.filter {
Some(filter) => Ms::render_where(filter, table, &op.message_type, &mut params)?,
None => None,
};
let ctx_body = Ms::context_predicates(table, ctx, &mut params);
match (user_body, ctx_body) {
(Some(user), Some(scope)) => sql.push_str(&format!(" WHERE ({user}) AND {scope}")),
(Some(user), None) => sql.push_str(&format!(" WHERE {user}")),
(None, Some(scope)) => sql.push_str(&format!(" WHERE {scope}")),
(None, None) => {}
}
if !group_columns.is_empty() {
sql.push_str(&format!(" GROUP BY {}", group_columns.join(", ")));
}
if let Some(having) = &op.having {
let body = Ms::render_having(having, op, table, &mut params)?;
sql.push_str(&format!(" HAVING {body}"));
}
if !op.sort.is_empty() {
let parts = op
.sort
.iter()
.map(|s| {
let token = Ms::resolve_sort_field(&s.field, op, table)?;
let direction = s.direction.token().to_uppercase();
Ok::<_, CompileError>(format!("{token} {direction}"))
})
.collect::<Result<Vec<_>, _>>()?;
sql.push_str(&format!(" ORDER BY {}", parts.join(", ")));
}
if let Some(pag) = &op.pagination {
if pag.uses_cursor() {
return Err(CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "keyset_cursor",
});
}
if (pag.limit.is_some() || pag.offset.is_some_and(|o| o > 0)) && op.sort.is_empty() {
return Err(CompileError::Malformed {
reason: "T-SQL OFFSET/FETCH NEXT requires ORDER BY".into(),
});
}
let offset = pag.offset.unwrap_or(0);
if pag.limit.is_some() || offset > 0 {
sql.push_str(&format!(" OFFSET {offset} ROWS"));
if let Some(limit) = pag.limit {
sql.push_str(&format!(" FETCH NEXT {limit} ROWS ONLY"));
}
}
}
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params,
})
}
fn compile_search(
&self,
op: &LogicalSearch,
ctx: &CompileContext<'_>,
) -> Result<CompiledRendering, CompileError> {
let text = op.text_query.as_deref().ok_or_else(|| {
CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "vector_search",
}
})?;
if text.trim().is_empty() {
return Err(CompileError::Malformed {
reason: "text_query must be non-empty for T-SQL CONTAINS()".into(),
});
}
let table = Ms::resolve_table(&op.message_type, ctx.manifest)?;
let mut params: Vec<LogicalValue> = Vec::new();
Ms::push_param(&mut params, LogicalValue::String(text.to_string()));
let mut sql = format!(
"SELECT TOP({top_k}) * FROM [{schema}].[{table}] WHERE CONTAINS(*, @P1)",
top_k = op.top_k,
schema = table.schema,
table = table.table,
);
if let Some(filter) = &op.filter
&& let Some(body) = Ms::render_where(filter, table, &op.message_type, &mut params)?
{
sql.push_str(&format!(" AND {body}"));
}
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params,
})
}
fn compile_resource_op(
&self,
op: &LogicalResourceOp,
_ctx: &CompileContext<'_>,
) -> Result<CompiledRendering, CompileError> {
if !matches!(op.resource_kind, ResourceKind::Table | ResourceKind::Index) {
return Err(CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "non_table_resource",
});
}
let sql = match (op.op, op.resource_kind) {
(ResourceOpKind::Ensure, ResourceKind::Table) => {
let spec = op.spec.as_ref().ok_or_else(|| CompileError::Malformed {
reason: "Ensure Table requires a spec with column definitions".into(),
})?;
let schema = spec.get("schema").and_then(|v| v.as_str()).unwrap_or("dbo");
let cols = spec
.get("columns")
.and_then(|v| v.as_array())
.ok_or_else(|| CompileError::Malformed {
reason: "Ensure Table spec.columns must be an array".into(),
})?;
let mut column_defs = Vec::with_capacity(cols.len());
let mut pk_cols: Vec<String> = Vec::new();
for c in cols {
let name = c.get("name").and_then(|v| v.as_str()).ok_or_else(|| {
CompileError::Malformed {
reason: "column missing 'name'".into(),
}
})?;
let ty = c.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
CompileError::Malformed {
reason: "column missing 'type'".into(),
}
})?;
let not_null = c.get("not_null").and_then(|v| v.as_bool()).unwrap_or(false);
let pk = c
.get("primary_key")
.and_then(|v| v.as_bool())
.unwrap_or(false);
if pk {
pk_cols.push(format!("[{name}]"));
}
let null_clause = if not_null { " NOT NULL" } else { " NULL" };
column_defs.push(format!("[{name}] {ty}{null_clause}"));
}
if !pk_cols.is_empty() {
column_defs.push(format!(
"CONSTRAINT [PK_{name}] PRIMARY KEY ({pk_list})",
name = op.resource_name,
pk_list = pk_cols.join(", "),
));
}
format!(
"IF NOT EXISTS (SELECT 1 FROM sys.tables t \
JOIN sys.schemas s ON t.schema_id = s.schema_id \
WHERE s.name = '{schema}' AND t.name = '{name}') \
CREATE TABLE [{schema}].[{name}] ({defs});",
name = op.resource_name,
defs = column_defs.join(", "),
)
}
(ResourceOpKind::Drop, ResourceKind::Table) => {
let schema = op
.spec
.as_ref()
.and_then(|s| s.get("schema"))
.and_then(|v| v.as_str())
.unwrap_or("dbo");
format!(
"IF OBJECT_ID('[{schema}].[{name}]', 'U') IS NOT NULL \
DROP TABLE [{schema}].[{name}];",
name = op.resource_name,
)
}
(ResourceOpKind::List, ResourceKind::Table) => {
let schema = op
.spec
.as_ref()
.and_then(|s| s.get("schema"))
.and_then(|v| v.as_str())
.unwrap_or("dbo");
format!(
"SELECT t.name FROM sys.tables t \
JOIN sys.schemas s ON t.schema_id = s.schema_id \
WHERE s.name = '{schema}' ORDER BY t.name"
)
}
(ResourceOpKind::Ensure, ResourceKind::Index) => {
let spec = op.spec.as_ref().ok_or_else(|| CompileError::Malformed {
reason: "Ensure Index requires a spec".into(),
})?;
let schema = spec.get("schema").and_then(|v| v.as_str()).unwrap_or("dbo");
let tbl = spec.get("table").and_then(|v| v.as_str()).ok_or_else(|| {
CompileError::Malformed {
reason: "Ensure Index spec missing 'table'".into(),
}
})?;
let cols = spec
.get("columns")
.and_then(|v| v.as_array())
.ok_or_else(|| CompileError::Malformed {
reason: "Ensure Index spec.columns must be an array".into(),
})?;
let col_list = cols
.iter()
.filter_map(|c| c.as_str())
.map(|c| format!("[{c}]"))
.collect::<Vec<_>>()
.join(", ");
let unique = spec
.get("unique")
.and_then(|v| v.as_bool())
.unwrap_or(false);
let kind = if unique { "UNIQUE INDEX" } else { "INDEX" };
format!(
"IF NOT EXISTS (SELECT 1 FROM sys.indexes \
WHERE name = '{name}' AND object_id = OBJECT_ID('[{schema}].[{tbl}]')) \
CREATE {kind} [{name}] ON [{schema}].[{tbl}] ({col_list});",
name = op.resource_name,
)
}
(ResourceOpKind::Drop, ResourceKind::Index) => {
let spec = op.spec.as_ref().ok_or_else(|| CompileError::Malformed {
reason: "Drop Index requires a spec with 'table'".into(),
})?;
let schema = spec.get("schema").and_then(|v| v.as_str()).unwrap_or("dbo");
let tbl = spec.get("table").and_then(|v| v.as_str()).ok_or_else(|| {
CompileError::Malformed {
reason: "Drop Index spec missing 'table'".into(),
}
})?;
format!("DROP INDEX [{}] ON [{schema}].[{tbl}];", op.resource_name)
}
(ResourceOpKind::List, ResourceKind::Index) => {
let spec = op.spec.as_ref().ok_or_else(|| CompileError::Malformed {
reason: "List Index requires a spec with 'table'".into(),
})?;
let schema = spec.get("schema").and_then(|v| v.as_str()).unwrap_or("dbo");
let tbl = spec.get("table").and_then(|v| v.as_str()).ok_or_else(|| {
CompileError::Malformed {
reason: "List Index spec missing 'table'".into(),
}
})?;
format!(
"SELECT name FROM sys.indexes WHERE object_id = OBJECT_ID('[{schema}].[{tbl}]') ORDER BY name"
)
}
_ => {
return Err(CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "unhandled_resource_op",
});
}
};
Ok(CompiledRendering::Sql {
backend: BackendKind::Mssql,
statement: sql,
params: Vec::new(),
})
}
}
fn render_belongs_to_include(
table: &ManifestTable,
manifest: &CatalogManifest,
include: &LogicalInclude,
message_type: &str,
) -> Result<String, CompileError> {
let relation = resolve_include_relation(table, manifest, include, message_type)?;
let alias = format!("_udb_include_{}", relation.name);
let predicates = relation
.local_columns
.iter()
.zip(relation.target_columns.iter())
.map(|(local, target)| {
format!(
"[{alias}].[{target}] = [{table}].[{local}]",
table = table.table
)
})
.collect::<Vec<_>>()
.join(" AND ");
let wrapper = if relation.many {
""
} else {
", WITHOUT_ARRAY_WRAPPER"
};
Ok(format!(
"(SELECT [{alias}].* FROM [{schema}].[{target_table}] [{alias}] \
WHERE {predicates} FOR JSON PATH{wrapper}) AS [{name}]",
schema = relation.target.schema,
target_table = relation.target.table,
name = relation.name,
))
}
fn build_merge_on_clause(
table: &ManifestTable,
message_type: &str,
conflict_fields: &[String],
) -> Result<String, CompileError> {
let parts = conflict_fields
.iter()
.map(|pk| {
let col = Ms::column_for(table, pk, message_type)?;
Ok(format!("target.[{col}] = source.[{col}]"))
})
.collect::<Result<Vec<_>, CompileError>>()?;
Ok(parts.join(" AND "))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::generation::{CatalogManifest, ManifestColumn, ManifestForeignKey, ManifestTable};
use crate::ir::filter::{ComparisonOp, LogicalFilter};
use crate::ir::operations::{
AggregateExpr, AggregateFunc, ConflictStrategy, LogicalAggregate, LogicalDelete,
LogicalRead, LogicalRecord, LogicalResourceOp, LogicalSearch, LogicalWrite, ResourceKind,
ResourceOpKind,
};
use crate::ir::projection::{LogicalPagination, LogicalSort, SortDirection};
use crate::ir::value::LogicalValue;
use serde_json::json;
fn fixture() -> CatalogManifest {
let table = ManifestTable {
message_name: "acme.billing.v1.Customer".into(),
schema: "billing".into(),
table: "customers".into(),
primary_key: vec!["id".into()],
columns: vec![
ManifestColumn {
field_name: "id".into(),
column_name: "id".into(),
proto_type: "string".into(),
sql_type: "nvarchar(64)".into(),
is_primary: true,
..Default::default()
},
ManifestColumn {
field_name: "name".into(),
column_name: "name".into(),
proto_type: "string".into(),
sql_type: "nvarchar(255)".into(),
..Default::default()
},
ManifestColumn {
field_name: "amount".into(),
column_name: "amount".into(),
proto_type: "int64".into(),
sql_type: "bigint".into(),
..Default::default()
},
],
..Default::default()
};
CatalogManifest {
tables: vec![table],
..Default::default()
}
}
fn relation_fixture() -> CatalogManifest {
CatalogManifest {
tables: vec![
ManifestTable {
message_name: "Invoice".into(),
proto_package: "billing.v1".into(),
schema: "billing".into(),
table: "invoices".into(),
primary_key: vec!["invoice_id".into()],
columns: vec![
ManifestColumn {
field_name: "invoice_id".into(),
column_name: "invoice_id".into(),
proto_type: "string".into(),
sql_type: "nvarchar(64)".into(),
is_primary: true,
..Default::default()
},
ManifestColumn {
field_name: "customer_id".into(),
column_name: "customer_id".into(),
proto_type: "string".into(),
sql_type: "nvarchar(64)".into(),
..Default::default()
},
],
foreign_keys: vec![ManifestForeignKey {
name: "fk_invoice_customer".into(),
columns: vec!["customer_id".into()],
ref_schema: "crm".into(),
ref_table: "customers".into(),
ref_columns: vec!["customer_id".into()],
..Default::default()
}],
..Default::default()
},
ManifestTable {
message_name: "Customer".into(),
proto_package: "crm.v1".into(),
schema: "crm".into(),
table: "customers".into(),
primary_key: vec!["customer_id".into()],
columns: vec![
ManifestColumn {
field_name: "customer_id".into(),
column_name: "customer_id".into(),
proto_type: "string".into(),
sql_type: "nvarchar(64)".into(),
is_primary: true,
..Default::default()
},
ManifestColumn {
field_name: "name".into(),
column_name: "name".into(),
proto_type: "string".into(),
sql_type: "nvarchar(255)".into(),
..Default::default()
},
],
..Default::default()
},
],
..Default::default()
}
}
fn sql(rendering: CompiledRendering) -> (String, Vec<LogicalValue>) {
match rendering {
CompiledRendering::Sql {
statement, params, ..
} => (statement, params),
other => panic!("expected Sql, got {other:?}"),
}
}
#[test]
fn select_uses_brackets_and_at_p_placeholders() {
let m = fixture();
let ctx = CompileContext::new(&m);
let read = LogicalRead::message("acme.billing.v1.Customer").with_filter(
LogicalFilter::Comparison {
field: "id".into(),
op: ComparisonOp::Eq,
value: LogicalValue::String("abc".into()),
},
);
let (statement, params) = sql(MssqlCompiler.compile_read(&read, &ctx).unwrap());
assert_eq!(
statement,
"SELECT * FROM [billing].[customers] WHERE [id] = @P1"
);
assert_eq!(params, vec![LogicalValue::String("abc".into())]);
}
#[test]
fn include_lowers_fk_belongs_to_as_for_json_subselect() {
let m = relation_fixture();
let ctx = CompileContext::new(&m);
let read = LogicalRead::message("billing.v1.Invoice").with_include("customer");
let (statement, params) = sql(MssqlCompiler.compile_read(&read, &ctx).unwrap());
assert!(params.is_empty());
assert!(
statement.contains(
"(SELECT [_udb_include_customer].* FROM [crm].[customers] \
[_udb_include_customer] WHERE [_udb_include_customer].[customer_id] = \
[invoices].[customer_id] FOR JSON PATH, WITHOUT_ARRAY_WRAPPER) AS [customer]"
),
"include must lower through the manifest FK; got: {statement}"
);
}
#[test]
fn pagination_without_order_by_is_rejected() {
let m = fixture();
let ctx = CompileContext::new(&m);
let read = LogicalRead::message("acme.billing.v1.Customer")
.with_pagination(LogicalPagination::limit(10));
let err = MssqlCompiler.compile_read(&read, &ctx).unwrap_err();
assert!(matches!(err, CompileError::Malformed { .. }));
}
#[test]
fn pagination_with_order_by_emits_offset_fetch_next() {
let m = fixture();
let ctx = CompileContext::new(&m);
let read = LogicalRead::message("acme.billing.v1.Customer")
.with_sort(vec![LogicalSort {
field: "id".into(),
direction: SortDirection::Asc,
nulls: crate::ir::projection::NullOrder::Default,
}])
.with_pagination(LogicalPagination::page(5, 10));
let (statement, _) = sql(MssqlCompiler.compile_read(&read, &ctx).unwrap());
assert!(statement.contains("ORDER BY [id] ASC"));
assert!(statement.contains("OFFSET 5 ROWS"));
assert!(statement.contains("FETCH NEXT 10 ROWS ONLY"));
}
#[test]
fn upsert_emits_merge_with_update_and_insert() {
let m = fixture();
let ctx = CompileContext::new(&m);
let mut rec = LogicalRecord::new();
rec.insert("id".into(), LogicalValue::String("abc".into()));
rec.insert("name".into(), LogicalValue::String("Alice".into()));
let write = LogicalWrite {
message_type: "acme.billing.v1.Customer".into(),
records: vec![rec],
conflict: ConflictStrategy::update(vec!["name".into()]),
return_fields: vec![],
};
let (statement, _) = sql(MssqlCompiler.compile_write(&write, &ctx).unwrap());
assert!(statement.starts_with("MERGE INTO [billing].[customers]"));
assert!(statement.contains("ON target.[id] = source.[id]"));
assert!(statement.contains("WHEN MATCHED THEN UPDATE SET target.[name] = source.[name]"));
assert!(statement.contains("WHEN NOT MATCHED THEN"));
}
#[test]
fn ignore_emits_merge_with_insert_only() {
let m = fixture();
let ctx = CompileContext::new(&m);
let mut rec = LogicalRecord::new();
rec.insert("id".into(), LogicalValue::String("abc".into()));
let write = LogicalWrite {
message_type: "acme.billing.v1.Customer".into(),
records: vec![rec],
conflict: ConflictStrategy::Ignore,
return_fields: vec![],
};
let (statement, _) = sql(MssqlCompiler.compile_write(&write, &ctx).unwrap());
assert!(statement.contains("WHEN NOT MATCHED THEN"));
assert!(!statement.contains("WHEN MATCHED"));
}
#[test]
fn batch_upsert_rejected_with_typed_error() {
let m = fixture();
let ctx = CompileContext::new(&m);
let mut rec_a = LogicalRecord::new();
rec_a.insert("id".into(), LogicalValue::String("a".into()));
let mut rec_b = LogicalRecord::new();
rec_b.insert("id".into(), LogicalValue::String("b".into()));
let write = LogicalWrite {
message_type: "acme.billing.v1.Customer".into(),
records: vec![rec_a, rec_b],
conflict: ConflictStrategy::Replace,
return_fields: vec![],
};
let err = MssqlCompiler.compile_write(&write, &ctx).unwrap_err();
assert!(matches!(
err,
CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "batch_upsert"
}
));
}
#[test]
fn aggregate_group_by_emits_bracketed_columns() {
let m = fixture();
let ctx = CompileContext::new(&m);
let agg = LogicalAggregate {
message_type: "acme.billing.v1.Customer".into(),
filter: None,
group_by: vec!["name".into()],
aggregates: vec![AggregateExpr {
func: AggregateFunc::Sum,
field: "amount".into(),
alias: "total".into(),
}],
having: None,
sort: vec![],
pagination: None,
};
let (statement, _) = sql(MssqlCompiler.compile_aggregate(&agg, &ctx).unwrap());
assert_eq!(
statement,
"SELECT [name], SUM([amount]) AS [total] FROM [billing].[customers] GROUP BY [name]"
);
}
#[test]
fn search_emits_contains_with_top() {
let m = fixture();
let ctx = CompileContext::new(&m);
let search = LogicalSearch {
message_type: "acme.billing.v1.Customer".into(),
vector: None,
text_query: Some("alice".into()),
filter: None,
top_k: 10,
score_threshold: None,
require_hybrid: false,
with_vector: false,
with_payload: true,
};
let (statement, _) = sql(MssqlCompiler.compile_search(&search, &ctx).unwrap());
assert!(statement.starts_with("SELECT TOP(10) * FROM [billing].[customers]"));
assert!(statement.contains("WHERE CONTAINS(*, @P1)"));
}
#[test]
fn search_vector_only_rejected() {
let m = fixture();
let ctx = CompileContext::new(&m);
let search = LogicalSearch {
message_type: "acme.billing.v1.Customer".into(),
vector: Some(vec![0.0]),
text_query: None,
filter: None,
top_k: 5,
score_threshold: None,
require_hybrid: false,
with_vector: false,
with_payload: true,
};
let err = MssqlCompiler.compile_search(&search, &ctx).unwrap_err();
assert!(matches!(
err,
CompileError::OperatorUnsupported {
backend: BackendKind::Mssql,
op: "vector_search"
}
));
}
#[test]
fn resource_op_create_table_uses_object_id_guard() {
let m = fixture();
let ctx = CompileContext::new(&m);
let op = LogicalResourceOp {
op: ResourceOpKind::Ensure,
resource_kind: ResourceKind::Table,
resource_name: "orders".into(),
spec: Some(json!({
"schema": "billing",
"columns": [
{"name": "id", "type": "bigint", "not_null": true, "primary_key": true}
]
})),
};
let (statement, _) = sql(MssqlCompiler.compile_resource_op(&op, &ctx).unwrap());
assert!(statement.starts_with("IF NOT EXISTS"));
assert!(statement.contains("sys.tables"));
assert!(statement.contains("CREATE TABLE [billing].[orders]"));
assert!(statement.contains("CONSTRAINT [PK_orders] PRIMARY KEY"));
}
#[test]
fn delete_with_filter_emits_bracketed_where() {
let m = fixture();
let ctx = CompileContext::new(&m);
let del = LogicalDelete {
message_type: "acme.billing.v1.Customer".into(),
filter: LogicalFilter::Comparison {
field: "id".into(),
op: ComparisonOp::Eq,
value: LogicalValue::String("abc".into()),
},
return_fields: vec![],
};
let (statement, _) = sql(MssqlCompiler.compile_delete(&del, &ctx).unwrap());
assert_eq!(
statement,
"DELETE FROM [billing].[customers] WHERE [id] = @P1"
);
}
}