@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())); }
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())); } }
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())); } } }
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())); } }  }
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(); }
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();  }