Function ParsePBFileHeader

be/src/kudu/util/pb_util.cc:400–451

// Read and parse the protobuf container file-level header documented in pb_util.h.
template<typename ReadableFileType>
Status ParsePBFileHeader(ReadableFileType* reader, optional<uint64_t>* cached_file_size,
                         uint64_t* offset, int* version) {
  RETURN_NOT_OK(CacheFileSize(reader, cached_file_size));
  const uint64_t file_size = *(*cached_file_size);

  // We initially read enough data for a V2+ file header. This optimizes for
  // V2+ and is valid on a V1 file because we don't consider these files valid
  // unless they contain a record in addition to the file header. The
  // additional 4 bytes required by a V2+ header (vs V1) is still less than the
  // minimum number of bytes required for a V1 format data record.
  uint64_t tmp_offset = *offset;
  faststring header;
  RETURN_NOT_OK_PREPEND(ValidateAndReadData(reader, file_size, &tmp_offset, kPBContainerV2HeaderLen,
                                            &header),
                        Substitute("Could not read header for proto container file $0",
                                   reader->filename()));
  Slice magic_and_version(header.data(), kPBContainerMagicLen + sizeof(uint32_t));
  Slice checksum(header.data() + kPBContainerMagicLen + sizeof(uint32_t), kPBContainerChecksumLen);

  // Validate magic number.
  if (PREDICT_FALSE(!strings::memeq(kPBContainerMagic, header.data(), kPBContainerMagicLen))) {
    string file_magic(reinterpret_cast<const char*>(header.data()), kPBContainerMagicLen);
    return Status::Corruption("Invalid magic number",
                              Substitute("Expected: $0, found: $1",
                                         Utf8SafeCEscape(kPBContainerMagic),
                                         Utf8SafeCEscape(file_magic)));
  }

  // Validate container file version.
  uint32_t tmp_version = DecodeFixed32(header.data() + kPBContainerMagicLen);
  if (PREDICT_FALSE(!IsSupportedContainerVersion(tmp_version))) {
    return Status::NotSupported(
        Substitute("Protobuf container has unsupported version: $0. Default version: $1",
                   tmp_version, kPBContainerDefaultVersion));
  }

  // Versions >= 2 have a checksum after the magic number and encoded version
  // to ensure the integrity of these fields.
  if (tmp_version >= 2) {
    RETURN_NOT_OK_PREPEND(ParseAndCompareChecksum(checksum.data(), { magic_and_version }),
        CHECKSUM_ERR_MSG("File header checksum does not match",
                         reader->filename(), tmp_offset - kPBContainerChecksumLen));
  } else {
    // Version 1 doesn't have a header checksum. Rewind our read offset so this
    // data will be read again when we next attempt to read a data record.
    tmp_offset -= kPBContainerChecksumLen;
  }

  *offset = tmp_offset;
  *version = tmp_version;
  return Status::OK();
}

// Read and parse the supplemental header from the container file.
template<typename ReadableFileType>