From 6c5103f0b284f63da207ec620277f86a2a74c0f8 Mon Sep 17 00:00:00 2001
From: Pierre Ducroquet <pducroquet@entrouvert.com>
Date: Sun, 16 Jan 2022 20:50:59 +0100
Subject: [PATCH 1/9] sql: switch wcs_all_forms to a table maintained by
 triggers (#60552)

Instead of a view, wcs_all_forms is now a table.

Direct benefits:
- we will no longer explode the collapse limit of the PG optimizer
- since we won't read every table, we can be **much** faster
  (we aren't yet, I have not done the indexes so far)
- the optimizer will have less work, thus it will be faster (I said it already)
---
 tests/test_sql.py |  34 +----
 wcs/sql.py        | 333 ++++++++++++++++++++++++++++++++++++++++------
 2 files changed, 299 insertions(+), 68 deletions(-)

diff --git a/tests/test_sql.py b/tests/test_sql.py
index f08f3a9bb..bd358f3d8 100644
--- a/tests/test_sql.py
+++ b/tests/test_sql.py
@@ -1115,7 +1115,7 @@ def test_migration_1_tracking_code(pub):
 def test_migration_2_formdef_id_in_views(formdef):
     conn, cur = sql.get_connection_and_cursor()
     cur.execute('UPDATE wcs_meta SET value = 1 WHERE key = %s', ('sql_level',))
-    cur.execute('DROP VIEW wcs_all_forms CASCADE')
+    cur.execute('DROP TABLE wcs_all_forms CASCADE')
 
     # hack a formdef table the wrong way, to check it is reconstructed
     # properly before the views are created
@@ -1127,29 +1127,6 @@ def test_migration_2_formdef_id_in_views(formdef):
     assert table_exists(cur, 'wcs_view_1_tests')
     assert not column_exists_in_table(cur, 'wcs_view_1_tests', 'f4_display')
 
-    view_names = []
-    cur.execute(
-        '''SELECT table_name FROM information_schema.views
-                   WHERE table_name LIKE %s''',
-        ('wcs_view_%',),
-    )
-    while True:
-        row = cur.fetchone()
-        if row is None:
-            break
-        view_names.append(row[0])
-
-    fake_formdef = FormDef()
-    common_fields = sql.get_view_fields(fake_formdef)
-    # remove formdef_id for the purpose of this test
-    common_fields.remove([x for x in common_fields if x[1] == 'formdef_id'][0])
-
-    union = ' UNION '.join(
-        ['''SELECT %s FROM %s''' % (', '.join([y[1] for y in common_fields]), x) for x in view_names]
-    )
-    assert 'formdef_id' not in union
-    cur.execute('''CREATE VIEW wcs_all_forms AS %s''' % union)
-
     sql.migrate()
 
     assert column_exists_in_table(cur, 'wcs_all_forms', 'formdef_id')
@@ -1162,7 +1139,7 @@ def test_migration_2_formdef_id_in_views(formdef):
 def test_migration_6_actions_roles(formdef):
     conn, cur = sql.get_connection_and_cursor()
     cur.execute('UPDATE wcs_meta SET value = 5 WHERE key = %s', ('sql_level',))
-    cur.execute('DROP VIEW wcs_all_forms CASCADE')
+    cur.execute('DROP TABLE wcs_all_forms CASCADE')
 
     # hack a formdef table the wrong way, to check it is reconstructed
     # properly before the views are created
@@ -1187,7 +1164,7 @@ def test_migration_6_actions_roles(formdef):
 def test_migration_10_submission_channel(formdef):
     conn, cur = sql.get_connection_and_cursor()
     cur.execute('UPDATE wcs_meta SET value = 9 WHERE key = %s', ('sql_level',))
-    cur.execute('DROP VIEW wcs_all_forms CASCADE')
+    cur.execute('DROP TABLE wcs_all_forms CASCADE')
 
     # hack a formdef table the wrong way, to check it is reconstructed
     # properly before the views are created
@@ -1338,6 +1315,7 @@ def drop_formdef_tables():
     for table_name in table_names:
         if table_name.startswith('formdata_'):
             cur.execute('DROP TABLE %s CASCADE' % table_name)
+    cur.execute('TRUNCATE wcs_all_forms;')
 
 
 def test_is_at_endpoint(pub):
@@ -2017,7 +1995,7 @@ def test_migration_30_anonymize_evo_who(pub):
 def test_migration_31_user_label(formdef):
     conn, cur = sql.get_connection_and_cursor()
     cur.execute('UPDATE wcs_meta SET value = 30 WHERE key = %s', ('sql_level',))
-    cur.execute('DROP VIEW wcs_all_forms CASCADE')
+    cur.execute('DROP TABLE wcs_all_forms CASCADE')
 
     cur.execute('DROP VIEW wcs_view_1_tests')
     cur.execute('ALTER TABLE formdata_1_tests DROP COLUMN user_label')
@@ -2047,7 +2025,7 @@ def test_migration_38_submission_agent_id(pub):
 
     conn, cur = sql.get_connection_and_cursor()
     cur.execute('UPDATE wcs_meta SET value = 37 WHERE key = %s', ('sql_level',))
-    cur.execute('DROP VIEW wcs_all_forms CASCADE')
+    cur.execute('DROP TABLE wcs_all_forms CASCADE')
 
     cur.execute('DROP VIEW wcs_view_1_tests')
     cur.execute('ALTER TABLE formdata_1_tests DROP COLUMN submission_agent_id')
diff --git a/wcs/sql.py b/wcs/sql.py
index f70983423..ad8ee963b 100644
--- a/wcs/sql.py
+++ b/wcs/sql.py
@@ -513,6 +513,16 @@ def get_formdef_table_name(formdef):
     return formdef.table_name
 
 
+def get_formdef_trigger_function_name(formdef):
+    assert formdef.id is not None
+    return '%s_%s_trigger_fn' % (formdef.data_sql_prefix, formdef.id)
+
+
+def get_formdef_trigger_name(formdef):
+    assert formdef.id is not None
+    return '%s_%s_trigger' % (formdef.data_sql_prefix, formdef.id)
+
+
 def get_formdef_new_id(id_start):
     new_id = id_start
     conn, cur = get_connection_and_cursor()
@@ -558,7 +568,13 @@ def formdef_wipe():
         ('formdata\\_%%\\_%%',),
     )
     for table_name in [x[0] for x in cur.fetchall()]:
+        cur.execute('DELETE FROM %s' % table_name)  # Force trigger execution
         cur.execute('''DROP TABLE %s CASCADE''' % table_name)
+    cur.execute("SELECT relkind FROM pg_class WHERE relname = 'wcs_all_forms'")
+    row = cur.fetchone()
+    # only do the delete if wcs_all_forms is a table and not still a view
+    if row is not None and row[0] == 'r':
+        cur.execute("DELETE FROM wcs_all_forms WHERE object_type = 'formdata'")
     conn.commit()
     cur.close()
 
@@ -572,6 +588,7 @@ def carddef_wipe():
         ('carddata\\_%%\\_%%',),
     )
     for table_name in [x[0] for x in cur.fetchall()]:
