Skip to content

fix: harden warehouse SQL identifier quoting #7187

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rudder-devbox wants to merge 5 commits into
base: master
Choose a base branch
from
Open

fix: harden warehouse SQL identifier quoting #7187

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9750c1b
fix: harden warehouse SQL identifier quoting
rudder-devbox Jul 19, 2026
e47314f
fix: use shared postgres identifier quoting
rudder-devbox Jul 19, 2026
ca04eb1
fix: quote snowflake users staging merge
rudder-devbox Jul 19, 2026
859bd74
chore: some more identifier changes
Jul 20, 2026
c99e78f
chore: review comments
Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
201 changes: 96 additions & 105 deletions warehouse/integrations/azure-synapse/azure-synapse.go
View file
Open in desktop

Large diffs are not rendered by default.

41 changes: 22 additions & 19 deletions warehouse/integrations/bigquery/bigquery.go
View file
Open in desktop
Comment thread
krishna2020 marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -230,7 +230,7 @@ func (bq *BigQuery) createTableView(ctx context.Context, tableName string, colum
}

viewName := tableName + "_view"
query := fmt.Sprintf("CREATE OR REPLACE VIEW `%s`.`%s` AS %s;", bq.namespace, viewName, deduplicationQuery)
query := fmt.Sprintf("CREATE OR REPLACE VIEW %s AS %s;", warehouseutils.BacktickQuoteQualifiedIdentifier(bq.namespace, viewName), deduplicationQuery)

