1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
use crate::{
ast::{ColumnConstraint, CreateTableStmt, TableConstraint},
binder::{BindError, Binder, bound::BoundCreateTableStmt},
catalog::objects::ColumnEntry,
common::symbol::Symbol,
};
impl<'c> Binder<'c> {
/// Binds a `CREATE TABLE` statement.
///
/// # Validation performed
///
/// 1. Confirms the target schema exists in the catalog.
/// 2. If `IF NOT EXISTS` is false, confirms the table does not already exist
/// in the target database and schema.
/// 3. Confirms no column name is defined more than once.
/// 4. For each table-level constraint, confirms every referenced column
/// name is defined on this table.
/// 5. Confirms at most one `PRIMARY KEY` is specified across all
/// column-level and table-level constraints combined.
///
/// # Resolution performed
///
/// - Resolves the schema and table name. If the schema is not specified in
/// the table path, it defaults to `default_schema`.
/// - Maps AST column definitions into [`ColumnEntry`] representations,
/// folding each column's [`ColumnConstraint`]s into resolved flags:
/// - `NULL` / `NOT NULL` → `nullable`
/// - `DEFAULT <expr>` → `default`
/// - `UNIQUE` → `is_unique`
/// - `PRIMARY KEY` → `is_primary_key` (and forces `nullable = false`)
/// - `CHECK`, `REFERENCES`, `AUTO_INCREMENT` are not yet enforced —
/// see inline notes below.
/// - Folds simple table-level `PRIMARY KEY (col)` / `UNIQUE (col)`
/// constraints (single column) into the matching [`ColumnEntry`] flags.
/// - Composite `PRIMARY KEY`/`UNIQUE` (multiple columns), `CHECK`, and
/// `FOREIGN KEY` constraints are validated (referenced columns must
/// exist) and carried through unchanged into
/// [`crate::catalog::objects::TableEntry::constraints`].
///
/// # Errors
///
/// Returns [`BindError`] if any validation fails.
pub fn bind_create_table(
&self,
db: Symbol,
default_schema: Symbol,
stmt: CreateTableStmt,
) -> Result<BoundCreateTableStmt, BindError> {
let (schema, name) = stmt.name.resolve_schema_table(default_schema);
// 1. Target schema must exist
if !self.catalog.schema_exists(db, schema) {
return Err(BindError::SchemaNotFound(schema));
}
// 2. Table must not already exist (unless IF NOT EXISTS)
if !stmt.if_not_exist && self.catalog.table_exists(db, schema, name) {
return Err(BindError::TableAlreadyExists(name));
}
// Tracks whether a PRIMARY KEY has been seen yet (column-level or
// table-level) — used to enforce at most one across the whole table.
let mut has_primary_key = false;
// 3. Build ColumnEntry list, folding column-level constraints into flags
let mut columns: Vec<ColumnEntry> = Vec::with_capacity(stmt.columns.len());
for col in stmt.columns {
// Duplicate column name check
if columns.iter().any(|c| c.name == col.name) {
return Err(BindError::DuplicateColumnName(col.name));
}
let mut nullable = true; // SQL default: columns are nullable
let mut default = None;
let mut is_unique = false;
let mut is_primary_key = false;
for constraint in col.constraints {
match constraint {
ColumnConstraint::Null => {
nullable = true;
}
ColumnConstraint::NotNull => {
nullable = false;
}
ColumnConstraint::Default(expr) => {
default = Some(expr);
}
ColumnConstraint::Unique => {
is_unique = true;
}
ColumnConstraint::PrimaryKey => {
if has_primary_key {
return Err(BindError::MultiplePrimaryKeys);
}
has_primary_key = true;
is_primary_key = true;
// PRIMARY KEY implies NOT NULL
nullable = false;
}
// TODO: CHECK requires an expression evaluator to enforce
// at INSERT/UPDATE time — parsed and accepted, but not
// validated or stored on the column. Once table-level
// CHECK support lands (TableConstraint::Check below),
// column-level CHECK should be normalized into the same
// representation rather than dropped silently.
ColumnConstraint::Check(_) => {}
// TODO: REFERENCES requires cross-table catalog lookups
// (the foreign table must exist and have matching
// columns) — deferred until multi-table validation is
// supported. Parsed and accepted, not stored or enforced.
ColumnConstraint::References { .. } => {}
// TODO: AUTO_INCREMENT requires sequence/identity support
// in storage — deferred. Parsed and accepted, not stored
// or enforced.
ColumnConstraint::AutoIncrement => {}
}
}
columns.push(ColumnEntry {
name: col.name,
data_type: col.data_type,
nullable,
default,
is_unique,
is_primary_key,
});
}
// 4. Process table-level constraints
let mut table_constraints = Vec::new();
for mut constraint in stmt.constraints {
match &mut constraint {
TableConstraint::PrimaryKey {
name: cname,
columns: cols,
} => {
for col_name in cols.iter() {
if !columns.iter().any(|c| c.name == *col_name) {
return Err(BindError::ColumnNotFound(*col_name));
}
}
if has_primary_key {
return Err(BindError::MultiplePrimaryKeys);
}
has_primary_key = true;
if cols.len() == 1 {
let col_name = cols[0];
let entry = columns
.iter_mut()
.find(|c| c.name == col_name)
.expect("column existence checked above");
entry.is_primary_key = true;
entry.nullable = false;
} else {
for col_name in cols.iter() {
let entry = columns
.iter_mut()
.find(|c| c.name == *col_name)
.expect("column existence checked above");
entry.is_primary_key = true;
entry.nullable = false;
}
if cname.is_none() {
let generated = format!("{}_pkey", self.catalog.interner.resolve(name));
*cname = Some(self.catalog.interner.intern(&generated));
}
table_constraints.push(constraint);
}
}
TableConstraint::Unique {
name: cname,
columns: cols,
} => {
for col_name in cols.iter() {
if !columns.iter().any(|c| c.name == *col_name) {
return Err(BindError::ColumnNotFound(*col_name));
}
}
if cols.len() == 1 {
let col_name = cols[0];
let entry = columns
.iter_mut()
.find(|c| c.name == col_name)
.expect("column existence checked above");
entry.is_unique = true;
} else {
if cname.is_none() {
let col_part = cols
.iter()
.map(|c| self.catalog.interner.resolve(*c))
.collect::<Vec<_>>()
.join("_");
let generated =
format!("{}_{}_key", self.catalog.interner.resolve(name), col_part);
*cname = Some(self.catalog.interner.intern(&generated));
}
table_constraints.push(constraint);
}
}
TableConstraint::Check { .. } => {
table_constraints.push(constraint);
}
TableConstraint::ForeignKey { columns: cols, .. } => {
for col_name in cols.iter() {
if !columns.iter().any(|c| c.name == *col_name) {
return Err(BindError::ColumnNotFound(*col_name));
}
}
table_constraints.push(constraint);
}
}
}
Ok(BoundCreateTableStmt {
db,
schema,
name,
columns,
constraints: table_constraints,
if_not_exists: stmt.if_not_exist,
})
}
}