karaolidis/telegraf-clickhouse-plugin
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix column type checking

Browse Source 
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
This commit is contained in:
Nick Karaolidis
2023-06-29 17:31:05 +03:00
parent 7c79958ae5
commit 8ddd67c1df
1 changed files with 23 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
plugins/outputs/clickhouse/clickhouse.go
View File

@@ -174,16 +174,17 @@ func (ch *ClickHouse) valueToDatatype(value interface{}) string {
return datatype return datatype
} }
func (ch *ClickHouse) toNullableType(pair *orderedmap.Pair[string, string]) string {
if pair.Key != "host" && pair.Key != ch.TimestampColumn && pair.Key != "measurement" {
return fmt.Sprintf("Nullable(%s)", pair.Value)
}
return pair.Value
}
func (ch *ClickHouse) generateCreateTable(tablename string, columns *orderedmap.OrderedMap[string, string]) string { func (ch *ClickHouse) generateCreateTable(tablename string, columns *orderedmap.OrderedMap[string, string]) string {
columnDefs := make([]string, 0, columns.Len()) columnDefs := make([]string, 0, columns.Len())
for pair := columns.Oldest(); pair != nil; pair = pair.Next() { for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
columnType := pair.Value columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdent(pair.Key), ch.toNullableType(pair)))
if pair.Key != "host" && pair.Key != ch.TimestampColumn && pair.Key != "measurement" {
columnType = fmt.Sprintf("Nullable(%s)", pair.Value)
}
columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdent(pair.Key), columnType))
} }
orderBy := make([]string, 0, 3) orderBy := make([]string, 0, 3)
@@ -210,21 +211,24 @@ func (ch *ClickHouse) generateCreateTable(tablename string, columns *orderedmap.
func (ch *ClickHouse) generateAlterTable(tablename string, columns *orderedmap.OrderedMap[string, string]) string { func (ch *ClickHouse) generateAlterTable(tablename string, columns *orderedmap.OrderedMap[string, string]) string {
alterDefs := make([]string, 0, columns.Len()) alterDefs := make([]string, 0, columns.Len())
modifyDefs := make([]string, 0, columns.Len())
for pair := columns.Oldest(); pair != nil; pair = pair.Next() { for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
columnType := pair.Value columnType := ch.toNullableType(pair)
if pair.Key != "host" && pair.Key != ch.TimestampColumn && pair.Key != "measurement" {
columnType = fmt.Sprintf("Nullable(%s)", pair.Value)
}
alterDefs = append(alterDefs, fmt.Sprintf("ADD COLUMN IF NOT EXISTS %s %s", alterDefs = append(alterDefs, fmt.Sprintf("ADD COLUMN IF NOT EXISTS %s %s",
quoteIdent(pair.Key), quoteIdent(pair.Key),
columnType)) columnType))
modifyDefs = append(modifyDefs, fmt.Sprintf("MODIFY COLUMN IF EXISTS %s %s",
quoteIdent(pair.Key),
columnType))
} }
return fmt.Sprintf("ALTER TABLE %s %s", return fmt.Sprintf("ALTER TABLE %s %s, %s",
quoteIdent(tablename), quoteIdent(tablename),
strings.Join(alterDefs, ",")) strings.Join(alterDefs, ","),
strings.Join(modifyDefs, ","))
} }
func (ch *ClickHouse) ensureTable(tablename string, columns *orderedmap.OrderedMap[string, string]) error { func (ch *ClickHouse) ensureTable(tablename string, columns *orderedmap.OrderedMap[string, string]) error {
@@ -251,21 +255,22 @@ func (ch *ClickHouse) ensureTable(tablename string, columns *orderedmap.OrderedM
break break
} }
tableColumns := make(map[string]struct{}) tableColumns := make(map[string]string)
for res.Next() { for res.Next() {
var name string var name string
var dtype string
var _i string var _i string
err = res.Scan(&name, &_i, &_i, &_i, &_i, &_i, &_i) err = res.Scan(&name, &dtype, &_i, &_i, &_i, &_i, &_i)
if err != nil { if err != nil {
return err return err
} }
tableColumns[name] = struct{}{} tableColumns[name] = dtype
} }
for pair := columns.Oldest(); pair != nil; pair = pair.Next() { for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
if _, ok := tableColumns[pair.Key]; !ok { if _, ok := tableColumns[pair.Key]; !ok || tableColumns[pair.Key] != ch.toNullableType(pair) {
_, err = ch.db.Exec(ch.generateAlterTable(tablename, columns)) _, err = ch.db.Exec(ch.generateAlterTable(tablename, columns))
if err != nil { if err != nil {
return err return err
@@ -289,7 +294,7 @@ func (ch *ClickHouse) generateInsert(tablename string, columns *orderedmap.Order
return fmt.Sprintf("INSERT INTO %s (%s) VALUES %s", return fmt.Sprintf("INSERT INTO %s (%s) VALUES %s",
quoteIdent(tablename), quoteIdent(tablename),
strings.Join(quotedColumns, ", "), strings.Join(quotedColumns, ","),
placeholders) placeholders)
} }
@@ -418,12 +423,6 @@ func (ch *ClickHouse) WriteSingleTable(metrics []telegraf.Metric) error {
columns.Set(colName, ch.valueToDatatype(field.Value)) columns.Set(colName, ch.valueToDatatype(field.Value))
} }
for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
if _, ok := metricEntry[pair.Key]; !ok {
metricEntry[pair.Key] = pair.Value
}
}
metricsData = append(metricsData, metricEntry) metricsData = append(metricsData, metricEntry)
} }
ch.Log.Infof("Prepared %d metrics for writing in %s\n", len(metrics), time.Since(start)) ch.Log.Infof("Prepared %d metrics for writing in %s\n", len(metrics), time.Since(start))