rework deposits sharding, towards making aggregator faster (not necessarily done)

1 files changed, 185 insertions, 81 deletions

diff --git a/src/exchangedb/exchange-0001.sql b/src/exchangedb/exchange-0001.sql
index e723a3677..568779f97 100644
--- a/src/exchangedb/exchange-0001.sql
+++ b/src/exchangedb/exchange-0001.sql
@@ -610,7 +610,7 @@ CREATE TABLE IF NOT EXISTS deposits
   (deposit_serial_id BIGINT GENERATED BY DEFAULT AS IDENTITY -- PRIMARY KEY
   ,shard INT8 NOT NULL
   ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32) -- REFERENCES known_coins (coin_pub) ON DELETE CASCADE
-  ,known_coin_id BIGINT NOT NULL -- REFERENCES known_coins (known_coin_id) ON DELETE CASCADE
+  ,known_coin_id BIGINT NOT NULL -- REFERENCES known_coins (known_coin_id) ON DELETE CASCADE --- FIXME: column needed???
   ,amount_with_fee_val INT8 NOT NULL
   ,amount_with_fee_frac INT4 NOT NULL
   ,wallet_timestamp INT8 NOT NULL
@@ -626,22 +626,11 @@ CREATE TABLE IF NOT EXISTS deposits
   ,done BOOLEAN NOT NULL DEFAULT FALSE
   ,extension_blocked BOOLEAN NOT NULL DEFAULT FALSE
   ,extension_details_serial_id INT8 REFERENCES extension_details (extension_details_serial_id) ON DELETE CASCADE
-  ,UNIQUE (shard, coin_pub, merchant_pub, h_contract_terms)
+  ,UNIQUE (coin_pub, merchant_pub, h_contract_terms)
   )
-  PARTITION BY HASH (shard); -- FIXME: why not BY RANGE? RANGE would seem better for 'deposits_get_ready'!
+  PARTITION BY HASH (coin_pub);
 -- FIXME:
--- new idea: partition deposits by coin_pub (remove deposits_by_coin)
--- define 'ready' == ! (tiny || done || blocked)
--- add new deposits_by_ready (on shard + wire_deadline), select by shard, then ready + deadline
---         -- use triggers to ONLY include 'ready' deposits (delete on update)!
---         -- use multi-level partitions: Hash(shard) + Range(wire_deadline/sec)
--- add new deposits_by_match (on shard + refund_deadline)
---         -- use triggers to ONLY include 'ready' deposits (delete on update)!
---         -- use multi-level partitions: Hash(shard) + Range(refund_deadline/sec)
--- => first we select per-merchant shard, basically stay on the same system as other ops for the same merchant
--- => second we select by deadline, use enough values so that _usually_ the aggregator
---    and the 'insert' process _can_ work on different shards!
--- => the latter could be achieved by dynamically (!) creating/deleting partitions:
+-- TODO: dynamically (!) creating/deleting partitions:
 --    create new partitions 'as needed', drop old ones once the aggregator has made
 --    them empty; as 'new' deposits will always have deadlines in the future, this
 --    would basically guarantee no conflict between aggregator and exchange service!
@@ -683,31 +672,15 @@ COMMENT ON COLUMN deposits.extension_details_serial_id
 COMMENT ON COLUMN deposits.tiny
   IS 'Set to TRUE if we decided that the amount is too small to ever trigger a wire transfer by itself (requires real aggregation)';
 
+-- FIXME: we sometimes go ONLY by 'deposit_serial_id',
+--        check if queries could be improved by adding shard or adding another index without shard here, or inverting the order of the index here!
 CREATE INDEX IF NOT EXISTS deposits_deposit_by_serial_id_index
   ON deposits
   (shard,deposit_serial_id);
-CREATE INDEX IF NOT EXISTS deposits_for_get_ready_index
-  ON deposits
-  (shard ASC
-  ,done
-  ,extension_blocked
-  ,tiny
-  ,wire_deadline ASC
-  );
-COMMENT ON INDEX deposits_for_get_ready_index
-  IS 'for deposits_get_ready';
 
