diff --git a/config.properties.example b/config.properties.example index 1715ab0b2..632137c9d 100644 --- a/config.properties.example +++ b/config.properties.example @@ -291,6 +291,12 @@ kafka.acks=1 # turn this option on and use filters to only include what you need. #ignore_missing_schema = false +# Resolve row column names and types from MySQL 8 FULL TABLE_MAP metadata instead +# of Maxwell's persisted DDL history. Requires binlog_row_metadata=FULL on MySQL +# before the configured/stored starting position. DDL and schema_id output are +# unavailable in this mode. +#schema_source = mysql + # javascript filter # maxwell can run a bit of javascript for each row if you need very custom filtering/data munging. # See http://maxwells-daemon.io/filtering/#javascript_filters for more details diff --git a/docs/docs/config.md b/docs/docs/config.md index 1b6ded4b1..7f1c1323f 100644 --- a/docs/docs/config.md +++ b/docs/docs/config.md @@ -28,6 +28,7 @@ replica_server_id | LONG | unique numeric identifie master_recovery | BOOLEAN | enable experimental master recovery code | false gtid_mode | BOOLEAN | enable GTID-based replication | false recapture_schema | BOOLEAN | recapture the latest schema. Not available in config.properties. | false +schema_source | mysql \| binlog | resolve row schemas from Maxwell's DDL history or MySQL 8 FULL TABLE_MAP metadata | mysql max_schemas | LONG | how many schema deltas to keep before triggering compaction operation | unlimited binlog_heartbeat | BOOLEAN | enable binlog heartbeats to detect stale connections | DISABLED   @@ -46,6 +47,19 @@ schema_ssl | [SSL_OPT](#sslopt) | SSL behavior for schema- schema_jdbc_options | STRING | mysql jdbc connection options for schema server | [DEFAULT_JDBC_OPTS](#jdbcopts)   +### Binlog table-metadata schema source + +On MySQL 8, `schema_source=binlog` makes the `TABLE_MAP` event immediately before each row event authoritative for column names, types, signedness, character sets, enum/set values, geometry types, and primary keys. Maxwell does not capture an initial schema or parse and persist DDL in this mode. This avoids schema-history drift during online schema changes such as gh-ost cutovers. + +The MySQL server must have `binlog_row_metadata=FULL`; Maxwell checks this at startup. The setting affects newly written binlog events only, so the starting position must not be older than the point where `FULL` was enabled. `output_ddl`, `output_schema_id`, and `recapture_schema` are unavailable in this mode. The Maxwell database is still used for positions, heartbeats, and bootstrap requests, and normal filters should still exclude gh-ost shadow tables if their row events should not be emitted. + +Example: + +``` +schema_source=binlog +filter=exclude: *.*, include: homs.*, exclude: homs./^_.*_(gho|ghc|del)$/ +``` + # producer options option | argument | description | default -------------------------------|-------------------------------------| --------------------------------------------------- | ------- @@ -324,5 +338,3 @@ A get request will return the live config state "filter": "exclude: noisy_db.*" } ``` - - diff --git a/src/main/java/com/zendesk/maxwell/Maxwell.java b/src/main/java/com/zendesk/maxwell/Maxwell.java index 04b0741d4..da6d1d598 100644 --- a/src/main/java/com/zendesk/maxwell/Maxwell.java +++ b/src/main/java/com/zendesk/maxwell/Maxwell.java @@ -99,22 +99,24 @@ private Position attemptMasterRecovery() throws Exception { recoveredHeartbeat = masterRecovery.recover(); if (recoveredHeartbeat != null) { - // load up the schema from the recovery position and chain it into the - // new server_id - MysqlSchemaStore oldServerSchemaStore = new MysqlSchemaStore( - context.getMaxwellConnectionPool(), - context.getReplicationConnectionPool(), - context.getSchemaConnectionPool(), - recoveryInfo.serverID, - recoveryInfo.position, - context.getCaseSensitivity(), - config.filter, - false - ); - - // Note we associate this schema to the start position of the heartbeat event, so that - // we pick it up when resuming at the event after the heartbeat. - oldServerSchemaStore.clone(context.getServerID(), recoveredHeartbeat.getPosition()); + if (!MaxwellConfig.SCHEMA_SOURCE_BINLOG.equals(config.schemaSource)) { + // Load up the schema from the recovery position and chain it into the + // new server_id. TABLE_MAP metadata mode only needs the recovered position. + MysqlSchemaStore oldServerSchemaStore = new MysqlSchemaStore( + context.getMaxwellConnectionPool(), + context.getReplicationConnectionPool(), + context.getSchemaConnectionPool(), + recoveryInfo.serverID, + recoveryInfo.position, + context.getCaseSensitivity(), + config.filter, + false + ); + + // Note we associate this schema to the start position of the heartbeat event, so that + // we pick it up when resuming at the event after the heartbeat. + oldServerSchemaStore.clone(context.getServerID(), recoveredHeartbeat.getPosition()); + } return recoveredHeartbeat.getNextPosition(); } } @@ -123,6 +125,10 @@ private Position attemptMasterRecovery() throws Exception { private void logColumnCastError(ColumnDefCastException e) throws SQLException, SchemaStoreException { LOGGER.error("checking for schema inconsistencies in " + e.database + "." + e.table); + if (this.replicator.usesBinlogRowMetadata()) { + LOGGER.error("row schema came from TABLE_MAP metadata; no persisted schema is available to compare"); + return; + } try ( Connection conn = context.getSchemaConnectionPool().getConnection(); SchemaCapturer capturer = new SchemaCapturer(conn, context.getCaseSensitivity(), e.database, e.table)) { Schema recaptured = capturer.capture(); @@ -242,6 +248,9 @@ private void startInner() throws Exception { try ( Connection connection = this.context.getReplicationConnection(); Connection rawConnection = this.context.getRawMaxwellConnection() ) { MaxwellMysqlStatus.ensureReplicationMysqlState(connection); + if (MaxwellConfig.SCHEMA_SOURCE_BINLOG.equals(config.schemaSource)) { + MaxwellMysqlStatus.ensureFullBinlogRowMetadata(connection); + } MaxwellMysqlStatus.ensureMaxwellMysqlState(rawConnection); if (config.gtidMode) { MaxwellMysqlStatus.ensureGtidMysqlState(connection); @@ -260,16 +269,25 @@ private void startInner() throws Exception { logBanner(producer, initPosition); this.context.setPosition(initPosition); - MysqlSchemaStore mysqlSchemaStore = new MysqlSchemaStore(this.context, initPosition); - BootstrapController bootstrapController = this.context.getBootstrapController(mysqlSchemaStore.getSchemaID()); + boolean useBinlogRowMetadata = MaxwellConfig.SCHEMA_SOURCE_BINLOG.equals(config.schemaSource); + MysqlSchemaStore mysqlSchemaStore = null; + Long initialSchemaID = null; - this.context.startSchemaCompactor(); + if (useBinlogRowMetadata) { + LOGGER.info("Using FULL TABLE_MAP metadata for row schemas; DDL schema tracking is disabled"); + } else { + mysqlSchemaStore = new MysqlSchemaStore(this.context, initPosition); + initialSchemaID = mysqlSchemaStore.getSchemaID(); + this.context.startSchemaCompactor(); + + if (config.recaptureSchema) { + mysqlSchemaStore.captureAndSaveSchema(); + } - if (config.recaptureSchema) { - mysqlSchemaStore.captureAndSaveSchema(); + mysqlSchemaStore.getSchema(); // trigger schema to load / capture before we start the replicator. } - mysqlSchemaStore.getSchema(); // trigger schema to load / capture before we start the replicator. + BootstrapController bootstrapController = this.context.getBootstrapController(initialSchemaID); this.replicator = new BinlogConnectorReplicator( mysqlSchemaStore, @@ -289,7 +307,8 @@ private void startInner() throws Exception { config.outputConfig, config.bufferMemoryUsage, config.replicationReconnectionRetries, - config.binlogEventQueueSize + config.binlogEventQueueSize, + useBinlogRowMetadata ); context.setReplicator(replicator); diff --git a/src/main/java/com/zendesk/maxwell/MaxwellConfig.java b/src/main/java/com/zendesk/maxwell/MaxwellConfig.java index 1c85bdeb5..b5ccd0246 100644 --- a/src/main/java/com/zendesk/maxwell/MaxwellConfig.java +++ b/src/main/java/com/zendesk/maxwell/MaxwellConfig.java @@ -33,6 +33,8 @@ * Configuration object for Maxwell */ public class MaxwellConfig extends AbstractConfig { + public static final String SCHEMA_SOURCE_MYSQL = "mysql"; + public static final String SCHEMA_SOURCE_BINLOG = "binlog"; static final Logger LOGGER = LoggerFactory.getLogger(MaxwellConfig.class); /** @@ -464,6 +466,13 @@ public class MaxwellConfig extends AbstractConfig { */ public boolean recaptureSchema; + /** + * Source used to resolve table definitions for row events. + * "mysql" uses Maxwell's persisted schema history; "binlog" uses MySQL 8 + * FULL TABLE_MAP metadata and does not replay DDL into the schema store. + */ + public String schemaSource; + /** * float between 0 and 1, defines percentage of JVM memory to use buffering rows. *

