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

!1422 oltp compression bug fix 

Merge pull request !1422 from 吴岳川/compress_newly
This commit is contained in:
opengauss-bot 2022-01-28 09:33:16 +00:00 committed by Gitee
parent 3166a5054d 2eb20665ee
commit fac95fa751
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
22 changed files with 144 additions and 173 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
contrib/pagehack/Makefile
View File

@ -14,7 +14,7 @@ subdir = contrib/pagehack
top_builddir = ../..
include $(top_builddir)/src/Makefile.global
enable_shared = false
override CFLAGS += -lzstd
ifeq ($(enable_debug), yes)
PG_CPPFLAGS += -DDEBUG
endif

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);
}

46
src/common/backend/utils/cache/partcache.cpp vendored
View File

@ -235,48 +235,6 @@ static Partition AllocatePartitionDesc(Form_pg_partition partp)
return partition;
}
void SetupPageCompressForPartition(RelFileNode* node, PageCompressOpts* compress_options, const char* relationName)
{
uint1 algorithm = compress_options->compressType;
if (algorithm == COMPRESS_TYPE_NONE) {
node->opt = 0;
} else {
if (!SUPPORT_PAGE_COMPRESSION) {
ereport(ERROR, (errmsg("unsupported page compression on this platform")));
}
uint1 compressLevel;
bool symbol = false;
if (compress_options->compressLevel >= 0) {
symbol = true;
compressLevel = compress_options->compressLevel;
} else {
symbol = false;
compressLevel = -compress_options->compressLevel;
}
bool success = false;
uint1 chunkSize = ConvertChunkSize(compress_options->compressChunkSize, &success);
if (!success) {
ereport(ERROR, (errmsg("invalid compress_chunk_size %d , must be one of %d, %d, %d or %d for %s",
compress_options->compressChunkSize, BLCKSZ / 16, BLCKSZ / 8, BLCKSZ / 4, BLCKSZ / 2,
relationName)));
}
uint1 preallocChunks;
if (compress_options->compressPreallocChunks >= BLCKSZ / compress_options->compressChunkSize) {
preallocChunks = (uint1)(BLCKSZ / compress_options->compressChunkSize - 1);
} else {
preallocChunks = (uint1)(compress_options->compressPreallocChunks);
}
Assert(preallocChunks <= MAX_PREALLOC_CHUNKS);
node->opt = 0;
SET_COMPRESS_OPTION((*node), compress_options->compressByteConvert, compress_options->compressDiffConvert,
preallocChunks, symbol, compressLevel, algorithm, chunkSize);
}
}
StorageType PartitionGetStorageType(Oid parentOid)
{
HeapTuple pg_class_tuple;
@ -432,7 +390,7 @@ static void PartitionInitPhysicalAddr(Partition partition)
partition->pd_node.opt = 0;
if (partition->rd_options) {
SetupPageCompressForPartition(&partition->pd_node, &((StdRdOptions*)(partition->rd_options))->compress,
SetupPageCompressForRelation(&partition->pd_node, &((StdRdOptions*)(partition->rd_options))->compress,
PartitionGetPartitionName(partition));
}
}
@ -575,7 +533,7 @@ Partition PartitionBuildLocalPartition(const char *relname, Oid partid, Oid part
/* compressed option was set by PartitionInitPhysicalAddr if part->rd_options != NULL */
if (part->rd_options == NULL && reloptions) {
StdRdOptions* options = (StdRdOptions*)default_reloptions(reloptions, false, RELOPT_KIND_HEAP);
SetupPageCompressForPartition(&part->pd_node, &options->compress, PartitionGetPartitionName(part));
SetupPageCompressForRelation(&part->pd_node, &options->compress, PartitionGetPartitionName(part));
}
}

34
src/common/backend/utils/cache/relcache.cpp vendored
View File

@ -1300,7 +1300,6 @@ static OpClassCacheEnt* LookupOpclassInfo(Oid operatorClassOid, StrategyNumber n
static void RelationCacheInitFileRemoveInDir(const char* tblspcpath);
static void unlink_initfile(const char* initfilename);
static void SetBackendId(Relation relation);
static void SetupPageCompressForRelation(Relation relation, PageCompressOpts *compress_options);
/*
* ScanPgRelation
*
@ -2500,7 +2499,7 @@ static void RelationInitPhysicalAddr(Relation relation)
// setup page compression options
relation->rd_node.opt = 0;
if (relation->rd_options && REL_SUPPORT_COMPRESSED(relation)) {
SetupPageCompressForRelation(relation, &((StdRdOptions*)(relation->rd_options))->compress);
SetupPageCompressForRelation(&relation->rd_node, &((StdRdOptions*)(relation->rd_options))->compress, RelationGetRelationName(relation));
}
}
@ -4390,7 +4389,7 @@ Relation RelationBuildLocalRelation(const char* relname, Oid relnamespace, Tuple
/* compressed option was set by RelationInitPhysicalAddr if rel->rd_options != NULL */
if (rel->rd_options == NULL && reloptions && SUPPORT_COMPRESSED(relkind, rel->rd_rel->relam)) {
StdRdOptions *options = (StdRdOptions *) default_reloptions(reloptions, false, RELOPT_KIND_HEAP);
SetupPageCompressForRelation(rel, &options->compress);
SetupPageCompressForRelation(&rel->rd_node, &options->compress, RelationGetRelationName(rel));
}
@ -7879,14 +7878,14 @@ char RelationGetRelReplident(Relation r)
return relreplident;
}
/* setup page compress options for relation */
static void SetupPageCompressForRelation(Relation relation, PageCompressOpts* compress_options)
void SetupPageCompressForRelation(RelFileNode* node, PageCompressOpts* compress_options, const char* relationName)
{
relation->rd_node.opt = 0;
uint1 algorithm = compress_options->compressType;
if (algorithm != COMPRESS_TYPE_NONE) {
if (algorithm == COMPRESS_TYPE_NONE) {
node->opt = 0;
} else {
if (!SUPPORT_PAGE_COMPRESSION) {
elog(ERROR, "unsupported page compression on this platform");
ereport(ERROR, (errmsg("unsupported page compression on this platform")));
}
uint1 compressLevel;
@ -7902,20 +7901,21 @@ static void SetupPageCompressForRelation(Relation relation, PageCompressOpts* co
bool success = false;
uint1 chunkSize = ConvertChunkSize(compress_options->compressChunkSize, &success);
if (!success) {
elog(ERROR, "invalid compress_chunk_size %d , must be one of %d, %d, %d or %d for %s",
compress_options->compressChunkSize, BLCKSZ / 16, BLCKSZ / 8, BLCKSZ / 4, BLCKSZ / 2,
RelationGetRelationName(relation));
ereport(ERROR, (errmsg("invalid compress_chunk_size %d , must be one of %d, %d, %d or %d for %s",
compress_options->compressChunkSize, BLCKSZ / 16, BLCKSZ / 8, BLCKSZ / 4, BLCKSZ / 2,
relationName)));
}
uint1 preallocChunks;
if (compress_options->compressPreallocChunks >= BLCKSZ / compress_options->compressChunkSize) {
preallocChunks = (uint1)(BLCKSZ / compress_options->compressChunkSize - 1);
ereport(ERROR, (errmsg("invalid compress_prealloc_chunks %d , must be less than %d for %s",
compress_options->compressPreallocChunks,
BLCKSZ / compress_options->compressChunkSize, relationName)));
} else {
preallocChunks = (uint1)(compress_options->compressPreallocChunks);
}
Assert(preallocChunks <= MAX_PREALLOC_CHUNKS);
SET_COMPRESS_OPTION(relation->rd_node, compress_options->compressByteConvert,
compress_options->compressDiffConvert, preallocChunks,
symbol, compressLevel, algorithm, chunkSize);
node->opt = 0;
SET_COMPRESS_OPTION((*node), compress_options->compressByteConvert, compress_options->compressDiffConvert,
preallocChunks, symbol, compressLevel, algorithm, 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, "", "");

12
src/gausskernel/optimizer/commands/indexcmds.cpp
View File

@ -927,6 +927,18 @@ Oid DefineIndex(Oid relationId, IndexStmt* stmt, Oid indexRelationId, bool is_al
}
}
TableCreateSupport indexCreateSupport{false,false,false,false,false,false};
ListCell* cell = NULL;
foreach (cell, stmt->options) {
DefElem* defElem = (DefElem*)lfirst(cell);
SetOneOfCompressOption(defElem->defname, &indexCreateSupport);
}
if (!indexCreateSupport.compressType && HasCompressOption(&indexCreateSupport)) {
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.")));
}
/*
* Parse AM-specific options, convert to text array form, validate.
*/

43
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);
@ -1117,14 +1114,8 @@ static List* AddDefaultOptionsIfNeed(List* options, const char relkind, CreateSt
ereport(ERROR,
(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;
} else if (pg_strcasecmp(def->defname, "compress_chunk_size") == 0) {
hasRowCompressChunk = true;
} else if (pg_strcasecmp(def->defname, "compress_prealloc_chunks") == 0) {
hasRowCompressPre = true;
} else if (pg_strcasecmp(def->defname, "compress_level") == 0) {
hasRowCompressLevel = true;
} else {
SetOneOfCompressOption(def->defname, &tableCreateSupport);
}
if (pg_strcasecmp(def->defname, "orientation") == 0 && pg_strcasecmp(defGetString(def), ORIENTATION_ORC) == 0) {
@ -1150,23 +1141,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 +1189,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);
}

17
src/gausskernel/storage/access/common/reloptions.cpp
View File

@ -2897,3 +2897,20 @@ bool is_cstore_option(char relkind, Datum reloptions)
pfree_ext(std_opt);
return result;
}
void SetOneOfCompressOption(const char* defname, TableCreateSupport* tableCreateSupport)
{
if (pg_strcasecmp(defname, "compresstype") == 0) {
tableCreateSupport->compressType = true;
} else if (pg_strcasecmp(defname, "compress_chunk_size") == 0) {
tableCreateSupport->compressChunkSize = true;
} else if (pg_strcasecmp(defname, "compress_prealloc_chunks") == 0) {
tableCreateSupport->compressPreAllocChunks = true;
} else if (pg_strcasecmp(defname, "compress_level") == 0) {
tableCreateSupport->compressLevel = true;
} else if (pg_strcasecmp(defname, "compress_byte_convert") == 0) {
tableCreateSupport->compressByteConvert = true;
} else if (pg_strcasecmp(defname, "compress_diff_convert") == 0) {
tableCreateSupport->compressDiffConvert = true;
}
}

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);

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

@ -146,21 +146,21 @@ bool check_unlink_rel_hashtbl(RelFileNode rnode, ForkNumber forknum)
return found;
}
static int OpenPcaFile(const char *path, const RelFileNodeBackend &node, const ForkNumber &forkNum, const uint32 &segNo)
static int OpenPcaFile(const char *path, const RelFileNodeBackend &node, const ForkNumber &forkNum, const uint32 &segNo, int oflags = 0)
{
Assert(node.node.opt != 0 && forkNum == MAIN_FORKNUM);
char dst[MAXPGPATH];
CopyCompressedPath(dst, path, COMPRESSED_TABLE_PCA_FILE);
uint32 flags = O_RDWR | PG_BINARY;
uint32 flags = O_RDWR | PG_BINARY | oflags;
return DataFileIdOpenFile(dst, RelFileNodeForkNumFill(node, PCA_FORKNUM, segNo), (int)flags, S_IRUSR | S_IWUSR);
}
static int OpenPcdFile(const char *path, const RelFileNodeBackend &node, const ForkNumber &forkNum, const uint32 &segNo)
static int OpenPcdFile(const char *path, const RelFileNodeBackend &node, const ForkNumber &forkNum, const uint32 &segNo, int oflags = 0)
{
Assert(node.node.opt != 0 && forkNum == MAIN_FORKNUM);
char dst[MAXPGPATH];
CopyCompressedPath(dst, path, COMPRESSED_TABLE_PCD_FILE);
uint32 flags = O_RDWR | PG_BINARY;
uint32 flags = O_RDWR | PG_BINARY | oflags;
return DataFileIdOpenFile(dst, RelFileNodeForkNumFill(node, PCD_FORKNUM, segNo), (int)flags, S_IRUSR | S_IWUSR);
}
@ -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;
@ -2472,12 +2471,12 @@ static MdfdVec *_mdfd_openseg(SMgrRelation reln, ForkNumber forknum, BlockNumber
if (IS_COMPRESSED_MAINFORK(reln, forknum)) {
FileClose(fd);
fd = -1;
fd_pca = OpenPcaFile(fullpath, reln->smgr_rnode, MAIN_FORKNUM, segno);
fd_pca = OpenPcaFile(fullpath, reln->smgr_rnode, MAIN_FORKNUM, segno, oflags);
if (fd_pca < 0) {
pfree(fullpath);
return NULL;
}
fd_pcd = OpenPcdFile(fullpath, reln->smgr_rnode, MAIN_FORKNUM, segno);
fd_pcd = OpenPcdFile(fullpath, reln->smgr_rnode, MAIN_FORKNUM, segno, oflags);
if (fd_pcd < 0) {
pfree(fullpath);
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;

17
src/include/access/reloptions.h
View File

@ -130,6 +130,22 @@ typedef struct {
int offset; /* offset of field in result struct */
} relopt_parse_elt;
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;
}
/*
* The following are the table append modes currently supported.
* on: mark the table on-line scaleout mode, when it is set, later data write by append mode.
@ -284,5 +300,6 @@ extern void forbid_to_set_options_for_timeseries_tbl(List* options);
extern List* RemoveRelOption(List* options, const char* optName, bool* removed);
void RowTblCheckCompressionOption(List *options);
void RowTblCheckHashBucketOption(List* options, StdRdOptions* std_opt);
void SetOneOfCompressOption(const char *defname, TableCreateSupport *tableCreateSupport);
#endif /* RELOPTIONS_H */

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

@ -113,7 +113,6 @@ typedef struct HeapPageCompressData {
char data[FLEXIBLE_ARRAY_MEMBER]; /* compressed page, except for the page header */
} HeapPageCompressData;
const uint4 CHUNK_SIZE_LIST[4] = {BLCKSZ / 2, BLCKSZ / 4, BLCKSZ / 8, BLCKSZ / 16};
constexpr uint4 INDEX_OF_HALF_BLCKSZ = 0;
constexpr uint4 INDEX_OF_QUARTER_BLCKSZ = 1;
@ -167,13 +166,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 +322,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);

2
src/include/utils/rel.h
View File

@ -804,6 +804,6 @@ extern void RelationDecrementReferenceCount(Oid relationId);
extern void GetTdeInfoFromRel(Relation rel, TdeInfo *tde_info);
extern char RelationGetRelReplident(Relation r);
extern void SetupPageCompressForRelation(RelFileNode* node, PageCompressOpts* compressOpts, const char* name);
#endif /* REL_H */

6
src/test/regress/expected/row_compression/normal_test.out
View File

@ -152,8 +152,12 @@ alter index normal_test.tbl_partition_id_idx set (compress_chunk_size=2048);
ERROR: change compress_chunk_size OPTION is not supported
alter index normal_test.tbl_partition_id_idx set (compress_prealloc_chunks=2);
ERROR: change partition compress_prealloc_chunks OPTION is not supported
create index rolcompress_index on normal_test.tbl_pc(id) with (compress_chunk_size=4096);
ERROR: compress_chunk_size/compress_prealloc_chunks/compress_level/compress_byte_convert/compress_diff_convert should be used with compresstype.
create table rolcompress_table_001(a int) with (compresstype=2, compress_prealloc_chunks=3);
ERROR: invalid compress_prealloc_chunks 3 , must be less than 2 for rolcompress_table_001
-- support
alter table normal_test.tbl_pc set (compress_prealloc_chunks=2);
alter table normal_test.tbl_pc set (compress_prealloc_chunks=1);
drop schema normal_test cascade;
NOTICE: drop cascades to 3 other objects
DETAIL: drop cascades to table normal_test.tbl_pc

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)

4
src/test/regress/sql/row_compression/normal_test.sql
View File

@ -55,7 +55,9 @@ select relname, reloptions from pg_partition where parentid in (Select relfileno
alter index normal_test.tbl_partition_id_idx set (compresstype=1);
alter index normal_test.tbl_partition_id_idx set (compress_chunk_size=2048);
alter index normal_test.tbl_partition_id_idx set (compress_prealloc_chunks=2);
create index rolcompress_index on normal_test.tbl_pc(id) with (compress_chunk_size=4096);
create table rolcompress_table_001(a int) with (compresstype=2, compress_prealloc_chunks=3);
-- support
alter table normal_test.tbl_pc set (compress_prealloc_chunks=2);
alter table normal_test.tbl_pc set (compress_prealloc_chunks=1);
drop schema normal_test cascade;

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);