Автоматизация тестирования таблиц в Postgresql на SQL29.11.2024 21:16

Пример описываемой автоматизации

Привет.

Предположим, вам пришла задача на тестирование какой-либо функциональности, которая относится к бэкенду. Вы переходите к документу с требованиями и видите, помимо прочего, описание таблиц базы данных (БД). Примерно так это выглядит:

Таблица 1. Требования к таблице price в схеме public

Следовательно, один из тест-кейсов должен содержать проверки на соответствие метаданных этих таблиц ожидаемым значениям. К таким метаданным относятся состав и тип столбцов, ограничения для значений, наличие связей и т.д. В Postgresql есть неочевидная возможность проводить подобные проверки автоматически с помощью SQL-запроса.

К слову, эта статья будет интересна не только с точки зрения обзора SQL-запроса для автоматизации тестирования, но и с точки зрения того, что и как можно писать на SQL.

Возьмем для примера пару таблиц:

-- drop table public.models

create table public.models (
    id int8 generated by default as identity( increment by 1 minvalue 1 maxvalue 9223372036854775807 start 1 cache 1 no cycle) not null,
    "name" varchar(50) not null,
    manufacturer varchar(50) not null,
    constraint models_pk primary key (id),
    constraint models_unique unique (name, manufacturer)
);

-- drop table public.price

create table public.price (
    id int8 generated by default as identity( increment by 1 minvalue 1 maxvalue 9223372036854775807 start 1 cache 1 no cycle) not null,
    price_value numeric(18,6) not null,
    model_id int4 not null,
    comment varchar(255) null,
    constraint price_pk primary key (id),
    constraint price_models_fk foreign key (model_id) references public.models(id),
    constraint price_unique unique (price_value, model_id)
);

Сосредоточимся на price. Таблица models нужна исключительно для того, чтобы добавить в price ограничение (constraint) по зависимому ключу (foreign key). SQL-запрос для автоматической проверки метаданных этой таблицы будет выглядеть так:

