Merge 'Implement UPSERT' from Preston Thorpe

This PR closes #2019
Implements https://sqlite.org/lang_upsert.html

Closes #2853

COMPAT.md

@@ -71,8 +71,8 @@ Turso aims to be fully compatible with SQLite, with opt-in features not supporte
 | END TRANSACTION           | Partial | Alias for `COMMIT TRANSACTION`                                                    |
 | EXPLAIN                   | Yes     |                                                                                   |
 | INDEXED BY                | No      |                                                                                   |
-| INSERT                    | Partial |                                                                                   |
-| ON CONFLICT clause        | No      |                                                                                   |
+| INSERT                    | Yes     |                                                                                   |
+| ON CONFLICT clause        | Yes     |                                                                                   |
 | REINDEX                   | No      |                                                                                   |
 | RELEASE SAVEPOINT         | No      |                                                                                   |
 | REPLACE                   | No      |                                                                                   |

core/translate/insert.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
-
 use turso_parser::ast::{
-    self, Expr, InsertBody, OneSelect, QualifiedName, ResolveType, ResultColumn, With,
+    self, Expr, InsertBody, OneSelect, QualifiedName, ResolveType, ResultColumn, Upsert, UpsertDo,
+    With,
 };
 
 use crate::error::{SQLITE_CONSTRAINT_NOTNULL, SQLITE_CONSTRAINT_PRIMARYKEY};
@@ -13,6 +13,9 @@ use crate::translate::expr::{
     emit_returning_results, process_returning_clause, ReturningValueRegisters,
 };
 use crate::translate::planner::ROWID;
+use crate::translate::upsert::{
+    collect_set_clauses_for_upsert, emit_upsert, upsert_matches_index, upsert_matches_pk,
+};
 use crate::util::normalize_ident;
 use crate::vdbe::builder::ProgramBuilderOpts;
 use crate::vdbe::insn::{IdxInsertFlags, InsertFlags, RegisterOrLiteral};
