LuoGe
/
openGauss-server
forked from huawei/openGauss-server
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. bugfix: remote read failed when opt > 0xff 

2. bugfix: byte_convert(diff_convert) can used independently
3. bugfix: checksum alway returns the same uint32
4. cleancode: remove dulicate check of mmap
5. bugfix: parallel repare pca when extreme rto
This commit is contained in:
wuyuechuan 2021-12-21 20:47:04 +08:00
parent 23ddd1ca38
commit d4a2c0cfc5
13 changed files with 84 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
contrib/pagehack/openGaussCompression.h
View File

@ -1,7 +1,3 @@
//
// Created by w00427717 on 2021/11/30.
//
#ifndef OPENGAUSS_SERVER_OPENGAUSSCOMPRESSION_H
#define OPENGAUSS_SERVER_OPENGAUSSCOMPRESSION_H
#define FRONTEND 1

2
src/bin/pg_rewind/file_ops.cpp
View File

@ -1366,5 +1366,5 @@ void FetchCompressedFile(char* buf, BlockNumber blockNumber, int32 size)
write_target_range(buffer_pos, seekpos, write_amount, 0, true);
}
pcAddr->nchunks = pcAddr->allocated_chunks;
pcAddr->checksum = AddrChecksum32(blockNumber, pcAddr);
pcAddr->checksum = AddrChecksum32(blockNumber, pcAddr, chunkSize);
}

2
src/gausskernel/cbb/grpc/remote_read_client.cpp
View File

@ -247,7 +247,7 @@ int RemoteGetPage(char* remoteAddress, uint32 spcnode, uint32 dbnode, uint32 rel
tnRet = snprintf_s(sqlCommands, MAX_PATH_LEN, MAX_PATH_LEN - 1,
"SELECT gs_read_block_from_remote(%u, %u, %u, %d, %d, %d, '%lu', %u, '%lu', false);", spcnode,
dbnode, relnode, bucketnode, opt, forknum, blocknum, blocksize, lsn);
dbnode, relnode, bucketnode, (int2)opt, forknum, blocknum, blocksize, lsn);
securec_check_ss(tnRet, "", "");

45
src/gausskernel/optimizer/commands/tablecmds.cpp
View File

