add gtid

index.js

@@ -171,14 +171,16 @@ ZongJi.prototype._options = function({
   filename,
   position,
   startAtEnd,
-  heartbeatInterval = 5
+  heartbeatInterval = 5,
+  gtidsData,
 }) {
   this.options = {
     serverId,
     filename,
     position,
     startAtEnd,
-    heartbeatInterval
+    heartbeatInterval,
+    gtidsData,
   };
 };
 
@@ -220,6 +222,7 @@ ZongJi.prototype.get = function(name) {
 // - `filename`, `position` the position of binlog to beigin with
 // - `startAtEnd` if true, will update filename / postion automatically
 // - `includeEvents`, `excludeEvents`, `includeSchema`, `exludeSchema` filter different binlog events bubbling
+// - `gtidsData` {[SID]: Array<[start, end]>} GTIDs-set with processed events GTIDs for server-side events filtering
 ZongJi.prototype.start = function(options = {}) {
 
   this._options(options);

lib/binlog_event.js

@@ -89,7 +89,7 @@ function Query(parser) {
   this.errorCode = parser.parseUnsignedNumber(2);
   this.statusVarsLength = parser.parseUnsignedNumber(2);
 
-  this.statusVars = parser.parseString(this.statusVarsLength);
+  this.statusVars = parser.parseBuffer(this.statusVarsLength);
   this.schema = parser.parseString(this.schemaLength);
   parser.parseUnsignedNumber(1);
 
@@ -191,6 +191,9 @@ TableMap.prototype.updateColumnInfo = function() {
   const columnSchemas = tableMap.columnSchemas;
   const columns = [];
   for (let j = 0; j < this.columnCount; j++) {
+    if (!columnSchemas[j]) {
+      continue;
+    }
     try {
       columns.push({
         name: columnSchemas[j].COLUMN_NAME,
@@ -200,7 +203,7 @@ TableMap.prototype.updateColumnInfo = function() {
         metadata: columnsMetadata[j]
       });
     } catch (e) {
-      throw new Error("Error on column index " + j.toString() + " table " + tableMap.tableName + " " + JSON.stringify(tableMap) + " " + e.message)
+      throw new Error("Error on column index " + j.toString() + " table " + tableMap.tableName + " " + JSON.stringify(tableMap) + " " + e.message);
     }
   }
 
@@ -289,6 +292,29 @@ TableMap.prototype.dump = function() {
   console.log('Column types:', this.columnTypes);
 };
 
+/**
+ * GTID_LOG_EVENT
+ * see https://dev.mysql.com/doc/dev/mysql-server/latest/classbinary__log_1_1Gtid__event.html
+ **/
+function GTIDLog(parser) {
+  BinlogEvent.apply(this, arguments);
+  this.GTID_FLAGS = parser.parseUnsignedNumber(1);
+
+  const SIDBuffer = parser._buffer.slice(parser._offset, parser._offset + 16);
+  parser._offset += 16;
+  this.SID = SIDBuffer.toString('hex');
+
+  this.GNO = BigInt(parser.parseUnsignedNumber(4));
+  this.GNO += BigInt(parser.parseUnsignedNumber(4)) << BigInt(32);
+}
+util.inherits(GTIDLog, BinlogEvent);
+
+GTIDLog.prototype.dump = function() {
+  BinlogEvent.prototype.dump.apply(this);
+  console.log('SID: %s', this.SID);
+  console.log('GNO: %d', this.GNO);
+};
+
 function Heartbeat() {
   BinlogEvent.apply(this, arguments);
 }
@@ -308,3 +334,4 @@ exports.Xid = Xid;
 exports.TableMap = TableMap;
 exports.Unknown = Unknown;
 exports.Heartbeat = Heartbeat;
+exports.GTIDLog = GTIDLog;

lib/code_map.js

@@ -48,6 +48,7 @@ const EventClass = {
   FORMAT_DESCRIPTION_EVENT: events.Format,
   XID_EVENT: events.Xid,
 
+  GTID_LOG_EVENT: events.GTIDLog,
   HEARTBEAT_LOG_EVENT: events.Heartbeat,
   TABLE_MAP_EVENT: events.TableMap,
   DELETE_ROWS_EVENT_V1: rowsEvents.DeleteRows,

lib/packet/combinloggtid.js

@@ -0,0 +1,73 @@
+const invariant = require('invariant');
+
+function ComBinlogGTID({serverId, nonBlock, filename, position, gtidsData}) {
+  this.command = 0x1e;
+  this.position = position || 4;
+
+  this.flags = 0;
+  this.flags |= (nonBlock ? 1 : 0);
+  this.flags |= 0x04; /*BINLOG_THROUGH_GTID*/
+
+  this.serverId = serverId || 1;
+  this.filename = filename || '';
+
+  this.gtidsData = gtidsData || {};
+}
+
+const writeUInt64 = (writer, _value /* BigInt */) => {
+  const value = BigInt(_value);
+
+  writer._allocate(8);
+
+  for (let i = 0; i < 8; i++) {
+    writer._buffer[writer._offset++] = Number((value >> BigInt(i * 8)) & BigInt(0xff));
+  }
+};
+
+/**
+ * https://dev.mysql.com/doc/internals/en/com-binlog-dump-gtid.html
+ */
+ComBinlogGTID.prototype.write = function(writer) {
+  writer.writeUnsignedNumber(1, this.command);
+  writer.writeUnsignedNumber(2, this.flags);
+  writer.writeUnsignedNumber(4, this.serverId);
+
+  writer.writeUnsignedNumber(4, Buffer.byteLength(this.filename, 'utf-8'));
+  writer.writeString(this.filename);
+
+  writer.writeUnsignedNumber(4, this.position);
+  writer.writeUnsignedNumber(4, 0); // high-part of this.position
+
+  if (!(this.flags & 0x04)) return;
+
+  const gtidsDataEntries = Object.entries(this.gtidsData);
+  // TODO: Support for multiple intervals per SID
+  writer.writeUnsignedNumber(4, 8 + gtidsDataEntries.length * 40); //data_length
+
+  writer.writeUnsignedNumber(4, gtidsDataEntries.length); //n_sids
+  writer.writeUnsignedNumber(4, 0); //n_sids
+
+  for (let i = 0; i < gtidsDataEntries.length; i++) {
+    const [sid, intervals] = gtidsDataEntries[i];
+
+    const sidBuffer = Buffer.from(sid, 'hex');
+    invariant(sidBuffer.length === 16, 'SID should be 16-bytes hex-string');
+    writer.writeBuffer(sidBuffer);  // sid
+    // Buffer.byteLength(value, 'utf-8')
+
+    writer.writeUnsignedNumber(4, intervals.length); // n_intervals
+    writer.writeUnsignedNumber(4, 0); // high-part of n_intervals
+
+    for (let j = 0; j < intervals.length; j++) {
+      const [start, end] = intervals[j];
+      writeUInt64(writer, start);
+      writeUInt64(writer, end);
+    }
+  }
+};
+
+ComBinlogGTID.prototype.parse = function() {
+  throw new Error('should never be called here');
+};
+
+module.exports = ComBinlogGTID;

lib/packet/index.js

@@ -63,4 +63,5 @@ ErrorPacket.prototype.write = function(writer) {
 exports.EofPacket = EofPacket;
 exports.ErrorPacket = ErrorPacket;
 exports.ComBinlog = require('./combinlog');
+exports.ComBinlogGTID = require('./combinloggtid');
 exports.initBinlogPacketClass = require('./binlog');

lib/sequence/binlog.js

@@ -1,5 +1,5 @@
 const Util = require('util');
-const { EofPacket, ErrorPacket, ComBinlog, initBinlogPacketClass } = require('../packet');
+const { EofPacket, ErrorPacket, ComBinlog, initBinlogPacketClass, ComBinlogGTID } = require('../packet');
 const Sequence = require('@vlasky/mysql/lib/protocol/sequences').Sequence;
 
 module.exports = function(zongji) {
@@ -19,9 +19,12 @@ module.exports = function(zongji) {
   Binlog.prototype.start = function() {
     // options include: position / nonBlock / serverId / filename
     let options = zongji.get([
-      'serverId', 'position', 'filename', 'nonBlock',
+      'serverId', 'position', 'filename', 'nonBlock', 'gtidsData'
     ]);
-    this.emit('packet', new ComBinlog(options));
+
+    const ComBinlogClass = options.gtidsData ? ComBinlogGTID : ComBinlog;
+
+    this.emit('packet', new ComBinlogClass(options));
   };
 
   Binlog.prototype.determinePacket = function(firstByte) {

package.json

@@ -38,8 +38,9 @@
     "@vlasky/mysql": "flydata/mysql#2b25650aa1c49d83b72e77f42bcceee9c0a67291",
     "big-integer": "1.6.48",
     "iconv-lite": "0.6.2",
-    "moment": "^2.29.3",
-    "moment-timezone": "0.5.33"
+    "invariant": "2.2.2",
+    "moment": "2.29.4",
+    "moment-timezone": "0.5.40"
   },
   "volta": {
     "node": "16.14.0",

yarn.lock

@@ -1368,6 +1368,13 @@ inquirer@^7.0.0:
     strip-ansi "^6.0.0"
     through "^2.3.6"
 
+invariant@2.2.2:
+  version "2.2.2"
+  resolved "https://registry.yarnpkg.com/invariant/-/invariant-2.2.2.tgz#9e1f56ac0acdb6bf303306f338be3b204ae60360"
+  integrity sha512-FUiAFCOgp7bBzHfa/fK+Uc/vqywvdN9Wg3CiTprLcE630mrhxjDS5MlBkHzeI6+bC/6bq9VX/hxBt05fPAT5WA==
+  dependencies:
+    loose-envify "^1.0.0"
+
 is-arrayish@^0.2.1:
   version "0.2.1"
   resolved "https://registry.yarnpkg.com/is-arrayish/-/is-arrayish-0.2.1.tgz#77c99840527aa8ecb1a8ba697b80645a7a926a9d"
@@ -1648,7 +1655,7 @@ log-update@^3.0.0:
     cli-cursor "^2.1.0"
     wrap-ansi "^5.0.0"
 
-loose-envify@^1.1.0, loose-envify@^1.4.0:
+loose-envify@^1.0.0, loose-envify@^1.1.0, loose-envify@^1.4.0:
   version "1.4.0"
   resolved "https://registry.yarnpkg.com/loose-envify/-/loose-envify-1.4.0.tgz#71ee51fa7be4caec1a63839f7e682d8132d30caf"
   integrity sha512-lyuxPGr/Wfhrlem2CL/UcnUc1zcqKAImBDzukY7Y5F/yQiNdko6+fRLevlw1HgMySw7f611UIY408EtxRSoK3Q==
@@ -1731,13 +1738,18 @@ mkdirp@^0.5.0, mkdirp@^0.5.1:
   dependencies:
     minimist "^1.2.5"
 
-moment-timezone@0.5.33:
-  version "0.5.33"
-  resolved "https://registry.yarnpkg.com/moment-timezone/-/moment-timezone-0.5.33.tgz#b252fd6bb57f341c9b59a5ab61a8e51a73bbd22c"
-  integrity sha512-PTc2vcT8K9J5/9rDEPe5czSIKgLoGsH8UNpA4qZTVw0Vd/Uz19geE9abbIOQKaAQFcnQ3v5YEXrbSc5BpshH+w==
+moment-timezone@0.5.40:
+  version "0.5.40"
+  resolved "https://registry.yarnpkg.com/moment-timezone/-/moment-timezone-0.5.40.tgz#c148f5149fd91dd3e29bf481abc8830ecba16b89"
+  integrity sha512-tWfmNkRYmBkPJz5mr9GVDn9vRlVZOTe6yqY92rFxiOdWXbjaR0+9LwQnZGGuNR63X456NqmEkbskte8tWL5ePg==
   dependencies:
     moment ">= 2.9.0"
 
+moment@2.29.4:
+  version "2.29.4"
+  resolved "https://registry.yarnpkg.com/moment/-/moment-2.29.4.tgz#3dbe052889fe7c1b2ed966fcb3a77328964ef108"
+  integrity sha512-5LC9SOxjSc2HF6vO2CyuTDNivEdoz2IvyJJGj6X8DJ0eFyfszE0QiEd+iXmBvUP3WHxSjFH/vIsA0EN00cgr8w==
+
 "moment@>= 2.9.0":
   version "2.29.1"
   resolved "https://registry.yarnpkg.com/moment/-/moment-2.29.1.tgz#b2be769fa31940be9eeea6469c075e35006fa3d3"