Есть несколько вариантов захвата изменений из СУБД Oracle Database (далее — «Oracle»), самыми распространенными являются API LogMiner и API XStream. В данной статье рассмотрим API LogMiner.

В статье будет рассмотрено:

Немного теории
Откуда LogMiner получает изменения
System change number
Redo log
— Online redo log
— — Ротация online redo log
— Archive redo log
Dictionary

Без чего нельзя начинать работу с LogMiner
Инициализация LogMiner
Необходимые привилегии для работы с LogMiner
Минимальное дополнительное ведение журнала

Теория без практики мертва...
Практика
— Забор текущих изменений
— Простые DML операции
— Длинные операции
— Транзакции
— — Получения изменений зафиксированных транзакций
— — Получения изменений незафиксированных транзакций
— — Получения изменений транзакции которую откатили
— — Получения изменений транзакции которую частично откатили
— DML операции с LOB полями
— — Гарантия получения операции
— — Почему не всегда можно использовать SEQUENCE#
— Переименование полей таблиц
— Сортировка в V$LOGMNR_CONTENTS
— Организация непрерывного сбора изменений

Термины

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

Транзакция — набор операций для работы с базой данных.

Захват изменений — получение SQL команды, которая повлияла на изменение данных в базе данных, но которая имеют весь набор данных как до, так и после изменения (далее – «Изменения»).

Откуда LogMiner получает изменения

В Oracle реализован набор требований к транзакционным базам данных, известный как ACID. Одно из ключевых требований — это durability. Durability (надежность) — гарантирует, что если пользователь получил подтверждение от системы (транзакция выполнена), он может быть уверен, что сделанные им изменения не будут отменены из‑за какого‑либо сбоя.

Объясню этот процесс очень упрощенно. Чтобы гарантировать пользователю выполнение транзакции, необходимо в самом идеальном случае записать внесенные изменения в сам файл базы данных. Но это дорогое удовольствие в плане производительности. Самое простое — это записать файл в специальный промежуточный журнал (Redo log file) и уже позже, при выполнении события контрольной точки (checkpoint), за раз перенести грязные данные в файл базы данных.

LogMiner работает с redo log files минуя сами файлы базы данных.

System change number

Для работы с LogMiner важно понимать, что такое System Change Number (SCN) — это логическая временная метка, используемая Oracle для упорядочивания событий, происходящих с базой данных. Для понимания, лучше представить его как Unix Time Stamp в разрезе сервера Oracle.

Текущий SCN можно посмотреть следующей командой:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2000

Мы получили SCN для текущего времени, так же это работает в обратную сторону:

SELECT SCN_TO_TIMESTAMP(2000) AS TIMESTAMP_SCN FROM DUAL;

          TIMESTAMP_SCN
-----------------------
2024-01-01 00:00:00.000

Redo log

Redo log — это файлы, в которых хранятся все изменения, внесенные в базу данных по мере их возникновения. Существует два вида redo log:

Online redo log состоит из двух или более предварительно выделенных файлов, содержащие текущие изменения. Как раз с этим видом логов и работает СУБД для поддержания надежности;
Archive redo log — это те же файлы, но хранят в себе более ранние изменения.

Online redo log зачастую хранится на более быстрых носителях, но использование archive redo log меньше влияет на СУБД.

Так же стоит отметить, что у вас может быть совершенно другой набор online redo log файлов, и может не быть вовсе archive redo log файлов.

Online redo log

SELECT
  T.MEMBER AS FILE_NAME,
  T.FIRST_CHANGE AS FIRST_CHANGE,
  T.NEXT_CHANGE AS NEXT_CHANGE,
  T.STATUS
FROM (
  SELECT
    F.MEMBER,
    L.FIRST_CHANGE#,
    L.NEXT_CHANGE#,
    L.STATUS,
    ROW_NUMBER() OVER(PARTITION BY F.GROUP# ORDER BY F.GROUP#) AS GROUP
  FROM V$LOGFILE F
    JOIN V$LOG L
      ON L.GROUP# = F.GROUP#
  WHERE F.TYPE = 'ONLINE'
) T
WHERE T.GROUP = 1;

       FILE_NAME FIRST_CHANGE          NEXT_CHANGE   STATUS 
---------------- ------------ -------------------- --------
/path/redo01.log 1700         1800                 INACTIVE
/path/redo02.log 1800         1900                 INACTIVE
/path/redo03.log 1900         18446744073709551615 CURRENT

В данном случае выделено три online redo log файла. Один из них текущий, у которого STATUS CURRENT. Можно увидеть начало и конец по SCN (FIRST_CHANGE <= SCN < NEXT_CHANGE) в каждом из файлов, и как они чередуются между собой.

Ротация online redo log

Запись происходит по кругу. Как только текущий online redo log достаточно заполнится, его место займет самый ранний online redo log по FIRST_CHANGE. А так же, замененный online redo log файл будет архивирован, то есть появится точно такой же archive redo log файл. Примерно ротации:

       FILE_NAME FIRST_CHANGE          NEXT_CHANGE   STATUS 
---------------- ------------ -------------------- --------
/path/redo01.log 2200         18446744073709551615 CURRENT
/path/redo02.log 1800         1900                 INACTIVE
/path/redo03.log 1900         2200                 INACTIVE

Archive redo log

SELECT
  AL.NAME AS FILE_NAME,
  AL.FIRST_CHANGE# AS FIRST_CHANGE,
  AL.NEXT_CHANGE# AS NEXT_CHANGE,
  AL.DICTIONARY_BEGIN,
  AL.DICTIONARY_END
FROM V$ARCHIVED_LOG AL
  JOIN V$ARCHIVE_DEST_STATUS ADS
    ON ADS.DEST_ID = AL.DEST_ID
WHERE AL.STATUS = 'A'
  AND ADS.STATUS = 'VALID'
  AND ADS.TYPE = 'LOCAL';

                     FILE_NAME FIRST_CHANGE NEXT_CHANGE DICTIONARY_BEGIN DICTIONARY_END 
------------------------------ ------------ ----------- ---------------- --------------
/path/arch1_214_1161333401.dbf 1100         1200        NO               NO
/path/arch1_215_1161333401.dbf 1200         1300        YES              YES
/path/arch1_216_1161333401.dbf 1300         1400        NO               NO
/path/arch1_217_1161333401.dbf 1400         1500        NO               NO
/path/arch1_218_1161333401.dbf 1500         1600        YES              NO
/path/arch1_219_1161333401.dbf 1600         1700        NO               YES
/path/arch1_220_1161333401.dbf 1700         1800        NO               NO
/path/arch1_221_1161333401.dbf 1800         1900        NO               NO

Содержимое archive redo log файла такое же, как и в online redo log файла. Но archive redo log может содержать в себе частично или полностью dictionary (словарь), о котором мы поговорим позже.

Вернемся к ротации online redo log. После ротации появится новый archive redo log файл, пример:

                     FILE_NAME FIRST_CHANGE NEXT_CHANGE DICTIONARY_BEGIN DICTIONARY_END 
------------------------------ ------------ ----------- ---------------- --------------
/path/arch1_222_1161333401.dbf 1900         2200        NO               NO

Dictionary

LogMiner требует dictionary (словарь) для преобразования идентификаторов объектов в имена объектов при возврате вам изменений.

LogMiner предоставляет три варианта предоставления словаря:

Online Catalog рекомендуется использовать, если вы имеете доступ к исходной базе данных, из которой были созданы redo log файлы. А так же, если не будут внесены изменения в названия полей интересующих таблиц;
Словарь из archive redo log рекомендуется использовать, если вы не планируете иметь доступ к исходной базе данных, из которой были созданы redo log файлы. А так же, если вы предполагаете, что будут внесены изменения в названия полей интересующих таблиц;
Словарь в отдельном файле рассматриваться не будет.

О наличии словаря в том или ином archive redo log файле говорят нам поля DICTIONARY_BEGIN и DICTIONARY_END.

Пример, когда словарь из archive redo log файла будет отдан полностью:

                     FILE_NAME FIRST_CHANGE NEXT_CHANGE DICTIONARY_BEGIN DICTIONARY_END 
------------------------------ ------------ ----------- ---------------- --------------
/path/arch1_215_1161333401.dbf 1200         1300        YES              YES

                     FILE_NAME FIRST_CHANGE NEXT_CHANGE DICTIONARY_BEGIN DICTIONARY_END 
------------------------------ ------------ ----------- ---------------- --------------
/path/arch1_218_1161333401.dbf 1500         1600        YES              NO
/path/arch1_219_1161333401.dbf 1600         1700        NO               YES

Инициализация LogMiner

При инициализации LogMiner необходимо указать redo log файлы и опции. Желательно инициализировать LogMiner каждый раз при получении изменений.

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('NAME_REDO_LOG_FILE_1', DBMS_LOGMNR.NEW);
  [
    DBMS_LOGMNR.ADD_LOGFILE('NAME_REDO_LOG_FILE_2', DBMS_LOGMNR.ADDFILE);
    | DBMS_LOGMNR.ADD_LOGFILE('NAME_REDO_LOG_FILE_3', DBMS_LOGMNR.ADDFILE);
    | ...
  ]
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>SCN_НАЧАЛА_СЧИТЫВАНИЯ_ФАЙЛА,
    ENDSCN=>SCN_КОНЦА_СЧИТЫВАНИЯ_ФАЙЛА,
    OPTIONS=>
      [
        DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
        | DBMS_LOGMNR.DICT_FROM_REDO_LOGS
      ]
      + DBMS_LOGMNR.COMMITTED_DATA_ONLY
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
      + DBMS_LOGMNR.DDL_DICT_TRACKING
  );
