firebird-qa/tests/bugs/gh_8091_test.py

#coding:utf-8

"""
ID:          issue-8091
ISSUE:       https://github.com/FirebirdSQL/firebird/issues/8091
TITLE:       Ability to create an inactive index
DESCRIPTION:
    Test creates a table with several indices, all of them are specified as INACTIVE.
    Then we check that these indices actually can *not* be used: explained plans for any query
    to this table that could relate to indexed columns must now contain 'Full Scan'.
    After this we extract metadata (with saving it to 'init_meta' variable) and drop test table.
    Applying of metada to the test database (which is empty now) must pass without errors and,
    more important, all indices must remain inactive after that.
    Finally, we change DB dialect to 1, make b/r and again do same actions.
    Result must be the same as for iteration with default dialect = 3.
NOTES:
    [25.10.2024] pzotov
    Checked on 6.0.0.508-67d8e39 (intermediate build).
"""
import time
from io import BytesIO
from firebird.driver import SrvRestoreFlag
import pytest
from firebird.qa import *

init_sql = """
    set bail on;
    recreate table test(id int generated by default as identity, x int, y int, z int);
    set term ^;
    execute block as
        declare n int = 100000;
        declare i int = 0;
    begin
        while (i < n) do
        begin
            insert into test(x, y, z) values( :i, null, :i);
            i = i + 1;
        end
    end^
    set term ;^
    commit;

    create unique ascending  index test_x_asc    inactive on test(x);
    create descending index test_y_desc   inactive on test(y);
    create unique descending index test_x_plus_y inactive on test computed by (x+y);

    create index test_z_partial         inactive on test(z) where mod(id,2) = 0;
    create unique index test_x_minus_y_partial inactive on test computed by (x-y) where mod(id,3) <= 1;
    commit;
"""
db = db_factory(init = init_sql)

act = python_act('db')

#-----------------------------------------------------------

def replace_leading(source, char="."):
    stripped = source.lstrip()
    return char * (len(source) - len(stripped)) + stripped

#-----------------------------------------------------------

def check_indices_inactive(act, qry_map, nr_block, capsys):
    with act.db.connect() as con:
        cur = con.cursor()
        for k,v in qry_map.items():
            ps = cur.prepare(v)
            # Print explained plan with padding eash line by dots in order to see indentations:
            print(v)
            print( '\n'.join([replace_leading(s) for s in ps.detailed_plan.split('\n')]) )
            print('')
            ps.free()
            
    expected_out = '\n'.join( [''.join( (qry_map[i],'\n',nr_block) ) for i in range(len(qry_map))] )
    act.expected_stdout = expected_out

    act.stdout = capsys.readouterr().out
    assert act.clean_stdout == act.clean_expected_stdout
    act.reset()
    
#-----------------------------------------------------------

@pytest.mark.version('>=6.0')
def test_1(act: Action, capsys):

    qry_map = {
         0 : 'select count(*) from test where x is null'
        ,1 : 'select count(*) from test where y is null'
        ,2 : 'select count(*) from test where x+y is null'
        ,3 : 'select count(*) from test where z is null and mod(id,2) = 0'
        ,4 : 'select count(*) from test where x-y is null and mod(id,3) <= 1'
        ,5 : 'select count(*) from test where x is not distinct from null'
        ,6 : 'select count(*) from test where y is not distinct from null'
        ,7 : 'select count(*) from test where x+y is not distinct from null'
        ,8 : 'select count(*) from test where z is not distinct from null and mod(id,2) = 0'
        ,9 : 'select count(*) from test where x-y is not distinct from null and mod(id,3) <= 1'
    }
    nr_block = """
        Select Expression
        ....-> Aggregate
        ........-> Filter
        ............-> Table "TEST" Full Scan
    """

    for iter in range(2):
    
        # check-1: ensure that all indices actually are INACTIVE, i.e. all queries will use full scan.
        ##########
        check_indices_inactive(act, qry_map, nr_block, capsys)
        
        #---===+++---===+++---===+++---===+++---===+++---===+++---===+++---===+++---===+++---===+++---

        # check-2: extract metadata, drop table and apply metadata which now contains 'INACTIVE' clause for indices.
        ##########
        act.isql(switches=['-x'])
        init_meta = '\n'.join( ('set bail on;', act.stdout) )

        with act.db.connect() as con:
            con.execute_immediate('drop table test')
            con.commit()

        # Apply metadata to main test database.
        act.isql(switches = [], input = init_meta)
        # NO errors must occur now:
        assert act.clean_stdout == ''
        act.reset()

        #---===+++---===+++---===+++---===+++---===+++---===+++---===+++---===+++---===+++---===+++---

        # check-3: ensure that all indices REMAIN INACTIVE, i.e. all queries will use full scan.
        ##########
        check_indices_inactive(act, qry_map, nr_block, capsys)

        if iter == 0:
            # change dialect to 1, make backup / restore and repeat all prev actions.
            act.gfix(switches = ['-sql_dialect', '1', act.db.dsn], combine_output = True)
            assert act.clean_stdout == ''
            act.reset()

            backup = BytesIO()
            with act.connect_server() as srv:
                srv.database.local_backup(database = act.db.db_path, backup_stream = backup)
                backup.seek(0)
                srv.database.local_restore(backup_stream = backup, database = act.db.db_path, flags = SrvRestoreFlag.REPLACE)