@ -1084,10 +1084,7 @@ static List* AddDefaultOptionsIfNeed(List* options, const char relkind, CreateSt
bool isUstore = false;
bool assignedStorageType = false;
bool hasRowCompressType = false;
bool hasRowCompressChunk = false;
bool hasRowCompressPre = false;
bool hasRowCompressLevel = false;
TableCreateSupport tableCreateSupport{false,false,false,false,false,false};
(void)isOrientationSet(options, NULL, false);
foreach (cell, options) {
DefElem* def = (DefElem*)lfirst(cell);
@ -1118,13 +1115,17 @@ static List* AddDefaultOptionsIfNeed(List* options, const char relkind, CreateSt
(errcode(ERRCODE_INVALID_OPTION),
errmsg("It is not allowed to assign version option for non-dfs table.")));
} else if (pg_strcasecmp(def->defname, "compresstype") == 0) {
hasRowCompressType = true;
tableCreateSupport.compressType = true;
} else if (pg_strcasecmp(def->defname, "compress_chunk_size") == 0) {
hasRowCompressChunk = true;
tableCreateSupport.compressChunkSize = true;
} else if (pg_strcasecmp(def->defname, "compress_prealloc_chunks") == 0) {
hasRowCompressPre = true;
tableCreateSupport.compressPreAllocChunks = true;
} else if (pg_strcasecmp(def->defname, "compress_level") == 0) {
hasRowCompressLevel = true;
tableCreateSupport.compressLevel = true;
} else if (pg_strcasecmp(def->defname, "compress_byte_convert") == 0) {
tableCreateSupport.compressByteConvert = true;
} else if (pg_strcasecmp(def->defname, "compress_diff_convert") == 0) {
tableCreateSupport.compressDiffConvert = true;
}
if (pg_strcasecmp(def->defname, "orientation") == 0 && pg_strcasecmp(defGetString(def), ORIENTATION_ORC) == 0) {
@ -1150,23 +1151,17 @@ static List* AddDefaultOptionsIfNeed(List* options, const char relkind, CreateSt
res = lappend(options, def);
}
if ((isCStore || isTsStore || relkind != RELKIND_RELATION ||
stmt->relation->relpersistence == RELPERSISTENCE_UNLOGGED ||
stmt->relation->relpersistence == RELPERSISTENCE_TEMP ||
stmt->relation->relpersistence == RELPERSISTENCE_GLOBAL_TEMP) &&
(hasRowCompressType || hasRowCompressChunk || hasRowCompressPre || hasRowCompressLevel)) {
ereport(ERROR,
(errcode(ERRCODE_INVALID_OPTION),
errmsg("only row orientation table support "
"compresstype/compress_chunk_size/compress_prealloc_chunks/compress_level.")));
bool noSupportTable = isCStore || isTsStore || relkind != RELKIND_RELATION ||
stmt->relation->relpersistence == RELPERSISTENCE_UNLOGGED ||
stmt->relation->relpersistence == RELPERSISTENCE_TEMP ||
stmt->relation->relpersistence == RELPERSISTENCE_GLOBAL_TEMP;
if (noSupportTable && tableCreateSupport.compressType) {
ereport(ERROR, (errcode(ERRCODE_INVALID_OPTION), errmsg("only row orientation table support compresstype.")));
}
if (!hasRowCompressType && (hasRowCompressChunk || hasRowCompressPre || hasRowCompressLevel)) {
ereport(ERROR,
(errcode(ERRCODE_INVALID_OPTION),
errmsg("compress_chunk_size/compress_prealloc_chunks/compress_level "
"should be used with compresstype.")));
if (!tableCreateSupport.compressType && HasCompressOption(&tableCreateSupport)) {
ereport(ERROR, (errcode(ERRCODE_INVALID_OPTION),
errmsg("compress_chunk_size/compress_prealloc_chunks/compress_level/compress_byte_convert/"
"compress_diff_convert should be used with compresstype.")));
}
if (isUstore && !isCStore && !hasCompression) {
@ -1204,7 +1199,7 @@ static List* AddDefaultOptionsIfNeed(List* options, const char relkind, CreateSt
DefElem *def1 = makeDefElem("orientation", (Node *)makeString(ORIENTATION_ROW));
res = lcons(def1, options);
}
if (!hasCompression && !hasRowCompressType) {
if (!hasCompression && !tableCreateSupport.compressType) {
DefElem *def2 = makeDefElem("compression", (Node *)rowCmprOpt);
res = lappend(options, def2);
}

25
src/gausskernel/storage/file/fd.cpp
View File

@ -3949,24 +3949,6 @@ void SetupPageCompressMemoryMap(File file, RelFileNode node, const RelFileNodeFo
RelFileNodeForkNum newOne(relFileNodeForkNum);
newOne.forknumber = PCA_FORKNUM;
PageCompressHeader *map = GetPageCompressHeader(vfdP, chunk_size, newOne);
if (map == (void *) (-1)) {
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg("Failed to mmap page compression address file %s: %m",
vfdP->fileName)));
}
if (map->chunk_size == 0 && map->algorithm == 0) {
map->chunk_size = chunk_size;
map->algorithm = GET_COMPRESS_ALGORITHM(node.opt);
if (pc_msync(map) != 0) {
ereport(data_sync_elevel(ERROR),
(errcode_for_file_access(), errmsg("could not msync file \"%s\": %m", vfdP->fileName)));
}
}
if (t_thrd.xlog_cxt.InRecovery) {
CheckAndRepairCompressAddress(map, chunk_size, map->algorithm, vfdP->fileName);
}
vfdP->with_pcmap = true;
vfdP->pcmap = map;
}
@ -3991,14 +3973,9 @@ PageCompressHeader *GetPageCompressMemoryMap(File file, uint32 chunk_size)
Assert(vfdP->with_pcmap);
if (vfdP->pcmap == NULL) {
map = GetPageCompressHeader(vfdP, chunk_size, vfdP->fileNode);
if (map == MAP_FAILED) {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES), errmsg(
"Failed to mmap page compression address file %s: %m", vfdP->fileName)));
}
vfdP->with_pcmap = true;
vfdP->pcmap = map;
}
return vfdP->pcmap;
}
}

