Синхронизация данных

gsqlcmd позволяет синхронизировать данные между базами данных на разных платформах серверов баз данных.

Ниже мы обсудим несколько техник.

Вы можете найти исходный код в папке "Examples\Sync Databases" загруженного пакета gsqlcmd.

Кроме того, вы можете попробовать любой пакетный файл с демонстрационным образцом, размещенным в базе данных Microsoft Azure SQL.

Содержание:

• Основы
• Двусторонняя синхронизация с использованием идентификаторов GUID
• Использование столбцов Rowversion/Timestamp SQL Server для синхронизации новых и измененных строк
• Использование столбцов последнего обновления для синхронизации новых и измененных строк
• Полная синхронизация с удалением
• Полная синхронизация между несколькими редактируемыми базами данных с целочисленными ключевыми столбцами
• Улучшение и унификация решений
• Заключение

Основы

Предположим, у нас есть две базы данных, db1 и db2, с двумя таблицами клиентов, table11 в db1 и table12 в db2.

CREATE TABLE [s61].[table11] (
      [id] int IDENTITY(1,1) NOT NULL
    , [name] nvarchar(50) NOT NULL
    , CONSTRAINT [PK_table11] PRIMARY KEY ([id])
);

CREATE TABLE [s61].[table12] (
      [id] int IDENTITY(1,1) NOT NULL
    , [name] nvarchar(50) NOT NULL
    , CONSTRAINT [PK_table12] PRIMARY KEY ([id])
);

Таблица 11 содержит следующие записи:

Чтобы объединить данные из table11 в table12, можно использовать следующий пакетный файл:

gsqlcmd exec db1 s61.table11 table11.csv

gsqlcmd make db2 table11.csv table12.sql /table=s61.table12 /merge /insertIdentity

gsqlcmd exec db2 table12.sql

Первая команда exec экспортирует данные из таблицы table11 в файл CSV:

id;"name"
1;"Customer C1"
2;"Customer C2"
3;"Customer C3"
4;"Customer C4"
5;"Customer C5"

Вторая команда make создает команды слияния в таблицу table12:

UPDATE [s61].[table12] SET [name] = N'Customer C1' WHERE [id] = 1;
UPDATE [s61].[table12] SET [name] = N'Customer C2' WHERE [id] = 2;
UPDATE [s61].[table12] SET [name] = N'Customer C3' WHERE [id] = 3;
UPDATE [s61].[table12] SET [name] = N'Customer C4' WHERE [id] = 4;
UPDATE [s61].[table12] SET [name] = N'Customer C5' WHERE [id] = 5;
GO
SET IDENTITY_INSERT [s61].[table12] ON;
INSERT INTO [s61].[table12] ([id], [name]) SELECT s.[id], s.[name] FROM (SELECT 1 AS [id], N'Customer C1' AS [name]) s LEFT OUTER JOIN [s61].[table12] t ON t.[id] = s.[id] WHERE t.[id] IS NULL;
INSERT INTO [s61].[table12] ([id], [name]) SELECT s.[id], s.[name] FROM (SELECT 2 AS [id], N'Customer C2' AS [name]) s LEFT OUTER JOIN [s61].[table12] t ON t.[id] = s.[id] WHERE t.[id] IS NULL;
INSERT INTO [s61].[table12] ([id], [name]) SELECT s.[id], s.[name] FROM (SELECT 3 AS [id], N'Customer C3' AS [name]) s LEFT OUTER JOIN [s61].[table12] t ON t.[id] = s.[id] WHERE t.[id] IS NULL;
INSERT INTO [s61].[table12] ([id], [name]) SELECT s.[id], s.[name] FROM (SELECT 4 AS [id], N'Customer C4' AS [name]) s LEFT OUTER JOIN [s61].[table12] t ON t.[id] = s.[id] WHERE t.[id] IS NULL;
INSERT INTO [s61].[table12] ([id], [name]) SELECT s.[id], s.[name] FROM (SELECT 5 AS [id], N'Customer C5' AS [name]) s LEFT OUTER JOIN [s61].[table12] t ON t.[id] = s.[id] WHERE t.[id] IS NULL;
SET IDENTITY_INSERT [s61].[table12] OFF;
GO
print 'Processed 5 total records';
GO

Эти команды SQL обновляют имена в целевой таблице и вставляют новые строки.

В команде make используется опция /merge для создания таких команд и опция /insertIdentity для сохранения значений идентификаторов.

