Skip to content

perf/refactor: throughput improvements, memory optimizations, and code quality enhancements #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yuzone merged 12 commits into from
Mar 27, 2026
Merged

perf/refactor: throughput improvements, memory optimizations, and code quality enhancements #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
57c7aa8
fix: support nested TIMESTAMP field conversion in RECORD types
yuzone Mar 27, 2026
986b412
fix: re-evaluate least loaded stream per chunk in FLBPluginFlushCtx
yuzone Mar 27, 2026
cfac0ae
fix test code
yuzone Mar 27, 2026
73f6efd
fix: replace gomonkey patch on output.NewDecoder with injectable func…
yuzone Mar 27, 2026
68f07ac
perf: pool nested sub-messages (RECORD types) in dynamicpb message tree
yuzone Mar 27, 2026
05b26f6
perf: reuse binaryData slice across flushes to eliminate per-flush al…
yuzone Mar 27, 2026
4482145
fix: drain pending AppendRows responses before closing streams on exit
yuzone Mar 27, 2026
5bcb910
Flush_Timeout_Sec default 30
yuzone Mar 27, 2026
4717315
docs: add comments to magic number constants and fix typo
yuzone Mar 27, 2026
e14ef73
perf: atomic configID, GC-safe response queue drain, and zero-alloc […
yuzone Mar 27, 2026
1c55fa6
refactor: extract toSubMessage helper to eliminate duplicate map→mess…
yuzone Mar 27, 2026
211f4e9
refactor: simplify **outputConfig to *outputConfig and split FLBPlugi…
yuzone Mar 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
179 changes: 96 additions & 83 deletions map_to_proto.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"fmt"
"strconv"
"sync"
"unsafe"

"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/reflect/protoreflect"
Expand Down Expand Up @@ -91,8 +92,24 @@ func getPooledMessage(md protoreflect.MessageDescriptor) *dynamicpb.Message {
// 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.
// Sub-messages (RECORD fields) are recursively returned to the pool before
// their parent field is cleared, enabling full reuse across the message tree.
func putPooledMessage(msg *dynamicpb.Message) {
msg.Range(func(fd protoreflect.FieldDescriptor, _ protoreflect.Value) bool {
msg.Range(func(fd protoreflect.FieldDescriptor, v protoreflect.Value) bool {
if fd.Kind() == protoreflect.MessageKind || fd.Kind() == protoreflect.GroupKind {
if fd.IsList() {
list := v.List()
for i := 0; i < list.Len(); i++ {
if subMsg, ok := list.Get(i).Message().(*dynamicpb.Message); ok {
putPooledMessage(subMsg)
}
}
} else {
if subMsg, ok := v.Message().(*dynamicpb.Message); ok {
putPooledMessage(subMsg)
}
}
}
msg.Clear(fd)
return true
})
Expand Down Expand Up @@ -156,16 +173,38 @@ func rawMapToBinary(md protoreflect.MessageDescriptor, rawData map[interface{}]i
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).
// mapToMessage retrieves a pooled dynamicpb.Message and populates it from a map[string]interface{}.
// Used for nested sub-messages. The caller must not release the returned message directly;
// it will be recursively released when the top-level message is passed to marshalAndRelease.
func mapToMessage(md protoreflect.MessageDescriptor, data map[string]interface{}, cache fieldLookupCache) (*dynamicpb.Message, error) {
msg := dynamicpb.NewMessage(md)
msg := getPooledMessage(md)
if err := populateMessage(msg, md, data, cache); err != nil {
putPooledMessage(msg)
return nil, err
}
return msg, nil
}

// toSubMessage converts val (map[string]interface{} or map[interface{}]interface{})
// into a populated *dynamicpb.Message using a pooled message.
// It is the single authoritative path for building sub-messages from either the
// typed (mapToBinary) or raw (rawMapToBinary) Fluent Bit record paths.
func toSubMessage(val interface{}, md protoreflect.MessageDescriptor, cache fieldLookupCache) (*dynamicpb.Message, error) {
switch sub := val.(type) {
case map[interface{}]interface{}:
subMsg := getPooledMessage(md)
if err := rawPopulateMessage(subMsg, md, sub, cache); err != nil {
putPooledMessage(subMsg)
return nil, err
}
return subMsg, nil
case map[string]interface{}:
return mapToMessage(md, sub, cache)
default:
return nil, fmt.Errorf("expected map for message field, got %T", val)
}
}

// 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 {
Expand Down Expand Up @@ -196,11 +235,7 @@ func populateMessage(msg *dynamicpb.Message, md protoreflect.MessageDescriptor,
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 fmt.Errorf("field %q: expected map for message field, got %T", key, val)
}
subMsg, err := mapToMessage(fd.Message(), subMap, cache)
subMsg, err := toSubMessage(val, fd.Message(), cache)
if err != nil {
return fmt.Errorf("field %q: %w", key, err)
}
Expand Down Expand Up @@ -247,26 +282,15 @@ func rawPopulateMessage(msg *dynamicpb.Message, md protoreflect.MessageDescripto
}

if fd.IsList() {
if err := rawSetRepeatedField(msg, fd, val, cache); err != nil {
if err := setRepeatedField(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)
subMsg, err := toSubMessage(val, fd.Message(), cache)
if err != nil {
return fmt.Errorf("field %q: %w", key, err)
}
msg.Set(fd, protoreflect.ValueOfMessage(subMsg))
} else {
pv, err := goToProtoScalar(fd, val)
if err != nil {
Expand Down Expand Up @@ -315,11 +339,7 @@ func setRepeatedField(msg *dynamicpb.Message, fd protoreflect.FieldDescriptor, v
continue
}
if fd.Kind() == protoreflect.MessageKind || fd.Kind() == protoreflect.GroupKind {
subMap, ok := item.(map[string]interface{})
if !ok {
return fmt.Errorf("element %d: expected map for repeated message, got %T", i, item)
}
subMsg, err := mapToMessage(fd.Message(), subMap, cache)
subMsg, err := toSubMessage(item, fd.Message(), cache)
if err != nil {
return fmt.Errorf("element %d: %w", i, err)
}
Expand All @@ -335,50 +355,6 @@ 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.
Expand Down Expand Up @@ -414,7 +390,7 @@ func toProtoString(val interface{}) (protoreflect.Value, error) {
case string:
return protoreflect.ValueOfString(v), nil
case []byte:
return protoreflect.ValueOfString(string(v)), nil
return protoreflect.ValueOfString(unsafe.String(unsafe.SliceData(v), len(v))), nil
case int:
return protoreflect.ValueOfString(strconv.Itoa(v)), nil
case int64:
Expand Down Expand Up @@ -458,7 +434,16 @@ func toProtoInt64(val interface{}) (protoreflect.Value, error) {
}
return protoreflect.ValueOfInt64(i), nil
case []byte:
return toProtoInt64(string(v))
s := unsafe.String(unsafe.SliceData(v), len(v))
i, err := strconv.ParseInt(s, 10, 64)
if err != nil {
f, ferr := strconv.ParseFloat(s, 64)
if ferr != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to int64: %w", v, err)
}
return protoreflect.ValueOfInt64(int64(f)), nil
}
return protoreflect.ValueOfInt64(i), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to int64", val)
}
Expand Down Expand Up @@ -487,7 +472,11 @@ func toProtoInt32(val interface{}) (protoreflect.Value, error) {
}
return protoreflect.ValueOfInt32(int32(i)), nil
case []byte:
return toProtoInt32(string(v))
i, err := strconv.ParseInt(unsafe.String(unsafe.SliceData(v), len(v)), 10, 32)
if err != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to int32: %w", v, err)
}
return protoreflect.ValueOfInt32(int32(i)), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to int32", val)
}
Expand All @@ -512,7 +501,11 @@ func toProtoUint64(val interface{}) (protoreflect.Value, error) {
}
return protoreflect.ValueOfUint64(u), nil
case []byte:
return toProtoUint64(string(v))
u, err := strconv.ParseUint(unsafe.String(unsafe.SliceData(v), len(v)), 10, 64)
if err != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to uint64: %w", v, err)
}
return protoreflect.ValueOfUint64(u), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to uint64", val)
}
Expand All @@ -537,7 +530,11 @@ func toProtoUint32(val interface{}) (protoreflect.Value, error) {
}
return protoreflect.ValueOfUint32(uint32(u)), nil
case []byte:
return toProtoUint32(string(v))
u, err := strconv.ParseUint(unsafe.String(unsafe.SliceData(v), len(v)), 10, 32)
if err != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to uint32: %w", v, err)
}
return protoreflect.ValueOfUint32(uint32(u)), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to uint32", val)
}
Expand All @@ -562,7 +559,11 @@ func toProtoDouble(val interface{}) (protoreflect.Value, error) {
}
return protoreflect.ValueOfFloat64(f), nil
case []byte:
return toProtoDouble(string(v))
f, err := strconv.ParseFloat(unsafe.String(unsafe.SliceData(v), len(v)), 64)
if err != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to float64: %w", v, err)
}
return protoreflect.ValueOfFloat64(f), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to float64", val)
}
Expand All @@ -585,7 +586,11 @@ func toProtoFloat(val interface{}) (protoreflect.Value, error) {
}
return protoreflect.ValueOfFloat32(float32(f)), nil
case []byte:
return toProtoFloat(string(v))
f, err := strconv.ParseFloat(unsafe.String(unsafe.SliceData(v), len(v)), 32)
if err != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to float32: %w", v, err)
}
return protoreflect.ValueOfFloat32(float32(f)), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to float32", val)
}
Expand All @@ -610,7 +615,11 @@ func toProtoBool(val interface{}) (protoreflect.Value, error) {
case float64:
return protoreflect.ValueOfBool(v != 0), nil
case []byte:
return toProtoBool(string(v))
b, err := strconv.ParseBool(unsafe.String(unsafe.SliceData(v), len(v)))
if err != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to bool: %w", v, err)
}
return protoreflect.ValueOfBool(b), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to bool", val)
}
Expand Down Expand Up @@ -656,7 +665,11 @@ func toProtoEnum(val interface{}) (protoreflect.Value, error) {
}
return protoreflect.ValueOfEnum(protoreflect.EnumNumber(i)), nil
case []byte:
return toProtoEnum(string(v))
i, err := strconv.ParseInt(unsafe.String(unsafe.SliceData(v), len(v)), 10, 32)
if err != nil {
return protoreflect.Value{}, fmt.Errorf("cannot convert []byte %q to enum: %w", v, err)
}
return protoreflect.ValueOfEnum(protoreflect.EnumNumber(i)), nil
default:
return protoreflect.Value{}, fmt.Errorf("cannot convert %T to enum", val)
}
Expand Down
Loading