[MySQL] Retrieve all tables in a particular schema (#627)

lib/mysql/schema/schema.go

@@ -70,6 +70,30 @@ func QuoteIdentifier(s string) string {
 	return fmt.Sprintf("`%s`", strings.ReplaceAll(s, "`", "``"))
 }
 
+func ListTables(db *sql.DB, dbName string) ([]string, error) {
+	rows, err := db.Query(fmt.Sprintf("SHOW FULL TABLES IN %s WHERE table_type = 'BASE TABLE'", QuoteIdentifier(dbName)))
+	if err != nil {
+		return nil, fmt.Errorf("failed to list tables: %w", err)
+	}
+
+	defer rows.Close()
+	var tables []string
+	for rows.Next() {
+		var table string
+		var unused string
+		if err = rows.Scan(&table, &unused); err != nil {
+			return nil, fmt.Errorf("failed to scan: %w", err)
+		}
+		tables = append(tables, table)
+	}
+
+	if err = rows.Err(); err != nil {
+		return nil, fmt.Errorf("failed to iterate over rows: %w", err)
+	}
+
+	return tables, nil
+}
+
 func GetCreateTableDDL(db *sql.DB, table string) (string, error) {
 	row := db.QueryRow("SHOW CREATE TABLE " + QuoteIdentifier(table))
 	var unused string

sources/mysql/streaming/iterator.go

@@ -39,11 +39,17 @@ func buildSchemaAdapter(db *sql.DB, cfg config.MySQL, schemaHistoryList persiste
 		return ddl.SchemaAdapter{}, fmt.Errorf("latest schema timestamp %d is greater than the current position's timestamp %d", latestSchemaUnixTs, pos.UnixTs)
 	}
 
-	// Check if there are any additional tables that we should be tracking that don't exist in our schema adapter
-	for _, tbl := range cfg.Tables {
-		if _, ok := schemaAdapter.GetTableAdapter(tbl.Name); !ok {
+	// Find all the tables in the schema, check if they are already in the schema adapter
+	// If not, then call [GetCreateTableDDL] to get the DDL and apply it to the schema adapter
+	tables, err := schema.ListTables(db, cfg.Database)
+	if err != nil {
+		return ddl.SchemaAdapter{}, fmt.Errorf("failed to list tables: %w", err)
+	}
+
+	for _, tbl := range tables {
+		if _, ok := schemaAdapter.GetTableAdapter(tbl); !ok {
 			now := time.Now().Unix()
-			ddlQuery, err := schema.GetCreateTableDDL(db, tbl.Name)
+			ddlQuery, err := schema.GetCreateTableDDL(db, tbl)
 			if err != nil {
 				return ddl.SchemaAdapter{}, fmt.Errorf("failed to get columns: %w", err)
 			}