bq.logger.Infon("Creating view", logger.NewStringField("view", viewName), logger.NewStringField("query", query))
job, err := bq.db.Query(query).Run(ctx)
Expand All @@ -252,10 +252,11 @@ func (bq *BigQuery) deduplicationQuery(tableName string, columnMap model.TableSc
if column, ok := partitionKeyMap[tableName]; ok {
partitionKey = column
}
partitionKey = warehouseutils.QuoteCommaSeparatedIdentifiers(partitionKey, warehouseutils.BacktickQuoteIdentifier)

var viewOrderByStmt string
if _, ok := columnMap["loaded_at"]; ok {
viewOrderByStmt = " ORDER BY loaded_at DESC "
viewOrderByStmt = " ORDER BY " + warehouseutils.BacktickQuoteIdentifier("loaded_at") + " DESC "
}

var (
Expand Down Expand Up @@ -285,7 +286,7 @@ func (bq *BigQuery) deduplicationQuery(tableName string, columnMap model.TableSc
logger.NewStringField("partitionColumn", partitionColumn),
)
granularity = string(bqPartitionType)
partitionFilter = `TIMESTAMP_TRUNC(` + partitionColumn + `, ` + granularity + `, 'UTC')`
partitionFilter = `TIMESTAMP_TRUNC(` + warehouseutils.BacktickQuoteIdentifier(partitionColumn) + `, ` + granularity + `, 'UTC')`
} else {
bq.logger.Warnn("Deduplication query: Partition column not found in schema",
logger.NewStringField("partitionColumn", partitionColumn),
Expand All @@ -300,7 +301,7 @@ func (bq *BigQuery) deduplicationQuery(tableName string, columnMap model.TableSc
// the following view takes the last two months into consideration i.e. 60 * 60 * 24 * 60 * 1000000
viewQuery := `SELECT * EXCEPT (__row_number) FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY ` + partitionKey + viewOrderByStmt + `) AS __row_number
FROM ` + "`" + bq.projectID + "." + bq.namespace + "." + tableName + "`" + `
FROM ` + warehouseutils.BacktickQuoteQualifiedIdentifier(bq.projectID, bq.namespace, tableName) + `
WHERE
` + partitionFilter + ` BETWEEN TIMESTAMP_TRUNC(
TIMESTAMP_MICROS(UNIX_MICROS(CURRENT_TIMESTAMP()) - 60 * 60 * 24 * 60 * 1000000),
Expand Down Expand Up @@ -395,7 +396,7 @@ func checkAndIgnoreAlreadyExistError(err error) bool {

func (bq *BigQuery) DeleteBy(ctx context.Context, tableNames []string, params warehouseutils.DeleteByParams) error {
for _, tb := range tableNames {
tableName := fmt.Sprintf("`%s`.`%s`", bq.namespace, tb)
tableName := warehouseutils.BacktickQuoteQualifiedIdentifier(bq.namespace, tb)
sqlStatement := fmt.Sprintf(`
DELETE FROM
%[1]s
Expand Down Expand Up @@ -658,7 +659,7 @@ func (bq *BigQuery) LoadUserTables(ctx context.Context) (errorMap map[string]err
}

firstValueSQL := func(column string) string {
return fmt.Sprintf("FIRST_VALUE(`%[1]s` IGNORE NULLS) OVER (PARTITION BY id ORDER BY received_at DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS `%[1]s`", column)
return fmt.Sprintf("FIRST_VALUE(%[1]s IGNORE NULLS) OVER (PARTITION BY %[2]s ORDER BY %[3]s DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS %[1]s", warehouseutils.BacktickQuoteIdentifier(column), warehouseutils.BacktickQuoteIdentifier("id"), warehouseutils.BacktickQuoteIdentifier("received_at"))
}

userColMap := bq.uploader.GetTableSchemaInWarehouse(warehouseutils.UsersTable)
Expand All @@ -667,7 +668,7 @@ func (bq *BigQuery) LoadUserTables(ctx context.Context) (errorMap map[string]err
if colName == "id" {
continue
}
userColNames = append(userColNames, fmt.Sprintf("`%s`", colName))
userColNames = append(userColNames, warehouseutils.BacktickQuoteIdentifier(colName))
firstValProps = append(firstValProps, firstValueSQL(colName))
}

Expand Down Expand Up @@ -695,7 +696,7 @@ func (bq *BigQuery) LoadUserTables(ctx context.Context) (errorMap map[string]err
strings.Join(firstValProps, ","),
strings.Join(userColNames, ","),
deduplicationQuery,
fmt.Sprintf("`%s`.`%s`", bq.namespace, stagingUsersTableName),
warehouseutils.BacktickQuoteQualifiedIdentifier(bq.namespace, stagingUsersTableName),
)

log.Infon("Loading data")
Expand Down Expand Up @@ -808,13 +809,14 @@ func (bq *BigQuery) dropDanglingStagingTables(ctx context.Context) error {
SELECT
table_name
FROM
%[1]s.INFORMATION_SCHEMA.TABLES
%[1]s
WHERE
table_schema = '%[1]s'
AND table_name LIKE '%[2]s';
table_schema = %[2]s
AND table_name LIKE %[3]s;
`,
bq.namespace,
fmt.Sprintf(`%s%%`, warehouseutils.StagingTablePrefix(provider)),
warehouseutils.BacktickQuoteQualifiedIdentifier(bq.namespace, "INFORMATION_SCHEMA", "TABLES"),
warehouseutils.SQLStringLiteral(bq.namespace),
warehouseutils.SQLStringLiteral(fmt.Sprintf(`%s%%`, warehouseutils.StagingTablePrefix(provider))),
)
query := bq.db.Query(sqlStatement)
it, err := bq.db.Read(ctx, query)
Expand Down Expand Up @@ -962,19 +964,20 @@ func (bq *BigQuery) FetchSchema(ctx context.Context) (model.Schema, error) {
c.column_name,
c.data_type
FROM
%[1]s.INFORMATION_SCHEMA.TABLES as t
LEFT JOIN %[1]s.INFORMATION_SCHEMA.COLUMNS as c ON (t.table_name = c.table_name)
%[1]s as t
LEFT JOIN %[2]s as c ON (t.table_name = c.table_name)
WHERE
(t.table_type != 'VIEW')
AND
(t.table_name NOT LIKE '%s')
(t.table_name NOT LIKE %[3]s)
AND (
c.column_name != '_PARTITIONTIME'
OR c.column_name IS NULL
);
`,
bq.namespace,
fmt.Sprintf(`%s%%`, warehouseutils.StagingTablePrefix(provider)),
warehouseutils.BacktickQuoteQualifiedIdentifier(bq.namespace, "INFORMATION_SCHEMA", "TABLES"),
warehouseutils.BacktickQuoteQualifiedIdentifier(bq.namespace, "INFORMATION_SCHEMA", "COLUMNS"),
warehouseutils.SQLStringLiteral(fmt.Sprintf(`%s%%`, warehouseutils.StagingTablePrefix(provider))),
)
query := bq.db.Query(sqlStatement)

Expand Down Expand Up @@ -1130,7 +1133,7 @@ func (bq *BigQuery) DownloadIdentityRules(ctx context.Context, gzWriter *misc.GZ
batchSize := int64(10000)
var offset int64
for {
sqlStatement := fmt.Sprintf(`SELECT DISTINCT %[1]s FROM %[2]s.%[3]s LIMIT %[4]d OFFSET %[5]d`, toSelectFields, bq.namespace, tableName, batchSize, offset)
sqlStatement := fmt.Sprintf(`SELECT DISTINCT %[1]s FROM %[2]s LIMIT %[3]d OFFSET %[4]d`, toSelectFields, warehouseutils.BacktickQuoteQualifiedIdentifier(bq.namespace, tableName), batchSize, offset)
bq.logger.Infon("Downloading distinct combinations of anonymous_id, user_id",
logger.NewStringField(logfield.Query, sqlStatement),
logger.NewIntField(logfield.TotalRows, totalRows),
Expand Down
70 changes: 33 additions & 37 deletions warehouse/integrations/clickhouse/clickhouse.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,7 +352,7 @@ func (ch *Clickhouse) ColumnsWithDataTypes(tableName string, columns model.Table
for columnName, dataType := range columns {
codec := ch.getClickHouseCodecForColumnType(dataType, tableName)
columnType := ch.getClickHouseColumnTypeForSpecificTable(tableName, columnName, rudderDataTypesMapToClickHouse[dataType], slices.Contains(notNullableColumns, columnName))
arr = append(arr, fmt.Sprintf(`%q %s %s`, columnName, columnType, codec))
arr = append(arr, fmt.Sprintf(`%s %s %s`, warehouseutils.DoubleQuoteIdentifier(columnName), columnType, codec))
}
return strings.Join(arr, ",")
}
Expand Down Expand Up @@ -584,9 +584,9 @@ func (ch *Clickhouse) loadByCopyCommand(ctx context.Context, tableName string, t

strKeys := warehouseutils.GetColumnsFromTableSchema(tableSchemaInUpload)
sort.Strings(strKeys)
sortedColumnNames := strings.Join(strKeys, ",")
sortedColumnNames := warehouseutils.DoubleQuoteAndJoinByComma(strKeys)
sortedColumnNamesWithDataTypes := warehouseutils.JoinWithFormatting(strKeys, func(idx int, name string) string {
return fmt.Sprintf(`%s %s`, name, rudderDataTypesMapToClickHouse[tableSchemaInUpload[name]])
return fmt.Sprintf(`%s %s`, warehouseutils.DoubleQuoteIdentifier(name), rudderDataTypesMapToClickHouse[tableSchemaInUpload[name]])
}, ",")

csvObjectLocation, err := ch.Uploader.GetSampleLoadFileLocation(ctx, tableName)
Expand All @@ -602,31 +602,30 @@ func (ch *Clickhouse) loadByCopyCommand(ctx context.Context, tableName string, t
}

sqlStatement := fmt.Sprintf(`
INSERT INTO %[1]q.%[2]q (
%[3]s
INSERT INTO %[1]s (
%[2]s
)
SELECT
*
FROM
s3(
'%[4]s',
%[3]s,
'%[4]s',
'%[5]s',
'%[6]s',
'CSV',
'%[7]s',
'%[6]s',
'gz'
)
settings
date_time_input_format = 'best_effort',
input_format_csv_arrays_as_nested_csv = 1;
`,
ch.Namespace, // 1
tableName, // 2
sortedColumnNames, // 3
loadFolder, // 4
accessKeyID, // 5
secretAccessKey, // 6
sortedColumnNamesWithDataTypes, // 7
warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Namespace, tableName), // 1
sortedColumnNames, // 2
warehouseutils.SQLStringLiteral(loadFolder), // 3
accessKeyID, // 4
secretAccessKey, // 5
sortedColumnNamesWithDataTypes, // 6
)
_, err = ch.DB.ExecContext(ctx, sqlStatement)
if err != nil {
Expand Down Expand Up @@ -691,7 +690,7 @@ func (ch *Clickhouse) loadTablesFromFilesNamesWithRetry(ctx context.Context, tab
sortedColumnKeys := warehouseutils.SortColumnKeysFromColumnMap(tableSchemaInUpload)
sortedColumnString := warehouseutils.DoubleQuoteAndJoinByComma(sortedColumnKeys)

sqlStatement := fmt.Sprintf(`INSERT INTO %q.%q (%v) VALUES (%s)`, ch.Namespace, tableName, sortedColumnString, generateArgumentString(len(sortedColumnKeys)))
sqlStatement := fmt.Sprintf(`INSERT INTO %s (%v) VALUES (%s)`, warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Namespace, tableName), sortedColumnString, generateArgumentString(len(sortedColumnKeys)))
ch.logger.Debugn("Preparing statement exec in db for loading in table",
logger.NewStringField("identifier", ch.GetLogIdentifier(tableName)),
logger.NewStringField(logfield.Query, sqlStatement),
Expand Down Expand Up @@ -853,7 +852,7 @@ func (ch *Clickhouse) createUsersTable(ctx context.Context, name string, columns
engineOptions := ""
cluster := ch.Warehouse.GetStringDestinationConfig(ch.conf, model.ClusterSetting)
if len(strings.TrimSpace(cluster)) > 0 {
clusterClause = fmt.Sprintf(`ON CLUSTER %q`, cluster)
clusterClause = fmt.Sprintf(`ON CLUSTER %s`, warehouseutils.DoubleQuoteIdentifier(cluster))
engine = fmt.Sprintf(`%s%s`, "Replicated", engine)
engineOptions = fmt.Sprintf(`'/clickhouse/{cluster}/tables/%s/{database}/{table}', '{replica}'`, uuid.New().String())
}
Expand All @@ -862,7 +861,7 @@ func (ch *Clickhouse) createUsersTable(ctx context.Context, name string, columns
return fmt.Errorf("getting partition by clause: %w", err)
}

sqlStatement := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %q.%q %s ( %v ) ENGINE = %s(%s) ORDER BY %s %s`, ch.Namespace, name, clusterClause, ch.ColumnsWithDataTypes(name, columns, notNullableColumns), engine, engineOptions, getSortKeyTuple(sortKeyFields), partitionByClause)
sqlStatement := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s %s ( %v ) ENGINE = %s(%s) ORDER BY %s %s`, warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Namespace, name), clusterClause, ch.ColumnsWithDataTypes(name, columns, notNullableColumns), engine, engineOptions, getSortKeyTuple(sortKeyFields), partitionByClause)
ch.logger.Infon("CH: Creating table in clickhouse for ch",
logger.NewStringField(logfield.DestinationID, ch.Warehouse.Destination.ID),
logger.NewStringField(logfield.Query, sqlStatement),
Expand Down Expand Up @@ -904,9 +903,9 @@ func getSortKeyTuple(sortKeyFields []string) string {
tuple.WriteString("(")
for index, field := range sortKeyFields {
if index == len(sortKeyFields)-1 {
tuple.WriteString(fmt.Sprintf(`%q`, field))
tuple.WriteString(warehouseutils.DoubleQuoteIdentifier(field))
} else {
tuple.WriteString(fmt.Sprintf(`%q,`, field))
tuple.WriteString(fmt.Sprintf(`%s,`, warehouseutils.DoubleQuoteIdentifier(field)))
}
}
tuple.WriteString(")")
Expand All @@ -932,7 +931,7 @@ func (ch *Clickhouse) CreateTable(ctx context.Context, tableName string, columns
engineOptions := ""
cluster := ch.Warehouse.GetStringDestinationConfig(ch.conf, model.ClusterSetting)
if len(strings.TrimSpace(cluster)) > 0 {
clusterClause = fmt.Sprintf(`ON CLUSTER %q`, cluster)
clusterClause = fmt.Sprintf(`ON CLUSTER %s`, warehouseutils.DoubleQuoteIdentifier(cluster))
engine = fmt.Sprintf(`%s%s`, "Replicated", engine)
engineOptions = fmt.Sprintf(`'/clickhouse/{cluster}/tables/%s/{database}/{table}', '{replica}'`, uuid.New().String())
}
Expand All @@ -949,7 +948,7 @@ func (ch *Clickhouse) CreateTable(ctx context.Context, tableName string, columns
}
}

sqlStatement = fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %q.%q %s ( %v ) ENGINE = %s(%s) %s %s`, ch.Namespace, tableName, clusterClause, ch.ColumnsWithDataTypes(tableName, columns, sortKeyFields), engine, engineOptions, orderByClause, partitionByClause)
sqlStatement = fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s %s ( %v ) ENGINE = %s(%s) %s %s`, warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Namespace, tableName), clusterClause, ch.ColumnsWithDataTypes(tableName, columns, sortKeyFields), engine, engineOptions, orderByClause, partitionByClause)

ch.logger.Infon("CH: Creating table in clickhouse for ch",
logger.NewStringField(logfield.DestinationID, ch.Warehouse.Destination.ID),
Expand All @@ -960,7 +959,7 @@ func (ch *Clickhouse) CreateTable(ctx context.Context, tableName string, columns
}

func (ch *Clickhouse) DropTable(ctx context.Context, tableName string) (err error) {
sqlStatement := fmt.Sprintf(`DROP TABLE %q.%q %s `, ch.Warehouse.Namespace, tableName, ch.clusterClause())
sqlStatement := fmt.Sprintf(`DROP TABLE %s %s `, warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Warehouse.Namespace, tableName), ch.clusterClause())
_, err = ch.DB.ExecContext(ctx, sqlStatement)
return err
}
Expand All @@ -972,10 +971,9 @@ func (ch *Clickhouse) AddColumns(ctx context.Context, tableName string, columnsI
)

queryBuilder.WriteString(fmt.Sprintf(`
ALTER TABLE
%q.%q %s`,
ch.Namespace,
tableName,
ALTER TABLE
%s %s`,
warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Namespace, tableName),
ch.clusterClause(),
))

Expand All @@ -986,7 +984,7 @@ func (ch *Clickhouse) AddColumns(ctx context.Context, tableName string, columnsI
rudderDataTypesMapToClickHouse[columnInfo.Type],
false,
)
queryBuilder.WriteString(fmt.Sprintf(` ADD COLUMN IF NOT EXISTS %q %s,`, columnInfo.Name, columnType))
queryBuilder.WriteString(fmt.Sprintf(` ADD COLUMN IF NOT EXISTS %s %s,`, warehouseutils.DoubleQuoteIdentifier(columnInfo.Name), columnType))
}

query = strings.TrimSuffix(queryBuilder.String(), ",")
Expand Down Expand Up @@ -1028,7 +1026,7 @@ func (ch *Clickhouse) CreateSchema(ctx context.Context) error {
logger.NewStringField("clusterClause", ch.clusterClause()),
)

query := fmt.Sprintf(`CREATE DATABASE IF NOT EXISTS %q %s`, ch.Namespace, ch.clusterClause())
query := fmt.Sprintf(`CREATE DATABASE IF NOT EXISTS %s %s`, warehouseutils.DoubleQuoteIdentifier(ch.Namespace), ch.clusterClause())
if _, err = db.ExecContext(ctx, query); err != nil {
return fmt.Errorf("creating database: %v", err)
}
Expand All @@ -1037,7 +1035,7 @@ func (ch *Clickhouse) CreateSchema(ctx context.Context) error {

func (ch *Clickhouse) clusterClause() string {
if cluster := ch.Warehouse.GetStringDestinationConfig(ch.conf, model.ClusterSetting); len(strings.TrimSpace(cluster)) > 0 {
return fmt.Sprintf(`ON CLUSTER %q`, cluster)
return fmt.Sprintf(`ON CLUSTER %s`, warehouseutils.DoubleQuoteIdentifier(cluster))
}
return ""
}
Expand Down Expand Up @@ -1202,10 +1200,9 @@ func (ch *Clickhouse) totalCountIntable(ctx context.Context, tableName string) (
sqlStatement string
)
sqlStatement = fmt.Sprintf(`
SELECT count(*) FROM "%[1]s"."%[2]s";
SELECT count(*) FROM %s;
`,
ch.Namespace,
tableName,
warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Namespace, tableName),
)
err = ch.DB.QueryRowContext(ctx, sqlStatement).Scan(&total)
return total, err
Expand Down Expand Up @@ -1244,10 +1241,9 @@ func (ch *Clickhouse) TestLoadTable(ctx context.Context, _, tableName string, pa
columns = append(columns, key)
}

sqlStatement := fmt.Sprintf(`INSERT INTO %q.%q (%v) VALUES (%s)`,
ch.Namespace,
tableName,
strings.Join(columns, ","),
sqlStatement := fmt.Sprintf(`INSERT INTO %s (%v) VALUES (%s)`,
warehouseutils.DoubleQuoteQualifiedIdentifier(ch.Namespace, tableName),
warehouseutils.DoubleQuoteAndJoinByComma(columns),
generateArgumentString(len(columns)),
)
txn, err := ch.DB.BeginTx(ctx, &sql.TxOptions{})
Expand Down
Loading
Loading