chore(outputs.sql): Make default create table template suitable for ClickHouse

This changes the default TableTemplate of the SQL output plugin so that
it can be used with ClickHouse. Previously it was invalid for ClickHouse
because it didn't include a PRIMARY KEY or ORDER BY clause, which is
required for ClickHouse.

This then also allows to remove the custom TableTemplate from the
ClickHouse unit test.

This also adds another placeholder {TAG_COLUMNS} that allows to get a
list of all tag columns in a TableTemplate. This can then be used to use
the tag columns as a primary or sort key, especially in ClickHouse.

Author: AndreKR

plugins/outputs/sql/README.md

@@ -99,6 +99,7 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ##  {TABLE} - table name as a quoted identifier
   ##  {TABLELITERAL} - table name as a quoted string literal
   ##  {COLUMNS} - column definitions (list of quoted identifiers and types)
+  ##  {TAG_COLUMNS} - tag column definitions (list of quoted identifiers)
   # table_template = "CREATE TABLE {TABLE}({COLUMNS})"
 
   ## Table existence check template

plugins/outputs/sql/sample.conf

@@ -18,6 +18,7 @@
   ##  {TABLE} - table name as a quoted identifier
   ##  {TABLELITERAL} - table name as a quoted string literal
   ##  {COLUMNS} - column definitions (list of quoted identifiers and types)
+  ##  {TAG_COLUMNS} - tag column definitions (list of quoted identifiers)
   # table_template = "CREATE TABLE {TABLE}({COLUMNS})"
 
   ## Table existence check template

plugins/outputs/sql/sql.go

@@ -153,13 +153,15 @@ func (p *SQL) deriveDatatype(value interface{}) string {
 
 func (p *SQL) generateCreateTable(metric telegraf.Metric) string {
 	columns := make([]string, 0, len(metric.TagList())+len(metric.FieldList())+1)
+	tagColumns := make([]string, 0, len(metric.TagList()))
 
 	if p.TimestampColumn != "" {
 		columns = append(columns, fmt.Sprintf("%s %s", quoteIdent(p.TimestampColumn), p.Convert.Timestamp))
 	}
 
 	for _, tag := range metric.TagList() {
 		columns = append(columns, fmt.Sprintf("%s %s", quoteIdent(tag.Key), p.Convert.Text))
+		tagColumns = append(tagColumns, quoteIdent(tag.Key))
 	}
 
 	var datatype string
@@ -172,6 +174,7 @@ func (p *SQL) generateCreateTable(metric telegraf.Metric) string {
 	query = strings.ReplaceAll(query, "{TABLE}", quoteIdent(metric.Name()))
 	query = strings.ReplaceAll(query, "{TABLELITERAL}", quoteStr(metric.Name()))
 	query = strings.ReplaceAll(query, "{COLUMNS}", strings.Join(columns, ","))
+	query = strings.ReplaceAll(query, "{TAG_COLUMNS}", strings.Join(tagColumns, ","))
 
 	return query
 }
@@ -274,13 +277,25 @@ func (p *SQL) Write(metrics []telegraf.Metric) error {
 	return nil
 }
 
+func (p *SQL) Init() error {
+	if p.TableTemplate == "" {
+		if p.Driver == "clickhouse" {
+			p.TableTemplate = "CREATE TABLE {TABLE}({COLUMNS}) ORDER BY ({TAG_COLUMNS}, " + p.TimestampColumn + ")"
+		} else {
+			p.TableTemplate = "CREATE TABLE {TABLE}({COLUMNS})"
+		}
+	}
+
+	return nil
+}
+
 func init() {
 	outputs.Add("sql", func() telegraf.Output { return newSQL() })
 }
 
 func newSQL() *SQL {
 	return &SQL{
-		TableTemplate:       "CREATE TABLE {TABLE}({COLUMNS})",
+		TableTemplate:       "",
 		TableExistsTemplate: "SELECT 1 FROM {TABLE} LIMIT 1",
 		TimestampColumn:     "timestamp",
 		Convert: ConvertStruct{

plugins/outputs/sql/sql_test.go

@@ -201,6 +201,8 @@ func TestMysqlIntegration(t *testing.T) {
 	p.Driver = "mysql"
 	p.DataSourceName = address
 	p.InitSQL = "SET sql_mode='ANSI_QUOTES';"
+	err = p.Init()
+	require.NoError(t, err)
 
 	require.NoError(t, p.Connect())
 	require.NoError(t, p.Write(
@@ -287,6 +289,8 @@ func TestPostgresIntegration(t *testing.T) {
 	p.Convert.Real = "double precision"
 	p.Convert.Unsigned = "bigint"
 	p.Convert.ConversionStyle = "literal"
+	err = p.Init()
+	require.NoError(t, err)
 
 	require.NoError(t, p.Connect())
 	defer p.Close()
@@ -372,14 +376,15 @@ func TestClickHouseIntegration(t *testing.T) {
 	p.Log = testutil.Logger{}
 	p.Driver = "clickhouse"
 	p.DataSourceName = address
-	p.TableTemplate = "CREATE TABLE {TABLE}({COLUMNS}) ENGINE MergeTree() ORDER by timestamp"
 	p.Convert.Integer = "Int64"
 	p.Convert.Text = "String"
 	p.Convert.Timestamp = "DateTime"
 	p.Convert.Defaultvalue = "String"
 	p.Convert.Unsigned = "UInt64"
 	p.Convert.Bool = "UInt8"
 	p.Convert.ConversionStyle = "literal"
+	err = p.Init()
+	require.NoError(t, err)
 
 	require.NoError(t, p.Connect())
 	require.NoError(t, p.Write(testMetrics))

plugins/outputs/sql/sqlite_test.go

@@ -25,6 +25,8 @@ func TestSqlite(t *testing.T) {
 	p.Log = testutil.Logger{}
 	p.Driver = "sqlite"
 	p.DataSourceName = address
+	err := p.Init()
+	require.NoError(t, err)
 
 	require.NoError(t, p.Connect())
 	defer p.Close()