Read internal::FlightData from arrow::Buffer containing FlightData protobuf without copying
| 770 | // Read internal::FlightData from arrow::Buffer containing FlightData |
| 771 | // protobuf without copying |
| 772 | arrow::Result<arrow::flight::internal::FlightData> DeserializeFlightData( |
| 773 | const std::shared_ptr<arrow::Buffer>& buffer) { |
| 774 | if (!buffer) { |
| 775 | return Status::Invalid("No payload"); |
| 776 | } |
| 777 | |
| 778 | arrow::flight::internal::FlightData out; |
| 779 | |
| 780 | auto buffer_length = static_cast<int>(buffer->size()); |
| 781 | CodedInputStream pb_stream(buffer->data(), buffer_length); |
| 782 | |
| 783 | pb_stream.SetTotalBytesLimit(buffer_length); |
| 784 | |
| 785 | // This is the bytes remaining when using CodedInputStream like this |
| 786 | while (pb_stream.BytesUntilTotalBytesLimit()) { |
| 787 | const uint32_t tag = pb_stream.ReadTag(); |
| 788 | const int field_number = WireFormatLite::GetTagFieldNumber(tag); |
| 789 | switch (field_number) { |
| 790 | case pb::FlightData::kFlightDescriptorFieldNumber: { |
| 791 | pb::FlightDescriptor pb_descriptor; |
| 792 | uint32_t length; |
| 793 | if (!pb_stream.ReadVarint32(&length)) { |
| 794 | return Status::Invalid("Unable to parse length of FlightDescriptor"); |
| 795 | } |
| 796 | // Can't use ParseFromCodedStream as this reads the entire |
| 797 | // rest of the stream into the descriptor command field. |
| 798 | std::string buffer; |
| 799 | if (!pb_stream.ReadString(&buffer, length)) { |
| 800 | return Status::Invalid("Unable to read FlightDescriptor from protobuf"); |
| 801 | } |
| 802 | if (!pb_descriptor.ParseFromString(buffer)) { |
| 803 | return Status::Invalid("Unable to parse FlightDescriptor"); |
| 804 | } |
| 805 | arrow::flight::FlightDescriptor descriptor; |
| 806 | ARROW_RETURN_NOT_OK( |
| 807 | arrow::flight::internal::FromProto(pb_descriptor, &descriptor)); |
| 808 | out.descriptor = std::make_unique<arrow::flight::FlightDescriptor>(descriptor); |
| 809 | } break; |
| 810 | case pb::FlightData::kDataHeaderFieldNumber: { |
| 811 | if (!ReadBytesZeroCopy(buffer, &pb_stream, &out.metadata)) { |
| 812 | return Status::Invalid("Unable to read FlightData metadata"); |
| 813 | } |
| 814 | } break; |
| 815 | case pb::FlightData::kAppMetadataFieldNumber: { |
| 816 | if (!ReadBytesZeroCopy(buffer, &pb_stream, &out.app_metadata)) { |
| 817 | return Status::Invalid("Unable to read FlightData application metadata"); |
| 818 | } |
| 819 | } break; |
| 820 | case pb::FlightData::kDataBodyFieldNumber: { |
| 821 | if (!ReadBytesZeroCopy(buffer, &pb_stream, &out.body)) { |
| 822 | return Status::Invalid("Unable to read FlightData body"); |
| 823 | } |
| 824 | } break; |
| 825 | default: { |
| 826 | // Unknown field. We should skip it for compatibility. |
| 827 | if (!WireFormatLite::SkipField(&pb_stream, tag)) { |
| 828 | return Status::Invalid("Could not skip unknown field tag in FlightData"); |
| 829 | } |