gis Database

Parameters

Name	Type	Mode
replay_count	integer	OUT

Definition

DECLARE
    sqltemplate text;
    sqlsession text;
    sqlupdatelogs text;
    p_central_id text;
    p_clone_id text;
    p_sync_id uuid;
    rec record;
    p_counter integer;
    final_counter integer;
    p_temporary_table text;
    dblink_connection_name text;
    dblink_msg text;
    current_status boolean;
    event_ids_on_error bigint[];
BEGIN
    -- Default values
    final_counter = 0;
    current_status = True;
    event_ids_on_error = array[]::bigint[];

-- Get the total number of logs to replay
    SELECT count(*) AS nb
    FROM temp_clone_audit
    INTO p_counter;
    -- RAISE NOTICE 'p_counter %', p_counter;

-- If there are some logs, process them
    IF p_counter > 0 THEN

BEGIN
            -- Get central server id
            SELECT server_id::text INTO p_central_id
            FROM central_lizsync.server_metadata
            LIMIT 1;
            -- RAISE NOTICE 'central id %', p_central_id;

-- Get clone server id
            SELECT server_id::text INTO p_clone_id
            FROM lizsync.server_metadata
            LIMIT 1;
            -- RAISE NOTICE 'clone id %', p_clone_id;

EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'LizSync: % %', SQLSTATE, SQLERRM;
            -- set status
            current_status = False;
            RETURN QUERY SELECT -1::integer;
        END;

-- Create dblink connection
        dblink_connection_name = (md5(((random())::text || (clock_timestamp())::text)))::text;
        -- RAISE NOTICE 'dblink_connection_name %', dblink_connection_name;
        SELECT dblink_connect(
            dblink_connection_name,
            'central_server'
        )
        INTO dblink_msg;

-- Add a new item in the central history table
        -- p_server_from = clone
        -- p_server_to = central
        BEGIN
            sqltemplate = format(
                'SELECT lizsync.insert_history_item(
                    ''%1$s''::text, ''%2$s''::text,
                    NULL, NULL, NULL,
                    ''partial'', ''pending''
                ) AS sync_id
                ',
                p_clone_id,
                p_central_id
            );
            SELECT sync_id FROM dblink(
                dblink_connection_name,
                sqltemplate
            ) AS foo(sync_id uuid)
            INTO p_sync_id;
            -- RAISE NOTICE 'sync id %', p_sync_id;
        EXCEPTION WHEN OTHERS THEN
            RAISE NOTICE 'LizSync: % %', SQLSTATE, SQLERRM;
            -- set status
            current_status = False;
            RETURN QUERY SELECT -1::integer;
        END;

-- Replay SQL queries in central db
        -- The session variables are used by the central server audit function
        -- to fill the sync_data field
        sqlsession = format(
            'SET SESSION "lizsync.server_from" = ''%1$s''; SET SESSION "lizsync.server_to" = ''%2$s''; SET SESSION "lizsync.sync_id" = ''%3$s'';',
            p_clone_id,
            p_central_id,
            p_sync_id
        );

-- Loop through logs and replay action
        -- We need to query one by one to be able
        -- to update the sync_data->action_tstamp_tx afterwards
        -- by searching in the central lizsync.logged_actions the action = sqltemplate
        -- for this sync_id
        FOR rec IN
            SELECT *
            FROM temp_clone_audit
            ORDER BY tid
        LOOP
            -- Run the query in the central database via dblink
            -- We add a session variable with the action timestamp in the clone
            -- This will let us compare clone and central timestamp when UPDATE conflicts appear
            sqltemplate = concat(
                sqlsession,
                format(
                    'SET SESSION "lizsync.clone_action_tstamp_tx" = ''%1$s'';',
                    rec.action_tstamp_tx
                ),
                trim(rec.action) || ';'
            );

BEGIN
                SELECT dblink_exec(
                    dblink_connection_name,
                    sqltemplate
                )
                INTO dblink_msg;

-- increase counter
                final_counter = final_counter + 1;
            EXCEPTION WHEN OTHERS THEN
                RAISE NOTICE 'LizSync: % %', SQLSTATE, SQLERRM;
                -- set status
                current_status = False;

-- add event id to keep because on error
                event_ids_on_error = event_ids_on_error || rec.event_id;
                CONTINUE;
            END;

END LOOP;

-- Update central history table item
        sqltemplate = format(
            'SELECT lizsync.update_history_item(''%1$s''::uuid, ''done'', NULL)',
            p_sync_id
        )
        ;
        PERFORM * FROM dblink(
            dblink_connection_name,
            sqltemplate
        ) AS foo(update_status boolean)
        ;

-- Disconnect dblink
        SELECT dblink_disconnect(dblink_connection_name)
        INTO dblink_msg;

END IF;

-- Remove successfully replayed logs from clone audit table
    DELETE FROM lizsync.logged_actions
    WHERE NOT (event_id = ANY (event_ids_on_error))
    ;

-- Set sequence to the minimum required
    PERFORM setval(
        '"lizsync"."logged_actions_event_id_seq"',
        (SELECT max(event_id) FROM lizsync.logged_actions)
    );

-- Return
    RETURN QUERY SELECT final_counter;
END;

replay_clone_logs_to_central

Description

Parameters

Definition