with
    -- Настраиваем запрос: задаем название таблицы и название схемы
    params(table_name, table_schema) as (
        values (
            'price',
            'public'
        )
    ),
    -- Описываем ожидаемые столбцы таблицы
    required_columns(column_name, data_type, is_nullable, character_maximum_length) as (
        values            
            ('id',          'bigint',               'NO', 	null),
            ('price_value', 'numeric',              'NO', 	18.6),
            ('model_id',    'integer',  		    'NO',	null),
            ('comment',     'character varying',    'YES',  255)
    ),
    -- Описываем ожидаемые ограничения
    required_constraints(column_name, constraint_type) as (
        values            
            ('id',          'PRIMARY KEY'),
            ('id',    		'FOREIGN KEY'),
            ('model_id',    'FOREIGN KEY'),
            ('price_value', 'UNIQUE'),
            ('model_id',    'UNIQUE')
    ),
    -- Находим информацию о столбцах тестируемой таблицы и добавляем обработку null'ов
    columns_info as (
        select
            column_name, data_type, is_nullable,
            coalesce(numeric_precision, 0)          as numeric_precision,
            coalesce(numeric_scale, 0)              as numeric_scale,
            coalesce(character_maximum_length, 0)   as character_maximum_length
        from information_schema.columns
        where
            table_name       = (select table_name   from params)
            and table_schema = (select table_schema from params)
    ),
    -- Проверяем существование таблицы и подсчитываем количество столбцов в ней
    check_table_exist as (
        select
            case when count_all_fields < 1 then false else true end table_exists,
            case when count_all_fields < 1 then 1 else 0 end table_exists_error,
            count_all_fields
        from (
            select count (*) as count_all_fields
            from columns_info
        ) sq
    ),
    -- Сравниваем ожидаемый и текущий наборы атрибутов таблицы
    fields_comparison as (
        select t.*
        from columns_info t
        inner join required_columns r
            on  t.column_name   = r.column_name
            and t.data_type     = r.data_type
            and t.is_nullable   = r.is_nullable
            and (
                -- Сравниваем целую часть десятичных значений
                case
                    when t.data_type = 'numeric'
                    then t.numeric_precision = trunc(r.character_maximum_length::numeric)
                end
                and
                -- Сравниваем дробную часть десятичных значений
                case
                    when t.data_type = 'numeric'
                    then t.numeric_scale = (r.character_maximum_length::numeric - trunc(r.character_maximum_length::numeric))
                    * power(10, length(split_part(r.character_maximum_length::text, '.', 2)))
                end
                or t.character_maximum_length = coalesce(r.character_maximum_length::numeric, 0)
           )
    ),
    -- Ищем лишние столбцы и считаем их количество
    check_unexpected_fields as (
        select
            count (column_name) as count_unexpected_fields,
            string_agg(column_name, ', ') as unexpected_fields
        from (
            select column_name
            from columns_info
            except
            select column_name
            from required_columns
        ) sq
    ),
    -- Ищем недостающие столбцы и считаем их количество
    check_missing_fields as (
        select
            count (column_name) as count_missing_fields,
            string_agg(column_name, ', ') as missing_fields
        from (
            select column_name
            from required_columns
            except
            select column_name
            from columns_info
        ) sq
    ),
    -- Ищем невалидные столбцы и считаем их количество
    check_invalid_fields as (
        select
            count (column_name) as count_invalid_fields,
            string_agg(column_name, ', ') as invalid_fields
        from (
            select column_name
            from required_columns
            except
            select column_name
            from fields_comparison
            except
            select string_to_table(missing_fields, ', ')
            from check_missing_fields
        ) sq
    ),
    -- Ищем все ограничения для таблицы
    constraints_query as(
        select
            t1.constraint_type,
            t2.column_name,
            t3.column_name as foreign_column
        from information_schema.table_constraints t1
        left join information_schema.constraint_column_usage t2
            on t1.constraint_name = t2.constraint_name
        left join information_schema.key_column_usage as t3
            on t1.constraint_name = t3.constraint_name
            and t1.table_schema = t3.table_schema
        where
            t1.table_name            = (select table_name   from params)
            and t1.constraint_schema = (select table_schema from params)
            and t2.column_name is not null
    ),
    -- Включаем значения зависимых ключей (foreign_column) в список ограничений (column_name)
    union_foreign_ref_columns as (
        select column_name, constraint_type
        from constraints_query
        union all
        select foreign_column as column_name, constraint_type
        from constraints_query
    ),
    -- Ищем лишние ограничения и считаем их количество
    check_unexpected_constraints as (
        select
            count (column_name) as count_unexpected_constraints,
            string_agg(column_name || ' (' || constraint_type || ')', ', ') as unexpected_constraints
        from (
            select *
            from union_foreign_ref_columns
            except
            select column_name, constraint_type
            from required_constraints
        ) sq
    ),
    -- Ищем недостающие ограничения и считаем их количество
    check_missing_constraints as (
        select
            count (column_name) as count_missing_constraints,
            string_agg(column_name || ' (' || constraint_type || ')', ', ') as missing_constraints
        from (
            select column_name, constraint_type
            from required_constraints
            except
            select *
            from union_foreign_ref_columns
        ) sq
    ),
    -- Собираем полученные данные
    checks as (
        select
            -- Выводим всю информацию об ошибках и суммируем их количество
            table_exists_error + count_unexpected_fields + count_missing_fields + count_invalid_fields +
            count_unexpected_constraints + count_missing_constraints as errors,
            *
        from        check_table_exist
        cross join  check_unexpected_fields
        cross join  check_missing_fields
        cross join  check_invalid_fields
        cross join  check_unexpected_constraints
        cross join  check_missing_constraints
)
select *
from checks

Важное примечание: чтобы использовать этот запрос для какой-либо вашей таблицы, нужно внести изменения только в CTE params (названия вашей таблицы и схемы, где она хранится),  required_columns (описание ожидаемых значений метаданных столбцов вашей таблицы) и required_constraints (описание ожидаемых ограничений для вашей таблицы). Остальной код не нужно менять.

