xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

add compress

This commit is contained in:
Yihao Deng 2024-04-17 07:35:00 +00:00
parent 1d9b65226e
commit 0f629bebda
1 changed files with 25 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -1601,29 +1601,36 @@ int32_t valueDecode(void* value, int32_t vlen, int64_t* ttl, char** dest) {
}
p = taosDecodeFixedI64(p, &key.unixTimestamp);
p = taosDecodeFixedI32(p, &key.len);
p = taosDecodeFixedI32(p, &key.rawLen);
p = taosDecodeFixedI8(p, &key.compress);
if (vlen != (sizeof(key.unixTimestamp) + sizeof(key.len) + sizeof(key.rawLen) + sizeof(key.compress) + key.len)) {
stError("vlen: %d, read len: %d", vlen, key.len);
goto _EXCEPT;
}
if (key.compress == 1) {
char* pCompressData = NULL;
if (key.len != 0) {
p = taosDecodeBinary(p, (void**)&pCompressData, key.len);
*dest = taosMemoryCalloc(1, key.rawLen);
int32_t decompressSize = LZ4_decompress_safe(pCompressData, *dest, key.len, key.rawLen);
ASSERT(decompressSize == key.rawLen);
key.len = decompressSize;
}
taosMemoryFree(pCompressData);
if (vlen == (sizeof(key.unixTimestamp) + sizeof(key.len) + key.len)) {
// compatiable with previous data
if (key.len != 0 && dest != NULL) p = taosDecodeBinary(p, (void**)dest, key.len);
} else {
if (key.len != 0 && dest != NULL) p = taosDecodeBinary(p, (void**)dest, key.len);
p = taosDecodeFixedI32(p, &key.rawLen);
p = taosDecodeFixedI8(p, &key.compress);
if (vlen != (sizeof(key.unixTimestamp) + sizeof(key.len) + sizeof(key.rawLen) + sizeof(key.compress) + key.len)) {
stError("vlen: %d, read len: %d", vlen, key.len);
goto _EXCEPT;
}
if (key.compress == 1) {
char* pCompressData = NULL;
if (key.len != 0) {
p = taosDecodeBinary(p, (void**)&pCompressData, key.len);
*dest = taosMemoryCalloc(1, key.rawLen);
int32_t decompressSize = LZ4_decompress_safe(pCompressData, *dest, key.len, key.rawLen);
ASSERT(decompressSize == key.rawLen);
key.len = decompressSize;
}
taosMemoryFree(pCompressData);
} else {
if (key.len != 0 && dest != NULL) p = taosDecodeBinary(p, (void**)dest, key.len);
}
if (ttl != NULL) *ttl = key.unixTimestamp == 0 ? 0 : key.unixTimestamp - taosGetTimestampMs();
}
if (ttl != NULL) *ttl = key.unixTimestamp == 0 ? 0 : key.unixTimestamp - taosGetTimestampMs();
return key.len;
_EXCEPT: