src/dirbs/partition_utils.py
def repartition_exceptions_lists(conn, *, num_physical_shards, src_filter_sql=None):
    """
    Function to repartition the exceptions_lists table.

    Arguments:
        conn: dirbs db connection instance
        num_physical_shards: number of physical shards to repartition
        src_filter_sql: custom filter sql, default None
    """
    with conn.cursor() as cursor, utils.db_role_setter(conn, role_name='dirbs_core_listgen'):
        # Create parent partition
        cursor.execute(
            """CREATE TABLE exceptions_lists_new (
                   LIKE exceptions_lists INCLUDING DEFAULTS
                                         INCLUDING IDENTITY
                                         INCLUDING CONSTRAINTS
                                         INCLUDING STORAGE
                                         INCLUDING COMMENTS
               )
               PARTITION BY LIST (operator_id)
            """
        )
        _grant_perms_list(conn, part_name='exceptions_lists_new')

        # Work out who the operators are
        imei_shard_names = utils.child_table_names(conn, 'exceptions_lists')
        operators = [x.operator_id for x in utils.table_invariants_list(conn, imei_shard_names, ['operator_id'])]

        # Create child partitions (operator at top level, then IMEI-sharded)
        for op_id in operators:
            tbl_name = per_mno_lists_partition(operator_id=op_id, suffix='_new', list_type='exceptions')
            create_per_mno_lists_partition(conn, parent_tbl_name='exceptions_lists_new', tbl_name=tbl_name,
                                           operator_id=op_id, num_physical_shards=num_physical_shards)

        # Insert data from original partition
        base_sql = sql.SQL("""INSERT INTO exceptions_lists_new
                                   SELECT *
                                     FROM exceptions_lists""")
        if src_filter_sql is not None:
            insert_sql = sql.SQL('{0} {1}').format(base_sql, sql.SQL(src_filter_sql))
        else:
            insert_sql = base_sql
        cursor.execute(insert_sql)

        # Add in indexes to each partition
        add_indices(conn, tbl_name='exceptions_lists_new', idx_metadata=exceptions_lists_indices())

        # Drop old table, after assigning sequence to new table
        cursor.execute('ALTER SEQUENCE exceptions_lists_row_id_seq OWNED BY exceptions_lists_new.row_id')
        cursor.execute('DROP TABLE exceptions_lists CASCADE')

        #  Rename tables, indexes and constraints
        rename_table_and_indices(conn, old_tbl_name='exceptions_lists_new',
                                 new_tbl_name='exceptions_lists', idx_metadata=exceptions_lists_indices())

    Function to repartition the exceptions_lists table.

    Arguments:
        conn: dirbs db connection instance
        num_physical_shards: number of physical shards to repartition
        src_filter_sql: custom filter sql, default None
    
Function to repartition the exceptions_lists table.
Function to repartition the exceptions_lists table .
def repartition_exceptions_lists(conn, *, num_physical_shards, src_filter_sql=None):
    with conn.cursor() as cursor, utils.db_role_setter(conn, role_name='dirbs_core_listgen'):
        cursor.execute(
            """CREATE TABLE exceptions_lists_new (
                   LIKE exceptions_lists INCLUDING DEFAULTS
                                         INCLUDING IDENTITY
                                         INCLUDING CONSTRAINTS
                                         INCLUDING STORAGE
                                         INCLUDING COMMENTS
               )
               PARTITION BY LIST (operator_id)
            """
        )
        _grant_perms_list(conn, part_name='exceptions_lists_new')
        imei_shard_names = utils.child_table_names(conn, 'exceptions_lists')
        operators = [x.operator_id for x in utils.table_invariants_list(conn, imei_shard_names, ['operator_id'])]
        for op_id in operators:
            tbl_name = per_mno_lists_partition(operator_id=op_id, suffix='_new', list_type='exceptions')
            create_per_mno_lists_partition(conn, parent_tbl_name='exceptions_lists_new', tbl_name=tbl_name,
                                           operator_id=op_id, num_physical_shards=num_physical_shards)
        base_sql = sql.SQL("""INSERT INTO exceptions_lists_new
                                   SELECT *
                                     FROM exceptions_lists""")
        if src_filter_sql is not None:
            insert_sql = sql.SQL('{0} {1}').format(base_sql, sql.SQL(src_filter_sql))
        else:
            insert_sql = base_sql
        cursor.execute(insert_sql)
        add_indices(conn, tbl_name='exceptions_lists_new', idx_metadata=exceptions_lists_indices())
        cursor.execute('ALTER SEQUENCE exceptions_lists_row_id_seq OWNED BY exceptions_lists_new.row_id')
        cursor.execute('DROP TABLE exceptions_lists CASCADE')
        rename_table_and_indices(conn, old_tbl_name='exceptions_lists_new',
                                 new_tbl_name='exceptions_lists', idx_metadata=exceptions_lists_indices())
