Function plan_upsert_with_on_conflict

nodedb-sql/src/planner/dml.rs:236–304

Plan an `INSERT ... ON CONFLICT DO UPDATE SET` statement.

(
    ins: &ast::Insert,
    catalog: &dyn SqlCatalog,
    on_conflict_updates: Vec<(String, SqlExpr)>,
)

/// Plan an `INSERT ... ON CONFLICT DO UPDATE SET` statement.
fn plan_upsert_with_on_conflict(
    ins: &ast::Insert,
    catalog: &dyn SqlCatalog,
    on_conflict_updates: Vec<(String, SqlExpr)>,
) -> Result<Vec<SqlPlan>> {
    let table_name = match &ins.table {
        ast::TableObject::TableName(name) => normalize_object_name_checked(name)?,
        ast::TableObject::TableFunction(_) => {
            return Err(SqlError::Unsupported {
                detail: "INSERT ... ON CONFLICT on a table function is not supported".into(),
            });
        }
    };
    let info = catalog
        .get_collection(DatabaseId::DEFAULT, &table_name)?
        .ok_or_else(|| SqlError::UnknownTable {
            name: table_name.clone(),
        })?;

    let columns: Vec<String> = ins.columns.iter().map(normalize_ident).collect();

    let source = ins.source.as_ref().ok_or_else(|| SqlError::Parse {
        detail: "INSERT ... ON CONFLICT requires VALUES".into(),
    })?;
    let rows_ast = match &*source.body {
        ast::SetExpr::Values(values) => &values.rows,
        _ => {
            return Err(SqlError::Unsupported {
                detail: "INSERT ... ON CONFLICT source must be VALUES".into(),
            });
        }
    };

    // KV: `INSERT ... ON CONFLICT (key) DO UPDATE SET ...` is an opt-in
    // overwrite — same physical semantics as UPSERT, with the optional
    // per-row assignments carried through for the Data Plane to apply
    // against the existing row.
    if info.engine == EngineType::KeyValue {
        return build_kv_insert_plan(
            table_name,
            &columns,
            rows_ast,
            KvInsertIntent::Put,
            on_conflict_updates,
            info.primary_key.as_deref(),
        );
    }

    let rows = convert_value_rows(&columns, rows_ast)?;
    let column_defaults: Vec<(String, String)> = info
        .columns
        .iter()
        .filter_map(|c| c.default.as_ref().map(|d| (c.name.clone(), d.clone())))
        .collect();
    let column_schema: Vec<(String, String)> = info
        .columns
        .iter()
        .filter_map(|c| c.raw_type.as_ref().map(|t| (c.name.clone(), t.clone())))