Третья команда exec выполняет команды SQL в базе данных db2.

Решение проблем с SET IDENTITY_INSERT

Чтобы выполнить сценарий с SET IDENTITY_INSERT в Microsoft SQL Server, пользователь должен иметь разрешение ALTER ON TABLE.

В противном случае возвращается ошибка: не удается найти объект "s61.table12", поскольку он не существует или у вас нет разрешений.

См. подробности на странице https://learn.microsoft.com/en-us/sql/t-sql/statements/set-identity-insert-transact-sql?view=sql-server-2017.

Чтобы решить эту проблему, можно использовать отдельного пользователя для синхронизации данных или создать целевую таблицу со столбцом первичного ключа без опции IDENTITY:

CREATE TABLE [s61].[table13] (
      [id] int NOT NULL
    , [name] nvarchar(50) NOT NULL
    , CONSTRAINT [PK_table13] PRIMARY KEY ([id])
);

Техника осталась прежней, но SQL-команды слияния не будут содержать SET IDENTITY_INSERT.

Если вы не можете изменить столбец идентификаторов, то можно использовать решение, приведенное ниже.

Использование разных значений идентификаторов в исходной и целевой таблицах

Если вам не нужно сохранять одинаковые значения ID в обеих таблицах, вы можете опустить столбцы идентификаторов при синхронизации.

Например:

gsqlcmd exec db1 "SELECT name FROM s61.table11" table11.csv

gsqlcmd make db2 table11.csv table14.sql /table=s61.table14 /keys=name /merge

gsqlcmd exec db2 table14.sql

Первая команда exec экспортирует данные без столбца ID:

"name"
"Customer C1"
"Customer C2"
"Customer C3"
"Customer C4"
"Customer C5"

Обратите внимание, что команда использует запрос прямо в командной строке: SELECT name FROM s61.table11

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

Кроме того, вы можете рассчитать значения для необходимых столбцов целевой таблицы.

Вторая команда создает следующие команды SQL для вставки отсутствующих имен в таблицу12:

INSERT INTO [s61].[table14] ([name]) SELECT s.[name] FROM (SELECT N'Customer C1' AS [name]) s LEFT OUTER JOIN [s61].[table14] t ON t.[name] = s.[name] WHERE t.[name] IS NULL;
INSERT INTO [s61].[table14] ([name]) SELECT s.[name] FROM (SELECT N'Customer C2' AS [name]) s LEFT OUTER JOIN [s61].[table14] t ON t.[name] = s.[name] WHERE t.[name] IS NULL;
INSERT INTO [s61].[table14] ([name]) SELECT s.[name] FROM (SELECT N'Customer C3' AS [name]) s LEFT OUTER JOIN [s61].[table14] t ON t.[name] = s.[name] WHERE t.[name] IS NULL;
INSERT INTO [s61].[table14] ([name]) SELECT s.[name] FROM (SELECT N'Customer C4' AS [name]) s LEFT OUTER JOIN [s61].[table14] t ON t.[name] = s.[name] WHERE t.[name] IS NULL;
INSERT INTO [s61].[table14] ([name]) SELECT s.[name] FROM (SELECT N'Customer C5' AS [name]) s LEFT OUTER JOIN [s61].[table14] t ON t.[name] = s.[name] WHERE t.[name] IS NULL;
GO
print 'Processed 5 total records';
GO

Обратите внимание, что команда make использует параметр /keys=name для указания столбцов, используемых в условиях JOIN.

Использование режима импорта

Односторонняя синхронизация, рассмотренная выше, всегда включает три части:

1. Экспорт данных из исходной таблицы
2. Создание команд SQL для обновления целевой таблицы
3. Выполнение команд SQL

Вы можете создавать пакетные файлы, используя файлы CSV на первом и втором шагах. Эта функция доступна во всех платных версиях.

Редакция gsqlcmd Enterprise позволяет использовать режим импорта, объединяющий эти три команды в одну.

Например, следующая команда объединяет данные из таблицы table11 в таблицу table13:

gsqlcmd import db2 db1::s61.table11 /table=s61.table13 /merge

Основная команда выполняется для соединения с целевой db2.

Однако команда загружает исходные данные из таблицы s61.table11 базы данных db1 в память, а не в CSV-файл, как команда:

gsqlcmd exec db1 s61.table11 table11.csv

