From 4e0b4d5d27167cb6902eba60b4f52560053eb88a Mon Sep 17 00:00:00 2001 From: Roopendra Talekar Date: Tue, 20 Dec 2022 09:47:50 +0530 Subject: [PATCH] add gtid --- index.js | 7 +++- lib/binlog_event.js | 31 +++++++++++++++- lib/code_map.js | 1 + lib/packet/combinloggtid.js | 73 +++++++++++++++++++++++++++++++++++++ lib/packet/index.js | 1 + lib/sequence/binlog.js | 9 +++-- package.json | 5 ++- yarn.lock | 22 ++++++++--- 8 files changed, 135 insertions(+), 14 deletions(-) create mode 100644 lib/packet/combinloggtid.js diff --git a/index.js b/index.js index 89213076..f49196e2 100644 --- a/index.js +++ b/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); diff --git a/lib/binlog_event.js b/lib/binlog_event.js index 91d07bfe..e809e3f8 100644 --- a/lib/binlog_event.js +++ b/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; diff --git a/lib/code_map.js b/lib/code_map.js index b6efe4f9..eaa6e6cd 100644 --- a/lib/code_map.js +++ b/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, diff --git a/lib/packet/combinloggtid.js b/lib/packet/combinloggtid.js new file mode 100644 index 00000000..49d90397 --- /dev/null +++ b/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; diff --git a/lib/packet/index.js b/lib/packet/index.js index 3f9f7da4..c07fadcf 100644 --- a/lib/packet/index.js +++ b/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'); diff --git a/lib/sequence/binlog.js b/lib/sequence/binlog.js index 5a7dbb24..4a6c5ea8 100644 --- a/lib/sequence/binlog.js +++ b/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) { diff --git a/package.json b/package.json index 725c0a5e..1103cc5e 100644 --- a/package.json +++ b/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", diff --git a/yarn.lock b/yarn.lock index 8def1dd1..a5b6bb1d 100644 --- a/yarn.lock +++ b/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"