✨ feat(frontend/backend): 批量操作与表格编辑增强并完善事务支持

- 批量导出/备份：表与数据库支持全选/反选/智能上下文
  - 右键菜单：单元格菜单支持设置 NULL
  - 编辑优化：大字段弹窗、仅值变化标记、提交只发送差异字段
  - 事务支持：PostgreSQL/SQLite/Oracle/DaMeng/KingBase ApplyChanges
  - MySQL 修复：提交前归一化 datetime，避免写入失败
  - 性能优化：移除 activeCell 重渲染、useRef 存储选中节点、防重加载
  - Redis 优化：二进制智能解码与视图模式切换
  - 资源更新：替换前端 favicon/logo

internal/app/app.go

@@ -10,23 +10,31 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"time"
 
 	"GoNavi-Wails/internal/connection"
 	"GoNavi-Wails/internal/db"
 	"GoNavi-Wails/internal/logger"
 )
 
+const dbCachePingInterval = 30 * time.Second
+
+type cachedDatabase struct {
+	inst     db.Database
+	lastPing time.Time
+}
+
 // App struct
 type App struct {
 	ctx     context.Context
-	dbCache map[string]db.Database // Cache for DB connections
-	mu      sync.Mutex             // Mutex for cache access
+	dbCache map[string]cachedDatabase // Cache for DB connections
+	mu      sync.RWMutex              // Mutex for cache access
 }
 
 // NewApp creates a new App application struct
 func NewApp() *App {
 	return &App{
-		dbCache: make(map[string]db.Database),
+		dbCache: make(map[string]cachedDatabase),
 	}
 }
 