Команда импорта с SELECT в командной строке выглядит так:

gsqlcmd import db2 "db1::SELECT name FROM s61.table11" /table=s61.table14 /keys=name /merge

Таким образом, вы можете легко изменить первоначальный make на режим import и обратно.

Следующие образцы будут использовать режим импорта.

Двусторонняя синхронизация с использованием идентификаторов GUID

В предыдущих примерах обе таблицы имели столбцы первичного ключа целочисленного типа.

Односторонняя синхронизация в этом случае проста и надежна.

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

Вот объявление таблиц для Microsoft SQL Server:

CREATE TABLE [s61].[table21] (
      [id] uniqueidentifier NOT NULL DEFAULT(newid())
    , [name] nvarchar(50) NOT NULL
    , CONSTRAINT [PK_table21] PRIMARY KEY ([id])
);

CREATE TABLE [s61].[table22] (
      [id] uniqueidentifier NOT NULL DEFAULT(newid())
    , [name] nvarchar(50) NOT NULL
    , CONSTRAINT [PK_table22] PRIMARY KEY ([id])
);

Столбец id имеет тип данных uniqueidentifier со значением по умолчанию newid().

В этом случае Microsoft SQL Server создает уникальные значения идентификатора для новых строк.

Ниже приведены команды для объединения данных из таблицы table21 в таблицу table22 и обратно:

gsqlcmd import db2 db1::s61.table21 /table=s61.table22 /merge

gsqlcmd import db1 db2::s61.table22 /table=s61.table21 /merge

Эта техника проста. Она не требует дополнительных таблиц или полей.

Однако она подходит для небольших и средних таблиц, так как все данные загружаются и обновляются каждый раз.

Использование столбцов Rowversion/Timestamp SQL Server для синхронизации новых и измененных строк

Microsoft SQL Server и Microsoft SQL Server Compact имеют встроенный тип данных rowversion с синонимом timestamp.

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

Сервер увеличивает значение на уровне сервера при каждой операции INSERT или UPDATE и обновляет значение поля.

Подробнее см. здесь:

• https://learn.microsoft.com/en-us/sql/t-sql/data-types/rowversion-transact-sql?view=sql-server-2017
• https://learn.microsoft.com/en-us/previous-versions/sql/compact/sql-server-compact-4.0/ms172424(v=sql.110)

Другие платформы баз данных позволяют реализовать функциональность версий строк с помощью триггеров.

Идея синхронизации проста:

• Импортируйте все строки один раз и сохраните максимальное значение версии строки.
• Синхронизируйте данные после последнего сохраненного значения rowversion и снова сохраните максимальное значение rowversion.

Вот объявление таблиц с полем rowversion (timestamp):

CREATE TABLE [s61].[table31] (
      [id] uniqueidentifier NOT NULL DEFAULT(newid())
    , [name] nvarchar(50) NOT NULL
    , [ts] timestamp NOT NULL
    , CONSTRAINT [PK_table31] PRIMARY KEY ([id])
);

CREATE TABLE [s61].[table32] (
      [id] uniqueidentifier NOT NULL DEFAULT(newid())
    , [name] nvarchar(50) NOT NULL
    , [ts] timestamp NOT NULL
    , CONSTRAINT [PK_table32] PRIMARY KEY ([id])
);

Ниже пример таблицы, которая содержит сохраненные метки времени:

CREATE TABLE [s61].[sync_timestamps] (
      [source] nvarchar(128) NOT NULL
    , [target] nvarchar(128) NOT NULL
    , [last_ts] binary(8) NOT NULL
    , [prev_ts] binary(8) NULL
    , CONSTRAINT [PK_sync_timestamps] PRIMARY KEY ([source], [target])
);

Благодаря столбцам source и target можно использовать всего одну таблицу для всех таблиц базы данных и пар синхронизации.

Ниже приведен пакетный файл для объединения новых и обновленных строк из таблицы table31 в таблицу table32:

gsqlcmd exec db1 update-sync-table31-to-table32.sql

if errorlevel 1 goto :EOF

gsqlcmd import db2 db1::select-table31-to-table32.sql /table=s61.table32 /merge

if errorlevel 1 goto :EOF

gsqlcmd exec db1 commit-sync-table31-to-table32.sql

Файл update-sync-table31-to-table32.sql содержит следующий код:

