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

Optimize NULL handling

Browse Source 
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
This commit is contained in:
Nick Karaolidis
2023-07-06 14:35:11 +03:00
parent 318155561b
commit 7c9a6cc6e0
1 changed files with 11 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -158,11 +158,12 @@ func (ch *ClickHouse) toDatatype(value interface{}) string {
return datatype
}
func (ch *ClickHouse) toNullable(pair *orderedmap.Pair[string, string]) string {
if pair.Key != "host" && pair.Key != ch.TimestampColumn && pair.Key != "measurement" {
return fmt.Sprintf("Nullable(%s)", pair.Value)
func (ch *ClickHouse) toNullableDatatype(key string, value interface{}) string {
datatype := ch.toDatatype(value)
if key != "host" && key != ch.TimestampColumn && key != "measurement" {
return fmt.Sprintf("Nullable(%s)", datatype)
}
return pair.Value
return datatype
}
func (ch *ClickHouse) pepareMetrics(metrics []telegraf.Metric, metricsData map[string][]map[string]interface{}, columns map[string]*orderedmap.OrderedMap[string, string]) {
@@ -188,12 +189,12 @@ func (ch *ClickHouse) pepareMetrics(metrics []telegraf.Metric, metricsData map[s
for _, tag := range metric.TagList() {
metricEntry[tag.Key] = tag.Value
columns[tablename].Set(tag.Key, ch.toDatatype(tag.Value))
columns[tablename].Set(tag.Key, ch.toNullableDatatype(tag.Key, tag.Value))
}
for _, field := range metric.FieldList() {
metricEntry[field.Key] = field.Value
columns[tablename].Set(field.Key, ch.toDatatype(field.Value))
columns[tablename].Set(field.Key, ch.toNullableDatatype(field.Key, field.Value))
}
metricsData[tablename] = append(metricsData[tablename], metricEntry)
@@ -203,7 +204,7 @@ func (ch *ClickHouse) pepareMetrics(metrics []telegraf.Metric, metricsData map[s
func (ch *ClickHouse) generateCreateTable(tablename string, columns *orderedmap.OrderedMap[string, string]) string {
columnDefs := make([]string, 0, columns.Len())
for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdent(pair.Key), ch.toNullable(pair)))
columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdent(pair.Key), pair.Value))
}
orderBy := make([]string, 0, 2)
@@ -230,15 +231,14 @@ func (ch *ClickHouse) generateAlterTable(tablename string, columns *orderedmap.O
modifyDefs := make([]string, 0, columns.Len())
for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
columnType := ch.toNullable(pair)
alterDefs = append(alterDefs, fmt.Sprintf("ADD COLUMN IF NOT EXISTS %s %s",
quoteIdent(pair.Key),
columnType))
pair.Value))
modifyDefs = append(modifyDefs, fmt.Sprintf("MODIFY COLUMN IF EXISTS %s %s",
quoteIdent(pair.Key),
columnType))
pair.Value))
}
return fmt.Sprintf("ALTER TABLE %s %s, %s",
@@ -286,7 +286,7 @@ func (ch *ClickHouse) ensureTable(tablename string, columns *orderedmap.OrderedM
}
for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
if _, ok := tableColumns[pair.Key]; !ok || tableColumns[pair.Key] != ch.toNullable(pair) {
if _, ok := tableColumns[pair.Key]; !ok || tableColumns[pair.Key] != pair.Value {
_, err = ch.db.Exec(ch.generateAlterTable(tablename, columns))
if err != nil {
return err