From 220e66939fbff603ecbd102cd4293fa0bb1ac77c Mon Sep 17 00:00:00 2001 From: yuzone Date: Thu, 5 Mar 2026 17:13:43 +0900 Subject: [PATCH] perf: skip parseMap allocation and pool dynamicpb.Message --- map_to_proto.go | 229 +++++++++++++++++++++--- map_to_proto_test.go | 404 ++++++++++++++++++++++++++++++++++++++++++- out_writeapi.go | 137 ++++++--------- out_writeapi_test.go | 336 +++++------------------------------ 4 files changed, 706 insertions(+), 400 deletions(-) diff --git a/map_to_proto.go b/map_to_proto.go index c5997bc..9cb7826 100644 --- a/map_to_proto.go +++ b/map_to_proto.go @@ -67,21 +67,49 @@ var marshalBufPool = sync.Pool{ }, } -// mapToBinary converts a map[string]interface{} directly to proto binary format, -// bypassing the intermediate JSON serialization (json.Marshal + protojson.Unmarshal) -// that the previous jsonToBinary implementation used. -// -// cache may be nil — in that case every field lookup falls back to -// ByName + linear JSON-name scan (still correct, just slower). -func mapToBinary(md protoreflect.MessageDescriptor, data map[string]interface{}, cache fieldLookupCache) ([]byte, error) { - msg, err := mapToMessage(md, data, cache) - if err != nil { - return nil, err +// messagePool caches *dynamicpb.Message instances keyed by MessageDescriptor +// identity to avoid per-row heap allocations in the hot path. +// We key by the descriptor itself (not FullName) because dynamicpb.Message.Set +// requires the field descriptor to belong to the exact same MessageDescriptor +// instance that was used to create the message. +var messagePool sync.Map // protoreflect.MessageDescriptor → *sync.Pool + +// getPooledMessage retrieves a dynamicpb.Message from the pool, +// or creates a new one if the pool is empty. +func getPooledMessage(md protoreflect.MessageDescriptor) *dynamicpb.Message { + p, ok := messagePool.Load(md) + if !ok { + p, _ = messagePool.LoadOrStore(md, &sync.Pool{ + New: func() interface{} { + return dynamicpb.NewMessage(md) + }, + }) } + return p.(*sync.Pool).Get().(*dynamicpb.Message) +} - // Marshal into a pooled scratch buffer, then copy the result out. +// putPooledMessage clears all populated fields and returns the message to the pool. +// Clearing via Range+Clear preserves internal map bucket memory so that +// subsequent reuse avoids re-growing the map. +func putPooledMessage(msg *dynamicpb.Message) { + msg.Range(func(fd protoreflect.FieldDescriptor, _ protoreflect.Value) bool { + msg.Clear(fd) + return true + }) + if p, ok := messagePool.Load(msg.Descriptor()); ok { + p.(*sync.Pool).Put(msg) + } +} + +// marshalAndRelease marshals a pooled dynamicpb.Message to bytes, +// then returns the message to the pool. Handles buffer pooling internally. +func marshalAndRelease(msg *dynamicpb.Message) ([]byte, error) { bufp := marshalBufPool.Get().(*[]byte) b, err := proto.MarshalOptions{}.MarshalAppend((*bufp)[:0], msg) + + // Return message to pool (clears fields, preserves map capacity) + putPooledMessage(msg) + if err != nil { *bufp = b marshalBufPool.Put(bufp) @@ -97,12 +125,50 @@ func mapToBinary(md protoreflect.MessageDescriptor, data map[string]interface{}, return result, nil } -// mapToMessage populates a dynamicpb.Message directly from a map[string]interface{}. -// It iterates the message descriptor's fields, looks up each field's name in the map, -// and sets the corresponding proto value. +// mapToBinary converts a map[string]interface{} directly to proto binary format, +// bypassing intermediate JSON serialization for better performance. +// +// cache may be nil — in that case every field lookup falls back to +// ByName + linear JSON-name scan (still correct, just slower). +func mapToBinary(md protoreflect.MessageDescriptor, data map[string]interface{}, cache fieldLookupCache) ([]byte, error) { + // Get a pooled top-level message to avoid per-row allocation + msg := getPooledMessage(md) + + if err := populateMessage(msg, md, data, cache); err != nil { + putPooledMessage(msg) + return nil, err + } + + return marshalAndRelease(msg) +} + +// rawMapToBinary converts a map[interface{}]interface{} (raw Fluent Bit record) +// directly to proto binary, avoiding intermediate map allocation. +// It also uses a pooled top-level message +func rawMapToBinary(md protoreflect.MessageDescriptor, rawData map[interface{}]interface{}, cache fieldLookupCache) ([]byte, error) { + msg := getPooledMessage(md) + + if err := rawPopulateMessage(msg, md, rawData, cache); err != nil { + putPooledMessage(msg) + return nil, err + } + + return marshalAndRelease(msg) +} + +// mapToMessage creates a new dynamicpb.Message and populates it from a map[string]interface{}. +// Used for nested sub-messages (which are not pooled). func mapToMessage(md protoreflect.MessageDescriptor, data map[string]interface{}, cache fieldLookupCache) (*dynamicpb.Message, error) { msg := dynamicpb.NewMessage(md) + if err := populateMessage(msg, md, data, cache); err != nil { + return nil, err + } + return msg, nil +} +// populateMessage fills an existing dynamicpb.Message from a map[string]interface{}. +// Extracted from mapToMessage so that callers can supply a pooled message +func populateMessage(msg *dynamicpb.Message, md protoreflect.MessageDescriptor, data map[string]interface{}, cache fieldLookupCache) error { // Use pre-built field map when available (O(1) per key). var fieldMap map[string]protoreflect.FieldDescriptor if cache != nil { @@ -127,27 +193,89 @@ func mapToMessage(md protoreflect.MessageDescriptor, data map[string]interface{} if fd.IsList() { if err := setRepeatedField(msg, fd, val, cache); err != nil { - return nil, fmt.Errorf("field %q: %w", key, err) + return fmt.Errorf("field %q: %w", key, err) } } else if fd.Kind() == protoreflect.MessageKind || fd.Kind() == protoreflect.GroupKind { subMap, ok := val.(map[string]interface{}) if !ok { - return nil, fmt.Errorf("field %q: expected map for message field, got %T", key, val) + return fmt.Errorf("field %q: expected map for message field, got %T", key, val) } subMsg, err := mapToMessage(fd.Message(), subMap, cache) if err != nil { - return nil, fmt.Errorf("field %q: %w", key, err) + return fmt.Errorf("field %q: %w", key, err) } msg.Set(fd, protoreflect.ValueOfMessage(subMsg)) } else { pv, err := goToProtoScalar(fd, val) if err != nil { - return nil, fmt.Errorf("field %q: %w", key, err) + return fmt.Errorf("field %q: %w", key, err) } msg.Set(fd, pv) } } - return msg, nil + return nil +} + +// rawPopulateMessage fills a dynamicpb.Message directly from a raw Fluent Bit +// record (map[interface{}]interface{}), handling key and value type conversions +// inline. +func rawPopulateMessage(msg *dynamicpb.Message, md protoreflect.MessageDescriptor, rawData map[interface{}]interface{}, cache fieldLookupCache) error { + var fieldMap map[string]protoreflect.FieldDescriptor + if cache != nil { + fieldMap = cache[md.FullName()] + } + + for k, val := range rawData { + if val == nil { + continue + } + + // Fluent Bit msgpack keys are always string-typed + key, ok := k.(string) + if !ok { + continue + } + + var fd protoreflect.FieldDescriptor + if fieldMap != nil { + fd = fieldMap[key] + } else { + fd = findFieldDescriptor(md.Fields(), key) + } + if fd == nil { + continue + } + + if fd.IsList() { + if err := rawSetRepeatedField(msg, fd, val, cache); err != nil { + return fmt.Errorf("field %q: %w", key, err) + } + } else if fd.Kind() == protoreflect.MessageKind || fd.Kind() == protoreflect.GroupKind { + switch sub := val.(type) { + case map[interface{}]interface{}: + subMsg := dynamicpb.NewMessage(fd.Message()) + if err := rawPopulateMessage(subMsg, fd.Message(), sub, cache); err != nil { + return fmt.Errorf("field %q: %w", key, err) + } + msg.Set(fd, protoreflect.ValueOfMessage(subMsg)) + case map[string]interface{}: + subMsg, err := mapToMessage(fd.Message(), sub, cache) + if err != nil { + return fmt.Errorf("field %q: %w", key, err) + } + msg.Set(fd, protoreflect.ValueOfMessage(subMsg)) + default: + return fmt.Errorf("field %q: expected map for message field, got %T", key, val) + } + } else { + pv, err := goToProtoScalar(fd, val) + if err != nil { + return fmt.Errorf("field %q: %w", key, err) + } + msg.Set(fd, pv) + } + } + return nil } // findFieldDescriptor looks up a field descriptor by name. @@ -207,6 +335,50 @@ func setRepeatedField(msg *dynamicpb.Message, fd protoreflect.FieldDescriptor, v return nil } +// rawSetRepeatedField populates a repeated proto field from a raw slice, +// handling map[interface{}]interface{} elements for nested messages +func rawSetRepeatedField(msg *dynamicpb.Message, fd protoreflect.FieldDescriptor, val interface{}, cache fieldLookupCache) error { + slice, ok := val.([]interface{}) + if !ok { + return fmt.Errorf("expected []interface{} for repeated field, got %T", val) + } + if len(slice) == 0 { + return nil + } + + list := msg.Mutable(fd).List() + for i, item := range slice { + if item == nil { + continue + } + if fd.Kind() == protoreflect.MessageKind || fd.Kind() == protoreflect.GroupKind { + switch sub := item.(type) { + case map[interface{}]interface{}: + subMsg := dynamicpb.NewMessage(fd.Message()) + if err := rawPopulateMessage(subMsg, fd.Message(), sub, cache); err != nil { + return fmt.Errorf("element %d: %w", i, err) + } + list.Append(protoreflect.ValueOfMessage(subMsg)) + case map[string]interface{}: + subMsg, err := mapToMessage(fd.Message(), sub, cache) + if err != nil { + return fmt.Errorf("element %d: %w", i, err) + } + list.Append(protoreflect.ValueOfMessage(subMsg)) + default: + return fmt.Errorf("element %d: expected map for repeated message, got %T", i, item) + } + } else { + pv, err := goToProtoScalar(fd, item) + if err != nil { + return fmt.Errorf("element %d: %w", i, err) + } + list.Append(pv) + } + } + return nil +} + // goToProtoScalar converts a Go interface{} value to the appropriate // protoreflect.Value for the given field descriptor's kind. // Handles all proto scalar types that appear in BigQuery Storage Write API schemas. @@ -285,6 +457,8 @@ func toProtoInt64(val interface{}) (protoreflect.Value, error) { return protoreflect.ValueOfInt64(int64(f)), nil } return protoreflect.ValueOfInt64(i), nil + case []byte: + return toProtoInt64(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to int64", val) } @@ -312,6 +486,8 @@ func toProtoInt32(val interface{}) (protoreflect.Value, error) { return protoreflect.Value{}, fmt.Errorf("cannot convert string %q to int32: %w", v, err) } return protoreflect.ValueOfInt32(int32(i)), nil + case []byte: + return toProtoInt32(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to int32", val) } @@ -335,6 +511,8 @@ func toProtoUint64(val interface{}) (protoreflect.Value, error) { return protoreflect.Value{}, fmt.Errorf("cannot convert string %q to uint64: %w", v, err) } return protoreflect.ValueOfUint64(u), nil + case []byte: + return toProtoUint64(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to uint64", val) } @@ -358,6 +536,8 @@ func toProtoUint32(val interface{}) (protoreflect.Value, error) { return protoreflect.Value{}, fmt.Errorf("cannot convert string %q to uint32: %w", v, err) } return protoreflect.ValueOfUint32(uint32(u)), nil + case []byte: + return toProtoUint32(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to uint32", val) } @@ -381,6 +561,8 @@ func toProtoDouble(val interface{}) (protoreflect.Value, error) { return protoreflect.Value{}, fmt.Errorf("cannot convert string %q to float64: %w", v, err) } return protoreflect.ValueOfFloat64(f), nil + case []byte: + return toProtoDouble(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to float64", val) } @@ -402,6 +584,8 @@ func toProtoFloat(val interface{}) (protoreflect.Value, error) { return protoreflect.Value{}, fmt.Errorf("cannot convert string %q to float32: %w", v, err) } return protoreflect.ValueOfFloat32(float32(f)), nil + case []byte: + return toProtoFloat(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to float32", val) } @@ -425,6 +609,8 @@ func toProtoBool(val interface{}) (protoreflect.Value, error) { return protoreflect.ValueOfBool(v != 0), nil case float64: return protoreflect.ValueOfBool(v != 0), nil + case []byte: + return toProtoBool(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to bool", val) } @@ -436,7 +622,8 @@ func toProtoBytes(val interface{}) (protoreflect.Value, error) { return protoreflect.ValueOfBytes(v), nil case string: // protojson expects base64-encoded strings for bytes fields. - // Since parseMap converts []byte → string, attempt base64 decode first. + // Since Fluent Bit's msgpack decoder may produce []byte for string values, + // attempt base64 decode first. b, err := base64.StdEncoding.DecodeString(v) if err != nil { // Try URL-safe base64 @@ -468,6 +655,8 @@ func toProtoEnum(val interface{}) (protoreflect.Value, error) { return protoreflect.Value{}, fmt.Errorf("cannot convert string %q to enum: %w", v, err) } return protoreflect.ValueOfEnum(protoreflect.EnumNumber(i)), nil + case []byte: + return toProtoEnum(string(v)) default: return protoreflect.Value{}, fmt.Errorf("cannot convert %T to enum", val) } diff --git a/map_to_proto_test.go b/map_to_proto_test.go index db01a0c..583544d 100644 --- a/map_to_proto_test.go +++ b/map_to_proto_test.go @@ -59,9 +59,9 @@ func TestMapToBinary_StringFields(t *testing.T) { md := buildMD(t, schema) tests := []struct { - name string - data map[string]interface{} - checkFn func(t *testing.T, msg *dynamicpb.Message) + name string + data map[string]interface{} + checkFn func(t *testing.T, msg *dynamicpb.Message) }{ { name: "string value", @@ -649,4 +649,400 @@ func BenchmarkMapToBinary(b *testing.B) { b.Fatal(err) } } -} \ No newline at end of file +} + +// helper: rawMapToBinary then unmarshal back to dynamicpb.Message for assertions. +func rawRoundTrip(t *testing.T, md protoreflect.MessageDescriptor, data map[interface{}]interface{}) *dynamicpb.Message { + t.Helper() + cache := buildFieldLookupCache(md) + b, err := rawMapToBinary(md, data, cache) + require.NoError(t, err) + msg := dynamicpb.NewMessage(md) + require.NoError(t, proto.Unmarshal(b, msg)) + return msg +} + +// TestRawMapToBinary_StringFieldsWithBytes tests STRING fields where values +// arrive as []byte (typical msgpack behavior from Fluent Bit). +func TestRawMapToBinary_StringFieldsWithBytes(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "Name", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "Tag", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + + tests := []struct { + name string + data map[interface{}]interface{} + checkFn func(t *testing.T, msg *dynamicpb.Message) + }{ + { + name: "[]byte values converted to string", + data: map[interface{}]interface{}{"Name": []byte("hello"), "Tag": []byte("world")}, + checkFn: func(t *testing.T, msg *dynamicpb.Message) { + assert.Equal(t, "hello", msg.Get(md.Fields().ByName("Name")).String()) + assert.Equal(t, "world", msg.Get(md.Fields().ByName("Tag")).String()) + }, + }, + { + name: "string values passed through", + data: map[interface{}]interface{}{"Name": "hello"}, + checkFn: func(t *testing.T, msg *dynamicpb.Message) { + assert.Equal(t, "hello", msg.Get(md.Fields().ByName("Name")).String()) + }, + }, + { + name: "nil value skipped", + data: map[interface{}]interface{}{"Name": []byte("hello"), "Tag": nil}, + checkFn: func(t *testing.T, msg *dynamicpb.Message) { + assert.Equal(t, "hello", msg.Get(md.Fields().ByName("Name")).String()) + assert.Equal(t, "", msg.Get(md.Fields().ByName("Tag")).String()) + }, + }, + { + name: "unknown field ignored", + data: map[interface{}]interface{}{"Name": []byte("hello"), "Unknown": []byte("ignored")}, + checkFn: func(t *testing.T, msg *dynamicpb.Message) { + assert.Equal(t, "hello", msg.Get(md.Fields().ByName("Name")).String()) + }, + }, + { + name: "int coerced to string", + data: map[interface{}]interface{}{"Name": 42}, + checkFn: func(t *testing.T, msg *dynamicpb.Message) { + assert.Equal(t, "42", msg.Get(md.Fields().ByName("Name")).String()) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + msg := rawRoundTrip(t, md, tt.data) + tt.checkFn(t, msg) + }) + } +} + +// TestRawMapToBinary_Int64WithBytes tests that []byte values (msgpack strings) +// are correctly parsed into int64 proto fields. +func TestRawMapToBinary_Int64WithBytes(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "count", Type: storagepb.TableFieldSchema_INT64, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + + tests := []struct { + name string + data map[interface{}]interface{} + expected int64 + }{ + {"int64", map[interface{}]interface{}{"count": int64(100)}, 100}, + {"uint64", map[interface{}]interface{}{"count": uint64(200)}, 200}, + {"float64", map[interface{}]interface{}{"count": float64(300)}, 300}, + {"[]byte", map[interface{}]interface{}{"count": []byte("500")}, 500}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + msg := rawRoundTrip(t, md, tt.data) + assert.Equal(t, tt.expected, msg.Get(md.Fields().ByName("count")).Int()) + }) + } +} + +// TestRawMapToBinary_NestedRawMap tests STRUCT fields where the nested value +// is a map[interface{}]interface{} (raw msgpack). +func TestRawMapToBinary_NestedRawMap(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "name", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + { + Name: "address", + Type: storagepb.TableFieldSchema_STRUCT, + Mode: storagepb.TableFieldSchema_NULLABLE, + Fields: []*storagepb.TableFieldSchema{ + {Name: "city", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "zip", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + }, + }, + } + md := buildMD(t, schema) + + data := map[interface{}]interface{}{ + "name": []byte("Alice"), + "address": map[interface{}]interface{}{ + "city": []byte("Tokyo"), + "zip": []byte("100-0001"), + }, + } + + msg := rawRoundTrip(t, md, data) + assert.Equal(t, "Alice", msg.Get(md.Fields().ByName("name")).String()) + + addrFD := md.Fields().ByName("address") + require.NotNil(t, addrFD) + addrMsg := msg.Get(addrFD).Message() + addrMD := addrFD.Message() + assert.Equal(t, "Tokyo", addrMsg.Get(addrMD.Fields().ByName("city")).String()) + assert.Equal(t, "100-0001", addrMsg.Get(addrMD.Fields().ByName("zip")).String()) +} + +// TestRawMapToBinary_RepeatedScalar tests REPEATED fields from raw maps. +func TestRawMapToBinary_RepeatedScalar(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "tags", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_REPEATED}, + }, + } + md := buildMD(t, schema) + + data := map[interface{}]interface{}{ + "tags": []interface{}{[]byte("alpha"), []byte("beta"), "gamma"}, + } + + msg := rawRoundTrip(t, md, data) + tagsFD := md.Fields().ByName("tags") + list := msg.Get(tagsFD).List() + require.Equal(t, 3, list.Len()) + assert.Equal(t, "alpha", list.Get(0).String()) + assert.Equal(t, "beta", list.Get(1).String()) + assert.Equal(t, "gamma", list.Get(2).String()) +} + +// TestRawMapToBinary_RepeatedMessage tests REPEATED STRUCT fields from raw maps. +func TestRawMapToBinary_RepeatedMessage(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + { + Name: "items", + Type: storagepb.TableFieldSchema_STRUCT, + Mode: storagepb.TableFieldSchema_REPEATED, + Fields: []*storagepb.TableFieldSchema{ + {Name: "key", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "value", Type: storagepb.TableFieldSchema_INT64, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + }, + }, + } + md := buildMD(t, schema) + + data := map[interface{}]interface{}{ + "items": []interface{}{ + map[interface{}]interface{}{"key": []byte("a"), "value": int64(1)}, + map[interface{}]interface{}{"key": []byte("b"), "value": int64(2)}, + }, + } + + msg := rawRoundTrip(t, md, data) + itemsFD := md.Fields().ByName("items") + list := msg.Get(itemsFD).List() + require.Equal(t, 2, list.Len()) + + itemMD := itemsFD.Message() + item0 := list.Get(0).Message() + assert.Equal(t, "a", item0.Get(itemMD.Fields().ByName("key")).String()) + assert.Equal(t, int64(1), item0.Get(itemMD.Fields().ByName("value")).Int()) + + item1 := list.Get(1).Message() + assert.Equal(t, "b", item1.Get(itemMD.Fields().ByName("key")).String()) + assert.Equal(t, int64(2), item1.Get(itemMD.Fields().ByName("value")).Int()) +} + +// TestRawMapToBinary_MatchesMapToBinary verifies that rawMapToBinary produces +// identical output to mapToBinary for the same logical data. +func TestRawMapToBinary_MatchesMapToBinary(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "Text", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "Time", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + cache := buildFieldLookupCache(md) + + // Same data in both formats ([]byte values in raw, string values in parsed) + rawData := map[interface{}]interface{}{ + "Text": []byte("FOO"), + "Time": []byte("000"), + } + parsedData := map[string]interface{}{ + "Text": "FOO", + "Time": "000", + } + + rawBytes, err := rawMapToBinary(md, rawData, cache) + require.NoError(t, err) + parsedBytes, err := mapToBinary(md, parsedData, cache) + require.NoError(t, err) + + assert.Equal(t, parsedBytes, rawBytes) +} + +// TestRawMapToBinary_BoolWithBytes tests []byte to bool conversion. +func TestRawMapToBinary_BoolWithBytes(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "active", Type: storagepb.TableFieldSchema_BOOL, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + + tests := []struct { + name string + data map[interface{}]interface{} + expected bool + }{ + {"[]byte_true", map[interface{}]interface{}{"active": []byte("true")}, true}, + {"[]byte_false", map[interface{}]interface{}{"active": []byte("false")}, false}, + {"[]byte_1", map[interface{}]interface{}{"active": []byte("1")}, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + msg := rawRoundTrip(t, md, tt.data) + assert.Equal(t, tt.expected, msg.Get(md.Fields().ByName("active")).Bool()) + }) + } +} + +// TestRawMapToBinary_DoubleWithBytes tests []byte to float64 conversion. +func TestRawMapToBinary_DoubleWithBytes(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "score", Type: storagepb.TableFieldSchema_DOUBLE, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + + data := map[interface{}]interface{}{"score": []byte("3.14")} + msg := rawRoundTrip(t, md, data) + assert.InDelta(t, 3.14, msg.Get(md.Fields().ByName("score")).Float(), 0.001) +} + +// TestMessagePoolReuse verifies that pooled messages produce correct output +// across multiple sequential calls (no stale field leaks after reset). +func TestMessagePoolReuse(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "name", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "count", Type: storagepb.TableFieldSchema_INT64, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + cache := buildFieldLookupCache(md) + + // Call mapToBinary multiple times; the pool should reuse messages + for i := 0; i < 100; i++ { + data := map[string]interface{}{ + "name": "user", + "count": int64(i), + } + b, err := mapToBinary(md, data, cache) + require.NoError(t, err) + + msg := dynamicpb.NewMessage(md) + require.NoError(t, proto.Unmarshal(b, msg)) + assert.Equal(t, "user", msg.Get(md.Fields().ByName("name")).String()) + assert.Equal(t, int64(i), msg.Get(md.Fields().ByName("count")).Int()) + } +} + +// TestMessagePoolReuse_NoStaleFields verifies that fields set in a previous +// row don't leak into the next row after pool reuse. +func TestMessagePoolReuse_NoStaleFields(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "a", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "b", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + cache := buildFieldLookupCache(md) + + // First call sets both fields + b1, err := mapToBinary(md, map[string]interface{}{"a": "hello", "b": "world"}, cache) + require.NoError(t, err) + msg1 := dynamicpb.NewMessage(md) + require.NoError(t, proto.Unmarshal(b1, msg1)) + assert.Equal(t, "hello", msg1.Get(md.Fields().ByName("a")).String()) + assert.Equal(t, "world", msg1.Get(md.Fields().ByName("b")).String()) + + // Second call sets only "a" — field "b" must NOT carry over from the first call + b2, err := mapToBinary(md, map[string]interface{}{"a": "only-a"}, cache) + require.NoError(t, err) + msg2 := dynamicpb.NewMessage(md) + require.NoError(t, proto.Unmarshal(b2, msg2)) + assert.Equal(t, "only-a", msg2.Get(md.Fields().ByName("a")).String()) + assert.Equal(t, "", msg2.Get(md.Fields().ByName("b")).String()) // must be default, not "world" +} + +// TestMessagePoolReuse_RawPath verifies pool reuse via rawMapToBinary. +func TestMessagePoolReuse_RawPath(t *testing.T) { + schema := &storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "x", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "y", Type: storagepb.TableFieldSchema_INT64, Mode: storagepb.TableFieldSchema_NULLABLE}, + }, + } + md := buildMD(t, schema) + cache := buildFieldLookupCache(md) + + // First call sets both fields + b1, err := rawMapToBinary(md, map[interface{}]interface{}{"x": []byte("first"), "y": int64(1)}, cache) + require.NoError(t, err) + msg1 := dynamicpb.NewMessage(md) + require.NoError(t, proto.Unmarshal(b1, msg1)) + assert.Equal(t, "first", msg1.Get(md.Fields().ByName("x")).String()) + assert.Equal(t, int64(1), msg1.Get(md.Fields().ByName("y")).Int()) + + // Second call sets only "x" — field "y" must not carry over + b2, err := rawMapToBinary(md, map[interface{}]interface{}{"x": []byte("second")}, cache) + require.NoError(t, err) + msg2 := dynamicpb.NewMessage(md) + require.NoError(t, proto.Unmarshal(b2, msg2)) + assert.Equal(t, "second", msg2.Get(md.Fields().ByName("x")).String()) + assert.Equal(t, int64(0), msg2.Get(md.Fields().ByName("y")).Int()) // must be default +} + +// BenchmarkRawMapToBinary benchmarks the raw map[interface{}]interface{} path. +func BenchmarkRawMapToBinary(b *testing.B) { + schema := mangleInputSchema(&storagepb.TableSchema{ + Fields: []*storagepb.TableFieldSchema{ + {Name: "name", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "age", Type: storagepb.TableFieldSchema_INT64, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "score", Type: storagepb.TableFieldSchema_DOUBLE, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "active", Type: storagepb.TableFieldSchema_BOOL, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "ts", Type: storagepb.TableFieldSchema_TIMESTAMP, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "numeric_val", Type: storagepb.TableFieldSchema_NUMERIC, Mode: storagepb.TableFieldSchema_NULLABLE}, + {Name: "tags", Type: storagepb.TableFieldSchema_STRING, Mode: storagepb.TableFieldSchema_REPEATED}, + }, + }, true) + descriptor, _ := adapt.StorageSchemaToProto2Descriptor(schema, "root") + md := descriptor.(protoreflect.MessageDescriptor) + cache := buildFieldLookupCache(md) + + // Raw data as it would come from Fluent Bit's msgpack decoder + data := map[interface{}]interface{}{ + "name": []byte("test-user"), + "age": int64(30), + "score": 95.5, + "active": true, + "ts": int64(1700000000000000), + "numeric_val": []byte("123.456789"), + "tags": []interface{}{[]byte("tag1"), []byte("tag2"), []byte("tag3")}, + } + + b.ResetTimer() + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _, err := rawMapToBinary(md, data, cache) + if err != nil { + b.Fatal(err) + } + } +} diff --git a/out_writeapi.go b/out_writeapi.go index cd3a8fe..9c8474b 100644 --- a/out_writeapi.go +++ b/out_writeapi.go @@ -184,91 +184,57 @@ func getDescriptors(curr_ctx context.Context, mw_client ManagedWriterClient, pro return messageDescriptor, dp, timestampFields, nil } -// This function handles the data transformation from a Go map to binary for a single row. -// It populates the proto message directly from the map, bypassing the intermediate -// JSON serialization (json.Marshal + protojson.Unmarshal) for better performance. -// The outputs of this function are the corresponding binary data as well as any error that occur. -func jsonToBinary(message_descriptor protoreflect.MessageDescriptor, jsonRow map[string]interface{}, cache fieldLookupCache) ([]byte, error) { - return mapToBinary(message_descriptor, jsonRow, cache) -} - -// From https://github.com/majst01/fluent-bit-go-redis-output.git -// Function is used to transform fluent-bit record to a JSON map -func parseMap(mapInterface map[interface{}]interface{}) map[string]interface{} { - if mapInterface == nil { - return nil - } - m := make(map[string]interface{}) - for k, v := range mapInterface { - switch t := v.(type) { - case []byte: - // Prevent encoding to base64 - m[k.(string)] = string(t) - case map[interface{}]interface{}: - m[k.(string)] = parseMap(t) - case []interface{}: - m[k.(string)] = parseSlice(t) - default: - m[k.(string)] = v +// convertTimestampFieldsRaw converts timestamp fields in a raw Fluent Bit record +// (map[interface{}]interface{}) from seconds to microseconds for BigQuery Storage Write API. +// BigQuery TIMESTAMP type expects microseconds since Unix epoch. +func convertTimestampFieldsRaw(data map[interface{}]interface{}, timestampFields []string) { + for _, field := range timestampFields { + // In Go, map[interface{}]interface{} lookup with a string key works + // because interface comparison uses underlying type+value equality. + if val, ok := data[field]; ok { + convertTimestampValueRaw(data, field, val) } } - return m } -// Function to handle slices that may contain nested maps -func parseSlice(sliceInterface []interface{}) []interface{} { - if sliceInterface == nil { - return nil - } - result := make([]interface{}, len(sliceInterface)) - for i, v := range sliceInterface { - switch t := v.(type) { - case []byte: - result[i] = string(t) - case map[interface{}]interface{}: - result[i] = parseMap(t) - case []interface{}: - result[i] = parseSlice(t) - default: - result[i] = v +// convertTimestampValueRaw handles the value conversion for map[interface{}]interface{} +func convertTimestampValueRaw(data map[interface{}]interface{}, field string, val interface{}) { + switch v := val.(type) { + case int64: + if v > 0 && v < maxUnixSeconds { + data[field] = v * 1000000 } - } - return result -} - -// Convert timestamp fields from seconds to microseconds for BigQuery Storage Write API -// BigQuery TIMESTAMP type expects microseconds since Unix epoch -func convertTimestampFields(data map[string]interface{}, timestampFields []string) { - for _, field := range timestampFields { - if val, ok := data[field]; ok { - switch v := val.(type) { - case int64: - // If value looks like seconds (less than year 2100 in seconds), convert to microseconds - if v > 0 && v < maxUnixSeconds { - data[field] = v * 1000000 - } - case int: - if v > 0 && v < maxUnixSeconds { - data[field] = int64(v) * 1000000 - } - case uint64: - if v > 0 && v < maxUnixSeconds { - data[field] = int64(v) * 1000000 - } - case float64: - if v > 0 && v < maxUnixSeconds { - data[field] = int64(v * 1000000) + case int: + if v > 0 && v < maxUnixSeconds { + data[field] = int64(v) * 1000000 + } + case uint64: + if v > 0 && v < maxUnixSeconds { + data[field] = int64(v) * 1000000 + } + case float64: + if v > 0 && v < maxUnixSeconds { + data[field] = int64(v * 1000000) + } + case string: + if v != "" { + if floatVal, err := strconv.ParseFloat(v, 64); err == nil { + if floatVal > 0 && floatVal < maxUnixSeconds { + data[field] = int64(floatVal * 1000000) + } else { + data[field] = int64(floatVal) } - case string: - // Handle string timestamp (e.g., "1769419316") - if v != "" { - if floatVal, err := strconv.ParseFloat(v, 64); err == nil { - if floatVal > 0 && floatVal < maxUnixSeconds { - data[field] = int64(floatVal * 1000000) - } else { - data[field] = int64(floatVal) - } - } + } + } + case []byte: + // Handle msgpack string values that arrive as []byte + s := string(v) + if s != "" { + if floatVal, err := strconv.ParseFloat(s, 64); err == nil { + if floatVal > 0 && floatVal < maxUnixSeconds { + data[field] = int64(floatVal * 1000000) + } else { + data[field] = int64(floatVal) } } } @@ -806,16 +772,13 @@ func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int break } - rowJSONMap := parseMap(record) - - // Convert timestamp fields from seconds to microseconds for BigQuery Storage Write API - convertTimestampFields(rowJSONMap, config.timestampFields) + // Convert timestamp fields in-place on the raw record + convertTimestampFieldsRaw(record, config.timestampFields) - // Serialize data - // Transform each row of data into binary using the jsonToBinary function and the message descriptor from the getDescriptors function - buf, err := jsonToBinary(config.messageDescriptor, rowJSONMap, config.fieldCache) + // Serialize data directly from raw Fluent Bit record + buf, err := rawMapToBinary(config.messageDescriptor, record, config.fieldCache) if err != nil { - log.Printf("Transforming row with value:%s from JSON to binary data for output instance with id: %d failed in FLBPluginFlushCtx: %s", rowJSONMap, id, err) + log.Printf("Transforming row with value:%v from map to binary data for output instance with id: %d failed in FLBPluginFlushCtx: %s", record, id, err) } else { // Successful data transformation if (currsize + len(buf)) >= config.maxChunkSize { diff --git a/out_writeapi_test.go b/out_writeapi_test.go index c1942d7..f71ca5e 100644 --- a/out_writeapi_test.go +++ b/out_writeapi_test.go @@ -1046,352 +1046,110 @@ func TestFLBPluginFlushCtxErrorHandling(t *testing.T) { assert.Equal(t, expectGotRecord, checks.gotRecord) } -// TestParseMap tests the parseMap function -func TestParseMap(t *testing.T) { - tests := []struct { - name string - input map[interface{}]interface{} - expected map[string]interface{} - }{ - { - name: "nil map returns nil", - input: nil, - expected: nil, - }, - { - name: "empty map", - input: map[interface{}]interface{}{}, - expected: map[string]interface{}{}, - }, - { - name: "simple string values", - input: map[interface{}]interface{}{ - "key1": "value1", - "key2": "value2", - }, - expected: map[string]interface{}{ - "key1": "value1", - "key2": "value2", - }, - }, - { - name: "byte slice values converted to string", - input: map[interface{}]interface{}{ - "text": []byte("hello"), - }, - expected: map[string]interface{}{ - "text": "hello", - }, - }, - { - name: "nested map", - input: map[interface{}]interface{}{ - "outer": map[interface{}]interface{}{ - "inner": "value", - }, - }, - expected: map[string]interface{}{ - "outer": map[string]interface{}{ - "inner": "value", - }, - }, - }, - { - name: "slice values", - input: map[interface{}]interface{}{ - "list": []interface{}{"a", "b", "c"}, - }, - expected: map[string]interface{}{ - "list": []interface{}{"a", "b", "c"}, - }, - }, - { - name: "slice with nested map", - input: map[interface{}]interface{}{ - "items": []interface{}{ - map[interface{}]interface{}{ - "name": "item1", - }, - map[interface{}]interface{}{ - "name": "item2", - }, - }, - }, - expected: map[string]interface{}{ - "items": []interface{}{ - map[string]interface{}{ - "name": "item1", - }, - map[string]interface{}{ - "name": "item2", - }, - }, - }, - }, - { - name: "slice with byte slice", - input: map[interface{}]interface{}{ - "data": []interface{}{ - []byte("first"), - []byte("second"), - }, - }, - expected: map[string]interface{}{ - "data": []interface{}{ - "first", - "second", - }, - }, - }, - { - name: "integer and float values", - input: map[interface{}]interface{}{ - "int": 42, - "float": 3.14, - }, - expected: map[string]interface{}{ - "int": 42, - "float": 3.14, - }, - }, - { - name: "boolean values", - input: map[interface{}]interface{}{ - "true": true, - "false": false, - }, - expected: map[string]interface{}{ - "true": true, - "false": false, - }, - }, - { - name: "deeply nested structure", - input: map[interface{}]interface{}{ - "level1": map[interface{}]interface{}{ - "level2": map[interface{}]interface{}{ - "level3": []interface{}{ - map[interface{}]interface{}{ - "data": []byte("deep"), - }, - }, - }, - }, - }, - expected: map[string]interface{}{ - "level1": map[string]interface{}{ - "level2": map[string]interface{}{ - "level3": []interface{}{ - map[string]interface{}{ - "data": "deep", - }, - }, - }, - }, - }, - }, - { - name: "mixed types", - input: map[interface{}]interface{}{ - "string": "text", - "bytes": []byte("binary"), - "number": 123, - "boolean": true, - "nested": map[interface{}]interface{}{ - "key": "value", - }, - "list": []interface{}{1, 2, 3}, - }, - expected: map[string]interface{}{ - "string": "text", - "bytes": "binary", - "number": 123, - "boolean": true, - "nested": map[string]interface{}{ - "key": "value", - }, - "list": []interface{}{1, 2, 3}, - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := parseMap(tt.input) - assert.Equal(t, tt.expected, result) - }) - } -} - -// TestParseSlice tests the parseSlice function -func TestParseSlice(t *testing.T) { - tests := []struct { - name string - input []interface{} - expected []interface{} - }{ - { - name: "nil slice returns nil", - input: nil, - expected: nil, - }, - { - name: "empty slice", - input: []interface{}{}, - expected: []interface{}{}, - }, - { - name: "simple values", - input: []interface{}{"a", "b", "c"}, - expected: []interface{}{"a", "b", "c"}, - }, - { - name: "byte slice converted to string", - input: []interface{}{[]byte("hello"), []byte("world")}, - expected: []interface{}{"hello", "world"}, - }, - { - name: "nested map in slice", - input: []interface{}{ - map[interface{}]interface{}{ - "key": "value", - }, - }, - expected: []interface{}{ - map[string]interface{}{ - "key": "value", - }, - }, - }, - { - name: "nested slice", - input: []interface{}{ - []interface{}{1, 2, 3}, - []interface{}{"a", "b"}, - }, - expected: []interface{}{ - []interface{}{1, 2, 3}, - []interface{}{"a", "b"}, - }, - }, - { - name: "mixed types", - input: []interface{}{"string", []byte("bytes"), 42, true}, - expected: []interface{}{"string", "bytes", 42, true}, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := parseSlice(tt.input) - assert.Equal(t, tt.expected, result) - }) - } -} - -// TestConvertTimestampFields tests the convertTimestampFields function -func TestConvertTimestampFields(t *testing.T) { +// TestConvertTimestampFieldsRaw tests the convertTimestampFieldsRaw function +// for raw Fluent Bit records (map[interface{}]interface{}). +func TestConvertTimestampFieldsRaw(t *testing.T) { tests := []struct { name string - input map[string]interface{} + input map[interface{}]interface{} timestampFields []string - expected map[string]interface{} + expected map[interface{}]interface{} }{ { name: "int64 seconds converted to microseconds", - input: map[string]interface{}{"time": int64(1700000000)}, + input: map[interface{}]interface{}{"time": int64(1700000000)}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(1700000000000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000)}, }, { name: "int seconds converted to microseconds", - input: map[string]interface{}{"time": 1700000000}, + input: map[interface{}]interface{}{"time": 1700000000}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(1700000000000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000)}, }, { name: "uint64 seconds converted to microseconds", - input: map[string]interface{}{"time": uint64(1700000000)}, + input: map[interface{}]interface{}{"time": uint64(1700000000)}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(1700000000000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000)}, }, { name: "float64 seconds converted to microseconds", - input: map[string]interface{}{"time": float64(1700000000.5)}, + input: map[interface{}]interface{}{"time": float64(1700000000.5)}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(1700000000500000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000500000)}, }, { name: "value already in microseconds not converted", - input: map[string]interface{}{"time": int64(1700000000000000)}, - timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(1700000000000000)}, - }, - { - name: "boundary value at maxUnixSeconds not converted", - input: map[string]interface{}{"time": int64(4102444800)}, + input: map[interface{}]interface{}{"time": int64(1700000000000000)}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(4102444800)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000)}, }, { - name: "boundary value just below maxUnixSeconds converted", - input: map[string]interface{}{"time": int64(4102444799)}, + name: "[]byte string timestamp converted", + input: map[interface{}]interface{}{"time": []byte("1700000000")}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(4102444799000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000)}, }, { - name: "zero value not converted", - input: map[string]interface{}{"time": int64(0)}, + name: "[]byte non-numeric string not converted", + input: map[interface{}]interface{}{"time": []byte("not-a-number")}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(0)}, + expected: map[interface{}]interface{}{"time": []byte("not-a-number")}, }, { - name: "negative value not converted", - input: map[string]interface{}{"time": int64(-1000)}, + name: "string timestamp converted", + input: map[interface{}]interface{}{"time": "1700000000"}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(-1000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000)}, }, { name: "field not in timestampFields not converted", - input: map[string]interface{}{"time": int64(1700000000), "other": int64(1700000000)}, + input: map[interface{}]interface{}{"time": int64(1700000000), "other": int64(1700000000)}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": int64(1700000000000000), "other": int64(1700000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000), "other": int64(1700000000)}, }, { name: "multiple timestamp fields", - input: map[string]interface{}{"time": int64(1700000000), "created_at": int64(1600000000)}, + input: map[interface{}]interface{}{"time": int64(1700000000), "created_at": int64(1600000000)}, timestampFields: []string{"time", "created_at"}, - expected: map[string]interface{}{"time": int64(1700000000000000), "created_at": int64(1600000000000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000000000), "created_at": int64(1600000000000000)}, }, { name: "field not present in data", - input: map[string]interface{}{"other": "value"}, + input: map[interface{}]interface{}{"other": "value"}, timestampFields: []string{"time"}, - expected: map[string]interface{}{"other": "value"}, - }, - { - name: "string value not converted", - input: map[string]interface{}{"time": "2023-11-14T00:00:00Z"}, - timestampFields: []string{"time"}, - expected: map[string]interface{}{"time": "2023-11-14T00:00:00Z"}, + expected: map[interface{}]interface{}{"other": "value"}, }, { name: "empty timestamp fields", - input: map[string]interface{}{"time": int64(1700000000)}, + input: map[interface{}]interface{}{"time": int64(1700000000)}, timestampFields: []string{}, - expected: map[string]interface{}{"time": int64(1700000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000)}, }, { name: "nil timestamp fields", - input: map[string]interface{}{"time": int64(1700000000)}, + input: map[interface{}]interface{}{"time": int64(1700000000)}, timestampFields: nil, - expected: map[string]interface{}{"time": int64(1700000000)}, + expected: map[interface{}]interface{}{"time": int64(1700000000)}, + }, + { + name: "zero value not converted", + input: map[interface{}]interface{}{"time": int64(0)}, + timestampFields: []string{"time"}, + expected: map[interface{}]interface{}{"time": int64(0)}, + }, + { + name: "negative value not converted", + input: map[interface{}]interface{}{"time": int64(-1000)}, + timestampFields: []string{"time"}, + expected: map[interface{}]interface{}{"time": int64(-1000)}, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - convertTimestampFields(tt.input, tt.timestampFields) + convertTimestampFieldsRaw(tt.input, tt.timestampFields) assert.Equal(t, tt.expected, tt.input) }) }