MERGE s61.sync_timestamps AS t
USING (SELECT 'table31' AS [source], 'table32' AS [target], COALESCE(MAX(ts), 0x) AS ts FROM s61.table31) AS s ([source], [target], ts)
ON (t.[source] = s.[source] AND t.[target] = s.[target])
WHEN MATCHED AND last_ts = prev_ts THEN
    UPDATE SET last_ts = s.ts
WHEN NOT MATCHED THEN
    INSERT ([source], [target], last_ts) VALUES (s.[source], s.[target], s.ts);

Команда merge вставляет строку или обновляет поле last_ts для синхронизируемой пары.

В таблице sync_timestamps будет содержать примерно такие значения:

Код обновляет поле last_ts, только если значения last_ts и prev_ts равны.

Это предотвратит обновление поля, если предыдущая синхронизация не удалась.

Для фиксации изменений пакет использует последнюю команду:

gsqlcmd exec db1 commit-sync-table31-to-table32.sql

Файл commit-sync-table31-to-table32.sql имеет следующий простой код:

UPDATE s61.sync_timestamps SET prev_ts = last_ts WHERE [source] = 'table31' AND [target] = 'table32'

Ниже приведен код файла select-table31-to-table32.sql, в котором используется выбор только новых и измененных строк:

SELECT
    id, name
FROM
    s61.table31
WHERE
    ts > COALESCE((
        SELECT
            prev_ts
        FROM
            s61.sync_timestamps
        WHERE
            [source] = 'table31' AND [target] = 'table32'
        ), 0x)

Обратите внимание, что он выбирает строки со значениями поля ts больше, чем значение поля prev_ts, зафиксированное при предыдущей синхронизации.

В этом разделе показано, как синхронизировать новые и измененные строки только с использованием столбцов с версиями строк.

Основным преимуществом этого способа является минимальное и безопасное изменение базовых таблиц. В SQL Server, в таблицы нужно добавить всего один столбец типа данных rowversion.

Кроме того, это единственный способ синхронизации данных с участием баз данных Microsoft SQL Server Compact, поскольку они не поддерживают триггеры, описанные ниже.

Использование столбцов последнего обновления для синхронизации новых и измененных строк

Другой способ выбрать новые и обновленные строки - использовать столбцы, содержащие время создания или последнего обновления.

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

Ниже приведен пример таблицы и ее триггеров для Microsoft SQL Server:

CREATE TABLE [s61].[table41] (
      [id] uniqueidentifier NOT NULL DEFAULT(newid())
    , [name] nvarchar(50) NOT NULL
    , [last_update] datetime NOT NULL DEFAULT(getutcdate())
    , CONSTRAINT [PK_table41] PRIMARY KEY ([id])
);

CREATE TRIGGER [s61].[trigger_table41_after_insert]
    ON [s61].[table41] AFTER INSERT
AS
BEGIN
SET NOCOUNT ON

UPDATE s61.table41
SET
    last_update = GETUTCDATE()
FROM
    s61.table41 t
    INNER JOIN inserted ON inserted.id = t.id
END;

CREATE TRIGGER [s61].[trigger_table41_after_update]
    ON [s61].[table41] AFTER UPDATE
AS
BEGIN
SET NOCOUNT ON

UPDATE s61.table41
SET
    last_update = GETUTCDATE()
FROM
    s61.table41 t
    INNER JOIN deleted ON deleted.id = t.id
END;

Чтобы сохранить последнее время синхронизированных данных для каждой пары синхронизации, можно использовать такую таблицу:

CREATE TABLE [s61].[sync_last_updates] (
      [source] nvarchar(128) NOT NULL
    , [target] nvarchar(128) NOT NULL
    , [last_update] datetime NOT NULL
    , [prev_update] datetime NULL
    , CONSTRAINT [PK_sync_last_updates] PRIMARY KEY ([source], [target])
);

Ниже приведен пакетный файл, используемый для синхронизации данных из таблицы table41 в db1 в таблицу table42 в db2:

gsqlcmd exec db1 update-sync-table41-to-table42.sql

if errorlevel 1 goto :EOF

gsqlcmd import db2 db1::select-table41-to-table42.sql /table=s61.table42 /merge

if errorlevel 1 goto :EOF

gsqlcmd exec db1 commit-sync-table41-to-table42.sql

Первая команда сохраняет значение last_update, которое будет зафиксировано в последней пакетной команде, если ошибок нет.

Файл update-sync-table41-to-table42.sql имеет следующий код:

