s3: add raw Protobuf input for typed Arrow/Parquet output by rituparnakhaund · Pull Request #12571 · fluent/fluent-bit · GitHub
Skip to content

s3: add raw Protobuf input for typed Arrow/Parquet output - #12571

Draft
rituparnakhaund wants to merge 1 commit into
fluent:masterfrom
rituparnakhaund:s3-parquet-protobuf-poc
Draft

rituparnakhaund wants to merge 1 commit into
fluent:masterfrom
rituparnakhaund:s3-parquet-protobuf-poc

Conversation

@rituparnakhaund

Copy link
Copy Markdown

s3: add raw Protobuf input for typed Arrow/Parquet output

Summary

Add an optional end-to-end raw Protobuf path to the S3 output. A Forward input
can carry each application message as a MessagePack binary field. The S3
output preserves those bytes and record boundaries in its local buffer, loads
a binary google.protobuf.FileDescriptorSet, decodes records with Protobuf
reflection, builds a deterministic Arrow table, and writes Arrow or Parquet
using the existing columnar writers and compression settings.

The payload is never Base64-encoded or converted to JSON. The existing
JSON-to-Arrow path remains unchanged and is still the default.

Motivation

The current Parquet path derives types from JSON values. JSON does not preserve
important source types such as int32 versus int64, float versus double,
or string versus bytes. Inference can also vary with the observed records,
and empty nested messages can produce unsupported zero-child structs.

Protobuf already has the required type information. Consuming its descriptor
and wire payload directly removes JSON schema inference from this path and
allows Parquet to expose typed nested columns.

Design

  • Add an opt-in FLB_PROTOBUF build option.
  • Load a reusable binary FileDescriptorSet and fully-qualified message name.
  • Use DescriptorPool, DynamicMessageFactory, and reflection; generated
    application message classes are not required.
  • Convert a caller-provided batch of individually framed messages into one
    Arrow table.
  • Reuse the existing Arrow/Feather and Parquet writers, including Zstd.
  • Keep libprotoc out of the runtime dependency; only libprotobuf is linked.
  • Add columnar_input_format protobuf to the S3 output, with a configured
    MessagePack binary key, descriptor file, and fully-qualified message name.
  • Persist buffered records as a sequence of standard MessagePack bin values,
    retaining exact payload bytes and message boundaries across restarts.
  • Isolate Protobuf buffers from JSON and from other message types using a
    framing-version/message-type identifier.
  • Reject missing, duplicate, or non-binary input fields instead of coercing
    data or producing corrupt columnar objects.
  • Retry or quarantine invalid buffered Protobuf rather than uploading raw
    bytes under a Parquet/Arrow object name.

Representative mappings:

Protobuf Arrow
int32, sint32, sfixed32 int32
int64, sint64, sfixed64 int64
unsigned integer types matching unsigned integer
float, double float32, float64
string, bytes utf8, binary
enum int32
message struct
repeated list
map map
zero-field message nullable boolean presence marker

Nested google.protobuf.Struct, Value, and ListValue values remain opaque
binary because their recursively dynamic values do not have one finite Arrow
schema. Other recursive message schemas are rejected with a configuration
error.

Descriptor size, file count, and nesting depth are bounded. C++ exceptions are
contained at the C API boundary, invalid payloads return an error, and one
dynamic message allocation is reused across a batch.

Configuration

format                    parquet
compression               zstd
columnar_input_format     protobuf
protobuf_data_key         wire_proto
protobuf_descriptor_file  /etc/fluent-bit/wire-log.desc
protobuf_message_type     com.example.logging.WireLog

The application sends WireLog.toByteArray() as the wire_proto MessagePack
bin value in a standard Forward record. Treating an arbitrary concatenated
byte stream as Protobuf remains unsupported because the Protobuf wire format
does not delimit top-level messages.

Testing

  • Protobuf scalar width and binary type preservation
  • nested messages, lists, maps, and zero-field message presence
  • two-record batch including an empty/default message
  • Zstd-compressed Parquet output and metadata
  • invalid descriptor, unknown message type, recursive schema, and malformed
    payload rejection
  • mocked Forward-over-Unix-socket to S3 PutObject integration with two binary
    records
  • text value rejection and incomplete configuration startup failure
  • existing JSON columnar compression regression test
  • complete mocked S3 runtime suite
  • S3 build with Arrow and Protobuf disabled
  • Valgrind: zero errors and zero definitely lost bytes

@coderabbitai

coderabbitai Bot commented Oct 13, 2026

Copy link
Copy Markdown

This branch had an error being deployed

1 failed deployment
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant