Merge 'Actually enforce uniqueness in create unique index' from Jussi Saurio

we just weren't doing it 🤡
Backport: 0.2
Closes #3568

Reviewed-by: Preston Thorpe <preston@turso.tech>

Closes #3571

core/translate/index.rs

@@ -1,11 +1,13 @@
 use std::sync::Arc;
 
 use crate::bail_parse_error;
+use crate::error::SQLITE_CONSTRAINT_UNIQUE;
 use crate::schema::{Table, RESERVED_TABLE_PREFIXES};
 use crate::translate::emitter::{
     emit_cdc_full_record, emit_cdc_insns, prepare_cdc_if_necessary, OperationMode, Resolver,
 };
 use crate::translate::expr::{translate_condition_expr, ConditionMetadata};
+use crate::translate::insert::format_unique_violation_desc;
 use crate::translate::plan::{
     ColumnUsedMask, IterationDirection, JoinedTable, Operation, Scan, TableReferences,
 };
@@ -91,6 +93,7 @@ pub fn translate_create_index(
     };
     let original_columns = columns;
     let columns = resolve_sorted_columns(&tbl, columns)?;
+    let unique = unique_if_not_exists.0;
 
     let idx = Arc::new(Index {
         name: idx_name.clone(),
@@ -106,7 +109,7 @@ pub fn translate_create_index(
                 default: col.default.clone(),
             })
             .collect(),
-        unique: unique_if_not_exists.0,
+        unique,
         ephemeral: false,
         has_rowid: tbl.has_rowid,
         // store the *original* where clause, because we need to rewrite it
@@ -295,8 +298,34 @@ pub fn translate_create_index(
         cursor_id: sorter_cursor_id,
         pc_if_empty: sorted_loop_end,
     });
-    program.preassign_label_to_next_insn(sorted_loop_start);
+
     let sorted_record_reg = program.alloc_register();