END;

ADD_LOGFILE
- DBMS_LOGMNR.NEW – первый redo log файла
- DBMS_LOGMNR.ADDFILE – последующие redo log файлы
START_LOGMNR – команда инициализации LogMiner
- STARTSCN – SCN начала считывания изменений
- ENDSCN – SCN конца считывания изменений
- OPTIONS – опции инициализации LogMiner
  - DICT_FROM_ONLINE_CATALOG – будет использован словарь Online catalog
  - DICT_FROM_REDO_LOGS – словарь будет извлечен из archive redo log файлов
  - COMMITTED_DATA_ONLY – будут получены только зафиксированные транзакции
  - PRINT_PRETTY_SQL – оформление выводимых SQL запросов
  - NO_SQL_DELIMITER – убрать ; из конца SQL запроса
  - NO_ROWID_IN_STMT – убрать ROW_ID из SQL запроса
  - DDL_DICT_TRACKING – актуализировать словарь встречаемыми DDL изменениями (работает только с DICT_FROM_REDO_LOGS)

В начале необходимо указать redo log фалы в рамках которых мы хотим получать изменения FIRST_CHANGE <= SCN ИЗМЕНЕНИЯ < NEXT_CHANGE.

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

Необходимые привилегии для работы с LogMiner

LogMiner использует представления для работы с redo log файлами и их содержимым. В рамках данной статьи понадобится этот набор привилегий:

GRANT LOGMINING TO ...

GRANT EXECUTE ON DBMS_LOGMNR TO ...
GRANT EXECUTE ON DBMS_LOGMNR_D TO ...

GRANT SELECT ON V$LOG TO ...
GRANT SELECT ON V$LOGMNR_LOGS TO ....
GRANT SELECT ON V$LOGMNR_CONTENTS TO ...
GRANT SELECT ON V$LOGFILE TO ...
GRANT SELECT ON V$ARCHIVED_LOG TO ...
GRANT SELECT ON V$ARCHIVE_DEST_STATUS TO ...

Минимальное дополнительное ведение журнала

Необходимо включить минимальное дополнительное ведение журнала в разрезе базы данных.

SELECT SUPPLEMENTAL_LOG_DATA_MIN FROM V$DATABASE;

SUPPLEMENTAL_LOG_DATA_MIN
-------------------------
NO

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

SELECT SUPPLEMENTAL_LOG_DATA_MIN FROM V$DATABASE;

SUPPLEMENTAL_LOG_DATA_MIN
-------------------------
YES

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

ALTER TABLE ... ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

Практика

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

Советую повторять все примеры параллельно со статьей, только так будет ясно как все это работает.

Забор текущих изменений

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

CREATE TABLE SC.TEST (
  ID INTEGER,
  COL1 VARCHAR2(4000),
  COL2 INTEGER
);

ALTER TABLE SC.TEST ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

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

Далее запомним текущий SCN (CURRENT_SCN) и добавим запись в таблицу:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2001

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (0, 'TEXT', 1);
COMMIT;

Теперь получим список redo log файлов в которых SCN больше нашего текущего CURRENT_SCN, который мы получили ранее:

SELECT
  T.NAME AS FILE_NAME
FROM (
  SELECT
    F.MEMBER AS NAME,
    ROW_NUMBER() OVER(PARTITION BY F.GROUP# ORDER BY F.GROUP#) AS GROUP
  FROM V$LOGFILE F
    JOIN V$LOG L
      ON L.GROUP# = F.GROUP#
  WHERE F.TYPE = 'ONLINE'
) T
WHERE T.GROUP = 1
  AND 2001 >= T.FIRST_CHANGE;

       FILE_NAME FIRST_CHANGE          NEXT_CHANGE  STATUS 
---------------- ------------ -------------------- -------
/path/redo03.log 1900         18446744073709551615 CURRENT

В основном советую получать незафиксированные транзакции.

Инициализируем LogMiner с redo log файлом, который получили ранее и с STARTSCN, равным нашему CURRENT_SCN:

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('/path/redo03.log', DBMS_LOGMNR.NEW);
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>2001,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );
END;

Вызовем представление V$LOGMNR_CONTENTS и отфильтруем его под нашу таблицу:

SELECT
  SCN,
  OPERATION,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  ROW_ID
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

 SCN OPERATION SEG_OWNER TABLE_NAME                SQL_REDO             ROW_ID  
---- --------- --------- ---------- ----------------------- ------------------
2002 INSERT    SC        TEST       insert into "SC"."TEST" AAAUkcAAHAAAAHeAAA
                                     values
                                        "ID" = 0,
                                        "COL1" = 'TEXT',
                                        "COL2" = 1