Возможность такой автоматической проверки обеспечивается наличием во всех БД Postgresql информационной схемы (information_schema). В Oracle, если не ошибаюсь, тоже такая есть, поэтому описываемый запрос может быть актуален и для других СУБД (полагаю, с корректировками).

Что показывает этот запрос?

Анализ запроса

В CTE params,  required_columns и required_constraints описываются входные данные. Это — ожидаемые датасеты значений. С params, думаю, все понятно — указываем название тестируемой таблицы и название схемы, где она хранится:

with
    -- Настраиваем запрос: задаем название таблицы и название схемы
    params(table_name, table_schema) as (
        values (
            'price',
            'smartphones'
        )
    )
select *
from params

Каждый CTE, о котором я пишу, буду оформлять с select’ом, который обращается к этому CTE. Весомой причины для этого нет — мне просто так удобнее:)
Этот код можно выполнить у себя локально. Он просто покажет значения внутри CTE:

required_column рассмотрим подробнее:

with
    -- Описываем ожидаемые столбцы таблицы
    required_columns(column_name, data_type, is_nullable, character_maximum_length) as (
        values            
            ('id',          'bigint',               'NO',   null),
            ('price_value', 'numeric',              'NO',   18.6),
            ('model_id',    'integer',  		    'NO',   null),
            ('comment',     'character varying',    'YES',  255)
    )
select *
from required_columns

Здесь описываются значения в 4-х столбцах:

• column_name: название столбца в тестируемой таблице;

• data_type: тип столбца. В текущей версии запроса указывать нужно изначальное название типа (character varying), а не синоним (varchar). Эти названия можно просмотреть здесь:
  https://www.postgresql.org/docs/8.1/datatype.html

• is_nullable: допускается ли в качестве значения для столбца null;

• character_maximum_length: длина строки. Иначе говоря, максимально допустимое количество символов (цифр) значения. Если ограничения нет — указывается null.

В CTE required_constraints содержатся ожидаемые ограничения по 2-м столбцам:

• column_name: название столбца, к которому применяется ограничение;

• constraint_type: тип ограничения.

Типы ограничений (constraint_type) бывают следующие:

• CHECK: значения в столбце должны удовлетворять какому-либо заданному условию;

• NOT NULL: название говорящее — значение в столбце не должно быть NULL;

• UNIQUE: в столбце не допускается наличие

• PRIMARY KEY: комбинация ограничений UNIQUE и NOT NULL;

• FOREIGN KEY: значение в столбце у отдельно взятой записи должно быть таким же, как значение в столбце другой таблицы.

Ограничение NOT NULL проверяется отдельно в рамках изучения соответствия фактических метаданных значениям в CTE required_columns. Как для этого там есть столбец is_nullable.

Теперь рассмотрим код, который описан после required_constraints. До этого CTE включительно, как упоминал выше, указываются входные данные. В CTE columns_info происходит получение метаданных о столбцах проверяемой таблицы с помощью запроса к columns:

select *
from information_schema.columns

Ограничим выборку только по тем метаданным, которые нужны для проверок. null’ы преобразуем в 0, чтобы значения можно было сравнивать. Также добавим фильтрацию (в where) по тестируемой таблице и по схеме, где она хранится:

select
    column_name, data_type, is_nullable,
    coalesce(numeric_precision, 0)          as numeric_precision,
    coalesce(numeric_scale, 0)              as numeric_scale,
    coalesce(character_maximum_length, 0)   as character_maximum_length
from information_schema.columns
where
    table_name       = 'price'
    and table_schema = 'public'

В select’е запроса:

Если у столбца тестируемой таблицы нет ограничений по количеству символов или цифр (например, при наличии типа boolean или jsonb), то в соответствующем столбце метаданных будет null.

Перенесем запрос в CTE и обратимся к записям, которые по нему возвращаются, через select:

with columns_info as (
    select
        column_name, data_type, is_nullable,
        coalesce(numeric_precision, 0)          as numeric_precision,
        coalesce(numeric_scale, 0)              as numeric_scale,
        coalesce(character_maximum_length, 0)   as character_maximum_length
    from information_schema.columns
    where
        table_name       = 'price'
        and table_schema = 'public'
)
select *
from columns_info

