take the benefit from using binlog_row_metadata=FULL

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

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.*"
 }
 ```
-
-

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);

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.
 	 * <p>
@@ -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");

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

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.");