Советую фильтровать представление V$LOGMNR_CONTENTS по интересующим нам таблицам или транзакциям.

Поздравляю, мы получили первые изменения)

Алгоритм получения текущих изменений такой:

Получение текущего SCN.
Совершить DML операцию.
Получить redo log файлы покрывающие текущий SCN.
Инициализировать LogMiner.
Вызвать представление V$LOGMNR_CONTENTS.

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

Простые DML операции

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

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2005

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (1, 'IVAN', 18);
UPDATE SC.TEST SET COL1 = 'IVAN2' WHERE ID = 1;
DELETE FROM SC.TEST WHERE ID = 1;
COMMIT;

Получим redo log файлы и инициализируем LogMiner. Воспользуемся представлением V$LOGMNR_CONTENTS для получение изменений, и отфильтруем данные под нужную нам таблицу:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

 SCN COMMIT_SCN COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                 SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ---------------- -------- --------- ---- --------- ---------- ------------------------ --- ------------------ ------ ------ ------
2006 NULL       NULL             0        INSERT    NULL SC        TEST       insert into "SC"."TEST"  0   AAAUkcAAHAAAAHeAAA 10     13     14218
                                                                               values
                                                                                  "ID" = 1,
                                                                                  "COL1" = 'IVAN',
                                                                                  "COL2" = 18
2006 NULL       NULL             0        UPDATE    NULL SC        TEST       update "SC"."TEST"       0   AAAUkcAAHAAAAHeAAA 10     13     14218
                                                                                set
                                                                                  "COL1" = 'IVAN2'
                                                                                where 
                                                                                  "ID" = 1 and 
                                                                                  "COL1" = 'IVAN' and
                                                                                  "COL2" = 18
2006 NULL       NULL             0        DELETE    NULL SC        TEST       delete from "SC"."TEST"  0   AAAUkcAAHAAAAHeAAA 10     13     14218
                                                                               where
                                                                                  "ID" = 1 and 
                                                                                  "COL1" = 'IVAN2' and
                                                                                  "COL2" = 18

Вот так выглядит каждая из DML операций как изменение. Вне зависимости от DML операции, у них всегда будет полный набор полей, кроме LOB полей (об этом будет позже). Так же, у этих изменений валидный ROW_ID и каждая из них считается законченной. Пройдемся немного по полям представления V$LOGMNR_CONTENTS:

SCN — метка времени совершения операции;
COMMIT_SCN — метку времени совершения фиксации транзакции. У DML изменений она будет только, если инициализация LogMiner будет выполнена с опцией COMMITTED_DATA_ONLY, а так будет встречаться только у операции COMMIT;
COMMIT_TIMESTAMP — выводит дату и время совершения фиксации транзакции. Условия появления информации у данного поля такие же, как и у поля COMMIT_SCN;
ROLLBACK — флаг отмены DML операции, если значение поля 1, то данная DML операция случит для отмены ранее совершенной DML операции;
OPERATION — тип операции (будут встречаться уже знакомые DML операции, так и COMMIT, ROLLBACK, INTERNAL (системная операция СУБД));
INFO — в основном поле содержит информацию по отклонению работы LogMiner, к примеру все ли поля были правильно интерпретированы словарем и т.д;
SEG_OWNER — схема таблицы;
TABLE_NAME — имя таблицы;
SQL_REDO — SQL запрос операции;
CSF — флаг законченности операции в поле SQL_REDO. Если операция длинная и не помещается в SQL_REDO, то значение поля 1 говорит о том, что операция будет продолжена в следующей записи в поле SQL_REDO;
ROW_ID — это уникальный адрес физического нахождения строки в файле, полезно для идентификации операции, которую нужно откатить;
XIDUSN, XIDSLT, XIDSQN — составной идентификатор транзакции.

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

Валидным (VALID) ROW_ID я буду называть в том случае, если оно однозначно идентифицирует строку в базе данных.

Шаблон простых операций:

OPERATION        SQL_REDO             ROW_ID 
--------- --------------- ------------------
INSERT    insert into ... VALID

UPDATE    update ...      VALID

DELETE    delete from ... VALID

Длинные операции

Длинные операции — это DML операции, которые не помещаются в рамках одной записи в поле SQL_REDO представления V$LOGMNR_CONTENTS и помечаются CSF равного 1.

Добавим такую запись в таблицу:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2010

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (0, DBMS_RANDOM.STRING('U', 4000), 20);
COMMIT;

Как и в предыдущих примерах, получим redo log файлы и инициализируем LogMiner. Теперь получим изменения:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

 SCN COMMIT_SCN COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ---------------- -------- --------- ---- --------- ---------- ----------------------- --- ------------------ ------ ------ ------
2012 NULL       NULL             0        INSERT    NULL SC        TEST       insert into "SC"."TEST" 1   AAAUkcAAHAAAAHeAAA 4      20     2108
                                                                               values
                                                                                  "ID" = 0,
                                                                                  "COL1" = 'ASDSFG...
2012 NULL       NULL             0        INSERT    NULL SC        TEST       ...ASDFKJBZDKSDFSFFKG', 0   AAAUkcAAHAAAAHeAAA 4      20     2108
                                                                                  "COL2" = 20

В примере видно, что поле CSF равно 1, это говорит нам о том, что операция не поместилась в поле SQL_REDO. В таком случае необходимо конкатенировать записи по полю SQL_REDO пока поле CSF не будет равным 0.

Шаблон длинной операций:

OPERATION        SQL_REDO CSF             ROW_ID 
--------- --------------- --- ------------------
INSERT    insert into ... 1   VALID
INSERT    ...             1   VALID
INSERT    ...             0   VALID

UPDATE    update ...      1   VALID
UPDATE    ...             0   VALID

DELETE    delete from ... 1   VALID
DELETE    ...             0   VALID

Транзакции

В данной главе будет показано как LogMiner получает изменения в разрезе транзакций.

Получения изменений зафиксированных транзакций

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

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('/path/redo03.log', DBMS_LOGMNR.NEW);
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>2020,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
      + DBMS_LOGMNR.COMMITTED_DATA_ONLY
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );
END;

В данном примере будем работать в двух сессиях. В первой сессии выполним следующий код и данную транзакцию не зафиксируем (без COMMIT):

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2020

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (1, 'PAVEL', 20);
UPDATE SC.TEST SET COL1 = 'PAVEL 2' WHERE ID = 1;

Во второй сессии выполним следующий код, но транзакцию зафиксируем (с COMMIT):

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (2, 'DMITRY', 25);
COMMIT;

Получим текущие изменения:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------- ---- --------- ---------- ----------------------- --- ------------------ ------ ------ ------
2024 2026       2024-01-01 00:00:10.000 0        INSERT    NULL SC        TEST       insert into "SC"."TEST" 0   AAAUkcAAHAAAAHeAAB 10     13     14218
                                                                                      values
                                                                                         "ID" = 2,
                                                                                         "COL1" = 'DMITRY',
                                                                                         "COL2" = 25

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

COMMIT;