@@ -44,7 +52,7 @@ func (a *App) Shutdown(ctx context.Context) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	for _, dbInst := range a.dbCache {
-		if err := dbInst.Close(); err != nil {
+		if err := dbInst.inst.Close(); err != nil {
 			logger.Error(err, "关闭数据库连接失败")
 		}
 	}
@@ -136,32 +144,63 @@ func formatConnSummary(config connection.ConnectionConfig) string {
 	return b.String()
 }
 
+func (a *App) getDatabaseForcePing(config connection.ConnectionConfig) (db.Database, error) {
+	return a.getDatabaseWithPing(config, true)
+}
+
 // Helper: Get or create a database connection
 func (a *App) getDatabase(config connection.ConnectionConfig) (db.Database, error) {
+	return a.getDatabaseWithPing(config, false)
+}
+
+func (a *App) getDatabaseWithPing(config connection.ConnectionConfig, forcePing bool) (db.Database, error) {
 	key := getCacheKey(config)
 	shortKey := key
 	if len(shortKey) > 12 {
 		shortKey = shortKey[:12]
 	}
-	logger.Infof("获取数据库连接：%s 缓存Key=%s", formatConnSummary(config), shortKey)
 
-	a.mu.Lock()
-	defer a.mu.Unlock()
+	a.mu.RLock()
+	entry, ok := a.dbCache[key]
+	a.mu.RUnlock()
+	if ok {
+		needPing := forcePing
+		if !needPing {
+			lastPing := entry.lastPing
+			if lastPing.IsZero() || time.Since(lastPing) >= dbCachePingInterval {
+				needPing = true
+			}
+		}
 
-	if dbInst, ok := a.dbCache[key]; ok {
-		logger.Infof("命中连接缓存，开始检测可用性：缓存Key=%s", shortKey)
-		if err := dbInst.Ping(); err == nil {
-			logger.Infof("缓存连接可用：缓存Key=%s", shortKey)
-			return dbInst, nil
+		if !needPing {
+			return entry.inst, nil
+		}
+
+		if err := entry.inst.Ping(); err == nil {
+			// Update lastPing (best effort)
+			a.mu.Lock()
+			if cur, exists := a.dbCache[key]; exists && cur.inst == entry.inst {
+				cur.lastPing = time.Now()
+				a.dbCache[key] = cur
+			}
+			a.mu.Unlock()
+			return entry.inst, nil
 		} else {
-			logger.Error(err, "缓存连接不可用，准备重建：缓存Key=%s", shortKey)
+			logger.Error(err, "缓存连接不可用，准备重建：%s 缓存Key=%s", formatConnSummary(config), shortKey)
 		}
-		if err := dbInst.Close(); err != nil {
-			logger.Error(err, "关闭失效缓存连接失败：缓存Key=%s", shortKey)
+
+		// Ping failed: remove cached instance (best effort)
+		a.mu.Lock()
+		if cur, exists := a.dbCache[key]; exists && cur.inst == entry.inst {
+			if err := cur.inst.Close(); err != nil {
+				logger.Error(err, "关闭失效缓存连接失败：缓存Key=%s", shortKey)
+			}
+			delete(a.dbCache, key)
 		}
-		delete(a.dbCache, key)
+		a.mu.Unlock()
 	}
 
+	logger.Infof("获取数据库连接：%s 缓存Key=%s", formatConnSummary(config), shortKey)
 	logger.Infof("创建数据库驱动实例：类型=%s 缓存Key=%s", config.Type, shortKey)
 	dbInst, err := db.NewDatabase(config.Type)
 	if err != nil {
@@ -175,7 +214,18 @@ func (a *App) getDatabase(config connection.ConnectionConfig) (db.Database, erro
 		return nil, wrapped
 	}
 
-	a.dbCache[key] = dbInst
+	now := time.Now()
+
+	a.mu.Lock()
+	if existing, exists := a.dbCache[key]; exists && existing.inst != nil {
+		a.mu.Unlock()
+		// Prefer existing cached connection to avoid cache racing duplicates.
+		_ = dbInst.Close()
+		return existing.inst, nil
+	}
+	a.dbCache[key] = cachedDatabase{inst: dbInst, lastPing: now}
+	a.mu.Unlock()
+
 	logger.Infof("数据库连接成功并写入缓存：%s 缓存Key=%s", formatConnSummary(config), shortKey)
 	return dbInst, nil
 }

internal/app/methods_db.go

@@ -14,8 +14,8 @@ import (
 // Generic DB Methods
 
 func (a *App) DBConnect(config connection.ConnectionConfig) connection.QueryResult {
-	// getDatabase checks cache and Pings. If valid, reuses. If not, connects.
-	_, err := a.getDatabase(config)
+	// 连接测试需要强制 ping，避免缓存命中但连接已失效时误判成功。
+	_, err := a.getDatabaseForcePing(config)
 	if err != nil {
 		logger.Error(err, "DBConnect 连接失败：%s", formatConnSummary(config))
 		return connection.QueryResult{Success: false, Message: err.Error()}
@@ -26,7 +26,7 @@ func (a *App) DBConnect(config connection.ConnectionConfig) connection.QueryResu
 }
 
 func (a *App) TestConnection(config connection.ConnectionConfig) connection.QueryResult {
-	_, err := a.getDatabase(config)
+	_, err := a.getDatabaseForcePing(config)
 	if err != nil {
 		logger.Error(err, "TestConnection 连接测试失败：%s", formatConnSummary(config))
 		return connection.QueryResult{Success: false, Message: err.Error()}

internal/app/methods_file.go

@@ -201,10 +201,10 @@ func (a *App) ApplyChanges(config connection.ConnectionConfig, dbName, tableName
 		if err != nil {
 			return connection.QueryResult{Success: false, Message: err.Error()}
 		}
-		return connection.QueryResult{Success: true, Message: "Changes applied successfully"}
+		return connection.QueryResult{Success: true, Message: "事务提交成功"}
 	}
-	
-	return connection.QueryResult{Success: false, Message: "Batch updates not supported for this database type"}
+
+	return connection.QueryResult{Success: false, Message: "当前数据库类型不支持批量提交"}
 }
 
 func (a *App) ExportTable(config connection.ConnectionConfig, dbName string, tableName string, format string) connection.QueryResult {

internal/db/custom_impl.go

@@ -248,7 +248,141 @@ func (c *CustomDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDe
 }
 
 func (c *CustomDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
-	return fmt.Errorf("read-only mode for custom")
+	if c.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+
+	tx, err := c.conn.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	driver := strings.ToLower(strings.TrimSpace(c.driver))
+	isMySQL := strings.Contains(driver, "mysql")
+	isPostgres := strings.Contains(driver, "postgres") || strings.Contains(driver, "kingbase") || strings.Contains(driver, "pg")
+	isOracle := strings.Contains(driver, "oracle") || strings.Contains(driver, "ora") || strings.Contains(driver, "dm") || strings.Contains(driver, "dameng")
+
+	quoteIdent := func(name string) string {
+		n := strings.TrimSpace(name)
+		if isMySQL {
+			n = strings.Trim(n, "`")
+			n = strings.ReplaceAll(n, "`", "``")
+			if n == "" {
+				return "``"
+			}
+			return "`" + n + "`"
+		}
+		n = strings.Trim(n, "\"")
+		n = strings.ReplaceAll(n, "\"", "\"\"")
+		if n == "" {
+			return "\"\""
+		}
+		return `"` + n + `"`
+	}
+
+	placeholder := func(idx int) string {
+		if isPostgres {
+			return fmt.Sprintf("$%d", idx)
+		}
+		if isOracle {
+			return fmt.Sprintf(":%d", idx)
+		}
+		// MySQL / SQLite / default
+		return "?"
+	}
+
+	schema := ""
+	table := strings.TrimSpace(tableName)
+	if parts := strings.SplitN(table, ".", 2); len(parts) == 2 {
+		schema = strings.TrimSpace(parts[0])
+		table = strings.TrimSpace(parts[1])
+	}
+
+	qualifiedTable := ""
+	if schema != "" {
+		qualifiedTable = fmt.Sprintf("%s.%s", quoteIdent(schema), quoteIdent(table))
+	} else {
+		qualifiedTable = quoteIdent(table)
+	}
+
+	// 1. Deletes
+	for _, pk := range changes.Deletes {
+		var wheres []string
+		var args []interface{}
+		idx := 0
+		for k, v := range pk {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = %s", quoteIdent(k), placeholder(idx)))
+			args = append(args, v)
+		}
+		if len(wheres) == 0 {
+			continue
+		}
+		query := fmt.Sprintf("DELETE FROM %s WHERE %s", qualifiedTable, strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("delete error: %v", err)
+		}
+	}
+
+	// 2. Updates
+	for _, update := range changes.Updates {
+		var sets []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range update.Values {
+			idx++
+			sets = append(sets, fmt.Sprintf("%s = %s", quoteIdent(k), placeholder(idx)))
+			args = append(args, v)
+		}
+
+		if len(sets) == 0 {
+			continue
+		}
+
+		var wheres []string
+		for k, v := range update.Keys {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = %s", quoteIdent(k), placeholder(idx)))
+			args = append(args, v)
+		}
+
+		if len(wheres) == 0 {
+			return fmt.Errorf("update requires keys")
+		}
+
+		query := fmt.Sprintf("UPDATE %s SET %s WHERE %s", qualifiedTable, strings.Join(sets, ", "), strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("update error: %v", err)
+		}
+	}
+
+	// 3. Inserts
+	for _, row := range changes.Inserts {
+		var cols []string
+		var placeholders []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range row {
+			idx++
+			cols = append(cols, quoteIdent(k))
+			placeholders = append(placeholders, placeholder(idx))
+			args = append(args, v)
+		}
+
+		if len(cols) == 0 {
+			continue
+		}
+
+		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", qualifiedTable, strings.Join(cols, ", "), strings.Join(placeholders, ", "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("insert error: %v", err)
+		}
+	}
+
+	return tx.Commit()
 }
 
 func (c *CustomDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {

internal/db/dameng_impl.go

@@ -373,7 +373,117 @@ func (d *DamengDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDe
 }
 
 func (d *DamengDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
-	return fmt.Errorf("read-only mode implemented for Dameng so far")
+	if d.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+
+	tx, err := d.conn.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	quoteIdent := func(name string) string {
+		n := strings.TrimSpace(name)
+		n = strings.Trim(n, "\"")
+		n = strings.ReplaceAll(n, "\"", "\"\"")
+		if n == "" {
+			return "\"\""
+		}
+		return `"` + n + `"`
+	}
+
+	schema := ""
+	table := strings.TrimSpace(tableName)
+	if parts := strings.SplitN(table, ".", 2); len(parts) == 2 {
+		schema = strings.TrimSpace(parts[0])
+		table = strings.TrimSpace(parts[1])
+	}
+
+	qualifiedTable := ""
+	if schema != "" {
+		qualifiedTable = fmt.Sprintf("%s.%s", quoteIdent(schema), quoteIdent(table))
+	} else {
+		qualifiedTable = quoteIdent(table)
+	}
+
+	// 1. Deletes
+	for _, pk := range changes.Deletes {
+		var wheres []string
+		var args []interface{}
+		idx := 0
+		for k, v := range pk {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = :%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+		if len(wheres) == 0 {
+			continue
+		}
+		query := fmt.Sprintf("DELETE FROM %s WHERE %s", qualifiedTable, strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("delete error: %v", err)
+		}
+	}
+
+	// 2. Updates
+	for _, update := range changes.Updates {
+		var sets []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range update.Values {
+			idx++
+			sets = append(sets, fmt.Sprintf("%s = :%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(sets) == 0 {
+			continue
+		}
+
+		var wheres []string
+		for k, v := range update.Keys {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = :%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(wheres) == 0 {
+			return fmt.Errorf("update requires keys")
+		}
+
+		query := fmt.Sprintf("UPDATE %s SET %s WHERE %s", qualifiedTable, strings.Join(sets, ", "), strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("update error: %v", err)
+		}
+	}
+
+	// 3. Inserts
+	for _, row := range changes.Inserts {
+		var cols []string
+		var placeholders []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range row {
+			idx++
+			cols = append(cols, quoteIdent(k))
+			placeholders = append(placeholders, fmt.Sprintf(":%d", idx))
+			args = append(args, v)
+		}
+
+		if len(cols) == 0 {
+			continue
+		}
+
+		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", qualifiedTable, strings.Join(cols, ", "), strings.Join(placeholders, ", "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("insert error: %v", err)
+		}
+	}
+
+	return tx.Commit()
 }
 
 func (d *DamengDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {

internal/db/kingbase_impl.go

@@ -597,7 +597,117 @@ func (k *KingbaseDB) GetTriggers(dbName, tableName string) ([]connection.Trigger
 }
 
 func (k *KingbaseDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
-	return fmt.Errorf("read-only mode implemented for Kingbase so far")
+	if k.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+
+	tx, err := k.conn.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	quoteIdent := func(name string) string {
+		n := strings.TrimSpace(name)
+		n = strings.Trim(n, "\"")
+		n = strings.ReplaceAll(n, "\"", "\"\"")
+		if n == "" {
+			return "\"\""
+		}
+		return `"` + n + `"`
+	}
+
+	schema := ""
+	table := strings.TrimSpace(tableName)
+	if parts := strings.SplitN(table, ".", 2); len(parts) == 2 {
+		schema = strings.TrimSpace(parts[0])
+		table = strings.TrimSpace(parts[1])
+	}
+
+	qualifiedTable := ""
+	if schema != "" {
+		qualifiedTable = fmt.Sprintf("%s.%s", quoteIdent(schema), quoteIdent(table))
+	} else {
+		qualifiedTable = quoteIdent(table)
+	}
+
+	// 1. Deletes
+	for _, pk := range changes.Deletes {
+		var wheres []string
+		var args []interface{}
+		idx := 0
+		for k, v := range pk {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = $%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+		if len(wheres) == 0 {
+			continue
+		}
+		query := fmt.Sprintf("DELETE FROM %s WHERE %s", qualifiedTable, strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("delete error: %v", err)
+		}
+	}
+
+	// 2. Updates
+	for _, update := range changes.Updates {
+		var sets []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range update.Values {
+			idx++
+			sets = append(sets, fmt.Sprintf("%s = $%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(sets) == 0 {
+			continue
+		}
+
+		var wheres []string
+		for k, v := range update.Keys {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = $%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(wheres) == 0 {
+			return fmt.Errorf("update requires keys")
+		}
+
+		query := fmt.Sprintf("UPDATE %s SET %s WHERE %s", qualifiedTable, strings.Join(sets, ", "), strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("update error: %v", err)
+		}
+	}
+
+	// 3. Inserts
+	for _, row := range changes.Inserts {
+		var cols []string
+		var placeholders []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range row {
+			idx++
+			cols = append(cols, quoteIdent(k))
+			placeholders = append(placeholders, fmt.Sprintf("$%d", idx))
+			args = append(args, v)
+		}
+
+		if len(cols) == 0 {
+			continue
+		}
+
+		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", qualifiedTable, strings.Join(cols, ", "), strings.Join(placeholders, ", "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("insert error: %v", err)
+		}
+	}
+
+	return tx.Commit()
 }
 
 func (k *KingbaseDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {

internal/db/mysql_impl.go

@@ -318,15 +318,19 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 		var args []interface{}
 		for k, v := range pk {
 			wheres = append(wheres, fmt.Sprintf("`%s` = ?", k))
-			args = append(args, v)
+			args = append(args, normalizeMySQLDateTimeValue(v))
 		}
 		if len(wheres) == 0 {
 			continue
 		}
 		query := fmt.Sprintf("DELETE FROM `%s` WHERE %s", tableName, strings.Join(wheres, " AND "))
-		if _, err := tx.Exec(query, args...); err != nil {
+		res, err := tx.Exec(query, args...)
+		if err != nil {
 			return fmt.Errorf("delete error: %v", err)
 		}
+		if affected, err := res.RowsAffected(); err == nil && affected == 0 {
+			return fmt.Errorf("删除未生效：未匹配到任何行")
+		}
 	}
 
 	// 2. Updates
@@ -336,7 +340,7 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 
 		for k, v := range update.Values {
 			sets = append(sets, fmt.Sprintf("`%s` = ?", k))
-			args = append(args, v)
+			args = append(args, normalizeMySQLDateTimeValue(v))
 		}
 
 		if len(sets) == 0 {
@@ -346,7 +350,7 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 		var wheres []string
 		for k, v := range update.Keys {
 			wheres = append(wheres, fmt.Sprintf("`%s` = ?", k))
-			args = append(args, v)
+			args = append(args, normalizeMySQLDateTimeValue(v))
 		}
 
 		if len(wheres) == 0 {
@@ -354,9 +358,13 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 		}
 
 		query := fmt.Sprintf("UPDATE `%s` SET %s WHERE %s", tableName, strings.Join(sets, ", "), strings.Join(wheres, " AND "))
-		if _, err := tx.Exec(query, args...); err != nil {
+		res, err := tx.Exec(query, args...)
+		if err != nil {
 			return fmt.Errorf("update error: %v", err)
 		}
+		if affected, err := res.RowsAffected(); err == nil && affected == 0 {
+			return fmt.Errorf("更新未生效：未匹配到任何行")
+		}
 	}
 
 	// 3. Inserts
@@ -368,7 +376,7 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 		for k, v := range row {
 			cols = append(cols, fmt.Sprintf("`%s`", k))
 			placeholders = append(placeholders, "?")
-			args = append(args, v)
+			args = append(args, normalizeMySQLDateTimeValue(v))
 		}
 
 		if len(cols) == 0 {
@@ -376,14 +384,93 @@ func (m *MySQLDB) ApplyChanges(tableName string, changes connection.ChangeSet) e
 		}
 
 		query := fmt.Sprintf("INSERT INTO `%s` (%s) VALUES (%s)", tableName, strings.Join(cols, ", "), strings.Join(placeholders, ", "))
-		if _, err := tx.Exec(query, args...); err != nil {
+		res, err := tx.Exec(query, args...)
+		if err != nil {
 			return fmt.Errorf("insert error: %v", err)
 		}
+		if affected, err := res.RowsAffected(); err == nil && affected == 0 {
+			return fmt.Errorf("插入未生效：未影响任何行")
+		}
 	}
 
 	return tx.Commit()
 }
 
+func normalizeMySQLDateTimeValue(value interface{}) interface{} {
+	text, ok := value.(string)
+	if !ok {
+		return value
+	}
+	raw := strings.TrimSpace(text)
+	if raw == "" {
+		return value
+	}
+
+	cleaned := strings.ReplaceAll(raw, "+ ", "+")
+	cleaned = strings.ReplaceAll(cleaned, "- ", "-")
+
+	if len(cleaned) >= 19 && cleaned[10] == 'T' {
+		if strings.HasSuffix(cleaned, "Z") || hasTimezoneOffset(cleaned) {
+			if t, err := time.Parse(time.RFC3339Nano, cleaned); err == nil {
+				return formatMySQLDateTime(t)
+			}
+			if t, err := time.Parse(time.RFC3339, cleaned); err == nil {
+				return formatMySQLDateTime(t)
+			}
+		}
+		return strings.Replace(cleaned, "T", " ", 1)
+	}
+
+	if strings.Contains(cleaned, " ") && (strings.HasSuffix(cleaned, "Z") || hasTimezoneOffset(cleaned)) {
+		candidate := strings.Replace(cleaned, " ", "T", 1)
+		if t, err := time.Parse(time.RFC3339Nano, candidate); err == nil {
+			return formatMySQLDateTime(t)
+		}
+		if t, err := time.Parse(time.RFC3339, candidate); err == nil {
+			return formatMySQLDateTime(t)
+		}
+	}
+
+	return value
+}
+
+func hasTimezoneOffset(text string) bool {
+	pos := strings.LastIndexAny(text, "+-")
+	if pos < 0 || pos < 10 || pos+1 >= len(text) {
+		return false
+	}
+	offset := text[pos+1:]
+	if len(offset) == 5 && offset[2] == ':' {
+		return isAllDigits(offset[:2]) && isAllDigits(offset[3:])
+	}
+	if len(offset) == 4 {
+		return isAllDigits(offset)
+	}
+	return false
+}
+
+func isAllDigits(text string) bool {
+	if text == "" {
+		return false
+	}
+	for _, r := range text {
+		if r < '0' || r > '9' {
+			return false
+		}
+	}
+	return true
+}
+
+func formatMySQLDateTime(t time.Time) string {
+	base := t.Format("2006-01-02 15:04:05")
+	nanos := t.Nanosecond()
+	if nanos == 0 {
+		return base
+	}
+	micro := nanos / 1000
+	return fmt.Sprintf("%s.%06d", base, micro)
+}
+
 func (m *MySQLDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {
 	query := fmt.Sprintf("SELECT TABLE_NAME, COLUMN_NAME, COLUMN_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '%s'", dbName)
 	if dbName == "" {

internal/db/oracle_impl.go

@@ -363,8 +363,117 @@ func (o *OracleDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDe
 }
 
 func (o *OracleDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
-	// TODO: Implement batch application for Oracle using correct syntax
-	return fmt.Errorf("read-only mode implemented for Oracle so far")
+	if o.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+
+	tx, err := o.conn.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	quoteIdent := func(name string) string {
+		n := strings.TrimSpace(name)
+		n = strings.Trim(n, "\"")
+		n = strings.ReplaceAll(n, "\"", "\"\"")
+		if n == "" {
+			return "\"\""
+		}
+		return `"` + n + `"`
+	}
+
+	schema := ""
+	table := strings.TrimSpace(tableName)
+	if parts := strings.SplitN(table, ".", 2); len(parts) == 2 {
+		schema = strings.TrimSpace(parts[0])
+		table = strings.TrimSpace(parts[1])
+	}
+
+	qualifiedTable := ""
+	if schema != "" {
+		qualifiedTable = fmt.Sprintf("%s.%s", quoteIdent(schema), quoteIdent(table))
+	} else {
+		qualifiedTable = quoteIdent(table)
+	}
+
+	// 1. Deletes
+	for _, pk := range changes.Deletes {
+		var wheres []string
+		var args []interface{}
+		idx := 0
+		for k, v := range pk {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = :%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+		if len(wheres) == 0 {
+			continue
+		}
+		query := fmt.Sprintf("DELETE FROM %s WHERE %s", qualifiedTable, strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("delete error: %v", err)
+		}
+	}
+
+	// 2. Updates
+	for _, update := range changes.Updates {
+		var sets []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range update.Values {
+			idx++
+			sets = append(sets, fmt.Sprintf("%s = :%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(sets) == 0 {
+			continue
+		}
+
+		var wheres []string
+		for k, v := range update.Keys {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = :%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(wheres) == 0 {
+			return fmt.Errorf("update requires keys")
+		}
+
+		query := fmt.Sprintf("UPDATE %s SET %s WHERE %s", qualifiedTable, strings.Join(sets, ", "), strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("update error: %v", err)
+		}
+	}
+
+	// 3. Inserts
+	for _, row := range changes.Inserts {
+		var cols []string
+		var placeholders []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range row {
+			idx++
+			cols = append(cols, quoteIdent(k))
+			placeholders = append(placeholders, fmt.Sprintf(":%d", idx))
+			args = append(args, v)
+		}
+
+		if len(cols) == 0 {
+			continue
+		}
+
+		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", qualifiedTable, strings.Join(cols, ", "), strings.Join(placeholders, ", "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("insert error: %v", err)
+		}
+	}
+
+	return tx.Commit()
 }
 
 func (o *OracleDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {

internal/db/postgres_impl.go

@@ -521,3 +521,117 @@ ORDER BY table_schema, table_name, ordinal_position`
 	}
 	return cols, nil
 }
+
+func (p *PostgresDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
+	if p.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+
+	tx, err := p.conn.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	quoteIdent := func(name string) string {
+		n := strings.TrimSpace(name)
+		n = strings.Trim(n, "\"")
+		n = strings.ReplaceAll(n, "\"", "\"\"")
+		if n == "" {
+			return "\"\""
+		}
+		return `"` + n + `"`
+	}
+
+	schema := ""
+	table := strings.TrimSpace(tableName)
+	if parts := strings.SplitN(table, ".", 2); len(parts) == 2 {
+		schema = strings.TrimSpace(parts[0])
+		table = strings.TrimSpace(parts[1])
+	}
+
+	qualifiedTable := ""
+	if schema != "" {
+		qualifiedTable = fmt.Sprintf("%s.%s", quoteIdent(schema), quoteIdent(table))
+	} else {
+		qualifiedTable = quoteIdent(table)
+	}
+
+	// 1. Deletes
+	for _, pk := range changes.Deletes {
+		var wheres []string
+		var args []interface{}
+		idx := 0
+		for k, v := range pk {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = $%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+		if len(wheres) == 0 {
+			continue
+		}
+		query := fmt.Sprintf("DELETE FROM %s WHERE %s", qualifiedTable, strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("delete error: %v", err)
+		}
+	}
+
+	// 2. Updates
+	for _, update := range changes.Updates {
+		var sets []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range update.Values {
+			idx++
+			sets = append(sets, fmt.Sprintf("%s = $%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(sets) == 0 {
+			continue
+		}
+
+		var wheres []string
+		for k, v := range update.Keys {
+			idx++
+			wheres = append(wheres, fmt.Sprintf("%s = $%d", quoteIdent(k), idx))
+			args = append(args, v)
+		}
+
+		if len(wheres) == 0 {
+			return fmt.Errorf("update requires keys")
+		}
+
+		query := fmt.Sprintf("UPDATE %s SET %s WHERE %s", qualifiedTable, strings.Join(sets, ", "), strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("update error: %v", err)
+		}
+	}
+
+	// 3. Inserts
+	for _, row := range changes.Inserts {
+		var cols []string
+		var placeholders []string
+		var args []interface{}
+		idx := 0
+
+		for k, v := range row {
+			idx++
+			cols = append(cols, quoteIdent(k))
+			placeholders = append(placeholders, fmt.Sprintf("$%d", idx))
+			args = append(args, v)
+		}
+
+		if len(cols) == 0 {
+			continue
+		}
+
+		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", qualifiedTable, strings.Join(cols, ", "), strings.Join(placeholders, ", "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("insert error: %v", err)
+		}
+	}
+
+	return tx.Commit()
+}

internal/db/sqlite_impl.go

@@ -445,6 +445,113 @@ func (s *SQLiteDB) GetTriggers(dbName, tableName string) ([]connection.TriggerDe
 	return triggers, nil
 }
 
+func (s *SQLiteDB) ApplyChanges(tableName string, changes connection.ChangeSet) error {
+	if s.conn == nil {
+		return fmt.Errorf("connection not open")
+	}
+
+	tx, err := s.conn.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	quoteIdent := func(name string) string {
+		n := strings.TrimSpace(name)
+		n = strings.Trim(n, "\"")
+		n = strings.ReplaceAll(n, "\"", "\"\"")
+		if n == "" {
+			return "\"\""
+		}
+		return `"` + n + `"`
+	}
+
+	schema := ""
+	table := strings.TrimSpace(tableName)
+	if parts := strings.SplitN(table, ".", 2); len(parts) == 2 {
+		schema = strings.TrimSpace(parts[0])
+		table = strings.TrimSpace(parts[1])
+	}
+
+	qualifiedTable := ""
+	if schema != "" {
+		qualifiedTable = fmt.Sprintf("%s.%s", quoteIdent(schema), quoteIdent(table))
+	} else {
+		qualifiedTable = quoteIdent(table)
+	}
+
+	// 1. Deletes
+	for _, pk := range changes.Deletes {
+		var wheres []string
+		var args []interface{}
+		for k, v := range pk {
+			wheres = append(wheres, fmt.Sprintf("%s = ?", quoteIdent(k)))
+			args = append(args, v)
+		}
+		if len(wheres) == 0 {
+			continue
+		}
+		query := fmt.Sprintf("DELETE FROM %s WHERE %s", qualifiedTable, strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("delete error: %v", err)
+		}
+	}
+
+	// 2. Updates
+	for _, update := range changes.Updates {
+		var sets []string
+		var args []interface{}
+
+		for k, v := range update.Values {
+			sets = append(sets, fmt.Sprintf("%s = ?", quoteIdent(k)))
+			args = append(args, v)
+		}
+
+		if len(sets) == 0 {
+			continue
+		}
+
+		var wheres []string
+		for k, v := range update.Keys {
+			wheres = append(wheres, fmt.Sprintf("%s = ?", quoteIdent(k)))
+			args = append(args, v)
+		}
+
+		if len(wheres) == 0 {
+			return fmt.Errorf("update requires keys")
+		}
+
+		query := fmt.Sprintf("UPDATE %s SET %s WHERE %s", qualifiedTable, strings.Join(sets, ", "), strings.Join(wheres, " AND "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("update error: %v", err)
+		}
+	}
+
+	// 3. Inserts
+	for _, row := range changes.Inserts {
+		var cols []string
+		var placeholders []string
+		var args []interface{}
+
+		for k, v := range row {
+			cols = append(cols, quoteIdent(k))
+			placeholders = append(placeholders, "?")
+			args = append(args, v)
+		}
+
+		if len(cols) == 0 {
+			continue
+		}
+
+		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", qualifiedTable, strings.Join(cols, ", "), strings.Join(placeholders, ", "))
+		if _, err := tx.Exec(query, args...); err != nil {
+			return fmt.Errorf("insert error: %v", err)
+		}
+	}
+
+	return tx.Commit()
+}
+
 func (s *SQLiteDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) {
 	tables, err := s.GetTables(dbName)
 	if err != nil {