use literals in distributed engine definition in ch (#6446)

* use literals in distributed engine definition in ch

* do not quote db

* use safe sql string

* quote database

runtime/drivers/clickhouse/olap.go

@@ -604,12 +604,12 @@ func (c *connection) createTable(ctx context.Context, name, sql string, outputPr
 	}
 	// create the distributed table
 	var distributed strings.Builder
-	database := c.config.Database
-	if c.config.Database == "" {
-		database = "currentDatabase()"
+	database := "currentDatabase()"
+	if c.config.Database != "" {
+		database = safeSQLString(c.config.Database)
 	}
 	fmt.Fprintf(&distributed, "CREATE OR REPLACE TABLE %s %s AS %s", safeSQLName(name), onClusterClause, safelocalTableName(name))
-	fmt.Fprintf(&distributed, " ENGINE = Distributed(%s, %s, %s", safeSQLName(c.config.Cluster), database, safelocalTableName(name))
+	fmt.Fprintf(&distributed, " ENGINE = Distributed(%s, %s, %s", safeSQLString(c.config.Cluster), database, safeSQLString(localTableName(name)))
 	if outputProps.DistributedShardingKey != "" {
 		fmt.Fprintf(&distributed, ", %s", outputProps.DistributedShardingKey)
 	} else {
@@ -1192,3 +1192,7 @@ func localTableName(name string) string {
 func tempTableForDictionary(name string) string {
 	return name + "_dict_temp_"
 }
+
+func safeSQLString(name string) string {
+	return drivers.DialectClickHouse.EscapeStringValue(name)
+}