From 2783cac6ba0b905a48f77f883daf82ad10b06215 Mon Sep 17 00:00:00 2001
From: Robin Tang <rtang.cs@gmail.com>
Date: Thu, 19 Dec 2024 13:01:22 -0800
Subject: [PATCH] More MySQL Improvements (#621)

---
 lib/antlr/parse.go                 | 2 ++
 lib/mysql/schema/schema.go         | 8 +++++---
 lib/mysql/schema/schema_test.go    | 7 +++++++
 sources/mysql/streaming/ddl/ddl.go | 6 +++++-
 4 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/lib/antlr/parse.go b/lib/antlr/parse.go
index 7ab31cef..6446f5c5 100644
--- a/lib/antlr/parse.go
+++ b/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,
diff --git a/lib/mysql/schema/schema.go b/lib/mysql/schema/schema.go
index 69113b7d..f2aab05f 100644
--- a/lib/mysql/schema/schema.go
+++ b/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, "(")
diff --git a/lib/mysql/schema/schema_test.go b/lib/mysql/schema/schema_test.go
index 53dc4d76..82febcb0 100644
--- a/lib/mysql/schema/schema_test.go
+++ b/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
diff --git a/sources/mysql/streaming/ddl/ddl.go b/sources/mysql/streaming/ddl/ddl.go
index 81c26b64..c346517f 100644
--- a/sources/mysql/streaming/ddl/ddl.go
+++ b/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{