From 924af86e5880b3d3c0741c8ea1b25f21307ea32d Mon Sep 17 00:00:00 2001 From: Parag Jain Date: Fri, 17 Jan 2025 21:23:04 +0530 Subject: [PATCH 1/4] use literals in distributed engine definition in ch --- runtime/drivers/clickhouse/olap.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/runtime/drivers/clickhouse/olap.go b/runtime/drivers/clickhouse/olap.go index 494160a2628..192104af7be 100644 --- a/runtime/drivers/clickhouse/olap.go +++ b/runtime/drivers/clickhouse/olap.go @@ -609,7 +609,7 @@ func (c *connection) createTable(ctx context.Context, name, sql string, outputPr database = "currentDatabase()" } 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'", c.config.Cluster, database, localTableName(name)) if outputProps.DistributedShardingKey != "" { fmt.Fprintf(&distributed, ", %s", outputProps.DistributedShardingKey) } else { From e3597f80f17c8e3880c6271ca40912679f463191 Mon Sep 17 00:00:00 2001 From: Parag Jain Date: Fri, 17 Jan 2025 21:31:15 +0530 Subject: [PATCH 2/4] do not quote db --- runtime/drivers/clickhouse/olap.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/runtime/drivers/clickhouse/olap.go b/runtime/drivers/clickhouse/olap.go index 192104af7be..3bf79c2e433 100644 --- a/runtime/drivers/clickhouse/olap.go +++ b/runtime/drivers/clickhouse/olap.go @@ -609,7 +609,7 @@ func (c *connection) createTable(ctx context.Context, name, sql string, outputPr database = "currentDatabase()" } fmt.Fprintf(&distributed, "CREATE OR REPLACE TABLE %s %s AS %s", safeSQLName(name), onClusterClause, safelocalTableName(name)) - fmt.Fprintf(&distributed, " ENGINE = Distributed('%s', '%s', '%s'", c.config.Cluster, database, localTableName(name)) + fmt.Fprintf(&distributed, " ENGINE = Distributed('%s', %s, '%s'", c.config.Cluster, database, localTableName(name)) if outputProps.DistributedShardingKey != "" { fmt.Fprintf(&distributed, ", %s", outputProps.DistributedShardingKey) } else { From 9e27f91b7e57fbc5954dbc3a0da069987f52c018 Mon Sep 17 00:00:00 2001 From: Parag Jain Date: Sat, 18 Jan 2025 00:22:27 +0530 Subject: [PATCH 3/4] use safe sql string --- runtime/drivers/clickhouse/olap.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/runtime/drivers/clickhouse/olap.go b/runtime/drivers/clickhouse/olap.go index 3bf79c2e433..2e7c34211c8 100644 --- a/runtime/drivers/clickhouse/olap.go +++ b/runtime/drivers/clickhouse/olap.go @@ -609,7 +609,7 @@ func (c *connection) createTable(ctx context.Context, name, sql string, outputPr database = "currentDatabase()" } fmt.Fprintf(&distributed, "CREATE OR REPLACE TABLE %s %s AS %s", safeSQLName(name), onClusterClause, safelocalTableName(name)) - fmt.Fprintf(&distributed, " ENGINE = Distributed('%s', %s, '%s'", c.config.Cluster, database, localTableName(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) +} From dfa62a5ddc4fe8c9cdb25a94b0acab2fbeb16fdb Mon Sep 17 00:00:00 2001 From: Parag Jain Date: Sat, 18 Jan 2025 21:03:35 +0530 Subject: [PATCH 4/4] quote database --- runtime/drivers/clickhouse/olap.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/runtime/drivers/clickhouse/olap.go b/runtime/drivers/clickhouse/olap.go index 2e7c34211c8..b9021ce5112 100644 --- a/runtime/drivers/clickhouse/olap.go +++ b/runtime/drivers/clickhouse/olap.go @@ -604,9 +604,9 @@ 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", safeSQLString(c.config.Cluster), database, safeSQLString(localTableName(name)))