+        cur.execute('DELETE FROM %s' % table_name)  # Force trigger execution
         cur.execute('''DROP TABLE %s CASCADE''' % table_name)
     conn.commit()
     cur.close()
@@ -775,6 +792,126 @@ def do_formdef_tables(formdef, conn=None, cur=None, rebuild_views=False, rebuild
     for field in existing_fields - needed_fields:
         cur.execute('''ALTER TABLE %s DROP COLUMN %s CASCADE''' % (table_name, field))
 
+    # recreate the trigger function, just so it's uptodate
+    category_value = formdef.category_id
+    geoloc_base_x_query = "NULL"
+    geoloc_base_y_query = "NULL"
+    if formdef.geolocations and 'base' in formdef.geolocations:
+        # default geolocation is in the 'base' key; we have to unstructure the
+        # field is the POINT type of postgresql cannot be used directly as it
+        # doesn't have an equality operator.
+        geoloc_base_x_query = "NEW.geoloc_base[0]"
+        geoloc_base_y_query = "NEW.geoloc_base[1]"
+    if formdef.category_id is None:
+        category_value = "NULL"
+    criticality_levels = len(formdef.workflow.criticality_levels or [0])
+    endpoint_status = formdef.workflow.get_endpoint_status()
+    endpoint_status_filter = ", ".join(["'wf-%s'" % x.id for x in endpoint_status])
+    if endpoint_status_filter == "":
+        # not the prettiest in town, but will do fine for now.
+        endpoint_status_filter = "'xxxx'"
+    object_type = formdef.data_sql_prefix  # shortcut.
+    formed_name_quotedstring = psycopg2.extensions.QuotedString(formdef.name)
+    formed_name_quotedstring.encoding = 'utf8'
+    formdef_name = formed_name_quotedstring.getquoted().decode()
+    cur.execute(
+        '''
+CREATE OR REPLACE FUNCTION {trg_fn_name}()
+RETURNS trigger
+LANGUAGE plpgsql
+AS $$
+BEGIN
+    -- TODO : sync back from users change !
+    IF TG_OP = 'DELETE' THEN
+        DELETE FROM wcs_all_forms WHERE formdef_id = {formdef_id} AND id = OLD.id AND object_type = '{object_type}';
+        RETURN OLD;
+    ELSEIF TG_OP = 'INSERT' THEN
+        INSERT INTO wcs_all_forms VALUES (
+            {category_id},
+            '{object_type}',
+            {formdef_id},
+            NEW.id,
+            NEW.user_id,
+            NEW.receipt_time,
+            NEW.status,
+            NEW.id_display,
+            NEW.submission_agent_id,
+            NEW.submission_channel,
+            NEW.backoffice_submission,
+            NEW.last_update_time,
+            NEW.digests,
+            NEW.user_label,
+            NEW.concerned_roles_array,
+            NEW.actions_roles_array,
+            NEW.fts,
+            NEW.status IN ({endpoint_status}),
+            {formdef_name},
+            (SELECT name FROM users WHERE users.id = CAST(NEW.user_id AS INTEGER)),
+            NEW.criticality_level - {criticality_levels},
+            {geoloc_base_x},
+            {geoloc_base_y},
+            NEW.anonymised);
+        RETURN NEW;
+    ELSE
+        UPDATE wcs_all_forms SET
+                user_id = NEW.user_id,
+                receipt_time = NEW.receipt_time,
+                status = NEW.status,
+                id_display = NEW.id_display,
+                submission_agent_id = NEW.submission_agent_id,
+                submission_channel = NEW.submission_channel,
+                backoffice_submission = NEW.backoffice_submission,
+                last_update_time = NEW.last_update_time,
+                digests = NEW.digests,
+                user_label = NEW.user_label,
+                concerned_roles_array = NEW.concerned_roles_array,
+                actions_roles_array = NEW.actions_roles_array,
+                fts = NEW.fts,
+                is_at_endpoint = NEW.status IN ({endpoint_status}),
+                formdef_name = {formdef_name},
+                user_name = (SELECT name FROM users WHERE users.id = CAST(NEW.user_id AS INTEGER)),
+                criticality_level = NEW.criticality_level - {criticality_levels},
+                geoloc_base_x = {geoloc_base_x},
+                geoloc_base_y = {geoloc_base_y},
+                anonymised = NEW.anonymised
+            WHERE formdef_id = {formdef_id}  AND id = OLD.id AND object_type = '{object_type}';
+        RETURN NEW;
+    END IF;
+END;
+$$;
+    '''.format(
+            trg_fn_name=get_formdef_trigger_function_name(formdef),
+            category_id=category_value,  # always valued ? need to handle null otherwise.
+            formdef_id=formdef.id,
+            geoloc_base_x=geoloc_base_x_query,
+            geoloc_base_y=geoloc_base_y_query,
+            formdef_name=formdef_name,
+            criticality_levels=criticality_levels,
+            endpoint_status=endpoint_status_filter,
+            object_type=object_type,
+        )
+    )
+
+    trg_name = get_formdef_trigger_name(formdef)
+    cur.execute(
+        '''SELECT 1 FROM pg_trigger
+            WHERE tgrelid = '%s'::regclass
+              AND tgname = '%s'
+        '''
+        % (table_name, trg_name)
+    )
+    if len(cur.fetchall()) == 0:
+        cur.execute(
+            '''CREATE TRIGGER {trg_name} AFTER INSERT OR UPDATE OR DELETE
+                ON {table_name}
+                FOR EACH ROW EXECUTE FUNCTION {trg_fn_name}();
+                '''.format(
+                trg_fn_name=get_formdef_trigger_function_name(formdef),
+                table_name=table_name,
+                trg_name=trg_name,
+            )
+        )
+
     # migrations on _evolutions table
     cur.execute(
         '''SELECT column_name FROM information_schema.columns
@@ -1524,56 +1661,85 @@ def drop_global_views(conn, cur):
     for view_name in view_names:
         cur.execute('''DROP VIEW IF EXISTS %s''' % view_name)
 
-    cur.execute('''DROP VIEW IF EXISTS wcs_all_forms''')
-
 
 def do_global_views(conn, cur):
     # recreate global views
-    from wcs.formdef import FormDef
-
-    view_names = [get_formdef_view_name(x) for x in FormDef.select(ignore_migration=True)]
+    from .carddef import CardDef
+    from .formdef import FormDef
 
+    # XXX TODO: make me dynamic, please ?
     cur.execute(
-        '''SELECT table_name FROM information_schema.views
-                    WHERE table_schema = 'public'
-                      AND table_name LIKE %s''',
-        ('wcs\\_view\\_%',),
+        """CREATE TABLE IF NOT EXISTS wcs_all_forms (
+        category_id integer,
+        object_type character varying NOT NULL, -- formdef or carddef
+        formdef_id integer NOT NULL,
+        id integer NOT NULL,
+        user_id character varying,
+        receipt_time timestamp without time zone,
+        status character varying,
+        id_display character varying,
+        submission_agent_id character varying,
+        submission_channel character varying,
+        backoffice_submission boolean,
+        last_update_time timestamp without time zone,
+        digests jsonb,
+        user_label character varying,
+        concerned_roles_array text[],
+        actions_roles_array text[],
+        fts tsvector,
+        is_at_endpoint boolean,
+        formdef_name text,
+        user_name character varying,
+        criticality_level integer,
+        geoloc_base_x double precision,
+        geoloc_base_y double precision,
+        anonymised timestamp with time zone
+        , PRIMARY KEY(object_type, formdef_id, id)
+    )"""
+    )
+    cur.execute(
+        '''CREATE INDEX IF NOT EXISTS %s_fts ON %s USING gin(fts)''' % ("wcs_all_forms", "wcs_all_forms")
     )
-    existing_views = set()
-    while True:
-        row = cur.fetchone()
-        if row is None:
-            break
-        existing_views.add(row[0])
 
-    view_names = existing_views.intersection(view_names)
-    if not view_names:
-        return
+    for attr in ('receipt_time', 'anonymised', 'user_id', 'status'):
+        cur.execute(
+            '''CREATE INDEX IF NOT EXISTS %s_%s ON %s (%s)''' % ("wcs_all_forms", attr, "wcs_all_forms", attr)
+        )
+    for attr in ('concerned_roles_array', 'actions_roles_array'):
+        cur.execute(
+            '''CREATE INDEX IF NOT EXISTS %s_%s ON %s USING gin (%s)'''
+            % ("wcs_all_forms", attr, "wcs_all_forms", attr)
+        )
 
-    cur.execute('''DROP VIEW IF EXISTS wcs_all_forms CASCADE''')
-
-    fake_formdef = FormDef()
-    common_fields = get_view_fields(fake_formdef)
-    common_fields.append(('concerned_roles_array', 'concerned_roles_array'))
-    common_fields.append(('actions_roles_array', 'actions_roles_array'))
-    common_fields.append(('fts', 'fts'))
-    common_fields.append(('is_at_endpoint', 'is_at_endpoint'))
-    common_fields.append(('formdef_name', 'formdef_name'))
-    common_fields.append(('user_name', 'user_name'))
-    common_fields.append(('criticality_level', 'criticality_level'))
-    common_fields.append(('geoloc_base_x', 'geoloc_base_x'))
-    common_fields.append(('geoloc_base_y', 'geoloc_base_y'))
-    common_fields.append(('anonymised', 'anonymised'))
-
-    union = ' UNION ALL '.join(
-        ['''SELECT %s FROM %s''' % (', '.join([y[1] for y in common_fields]), x) for x in view_names]
-    )
-    cur.execute('''CREATE VIEW wcs_all_forms AS %s''' % union)
+    # Purge of any dead data
+    valid_data = {FormDef.data_sql_prefix: set(), CardDef.data_sql_prefix: set()}
+    for formdef in FormDef.select(ignore_migration=True):
+        valid_data[formdef.data_sql_prefix].add(formdef.id)
+    for object_type, valid_ids in valid_data.items():
+        if valid_ids:
+            cur.execute(
+                "DELETE FROM wcs_all_forms WHERE object_type = '%s' AND formdef_id NOT IN (%s)"
+                % (object_type, ", ".join(valid_ids))
+            )
+        else:
+            cur.execute("DELETE FROM wcs_all_forms WHERE object_type = '%s'" % object_type)
+    ###fake_formdef = FormDef()
+    ###common_fields = get_view_fields(fake_formdef)
+    ###common_fields.append(('concerned_roles_array', 'concerned_roles_array'))
+    ###common_fields.append(('actions_roles_array', 'actions_roles_array'))
+    ###common_fields.append(('fts', 'fts'))
+    ###common_fields.append(('is_at_endpoint', 'is_at_endpoint'))
+    ###common_fields.append(('formdef_name', 'formdef_name'))
+    ###common_fields.append(('user_name', 'user_name'))
+    ###common_fields.append(('criticality_level', 'criticality_level'))
+    ###common_fields.append(('geoloc_base_x', 'geoloc_base_x'))
+    ###common_fields.append(('geoloc_base_y', 'geoloc_base_y'))
+    ###common_fields.append(('anonymised', 'anonymised'))
 
     for category in wcs.categories.Category.select():
         name = get_name_as_sql_identifier(category.url_name)[:40]
         cur.execute(
-            '''CREATE VIEW wcs_category_%s AS SELECT * from wcs_all_forms
+            '''CREATE OR REPLACE VIEW wcs_category_%s AS SELECT * from wcs_all_forms
                         WHERE category_id = %s'''
             % (name, category.id)
         )
@@ -2588,6 +2754,7 @@ class SqlDataMixin(SqlMixin):
         conn, cur = get_connection_and_cursor()
         if drop:
             cur.execute('''DROP TABLE %s_evolutions CASCADE''' % cls._table_name)
+            cur.execute('''DELETE FROM %s''' % cls._table_name)  # force trigger execution first.
             cur.execute('''DROP TABLE %s CASCADE''' % cls._table_name)
         else:
             cur.execute('''DELETE FROM %s_evolutions''' % cls._table_name)
@@ -3776,7 +3943,7 @@ def get_period_total(
 # latest migration, number + description (description is not used
 # programmaticaly but will make sure git conflicts if two migrations are
 # separately added with the same number)
-SQL_LEVEL = (58, 'add workflow_merged_roles_dict')
+SQL_LEVEL = (59, 'switch wcs_all_forms to a trigger-maintained table')
 
 
 def migrate_global_views(conn, cur):
@@ -3855,7 +4022,7 @@ def migrate():
     conn, cur = get_connection_and_cursor()
     sql_level = get_sql_level(conn, cur)
     if sql_level < 0:
-        # fake code to help in tetsting the error code path.
+        # fake code to help in testing the error code path.
         raise RuntimeError()
     if sql_level < 1:  # 1: introduction of tracking_code table
         do_tracking_code_table()
@@ -3982,6 +4149,92 @@ def migrate():
             do_formdef_tables(formdef, rebuild_views=False, rebuild_global_views=False)
         migrate_views(conn, cur)
         set_reindex('formdata', 'needed', conn=conn, cur=cur)
+    if sql_level < 59:
+        # switch wcs_all_forms to a trigger-maintained table
+        from wcs.carddef import CardDef
+        from wcs.formdef import FormDef
+
+        cur.execute("SELECT relkind FROM pg_class WHERE relname = 'wcs_all_forms';")
+        rows = cur.fetchall()
+        if len(rows) != 0:
+            if rows[0][0] == 'v':
+                # force wcs_all_forms table creation
+                cur.execute('DROP VIEW IF EXISTS wcs_all_forms CASCADE;')
+                do_global_views(conn, cur)
+            else:
+                assert rows[0][0] == 'r'
+        else:
+            do_global_views(conn, cur)
+
+        # assert there is no row, ie we are doing a clean migration (special case with unit tests likely) ?
+        # cur.execute("SELECT COUNT(*) FROM wcs_all_forms;")
+        # assert(cur.fetchone()[0] == 0)
+
+        # now copy all data into the table
+        for formdef in FormDef.select() + CardDef.select():
+            category_value = formdef.category_id
+            if formdef.category_id is None:
+                category_value = "NULL"
+            geoloc_base_x_query = "NULL"
+            geoloc_base_y_query = "NULL"
+            if formdef.geolocations and 'base' in formdef.geolocations:
+                # default geolocation is in the 'base' key; we have to unstructure the
+                # field is the POINT type of postgresql cannot be used directly as it
+                # doesn't have an equality operator.
+                geoloc_base_x_query = "geoloc_base[0]"
+                geoloc_base_y_query = "geoloc_base[1]"
+            criticality_levels = len(formdef.workflow.criticality_levels or [0])
+            endpoint_status = formdef.workflow.get_endpoint_status()
+            endpoint_status_filter = ", ".join(["'wf-%s'" % x.id for x in endpoint_status])
+            if endpoint_status_filter == "":
+                # not the prettiest in town, but will do fine for now.
+                endpoint_status_filter = "'xxxx'"
+            object_type = formdef.data_sql_prefix  # shortcut.
+            formed_name_quotedstring = psycopg2.extensions.QuotedString(formdef.name)
+            formed_name_quotedstring.encoding = 'utf8'
+            formdef_name = formed_name_quotedstring.getquoted().decode()
+            cur.execute(
+                """
+                INSERT INTO wcs_all_forms
+                SELECT
+                    {category_id},
+                    '{object_type}',
+                    {formdef_id},
+                    id,
+                    user_id,
+                    receipt_time,
+                    status,
+                    id_display,
+                    submission_agent_id,
+                    submission_channel,
+                    backoffice_submission,
+                    last_update_time,
+                    digests,
+                    user_label,
+                    concerned_roles_array,
+                    actions_roles_array,
+                    fts,
+                    status IN ({endpoint_status}),
+                    {formdef_name},
+                    (SELECT name FROM users WHERE users.id = CAST(user_id AS INTEGER)),
+                    criticality_level - {criticality_levels},
+                    {geoloc_base_x},
+                    {geoloc_base_y},
+                    anonymised
+                FROM {table_name}
+                ON CONFLICT DO NOTHING;
+                    """.format(
+                    table_name=get_formdef_table_name(formdef),
+                    category_id=category_value,  # always valued ? need to handle null otherwise.
+                    formdef_id=formdef.id,
+                    geoloc_base_x=geoloc_base_x_query,
+                    geoloc_base_y=geoloc_base_y_query,
+                    formdef_name=formdef_name,
+                    criticality_levels=criticality_levels,
+                    endpoint_status=endpoint_status_filter,
+                    object_type=object_type,
+                )
+            )
 
     cur.execute('''UPDATE wcs_meta SET value = %s WHERE key = %s''', (str(SQL_LEVEL[0]), 'sql_level'))
 
-- 
2.35.1