Добавим код для проверки существования таблицы. Сделаем это через подсчет количества значений в запросе, которые мы описали ранее (меняем select * на select count (*) ):

with columns_info as (
    select
        column_name, data_type, is_nullable,
        coalesce(numeric_precision, 0)          as numeric_precision,
        coalesce(numeric_scale, 0)              as numeric_scale,
        coalesce(character_maximum_length, 0)   as character_maximum_length
    from information_schema.columns
    where
        table_name       = 'price'
        and table_schema = 'public'
)
select count (*)
from columns_info

Логика проста: нет таблицы — нет и записей, которые возвращаются по значениям, указанным в фильтрах. Добавим сюда case для вывода boolean’а, который будет отображать статус проверки. А также счетчик, который будет показывать 1, если boolean — true и 0, если boolean — false. Он пригодится для подсчета суммы ошибок. Еще выведем фактическое количество столбцов в тестируемой таблице — это просто полезная информация. Все эти данные возьмем из select’а, описанного выше под CTE. Выполнять его будем через подзапрос:

with columns_info as (
    select
        column_name, data_type, is_nullable,
        coalesce(numeric_precision, 0)          as numeric_precision,
        coalesce(numeric_scale, 0)              as numeric_scale,
        coalesce(character_maximum_length, 0)   as character_maximum_length
    from information_schema.columns
    where
        table_name       = 'price'
        and table_schema = 'public'
)
select
    case when count_all_fields < 1 then false   else true   end table_exists,
    case when count_all_fields < 1 then 1       else 0      end table_exists_error,
    count_all_fields
from (
    select count (*) as count_all_fields
    from columns_info
) sq

В следующем блоке кода будем проводить сравнение — ожидаемых значений, описанных в CTE required_columns, и фактических значений, описанных в CTE columns_info. Уберем код для проверки существования таблицы в отдельный CTE check_table_exists. Также добавим 3 CTE с ожидаемыми данными, чтобы можно было проводить сравнение. Название таблицы (table_name) и название схемы (schema_name) для columns_info теперь будем получать через подзапросы. Пишем код дальше:

with
    -- Настраиваем запрос: задаем название таблицы и название схемы
    params(table_name, table_schema) as (
        values (
            'price',
            'public'
        )
    ),
    -- Описываем ожидаемые столбцы таблицы
    required_columns(column_name, data_type, is_nullable, character_maximum_length) as (
        values            
            ('id',          'bigint',                       'NO', 	null),
            ('price_value', 'numeric',            			'NO', 	18.6),
            ('model_id',    'integer',  					'NO',	null),
            ('comment',     'character varying',            'YES',	255)
    ),
    -- Описываем ожидаемые ограничения
    required_constraints(column_name, constraint_type) as (
        values            
            ('id',          'PRIMARY KEY'),
            ('id',    		'FOREIGN KEY'),
            ('model_id',    'FOREIGN KEY'),
            ('price_value', 'UNIQUE'),
            ('model_id',    'UNIQUE')
    ),
    -- Находим информацию о столбцах тестируемой таблицы и добавляем обработку null'ов
    columns_info as (
        select
            column_name, data_type, is_nullable,
            coalesce(numeric_precision, 0)          as numeric_precision,
            coalesce(numeric_scale, 0)              as numeric_scale,
            coalesce(character_maximum_length, 0)   as character_maximum_length
        from information_schema.columns
        where
            table_name       = (select table_name   from params)
            and table_schema = (select table_schema from params)
    ),
    -- Проверяем существование таблицы и подсчитываем количество столбцов в ней
    check_table_exist as (
        select
            case when count_all_fields < 1 then false else true end table_exists,
            case when count_all_fields < 1 then 1 else 0 end table_exists_error,
            count_all_fields
        from (
            select count (*) as count_all_fields
            from columns_info
        ) sq
    )
select *
from columns_info

Сравнить датасеты в SQL можно разными способами. Мне известны три — через:

