Informix replication to SPL routine

CREATE TABLE state_log
(
    op_type     CHAR(1),
    source_id   INT,
    commit_time INT,
    txn_id      BIGINT,
    userid      INT, 
    session_id  INT,
    old_code    CHAR(2),
    old_sname   CHAR(15),
    new_code    CHAR(2),
    new_sname   CHAR(15)
);

CREATE PROCEDURE sp_state_log
(
    op_type     CHAR(1),
    source_id   INT,
    commit_time INT,
    txn_id      BIGINT,
    userid      INT, 
    session_id  INT,
    old_code    CHAR(2),
    old_sname   CHAR(15),
    new_code    CHAR(2),
    new_sname   CHAR(15)
)

INSERT INTO state_log VALUES
(
    op_type,
    source_id,
    commit_time,
    txn_id,
    userid, 
    session_id,
    old_code,
    old_sname,
    new_code,
    new_sname
);

END PROCEDURE;

cdr define replicate repl_state -C always -S row -A -R \
    --splname=sp_state_log \
    "P stores_demo@g_informix:informix.state" "select * from informix.state" \
    "R stores_demo@g_loopback:informix.state" "select * from informix.state"

cdr start replicate repl_state

INSERT INTO state VALUES (53, 'GB');
UPDATE state SET sname = 'UK' WHERE code = 53;
DELETE FROM state WHERE code = 53;
SELECT * FROM state_log;

CREATE TABLE data_change_log (data BSON);

CREATE INDEX data_change_log ON data_change_log
    (BSON_VALUE_BIGINT(data, 'commit_time')) USING BSON;

CREATE PROCEDURE sp_data_change_capture (data JSON)
    INSERT INTO data_change_log VALUES (data);
END PROCEDURE;

cdr define replicate repl_state -C always -S row -A -R \
    --jsonsplname=sp_data_change_capture \
    "P stores_demo@g_informix:informix.state" "select * from informix.state" \
    "R stores_demo@g_loopback:informix.state" "select * from informix.state"

cdr start replicate repl_state

SELECT REPLACE(data::JSON::LVARCHAR, ',"', ', "') AS data
FROM data_change_log;

CREATE FUNCTION sp_data_change_diff
(
    p_tabname       VARCHAR(128) DEFAULT '*',
    p_datetime_1    DATETIME YEAR TO SECOND DEFAULT NULL,
    p_datetime_2    DATETIME YEAR TO SECOND DEFAULT NULL
)
RETURNING
    DATETIME YEAR TO SECOND     AS date_time,
    VARCHAR(128)                AS table_name,
    VARCHAR(255)                AS key_value,
    VARCHAR(128)                AS column_name,
    VARCHAR(255)                AS value_before,
    VARCHAR(255)                AS value_after;
{
    Show columns updated in JSON loopback replication output
    See also https://en.wikipedia.org/wiki/Year_2038_problem
    Doug Lawry, Oninit Consulting, May 2021

    Examples usage:

    EXECUTE FUNCTION sp_data_change_diff();
    EXECUTE FUNCTION sp_data_change_diff('customer');
    EXECUTE FUNCTION sp_data_change_diff
    (
        '*',
        '2021-05-19 12:27:00',
        '2021-05-19 12:29:00'
    );
}
    DEFINE l_data       BSON;
    DEFINE l_tabname    VARCHAR(128);
    DEFINE l_utc_time   BIGINT;
    DEFINE l_datetime   DATETIME YEAR TO SECOND;
    DEFINE l_colno      SMALLINT;
    DEFINE l_colname    VARCHAR(128);
    DEFINE l_key_value  VARCHAR(255);
    DEFINE l_value_1    VARCHAR(255);
    DEFINE l_value_2    VARCHAR(255);

    FOREACH

        SELECT  data
        INTO    l_data
        FROM    data_change_log
        WHERE   BSON_VALUE_VARCHAR (data, 'table') MATCHES p_tabname
        AND     BSON_VALUE_VARCHAR (data, 'operation') = 'update'
        AND     BSON_VALUE_BIGINT  (data, 'commit_time')
        BETWEEN sp_datetime_to_utc (NVL(p_datetime_1, MDY(01,02,1970)))
        AND     sp_datetime_to_utc (NVL(p_datetime_2, MDY(01,18,2038)))

        LET l_tabname  = BSON_VALUE_VARCHAR (l_data, 'table');
        LET l_utc_time = BSON_VALUE_BIGINT  (l_data, 'commit_time');
        LET l_datetime = DBINFO ('UTC_TO_DATETIME', l_utc_time);

        FOREACH

            SELECT colno, colname
            INTO l_colno, l_colname
            FROM systables AS t
            JOIN syscolumns AS c ON c.tabid = t.tabid
            WHERE tabname = l_tabname
            ORDER BY colno

            LET l_value_1 = BSON_VALUE_VARCHAR (l_data, 'before_rowdata.' || l_colname);
            LET l_value_2 = BSON_VALUE_VARCHAR (l_data,        'rowdata.' || l_colname);

            IF l_colno = 1 THEN
                LET l_key_value = l_value_1;
            END IF

            IF NVL(l_value_1,'')
            != NVL(l_value_2,'')
            THEN
                RETURN
                    l_datetime,
                    l_tabname,
                    l_key_value,
                    l_colname,
                    l_value_1,
                    l_value_2
                WITH RESUME;
            END IF

        END FOREACH

    END FOREACH

