fix(outputs.parquet): Reserve the timestamp column for the metric time

plugins/outputs/parquet/README.md

@@ -62,6 +62,10 @@ When writing to a file, the schema is used to look for each value and if it is
 not present a null value is added. The result is that if additional fields are
 present after the first metric flush those fields are omitted.
 
+The `timestamp_field_name` column holds the metric time, so a field or tag with
+that same name is dropped and logged. Set `timestamp_field_name` to another name
+or rename the field or tag to keep both.
+
 ### Write
 
 The plugin makes use of the buffered writer. This may buffer some metrics into

plugins/outputs/parquet/parquet.go

@@ -260,6 +260,11 @@ func (p *Parquet) createSchema(metrics []telegraf.Metric) (*arrow.Schema, error)
 
 	fields := make([]arrow.Field, 0)
 	for key, value := range rawFields {
+		if p.TimestampFieldName != "" && key == p.TimestampFieldName {
+			p.Log.Warnf("Ignoring the %q field or tag as that column holds the metric time; "+
+				"set 'timestamp_field_name' to another name to keep it", key)
+			continue
+		}
 		fields = append(fields, arrow.Field{
 			Name:     key,
 			Type:     value,

plugins/outputs/parquet/parquet_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/apache/arrow-go/v18/parquet"
 	"github.com/apache/arrow-go/v18/parquet/file"
 	"github.com/stretchr/testify/require"
 
@@ -276,3 +277,30 @@ func TestMissingValuesReadBackAsNull(t *testing.T) {
 		require.Equalf(t, int16(1), column.MaxDefinitionLevel(), "column %q is required, nulls would be written as zero", column.Name())
 	}
 }
+
+func TestTimestampFieldNameCollisionKeepsOneColumn(t *testing.T) {
+	dir := t.TempDir()
+	p := &Parquet{Directory: dir, TimestampFieldName: "timestamp", Log: testutil.Logger{}}
+	require.NoError(t, p.Init())
+
+	require.NoError(t, p.Write([]telegraf.Metric{
+		metric.New("demo", nil, map[string]interface{}{"timestamp": "x", "value": int64(1)}, time.Now()),
+	}))
+	require.NoError(t, p.Close())
+
+	written, err := filepath.Glob(filepath.Join(dir, "*.parquet"))
+	require.NoError(t, err)
+	require.Len(t, written, 1)
+
+	reader, err := file.OpenParquetFile(written[0], false)
+	require.NoError(t, err)
+	defer reader.Close()
+
+	schema := reader.MetaData().Schema
+	names := make([]string, 0, schema.NumColumns())
+	for i := 0; i < schema.NumColumns(); i++ {
+		names = append(names, schema.Column(i).Name())
+	}
+	require.ElementsMatch(t, []string{"value", "timestamp"}, names)
+	require.Equal(t, parquet.Types.Int64, schema.Column(schema.ColumnIndexByName("timestamp")).PhysicalType())
+}