Native Protobuf Record Payload

[gudladona]

Hello,

I'd like to pitch the idea of making Hudi Record payload support native protobuf bytes. Protobuf as the wire format has become very popular for its small payload size per message, particularly over streaming. Also, Protobuf data can be represented as a spark InternalRow type without needing to convert to an Avro record. Additionally, the Parquet also supports Proto native writer (ProtoParquetWriter) which can take a Dynamic message as source that can be easily built from proto bytes. Doing this would alleviate the expensive conversion of Proto --> Avro --> HudiRecordPayload(contains re-serialized avro bytes) and then use a Proto Avro writer during the file write process.

Kindly let me know your thoughts, I am happy to start a PR if the idea(improvement) sounds legit.

[danny0405]

The payload API is deprecated since 1.1, what you really want to propose here is a new file format encoded using Protobuf I guess? In general two things need clarification:

1. the in-memry representation of the PB payload, because the payload class is deprecated, you might need to impl a new HoodieRecord backed up by PB bytes;
2. the reader/writer API adapter.

[vinothchandar]

@gudladona We plan to move towards RecordMerger and the 1.x merge modes to handle data using engine native POJOs. As long as protobuf can be converted into Spark internalRow at entry, would that address your concern? or are you proposing storing records at rest in protobuf

[gudladona]

Thanks for the feedback, I think I need to familiarize myself with the RecordMerger. Historically, the (Avro) data itself is stored in the Payload as bytes. See the base class below

  /**
   * Instantiate {@link BaseAvroPayload}.
   *
   * @param record      Generic record for the payload.
   * @param orderingVal {@link Comparable} to be used in pre combine.
   */
  public BaseAvroPayload(GenericRecord record, Comparable orderingVal) {
    this.recordBytes = record != null ? HoodieAvroUtils.avroToBytes(record) : new byte[0];
    this.orderingVal = orderingVal;
    this.isDeletedRecord = record == null || isDeleteRecord(record);

    if (orderingVal == null) {
      throw new HoodieException("Ordering value is null for record: " + record);
    }
  }

Here we store the avro as bytes anyway to use the getInsertValue function later to create a GenericRecord. My thinking being we could accomplish the same with proto bytes and a Schema that can prevent the current implementation of a converstion to avro and then conversion to recordBytes and then convert back to GenericRecord. We could have only 1 pass of Proto to avro conversion OR, if a proto parquet writer is used simply pass the proto bytes as the source along with the schema.

[vinothchandar]

@gudladona So you want to shuffle proto bytes all the way to storage?

My thinking being we could accomplish the same with proto bytes

If we are thinking Spark-centric, then it may be better to convert to Row upfront and let Spark optimize it. (for e.g may be down the line, it gets columnar shuffle)

We could have only 1 pass of Proto to avro conversion

That approach will have 1 pass to Row. like today, where we convert 1 pass to avro.

[danny0405]

The whole payload stuff has been deprecated since 1.1: BaseAvroPayload, you need to impl a new HoodieRecord POJO there instead. Takes the master HoodieAvroIndexedRecord as an example.

[vinothchandar]

I'd discourage users from writing any custom records

[danny0405]

yeah, a new record type is deeply plugged into the system of Hudi which I don't like too, just need a way to make the new encoding like PB pluggable here. I mean the request is there.