From afb3678e0c16c7dde344fbcd614e4b0af9a711e8 Mon Sep 17 00:00:00 2001 From: 81reap Date: Tue, 25 Aug 2026 23:18:59 -0400 Subject: [PATCH 1/2] fix(outputs.parquet): Reserve the timestamp column for the metric time Telegraf reserves an int64 `timestamp_field_name` column for the metric time, but gets conflicting names would carry the same name and two different types. A bad input would go to the int64 builder and panic. (eg :: `demo timestamp="x"`) Now telegraf drops fields or tags that collide with reserved column names and log the exception. --- plugins/outputs/parquet/README.md | 8 +++++++ plugins/outputs/parquet/parquet.go | 11 ++++++++++ plugins/outputs/parquet/parquet_test.go | 28 +++++++++++++++++++++++++ 3 files changed, 47 insertions(+) diff --git a/plugins/outputs/parquet/README.md b/plugins/outputs/parquet/README.md index 19627cd5da2c4..d765defd3e3c1 100644 --- a/plugins/outputs/parquet/README.md +++ b/plugins/outputs/parquet/README.md @@ -62,6 +62,14 @@ 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. +Since column types are fixed at file creation, when an unknown value is logged +it will be logged as `null` and once per column. +Inputs that collide with reserved columns like `timestamp_field_name` are +dropped and logged. Rename `timestamp_field_name` or your field to fix it. + +Since parquet column schemas are fixed at file creation, new values that +do not fit the column schema are written as `null` and logged once per column. + ### Write The plugin makes use of the buffered writer. This may buffer some metrics into diff --git a/plugins/outputs/parquet/parquet.go b/plugins/outputs/parquet/parquet.go index a4cf876d0c45b..f6be53b3dfaf4 100644 --- a/plugins/outputs/parquet/parquet.go +++ b/plugins/outputs/parquet/parquet.go @@ -258,6 +258,17 @@ func (p *Parquet) createSchema(metrics []telegraf.Metric) (*arrow.Schema, error) } } + if p.TimestampFieldName != "" { + if _, taken := rawFields[p.TimestampFieldName]; taken { + delete(rawFields, 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", + p.TimestampFieldName, + ) + } + } + fields := make([]arrow.Field, 0) for key, value := range rawFields { fields = append(fields, arrow.Field{ diff --git a/plugins/outputs/parquet/parquet_test.go b/plugins/outputs/parquet/parquet_test.go index b063ce0997605..3d675a31fe388 100644 --- a/plugins/outputs/parquet/parquet_test.go +++ b/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()) +} From aa5e46e670837e6655bde0d00bfeba0166bd0149 Mon Sep 17 00:00:00 2001 From: 81reap Date: Wed, 26 Aug 2026 16:24:28 -0400 Subject: [PATCH 2/2] rev2 (please squash + merge) --- plugins/outputs/parquet/README.md | 10 +++------- plugins/outputs/parquet/parquet.go | 16 +++++----------- 2 files changed, 8 insertions(+), 18 deletions(-) diff --git a/plugins/outputs/parquet/README.md b/plugins/outputs/parquet/README.md index d765defd3e3c1..7fc4d3f36efb0 100644 --- a/plugins/outputs/parquet/README.md +++ b/plugins/outputs/parquet/README.md @@ -62,13 +62,9 @@ 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. -Since column types are fixed at file creation, when an unknown value is logged -it will be logged as `null` and once per column. -Inputs that collide with reserved columns like `timestamp_field_name` are -dropped and logged. Rename `timestamp_field_name` or your field to fix it. - -Since parquet column schemas are fixed at file creation, new values that -do not fit the column schema are written as `null` and logged once per column. +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 diff --git a/plugins/outputs/parquet/parquet.go b/plugins/outputs/parquet/parquet.go index f6be53b3dfaf4..6069b62f2baae 100644 --- a/plugins/outputs/parquet/parquet.go +++ b/plugins/outputs/parquet/parquet.go @@ -258,19 +258,13 @@ func (p *Parquet) createSchema(metrics []telegraf.Metric) (*arrow.Schema, error) } } - if p.TimestampFieldName != "" { - if _, taken := rawFields[p.TimestampFieldName]; taken { - delete(rawFields, 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", - p.TimestampFieldName, - ) - } - } - 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,