В данном случае я воспользуюсь join’ом. Соединение будем делать по столбцам, описанным в CTE columns_info. Начнем с первых 3-х. Таблицам добавим алиасы t и r. Возвращать записи будем только из columns_info. Иначе говоря, только те, которые фактически есть в БД. Код в with идентичен блоку, описанному выше, поэтому его далее буду сокращать для удобства:

with (...)
select t.*
from columns_info t
inner join required_columns r
    on  t.column_name               = r.column_name
    and t.data_type                 = r.data_type
    and t.is_nullable               = r.is_nullable

Остаются еще 3 столбца. С ними сложнее. В CTE required_columns есть только 4 столбца, тогда в columns_info возвращается 6. Это связано с тем, что ожидаемое ограничение по количеству символов и цифр мы задаем в 1-м столбце для всех типов. При этом фактически в метаданных есть разделение такого ограничения на 3 столбца.

Начнем с простого: сравнение для столбцов varchar. null’ы в датасете с ожидаемыми значениями заменяем на 0. На всякий случай повторю, что это нужно для сравнения: null’ы через равенство проверить не получится:

with (...)
select t.*
from columns_info t
inner join required_columns r
    on  t.column_name               = r.column_name
    and t.data_type                 = r.data_type
    and t.is_nullable               = r.is_nullable
    and t.character_maximum_length  = coalesce(r.character_maximum_length, 0)

Теперь нужно сравнить целую и дробную части числовых значений ограничений. Но только для типа numeric и decimal, т.к. остальные числовые типы имеют фиксированный диапазон значений. Начнем с целой части чисел. Здесь нужно использовать case для отбора значений с типами numeric и decimal. С учетом того, что сравнивается целая часть, у числа из столбца character_maximum_length (из CTE required_columns) нужно отсекать дробную часть с помощью функции trunc (). Проверяемый столбец может быть единовременно только одного типа, поэтому в ранее написанную обработку для ограничений varchar новый код добавляем через or. Всю обработку при этом оборачиваем в скобки:

with (...)
select t.*
from columns_info t
inner join required_columns r
    on  t.column_name               = r.column_name
    and t.data_type                 = r.data_type
    and t.is_nullable               = r.is_nullable
    and t.character_maximum_length  = coalesce(r.character_maximum_length::numeric, 0)
    and (
    	case
            when t.data_type in ('numeric', 'decimal')
            then t.numeric_precision = trunc(r.character_maximum_length::numeric)
        end
        or t.character_maximum_length = coalesce(r.character_maximum_length::numeric, 0)
    )

И теперь самое сложное: сравнение дробной части. Вычислить её просто — нужно лишь из числа вычесть его целую часть. Пример для наглядности:

18,6 — 18 = 0,6

Но описать кодом это будет сложнее. В таблице price, которая используется в этой статье как объект тестирования, есть столбец price_value. Он имеет тип numeric и ограничение 18.6, то есть 18 цифр в целой части и 6 в дробной. То, что нужно для примера. Попробуем в select добавить разность, чтобы посмотреть, что получится:

Это ожидаемое значение. А что в фактическом?

То есть чтобы корректно провести сравнение, нужно дробную часть ожидаемого значения перевести в целую. В данном случае достаточно выполнить умножение на 10. Но что если бы ограничение было не 18.6, а, скажем, 18.15? Тогда умножение на 10 нужно проводить во 2-й степени. И значение этой степени — переменное, равное количеству знаков после запятой. Иначе говоря, для столбца price_value, у которого есть ограничение 18.6, степень будет равна 1 (после запятой только 1 цифра — 6). А если ограничение было 18.15, то степень должна была быть 2.

Итак, разность числа и его целой части, которые берутся из character_maximum_length (CTE required_columns), показанная в виде кода на скриншоте выше, выглядит так:

r.character_maximum_length::numeric - trunc(r.character_maximum_length::numeric)

Теперь её нужно умножить на 10 в степени:

(r.character_maximum_length::numeric - trunc(r.character_maximum_length::numeric)) * power(10, 1)

