Plugin mongodb_events_plugin

← Firebird streaming / CDC

The mongodb_events_plugin plugin is designed to register events in the Firebird database and write them to the MongoDB NoSQL DBMS.

Each event is a document in the events collection.

The event contains the following information:

event type (INSERT, UPDATE, DELETE, SET SEQUENCE, EXECUTE SQL);
for INSERT, UPDATE, DELETE events:
- the name of the table for which this event occurred;
- the number of the transaction in which this event occurred;
- old and new values of the table record fields;
- an array of changed fields for the UPDATE event;
- new values of BLOB fields;
for the SET SEQUENCE event:
- sequence (generator) name;
- sequence value;
for the EXECUTE SQL event:
- the number of the transaction in which this event occurred;
- SQL query text.

Only new field values are present in the INSERT event. The UPDATE event contains both old and new field values. Only the old field values are present in the DELETE event.

The EXECUTE SQL event occurs only for DDL queries.

Note: for BLOB fields, the BlobId is always stored. If the configuration parameter is dumpBlobs, then in addition to this, new BLOB field values will be published for the INSERT and UPDATE events. For old field values, this is not possible because the old BLOB field values are not present in the replication log files.

Algorithm for the mongodb_events_plugin

Each event in the MongoDB database is stored as a document in the events collection.

the fb_streaming service analyzes the replication log directory set for the task for unprocessed files;
for unprocessed replication files, the following events are analyzed:
- start of transaction. Allocates a structure in memory to store savepoints;
- сreate a savepoint. A structure in memory is allocated for storing documents;
- release the savepoint. Documents saved in this savepoint are copied to a higher-level savepoint;
- rollback savepoint. The savepoint is destroyed along with all documents;
- commit the transaction. Documents created in the INSERT, UPDATE, DELETE and EXECUTE SQL events are written to the MongoDB database;
- transaction rollback. All savepoints and documents stored in them are destroyed;
- DDL query execution (EXECUTE SQL). A new document is created and saved to the list of documents for the savepoint. The statement text, event type, and transaction number are stored in the document;
- setting a new sequence value (SET SEQUENCE). A new document is created and immediately written to the MongoDB database. The sequence name, event type, and new sequence value are stored in the document;
- INSERT, UPDATE and DELETE events. The table name is checked in the include_tables and exclude_tables filters, and if the table name satisfies the filtering criteria, then a new document is created and saved to the list of documents for the savepoint. The document saves the table name, event type, transaction number, old and new field values.
after processing the file, its number is fixed in the control file.

Other replication events are ignored.

mongodb_events_plugin configuration

First of all, you need to set up asynchronous replication for your database, for this you need to add the following lines to the replication.conf file:

database = d:\fbdata\4.0\horses.fdb
{
    journal_directory = d:\fbdata\4.0\replication\horses\journal
    journal_archive_directory = d:\fbdata\4.0\replication\horses\archive
    journal_archive_command = "copy $(pathname) $(archivepathname) && copy $(pathname) d:\fbdata\4.0\replication\horses\archive_mongo
}

Please note: the log archive files are duplicated here so that logical replication and the task of logging events to MongoDB work at the same time. This is necessary because log archive files are deleted after processing and cannot be used by another task.

Now you need to enable replication in the database:

ALTER DATABASE INCLUDE ALL TO PUBLICATION;
ALTER DATABASE ENABLE PUBLICATION;

Next, configure fb_streaming.conf so that fb_streaming automatically logs events to MongoDB.

task = d:\fbdata\4.0\replication\horses\archive_mongo
{
   database = inet://localhost:3054/d:\fbdata\4.0\horses.fdb
   username = SYSDBA
   password = masterkey
   plugin = mongodb_events_plugin
   mongo_uri = mongodb://localhost:27017
   mongo_database = horses
   dumpBlobs = true
   register_ddl_events = true
   register_sequence_events = true
   # include_tables =
   # exclude_tables =
}

There are a number of additional settings for this plugin:

mongo_uri - URI to connect to the MongoDB server;
mongo_database - the name of the database in MongoDB to which the events are saved;
dumpBlobs - whether to publish new BLOB field values (false by default);
register_ddl_events - whether to register DDL events (true by default);
register_sequence_events - whether to register sequence value setting events (true by default);
include_tables - a regular expression that defines the names of tables for which it is necessary to track events;
exclude_tables - a regular expression that defines the names of tables for which events should not be tracked;
charset_for_none_fields - character set used for converting data from string fields with the character set type NONE. Typically, data in string fields with the NONE character set is stored “as is,” meaning in the original encoding used by the application when saving it. Firebird does not re-encode such strings. The JSON format requires all strings to be in UTF-8 encoding. To correctly convert strings from NONE fields to UTF8, the system needs to know the original encoding of the data. This is precisely what the charset_for_none_fields parameter specifies.