3
src/gausskernel/storage/replication/basebackup.cpp
View File

@ -1994,8 +1994,7 @@ static void SendCompressedFile(char* readFileName, int basePathLen, struct stat&
for (size_t i = 0; i < nchunks; i++) {
addr->chunknos[i] = chunkIndex++;
}
addr->checksum = AddrChecksum32(blockNum, addr);
addr->checksum = AddrChecksum32(blockNum, addr, chunkSize);
totalLen += len;
}
ReleaseMap(map, readFileName);

10
src/gausskernel/storage/smgr/md.cpp
View File

@ -835,7 +835,7 @@ static void mdextend_pc(SMgrRelation reln, ForkNumber forknum, BlockNumber block
}
/* write checksum */
pcAddr->checksum = AddrChecksum32(blocknum, pcAddr);
pcAddr->checksum = AddrChecksum32(blocknum, pcAddr, chunk_size);
if (pg_atomic_read_u32(&pcMap->nblocks) < blocknum % RELSEG_SIZE + 1) {
pg_atomic_write_u32(&pcMap->nblocks, blocknum % RELSEG_SIZE + 1);
@ -1890,11 +1890,11 @@ static void mdwrite_pc(SMgrRelation reln, ForkNumber forknum, BlockNumber blockn
/* write checksum */
if (mmapSync) {
pcMap->sync = false;
pcAddr->checksum = AddrChecksum32(blocknum, pcAddr);
pcAddr->checksum = AddrChecksum32(blocknum, pcAddr, chunk_size);
}
/* write checksum */
pcAddr->checksum = AddrChecksum32(blocknum, pcAddr);
pcAddr->checksum = AddrChecksum32(blocknum, pcAddr, chunk_size);
mmapSync = false;
if (work_buffer != NULL && work_buffer != buffer) {
@ -2225,7 +2225,7 @@ void mdtruncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
for (BlockNumber blk = 0; blk < RELSEG_SIZE; ++blk) {
pcAddr = GET_PAGE_COMPRESS_ADDR(pcMap, chunk_size, blk);
pcAddr->nchunks = 0;
pcAddr->checksum = AddrChecksum32(blk, pcAddr);
pcAddr->checksum = AddrChecksum32(blk, pcAddr, chunk_size);
}
pg_atomic_write_u32(&pcMap->nblocks, last_seg_blocks);
pcMap->sync = false;
@ -2461,7 +2461,6 @@ static MdfdVec *_mdfd_openseg(SMgrRelation reln, ForkNumber forknum, BlockNumber
/* open the file */
fd = DataFileIdOpenFile(fullpath, filenode, O_RDWR | PG_BINARY | oflags, FILE_RW_PERMISSION);
pfree(fullpath);
if (fd < 0) {
return NULL;
@ -2485,6 +2484,7 @@ static MdfdVec *_mdfd_openseg(SMgrRelation reln, ForkNumber forknum, BlockNumber
SetupPageCompressMemoryMap(fd_pca, reln->smgr_rnode.node, filenode);
}
pfree(fullpath);
/* allocate an mdfdvec entry for it */
v = _fdvec_alloc();

35
src/gausskernel/storage/smgr/mmap_shared.cpp
View File

@ -55,21 +55,19 @@ static inline pthread_mutex_t *MmapPartitionLock(size_t hashCode)
return &mmapLockArray[hashCode % LOCK_ARRAY_SIZE];
}
static inline PageCompressHeader *MmapSharedMapFile(Vfd *vfdP, int chunkSize, bool readonly)
static inline PageCompressHeader *MmapSharedMapFile(Vfd *vfdP, uint16 chunkSize, uint2 opt, bool readonly)
{
PageCompressHeader *map = NULL;
size_t pcMapSize = SIZE_OF_PAGE_COMPRESS_ADDR_FILE(chunkSize);
bool status = compressed_mem_reserve(pcMapSize, false);
if (status) {
map = pc_mmap_real_size(vfdP->fd, pcMapSize, false);
if (map == MAP_FAILED) {
compressed_mem_release(pcMapSize);
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
errmsg("Failed to mmap page compression address file %s: %m", vfdP->fileName)));
auto map = pc_mmap_real_size(vfdP->fd, SIZE_OF_PAGE_COMPRESS_ADDR_FILE(chunkSize), false);
if (map->chunk_size == 0 || map->algorithm == 0) {
map->chunk_size = chunkSize;
map->algorithm = GET_COMPRESS_ALGORITHM(opt);
if (pc_msync(map) != 0) {
ereport(data_sync_elevel(ERROR),
(errcode_for_file_access(), errmsg("could not msync file \"%s\": %m", vfdP->fileName)));
}
} else {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
errmsg("Failed to mmap page compression address file %s: %m", vfdP->fileName)));
}
if (RecoveryInProgress() && !map->sync) {
CheckAndRepairCompressAddress(map, chunkSize, map->algorithm, vfdP->fileName);
}
return map;
}
@ -96,11 +94,8 @@ void RealInitialMMapLockArray()
HASH_ELEM | HASH_FUNCTION | HASH_PARTITION);
}
PageCompressHeader *GetPageCompressHeader(void *vfd, int chunkSize, const RelFileNodeForkNum &relFileNodeForkNum)
PageCompressHeader *GetPageCompressHeader(void *vfd, uint16 chunkSize, const RelFileNodeForkNum &relFileNodeForkNum)
{
if (IsInitdb && g_instance.mmapCache == NULL) {
RealInitialMMapLockArray();
}
Vfd *currentVfd = (Vfd *)vfd;
uint32 hashCode = MmapTableHashCode(relFileNodeForkNum);
AutoMutexLock mmapLock(MmapPartitionLock(hashCode));
@ -114,7 +109,7 @@ PageCompressHeader *GetPageCompressHeader(void *vfd, int chunkSize, const RelFil
mmapEntry->reference = 0;
}
if (mmapEntry->pcmap == NULL) {
mmapEntry->pcmap = MmapSharedMapFile(currentVfd, chunkSize, false);
mmapEntry->pcmap = MmapSharedMapFile(currentVfd, chunkSize, relFileNodeForkNum.rnode.node.opt, false);
}
++mmapEntry->reference;
mmapLock.unLock();
@ -137,12 +132,10 @@ void UnReferenceAddrFile(void *vfd)
}
--mmapEntry->reference;
if (mmapEntry->reference == 0) {
size_t chunkSize = mmapEntry->pcmap->chunk_size;
if (pc_munmap(mmapEntry->pcmap) != 0) {
ereport(ERROR,
(errcode_for_file_access(), errmsg("could not munmap file \"%s\": %m", currentVfd->fileName)));
}
compressed_mem_release(SIZE_OF_PAGE_COMPRESS_ADDR_FILE(chunkSize));
if (hash_search_with_hash_value(g_instance.mmapCache, (void *)&relFileNodeForkNum, hashCode, HASH_REMOVE,
NULL) == NULL) {
ereport(ERROR,
@ -153,4 +146,4 @@ void UnReferenceAddrFile(void *vfd)
ereport(FATAL, (errcode_for_file_access(), errmsg("could not munmap file \"%s\": %m", currentVfd->fileName)));
}
mmapLock.unLock();
}
}

