add sharding logic

src/exchange-tools/taler-exchange-dbinit.c

@@ -49,6 +49,10 @@ static int gc_db;
  */
 static uint32_t num_partitions;
 
+/**
+ * -S option: setup a sharded database
+ */
+static uint32_t num_shards;
 
 /**
  * Main function that will be run.
@@ -113,6 +117,24 @@ run (void *cls,
       return;
     }
   }
+  else if (1 <
+           num_shards)
+  {
+    if (GNUNET_OK != plugin->setup_shards (plugin->cls, num_shards))
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Could not setup shards. Aborting\n");
+      if (GNUNET_OK != plugin->drop_tables (plugin->cls))
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                    "Could not drop tables after failed shard setup, please delete the DB manually\n");
+      }
+      TALER_EXCHANGEDB_plugin_unload (plugin);
+      plugin = NULL;
+      global_ret = EXIT_NOTINSTALLED;
+      return;
+    }
+  }
   if (gc_db || clear_shards)
   {
     if (GNUNET_OK !=
@@ -178,6 +200,11 @@ main (int argc,
                                "NUMBER",
                                "Setup a partitioned database where each table which can be partitioned holds NUMBER partitions on a single DB node (NOTE: this is different from sharding)",
                                &num_partitions),
+    GNUNET_GETOPT_option_uint ('S',
+                               "shard",
+                               "NUMBER",
+                               "Setup a sharded database whit N shards",
+                               &num_shards),
     GNUNET_GETOPT_OPTION_END
   };
   enum GNUNET_GenericReturnValue ret;

src/exchangedb/Makefile.am

@@ -19,9 +19,19 @@ sql_DATA = \
   benchmark-0001.sql \
   exchange-0000.sql \
   exchange-0001.sql \
-  partition-0001.sql \
+  shard-0001.sql \
   drop0001.sql
 
+BUILT_SOURCES = \
+  shard-0001.sql \
+  exchange-0001.sql
+
+exchange-0001.sql: common-0001.sql exchange-0001-part.sql
+	cat common-0001.sql exchange-0001-part.sql >$@
+
+shard-0001.sql: common-0001.sql
+	cp common-0001.sql $@
+
 EXTRA_DIST = \
   exchangedb.conf \
   exchangedb-postgres.conf \

src/exchangedb/drop0001.sql

@@ -43,43 +43,65 @@ DROP TABLE IF EXISTS wire_accounts CASCADE;
 DROP TABLE IF EXISTS signkey_revocations CASCADE;
 DROP TABLE IF EXISTS work_shards CASCADE;
 DROP TABLE IF EXISTS prewire CASCADE;
+DROP FUNCTION IF EXISTS create_table_prewire;
 DROP TABLE IF EXISTS recoup CASCADE;
+DROP FUNCTION IF EXISTS create_table_recoup;
 DROP FUNCTION IF EXISTS add_constraints_to_recoup_partition;
 DROP TABLE IF EXISTS recoup_refresh CASCADE;
+DROP FUNCTION IF EXISTS create_table_recoup_refresh;
 DROP FUNCTION IF EXISTS add_constraints_to_recoup_refresh_partition;
+DROP TABLE IF EXISTS aggregation_transient CASCADE;
+DROP FUNCTION IF EXISTS create_table_aggregation_transient;
 DROP TABLE IF EXISTS aggregation_tracking CASCADE;
+DROP FUNCTION IF EXISTS create_table_aggregation_tracking;
 DROP FUNCTION IF EXISTS add_constraints_to_aggregation_tracking_partition;
 DROP TABLE IF EXISTS wire_out CASCADE;
+DROP FUNCTION IF EXISTS create_table_wire_out;
 DROP FUNCTION IF EXISTS add_constraints_to_wire_out_partition;
 DROP TABLE IF EXISTS wire_targets CASCADE;
+DROP FUNCTION IF EXISTS create_table_wire_targets;
 DROP FUNCTION IF EXISTS add_constraints_to_wire_targets_partition;
 DROP TABLE IF EXISTS wire_fee CASCADE;
 DROP TABLE IF EXISTS deposits CASCADE;
+DROP FUNCTION IF EXISTS create_table_deposits;
 DROP TABLE IF EXISTS deposits_by_ready CASCADE;
+DROP FUNCTION IF EXISTS create_table_deposits_by_ready;
 DROP TABLE IF EXISTS deposits_for_matching CASCADE;
+DROP FUNCTION IF EXISTS create_table_deposits_for_matching;
 DROP FUNCTION IF EXISTS add_constraints_to_deposits_partition;
 DROP TABLE IF EXISTS extension_details CASCADE;
 DROP TABLE IF EXISTS refunds CASCADE;
+DROP FUNCTION IF EXISTS create_table_refunds;
 DROP FUNCTION IF EXISTS add_constraints_to_refunds_partition;
 DROP TABLE IF EXISTS refresh_commitments CASCADE;
+DROP FUNCTION IF EXISTS create_table_refresh_commitments;
 DROP FUNCTION IF EXISTS add_constraints_to_refresh_commitments_partition;
 DROP TABLE IF EXISTS refresh_revealed_coins CASCADE;
+DROP FUNCTION IF EXISTS create_table_refresh_revealed_coins;
 DROP FUNCTION IF EXISTS add_constraints_to_refresh_revealed_coins_partition;
 DROP TABLE IF EXISTS refresh_transfer_keys CASCADE;
+DROP FUNCTION IF EXISTS create_table_refresh_transfer_keys;
 DROP FUNCTION IF EXISTS add_constraints_to_refresh_transfer_keys_partition;
 DROP TABLE IF EXISTS known_coins CASCADE;
+DROP FUNCTION IF EXISTS create_table_known_coins;
 DROP FUNCTION IF EXISTS add_constraints_to_known_coins_partition;
 DROP TABLE IF EXISTS reserves_close CASCADE;
+DROP FUNCTION IF EXISTS create_table_reserves_close;
 DROP FUNCTION IF EXISTS add_constraints_to_reserves_close_partition;
 DROP TABLE IF EXISTS reserves_out CASCADE;
+DROP FUNCTION IF EXISTS create_table_reserves_out;
 DROP TABLE IF EXISTS reserves_out_by_reserve CASCADE;
+DROP FUNCTION IF EXISTS create_table_reserves_out_by_reserve;
 DROP FUNCTION IF EXISTS add_constraints_to_reserves_out_partition;
 DROP TABLE IF EXISTS reserves_in CASCADE;
+DROP FUNCTION IF EXISTS create_table_reserves_in;
 DROP FUNCTION IF EXISTS add_constraints_to_reserves_in_partition;
 DROP TABLE IF EXISTS reserves CASCADE;
+DROP FUNCTION IF EXISTS create_table_reserves;
 DROP TABLE IF EXISTS denomination_revocations CASCADE;
 DROP TABLE IF EXISTS denominations CASCADE;
 DROP TABLE IF EXISTS cs_nonce_locks CASCADE;
+DROP FUNCTION IF EXISTS create_table_cs_nonce_locks;
 DROP FUNCTION IF EXISTS add_constraints_to_cs_nonce_locks_partition;
 
 DROP TABLE IF EXISTS global_fee CASCADE;
@@ -122,14 +144,14 @@ DROP FUNCTION IF EXISTS exchange_do_account_merge;
 DROP FUNCTION IF EXISTS exchange_do_history_request;
 DROP FUNCTION IF EXISTS exchange_do_close_request;
 
--- Unregister patch (partition-0001.sql)
--- SELECT _v.unregister_patch('partition-0001');
 
--- Drops for partition-0001.sql
 DROP FUNCTION IF EXISTS create_table_partition;
 DROP FUNCTION IF EXISTS create_partitions;
 DROP FUNCTION IF EXISTS detach_default_partitions;
 DROP FUNCTION IF EXISTS drop_default_partitions;
+DROP FUNCTION IF EXISTS master_prepare_sharding;
+DROP FUNCTION IF EXISTS create_foreign_table;
+DROP FUNCTION IF EXISTS create_shard_server;
 
 
 -- And we're out of here...

src/exchangedb/partition-0001.sql

@@ -1,312 +0,0 @@
---
--- This file is part of TALER
--- Copyright (C) 2014--2022 Taler Systems SA
---
--- TALER is free software; you can redistribute it and/or modify it under the
--- terms of the GNU General Public License as published by the Free Software
--- Foundation; either version 3, or (at your option) any later version.
---
--- TALER is distributed in the hope that it will be useful, but WITHOUT ANY
--- WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
--- A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
---
--- You should have received a copy of the GNU General Public License along with
--- TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
---
-
--- Everything in one big transaction
-BEGIN;
-
--- Check patch versioning is in place.
--- SELECT _v.register_patch('partition-0001', NULL, NULL);
-
-CREATE OR REPLACE FUNCTION create_table_partition(
-    source_table_name VARCHAR
-    ,modulus INTEGER
-    ,partition_num INTEGER
-  )
-  RETURNS VOID
-  LANGUAGE plpgsql
-AS $$
-BEGIN
-
-  RAISE NOTICE 'Creating partition %_%', source_table_name, partition_num;
-
-  EXECUTE FORMAT(
-    'CREATE TABLE IF NOT EXISTS %I '
-      'PARTITION OF %I '
-      'FOR VALUES WITH (MODULUS %s, REMAINDER %s)'
-    ,source_table_name || '_' || partition_num
-    ,source_table_name
-    ,modulus
-    ,partition_num-1
-  );
-
-END
-$$;
-
-CREATE OR REPLACE FUNCTION detach_default_partitions()
-  RETURNS VOID
-  LANGUAGE plpgsql
-AS $$
-BEGIN
-
-  RAISE NOTICE 'Detaching all default table partitions';
-
-  ALTER TABLE IF EXISTS wire_targets
-    DETACH PARTITION wire_targets_default;
-
-  ALTER TABLE IF EXISTS reserves
-    DETACH PARTITION reserves_default;
-
-  ALTER TABLE IF EXISTS reserves_in
-    DETACH PARTITION reserves_in_default;
-  
-  ALTER TABLE IF EXISTS reserves_close
-    DETACH PARTITION reserves_close_default;
-  
-  ALTER TABLE IF EXISTS reserves_out
-    DETACH PARTITION reserves_out_default;
-  
-  ALTER TABLE IF EXISTS known_coins
-    DETACH PARTITION known_coins_default;
-  
-  ALTER TABLE IF EXISTS refresh_commitments
-    DETACH PARTITION refresh_commitments_default;
-  
-  ALTER TABLE IF EXISTS refresh_revealed_coins
-    DETACH PARTITION refresh_revealed_coins_default;
-  
-  ALTER TABLE IF EXISTS refresh_transfer_keys
-    DETACH PARTITION refresh_transfer_keys_default;
-  
-  ALTER TABLE IF EXISTS deposits
-    DETACH PARTITION deposits_default;
-  
-  ALTER TABLE IF EXISTS refunds
-    DETACH PARTITION refunds_default;
-  
-  ALTER TABLE IF EXISTS wire_out
-    DETACH PARTITION wire_out_default;
-  
-  ALTER TABLE IF EXISTS aggregation_tracking
-    DETACH PARTITION aggregation_tracking_default;
-  
-  ALTER TABLE IF EXISTS recoup
-    DETACH PARTITION recoup_default;
-  
-  ALTER TABLE IF EXISTS recoup_refresh
-    DETACH PARTITION recoup_refresh_default;
-  
-  ALTER TABLE IF EXISTS prewire
-    DETACH PARTITION prewire_default;
-  
-  ALTER TABLE IF EXISTS cs_nonce_locks
-    DETACH partition cs_nonce_locks_default;
-
-END
-$$;
-
-COMMENT ON FUNCTION detach_default_partitions
-  IS 'We need to drop default and create new one before deleting the default partitions
-      otherwise constraints get lost too. Might be needed in shardig too';
-
-
-CREATE OR REPLACE FUNCTION drop_default_partitions()
-  RETURNS VOID
-  LANGUAGE plpgsql
-AS $$
-BEGIN
-
-  RAISE NOTICE 'Dropping default table partitions';
-
-  DROP TABLE IF EXISTS wire_targets_default;
-  DROP TABLE IF EXISTS reserves_default;
-  DROP TABLE IF EXISTS reserves_in_default;
-  DROP TABLE IF EXISTS reserves_close_default;
-  DROP TABLE IF EXISTS reserves_out_default;
-  DROP TABLE IF EXISTS known_coins_default;
-  DROP TABLE IF EXISTS refresh_commitments_default;
-  DROP TABLE IF EXISTS refresh_revealed_coins_default;
-  DROP TABLE IF EXISTS refresh_transfer_keys_default;
-  DROP TABLE IF EXISTS deposits_default;
-  DROP TABLE IF EXISTS refunds_default;
-  DROP TABLE IF EXISTS wire_out_default;
-  DROP TABLE IF EXISTS aggregation_tracking_default;
-  DROP TABLE IF EXISTS recoup_default;
-  DROP TABLE IF EXISTS recoup_refresh_default;
-  DROP TABLE IF EXISTS prewire_default;
-  DROP TABLE IF EXISTS cs_nonce_locks_default;
-
-END
-$$;
-
-COMMENT ON FUNCTION drop_default_partitions
-  IS 'Drop all default partitions once other partitions are attached.
-      Might be needed in sharding too.';
-
-CREATE OR REPLACE FUNCTION create_partitions(
-    num_partitions INTEGER
-)
-RETURNS VOID
-LANGUAGE plpgsql
-AS $$
-DECLARE
-  modulus INTEGER;
-BEGIN
-
-  modulus := num_partitions;
-
-  PERFORM detach_default_partitions();
-
-  LOOP
-    PERFORM create_table_partition(
-      'wire_targets'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_wire_targets_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'reserves'
-      ,modulus
-      ,num_partitions
-    );
-
-    PERFORM create_table_partition(
-      'reserves_in'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_reserves_in_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'reserves_close'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_reserves_close_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'reserves_out'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_reserves_out_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'known_coins'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_known_coins_partition(num_partitions::varchar);
-    
-    PERFORM create_table_partition(
-      'refresh_commitments'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_refresh_commitments_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'refresh_revealed_coins'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_refresh_revealed_coins_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'refresh_transfer_keys'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_refresh_transfer_keys_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'deposits'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_deposits_partition(num_partitions::varchar);
-
--- TODO: dynamically (!) creating/deleting deposits 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!
--- SEE also: https://www.cybertec-postgresql.com/en/automatic-partition-creation-in-postgresql/
--- (article is slightly wrong, as this works:)
---CREATE TABLE tab (
---  id bigint GENERATED ALWAYS AS IDENTITY,
---  ts timestamp NOT NULL,
---  data text
--- PARTITION BY LIST ((ts::date));
--- CREATE TABLE tab_def PARTITION OF tab DEFAULT;
--- BEGIN
--- CREATE TABLE tab_part2 (LIKE tab);
--- insert into tab_part2 (id,ts, data) values (5,'2022-03-21', 'foo');
--- alter table tab attach partition tab_part2 for values in ('2022-03-21');
--- commit;
--- Naturally, to ensure this is actually 100% conflict-free, we'd
--- need to create tables at the granularity of the wire/refund deadlines;
--- that is right now configurable via AGGREGATOR_SHIFT option.
-
-
-    PERFORM create_table_partition(
-      'refunds'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_refunds_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'wire_out'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_wire_out_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'aggregation_tracking'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_aggregation_tracking_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'recoup'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_recoup_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'recoup_refresh'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_recoup_refresh_partition(num_partitions::varchar);
-
-    PERFORM create_table_partition(
-      'prewire'
-      ,modulus
-      ,num_partitions
-    );
-
-    PERFORM create_table_partition(
-      'cs_nonce_locks'
-      ,modulus
-      ,num_partitions
-    );
-    PERFORM add_constraints_to_cs_nonce_locks_partition(num_partitions::varchar);
-
-    num_partitions=num_partitions-1;
-    EXIT WHEN num_partitions=0;
-
-  END LOOP;
-
-  PERFORM drop_default_partitions();
-
-END
-$$;
-
-COMMIT;

src/exchangedb/plugin_exchangedb_postgres.c

@@ -227,7 +227,7 @@ postgres_setup_partitions (void *cls,
 
   conn = GNUNET_PQ_connect_with_cfg (pg->cfg,
                                      "exchangedb-postgres",
-                                     "partition-",
+                                     "exchage-",
                                      NULL,
                                      ps);
   if (NULL == conn)
@@ -242,6 +242,59 @@ postgres_setup_partitions (void *cls,
 }
 
 
+/**
+ * Setup shards for already existing tables
+ *
+ * @param cls the `struct PostgresClosure` with the plugin-specific state
+ * @param num the number of shard servers to create for each partitioned table
+ * @return #GNUNET_OK upon success; #GNUNET_SYSERR upon failure
+ */
+static enum GNUNET_GenericReturnValue
+postgres_setup_shards (void *cls,
+                       const uint32_t num)
+{
+  struct PostgresClosure *pg = cls;
+  struct GNUNET_PQ_Context *conn;
+  enum GNUNET_GenericReturnValue ret;
+  struct GNUNET_PQ_QueryParam params[] = {
+    GNUNET_PQ_query_param_uint32 (&num),
+    GNUNET_PQ_query_param_end
+  };
+  struct GNUNET_PQ_PreparedStatement ps[] = {
+    GNUNET_PQ_make_prepare ("setup_shards",
+                            "SELECT"
+                            " master_prepare_sharding"
+                            " ();",
+                            0),
+    GNUNET_PQ_make_prepare ("create_shards",
+                            "SELECT"
+                            " create_shards"
+                            " ($1);",
+                            1),
+    GNUNET_PQ_PREPARED_STATEMENT_END
+  };
+
+  conn = GNUNET_PQ_connect_with_cfg (pg->cfg,
+                                     "exchangedb-postgres",
+                                     "exchage-",
+                                     NULL,
+                                     ps);
+  if (NULL == conn)
+    return GNUNET_SYSERR;
+  ret = GNUNET_OK;
+  if (0 > GNUNET_PQ_eval_prepared_non_select (conn,
+                                              "setup_shards",
+                                              NULL))
+    ret = GNUNET_SYSERR;
+  if (0 > GNUNET_PQ_eval_prepared_non_select (conn,
+                                              "create_shards",
+                                              params))
+    ret = GNUNET_SYSERR;
+  GNUNET_PQ_disconnect (conn);
+  return ret;
+}
+
+
 /**
  * Initialize prepared statements for @a pg.
  *
@@ -13062,6 +13115,7 @@ libtaler_plugin_exchangedb_postgres_init (void *cls)
   plugin->drop_tables = &postgres_drop_tables;
   plugin->create_tables = &postgres_create_tables;
   plugin->setup_partitions = &postgres_setup_partitions;
+  plugin->setup_shards = &postgres_setup_shards;
   plugin->start = &postgres_start;
   plugin->start_read_committed = &postgres_start_read_committed;
   plugin->commit = &postgres_commit;

src/include/taler_exchangedb_plugin.h

@@ -2241,6 +2241,21 @@ struct TALER_EXCHANGEDB_Plugin
   (*setup_partitions)(void *cls,
                       const uint32_t num);
 
+  /**
+   * Change already present tables of the database to num shards
+   * Only has an effect if there are default partitions only
+   *
+   * @param cls the @e cls of this struct with the plugin-specific state
+   * @param num the number of shard servers to create. The shard servers
+   *            must follow the numbering of 1-N, have the same user as
+   *            the master and have tables named <table>_n where n is the same
+   *            as the servers index of N.
+   * @return #GNUNET_OK upon success; #GNUNET_SYSERR upon failure
+   */
+  enum GNUNET_GenericReturnValue
+  (*setup_shards)(void *cls,
+                  const uint32_t num);
+
   /**
    * Start a transaction.
    *