diff --git a/agents/drivers/kingbase-go/kingbase_metadata.go b/agents/drivers/kingbase-go/kingbase_metadata.go
index 86b6277ddc..129b221b15 100644
--- a/agents/drivers/kingbase-go/kingbase_metadata.go
+++ b/agents/drivers/kingbase-go/kingbase_metadata.go
@@ -179,6 +179,18 @@ func (s *server) identifierQuote() string {
 	return `"`
 }
 
+// quoteDDLIdentifier quotes an identifier inside generated DDL using the mode's
+// identifier quote. MySQL compatibility mode uses backticks (and backtick
+// escaping); everything else keeps the PostgreSQL-compatible double quote, so
+// schema/table names containing hyphens or other special characters render as
+// valid SQL instead of being parsed as operators or bare tokens.
+func (s *server) quoteDDLIdentifier(value string) string {
+	if s.mode.mysqlCompat {
+		return "`" + strings.ReplaceAll(value, "`", "``") + "`"
+	}
+	return quoteIdentifier(value)
+}
+
 func (s *server) connectionInfo() (map[string]any, error) {
 	db, err := s.requireDB()
 	if err != nil {
@@ -700,7 +712,7 @@ func (s *server) getTypeDetails(schema, name string) (*customTypeDetails, error)
 		warnings = append(warnings, s.customTypeRangeAttributes(queries, &details.Properties, oid, true)...)
 	case customTypeKindBase:
 	}
-	details.DDL = buildCustomTypeDDL(schema, name, kind, inputFn, &details.Members, &details.Properties, warnings)
+	details.DDL = s.buildCustomTypeDDL(schema, name, kind, inputFn, &details.Members, &details.Properties, warnings)
 	return details, nil
 }
 
@@ -924,22 +936,22 @@ func resolveCustomTypeDomainDefault(typdefaultbin, typdefault sql.NullString, re
 	return &value, nil
 }
 