Но единица (1) в power () покрывает не все случаи, поэтому добавим код для динамического расчета:

(r.character_maximum_length::numeric - trunc(r.character_maximum_length::numeric)) * power(10, length(split_part(r.character_maximum_length::text, '.', 2)))

Прокомментирую. r.character_maximum_length: text — выполняем приведение типов, т.к. этот столбец имеет тип numeric, но его значения будут использоваться в функциях, которые работают только с символьными типами. Функция split_part () используется для отбора цифр после запятой (но в качестве разделителя указывается именно точка, т.к. дробное значение в метаданных хранится именно в таком виде). А length подсчитывает количество этих цифр.

Добавляем этот код в выражения для соединения:

with(...)
select t.*
from columns_info t
inner join required_columns r
    on  t.column_name   = r.column_name
    and t.data_type     = r.data_type
    and t.is_nullable   = r.is_nullable
    and (
        -- Сравниваем целую часть десятичных значений
        case
            when t.data_type in ('numeric', 'decimal')
            then t.numeric_precision = trunc(r.character_maximum_length::numeric)
        end
        and
        -- Сравниваем дробную часть десятичных значений
        case
            when t.data_type in ('numeric', 'decimal')
            then t.numeric_scale = (r.character_maximum_length::numeric - trunc(r.character_maximum_length::numeric))
            * power(10, length(split_part(r.character_maximum_length::text, '.', 2)))
        end
        or t.character_maximum_length = coalesce(r.character_maximum_length::numeric, 0)
   )

Результат сравнения покажет, какие столбцы тестируемой таблицы имеют ожидаемые метаданные. В дальнейшем отобранные таким образом записи будем сравнивать с датасетом из CTE required_columns. Так мы найдем столбцы, у которых какие-либо метаданные не соответствуют ожидаемым. Добавим этот запрос в CTE с названием fields_comparison:

-- Сравниваем ожидаемый и текущий наборы атрибутов таблицы
with 
	(...),
    fields_comparison as (
        select t.*
        from columns_info t
        inner join required_columns r
            on  t.column_name   = r.column_name
            and t.data_type     = r.data_type
            and t.is_nullable   = r.is_nullable
            and (
                -- Сравниваем целую часть десятичных значений
                case
                    when t.data_type = 'numeric'
                    then t.numeric_precision = trunc(r.character_maximum_length::numeric)
                end
                and
                -- Сравниваем дробную часть десятичных значений
                case
                    when t.data_type = 'numeric'
                    then t.numeric_scale = (r.character_maximum_length::numeric - trunc(r.character_maximum_length::numeric))
                    * power(10, length(split_part(r.character_maximum_length::text, '.', 2)))
                end
                or t.character_maximum_length = coalesce(r.character_maximum_length::numeric, 0)
           )
	)
select *
from fields_comparison

Едем дальше. В следующем CTE,  check_unexpected_fields, выполняется поиск столбцов таблицы, которые есть в фактическом датасете, но отсутствуют в ожидаемом (их условно можно назвать «лишними»). Это реализовано через сравнение записей в CTE columns_info и CTE required_columns. Запрос довольно простой:

select column_name
from columns_info
except
select column_name
from required_columns

По такому запросу каждое название «лишнего» столбца будет выведено в отдельной строке. Однако с учетом того, что это не единственный CTE, из которого будет формироваться результирующая выборка, вернувшиеся значения нужно собрать в одной ячейке. Я это делаю через перечисление в строке, с запятой в качестве разделителя, с помощью функции string_agg (). И дополнительно подсчитываю количество значений, вернувшихся по запросу:

with
	(...),
    -- Ищем лишние столбцы и считаем их количество
    check_unexpected_fields as (
        select
            count (column_name) as count_unexpected_fields,
            string_agg(column_name, ', ') as unexpected_fields
        from (
            select column_name
            from columns_info
            except
            select column_name
            from required_columns
        ) sq
    )
select *
from check_unexpected_fields

