doc.prototypes.ru: Проекты: PostgreSQL: Кластерные решения: Распределение данных с помощью dblink: Проксирующая база данных: Выборка данных

Равномерное распределение с помощью dblink

Проксирующая база данных

Выборка данных

План выборки данных

Выборки данных можно разделить на три вида:

Выборки из одной базы данных без учета распределения - это данные из индексной базы;
Выборки из одной базы данных c учетом распределения - это данные из распределенной базы, но в её пределах, например, список альбомов пользователя;
Выборки из нескольких баз данных - это данные которые аггрегируются из нескольких хранилищ, например список фото для определенного тега;

Определим, какие данные будет выбирать наше приложение.

Независимые списки:
- Список ТОПовых фото, с возможностью постраничного вывода (limit, offset);
- Список последних фото, с возможностью постраничного вывода (limit, offset);
Зависимые списки:
- Для пользователя:
- - Список альбомов пользователя, с возможностью сортировки и постраничного вывода (user_id, user_hash, order, limit, offset);
  - Список тегов пользователя (user_id, user_hash);
  - Список последних фото пользователя, с возможностью постраничного вывода (user_id, user_hash, limit, offset);
- Для альбома:
- - Список фото альбома, с возможностью сортировки и постраничного вывода (album_id, user_hash, order, limit, offset);
  - Список тегов альбома (album_id, user_hash);
- Для фото:
- - Список тегов фото (foto_id, user_hash);
- Для тегов:
- - Список фото для тега, с возможностью постраничного вывода (tag_id, tag_hash, limit, offset);
Строковые данные:
- Пользователь, (user_id + user_hash или login +? pass);
- Альбом (album_id, user_hash);
- Фото (foto_id, user_hash);
- Фото + Альбом + Пользователь (foto_id, user_hash);
- Тег - у нас уже есть процедура получения по контенту;

Независимые списки

У нас есть 2 таблицы: топовые и последние фото в индексной базе данных, из них выбираем только с учетом постраницного вывода.

Итак, для топовых фото:

SQL код (1)

CREATE OR REPLACE FUNCTION "public"."get_index_top_foto" (
    limit_param         integer,
    offset_param        integer
) RETURNS SETOF "public"."fotos" AS
$body$
DECLARE
    index_connect       TEXT;
    foto_connect        TEXT;
    foto_connect_now    TEXT;
    query               TEXT;
    top_foto_row        index_top_foto;
    top_foto_list       index_top_foto[];
    fotos_row           fotos;
    fotos_list          fotos[];
    foto_id             TEXT;
    foto_ids            TEXT[];
    t_limit             INTEGER             := 30;
    t_offset            INTEGER             := 0;
BEGIN
-- Выбираем список топовых фото
    SELECT index_db_select(TRUE) INTO index_connect;
    IF limit_param IS NOT NULL THEN t_limit := limit_param; END IF;
    IF offset_param IS NOT NULL THEN t_offset := offset_param; END IF;
    query := 'SELECT * 
                FROM index_top_foto 
                ORDER BY rating DESC 
                LIMIT ' || t_limit || '
                OFFSET ' || t_offset;
-- Формируем массив топовых фото
    FOR top_foto_row IN
        SELECT * FROM _dblink_index_top_foto(index_connect, query, TRUE) 
    LOOP
        top_foto_list := array_append(top_foto_list, top_foto_row);
    END LOOP;