-func quoteCatalogIdentifier(value string) string {
+func (s *server) quoteCatalogIdentifier(value string) string {
 	value = strings.TrimSpace(value)
 	if strings.HasPrefix(value, `"`) && strings.HasSuffix(value, `"`) && strings.Contains(value, `"."`) {
 		return value
 	}
 	if separator := strings.LastIndexByte(value, '.'); separator >= 0 {
-		return quoteIdentifier(value[:separator]) + "." + quoteIdentifier(value[separator+1:])
+		return s.quoteDDLIdentifier(value[:separator]) + "." + s.quoteDDLIdentifier(value[separator+1:])
 	}
-	return quoteIdentifier(value)
+	return s.quoteDDLIdentifier(value)
 }
 
 // buildCustomTypeDDL generates normalized CREATE TYPE text. complete is only
 // true when the text can be executed standalone; multiranges and base types
 // are marked incomplete with visible warnings.
-func buildCustomTypeDDL(schema, name string, kind customTypeKind, inputFn sql.NullString, members *[]customTypeMember, properties *customTypeProperties, warnings []string) *customTypeDdl {
-	qualified := quoteIdentifier(schema) + "." + quoteIdentifier(name)
+func (s *server) buildCustomTypeDDL(schema, name string, kind customTypeKind, inputFn sql.NullString, members *[]customTypeMember, properties *customTypeProperties, warnings []string) *customTypeDdl {
+	qualified := s.quoteDDLIdentifier(schema) + "." + s.quoteDDLIdentifier(name)
 	switch kind {
 	case customTypeKindEnum:
 		values := make([]string, 0, len(*members))
@@ -957,9 +969,9 @@ func buildCustomTypeDDL(schema, name string, kind customTypeKind, inputFn sql.Nu
 		fields := make([]string, 0, len(*members))
 		comments := make([]string, 0, len(*members))
 		for _, member := range *members {
-			fields = append(fields, quoteIdentifier(member.Name)+" "+member.DataType)
+			fields = append(fields, s.quoteDDLIdentifier(member.Name)+" "+member.DataType)
 			if member.Comment != nil {
-				comments = append(comments, fmt.Sprintf("COMMENT ON COLUMN %s.%s IS %s;", qualified, quoteIdentifier(member.Name), quoteLiteral(*member.Comment)))
+				comments = append(comments, fmt.Sprintf("COMMENT ON COLUMN %s.%s IS %s;", qualified, s.quoteDDLIdentifier(member.Name), quoteLiteral(*member.Comment)))
 			}
 		}
 		sql := fmt.Sprintf("CREATE TYPE %s AS (\n  %s\n);", qualified, strings.Join(fields, ",\n  "))
@@ -978,7 +990,7 @@ func buildCustomTypeDDL(schema, name string, kind customTypeKind, inputFn sql.Nu
 		}
 		parts := []string{fmt.Sprintf("CREATE DOMAIN %s AS %s", qualified, base)}
 		if properties.Collation != nil && *properties.Collation != "" {
-			parts = append(parts, "COLLATE "+quoteCatalogIdentifier(*properties.Collation))
+			parts = append(parts, "COLLATE "+s.quoteCatalogIdentifier(*properties.Collation))
 		}
 		if properties.Default != nil && *properties.Default != "" {
 			parts = append(parts, "DEFAULT "+*properties.Default)
@@ -992,7 +1004,7 @@ func buildCustomTypeDDL(schema, name string, kind customTypeKind, inputFn sql.Nu
 			if constraintName == "" {
 				constraintName = name + "_check"
 			}
-			parts = append(parts, fmt.Sprintf("CONSTRAINT %s CHECK %s", quoteIdentifier(constraintName), body))
+			parts = append(parts, fmt.Sprintf("CONSTRAINT %s CHECK %s", s.quoteDDLIdentifier(constraintName), body))
 		}
 		for _, warning := range warnings {
 			if strings.Contains(warning, "domain constraints") || strings.Contains(warning, "default value could not be rendered") {
@@ -1006,16 +1018,16 @@ func buildCustomTypeDDL(schema, name string, kind customTypeKind, inputFn sql.Nu
 			args = append(args, "subtype = "+*properties.RangeSubtype)
 		}
 		if properties.RangeSubtypeOpclass != nil && *properties.RangeSubtypeOpclass != "" {
-			args = append(args, "subtype_opclass = "+quoteCatalogIdentifier(*properties.RangeSubtypeOpclass))
+			args = append(args, "subtype_opclass = "+s.quoteCatalogIdentifier(*properties.RangeSubtypeOpclass))
 		}
 		if properties.RangeCanonicalFunction != nil && *properties.RangeCanonicalFunction != "" {
-			args = append(args, "canonical = "+quoteCatalogIdentifier(*properties.RangeCanonicalFunction))
+			args = append(args, "canonical = "+s.quoteCatalogIdentifier(*properties.RangeCanonicalFunction))
 		}
 		if properties.RangeSubtypeDiffFunction != nil && *properties.RangeSubtypeDiffFunction != "" {
-			args = append(args, "subtype_diff = "+quoteCatalogIdentifier(*properties.RangeSubtypeDiffFunction))
+			args = append(args, "subtype_diff = "+s.quoteCatalogIdentifier(*properties.RangeSubtypeDiffFunction))
 		}
 		if properties.RangeMultirangeName != nil && *properties.RangeMultirangeName != "" {
-			args = append(args, "multirange_type_name = "+quoteCatalogIdentifier(*properties.RangeMultirangeName))
+			args = append(args, "multirange_type_name = "+s.quoteCatalogIdentifier(*properties.RangeMultirangeName))
 		}
 		if properties.RangeSubtype == nil || *properties.RangeSubtype == "" {
 			result := &customTypeDdl{
@@ -1552,7 +1564,7 @@ func (s *server) getTableDDL(schema, table string) (string, error) {
 		return "", err
 	}
 	tableComment, _ := s.getTableComment(effective, table)
-	ddl := renderTableDDL(effective, table, columns, tableComment)
+	ddl := s.renderTableDDL(effective, table, columns, tableComment)
 	ddl, err = s.appendTableIndexDDL(effective, table, ddl)
 	if err != nil {
 		return "", err
@@ -1564,19 +1576,19 @@ func (s *server) getTableDDL(schema, table string) (string, error) {
 	return ddl, nil
 }
 
-func renderTableDDL(schema, table string, columns []columnInfo, tableComment *string) string {
+func (s *server) renderTableDDL(schema, table string, columns []columnInfo, tableComment *string) string {
 	definitions := make([]string, 0, len(columns)+1)
 	primary := []string{}
 	for _, column := range columns {
-		definitions = append(definitions, columnDDLDefinition(column))
+		definitions = append(definitions, s.columnDDLDefinition(column))
 		if column.IsPrimaryKey {
-			primary = append(primary, quoteIdentifier(column.Name))
+			primary = append(primary, s.quoteDDLIdentifier(column.Name))
 		}
 	}
 	if len(primary) > 0 {
 		definitions = append(definitions, "PRIMARY KEY ("+strings.Join(primary, ", ")+")")
 	}
-	qualifiedTable := quoteIdentifier(schema) + "." + quoteIdentifier(table)
+	qualifiedTable := s.quoteDDLIdentifier(schema) + "." + s.quoteDDLIdentifier(table)
 	ddl := "CREATE TABLE " + qualifiedTable + " (\n  " + strings.Join(definitions, ",\n  ") + "\n);"
 	if tableComment != nil && strings.TrimSpace(*tableComment) != "" {
 		ddl += "\nCOMMENT ON TABLE " + qualifiedTable + " IS " + quoteLiteral(*tableComment) + ";"
@@ -1585,7 +1597,7 @@ func renderTableDDL(schema, table string, columns []columnInfo, tableComment *st
 		if column.Comment == nil || strings.TrimSpace(*column.Comment) == "" {
 			continue
 		}
-		ddl += "\nCOMMENT ON COLUMN " + qualifiedTable + "." + quoteIdentifier(column.Name) + " IS " + quoteLiteral(*column.Comment) + ";"
+		ddl += "\nCOMMENT ON COLUMN " + qualifiedTable + "." + s.quoteDDLIdentifier(column.Name) + " IS " + quoteLiteral(*column.Comment) + ";"
 	}
 	return ddl
 }
@@ -1637,7 +1649,7 @@ WHERE n.nspname = %s AND t.relname = %s AND NOT ix.indisprimary ORDER BY i.relna
 			result = append(result, definition)
 		}
 		if comment.Valid && strings.TrimSpace(comment.String) != "" {
-			result = append(result, "COMMENT ON INDEX "+quoteIdentifier(effective)+"."+quoteIdentifier(name)+" IS "+quoteLiteral(comment.String))
+			result = append(result, "COMMENT ON INDEX "+s.quoteDDLIdentifier(effective)+"."+s.quoteDDLIdentifier(name)+" IS "+quoteLiteral(comment.String))
 		}
 	}
 	return result, rows.Err()
@@ -1704,8 +1716,8 @@ func ensureStatementTerminator(statement string) string {
 	return trimmed + ";"
 }
 
-func columnDDLDefinition(column columnInfo) string {
-	definition := quoteIdentifier(column.Name) + " " + columnDDLDataType(column)
+func (s *server) columnDDLDefinition(column columnInfo) string {
+	definition := s.quoteDDLIdentifier(column.Name) + " " + columnDDLDataType(column)
 	if column.Extra != nil && *column.Extra != "" {
 		// Identity clauses belong immediately after the data type in both
 		// PostgreSQL-compatible and SQL Server-compatible Kingbase modes.
diff --git a/agents/drivers/kingbase-go/main_test.go b/agents/drivers/kingbase-go/main_test.go
index 932d3dc4db..0b511db2ab 100644
--- a/agents/drivers/kingbase-go/main_test.go
+++ b/agents/drivers/kingbase-go/main_test.go
@@ -1791,7 +1791,7 @@ func TestKingbaseDomainConstraintReadFailuresMarkDDLIncomplete(t *testing.T) {
 	if len(warnings) != 1 || !strings.Contains(warnings[0], "domain constraints could not be decoded") {
 		t.Fatalf("constraint scan failures must be retained as warnings: %v", warnings)
 	}
-	ddl := buildCustomTypeDDL("app", "email", customTypeKindDomain, sql.NullString{}, &[]customTypeMember{}, &properties, warnings)
+	ddl := server.buildCustomTypeDDL("app", "email", customTypeKindDomain, sql.NullString{}, &[]customTypeMember{}, &properties, warnings)
 	if ddl.Complete {
 		t.Fatalf("domain DDL must be incomplete after a constraint scan failure: %+v", ddl)
 	}
@@ -1849,12 +1849,13 @@ func TestKingbaseSystemSchemasAreRejectedForCustomTypeDetails(t *testing.T) {
 }
 
 func TestKingbaseCustomTypeDDL(t *testing.T) {
+	srv := newServer()
 	nullInput := sql.NullString{}
 	enumMembers := []customTypeMember{
 		{Ordinal: 1, EnumValue: stringPtr("draft")},
 		{Ordinal: 2, EnumValue: stringPtr("已归档")},
 	}
-	enumDDL := buildCustomTypeDDL("app", "status", customTypeKindEnum, nullInput, &enumMembers, &customTypeProperties{}, nil)
+	enumDDL := srv.buildCustomTypeDDL("app", "status", customTypeKindEnum, nullInput, &enumMembers, &customTypeProperties{}, nil)
 	if enumDDL.SQL != "CREATE TYPE \"app\".\"status\" AS ENUM ('draft', '已归档');" || !enumDDL.Complete {
 		t.Fatalf("unexpected enum DDL: %+v", enumDDL)
 	}
@@ -1862,34 +1863,34 @@ func TestKingbaseCustomTypeDDL(t *testing.T) {
 	compositeMembers := []customTypeMember{
 		{Name: "city", DataType: "text", Ordinal: 1, Comment: stringPtr("city name")},
 	}
-	compositeDDL := buildCustomTypeDDL("app", "address", customTypeKindComposite, nullInput, &compositeMembers, &customTypeProperties{}, nil)
+	compositeDDL := srv.buildCustomTypeDDL("app", "address", customTypeKindComposite, nullInput, &compositeMembers, &customTypeProperties{}, nil)
 	if !strings.Contains(compositeDDL.SQL, "\"city\" text") || !strings.Contains(compositeDDL.SQL, "COMMENT ON COLUMN \"app\".\"address\".\"city\" IS 'city name';") {
 		t.Fatalf("unexpected composite DDL: %+v", compositeDDL)
 	}
 
 	notNull := true
 	domainProps := customTypeProperties{BaseType: stringPtr("text"), NotNull: &notNull, DomainConstraints: []customTypeDomainConstraint{{Name: "email_valid", Definition: "CHECK ((VALUE <> ''::text))"}}}
-	domainDDL := buildCustomTypeDDL("app", "email", customTypeKindDomain, nullInput, &[]customTypeMember{}, &domainProps, nil)
+	domainDDL := srv.buildCustomTypeDDL("app", "email", customTypeKindDomain, nullInput, &[]customTypeMember{}, &domainProps, nil)
 	if !strings.Contains(domainDDL.SQL, "CREATE DOMAIN \"app\".\"email\" AS text") || !strings.Contains(domainDDL.SQL, "NOT NULL") || !strings.Contains(domainDDL.SQL, "CHECK ((VALUE <> ''::text))") {
 		t.Fatalf("unexpected domain DDL: %+v", domainDDL)
 	}
 
 	rangeProps := customTypeProperties{RangeSubtype: stringPtr("numeric"), RangeCanonicalFunction: stringPtr("\"extensions\".\"numeric_range_canonical\"")}
-	rangeDDL := buildCustomTypeDDL("app", "price_range", customTypeKindRange, nullInput, &[]customTypeMember{}, &rangeProps, nil)
+	rangeDDL := srv.buildCustomTypeDDL("app", "price_range", customTypeKindRange, nullInput, &[]customTypeMember{}, &rangeProps, nil)
 	if !rangeDDL.Complete || !strings.Contains(rangeDDL.SQL, "subtype = numeric") || !strings.Contains(rangeDDL.SQL, "canonical = \"extensions\".\"numeric_range_canonical\"") {
 		t.Fatalf("unexpected range DDL: %+v", rangeDDL)
 	}
-	missingSubtype := buildCustomTypeDDL("app", "price_range", customTypeKindRange, nullInput, &[]customTypeMember{}, &customTypeProperties{RangeMultirangeName: stringPtr("price_multirange")}, nil)
+	missingSubtype := srv.buildCustomTypeDDL("app", "price_range", customTypeKindRange, nullInput, &[]customTypeMember{}, &customTypeProperties{RangeMultirangeName: stringPtr("price_multirange")}, nil)
 	if missingSubtype.Complete || missingSubtype.SQL != "CREATE TYPE \"app\".\"price_range\" AS RANGE (subtype = unknown);" {
 		t.Fatalf("range DDL without subtype must be incomplete: %+v", missingSubtype)
 	}
 
-	multirangeDDL := buildCustomTypeDDL("app", "_price_range", customTypeKindMultirange, nullInput, &[]customTypeMember{}, &customTypeProperties{}, nil)
+	multirangeDDL := srv.buildCustomTypeDDL("app", "_price_range", customTypeKindMultirange, nullInput, &[]customTypeMember{}, &customTypeProperties{}, nil)
 	if multirangeDDL.Complete || len(multirangeDDL.Warnings) == 0 {
 		t.Fatalf("multirange DDL must be incomplete with warnings: %+v", multirangeDDL)
 	}
 
-	baseDDL := buildCustomTypeDDL("app", "point2d", customTypeKindBase, nullInput, &[]customTypeMember{}, &customTypeProperties{}, nil)
+	baseDDL := srv.buildCustomTypeDDL("app", "point2d", customTypeKindBase, nullInput, &[]customTypeMember{}, &customTypeProperties{}, nil)
 	if baseDDL.Complete || len(baseDDL.Warnings) == 0 {
 		t.Fatalf("base DDL must be incomplete with warnings: %+v", baseDDL)
 	}
@@ -2136,6 +2137,7 @@ func TestColumnsFallbackWhenCatalogHasNoAttidentityAndCacheChoice(t *testing.T)
 }
 
 func TestKingbaseIdentityClausesAreExposedAndRendered(t *testing.T) {
+	srv := newServer()
 	tests := []struct {
 		name     string
 		code     string
@@ -2151,7 +2153,7 @@ func TestKingbaseIdentityClausesAreExposedAndRendered(t *testing.T) {
 			if extra == nil || *extra != test.expected {
 				t.Fatalf("unexpected identity clause for %q: %#v", test.code, extra)
 			}
-			definition := columnDDLDefinition(columnInfo{Name: "id", DataType: "integer", IsNullable: false, Extra: extra})
+			definition := srv.columnDDLDefinition(columnInfo{Name: "id", DataType: "integer", IsNullable: false, Extra: extra})
 			expected := `"id" integer ` + test.expected + " NOT NULL"
 			if definition != expected {
 				t.Fatalf("unexpected column DDL: %s", definition)
@@ -2199,10 +2201,11 @@ func TestTableDDLIncludesIdentityIndexesTriggersAndComments(t *testing.T) {
 }
 
 func TestRenderTableDDLIncludesEscapedComments(t *testing.T) {
+	srv := newServer()
 	primaryComment := "主键'编号"
 	emptyComment := "  "
 	tableComment := "订单'表"
-	ddl := renderTableDDL(
+	ddl := srv.renderTableDDL(
 		`app"schema`,
 		`order"items`,
 		[]columnInfo{
@@ -2228,12 +2231,13 @@ func TestRenderTableDDLIncludesEscapedComments(t *testing.T) {
 }
 
 func TestColumnDDLDefinitionPreservesCompatibilityExtras(t *testing.T) {
+	srv := newServer()
 	identity := "IDENTITY(1,1)"
 	defaultValue := "0"
-	if definition := columnDDLDefinition(columnInfo{Name: "id", DataType: "integer", IsNullable: false, Extra: &identity}); definition != `"id" integer IDENTITY(1,1) NOT NULL` {
+	if definition := srv.columnDDLDefinition(columnInfo{Name: "id", DataType: "integer", IsNullable: false, Extra: &identity}); definition != `"id" integer IDENTITY(1,1) NOT NULL` {
 		t.Fatalf("unexpected SQL Server-compatible DDL: %s", definition)
 	}
-	if definition := columnDDLDefinition(columnInfo{Name: "count", DataType: "integer", IsNullable: true, ColumnDefault: &defaultValue}); definition != `"count" integer DEFAULT 0` {
+	if definition := srv.columnDDLDefinition(columnInfo{Name: "count", DataType: "integer", IsNullable: true, ColumnDefault: &defaultValue}); definition != `"count" integer DEFAULT 0` {
 		t.Fatalf("unexpected regular column DDL: %s", definition)
 	}
 	if extra := kingbaseIdentityClause(""); extra != nil {
@@ -2241,7 +2245,25 @@ func TestColumnDDLDefinitionPreservesCompatibilityExtras(t *testing.T) {
 	}
 }
 
+func TestRenderTableDDLUsesBacktickIdentifiersInMySQLCompatMode(t *testing.T) {
+	srv := newServer()
+	srv.mode.mysqlCompat = true
+	ddl := srv.renderTableDDL(
+		"audit-schema",
+		"events",
+		[]columnInfo{
+			{Name: "id", DataType: "integer", IsNullable: false, IsPrimaryKey: true},
+		},
+		nil,
+	)
+	expected := "CREATE TABLE `audit-schema`.`events` (\n  `id` integer NOT NULL,\n  PRIMARY KEY (`id`)\n);"
+	if ddl != expected {
+		t.Fatalf("MySQL-compat DDL must use backtick identifiers:\ngot:  %s\nwant: %s", ddl, expected)
+	}
+}
+
 func TestColumnDDLDefinitionRestoresMySQLCompatibilityTypeModifiers(t *testing.T) {
+	srv := newServer()
 	length := 64
 	precision := 12
 	scale := 4
@@ -2261,7 +2283,7 @@ func TestColumnDDLDefinitionRestoresMySQLCompatibilityTypeModifiers(t *testing.T
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			if got := columnDDLDefinition(test.column); got != test.want {
+			if got := srv.columnDDLDefinition(test.column); got != test.want {
 				t.Fatalf("unexpected column DDL: got %q, want %q", got, test.want)
 			}
 		})
@@ -2293,7 +2315,7 @@ func TestInformationSchemaColumnsPreserveFullTypesInDDL(t *testing.T) {
 	if len(columns) != 4 || columns[0].DataType != "integer" || columns[0].FullDataType != "integer unsigned" {
 		t.Fatalf("unexpected metadata columns: %#v", columns)
 	}
-	ddl := renderTableDDL("public", "orders", columns, nil)
+	ddl := server.renderTableDDL("public", "orders", columns, nil)
 	for _, expected := range []string{
 		`"count" integer unsigned`,
 		`"status" enum('new','done')`,
@@ -2336,7 +2358,7 @@ func TestInformationSchemaColumnsResolveUserDefinedTypeWithoutColumnType(t *test
 	if !strings.Contains(string(payload), `"data_type":"datetime"`) || strings.Contains(string(payload), "USER-DEFINED") {
 		t.Fatalf("unresolved user-defined type leaked into get_columns payload: %s", payload)
 	}
-	if ddl := renderTableDDL("public", "orders", columns, nil); !strings.Contains(ddl, `"created_at" datetime`) || strings.Contains(ddl, "USER-DEFINED") {
+	if ddl := server.renderTableDDL("public", "orders", columns, nil); !strings.Contains(ddl, `"created_at" datetime`) || strings.Contains(ddl, "USER-DEFINED") {
 		t.Fatalf("unexpected user-defined type DDL:\n%s", ddl)
 	}
 
@@ -2398,7 +2420,7 @@ func TestInformationSchemaColumnsPreserveColumnTypeWithoutUdtName(t *testing.T)
 		if len(columns) != 1 || columns[0].FullDataType != "enum('new','done')" {
 			t.Fatalf("unexpected metadata columns: %#v", columns)
 		}
-		if ddl := renderTableDDL("public", table, columns, nil); !strings.Contains(ddl, `"status" enum('new','done')`) {
+		if ddl := server.renderTableDDL("public", table, columns, nil); !strings.Contains(ddl, `"status" enum('new','done')`) {
 			t.Fatalf("unexpected table DDL:\n%s", ddl)
 		}
 	}
@@ -2434,7 +2456,7 @@ func TestInformationSchemaColumnsFallbackWithoutExtendedTypeColumns(t *testing.T
 		if err != nil {
 			t.Fatal(err)
 		}
-		if len(columns) != 1 || columnDDLDefinition(columns[0]) != `"label" varchar(64)` {
+		if len(columns) != 1 || server.columnDDLDefinition(columns[0]) != `"label" varchar(64)` {
 			t.Fatalf("unexpected fallback columns: %#v", columns)
 		}
 	}
@@ -2479,7 +2501,7 @@ func TestInformationSchemaColumnsRetriesOnlyForMissingTypeMetadataColumns(t *tes
 				if err != nil {
 					t.Fatal(err)
 				}
-				if len(columns) != 1 || columnDDLDefinition(columns[0]) != `"label" varchar(64)` {
+				if len(columns) != 1 || server.columnDDLDefinition(columns[0]) != `"label" varchar(64)` {
 					t.Fatalf("unexpected fallback columns: %#v", columns)
 				}
 			} else if !errors.Is(err, test.firstError) {
diff --git a/apps/desktop/src/App.vue b/apps/desktop/src/App.vue
index 2f856b5273..a668cdb488 100644
--- a/apps/desktop/src/App.vue
+++ b/apps/desktop/src/App.vue
@@ -1564,6 +1564,7 @@ async function newQuery() {
     targetConnectionId: target.connectionId,
     targetDatabase: target.database,
     databaseType: effectiveDatabaseTypeForConnection(conn),
+    identifierQuote: connectionStore.connectionIdentifierQuote?.(target.connectionId),
   });
   const tabId = queryStore.createTab(conn.id, target.database, undefined, "query", target.schema, initialSql, target.catalog);
   if (initialSql) {
diff --git a/apps/desktop/src/components/sidebar/SidebarTreeRuntimeHost.vue b/apps/desktop/src/components/sidebar/SidebarTreeRuntimeHost.vue
index db8a4781e0..69d5e9eb6a 100644
--- a/apps/desktop/src/components/sidebar/SidebarTreeRuntimeHost.vue
+++ b/apps/desktop/src/components/sidebar/SidebarTreeRuntimeHost.vue
@@ -76,6 +76,7 @@ import { isXuguTypeMemberContainer } from "@/lib/sidebar/xuguTypeMembers";
 import { isXuguPublicSynonymTreeNode } from "@/lib/sidebar/xuguPublicSynonyms";
 import { mysqlObjectTemplateForGroup } from "@/lib/sidebar/mysqlObjectTemplates";
 import { buildTableDeleteTemplate, buildTableInsertTemplate, buildTableSelectTemplate, buildTableUpdateTemplate } from "@/lib/table/tableSqlTemplates";
+import { qualifiedTableName } from "@/lib/table/tableSelectSql";
 import { driverStoreFocusForInstallError } from "@/lib/connection/agentDriverInstallHint";
 import {
   canCreateConnectionNamespace,
@@ -1540,6 +1541,7 @@ async function generateDdlTemplate() {
         const result = await api.getObjectSource(target.connectionId, target.database, schema, target.label, "VIEW");
         return buildViewDdl({
           databaseType: databaseTypeForNode(target),
+          identifierQuote: connectionStore.connectionIdentifierQuote?.(target.connectionId),
           schema,
           name: target.label,
           source: result.source,
@@ -3486,7 +3488,17 @@ function createView() {
   connectionStore.activeConnectionId = node.connectionId;
   const viewName = "new_view";
   const effectiveDbType = effectiveDatabaseTypeForConnection(connectionStore.getConfig(node.connectionId));
-  const viewSqlName = effectiveDbType === "informix" || !node.schema ? viewName : `${node.schema}.${viewName}`;
+  // Quote the schema/name with the driver-reported identifier quote so
+  // hyphenated schemas (Kingbase MySQL compat) render as valid identifiers.
+  const viewSqlName =
+    effectiveDbType === "informix" || !node.schema
+      ? viewName
+      : qualifiedTableName({
+          databaseType: effectiveDbType,
+          identifierQuote: connectionStore.connectionIdentifierQuote?.(node.connectionId),
+          schema: node.schema,
+          tableName: viewName,
+        });
   const tabId = queryStore.createTab(node.connectionId, node.database, t("contextMenu.createView"), "query", node.schema, undefined, node.catalog);
   queryStore.updateSql(tabId, `CREATE VIEW ${viewSqlName} AS\nSELECT\n  *\nFROM table_name;\n`);
   queryStore.setObjectSource(tabId, {
diff --git a/apps/desktop/src/lib/__tests__/sql/newQueryContext.spec.ts b/apps/desktop/src/lib/__tests__/sql/newQueryContext.spec.ts
index b2e1a6309e..60e3032c67 100644
--- a/apps/desktop/src/lib/__tests__/sql/newQueryContext.spec.ts
+++ b/apps/desktop/src/lib/__tests__/sql/newQueryContext.spec.ts
@@ -216,6 +216,15 @@ describe("buildSelectAllSql", () => {
   it("qualifies a StarRocks external-catalog table with catalog and database", () => {
     expect(buildSelectAllSql("starrocks", { catalog: "paimon_catalog", database: "bi", tableName: "events" })).toBe("SELECT * FROM `paimon_catalog`.`bi`.`events`");
   });
+  it("uses the driver-reported identifier quote for Kingbase MySQL compat mode", () => {
+    expect(buildSelectAllSql("kingbase", { schema: "audit_schema", tableName: "events" }, "`")).toBe("SELECT * FROM `audit_schema`.`events`");
+  });
+  it("uses the driver-reported identifier quote for Kingbase PostgreSQL mode", () => {
+    expect(buildSelectAllSql("kingbase", { schema: "audit_schema", tableName: "events" }, '"')).toBe('SELECT * FROM "audit_schema"."events"');
+  });
+  it("falls back to double quotes for Kingbase when no identifier quote is reported", () => {
+    expect(buildSelectAllSql("kingbase", { schema: "audit_schema", tableName: "events" })).toBe('SELECT * FROM "audit_schema"."events"');
+  });
 });
 
 describe("isNewQueryPrefillSupported", () => {
diff --git a/apps/desktop/src/lib/sql/newQueryContext.ts b/apps/desktop/src/lib/sql/newQueryContext.ts
index 2da940d6e3..50636658fe 100644
--- a/apps/desktop/src/lib/sql/newQueryContext.ts
+++ b/apps/desktop/src/lib/sql/newQueryContext.ts
@@ -90,6 +90,7 @@ export interface ResolveNewQueryInitialSqlInput extends ResolveNewQueryTableInpu
   targetConnectionId: string;
   targetDatabase: string;
   databaseType?: DatabaseType;
+  identifierQuote?: string;
 }
 
 // Database types whose "table" view does not use standard SQL `SELECT * FROM <table>`
@@ -145,9 +146,9 @@ export function resolveNewQueryTable(input: ResolveNewQueryTableInput): NewQuery
  * the same per-dialect identifier quoting and schema/catalog qualification used
  * by the table-data view.
  */
-export function buildSelectAllSql(databaseType: DatabaseType | undefined, table: Pick<NewQueryTable, "schema" | "catalog" | "tableName"> & Partial<Pick<NewQueryTable, "database">>): string {
+export function buildSelectAllSql(databaseType: DatabaseType | undefined, table: Pick<NewQueryTable, "schema" | "catalog" | "tableName"> & Partial<Pick<NewQueryTable, "database">>, identifierQuote?: string): string {
   if (databaseType === "victoriametrics") return metricRangeQuery(table.tableName);
-  const ref = qualifiedTableName({ databaseType, database: table.database, schema: table.schema, catalog: table.catalog, tableName: table.tableName });
+  const ref = qualifiedTableName({ databaseType, identifierQuote, database: table.database, schema: table.schema, catalog: table.catalog, tableName: table.tableName });
   return `SELECT * FROM ${ref}`;
 }
 
@@ -162,5 +163,5 @@ export function resolveNewQueryInitialSql(input: ResolveNewQueryInitialSqlInput)
   const table = resolveNewQueryTable(input);
   if (!table || table.connectionId !== input.targetConnectionId || table.database !== input.targetDatabase) return undefined;
 
-  return buildSelectAllSql(input.databaseType, table);
+  return buildSelectAllSql(input.databaseType, table, input.identifierQuote);
 }
diff --git a/apps/desktop/src/lib/table/viewDdl.ts b/apps/desktop/src/lib/table/viewDdl.ts
index 7eed0c814a..5295991afc 100644
--- a/apps/desktop/src/lib/table/viewDdl.ts
+++ b/apps/desktop/src/lib/table/viewDdl.ts
@@ -6,6 +6,8 @@ export type BuildViewDdlInput = {
   schema?: string | null;
   name: string;
   source: string;
+  /** Driver-reported identifier quote (e.g. `` ` `` for Kingbase MySQL compat). */
+  identifierQuote?: string;
 };
 
 export function buildViewDdl(input: BuildViewDdlInput): Promise<string> {
diff --git a/crates/dbx-core/src/object_source_sql.rs b/crates/dbx-core/src/object_source_sql.rs
index bf3022ed5f..a8b28f5ff4 100644
--- a/crates/dbx-core/src/object_source_sql.rs
+++ b/crates/dbx-core/src/object_source_sql.rs
@@ -36,6 +36,11 @@ pub struct BuildViewDdlInput {
     pub schema: Option<String>,
     pub name: String,
     pub source: String,
+    /// Driver-reported identifier quote (e.g. `` ` `` for Kingbase MySQL
+    /// compatibility mode). When set, overrides the database_type-based quote
+    /// selection so hyphenated schemas render as valid identifiers.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub identifier_quote: Option<String>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -224,7 +229,13 @@ pub fn build_view_ddl_sql(input: BuildViewDdlInput) -> String {
         return ensure_semicolon(source);
     }
 
-    let qualified_name = if matches!(input.database_type, Some(DatabaseType::Mysql | DatabaseType::Goldendb)) {
+    // Backtick-quoting is used for MySQL-family engines and for connections
+    // whose driver reports a backtick identifier quote (Kingbase MySQL
+    // compatibility mode); everything else keeps the PostgreSQL double quote.
+    let use_backtick = input.identifier_quote.as_deref() == Some("`")
+        || (input.identifier_quote.is_none()
+            && matches!(input.database_type, Some(DatabaseType::Mysql | DatabaseType::Goldendb)));
+    let qualified_name = if use_backtick {
         mysql_qualified_name(input.schema.as_deref(), &input.name)
     } else {
         postgres_qualified_name(input.schema.as_deref(), &input.name)
@@ -1582,6 +1593,7 @@ mod tests {
             schema: Some("public".to_string()),
             name: "active users".to_string(),
             source: " SELECT id, name FROM users WHERE active ".to_string(),
+            identifier_quote: None,
         });
 
         assert_eq!(
@@ -1597,6 +1609,7 @@ mod tests {
             schema: Some("reporting".to_string()),
             name: "active_users".to_string(),
             source: "CREATE ALGORITHM=UNDEFINED VIEW `active_users` AS SELECT `id` FROM `users`".to_string(),
+            identifier_quote: None,
         });
 
         assert_eq!(sql, "CREATE ALGORITHM=UNDEFINED VIEW `active_users` AS SELECT `id` FROM `users`;");
@@ -1609,11 +1622,38 @@ mod tests {
             schema: Some("reporting".to_string()),
             name: "active_users".to_string(),
             source: "SELECT id FROM users".to_string(),
+            identifier_quote: None,
         });
 
         assert_eq!(sql, "CREATE VIEW `reporting`.`active_users` AS\nSELECT id FROM users;");
     }
 
+    #[test]
+    fn view_ddl_uses_backtick_quote_when_identifier_quote_reports_mysql_compat() {
+        let sql = build_view_ddl_sql(BuildViewDdlInput {
+            database_type: Some(DatabaseType::Kingbase),
+            schema: Some("audit-schema".to_string()),
+            name: "active_users".to_string(),
+            source: "SELECT id FROM users".to_string(),
+            identifier_quote: Some("`".to_string()),
+        });
+
+        assert_eq!(sql, "CREATE OR REPLACE VIEW `audit-schema`.`active_users` AS\nSELECT id FROM users;");
+    }
+
+    #[test]
+    fn view_ddl_keeps_double_quote_without_mysql_compat_identifier_quote() {
+        let sql = build_view_ddl_sql(BuildViewDdlInput {
+            database_type: Some(DatabaseType::Kingbase),
+            schema: Some("audit-schema".to_string()),
+            name: "active_users".to_string(),
+            source: "SELECT id FROM users".to_string(),
+            identifier_quote: None,
+        });
+
+        assert_eq!(sql, "CREATE OR REPLACE VIEW \"audit-schema\".\"active_users\" AS\nSELECT id FROM users;");
+    }
+
     #[test]
     fn oracle_package_source_saves_as_single_create_or_replace_statement() {
         let sql = build_executable_object_source_sql(EditableObjectSourceSqlInput {
diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs
index c0f2b7fcad..346ce75204 100644
--- a/crates/dbx-core/src/schema.rs
+++ b/crates/dbx-core/src/schema.rs
@@ -6397,11 +6397,16 @@ async fn get_table_ddl_core_with_options(
         )
         .await?;
         let database_type = connection_config(state, connection_id).await.map(|config| config.db_type);
+        // Kingbase MySQL compatibility mode reports a backtick identifier
+        // quote; thread it through so the view DDL wraps hyphenated schema
+        // names in backticks instead of double quotes the server rejects.
+        let identifier_quote = state.connection_identifier_quote(connection_id, Some(database)).await.ok().flatten();
         return Ok(crate::object_source_sql::build_view_ddl_sql(crate::object_source_sql::BuildViewDdlInput {
             database_type,
             schema: if schema.trim().is_empty() { None } else { Some(schema.to_string()) },
             name: table.to_string(),
             source: source.source,
+            identifier_quote,
         }));
     }
     if matches!(object_type, Some(db::ObjectSourceKind::MaterializedView)) {