MERGE s61.sync_last_updates AS t
USING (SELECT 'table41' AS [source], 'table42' AS [target], COALESCE(MAX(last_update), 0) AS last_update FROM s61.table41) AS s ([source], [target], last_update)
ON (t.[source] = s.[source] AND t.[target] = s.[target])
WHEN MATCHED AND t.last_update = t.prev_update THEN
    UPDATE SET last_update = s.last_update
WHEN NOT MATCHED THEN
    INSERT ([source], [target], last_update) VALUES (s.[source], s.[target], s.last_update);

Файл commit-sync-table41-to-table42.sql, использованный в последней пакетной команде, имеет следующий простой код:

UPDATE s61.sync_last_updates SET prev_update = last_update WHERE [source] = 'table41' AND [target] = 'table42'

Ниже приведен код файла select-table41-to-table42.sql, используемый для выборки строк, добавленных или обновленных после времени последней синхронизации, сохраненного в поле prev_update:

SELECT
    id, name
FROM
    s61.table41
WHERE
    last_update > COALESCE((
        SELECT
            prev_update
        FROM
            s61.sync_last_updates
        WHERE
            [source] = 'table41' AND [target] = 'table42'
        ), 0)

Полная синхронизация с удалением

В приведенных выше разделах обсуждается только добавление и обновление данных из исходных таблиц в целевые.

Такая синхронизация может решить ваши задачи полностью, если вы будете скрывать строки, а не удалять их.

Для репликации удаления строк необходимо иметь данные об удалении.

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

Ниже приведен пример таблицы, которая содержит удаленные идентификаторы:

CREATE TABLE [s61].[deleted_guids] (
      [id] uniqueidentifier NOT NULL
    , [source] nvarchar(128) NOT NULL
    , [last_update] datetime NOT NULL
    , CONSTRAINT [PK_deleted_guids] PRIMARY KEY ([id])
);

Поскольку идентификаторы GUID уникальны глобально, то таблица использует их в качестве первичного ключа.

Столбец source содержит таблицу удаленной строки.

Столбец last_update содержит время удаления, используемое для однократной синхронизации удаления.

Вот пример триггера:

CREATE TRIGGER [s61].[trigger_table51_after_delete]
    ON [s61].[table51] AFTER DELETE
AS
BEGIN
SET NOCOUNT ON

INSERT INTO deleted_guids (id, [source], last_update)
SELECT
    deleted.id
    , 'table51' AS [source]
    , GETUTCDATE() AS last_update
FROM
    deleted
END;

Ниже приведен модифицированный пакетный файл, который синхронизирует также удаление строк в целевой таблице:

gsqlcmd exec db1 update-sync-table51-to-table52.sql

if errorlevel 1 goto :EOF

gsqlcmd import db2 db1::select-table51-to-table52.sql /table=s61.table52 /merge

if errorlevel 1 goto :EOF

gsqlcmd import db2 db1::select-table51-to-table52-deleted.sql /table=s61.table52 /delete

if errorlevel 1 goto :EOF

gsqlcmd exec db1 commit-sync-table51-to-table52.sql

Пакет аналогичен рассмотренному выше. Однако он включает дополнительные действия для удаления.

Код update-sync-table51-to-table52.sql должен сохранять время last_update для обеих таблиц, включая deleted_guids:

MERGE s61.sync_last_updates AS t
USING (SELECT 'table51' AS [source], 'table52' AS [target], COALESCE(MAX(last_update), 0) AS last_update FROM s61.table51) AS s ([source], [target], last_update)
ON (t.[source] = s.[source] AND t.[target] = s.[target])
WHEN MATCHED AND t.last_update = t.prev_update THEN
    UPDATE SET last_update = s.last_update
WHEN NOT MATCHED THEN
    INSERT ([source], [target], last_update) VALUES (s.[source], s.[target], s.last_update);

MERGE s61.sync_last_updates AS t
USING (SELECT 'deleted_guids' AS [source], 'table52' AS [target], COALESCE(MAX(last_update), 0) AS last_update FROM s61.deleted_guids) AS s ([source], [target], last_update)
ON (t.[source] = s.[source] AND t.[target] = s.[target])
WHEN MATCHED AND t.last_update = t.prev_update THEN
    UPDATE SET last_update = s.last_update
WHEN NOT MATCHED THEN
    INSERT ([source], [target], last_update) VALUES (s.[source], s.[target], s.last_update);