И заново вызовем представление V$LOGMNR_CONTENTS:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                 SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------- ---- --------- ---------- ------------------------ --- ------------------ ------ ------ ------
2022 2028       2024-01-01 00:00:12.000 0        INSERT    NULL SC        TEST       insert into "SC"."TEST"  0   AAAUkcAAHAAAAHeAAA 8      15     1263
                                                                                      values
                                                                                         "ID" = 1,
                                                                                         "COL1" = 'PAVEL',
                                                                                         "COL2" = 20
2023 2028       2024-01-01 00:00:12.000 0        UPDATE    NULL SC        TEST       update "SC"."TEST"       0   AAAUkcAAHAAAAHeAAA 8      15     1263
                                                                                       set
                                                                                         "COL1" = 'PAVEL 2'
                                                                                       where 
                                                                                         "ID" = 1 and 
                                                                                         "COL1" = 'PAVEL' and
                                                                                         "COL2" = 20
2024 2026       2024-01-01 00:00:10.000 0        INSERT    NULL SC        TEST       insert into "SC"."TEST"  0   AAAUkcAAHAAAAHeAAB 10     13     14218
                                                                                      values
                                                                                         "ID" = 2,
                                                                                         "COL1" = 'DMITRY',
                                                                                         "COL2" = 25

Теперь есть все изменения из двух сессий. Стоит обратить внимание на сортировку по SCN и по COMMIT_SCN.

Допустим, мы хотим прочитать только операцию UPDATE из первой сессии. Инициализируем LogMiner следующим образом:

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('/path/redo03.log', DBMS_LOGMNR.NEW);
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>2023,
    ENDSCN=>2028,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
      + DBMS_LOGMNR.COMMITTED_DATA_ONLY
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );
END;

В данном случае мы не получим ничего:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------- ---- --------- ---------- ----------------------- --- ------------------ ------ ------ ------

В режиме COMMITTED_DATA_ONLY транзакцию можно получить только атомарно, то есть полностью.

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

Сложность в отслеживании долгих транзакций;
Чтение транзакций атомарно (необходимо знать начало и конец транзакции).

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

Получения изменений незафиксированных транзакций

Добавим две записи и пока что не зафиксируем транзакцию:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2030

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (3, 'FEDOR', 20);
INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (4, 'ALEXANDER', 30);

Получим изменения:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                         SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------- ---- --------- ---------- -------------------------------- --- ------------------ ------ ------ ------
2032 NULL       NULL                    0        START     NULL NULL      NULL       set transaction read write       0   AAAUomAAAAAAAAAAAA 10     27     14783
2032 NULL       NULL                    0        INSERT    NULL SC        TEST       insert into "SC"."TEST"          0   AAAUomAAHAAAAWvAAA 10     27     14783
                                                                                       values
                                                                                          "ID" = 3,
                                                                                          "COL1" = 'FEDOR',
                                                                                          "COL2" = 20
2032 NULL       NULL                    0        INSERT    NULL SC        TEST       insert into "SC"."TEST"          0   AAAUomAAHAAAAWvAAB 10     27     14783
                                                                                       values
                                                                                          "ID" = 4,
                                                                                          "COL1" = 'ALEXANDER',
                                                                                          "COL2" = 30

Мы получили грязные данные. После того, как зафиксируем транзакцию появится операция COMMIT:

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                         SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------- ---- --------- ---------- -------------------------------- --- ------------------ ------ ------ ------
2038 2038       2024-01-01 00:00:20.000 0        COMMIT    NULL NULL      NULL       commit                           0   AAAAAAAAAAAAAAAAAA 10     27     14783

Получения изменений транзакции которую откатили

Добавим запись и откатим транзакцию:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2040

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (5, 'ALEKSEY', 20);
ROLLBACK;

Получим идентификатор транзакции:

SELECT
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

XIDUSN XIDSLT XIDSQN 
------ ------ ------
10     13     14229

Получим изменения по транзакции:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE XIDUSN = 10
  AND XIDSLT = 13
  AND XIDSQN = 14229;

 SCN COMMIT_SCN COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                         SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ---------------- -------- --------- ---- --------- ---------- -------------------------------- --- ------------------ ------ ------ ------
2042 NULL       NULL             0        START     NULL NULL      NULL       set transaction read write       0   AAAUkcAAAAAAAAAAAA 10     13     14229
2042 NULL       NULL             0        INSERT    NULL SC        TEST       insert into "SC"."TEST"          0   AAAUkcAAHAAAAHdAAA 10     13     14229
                                                                               values
                                                                                  "ID" = 5,
                                                                                  "COL1" = 'ALEKSEY',
                                                                                  "COL2" = 20
2045 NULL      NULL              1        DELETE    NULL SC        TEST       delete from "SC"."TEST"
                                                                               where
                                                                                  ROWID = 'AAAUkcAAHAAAAHdAAA' 0   AAAUkcAAHAAAAHdAAA 10     13     14229
2046 NULL      NULL              0        ROLLBACK NULL  NULL      NULL       rollback                         0   AAAAAAAAAAAAAAAAAA 10     13     14229

Шаблон транзакции которую откатили:

ROLLBACK OPERATION                   SQL_REDO 
-------- --------- --------------------------
0        START     set transaction read write
0        INSERT    insert ...
0        INSERT    insert ...
...
1        DELETE    delete ...
1        DELETE    delete ...
...
         ROLLBACK  rollback

Получения изменений транзакции которую частично откатили

Надеюсь вы знакомы с savepoint, она же «‎точка сохранения». Выполним следующие команды:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2050

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (6, 'ILYA', 20);
SAVEPOINT S1;
INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (7, 'LEV', 30);
ROLLBACK TO SAVEPOINT S1;
INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (8, 'ARTHUR', 40);
COMMIT;

В даном примере мы откатили нашу транзакцию до точки сохранения и зафиксировали транзакцию. Получим идентификатор транзакции:

SELECT
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST';

XIDUSN XIDSLT XIDSQN 
------ ------ ------
3      16     2004

Теперь посмотрим, как это выглядит в разрезе транзакции:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE XIDUSN = 3
  AND XIDSLT = 16
  AND XIDSQN = 2004;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                         SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------- ---- --------- ---------- -------------------------------- --- ------------------ ------ ------ ------
2052 NULL       NULL                    0        START     NULL NULL      NULL       set transaction read write       0   AAAUkcAAAAAAAAAAAA 3      16     2004
2052 NULL       NULL                    0        INSERT    NULL SC        TEST       insert into "SC"."TEST"          0   AAAUkcAAHAAAAHfAAA 3      16     2004
                                                                                      values
                                                                                         "ID" = 6,
                                                                                         "COL1" = 'ILYA',
                                                                                         "COL2" = 20
2052 NULL       NULL                    0        INSERT    NULL SC        TEST       insert into "SC"."TEST"          0   AAAUkcAAHAAAAHfAAB 3      16     2004
                                                                                      values
                                                                                         "ID" = 7,
                                                                                         "COL1" = 'LEV',
                                                                                         "COL2" = 30
2052 NULL       NULL                    1        DELETE    NULL SC        TEST       delete from "SC"."TEST"          0   AAAUkcAAHAAAAHfAAB 3      16     2004
                                                                                      where
                                                                                         ROWID = 'AAAUkcAAHAAAAHfAAB'
