Examine query inside postgres function

Question

I would like to use a query result as a condition on what action to perform inside a postgres function. Here is my attempt:

CREATE OR REPLACE FUNCTION upsert_bin_effect(operation_id integer, typ text, asset_id integer, qty double precision) RETURNS boolean AS $$
BEGIN
    existing_locked := (SELECT SUM(qty) AS qty INTO existing FROM bin_effect be WHERE be.operation_id = operation_id AND be.type = typ AND be.asset_id = asset_id AND be.locked = true GROUP BY be.operation_id, be.type, be.asset_id);
    qty = qty - existing.locked.qty
    
    existing_unlocked := (SELECT * INTO existing FROM bin_effect be WHERE be.operation_id = operation_id AND be.type = typ AND be.asset_id = asset_id AND (be.locked = false OR be.locked IS NULL));
    IF EXISTS(existing_unlocked)
    THEN            
        UPDATE bin_effect be SET be.qty = qty WHERE be.id = existing_unlocked.id
    ELSE
        INSERT INTO bin_effect be (be.operation_id, be."type", be.asset_id, be.qty) VALUES (operation_id, typ, asset_id, qty);
    END IF;
    
END;
$$ LANGUAGE plpgsql;

existing_locked can have multiple rows, I'd like to subtract the sum of existing_locked.qty from the incoming qty. Then, update the record that is not locked (i.e. in existing_unlocked), if it exists, with the net qty - otherwise insert a new row with the net qty.

If we assume there is a table with the following data:

operation_id, type, asset_id, qty, locked
1, 'A', 1, 10, true
1, 'A', 1, 20, true
1, 'A', 2, 5, true
1, 'A', 2, 15, null

The following call:

upsert_bin_effect(1, 'A', 1, 100)

should result in:

operation_id, type, asset_id, qty, locked
1, 'A', 1, 10, true
1, 'A', 1, 20, true
1, 'A', 2, 5, true
1, 'A', 2, 15, null
1, 'A', 1, 70, null

The following call:

upsert_bin_effect(1, 'A', 2, 100)

should result in:

operation_id, type, asset_id, qty, locked
1, 'A', 1, 10, true
1, 'A', 1, 20, true
1, 'A', 2, 5, true
1, 'A', 2, 95, null

The following call:

upsert_bin_effect(1, 'A', 3, 100)

should result in:

operation_id, type, asset_id, qty, locked
1, 'A', 1, 10, true
1, 'A', 1, 20, true
1, 'A', 2, 5, true
1, 'A', 2, 95, null
1, 'A', 3, 100, null

To better describe how I'd like this function to work here is some javascript pseudocode which implements the desired functionality:

// these are mock result sets, assume they were queried where operation_id, type, asset_id are equal and locked is true/falsy respectively.
const existingLocked = [];
const existingUnlocked = [];

function upsert_bin_effect(operationId, typ, assetId, qty) {
    const lockedQty = existingLocked.reduce((sum, r) => sum + r.qty, 0);
  
  // incoming qty represents the total qty. lockedQty represents qty for locked rows (rows we cannot update)
  // If there is nonzero lockedQty, we subtract it from qty because when we upsert qty
  // we need to ensure that all rows qty will sum to the incoming qty.
  qty = qty - lockedQty;
  
  // existingUnlocked should only ever contain a single row (because of the upsert logic below)
  if (!!existingUnlocked[0]) {
    // if there is an existing unlocked row, update it with the (incoming qty - any locked qty)
    existingUnlocked[0].update('qty', qty);
  }
  else {
    // otherwise insert a new row with qty = (incoming qty - any locked qty)
    db.binEffect.insert(operationId, typ, assetId, qty)
  }
}

I am pretty new to sql function programming. Does this make sense? If not, how can I accomplish what I am trying to do?

Answer 1

Let me just preface this by saying, this could be a bit of an XY problem. Postgres has mechanism for locking rows in a transaction, but without having to modify them. This means other transactions cannot update them, but may still be able to read their pre-locked state. This is known as SELECT FOR UPDATE. So your upsert_bin_effect could read the locked rows, but would not be able to modify them.

If you create a unique index on unlocked rows, then you can do this in a single query, using INSERT ... ON CONFLICT UPDATE. Given this example schema:

create table bin_effect (
    operation_id integer not null,
    typ text not null,
    asset_id integer not null,
    qty double precision not null,
    locked boolean not null
    -- don't understand why you were using null to indicated that the row was not locked
);

-- create a partial unique index that guarantees at most one unlocked row per "key"
create unique index bin_effect_unqiue_unlocked
    on bin_effect (operation_id, typ, asset_id) where not locked;

Then given a table initialised from the following insert:

insert into bin_effect values
    (1, 'A', 1, 10, true),
    (1, 'A', 1, 20, true),
    (1, 'A', 2, 5, true),
    (1, 'A', 2, 15, false);

Then the following query will insert an unlocked row or update the unlocked row for the given operation_id, typ and asset_id. You can then either use this query as a parameterised query in either directly or as part of a stored function. NB. This is a raw query for operation_id = 1, typ = 'A', asset_id = 1 and a new quantity of 100.