@@ -832,6 +841,8 @@ protected MaxwellOptionParser buildOptionParser() { parser.accepts( "recapture_schema", "recapture the latest schema. Only use if Maxwell's schema has fallen out of sync" ) .withOptionalArg().ofType(Boolean.class); + parser.accepts( "schema_source", "table schema source: mysql|binlog. binlog requires MySQL binlog_row_metadata=FULL" ) + .withRequiredArg(); parser.accepts( "buffer_memory_usage", "Percentage of JVM memory available for transaction buffer. Floating point between 0 and 1." ) .withRequiredArg().ofType(Float.class); parser.accepts("binlog_event_queue_size", "Size of queue to buffer events parsed from binlog.") @@ -1208,6 +1219,7 @@ private void setup(OptionSet options, Properties properties) { this.masterRecovery = fetchBooleanOption("master_recovery", options, properties, false); this.ignoreProducerError = fetchBooleanOption("ignore_producer_error", options, properties, true); this.recaptureSchema = fetchBooleanOption("recapture_schema", options, null, false); + this.schemaSource = fetchStringOption("schema_source", options, properties, SCHEMA_SOURCE_MYSQL).toLowerCase(); this.bufferMemoryUsage = fetchFloatOption("buffer_memory_usage", options, properties, 0.25f); this.maxSchemaDeltas = fetchIntegerOption("max_schemas", options, properties, null); @@ -1333,6 +1345,18 @@ public void validate() { validatePartitionBy(); validateFilter(); + if (!this.schemaSource.equals(SCHEMA_SOURCE_MYSQL) && !this.schemaSource.equals(SCHEMA_SOURCE_BINLOG)) + usageForOptions("please specify --schema_source=mysql|binlog", "--schema_source"); + + if (this.schemaSource.equals(SCHEMA_SOURCE_BINLOG) && this.recaptureSchema) + usageForOptions("--recapture_schema cannot be used with --schema_source=binlog", "--recapture_schema", "--schema_source"); + + if (this.schemaSource.equals(SCHEMA_SOURCE_BINLOG) && this.outputConfig.outputDDL) + usageForOptions("--output_ddl cannot be used with --schema_source=binlog", "--output_ddl", "--schema_source"); + + if (this.schemaSource.equals(SCHEMA_SOURCE_BINLOG) && this.outputConfig.includesSchemaId) + usageForOptions("--output_schema_id cannot be used with --schema_source=binlog", "--output_schema_id", "--schema_source"); + if ( this.producerType.equals("kafka") ) { if ( !this.kafkaProperties.containsKey("bootstrap.servers") ) { usageForOptions("Please specify kafka.bootstrap.servers", "kafka"); diff --git a/src/main/java/com/zendesk/maxwell/MaxwellMysqlStatus.java b/src/main/java/com/zendesk/maxwell/MaxwellMysqlStatus.java index 4c2fd6839..8568b1f67 100644 --- a/src/main/java/com/zendesk/maxwell/MaxwellMysqlStatus.java +++ b/src/main/java/com/zendesk/maxwell/MaxwellMysqlStatus.java @@ -151,6 +151,14 @@ public static void ensureReplicationMysqlState(Connection c) throws SQLException m.ensureRowImageFormat(); } + /** + * Verify that TABLE_MAP events contain the metadata required to decode rows + * without Maxwell's persisted schema history. + */ + public static void ensureFullBinlogRowMetadata(Connection c) throws SQLException, MaxwellCompatibilityError { + new MaxwellMysqlStatus(c).ensureVariableState("binlog_row_metadata", "FULL"); + } + /** * Verify that the maxwell database is in the expected state * @param c a JDBC connection diff --git a/src/main/java/com/zendesk/maxwell/replication/BinlogConnectorReplicator.java b/src/main/java/com/zendesk/maxwell/replication/BinlogConnectorReplicator.java index fe24f17ba..9285705ea 100644 --- a/src/main/java/com/zendesk/maxwell/replication/BinlogConnectorReplicator.java +++ b/src/main/java/com/zendesk/maxwell/replication/BinlogConnectorReplicator.java @@ -76,6 +76,7 @@ public class BinlogConnectorReplicator extends RunLoopProcess implements Replica private final Meter rowMeter; private SchemaStore schemaStore; + private final boolean useBinlogRowMetadata; private Histogram transactionRowCount; private Histogram transactionExecutionTime; @@ -124,7 +125,8 @@ public BinlogConnectorReplicator( outputConfig, bufferMemoryUsage, replicationReconnectionRetries, - BINLOG_QUEUE_SIZE + BINLOG_QUEUE_SIZE, + false ); } @@ -147,6 +149,50 @@ public BinlogConnectorReplicator( float bufferMemoryUsage, int replicationReconnectionRetries, int binlogEventQueueSize + ) { + this( + schemaStore, + producer, + bootstrapper, + mysqlConfig, + replicaServerID, + maxwellSchemaDatabaseName, + metrics, + start, + stopOnEOF, + clientID, + heartbeatNotifier, + scripting, + filter, + ignoreMissingSchema, + outputConfig, + bufferMemoryUsage, + replicationReconnectionRetries, + binlogEventQueueSize, + false + ); + } + + public BinlogConnectorReplicator( + SchemaStore schemaStore, + AbstractProducer producer, + BootstrapController bootstrapper, + MaxwellMysqlConfig mysqlConfig, + Long replicaServerID, + String maxwellSchemaDatabaseName, + Metrics metrics, + Position start, + boolean stopOnEOF, + String clientID, + HeartbeatNotifier heartbeatNotifier, + Scripting scripting, + Filter filter, + boolean ignoreMissingSchema, + MaxwellOutputConfig outputConfig, + float bufferMemoryUsage, + int replicationReconnectionRetries, + int binlogEventQueueSize, + boolean useBinlogRowMetadata ) { this.clientID = clientID; this.bootstrapper = bootstrapper; @@ -157,6 +203,7 @@ public BinlogConnectorReplicator( this.stopOnEOF = stopOnEOF; this.scripting = scripting; this.schemaStore = schemaStore; + this.useBinlogRowMetadata = useBinlogRowMetadata; this.tableCache = new TableCache(maxwellSchemaDatabaseName); this.filter = filter; this.ignoreMissingSchema = ignoreMissingSchema; @@ -403,6 +450,14 @@ private void processQueryEvent(String dbName, String sql, SchemaStore schemaStor } private void processQueryEvent(BinlogConnectorEvent event) throws Exception { + if (useBinlogRowMetadata) { + // TABLE_MAP is authoritative in this mode. DDL is deliberately not + // parsed, persisted, or emitted; the next TABLE_MAP supplies the new + // definition. + tableCache.clear(); + return; + } + QueryEventData data = event.queryData(); processQueryEvent( data.getDatabase(), @@ -589,7 +644,7 @@ private RowMapBuffer getTransactionRows(BinlogConnectorEvent beginEvent) throws break; case TABLE_MAP: TableMapEventData data = event.tableMapData(); - tableCache.processEvent(getSchema(), this.filter, this.ignoreMissingSchema, data.getTableId(), data.getDatabase(), data.getTable()); + processTableMapEvent(data); break; case ROWS_QUERY: RowsQueryEventData rqed = event.getEvent().getData(); @@ -717,7 +772,7 @@ public RowMap getRow() throws Exception { break; case TABLE_MAP: TableMapEventData data = event.tableMapData(); - tableCache.processEvent(getSchema(), this.filter,this.ignoreMissingSchema, data.getTableId(), data.getDatabase(), data.getTable()); + processTableMapEvent(data); break; case QUERY: QueryEventData qe = event.queryData(); @@ -775,13 +830,28 @@ protected BinlogConnectorEvent pollEvent() throws InterruptedException { } public Schema getSchema() throws SchemaStoreException { + if (this.schemaStore == null) + throw new SchemaStoreException("schema history is disabled by --schema_source=binlog"); return this.schemaStore.getSchema(); } public Long getSchemaId() throws SchemaStoreException { + if (this.schemaStore == null) + return null; return this.schemaStore.getSchemaID(); } + public boolean usesBinlogRowMetadata() { + return useBinlogRowMetadata; + } + + private void processTableMapEvent(TableMapEventData data) throws SchemaStoreException { + if (useBinlogRowMetadata) + tableCache.processEvent(data, this.filter); + else + tableCache.processEvent(getSchema(), this.filter, this.ignoreMissingSchema, data.getTableId(), data.getDatabase(), data.getTable()); + } + @Override public void onConnect(BinaryLogClient client) { LOGGER.info("Binlog connected."); diff --git a/src/main/java/com/zendesk/maxwell/replication/BinlogTableMetadata.java b/src/main/java/com/zendesk/maxwell/replication/BinlogTableMetadata.java new file mode 100644 index 000000000..faedcfcb1 --- /dev/null +++ b/src/main/java/com/zendesk/maxwell/replication/BinlogTableMetadata.java @@ -0,0 +1,321 @@ +package com.zendesk.maxwell.replication; + +import com.github.shyiko.mysql.binlog.event.TableMapEventData; +import com.github.shyiko.mysql.binlog.event.TableMapEventMetadata; +import com.github.shyiko.mysql.binlog.event.deserialization.ColumnType; +import com.mysql.cj.CharsetMapping; +import com.zendesk.maxwell.schema.Table; +import com.zendesk.maxwell.schema.columndef.ColumnDef; + +import java.util.ArrayList; +import java.util.BitSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Builds the table definition needed to decode a row directly from a MySQL 8 + * TABLE_MAP event generated with binlog_row_metadata=FULL. + */ +final class BinlogTableMetadata { + private BinlogTableMetadata() { } + + static Table buildTable(TableMapEventData event) { + TableMapEventMetadata metadata = event.getEventMetadata(); + byte[] columnTypes = event.getColumnTypes(); + int[] columnMetadata = event.getColumnMetadata(); + + if (metadata == null || metadata.getColumnNames() == null) { + throw metadataError(event, "column names are absent"); + } + + List names = metadata.getColumnNames(); + if (columnTypes == null || columnMetadata == null || + columnTypes.length != names.size() || columnMetadata.length != names.size()) { + throw metadataError(event, "column name, type, and metadata counts do not match"); + } + + List columns = new ArrayList<>(names.size()); + int characterIndex = 0; + int enumIndex = 0; + int setIndex = 0; + int geometryIndex = 0; + + for (int i = 0; i < names.size(); i++) { + ColumnType type = effectiveType(columnTypes[i], columnMetadata[i]); + String charset = null; + String[] enumValues = null; + + if (isCharacterType(type)) { + Integer collation = collationForColumn( + characterIndex++, + metadata.getColumnCharsets(), + metadata.getDefaultCharset() + ); + charset = charsetForCollation(event, names.get(i), collation); + } else if (type == ColumnType.ENUM) { + enumValues = typeValues(event, names.get(i), metadata.getEnumStrValues(), enumIndex++, "ENUM"); + } else if (type == ColumnType.SET) { + enumValues = typeValues(event, names.get(i), metadata.getSetStrValues(), setIndex++, "SET"); + } + + String sqlType = sqlType(type, columnMetadata[i], charset, metadata, geometryIndex); + if (type == ColumnType.GEOMETRY) + geometryIndex++; + + boolean signed = isSigned(event, type, i, metadata.getSignedness()); + Long columnLength = temporalPrecision(type, columnMetadata[i]); + columns.add(ColumnDef.build(names.get(i), charset, sqlType, (short) i, signed, enumValues, columnLength)); + } + + return new Table(event.getDatabase(), event.getTable(), null, columns, primaryKeys(event, metadata, names)); + } + + private static ColumnType effectiveType(byte rawType, int metadata) { + ColumnType type = ColumnType.byCode(rawType & 0xff); + if (type == null) + throw new IllegalArgumentException("Unsupported binlog column type code " + (rawType & 0xff)); + + // MySQL encodes ENUM, SET, and long CHAR variants inside MYSQL_TYPE_STRING metadata. + if (type == ColumnType.STRING && metadata >= 256) { + int encodedType = metadata >> 8; + if ((encodedType & 0x30) != 0x30) + encodedType |= 0x30; + + ColumnType effective = ColumnType.byCode(encodedType); + if (effective == ColumnType.ENUM || effective == ColumnType.SET) + return effective; + } + + return type; + } + + private static boolean isCharacterType(ColumnType type) { + switch (type) { + case VARCHAR: + case VAR_STRING: + case STRING: + case TINY_BLOB: + case MEDIUM_BLOB: + case LONG_BLOB: + case BLOB: + return true; + default: + return false; + } + } + + private static Integer collationForColumn( + int characterIndex, + List columnCharsets, + TableMapEventMetadata.DefaultCharset defaultCharset + ) { + if (columnCharsets != null) { + if (characterIndex >= columnCharsets.size()) + return null; + return columnCharsets.get(characterIndex); + } + + if (defaultCharset == null) + return null; + + Map overrides = defaultCharset.getCharsetCollations(); + // DEFAULT_CHARSET override indexes are relative to character columns, + // unlike primary-key and column-name indexes, which address all columns. + if (overrides != null && overrides.containsKey(characterIndex)) + return overrides.get(characterIndex); + + return defaultCharset.getDefaultCharsetCollation(); + } + + private static String charsetForCollation(TableMapEventData event, String column, Integer collation) { + if (collation == null) + throw metadataError(event, "character set metadata is absent for column " + column); + + String charset = CharsetMapping.getStaticMysqlCharsetNameForCollationIndex(collation); + if (charset == null) + throw metadataError(event, "unsupported collation " + collation + " for column " + column); + return charset; + } + + private static String[] typeValues( + TableMapEventData event, + String column, + List values, + int index, + String type + ) { + if (values == null || index >= values.size()) + throw metadataError(event, type + " values are absent for column " + column); + return values.get(index); + } + + private static boolean isSigned(TableMapEventData event, ColumnType type, int columnIndex, BitSet unsignedColumns) { + if (!isIntegerType(type)) + return true; + if (unsignedColumns == null) + throw metadataError(event, "signedness metadata is absent for integer column " + columnIndex); + + // MySQL's SIGNEDNESS optional metadata bitmap has a set bit for UNSIGNED columns. + return !unsignedColumns.get(columnIndex); + } + + private static boolean isIntegerType(ColumnType type) { + switch (type) { + case TINY: + case SHORT: + case INT24: + case LONG: + case LONGLONG: + return true; + default: + return false; + } + } + + private static Long temporalPrecision(ColumnType type, int metadata) { + switch (type) { + case TIMESTAMP_V2: + case DATETIME_V2: + case TIME_V2: + return (long) metadata; + default: + return null; + } + } + + private static String sqlType( + ColumnType type, + int metadata, + String charset, + TableMapEventMetadata eventMetadata, + int geometryIndex + ) { + switch (type) { + case DECIMAL: + case NEWDECIMAL: + return "decimal"; + case TINY: + return "tinyint"; + case SHORT: + return "smallint"; + case INT24: + return "mediumint"; + case LONG: + return "int"; + case LONGLONG: + return "bigint"; + case FLOAT: + return "float"; + case DOUBLE: + return "double"; + case TIMESTAMP: + case TIMESTAMP_V2: + return "timestamp"; + case DATE: + case NEWDATE: + return "date"; + case TIME: + case TIME_V2: + return "time"; + case DATETIME: + case DATETIME_V2: + return "datetime"; + case YEAR: + return "year"; + case VARCHAR: + case VAR_STRING: + return isBinary(charset) ? "varbinary" : "varchar"; + case STRING: + return isBinary(charset) ? "binary" : "char"; + case BIT: + return "bit"; + case JSON: + return "json"; + case ENUM: + return "enum"; + case SET: + return "set"; + case TINY_BLOB: + return isBinary(charset) ? "tinyblob" : "tinytext"; + case MEDIUM_BLOB: + return isBinary(charset) ? "mediumblob" : "mediumtext"; + case LONG_BLOB: + return isBinary(charset) ? "longblob" : "longtext"; + case BLOB: + return blobType(metadata, isBinary(charset)); + case GEOMETRY: + return geometryType(eventMetadata.getGeometryTypes(), geometryIndex); + default: + throw new IllegalArgumentException("Unsupported binlog column type " + type); + } + } + + private static boolean isBinary(String charset) { + return "binary".equalsIgnoreCase(charset); + } + + private static String blobType(int metadata, boolean binary) { + String prefix; + switch (metadata) { + case 1: + prefix = "tiny"; + break; + case 3: + prefix = "medium"; + break; + case 4: + prefix = "long"; + break; + case 2: + default: + prefix = ""; + break; + } + return prefix + (binary ? "blob" : "text"); + } + + private static String geometryType(List geometryTypes, int index) { + if (geometryTypes == null || index >= geometryTypes.size()) + return "geometry"; + + switch (geometryTypes.get(index)) { + case 1: return "point"; + case 2: return "linestring"; + case 3: return "polygon"; + case 4: return "multipoint"; + case 5: return "multilinestring"; + case 6: return "multipolygon"; + case 7: return "geometrycollection"; + default: return "geometry"; + } + } + + private static List primaryKeys( + TableMapEventData event, + TableMapEventMetadata metadata, + List names + ) { + Set indexes = new LinkedHashSet<>(); + if (metadata.getSimplePrimaryKeys() != null) + indexes.addAll(metadata.getSimplePrimaryKeys()); + if (metadata.getPrimaryKeysWithPrefix() != null) + indexes.addAll(metadata.getPrimaryKeysWithPrefix().keySet()); + + List keys = new ArrayList<>(indexes.size()); + for (Integer index : indexes) { + if (index == null || index < 0 || index >= names.size()) + throw metadataError(event, "invalid primary-key column index " + index); + keys.add(names.get(index)); + } + return keys; + } + + private static IllegalStateException metadataError(TableMapEventData event, String detail) { + return new IllegalStateException( + "TABLE_MAP metadata for " + event.getDatabase() + "." + event.getTable() + " is incomplete: " + detail + + ". Enable binlog_row_metadata=FULL before the starting binlog position or use --schema_source=mysql." + ); + } +} diff --git a/src/main/java/com/zendesk/maxwell/replication/Replicator.java b/src/main/java/com/zendesk/maxwell/replication/Replicator.java index 6e9468266..29cccda58 100644 --- a/src/main/java/com/zendesk/maxwell/replication/Replicator.java +++ b/src/main/java/com/zendesk/maxwell/replication/Replicator.java @@ -14,6 +14,7 @@ public interface Replicator extends StoppableTask { Long getLastHeartbeatRead(); Schema getSchema() throws SchemaStoreException; Long getSchemaId() throws SchemaStoreException; + default boolean usesBinlogRowMetadata() { return false; } void stopAtHeartbeat(long heartbeat); void runLoop() throws Exception; diff --git a/src/main/java/com/zendesk/maxwell/replication/TableCache.java b/src/main/java/com/zendesk/maxwell/replication/TableCache.java index 07d4a5dc4..ad27c1fd9 100644 --- a/src/main/java/com/zendesk/maxwell/replication/TableCache.java +++ b/src/main/java/com/zendesk/maxwell/replication/TableCache.java @@ -2,6 +2,7 @@ import java.util.HashMap; +import com.github.shyiko.mysql.binlog.event.TableMapEventData; import com.zendesk.maxwell.filtering.Filter; import com.zendesk.maxwell.schema.Database; import com.zendesk.maxwell.schema.Schema; @@ -42,6 +43,20 @@ public void processEvent(Schema schema, Filter filter, Boolean ignoreMissingSche } + /** + * Cache a table definition carried by a MySQL 8 TABLE_MAP event. Unlike the + * persisted-schema path, replace the entry on every event because TABLE_MAP + * is the authoritative schema generation for the following row events. + */ + public void processEvent(TableMapEventData event, Filter filter) { + if (filter.isTableBlacklisted(event.getDatabase(), event.getTable())) { + tableMapCache.remove(event.getTableId()); + return; + } + + tableMapCache.put(event.getTableId(), BinlogTableMetadata.buildTable(event)); + } + public Table getTable(Long tableId) { return tableMapCache.get(tableId); } diff --git a/src/test/java/com/zendesk/maxwell/MaxwellConfigTest.java b/src/test/java/com/zendesk/maxwell/MaxwellConfigTest.java index 1f2cb2a9e..6b970bb3a 100644 --- a/src/test/java/com/zendesk/maxwell/MaxwellConfigTest.java +++ b/src/test/java/com/zendesk/maxwell/MaxwellConfigTest.java @@ -153,6 +153,18 @@ public void testPubsubConfigDefault() { assertEquals(config.pubsubRpcTimeoutMultiplier, 1.0f, 0.0f); } + @Test + public void testSchemaSourceDefaultsToMysql() { + config = new MaxwellConfig(); + assertEquals("mysql", config.schemaSource); + } + + @Test + public void testBinlogSchemaSourceFromArgs() { + config = new MaxwellConfig(new String[] { "--schema_source=binlog" }); + assertEquals("binlog", config.schemaSource); + } + private String getTestConfigDir() { return System.getProperty("user.dir") + "/src/test/resources/config/"; diff --git a/src/test/java/com/zendesk/maxwell/replication/BinlogRowMetadataIntegrationTest.java b/src/test/java/com/zendesk/maxwell/replication/BinlogRowMetadataIntegrationTest.java new file mode 100644 index 000000000..c1c8d33db --- /dev/null +++ b/src/test/java/com/zendesk/maxwell/replication/BinlogRowMetadataIntegrationTest.java @@ -0,0 +1,72 @@ +package com.zendesk.maxwell.replication; + +import com.zendesk.maxwell.MaxwellConfig; +import com.zendesk.maxwell.MaxwellTestSupport; +import com.zendesk.maxwell.MysqlIsolatedServer; +import com.zendesk.maxwell.filtering.Filter; +import com.zendesk.maxwell.row.RowMap; +import org.junit.Test; + +import java.sql.ResultSet; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assume.assumeTrue; + +public class BinlogRowMetadataIntegrationTest { + @Test + public void followsGhostCutoverWithoutProcessingDDL() throws Exception { + assumeTrue(MysqlIsolatedServer.getVersion().atLeast(8, 0)); + assumeTrue(!MysqlIsolatedServer.getVersion().isMariaDB); + + MysqlIsolatedServer server = MaxwellTestSupport.setupServer("--binlog-row-metadata=FULL"); + try { + Filter filter = new Filter( + "exclude: *.*,include: homs.*,exclude: homs./^_.*_(gho|ghc|del)$/" + ); + + String[] before = { + "DROP DATABASE IF EXISTS homs", + "CREATE DATABASE homs", + "CREATE TABLE homs.order_line (id BIGINT UNSIGNED PRIMARY KEY, routing_approach VARCHAR(32))" + }; + String[] changes = { + "INSERT INTO homs.order_line VALUES (1, 'legacy')", + "CREATE TABLE homs._order_line_gho LIKE homs.order_line", + "ALTER TABLE homs._order_line_gho MODIFY routing_approach INT NOT NULL", + "INSERT INTO homs._order_line_gho VALUES (2, 3)", + "RENAME TABLE homs.order_line TO homs._order_line_del, homs._order_line_gho TO homs.order_line", + "INSERT INTO homs.order_line VALUES (3, -4)", + "DROP TABLE homs._order_line_del" + }; + + List rows = MaxwellTestSupport.getRowsWithReplicator( + server, + changes, + before, + config -> configureBinlogMetadataMode(config, filter) + ); + + assertEquals(2, rows.size()); + assertEquals("order_line", rows.get(0).getTable()); + assertEquals("legacy", rows.get(0).getData("routing_approach")); + assertEquals(1L, ((Number) rows.get(0).getData("id")).longValue()); + assertEquals(-4L, ((Number) rows.get(1).getData("routing_approach")).longValue()); + assertEquals(3L, ((Number) rows.get(1).getData("id")).longValue()); + assertFalse(rows.stream().anyMatch(row -> row.getTable().startsWith("_"))); + + try (ResultSet schemas = server.query("SELECT COUNT(*) FROM maxwell.schemas")) { + schemas.next(); + assertEquals(0, schemas.getInt(1)); + } + } finally { + server.shutDown(); + } + } + + private static void configureBinlogMetadataMode(MaxwellConfig config, Filter filter) { + config.schemaSource = MaxwellConfig.SCHEMA_SOURCE_BINLOG; + config.filter = filter; + } +} diff --git a/src/test/java/com/zendesk/maxwell/replication/BinlogTableMetadataTest.java b/src/test/java/com/zendesk/maxwell/replication/BinlogTableMetadataTest.java new file mode 100644 index 000000000..f7766fbc9 --- /dev/null +++ b/src/test/java/com/zendesk/maxwell/replication/BinlogTableMetadataTest.java @@ -0,0 +1,150 @@ +package com.zendesk.maxwell.replication; + +import com.github.shyiko.mysql.binlog.event.TableMapEventData; +import com.github.shyiko.mysql.binlog.event.TableMapEventMetadata; +import com.github.shyiko.mysql.binlog.event.deserialization.ColumnType; +import com.zendesk.maxwell.filtering.Filter; +import com.zendesk.maxwell.schema.Table; +import com.zendesk.maxwell.schema.columndef.BigIntColumnDef; +import com.zendesk.maxwell.schema.columndef.ColumnDefWithLength; +import com.zendesk.maxwell.schema.columndef.EnumeratedColumnDef; +import com.zendesk.maxwell.schema.columndef.IntColumnDef; +import com.zendesk.maxwell.schema.columndef.StringColumnDef; +import org.junit.Test; + +import java.util.Arrays; +import java.util.BitSet; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class BinlogTableMetadataTest { + @Test + public void buildsTableFromFullMetadata() { + TableMapEventData event = event( + new ColumnType[] { + ColumnType.LONG, + ColumnType.LONGLONG, + ColumnType.VARCHAR, + ColumnType.BLOB, + ColumnType.STRING, + ColumnType.STRING, + ColumnType.DATETIME_V2, + ColumnType.GEOMETRY + }, + new int[] { + 0, + 0, + 255, + 2, + (ColumnType.ENUM.getCode() << 8) | 1, + (ColumnType.SET.getCode() << 8) | 1, + 6, + 4 + }, + "unsigned_id", "signed_id", "label", "payload", "state", "flags", "created_at", "point" + ); + + TableMapEventMetadata metadata = event.getEventMetadata(); + BitSet unsignedColumns = new BitSet(); + unsignedColumns.set(0); + metadata.setSignedness(unsignedColumns); + metadata.setColumnCharsets(Arrays.asList(255, 63)); // utf8mb4, binary + metadata.setEnumStrValues(Collections.singletonList(new String[] { "new", "done" })); + metadata.setSetStrValues(Collections.singletonList(new String[] { "a", "b" })); + metadata.setGeometryTypes(Collections.singletonList(1)); + metadata.setSimplePrimaryKeys(Collections.singletonList(0)); + + Table table = BinlogTableMetadata.buildTable(event); + + assertEquals("orders", table.getName()); + assertEquals(Collections.singletonList("unsigned_id"), table.getPKList()); + assertFalse(((IntColumnDef) table.findColumn(0)).isSigned()); + assertTrue(((BigIntColumnDef) table.findColumn(1)).isSigned()); + assertEquals("utf8mb4", ((StringColumnDef) table.findColumn(2)).getCharset()); + assertEquals("blob", table.findColumn(3).getType()); + assertEquals("binary", ((StringColumnDef) table.findColumn(3)).getCharset()); + assertEquals(Arrays.asList("new", "done"), ((EnumeratedColumnDef) table.findColumn(4)).getEnumValues()); + assertEquals(Arrays.asList("a", "b"), ((EnumeratedColumnDef) table.findColumn(5)).getEnumValues()); + assertEquals(Long.valueOf(6), ((ColumnDefWithLength) table.findColumn(6)).getColumnLength()); + assertEquals("point", table.findColumn(7).getType()); + } + + @Test + public void supportsDefaultCharsetWithPerColumnOverride() { + TableMapEventData event = event( + new ColumnType[] { ColumnType.LONG, ColumnType.VARCHAR, ColumnType.BLOB }, + new int[] { 0, 100, 2 }, + "number_value", "text_value", "binary_value" + ); + event.getEventMetadata().setSignedness(new BitSet()); + + TableMapEventMetadata.DefaultCharset defaultCharset = new TableMapEventMetadata.DefaultCharset(); + defaultCharset.setDefaultCharsetCollation(255); + Map overrides = new LinkedHashMap<>(); + overrides.put(1, 63); + defaultCharset.setCharsetCollations(overrides); + event.getEventMetadata().setDefaultCharset(defaultCharset); + + Table table = BinlogTableMetadata.buildTable(event); + assertEquals("varchar", table.findColumn(1).getType()); + assertEquals("blob", table.findColumn(2).getType()); + } + + @Test + public void rejectsTableMapWithoutFullMetadata() { + TableMapEventData event = new TableMapEventData(); + event.setDatabase("homs"); + event.setTable("orders"); + event.setColumnTypes(new byte[] { (byte) ColumnType.LONG.getCode() }); + event.setColumnMetadata(new int[] { 0 }); + + IllegalStateException error = assertThrows( + IllegalStateException.class, + () -> BinlogTableMetadata.buildTable(event) + ); + assertTrue(error.getMessage().contains("binlog_row_metadata=FULL")); + } + + @Test + public void tableCacheReplacesDefinitionOnEveryTableMap() { + TableCache cache = new TableCache("maxwell"); + Filter filter = new Filter(); + + TableMapEventData first = integerEvent("old_name"); + TableMapEventData second = integerEvent("new_name"); + cache.processEvent(first, filter); + cache.processEvent(second, filter); + + assertEquals("new_name", cache.getTable(42L).findColumn(0).getName()); + } + + private static TableMapEventData integerEvent(String columnName) { + TableMapEventData event = event(new ColumnType[] { ColumnType.LONG }, new int[] { 0 }, columnName); + event.setTableId(42L); + event.getEventMetadata().setSignedness(new BitSet()); + return event; + } + + private static TableMapEventData event(ColumnType[] types, int[] typeMetadata, String... names) { + TableMapEventData event = new TableMapEventData(); + event.setTableId(42L); + event.setDatabase("homs"); + event.setTable("orders"); + byte[] typeCodes = new byte[types.length]; + for (int i = 0; i < types.length; i++) + typeCodes[i] = (byte) types[i].getCode(); + event.setColumnTypes(typeCodes); + event.setColumnMetadata(typeMetadata); + + TableMapEventMetadata metadata = new TableMapEventMetadata(); + metadata.setColumnNames(Arrays.asList(names)); + event.setEventMetadata(metadata); + return event; + } +}