2052 NULL       NULL                    0        INSERT    NULL SC        TEST       insert into "C##DBO"."TEST"      0   AAAUkcAAHAAAAHfAAB 3      16     2004
                                                                                      values
                                                                                         "ID" = 8,
                                                                                         "COL1" = 'ARTHUR',
                                                                                         "COL2" = 40
2054 2054       2024-01-01 00:00:30.000 0        COMMIT    NULL NULL       NULL      commit                           0   AAAAAAAAAAAAAAAAAA

Обратите внимание на поле ROW_ID операции, которую откатили и на ROW_ID операции, которая идет далее. Операция вставки новой записи заняла место записи операции, которую откатили. Это важный нюанс идентификатора записи ROW_ID, который следует из его определения.

Попробуем выразить все это в шаблоне:

OPERATION ROLLBACK                         SQL_REDO CSF             ROW_ID 
--------- -------- -------------------------------- --- ------------------
START     0        set transaction read write       0   

INSERT    0        insert into ...                  0   

INSERT    0        insert into ...                  0   
DELETE    1        delete from ...                  0   

INSERT    0        insert into ...                  0   PREVIOUS ROW_ID

COMMIT    0        commit                           0

DML операции с LOB полями

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

CREATE TABLE SC.TEST_LOB (
  ID INTEGER,
  COL1 VARCHAR2(4000),
  COL2 INTEGER,
  TEXT1 CLOB
);

ALTER TABLE SC.TEST_LOB ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

Добавим запись с небольшим LOB контентом и обновим ее:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2060

INSERT INTO SC.TEST_LOB (ID, COL1, COL2, TEXT1) VALUES (1, 'TEXT', 10, DBMS_RANDOM.STRING('U', 10));
UPDATE SC.TEST_LOB SET COL1 = 'TEXT_UPDATE' WHERE ID = 1;

COMMIT;

Теперь получим redo log файлы в которых есть изменения по нашему SCN, инициализируем LogMiner и получим идентификатор транзакции которую мы провели:

SELECT DISTINCT
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST_LOB';

XIDUSN XIDSLT XIDSQN 
------ ------ ------
9      10     2331

И получим изменения данной транзакции:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE XIDUSN = 9
  AND XIDSLT = 10
  AND XIDSQN = 2331;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK OPERATION INFO SEG_OWNER TABLE_NAME                    SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------- ---- --------- ---------- --------------------------- --- ------------------ ------ ------ ------
2062 NULL       NULL                    0        START     NULL NULL      NULL       set transaction read write  0   AAAAAAAAAAAAAAAAAA 9      10     2331
2062 NULL       NULL                    0        INSERT    NULL SC        TEST_LOB   insert into "SC"."TEST_LOB" 0   AAAAAAAAAAAAAAAAAA 9      10     2331
                                                                                      values
                                                                                         "ID" = 1,
                                                                                         "COL1" = 'TEXT',
                                                                                         "COL2" = 10,
                                                                                         "TEXT1" = EMPTY_CLOB() 
2064 NULL       NULL                    0        UPDATE    NULL SC        TEST_LOB   update "SC"."TEST_LOB"      0   AAAUoIAAHAAAMl/AAB 9      10     2331
                                                                                       set
                                                                                         "TEXT1" = 'ENWNSJMDJG'
                                                                                       where 
                                                                                         "ID" = 1 and 
                                                                                         "COL1" = 'TEXT' and
                                                                                         "COL2" = 10
2066 NULL       NULL                    0        UPDATE    NULL SC        TEST_LOB   update "SC"."TEST_LOB"      0   AAAUoIAAHAAAMl/AAB 9      10     2331
                                                                                      set
                                                                                        "TEXT1" = 'TEXT_UPDATE'
                                                                                      where 
                                                                                        "ID" = 1 and 
                                                                                        "COL1" = 'TEXT' and
                                                                                        "COL2" = 10
2068 2068       2024-01-01 00:00:40.000 0        COMMIT    NULL NULL      NULL       commit                      0   AAAAAAAAAAAAAAAAAA 9      10     2331

Как видно, INSERT не имеет валидный ROW_ID и содержит "TEXT1" = EMPTY_CLOB(), который указывает на LOB поле. Далее идет UPDATE — это UPDATE LOB поля. Как видно, у операции INSERT ROW_ID не валидный (как было в простых примерах), а уже у UPDATE LOB поля ROW_ID валидный и принадлежит операции INSERT.

Мы не будем видеть в будущих изменениях LOB поле, пока оно само не будет изменено.

Теперь, по аналогии с предыдущим примером, необходимо добавить запись, но теперь с большим LOB контентом:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2070

INSERT INTO SC.TEST_LOB (ID, COL1, COL2, TEXT1) VALUES (2, 'TEXT', 20, DBMS_RANDOM.STRING('U', 20000));
COMMIT;

Получим идентификатор транзакции, которую мы провели и посмотрим на нее:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE XIDUSN = 10
  AND XIDSLT = 3
  AND XIDSQN = 14629;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK       OPERATION   INFO SEG_OWNER TABLE_NAME                    SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------------- ------ --------- ---------- --------------------------- --- ------------------ ------ ------ ------
2072 NULL       NULL                    0        START           NULL   NULL      NULL       set transaction read write  0   AAAAAAAAAAAAAAAAAA 10     3      14629
2072 NULL       NULL                    0        INSERT          NULL   SC        TEST_LOB   insert into "SC"."TEST_LOB" 0   AAAAAAAAAAAAAAAAAA 10     3      14629
                                                                                               values
                                                                                                  "ID" = 2,
                                                                                                  "COL1" = 'TEXT',
                                                                                                  "COL2" = 20,
                                                                                                  "TEXT1" = EMPTY_CLOB()
2074 NULL       NULL                    0        INTERNAL        NULL   NULL      NULL       NULL                        0   AAAUoJAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        SEL_LOB_LOCATOR LOB... SC        TEST_LOB   DECLARE ...                 0   AAAAAAAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        INTERNAL        NULL   SC        TEST_LOB   NULL                        0   AAAAAAAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        LOB_WRITE       LOB... SC        TEST_LOB   buf_c := 'ODWCBXRAIEGQQW... 0   AAAAAAAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        LOB_WRITE       LOB... SC        TEST_LOB   buf_c := 'NHJGNONUXGBBBS... 0   AAAAAAAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        INTERNAL        NULL   SC        TEST_LOB   NULL                        0   AAAAAAAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        INTERNAL        NULL   SC        TEST_LOB   END;                        0   AAAAAAAAAAAAAAAAAA 10     3      14629
2076 NULL       NULL                    0        INTERNAL        NULL   NULL      NULL       NULL                        0   AAAUoIAAHAAAMl/AAC 10     3      14629
2078 2078       2024-01-01 00:00:50.000 0        COMMIT          NULL   NULL      NULL       commit                      0   AAAAAAAAAAAAAAAAAA 10     3      14629

Как видно, INSERT так же не имеет валидный ROW_ID и содержит "TEXT1" = EMPTY_CLOB(), который указывает на LOB поле. Далее идут операции SEL_LOB_LOCATOR и LOB_WRITE:

SEL_LOB_LOCATOR — операция, содержащая идентификатор LOB поля, который будет выводиться в следующим LOB_WRITE.
LOB_WRITE — операция, в которой находится содержимое LOB поля.