END FUNCTION;

CREATE FUNCTION sp_datetime_to_utc
(
    date_time DATETIME YEAR TO SECOND
    DEFAULT   CURRENT  YEAR TO SECOND
)
RETURNING
    BIGINT AS utc_time;
{
    Inverse of DBINFO('utc_to_datetime',INT)
    Doug Lawry, Oninit Consulting, May 2021

    Check same result to test correctness:

        SELECT
            DBINFO('utc_current'),
            sp_datetime_to_utc()
        FROM systables
        WHERE tabid = 1;

    Maximum safe range with up to 24 hours TZ offset:

        '1970-01-02 00:00:00' -- zero is 1970-01-01
        '2038-01-18 00:00:00' -- Google "Year 2038"

    This function will cope after that using BIGINT,
    but DBINFO('utc_to_datetime',INT) will need fixing.
}
    DEFINE t INTERVAL DAY(7) TO SECOND;
    DEFINE d INTERVAL DAY(7) TO DAY;
    DEFINE s INTERVAL SECOND(7) TO SECOND;

    -- Time since UTC start
    LET t = date_time - DBINFO('utc_to_datetime', 0);

    -- Split days/seconds as max INTERVAL scale is 9:
    LET d = t;
    LET s = t - d;

    -- Recombine total seconds, allowed via CHAR:
    RETURN
        d :: CHAR(9) * 24 * 60 * 60 +
        s :: CHAR(9);

END FUNCTION;

$ cat repl_define.sh
[ $# -lt 2 ] && exec echo "Usage: $0 database table-names"

set -e # exit on error

DB=$1 ; shift   # database name
TO=informix     # table owner
G1=g_informix   # groups in sqlhosts
G2=g_loopback

for TN # table name
do

    RP=repl_$TN # replicate name
    T1=$TO.$TN  # owner + table names
    T2=$T1

    cdr define replicate $RP -C always -S row -A -R \
        --jsonsplname=sp_data_change_capture \
        "P $DB@$G1:$T1" "select * from $T1" \
        "R $DB@$G2:$T2" "select * from $T2"

    cdr start replicate $RP

done

$ repl_define.sh stores_demo customer stock
Interpreting this replicate as a master replicate.
Verification of stores_demo@g_informix:informix.customer started
Verification of stores_demo@g_informix:informix.customer is successful
Verification of stores_demo@g_loopback:informix.customer started
Verification of stores_demo@g_loopback:informix.customer is successful
Interpreting this replicate as a master replicate.
Verification of stores_demo@g_informix:informix.stock started
Verification of stores_demo@g_informix:informix.stock is successful
Verification of stores_demo@g_loopback:informix.stock started
Verification of stores_demo@g_loopback:informix.stock is successful

TRUNCATE data_change_log;

UPDATE  customer
SET     city = 'Sunny Vale'
WHERE   city = 'Sunnyvale';

UPDATE  customer
SET     city = 'Sunnyvale'
WHERE   city = 'Sunny Vale';

UPDATE  stock
SET     description = REPLACE(description, 'tires', 'tyres')
WHERE   description MATCHES '* tires';

UPDATE  stock
SET     description = REPLACE(description, 'tyres', 'tires')
WHERE   description MATCHES '* tyres';

EXECUTE FUNCTION sp_data_change_diff();

SELECT * FROM TABLE (FUNCTION sp_data_change_diff())
AS data (date_time, table_name, key_value, column_name, value_before, value_after)
WHERE table_name IN ('customer', 'stock');