diff --git a/plugins/outputs/parquet/README.md b/plugins/outputs/parquet/README.md index 19627cd5da2c4..7fc4d3f36efb0 100644 --- a/plugins/outputs/parquet/README.md +++ b/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 diff --git a/plugins/outputs/parquet/parquet.go b/plugins/outputs/parquet/parquet.go index a4cf876d0c45b..6069b62f2baae 100644 --- a/plugins/outputs/parquet/parquet.go +++ b/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, 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()) +}