Валидный ROW_ID содержится в последней операции INTERNAL. Тут можно сказать, что операция длилась от SCN 2072 до SCN 2076 (далее я обьясню зачем это знать).

В одном случае операция с LOB полями заканчивается на UPDATE, а в другом на INTERNAL.

Так же в одной таблице может быть несколько LOB полей, вести они себя будет аналогично.

Шаблон операции с LOB полями:

      OPERATION               SQL_REDO       ROW_ID
--------------- ---------------------- ------------
INSERT          insert into ...        NO VALID
UPDATE          update ...             VALID

INSERT          insert into ...        NO VALID
SEL_LOB_LOCATOR ...                    NO VALID
LOB_WRITE       ...                    NO VALID
INTERNAL        ...                    VALID

INSERT          insert into ...        NO VALID
UPDATE          update ...             NO VALID
SEL_LOB_LOCATOR ...                    NO VALID
LOB_WRITE       ...                    NO VALID
INTERNAL        ...                    VALID

INSERT          insert into ...        NO VALID
SEL_LOB_LOCATOR ...                    NO VALID
LOB_WRITE       ...                    NO VALID
UPDATE          update ...             VALID

Гарантия получения операции

Из прошлого примера видно, что операция началась с SCN 2072 и длилась до SCN 2076. А что если инициализировать LogMiner с STARTSCN 2076:

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('/path/redo03.log', DBMS_LOGMNR.NEW);
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>2076,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );
END;

Получим изменения:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE XIDUSN = 10
  AND XIDSLT = 3
  AND XIDSQN = 14629;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK       OPERATION   INFO SEG_OWNER TABLE_NAME                    SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------------- ------ --------- ---------- --------------------------- --- ------------------ ------ ------ ------
2076 NULL       NULL                    0        INSERT          NULL   SC        TEST_LOB insert into "SC"."TEST_LOB"   0   AAAUoIAAHAAAMl/AAC 10     3      14629
                                                                                            values
                                                                                               "ID" = 2,
                                                                                               "COL1" = 'TEXT',
                                                                                               "COL2" = 20,
                                                                                               "TEXT1" = EMPTY_CLOB()

Может показаться странным, что на месте операции INTERNAL (из прошлого примера) мы получили INSERT, где есть LOB поле. И самое неожиданное, что получили валидный ROW_ID.

Странного ничего нет, мы начали считывать LogMiner с середины данной операции. Он обязан так или иначе отдать изменение. INTERNAL только указывал на физическое расположение записи, в данном случае он всегда будет заменен на DML операцию.

Но есть нюанс: можно получить дубль, если считывать LogMiner последовательно:

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('/path/redo03.log', DBMS_LOGMNR.NEW);
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>2072,
    ENDSCN=>2074,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );
END;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK       OPERATION   INFO SEG_OWNER TABLE_NAME                    SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------------- ------ --------- ---------- --------------------------- --- ------------------ ------ ------ ------
2072 NULL       NULL                    0        START           NULL   NULL      NULL       set transaction read write  0   AAAAAAAAAAAAAAAAAA 10     3      14629
2072 NULL       NULL                    0        INSERT          NULL   SC        TEST_LOB   insert into "SC"."TEST_LOB" 0   AAAAAAAAAAAAAAAAAA 10     3      14629
                                                                                               values
                                                                                                  "ID" = 2,
                                                                                                  "COL1" = 'TEXT',
                                                                                                  "COL2" = 20,
                                                                                                  "TEXT1" = EMPTY_CLOB()
2074 NULL       NULL                    0        INTERNAL        NULL   NULL      NULL       NULL                        0   AAAUoJAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        SEL_LOB_LOCATOR LOB... SC        TEST_LOB   DECLARE ...                 0   AAAAAAAAAAAAAAAAAA 10     3      14629
2074 NULL       NULL                    0        INTERNAL        NULL   SC        TEST_LOB   NULL                        0   AAAAAAAAAAAAAAAAAA 10     3      14629

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('/path/redo03.log', DBMS_LOGMNR.NEW);
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>2076,
    ENDSCN=>2076,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );
END;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK       OPERATION   INFO SEG_OWNER TABLE_NAME                    SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------------- ------ --------- ---------- --------------------------- --- ------------------ ------ ------ ------
2076 NULL       NULL                    0        INSERT          NULL   SC        TEST_LOB insert into "SC"."TEST_LOB"   0   AAAUoIAAHAAAMl/AAC 10     3      14629
                                                                                            values
                                                                                               "ID" = 2,
                                                                                               "COL1" = 'TEXT',
                                                                                               "COL2" = 20,
                                                                                               "TEXT1" = EMPTY_CLOB()

В этом случае вы потеряете содержимое LOB поля, но главное получите саму запись, даже два раза).

Не нужно считывать сами LOB поля грязно (без опции COMMITTED_DATA_ONLY). Необходимо собирать информацию о записях (SCN начала транзакции и COMMIT), где были изменены LOB поля, и параллельным процессом забирать изменения у LOB полей.

Почему не всегда можно использовать SEQUENCE#

Поле SEQUENCE# из представления V$LOGMNR_CONTENTS указывает на порядок операций в транзакции. Но это работает только при инициализации LogMier с опцией COMMITED_DATA_ONLY. Некоторые отсеивают UPDATE LOB поля по SEQUENCE# не равным 1 (делать этого не стоит, сейчас объясню почему).

Добавим запись в таблицу с одним LOB полем:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2080

INSERT INTO SC.TEST_LOB (ID, COL1, COL2, TEXT1) VALUES (3, 'TEXT', 20, DBMS_RANDOM.STRING('U', 20));
COMMIT;

Получим изменения в разрезе транзакции:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SEQUENCE#,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE XIDUSN = 10
  AND XIDSLT = 22
  AND XIDSQN = 14776;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK       OPERATION   INFO SEG_OWNER TABLE_NAME SEQUENCE#                    SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------------- ------ --------- ---------- --------- --------------------------- --- ------------------ ------ ------ ------
2082 NULL       NULL                    0        START           NULL   NULL      NULL       1         set transaction read write  0   AAAAAAAAAAAAAAAAAA 10     22     14776
2082 NULL       NULL                    0        INSERT          NULL   SC        TEST_LOB   1         insert into "SC"."TEST_LOB" 0   AAAAAAAAAAAAAAAAAA 10     22     14776
                                                                                                        values
                                                                                                           "ID" = 3,
                                                                                                           "COL1" = 'TEXT',
                                                                                                           "COL2" = 20,
                                                                                                           "TEXT1" = EMPTY_CLOB()
2083 NULL       NULL                    0        UPDATE          NULL   SC        TEST_LOB   2         update "C##DBO"."TEST_LOB"  0   AAAUoyAAHAAAAXvAAA 10     22     14776
                                                                                                         set
                                                                                                           "TEXT1" = 'AGSRJNFFXN'
                                                                                                         where 
                                                                                                           "ID" = 3 and 
                                                                                                           "COL1" = 'TEXT' and 
                                                                                                           "COL2" = 20