def repartition_exceptions_lists ( conn , * , num_physical_shards , src_filter_sql = None ) : with conn . cursor ( ) as cursor , utils . db_role_setter ( conn , role_name = 'dirbs_core_listgen' ) : cursor . execute ( """CREATE TABLE exceptions_lists_new (
                   LIKE exceptions_lists INCLUDING DEFAULTS
                                         INCLUDING IDENTITY
                                         INCLUDING CONSTRAINTS
                                         INCLUDING STORAGE
                                         INCLUDING COMMENTS
               )
               PARTITION BY LIST (operator_id)
            """ ) _grant_perms_list ( conn , part_name = 'exceptions_lists_new' ) imei_shard_names = utils . child_table_names ( conn , 'exceptions_lists' ) operators = [ x . operator_id for x in utils . table_invariants_list ( conn , imei_shard_names , [ 'operator_id' ] ) ] for op_id in operators : tbl_name = per_mno_lists_partition ( operator_id = op_id , suffix = '_new' , list_type = 'exceptions' ) create_per_mno_lists_partition ( conn , parent_tbl_name = 'exceptions_lists_new' , tbl_name = tbl_name , operator_id = op_id , num_physical_shards = num_physical_shards ) base_sql = sql . SQL ( """INSERT INTO exceptions_lists_new
                                   SELECT *
                                     FROM exceptions_lists""" ) if src_filter_sql is not None : insert_sql = sql . SQL ( '{0} {1}' ) . format ( base_sql , sql . SQL ( src_filter_sql ) ) else : insert_sql = base_sql cursor . execute ( insert_sql ) add_indices ( conn , tbl_name = 'exceptions_lists_new' , idx_metadata = exceptions_lists_indices ( ) ) cursor . execute ( 'ALTER SEQUENCE exceptions_lists_row_id_seq OWNED BY exceptions_lists_new.row_id' ) cursor . execute ( 'DROP TABLE exceptions_lists CASCADE' ) rename_table_and_indices ( conn , old_tbl_name = 'exceptions_lists_new' , new_tbl_name = 'exceptions_lists' , idx_metadata = exceptions_lists_indices ( ) )
Function to repartition the exceptions_lists table.
Function to repartition the exceptions_lists table .
"""
    Function to repartition the exceptions_lists table.

    Arguments:
        conn: dirbs db connection instance
        num_physical_shards: number of physical shards to repartition
        src_filter_sql: custom filter sql, default None
    """ # Create parent partition # Work out who the operators are # Create child partitions (operator at top level, then IMEI-sharded) # Insert data from original partition # Add in indexes to each partition # Drop old table, after assigning sequence to new table #  Rename tables, indexes and constraints
conn
dirbs db connection instance
dirbs db connection instance
num_physical_shards
number of physical shards to repartition
number of physical shards to repartition
src_filter_sql
custom filter sql, default None
custom filter sql default None