0001-Switch-wcs_all_forms-to-a-table-maintained-by-trigge.patch

Pierre Ducroquet, 19 janvier 2022 23:49

Voir les différences: en ligne côte à côte

Subject: [PATCH] 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        | 332 ++++++++++++++++++++++++++++++++++++++++------
 2 files changed, 298 insertions(+), 68 deletions(-)

     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
-...
         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')
-...
     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
-...
     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
-...
         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):
-...
     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')
-...
         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')

         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()
-...
             ('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()
-...
             ('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()
-...
         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
-...
         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)
+            )
-...
             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)
-...
     # 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 = (56, 'add gin indexes to concerned_roles_array and actions_roles_array')
     SQL_LEVEL = (57, 'switch wcs_all_forms to a trigger-maintained table')
     def migrate_global_views(conn, cur):
-...
         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()
-...
             # 50: switch role uuid column to varchar
             do_role_table()
             migrate_legacy_roles()
         if sql_level < 57:
             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
             from wcs.carddef import CardDef
             from wcs.formdef import FormDef
             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'))
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-Switch-wcs_all_forms-to-a-table-maintained-by-trigge.patch