2085 2085       2024-01-01 00:01:00.000 0        COMMIT          NULL   NULL      NULL       1         commit                      0   AAAAAAAAAAAAAAAAAA 10     22     14776

В данном примере это работает, но есть пример, когда это не так.

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

CREATE TABLE SC.TEST_D_LOB (
  ID INTEGER,
  COL1 VARCHAR2(4000),
  TEXT1 CLOB,
  TEXT2 CLOB
);

ALTER TABLE SC.TEST_D_LOB ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

Добавим запись:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2090

INSERT INTO SC.TEST_D_LOB (ID, COL1, TEXT1, TEXT2)
VALUES (1, 'TEXT', DBMS_RANDOM.STRING('U', 20000), DBMS_RANDOM.STRING('U', 10));

COMMIT;

Получим изменения:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SEQUENCE#,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE XIDUSN = 10
  AND XIDSLT = 2
  AND XIDSQN = 13646;

 SCN COMMIT_SCN        COMMIT_TIMESTAMP ROLLBACK       OPERATION   INFO SEG_OWNER TABLE_NAME SEQUENCE#                      SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ----------------------- -------- --------------- ------ --------- ---------- --------- ----------------------------- --- ------------------ ------ ------ ------
2092 NULL       NULL                    0        START           NULL   NULL      NULL       1         set transaction read write    0   AAAAAAAAAAAAAAAAAA 10     2      13646
2092 NULL       NULL                    0        INSERT          NULL   SC        TEST_D_LOB 1         insert into "SC"."TEST_D_LOB" 0   AAAAAAAAAAAAAAAAAA 10     2      13646
                                                                                                        values
                                                                                                           "ID" = 1,
                                                                                                           "COL1" = 'TEXT',
                                                                                                           "TEXT1" = EMPTY_CLOB(),
                                                                                                           "TEXT2" = EMPTY_CLOB()
2094 NULL       NULL                    0        INTERNAL        NULL   NULL      NULL       2         NULL                          0   AAAUpFAAAAAAAAAAAA 10     2      13646
2094 NULL       NULL                    0        SEL_LOB_LOCATOR LOB... SC        TEST_D_LOB 3         DECLARE...                    0   AAAAAAAAAAAAAAAAAA 10     2      13646
2094 NULL       NULL                    0        INTERNAL        NULL   SC        TEST_D_LOB 4         NULL                          0   AAAAAAAAAAAAAAAAAA 10     2      13646
2094 NULL       NULL                    0        LOB_WRITE       LOB... SC        TEST_D_LOB 1         buf_c := 'NMIOJVMPOFAMXWID... 0   AAAAAAAAAAAAAAAAAA 10     2      13646
2094 NULL       NULL                    0        LOB_WRITE       LOB... SC        TEST_D_LOB 1         buf_c := 'GTGIKSLKXDGRQVZV... 0   AAAAAAAAAAAAAAAAAA 10     2      13646
2094 NULL       NULL                    0        INTERNAL        NULL   SC        TEST_D_LOB 2         END;                          0   AAAAAAAAAAAAAAAAAA 10     2      13646
2095 NULL       NULL                    0        UPDATE          NULL   SC        TEST_D_LOB 1         update "SC"."TEST_D_LOB"      0   AAAUpEAAHAAAAdnAAA 10     2      13646
                                                                                                         set
                                                                                                           "TEXT2" = 'SLMNCKEOPB'
                                                                                                         where 
                                                                                                           "ID" = 1 and 
                                                                                                           "COL1" = 'TEXT'
2096 2096       2024-01-01 00:01:10.000 0        COMMIT          NULL   NULL      NULL       1         commit                        0   AAAAAAAAAAAAAAAAAA 10     2      13646

В данном примере видно, что UPDATE LOB поля имеет SEQUENCE# 1.

Нельзя ориентироваться на SEQUENCE#, если при инициализации LogMiner нет опции COMMITED_DATA_ONLY.

Переименование полей таблиц

После переименования поля таблицы, online catalog уже не будет знать старого имени поля до DDL операции. Добавим запись в таблицу и переименуем поле из этой таблицы:

SELECT TIMESTAMP_TO_SCN(SYSDATE) AS CURRENT_SCN FROM DUAL;

CURRENT_SCN
-----------
2100

INSERT INTO SC.TEST (ID, COL1, COL2) VALUES (9, 'ANATOLY', 40);
ALTER TABLE SC.TEST RENAME COLUMN COL2 TO COL22;
INSERT INTO SC.TEST (ID, COL1, COL22) VALUES (10, 'IGOR', 30);
COMMIT;

Посмотрим изменения:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SEQUENCE#,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST_LOB';

 SCN COMMIT_SCN COMMIT_TIMESTAMP ROLLBACK       OPERATION                INFO SEG_OWNER TABLE_NAME                      SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ---------------- -------- --------------- ------------------- --------- ---------- ----------------------------- --- ------------------ ------ ------ ------
2102 NULL       NULL             0        INSERT          Dictionary Mismatch SC        TEST       insert into "SC"."TEST"       0   AAAUomAAHAAAAWvAAE 1      9      2050
                                                                                                    values
                                                                                                       "COL 1" = HEXTORAW('c10a'),
                                                                                                       "COL 2" = HEXTORAW('414e41544f4c59'),
                                                                                                       "COL 3" = HEXTORAW('c129')
2104 NULL       NULL             0        DDL             USER DDL...         SC        TEST       ALTER TABLE "SC"."TEST"...    0   AAAAAAAAAAAAAAAAAB 3      3      2036
2106 NULL       NULL             0        INSERT          NULL                SC        TEST       insert into "SC"."TEST"       0   AAAUomAAHAAAAWvAAC 9      20     2356
                                                                                                    values
                                                                                                       "ID" = 10,
                                                                                                       "COL1" = 'IGOR',
                                                                                                       "COL22" = 30

Из поля INFO мы получили следующую ошибку: Dictionary Mismatch (несоответствие словаря). Для исправления данной ошибки необходимо воспользоваться словарем из archive redo log файлов.

Для начала получим нужные логи:

SELECT
  AL.NAME AS FILE_NAME,
  AL.FIRST_CHANGE# AS FIRST_CHANGE,
  AL.NEXT_CHANGE# AS NEXT_CHANGE,
  AL.DICTIONARY_BEGIN,
  AL.DICTIONARY_END
FROM V$ARCHIVED_LOG AL
  JOIN V$ARCHIVE_DEST_STATUS ADS
    ON ADS.DEST_ID = AL.DEST_ID
WHERE AL.STATUS = 'A'
  AND ADS.STATUS = 'VALID'
  AND ADS.TYPE = 'LOCAL';


                     FILE_NAME FIRST_CHANGE NEXT_CHANGE DICTIONARY_BEGIN DICTIONARY_END 
------------------------------ ------------ ----------- ---------------- --------------
/path/arch1_214_1161333401.dbf 1100         1200        NO               NO
/path/arch1_215_1161333401.dbf 1200         1300        YES              YES
/path/arch1_216_1161333401.dbf 1300         1400        NO               NO
/path/arch1_217_1161333401.dbf 1400         1500        NO               NO
/path/arch1_218_1161333401.dbf 1500         1600        YES              NO
/path/arch1_219_1161333401.dbf 1600         1700        NO               YES
/path/arch1_220_1161333401.dbf 1700         1800        NO               NO
/path/arch1_221_1161333401.dbf 1800         1900        NO               NO

