| 862 | } |
| 863 | |
| 864 | Status HdfsScanner::DecompressStream(int64_t bytes_to_read, |
| 865 | uint8_t** decompressed_buffer, int64_t* decompressed_len, bool *eosr) { |
| 866 | // Some decompressors, such as Bzip2 API (version 0.9 and later) and Gzip can |
| 867 | // decompress buffers that are read from stream_, so we don't need to read the |
| 868 | // whole file in once. A compressed buffer is passed to ProcessBlockStreaming |
| 869 | // but it may not consume all of the input. |
| 870 | uint8_t* compressed_buffer_ptr = nullptr; |
| 871 | int64_t compressed_buffer_size = 0; |
| 872 | // We don't know how many bytes ProcessBlockStreaming() will consume so we set |
| 873 | // peek=true and then later advance the stream using SkipBytes(). |
| 874 | if (bytes_to_read == -1) { |
| 875 | RETURN_IF_ERROR(stream_->GetBuffer(true, &compressed_buffer_ptr, |
| 876 | &compressed_buffer_size)); |
| 877 | } else { |
| 878 | DCHECK_GT(bytes_to_read, 0); |
| 879 | Status status; |
| 880 | if (!stream_->GetBytes(bytes_to_read, &compressed_buffer_ptr, &compressed_buffer_size, |
| 881 | &status, true)) { |
| 882 | DCHECK(!status.ok()); |
| 883 | return status; |
| 884 | } |
| 885 | } |
| 886 | int64_t compressed_buffer_bytes_read = 0; |
| 887 | bool stream_end = false; |
| 888 | { |
| 889 | SCOPED_TIMER(decompress_timer_); |
| 890 | Status status = decompressor_->ProcessBlockStreaming(compressed_buffer_size, |
| 891 | compressed_buffer_ptr, &compressed_buffer_bytes_read, decompressed_len, |
| 892 | decompressed_buffer, &stream_end); |
| 893 | if (!status.ok()) { |
| 894 | status.AddDetail(Substitute("$0file=$1, offset=$2", status.GetDetail(), |
| 895 | stream_->filename(), stream_->file_offset())); |
| 896 | return status; |
| 897 | } |
| 898 | DCHECK_GE(compressed_buffer_size, compressed_buffer_bytes_read); |
| 899 | } |
| 900 | // Skip the bytes in stream_ that were decompressed. |
| 901 | Status status; |
| 902 | if (!stream_->SkipBytes(compressed_buffer_bytes_read, &status)) { |
| 903 | DCHECK(!status.ok()); |
| 904 | return status; |
| 905 | } |
| 906 | |
| 907 | if (stream_->eosr()) { |
| 908 | if (stream_end) { |
| 909 | *eosr = true; |
| 910 | } else { |
| 911 | return Status(TErrorCode::COMPRESSED_FILE_TRUNCATED, stream_->filename()); |
| 912 | } |
| 913 | } else if (*decompressed_len == 0) { |
| 914 | return Status(TErrorCode::COMPRESSED_FILE_DECOMPRESSOR_NO_PROGRESS, |
| 915 | stream_->filename()); |
| 916 | } |
| 917 | |
| 918 | return Status::OK(); |
| 919 | } |
| 920 | |
| 921 | bool HdfsScanner::ReportTupleParseError(FieldLocation* fields, uint8_t* errors) { |
nothing calls this directly
no test coverage detected