external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix style

pull/4311/head
Nils Dijk 2020-11-05 16:38:57 +01:00
parent 3bb6554976
commit b6d4a1bbe2
No known key found for this signature in database
GPG Key ID: CA1177EF9434F241
12 changed files with 362 additions and 554 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/backend/columnar/cstore_customscan.c
View File

@ -136,8 +136,6 @@ static void
CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti, CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
RangeTblEntry *rte) RangeTblEntry *rte)
{ {
Relation relation;
/* call into previous hook if assigned */ /* call into previous hook if assigned */
if (PreviousSetRelPathlistHook) if (PreviousSetRelPathlistHook)
{ {
@ -161,7 +159,7 @@ CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
* If that is the case we want to insert an extra path that pushes down the projection * If that is the case we want to insert an extra path that pushes down the projection
* into the scan of the table to minimize the data read. * into the scan of the table to minimize the data read.
*/ */
relation = RelationIdGetRelation(rte->relid); Relation relation = RelationIdGetRelation(rte->relid);
if (relation->rd_tableam == GetCstoreTableAmRoutine()) if (relation->rd_tableam == GetCstoreTableAmRoutine())
{ {
Path *customPath = CreateCStoreScanPath(rel, rte); Path *customPath = CreateCStoreScanPath(rel, rte);
@ -181,19 +179,17 @@ CreateCStoreScanPath(RelOptInfo *rel, RangeTblEntry *rte)
{ {
CStoreScanPath *cspath = (CStoreScanPath *) newNode(sizeof(CStoreScanPath), CStoreScanPath *cspath = (CStoreScanPath *) newNode(sizeof(CStoreScanPath),
T_CustomPath); T_CustomPath);
CustomPath *cpath;
Path *path;
/* /*
* popuate custom path information * popuate custom path information
*/ */
cpath = &cspath->custom_path; CustomPath *cpath = &cspath->custom_path;
cpath->methods = &CStoreScanPathMethods; cpath->methods = &CStoreScanPathMethods;
/* /*
* populate generic path information * populate generic path information
*/ */
path = &cpath->path; Path *path = &cpath->path;
path->pathtype = T_CustomScan; path->pathtype = T_CustomScan;
path->parent = rel; path->parent = rel;
path->pathtarget = rel->reltarget; path->pathtarget = rel->reltarget;
@ -329,18 +325,14 @@ static TupleTableSlot *
CStoreScanNext(CStoreScanState *cstorescanstate) CStoreScanNext(CStoreScanState *cstorescanstate)
{ {
CustomScanState *node = (CustomScanState *) cstorescanstate; CustomScanState *node = (CustomScanState *) cstorescanstate;
TableScanDesc scandesc;
EState *estate;
ScanDirection direction;
TupleTableSlot *slot;
/* /*
* get information from the estate and scan state * get information from the estate and scan state
*/ */
scandesc = node->ss.ss_currentScanDesc; TableScanDesc scandesc = node->ss.ss_currentScanDesc;
estate = node->ss.ps.state; EState *estate = node->ss.ps.state;
direction = estate->es_direction; ScanDirection direction = estate->es_direction;
slot = node->ss.ss_ScanTupleSlot; TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
if (scandesc == NULL) if (scandesc == NULL)
{ {
@ -394,12 +386,10 @@ CStoreScan_ExecCustomScan(CustomScanState *node)
static void static void
CStoreScan_EndCustomScan(CustomScanState *node) CStoreScan_EndCustomScan(CustomScanState *node)
{ {
TableScanDesc scanDesc;
/* /*
* get information from node * get information from node
*/ */
scanDesc = node->ss.ss_currentScanDesc; TableScanDesc scanDesc = node->ss.ss_currentScanDesc;
/* /*
* Free the exprcontext * Free the exprcontext

181
src/backend/columnar/cstore_fdw.c
View File

@ -255,17 +255,14 @@ cstore_fdw_finish()
Datum Datum
cstore_ddl_event_end_trigger(PG_FUNCTION_ARGS) cstore_ddl_event_end_trigger(PG_FUNCTION_ARGS)
{ {
EventTriggerData *triggerData = NULL;
Node *parseTree = NULL;
/* error if event trigger manager did not call this function */ /* error if event trigger manager did not call this function */
if (!CALLED_AS_EVENT_TRIGGER(fcinfo)) if (!CALLED_AS_EVENT_TRIGGER(fcinfo))
{ {
ereport(ERROR, (errmsg("trigger not fired by event trigger manager"))); ereport(ERROR, (errmsg("trigger not fired by event trigger manager")));
} }
triggerData = (EventTriggerData *) fcinfo->context; EventTriggerData *triggerData = (EventTriggerData *) fcinfo->context;
parseTree = triggerData->parsetree; Node *parseTree = triggerData->parsetree;
if (nodeTag(parseTree) == T_CreateForeignTableStmt) if (nodeTag(parseTree) == T_CreateForeignTableStmt)
{ {
@ -495,16 +492,9 @@ CopyIntoCStoreTable(const CopyStmt *copyStatement, const char *queryString)
{ {
uint64 processedRowCount = 0; uint64 processedRowCount = 0;
Relation relation = NULL; Relation relation = NULL;
Oid relationId = InvalidOid;
TupleDesc tupleDescriptor = NULL;
uint32 columnCount = 0;
CopyState copyState = NULL; CopyState copyState = NULL;
bool nextRowFound = true; bool nextRowFound = true;
Datum *columnValues = NULL;
bool *columnNulls = NULL;
TableWriteState *writeState = NULL; TableWriteState *writeState = NULL;
CStoreOptions *cstoreOptions = NULL;
MemoryContext tupleContext = NULL;
/* Only superuser can copy from or to local file */ /* Only superuser can copy from or to local file */
CheckSuperuserPrivilegesForCopy(copyStatement); CheckSuperuserPrivilegesForCopy(copyStatement);
@ -516,15 +506,15 @@ CopyIntoCStoreTable(const CopyStmt *copyStatement, const char *queryString)
* concurrent reads and writes. * concurrent reads and writes.
*/ */
relation = cstore_fdw_openrv(copyStatement->relation, RowExclusiveLock); relation = cstore_fdw_openrv(copyStatement->relation, RowExclusiveLock);
relationId = RelationGetRelid(relation); Oid relationId = RelationGetRelid(relation);
/* allocate column values and nulls arrays */ /* allocate column values and nulls arrays */
tupleDescriptor = RelationGetDescr(relation); TupleDesc tupleDescriptor = RelationGetDescr(relation);
columnCount = tupleDescriptor->natts; uint32 columnCount = tupleDescriptor->natts;
columnValues = palloc0(columnCount * sizeof(Datum)); Datum *columnValues = palloc0(columnCount * sizeof(Datum));
columnNulls = palloc0(columnCount * sizeof(bool)); bool *columnNulls = palloc0(columnCount * sizeof(bool));
cstoreOptions = CStoreGetOptions(relationId); CStoreOptions *cstoreOptions = CStoreGetOptions(relationId);
/* /*
* We create a new memory context called tuple context, and read and write * We create a new memory context called tuple context, and read and write
@ -533,7 +523,7 @@ CopyIntoCStoreTable(const CopyStmt *copyStatement, const char *queryString)
* allocated for each row, and don't bloat memory usage with large input * allocated for each row, and don't bloat memory usage with large input
* files. * files.
*/ */
tupleContext = AllocSetContextCreate(CurrentMemoryContext, MemoryContext tupleContext = AllocSetContextCreate(CurrentMemoryContext,
"CStore COPY Row Memory Context", "CStore COPY Row Memory Context",
ALLOCSET_DEFAULT_SIZES); ALLOCSET_DEFAULT_SIZES);
@ -606,10 +596,6 @@ static uint64
CopyOutCStoreTable(CopyStmt *copyStatement, const char *queryString) CopyOutCStoreTable(CopyStmt *copyStatement, const char *queryString)
{ {
uint64 processedCount = 0; uint64 processedCount = 0;
RangeVar *relation = NULL;
char *qualifiedName = NULL;
List *queryList = NIL;
Node *rawQuery = NULL;
StringInfo newQuerySubstring = makeStringInfo(); StringInfo newQuerySubstring = makeStringInfo();
@ -621,14 +607,14 @@ CopyOutCStoreTable(CopyStmt *copyStatement, const char *queryString)
"...' instead"))); "...' instead")));
} }
relation = copyStatement->relation; RangeVar *relation = copyStatement->relation;
qualifiedName = quote_qualified_identifier(relation->schemaname, char *qualifiedName = quote_qualified_identifier(relation->schemaname,
relation->relname); relation->relname);
appendStringInfo(newQuerySubstring, "select * from %s", qualifiedName); appendStringInfo(newQuerySubstring, "select * from %s", qualifiedName);
queryList = raw_parser(newQuerySubstring->data); List *queryList = raw_parser(newQuerySubstring->data);
/* take the first parse tree */ /* take the first parse tree */
rawQuery = linitial(queryList); Node *rawQuery = linitial(queryList);
/* /*
* Set the relation field to NULL so that COPY command works on * Set the relation field to NULL so that COPY command works on
@ -674,7 +660,6 @@ CStoreProcessAlterTableCommand(AlterTableStmt *alterStatement)
{ {
ObjectType objectType = alterStatement->relkind; ObjectType objectType = alterStatement->relkind;
RangeVar *relationRangeVar = alterStatement->relation; RangeVar *relationRangeVar = alterStatement->relation;
Oid relationId = InvalidOid;
List *commandList = alterStatement->cmds; List *commandList = alterStatement->cmds;
ListCell *commandCell = NULL; ListCell *commandCell = NULL;
@ -684,7 +669,7 @@ CStoreProcessAlterTableCommand(AlterTableStmt *alterStatement)
return; return;
} }
relationId = RangeVarGetRelid(relationRangeVar, AccessShareLock, true); Oid relationId = RangeVarGetRelid(relationRangeVar, AccessShareLock, true);
if (!IsCStoreFdwTable(relationId)) if (!IsCStoreFdwTable(relationId))
{ {
return; return;
@ -700,7 +685,6 @@ CStoreProcessAlterTableCommand(AlterTableStmt *alterStatement)
Oid targetTypeId = typenameTypeId(NULL, columnDef->typeName); Oid targetTypeId = typenameTypeId(NULL, columnDef->typeName);
char *typeName = TypeNameToString(columnDef->typeName); char *typeName = TypeNameToString(columnDef->typeName);
AttrNumber attributeNumber = get_attnum(relationId, columnName); AttrNumber attributeNumber = get_attnum(relationId, columnName);
Oid currentTypeId = InvalidOid;
if (attributeNumber <= 0) if (attributeNumber <= 0)
{ {
@ -708,7 +692,7 @@ CStoreProcessAlterTableCommand(AlterTableStmt *alterStatement)
continue; continue;
} }
currentTypeId = get_atttype(relationId, attributeNumber); Oid currentTypeId = get_atttype(relationId, attributeNumber);
/* /*
* We are only interested in implicit coersion type compatibility. * We are only interested in implicit coersion type compatibility.
@ -811,34 +795,28 @@ TruncateCStoreTables(List *cstoreRelationList)
static void static void
FdwNewRelFileNode(Relation relation) FdwNewRelFileNode(Relation relation)
{ {
Relation pg_class; Relation pg_class = heap_open(RelationRelationId, RowExclusiveLock);
HeapTuple tuple;
Form_pg_class classform;
pg_class = heap_open(RelationRelationId, RowExclusiveLock); HeapTuple tuple = SearchSysCacheCopy1(RELOID,
tuple = SearchSysCacheCopy1(RELOID,
ObjectIdGetDatum(RelationGetRelid(relation))); ObjectIdGetDatum(RelationGetRelid(relation)));
if (!HeapTupleIsValid(tuple)) if (!HeapTupleIsValid(tuple))
{ {
elog(ERROR, "could not find tuple for relation %u", elog(ERROR, "could not find tuple for relation %u",
RelationGetRelid(relation)); RelationGetRelid(relation));
} }
classform = (Form_pg_class) GETSTRUCT(tuple); Form_pg_class classform = (Form_pg_class) GETSTRUCT(tuple);
if (true) if (true)
{ {
char persistence = relation->rd_rel->relpersistence; char persistence = relation->rd_rel->relpersistence;
Relation tmprel;
Oid tablespace; Oid tablespace;
Oid filenode;
/* /*
* Upgrade to AccessExclusiveLock, and hold until the end of the * Upgrade to AccessExclusiveLock, and hold until the end of the
* transaction. This shouldn't happen during a read, but it's hard to * transaction. This shouldn't happen during a read, but it's hard to
* prove that because it happens lazily. * prove that because it happens lazily.
*/ */
tmprel = heap_open(relation->rd_id, AccessExclusiveLock); Relation tmprel = heap_open(relation->rd_id, AccessExclusiveLock);
heap_close(tmprel, NoLock); heap_close(tmprel, NoLock);
if (OidIsValid(relation->rd_rel->relfilenode)) if (OidIsValid(relation->rd_rel->relfilenode))
@ -856,7 +834,7 @@ FdwNewRelFileNode(Relation relation)
tablespace = MyDatabaseTableSpace; tablespace = MyDatabaseTableSpace;
} }
filenode = GetNewRelFileNode(tablespace, NULL, persistence); Oid filenode = GetNewRelFileNode(tablespace, NULL, persistence);
classform->relfilenode = filenode; classform->relfilenode = filenode;
classform->relpages = 0; /* it's empty until further notice */ classform->relpages = 0; /* it's empty until further notice */
@ -886,8 +864,7 @@ FdwCreateStorage(Relation relation)
if (!smgrexists(relation->rd_smgr, MAIN_FORKNUM)) if (!smgrexists(relation->rd_smgr, MAIN_FORKNUM))
{ {
#if PG_VERSION_NUM >= 120000 #if PG_VERSION_NUM >= 120000
SMgrRelation srel; SMgrRelation srel = RelationCreateStorage(relation->rd_node,
srel = RelationCreateStorage(relation->rd_node,
relation->rd_rel->relpersistence); relation->rd_rel->relpersistence);
smgrclose(srel); smgrclose(srel);
#else #else
@ -906,14 +883,13 @@ bool
IsCStoreFdwTable(Oid relationId) IsCStoreFdwTable(Oid relationId)
{ {
bool cstoreTable = false; bool cstoreTable = false;
char relationKind = 0;
if (relationId == InvalidOid) if (relationId == InvalidOid)
{ {
return false; return false;
} }
relationKind = get_rel_relkind(relationId); char relationKind = get_rel_relkind(relationId);
if (relationKind == RELKIND_FOREIGN_TABLE) if (relationKind == RELKIND_FOREIGN_TABLE)
{ {
ForeignTable *foreignTable = GetForeignTable(relationId); ForeignTable *foreignTable = GetForeignTable(relationId);
@ -956,13 +932,8 @@ IsCStoreServer(ForeignServer *server)
static bool static bool
DistributedTable(Oid relationId) DistributedTable(Oid relationId)
{ {
bool distributedTable = false;
Oid partitionOid = InvalidOid;
Relation heapRelation = NULL;
TableScanDesc scanDesc = NULL;
const int scanKeyCount = 1; const int scanKeyCount = 1;
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
HeapTuple heapTuple = NULL;
bool missingOK = true; bool missingOK = true;
Oid extensionOid = get_extension_oid(CITUS_EXTENSION_NAME, missingOK); Oid extensionOid = get_extension_oid(CITUS_EXTENSION_NAME, missingOK);
@ -972,23 +943,25 @@ DistributedTable(Oid relationId)
return false; return false;
} }
partitionOid = get_relname_relid(CITUS_PARTITION_TABLE_NAME, PG_CATALOG_NAMESPACE); Oid partitionOid = get_relname_relid(CITUS_PARTITION_TABLE_NAME,
PG_CATALOG_NAMESPACE);
if (partitionOid == InvalidOid) if (partitionOid == InvalidOid)
{ {
/* the pg_dist_partition table does not exist */ /* the pg_dist_partition table does not exist */
return false; return false;
} }
heapRelation = heap_open(partitionOid, AccessShareLock); Relation heapRelation = heap_open(partitionOid, AccessShareLock);
ScanKeyInit(&scanKey[0], ATTR_NUM_PARTITION_RELATION_ID, InvalidStrategy, ScanKeyInit(&scanKey[0], ATTR_NUM_PARTITION_RELATION_ID, InvalidStrategy,
F_OIDEQ, ObjectIdGetDatum(relationId)); F_OIDEQ, ObjectIdGetDatum(relationId));
scanDesc = table_beginscan(heapRelation, SnapshotSelf, scanKeyCount, scanKey); TableScanDesc scanDesc = table_beginscan(heapRelation, SnapshotSelf, scanKeyCount,
scanKey);
heapTuple = heap_getnext(scanDesc, ForwardScanDirection); HeapTuple heapTuple = heap_getnext(scanDesc, ForwardScanDirection);
distributedTable = HeapTupleIsValid(heapTuple); bool distributedTable = HeapTupleIsValid(heapTuple);
table_endscan(scanDesc); table_endscan(scanDesc);
relation_close(heapRelation, AccessShareLock); relation_close(heapRelation, AccessShareLock);
@ -1027,17 +1000,15 @@ cstore_table_size(PG_FUNCTION_ARGS)
{ {
Oid relationId = PG_GETARG_OID(0); Oid relationId = PG_GETARG_OID(0);
bool cstoreTable = IsCStoreFdwTable(relationId); bool cstoreTable = IsCStoreFdwTable(relationId);
Relation relation;
BlockNumber nblocks;
if (!cstoreTable) if (!cstoreTable)
{ {
ereport(ERROR, (errmsg("relation is not a cstore table"))); ereport(ERROR, (errmsg("relation is not a cstore table")));
} }
relation = cstore_fdw_open(relationId, AccessShareLock); Relation relation = cstore_fdw_open(relationId, AccessShareLock);
RelationOpenSmgr(relation); RelationOpenSmgr(relation);
nblocks = smgrnblocks(relation->rd_smgr, MAIN_FORKNUM); BlockNumber nblocks = smgrnblocks(relation->rd_smgr, MAIN_FORKNUM);
heap_close(relation, AccessShareLock); heap_close(relation, AccessShareLock);
PG_RETURN_INT64(nblocks * BLCKSZ); PG_RETURN_INT64(nblocks * BLCKSZ);
} }
@ -1205,19 +1176,15 @@ GetSlotHeapTuple(TupleTableSlot *tts)
static CStoreOptions * static CStoreOptions *
CStoreGetOptions(Oid foreignTableId) CStoreGetOptions(Oid foreignTableId)
{ {
CStoreOptions *cstoreOptions = NULL;
CompressionType compressionType = cstore_compression; CompressionType compressionType = cstore_compression;
int32 stripeRowCount = cstore_stripe_row_count; int32 stripeRowCount = cstore_stripe_row_count;
int32 blockRowCount = cstore_block_row_count; int32 blockRowCount = cstore_block_row_count;
char *compressionTypeString = NULL;
char *stripeRowCountString = NULL;
char *blockRowCountString = NULL;
compressionTypeString = CStoreGetOptionValue(foreignTableId, char *compressionTypeString = CStoreGetOptionValue(foreignTableId,
OPTION_NAME_COMPRESSION_TYPE); OPTION_NAME_COMPRESSION_TYPE);
stripeRowCountString = CStoreGetOptionValue(foreignTableId, char *stripeRowCountString = CStoreGetOptionValue(foreignTableId,
OPTION_NAME_STRIPE_ROW_COUNT); OPTION_NAME_STRIPE_ROW_COUNT);
blockRowCountString = CStoreGetOptionValue(foreignTableId, char *blockRowCountString = CStoreGetOptionValue(foreignTableId,
OPTION_NAME_BLOCK_ROW_COUNT); OPTION_NAME_BLOCK_ROW_COUNT);
ValidateForeignTableOptions(compressionTypeString, ValidateForeignTableOptions(compressionTypeString,
@ -1237,7 +1204,7 @@ CStoreGetOptions(Oid foreignTableId)
blockRowCount = pg_atoi(blockRowCountString, sizeof(int32), 0); blockRowCount = pg_atoi(blockRowCountString, sizeof(int32), 0);
} }
cstoreOptions = palloc0(sizeof(CStoreOptions)); CStoreOptions *cstoreOptions = palloc0(sizeof(CStoreOptions));
cstoreOptions->compressionType = compressionType; cstoreOptions->compressionType = compressionType;
cstoreOptions->stripeRowCount = stripeRowCount; cstoreOptions->stripeRowCount = stripeRowCount;
cstoreOptions->blockRowCount = blockRowCount; cstoreOptions->blockRowCount = blockRowCount;
@ -1254,14 +1221,12 @@ CStoreGetOptions(Oid foreignTableId)
static char * static char *
CStoreGetOptionValue(Oid foreignTableId, const char *optionName) CStoreGetOptionValue(Oid foreignTableId, const char *optionName)
{ {
ForeignTable *foreignTable = NULL;
ForeignServer *foreignServer = NULL;
List *optionList = NIL; List *optionList = NIL;
ListCell *optionCell = NULL; ListCell *optionCell = NULL;
char *optionValue = NULL; char *optionValue = NULL;
foreignTable = GetForeignTable(foreignTableId); ForeignTable *foreignTable = GetForeignTable(foreignTableId);
foreignServer = GetForeignServer(foreignTable->serverid); ForeignServer *foreignServer = GetForeignServer(foreignTable->serverid);
optionList = list_concat(optionList, foreignTable->options); optionList = list_concat(optionList, foreignTable->options);
optionList = list_concat(optionList, foreignServer->options); optionList = list_concat(optionList, foreignServer->options);
@ -1451,8 +1416,6 @@ CStoreGetForeignPlan(PlannerInfo * root, RelOptInfo * baserel, Oid foreignTableI
#endif #endif
{ {
ForeignScan *foreignScan = NULL; ForeignScan *foreignScan = NULL;
List *columnList = NIL;
List *foreignPrivateList = NIL;
/* /*
* Although we skip row blocks that are refuted by the WHERE clause, but * Although we skip row blocks that are refuted by the WHERE clause, but
@ -1469,8 +1432,8 @@ CStoreGetForeignPlan(PlannerInfo * root, RelOptInfo * baserel, Oid foreignTableI
* in executor's callback functions, so we get the column list here and put * in executor's callback functions, so we get the column list here and put
* it into foreign scan node's private list. * it into foreign scan node's private list.
*/ */
columnList = ColumnList(baserel, foreignTableId); List *columnList = ColumnList(baserel, foreignTableId);
foreignPrivateList = list_make1(columnList); List *foreignPrivateList = list_make1(columnList);
/* create the foreign scan node */ /* create the foreign scan node */
#if PG_VERSION_NUM >= 90500 #if PG_VERSION_NUM >= 90500
@ -1525,10 +1488,8 @@ TupleCountEstimate(Relation relation, RelOptInfo *baserel)
static BlockNumber static BlockNumber
PageCount(Relation relation) PageCount(Relation relation)
{ {
BlockNumber nblocks;
RelationOpenSmgr(relation); RelationOpenSmgr(relation);
nblocks = smgrnblocks(relation->rd_smgr, MAIN_FORKNUM); BlockNumber nblocks = smgrnblocks(relation->rd_smgr, MAIN_FORKNUM);
return (nblocks > 0) ? nblocks : 1; return (nblocks > 0) ? nblocks : 1;
} }
@ -1655,9 +1616,8 @@ CStoreExplainForeignScan(ForeignScanState *scanState, ExplainState *explainState
/* supress file size if we're not showing cost details */ /* supress file size if we're not showing cost details */
if (explainState->costs) if (explainState->costs)
{ {
long nblocks;
RelationOpenSmgr(relation); RelationOpenSmgr(relation);
nblocks = smgrnblocks(relation->rd_smgr, MAIN_FORKNUM); long nblocks = smgrnblocks(relation->rd_smgr, MAIN_FORKNUM);
ExplainPropertyLong("CStore File Size", (long) (nblocks * BLCKSZ), ExplainPropertyLong("CStore File Size", (long) (nblocks * BLCKSZ),
explainState); explainState);
} }
@ -1668,15 +1628,8 @@ CStoreExplainForeignScan(ForeignScanState *scanState, ExplainState *explainState
static void static void
CStoreBeginForeignScan(ForeignScanState *scanState, int executorFlags) CStoreBeginForeignScan(ForeignScanState *scanState, int executorFlags)
{ {
TableReadState *readState = NULL;
Oid foreignTableId = InvalidOid;
Relation currentRelation = scanState->ss.ss_currentRelation; Relation currentRelation = scanState->ss.ss_currentRelation;
TupleDesc tupleDescriptor = RelationGetDescr(currentRelation); TupleDesc tupleDescriptor = RelationGetDescr(currentRelation);
List *columnList = NIL;
ForeignScan *foreignScan = NULL;
List *foreignPrivateList = NIL;
List *whereClauseList = NIL;
Relation relation = NULL;
cstore_fdw_initrel(currentRelation); cstore_fdw_initrel(currentRelation);
@ -1686,15 +1639,16 @@ CStoreBeginForeignScan(ForeignScanState *scanState, int executorFlags)
return; return;
} }
foreignTableId = RelationGetRelid(scanState->ss.ss_currentRelation); Oid foreignTableId = RelationGetRelid(scanState->ss.ss_currentRelation);
foreignScan = (ForeignScan *) scanState->ss.ps.plan; ForeignScan *foreignScan = (ForeignScan *) scanState->ss.ps.plan;
foreignPrivateList = (List *) foreignScan->fdw_private; List *foreignPrivateList = (List *) foreignScan->fdw_private;
whereClauseList = foreignScan->scan.plan.qual; List *whereClauseList = foreignScan->scan.plan.qual;
columnList = (List *) linitial(foreignPrivateList); List *columnList = (List *) linitial(foreignPrivateList);
relation = cstore_fdw_open(foreignTableId, AccessShareLock); Relation relation = cstore_fdw_open(foreignTableId, AccessShareLock);
readState = CStoreBeginRead(relation, tupleDescriptor, columnList, whereClauseList); TableReadState *readState = CStoreBeginRead(relation, tupleDescriptor, columnList,
whereClauseList);
scanState->fdw_state = (void *) readState; scanState->fdw_state = (void *) readState;
} }
@ -1710,7 +1664,6 @@ CStoreIterateForeignScan(ForeignScanState *scanState)
{ {
TableReadState *readState = (TableReadState *) scanState->fdw_state; TableReadState *readState = (TableReadState *) scanState->fdw_state;
TupleTableSlot *tupleSlot = scanState->ss.ss_ScanTupleSlot; TupleTableSlot *tupleSlot = scanState->ss.ss_ScanTupleSlot;
bool nextRowFound = false;
TupleDesc tupleDescriptor = tupleSlot->tts_tupleDescriptor; TupleDesc tupleDescriptor = tupleSlot->tts_tupleDescriptor;
Datum *columnValues = tupleSlot->tts_values; Datum *columnValues = tupleSlot->tts_values;
@ -1723,7 +1676,7 @@ CStoreIterateForeignScan(ForeignScanState *scanState)
ExecClearTuple(tupleSlot); ExecClearTuple(tupleSlot);
nextRowFound = CStoreReadNextRow(readState, columnValues, columnNulls); bool nextRowFound = CStoreReadNextRow(readState, columnValues, columnNulls);
if (nextRowFound) if (nextRowFound)
{ {
ExecStoreVirtualTuple(tupleSlot); ExecStoreVirtualTuple(tupleSlot);
@ -1797,13 +1750,9 @@ CStoreAcquireSampleRows(Relation relation, int logLevel,
double selectionState = 0; double selectionState = 0;
MemoryContext oldContext = CurrentMemoryContext; MemoryContext oldContext = CurrentMemoryContext;
MemoryContext tupleContext = NULL; MemoryContext tupleContext = NULL;
Datum *columnValues = NULL;
bool *columnNulls = NULL;
TupleTableSlot *scanTupleSlot = NULL; TupleTableSlot *scanTupleSlot = NULL;
List *columnList = NIL; List *columnList = NIL;
List *foreignPrivateList = NULL;
ForeignScanState *scanState = NULL; ForeignScanState *scanState = NULL;
ForeignScan *foreignScan = NULL;
char *relationName = NULL; char *relationName = NULL;
int executorFlags = 0; int executorFlags = 0;
uint32 columnIndex = 0; uint32 columnIndex = 0;
@ -1829,13 +1778,13 @@ CStoreAcquireSampleRows(Relation relation, int logLevel,
} }
/* setup foreign scan plan node */ /* setup foreign scan plan node */
foreignPrivateList = list_make1(columnList); List *foreignPrivateList = list_make1(columnList);
foreignScan = makeNode(ForeignScan); ForeignScan *foreignScan = makeNode(ForeignScan);
foreignScan->fdw_private = foreignPrivateList; foreignScan->fdw_private = foreignPrivateList;
/* set up tuple slot */ /* set up tuple slot */
columnValues = palloc0(columnCount * sizeof(Datum)); Datum *columnValues = palloc0(columnCount * sizeof(Datum));
columnNulls = palloc0(columnCount * sizeof(bool)); bool *columnNulls = palloc0(columnCount * sizeof(bool));
#if PG_VERSION_NUM >= 120000 #if PG_VERSION_NUM >= 120000
scanTupleSlot = MakeTupleTableSlot(NULL, &TTSOpsVirtual); scanTupleSlot = MakeTupleTableSlot(NULL, &TTSOpsVirtual);
#elif PG_VERSION_NUM >= 110000 #elif PG_VERSION_NUM >= 110000
@ -1968,13 +1917,12 @@ CStorePlanForeignModify(PlannerInfo *plannerInfo, ModifyTable *plan,
if (plan->operation == CMD_INSERT) if (plan->operation == CMD_INSERT)
{ {
ListCell *tableCell = NULL; ListCell *tableCell = NULL;
Query *query = NULL;
/* /*
* Only insert operation with select subquery is supported. Other forms * Only insert operation with select subquery is supported. Other forms
* of insert, update, and delete operations are not supported. * of insert, update, and delete operations are not supported.
*/ */
query = plannerInfo->parse; Query *query = plannerInfo->parse;
foreach(tableCell, query->rtable) foreach(tableCell, query->rtable)
{ {
RangeTblEntry *tableEntry = lfirst(tableCell); RangeTblEntry *tableEntry = lfirst(tableCell);
@ -2027,18 +1975,12 @@ CStoreBeginForeignModify(ModifyTableState *modifyTableState,
static void static void
CStoreBeginForeignInsert(ModifyTableState *modifyTableState, ResultRelInfo *relationInfo) CStoreBeginForeignInsert(ModifyTableState *modifyTableState, ResultRelInfo *relationInfo)
{ {
Oid foreignTableOid = InvalidOid; Oid foreignTableOid = RelationGetRelid(relationInfo->ri_RelationDesc);
CStoreOptions *cstoreOptions = NULL; Relation relation = cstore_fdw_open(foreignTableOid, RowExclusiveLock);
TupleDesc tupleDescriptor = NULL; CStoreOptions *cstoreOptions = CStoreGetOptions(foreignTableOid);
TableWriteState *writeState = NULL; TupleDesc tupleDescriptor = RelationGetDescr(relationInfo->ri_RelationDesc);
Relation relation = NULL;
foreignTableOid = RelationGetRelid(relationInfo->ri_RelationDesc); TableWriteState *writeState = CStoreBeginWrite(relation,
relation = cstore_fdw_open(foreignTableOid, RowExclusiveLock);
cstoreOptions = CStoreGetOptions(foreignTableOid);
tupleDescriptor = RelationGetDescr(relationInfo->ri_RelationDesc);
writeState = CStoreBeginWrite(relation,
cstoreOptions->compressionType, cstoreOptions->compressionType,
cstoreOptions->stripeRowCount, cstoreOptions->stripeRowCount,
cstoreOptions->blockRowCount, cstoreOptions->blockRowCount,
@ -2057,11 +1999,10 @@ CStoreExecForeignInsert(EState *executorState, ResultRelInfo *relationInfo,
TupleTableSlot *tupleSlot, TupleTableSlot *planSlot) TupleTableSlot *tupleSlot, TupleTableSlot *planSlot)
{ {
TableWriteState *writeState = (TableWriteState *) relationInfo->ri_FdwState; TableWriteState *writeState = (TableWriteState *) relationInfo->ri_FdwState;
HeapTuple heapTuple;
Assert(writeState != NULL); Assert(writeState != NULL);
heapTuple = GetSlotHeapTuple(tupleSlot); HeapTuple heapTuple = GetSlotHeapTuple(tupleSlot);
if (HeapTupleHasExternal(heapTuple)) if (HeapTupleHasExternal(heapTuple))
{ {

145
src/backend/columnar/cstore_metadata_tables.c
View File

@ -130,9 +130,6 @@ void
InitCStoreDataFileMetadata(Oid relfilenode, int blockRowCount, int stripeRowCount, InitCStoreDataFileMetadata(Oid relfilenode, int blockRowCount, int stripeRowCount,
CompressionType compression) CompressionType compression)
{ {
Oid cstoreDataFilesOid = InvalidOid;
Relation cstoreDataFiles = NULL;
ModifyState *modifyState = NULL;
NameData compressionName = { 0 }; NameData compressionName = { 0 };
namestrcpy(&compressionName, CompressionTypeStr(compression)); namestrcpy(&compressionName, CompressionTypeStr(compression));
@ -149,10 +146,10 @@ InitCStoreDataFileMetadata(Oid relfilenode, int blockRowCount, int stripeRowCoun
DeleteDataFileMetadataRowIfExists(relfilenode); DeleteDataFileMetadataRowIfExists(relfilenode);
cstoreDataFilesOid = CStoreDataFilesRelationId(); Oid cstoreDataFilesOid = CStoreDataFilesRelationId();
cstoreDataFiles = heap_open(cstoreDataFilesOid, RowExclusiveLock); Relation cstoreDataFiles = heap_open(cstoreDataFilesOid, RowExclusiveLock);
modifyState = StartModifyRelation(cstoreDataFiles); ModifyState *modifyState = StartModifyRelation(cstoreDataFiles);
InsertTupleAndEnforceConstraints(modifyState, values, nulls); InsertTupleAndEnforceConstraints(modifyState, values, nulls);
FinishModifyRelation(modifyState); FinishModifyRelation(modifyState);
@ -169,9 +166,6 @@ UpdateCStoreDataFileMetadata(Oid relfilenode, int blockRowCount, int stripeRowCo
const int scanKeyCount = 1; const int scanKeyCount = 1;
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
bool indexOK = true; bool indexOK = true;
SysScanDesc scanDescriptor = NULL;
Form_cstore_data_files metadata = NULL;
HeapTuple heapTuple = NULL;
Datum values[Natts_cstore_data_files] = { 0 }; Datum values[Natts_cstore_data_files] = { 0 };
bool isnull[Natts_cstore_data_files] = { 0 }; bool isnull[Natts_cstore_data_files] = { 0 };
bool replace[Natts_cstore_data_files] = { 0 }; bool replace[Natts_cstore_data_files] = { 0 };
@ -182,19 +176,19 @@ UpdateCStoreDataFileMetadata(Oid relfilenode, int blockRowCount, int stripeRowCo
ScanKeyInit(&scanKey[0], Anum_cstore_data_files_relfilenode, BTEqualStrategyNumber, ScanKeyInit(&scanKey[0], Anum_cstore_data_files_relfilenode, BTEqualStrategyNumber,
F_INT8EQ, ObjectIdGetDatum(relfilenode)); F_INT8EQ, ObjectIdGetDatum(relfilenode));
scanDescriptor = systable_beginscan(cstoreDataFiles, SysScanDesc scanDescriptor = systable_beginscan(cstoreDataFiles,
CStoreDataFilesIndexRelationId(), CStoreDataFilesIndexRelationId(),
indexOK, indexOK,
NULL, scanKeyCount, scanKey); NULL, scanKeyCount, scanKey);
heapTuple = systable_getnext(scanDescriptor); HeapTuple heapTuple = systable_getnext(scanDescriptor);
if (heapTuple == NULL) if (heapTuple == NULL)
{ {
ereport(ERROR, (errmsg("relfilenode %d doesn't belong to a cstore table", ereport(ERROR, (errmsg("relfilenode %d doesn't belong to a cstore table",
relfilenode))); relfilenode)));
} }
metadata = (Form_cstore_data_files) GETSTRUCT(heapTuple); Form_cstore_data_files metadata = (Form_cstore_data_files) GETSTRUCT(heapTuple);
bool changed = false; bool changed = false;
if (metadata->block_row_count != blockRowCount) if (metadata->block_row_count != blockRowCount)
@ -250,14 +244,11 @@ SaveStripeSkipList(Oid relfilenode, uint64 stripe, StripeSkipList *stripeSkipLis
{ {
uint32 columnIndex = 0; uint32 columnIndex = 0;
uint32 blockIndex = 0; uint32 blockIndex = 0;
Oid cstoreSkipNodesOid = InvalidOid;
Relation cstoreSkipNodes = NULL;
ModifyState *modifyState = NULL;
uint32 columnCount = stripeSkipList->columnCount; uint32 columnCount = stripeSkipList->columnCount;
cstoreSkipNodesOid = CStoreSkipNodesRelationId(); Oid cstoreSkipNodesOid = CStoreSkipNodesRelationId();
cstoreSkipNodes = heap_open(cstoreSkipNodesOid, RowExclusiveLock); Relation cstoreSkipNodes = heap_open(cstoreSkipNodesOid, RowExclusiveLock);
modifyState = StartModifyRelation(cstoreSkipNodes); ModifyState *modifyState = StartModifyRelation(cstoreSkipNodes);
for (columnIndex = 0; columnIndex < columnCount; columnIndex++) for (columnIndex = 0; columnIndex < columnCount; columnIndex++)
{ {
@ -316,28 +307,24 @@ StripeSkipList *
ReadStripeSkipList(Oid relfilenode, uint64 stripe, TupleDesc tupleDescriptor, ReadStripeSkipList(Oid relfilenode, uint64 stripe, TupleDesc tupleDescriptor,
uint32 blockCount) uint32 blockCount)
{ {
StripeSkipList *skipList = NULL;
int32 columnIndex = 0; int32 columnIndex = 0;
Oid cstoreSkipNodesOid = InvalidOid;
Relation cstoreSkipNodes = NULL;
Relation index = NULL;
HeapTuple heapTuple = NULL; HeapTuple heapTuple = NULL;
uint32 columnCount = tupleDescriptor->natts; uint32 columnCount = tupleDescriptor->natts;
ScanKeyData scanKey[2]; ScanKeyData scanKey[2];
SysScanDesc scanDescriptor = NULL;
cstoreSkipNodesOid = CStoreSkipNodesRelationId(); Oid cstoreSkipNodesOid = CStoreSkipNodesRelationId();
cstoreSkipNodes = heap_open(cstoreSkipNodesOid, AccessShareLock); Relation cstoreSkipNodes = heap_open(cstoreSkipNodesOid, AccessShareLock);
index = index_open(CStoreSkipNodesIndexRelationId(), AccessShareLock); Relation index = index_open(CStoreSkipNodesIndexRelationId(), AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_cstore_skipnodes_relfilenode, ScanKeyInit(&scanKey[0], Anum_cstore_skipnodes_relfilenode,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode)); BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode));
ScanKeyInit(&scanKey[1], Anum_cstore_skipnodes_stripe, ScanKeyInit(&scanKey[1], Anum_cstore_skipnodes_stripe,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(stripe)); BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(stripe));
scanDescriptor = systable_beginscan_ordered(cstoreSkipNodes, index, NULL, 2, scanKey); SysScanDesc scanDescriptor = systable_beginscan_ordered(cstoreSkipNodes, index, NULL,
2, scanKey);
skipList = palloc0(sizeof(StripeSkipList)); StripeSkipList *skipList = palloc0(sizeof(StripeSkipList));
skipList->blockCount = blockCount; skipList->blockCount = blockCount;
skipList->columnCount = columnCount; skipList->columnCount = columnCount;
skipList->blockSkipNodeArray = palloc0(columnCount * sizeof(ColumnBlockSkipNode *)); skipList->blockSkipNodeArray = palloc0(columnCount * sizeof(ColumnBlockSkipNode *));
@ -349,18 +336,14 @@ ReadStripeSkipList(Oid relfilenode, uint64 stripe, TupleDesc tupleDescriptor,
while (HeapTupleIsValid(heapTuple = systable_getnext(scanDescriptor))) while (HeapTupleIsValid(heapTuple = systable_getnext(scanDescriptor)))
{ {
int32 attr = 0;
int32 blockIndex = 0;
ColumnBlockSkipNode *skipNode = NULL;
Datum datumArray[Natts_cstore_skipnodes]; Datum datumArray[Natts_cstore_skipnodes];
bool isNullArray[Natts_cstore_skipnodes]; bool isNullArray[Natts_cstore_skipnodes];
heap_deform_tuple(heapTuple, RelationGetDescr(cstoreSkipNodes), datumArray, heap_deform_tuple(heapTuple, RelationGetDescr(cstoreSkipNodes), datumArray,
isNullArray); isNullArray);
attr = DatumGetInt32(datumArray[Anum_cstore_skipnodes_attr - 1]); int32 attr = DatumGetInt32(datumArray[Anum_cstore_skipnodes_attr - 1]);
blockIndex = DatumGetInt32(datumArray[Anum_cstore_skipnodes_block - 1]); int32 blockIndex = DatumGetInt32(datumArray[Anum_cstore_skipnodes_block - 1]);
if (attr <= 0 || attr > columnCount) if (attr <= 0 || attr > columnCount)
{ {
@ -376,7 +359,8 @@ ReadStripeSkipList(Oid relfilenode, uint64 stripe, TupleDesc tupleDescriptor,
columnIndex = attr - 1; columnIndex = attr - 1;
skipNode = &skipList->blockSkipNodeArray[columnIndex][blockIndex]; ColumnBlockSkipNode *skipNode =
&skipList->blockSkipNodeArray[columnIndex][blockIndex];
skipNode->rowCount = DatumGetInt64(datumArray[Anum_cstore_skipnodes_row_count - skipNode->rowCount = DatumGetInt64(datumArray[Anum_cstore_skipnodes_row_count -
1]); 1]);
skipNode->valueBlockOffset = skipNode->valueBlockOffset =
@ -507,12 +491,11 @@ GetHighestUsedAddressAndId(Oid relfilenode,
uint64 *highestUsedId) uint64 *highestUsedId)
{ {
ListCell *stripeMetadataCell = NULL; ListCell *stripeMetadataCell = NULL;
List *stripeMetadataList = NIL;
SnapshotData SnapshotDirty; SnapshotData SnapshotDirty;
InitDirtySnapshot(SnapshotDirty); InitDirtySnapshot(SnapshotDirty);
stripeMetadataList = ReadDataFileStripeList(relfilenode, &SnapshotDirty); List *stripeMetadataList = ReadDataFileStripeList(relfilenode, &SnapshotDirty);
*highestUsedId = 0; *highestUsedId = 0;
*highestUsedAddress = 0; *highestUsedAddress = 0;
@ -538,14 +521,7 @@ ReserveStripe(Relation rel, uint64 sizeBytes,
uint64 blockCount, uint64 blockRowCount) uint64 blockCount, uint64 blockRowCount)
{ {
StripeMetadata stripe = { 0 }; StripeMetadata stripe = { 0 };
Oid relfilenode = InvalidOid;
uint64 currLogicalHigh = 0; uint64 currLogicalHigh = 0;
SmgrAddr currSmgrHigh;
uint64 nblocks = 0;
uint64 resLogicalStart = 0;
SmgrAddr resSmgrStart;
uint64 resLogicalEnd = 0;
SmgrAddr resSmgrEnd;
uint64 highestId = 0; uint64 highestId = 0;
/* /*
@ -556,18 +532,18 @@ ReserveStripe(Relation rel, uint64 sizeBytes,
*/ */
LockRelation(rel, ShareUpdateExclusiveLock); LockRelation(rel, ShareUpdateExclusiveLock);
relfilenode = rel->rd_node.relNode; Oid relfilenode = rel->rd_node.relNode;
GetHighestUsedAddressAndId(relfilenode, &currLogicalHigh, &highestId); GetHighestUsedAddressAndId(relfilenode, &currLogicalHigh, &highestId);
currSmgrHigh = logical_to_smgr(currLogicalHigh); SmgrAddr currSmgrHigh = logical_to_smgr(currLogicalHigh);
resSmgrStart = next_block_start(currSmgrHigh); SmgrAddr resSmgrStart = next_block_start(currSmgrHigh);
resLogicalStart = smgr_to_logical(resSmgrStart); uint64 resLogicalStart = smgr_to_logical(resSmgrStart);
resLogicalEnd = resLogicalStart + sizeBytes - 1; uint64 resLogicalEnd = resLogicalStart + sizeBytes - 1;
resSmgrEnd = logical_to_smgr(resLogicalEnd); SmgrAddr resSmgrEnd = logical_to_smgr(resLogicalEnd);
RelationOpenSmgr(rel); RelationOpenSmgr(rel);
nblocks = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM); uint64 nblocks = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM);
while (resSmgrEnd.blockno >= nblocks) while (resSmgrEnd.blockno >= nblocks)
{ {
@ -602,34 +578,29 @@ static List *
ReadDataFileStripeList(Oid relfilenode, Snapshot snapshot) ReadDataFileStripeList(Oid relfilenode, Snapshot snapshot)
{ {
List *stripeMetadataList = NIL; List *stripeMetadataList = NIL;
Oid cstoreStripesOid = InvalidOid;
Relation cstoreStripes = NULL;
Relation index = NULL;
TupleDesc tupleDescriptor = NULL;
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
SysScanDesc scanDescriptor = NULL;
HeapTuple heapTuple; HeapTuple heapTuple;
ScanKeyInit(&scanKey[0], Anum_cstore_stripes_relfilenode, ScanKeyInit(&scanKey[0], Anum_cstore_stripes_relfilenode,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode)); BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode));
cstoreStripesOid = CStoreStripesRelationId(); Oid cstoreStripesOid = CStoreStripesRelationId();
cstoreStripes = heap_open(cstoreStripesOid, AccessShareLock); Relation cstoreStripes = heap_open(cstoreStripesOid, AccessShareLock);
index = index_open(CStoreStripesIndexRelationId(), AccessShareLock); Relation index = index_open(CStoreStripesIndexRelationId(), AccessShareLock);
tupleDescriptor = RelationGetDescr(cstoreStripes); TupleDesc tupleDescriptor = RelationGetDescr(cstoreStripes);
scanDescriptor = systable_beginscan_ordered(cstoreStripes, index, snapshot, 1, SysScanDesc scanDescriptor = systable_beginscan_ordered(cstoreStripes, index,
snapshot, 1,
scanKey); scanKey);
while (HeapTupleIsValid(heapTuple = systable_getnext(scanDescriptor))) while (HeapTupleIsValid(heapTuple = systable_getnext(scanDescriptor)))
{ {
StripeMetadata *stripeMetadata = NULL;
Datum datumArray[Natts_cstore_stripes]; Datum datumArray[Natts_cstore_stripes];
bool isNullArray[Natts_cstore_stripes]; bool isNullArray[Natts_cstore_stripes];
heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray);
stripeMetadata = palloc0(sizeof(StripeMetadata)); StripeMetadata *stripeMetadata = palloc0(sizeof(StripeMetadata));
stripeMetadata->id = DatumGetInt64(datumArray[Anum_cstore_stripes_stripe - 1]); stripeMetadata->id = DatumGetInt64(datumArray[Anum_cstore_stripes_stripe - 1]);
stripeMetadata->fileOffset = DatumGetInt64( stripeMetadata->fileOffset = DatumGetInt64(
datumArray[Anum_cstore_stripes_file_offset - 1]); datumArray[Anum_cstore_stripes_file_offset - 1]);
@ -663,19 +634,13 @@ static bool
ReadCStoreDataFiles(Oid relfilenode, DataFileMetadata *metadata) ReadCStoreDataFiles(Oid relfilenode, DataFileMetadata *metadata)
{ {
bool found = false; bool found = false;
Oid cstoreDataFilesOid = InvalidOid;
Relation cstoreDataFiles = NULL;
Relation index = NULL;
TupleDesc tupleDescriptor = NULL;
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
SysScanDesc scanDescriptor = NULL;
HeapTuple heapTuple = NULL;
ScanKeyInit(&scanKey[0], Anum_cstore_data_files_relfilenode, ScanKeyInit(&scanKey[0], Anum_cstore_data_files_relfilenode,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode)); BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode));
cstoreDataFilesOid = CStoreDataFilesRelationId(); Oid cstoreDataFilesOid = CStoreDataFilesRelationId();
cstoreDataFiles = try_relation_open(cstoreDataFilesOid, AccessShareLock); Relation cstoreDataFiles = try_relation_open(cstoreDataFilesOid, AccessShareLock);
if (cstoreDataFiles == NULL) if (cstoreDataFiles == NULL)
{ {
/* /*
@ -685,7 +650,7 @@ ReadCStoreDataFiles(Oid relfilenode, DataFileMetadata *metadata)
return false; return false;
} }
index = try_relation_open(CStoreDataFilesIndexRelationId(), AccessShareLock); Relation index = try_relation_open(CStoreDataFilesIndexRelationId(), AccessShareLock);
if (index == NULL) if (index == NULL)
{ {
heap_close(cstoreDataFiles, NoLock); heap_close(cstoreDataFiles, NoLock);
@ -694,11 +659,12 @@ ReadCStoreDataFiles(Oid relfilenode, DataFileMetadata *metadata)
return false; return false;
} }
tupleDescriptor = RelationGetDescr(cstoreDataFiles); TupleDesc tupleDescriptor = RelationGetDescr(cstoreDataFiles);
scanDescriptor = systable_beginscan_ordered(cstoreDataFiles, index, NULL, 1, scanKey); SysScanDesc scanDescriptor = systable_beginscan_ordered(cstoreDataFiles, index, NULL,
1, scanKey);
heapTuple = systable_getnext(scanDescriptor); HeapTuple heapTuple = systable_getnext(scanDescriptor);
if (HeapTupleIsValid(heapTuple)) if (HeapTupleIsValid(heapTuple))
{ {
Datum datumArray[Natts_cstore_data_files]; Datum datumArray[Natts_cstore_data_files];
@ -707,13 +673,11 @@ ReadCStoreDataFiles(Oid relfilenode, DataFileMetadata *metadata)
if (metadata) if (metadata)
{ {
Name compressionName = NULL;
metadata->blockRowCount = DatumGetInt32( metadata->blockRowCount = DatumGetInt32(
datumArray[Anum_cstore_data_files_block_row_count - 1]); datumArray[Anum_cstore_data_files_block_row_count - 1]);
metadata->stripeRowCount = DatumGetInt32( metadata->stripeRowCount = DatumGetInt32(
datumArray[Anum_cstore_data_files_stripe_row_count - 1]); datumArray[Anum_cstore_data_files_stripe_row_count - 1]);
compressionName = DatumGetName( Name compressionName = DatumGetName(
datumArray[Anum_cstore_data_files_compression - 1]); datumArray[Anum_cstore_data_files_compression - 1]);
metadata->compression = ParseCompressionType(NameStr(*compressionName)); metadata->compression = ParseCompressionType(NameStr(*compressionName));
} }
@ -734,12 +698,7 @@ ReadCStoreDataFiles(Oid relfilenode, DataFileMetadata *metadata)
void void
DeleteDataFileMetadataRowIfExists(Oid relfilenode) DeleteDataFileMetadataRowIfExists(Oid relfilenode)
{ {
Oid cstoreDataFilesOid = InvalidOid;
Relation cstoreDataFiles = NULL;
Relation index = NULL;
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
SysScanDesc scanDescriptor = NULL;
HeapTuple heapTuple = NULL;
/* /*
* During a restore for binary upgrade, metadata tables and indexes may or * During a restore for binary upgrade, metadata tables and indexes may or
@ -753,19 +712,20 @@ DeleteDataFileMetadataRowIfExists(Oid relfilenode)
ScanKeyInit(&scanKey[0], Anum_cstore_data_files_relfilenode, ScanKeyInit(&scanKey[0], Anum_cstore_data_files_relfilenode,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode)); BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(relfilenode));
cstoreDataFilesOid = CStoreDataFilesRelationId(); Oid cstoreDataFilesOid = CStoreDataFilesRelationId();
cstoreDataFiles = try_relation_open(cstoreDataFilesOid, AccessShareLock); Relation cstoreDataFiles = try_relation_open(cstoreDataFilesOid, AccessShareLock);
if (cstoreDataFiles == NULL) if (cstoreDataFiles == NULL)
{ {
/* extension has been dropped */ /* extension has been dropped */
return; return;
} }
index = index_open(CStoreDataFilesIndexRelationId(), AccessShareLock); Relation index = index_open(CStoreDataFilesIndexRelationId(), AccessShareLock);
scanDescriptor = systable_beginscan_ordered(cstoreDataFiles, index, NULL, 1, scanKey); SysScanDesc scanDescriptor = systable_beginscan_ordered(cstoreDataFiles, index, NULL,
1, scanKey);
heapTuple = systable_getnext(scanDescriptor); HeapTuple heapTuple = systable_getnext(scanDescriptor);
if (HeapTupleIsValid(heapTuple)) if (HeapTupleIsValid(heapTuple))
{ {
ModifyState *modifyState = StartModifyRelation(cstoreDataFiles); ModifyState *modifyState = StartModifyRelation(cstoreDataFiles);
@ -785,13 +745,12 @@ DeleteDataFileMetadataRowIfExists(Oid relfilenode)
static ModifyState * static ModifyState *
StartModifyRelation(Relation rel) StartModifyRelation(Relation rel)
{ {
ModifyState *modifyState = NULL;
EState *estate = create_estate_for_relation(rel); EState *estate = create_estate_for_relation(rel);
/* ExecSimpleRelationInsert, ... require caller to open indexes */ /* ExecSimpleRelationInsert, ... require caller to open indexes */
ExecOpenIndices(estate->es_result_relation_info, false); ExecOpenIndices(estate->es_result_relation_info, false);
modifyState = palloc(sizeof(ModifyState)); ModifyState *modifyState = palloc(sizeof(ModifyState));
modifyState->rel = rel; modifyState->rel = rel;
modifyState->estate = estate; modifyState->estate = estate;
@ -869,13 +828,11 @@ FinishModifyRelation(ModifyState *state)
static EState * static EState *
create_estate_for_relation(Relation rel) create_estate_for_relation(Relation rel)
{ {
EState *estate;
ResultRelInfo *resultRelInfo; ResultRelInfo *resultRelInfo;
RangeTblEntry *rte;
estate = CreateExecutorState(); EState *estate = CreateExecutorState();
rte = makeNode(RangeTblEntry); RangeTblEntry *rte = makeNode(RangeTblEntry);
rte->rtekind = RTE_RELATION; rte->rtekind = RTE_RELATION;
rte->relid = RelationGetRelid(rel); rte->relid = RelationGetRelid(rel);
rte->relkind = rel->rd_rel->relkind; rte->relkind = rel->rd_rel->relkind;

123
src/backend/columnar/cstore_reader.c
View File

@ -84,23 +84,20 @@ TableReadState *
CStoreBeginRead(Relation relation, TupleDesc tupleDescriptor, CStoreBeginRead(Relation relation, TupleDesc tupleDescriptor,
List *projectedColumnList, List *whereClauseList) List *projectedColumnList, List *whereClauseList)
{ {
TableReadState *readState = NULL;
DataFileMetadata *datafileMetadata = NULL;
MemoryContext stripeReadContext = NULL;
Oid relNode = relation->rd_node.relNode; Oid relNode = relation->rd_node.relNode;
datafileMetadata = ReadDataFileMetadata(relNode, false); DataFileMetadata *datafileMetadata = ReadDataFileMetadata(relNode, false);
/* /*
* We allocate all stripe specific data in the stripeReadContext, and reset * We allocate all stripe specific data in the stripeReadContext, and reset
* this memory context before loading a new stripe. This is to avoid memory * this memory context before loading a new stripe. This is to avoid memory
* leaks. * leaks.
*/ */
stripeReadContext = AllocSetContextCreate(CurrentMemoryContext, MemoryContext stripeReadContext = AllocSetContextCreate(CurrentMemoryContext,
"Stripe Read Memory Context", "Stripe Read Memory Context",
ALLOCSET_DEFAULT_SIZES); ALLOCSET_DEFAULT_SIZES);
readState = palloc0(sizeof(TableReadState)); TableReadState *readState = palloc0(sizeof(TableReadState));
readState->relation = relation; readState->relation = relation;
readState->datafileMetadata = datafileMetadata; readState->datafileMetadata = datafileMetadata;
readState->projectedColumnList = projectedColumnList; readState->projectedColumnList = projectedColumnList;
@ -125,8 +122,6 @@ CStoreBeginRead(Relation relation, TupleDesc tupleDescriptor,
bool bool
CStoreReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNulls) CStoreReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNulls)
{ {
uint32 blockIndex = 0;
uint32 blockRowIndex = 0;
StripeMetadata *stripeMetadata = readState->currentStripeMetadata; StripeMetadata *stripeMetadata = readState->currentStripeMetadata;
MemoryContext oldContext = NULL; MemoryContext oldContext = NULL;
@ -138,7 +133,6 @@ CStoreReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNu
*/ */
while (readState->stripeBuffers == NULL) while (readState->stripeBuffers == NULL)
{ {
StripeBuffers *stripeBuffers = NULL;
List *stripeMetadataList = readState->datafileMetadata->stripeMetadataList; List *stripeMetadataList = readState->datafileMetadata->stripeMetadataList;
uint32 stripeCount = list_length(stripeMetadataList); uint32 stripeCount = list_length(stripeMetadataList);
@ -153,11 +147,14 @@ CStoreReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNu
readState->blockData = NULL; readState->blockData = NULL;
stripeMetadata = list_nth(stripeMetadataList, readState->readStripeCount); stripeMetadata = list_nth(stripeMetadataList, readState->readStripeCount);
stripeBuffers = LoadFilteredStripeBuffers(readState->relation, StripeBuffers *stripeBuffers = LoadFilteredStripeBuffers(readState->relation,
stripeMetadata, stripeMetadata,
readState->tupleDescriptor, readState->
readState->projectedColumnList, tupleDescriptor,
readState->whereClauseList); readState->
projectedColumnList,
readState->
whereClauseList);
readState->readStripeCount++; readState->readStripeCount++;
readState->currentStripeMetadata = stripeMetadata; readState->currentStripeMetadata = stripeMetadata;
@ -172,17 +169,15 @@ CStoreReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNu
} }
} }
blockIndex = readState->stripeReadRowCount / stripeMetadata->blockRowCount; uint32 blockIndex = readState->stripeReadRowCount / stripeMetadata->blockRowCount;
blockRowIndex = readState->stripeReadRowCount % stripeMetadata->blockRowCount; uint32 blockRowIndex = readState->stripeReadRowCount % stripeMetadata->blockRowCount;
if (blockIndex != readState->deserializedBlockIndex) if (blockIndex != readState->deserializedBlockIndex)
{ {
uint32 lastBlockIndex = 0;
uint32 blockRowCount = 0; uint32 blockRowCount = 0;
uint32 stripeRowCount = 0;
stripeRowCount = stripeMetadata->rowCount; uint32 stripeRowCount = stripeMetadata->rowCount;
lastBlockIndex = stripeRowCount / stripeMetadata->blockRowCount; uint32 lastBlockIndex = stripeRowCount / stripeMetadata->blockRowCount;
if (blockIndex == lastBlockIndex) if (blockIndex == lastBlockIndex)
{ {
blockRowCount = stripeRowCount % stripeMetadata->blockRowCount; blockRowCount = stripeRowCount % stripeMetadata->blockRowCount;
@ -317,11 +312,11 @@ FreeBlockData(BlockData *blockData)
uint64 uint64
CStoreTableRowCount(Relation relation) CStoreTableRowCount(Relation relation)
{ {
DataFileMetadata *datafileMetadata = NULL;
ListCell *stripeMetadataCell = NULL; ListCell *stripeMetadataCell = NULL;
uint64 totalRowCount = 0; uint64 totalRowCount = 0;
datafileMetadata = ReadDataFileMetadata(relation->rd_node.relNode, false); DataFileMetadata *datafileMetadata = ReadDataFileMetadata(relation->rd_node.relNode,
false);
foreach(stripeMetadataCell, datafileMetadata->stripeMetadataList) foreach(stripeMetadataCell, datafileMetadata->stripeMetadataList)
{ {
@ -343,8 +338,6 @@ LoadFilteredStripeBuffers(Relation relation, StripeMetadata *stripeMetadata,
TupleDesc tupleDescriptor, List *projectedColumnList, TupleDesc tupleDescriptor, List *projectedColumnList,
List *whereClauseList) List *whereClauseList)
{ {
StripeBuffers *stripeBuffers = NULL;
ColumnBuffers **columnBuffersArray = NULL;
uint32 columnIndex = 0; uint32 columnIndex = 0;
uint32 columnCount = tupleDescriptor->natts; uint32 columnCount = tupleDescriptor->natts;
@ -363,7 +356,7 @@ LoadFilteredStripeBuffers(Relation relation, StripeMetadata *stripeMetadata,
selectedBlockMask); selectedBlockMask);
/* load column data for projected columns */ /* load column data for projected columns */
columnBuffersArray = palloc0(columnCount * sizeof(ColumnBuffers *)); ColumnBuffers **columnBuffersArray = palloc0(columnCount * sizeof(ColumnBuffers *));
for (columnIndex = 0; columnIndex < stripeMetadata->columnCount; columnIndex++) for (columnIndex = 0; columnIndex < stripeMetadata->columnCount; columnIndex++)
{ {
@ -383,7 +376,7 @@ LoadFilteredStripeBuffers(Relation relation, StripeMetadata *stripeMetadata,
} }
} }
stripeBuffers = palloc0(sizeof(StripeBuffers)); StripeBuffers *stripeBuffers = palloc0(sizeof(StripeBuffers));
stripeBuffers->columnCount = columnCount; stripeBuffers->columnCount = columnCount;
stripeBuffers->rowCount = StripeSkipListRowCount(selectedBlockSkipList); stripeBuffers->rowCount = StripeSkipListRowCount(selectedBlockSkipList);
stripeBuffers->columnBuffersArray = columnBuffersArray; stripeBuffers->columnBuffersArray = columnBuffersArray;
@ -432,7 +425,6 @@ LoadColumnBuffers(Relation relation, ColumnBlockSkipNode *blockSkipNodeArray,
uint32 blockCount, uint64 stripeOffset, uint32 blockCount, uint64 stripeOffset,
Form_pg_attribute attributeForm) Form_pg_attribute attributeForm)
{ {
ColumnBuffers *columnBuffers = NULL;
uint32 blockIndex = 0; uint32 blockIndex = 0;
ColumnBlockBuffers **blockBuffersArray = ColumnBlockBuffers **blockBuffersArray =
palloc0(blockCount * sizeof(ColumnBlockBuffers *)); palloc0(blockCount * sizeof(ColumnBlockBuffers *));
@ -470,7 +462,7 @@ LoadColumnBuffers(Relation relation, ColumnBlockSkipNode *blockSkipNodeArray,
blockBuffersArray[blockIndex]->valueCompressionType = compressionType; blockBuffersArray[blockIndex]->valueCompressionType = compressionType;
} }
columnBuffers = palloc0(sizeof(ColumnBuffers)); ColumnBuffers *columnBuffers = palloc0(sizeof(ColumnBuffers));
columnBuffers->blockBuffersArray = blockBuffersArray; columnBuffers->blockBuffersArray = blockBuffersArray;
return columnBuffers; return columnBuffers;
@ -486,34 +478,31 @@ static bool *
SelectedBlockMask(StripeSkipList *stripeSkipList, List *projectedColumnList, SelectedBlockMask(StripeSkipList *stripeSkipList, List *projectedColumnList,
List *whereClauseList) List *whereClauseList)
{ {
bool *selectedBlockMask = NULL;
ListCell *columnCell = NULL; ListCell *columnCell = NULL;
uint32 blockIndex = 0; uint32 blockIndex = 0;
List *restrictInfoList = BuildRestrictInfoList(whereClauseList); List *restrictInfoList = BuildRestrictInfoList(whereClauseList);
selectedBlockMask = palloc0(stripeSkipList->blockCount * sizeof(bool)); bool *selectedBlockMask = palloc0(stripeSkipList->blockCount * sizeof(bool));
memset(selectedBlockMask, true, stripeSkipList->blockCount * sizeof(bool)); memset(selectedBlockMask, true, stripeSkipList->blockCount * sizeof(bool));
foreach(columnCell, projectedColumnList) foreach(columnCell, projectedColumnList)
{ {
Var *column = lfirst(columnCell); Var *column = lfirst(columnCell);
uint32 columnIndex = column->varattno - 1; uint32 columnIndex = column->varattno - 1;
FmgrInfo *comparisonFunction = NULL;
Node *baseConstraint = NULL;
/* if this column's data type doesn't have a comparator, skip it */ /* if this column's data type doesn't have a comparator, skip it */
comparisonFunction = GetFunctionInfoOrNull(column->vartype, BTREE_AM_OID, FmgrInfo *comparisonFunction = GetFunctionInfoOrNull(column->vartype,
BTREE_AM_OID,
BTORDER_PROC); BTORDER_PROC);
if (comparisonFunction == NULL) if (comparisonFunction == NULL)
{ {
continue; continue;
} }
baseConstraint = BuildBaseConstraint(column); Node *baseConstraint = BuildBaseConstraint(column);
for (blockIndex = 0; blockIndex < stripeSkipList->blockCount; blockIndex++) for (blockIndex = 0; blockIndex < stripeSkipList->blockCount; blockIndex++)
{ {
bool predicateRefuted = false; bool predicateRefuted = false;
List *constraintList = NIL;
ColumnBlockSkipNode *blockSkipNodeArray = ColumnBlockSkipNode *blockSkipNodeArray =
stripeSkipList->blockSkipNodeArray[columnIndex]; stripeSkipList->blockSkipNodeArray[columnIndex];
ColumnBlockSkipNode *blockSkipNode = &blockSkipNodeArray[blockIndex]; ColumnBlockSkipNode *blockSkipNode = &blockSkipNodeArray[blockIndex];
@ -530,7 +519,7 @@ SelectedBlockMask(StripeSkipList *stripeSkipList, List *projectedColumnList,
UpdateConstraint(baseConstraint, blockSkipNode->minimumValue, UpdateConstraint(baseConstraint, blockSkipNode->minimumValue,
blockSkipNode->maximumValue); blockSkipNode->maximumValue);
constraintList = list_make1(baseConstraint); List *constraintList = list_make1(baseConstraint);
#if (PG_VERSION_NUM >= 100000) #if (PG_VERSION_NUM >= 100000)
predicateRefuted = predicate_refuted_by(constraintList, restrictInfoList, predicateRefuted = predicate_refuted_by(constraintList, restrictInfoList,
false); false);
@ -558,24 +547,21 @@ FmgrInfo *
GetFunctionInfoOrNull(Oid typeId, Oid accessMethodId, int16 procedureId) GetFunctionInfoOrNull(Oid typeId, Oid accessMethodId, int16 procedureId)
{ {
FmgrInfo *functionInfo = NULL; FmgrInfo *functionInfo = NULL;
Oid operatorClassId = InvalidOid;
Oid operatorFamilyId = InvalidOid;
Oid operatorId = InvalidOid;
/* get default operator class from pg_opclass for datum type */ /* get default operator class from pg_opclass for datum type */
operatorClassId = GetDefaultOpClass(typeId, accessMethodId); Oid operatorClassId = GetDefaultOpClass(typeId, accessMethodId);
if (operatorClassId == InvalidOid) if (operatorClassId == InvalidOid)
{ {
return NULL; return NULL;
} }
operatorFamilyId = get_opclass_family(operatorClassId); Oid operatorFamilyId = get_opclass_family(operatorClassId);
if (operatorFamilyId == InvalidOid) if (operatorFamilyId == InvalidOid)
{ {
return NULL; return NULL;
} }
operatorId = get_opfamily_proc(operatorFamilyId, typeId, typeId, procedureId); Oid operatorId = get_opfamily_proc(operatorFamilyId, typeId, typeId, procedureId);
if (operatorId != InvalidOid) if (operatorId != InvalidOid)
{ {
functionInfo = (FmgrInfo *) palloc0(sizeof(FmgrInfo)); functionInfo = (FmgrInfo *) palloc0(sizeof(FmgrInfo));
@ -601,10 +587,9 @@ BuildRestrictInfoList(List *whereClauseList)
ListCell *qualCell = NULL; ListCell *qualCell = NULL;
foreach(qualCell, whereClauseList) foreach(qualCell, whereClauseList)
{ {
RestrictInfo *restrictInfo = NULL;
Node *qualNode = (Node *) lfirst(qualCell); Node *qualNode = (Node *) lfirst(qualCell);
restrictInfo = make_simple_restrictinfo((Expr *) qualNode); RestrictInfo *restrictInfo = make_simple_restrictinfo((Expr *) qualNode);
restrictInfoList = lappend(restrictInfoList, restrictInfo); restrictInfoList = lappend(restrictInfoList, restrictInfo);
} }
@ -622,14 +607,10 @@ BuildRestrictInfoList(List *whereClauseList)
static Node * static Node *
BuildBaseConstraint(Var *variable) BuildBaseConstraint(Var *variable)
{ {
Node *baseConstraint = NULL; OpExpr *lessThanExpr = MakeOpExpression(variable, BTLessEqualStrategyNumber);
OpExpr *lessThanExpr = NULL; OpExpr *greaterThanExpr = MakeOpExpression(variable, BTGreaterEqualStrategyNumber);
OpExpr *greaterThanExpr = NULL;
lessThanExpr = MakeOpExpression(variable, BTLessEqualStrategyNumber); Node *baseConstraint = make_and_qual((Node *) lessThanExpr, (Node *) greaterThanExpr);
greaterThanExpr = MakeOpExpression(variable, BTGreaterEqualStrategyNumber);
baseConstraint = make_and_qual((Node *) lessThanExpr, (Node *) greaterThanExpr);
return baseConstraint; return baseConstraint;
} }
@ -648,17 +629,14 @@ MakeOpExpression(Var *variable, int16 strategyNumber)
Oid collationId = variable->varcollid; Oid collationId = variable->varcollid;
Oid accessMethodId = BTREE_AM_OID; Oid accessMethodId = BTREE_AM_OID;
Oid operatorId = InvalidOid;
Const *constantValue = NULL;
OpExpr *expression = NULL;
/* Load the operator from system catalogs */ /* Load the operator from system catalogs */
operatorId = GetOperatorByType(typeId, accessMethodId, strategyNumber); Oid operatorId = GetOperatorByType(typeId, accessMethodId, strategyNumber);
constantValue = makeNullConst(typeId, typeModId, collationId); Const *constantValue = makeNullConst(typeId, typeModId, collationId);
/* Now make the expression with the given variable and a null constant */ /* Now make the expression with the given variable and a null constant */
expression = (OpExpr *) make_opclause(operatorId, OpExpr *expression = (OpExpr *) make_opclause(operatorId,
InvalidOid, /* no result type yet */ InvalidOid, /* no result type yet */
false, /* no return set */ false, /* no return set */
(Expr *) variable, (Expr *) variable,
@ -707,14 +685,12 @@ UpdateConstraint(Node *baseConstraint, Datum minValue, Datum maxValue)
Node *minNode = get_rightop((Expr *) greaterThanExpr); Node *minNode = get_rightop((Expr *) greaterThanExpr);
Node *maxNode = get_rightop((Expr *) lessThanExpr); Node *maxNode = get_rightop((Expr *) lessThanExpr);
Const *minConstant = NULL;
Const *maxConstant = NULL;
Assert(IsA(minNode, Const)); Assert(IsA(minNode, Const));
Assert(IsA(maxNode, Const)); Assert(IsA(maxNode, Const));
minConstant = (Const *) minNode; Const *minConstant = (Const *) minNode;
maxConstant = (Const *) maxNode; Const *maxConstant = (Const *) maxNode;
minConstant->constvalue = minValue; minConstant->constvalue = minValue;
maxConstant->constvalue = maxValue; maxConstant->constvalue = maxValue;
@ -735,8 +711,6 @@ static StripeSkipList *
SelectedBlockSkipList(StripeSkipList *stripeSkipList, bool *projectedColumnMask, SelectedBlockSkipList(StripeSkipList *stripeSkipList, bool *projectedColumnMask,
bool *selectedBlockMask) bool *selectedBlockMask)
{ {
StripeSkipList *SelectedBlockSkipList = NULL;
ColumnBlockSkipNode **selectedBlockSkipNodeArray = NULL;
uint32 selectedBlockCount = 0; uint32 selectedBlockCount = 0;
uint32 blockIndex = 0; uint32 blockIndex = 0;
uint32 columnIndex = 0; uint32 columnIndex = 0;
@ -750,7 +724,9 @@ SelectedBlockSkipList(StripeSkipList *stripeSkipList, bool *projectedColumnMask,
} }
} }
selectedBlockSkipNodeArray = palloc0(columnCount * sizeof(ColumnBlockSkipNode *)); ColumnBlockSkipNode **selectedBlockSkipNodeArray = palloc0(columnCount *
sizeof(ColumnBlockSkipNode
*));
for (columnIndex = 0; columnIndex < columnCount; columnIndex++) for (columnIndex = 0; columnIndex < columnCount; columnIndex++)
{ {
uint32 selectedBlockIndex = 0; uint32 selectedBlockIndex = 0;
@ -779,7 +755,7 @@ SelectedBlockSkipList(StripeSkipList *stripeSkipList, bool *projectedColumnMask,
} }
} }
SelectedBlockSkipList = palloc0(sizeof(StripeSkipList)); StripeSkipList *SelectedBlockSkipList = palloc0(sizeof(StripeSkipList));
SelectedBlockSkipList->blockSkipNodeArray = selectedBlockSkipNodeArray; SelectedBlockSkipList->blockSkipNodeArray = selectedBlockSkipNodeArray;
SelectedBlockSkipList->blockCount = selectedBlockCount; SelectedBlockSkipList->blockCount = selectedBlockCount;
SelectedBlockSkipList->columnCount = stripeSkipList->columnCount; SelectedBlockSkipList->columnCount = stripeSkipList->columnCount;
@ -882,14 +858,12 @@ DeserializeDatumArray(StringInfo datumBuffer, bool *existsArray, uint32 datumCou
for (datumIndex = 0; datumIndex < datumCount; datumIndex++) for (datumIndex = 0; datumIndex < datumCount; datumIndex++)
{ {
char *currentDatumDataPointer = NULL;
if (!existsArray[datumIndex]) if (!existsArray[datumIndex])
{ {
continue; continue;
} }
currentDatumDataPointer = datumBuffer->data + currentDatumDataOffset; char *currentDatumDataPointer = datumBuffer->data + currentDatumDataOffset;
datumArray[datumIndex] = fetch_att(currentDatumDataPointer, datumTypeByValue, datumArray[datumIndex] = fetch_att(currentDatumDataPointer, datumTypeByValue,
datumTypeLength); datumTypeLength);
@ -940,10 +914,9 @@ DeserializeBlockData(StripeBuffers *stripeBuffers, uint64 blockIndex,
{ {
ColumnBlockBuffers *blockBuffers = ColumnBlockBuffers *blockBuffers =
columnBuffers->blockBuffersArray[blockIndex]; columnBuffers->blockBuffersArray[blockIndex];
StringInfo valueBuffer = NULL;
/* decompress and deserialize current block's data */ /* decompress and deserialize current block's data */
valueBuffer = DecompressBuffer(blockBuffers->valueBuffer, StringInfo valueBuffer = DecompressBuffer(blockBuffers->valueBuffer,
blockBuffers->valueCompressionType); blockBuffers->valueCompressionType);
if (blockBuffers->valueCompressionType != COMPRESSION_NONE) if (blockBuffers->valueCompressionType != COMPRESSION_NONE)
@ -1045,17 +1018,13 @@ ReadFromSmgr(Relation rel, uint64 offset, uint32 size)
while (read < size) while (read < size)
{ {
Buffer buffer;
Page page;
PageHeader phdr;
uint32 to_read;
SmgrAddr addr = logical_to_smgr(offset + read); SmgrAddr addr = logical_to_smgr(offset + read);
buffer = ReadBuffer(rel, addr.blockno); Buffer buffer = ReadBuffer(rel, addr.blockno);
page = BufferGetPage(buffer); Page page = BufferGetPage(buffer);
phdr = (PageHeader) page; PageHeader phdr = (PageHeader) page;
to_read = Min(size - read, phdr->pd_upper - addr.offset); uint32 to_read = Min(size - read, phdr->pd_upper - addr.offset);
memcpy(resultBuffer->data + read, page + addr.offset, to_read); memcpy(resultBuffer->data + read, page + addr.offset, to_read);
ReleaseBuffer(buffer); ReleaseBuffer(buffer);
read += to_read; read += to_read;

95
src/backend/columnar/cstore_tableam.c
View File

@ -125,13 +125,10 @@ CStoreTableAMDefaultOptions()
static CStoreOptions * static CStoreOptions *
CStoreTableAMGetOptions(Relation rel) CStoreTableAMGetOptions(Relation rel)
{ {
CStoreOptions *cstoreOptions = NULL;
DataFileMetadata *metadata = NULL;
Assert(rel != NULL); Assert(rel != NULL);
cstoreOptions = palloc0(sizeof(CStoreOptions)); CStoreOptions *cstoreOptions = palloc0(sizeof(CStoreOptions));
metadata = ReadDataFileMetadata(rel->rd_node.relNode, false); DataFileMetadata *metadata = ReadDataFileMetadata(rel->rd_node.relNode, false);
cstoreOptions->compressionType = metadata->compression; cstoreOptions->compressionType = metadata->compression;
cstoreOptions->stripeRowCount = metadata->stripeRowCount; cstoreOptions->stripeRowCount = metadata->stripeRowCount;
cstoreOptions->blockRowCount = metadata->blockRowCount; cstoreOptions->blockRowCount = metadata->blockRowCount;
@ -213,14 +210,13 @@ RelationColumnList(Relation rel)
int32 vartypmod = tupdesc->attrs[i].atttypmod; int32 vartypmod = tupdesc->attrs[i].atttypmod;
Oid varcollid = tupdesc->attrs[i].attcollation; Oid varcollid = tupdesc->attrs[i].attcollation;
Index varlevelsup = 0; Index varlevelsup = 0;
Var *var;
if (tupdesc->attrs[i].attisdropped) if (tupdesc->attrs[i].attisdropped)
{ {
continue; continue;
} }
var = makeVar(varno, varattno, vartype, vartypmod, Var *var = makeVar(varno, varattno, vartype, vartypmod,
varcollid, varlevelsup); varcollid, varlevelsup);
columnList = lappend(columnList, var); columnList = lappend(columnList, var);
} }
@ -242,7 +238,6 @@ cstore_beginscan(Relation relation, Snapshot snapshot,
ParallelTableScanDesc parallel_scan, ParallelTableScanDesc parallel_scan,
uint32 flags) uint32 flags)
{ {
TableScanDesc scandesc;
int natts = relation->rd_att->natts; int natts = relation->rd_att->natts;
Bitmapset *attr_needed = NULL; Bitmapset *attr_needed = NULL;
@ -251,7 +246,8 @@ cstore_beginscan(Relation relation, Snapshot snapshot,
/* the cstore access method does not use the flags, they are specific to heap */ /* the cstore access method does not use the flags, they are specific to heap */
flags = 0; flags = 0;
scandesc = cstore_beginscan_extended(relation, snapshot, nkeys, key, parallel_scan, TableScanDesc scandesc = cstore_beginscan_extended(relation, snapshot, nkeys, key,
parallel_scan,
flags, attr_needed, NULL); flags, attr_needed, NULL);
pfree(attr_needed); pfree(attr_needed);
@ -267,9 +263,7 @@ cstore_beginscan_extended(Relation relation, Snapshot snapshot,
uint32 flags, Bitmapset *attr_needed, List *scanQual) uint32 flags, Bitmapset *attr_needed, List *scanQual)
{ {
TupleDesc tupdesc = relation->rd_att; TupleDesc tupdesc = relation->rd_att;
TableReadState *readState = NULL;
CStoreScanDesc scan = palloc(sizeof(CStoreScanDescData)); CStoreScanDesc scan = palloc(sizeof(CStoreScanDescData));
List *columnList = NIL;
List *neededColumnList = NIL; List *neededColumnList = NIL;
MemoryContext oldContext = MemoryContextSwitchTo(GetCStoreMemoryContext()); MemoryContext oldContext = MemoryContextSwitchTo(GetCStoreMemoryContext());
ListCell *columnCell = NULL; ListCell *columnCell = NULL;
@ -281,7 +275,7 @@ cstore_beginscan_extended(Relation relation, Snapshot snapshot,
scan->cs_base.rs_flags = flags; scan->cs_base.rs_flags = flags;
scan->cs_base.rs_parallel = parallel_scan; scan->cs_base.rs_parallel = parallel_scan;
columnList = RelationColumnList(relation); List *columnList = RelationColumnList(relation);
/* only collect columns that we need for the scan */ /* only collect columns that we need for the scan */
foreach(columnCell, columnList) foreach(columnCell, columnList)
@ -293,7 +287,8 @@ cstore_beginscan_extended(Relation relation, Snapshot snapshot,
} }
} }
readState = CStoreBeginRead(relation, tupdesc, neededColumnList, scanQual); TableReadState *readState = CStoreBeginRead(relation, tupdesc, neededColumnList,
scanQual);
scan->cs_readState = readState; scan->cs_readState = readState;
@ -324,12 +319,11 @@ static bool
cstore_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot) cstore_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot)
{ {
CStoreScanDesc scan = (CStoreScanDesc) sscan; CStoreScanDesc scan = (CStoreScanDesc) sscan;
bool nextRowFound;
MemoryContext oldContext = MemoryContextSwitchTo(GetCStoreMemoryContext()); MemoryContext oldContext = MemoryContextSwitchTo(GetCStoreMemoryContext());
ExecClearTuple(slot); ExecClearTuple(slot);
nextRowFound = CStoreReadNextRow(scan->cs_readState, slot->tts_values, bool nextRowFound = CStoreReadNextRow(scan->cs_readState, slot->tts_values,
slot->tts_isnull); slot->tts_isnull);
MemoryContextSwitchTo(oldContext); MemoryContextSwitchTo(oldContext);
@ -443,12 +437,11 @@ static void
cstore_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid, cstore_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
int options, BulkInsertState bistate) int options, BulkInsertState bistate)
{ {
HeapTuple heapTuple;
MemoryContext oldContext = MemoryContextSwitchTo(GetCStoreMemoryContext()); MemoryContext oldContext = MemoryContextSwitchTo(GetCStoreMemoryContext());
cstore_init_write_state(relation); cstore_init_write_state(relation);
heapTuple = ExecCopySlotHeapTuple(slot); HeapTuple heapTuple = ExecCopySlotHeapTuple(slot);
if (HeapTupleHasExternal(heapTuple)) if (HeapTupleHasExternal(heapTuple))
{ {
/* detoast any toasted attributes */ /* detoast any toasted attributes */
@ -559,7 +552,6 @@ cstore_relation_set_new_filenode(Relation rel,
TransactionId *freezeXid, TransactionId *freezeXid,
MultiXactId *minmulti) MultiXactId *minmulti)
{ {
SMgrRelation srel;
DataFileMetadata *metadata = ReadDataFileMetadata(rel->rd_node.relNode, true); DataFileMetadata *metadata = ReadDataFileMetadata(rel->rd_node.relNode, true);
uint64 blockRowCount = 0; uint64 blockRowCount = 0;
uint64 stripeRowCount = 0; uint64 stripeRowCount = 0;
@ -587,7 +579,7 @@ cstore_relation_set_new_filenode(Relation rel,
Assert(persistence == RELPERSISTENCE_PERMANENT); Assert(persistence == RELPERSISTENCE_PERMANENT);
*freezeXid = RecentXmin; *freezeXid = RecentXmin;
*minmulti = GetOldestMultiXactId(); *minmulti = GetOldestMultiXactId();
srel = RelationCreateStorage(*newrnode, persistence); SMgrRelation srel = RelationCreateStorage(*newrnode, persistence);
InitCStoreDataFileMetadata(newrnode->relNode, blockRowCount, stripeRowCount, InitCStoreDataFileMetadata(newrnode->relNode, blockRowCount, stripeRowCount,
compression); compression);
smgrclose(srel); smgrclose(srel);
@ -639,11 +631,6 @@ cstore_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
double *tups_vacuumed, double *tups_vacuumed,
double *tups_recently_dead) double *tups_recently_dead)
{ {
TableWriteState *writeState = NULL;
TableReadState *readState = NULL;
CStoreOptions *cstoreOptions = NULL;
Datum *values = NULL;
bool *nulls = NULL;
TupleDesc sourceDesc = RelationGetDescr(OldHeap); TupleDesc sourceDesc = RelationGetDescr(OldHeap);
TupleDesc targetDesc = RelationGetDescr(NewHeap); TupleDesc targetDesc = RelationGetDescr(NewHeap);
@ -664,7 +651,7 @@ cstore_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
* relation first. * relation first.
*/ */
cstoreOptions = CStoreTableAMGetOptions(OldHeap); CStoreOptions *cstoreOptions = CStoreTableAMGetOptions(OldHeap);
UpdateCStoreDataFileMetadata(NewHeap->rd_node.relNode, UpdateCStoreDataFileMetadata(NewHeap->rd_node.relNode,
cstoreOptions->blockRowCount, cstoreOptions->blockRowCount,
@ -673,16 +660,17 @@ cstore_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
cstoreOptions = CStoreTableAMGetOptions(NewHeap); cstoreOptions = CStoreTableAMGetOptions(NewHeap);
writeState = CStoreBeginWrite(NewHeap, TableWriteState *writeState = CStoreBeginWrite(NewHeap,
cstoreOptions->compressionType, cstoreOptions->compressionType,
cstoreOptions->stripeRowCount, cstoreOptions->stripeRowCount,
cstoreOptions->blockRowCount, cstoreOptions->blockRowCount,
targetDesc); targetDesc);
readState = CStoreBeginRead(OldHeap, sourceDesc, RelationColumnList(OldHeap), NULL); TableReadState *readState = CStoreBeginRead(OldHeap, sourceDesc, RelationColumnList(
OldHeap), NULL);
values = palloc0(sourceDesc->natts * sizeof(Datum)); Datum *values = palloc0(sourceDesc->natts * sizeof(Datum));
nulls = palloc0(sourceDesc->natts * sizeof(bool)); bool *nulls = palloc0(sourceDesc->natts * sizeof(bool));
*num_tuples = 0; *num_tuples = 0;
@ -727,7 +715,6 @@ cstore_vacuum_rel(Relation rel, VacuumParams *params,
static void static void
LogRelationStats(Relation rel, int elevel) LogRelationStats(Relation rel, int elevel)
{ {
DataFileMetadata *datafileMetadata = NULL;
ListCell *stripeMetadataCell = NULL; ListCell *stripeMetadataCell = NULL;
Oid relfilenode = rel->rd_node.relNode; Oid relfilenode = rel->rd_node.relNode;
StringInfo infoBuf = makeStringInfo(); StringInfo infoBuf = makeStringInfo();
@ -736,13 +723,11 @@ LogRelationStats(Relation rel, int elevel)
uint64 totalStripeLength = 0; uint64 totalStripeLength = 0;
uint64 tupleCount = 0; uint64 tupleCount = 0;
uint64 blockCount = 0; uint64 blockCount = 0;
uint64 relPages = 0;
int stripeCount = 0;
TupleDesc tupdesc = RelationGetDescr(rel); TupleDesc tupdesc = RelationGetDescr(rel);
uint64 droppedBlocksWithData = 0; uint64 droppedBlocksWithData = 0;
datafileMetadata = ReadDataFileMetadata(relfilenode, false); DataFileMetadata *datafileMetadata = ReadDataFileMetadata(relfilenode, false);
stripeCount = list_length(datafileMetadata->stripeMetadataList); int stripeCount = list_length(datafileMetadata->stripeMetadataList);
foreach(stripeMetadataCell, datafileMetadata->stripeMetadataList) foreach(stripeMetadataCell, datafileMetadata->stripeMetadataList)
{ {
@ -777,7 +762,7 @@ LogRelationStats(Relation rel, int elevel)
} }
RelationOpenSmgr(rel); RelationOpenSmgr(rel);
relPages = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM); uint64 relPages = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM);
RelationCloseSmgr(rel); RelationCloseSmgr(rel);
appendStringInfo(infoBuf, "total file size: %ld, total data size: %ld\n", appendStringInfo(infoBuf, "total file size: %ld, total data size: %ld\n",
@ -815,9 +800,6 @@ static void
TruncateCStore(Relation rel, int elevel) TruncateCStore(Relation rel, int elevel)
{ {
PGRUsage ru0; PGRUsage ru0;
BlockNumber old_rel_pages = 0;
BlockNumber new_rel_pages = 0;
SmgrAddr highestPhysicalAddress;
pg_rusage_init(&ru0); pg_rusage_init(&ru0);
@ -851,7 +833,7 @@ TruncateCStore(Relation rel, int elevel)
} }
RelationOpenSmgr(rel); RelationOpenSmgr(rel);
old_rel_pages = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM); BlockNumber old_rel_pages = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM);
RelationCloseSmgr(rel); RelationCloseSmgr(rel);
/* /*
@ -859,10 +841,10 @@ TruncateCStore(Relation rel, int elevel)
* new stripes be added beyond highestPhysicalAddress while * new stripes be added beyond highestPhysicalAddress while
* we're truncating. * we're truncating.
*/ */
highestPhysicalAddress = SmgrAddr highestPhysicalAddress =
logical_to_smgr(GetHighestUsedAddress(rel->rd_node.relNode)); logical_to_smgr(GetHighestUsedAddress(rel->rd_node.relNode));
new_rel_pages = highestPhysicalAddress.blockno + 1; BlockNumber new_rel_pages = highestPhysicalAddress.blockno + 1;
if (new_rel_pages == old_rel_pages) if (new_rel_pages == old_rel_pages)
{ {
UnlockRelation(rel, AccessExclusiveLock); UnlockRelation(rel, AccessExclusiveLock);
@ -1104,11 +1086,9 @@ CStoreTableAMProcessUtility(PlannedStmt * plannedStatement,
if (nodeTag(parseTree) == T_CreateTrigStmt) if (nodeTag(parseTree) == T_CreateTrigStmt)
{ {
CreateTrigStmt *createTrigStmt = (CreateTrigStmt *) parseTree; CreateTrigStmt *createTrigStmt = (CreateTrigStmt *) parseTree;
Relation rel;
bool isCStore;
rel = relation_openrv(createTrigStmt->relation, AccessShareLock); Relation rel = relation_openrv(createTrigStmt->relation, AccessShareLock);
isCStore = rel->rd_tableam == GetCstoreTableAmRoutine(); bool isCStore = rel->rd_tableam == GetCstoreTableAmRoutine();
relation_close(rel, AccessShareLock); relation_close(rel, AccessShareLock);
if (isCStore && if (isCStore &&
@ -1201,9 +1181,6 @@ CStoreTableAMObjectAccessHook(ObjectAccessType access, Oid classId, Oid objectId
static bool static bool
IsCStoreTableAmTable(Oid relationId) IsCStoreTableAmTable(Oid relationId)
{ {
bool result;
Relation rel;
if (!OidIsValid(relationId)) if (!OidIsValid(relationId))
{ {
return false; return false;
@ -1213,8 +1190,8 @@ IsCStoreTableAmTable(Oid relationId)
* Lock relation to prevent it from being dropped & * Lock relation to prevent it from being dropped &
* avoid race conditions. * avoid race conditions.
*/ */
rel = relation_open(relationId, AccessShareLock); Relation rel = relation_open(relationId, AccessShareLock);
result = rel->rd_tableam == GetCstoreTableAmRoutine(); bool result = rel->rd_tableam == GetCstoreTableAmRoutine();
relation_close(rel, NoLock); relation_close(rel, NoLock);
return result; return result;
@ -1317,9 +1294,6 @@ Datum
alter_cstore_table_set(PG_FUNCTION_ARGS) alter_cstore_table_set(PG_FUNCTION_ARGS)
{ {
Oid relationId = PG_GETARG_OID(0); Oid relationId = PG_GETARG_OID(0);
int blockRowCount = 0;
int stripeRowCount = 0;
CompressionType compression = COMPRESSION_TYPE_INVALID;
Relation rel = table_open(relationId, AccessExclusiveLock); /* ALTER TABLE LOCK */ Relation rel = table_open(relationId, AccessExclusiveLock); /* ALTER TABLE LOCK */
DataFileMetadata *metadata = ReadDataFileMetadata(rel->rd_node.relNode, true); DataFileMetadata *metadata = ReadDataFileMetadata(rel->rd_node.relNode, true);
@ -1329,9 +1303,9 @@ alter_cstore_table_set(PG_FUNCTION_ARGS)
quote_identifier(RelationGetRelationName(rel))))); quote_identifier(RelationGetRelationName(rel)))));
} }
blockRowCount = metadata->blockRowCount; int blockRowCount = metadata->blockRowCount;
stripeRowCount = metadata->stripeRowCount; int stripeRowCount = metadata->stripeRowCount;
compression = metadata->compression; CompressionType compression = metadata->compression;
/* block_row_count => not null */ /* block_row_count => not null */
if (!PG_ARGISNULL(1)) if (!PG_ARGISNULL(1))
@ -1375,9 +1349,6 @@ Datum
alter_cstore_table_reset(PG_FUNCTION_ARGS) alter_cstore_table_reset(PG_FUNCTION_ARGS)
{ {
Oid relationId = PG_GETARG_OID(0); Oid relationId = PG_GETARG_OID(0);
int blockRowCount = 0;
int stripeRowCount = 0;
CompressionType compression = COMPRESSION_TYPE_INVALID;
Relation rel = table_open(relationId, AccessExclusiveLock); /* ALTER TABLE LOCK */ Relation rel = table_open(relationId, AccessExclusiveLock); /* ALTER TABLE LOCK */
DataFileMetadata *metadata = ReadDataFileMetadata(rel->rd_node.relNode, true); DataFileMetadata *metadata = ReadDataFileMetadata(rel->rd_node.relNode, true);
@ -1387,9 +1358,9 @@ alter_cstore_table_reset(PG_FUNCTION_ARGS)
quote_identifier(RelationGetRelationName(rel))))); quote_identifier(RelationGetRelationName(rel)))));
} }
blockRowCount = metadata->blockRowCount; int blockRowCount = metadata->blockRowCount;
stripeRowCount = metadata->stripeRowCount; int stripeRowCount = metadata->stripeRowCount;
compression = metadata->compression; CompressionType compression = metadata->compression;
/* block_row_count => true */ /* block_row_count => true */
if (!PG_ARGISNULL(1) && PG_GETARG_BOOL(1)) if (!PG_ARGISNULL(1) && PG_GETARG_BOOL(1))

62
src/backend/columnar/cstore_writer.c
View File

@ -61,17 +61,11 @@ CStoreBeginWrite(Relation relation,
uint64 stripeMaxRowCount, uint32 blockRowCount, uint64 stripeMaxRowCount, uint32 blockRowCount,
TupleDesc tupleDescriptor) TupleDesc tupleDescriptor)
{ {
TableWriteState *writeState = NULL;
FmgrInfo **comparisonFunctionArray = NULL;
MemoryContext stripeWriteContext = NULL;
uint32 columnCount = 0;
uint32 columnIndex = 0; uint32 columnIndex = 0;
bool *columnMaskArray = NULL;
BlockData *blockData = NULL;
/* get comparison function pointers for each of the columns */ /* get comparison function pointers for each of the columns */
columnCount = tupleDescriptor->natts; uint32 columnCount = tupleDescriptor->natts;
comparisonFunctionArray = palloc0(columnCount * sizeof(FmgrInfo *)); FmgrInfo **comparisonFunctionArray = palloc0(columnCount * sizeof(FmgrInfo *));
for (columnIndex = 0; columnIndex < columnCount; columnIndex++) for (columnIndex = 0; columnIndex < columnCount; columnIndex++)
{ {
FmgrInfo *comparisonFunction = NULL; FmgrInfo *comparisonFunction = NULL;
@ -94,16 +88,17 @@ CStoreBeginWrite(Relation relation,
* reset this memory context once we have flushed the stripe to the file. * reset this memory context once we have flushed the stripe to the file.
* This is to avoid memory leaks. * This is to avoid memory leaks.
*/ */
stripeWriteContext = AllocSetContextCreate(CurrentMemoryContext, MemoryContext stripeWriteContext = AllocSetContextCreate(CurrentMemoryContext,
"Stripe Write Memory Context", "Stripe Write Memory Context",
ALLOCSET_DEFAULT_SIZES); ALLOCSET_DEFAULT_SIZES);
columnMaskArray = palloc(columnCount * sizeof(bool)); bool *columnMaskArray = palloc(columnCount * sizeof(bool));
memset(columnMaskArray, true, columnCount); memset(columnMaskArray, true, columnCount);
blockData = CreateEmptyBlockData(columnCount, columnMaskArray, blockRowCount); BlockData *blockData = CreateEmptyBlockData(columnCount, columnMaskArray,
blockRowCount);
writeState = palloc0(sizeof(TableWriteState)); TableWriteState *writeState = palloc0(sizeof(TableWriteState));
writeState->relation = relation; writeState->relation = relation;
writeState->compressionType = compressionType; writeState->compressionType = compressionType;
writeState->stripeMaxRowCount = stripeMaxRowCount; writeState->stripeMaxRowCount = stripeMaxRowCount;
@ -132,8 +127,6 @@ void
CStoreWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNulls) CStoreWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNulls)
{ {
uint32 columnIndex = 0; uint32 columnIndex = 0;
uint32 blockIndex = 0;
uint32 blockRowIndex = 0;
StripeBuffers *stripeBuffers = writeState->stripeBuffers; StripeBuffers *stripeBuffers = writeState->stripeBuffers;
StripeSkipList *stripeSkipList = writeState->stripeSkipList; StripeSkipList *stripeSkipList = writeState->stripeSkipList;
uint32 columnCount = writeState->tupleDescriptor->natts; uint32 columnCount = writeState->tupleDescriptor->natts;
@ -161,8 +154,8 @@ CStoreWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNul
} }
} }
blockIndex = stripeBuffers->rowCount / blockRowCount; uint32 blockIndex = stripeBuffers->rowCount / blockRowCount;
blockRowIndex = stripeBuffers->rowCount % blockRowCount; uint32 blockRowIndex = stripeBuffers->rowCount % blockRowCount;
for (columnIndex = 0; columnIndex < columnCount; columnIndex++) for (columnIndex = 0; columnIndex < columnCount; columnIndex++)
{ {
@ -257,7 +250,6 @@ static StripeBuffers *
CreateEmptyStripeBuffers(uint32 stripeMaxRowCount, uint32 blockRowCount, CreateEmptyStripeBuffers(uint32 stripeMaxRowCount, uint32 blockRowCount,
uint32 columnCount) uint32 columnCount)
{ {
StripeBuffers *stripeBuffers = NULL;
uint32 columnIndex = 0; uint32 columnIndex = 0;
uint32 maxBlockCount = (stripeMaxRowCount / blockRowCount) + 1; uint32 maxBlockCount = (stripeMaxRowCount / blockRowCount) + 1;
ColumnBuffers **columnBuffersArray = palloc0(columnCount * sizeof(ColumnBuffers *)); ColumnBuffers **columnBuffersArray = palloc0(columnCount * sizeof(ColumnBuffers *));
@ -280,7 +272,7 @@ CreateEmptyStripeBuffers(uint32 stripeMaxRowCount, uint32 blockRowCount,
columnBuffersArray[columnIndex]->blockBuffersArray = blockBuffersArray; columnBuffersArray[columnIndex]->blockBuffersArray = blockBuffersArray;
} }
stripeBuffers = palloc0(sizeof(StripeBuffers)); StripeBuffers *stripeBuffers = palloc0(sizeof(StripeBuffers));
stripeBuffers->columnBuffersArray = columnBuffersArray; stripeBuffers->columnBuffersArray = columnBuffersArray;
stripeBuffers->columnCount = columnCount; stripeBuffers->columnCount = columnCount;
stripeBuffers->rowCount = 0; stripeBuffers->rowCount = 0;
@ -298,7 +290,6 @@ static StripeSkipList *
CreateEmptyStripeSkipList(uint32 stripeMaxRowCount, uint32 blockRowCount, CreateEmptyStripeSkipList(uint32 stripeMaxRowCount, uint32 blockRowCount,
uint32 columnCount) uint32 columnCount)
{ {
StripeSkipList *stripeSkipList = NULL;
uint32 columnIndex = 0; uint32 columnIndex = 0;
uint32 maxBlockCount = (stripeMaxRowCount / blockRowCount) + 1; uint32 maxBlockCount = (stripeMaxRowCount / blockRowCount) + 1;
@ -310,7 +301,7 @@ CreateEmptyStripeSkipList(uint32 stripeMaxRowCount, uint32 blockRowCount,
palloc0(maxBlockCount * sizeof(ColumnBlockSkipNode)); palloc0(maxBlockCount * sizeof(ColumnBlockSkipNode));
} }
stripeSkipList = palloc0(sizeof(StripeSkipList)); StripeSkipList *stripeSkipList = palloc0(sizeof(StripeSkipList));
stripeSkipList->columnCount = columnCount; stripeSkipList->columnCount = columnCount;
stripeSkipList->blockCount = 0; stripeSkipList->blockCount = 0;
stripeSkipList->blockSkipNodeArray = blockSkipNodeArray; stripeSkipList->blockSkipNodeArray = blockSkipNodeArray;
@ -328,13 +319,9 @@ WriteToSmgr(Relation rel, uint64 logicalOffset, char *data, uint32 dataLength)
while (remaining > 0) while (remaining > 0)
{ {
SmgrAddr addr = logical_to_smgr(logicalOffset); SmgrAddr addr = logical_to_smgr(logicalOffset);
BlockNumber nblocks;
Page page;
PageHeader phdr;
uint64 to_write;
RelationOpenSmgr(rel); RelationOpenSmgr(rel);
nblocks = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM); BlockNumber nblocks = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM);
Assert(addr.blockno < nblocks); Assert(addr.blockno < nblocks);
(void) nblocks; /* keep compiler quiet */ (void) nblocks; /* keep compiler quiet */
RelationCloseSmgr(rel); RelationCloseSmgr(rel);
@ -342,8 +329,8 @@ WriteToSmgr(Relation rel, uint64 logicalOffset, char *data, uint32 dataLength)
buffer = ReadBuffer(rel, addr.blockno); buffer = ReadBuffer(rel, addr.blockno);
LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE); LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
page = BufferGetPage(buffer); Page page = BufferGetPage(buffer);
phdr = (PageHeader) page; PageHeader phdr = (PageHeader) page;
if (PageIsNew(page)) if (PageIsNew(page))
{ {
PageInit(page, BLCKSZ, 0); PageInit(page, BLCKSZ, 0);
@ -366,7 +353,7 @@ WriteToSmgr(Relation rel, uint64 logicalOffset, char *data, uint32 dataLength)
START_CRIT_SECTION(); START_CRIT_SECTION();
to_write = Min(phdr->pd_upper - phdr->pd_lower, remaining); uint64 to_write = Min(phdr->pd_upper - phdr->pd_lower, remaining);
memcpy(page + phdr->pd_lower, data, to_write); memcpy(page + phdr->pd_lower, data, to_write);
phdr->pd_lower += to_write; phdr->pd_lower += to_write;
@ -374,8 +361,6 @@ WriteToSmgr(Relation rel, uint64 logicalOffset, char *data, uint32 dataLength)
if (RelationNeedsWAL(rel)) if (RelationNeedsWAL(rel))
{ {
XLogRecPtr recptr = 0;
XLogBeginInsert(); XLogBeginInsert();
/* /*
@ -384,7 +369,7 @@ WriteToSmgr(Relation rel, uint64 logicalOffset, char *data, uint32 dataLength)
*/ */
XLogRegisterBuffer(0, buffer, REGBUF_FORCE_IMAGE); XLogRegisterBuffer(0, buffer, REGBUF_FORCE_IMAGE);
recptr = XLogInsert(RM_GENERIC_ID, 0); XLogRecPtr recptr = XLogInsert(RM_GENERIC_ID, 0);
PageSetLSN(page, recptr); PageSetLSN(page, recptr);
} }
@ -420,7 +405,6 @@ FlushStripe(TableWriteState *writeState)
uint32 blockRowCount = writeState->blockRowCount; uint32 blockRowCount = writeState->blockRowCount;
uint32 lastBlockIndex = stripeBuffers->rowCount / blockRowCount; uint32 lastBlockIndex = stripeBuffers->rowCount / blockRowCount;
uint32 lastBlockRowCount = stripeBuffers->rowCount % blockRowCount; uint32 lastBlockRowCount = stripeBuffers->rowCount % blockRowCount;
uint64 currentFileOffset = 0;
uint64 stripeSize = 0; uint64 stripeSize = 0;
uint64 stripeRowCount = 0; uint64 stripeRowCount = 0;
@ -477,7 +461,7 @@ FlushStripe(TableWriteState *writeState)
stripeRowCount, columnCount, blockCount, stripeRowCount, columnCount, blockCount,
blockRowCount); blockRowCount);
currentFileOffset = stripeMetadata.fileOffset; uint64 currentFileOffset = stripeMetadata.fileOffset;
/* /*
* Each stripe has only one section: * Each stripe has only one section:
@ -531,11 +515,10 @@ FlushStripe(TableWriteState *writeState)
static StringInfo static StringInfo
SerializeBoolArray(bool *boolArray, uint32 boolArrayLength) SerializeBoolArray(bool *boolArray, uint32 boolArrayLength)
{ {
StringInfo boolArrayBuffer = NULL;
uint32 boolArrayIndex = 0; uint32 boolArrayIndex = 0;
uint32 byteCount = (boolArrayLength + 7) / 8; uint32 byteCount = (boolArrayLength + 7) / 8;
boolArrayBuffer = makeStringInfo(); StringInfo boolArrayBuffer = makeStringInfo();
enlargeStringInfo(boolArrayBuffer, byteCount); enlargeStringInfo(boolArrayBuffer, byteCount);
boolArrayBuffer->len = byteCount; boolArrayBuffer->len = byteCount;
memset(boolArrayBuffer->data, 0, byteCount); memset(boolArrayBuffer->data, 0, byteCount);
@ -564,11 +547,10 @@ SerializeSingleDatum(StringInfo datumBuffer, Datum datum, bool datumTypeByValue,
{ {
uint32 datumLength = att_addlength_datum(0, datumTypeLength, datum); uint32 datumLength = att_addlength_datum(0, datumTypeLength, datum);
uint32 datumLengthAligned = att_align_nominal(datumLength, datumTypeAlign); uint32 datumLengthAligned = att_align_nominal(datumLength, datumTypeAlign);
char *currentDatumDataPointer = NULL;
enlargeStringInfo(datumBuffer, datumLengthAligned); enlargeStringInfo(datumBuffer, datumLengthAligned);
currentDatumDataPointer = datumBuffer->data + datumBuffer->len; char *currentDatumDataPointer = datumBuffer->data + datumBuffer->len;
memset(currentDatumDataPointer, 0, datumLengthAligned); memset(currentDatumDataPointer, 0, datumLengthAligned);
if (datumTypeLength > 0) if (datumTypeLength > 0)
@ -624,11 +606,9 @@ SerializeBlockData(TableWriteState *writeState, uint32 blockIndex, uint32 rowCou
{ {
ColumnBuffers *columnBuffers = stripeBuffers->columnBuffersArray[columnIndex]; ColumnBuffers *columnBuffers = stripeBuffers->columnBuffersArray[columnIndex];
ColumnBlockBuffers *blockBuffers = columnBuffers->blockBuffersArray[blockIndex]; ColumnBlockBuffers *blockBuffers = columnBuffers->blockBuffersArray[blockIndex];
StringInfo serializedValueBuffer = NULL;
CompressionType actualCompressionType = COMPRESSION_NONE; CompressionType actualCompressionType = COMPRESSION_NONE;
bool compressed = false;
serializedValueBuffer = blockData->valueBufferArray[columnIndex]; StringInfo serializedValueBuffer = blockData->valueBufferArray[columnIndex];
/* the only other supported compression type is pg_lz for now */ /* the only other supported compression type is pg_lz for now */
Assert(requestedCompressionType == COMPRESSION_NONE || Assert(requestedCompressionType == COMPRESSION_NONE ||
@ -638,7 +618,7 @@ SerializeBlockData(TableWriteState *writeState, uint32 blockIndex, uint32 rowCou
* if serializedValueBuffer is be compressed, update serializedValueBuffer * if serializedValueBuffer is be compressed, update serializedValueBuffer
* with compressed data and store compression type. * with compressed data and store compression type.
*/ */
compressed = CompressBuffer(serializedValueBuffer, compressionBuffer, bool compressed = CompressBuffer(serializedValueBuffer, compressionBuffer,
requestedCompressionType); requestedCompressionType);
if (compressed) if (compressed)
{ {