MysqlSchemaManager implements MysqlSchemaArchiver { public void processDDL(QueryEvent event, String gtid) { String sql = event.getSql(); BinlogFilePos pos = event.getBinlogFilePos(); String database = event.getDatabase(); if (!isSchemaVersionEnabled) { log.info("Skip processing DDL {} because schema versioning is not enabled.", sql); return; } if (!shouldProcessDDL(sql)) { log.info("Not processing DDL {} because it is not our interest.", sql); return; } List<MysqlTableSchema> schemas = gtid == null ? schemaStore.queryByBinlogFilePos(pos) : schemaStore.queryByGTID(gtid); if (!schemas.isEmpty()) { log.info("DDL {} is already processed at BinlogFilePos: {}, GTID: {}", sql, pos, gtid); schemas.forEach(schemaStore::updateSchemaCache); return; } schemaDatabase.applyDDL(sql, DATABASE_DDL_SQL_PATTERN.matcher(sql).find() ? null : database); Set<String> databasesInSchemaStore = ImmutableSet.copyOf(schemaStore.getSchemaCache().rowKeySet()); Set<String> databasesInSchemaDatabase = ImmutableSet.copyOf(schemaDatabase.listDatabases()); boolean isTableColumnsChanged = false; for (String newDatabase : Sets.difference(databasesInSchemaDatabase, databasesInSchemaStore)) { boolean isColumnChangedForNewDB = processTableSchemaChanges( newDatabase, event, gtid, Collections.emptyMap(), schemaDatabase.getColumnsForAllTables(newDatabase)); isTableColumnsChanged = isTableColumnsChanged || isColumnChangedForNewDB; } for (String existingDatbase : databasesInSchemaStore) { boolean isColumnChangedForExistingDB = processTableSchemaChanges( existingDatbase, event, gtid, schemaStore.getSchemaCache().row(existingDatbase), schemaDatabase.getColumnsForAllTables(existingDatbase)); isTableColumnsChanged = isTableColumnsChanged || isColumnChangedForExistingDB; } if (!isTableColumnsChanged) { schemaStore.put( new MysqlTableSchema( 0, database, null, pos, gtid, sql, event.getTimestamp(), Collections.emptyList(), Collections.emptyMap())); } }  List<MysqlColumn> getTableColumns(String database, String table); void processDDL(QueryEvent event, String gtid); void initialize(BinlogFilePos pos); @Override void archive(); void compress();  }
@Test public void testCreateTable() { String sql = "CREATE TABLE table123 (uid INT)"; QueryEvent event = new QueryEvent(1L, TIMESTAMP, BINLOG_FILE_POS, "db1", sql); List<MysqlColumn> newTableColumns = Collections.singletonList(new MysqlColumn("uid", "int", "int", false)); when(schemaDatabase.getColumnsForAllTables("db1")) .thenReturn( ImmutableMap.of( "table1", TABLE1_COLUMNS, "table2", TABLE2_COLUMNS, "table123", newTableColumns)); schemaManager.processDDL(event, GTID); verify(schemaDatabase).applyDDL(sql, "db1"); verify(schemaStore) .put( new MysqlTableSchema( 0, "db1", "table123", BINLOG_FILE_POS, GTID, sql, TIMESTAMP, newTableColumns, Collections.emptyMap())); }
@Test public void testDropTable() { String sql = "DROP TABLE db1.table2"; QueryEvent event = new QueryEvent(1L, TIMESTAMP, BINLOG_FILE_POS, "db1", sql); when(schemaDatabase.getColumnsForAllTables("db1")) .thenReturn(ImmutableMap.of("table1", TABLE1_COLUMNS)); schemaManager.processDDL(event, GTID); verify(schemaDatabase).applyDDL(sql, "db1"); verify(schemaStore) .put( new MysqlTableSchema( 0, "db1", "table2", BINLOG_FILE_POS, GTID, sql, TIMESTAMP, Collections.emptyList(), Collections.emptyMap())); }
@Test public void testAlterTable() { String sql = "ALTER TABLE db1.table1 ADD account VARCHAR(255)"; QueryEvent event = new QueryEvent(1L, TIMESTAMP, BINLOG_FILE_POS, "db1", sql); List<MysqlColumn> newTableColumns = Arrays.asList( new MysqlColumn("id", "int", "int", false), new MysqlColumn("account", "varchar", "varchar(255)", false)); when(schemaDatabase.getColumnsForAllTables("db1")) .thenReturn(ImmutableMap.of("table1", newTableColumns, "table2", TABLE2_COLUMNS)); schemaManager.processDDL(event, GTID); verify(schemaDatabase).applyDDL(sql, "db1"); verify(schemaStore) .put( new MysqlTableSchema( 0, "db1", "table1", BINLOG_FILE_POS, GTID, sql, TIMESTAMP, newTableColumns, Collections.emptyMap())); }
@Test public void testRenameTable() { String sql = "RENAME TABLE db2.table3 TO db2.table4"; QueryEvent event = new QueryEvent(1L, TIMESTAMP, BINLOG_FILE_POS, "db2", sql); when(schemaDatabase.getColumnsForAllTables("db2")) .thenReturn(ImmutableMap.of("table4", TABLE3_COLUMNS)); schemaManager.processDDL(event, GTID); verify(schemaDatabase).applyDDL(sql, "db2"); verify(schemaStore) .put( new MysqlTableSchema( 0, "db2", "table3", BINLOG_FILE_POS, GTID, sql, TIMESTAMP, Collections.emptyList(), Collections.emptyMap())); verify(schemaStore) .put( new MysqlTableSchema( 0, "db2", "table4", BINLOG_FILE_POS, GTID, sql, TIMESTAMP, TABLE3_COLUMNS, Collections.emptyMap())); }
@Test public void testCreateDatabase() { String sql = "CREATE DATABASE db3"; QueryEvent event = new QueryEvent(1L, TIMESTAMP, BINLOG_FILE_POS, "db3", sql); when(schemaDatabase.listDatabases()).thenReturn(ImmutableList.of("db1", "db2", "db3")); when(schemaDatabase.getColumnsForAllTables("db3")).thenReturn(Collections.emptyMap()); schemaManager.processDDL(event, GTID); verify(schemaDatabase).applyDDL(sql, null); verify(schemaStore) .put( new MysqlTableSchema( 0, "db3", null, BINLOG_FILE_POS, GTID, sql, TIMESTAMP, Collections.emptyList(), Collections.emptyMap())); }
@Test public void testDropDatabase() { String sql = "DROP DATABASE db1"; QueryEvent event = new QueryEvent(1, TIMESTAMP, BINLOG_FILE_POS, null, sql); when(schemaDatabase.listDatabases()).thenReturn(ImmutableList.of("db2")); when(schemaDatabase.getColumnsForAllTables("db1")).thenReturn(Collections.emptyMap()); schemaManager.processDDL(event, GTID); verify(schemaDatabase).applyDDL(sql, null); verify(schemaStore) .put( new MysqlTableSchema( 0, "db1", "table1", BINLOG_FILE_POS, GTID, sql, TIMESTAMP, Collections.emptyList(), Collections.emptyMap())); verify(schemaStore) .put( new MysqlTableSchema( 0, "db1", "table2", BINLOG_FILE_POS, GTID, sql, TIMESTAMP, Collections.emptyList(), Collections.emptyMap())); }