-- pre-calculate the desired qty, so we only compute it once
with new_value as (
    select
        -- use coalesce for when there no matching rows in table (locked or otherwise)
        100 - coalesce(sum(qty), 0) as qty
    from bin_effect
    where 
        operation_id = 1
        and typ = 'A'
        and asset_id = 1
        and locked
)
-- try to insert a new row
insert into bin_effect (operation_id, typ, asset_id, locked, qty)
    values
        (1, 'A', 1, false, (select qty from new_value))
    -- if the insertion fails, then update the pre-existing row
    on conflict (operation_id, typ, asset_id) where not locked 
        do update set qty = (select qty from new_value) 
;

As a stored function:

create or replace function upsert_bin_effect(operation_id_ integer, typ_ text, asset_id_ integer, new_qty double precision)
returns double precision as $$
    with new_value as (
        select 
            new_qty - coalesce(sum(qty), 0) as qty
        from bin_effect
        where 
            operation_id = operation_id_
            and typ = typ_
            and asset_id = asset_id_
            and locked
    )
    insert into bin_effect (operation_id, typ, asset_id, locked, qty)
        values
            (operation_id_, typ_, asset_id_, false, (select qty from new_value))
        on conflict (operation_id, typ, asset_id) where not locked
            do update set qty = (select qty from new_value)
        returning qty
    ;
$$
language sql;

Example usage and output:

postgres=# select upsert_bin_effect(1, 'A', 1, 100);
 upsert_bin_effect
-------------------
                70
(1 row)


postgres=# select upsert_bin_effect(1, 'A', 2, 100);
 upsert_bin_effect
-------------------
                95
(1 row)


postgres=# select upsert_bin_effect(1, 'A', 3, 100);
 upsert_bin_effect
-------------------
               100
(1 row)


postgres=# table bin_effect;
 operation_id | typ | asset_id | qty | locked
--------------+-----+----------+-----+--------
            1 | A   |        1 |  10 | t
            1 | A   |        1 |  20 | t
            1 | A   |        2 |   5 | t
            1 | A   |        1 |  70 | f
            1 | A   |        2 |  95 | f
            1 | A   |        3 | 100 | f
(6 rows)

Answer 2

There several issues with this function before you get to what you want:

DECLARE existing RESULT

--There is no RESULT type and you do end with ; So:

DECLARE existing RECORD;

existing_locked is not declared so assignment to it will fail. Same for existing_unlocked.

qty = qty - existing.locked.qty does not end with a ;

I would spend some time here:

https://www.postgresql.org/docs/12/plpgsql-structure.html

From your comment below I'm not seeing that the new examples match what you say you want:

operation_id, type, asset_id, qty, locked
1, 'A', 1, 10, true
1, 'A', 1, 20, true
1, 'A', 2, 5, true
1, 'A', 2, 15, null
1, 'A', 1, 70, null

--The following call:

upsert_bin_effect(1, 'A', 2, 100)

--should result in:

operation_id, type, asset_id, qty, locked
1, 'A', 1, 10, true
1, 'A', 1, 20, true
1, 'A', 2, 5, true
1, 'A', 2, 95, null

"existing can have multiple rows, I'd like to subtract the sum of existing.qty from the incoming qty for any locked rows. Then, update any record that is not locked with the incoming qty if an unlocked row exists, otherwise insert a new one."

I would have thought the result would be:

operation_id, type, asset_id, qty, locked
1, 'A', 1, 10, true
1, 'A', 1, 20, true
1, 'A', 2, 5, true
1, 'A', 2, 115, null

UPDATE

A stab at what I think you want. Obviously not tested, but it should be a reasonable starting point.

CREATE OR REPLACE FUNCTION upsert_bin_effect(operation_id integer, typ text, asset_id integer, qty double precision) RETURNS boolean AS $$
DECLARE
    existing_locked RECORD;
    existing_unlocked RECORD;
    net_qty float;
    unlocked_ct integer;

BEGIN
    existing_locked := (SELECT SUM(qty) AS qty INTO existing FROM bin_effect be WHERE be.operation_id = operation_id AND be.type = typ AND be.asset_id = asset_id AND be.locked = true GROUP BY be.operation_id, be.type, be.asset_id);
    net_qty = qty - existing.locked.qty;
    
    existing_unlocked := (SELECT * INTO existing FROM bin_effect be WHERE be.operation_id = operation_id AND be.type = typ AND be.asset_id = asset_id AND (be.locked = false OR be.locked IS NULL));
GET DIAGNOSTICS unlocked_ct = ROW_COUNT;
    IF EXISTS(existing_unlocked)
    THEN
        IF unlocked_ct = 1 THEN
           UPDATE bin_effect be SET be.qty = net_qty WHERE be.id = existing_unlocked.id;
        ELSEIF unlocked_ct > 1
            --Not sure if you want this to happen, included as example.
            RAISE EXCEPTION 'Too many unlocked row';
        END IF;
    ELSE
        INSERT INTO bin_effect be (be.operation_id, be."type", be.asset_id, be.qty) VALUES (operation_id, typ, asset_id, qty);
    END IF;
    
END;
$$ LANGUAGE plpgsql;