diff --git a/crates/dbx-core/src/table_structure_sql/create_table.rs b/crates/dbx-core/src/table_structure_sql/create_table.rs
index ae6a8df09f..1989550960 100644
--- a/crates/dbx-core/src/table_structure_sql/create_table.rs
+++ b/crates/dbx-core/src/table_structure_sql/create_table.rs
@@ -1,5 +1,6 @@
 use super::column_format::{
     column_data_type, column_extra_clause, has_dameng_identity, is_dameng_identity_compatible_type,
+    is_mysql_character_data_type,
 };
 use super::comments::{build_sqlserver_column_comment_sql, build_sqlserver_table_comment_sql};
 use super::dialect::{capabilities_for, database_label, StructureDialect};
@@ -9,6 +10,7 @@ use super::triggers::build_trigger_sql;
 use super::types::{TableStructureSqlOptions, TableStructureSqlResult};
 use super::util::{clean, format_default_for_sql, normalize_default, qualified_table, quote_ident, quote_string};
 use super::validation::{validate_columns, validate_dameng_identity};
+use crate::models::connection::DatabaseType;
 
 pub fn build_create_table_sql(mut options: TableStructureSqlOptions) -> TableStructureSqlResult {
     options.table_name = clean(&options.table_name);
@@ -48,6 +50,14 @@ pub fn build_create_table_sql(mut options: TableStructureSqlOptions) -> TableStr
     for column in &active_columns {
         let data_type = column_data_type(dialect, column);
         let mut parts = vec![quote_ident(dialect, &column.name), data_type];
+        if options.database_type == Some(DatabaseType::Mysql) && is_mysql_character_data_type(&column.data_type) {
+            if !column.character_set.trim().is_empty() {
+                parts.push(format!("CHARACTER SET {}", quote_ident(dialect, &column.character_set)));
+            }
+            if !column.collation.trim().is_empty() {
+                parts.push(format!("COLLATE {}", quote_ident(dialect, &column.collation)));
+            }
+        }
         if !column.is_nullable
             && !column.is_primary_key
             && !matches!(dialect, StructureDialect::ClickHouse | StructureDialect::ManticoreSearch)
diff --git a/crates/dbx-core/src/table_structure_sql/tests.rs b/crates/dbx-core/src/table_structure_sql/tests.rs
index 84964e3805..f08f701d09 100644
--- a/crates/dbx-core/src/table_structure_sql/tests.rs
+++ b/crates/dbx-core/src/table_structure_sql/tests.rs
@@ -2756,6 +2756,61 @@ fn mysql_create_table_with_auto_increment() {
     assert!(result.statements[0].contains("AUTO_INCREMENT"));
 }
 
+#[test]
+fn mysql_create_table_keeps_column_charset_collation_and_comment() {
+    let mut name = column("name");
+    name.data_type = "varchar(255)".to_string();
+    name.character_set = "gbk".to_string();
+    name.collation = "gbk_bin".to_string();
+    name.comment = "测试".to_string();
+
+    let result = build_create_table_sql(TableStructureSqlOptions {
+        database_type: Some(DatabaseType::Mysql),
+        schema: None,
+        table_name: "users".to_string(),
+        columns: vec![name],
+        indexes: Vec::new(),
+        foreign_keys: Vec::new(),
+        triggers: Vec::new(),
+        table_comment: Some("User accounts".to_string()),
+        original_table_comment: None,
+    });
+
+    assert_eq!(result.warnings, Vec::<String>::new());
+    assert_eq!(
+        result.statements,
+        vec![
+            "CREATE TABLE `users` (\n  `name` varchar(255) CHARACTER SET `gbk` COLLATE `gbk_bin` COMMENT '测试'\n) COMMENT = 'User accounts';"
+        ]
+    );
+}
+
+#[test]
+fn mysql_compatible_databases_do_not_emit_mysql_column_charset_clauses() {
+    for database_type in [DatabaseType::StarRocks, DatabaseType::Databend, DatabaseType::Gbase] {
+        let mut name = column("name");
+        name.data_type = "varchar(255)".to_string();
+        name.character_set = "utf8mb4".to_string();
+        name.collation = "utf8mb4_bin".to_string();
+
+        let result = build_create_table_sql(TableStructureSqlOptions {
+            database_type: Some(database_type),
+            schema: None,
+            table_name: "users".to_string(),
+            columns: vec![name],
+            indexes: Vec::new(),
+            foreign_keys: Vec::new(),
+            triggers: Vec::new(),
+            table_comment: None,
+            original_table_comment: None,
+        });
+
+        assert_eq!(result.warnings, Vec::<String>::new());
+        assert!(!result.statements[0].contains("CHARACTER SET"));
+        assert!(!result.statements[0].contains("COLLATE"));
+    }
+}
+
 #[test]
 fn mysql_create_table_with_on_update_current_timestamp() {
     let mut col = column("updated_at");