src/dirbs/schema_migrators/v85_upgrade.py
def _repartition_pairing_list(self, conn, *, num_physical_shards):
        """Repartition pairing list to implement change in structure."""
        with conn.cursor() as cursor, utils.db_role_setter(conn, role_name='dirbs_core_power_user'):
            cursor.execute(
                """CREATE TABLE historic_pairing_list_new (
                       LIKE historic_pairing_list INCLUDING DEFAULTS
                                                  INCLUDING IDENTITY
                                                  INCLUDING CONSTRAINTS
                                                  INCLUDING STORAGE
                                                  INCLUDING COMMENTS
                   )
                   PARTITION BY RANGE (virt_imei_shard)
                """
            )
            cursor.execute("""ALTER TABLE historic_pairing_list_new ADD COLUMN msisdn TEXT NOT NULL""")

            # grant permissions
            part_utils._grant_perms_pairing_list(conn, part_name='historic_pairing_list_new')

            # create child partitions
            part_utils.create_imei_shard_partitions(conn, tbl_name='historic_pairing_list_new',
                                                    num_physical_shards=num_physical_shards,
                                                    perms_func=part_utils._grant_perms_pairing_list, fillfactor=80)

            # copy data from original partition
            cursor.execute("""INSERT INTO historic_pairing_list_new
                                   SELECT DISTINCT p.imei_norm, p.imsi, p.start_date, p.end_date, p.virt_imei_shard,
                                          m.msisdn
                                     FROM historic_pairing_list p
                               INNER JOIN monthly_network_triplets_country m ON p.imsi = m.imsi""")

            # add indexes
            idx_metadata = [part_utils.IndexMetadatum(idx_cols=['imei_norm', 'imsi', 'msisdn'],
                                                      is_unique=True,
                                                      partial_sql='WHERE end_date IS NULL')]
            part_utils.add_indices(conn, tbl_name='historic_pairing_list_new', idx_metadata=idx_metadata)

            # drop old views, tables, indexes and constraints
            cursor.execute('DROP VIEW pairing_list')
            cursor.execute('DROP TABLE historic_pairing_list CASCADE')
            part_utils.rename_table_and_indices(conn, old_tbl_name='historic_pairing_list_new',
                                                new_tbl_name='historic_pairing_list', idx_metadata=idx_metadata)

            # create new view and grant permissions
            cursor.execute("""CREATE VIEW pairing_list AS
                                   SELECT imei_norm, imsi, msisdn, virt_imei_shard
                                     FROM historic_pairing_list
                                    WHERE end_date IS NULL WITH CHECK OPTION""")
            cursor.execute("""GRANT SELECT ON pairing_list
                              TO dirbs_core_listgen, dirbs_core_report, dirbs_core_api, dirbs_core_import_pairing_list
                           """)

            # drop and recreate staging data insert trigger
            cursor.execute("""
                            DROP FUNCTION pairing_list_staging_data_insert_trigger_fn() CASCADE;

                            CREATE FUNCTION pairing_list_staging_data_insert_trigger_fn() RETURNS trigger
                                LANGUAGE plpgsql
                                AS $$
                            BEGIN
                                -- Clean/normalize data before inserting
                                NEW.imei_norm = normalize_imei(NULLIF(TRIM(NEW.imei), ''));
                                NEW.imsi = NULLIF(TRIM(new.imsi), '');
                                NEW.msisdn = NULLIF(TRIM(new.msisdn), '');
                                RETURN NEW;
                            END
                            $$;
            """)
Repartition pairing list to implement change in structure.
Repartition pairing list to implement change in structure.
Repartition pairing list to implement change in structure .
def _repartition_pairing_list(self, conn, *, num_physical_shards):
        with conn.cursor() as cursor, utils.db_role_setter(conn, role_name='dirbs_core_power_user'):
            cursor.execute(
                """CREATE TABLE historic_pairing_list_new (
                       LIKE historic_pairing_list INCLUDING DEFAULTS
                                                  INCLUDING IDENTITY
                                                  INCLUDING CONSTRAINTS
                                                  INCLUDING STORAGE
                                                  INCLUDING COMMENTS
                   )
                   PARTITION BY RANGE (virt_imei_shard)
                """
            )
            cursor.execute("""ALTER TABLE historic_pairing_list_new ADD COLUMN msisdn TEXT NOT NULL""")
            part_utils._grant_perms_pairing_list(conn, part_name='historic_pairing_list_new')
            part_utils.create_imei_shard_partitions(conn, tbl_name='historic_pairing_list_new',
                                                    num_physical_shards=num_physical_shards,
                                                    perms_func=part_utils._grant_perms_pairing_list, fillfactor=80)
            cursor.execute("""INSERT INTO historic_pairing_list_new
                                   SELECT DISTINCT p.imei_norm, p.imsi, p.start_date, p.end_date, p.virt_imei_shard,
                                          m.msisdn
                                     FROM historic_pairing_list p
                               INNER JOIN monthly_network_triplets_country m ON p.imsi = m.imsi""")
            idx_metadata = [part_utils.IndexMetadatum(idx_cols=['imei_norm', 'imsi', 'msisdn'],
                                                      is_unique=True,
                                                      partial_sql='WHERE end_date IS NULL')]
            part_utils.add_indices(conn, tbl_name='historic_pairing_list_new', idx_metadata=idx_metadata)
            cursor.execute('DROP VIEW pairing_list')
            cursor.execute('DROP TABLE historic_pairing_list CASCADE')
            part_utils.rename_table_and_indices(conn, old_tbl_name='historic_pairing_list_new',
                                                new_tbl_name='historic_pairing_list', idx_metadata=idx_metadata)
            cursor.execute("""CREATE VIEW pairing_list AS
                                   SELECT imei_norm, imsi, msisdn, virt_imei_shard
                                     FROM historic_pairing_list
                                    WHERE end_date IS NULL WITH CHECK OPTION""")
            cursor.execute("""GRANT SELECT ON pairing_list
                              TO dirbs_core_listgen, dirbs_core_report, dirbs_core_api, dirbs_core_import_pairing_list
                           """)
            cursor.execute("""
                            DROP FUNCTION pairing_list_staging_data_insert_trigger_fn() CASCADE;
                            CREATE FUNCTION pairing_list_staging_data_insert_trigger_fn() RETURNS trigger
                                LANGUAGE plpgsql
                                AS $$
                            BEGIN
                                -- Clean/normalize data before inserting
                                NEW.imei_norm = normalize_imei(NULLIF(TRIM(NEW.imei), ''));
                                NEW.imsi = NULLIF(TRIM(new.imsi), '');
                                NEW.msisdn = NULLIF(TRIM(new.msisdn), '');
                                RETURN NEW;
                            END
                            $$;
            """)
