Merge pull request #2252 from hemanth132/fix/schema-restore-binlog-numeric-sort

Fix schema restore selecting stale snapshot due to lexicographic binlog file ordering

Author: osheroff

src/main/java/com/zendesk/maxwell/schema/MysqlSavedSchema.java

@@ -592,29 +592,65 @@ private static Long findSchema(Connection connection, Position targetPosition, L
 		} else {
 			// Only consider binlog positions before the target position on the current server.
 			// Within those, sort for the latest binlog file, then the latest binlog position.
-			String sql = "SELECT id from `schemas` "
-					+ "WHERE deleted = 0 "
-					+ "AND last_heartbeat_read <= ? AND ("
-					+ "(binlog_file < ?) OR "
-					+ "(binlog_file = ? and binlog_position < ? and base_schema_id is not null) OR "
-					+ "(binlog_file = ? and binlog_position <= ? and base_schema_id is null) "
-					+ ") AND server_id = ? "
-					+ "ORDER BY last_heartbeat_read DESC, binlog_file DESC, binlog_position DESC limit 1";
-			try ( PreparedStatement s = connection.prepareStatement(sql) ) {
-
-				s.setLong(1, targetPosition.getLastHeartbeatRead());
-				s.setString(2, targetBinlogPosition.getFile());
-				s.setString(3, targetBinlogPosition.getFile());
-				s.setLong(4, targetBinlogPosition.getOffset());
-				s.setString(5, targetBinlogPosition.getFile());
-				s.setLong(6, targetBinlogPosition.getOffset());
-				s.setLong(7, serverID);
-
-				try ( ResultSet rs = s.executeQuery() ) {
-					if (rs.next()) {
-						return rs.getLong("id");
-					} else
-						return null;
+			//
+			// Binlog file names contain a numeric suffix (e.g. "mysql-bin-changelog.1215501").
+			// When that suffix is parseable, we compare and sort numerically via CAST to avoid
+			// lexicographic ordering bugs where a shorter number string like "122046" compares
+			// greater than a longer one like "1215501". If the suffix cannot be parsed we fall
+			// back to string comparison so that non-standard formats still work.
+			Long targetFileNumber = targetBinlogPosition.getFileNumber();
+			if ( targetFileNumber != null ) {
+				String sql = "SELECT id from `schemas` "
+						+ "WHERE deleted = 0 "
+						+ "AND last_heartbeat_read <= ? AND ("
+						+ "(CAST(SUBSTRING_INDEX(binlog_file, '.', -1) AS UNSIGNED) < ?) OR "
+						+ "(binlog_file = ? and binlog_position < ? and base_schema_id is not null) OR "
+						+ "(binlog_file = ? and binlog_position <= ? and base_schema_id is null) "
+						+ ") AND server_id = ? "
+						+ "ORDER BY last_heartbeat_read DESC, "
+						+ "CAST(SUBSTRING_INDEX(binlog_file, '.', -1) AS UNSIGNED) DESC, "
+						+ "binlog_position DESC limit 1";
+				try ( PreparedStatement s = connection.prepareStatement(sql) ) {
+					s.setLong(1, targetPosition.getLastHeartbeatRead());
+					s.setLong(2, targetFileNumber);
+					s.setString(3, targetBinlogPosition.getFile());
+					s.setLong(4, targetBinlogPosition.getOffset());
+					s.setString(5, targetBinlogPosition.getFile());
+					s.setLong(6, targetBinlogPosition.getOffset());
+					s.setLong(7, serverID);
+
+					try ( ResultSet rs = s.executeQuery() ) {
+						if (rs.next()) {
+							return rs.getLong("id");
+						} else
+							return null;
+					}
+				}
+			} else {
+				// Fallback: file name has no parseable numeric suffix; use string comparison.
+				String sql = "SELECT id from `schemas` "
+						+ "WHERE deleted = 0 "
+						+ "AND last_heartbeat_read <= ? AND ("
+						+ "(binlog_file < ?) OR "
+						+ "(binlog_file = ? and binlog_position < ? and base_schema_id is not null) OR "
+						+ "(binlog_file = ? and binlog_position <= ? and base_schema_id is null) "
+						+ ") AND server_id = ? "
+						+ "ORDER BY last_heartbeat_read DESC, binlog_file DESC, binlog_position DESC limit 1";
+				try ( PreparedStatement s = connection.prepareStatement(sql) ) {
+					s.setLong(1, targetPosition.getLastHeartbeatRead());
+					s.setString(2, targetBinlogPosition.getFile());
+					s.setString(3, targetBinlogPosition.getFile());
+					s.setLong(4, targetBinlogPosition.getOffset());
+					s.setString(5, targetBinlogPosition.getFile());
+					s.setLong(6, targetBinlogPosition.getOffset());
+					s.setLong(7, serverID);
+
+					try ( ResultSet rs = s.executeQuery() ) {
+						if (rs.next()) {
+							return rs.getLong("id");
+						} else
+							return null;
+					}
 				}
 			}
 		}

src/test/java/com/zendesk/maxwell/MysqlSavedSchemaTest.java

@@ -220,6 +220,55 @@ public void testTableWithSameNameInTwoDBs() throws Exception {
 	}
 
 
+	/**
+	 * Regression test for https://github.com/zendesk/maxwell/issues/2251.
+	 *
+	 * Binlog file numbers are variable-length on Aurora/RDS. String (lexicographic)
+	 * comparison of file names produces wrong ordering when numbers cross a digit-count
+	 * boundary: e.g. "mysql-bin-changelog.122046" sorts AFTER "mysql-bin-changelog.1215501"
+	 * as a string even though 122046 < 1215501 numerically. This causes findSchema to pick
+	 * a stale snapshot, skipping all deltas recorded at those incorrectly-ordered files.
+	 */
+	@Test
+	public void testFindSchemaUsesNumericFileOrdering() throws Exception {
+		if (context.getConfig().gtidMode) {
+			return;
+		}
+
+		Connection c = context.getMaxwellConnection();
+		long serverId = 200;
+		long heartbeat = 0L;
+
+		// Target position: 7-digit file number
+		Position targetPosition = makePosition(500L, "mysql-bin-changelog.1215501", heartbeat + 1);
+
+		// This schema is at a 6-digit file that sorts AFTER the target under string comparison
+		// ("122046" > "1215501" lexicographically) but is numerically BEFORE it.
+		// It must be excluded from the result.
+		new MysqlSavedSchema(serverId, caseSensitivity, buildSchema(),
+				makePosition(100L, "mysql-bin-changelog.122046", heartbeat)
+		).saveSchema(c);
+
+		// This schema is at a 7-digit file numerically closest to the target and before it.
+		// It should be selected.
+		MysqlSavedSchema expectedSchema = new MysqlSavedSchema(serverId, caseSensitivity, buildSchema(),
+				makePosition(200L, "mysql-bin-changelog.1203422", heartbeat)
+		);
+		expectedSchema.save(c);
+
+		// This schema is at a 6-digit file that is also numerically before the target but
+		// older than expectedSchema — it must not be preferred over expectedSchema.
+		new MysqlSavedSchema(serverId, caseSensitivity, buildSchema(),
+				makePosition(300L, "mysql-bin-changelog.121345", heartbeat)
+		).saveSchema(c);
+
+		MysqlSavedSchema foundSchema = MysqlSavedSchema.restore(
+				context.getMaxwellConnectionPool(), serverId, caseSensitivity, targetPosition);
+
+		assertThat(foundSchema.getBinlogPosition(), equalTo(expectedSchema.getBinlogPosition()));
+		assertThat(foundSchema.getSchemaID(), equalTo(expectedSchema.getSchemaID()));
+	}
+
 	private Schema buildSchema() {
 		String charset = Charset.defaultCharset().toString();
 		List<Database> databases = new ArrayList<>();