diff --git a/src/main/java/io/debezium/connector/vitess/connection/DdlMetadataExtractor.java b/src/main/java/io/debezium/connector/vitess/connection/DdlMetadataExtractor.java
index 4ffd7848..2ced72a3 100644
--- a/src/main/java/io/debezium/connector/vitess/connection/DdlMetadataExtractor.java
+++ b/src/main/java/io/debezium/connector/vitess/connection/DdlMetadataExtractor.java
@@ -24,12 +24,15 @@ public class DdlMetadataExtractor {
 
     // VStream DDL statements do not contain any database/keyspace, only contains the table name
     private static final Pattern TABLE_NAME_PATTERN = Pattern.compile(
-            "(?i)(CREATE|ALTER|TRUNCATE|DROP|RENAME)\\s+TABLE\\s+['\\\"`]?([\\w]+)['\\\"`]?",
+            "(CREATE|ALTER|TRUNCATE|DROP|RENAME)\\s+TABLE\\s+([\\w`\\.]+)",
             Pattern.CASE_INSENSITIVE);
 
     // Regex to match in-line or multi-line comments (e.g., /* comment */)
     private static final Pattern COMMENT_PATTERN = Pattern.compile("/\\*.*?\\*/", Pattern.DOTALL);
 
+    // Regex to match single-line comments (e.g., -- comment and # comment)
+    private static final Pattern SINGLE_LINE_COMMENT_PATTERN = Pattern.compile("(--|#).*?(\r?\n|$)");
+
     private static final String UNKNOWN_TABLE_NAME = "<UNKNOWN>";
 
     private final DdlMessage ddlMessage;
@@ -49,12 +52,20 @@ public void extractMetadata() {
             if (operation.equals("RENAME")) {
                 operation = "ALTER";
             }
-            table = matcher.group(2);
+            String tableName = matcher.group(2);
+            if (tableName.contains(".")) {
+                String[] parts = tableName.split("\\.");
+                tableName = parts[1];
+            }
+            table = tableName.replaceAll("`", "");
         }
     }
 
     private String removeComments(String statement) {
-        return COMMENT_PATTERN.matcher(statement).replaceAll("");
+        statement = COMMENT_PATTERN.matcher(statement).replaceAll("");
+        statement = SINGLE_LINE_COMMENT_PATTERN.matcher(statement).replaceAll("");
+        statement = statement.replaceAll("\\s+", " ").trim();
+        return statement;
     }
 
     public SchemaChangeEvent.SchemaChangeEventType getSchemaChangeEventType() {
diff --git a/src/test/java/io/debezium/connector/vitess/connection/DdlMetadataExtractorTest.java b/src/test/java/io/debezium/connector/vitess/connection/DdlMetadataExtractorTest.java
index e33b107e..251d9688 100644
--- a/src/test/java/io/debezium/connector/vitess/connection/DdlMetadataExtractorTest.java
+++ b/src/test/java/io/debezium/connector/vitess/connection/DdlMetadataExtractorTest.java
@@ -11,6 +11,7 @@
 import org.junit.Test;
 
 import io.debezium.connector.vitess.TestHelper;
+import io.debezium.connector.vitess.VitessDatabaseSchema;
 import io.debezium.junit.logging.LogInterceptor;
 import io.debezium.schema.SchemaChangeEvent;
 
@@ -19,61 +20,87 @@
  */
 public class DdlMetadataExtractorTest {
 
+    private static String expectedTableName = VitessDatabaseSchema.buildTableId(
+            TestHelper.TEST_SHARD, TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_TABLE).toDoubleQuotedString();
+
     @Test
     public void shouldGetAlterType() {
-        DdlMessage ddlMessage = new DdlMessage(null, null, "ALTER TABLE foo ADD COLUMN bar",
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("ALTER TABLE %s ADD COLUMN bar", TestHelper.TEST_TABLE),
                 TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
         DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
         assertThat(extractor.getSchemaChangeEventType()).isEqualTo(SchemaChangeEvent.SchemaChangeEventType.ALTER);
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
     }
 
     @Test
     public void shouldGetCreateType() {
-        DdlMessage ddlMessage = new DdlMessage(null, null, "CREATE    TABLE foo",
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("CREATE    TABLE %s", TestHelper.TEST_TABLE),
                 TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
         DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
         assertThat(extractor.getSchemaChangeEventType()).isEqualTo(SchemaChangeEvent.SchemaChangeEventType.CREATE);
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
     }
 
     @Test
     public void shouldGetTruncateType() {
-        DdlMessage ddlMessage = new DdlMessage(null, null, "TRUNCATE    TABLE foo",
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("TRUNCATE    TABLE %s", TestHelper.TEST_TABLE),
                 TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
         DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
         assertThat(extractor.getSchemaChangeEventType()).isEqualTo(SchemaChangeEvent.SchemaChangeEventType.TRUNCATE);
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
     }
 
     @Test
     public void shouldGetTable() {
-        DdlMessage ddlMessage = new DdlMessage(null, null, "TRUNCATE    TABLE foo",
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("TRUNCATE    TABLE %s", TestHelper.TEST_TABLE),
                 TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
         DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
-        assertThat(extractor.getTable()).isEqualTo("\"0\".\"test_unsharded_keyspace\".\"foo\"");
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
     }
 
     @Test
     public void shouldGetDropType() {
-        DdlMessage ddlMessage = new DdlMessage(null, null, "DROP TABLE foo",
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("DROP TABLE %s", TestHelper.TEST_TABLE),
                 TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
         DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
         assertThat(extractor.getSchemaChangeEventType()).isEqualTo(SchemaChangeEvent.SchemaChangeEventType.DROP);
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
     }
 
     @Test
     public void shouldGetRenameType() {
-        DdlMessage ddlMessage = new DdlMessage(null, null, "RENAME TABLE foo TO bar",
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("RENAME TABLE %s TO %s_suffix", TestHelper.TEST_TABLE, TestHelper.TEST_TABLE),
                 TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
         DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
         assertThat(extractor.getSchemaChangeEventType()).isEqualTo(SchemaChangeEvent.SchemaChangeEventType.ALTER);
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
     }
 
     @Test
     public void shouldParseStatementWithComments() {
-        DdlMessage ddlMessage = new DdlMessage(null, null, "rename /* gh-ost */ table `keyspace`.`table1` to " +
-                "`keyspace`.`_table1_del`, `keyspace`.`_table_gho` to `keyspace`.`table`",
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("rename /* gh-ost */ table `keyspace`.`%s` to `keyspace`.`_table1_del`, `keyspace`.`_table_gho` to `keyspace`.`table`",
+                        TestHelper.TEST_TABLE),
                 TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
         DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
         assertThat(extractor.getSchemaChangeEventType()).isEqualTo(SchemaChangeEvent.SchemaChangeEventType.ALTER);
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
+    }
+
+    @Test
+    public void shouldParseStatementWithSingleLineComment() {
+        DdlMessage ddlMessage = new DdlMessage(null, null,
+                String.format("create \n # gh-ost \n table `keyspace`.`%s`;", TestHelper.TEST_TABLE),
+                TestHelper.TEST_UNSHARDED_KEYSPACE, TestHelper.TEST_SHARD);
+        DdlMetadataExtractor extractor = new DdlMetadataExtractor(ddlMessage);
+        assertThat(extractor.getSchemaChangeEventType()).isEqualTo(SchemaChangeEvent.SchemaChangeEventType.CREATE);
+        assertThat(extractor.getTable()).isEqualTo(expectedTableName);
     }
 
     @Test