Далее в CTE check_missing_fields содержится почти такой же запрос. Разница лишь в том, что запросы в except меняются друг с другом местами. Этот CTE показывает недостающие столбцы тестируемой таблицы. Иначе говоря, те, что есть в ожидаемом датасете, но отсутствуют в фактическом:

with 
	(...),
    -- Ищем недостающие столбцы и считаем их количество
    check_missing_fields as (
        select
            count (column_name) as count_missing_fields,
            string_agg(column_name, ', ') as missing_fields
        from (
            select column_name
            from required_columns
            except
            select column_name
            from columns_info
        ) sq
    )
select *
from check_missing_fields

Итак, на текущей стадии нам известны столбцы:

• Ожидаемые (CTE required_columns);

• Полностью совпавшие с ожидаемыми (CTE fields_comparison);

• Лишние (CTE check_unexpected_fields);

• Недостающие (CTE check_missing_fields).

С помощью этих данных можно найти еще одно множество. В нем будут содержаться столбцы, которые совпадают с ожидаемым датасетом лишь частично. Соответственно, CTE с таким запросом будет возвращать невалидные значения. Следовательно, для него выглядит логичным название check_invalid_fields.

Запрос для поиска невалидных столбцов также содержит в себе оператор except, но в данном случае он используется дважды. Из множества записей ожидаемых столбцов (CTE required_columns) вычитается множество полностью совпавших столбцов (CTE fields_comparison) и недостающих столбцов (CTE check_missing_fields). Для CTE check_missing_fields в рамках такого запроса нужно «развернуть» строку со значениями обратно в набор записей. Это можно сделать с помощью функции string_to_table (). Получаем такой код:

with 
	(...),
    -- Ищем невалидные столбцы и считаем их количество
    check_invalid_fields as (
        select
            count (column_name) as count_invalid_fields,
            string_agg(column_name, ', ') as invalid_fields
        from (
            select column_name
            from required_columns
            except
            select column_name
            from fields_comparison
            except
            select string_to_table(missing_fields, ', ')
            from check_missing_fields
        ) sq
select *
from check_invalid_fields

Вся необходимая информация для тестирования столбцов — собрана. Перейдем к ограничениям (constraints). К слову, к ним в контексте метаданных, в частности, относятся зависимые ключи (foreign keys). Для получения необходимых сведений будем запрашивать из информационной схемы столбцы таблиц:

• constraint_type из table_constraints;

• column_name из constraint_column_usage;

• column_name из key_column_usage;

Если какое-либо ограничение не удовлетворяется при изменении данных таблицы, то по запросу возвращается ошибка. В её формулировке будет указано название нарушаемого в таком случае ограничения (violates constraint):

Ограничения помогают поддерживать целостность данных. По этой причине важно проверять их наличие и логику. Информацию об ограничениях тестируемой таблицы будем получать по отдельному запросу. Его опишем в CTE constraints_query:

with
	(...),
    -- Ищем все ограничения для таблицы
    constraints_query as(
        select
            t1.constraint_type,
            t2.column_name,
            t3.column_name as foreign_column
        from information_schema.table_constraints t1
        left join information_schema.constraint_column_usage t2
            on t1.constraint_name = t2.constraint_name
        left join information_schema.key_column_usage as t3
            on t1.constraint_name = t3.constraint_name
            and t1.table_schema = t3.table_schema
        where
            t1.table_name            = (select table_name   from params)
            and t1.constraint_schema = (select table_schema from params)
            and t2.column_name is not null
    )
select *
from constraints_query

По этому запросу будут возвращаться 3 столбца:

При этом в ожидаемом датасете, в CTE required_constraints, у нас всего 2 столбца:

with
    (...)
    -- Описываем ожидаемые ограничения
    required_constraints(column_name, constraint_type) as (
        values            
            ('id',          'PRIMARY KEY'),
            ('id',    		'FOREIGN KEY'),
            ('model_id',    'FOREIGN KEY'),
            ('price_value', 'UNIQUE'),
            ('model_id',    'UNIQUE')
    )
select *
from required_constraints

Поэтому в следующем CTE (назовем его union_foreign_ref_columns) объединим значения из столбцов column_name и foreign_column в 1 столбец:

with
	(...),
    -- Включаем значения зависимых ключей (foreign_column) в список ограничений (column_name)
    union_foreign_ref_columns as (
        select column_name, constraint_type
        from constraints_query
        union all
        select foreign_column as column_name, constraint_type
        from constraints_query
    )
select *
from union_foreign_ref_columns

Фрагмент as column_name необязателен, но я его оставил его для улучшения читаемости кода.

Фактический набор ограничений получили. Теперь можно проводить сравнение. Я решил не делать отдельный CTE под поиск невалидных ограничений, т.к. здесь рассматриваются всего 2 столбца (column_name и constraints_type). В конечном запросе рассматриваются только «лишние» и недостающие ограничения в 2-х соответствующих CTE. И это почти конец запроса:

with
	(...),
    -- Ищем лишние ограничения и считаем их количество
    check_unexpected_constraints as (
        select
            count (column_name) as count_unexpected_constraints,
            string_agg(column_name || ' (' || constraint_type || ')', ', ') as unexpected_constraints
        from (
            select *
            from union_foreign_ref_columns
            except
            select column_name, constraint_type
            from required_constraints
        ) sq
    )
select *
from check_unexpected_constraints

и

with
	(...),
    -- Ищем недостающие ограничения и считаем их количество
    check_missing_constraints as (
        select
            count (column_name) as count_missing_constraints,
            string_agg(column_name || ' (' || constraint_type || ')', ', ') as missing_constraints
        from (
            select column_name, constraint_type
            from required_constraints
            except
            select *
            from union_foreign_ref_columns
        ) sq
    )
select *
from check_missing_constraints

Логика сравнения здесь такая же, как при сравнении столбцов. Отличие между этими 2-мя CTE — только в порядке расположения относительно except: в check_unexpected_constraints запрос к union_foreign_ref_columns идет до except, а в check_missing_constraints — после. Из интересного здесь разве что данная конструкция:

string_agg(column_name || ' (' || constraint_type || ')', ', ')

string_agg () здесь по-прежнему нужен для вывода нескольких записей в 1-й строке. Но здесь 1-й аргумент функции логически немного отличается от этого же аргумента функции в коде сравнения столбцов. Это нужно исключительно для того, чтобы в 1-й ячейке уместить не только несколько строк, но и информацию сразу по 2-м столбцам. В итоговой выборке значение в этой ячейке будет выглядеть так:

По этому скриншоту проще понять, почему конструкция с string_agg () именно такая, какая она есть. В нее включается информацию из 2-х столбцов, при этом значения 2-го столбца берется в скобки с пробелом перед 1-м значением.

Ну и в последнем CTE собираем информацию, полученную в других CTE, и суммируем количество найденных ошибок, которые мы получали с помощью count ():

with
	(...),
    -- Собираем полученные данные
    checks as (
        select
            -- Выводим всю информацию об ошибках и суммируем их количество
            table_exists_error + count_unexpected_fields + count_missing_fields + count_invalid_fields +
            count_unexpected_constraints + count_missing_constraints as errors,
            *
        from        check_table_exist
        cross join  check_unexpected_fields
        cross join  check_missing_fields
        cross join  check_invalid_fields
        cross join  check_unexpected_constraints
        cross join  check_missing_constraints
)
select *
from checks

Значения, выводимые этим запросом в строках, видны в самих строках, в тултипе (если в строке большое значение), и их также можно скопировать и вставить блокнот:

Почему этот запрос полезен?

• По моему опыту, почему-то новые таблицы БД, которые создают разработчики, часто не соответствуют тому, что описано в требованиях. Исходя из этого, проще один раз описать ожидаемые значения в запросе и запускать его после правок. Это сэкономит время при сравнении и исключит человеческий фактор. Особенно если итераций правок будет несколько;

• Польза от таких проверочных запросов тем выше, чем больше таблиц добавляется в рамках доработки;

• Пригодится для быстрого тестирования метаданных таблиц после миграции баз данных на другой стенд.

© Habrahabr.ru