| 421 | } |
| 422 | |
| 423 | Status HdfsSequenceScanner::ReadFileHeader() { |
| 424 | uint8_t* header; |
| 425 | |
| 426 | RETURN_IF_FALSE(stream_->ReadBytes( |
| 427 | sizeof(SEQFILE_VERSION_HEADER), &header, &parse_status_)); |
| 428 | |
| 429 | if (memcmp(header, SEQFILE_VERSION_HEADER, sizeof(SEQFILE_VERSION_HEADER))) { |
| 430 | stringstream ss; |
| 431 | ss << stream_->filename() << " Invalid SEQFILE_VERSION_HEADER: '" |
| 432 | << ReadWriteUtil::HexDump(header, sizeof(SEQFILE_VERSION_HEADER)) << "'"; |
| 433 | return Status(ss.str()); |
| 434 | } |
| 435 | |
| 436 | // We don't care what this is since we don't use the keys. |
| 437 | RETURN_IF_FALSE(stream_->SkipText(&parse_status_)); |
| 438 | |
| 439 | uint8_t* class_name; |
| 440 | int64_t len; |
| 441 | RETURN_IF_FALSE(stream_->ReadText(&class_name, &len, &parse_status_)); |
| 442 | if (memcmp(class_name, HdfsSequenceScanner::SEQFILE_VALUE_CLASS_NAME, len)) { |
| 443 | stringstream ss; |
| 444 | ss << stream_->filename() << " Invalid SEQFILE_VALUE_CLASS_NAME: '" |
| 445 | << string(reinterpret_cast<char*>(class_name), len) << "'"; |
| 446 | return Status(ss.str()); |
| 447 | } |
| 448 | |
| 449 | SeqFileHeader* seq_header = reinterpret_cast<SeqFileHeader*>(header_); |
| 450 | bool is_blk_compressed; |
| 451 | RETURN_IF_FALSE( |
| 452 | stream_->ReadBoolean(&header_->is_compressed, &parse_status_)); |
| 453 | RETURN_IF_FALSE( |
| 454 | stream_->ReadBoolean(&is_blk_compressed, &parse_status_)); |
| 455 | seq_header->is_row_compressed = !is_blk_compressed; |
| 456 | |
| 457 | if (header_->is_compressed) { |
| 458 | uint8_t* codec_ptr; |
| 459 | RETURN_IF_FALSE(stream_->ReadText(&codec_ptr, &len, &parse_status_)); |
| 460 | header_->codec = string(reinterpret_cast<char*>(codec_ptr), len); |
| 461 | Codec::CodecMap::const_iterator it = Codec::CODEC_MAP.find(header_->codec); |
| 462 | if (it == Codec::CODEC_MAP.end()) { |
| 463 | return Status(TErrorCode::COMPRESSED_FILE_BLOCK_CORRUPTED, header_->codec); |
| 464 | } |
| 465 | header_->compression_type = it->second; |
| 466 | } else { |
| 467 | header_->compression_type = THdfsCompression::NONE; |
| 468 | } |
| 469 | VLOG_FILE << stream_->filename() << ": " |
| 470 | << (header_->is_compressed ? |
| 471 | (seq_header->is_row_compressed ? "row compressed" : "block compressed") : |
| 472 | "not compressed"); |
| 473 | if (header_->is_compressed) VLOG_FILE << header_->codec; |
| 474 | |
| 475 | // Skip file metadata |
| 476 | int map_size = 0; |
| 477 | RETURN_IF_FALSE(stream_->ReadInt(&map_size, &parse_status_)); |
| 478 | |
| 479 | for (int i = 0; i < map_size; ++i) { |
| 480 | RETURN_IF_FALSE(stream_->SkipText(&parse_status_)); |
nothing calls this directly
no test coverage detected