-CREATE INDEX IF NOT EXISTS deposits_for_iterate_matching_index
+CREATE INDEX IF NOT EXISTS deposits_by_coin_pub_index
   ON deposits
-  (shard
-  ,merchant_pub
-  ,wire_target_h_payto
-  ,done
-  ,extension_blocked
-  ,refund_deadline ASC
-  );
-COMMENT ON INDEX deposits_for_iterate_matching_index
-  IS 'for deposits_iterate_matching';
+  (coin_pub);
 
 
 CREATE TABLE IF NOT EXISTS deposits_default
@@ -732,66 +705,198 @@ $$;
 SELECT add_constraints_to_deposits_partition('default');
 
 
+CREATE TABLE IF NOT EXISTS deposits_by_ready
+  (wire_deadline INT8 NOT NULL
+  ,shard INT8 NOT NULL
+  ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32)
+  ,deposit_serial_id INT8
+  )
+  PARTITION BY RANGE (wire_deadline);
+COMMENT ON TABLE deposits_by_ready
+  IS 'Enables fast lookups for deposits_get_ready, auto-populated via TRIGGER below';
+
+CREATE INDEX IF NOT EXISTS deposits_by_ready_main_index
+  ON deposits_by_ready
+  (wire_deadline ASC, shard ASC, coin_pub);
+
+CREATE TABLE IF NOT EXISTS deposits_by_ready_default
+  PARTITION OF deposits_by_ready
+  DEFAULT;
 
