From 8d89f84741bd896f56745b3a83cb1740a8758839 Mon Sep 17 00:00:00 2001 From: 81reap Date: Tue, 25 Aug 2026 23:10:45 -0400 Subject: [PATCH] fix(outputs.parquet): Write null instead of panicking on a type mismatch Parquet files have fixed schemas once created. A value that didn't match the column schema would panic on the output flush goroutine (eg :: (1) `demo k-1i` (2) `demo,k=v other=2i`). Since this is uncaught, it would crash Telegraf. Now values that don't fit the column schema are logged as null and one per column without crashing. The record batch is now also released when the write fails, which previously leaked its arrow buffers. --- plugins/outputs/parquet/README.md | 3 + plugins/outputs/parquet/parquet.go | 169 ++++++++++++------------ plugins/outputs/parquet/parquet_test.go | 58 ++++++++ 3 files changed, 149 insertions(+), 81 deletions(-) diff --git a/plugins/outputs/parquet/README.md b/plugins/outputs/parquet/README.md index 19627cd5da2c4..05a12f89d4605 100644 --- a/plugins/outputs/parquet/README.md +++ b/plugins/outputs/parquet/README.md @@ -62,6 +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. + ### 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..81c35d296405d 100644 --- a/plugins/outputs/parquet/parquet.go +++ b/plugins/outputs/parquet/parquet.go @@ -27,6 +27,7 @@ var defaultTimestampFieldName = "timestamp" type metricGroup struct { filename string + warned map[string]bool builder *array.RecordBuilder schema *arrow.Schema writer *pqarrow.FileWriter @@ -106,6 +107,7 @@ func (p *Parquet) Write(metrics []telegraf.Metric) error { p.metricGroups[name] = &metricGroup{ builder: array.NewRecordBuilder(memory.DefaultAllocator, schema), filename: filename, + warned: make(map[string]bool), schema: schema, writer: writer, } @@ -117,14 +119,13 @@ func (p *Parquet) Write(metrics []telegraf.Metric) error { } } - record, err := p.createRecordBatch(metrics, p.metricGroups[name].builder, p.metricGroups[name].schema) + group := p.metricGroups[name] + record := p.createRecordBatch(group, metrics) + err := group.writer.WriteBuffered(record) + record.Release() if err != nil { - return fmt.Errorf("failed to create record for file %q: %w", p.metricGroups[name].filename, err) - } - if err = p.metricGroups[name].writer.WriteBuffered(record); err != nil { - return fmt.Errorf("failed to write to file %q: %w", p.metricGroups[name].filename, err) + return fmt.Errorf("failed to write to file %q: %w", group.filename, err) } - record.Release() } return nil @@ -154,89 +155,95 @@ func (p *Parquet) rotateIfNeeded(name string) error { return nil } -func (p *Parquet) createRecordBatch(metrics []telegraf.Metric, builder *array.RecordBuilder, schema *arrow.Schema) (arrow.RecordBatch, error) { - for index, col := range schema.Fields() { +func (p *Parquet) createRecordBatch(group *metricGroup, metrics []telegraf.Metric) arrow.RecordBatch { + for index, column := range group.schema.Fields() { + builder := group.builder.Field(index) + for _, m := range metrics { - if p.TimestampFieldName != "" && col.Name == p.TimestampFieldName { - builder.Field(index).(*array.Int64Builder).Append(m.Time().UnixNano()) - continue + value := p.valueFor(m, column.Name) + if !appendValue(builder, value) { + p.warnOncef( + group, column.Name, + "Writing null for column %q of file %q as a %T value does not fit its %s column", + column.Name, group.filename, value, column.Type, + ) } + } + } - // Try to get the value from a field first, then from a tag. - var value any - var ok bool - value, ok = m.GetField(col.Name) - if !ok { - value, ok = m.GetTag(col.Name) - } + return group.builder.NewRecordBatch() +} - // if neither field nor tag exists, append a null value - if !ok { - switch col.Type { - case arrow.PrimitiveTypes.Int8: - builder.Field(index).(*array.Int8Builder).AppendNull() - case arrow.PrimitiveTypes.Int16: - builder.Field(index).(*array.Int16Builder).AppendNull() - case arrow.PrimitiveTypes.Int32: - builder.Field(index).(*array.Int32Builder).AppendNull() - case arrow.PrimitiveTypes.Int64: - builder.Field(index).(*array.Int64Builder).AppendNull() - case arrow.PrimitiveTypes.Uint8: - builder.Field(index).(*array.Uint8Builder).AppendNull() - case arrow.PrimitiveTypes.Uint16: - builder.Field(index).(*array.Uint16Builder).AppendNull() - case arrow.PrimitiveTypes.Uint32: - builder.Field(index).(*array.Uint32Builder).AppendNull() - case arrow.PrimitiveTypes.Uint64: - builder.Field(index).(*array.Uint64Builder).AppendNull() - case arrow.PrimitiveTypes.Float32: - builder.Field(index).(*array.Float32Builder).AppendNull() - case arrow.PrimitiveTypes.Float64: - builder.Field(index).(*array.Float64Builder).AppendNull() - case arrow.BinaryTypes.String: - builder.Field(index).(*array.StringBuilder).AppendNull() - case arrow.FixedWidthTypes.Boolean: - builder.Field(index).(*array.BooleanBuilder).AppendNull() - default: - return nil, fmt.Errorf("unsupported type: %T", value) - } +func (p *Parquet) valueFor(m telegraf.Metric, column string) interface{} { + if p.TimestampFieldName != "" && column == p.TimestampFieldName { + return m.Time().UnixNano() + } + if value, found := m.GetField(column); found { + return value + } + if value, found := m.GetTag(column); found { + return value + } - continue - } + return nil +} - switch col.Type { - case arrow.PrimitiveTypes.Int8: - builder.Field(index).(*array.Int8Builder).Append(value.(int8)) - case arrow.PrimitiveTypes.Int16: - builder.Field(index).(*array.Int16Builder).Append(value.(int16)) - case arrow.PrimitiveTypes.Int32: - builder.Field(index).(*array.Int32Builder).Append(value.(int32)) - case arrow.PrimitiveTypes.Int64: - builder.Field(index).(*array.Int64Builder).Append(value.(int64)) - case arrow.PrimitiveTypes.Uint8: - builder.Field(index).(*array.Uint8Builder).Append(value.(uint8)) - case arrow.PrimitiveTypes.Uint16: - builder.Field(index).(*array.Uint16Builder).Append(value.(uint16)) - case arrow.PrimitiveTypes.Uint32: - builder.Field(index).(*array.Uint32Builder).Append(value.(uint32)) - case arrow.PrimitiveTypes.Uint64: - builder.Field(index).(*array.Uint64Builder).Append(value.(uint64)) - case arrow.PrimitiveTypes.Float32: - builder.Field(index).(*array.Float32Builder).Append(value.(float32)) - case arrow.PrimitiveTypes.Float64: - builder.Field(index).(*array.Float64Builder).Append(value.(float64)) - case arrow.BinaryTypes.String: - builder.Field(index).(*array.StringBuilder).Append(value.(string)) - case arrow.FixedWidthTypes.Boolean: - builder.Field(index).(*array.BooleanBuilder).Append(value.(bool)) - default: - return nil, fmt.Errorf("unsupported type: %T", value) - } - } +func (p *Parquet) warnOncef(group *metricGroup, column, format string, args ...interface{}) { + if group.warned[column] { + return + } + group.warned[column] = true + p.Log.Warnf(format, args...) +} + +func appendValue(builder array.Builder, value interface{}) bool { + switch v := value.(type) { + case nil: + builder.AppendNull() + return true + case int8: + return appendTyped(builder, v) + case int16: + return appendTyped(builder, v) + case int32: + return appendTyped(builder, v) + case int64: + return appendTyped(builder, v) + case int: + return appendTyped(builder, int64(v)) + case uint8: + return appendTyped(builder, v) + case uint16: + return appendTyped(builder, v) + case uint32: + return appendTyped(builder, v) + case uint64: + return appendTyped(builder, v) + case uint: + return appendTyped(builder, uint64(v)) + case float32: + return appendTyped(builder, v) + case float64: + return appendTyped(builder, v) + case string: + return appendTyped(builder, v) + case bool: + return appendTyped(builder, v) + default: + builder.AppendNull() + return false + } +} + +func appendTyped[T any](builder array.Builder, value T) bool { + column, ok := builder.(interface{ Append(T) }) + if !ok { + builder.AppendNull() + return false } + column.Append(value) - record := builder.NewRecordBatch() - return record, nil + return true } func (p *Parquet) createSchema(metrics []telegraf.Metric) (*arrow.Schema, error) { diff --git a/plugins/outputs/parquet/parquet_test.go b/plugins/outputs/parquet/parquet_test.go index b063ce0997605..0584a9c024c91 100644 --- a/plugins/outputs/parquet/parquet_test.go +++ b/plugins/outputs/parquet/parquet_test.go @@ -6,6 +6,9 @@ import ( "testing" "time" + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" "github.com/apache/arrow-go/v18/parquet/file" "github.com/stretchr/testify/require" @@ -276,3 +279,58 @@ 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 TestConflictingValueTypesDoNotPanic(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{}{"k": int64(1)}, time.Now()), + })) + require.NoError(t, p.Write([]telegraf.Metric{ + metric.New("demo", map[string]string{"k": "v"}, map[string]interface{}{"other": int64(2)}, 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() + require.Equal(t, int64(2), reader.NumRows()) +} + +func TestEveryConvertibleTypeRoundTrips(t *testing.T) { + values := map[string]interface{}{ + "int8": int8(1), "int16": int16(2), "int32": int32(3), "int64": int64(4), "int": 5, + "uint8": uint8(6), "uint16": uint16(7), "uint32": uint32(8), "uint64": uint64(9), "uint": uint(10), + "float32": float32(11), "float64": float64(12), + "string": "thirteen", "bool": true, + } + + for name, value := range values { + t.Run(name, func(t *testing.T) { + datatype, err := goToArrowType(value) + require.NoError(t, err) + + builder := array.NewBuilder(memory.DefaultAllocator, datatype) + defer builder.Release() + + require.True(t, appendValue(builder, value)) + require.Equal(t, 0, builder.NullN()) + }) + } +} + +func TestAppendValueNullsWhatItCannotWrite(t *testing.T) { + builder := array.NewBuilder(memory.DefaultAllocator, arrow.PrimitiveTypes.Int64) + defer builder.Release() + + require.False(t, appendValue(builder, "not an int")) + require.False(t, appendValue(builder, []string{"unsupported"})) + require.True(t, appendValue(builder, nil)) + require.Equal(t, 3, builder.NullN()) +}