Note: if the database specified in the mongo_database parameter does not exist, then it will be created when you first write to it.

Now you can install and start the service:

fb_streaming install

fb_streaming start

On Linux:

sudo systemctl enable fb_streaming

sudo systemctl start fb_streaming

An example of the contents of the event log in the MongoDB database

To get all events, type the following commands in mongosh

use horses;
'switched to db horses'
db.events.find();

Here, with the first command, we switched to the horses database in which events were recorded.

The second command is a request to fetch data from the events collection. The mongodb_events_plugin plugin writes its events to this collection.

The content of the collection looks like this:

{ _id: ObjectId("638f37f5022b0000ad005775"),
  event: 'EXECUTE SQL',
  sql: 'CREATE SEQUENCE SEQ1',
  tnx: 6590 }
{ _id: ObjectId("638f37f9022b0000ad005776"),
  event: 'EXECUTE SQL',
  sql: 'CREATE TABLE TABLE1 (\r\n  ID INT NOT NULL,\r\n  S VARCHAR(10),\r\n  PRIMARY KEY(ID)\r\n)',
  tnx: 6591 }
{ _id: ObjectId("638f37f9022b0000ad005777"),
  event: 'EXECUTE SQL',
  sql: 'ALTER TABLE TABLE1\r\nENABLE PUBLICATION',
  tnx: 6594 }
{ _id: ObjectId("638f37f9022b0000ad005778"),
  event: 'SET SEQUENCE',
  sequence: 'SEQ1',
  value: 1 }
{ _id: ObjectId("638f37f9022b0000ad005779"),
  event: 'INSERT',
  table: 'TABLE1',
  tnx: 6597,
  record: { ID: 1, S: 'Hello' } }
{ _id: ObjectId("638f3823022b0000ad00577b"),
  event: 'EXECUTE SQL',
  sql: 'DROP TABLE TABLE1',
  tnx: 6608 }
{ _id: ObjectId("638f3827022b0000ad00577c"),
  event: 'EXECUTE SQL',
  sql: 'DROP SEQUENCE SEQ1',
  tnx: 6609 }
{ _id: ObjectId("633d8c86873d0000d8004172"),
  event: 'UPDATE',
  table: 'COLOR',
  tnx: 11771,
  changedFields: [ 'NAME_DE' ],
  oldRecord:
   { NAME: 'dun',
     CODE_COLOR: 14,
     CODE_SENDER: 1,
     NAME_DE: '',
     SHORTNAME: 'dun' },
  record:
   { NAME: 'dun',
     CODE_COLOR: 14,
     CODE_SENDER: 1,
     NAME_DE: 'g',
     SHORTNAME: 'dun' } }
{ _id: ObjectId("633d8c8a873d0000d8004173"),
  event: 'UPDATE',
  table: 'COLOR',
  tnx: 11790,
  changedFields: [ 'NAME_DE' ],
  oldRecord:
   { NAME: 'dun',
     CODE_COLOR: 14,
     CODE_SENDER: 1,
     NAME_DE: 'g',
     SHORTNAME: 'dun' },
  record:
   { NAME: 'dun',
     CODE_COLOR: 14,
     CODE_SENDER: 1,
     NAME_DE: '',
     SHORTNAME: 'dun' } }
{ _id: ObjectId("633d8c8a873d0000d8004174"),
  event: 'INSERT',
  table: 'CLIP',
  tnx: 11821,
  record:
   { AVALUE: 44,
     CODE_CLIP: 1,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 345,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d8004175"),
  event: 'DELETE',
  table: 'CLIP',
  tnx: 11849,
  record:
   { AVALUE: 44,
     CODE_CLIP: 1,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 345,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d8004176"),
  event: 'UPDATE',
  table: 'BREED',
  tnx: 11891,
  changedFields: [ 'MARK' ],
  oldRecord:
   { CODE_DEPARTURE: 15,
     CODE_BREED: 55,
     CODE_SENDER: 1,
     NAME: 'Orlov trotter',
     SHORTNAME: 'orl. trot.'
     MARK: '' },
  record:
   { CODE_DEPARTURE: 15,
     CODE_BREED: 55,
     CODE_SENDER: 1,
     NAME: 'Orlov trotter',
     SHORTNAME: 'orl. trot.',
     MARK: '5' } }
{ _id: ObjectId("633d8c8a873d0000d8004177"),
  event: 'INSERT',
  table: 'CLIP',
  tnx: 11913,
  record:
   { AVALUE: 1,
     CODE_CLIP: 2,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 1,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d8004178"),
  event: 'DELETE',
  table: 'CLIP',
  tnx: 11942,
  record:
   { AVALUE: 1,
     CODE_CLIP: 2,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 1,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d8004179"),
  event: 'INSERT',
  table: 'CLIP',
  tnx: 12001,
  record:
   { AVALUE: 3,
     CODE_CLIP: 5,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 1,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d800417a"),
  event: 'DELETE',
  table: 'CLIP',
  tnx: 12039,
  record:
   { AVALUE: 3,
     CODE_CLIP: 5,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 1,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d800417b"),
  event: 'UPDATE',
  table: 'COLOR',
  tnx: 11799,
  changedFields: [ 'NAME_DE' ],
  oldRecord:
   { NAME: 'clay with mixed hairs',
     CODE_COLOR: 118,
     CODE_SENDER: 1,
     NAME_DE: '',
     SHORTNAME: 'c.m.h.' },
  record:
   { NAME: 'clay with mixed hairs',
     CODE_COLOR: 118,
     CODE_SENDER: 1,
     NAME_DE: '3',
     SHORTNAME: 'c.m.h.' } }
{ _id: ObjectId("633d8c8a873d0000d800417c"),
  event: 'INSERT',
  table: 'CLIP',
  tnx: 12087,
  record:
   { AVALUE: 1,
     CODE_CLIP: 6,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 2,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d800417d"),
  event: 'INSERT',
  table: 'CLIP',
  tnx: 12087,
  record:
   { AVALUE: 3,
     CODE_CLIP: 7,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 3,
     REMARK: 'Other' } }
{ _id: ObjectId("633d8c8a873d0000d800417e"),
  event: 'UPDATE',
  table: 'BREED',
  tnx: 12197,
  changedFields: [ 'MARK' ],
  oldRecord:
   { CODE_DEPARTURE: 17,
     CODE_BREED: 58,
     CODE_SENDER: 1,
     NAME: 'Trotter',
     SHORTNAME: 'rus.rys.',
     MARK: '' },
  record:
   { CODE_DEPARTURE: 17,
     CODE_BREED: 58,
     CODE_SENDER: 1,
     NAME_EN: 'Trotter',
     SHORTNAME: 'rus.rys.',
     MARK: '3' } }
{ _id: ObjectId("633d8c8a873d0000d800417f"),
  event: 'UPDATE',
  table: 'COLOR',
  tnx: 12218,
  changedFields: [ 'NAME_DE', 'SHORTNAME' ],
  oldRecord:
   { NAME: 'red grey',
     CODE_COLOR: 3,
     CODE_SENDER: 1,
     NAME_DE: '',
     SHORTNAME: '' },
  record:
   { NAME: 'red grey',
     CODE_COLOR: 3,
     CODE_SENDER: 1,
     NAME_DE: '5',
     SHORTNAME_EN: 'r.g.' } }
{ _id: ObjectId("633d8c8a873d0000d8004180"),
  event: 'INSERT',
  table: 'CLIP',
  tnx: 12287,
  record:
   { AVALUE: 0,
     CODE_CLIP: 8,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 5,
     REMARK: null } }
{ _id: ObjectId("633d8c8a873d0000d8004181"),
  event: 'UPDATE',
  table: 'CLIP',
  tnx: 12287,
  changedFields: [ 'REMARK' ],
  oldRecord:
   { AVALUE: 3,
     CODE_CLIP: 7,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 3,
     REMARK: 'Other' },
  record:
   { AVALUE: 3,
     CODE_CLIP: 7,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 3,
     REMARK: 'Other 2' } }
{ _id: ObjectId("633d8c8a873d0000d8004182"),
  event: 'DELETE',
  table: 'CLIP',
  tnx: 12287,
  record:
   { AVALUE: 1,
     CODE_CLIP: 6,
     CODE_CLIPTYPE: 1,
     CODE_RECORD: 2,
     REMARK: null } }

Description of fields:

_id - internal primary key for the MongoDB collection;
event - event type;
table - table name for which the event occurred;
tnx - transaction number in which the event occurred;
record - new record in INSERT and UPDATE events, old one - in DELETE event;
oldRecord - old record in the UPDATE event;
changedFields - list of column names that were changed in the UPDATE event;
newBlobs - new values of BLOB fields;
sql - SQL query text for DDL statements;
sequence - sequence name;
value - new value of the sequence.