-CREATE TABLE IF NOT EXISTS deposits_by_coin
-  (deposit_serial_id BIGINT
+
+CREATE TABLE IF NOT EXISTS deposits_for_matching
+  (refund_deadline INT8 NOT NULL
   ,shard INT8 NOT NULL
   ,coin_pub BYTEA NOT NULL CHECK (LENGTH(coin_pub)=32)
+  ,deposit_serial_id INT8
   )
-  PARTITION BY HASH (coin_pub);
-COMMENT ON TABLE deposits_by_coin
-  IS 'Enables fast lookups of deposit by coin_pub, auto-populated via TRIGGER below';
+  PARTITION BY RANGE (refund_deadline);
+COMMENT ON TABLE deposits_for_matching
+  IS 'Enables fast lookups for deposits_iterate_matching, auto-populated via TRIGGER below';
 
-CREATE INDEX IF NOT EXISTS deposits_by_coin_main_index
-  ON deposits_by_coin
-  (coin_pub);
+CREATE INDEX IF NOT EXISTS deposits_for_matching_main_index
+  ON deposits_for_matching
+  (refund_deadline ASC, shard, coin_pub);
 
-CREATE TABLE IF NOT EXISTS deposits_by_coin_default
-  PARTITION OF deposits_by_coin
-  FOR VALUES WITH (MODULUS 1, REMAINDER 0);
+CREATE TABLE IF NOT EXISTS deposits_for_matching_default
+  PARTITION OF deposits_for_matching
+  DEFAULT;
+  
 
-CREATE OR REPLACE FUNCTION deposits_by_coin_insert_trigger()
+CREATE OR REPLACE FUNCTION deposits_insert_trigger()
   RETURNS trigger
   LANGUAGE plpgsql
   AS $$
+DECLARE
+  is_ready BOOLEAN;
+DECLARE
+  is_tready BOOLEAN; -- is ready, but may be tiny
 BEGIN
-  INSERT INTO deposits_by_coin
-    (deposit_serial_id
-    ,shard
-    ,coin_pub)
-  VALUES
-    (NEW.deposit_serial_id
-    ,NEW.shard
-    ,NEW.coin_pub);
+  is_ready  = NOT (NEW.done OR NEW.tiny OR NEW.extension_blocked);
+  is_tready = NOT (NEW.done OR NEW.extension_blocked);
+
+  IF (is_ready)
+  THEN
+    INSERT INTO deposits_by_ready
+      (wire_deadline
+      ,shard
+      ,coin_pub
+      ,deposit_serial_id)
+    VALUES
+      (NEW.wire_deadline
+      ,NEW.shard
+      ,NEW.coin_pub
+      ,NEW.deposit_serial_id);
+  END IF;
+  IF (is_tready)
+  THEN
+    INSERT INTO deposits_for_matching
+      (refund_deadline
+      ,shard
+      ,coin_pub
+      ,deposit_serial_id)
+    VALUES
+      (NEW.refund_deadline
+      ,NEW.shard
+      ,NEW.coin_pub
+      ,NEW.deposit_serial_id);
+  END IF;
   RETURN NEW;
 END $$;  
-COMMENT ON FUNCTION deposits_by_coin_insert_trigger()
-  IS 'Replicate deposit inserts into deposits_by_coin table.';
+COMMENT ON FUNCTION deposits_insert_trigger()
+  IS 'Replicate deposit inserts into materialized indices.';
 
 CREATE TRIGGER deposits_on_insert
   AFTER INSERT
    ON deposits
-   FOR EACH ROW EXECUTE FUNCTION deposits_by_coin_insert_trigger();
+   FOR EACH ROW EXECUTE FUNCTION deposits_insert_trigger();
 
-CREATE OR REPLACE FUNCTION deposits_by_coin_delete_trigger()
+CREATE OR REPLACE FUNCTION deposits_update_trigger()
   RETURNS trigger
   LANGUAGE plpgsql
   AS $$
+DECLARE
+  was_ready BOOLEAN;
+DECLARE
+  is_ready BOOLEAN;
+DECLARE
+  was_tready BOOLEAN; -- was ready, but may be tiny
+DECLARE
+  is_tready BOOLEAN; -- is ready, but may be tiny
 BEGIN
-  DELETE FROM deposits_by_coin
-   WHERE coin_pub = OLD.coin_pub
-     AND shard = OLD.shard
-     AND deposit_serial_id = OLD.deposit_serial_id;
-  RETURN OLD;
+  was_ready = NOT (OLD.done OR OLD.tiny OR OLD.extension_blocked);
+  is_ready  = NOT (NEW.done OR NEW.tiny OR NEW.extension_blocked);
+  was_tready = NOT (OLD.done OR OLD.extension_blocked);
+  is_tready  = NOT (NEW.done OR NEW.extension_blocked);
+  IF (was_ready AND NOT is_ready)
+  THEN
+    DELETE FROM deposits_by_ready
+     WHERE wire_deadline = OLD.wire_deadline
+       AND shard = OLD.shard
+       AND coin_pub = OLD.coin_pub
+       AND deposit_serial_id = OLD.deposit_serial_id;
+  END IF;
+  IF (was_tready AND NOT is_tready)
+  THEN
+    DELETE FROM deposits_for_matching
+     WHERE refund_deadline = OLD.refund_deadline
+       AND shard = OLD.shard
+       AND coin_pub = OLD.coin_pub
+       AND deposit_serial_id = OLD.deposit_serial_id;
+  END IF;
+  IF (is_ready AND NOT was_ready)
+  THEN
+    INSERT INTO deposits_by_ready
+      (wire_deadline
+      ,shard
+      ,coin_pub
+      ,deposit_serial_id)
+    VALUES
+      (NEW.wire_deadline
+      ,NEW.shard
+      ,NEW.coin_pub
+      ,NEW.deposit_serial_id);
+  END IF;
+  IF (is_tready AND NOT was_tready)
+  THEN
+    INSERT INTO deposits_for_matching
+      (refund_deadline
+      ,shard
+      ,coin_pub
+      ,deposit_serial_id)
+    VALUES
+      (NEW.refund_deadline
+      ,NEW.shard
+      ,NEW.coin_pub
+      ,NEW.deposit_serial_id);
+  END IF;
+  RETURN NEW;
+END $$;
+COMMENT ON FUNCTION deposits_update_trigger()
+  IS 'Replicate deposits changes into materialized indices.';
+
+CREATE TRIGGER deposits_on_update
+  AFTER UPDATE
+    ON deposits
+   FOR EACH ROW EXECUTE FUNCTION deposits_update_trigger();
+
+CREATE OR REPLACE FUNCTION deposits_delete_trigger()
+  RETURNS trigger
+  LANGUAGE plpgsql
+  AS $$
+DECLARE
+  was_ready BOOLEAN;
+DECLARE
+  was_tready BOOLEAN; -- is ready, but may be tiny
+BEGIN
+  was_ready  = NOT (OLD.done OR OLD.tiny OR OLD.extension_blocked);
+  was_tready = NOT (OLD.done OR OLD.extension_blocked);
+
+  IF (was_ready)
+  THEN
+    DELETE FROM deposits_by_ready
+     WHERE wire_deadline = OLD.wire_deadline
+       AND shard = OLD.shard
+       AND coin_pub = OLD.coin_pub
+       AND deposit_serial_id = OLD.deposit_serial_id;
+  END IF;
+  IF (was_tready)
+  THEN
+    DELETE FROM deposits_for_matching
+     WHERE refund_deadline = OLD.refund_deadline
+       AND shard = OLD.shard
+       AND coin_pub = OLD.coin_pub
+       AND deposit_serial_id = OLD.deposit_serial_id;
+  END IF;
+  RETURN NEW;
 END $$;  
-COMMENT ON FUNCTION deposits_by_coin_delete_trigger()
-  IS 'Replicate deposits deletions into deposits_by_coin table.';
+COMMENT ON FUNCTION deposits_delete_trigger()
+  IS 'Replicate deposit deletions into materialized indices.';
 
 CREATE TRIGGER deposits_on_delete
   AFTER DELETE
-    ON deposits
-   FOR EACH ROW EXECUTE FUNCTION deposits_by_coin_delete_trigger();
-
+   ON deposits
+   FOR EACH ROW EXECUTE FUNCTION deposits_delete_trigger();
 
 
 CREATE TABLE IF NOT EXISTS refunds
@@ -2011,7 +2116,7 @@ DECLARE
 BEGIN
 -- Shards: INSERT extension_details (by extension_details_serial_id)
 --         INSERT wire_targets (by h_payto), on CONFLICT DO NOTHING;
---         INSERT deposits (by shard + merchant_pub + h_payto), ON CONFLICT DO NOTHING;
+--         INSERT deposits (by coin_pub, shard), ON CONFLICT DO NOTHING;
 --         UPDATE known_coins (by coin_pub)
 
 IF NOT NULL in_extension_details
@@ -2356,27 +2461,26 @@ DECLARE
 DECLARE
   deposit_frac INT8; -- amount that was originally deposited
 BEGIN
--- Shards: SELECT deposits (by shard, coin_pub, h_contract_terms, merchant_pub)
+-- Shards: SELECT deposits (coin_pub, shard, h_contract_terms, merchant_pub)
 --         INSERT refunds (by deposit_serial_id, rtransaction_id) ON CONFLICT DO NOTHING
 --         SELECT refunds (by deposit_serial_id)
 --         UPDATE known_coins (by coin_pub)
 
 SELECT
-   dep.deposit_serial_id
-  ,dep.amount_with_fee_val
-  ,dep.amount_with_fee_frac
-  ,dep.done
+   deposit_serial_id
+  ,amount_with_fee_val
+  ,amount_with_fee_frac
+  ,done
 INTO
    dsi
   ,deposit_val
   ,deposit_frac
   ,out_gone
-FROM deposits_by_coin dbc
-  JOIN deposits dep USING (shard,deposit_serial_id)
- WHERE dbc.coin_pub=in_coin_pub
-  AND dep.shard=in_deposit_shard
-  AND dep.merchant_pub=in_merchant_pub
-  AND dep.h_contract_terms=in_h_contract_terms;
+FROM deposits
+ WHERE coin_pub=in_coin_pub
+  AND shard=in_deposit_shard
+  AND merchant_pub=in_merchant_pub
+  AND h_contract_terms=in_h_contract_terms;
 
 IF NOT FOUND
 THEN