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

Merge pull request #4667 from citusdata/private-structs 

Columnar: make read and write state private.
pull/4666/head
Hadi Moshayedi 2021-02-08 10:24:02 -08:00 committed by GitHub
parent 353b080474 2ea31c899e
commit 1d3b866df5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 86 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -490,7 +490,7 @@ ColumnarScan_ExplainCustomScan(CustomScanState *node, List *ancestors,
if (scanDesc != NULL)
{
int64 chunkGroupsFiltered = ColumnarGetChunkGroupsFiltered(scanDesc);
int64 chunkGroupsFiltered = ColumnarScanChunkGroupsFiltered(scanDesc);
ExplainPropertyInteger("Columnar Chunk Groups Removed by Filter", NULL,
chunkGroupsFiltered, es);
}

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

@ -39,6 +39,30 @@
#include "columnar/columnar.h"
#include "columnar/columnar_version_compat.h"
struct TableReadState
{
List *stripeList;
StripeMetadata *currentStripeMetadata;
TupleDesc tupleDescriptor;
Relation relation;
/*
* List of Var pointers for columns in the query. We use this both for
* getting vector of projected columns, and also when we want to build
* base constraint to find selected row chunks.
*/
List *projectedColumnList;
List *whereClauseList;
MemoryContext stripeReadContext;
StripeBuffers *stripeBuffers;
uint32 readStripeCount;
uint64 stripeReadRowCount;
int64 chunkGroupsFiltered;
ChunkData *chunkData;
int32 deserializedChunkIndex;
};
/* static function declarations */
static StripeBuffers * LoadFilteredStripeBuffers(Relation relation,
StripeMetadata *stripeMetadata,
@ -235,6 +259,18 @@ ColumnarEndRead(TableReadState *readState)
}
/*
* ColumnarReadChunkGroupsFiltered
*
* Return the number of chunk groups filtered during this read operation.
*/
int64
ColumnarReadChunkGroupsFiltered(TableReadState *state)
{
return state->chunkGroupsFiltered;
}
/*
* CreateEmptyChunkDataArray creates data buffers to keep deserialized exist and
* value arrays for requested columns in columnMask.

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

@ -443,7 +443,8 @@ columnar_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
TableWriteState *writeState = columnar_init_write_state(relation,
RelationGetDescr(relation),
GetCurrentSubTransactionId());
MemoryContext oldContext = MemoryContextSwitchTo(writeState->perTupleContext);
MemoryContext oldContext = MemoryContextSwitchTo(ColumnarWritePerTupleContext(
writeState));
ColumnarCheckLogicalReplication(relation);
@ -455,7 +456,7 @@ columnar_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
ColumnarWriteRow(writeState, values, slot->tts_isnull);
MemoryContextSwitchTo(oldContext);
MemoryContextReset(writeState->perTupleContext);
MemoryContextReset(ColumnarWritePerTupleContext(writeState));
}
@ -486,7 +487,8 @@ columnar_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
ColumnarCheckLogicalReplication(relation);
MemoryContext oldContext = MemoryContextSwitchTo(writeState->perTupleContext);
MemoryContext oldContext = MemoryContextSwitchTo(ColumnarWritePerTupleContext(
writeState));
for (int i = 0; i < ntuples; i++)
{
@ -498,7 +500,7 @@ columnar_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
tupleSlot->tts_values, tupleSlot->tts_isnull);
ColumnarWriteRow(writeState, values, tupleSlot->tts_isnull);
MemoryContextReset(writeState->perTupleContext);
MemoryContextReset(ColumnarWritePerTupleContext(writeState));
}
MemoryContextSwitchTo(oldContext);
@ -1158,7 +1160,7 @@ columnar_tableam_finish()
* Get the number of chunks filtered out during the given scan.
*/
int64
ColumnarGetChunkGroupsFiltered(TableScanDesc scanDesc)
ColumnarScanChunkGroupsFiltered(TableScanDesc scanDesc)
{
ColumnarScanDesc columnarScanDesc = (ColumnarScanDesc) scanDesc;
TableReadState *readState = columnarScanDesc->cs_readState;
@ -1166,7 +1168,7 @@ ColumnarGetChunkGroupsFiltered(TableScanDesc scanDesc)
/* readState is initialized lazily */
if (readState != NULL)
{
return readState->chunkGroupsFiltered;
return ColumnarReadChunkGroupsFiltered(readState);
}
else
{

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

@ -32,6 +32,28 @@
#include "columnar/columnar.h"
#include "columnar/columnar_version_compat.h"
struct TableWriteState
{
TupleDesc tupleDescriptor;
FmgrInfo **comparisonFunctionArray;
RelFileNode relfilenode;
MemoryContext stripeWriteContext;
MemoryContext perTupleContext;
StripeBuffers *stripeBuffers;
StripeSkipList *stripeSkipList;
ColumnarOptions options;
ChunkData *chunkData;
/*
* compressionBuffer buffer is used as temporary storage during
* data value compression operation. It is kept here to minimize
* memory allocations. It lives in stripeWriteContext and gets
* deallocated when memory context is reset.
*/
StringInfo compressionBuffer;
};
static StripeBuffers * CreateEmptyStripeBuffers(uint32 stripeMaxRowCount,
uint32 chunkRowCount,
uint32 columnCount);
@ -247,6 +269,18 @@ ColumnarFlushPendingWrites(TableWriteState *writeState)
}
/*
* ColumnarWritePerTupleContext
*
* Return per-tuple context for columnar write operation.
*/
MemoryContext
ColumnarWritePerTupleContext(TableWriteState *state)
{
return state->perTupleContext;
}
/*
* CreateEmptyStripeBuffers allocates an empty StripeBuffers structure with the given
* column count.

50
src/include/columnar/columnar.h
View File

@ -213,53 +213,13 @@ typedef struct StripeBuffers
/* TableReadState represents state of a columnar scan. */
typedef struct TableReadState
{
List *stripeList;
StripeMetadata *currentStripeMetadata;
TupleDesc tupleDescriptor;
Relation relation;
/*
* List of Var pointers for columns in the query. We use this both for
* getting vector of projected columns, and also when we want to build
* base constraint to find selected row chunks.
*/
List *projectedColumnList;
List *whereClauseList;
MemoryContext stripeReadContext;
StripeBuffers *stripeBuffers;
uint32 readStripeCount;
uint64 stripeReadRowCount;
int64 chunkGroupsFiltered;
ChunkData *chunkData;
int32 deserializedChunkIndex;
} TableReadState;
struct TableReadState;
typedef struct TableReadState TableReadState;
/* TableWriteState represents state of a columnar write operation. */
typedef struct TableWriteState
{
TupleDesc tupleDescriptor;
FmgrInfo **comparisonFunctionArray;
RelFileNode relfilenode;
MemoryContext stripeWriteContext;
MemoryContext perTupleContext;
StripeBuffers *stripeBuffers;
StripeSkipList *stripeSkipList;
ColumnarOptions options;
ChunkData *chunkData;
/*
* compressionBuffer buffer is used as temporary storage during
* data value compression operation. It is kept here to minimize
* memory allocations. It lives in stripeWriteContext and gets
* deallocated when memory context is reset.
*/
StringInfo compressionBuffer;
} TableWriteState;
struct TableWriteState;
typedef struct TableWriteState TableWriteState;
extern int columnar_compression;
extern int columnar_stripe_row_limit;
@ -279,6 +239,7 @@ extern void ColumnarWriteRow(TableWriteState *state, Datum *columnValues,
extern void ColumnarFlushPendingWrites(TableWriteState *state);
extern void ColumnarEndWrite(TableWriteState *state);
extern bool ContainsPendingWrites(TableWriteState *state);
extern MemoryContext ColumnarWritePerTupleContext(TableWriteState *state);
/* Function declarations for reading from columnar table */
extern TableReadState * ColumnarBeginRead(Relation relation,
@ -289,6 +250,7 @@ extern bool ColumnarReadNextRow(TableReadState *state, Datum *columnValues,
bool *columnNulls);
extern void ColumnarRescan(TableReadState *readState);
extern void ColumnarEndRead(TableReadState *state);
extern int64 ColumnarReadChunkGroupsFiltered(TableReadState *state);
/* Function declarations for common functions */
extern FmgrInfo * GetFunctionInfoOrNull(Oid typeId, Oid accessMethodId,

2
src/include/columnar/columnar_tableam.h
View File

@ -18,7 +18,7 @@ extern TableScanDesc columnar_beginscan_extended(Relation relation, Snapshot sna
ParallelTableScanDesc parallel_scan,
uint32 flags, Bitmapset *attr_needed,
List *scanQual);
extern int64 ColumnarGetChunkGroupsFiltered(TableScanDesc scanDesc);
extern int64 ColumnarScanChunkGroupsFiltered(TableScanDesc scanDesc);
extern bool IsColumnarTableAmTable(Oid relationId);
extern TableDDLCommand * ColumnarGetTableOptionsDDL(Oid relationId);
extern char * GetShardedTableDDLCommandColumnar(uint64 shardId, void *context);