Implement SQL Server 2025 Parity Features (Group 1)

crates/iridium_core/src/ast/expressions.rs

@@ -67,6 +67,7 @@ pub enum Expr {
     Like {
         expr: Box<Expr>,
         pattern: Box<Expr>,
+        escape: Option<Box<Expr>>,
         negated: bool,
     },
     WindowFunction {

crates/iridium_core/src/ast/statements/ddl.rs

@@ -47,7 +47,21 @@ pub struct DropTableStmt {
 pub struct CreateIndexStmt {
     pub name: ObjectName,
     pub table: ObjectName,
-    pub columns: Vec<String>,
+    pub is_unique: bool,
+    pub is_clustered: bool,
+    pub columns: Vec<IndexColumnSpec>,
+    pub options: Vec<IndexOptionSpec>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct IndexColumnSpec {
+    pub name: String,
+    pub is_desc: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum IndexOptionSpec {
+    FillFactor(u8),
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -129,6 +143,8 @@ pub struct ColumnSpec {
     pub check_constraint_name: Option<String>,
     pub computed_expr: Option<Expr>,
     pub foreign_key: Option<ForeignKeyRef>,
+    pub collation: Option<String>,
+    pub is_clustered: bool,
     pub ansi_padding_on: bool,
 }
 
@@ -169,10 +185,12 @@ pub enum TableConstraintSpec {
     },
     PrimaryKey {
         name: String,
-        columns: Vec<String>,
+        columns: Vec<IndexColumnSpec>,
+        is_clustered: bool,
     },
     Unique {
         name: String,
-        columns: Vec<String>,
+        columns: Vec<IndexColumnSpec>,
+        is_clustered: bool,
     },
 }

crates/iridium_core/src/catalog/mod.rs

@@ -70,6 +70,8 @@ pub struct ColumnDef {
     pub check: Option<Expr>,
     pub check_constraint_name: Option<String>,
     pub computed_expr: Option<Expr>,
+    pub collation: Option<String>,
+    pub is_clustered: bool,
     #[serde(default = "default_ansi_padding_on")]
     pub ansi_padding_on: bool,
 }

crates/iridium_core/src/executor/database/dispatch.rs

@@ -101,8 +101,17 @@ fn handle_session_statement<C: Catalog, S: Storage>(
             Err(e) => Some(Err(e)),
         }
     } else if let Statement::Session(SessionStatement::SetIdentityInsert(ref id_stmt)) = stmt {
+        let storage_guard = state.storage.read();
+        let (catalog, _) = storage_guard.get_refs();
+        let schema = id_stmt.table.schema_or_dbo();
+        if catalog.find_table(schema, &id_stmt.table.name).is_none() {
+            return Some(Err(DbError::table_not_found(schema, &id_stmt.table.name)));
+        }
+
         let table_name = normalize_identifier(&id_stmt.table.name);
         if id_stmt.on {
+            // SQL Server only allows one table to have IDENTITY_INSERT ON at a time in a session.
+            session_options.identity_insert.clear();
             session_options.identity_insert.insert(table_name);
         } else {
             session_options.identity_insert.remove(&table_name);

crates/iridium_core/src/executor/evaluator.rs

@@ -261,9 +261,18 @@ fn eval_expr_inner(
         Expr::Like {
             expr: like_expr,
             pattern,
+            escape,
             negated,
         } => eval_like(
-            like_expr, pattern, *negated, row, ctx, catalog, storage, clock,
+            like_expr,
+            pattern,
+            escape.as_deref(),
+            *negated,
+            row,
+            ctx,
+            catalog,
+            storage,
+            clock,
         ),
         Expr::Subquery(stmt) => eval_scalar_subquery(stmt, row, ctx, catalog, storage, clock),
         Expr::Exists { subquery, negated } => {

crates/iridium_core/src/executor/identifier.rs

@@ -23,14 +23,21 @@ pub(crate) fn resolve_identifier(
         }
     }
 
+    let is_identity_col = name.eq_ignore_ascii_case("IDENTITYCOL");
+
     let mut matches: Vec<(usize, Value)> = Vec::new();
     for (binding_idx, binding) in row.iter().enumerate() {
-        if let Some(col_idx) = binding
-            .table
-            .columns
-            .iter()
-            .position(|c| c.name.eq_ignore_ascii_case(name))
-        {
+        let col_idx = if is_identity_col {
+            binding.table.columns.iter().position(|c| c.identity.is_some())
+        } else {
+            binding
+                .table
+                .columns
+                .iter()
+                .position(|c| c.name.eq_ignore_ascii_case(name))
+        };
+
+        if let Some(col_idx) = col_idx {
             let value = binding
                 .row
                 .as_ref()
@@ -43,12 +50,16 @@ pub(crate) fn resolve_identifier(
     if matches.is_empty() {
         for apply_row in ctx.row.apply_stack.iter().rev() {
             for binding in apply_row.iter() {
-                if let Some(col_idx) = binding
-                    .table
-                    .columns
-                    .iter()
-                    .position(|c| c.name.eq_ignore_ascii_case(name))
-                {
+                let col_idx = if is_identity_col {
+                    binding.table.columns.iter().position(|c| c.identity.is_some())
+                } else {
+                    binding
+                        .table
+                        .columns
+                        .iter()
+                        .position(|c| c.name.eq_ignore_ascii_case(name))
+                };
+                if let Some(col_idx) = col_idx {
                     let value = binding
                         .row
                         .as_ref()
@@ -66,12 +77,16 @@ pub(crate) fn resolve_identifier(
     if matches.is_empty() {
         for outer_row in ctx.row.outer_stack.iter().rev() {
             for binding in outer_row.iter() {
-                if let Some(col_idx) = binding
-                    .table
-                    .columns
-                    .iter()
-                    .position(|c| c.name.eq_ignore_ascii_case(name))
-                {
+                let col_idx = if is_identity_col {
+                    binding.table.columns.iter().position(|c| c.identity.is_some())
+                } else {
+                    binding
+                        .table
+                        .columns
+                        .iter()
+                        .position(|c| c.name.eq_ignore_ascii_case(name))
+                };
+                if let Some(col_idx) = col_idx {
                     let value = binding
                         .row
                         .as_ref()
@@ -109,6 +124,7 @@ pub(crate) fn resolve_qualified_identifier(
 
     let table_name = &parts[0];
     let column_name = &parts[1];
+    let is_identity_col = column_name.eq_ignore_ascii_case("IDENTITYCOL");
 
     let search_row = |row: &[ContextTable]| -> Result<Option<Value>, DbError> {
         for binding in row {
@@ -119,11 +135,15 @@ pub(crate) fn resolve_qualified_identifier(
                     .iter()
                     .any(|a| a.eq_ignore_ascii_case(table_name))
             {
-                let idx = binding
-                    .table
-                    .columns
-                    .iter()
-                    .position(|c| c.name.eq_ignore_ascii_case(column_name));
+                let idx = if is_identity_col {
+                    binding.table.columns.iter().position(|c| c.identity.is_some())
+                } else {
+                    binding
+                        .table
+                        .columns
+                        .iter()
+                        .position(|c| c.name.eq_ignore_ascii_case(column_name))
+                };
 
                 if let Some(idx) = idx {
                     return Ok(Some(

crates/iridium_core/src/executor/metadata/mod.rs

@@ -61,6 +61,8 @@ pub(super) fn virtual_table_def(name: &str, cols: Vec<(&str, DataType, bool)>) -
                 check: None,
                 check_constraint_name: None,
                 computed_expr: None,
+                collation: None,
+                is_clustered: false,
                 ansi_padding_on: true,
             })
             .collect(),

crates/iridium_core/src/executor/mutation/insert.rs

@@ -284,7 +284,7 @@ impl<'a> MutationExecutor<'a> {
                 }
             } else if col.identity.is_some() {
                 let table_upper = normalize_identifier(&table.name);
-                if !ctx.session.identity_insert.contains(&table_upper) {
+                if !ctx.options.identity_insert.contains(&table_upper) {
                     return Err(DbError::Execution(format!(
                         "Cannot insert explicit value for identity column '{}' in table '{}' when IDENTITY_INSERT is set to OFF.",
                         col.name, table.name

crates/iridium_core/src/executor/predicates.rs

@@ -132,6 +132,7 @@ pub(crate) fn eval_between(
 pub(crate) fn eval_like(
     like_expr: &Expr,
     pattern: &Expr,
+    escape: Option<&Expr>,
     negated: bool,
     row: &[ContextTable],
     ctx: &mut ExecutionContext,
@@ -148,40 +149,66 @@ pub(crate) fn eval_like(
 
     let s = val.to_string_value();
     let p = pat.to_string_value();
-    let matched = like_match(&s, &p);
+    let esc = match escape {
+        Some(e) => {
+            let ev = eval_expr(e, row, ctx, catalog, storage, clock)?;
+            if ev.is_null() {
+                return Ok(Value::Null);
+            }
+            Some(ev.to_string_value())
+        }
+        None => None,
+    };
+    let matched = like_match(&s, &p, esc.as_deref());
     Ok(Value::Bit(if negated { !matched } else { matched }))
 }
 
-fn like_match(s: &str, pattern: &str) -> bool {
+fn like_match(s: &str, pattern: &str, escape: Option<&str>) -> bool {
     let s: Vec<char> = s.to_ascii_uppercase().chars().collect();
-    let p: Vec<char> = pattern.to_ascii_uppercase().chars().collect();
+    let p_raw: Vec<char> = pattern.to_ascii_uppercase().chars().collect();
+    let esc_char = escape.and_then(|e| e.chars().next().map(|c| c.to_ascii_uppercase()));
+
+    let mut p = Vec::new();
+    let mut escaped = Vec::new();
+    let mut i = 0;
+    while i < p_raw.len() {
+        if let Some(ec) = esc_char {
+            if p_raw[i] == ec && i + 1 < p_raw.len() {
+                p.push(p_raw[i + 1]);
+                escaped.push(true);
+                i += 2;
+                continue;
+            }
+        }
+        p.push(p_raw[i]);
+        escaped.push(false);
+        i += 1;
+    }
+
     let sn = s.len();
     let pn = p.len();
 
-    // dp[j] = whether s[0..i] matches p[0..j]
     let mut dp = vec![false; pn + 1];
     dp[0] = true;
-    // leading '%' can match empty string
     for j in 0..pn {
-        if p[j] == '%' {
+        if p[j] == '%' && !escaped[j] {
             dp[j + 1] = dp[j];
         } else {
             break;
         }
     }
 
-    for i in 0..sn {
-        let mut prev = dp[0]; // dp_prev[0] (previous row, col 0)
+    for i_s in 0..sn {
+        let mut prev = dp[0];
         dp[0] = false;
         for j in 0..pn {
-            let tmp = dp[j + 1]; // save dp_prev[j+1] before overwrite
-            dp[j + 1] = match p[j] {
-                '%' => {
-                    // dp_prev[j+1] (skip char in s) || dp[j] (skip '%' in pattern)
-                    tmp || dp[j]
-                }
-                '_' => prev, // dp_prev[j]: both advance by one
-                c => prev && s[i] == c,
+            let tmp = dp[j + 1];
+            dp[j + 1] = if p[j] == '%' && !escaped[j] {
+                tmp || dp[j]
+            } else if (p[j] == '_' && !escaped[j]) || (p[j] == s[i_s]) {
+                prev
+            } else {
+                false
             };
             prev = tmp;
         }

crates/iridium_core/src/executor/query/binder/mod.rs

@@ -92,6 +92,8 @@ pub(super) fn query_result_to_bound_table(
                 check: None,
                 check_constraint_name: None,
                 computed_expr: None,
+                collation: None,
+                is_clustered: false,
                 ansi_padding_on: true,
             })
             .collect(),

crates/iridium_core/src/executor/query/binder/tvf.rs

@@ -108,6 +108,8 @@ pub(super) fn bind_builtin_tvf(
         check: None,
         check_constraint_name: None,
         computed_expr: None,
+        collation: None,
+        is_clustered: false,
         ansi_padding_on: true,
     }];
 
@@ -125,6 +127,8 @@ pub(super) fn bind_builtin_tvf(
             check: None,
             check_constraint_name: None,
             computed_expr: None,
+            collation: None,
+            is_clustered: false,
             ansi_padding_on: true,
         });
     }
@@ -328,6 +332,8 @@ fn bind_openjson(
             check: None,
             check_constraint_name: None,
             computed_expr: None,
+            collation: None,
+            is_clustered: false,
             ansi_padding_on: true,
         },
         ColumnDef {
@@ -343,6 +349,8 @@ fn bind_openjson(
             check: None,
             check_constraint_name: None,
             computed_expr: None,
+            collation: None,
+            is_clustered: false,
             ansi_padding_on: true,
         },
         ColumnDef {
@@ -358,6 +366,8 @@ fn bind_openjson(
             check: None,
             check_constraint_name: None,
             computed_expr: None,
+            collation: None,
+            is_clustered: false,
             ansi_padding_on: true,
         },
     ];

crates/iridium_core/src/executor/query/binder/values.rs

@@ -56,6 +56,8 @@ pub(super) fn bind_plain_table(
                     check: None,
                     check_constraint_name: None,
                     computed_expr: None,
+                    collation: None,
+                    is_clustered: false,
                     ansi_padding_on: true,
                 });
             }

crates/iridium_core/src/executor/query/from_tree.rs

@@ -398,6 +398,8 @@ fn apply_from_alias(
                 check: None,
                 check_constraint_name: None,
                 computed_expr: None,
+                collation: None,
+                is_clustered: false,
                 ansi_padding_on: true,
             });
         }

crates/iridium_core/src/executor/query/transformer/apply.rs

@@ -80,6 +80,8 @@ fn build_virtual_table(alias: &str, sub_result: &QueryResult) -> TableDef {
                 check: None,
                 check_constraint_name: None,
                 computed_expr: None,
+                collation: None,
+                is_clustered: false,
                 ansi_padding_on: true,
             })
             .collect(),