This is an automated email from the ASF dual-hosted git repository.
liaoxin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 73590e6e0ec [refactor](compress) Unify the decompressin error msg
(#56299)
73590e6e0ec is described below
commit 73590e6e0eca0acabf9588b882b3b9236d6d5eea
Author: Refrain <[email protected]>
AuthorDate: Thu Sep 25 23:25:53 2025 +0800
[refactor](compress) Unify the decompressin error msg (#56299)
---
be/src/exec/decompressor.cpp | 36 ++++++++++++++++++++++--------------
1 file changed, 22 insertions(+), 14 deletions(-)
diff --git a/be/src/exec/decompressor.cpp b/be/src/exec/decompressor.cpp
index d117ba2236b..44893af849c 100644
--- a/be/src/exec/decompressor.cpp
+++ b/be/src/exec/decompressor.cpp
@@ -177,7 +177,7 @@ Status GzipDecompressor::init() {
int window_bits = _is_deflate ? WINDOW_BITS : (WINDOW_BITS | DETECT_CODEC);
int ret = inflateInit2(&_z_strm, window_bits);
if (ret < 0) {
- return Status::InternalError("Failed to init inflate. status code:
{}", ret);
+ return Status::InternalError("Failed to do gzip decompress. status
code: {}", ret);
}
return Status::OK();
@@ -221,10 +221,21 @@ Status GzipDecompressor::decompress(uint8_t* input,
uint32_t input_len, size_t*
// reset _z_strm to continue decoding a subsequent gzip stream
ret = inflateReset(&_z_strm);
if (ret != Z_OK) {
- return Status::InternalError("Failed to inflateReset. return
code: {}", ret);
+ if (_is_deflate) {
+ return Status::InternalError("Failed to do deflate
decompress. return code: {}",
+ ret);
+ } else {
+ return Status::InternalError("Failed to do gzip
decompress. return code: {}",
+ ret);
+ }
}
} else if (ret != Z_OK) {
- return Status::InternalError("Failed to inflate. return code: {}",
ret);
+ if (_is_deflate) {
+ return Status::InternalError("Failed to do deflate decompress.
return code: {}",
+ ret);
+ } else {
+ return Status::InternalError("Failed to do gzip decompress.
return code: {}", ret);
+ }
} else {
// here ret must be Z_OK.
// we continue if avail_out and avail_in > 0.
@@ -251,7 +262,7 @@ Status Bzip2Decompressor::init() {
bzero(&_bz_strm, sizeof(_bz_strm));
int ret = BZ2_bzDecompressInit(&_bz_strm, 0, 0);
if (ret != BZ_OK) {
- return Status::InternalError("Failed to init bz2. status code: {}",
ret);
+ return Status::InternalError("Failed to do bz2 decompress. status
code: {}", ret);
}
return Status::OK();
@@ -277,19 +288,17 @@ Status Bzip2Decompressor::decompress(uint8_t* input,
uint32_t input_len, size_t*
if (ret == BZ_DATA_ERROR || ret == BZ_DATA_ERROR_MAGIC) {
LOG(INFO) << "input_bytes_read: " << *input_bytes_read
<< " decompressed_len: " << *decompressed_len;
- return Status::InternalError("Failed to bz2 decompress. status
code: {}", ret);
+ return Status::InternalError("Failed to do bz2 decompress. status
code: {}", ret);
} else if (ret == BZ_STREAM_END) {
*stream_end = true;
ret = BZ2_bzDecompressEnd(&_bz_strm);
if (ret != BZ_OK) {
- return Status::InternalError(
- "Failed to end bz2 after meet BZ_STREAM_END. status
code: {}", ret);
+ return Status::InternalError("Failed to do bz2 decompress.
status code: {}", ret);
}
ret = BZ2_bzDecompressInit(&_bz_strm, 0, 0);
if (ret != BZ_OK) {
- return Status::InternalError(
- "Failed to init bz2 after meet BZ_STREAM_END. status
code: {}", ret);
+ return Status::InternalError("Failed to do bz2 decompress.
status code: {}", ret);
}
} else if (ret != BZ_OK) {
return Status::InternalError("Failed to bz2 decompress. status
code: {}", ret);
@@ -338,7 +347,7 @@ Status ZstdDecompressor::decompress(uint8_t* input,
uint32_t input_len, size_t*
*decompressed_len = outputBuffer.pos;
if (ZSTD_isError(ret)) {
- return Status::InternalError("Failed to zstd decompress: {}",
ZSTD_getErrorName(ret));
+ return Status::InternalError("Failed to do zstd decompress: {}",
ZSTD_getErrorName(ret));
}
*stream_end = ret == 0;
@@ -549,7 +558,7 @@ Status Lz4BlockDecompressor::decompress(uint8_t* input,
uint32_t input_len,
reinterpret_cast<const char*>(input_ptr),
reinterpret_cast<char*>(output_ptr),
compressed_small_block_len, remaining_output_len);
if (decompressed_small_block_len < 0) {
- return Status::InvalidArgument("fail to do LZ4 decompress,
error = {}",
+ return Status::InvalidArgument("Failed to do Lz4Block
decompress, error = {}",
LZ4F_getErrorName(decompressed_small_block_len));
}
input_ptr += compressed_small_block_len;
@@ -666,14 +675,13 @@ Status SnappyBlockDecompressor::decompress(uint8_t*
input, uint32_t input_len,
if (!snappy::GetUncompressedLength(reinterpret_cast<const
char*>(input_ptr),
compressed_small_block_len,
&decompressed_small_block_len))
{
- return Status::InternalError(
- "snappy block decompress failed to get uncompressed
len");
+ return Status::InternalError("Failed to do snappy
decompress.");
}
if (!snappy::RawUncompress(reinterpret_cast<const
char*>(input_ptr),
compressed_small_block_len,
reinterpret_cast<char*>(output_ptr))) {
return Status::InternalError(
- "snappy block decompress failed. uncompressed_len: {},
compressed_len: {}",
+ "Failed to do snappy decompress. uncompressed_len: {},
compressed_len: {}",
decompressed_small_block_len,
compressed_small_block_len);
}
input_ptr += compressed_small_block_len;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