Нам нужны все archive redo log файлы до ближайшего DICTIONARY_BEGIN, то есть эти:

                          NAME 
------------------------------
/path/arch1_218_1161333401.dbf
/path/arch1_219_1161333401.dbf
/path/arch1_220_1161333401.dbf
/path/arch1_221_1161333401.dbf

Еще нужно забрать текущий online redo log файл:

SELECT
  T.NAME AS FILE_NAME
FROM (
  SELECT
    F.MEMBER AS NAME,
    ROW_NUMBER() OVER(PARTITION BY F.GROUP# ORDER BY F.GROUP#) AS GROUP
  FROM V$LOGFILE F
    JOIN V$LOG L
      ON L.GROUP# = F.GROUP#
  WHERE F.TYPE = 'ONLINE'
    AND L.STATUS = 'CURRENT'
) T
WHERE T.GROUP = 1;

       FILE_NAME 
----------------
/path/redo03.log

Далее инициализируем LogMiner с параметрами DICT_FROM_REDO_LOGS и DDL_DICT_TRACKING:

BEGIN
  DBMS_LOGMNR.ADD_LOGFILE('/path/arch1_218_1161333401.dbf', DBMS_LOGMNR.NEW);
  DBMS_LOGMNR.ADD_LOGFILE('/path/arch1_219_1161333401.dbf', DBMS_LOGMNR.ADDFILE);
  DBMS_LOGMNR.ADD_LOGFILE('/path/arch1_220_1161333401.dbf', DBMS_LOGMNR.ADDFILE);
  DBMS_LOGMNR.ADD_LOGFILE('/path/arch1_221_1161333401.dbf', DBMS_LOGMNR.ADDFILE);
  DBMS_LOGMNR.ADD_LOGFILE('/path/redo03.log', DBMS_LOGMNR.ADDFILE);
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>2100,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_REDO_LOGS
      + DBMS_LOGMNR.DDL_DICT_TRACKING
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );
END;

Вот теперь можно получить корректные изменения:

SELECT
  SCN,
  COMMIT_SCN,
  COMMIT_TIMESTAMP,
  ROLLBACK,
  OPERATION,
  INFO,
  SEG_OWNER,
  TABLE_NAME,
  SEQUENCE#,
  SQL_REDO,
  CSF,
  ROW_ID,
  XIDUSN,
  XIDSLT,
  XIDSQN
FROM V$LOGMNR_CONTENTS
WHERE SEG_OWNER = 'SC'
  AND TABLE_NAME = 'TEST_LOB';

 SCN COMMIT_SCN COMMIT_TIMESTAMP ROLLBACK       OPERATION                INFO SEG_OWNER TABLE_NAME                      SQL_REDO CSF             ROW_ID XIDUSN XIDSLT XIDSQN 
---- ---------- ---------------- -------- --------------- ------------------- --------- ---------- ----------------------------- --- ------------------ ------ ------ ------
2102 NULL       NULL             0        INSERT          NULL                SC        TEST       insert into "SC"."TEST"       0   AAAUomAAHAAAAWvAAE 1      9      2050
                                                                                                    values
                                                                                                       "ID" = 9,
                                                                                                       "COL1" = 'ANATOLY',
                                                                                                       "COL2" = 40
2104 NULL       NULL             0        DDL             USER DDL...         SC        TEST       ALTER TABLE "SC"."TEST"...    0   AAAAAAAAAAAAAAAAAB 3      3      2036
2106 NULL       NULL             0        INSERT          NULL                SC        TEST       insert into "SC"."TEST"       0   AAAUomAAHAAAAWvAAC 9      20     2356
                                                                                                    values
                                                                                                       "ID" = 10,
                                                                                                       "COL1" = 'IGOR',
                                                                                                       "COL22" = 30

Сортировка в V$LOGMNR_CONTENTS

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

SELECT
  T.NUM,
  T.SCN,
  T.COMMIT_SCN,
  T.COMMIT_TIMESTAMP,
  ...
FROM ( 
  SELECT 
    ROWNUM AS NUM, 
    SCN, 
    COMMIT_SCN, 
    COMMIT_TIMESTAMP, 
    ...
  FROM V$LOGMNR_CONTENTS
) T
ORDER BY T.SCN, T.NUM

Организация непрерывного сбора изменений

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

ЦИКЛ BEGIN:

  -- Для первой итерации, когда CURRENT_SCN попросту нет, мы берем его от текущего времени
  -- Советую где-нибудь хранить CURRENT_SCN - это точка начала забора изменений
  SELECT COALESCE(CURRENT_SCN, TIMESTAMP_TO_SCN(SYSDATE)) FROM DUAL INTO CURRENT_SCN;

  -- Далее получаем список online redo log files по нашему CURRENT_SCN
  CREATE TEMP TABLE LIST_REDO_LOG_FILES AS
  SELECT
    T.NAME AS FILE_NAME,
  FROM (
    SELECT
      F.MEMBER AS NAME,
      ROW_NUMBER() OVER(PARTITION BY F.GROUP# ORDER BY F.GROUP#) AS GROUP
    FROM V$LOGFILE F
      JOIN V$LOG L
        ON L.GROUP# = F.GROUP#
    WHERE F.TYPE = 'ONLINE'
  ) T
  WHERE T.GROUP = 1
    AND CURRENT_SCN >= T.FIRST_CHANGE;

  -- Добавляем в LogMiner список redo log files
  ЦИКЛ ПО LIST_REDO_LOG_FILES
    ЕСЛИ ПЕРВЫЙ ФАЙЛ, ТО
      DBMS_LOGMNR.ADD_LOGFILE(LIST_REDO_LOG_FILES.FILE_NAME, DBMS_LOGMNR.NEW);
    ИНАЧЕ
      DBMS_LOGMNR.ADD_LOGFILE(LIST_REDO_LOG_FILES.FILE_NAME, DBMS_LOGMNR.ADDFILE);
  ЦИКЛ ПО LIST_REDO_LOG_FILES END;

  -- Инициализируем LogMiner с нужного нам STARTSCN
  DBMS_LOGMNR.START_LOGMNR(
    STARTSCN=>CURRENT_SCN,
    OPTIONS=>
      DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
      + DBMS_LOGMNR.PRINT_PRETTY_SQL
      + DBMS_LOGMNR.NO_SQL_DELIMITER
      + DBMS_LOGMNR.NO_ROWID_IN_STMT
  );

  -- Получаем изменения через представление V$LOGMNR_CONTENTS
  -- Отфильтровываем нужные нам операции и таблицы
  CREATE TEMP TABLE CONTENTS AS
  SELECT
    SCN,
    ...
  FROM V$LOGMNR_CONTENTS
  WHERE ...;

  -- Обновляем CURRENT_SCN
  SELECT MAX(SCN) FROM CONTENTS INTO CURRENT_SCN;
  
  -- Любумся изменениями
  SELECT * FROM CONTENTS;

ЦИКЛ END;

Получение изменений из Oracle Database (LogMiner)