+
+    if unique {
+        // Since the records to be inserted are sorted, we can compare prev with current and if they are equal,
+        // we fall through to Halt with a unique constraint violation error.
+        let goto_label = program.allocate_label();
+        let label_after_sorter_compare = program.allocate_label();
+        program.resolve_label(goto_label, program.offset());
+        program.emit_insn(Insn::Goto {
+            target_pc: label_after_sorter_compare,
+        });
+        program.preassign_label_to_next_insn(sorted_loop_start);
+        program.emit_insn(Insn::SorterCompare {
+            cursor_id: sorter_cursor_id,
+            sorted_record_reg,
+            num_regs: columns.len(),
+            pc_when_nonequal: goto_label,
+        });
+        program.emit_insn(Insn::Halt {
+            err_code: SQLITE_CONSTRAINT_UNIQUE,
+            description: format_unique_violation_desc(tbl_name.as_str(), &idx),
+        });
+        program.preassign_label_to_next_insn(label_after_sorter_compare);
+    } else {
+        program.preassign_label_to_next_insn(sorted_loop_start);
+    }
+
     program.emit_insn(Insn::SorterData {
         pseudo_cursor: pseudo_cursor_id,
         cursor_id: sorter_cursor_id,

core/vdbe/builder.rs

@@ -687,6 +687,12 @@ impl ProgramBuilder {
                 Insn::SorterSort { pc_if_empty, .. } => {
                     resolve(pc_if_empty, "SorterSort");
                 }
+                Insn::SorterCompare {
+                    pc_when_nonequal: target_pc,
+                    ..
+                } => {
+                    resolve(target_pc, "SorterCompare");
+                }
                 Insn::NotNull {
                     reg: _reg,
                     target_pc,

core/vdbe/execute.rs

@@ -4247,6 +4247,59 @@ pub fn op_sorter_next(
     Ok(InsnFunctionStepResult::Step)
 }
 
+pub fn op_sorter_compare(
+    program: &Program,
+    state: &mut ProgramState,
+    insn: &Insn,
+    pager: &Arc<Pager>,
+    mv_store: Option<&Arc<MvStore>>,
+) -> Result<InsnFunctionStepResult> {
+    load_insn!(
+        SorterCompare {
+            cursor_id,
+            sorted_record_reg,
+            num_regs,
+            pc_when_nonequal,
+        },
+        insn
+    );
+
+    let previous_sorter_values = {
+        let Register::Record(record) = &state.registers[*sorted_record_reg] else {
+            return Err(LimboError::InternalError(
+                "Sorted record must be a record".to_string(),
+            ));
+        };
+        &record.get_values()[..*num_regs]
+    };
+
+    let cursor = state.get_cursor(*cursor_id);
+    let cursor = cursor.as_sorter_mut();
+    let Some(current_sorter_record) = cursor.record() else {
+        return Err(LimboError::InternalError(
+            "Sorter must have a record".to_string(),
+        ));
+    };
+
+    let current_sorter_values = &current_sorter_record.get_values()[..*num_regs];
+    // If the current sorter record has a NULL in any of the significant fields, the comparison is not equal.
+    let is_equal = current_sorter_values
+        .iter()
+        .all(|v| !matches!(v, RefValue::Null))
+        && compare_immutable(
+            previous_sorter_values,
+            current_sorter_values,
+            &cursor.index_key_info,
+        )
+        .is_eq();
+    if is_equal {
+        state.pc += 1;
+    } else {
+        state.pc = pc_when_nonequal.as_offset_int();
+    }
+    Ok(InsnFunctionStepResult::Step)
+}
+
 pub fn op_function(
     program: &Program,
     state: &mut ProgramState,

core/vdbe/explain.rs

@@ -1042,6 +1042,20 @@ pub fn insn_to_row(
                 0,
                 "".to_string(),
             ),
+            Insn::SorterCompare {
+                cursor_id,
+                pc_when_nonequal,
+                sorted_record_reg,
+                num_regs,
+            } => (
+                "SorterCompare",
+                *cursor_id as i32,
+                pc_when_nonequal.as_debug_int(),
+                *sorted_record_reg as i32,
+                Value::build_text(num_regs.to_string()),
+                0,
+                "".to_string(),
+            ),
             Insn::Function {
                 constant_mask,
                 start_reg,

core/vdbe/insn.rs

@@ -713,6 +713,18 @@ pub enum Insn {
         record_reg: usize,
     },
 
+    /// `cursor_id` is a sorter cursor. This instruction compares a prefix of the record blob in register `sorted_record_reg`
+    /// against a prefix of the entry that the sorter cursor currently points to.
+    /// Only the first `num_regs` fields of `sorted_record_reg` and the sorter record are compared.
+    /// Fall through to next instruction if the two records compare equal to each other.
+    /// Jump to `pc_when_nonequal` if they are different.
+    SorterCompare {
+        cursor_id: CursorID,
+        pc_when_nonequal: BranchOffset,
+        sorted_record_reg: usize,
+        num_regs: usize,
+    },
+
     /// Sort the rows in the sorter.
     SorterSort {
         cursor_id: CursorID,
@@ -1263,6 +1275,7 @@ impl InsnVariants {
             InsnVariants::SorterSort => execute::op_sorter_sort,
             InsnVariants::SorterData => execute::op_sorter_data,
             InsnVariants::SorterNext => execute::op_sorter_next,
+            InsnVariants::SorterCompare => execute::op_sorter_compare,
             InsnVariants::Function => execute::op_function,
             InsnVariants::Cast => execute::op_cast,
             InsnVariants::InitCoroutine => execute::op_init_coroutine,

core/vdbe/sorter.rs

@@ -62,7 +62,7 @@ pub struct Sorter {
     /// The number of values in the key.
     key_len: usize,
     /// The key info.
-    index_key_info: Rc<Vec<KeyInfo>>,
+    pub index_key_info: Rc<Vec<KeyInfo>>,
     /// Sorted chunks stored on disk.
     chunks: Vec<SortedChunk>,
     /// The heap of records consumed from the chunks and their corresponding chunk index.

testing/create_index.test

@@ -13,3 +13,49 @@ do_execsql_test_on_specific_db {:memory:} create-index-quoted-identifiers {
     "CREATE INDEX \"idx idx\" ON \"t t\" (\"a a\")"
     "CREATE UNIQUE INDEX \"unique idx idx\" ON \"t t\" (\"a a\")"
 }
+
+
+# single-column index key: creating unique index fails with duplicates
+do_execsql_test_in_memory_any_error create-unique-index-with-duplicates-1 {
+    CREATE TABLE t1(a, b);
+    INSERT INTO t1 VALUES(1, 1);
+    INSERT INTO t1 VALUES(1, 2);
+    CREATE UNIQUE INDEX idx1 ON t1(a);
+}
+
+# multi-column index key: creating unique index fails with duplicates
+do_execsql_test_in_memory_any_error create-unique-index-with-duplicates-2 {
+    CREATE TABLE t2(a, b, c);
+    INSERT INTO t2 VALUES(1, 2, 3);
+    INSERT INTO t2 VALUES(1, 2, 4);
+    CREATE UNIQUE INDEX idx2 ON t2(a, b);
+}
+
+
+# single-column index key: creating unique index succeeds because NULLs are never equal
+do_execsql_test_on_specific_db {:memory:} create-unique-index-with-duplicates-3 {
+    CREATE TABLE t3(a);
+    INSERT INTO t3 VALUES(NULL);
+    INSERT INTO t3 VALUES(NULL);
+    CREATE UNIQUE INDEX idx3 ON t3(a);
+    SELECT count(*) FROM t3;
+} {2}
+
+# multi-column index key: creating unique index succeeds because NULLs are never equal
+do_execsql_test_on_specific_db {:memory:} create-unique-index-with-duplicates-4 {
+    CREATE TABLE t4(a, b);
+    INSERT INTO t4 VALUES(1, NULL);
+    INSERT INTO t4 VALUES(1, NULL);
+    CREATE UNIQUE INDEX idx4 ON t4(a, b);
+    SELECT count(*) FROM t4;
+} {2}
+
+# multi-column index key: creating unique index succeeds when all NULLs
+do_execsql_test_on_specific_db {:memory:} create-unique-index-with-duplicates-5 {
+    CREATE TABLE t5(a, b);
+    INSERT INTO t5 VALUES(NULL, NULL);
+    INSERT INTO t5 VALUES(NULL, NULL);
+    CREATE UNIQUE INDEX idx5 ON t5(a, b);
+    SELECT count(*) FROM t5;
+} {2}
+