2
src/gausskernel/storage/smgr/page_compression.cpp
View File

@ -88,7 +88,7 @@ void CheckAndRepairCompressAddress(PageCompressHeader *pcMap, uint16 chunk_size,
/* check compress address of every pages */
for (BlockNumber blocknum = 0; blocknum < (BlockNumber)RELSEG_SIZE; ++blocknum) {
PageCompressAddr *pcAddr = GET_PAGE_COMPRESS_ADDR(pcMap, chunk_size, blocknum);
if (pcAddr->checksum != AddrChecksum32(blocknum, pcAddr)) {
if (pcAddr->checksum != AddrChecksum32(blocknum, pcAddr, chunk_size)) {
ereport(WARNING, (errcode(ERRCODE_DATA_CORRUPTED), errmsg("invalid checkum %u of block %u in file \"%s\"",
pcAddr->checksum, blocknum, path)));
pcAddr->allocated_chunks = pcAddr->nchunks = 0;

23
src/include/storage/page_compression.h
View File

@ -113,6 +113,21 @@ typedef struct HeapPageCompressData {
char data[FLEXIBLE_ARRAY_MEMBER]; /* compressed page, except for the page header */
} HeapPageCompressData;
struct TableCreateSupport {
bool compressType;
bool compressLevel;
bool compressChunkSize;
bool compressPreAllocChunks;
bool compressByteConvert;
bool compressDiffConvert;
};
inline bool HasCompressOption(TableCreateSupport *tableCreateSupport)
{
return tableCreateSupport->compressLevel || tableCreateSupport->compressChunkSize ||
tableCreateSupport->compressPreAllocChunks || tableCreateSupport->compressByteConvert ||
tableCreateSupport->compressDiffConvert;
}
const uint4 CHUNK_SIZE_LIST[4] = {BLCKSZ / 2, BLCKSZ / 4, BLCKSZ / 8, BLCKSZ / 16};
constexpr uint4 INDEX_OF_HALF_BLCKSZ = 0;
@ -167,13 +182,13 @@ constexpr unsigned CMP_LEVEL_INDEX = 4;
constexpr unsigned CMP_ALGORITHM_INDEX = 5;
constexpr unsigned CMP_CHUNK_SIZE_INDEX = 6;
struct CmpBitStuct {
struct CmpBitStruct {
unsigned int bitLen;
unsigned int mask;
unsigned int moveBit;
};
constexpr CmpBitStuct g_cmpBitStruct[] = {{CMP_BYTE_CONVERT_LEN, 0x01, 15},
constexpr CmpBitStruct g_cmpBitStruct[] = {{CMP_BYTE_CONVERT_LEN, 0x01, 15},
{CMP_DIFF_CONVERT_LEN, 0x01, 14},
{CMP_PRE_CHUNK_LEN, 0x07, 11},
{CMP_LEVEL_SYMBOL_LEN, 0x01, 10},
@ -323,11 +338,11 @@ extern uint1 ConvertChunkSize(uint32 compressedChunkSize, bool* success);
* @param pageCompressAddr addr of block
* @return checksum uint32
*/
extern uint32 AddrChecksum32(BlockNumber blockNumber, const PageCompressAddr* pageCompressAddr);
extern uint32 AddrChecksum32(BlockNumber blockNumber, const PageCompressAddr* pageCompressAddr, uint16 chunkSize);
#ifndef FRONTEND
extern void CheckAndRepairCompressAddress(PageCompressHeader *pcMap, uint16 chunk_size, uint8 algorithm, const char *path);
PageCompressHeader* GetPageCompressHeader(void* vfd, int chunkSize, const RelFileNodeForkNum &relFileNodeForkNum);
PageCompressHeader* GetPageCompressHeader(void* vfd, uint16 chunkSize, const RelFileNodeForkNum &relFileNodeForkNum);
void UnReferenceAddrFile(void* vfd);
void RealInitialMMapLockArray();
#endif

12
src/include/storage/page_compression_impl.h
View File

@ -333,6 +333,12 @@ void CompressPagePrepareConvert(char *src, bool diff_convert, bool *real_ByteCon
FreePointer((void*)aux_buf);
}
inline size_t CompressReservedLen(const char* page)
{
auto length = offsetof(HeapPageCompressData, page_header) - offsetof(HeapPageCompressData, data);
return GetPageHeaderSize(page) + length;
}
/**
* CompressPageBufferBound()
* -- Get the destination buffer boundary to compress one page.
@ -345,7 +351,7 @@ int CompressPageBufferBound(const char* page, uint8 algorithm)
case COMPRESS_ALGORITHM_PGLZ:
return BLCKSZ + 4;
case COMPRESS_ALGORITHM_ZSTD:
return ZSTD_compressBound(BLCKSZ - GetPageHeaderSize(page));
return ZSTD_compressBound(BLCKSZ - CompressReservedLen(page));
default:
return -1;
}
@ -688,12 +694,12 @@ int pc_msync(PageCompressHeader *map)
}
uint32 AddrChecksum32(BlockNumber blockNumber, const PageCompressAddr* pageCompressAddr)
uint32 AddrChecksum32(BlockNumber blockNumber, const PageCompressAddr* pageCompressAddr, uint16 chunkSize)
{
#define UINT_LEN sizeof(uint32)
uint32 checkSum = 0;
char* addr = ((char*) pageCompressAddr) + UINT_LEN;
size_t len = sizeof(PageCompressAddr) - UINT_LEN;
size_t len = SIZE_OF_PAGE_COMPRESS_ADDR(chunkSize) - UINT_LEN;
do {
if (len >= UINT_LEN) {
checkSum += *((uint32*) addr);

16
src/test/regress/expected/row_compression/unsupported_feature.out
View File

@ -20,20 +20,24 @@ ERROR: value 128 out of bounds for option "compress_level"
DETAIL: Valid values are between "-31" and "31".
-- compresstype cant be used with column table
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(ORIENTATION = 'column', compresstype=2);
ERROR: only row orientation table support compresstype/compress_chunk_size/compress_prealloc_chunks/compress_level.
ERROR: only row orientation table support compresstype.
-- compresstype cant be used with temp table
CREATE TEMP TABLE compressed_temp_table_1024(id int) WITH(compresstype=2);
ERROR: only row orientation table support compresstype/compress_chunk_size/compress_prealloc_chunks/compress_level.
ERROR: only row orientation table support compresstype.
-- compresstype cant be used with unlogged table
CREATE unlogged TABLE compressed_unlogged_table_1024(id int) WITH(compresstype=2);
ERROR: only row orientation table support compresstype/compress_chunk_size/compress_prealloc_chunks/compress_level.
ERROR: only row orientation table support compresstype.
-- use compress_prealloc_chunks\compress_chunk_size\compress_level without compresstype
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_prealloc_chunks=5);
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level should be used with compresstype.
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level/compress_byte_convert/compress_diff_convert should be used with compresstype.
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_chunk_size=1024);
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level should be used with compresstype.
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level/compress_byte_convert/compress_diff_convert should be used with compresstype.
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_byte_convert=true);
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level/compress_byte_convert/compress_diff_convert should be used with compresstype.
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_diff_convert=true);
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level/compress_byte_convert/compress_diff_convert should be used with compresstype.
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_level=5);
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level should be used with compresstype.
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level/compress_byte_convert/compress_diff_convert should be used with compresstype.
-- unspport exchange
CREATE TABLE unspported_feature.exchange_table(id int) WITH(compresstype=2);
CREATE TABLE unspported_feature.alter_table(id int) partition by range(id)

2
src/test/regress/sql/row_compression/unsupported_feature.sql
View File

@ -18,6 +18,8 @@ CREATE unlogged TABLE compressed_unlogged_table_1024(id int) WITH(compresstype=2
-- use compress_prealloc_chunks\compress_chunk_size\compress_level without compresstype
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_prealloc_chunks=5);
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_chunk_size=1024);
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_byte_convert=true);
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_diff_convert=true);
CREATE TABLE unspported_feature.compressed_table_1024(id int) WITH(compress_level=5);
-- unspport exchange
CREATE TABLE unspported_feature.exchange_table(id int) WITH(compresstype=2);