Последняя команда также должна зафиксировать время последнего обновления для обеих таблиц.

Файл commit-sync-table51-to-table52.sql содержит следующий код:

UPDATE s61.sync_last_updates SET prev_update = last_update WHERE [source] = 'table51' AND [target] = 'table52';

UPDATE s61.sync_last_updates SET prev_update = last_update WHERE [source] = 'deleted_guids' AND [target] = 'table52';

Ниже приведен код select-table51-to-table52-deleted.sql, используемый для выборки идентификаторов, удаленных в таблице table51 после последней синхронизации:

SELECT
    id
FROM
    s61.deleted_guids
WHERE
    last_update > COALESCE((
        SELECT
            prev_update
        FROM
            s61.sync_last_updates
        WHERE
            [source] = 'deleted_guids' AND [target] = 'table52'
        ), 0)
    AND [source] = 'table51'

Пакетный файл использует этот файл в следующей команде:

gsqlcmd import db2 db1::select-table51-to-table52-deleted.sql /table=s61.table52 /delete

Эта команда использует опцию /delete вместо /merge.

Полная синхронизация между несколькими редактируемыми базами данных с целочисленными ключевыми столбцами

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

Например, у вас есть автономное настольное приложение для ваших продавцов, которое необходимо синхронизировать с центральной базой данных.

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

Возможным решением является добавление столбцов исходной таблицы и ее идентификаторов.

Например:

CREATE TABLE [s61].[table71] (
      [id] int IDENTITY(1,1) NOT NULL
    , [name] nvarchar(50) NOT NULL
    , [source_table] nvarchar(128) NULL
    , [source_row_id] int NULL
    , [last_update] datetime NULL
    , CONSTRAINT [PK_table71] PRIMARY KEY ([id])
);

Столбец source_table содержит имя таблицы, в которую изначально была добавлена ​​строка. Столбец source_row_id содержит идентификатор строки.

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

Триггер вставки должен автоматически обновлять эти поля. Вот пример:

CREATE TRIGGER [s61].[trigger_table71_after_insert]
    ON [s61].[table71] AFTER INSERT
AS
BEGIN
SET NOCOUNT ON

IF USER_NAME() IN ('sample61_user2')
    RETURN

UPDATE s61.table71
SET
    source_table = 'table71'
    , source_row_id = t.id
    , last_update = GETUTCDATE()
FROM
    s61.table71 t
    INNER JOIN inserted ON inserted.id = t.id
END;

Триггер обновляет столбец source_row_id начальным идентификатором, сгенерированным базой данных.

Чтобы предотвратить изменение значений source_table, source_row_id и last_update при синхронизации, триггер использует следующий код:

IF USER_NAME() IN ('sample61_user2')
    RETURN

В примере в операциях синхронизации используется имя пользователя sample61_user2. Таким образом, он импортирует данные как есть.

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

Триггер обновления также необходимо изменить, чтобы сохранить исходные значения синхронизации строки независимо от возможных изменений пользователем:

CREATE TRIGGER [s61].[trigger_table71_after_update]
    ON s61.table71 AFTER UPDATE
AS
BEGIN
SET NOCOUNT ON

IF @@NESTLEVEL > 1
    RETURN;

IF USER_NAME() IN ('sample61_user2')
    RETURN

UPDATE s61.table71
SET
    source_table = deleted.source_table
    , source_row_id = deleted.source_row_id
    , last_update = GETUTCDATE()
FROM
    s61.table71 t
    INNER JOIN deleted ON deleted.id = t.id
END;

Этот триггер также пропускает обновление полей в операциях синхронизации, выполняемых под именем пользователя sample61_user2, что позволяет обновлять значения столбца last_update импортированных строк.

Удаление строк также является определенной проблемой, поскольку нам нужно удалить записи в других базах данных, используя значения source_table и source_row_id.

Вот пример таблицы, в которой хранятся идентификаторы удаленных строк:

CREATE TABLE [s61].[deleted_ints] (
      [id] int NOT NULL
    , [source] nvarchar(128) NOT NULL
    , [source_table] nvarchar(128) NOT NULL
    , [source_row_id] int NOT NULL
    , [last_update] datetime NOT NULL
    , CONSTRAINT [PK_deleted_ints] PRIMARY KEY ([id], [source])
);

В таблице поля id и source используются в качестве первичного ключа для хранения значений для любой таблицы базы данных.

Кроме того, она содержит столбцы source_table и source_row_id, содержащие значения, используемые при синхронизации.

Столбец last_update содержит время удаления, используемое для однократной синхронизации удаления.

Триггер удаления имеет следующий код:

CREATE TRIGGER [s61].[trigger_table71_after_delete]
    ON s61.table71 AFTER DELETE
AS
BEGIN
SET NOCOUNT ON

INSERT INTO deleted_ints (id, [source], source_table, source_row_id, last_update)
SELECT
    deleted.id
    , 'table71' AS [source]
    , source_table
    , source_row_id
    , GETUTCDATE() AS last_update
FROM
    deleted
END;

Пакетный файл синхронизации имеет следующий код:

gsqlcmd exec db1 update-sync-table71-to-table72.sql

if errorlevel 1 goto :EOF

gsqlcmd import db2 db1::select-table71-to-table72.sql /table=s61.table72 /merge /keys=source_table,source_row_id

if errorlevel 1 goto :EOF

gsqlcmd import db2 db1::select-table71-to-table72-deleted.sql /table=s61.table72 /delete /keys=source_table,source_row_id

if errorlevel 1 goto :EOF

gsqlcmd exec db1 commit-sync-table71-to-table72.sql

Команды аналогичны тем, которые обсуждались в предыдущем разделе.

Однако команды import используют параметр /keys=source_table,source_row_id для использования столбцов source_table и source_row_id в условиях JOIN и WHERE.

Вы должны создать индексы для этих столбцов, чтобы повысить производительность.

Файлы update-sync-table71-to-table72.sql и commit-sync-table71-to-table72.sql аналогичны рассмотренным в предыдущем разделе, за исключением имен таблиц.

Файл select-table71-to-table72.sql похож на предыдущий, за исключением предложения SELECT:

SELECT
    name, source_table, source_row_id, last_update

Он не содержит столбца локального идентификатора, но содержит глобальные source_table и source_row_id.

Файл select-table71-to-table72-deleted.sql содержит измененные предложения SELECT и FROM:

SELECT
    source_table, source_row_id
FROM
    s61.deleted_ints
WHERE
    last_update > COALESCE((
        SELECT
            prev_update
        FROM
            s61.sync_last_updates
        WHERE
            [source] = 'deleted_ints' AND [target] = 'table72'
        ), 0)
    AND [source] = 'table71'

Улучшение и унификация решений

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

Это сделано в образовательных целях, чтобы шаг за шагом увеличивать сложность и показывать различия подходов.

Вы можете поместить код в хранимые процедуры и унифицировать пакетный файл синхронизации.

Например, пакетный файл может выглядеть так:

@echo off

set schema=s61
set source=%1
set target=%2

gsqlcmd exec   db1      "exec %schema%.usp_update_last_updates @source='%source%', @target='%target%'"

if errorlevel 1 goto :EOF

gsqlcmd import db2 "db1::exec %schema%.usp_select_upsert_rows  @source='%source%', @target='%target%'" /table=%schema%.%source% /merge  /keys=source_table,source_row_id

if errorlevel 1 goto :EOF

gsqlcmd import db2 "db1::exec %schema%.usp_select_delete_rows  @source='%source%', @target='%target%'" /table=%schema%.%source% /delete /keys=source_table,source_row_id

if errorlevel 1 goto :EOF

gsqlcmd exec   db1      "exec %schema%.usp_commit_last_updates @source='%source%', @target='%target%'"

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

Процедуры генерируют коды SQL для выбора фактических имен столбцов исходных таблиц.

Вы можете найти исходный код в папке "Examples\Sync Databases" загружаемого пакета gsqlcmd.

Заключение

Любой процесс синхронизации должен включать как минимум три шага:

• Получение данных из источника
• Создание команд SQL для изменения целевой таблицы
• Выполнение SQL-команд

gsqlcmd позволяет получать данные из баз данных, файлов и веб-ресурсов.

Кроме того, он позволяет создавать и выполнять команды для разных платформ баз данных, включая SQL Server, Oracle, DB2, MySQL, PostgreSQL, NuoDB, SQLite и SQLCE.

Таким образом, вы можете синхронизировать данные на любых поддерживаемых платформах базах данных.

Вы можете найти исходный код приведенных решений в папке "Examples\Sync Databases" загруженного пакета gsqlcmd.

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