@@ -101,48 +104,56 @@ pub fn translate_insert(
     let root_page = btree_table.root_page;
 
     let mut values: Option<Vec<Box<Expr>>> = None;
+    let mut upsert_opt: Option<Upsert> = None;
+
     let inserting_multiple_rows = match &mut body {
-        InsertBody::Select(select, _) => match &mut select.body.select {
-            // TODO see how to avoid clone
-            OneSelect::Values(values_expr) if values_expr.len() <= 1 => {
-                if values_expr.is_empty() {
-                    crate::bail_parse_error!("no values to insert");
-                }
-                let mut param_idx = 1;
-                for expr in values_expr.iter_mut().flat_map(|v| v.iter_mut()) {
-                    match expr.as_mut() {
-                        Expr::Id(name) => {
-                            if name.is_double_quoted() {
-                                *expr =
-                                    Expr::Literal(ast::Literal::String(name.to_string())).into();
-                            } else {
-                                // an INSERT INTO ... VALUES (...) cannot reference columns
-                                crate::bail_parse_error!("no such column: {name}");
-                            }
-                        }
-                        Expr::Qualified(first_name, second_name) => {
-                            // an INSERT INTO ... VALUES (...) cannot reference columns
-                            crate::bail_parse_error!("no such column: {first_name}.{second_name}");
-                        }
-                        _ => {}
+        InsertBody::Select(select, upsert) => {
+            upsert_opt = upsert.as_deref().cloned();
+            match &mut select.body.select {
+                // TODO see how to avoid clone
+                OneSelect::Values(values_expr) if values_expr.len() <= 1 => {
+                    if values_expr.is_empty() {
+                        crate::bail_parse_error!("no values to insert");
                     }
-                    rewrite_expr(expr, &mut param_idx)?;
+                    let mut param_idx = 1;
+                    for expr in values_expr.iter_mut().flat_map(|v| v.iter_mut()) {
+                        match expr.as_mut() {
+                            Expr::Id(name) => {
+                                if name.is_double_quoted() {
+                                    *expr = Expr::Literal(ast::Literal::String(name.to_string()))
+                                        .into();
+                                } else {
+                                    // an INSERT INTO ... VALUES (...) cannot reference columns
+                                    crate::bail_parse_error!("no such column: {name}");
+                                }
+                            }
+                            Expr::Qualified(first_name, second_name) => {
+                                // an INSERT INTO ... VALUES (...) cannot reference columns
+                                crate::bail_parse_error!(
+                                    "no such column: {first_name}.{second_name}"
+                                );
+                            }
+                            _ => {}
+                        }
+                        rewrite_expr(expr, &mut param_idx)?;
+                    }
+                    values = values_expr.pop();
+                    false
                 }
-                values = values_expr.pop();
-                false
+                _ => true,
             }
-            _ => true,
-        },
+        }
         InsertBody::DefaultValues => false,
     };
 
     let halt_label = program.allocate_label();
     let loop_start_label = program.allocate_label();
+    let row_done_label = program.allocate_label();
 
     let cdc_table = prepare_cdc_if_necessary(&mut program, schema, table.get_name())?;
 
     // Process RETURNING clause using shared module
-    let (result_columns, _) = process_returning_clause(
+    let (mut result_columns, _) = process_returning_clause(
         &mut returning,
         &table,
         table_name.as_str(),
@@ -158,7 +169,6 @@ pub fn translate_insert(
     let mut yield_reg_opt = None;
     let mut temp_table_ctx = None;
     let (num_values, cursor_id) = match body {
-        // TODO: upsert
         InsertBody::Select(select, _) => {
             // Simple Common case of INSERT INTO <table> VALUES (...)
             if matches!(&select.body.select, OneSelect::Values(values) if values.len() <= 1) {
@@ -336,6 +346,7 @@ pub fn translate_insert(
             db: 0,
         });
     }
+
     // Common record insertion logic for both single and multiple rows
     let has_user_provided_rowid = insertion.key.is_provided_by_user();
     let check_rowid_is_integer_label = if has_user_provided_rowid {
@@ -365,6 +376,17 @@ pub fn translate_insert(
         });
     }
 
+    let emit_halt_with_constraint = |program: &mut ProgramBuilder, col_name: &str| {
+        let mut description = String::with_capacity(table_name.as_str().len() + col_name.len() + 2);
+        description.push_str(table_name.as_str());
+        description.push('.');
+        description.push_str(col_name);
+        program.emit_insn(Insn::Halt {
+            err_code: SQLITE_CONSTRAINT_PRIMARYKEY,
+            description,
+        });
+    };
+
     // Check uniqueness constraint for rowid if it was provided by user.
     // When the DB allocates it there are no need for separate uniqueness checks.
     if has_user_provided_rowid {
@@ -375,15 +397,46 @@ pub fn translate_insert(
             target_pc: make_record_label,
         });
         let rowid_column_name = insertion.key.column_name();
-        let mut description =
-            String::with_capacity(table_name.as_str().len() + rowid_column_name.len() + 2);
-        description.push_str(table_name.as_str());
-        description.push('.');
-        description.push_str(rowid_column_name);
-        program.emit_insn(Insn::Halt {
-            err_code: SQLITE_CONSTRAINT_PRIMARYKEY,
-            description,
-        });
+
+        // Conflict on rowid: attempt to route through UPSERT if it targets the PK, otherwise raise constraint.
+        // emit Halt for every case *except* when upsert handles the conflict
+        'emit_halt: {
+            if let Some(ref mut upsert) = upsert_opt.as_mut() {
+                if upsert_matches_pk(upsert, &table) {
+                    match upsert.do_clause {
+                        UpsertDo::Nothing => {
+                            program.emit_insn(Insn::Goto {
+                                target_pc: row_done_label,
+                            });
+                        }
+                        UpsertDo::Set {
+                            ref mut sets,
+                            ref mut where_clause,
+                        } => {
+                            let mut rewritten_sets = collect_set_clauses_for_upsert(&table, sets)?;
+
+                            emit_upsert(
+                                &mut program,
+                                schema,
+                                &table,
+                                &insertion,
+                                cursor_id,
+                                insertion.key_register(),
+                                &mut rewritten_sets,
+                                where_clause,
+                                &resolver,
+                                &idx_cursors,
+                                &mut result_columns,
+                                cdc_table.as_ref().map(|c| c.0),
+                                row_done_label,
+                            )?;
+                        }
+                    }
+                    break 'emit_halt;
+                }
+            }
+            emit_halt_with_constraint(&mut program, rowid_column_name);
+        }
         program.preassign_label_to_next_insn(make_record_label);
     }
 
@@ -418,8 +471,13 @@ pub fn translate_insert(
         // copy each index column from the table's column registers into these scratch regs
         for (i, column_mapping) in column_mappings.clone().enumerate() {
             // copy from the table's column register over to the index's scratch register
+            let Some(col_mapping) = column_mapping else {
+                return Err(crate::LimboError::PlanningError(
+                    "Column not found in INSERT".to_string(),
+                ));
+            };
             program.emit_insn(Insn::Copy {
-                src_reg: column_mapping.register,
+                src_reg: col_mapping.register,
                 dst_reg: idx_start_reg + i,
                 extra_amount: 0,
             });
@@ -460,14 +518,55 @@ pub fn translate_insert(
                 },
             );
 
-            program.emit_insn(Insn::Halt {
-                err_code: SQLITE_CONSTRAINT_PRIMARYKEY,
-                description: column_names,
-            });
-
+            // again, emit halt for every case *except* when upsert handles the conflict
+            'emit_halt: {
+                if let Some(ref mut upsert) = upsert_opt.as_mut() {
+                    if upsert_matches_index(upsert, index, &table) {
+                        match upsert.do_clause {
+                            UpsertDo::Nothing => {
+                                program.emit_insn(Insn::Goto {
+                                    target_pc: row_done_label,
+                                });
+                            }
+                            UpsertDo::Set {
+                                ref mut sets,
+                                ref mut where_clause,
+                            } => {
+                                let mut rewritten_sets =
+                                    collect_set_clauses_for_upsert(&table, sets)?;
+                                let conflict_rowid_reg = program.alloc_register();
+                                program.emit_insn(Insn::IdxRowId {
+                                    cursor_id: idx_cursor_id,
+                                    dest: conflict_rowid_reg,
+                                });
+                                emit_upsert(
+                                    &mut program,
+                                    schema,
+                                    &table,
+                                    &insertion,
+                                    cursor_id,
+                                    conflict_rowid_reg,
+                                    &mut rewritten_sets,
+                                    where_clause,
+                                    &resolver,
+                                    &idx_cursors,
+                                    &mut result_columns,
+                                    cdc_table.as_ref().map(|c| c.0),
+                                    row_done_label,
+                                )?;
+                            }
+                        }
+                        break 'emit_halt;
+                    }
+                }
+                // No matching UPSERT rule: unique constraint violation.
+                program.emit_insn(Insn::Halt {
+                    err_code: SQLITE_CONSTRAINT_PRIMARYKEY,
+                    description: column_names,
+                });
+            }
             program.resolve_label(label_idx_insert, program.offset());
         }
-
         // now do the actual index insertion using the unpacked registers
         program.emit_insn(Insn::IdxInsert {
             cursor_id: idx_cursor_id,
@@ -570,6 +669,8 @@ pub fn translate_insert(
 
     if inserting_multiple_rows {
         if let Some(temp_table_ctx) = temp_table_ctx {
+            program.resolve_label(row_done_label, program.offset());
+
             program.emit_insn(Insn::Next {
                 cursor_id: temp_table_ctx.cursor_id,
                 pc_if_next: temp_table_ctx.loop_start_label,
@@ -581,10 +682,13 @@ pub fn translate_insert(
             });
         } else {
             // For multiple rows which not require a temp table, loop back
+            program.resolve_label(row_done_label, program.offset());
             program.emit_insn(Insn::Goto {
                 target_pc: loop_start_label,
             });
         }
+    } else {
+        program.resolve_label(row_done_label, program.offset());
     }
 
     program.resolve_label(halt_label, program.offset());
@@ -607,7 +711,7 @@ pub const ROWID_COLUMN: &Column = &Column {
 
 /// Represents how a table should be populated during an INSERT.
 #[derive(Debug)]
-struct Insertion<'a> {
+pub struct Insertion<'a> {
     /// The integer key ("rowid") provided to the VDBE.
     key: InsertionKey<'a>,
     /// The column values that will be fed to the MakeRecord instruction to insert the row.
@@ -639,7 +743,7 @@ impl<'a> Insertion<'a> {
     }
 
     /// Returns the column mapping for a given column name.
-    pub fn get_col_mapping_by_name(&self, name: &str) -> &ColMapping<'a> {
+    pub fn get_col_mapping_by_name(&self, name: &str) -> Option<&ColMapping<'a>> {
         if let InsertionKey::RowidAlias(mapping) = &self.key {
             // If the key is a rowid alias, a NULL is emitted as the column value,
             // so we need to return the key mapping instead so that the non-NULL rowid is used
@@ -650,18 +754,15 @@ impl<'a> Insertion<'a> {
                 .as_ref()
                 .is_some_and(|n| n.eq_ignore_ascii_case(name))
             {
-                return mapping;
+                return Some(mapping);
             }
         }
-        self.col_mappings
-            .iter()
-            .find(|col| {
-                col.column
-                    .name
-                    .as_ref()
-                    .is_some_and(|n| n.eq_ignore_ascii_case(name))
-            })
-            .unwrap_or_else(|| panic!("column {name} not found in insertion"))
+        self.col_mappings.iter().find(|col| {
+            col.column
+                .name
+                .as_ref()
+                .is_some_and(|n| n.eq_ignore_ascii_case(name))
+        })
     }
 }
 
@@ -708,18 +809,18 @@ impl InsertionKey<'_> {
 /// In a vector of [ColMapping], the index of a given [ColMapping] is
 /// the position of the column in the table.
 #[derive(Debug)]
-struct ColMapping<'a> {
+pub struct ColMapping<'a> {
     /// Column definition
-    column: &'a Column,
+    pub column: &'a Column,
     /// Index of the value to use from a tuple in the insert statement.
     /// This is needed because the values in the insert statement are not necessarily
     /// in the same order as the columns in the table, nor do they necessarily contain
     /// all of the columns in the table.
     /// If None, a NULL will be emitted for the column, unless it has a default value.
     /// A NULL rowid alias column's value will be autogenerated.
-    value_index: Option<usize>,
+    pub value_index: Option<usize>,
     /// Register where the value will be stored for insertion into the table.
-    register: usize,
+    pub register: usize,
 }
 
 /// Resolves how each column in a table should be populated during an INSERT.

core/translate/mod.rs

@@ -35,6 +35,7 @@ pub(crate) mod select;
 pub(crate) mod subquery;
 pub(crate) mod transaction;
 pub(crate) mod update;
+pub(crate) mod upsert;
 mod values;
 pub(crate) mod view;
 

core/translate/upsert.rs

@@ -0,0 +1,816 @@
+use std::{collections::HashMap, sync::Arc};
+
+use turso_parser::ast::{self, Upsert};
+
+use crate::{
+    bail_parse_error,
+    error::SQLITE_CONSTRAINT_NOTNULL,
+    schema::{Index, IndexColumn, Schema, Table},
+    translate::{
+        emitter::{
+            emit_cdc_full_record, emit_cdc_insns, emit_cdc_patch_record, OperationMode, Resolver,
+        },
+        expr::{
+            emit_returning_results, translate_expr, translate_expr_no_constant_opt,
+            NoConstantOptReason, ReturningValueRegisters,
+        },
+        insert::{Insertion, ROWID_COLUMN},
+        plan::ResultSetColumn,
+    },
+    util::normalize_ident,
+    vdbe::{
+        builder::ProgramBuilder,
+        insn::{IdxInsertFlags, InsertFlags, Insn},
+        BranchOffset,
+    },
+};
+
+/// A ConflictTarget is extracted from each ON CONFLICT target,
+// e.g. INSERT INTO x(a) ON CONFLICT  *(a COLLATE nocase)*
+#[derive(Debug, Clone)]
+pub struct ConflictTarget {
+    /// The normalized column name in question
+    col_name: String,
+    /// Possible collation name, normalized to lowercase
+    collate: Option<String>,
+}
+
+// Extract `(column, optional_collate)` from an ON CONFLICT target Expr.
+// Accepts: Id, Qualified, DoublyQualified, Parenthesized, Collate
+fn extract_target_key(e: &ast::Expr) -> Option<ConflictTarget> {
+    match e {
+        // expr COLLATE c: carry c and keep descending into expr
+        ast::Expr::Collate(inner, c) => {
+            let mut tk = extract_target_key(inner.as_ref())?;
+            let cstr = match c {
+                ast::Name::Ident(s) => s.as_str(),
+                _ => return None,
+            };
+            tk.collate = Some(cstr.to_ascii_lowercase());
+            Some(tk)
+        }
+        ast::Expr::Parenthesized(v) if v.len() == 1 => extract_target_key(&v[0]),
+        // Bare identifier
+        ast::Expr::Id(ast::Name::Ident(name)) => Some(ConflictTarget {
+            col_name: normalize_ident(name),
+            collate: None,
+        }),
+        // t.a or db.t.a
+        ast::Expr::Qualified(_, col) | ast::Expr::DoublyQualified(_, _, col) => {
+            let cname = match col {
+                ast::Name::Ident(s) => s.as_str(),
+                _ => return None,
+            };
+            Some(ConflictTarget {
+                col_name: normalize_ident(cname),
+                collate: None,
+            })
+        }
+        _ => None,
+    }
+}
+
+// Return the index key’s effective collation.
+// If `idx_col.collation` is None, fall back to the column default or "BINARY".
+fn effective_collation_for_index_col(idx_col: &IndexColumn, table: &Table) -> String {
+    if let Some(c) = idx_col.collation.as_ref() {
+        return c.to_string().to_ascii_lowercase();
+    }
+    // Otherwise use the table default, or default to BINARY
+    table
+        .get_column_by_name(&idx_col.name)
+        .map(|s| {
+            s.1.collation
+                .map(|c| c.to_string().to_ascii_lowercase())
+                .unwrap_or_else(|| "binary".to_string())
+        })
+        .unwrap_or_else(|| "binary".to_string())
+}
+
+/// Match ON CONFLICT target to the PRIMARY KEY, if any.
+/// If no target is specified, it is an automatic match for PRIMARY KEY
+pub fn upsert_matches_pk(upsert: &Upsert, table: &Table) -> bool {
+    let Some(t) = upsert.index.as_ref() else {
+        // Omitted target is automatic
+        return true;
+    };
+    if !t.targets.len().eq(&1) {
+        return false;
+    }
+    let pk = table
+        .columns()
+        .iter()
+        .find(|c| c.is_rowid_alias || c.primary_key)
+        .unwrap_or(ROWID_COLUMN);
+    extract_target_key(&t.targets[0].expr).is_some_and(|tk| {
+        tk.col_name
+            .eq_ignore_ascii_case(pk.name.as_ref().unwrap_or(&String::new()))
+    })
+}
+
+#[derive(Hash, Debug, Eq, PartialEq, Clone)]
+/// A hashable descriptor of a single index key term used when
+/// matching an `ON CONFLICT` target against a UNIQUE index.
+/// captures only the attributes (name and effective collation) that
+/// determine whether two key terms are equivalent for conflict detection.
+pub struct KeySig {
+    /// column name, normalized to lowercase
+    name: String,
+    /// defaults to "binary" if not specified on the target or col
+    coll: String,
+}
+
+/// Match ON CONFLICT target to a UNIQUE index, ignoring order, requiring exact
+/// coverage, and honoring collations. `table` is used to derive effective collation.
+pub fn upsert_matches_index(upsert: &Upsert, index: &Index, table: &Table) -> bool {
+    let Some(target) = upsert.index.as_ref() else {
+        // catch-all
+        return true;
+    };
+    // if not unique or column count differs, no match
+    if !index.unique || target.targets.len() != index.columns.len() {
+        return false;
+    }
+
+    let mut need: HashMap<KeySig, usize> = HashMap::new();
+    for ic in &index.columns {
+        let sig = KeySig {
+            name: normalize_ident(&ic.name).to_string(),
+            coll: effective_collation_for_index_col(ic, table),
+        };
+        *need.entry(sig).or_insert(0) += 1;
+    }
+
+    // Consume from the multiset using target entries, order-insensitive
+    for te in &target.targets {
+        let tk = match extract_target_key(&te.expr) {
+            Some(x) => x,
+            None => return false, // not a simple column ref
+        };
+
+        // Candidate signatures for this target:
+        // If target specifies COLLATE, require exact match on (name, coll).
+        // Otherwise, accept any collation currently present for that name.
+        let mut matched = false;
+        if let Some(ref coll) = tk.collate {
+            let sig = KeySig {
+                name: tk.col_name.to_string(),
+                coll: coll.clone(),
+            };
+            if let Some(cnt) = need.get_mut(&sig) {
+                *cnt -= 1;
+                if *cnt == 0 {
+                    need.remove(&sig);
+                }
+                matched = true;
+            }
+        } else {
+            // Try any available collation for this column name
+            if let Some((sig, cnt)) = need
+                .iter_mut()
+                .find(|(k, _)| k.name.eq_ignore_ascii_case(&tk.col_name))
+            {
+                *cnt -= 1;
+                if *cnt == 0 {
+                    let key = sig.clone();
+                    need.remove(&key);
+                }
+                matched = true;
+            }
+        }
+        if !matched {
+            return false;
+        }
+    }
+    // All targets matched exactly.
+    need.is_empty()
+}
+
+#[allow(clippy::too_many_arguments)]
+/// Emit the bytecode to implement the `DO UPDATE` arm of an UPSERT.
+///
+/// This routine is entered after the caller has determined that an INSERT
+/// would violate a UNIQUE/PRIMARY KEY constraint and that the user requested
+/// `ON CONFLICT ... DO UPDATE`.
+///
+/// High-level flow:
+/// 1. Seek to the conflicting row by rowid and load the current row snapshot
+///    into a contiguous set of registers.
+/// 2. Optionally duplicate CURRENT into BEFORE* (for index rebuild and CDC).
+/// 3. Copy CURRENT into NEW, then evaluate SET expressions into NEW,
+///    with all references to the target table columns rewritten to read from
+///    the CURRENT registers (per SQLite semantics).
+/// 4. Enforce NOT NULL constraints and (if STRICT) type checks on NEW.
+/// 5. Rebuild indexes (delete keys using BEFORE, insert keys using NEW).
+/// 6. Rewrite the table row payload at the same rowid with NEW.
+/// 7. Emit CDC rows and RETURNING output if requested.
+/// 8. Jump to `row_done_label`.
+///
+/// Semantics reference: https://sqlite.org/lang_upsert.html
+/// Column references in the DO UPDATE expressions refer to the original
+/// (unchanged) row. To refer to would-be inserted values, use `excluded.x`.
+pub fn emit_upsert(
+    program: &mut ProgramBuilder,
+    schema: &Schema,
+    table: &Table,
+    insertion: &Insertion,
+    tbl_cursor_id: usize,
+    conflict_rowid_reg: usize,
+    set_pairs: &mut [(usize, Box<ast::Expr>)],
+    where_clause: &mut Option<Box<ast::Expr>>,
+    resolver: &Resolver,
+    idx_cursors: &[(&String, usize, usize)],
+    returning: &mut [ResultSetColumn],
+    cdc_cursor_id: Option<usize>,
+    row_done_label: BranchOffset,
+) -> crate::Result<()> {
+    // Seek and snapshot current row
+    program.emit_insn(Insn::SeekRowid {
+        cursor_id: tbl_cursor_id,
+        src_reg: conflict_rowid_reg,
+        target_pc: row_done_label,
+    });
+    let num_cols = table.columns().len();
+    let current_start = program.alloc_registers(num_cols);
+    for (i, col) in table.columns().iter().enumerate() {
+        if col.is_rowid_alias {
+            program.emit_insn(Insn::RowId {
+                cursor_id: tbl_cursor_id,
+                dest: current_start + i,
+            });
+        } else {
+            program.emit_insn(Insn::Column {
+                cursor_id: tbl_cursor_id,
+                column: i,
+                dest: current_start + i,
+                default: None,
+            });
+        }
+    }
+
+    // Keep BEFORE snapshot if needed
+    let before_start = if cdc_cursor_id.is_some() || !idx_cursors.is_empty() {
+        let s = program.alloc_registers(num_cols);
+        program.emit_insn(Insn::Copy {
+            src_reg: current_start,
+            dst_reg: s,
+            extra_amount: num_cols - 1,
+        });
+        Some(s)
+    } else {
+        None
+    };
+
+    // NEW snapshot starts as a copy of CURRENT, then SET expressions overwrite
+    // the assigned columns. matching SQLite semantics of UPDATE reading the old row.
+    let new_start = program.alloc_registers(num_cols);
+    program.emit_insn(Insn::Copy {
+        src_reg: current_start,
+        dst_reg: new_start,
+        extra_amount: num_cols - 1,
+    });
+
+    // WHERE predicate on the target row. If false or NULL, skip the UPDATE.
+    if let Some(pred) = where_clause.as_mut() {
+        rewrite_upsert_expr_in_place(
+            pred,
+            table,
+            table.get_name(),
+            current_start,
+            conflict_rowid_reg,
+            insertion,
+        )?;
+        let pr = program.alloc_register();
+        translate_expr(program, None, pred, pr, resolver)?;
+        program.emit_insn(Insn::IfNot {
+            reg: pr,
+            target_pc: row_done_label,
+            jump_if_null: true,
+        });
+    }
+
+    // Evaluate each SET expression into the NEW row img
+    for (col_idx, expr) in set_pairs.iter_mut() {
+        rewrite_upsert_expr_in_place(
+            expr,
+            table,
+            table.get_name(),
+            current_start,
+            conflict_rowid_reg,
+            insertion,
+        )?;
+        translate_expr_no_constant_opt(
+            program,
+            None,
+            expr,
+            new_start + *col_idx,
+            resolver,
+            NoConstantOptReason::RegisterReuse,
+        )?;
+        let col = &table.columns()[*col_idx];
+        if col.notnull && !col.is_rowid_alias {
+            program.emit_insn(Insn::HaltIfNull {
+                target_reg: new_start + *col_idx,
+                err_code: SQLITE_CONSTRAINT_NOTNULL,
+                description: format!("{}.{}", table.get_name(), col.name.as_ref().unwrap()),
+            });
+        }
+    }
+
+    // If STRICT, perform type checks on the NEW image
+    if let Some(bt) = table.btree() {
+        if bt.is_strict {
+            program.emit_insn(Insn::TypeCheck {
+                start_reg: new_start,
+                count: num_cols,
+                check_generated: true,
+                table_reference: Arc::clone(&bt),
+            });
+        }
+    }
+
+    // Rebuild indexes: remove keys corresponding to BEFORE and insert keys for NEW.
+    if let Some(before) = before_start {
+        for (idx_name, _root, idx_cid) in idx_cursors {
+            let idx_meta = schema
+                .get_index(table.get_name(), idx_name)
+                .expect("index exists");
+            let k = idx_meta.columns.len();
+
+            let del = program.alloc_registers(k + 1);
+            for (i, ic) in idx_meta.columns.iter().enumerate() {
+                let (ci, _) = table.get_column_by_name(&ic.name).unwrap();
+                program.emit_insn(Insn::Copy {
+                    src_reg: before + ci,
+                    dst_reg: del + i,
+                    extra_amount: 0,
+                });
+            }
+            program.emit_insn(Insn::Copy {
+                src_reg: conflict_rowid_reg,
+                dst_reg: del + k,
+                extra_amount: 0,
+            });
+            program.emit_insn(Insn::IdxDelete {
+                start_reg: del,
+                num_regs: k + 1,
+                cursor_id: *idx_cid,
+                raise_error_if_no_matching_entry: false,
+            });
+
+            let ins = program.alloc_registers(k + 1);
+            for (i, ic) in idx_meta.columns.iter().enumerate() {
+                let (ci, _) = table.get_column_by_name(&ic.name).unwrap();
+                program.emit_insn(Insn::Copy {
+                    src_reg: new_start + ci,
+                    dst_reg: ins + i,
+                    extra_amount: 0,
+                });
+            }
+            program.emit_insn(Insn::Copy {
+                src_reg: conflict_rowid_reg,
+                dst_reg: ins + k,
+                extra_amount: 0,
+            });
+
+            let rec = program.alloc_register();
+            program.emit_insn(Insn::MakeRecord {
+                start_reg: ins,
+                count: k + 1,
+                dest_reg: rec,
+                index_name: Some((*idx_name).clone()),
+            });
+            program.emit_insn(Insn::IdxInsert {
+                cursor_id: *idx_cid,
+                record_reg: rec,
+                unpacked_start: Some(ins),
+                unpacked_count: Some((k + 1) as u16),
+                flags: IdxInsertFlags::new().nchange(true),
+            });
+        }
+    }
+
+    // Write table row (same rowid, new payload)
+    let rec = program.alloc_register();
+    program.emit_insn(Insn::MakeRecord {
+        start_reg: new_start,
+        count: num_cols,
+        dest_reg: rec,
+        index_name: None,
+    });
+    program.emit_insn(Insn::Insert {
+        cursor: tbl_cursor_id,
+        key_reg: conflict_rowid_reg,
+        record_reg: rec,
+        flag: InsertFlags::new(),
+        table_name: table.get_name().to_string(),
+    });
+
+    if let Some(cdc_id) = cdc_cursor_id {
+        let after_rec = if program.capture_data_changes_mode().has_after() {
+            Some(emit_cdc_patch_record(
+                program,
+                table,
+                new_start,
+                rec,
+                conflict_rowid_reg,
+            ))
+        } else {
+            None
+        };
+        // Build BEFORE if needed
+        let before_rec = if program.capture_data_changes_mode().has_before() {
+            Some(emit_cdc_full_record(
+                program,
+                table.columns(),
+                tbl_cursor_id,
+                conflict_rowid_reg,
+            ))
+        } else {
+            None
+        };
+        emit_cdc_insns(
+            program,
+            resolver,
+            OperationMode::UPDATE,
+            cdc_id,
+            conflict_rowid_reg,
+            before_rec,
+            after_rec,
+            None,
+            table.get_name(),
+        )?;
+    }
+
+    if !returning.is_empty() {
+        let regs = ReturningValueRegisters {
+            rowid_register: conflict_rowid_reg,
+            columns_start_register: new_start,
+            num_columns: num_cols,
+        };
+
+        emit_returning_results(program, returning, &regs)?;
+    }
+    program.emit_insn(Insn::Goto {
+        target_pc: row_done_label,
+    });
+    Ok(())
+}
+
+/// Normalize the `SET` clause into `(column_index, Expr)` pairs using table layout.
+///
+/// Supports multi-target row-value SETs: `SET (a, b) = (expr1, expr2)`.
+/// Enforces same number of column names and RHS values.
+/// Rewrites `EXCLUDED.*` references to direct `Register` reads from the insertion registers
+/// If the same column is assigned multiple times, the last assignment wins.
+pub fn collect_set_clauses_for_upsert(
+    table: &Table,
+    set_items: &mut [ast::Set],
+) -> crate::Result<Vec<(usize, Box<ast::Expr>)>> {
+    let lookup: HashMap<String, usize> = table
+        .columns()
+        .iter()
+        .enumerate()
+        .filter_map(|(i, c)| c.name.as_ref().map(|n| (n.to_lowercase(), i)))
+        .collect();
+
+    let mut out: Vec<(usize, Box<ast::Expr>)> = vec![];
+
+    for set in set_items {
+        let values: Vec<Box<ast::Expr>> = match set.expr.as_ref() {
+            ast::Expr::Parenthesized(v) => v.clone(),
+            e => vec![e.clone().into()],
+        };
+        if set.col_names.len() != values.len() {
+            bail_parse_error!(
+                "{} columns assigned {} values",
+                set.col_names.len(),
+                values.len()
+            );
+        }
+        for (cn, e) in set.col_names.iter().zip(values.into_iter()) {
+            let Some(idx) = lookup.get(&normalize_ident(cn.as_str())) else {
+                bail_parse_error!("no such column: {}", cn);
+            };
+            if let Some(existing) = out.iter_mut().find(|(i, _)| *i == *idx) {
+                existing.1 = e;
+            } else {
+                out.push((*idx, e));
+            }
+        }
+    }
+    Ok(out)
+}
+
+/// Rewrite an UPSERT expression so that:
+/// EXCLUDED.x -> Register(insertion.x)
+/// t.x / x    -> Register(CURRENT.x) when t == target table or unqualified
+/// rowid      -> Register(conflict_rowid_reg)
+///
+/// Only rewrites names in the current expression scope, does not enter subqueries.
+fn rewrite_upsert_expr_in_place(
+    e: &mut ast::Expr,
+    table: &Table,
+    table_name: &str,
+    current_start: usize,
+    conflict_rowid_reg: usize,
+    insertion: &Insertion,
+) -> crate::Result<()> {
+    use ast::Expr::*;
+
+    // helper: return the CURRENT-row register for a column (including rowid alias)
+    let col_reg = |name: &str| -> Option<usize> {
+        if name.eq_ignore_ascii_case("rowid") {
+            return Some(conflict_rowid_reg);
+        }
+        let (idx, _c) = table.get_column_by_name(&normalize_ident(name))?;
+        Some(current_start + idx)
+    };
+
+    match e {
+        // EXCLUDED.x -> insertion register
+        Qualified(ns, ast::Name::Ident(c)) if ns.as_str().eq_ignore_ascii_case("excluded") => {
+            let Some(reg) = insertion.get_col_mapping_by_name(c.as_str()) else {
+                bail_parse_error!("no such column in EXCLUDED: {}", c);
+            };
+            *e = Register(reg.register);
+        }
+
+        // t.x -> CURRENT, only if t matches the target table name (never "excluded")
+        Qualified(ns, ast::Name::Ident(c)) if ns.as_str().eq_ignore_ascii_case(table_name) => {
+            if let Some(reg) = col_reg(c.as_str()) {
+                *e = Register(reg);
+            }
+        }
+        // Unqualified column id -> CURRENT
+        Id(ast::Name::Ident(name)) => {
+            if let Some(reg) = col_reg(name.as_str()) {
+                *e = Register(reg);
+            }
+        }
+        RowId { .. } => {
+            *e = Register(conflict_rowid_reg);
+        }
+        Collate(inner, _) => rewrite_upsert_expr_in_place(
+            inner,
+            table,
+            table_name,
+            current_start,
+            conflict_rowid_reg,
+            insertion,
+        )?,
+        Parenthesized(v) => {
+            for ex in v {
+                rewrite_upsert_expr_in_place(
+                    ex,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+        }
+        Between {
+            lhs, start, end, ..
+        } => {
+            rewrite_upsert_expr_in_place(
+                lhs,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+            rewrite_upsert_expr_in_place(
+                start,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+            rewrite_upsert_expr_in_place(
+                end,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+        Binary(l, _, r) => {
+            rewrite_upsert_expr_in_place(
+                l,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+            rewrite_upsert_expr_in_place(
+                r,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+        Case {
+            base,
+            when_then_pairs,
+            else_expr,
+        } => {
+            if let Some(b) = base {
+                rewrite_upsert_expr_in_place(
+                    b,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+            for (w, t) in when_then_pairs.iter_mut() {
+                rewrite_upsert_expr_in_place(
+                    w,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+                rewrite_upsert_expr_in_place(
+                    t,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+            if let Some(e2) = else_expr {
+                rewrite_upsert_expr_in_place(
+                    e2,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+        }
+        Cast { expr: inner, .. } => {
+            rewrite_upsert_expr_in_place(
+                inner,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+        FunctionCall {
+            args,
+            order_by,
+            filter_over,
+            ..
+        } => {
+            for a in args {
+                rewrite_upsert_expr_in_place(
+                    a,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+            for sc in order_by {
+                rewrite_upsert_expr_in_place(
+                    &mut sc.expr,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+            if let Some(ref mut f) = &mut filter_over.filter_clause {
+                rewrite_upsert_expr_in_place(
+                    f,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+        }
+        InList { lhs, rhs, .. } => {
+            rewrite_upsert_expr_in_place(
+                lhs,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+            for ex in rhs {
+                rewrite_upsert_expr_in_place(
+                    ex,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+        }
+        InSelect { lhs, .. } => {
+            // rewrite only `lhs`, not the subselect
+            rewrite_upsert_expr_in_place(
+                lhs,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+        InTable { lhs, .. } => {
+            rewrite_upsert_expr_in_place(
+                lhs,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+        IsNull(inner) => {
+            rewrite_upsert_expr_in_place(
+                inner,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+        Like {
+            lhs, rhs, escape, ..
+        } => {
+            rewrite_upsert_expr_in_place(
+                lhs,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+            rewrite_upsert_expr_in_place(
+                rhs,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+            if let Some(e3) = escape {
+                rewrite_upsert_expr_in_place(
+                    e3,
+                    table,
+                    table_name,
+                    current_start,
+                    conflict_rowid_reg,
+                    insertion,
+                )?;
+            }
+        }
+        NotNull(inner) => {
+            rewrite_upsert_expr_in_place(
+                inner,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+        Unary(_, inner) => {
+            rewrite_upsert_expr_in_place(
+                inner,
+                table,
+                table_name,
+                current_start,
+                conflict_rowid_reg,
+                insertion,
+            )?;
+        }
+
+        _ => {}
+    }
+    Ok(())
+}

testing/all.test

@@ -41,3 +41,4 @@ source $testdir/integrity_check.test
 source $testdir/rollback.test
 source $testdir/views.test
 source $testdir/vtab.test
+source $testdir/upsert.test

testing/upsert.test

@@ -0,0 +1,348 @@
+#!/usr/bin/env tclsh
+set testdir [file dirname $argv0]
+source $testdir/tester.tcl
+
+do_execsql_test_on_specific_db {:memory:} upsert-pk-update {
+    CREATE TABLE t (id INTEGER PRIMARY KEY, name);
+    INSERT INTO t VALUES (1,'old');
+    INSERT INTO t VALUES (1,'new') ON CONFLICT DO UPDATE SET name = excluded.name;
+    SELECT * FROM t;
+} {1|new}
+
+do_execsql_test_on_specific_db {:memory:} upsert-pk-do-nothing {
+    CREATE TABLE t (id INTEGER PRIMARY KEY, name);
+    INSERT INTO t VALUES (1,'new');
+    INSERT INTO t VALUES (1,'ignored') ON CONFLICT DO NOTHING;
+    SELECT * FROM t;
+} {1|new}
+
+
+do_execsql_test_on_specific_db {:memory:} upsert-unique-update {
+    CREATE TABLE u (a, b, c);
+    CREATE UNIQUE INDEX u_a ON u(a);
+    INSERT INTO u VALUES (1,10,100);
+    INSERT INTO u VALUES (1,20,200) ON CONFLICT(a) DO UPDATE SET b = excluded.b, c = excluded.c;
+    SELECT * FROM u;
+} {1|20|200}
+
+do_execsql_test_on_specific_db {:memory:} upsert-unique-do-nothing {
+    CREATE TABLE u (a, b, c);
+    CREATE UNIQUE INDEX u_a ON u(a);
+    INSERT INTO u VALUES (1,10,100);
+    INSERT INTO u VALUES (2,30,300) ON CONFLICT(a) DO NOTHING;
+    SELECT * FROM u ORDER BY a;
+} {1|10|100
+2|30|300}
+
+do_execsql_test_on_specific_db {:memory:} upsert-where-guard-no-change {
+    CREATE TABLE g (a UNIQUE, b);
+    INSERT INTO g VALUES (1,'x');
+    INSERT INTO g VALUES (1,'y') ON CONFLICT(a) DO UPDATE SET b = excluded.b WHERE b IS NULL;
+    SELECT * FROM g;
+} {1|x}
+
+do_execsql_test_on_specific_db {:memory:} upsert-where-guard-apply {
+    CREATE TABLE g (a UNIQUE, b);
+    INSERT INTO g VALUES (1,NULL);
+    INSERT INTO g VALUES (1,'y') ON CONFLICT(a) DO UPDATE SET b = excluded.b WHERE b IS NULL;
+    SELECT * FROM g;
+} {1|y}
+
+do_execsql_test_on_specific_db {:memory:} upsert-selfref-and-excluded {
+    CREATE TABLE s (a UNIQUE, b, c);
+    INSERT INTO s VALUES (1,10,'old');
+    INSERT INTO s VALUES (1,99,'new')
+      ON CONFLICT(a) DO UPDATE SET b = b + 1, c = excluded.c;
+    SELECT * FROM s;
+} {1|11|new}
+
+do_execsql_test_on_specific_db {:memory:} upsert-values-mixed-insert-update {
+    CREATE TABLE m (a UNIQUE, b);
+    INSERT INTO m VALUES (1,'one');
+    INSERT INTO m VALUES (1,'uno'), (2,'dos')
+      ON CONFLICT(a) DO UPDATE SET b = excluded.b;
+    SELECT * FROM m ORDER BY a;
+} {1|uno
+2|dos}
+
+do_execsql_test_on_specific_db {:memory:} upsert-select-single {
+    CREATE TABLE s1 (a UNIQUE, b);
+    INSERT INTO s1 VALUES (1,'old');
+    INSERT INTO s1 SELECT 1,'NEW' ON CONFLICT(a) DO UPDATE SET b = excluded.b;
+    SELECT * FROM s1;
+} {1|NEW}
+
+do_execsql_test_on_specific_db {:memory:} upsert-composite-target-orderless {
+    CREATE TABLE c (a, b, val);
+    CREATE UNIQUE INDEX c_ab ON c(a,b);
+    INSERT INTO c VALUES (1,1,'x');
+    INSERT INTO c VALUES (1,1,'y') ON CONFLICT(b,a) DO UPDATE SET val = excluded.val;
+    SELECT val FROM c WHERE a=1 AND b=1;
+} {y}
+
+do_execsql_test_on_specific_db {:memory:} upsert-collate-nocase {
+    CREATE TABLE nc (name TEXT COLLATE NOCASE UNIQUE, v);
+    INSERT INTO nc VALUES ('Alice', 1);
+    INSERT INTO nc VALUES ('aLiCe', 2)
+      ON CONFLICT(name COLLATE NOCASE) DO UPDATE SET v = excluded.v;
+    SELECT * FROM nc;
+} {Alice|2}
+
+do_execsql_test_on_specific_db {:memory:} upsert-returning-update {
+    CREATE TABLE r (id INTEGER PRIMARY KEY, name);
+    INSERT INTO r VALUES (1,'a');
+    INSERT INTO r VALUES (1,'b')
+      ON CONFLICT DO UPDATE SET name = excluded.name
+      RETURNING id, name;
+} {1|b}
+
+do_execsql_test_on_specific_db {:memory:} upsert-returning-insert {
+    CREATE TABLE r2 (id INTEGER PRIMARY KEY, name);
+    INSERT INTO r2 VALUES (2,'c')
+      ON CONFLICT DO UPDATE SET name = excluded.name
+      RETURNING id, name;
+} {2|c}
+
+do_execsql_test_on_specific_db {:memory:} upsert-returning-do-nothing-empty {
+    CREATE TABLE r3 (id INTEGER PRIMARY KEY, name);
+    INSERT INTO r3 VALUES (2,'orig');
+    INSERT INTO r3 VALUES (2,'ignored')
+      ON CONFLICT DO NOTHING
+      RETURNING id, name;
+} {}
+
+do_execsql_test_on_specific_db {:memory:} upsert-rowid-in-set {
+    CREATE TABLE rid (id INTEGER PRIMARY KEY, name);
+    INSERT INTO rid VALUES (5,'foo');
+    INSERT INTO rid VALUES (5,'bar')
+      ON CONFLICT DO UPDATE SET name = printf('id=%d', rowid);
+    SELECT * FROM rid;
+} {5|id=5}
+
+do_execsql_test_in_memory_any_error upsert-notnull-violation {
+    CREATE TABLE nn (a UNIQUE, b NOT NULL);
+    INSERT INTO nn VALUES (1,'x');
+    INSERT INTO nn VALUES (1,'y') ON CONFLICT(a) DO UPDATE SET b = NULL;
+}
+
+do_execsql_test_on_specific_db {:memory:} upsert-updates-other-unique-key {
+    CREATE TABLE idx (a UNIQUE, b UNIQUE);
+    INSERT INTO idx VALUES (1,1);
+    INSERT INTO idx VALUES (1,2) ON CONFLICT(a) DO UPDATE SET b = excluded.b;
+    SELECT * FROM idx;
+} {1|2}
+
+
+do_execsql_test_in_memory_any_error upsert-target-mismatch-errors {
+    CREATE TABLE tm (a, b UNIQUE);
+    INSERT INTO tm VALUES (1,1);
+    INSERT INTO tm VALUES (2,1)
+      ON CONFLICT(a) DO UPDATE SET a = excluded.a;  -- conflict is on b, target is a → error
+}
+
+do_execsql_test_on_specific_db {:memory:} upsert-omitted-target-matches-pk {
+    CREATE TABLE pkalias (a INTEGER PRIMARY KEY, b);
+    INSERT INTO pkalias VALUES (42,'old');
+    INSERT INTO pkalias (a,b) VALUES (42,'new') ON CONFLICT DO UPDATE SET b = excluded.b;
+    SELECT * FROM pkalias;
+} {42|new}
+
+do_execsql_test_on_specific_db {:memory:} upsert-rowvalue-set {
+    CREATE TABLE rv (a INTEGER PRIMARY KEY, b, c);
+    INSERT INTO rv VALUES (1,'x','y');
+    INSERT INTO rv VALUES (1,'B','C')
+      ON CONFLICT DO UPDATE SET (b,c) = (excluded.b, excluded.c);
+    SELECT * FROM rv;
+} {1|B|C}
+
+do_execsql_test_on_specific_db {:memory:} upsert-where-excluded-vs-target {
+    CREATE TABLE wh (a UNIQUE, b);
+    INSERT INTO wh VALUES (1,5);
+    INSERT INTO wh VALUES (1,3)
+      ON CONFLICT(a) DO UPDATE SET b = excluded.b WHERE excluded.b > b;  -- 3 > 5 → no
+    INSERT INTO wh VALUES (1,10)
+      ON CONFLICT(a) DO UPDATE SET b = excluded.b WHERE excluded.b > b; -- 10 > 5 → yes
+    SELECT * FROM wh;
+} {1|10}
+
+do_execsql_test_in_memory_any_error upsert-invalid-qualified-lhs {
+    CREATE TABLE bad (a UNIQUE, b);
+    INSERT INTO bad VALUES (1,'x');
+    INSERT INTO bad VALUES (1,'y')
+      ON CONFLICT(a) DO UPDATE SET excluded.b = 'z';
+}
+
+do_execsql_test_on_specific_db {:memory:} upsert-values-returning-mixed {
+    CREATE TABLE mix (k UNIQUE, v);
+    INSERT INTO mix VALUES (1,'one');
+    INSERT INTO mix VALUES (1,'uno'), (2,'dos')
+      ON CONFLICT(k) DO UPDATE SET v = excluded.v
+      RETURNING k, v
+      ;
+} {1|uno
+2|dos}
+
+do_execsql_test_on_specific_db {:memory:} upsert-collate-implicit-match {
+    CREATE TABLE ci (name TEXT COLLATE NOCASE, v);
+    -- no explicit collate on index
+    CREATE UNIQUE INDEX ci_name ON ci(name);
+    INSERT INTO ci VALUES ('Alice', 1);
+    INSERT INTO ci VALUES ('aLiCe', 2)
+      ON CONFLICT(name COLLATE NOCASE) DO UPDATE SET v = excluded.v;
+    SELECT * FROM ci;
+} {Alice|2}
+
+# Composite index requires exact coverage, targeting too few columns must not match.
+do_execsql_test_in_memory_any_error upsert-composite-target-too-few {
+    CREATE TABLE ct (a, b, val);
+    CREATE UNIQUE INDEX ct_ab ON ct(a,b);
+    INSERT INTO ct VALUES (1,1,'x');
+    INSERT INTO ct VALUES (1,1,'y')
+      ON CONFLICT(a) DO UPDATE SET val = excluded.val;   -- only "a" given → no match → error
+}
+
+# Qualified target (t.a) should match unique index on a.
+do_execsql_test_on_specific_db {:memory:} upsert-qualified-target {
+    CREATE TABLE qt (a UNIQUE, b);
+    INSERT INTO qt VALUES (1,'old');
+    INSERT INTO qt VALUES (1,'new')
+      ON CONFLICT(qt.a) DO UPDATE SET b = excluded.b;
+    SELECT * FROM qt;
+} {1|new}
+
+# Non-simple target expression is not allowed (e.g., lower(name)) → no match → error.
+do_execsql_test_in_memory_any_error upsert-invalid-target-expression {
+    CREATE TABLE it (name, v);
+    CREATE UNIQUE INDEX it_name ON it(name);
+    INSERT INTO it VALUES ('x',1);
+    INSERT INTO it VALUES ('x',2)
+      ON CONFLICT(lower(name)) DO UPDATE SET v = excluded.v;
+}
+
+
+# WHERE with three-valued logic: b < excluded.b is NULL if b IS NULL, should NOT update.
+do_execsql_test_on_specific_db {:memory:} upsert-where-null-3vl-no-update {
+    CREATE TABLE w3 (a UNIQUE, b);
+    INSERT INTO w3 VALUES (1, NULL);
+    INSERT INTO w3 VALUES (1, 5)
+      ON CONFLICT(a) DO UPDATE SET b = excluded.b WHERE b < excluded.b;
+    SELECT * FROM w3;
+} {1|}
+
+# WHERE false on PK conflict → behaves like DO NOTHING.
+do_execsql_test_on_specific_db {:memory:} upsert-pk-where-false {
+    CREATE TABLE pw (id INTEGER PRIMARY KEY, name);
+    INSERT INTO pw VALUES (7,'keep');
+    INSERT INTO pw VALUES (7,'drop')
+      ON CONFLICT DO UPDATE SET name = excluded.name WHERE 0;
+    SELECT * FROM pw;
+} {7|keep}
+
+# WHERE referencing both target and EXCLUDED with arithmetic.
+do_execsql_test_on_specific_db {:memory:} upsert-where-combo {
+    CREATE TABLE wc (a UNIQUE, b);
+    INSERT INTO wc VALUES (1, 10);
+    INSERT INTO wc VALUES (1, 12)
+      ON CONFLICT(a) DO UPDATE SET b = excluded.b
+      WHERE excluded.b >= b + 2;   -- 12 >= 10 + 2 → yes
+    SELECT * FROM wc;
+} {1|12}
+
+# Invalid EXCLUDED reference should error.
+do_execsql_test_in_memory_error_content upsert-invalid-excluded-column {
+    CREATE TABLE xe (a UNIQUE, v);
+    INSERT INTO xe VALUES (1, 'ok');
+    INSERT INTO xe VALUES (1, 'nope')
+      ON CONFLICT(a) DO UPDATE SET v = excluded.not_a_column;
+} {".*no such column.*"}
+
+
+# DO UPDATE changes the *conflicting key column* to a different unique value.
+do_execsql_test_on_specific_db {:memory:} upsert-update-conflicting-key {
+    CREATE TABLE uk (a UNIQUE, b);
+    INSERT INTO uk VALUES (1,'old');
+    INSERT INTO uk VALUES (1,'new')
+      ON CONFLICT(a) DO UPDATE SET a = 2, b = excluded.b;
+    SELECT * FROM uk;
+} {2|new}
+
+# DO UPDATE that writes the same values (no-op) should still succeed.
+do_execsql_test_on_specific_db {:memory:} upsert-noop-update-ok {
+    CREATE TABLE nu (a UNIQUE, b);
+    INSERT INTO nu VALUES (5,'same');
+    INSERT INTO nu VALUES (5,'irrelevant')
+      ON CONFLICT(a) DO UPDATE SET b = b;  -- no change
+    SELECT * FROM nu;
+} {5|same}
+
+# DO UPDATE that would violate a different UNIQUE constraint should error.
+do_execsql_test_in_memory_any_error upsert-update-causes-second-unique-violation {
+    CREATE TABLE uv (a UNIQUE, b UNIQUE);
+    INSERT INTO uv VALUES (1, 10);
+    INSERT INTO uv VALUES (2, 20);
+    INSERT INTO uv VALUES (1, 20)
+      ON CONFLICT(a) DO UPDATE SET b = excluded.b; # would duplicate b=20 from row a=2
+}
+
+# Multi-row VALUES with mixed conflict/non-conflict and WHERE filter in the DO UPDATE.
+do_execsql_test_on_specific_db {:memory:} upsert-multirow-mixed-where {
+    CREATE TABLE mm (k UNIQUE, v);
+    INSERT INTO mm VALUES (1,'one');
+    INSERT INTO mm VALUES (1,'two'), (2,'dos'), (1,'zzz')
+      ON CONFLICT(k) DO UPDATE SET v = excluded.v
+      WHERE excluded.v != 'zzz';  -- skip the 'zzz' update
+    SELECT * FROM mm ORDER BY k;
+} {1|two
+2|dos}
+
+# Omitted target with UNIQUE index: confirm it updates.
+do_execsql_test_on_specific_db {:memory:} upsert-omitted-target-updates-unique {
+    CREATE TABLE ou (a, b);
+    CREATE UNIQUE INDEX ou_a ON ou(a);
+    INSERT INTO ou VALUES (3,'x');
+    INSERT INTO ou VALUES (3,'y')
+      ON CONFLICT DO UPDATE SET b = excluded.b;
+    SELECT * FROM ou;
+} {3|y}
+
+# Target qualified with database.table.column should match too (assuming single db).
+do_execsql_test_on_specific_db {:memory:} upsert-doubly-qualified-target {
+    CREATE TABLE dq (a UNIQUE, b);
+    INSERT INTO dq VALUES (1,'old');
+    INSERT INTO dq VALUES (1,'new')
+      ON CONFLICT(main.dq.a) DO UPDATE SET b = excluded.b;
+    SELECT * FROM dq;
+} {1|new}
+
+
+# TODO: uncomment these when we support collations in indexes 
+# (right now it errors on Parse Error: cannot use expressions in CREATE INDEX)
+#
+# Target specifies BINARY but the unique index is NOCASE: target should NOT match, so expect error
+# do_execsql_test_in_memory_any_error upsert-collate-target-mismatch {
+#     CREATE TABLE cm (name TEXT, v);
+#     CREATE UNIQUE INDEX cm_name_nocase ON cm(name COLLATE NOCASE);
+#     INSERT INTO cm VALUES ('Alice', 1);
+#     INSERT INTO cm VALUES ('aLiCe', 2)
+#       ON CONFLICT(name COLLATE BINARY) DO UPDATE SET v = excluded.v;
+# }
+#
+# do_execsql_test_on_specific_db {:memory:} upsert-collate-omitted-target-matches {
+#     CREATE TABLE co (name TEXT, v);
+#     CREATE UNIQUE INDEX co_name_nocase ON co(name COLLATE NOCASE);
+#     INSERT INTO co VALUES ('Alice', 1);
+#     INSERT INTO co VALUES ('aLiCe', 9)
+#       ON CONFLICT DO UPDATE SET v = excluded.v;
+#     SELECT * FROM co;
+# } {Alice|9}
+#
+#
+# do_execsql_test_on_specific_db {:memory:} upsert-composite-collate-orderless {
+#     CREATE TABLE cc (name TEXT, city TEXT, val);
+#     CREATE UNIQUE INDEX cc_nc ON cc(name COLLATE NOCASE, city);
+#     INSERT INTO cc VALUES ('Alice','SF','old');
+#     INSERT INTO cc VALUES ('aLiCe','SF','new')
+#       ON CONFLICT(city, name COLLATE NOCASE) DO UPDATE SET val = excluded.val;
+#     SELECT * FROM cc;
+# } {Alice|SF|new}
+