More MySQL Improvements (#621)

lib/antlr/parse.go

@@ -86,6 +86,8 @@ func visit(tree antlr.Tree) ([]Event, error) {
 	case *generated.RenameTableContext:
 		return processRenameTable(ctx)
 	case
+		*generated.CreateEventContext,
+		*generated.DropEventContext,
 		*generated.EmptyStatement_Context,
 		*generated.TruncateTableContext,
 		*generated.AdministrationStatementContext,

lib/mysql/schema/schema.go

@@ -131,9 +131,11 @@ func ParseColumnDataType(originalS string, optionalSQLMode []string) (DataType,
 		s = s[:collateIdx]
 	}
 
-	if charSetIdx := strings.Index(s, " character set"); charSetIdx != -1 {
-		// Strip character set
-		s = s[:charSetIdx]
+	for _, charSetPermutation := range []string{" character set", " charset"} {
+		if charSetIdx := strings.Index(s, charSetPermutation); charSetIdx != -1 {
+			// Strip character set
+			s = s[:charSetIdx]
+		}
 	}
 
 	parenIndex := strings.Index(s, "(")

lib/mysql/schema/schema_test.go

@@ -116,6 +116,13 @@ func TestParseColumnDataType(t *testing.T) {
 			assert.Equal(t, Varchar, dataType)
 			assert.Equal(t, &Opts{Size: typing.ToPtr(255)}, opts)
 		}
+		{
+			// Using charset instead
+			dataType, opts, err := ParseColumnDataType(`varchar(255) CHARSET utf8mb3 COLLATE utf8mb3_unicode_ci`, nil)
+			assert.NoError(t, err)
+			assert.Equal(t, Varchar, dataType)
+			assert.Equal(t, &Opts{Size: typing.ToPtr(255)}, opts)
+		}
 	}
 	{
 		// Decimal

sources/mysql/streaming/ddl/ddl.go

@@ -55,7 +55,11 @@ func (s *SchemaAdapter) ApplyDDL(unixTs int64, query string) error {
 }
 
 func (s *SchemaAdapter) applyDDL(unixTs int64, result antlr.Event) error {
-	if _, ok := result.(antlr.CreateTableEvent); ok {
+	switch result.(type) {
+	case antlr.DropTableEvent:
+		delete(s.adapters, result.GetTable())
+		return nil
+	case antlr.CreateTableEvent:
 		var cols []Column
 		for _, col := range result.GetColumns() {
 			cols = append(cols, Column{