def _repartition_pairing_list ( self , conn , * , num_physical_shards ) : with conn . cursor ( ) as cursor , utils . db_role_setter ( conn , role_name = 'dirbs_core_power_user' ) : cursor . execute ( """CREATE TABLE historic_pairing_list_new (
                       LIKE historic_pairing_list INCLUDING DEFAULTS
                                                  INCLUDING IDENTITY
                                                  INCLUDING CONSTRAINTS
                                                  INCLUDING STORAGE
                                                  INCLUDING COMMENTS
                   )
                   PARTITION BY RANGE (virt_imei_shard)
                """ ) cursor . execute ( """ALTER TABLE historic_pairing_list_new ADD COLUMN msisdn TEXT NOT NULL""" ) part_utils . _grant_perms_pairing_list ( conn , part_name = 'historic_pairing_list_new' ) part_utils . create_imei_shard_partitions ( conn , tbl_name = 'historic_pairing_list_new' , num_physical_shards = num_physical_shards , perms_func = part_utils . _grant_perms_pairing_list , fillfactor = 80 ) cursor . execute ( """INSERT INTO historic_pairing_list_new
                                   SELECT DISTINCT p.imei_norm, p.imsi, p.start_date, p.end_date, p.virt_imei_shard,
                                          m.msisdn
                                     FROM historic_pairing_list p
                               INNER JOIN monthly_network_triplets_country m ON p.imsi = m.imsi""" ) idx_metadata = [ part_utils . IndexMetadatum ( idx_cols = [ 'imei_norm' , 'imsi' , 'msisdn' ] , is_unique = True , partial_sql = 'WHERE end_date IS NULL' ) ] part_utils . add_indices ( conn , tbl_name = 'historic_pairing_list_new' , idx_metadata = idx_metadata ) cursor . execute ( 'DROP VIEW pairing_list' ) cursor . execute ( 'DROP TABLE historic_pairing_list CASCADE' ) part_utils . rename_table_and_indices ( conn , old_tbl_name = 'historic_pairing_list_new' , new_tbl_name = 'historic_pairing_list' , idx_metadata = idx_metadata ) cursor . execute ( """CREATE VIEW pairing_list AS
                                   SELECT imei_norm, imsi, msisdn, virt_imei_shard
                                     FROM historic_pairing_list
                                    WHERE end_date IS NULL WITH CHECK OPTION""" ) cursor . execute ( """GRANT SELECT ON pairing_list
                              TO dirbs_core_listgen, dirbs_core_report, dirbs_core_api, dirbs_core_import_pairing_list
                           """ ) cursor . execute ( """
                            DROP FUNCTION pairing_list_staging_data_insert_trigger_fn() CASCADE;

                            CREATE FUNCTION pairing_list_staging_data_insert_trigger_fn() RETURNS trigger
                                LANGUAGE plpgsql
                                AS $$
                            BEGIN
                                -- Clean/normalize data before inserting
                                NEW.imei_norm = normalize_imei(NULLIF(TRIM(NEW.imei), ''));
                                NEW.imsi = NULLIF(TRIM(new.imsi), '');
                                NEW.msisdn = NULLIF(TRIM(new.msisdn), '');
                                RETURN NEW;
                            END
                            $$;
            """ )
Repartition pairing list to implement change in structure.
Repartition pairing list to implement change in structure .
"""Repartition pairing list to implement change in structure.""" # grant permissions # create child partitions # copy data from original partition # add indexes # drop old views, tables, indexes and constraints # create new view and grant permissions # drop and recreate staging data insert trigger
self
conn
num_physical_shards