-- Выбираем фото из хранилищ, при этом группируем по hash
    FOR top_foto_row IN 
        SELECT * FROM unnest(top_foto_list) ORDER BY user_hash
    LOOP
        SELECT storage_select_by_hash(top_foto_row.user_hash, TRUE) INTO foto_connect;
        foto_id := '(' || 
                        quote_literal(top_foto_row.user_hash) || ',' || 
                        top_foto_row.foto_id || ')';
        IF foto_connect <> foto_connect_now AND foto_connect_now IS NOT NULL THEN
            query := 'SELECT * 
                        FROM fotos 
                        WHERE (user_hash, id) IN (' || 
                              array_to_string(foto_ids, ',') || ');';
            FOR fotos_row IN 
                SELECT * FROM _dblink_fotos(foto_connect_now, query, TRUE) 
            LOOP
                fotos_list := array_append(fotos_list, fotos_row);
            END LOOP;
            foto_ids := NULL;
            foto_connect_now := foto_connect;
        ELSIF foto_connect_now IS NULL THEN
            foto_connect_now := foto_connect;
        END IF;
        foto_ids := array_append(foto_ids, foto_id);
    END LOOP;
-- Забираем последний список
    query := 'SELECT * 
                FROM fotos 
                WHERE (user_hash, id) IN (' || array_to_string(foto_ids, ',') || ');';
    FOR fotos_row IN 
        SELECT * FROM _dblink_fotos(foto_connect_now, query, TRUE) 
    LOOP
        fotos_list := array_append(fotos_list, fotos_row);
    END LOOP;
-- Сортируем список как надо и возвращаем
    FOR fotos_row IN SELECT * FROM unnest(fotos_list) ORDER BY rating DESC LOOP
        RETURN NEXT fotos_row;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

-- И вспомогательные процедуры, без ненужных параметров
CREATE FUNCTION "public"."get_fotos_top" () RETURNS "public"."fotos" AS
$body$
SELECT * FROM get_fotos_top(NULL, NULL);
$body$
LANGUAGE 'sql';

CREATE FUNCTION "public"."get_fotos_top" (offset_param integer) RETURNS "public"."fotos" AS
$body$
SELECT * FROM get_fotos_top(NULL, $1);
$body$
LANGUAGE 'sql';

-- Вызов:
SELECT * FROM get_index_top_foto();
SELECT * FROM get_index_top_foto(20);
SELECT * FROM get_index_top_foto(20,20);

При штатной работе серверов баз данных, у нас все бедет хорошо, но в случае если одно из хранилищ остановлено, то мы получим ошибку. В принципе, это нормально для данных, относящихся только к этому хранилищу, но в случае с выборками данных с нескольких серверов, мы все таки можем выбрать список пусть не полный. Это позволит нам не останавливать проект, а включить режим частичной работы. Хранимую процедуру для выборки последних фото так и сделаем:

SQL код (2)

CREATE OR REPLACE FUNCTION "public"."get_index_last_foto" (
    limit_param             integer,
    offset_param            integer
) RETURNS SETOF "public"."fotos" AS
$body$
DECLARE
    index_connect           TEXT;
    foto_connect            TEXT;
    foto_connect_now        TEXT;
    query                   TEXT;
    last_foto_row           index_last_foto;
    last_foto_list          index_last_foto[];
    fotos_row               fotos;
    fotos_list              fotos[];
    fotos_count             INTEGER;
    foto_id                 TEXT;
    foto_ids                TEXT[];
    t_limit                 INTEGER             := 30;
    t_offset                INTEGER             := 0;
BEGIN
-- Выбираем список топовых фото
    SELECT index_db_select(TRUE) INTO index_connect;
    IF limit_param IS NOT NULL THEN t_limit := limit_param; END IF;
    IF offset_param IS NOT NULL THEN t_offset := offset_param; END IF;
    query := 'SELECT * 
                FROM index_last_foto 
                ORDER BY added DESC 
                LIMIT ' || t_limit || '
                OFFSET ' || t_offset;
-- Формируем массив топовых фото
    FOR last_foto_row IN 
        SELECT * FROM _dblink_index_last_foto(index_connect, query, TRUE) 
    LOOP
        last_foto_list := array_append(last_foto_list, last_foto_row);
    END LOOP;
-- Выбираем фото из хранилищ, при этом группируем по hash
    FOR last_foto_row IN 
        SELECT * FROM unnest(last_foto_list) ORDER BY user_hash
    LOOP
        SELECT storage_select_by_hash(last_foto_row.user_hash, FALSE) INTO foto_connect;
-- Коннект к удаленной базе получили
        IF foto_connect <> 'ERR' THEN
            foto_id := '(' || 
                            quote_literal(last_foto_row.user_hash) || ',' || 
                            last_foto_row.foto_id || ')';
            IF foto_connect <> foto_connect_now AND foto_connect_now IS NOT NULL THEN
                query := 'SELECT * 
                                FROM fotos 
                                WHERE (user_hash, id) IN (' || 
                                  array_to_string(foto_ids, ',') || ');';
                FOR fotos_row IN 
                    SELECT * FROM _dblink_fotos(foto_connect_now, query, TRUE) 
                LOOP
                    fotos_list := array_append(fotos_list, fotos_row);
                END LOOP;
                foto_ids := NULL;
                foto_connect_now := foto_connect;
            ELSIF foto_connect_now IS NULL THEN
                foto_connect_now := foto_connect;
            END IF;
            foto_ids := array_append(foto_ids, foto_id);
        END IF;
    END LOOP;
-- Забираем последний список
    query := 'SELECT * 
                    FROM fotos 
                    WHERE (user_hash, id) IN (' || 
                      array_to_string(foto_ids, ',') || ');';
    FOR fotos_row IN 
        SELECT * FROM _dblink_fotos(foto_connect_now, query, TRUE) 
    LOOP
        fotos_list := array_append(fotos_list, fotos_row);
    END LOOP;
-- Сортируем список как надо и возвращаем
    FOR fotos_row IN 
        SELECT t2.* FROM unnest(last_foto_list) AS t1 
            JOIN unnest(fotos_list) AS t2 ON
            t1.foto_id = t2.id AND t1.user_hash = t2.user_hash
    LOOP
        RETURN NEXT fotos_row;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

-- И вспомогательные процедуры, без ненужных параметров
CREATE OR REPLACE FUNCTION "public"."get_index_last_foto" ()
    RETURNS SETOF "public"."fotos" AS
$body$
SELECT * FROM get_index_last_foto(NULL,NULL);
$body$
LANGUAGE 'sql';

CREATE OR REPLACE FUNCTION "public"."get_fotos_last" (
    offset_param    integer
) RETURNS "public"."fotos" AS
$body$
SELECT * FROM get_fotos_last(NULL,$1);
$body$
LANGUAGE 'sql';

-- Вызов:
SELECT * FROM get_index_last_foto();
SELECT * FROM get_index_last_foto(20);
SELECT * FROM get_index_last_foto(20,20);

В итоге, му можем получить не полный список последних фото, но мы его получим все равно. Можно, конечно, добирать список если не достает строк, но при этом у нас сильно пострадает постраничное разделение.

Зависимые списки

Данные по пользователям, альбомам и фото выбираются в пределах одного хранилища (ктоме выборки тегов, тогда еще используется индексная база), поэтому выборки относительно простые, можно даже не заострять на них внимание:

SQL код (3)

-- Список альбомов пользователя
CREATE OR REPLACE FUNCTION "public"."get_albums_by_user" (
    user_id             integer,
    user_hash           text,
    order_param         text,
    limit_param         integer,
    offset_param        integer
) RETURNS SETOF "public"."albums" AS
$body$
DECLARE
    album               albums;
    storage             TEXT;
    query               TEXT;
    t_limit             INTEGER             := 30;
    t_offset            INTEGER             := 0;
    t_order             TEXT                := 'id';
BEGIN
-- Обрабатываем входящие данные
    IF limit_param IS NOT NULL THEN t_limit := limit_param; END IF;
    IF offset_param IS NOT NULL THEN t_offset := offset_param; END IF;
    IF order_param IS NOT NULL THEN t_order := order_param; END IF;
-- Выбираем распределенный сервер для пользователя
    SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
    query   := 'SELECT * 
                FROM albums 
                WHERE   user_id = ' || user_id || ' AND 
                        user_hash = ' || quote_literal(user_hash) || '
                ORDER BY ' || t_order || '
                LIMIT ' || t_limit || '
                OFFSET ' || t_offset;
    FOR album IN SELECT * FROM _dblink_albums(storage, query, TRUE) LOOP
        RETURN NEXT album;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

-- Список последних фото пользователя
CREATE OR REPLACE FUNCTION "public"."get_fotos_last_by_user" (
    user_id                 integer,
    user_hash               text,
    limit_param             integer,
    offset_param            integer
) RETURNS SETOF "public"."fotos" AS
$body$
DECLARE
    storage                 TEXT;
    foto                    fotos;
    query                   TEXT;
    t_limit                 INTEGER             := 30;
    t_offset                INTEGER             := 0;
BEGIN
-- Выбираем хранилище
    SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
    IF limit_param IS NOT NULL THEN t_limit := limit_param; END IF;
    IF offset_param IS NOT NULL THEN t_offset := offset_param; END IF;
    query := 'SELECT * 
                FROM fotos 
                WHERE   user_id = ' || user_id || ' AND
                    user_hash = ' || quote_literal(user_hash) || '
                ORDER BY id DESC
                LIMIT ' || t_limit || '
                OFFSET ' || t_offset;
    FOR foto IN SELECT * FROM _dblink_fotos(storage, query, TRUE) LOOP
        RETURN NEXT foto;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

-- Список фото альбома
CREATE OR REPLACE FUNCTION "public"."get_fotos_by_album" (
    album_id            integer,
    user_hash           text,
    order_param         text,
    limit_param         integer,
    offset_param        integer
) RETURNS SETOF "public"."fotos" AS
$body$
DECLARE
    storage             TEXT;
    foto                fotos;
    query               TEXT;
    t_limit             INTEGER             := 30;
    t_offset            INTEGER             := 0;
    t_order             TEXT                := 'id';
BEGIN
-- Обрабатываем входящие данные
    IF limit_param IS NOT NULL THEN t_limit := limit_param; END IF;
    IF offset_param IS NOT NULL THEN t_offset := offset_param; END IF;
    IF order_param IS NOT NULL THEN t_order := order_param; END IF;
-- Выбираем распределенный сервер для пользователя
    SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
    query   := 'SELECT * 
            FROM fotos 
            WHERE   album_id = ' || album_id || ' AND 
                        user_hash = ' || quote_literal(user_hash) || '
            ORDER BY ' || t_order || '
            LIMIT ' || t_limit || '
            OFFSET ' || t_offset;
    FOR foto IN SELECT * FROM _dblink_fotos(storage, query, TRUE) LOOP
        RETURN NEXT foto;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

Отдельно рассмотрим выборки тегов для сущностей (пользователь, альбом, фото). В разделе структура распределенной DB, мы ввели несколько денормализованных связей (таблиц), теперь рассмотрим для чего они нам потребовались. Итак, у нас есть таблица fotos_tags в которой добавлены избыточные поля - album_id и user_id, по сути, для выборки тегов альбома или пользователя, этой таблицы будет достаточно, но, для этого нам потребуется применять DISTINCT, что бы получить уникальные теги в пределах условия, или GROUP BY, что бы получить уникальные теги в пределах условия и посчитать частоту их использования. Ясно, что применение этих директив уменьшит производительность, хотя и позволит нам избежать дополнительных денормализаций, но тут стоит посмотреть на приоритет и частоту использования задач этих данных, а приоритет таков - нужно быстро отдать данные, а изменения не так часты.

Решений несколько, рассмотрим два из них:

Отдельная таблица связей, с количественными показателями (для альбомов) - позволит нам двумя запросами получить нужные теги и частоту их использования. Такой связью просто управлять и обновлять, но в нашем случае, это запросы к двум базам: хранилищу, для получени списка tag_id и к индексной базе для получения тегов;
Отдельная поле у сущности, с количественными показателями (для пользователей) - позволит нам одним запросом получить нужные теги и частоту их использования. Такой связью управлять сложнее, как было видно из триггеров, так как мы работаем не со строками, а с массивом в поле, Но зато, список tag_id у нас есть у сущности;

В принципе, можно вообще уйти от запросов, как таковых, и хранить в отдельном поле сущности все данные от тегах, но в распределенном хранилище, у нас используется только tag_id и tag_hash, а контент тега не никогда передается, поэтому усложнять логику мне кажется не совсем целесообразно.

Итого получаем три вида хранимых процедур для выборки тегов, какой способ выбирать - оставляю на совести разработчика.

SQL код (4)

-- Выборка тегов для фото
CREATE OR REPLACE FUNCTION "public"."get_tags_by_foto" (
    foto_id             integer,
    user_hash           text
) RETURNS SETOF "public"."index_tags" AS
$body$
DECLARE
    storage             TEXT;
    index_db            TEXT;
    query               TEXT;
    fotos_tag           fotos_tags;
    tag_ids             INTEGER[];
    tag                 index_tags;
BEGIN
-- Выбираем хранилище
    SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
-- Формируем запрос к хранилищу и выбираем ID тегов
    query := 'SELECT * 
            FROM fotos_tags 
            WHERE foto_id = ' || foto_id || ' AND 
                user_hash = ' || quote_literal(user_hash);
    FOR fotos_tag IN SELECT * FROM _dblink_fotos_tags(storage, query, TRUE) LOOP
        tag_ids := array_append(tag_ids, fotos_tag.tag_id);
    END LOOP;
-- Формируем запрос к индексной базе и выбираем теги
    SELECT index_db_select(TRUE) INTO index_db;
    query := 'SELECT * 
                FROM index_tags 
                WHERE id IN(' || array_to_string(tag_ids, ',') || ')';
    FOR tag IN SELECT * FROM _dblink_index_tags(index_db, query, TRUE) LOOP
        RETURN NEXT tag;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

-- Выборка тегов для альбома, практически один в один - как для фото
CREATE OR REPLACE FUNCTION "public"."get_tags_by_album" (
    album_id            integer,
    user_hash           text
) RETURNS SETOF "public"."index_tags" AS
$body$
DECLARE
    storage             TEXT;
    index_db            TEXT;
    query               TEXT;
    albums_tag          albums_tags;
    tag_ids             INTEGER[];
    tag                 index_tags;
BEGIN
-- Выбираем хранилище
    SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
-- Формируем запрос к хранилищу и выбираем ID тегов
    query := 'SELECT * 
                FROM albums_tags 
                WHERE album_id = ' || foto_id || ' AND 
                    user_hash = ' || quote_literal(user_hash);
    FOR albums_tag IN SELECT * FROM _dblink_albums_tags(storage, query, TRUE) LOOP
        tag_ids := array_append(tag_ids, albums_tag.tag_id);
    END LOOP;
-- Формируем запрос к индексной базе и выбираем теги
    SELECT index_db_select(TRUE) INTO index_db;
    query := 'SELECT * 
                FROM index_tags 
                WHERE id IN(' || array_to_string(tag_ids, ',') || ')';
    FOR tag IN SELECT * FROM _dblink_index_tags(index_db, query, TRUE) LOOP
        RETURN NEXT tag;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

-- Выборка тегов для пользователя - процедура выборки тегов по id
CREATE OR REPLACE FUNCTION "public"."get_tags_by_ids" (
    ids                 integer[]
) RETURNS SETOF "public"."index_tags" AS
$body$
DECLARE
    tag                 index_tags;
    index_db            TEXT;
    query               TEXT;
BEGIN
-- Формируем запрос к индексной базе и выбираем теги
    SELECT index_db_select(TRUE) INTO index_db;
    query := 'SELECT * 
                FROM index_tags 
                WHERE id IN(' || array_to_string(ids, ',') || ')';
    FOR tag IN SELECT * FROM _dblink_index_tags(index_db, query, TRUE) LOOP
        RETURN NEXT tag;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

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

SQL код (5)

CREATE OR REPLACE FUNCTION "public"."get_fotos_by_tag" (
    tag_id                  integer,
    tag_hash                text,
    limit_param             integer,
    offset_param            integer
) RETURNS SETOF "public"."fotos" AS
$body$
DECLARE
    tag_connect             TEXT;
    foto_connect            TEXT;
    foto_connect_now        TEXT;
    query                   TEXT;
    foto_tag                fotos_tags;
    foto_tags               fotos_tags[];
    fotos_row               fotos;
    fotos_list              fotos[];
    foto_id                 TEXT;
    foto_ids                TEXT[];
    t_limit                 INTEGER             := 30;
    t_offset                INTEGER             := 0;
BEGIN
-- Выбираем список фото для тега
    SELECT storage_select_by_hash(tag_hash, TRUE) INTO tag_connect;
    IF limit_param IS NOT NULL THEN t_limit := limit_param; END IF;
    IF offset_param IS NOT NULL THEN t_offset := offset_param; END IF;
    query := 'SELECT * 
                FROM fotos_tags
                WHERE tag_id = ' || tag_id || '
                ORDER BY foto_id 
                LIMIT ' || t_limit || '
                OFFSET ' || t_offset;
-- Формируем массив ID фото
    FOR foto_tag IN 
        SELECT * FROM _dblink_fotos_tags(tag_connect, query, TRUE) 
    LOOP
        foto_tags := array_append(foto_tags, foto_tag);
    END LOOP;
-- Выбираем фото из хранилищ, при этом группируем по hash
    FOR foto_tag IN 
        SELECT * FROM unnest(foto_tags) ORDER BY user_hash
    LOOP
        SELECT storage_select_by_hash(foto_tag.user_hash, TRUE) INTO foto_connect;
        foto_id := '(' || 
                        quote_literal(foto_tag.user_hash) || ',' || 
                        foto_tag.foto_id || ')';
        IF foto_connect <> foto_connect_now AND foto_connect_now IS NOT NULL THEN
            query := 'SELECT * 
                        FROM fotos 
                        WHERE (user_hash, id) IN (' || 
                              array_to_string(foto_ids, ',') || ');';
            FOR fotos_row IN 
                SELECT * FROM _dblink_fotos(foto_connect_now, query, TRUE) 
            LOOP
                fotos_list := array_append(fotos_list, fotos_row);
            END LOOP;
            foto_ids := NULL;
            foto_connect_now := foto_connect;
        ELSIF foto_connect_now IS NULL THEN
            foto_connect_now := foto_connect;
        END IF;
        foto_ids := array_append(foto_ids, foto_id);
    END LOOP;
-- Забираем последний список
    query := 'SELECT * 
                    FROM fotos 
                    WHERE (user_hash, id) IN (' || 
                      array_to_string(foto_ids, ',') || ');';
    FOR fotos_row IN 
        SELECT * FROM _dblink_fotos(foto_connect_now, query, TRUE) 
    LOOP
        fotos_list := array_append(fotos_list, fotos_row);
    END LOOP;
-- Сортируем список как надо и возвращаем
    FOR fotos_row IN SELECT * FROM unnest(fotos_list) ORDER BY rating DESC LOOP
        RETURN NEXT fotos_row;
    END LOOP;
    RETURN;
END;
$body$
LANGUAGE 'plpgsql';

Можно, конечно, сделать еще процедуру, в которую будем передавать контент тега, а не id и hash, но я не думаю, что для приложения это будет актуально.

Строковые данные

Строковые данные выбираются довольно просто, мы определяем хранилище, выбираем из него и отдаем приложению.

SQL код (6)

-- Выборка пользователя
CREATE OR REPLACE FUNCTION "public"."get_user" (
    user_id         integer,
    user_hash       text,
    user_login      text,
    user_pass       text
) RETURNS "public"."users" AS
$body$
DECLARE
    storage         TEXT;
    query           TEXT;
    user_row        users;
BEGIN
    IF user_id IS NOT NULL AND user_hash IS NOT NULL THEN
        SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
        query := 'SELECT * FROM users
                   WHERE id = ' || user_id || ' AND 
                      user_hash = ' || quote_literal(user_hash);
    ELSIF user_login IS NOT NULL THEN
        SELECT storage_select_by_text(user_login, TRUE) INTO storage;
        query := 'SELECT * FROM users 
                WHERE login = ' || quote_literal(user_login) ||
                CASE WHEN user_pass IS NOT NULL
                    THEN ' AND pass = ' || quote_literal(user_pass)
                    ELSE ''
                END;
    ELSE
        RETURN NULL;
    END IF;
    SELECT * INTO user_row 
        FROM _dblink_users(storage, query, TRUE);
    RETURN user_row;
END;
$body$
LANGUAGE 'plpgsql';

-- Дополнительные процедуры, без лишних параметров
CREATE OR REPLACE FUNCTION "public"."get_user" (
    user_id         integer,
    user_hash       text
) RETURNS "public"."users" AS
$body$
SELECT * FROM get_user($1, $2, NULL, NULL);
$body$
LANGUAGE 'sql';

CREATE OR REPLACE FUNCTION "public"."get_user" (
    user_login      text,
    user_pass       text
) RETURNS "public"."users" AS
$body$
SELECT * FROM get_user(NULL, NULL, $1, $2);
$body$
LANGUAGE 'sql';

CREATE OR REPLACE FUNCTION "public"."get_user" (
    user_login      text
) RETURNS "public"."users" AS
$body$
SELECT * FROM get_user(NULL, NULL, $1, NULL);
$body$
LANGUAGE 'sql';

-- Выборка альбома
CREATE OR REPLACE FUNCTION "public"."get_album" (
    album_id        integer,
    user_hash       text
) RETURNS "public"."albums" AS
$body$
DECLARE
    storage         TEXT;
    query           TEXT;
    album           albums;
BEGIN
    IF user_hash IS NULL OR album_id IS NULL THEN
        RETURN NULL;
    END IF;
    SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
    query := 'SELECT * FROM albums 
            WHERE id = ' || album_id || ' AND 
                user_hash = ' || quote_literal(user_hash);
    SELECT * INTO album FROM _dblink_albums(storage, query, TRUE);
    RETURN album;
END;
$body$
LANGUAGE 'plpgsql';

-- Выборка фото
CREATE OR REPLACE FUNCTION "public"."get_foto" (
    foto_id         integer,
    user_hash       text
) RETURNS "public"."fotos" AS
$body$
DECLARE
    storage         TEXT;
    query           TEXT;
    foto            fotos;
BEGIN
    IF user_hash IS NULL OR foto_id IS NULL THEN
        RETURN NULL;
    END IF;
    SELECT storage_select_by_hash(user_hash, TRUE) INTO storage;
    query := 'SELECT * FROM fotos 
            WHERE id = ' || foto_id || ' AND 
                user_hash = ' || quote_literal(user_hash);
    SELECT * INTO foto FROM _dblink_fotos(storage, query, TRUE);
    RETURN foto;
END;
$body$
LANGUAGE 'plpgsql';

Единственно, что может понадобится не "чистая" строка из таблицы, а результат JOIN таблиц. Так, например, нам нужно взять строку фото, с данными о альбоме и пользователе. Сложности возникают только могут возникнуть только при выдаче результатов, так как нам прийдется проименовать каждое возвращаемое поле. Составной индекс для этого я бы не рекомендовал использовать так как его потом невозможно будет изменить без удаления зависимостей - хранимых процедур. Лучше создать пустую таблицу, её можно относительно спокойно менять, относительно - значит что при изменении запроса к хранилищу её потребуется изменить и наоборот.

Миграция данных

Сергей Томулевич aka Phoinix (01.12.2009 г.)