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

Merge pull request #3 from citusdata/add_reindent 

Add 'make reindent'
merge-cstore-pykello
Hadi Moshayedi 2020-09-07 15:51:09 -07:00 committed by GitHub
parent 4b1e80a19f b74de68ce3
commit 85a51fb2ef
13 changed files with 143 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
.gitattributes vendored Normal file
View File

@ -0,0 +1,26 @@
* whitespace=space-before-tab,trailing-space
*.[chly] whitespace=space-before-tab,trailing-space,indent-with-non-tab,tabwidth=4
*.dsl whitespace=space-before-tab,trailing-space,tab-in-indent
*.patch -whitespace
*.pl whitespace=space-before-tab,trailing-space,tabwidth=4
*.po whitespace=space-before-tab,trailing-space,tab-in-indent,-blank-at-eof
*.sgml whitespace=space-before-tab,trailing-space,tab-in-indent,-blank-at-eol
*.x[ms]l whitespace=space-before-tab,trailing-space,tab-in-indent
# Avoid confusing ASCII underlines with leftover merge conflict markers
README conflict-marker-size=32
README.* conflict-marker-size=32
# Certain data files that contain special whitespace, and other special cases
*.data -whitespace
# Test output files that contain extra whitespace
*.out -whitespace
src/test/regress/output/*.source -whitespace
# These files are maintained or generated elsewhere. We take them as is.
configure -whitespace
# all C files (implementation and header) use our style...
*.[ch] citus-style

3
Makefile
View File

@ -58,3 +58,6 @@ installcheck: remove_cstore_files
remove_cstore_files: remove_cstore_files:
rm -f data/*.cstore data/*.cstore.footer rm -f data/*.cstore data/*.cstore.footer
reindent:
citus_indent .

8
cstore.c
View File

@ -43,6 +43,7 @@ ParseCompressionType(const char *compressionTypeString)
return compressionType; return compressionType;
} }
/* CreateDirectory creates a new directory with the given directory name. */ /* CreateDirectory creates a new directory with the given directory name. */
static void static void
CreateDirectory(StringInfo directoryName) CreateDirectory(StringInfo directoryName)
@ -56,6 +57,7 @@ CreateDirectory(StringInfo directoryName)
} }
} }
/* DirectoryExists checks if a directory exists for the given directory name. */ /* DirectoryExists checks if a directory exists for the given directory name. */
static bool static bool
DirectoryExists(StringInfo directoryName) DirectoryExists(StringInfo directoryName)
@ -91,6 +93,7 @@ DirectoryExists(StringInfo directoryName)
return directoryExists; return directoryExists;
} }
/* /*
* RemoveCStoreDatabaseDirectory removes CStore directory previously * RemoveCStoreDatabaseDirectory removes CStore directory previously
* created for this database. * created for this database.
@ -132,8 +135,9 @@ InitializeCStoreTableFile(Oid relationId, Relation relation, CStoreOptions *csto
* empty data file and a valid footer file for the table. * empty data file and a valid footer file for the table.
*/ */
writeState = CStoreBeginWrite(relationId, cstoreOptions->filename, writeState = CStoreBeginWrite(relationId, cstoreOptions->filename,
cstoreOptions->compressionType, cstoreOptions->stripeRowCount, cstoreOptions->compressionType,
cstoreOptions->blockRowCount, tupleDescriptor); cstoreOptions->stripeRowCount,
cstoreOptions->blockRowCount, tupleDescriptor);
CStoreEndWrite(writeState); CStoreEndWrite(writeState);
} }

16
cstore.h
View File

@ -60,7 +60,6 @@ typedef enum
COMPRESSION_PG_LZ = 1, COMPRESSION_PG_LZ = 1,
COMPRESSION_COUNT COMPRESSION_COUNT
} CompressionType; } CompressionType;
@ -75,7 +74,6 @@ typedef struct CStoreOptions
CompressionType compressionType; CompressionType compressionType;
uint64 stripeRowCount; uint64 stripeRowCount;
uint32 blockRowCount; uint32 blockRowCount;
} CStoreOptions; } CStoreOptions;
@ -90,7 +88,6 @@ typedef struct StripeMetadata
uint64 dataLength; uint64 dataLength;
uint64 footerLength; uint64 footerLength;
uint64 id; uint64 id;
} StripeMetadata; } StripeMetadata;
@ -99,7 +96,6 @@ typedef struct TableFooter
{ {
List *stripeMetadataList; List *stripeMetadataList;
uint64 blockRowCount; uint64 blockRowCount;
} TableFooter; } TableFooter;
@ -123,7 +119,6 @@ typedef struct ColumnBlockSkipNode
uint64 existsLength; uint64 existsLength;
CompressionType valueCompressionType; CompressionType valueCompressionType;
} ColumnBlockSkipNode; } ColumnBlockSkipNode;
@ -137,7 +132,6 @@ typedef struct StripeSkipList
ColumnBlockSkipNode **blockSkipNodeArray; ColumnBlockSkipNode **blockSkipNodeArray;
uint32 columnCount; uint32 columnCount;
uint32 blockCount; uint32 blockCount;
} StripeSkipList; } StripeSkipList;
@ -155,7 +149,6 @@ typedef struct ColumnBlockData
/* valueBuffer keeps actual data for type-by-reference datums from valueArray. */ /* valueBuffer keeps actual data for type-by-reference datums from valueArray. */
StringInfo valueBuffer; StringInfo valueBuffer;
} ColumnBlockData; } ColumnBlockData;
@ -171,7 +164,6 @@ typedef struct ColumnBlockBuffers
StringInfo existsBuffer; StringInfo existsBuffer;
StringInfo valueBuffer; StringInfo valueBuffer;
CompressionType valueCompressionType; CompressionType valueCompressionType;
} ColumnBlockBuffers; } ColumnBlockBuffers;
@ -182,7 +174,6 @@ typedef struct ColumnBlockBuffers
typedef struct ColumnBuffers typedef struct ColumnBuffers
{ {
ColumnBlockBuffers **blockBuffersArray; ColumnBlockBuffers **blockBuffersArray;
} ColumnBuffers; } ColumnBuffers;
@ -192,7 +183,6 @@ typedef struct StripeBuffers
uint32 columnCount; uint32 columnCount;
uint32 rowCount; uint32 rowCount;
ColumnBuffers **columnBuffersArray; ColumnBuffers **columnBuffersArray;
} StripeBuffers; } StripeBuffers;
@ -207,7 +197,6 @@ typedef struct StripeFooter
uint64 *skipListSizeArray; uint64 *skipListSizeArray;
uint64 *existsSizeArray; uint64 *existsSizeArray;
uint64 *valueSizeArray; uint64 *valueSizeArray;
} StripeFooter; } StripeFooter;
@ -234,7 +223,6 @@ typedef struct TableReadState
uint64 stripeReadRowCount; uint64 stripeReadRowCount;
ColumnBlockData **blockDataArray; ColumnBlockData **blockDataArray;
int32 deserializedBlockIndex; int32 deserializedBlockIndex;
} TableReadState; } TableReadState;
@ -257,6 +245,7 @@ typedef struct TableWriteState
StripeSkipList *stripeSkipList; StripeSkipList *stripeSkipList;
uint32 stripeMaxRowCount; uint32 stripeMaxRowCount;
ColumnBlockData **blockDataArray; ColumnBlockData **blockDataArray;
/* /*
* compressionBuffer buffer is used as temporary storage during * compressionBuffer buffer is used as temporary storage during
* data value compression operation. It is kept here to minimize * data value compression operation. It is kept here to minimize
@ -264,7 +253,6 @@ typedef struct TableWriteState
* deallocated when memory context is reset. * deallocated when memory context is reset.
*/ */
StringInfo compressionBuffer; StringInfo compressionBuffer;
} TableWriteState; } TableWriteState;
extern CompressionType ParseCompressionType(const char *compressionTypeString); extern CompressionType ParseCompressionType(const char *compressionTypeString);
@ -283,7 +271,7 @@ extern TableWriteState * CStoreBeginWrite(Oid relationId,
TupleDesc tupleDescriptor); TupleDesc tupleDescriptor);
extern void CStoreWriteRow(TableWriteState *state, Datum *columnValues, extern void CStoreWriteRow(TableWriteState *state, Datum *columnValues,
bool *columnNulls); bool *columnNulls);
extern void CStoreEndWrite(TableWriteState * state); extern void CStoreEndWrite(TableWriteState *state);
/* Function declarations for reading from a cstore file */ /* Function declarations for reading from a cstore file */
extern TableReadState * CStoreBeginRead(Oid relationId, const char *filename, extern TableReadState * CStoreBeginRead(Oid relationId, const char *filename,

17
cstore_compression.c
View File

@ -22,38 +22,39 @@
#include "cstore.h" #include "cstore.h"
#if PG_VERSION_NUM >= 90500 #if PG_VERSION_NUM >= 90500
/* /*
* The information at the start of the compressed data. This decription is taken * The information at the start of the compressed data. This decription is taken
* from pg_lzcompress in pre-9.5 version of PostgreSQL. * from pg_lzcompress in pre-9.5 version of PostgreSQL.
*/ */
typedef struct CStoreCompressHeader typedef struct CStoreCompressHeader
{ {
int32 vl_len_; /* varlena header (do not touch directly!) */ int32 vl_len_; /* varlena header (do not touch directly!) */
int32 rawsize; int32 rawsize;
} CStoreCompressHeader; } CStoreCompressHeader;
/* /*
* Utilities for manipulation of header information for compressed data * Utilities for manipulation of header information for compressed data
*/ */
#define CSTORE_COMPRESS_HDRSZ ((int32) sizeof(CStoreCompressHeader)) #define CSTORE_COMPRESS_HDRSZ ((int32) sizeof(CStoreCompressHeader))
#define CSTORE_COMPRESS_RAWSIZE(ptr) (((CStoreCompressHeader *) (ptr))->rawsize) #define CSTORE_COMPRESS_RAWSIZE(ptr) (((CStoreCompressHeader *) (ptr))->rawsize)
#define CSTORE_COMPRESS_RAWDATA(ptr) (((char *) (ptr)) + CSTORE_COMPRESS_HDRSZ) #define CSTORE_COMPRESS_RAWDATA(ptr) (((char *) (ptr)) + CSTORE_COMPRESS_HDRSZ)
#define CSTORE_COMPRESS_SET_RAWSIZE(ptr, len) (((CStoreCompressHeader *) (ptr))->rawsize = (len)) #define CSTORE_COMPRESS_SET_RAWSIZE(ptr, len) (((CStoreCompressHeader *) (ptr))->rawsize = \
(len))
#else #else
#define CSTORE_COMPRESS_HDRSZ (0) #define CSTORE_COMPRESS_HDRSZ (0)
#define CSTORE_COMPRESS_RAWSIZE(ptr) (PGLZ_RAW_SIZE((PGLZ_Header *) buffer->data)) #define CSTORE_COMPRESS_RAWSIZE(ptr) (PGLZ_RAW_SIZE((PGLZ_Header *) buffer->data))
#define CSTORE_COMPRESS_RAWDATA(ptr) (((PGLZ_Header *) (ptr))) #define CSTORE_COMPRESS_RAWDATA(ptr) (((PGLZ_Header *) (ptr)))
#define CSTORE_COMPRESS_SET_RAWSIZE(ptr, len) (((CStoreCompressHeader *) (ptr))->rawsize = (len)) #define CSTORE_COMPRESS_SET_RAWSIZE(ptr, len) (((CStoreCompressHeader *) (ptr))->rawsize = \
(len))
#endif #endif
/* /*
* CompressBuffer compresses the given buffer with the given compression type * CompressBuffer compresses the given buffer with the given compression type
* outputBuffer enlarged to contain compressed data. The function returns true * outputBuffer enlarged to contain compressed data. The function returns true

112
cstore_fdw.c
View File

@ -85,7 +85,6 @@ typedef struct CStoreValidOption
{ {
const char *optionName; const char *optionName;
Oid optionContextId; Oid optionContextId;
} CStoreValidOption; } CStoreValidOption;
#define COMPRESSION_STRING_DELIMITED_LIST "none, pglz" #define COMPRESSION_STRING_DELIMITED_LIST "none, pglz"
@ -114,13 +113,13 @@ static void CStoreProcessUtility(Node *parseTree, const char *queryString,
ParamListInfo paramListInfo, ParamListInfo paramListInfo,
DestReceiver *destReceiver, char *completionTag); DestReceiver *destReceiver, char *completionTag);
#endif #endif
static bool CopyCStoreTableStatement(CopyStmt* copyStatement); static bool CopyCStoreTableStatement(CopyStmt *copyStatement);
static void CheckSuperuserPrivilegesForCopy(const CopyStmt* copyStatement); static void CheckSuperuserPrivilegesForCopy(const CopyStmt *copyStatement);
static void CStoreProcessCopyCommand(CopyStmt *copyStatement, const char *queryString, static void CStoreProcessCopyCommand(CopyStmt *copyStatement, const char *queryString,
char *completionTag); char *completionTag);
static uint64 CopyIntoCStoreTable(const CopyStmt *copyStatement, static uint64 CopyIntoCStoreTable(const CopyStmt *copyStatement,
const char *queryString); const char *queryString);
static uint64 CopyOutCStoreTable(CopyStmt* copyStatement, const char* queryString); static uint64 CopyOutCStoreTable(CopyStmt *copyStatement, const char *queryString);
static void CStoreProcessAlterTableCommand(AlterTableStmt *alterStatement); static void CStoreProcessAlterTableCommand(AlterTableStmt *alterStatement);
static List * DroppedCStoreFilenameList(DropStmt *dropStatement); static List * DroppedCStoreFilenameList(DropStmt *dropStatement);
static List * FindCStoreTables(List *tableList); static List * FindCStoreTables(List *tableList);
@ -168,7 +167,7 @@ static int CStoreAcquireSampleRows(Relation relation, int logLevel,
HeapTuple *sampleRows, int targetRowCount, HeapTuple *sampleRows, int targetRowCount,
double *totalRowCount, double *totalDeadRowCount); double *totalRowCount, double *totalDeadRowCount);
static List * CStorePlanForeignModify(PlannerInfo *plannerInfo, ModifyTable *plan, static List * CStorePlanForeignModify(PlannerInfo *plannerInfo, ModifyTable *plan,
Index resultRelation, int subplanIndex); Index resultRelation, int subplanIndex);
static void CStoreBeginForeignModify(ModifyTableState *modifyTableState, static void CStoreBeginForeignModify(ModifyTableState *modifyTableState,
ResultRelInfo *relationInfo, List *fdwPrivate, ResultRelInfo *relationInfo, List *fdwPrivate,
int subplanIndex, int executorflags); int subplanIndex, int executorflags);
@ -201,7 +200,8 @@ static ProcessUtility_hook_type PreviousProcessUtilityHook = NULL;
* previous utility hook, and then install our hook to pre-intercept calls to * previous utility hook, and then install our hook to pre-intercept calls to
* the copy command. * the copy command.
*/ */
void cstore_fdw_init() void
cstore_fdw_init()
{ {
PreviousProcessUtilityHook = ProcessUtility_hook; PreviousProcessUtilityHook = ProcessUtility_hook;
ProcessUtility_hook = CStoreProcessUtility; ProcessUtility_hook = CStoreProcessUtility;
@ -212,7 +212,8 @@ void cstore_fdw_init()
* Called when the module is unloaded. This function uninstalls the * Called when the module is unloaded. This function uninstalls the
* extension's hooks. * extension's hooks.
*/ */
void cstore_fdw_finish() void
cstore_fdw_finish()
{ {
ProcessUtility_hook = PreviousProcessUtilityHook; ProcessUtility_hook = PreviousProcessUtilityHook;
} }
@ -296,10 +297,10 @@ CStoreProcessUtility(PlannedStmt *plannedStatement, const char *queryString,
DestReceiver *destReceiver, char *completionTag) DestReceiver *destReceiver, char *completionTag)
#else #else
static void static void
CStoreProcessUtility(Node * parseTree, const char *queryString, CStoreProcessUtility(Node * parseTree, const char * queryString,
ProcessUtilityContext context, ProcessUtilityContext context,
ParamListInfo paramListInfo, ParamListInfo paramListInfo,
DestReceiver *destReceiver, char *completionTag) DestReceiver * destReceiver, char * completionTag)
#endif #endif
{ {
#if PG_VERSION_NUM >= 100000 #if PG_VERSION_NUM >= 100000
@ -387,11 +388,12 @@ CStoreProcessUtility(Node * parseTree, const char *queryString,
CALL_PREVIOUS_UTILITY(parseTree, queryString, context, paramListInfo, CALL_PREVIOUS_UTILITY(parseTree, queryString, context, paramListInfo,
destReceiver, completionTag); destReceiver, completionTag);
/* restore the former relation list. Our
* replacement could be freed but still needed /* restore the former relation list. Our
* in a cached plan. A truncate can be cached * replacement could be freed but still needed
* if run from a pl/pgSQL function */ * in a cached plan. A truncate can be cached
truncateStatement->relations = allTablesList; * if run from a pl/pgSQL function */
truncateStatement->relations = allTablesList;
} }
TruncateCStoreTables(cstoreRelationList); TruncateCStoreTables(cstoreRelationList);
@ -439,7 +441,7 @@ CStoreProcessUtility(Node * parseTree, const char *queryString,
* true. The function returns false otherwise. * true. The function returns false otherwise.
*/ */
static bool static bool
CopyCStoreTableStatement(CopyStmt* copyStatement) CopyCStoreTableStatement(CopyStmt *copyStatement)
{ {
bool copyCStoreTableStatement = false; bool copyCStoreTableStatement = false;
@ -474,7 +476,7 @@ CopyCStoreTableStatement(CopyStmt* copyStatement)
* copy operation and reports error if user does not have superuser rights. * copy operation and reports error if user does not have superuser rights.
*/ */
static void static void
CheckSuperuserPrivilegesForCopy(const CopyStmt* copyStatement) CheckSuperuserPrivilegesForCopy(const CopyStmt *copyStatement)
{ {
/* /*
* We disallow copy from file or program except to superusers. These checks * We disallow copy from file or program except to superusers. These checks
@ -485,16 +487,16 @@ CheckSuperuserPrivilegesForCopy(const CopyStmt* copyStatement)
if (copyStatement->is_program) if (copyStatement->is_program)
{ {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("must be superuser to COPY to or from a program"), errmsg("must be superuser to COPY to or from a program"),
errhint("Anyone can COPY to stdout or from stdin. " errhint("Anyone can COPY to stdout or from stdin. "
"psql's \\copy command also works for anyone."))); "psql's \\copy command also works for anyone.")));
} }
else else
{ {
ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("must be superuser to COPY to or from a file"), errmsg("must be superuser to COPY to or from a file"),
errhint("Anyone can COPY to stdout or from stdin. " errhint("Anyone can COPY to stdout or from stdin. "
"psql's \\copy command also works for anyone."))); "psql's \\copy command also works for anyone.")));
} }
} }
} }
@ -505,7 +507,7 @@ CheckSuperuserPrivilegesForCopy(const CopyStmt* copyStatement)
* It determines the copy direction and forwards execution to appropriate function. * It determines the copy direction and forwards execution to appropriate function.
*/ */
static void static void
CStoreProcessCopyCommand(CopyStmt *copyStatement, const char* queryString, CStoreProcessCopyCommand(CopyStmt *copyStatement, const char *queryString,
char *completionTag) char *completionTag)
{ {
uint64 processedCount = 0; uint64 processedCount = 0;
@ -648,7 +650,7 @@ CopyIntoCStoreTable(const CopyStmt *copyStatement, const char *queryString)
* stream. Copying selected columns from cstore table is not currently supported. * stream. Copying selected columns from cstore table is not currently supported.
*/ */
static uint64 static uint64
CopyOutCStoreTable(CopyStmt* copyStatement, const char* queryString) CopyOutCStoreTable(CopyStmt *copyStatement, const char *queryString)
{ {
uint64 processedCount = 0; uint64 processedCount = 0;
RangeVar *relation = NULL; RangeVar *relation = NULL;
@ -682,6 +684,7 @@ CopyOutCStoreTable(CopyStmt* copyStatement, const char* queryString)
copyStatement->relation = NULL; copyStatement->relation = NULL;
#if (PG_VERSION_NUM >= 100000) #if (PG_VERSION_NUM >= 100000)
/* /*
* raw_parser returns list of RawStmt* in PG 10+ we need to * raw_parser returns list of RawStmt* in PG 10+ we need to
* extract actual query from it. * extract actual query from it.
@ -737,7 +740,7 @@ CStoreProcessAlterTableCommand(AlterTableStmt *alterStatement)
foreach(commandCell, commandList) foreach(commandCell, commandList)
{ {
AlterTableCmd *alterCommand = (AlterTableCmd *) lfirst(commandCell); AlterTableCmd *alterCommand = (AlterTableCmd *) lfirst(commandCell);
if(alterCommand->subtype == AT_AlterColumnType) if (alterCommand->subtype == AT_AlterColumnType)
{ {
char *columnName = alterCommand->name; char *columnName = alterCommand->name;
ColumnDef *columnDef = (ColumnDef *) alterCommand->def; ColumnDef *columnDef = (ColumnDef *) alterCommand->def;
@ -849,7 +852,7 @@ OpenRelationsForTruncate(List *cstoreTableList)
Relation relation = heap_openrv(rangeVar, AccessExclusiveLock); Relation relation = heap_openrv(rangeVar, AccessExclusiveLock);
Oid relationId = relation->rd_id; Oid relationId = relation->rd_id;
AclResult aclresult = pg_class_aclcheck(relationId, GetUserId(), AclResult aclresult = pg_class_aclcheck(relationId, GetUserId(),
ACL_TRUNCATE); ACL_TRUNCATE);
if (aclresult != ACLCHECK_OK) if (aclresult != ACLCHECK_OK)
{ {
aclcheck_error(aclresult, ACLCHECK_OBJECT_TABLE, get_rel_name(relationId)); aclcheck_error(aclresult, ACLCHECK_OBJECT_TABLE, get_rel_name(relationId));
@ -890,6 +893,7 @@ TruncateCStoreTables(List *cstoreRelationList)
} }
} }
/* /*
* CStoreTable checks if the given table name belongs to a foreign columnar store * CStoreTable checks if the given table name belongs to a foreign columnar store
* table. If it does, the function returns true. Otherwise, it returns false. * table. If it does, the function returns true. Otherwise, it returns false.
@ -996,23 +1000,20 @@ DistributedTable(Oid relationId)
static bool static bool
DistributedWorkerCopy(CopyStmt *copyStatement) DistributedWorkerCopy(CopyStmt *copyStatement)
{ {
ListCell *optionCell = NULL; ListCell *optionCell = NULL;
foreach(optionCell, copyStatement->options) foreach(optionCell, copyStatement->options)
{ {
DefElem *defel = (DefElem *) lfirst(optionCell); DefElem *defel = (DefElem *) lfirst(optionCell);
if (strncmp(defel->defname, "master_host", NAMEDATALEN) == 0) if (strncmp(defel->defname, "master_host", NAMEDATALEN) == 0)
{ {
return true; return true;
} }
} }
return false; return false;
} }
/* /*
* cstore_table_size returns the total on-disk size of a cstore table in bytes. * cstore_table_size returns the total on-disk size of a cstore table in bytes.
* The result includes the sizes of data file and footer file. * The result includes the sizes of data file and footer file.
@ -1056,7 +1057,7 @@ cstore_table_size(PG_FUNCTION_ARGS)
{ {
ereport(ERROR, (errcode_for_file_access(), ereport(ERROR, (errcode_for_file_access(),
errmsg("could not stat file \"%s\": %m", errmsg("could not stat file \"%s\": %m",
footerFilename->data))); footerFilename->data)));
} }
tableSize += dataFileStatBuffer.st_size; tableSize += dataFileStatBuffer.st_size;
@ -1428,7 +1429,6 @@ CStoreDefaultFilePath(Oid foreignTableId)
{ {
databaseOid = MyDatabaseId; databaseOid = MyDatabaseId;
relationFileOid = foreignTableId; relationFileOid = foreignTableId;
} }
cstoreFilePath = makeStringInfo(); cstoreFilePath = makeStringInfo();
@ -1447,7 +1447,8 @@ static void
CStoreGetForeignRelSize(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId) CStoreGetForeignRelSize(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId)
{ {
CStoreOptions *cstoreOptions = CStoreGetOptions(foreignTableId); CStoreOptions *cstoreOptions = CStoreGetOptions(foreignTableId);
double tupleCountEstimate = TupleCountEstimate(foreignTableId, baserel, cstoreOptions->filename); double tupleCountEstimate = TupleCountEstimate(foreignTableId, baserel,
cstoreOptions->filename);
double rowSelectivity = clauselist_selectivity(root, baserel->baserestrictinfo, double rowSelectivity = clauselist_selectivity(root, baserel->baserestrictinfo,
0, JOIN_INNER, NULL); 0, JOIN_INNER, NULL);
@ -1494,7 +1495,8 @@ CStoreGetForeignPaths(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId
double queryPageCount = relationPageCount * queryColumnRatio; double queryPageCount = relationPageCount * queryColumnRatio;
double totalDiskAccessCost = seq_page_cost * queryPageCount; double totalDiskAccessCost = seq_page_cost * queryPageCount;
double tupleCountEstimate = TupleCountEstimate(foreignTableId, baserel, cstoreOptions->filename); double tupleCountEstimate = TupleCountEstimate(foreignTableId, baserel,
cstoreOptions->filename);
/* /*
* We estimate costs almost the same way as cost_seqscan(), thus assuming * We estimate costs almost the same way as cost_seqscan(), thus assuming
@ -1505,7 +1507,7 @@ CStoreGetForeignPaths(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId
double totalCpuCost = cpuCostPerTuple * tupleCountEstimate; double totalCpuCost = cpuCostPerTuple * tupleCountEstimate;
double startupCost = baserel->baserestrictcost.startup; double startupCost = baserel->baserestrictcost.startup;
double totalCost = startupCost + totalCpuCost + totalDiskAccessCost; double totalCost = startupCost + totalCpuCost + totalDiskAccessCost;
/* create a foreign path node and add it as the only possible path */ /* create a foreign path node and add it as the only possible path */
#if PG_VERSION_NUM >= 90600 #if PG_VERSION_NUM >= 90600
@ -1550,8 +1552,8 @@ CStoreGetForeignPlan(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId,
Plan *outerPlan) Plan *outerPlan)
#else #else
static ForeignScan * static ForeignScan *
CStoreGetForeignPlan(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId, CStoreGetForeignPlan(PlannerInfo * root, RelOptInfo * baserel, Oid foreignTableId,
ForeignPath *bestPath, List *targetList, List *scanClauses) ForeignPath * bestPath, List * targetList, List * scanClauses)
#endif #endif
{ {
ForeignScan *foreignScan = NULL; ForeignScan *foreignScan = NULL;
@ -1720,7 +1722,7 @@ ColumnList(RelOptInfo *baserel, Oid foreignTableId)
{ {
ListCell *neededColumnCell = NULL; ListCell *neededColumnCell = NULL;
Var *column = NULL; Var *column = NULL;
Form_pg_attribute attributeForm = TupleDescAttr(tupleDescriptor, columnIndex - 1); Form_pg_attribute attributeForm = TupleDescAttr(tupleDescriptor, columnIndex - 1);
if (attributeForm->attisdropped) if (attributeForm->attisdropped)
{ {
@ -1920,7 +1922,7 @@ CStoreAcquireSampleRows(Relation relation, int logLevel,
{ {
int sampleRowCount = 0; int sampleRowCount = 0;
double rowCount = 0.0; double rowCount = 0.0;
double rowCountToSkip = -1; /* -1 means not set yet */ double rowCountToSkip = -1; /* -1 means not set yet */
double selectionState = 0; double selectionState = 0;
MemoryContext oldContext = CurrentMemoryContext; MemoryContext oldContext = CurrentMemoryContext;
MemoryContext tupleContext = NULL; MemoryContext tupleContext = NULL;
@ -1948,7 +1950,8 @@ CStoreAcquireSampleRows(Relation relation, int logLevel,
if (!attributeForm->attisdropped) if (!attributeForm->attisdropped)
{ {
Var *column = makeVar(tableId, columnIndex + 1, attributeForm->atttypid, Var *column = makeVar(tableId, columnIndex + 1, attributeForm->atttypid,
attributeForm->atttypmod, attributeForm->attcollation, 0); attributeForm->atttypmod, attributeForm->attcollation,
0);
columnList = lappend(columnList, column); columnList = lappend(columnList, column);
} }
} }
@ -2139,7 +2142,7 @@ CStoreBeginForeignModify(ModifyTableState *modifyTableState,
return; return;
} }
Assert (modifyTableState->operation == CMD_INSERT); Assert(modifyTableState->operation == CMD_INSERT);
CStoreBeginForeignInsert(modifyTableState, relationInfo); CStoreBeginForeignInsert(modifyTableState, relationInfo);
} }
@ -2152,7 +2155,7 @@ CStoreBeginForeignModify(ModifyTableState *modifyTableState,
static void static void
CStoreBeginForeignInsert(ModifyTableState *modifyTableState, ResultRelInfo *relationInfo) CStoreBeginForeignInsert(ModifyTableState *modifyTableState, ResultRelInfo *relationInfo)
{ {
Oid foreignTableOid = InvalidOid; Oid foreignTableOid = InvalidOid;
CStoreOptions *cstoreOptions = NULL; CStoreOptions *cstoreOptions = NULL;
TupleDesc tupleDescriptor = NULL; TupleDesc tupleDescriptor = NULL;
TableWriteState *writeState = NULL; TableWriteState *writeState = NULL;
@ -2183,7 +2186,7 @@ static TupleTableSlot *
CStoreExecForeignInsert(EState *executorState, ResultRelInfo *relationInfo, 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; HeapTuple heapTuple;
Assert(writeState != NULL); Assert(writeState != NULL);
@ -2224,7 +2227,7 @@ CStoreEndForeignModify(EState *executorState, ResultRelInfo *relationInfo)
static void static void
CStoreEndForeignInsert(EState *executorState, ResultRelInfo *relationInfo) CStoreEndForeignInsert(EState *executorState, ResultRelInfo *relationInfo)
{ {
TableWriteState *writeState = (TableWriteState*) relationInfo->ri_FdwState; TableWriteState *writeState = (TableWriteState *) relationInfo->ri_FdwState;
/* writeState is NULL during Explain queries */ /* writeState is NULL during Explain queries */
if (writeState != NULL) if (writeState != NULL)
@ -2238,6 +2241,7 @@ CStoreEndForeignInsert(EState *executorState, ResultRelInfo *relationInfo)
#if PG_VERSION_NUM >= 90600 #if PG_VERSION_NUM >= 90600
/* /*
* CStoreIsForeignScanParallelSafe always returns true to indicate that * CStoreIsForeignScanParallelSafe always returns true to indicate that
* reading from a cstore_fdw table in a parallel worker is safe. This * reading from a cstore_fdw table in a parallel worker is safe. This
@ -2254,4 +2258,6 @@ CStoreIsForeignScanParallelSafe(PlannerInfo *root, RelOptInfo *rel,
{ {
return true; return true;
} }
#endif #endif

2
cstore_fdw.h
View File

@ -32,4 +32,4 @@ extern Datum cstore_clean_table_resources(PG_FUNCTION_ARGS);
extern Datum cstore_fdw_handler(PG_FUNCTION_ARGS); extern Datum cstore_fdw_handler(PG_FUNCTION_ARGS);
extern Datum cstore_fdw_validator(PG_FUNCTION_ARGS); extern Datum cstore_fdw_validator(PG_FUNCTION_ARGS);
#endif /* CSTORE_FDW_H */ #endif /* CSTORE_FDW_H */

8
cstore_metadata_serialization.c
View File

@ -143,8 +143,8 @@ SerializeColumnSkipList(ColumnBlockSkipNode *blockSkipNodeArray, uint32 blockCou
{ {
ColumnBlockSkipNode blockSkipNode = blockSkipNodeArray[blockIndex]; ColumnBlockSkipNode blockSkipNode = blockSkipNodeArray[blockIndex];
Protobuf__ColumnBlockSkipNode *protobufBlockSkipNode = NULL; Protobuf__ColumnBlockSkipNode *protobufBlockSkipNode = NULL;
ProtobufCBinaryData binaryMinimumValue = {0, 0}; ProtobufCBinaryData binaryMinimumValue = { 0, 0 };
ProtobufCBinaryData binaryMaximumValue = {0, 0}; ProtobufCBinaryData binaryMaximumValue = { 0, 0 };
if (blockSkipNode.hasMinMax) if (blockSkipNode.hasMinMax)
{ {
@ -352,7 +352,7 @@ DeserializeRowCount(StringInfo buffer)
for (blockIndex = 0; blockIndex < blockCount; blockIndex++) for (blockIndex = 0; blockIndex < blockCount; blockIndex++)
{ {
Protobuf__ColumnBlockSkipNode *protobufBlockSkipNode = Protobuf__ColumnBlockSkipNode *protobufBlockSkipNode =
protobufBlockSkipList->blockskipnodearray[blockIndex]; protobufBlockSkipList->blockskipnodearray[blockIndex];
rowCount += protobufBlockSkipNode->rowcount; rowCount += protobufBlockSkipNode->rowcount;
} }
@ -452,7 +452,7 @@ DeserializeColumnSkipList(StringInfo buffer, bool typeByValue, int typeLength,
static ProtobufCBinaryData static ProtobufCBinaryData
DatumToProtobufBinary(Datum datum, bool datumTypeByValue, int datumTypeLength) DatumToProtobufBinary(Datum datum, bool datumTypeByValue, int datumTypeLength)
{ {
ProtobufCBinaryData protobufBinary = {0, 0}; ProtobufCBinaryData protobufBinary = { 0, 0 };
int datumLength = att_addlength_datum(0, datumTypeLength, datum); int datumLength = att_addlength_datum(0, datumTypeLength, datum);
char *datumBuffer = palloc0(datumLength); char *datumBuffer = palloc0(datumLength);

2
cstore_metadata_serialization.h
View File

@ -31,4 +31,4 @@ extern ColumnBlockSkipNode * DeserializeColumnSkipList(StringInfo buffer,
uint32 blockCount); uint32 blockCount);
#endif /* CSTORE_SERIALIZATION_H */ #endif /* CSTORE_SERIALIZATION_H */

31
cstore_reader.c
View File

@ -67,7 +67,7 @@ static OpExpr * MakeOpExpression(Var *variable, int16 strategyNumber);
static Oid GetOperatorByType(Oid typeId, Oid accessMethodId, int16 strategyNumber); static Oid GetOperatorByType(Oid typeId, Oid accessMethodId, int16 strategyNumber);
static void UpdateConstraint(Node *baseConstraint, Datum minValue, Datum maxValue); static void UpdateConstraint(Node *baseConstraint, Datum minValue, Datum maxValue);
static StripeSkipList * SelectedBlockSkipList(StripeSkipList *stripeSkipList, static StripeSkipList * SelectedBlockSkipList(StripeSkipList *stripeSkipList,
bool *projectedColumnMask, bool *projectedColumnMask,
bool *selectedBlockMask); bool *selectedBlockMask);
static uint32 StripeSkipListRowCount(StripeSkipList *stripeSkipList); static uint32 StripeSkipListRowCount(StripeSkipList *stripeSkipList);
static bool * ProjectedColumnMask(uint32 columnCount, List *projectedColumnList); static bool * ProjectedColumnMask(uint32 columnCount, List *projectedColumnList);
@ -104,7 +104,7 @@ CStoreBeginRead(Oid relationId, const char *filename, TupleDesc tupleDescriptor,
MemoryContext stripeReadContext = NULL; MemoryContext stripeReadContext = NULL;
uint32 columnCount = 0; uint32 columnCount = 0;
bool *projectedColumnMask = NULL; bool *projectedColumnMask = NULL;
ColumnBlockData **blockDataArray = NULL; ColumnBlockData **blockDataArray = NULL;
StringInfo tableFooterFilename = makeStringInfo(); StringInfo tableFooterFilename = makeStringInfo();
appendStringInfo(tableFooterFilename, "%s%s", filename, CSTORE_FOOTER_FILE_SUFFIX); appendStringInfo(tableFooterFilename, "%s%s", filename, CSTORE_FOOTER_FILE_SUFFIX);
@ -134,7 +134,7 @@ CStoreBeginRead(Oid relationId, const char *filename, TupleDesc tupleDescriptor,
columnCount = tupleDescriptor->natts; columnCount = tupleDescriptor->natts;
projectedColumnMask = ProjectedColumnMask(columnCount, projectedColumnList); projectedColumnMask = ProjectedColumnMask(columnCount, projectedColumnList);
blockDataArray = CreateEmptyBlockDataArray(columnCount, projectedColumnMask, blockDataArray = CreateEmptyBlockDataArray(columnCount, projectedColumnMask,
tableFooter->blockRowCount); tableFooter->blockRowCount);
readState = palloc0(sizeof(TableReadState)); readState = palloc0(sizeof(TableReadState));
readState->relationId = relationId; readState->relationId = relationId;
@ -356,7 +356,7 @@ ColumnBlockData **
CreateEmptyBlockDataArray(uint32 columnCount, bool *columnMask, uint32 blockRowCount) CreateEmptyBlockDataArray(uint32 columnCount, bool *columnMask, uint32 blockRowCount)
{ {
uint32 columnIndex = 0; uint32 columnIndex = 0;
ColumnBlockData **blockDataArray = palloc0(columnCount * sizeof(ColumnBlockData*)); ColumnBlockData **blockDataArray = palloc0(columnCount * sizeof(ColumnBlockData *));
/* allocate block memory for deserialized data */ /* allocate block memory for deserialized data */
for (columnIndex = 0; columnIndex < columnCount; columnIndex++) for (columnIndex = 0; columnIndex < columnCount; columnIndex++)
@ -448,12 +448,12 @@ StripeRowCount(Oid relid, FILE *tableFile, StripeMetadata *stripeMetadata)
uint64 rowCount = 0; uint64 rowCount = 0;
StringInfo firstColumnSkipListBuffer = NULL; StringInfo firstColumnSkipListBuffer = NULL;
StripeFooter * stripeFooter = ReadStripeFooter(relid, stripeMetadata->id, StripeFooter *stripeFooter = ReadStripeFooter(relid, stripeMetadata->id,
RelationColumnCount(relid)); RelationColumnCount(relid));
firstColumnSkipListBuffer = ReadFromFile(tableFile, stripeMetadata->fileOffset, firstColumnSkipListBuffer = ReadFromFile(tableFile, stripeMetadata->fileOffset,
stripeFooter->skipListSizeArray[0]); stripeFooter->skipListSizeArray[0]);
rowCount = DeserializeRowCount(firstColumnSkipListBuffer); rowCount = DeserializeRowCount(firstColumnSkipListBuffer);
return rowCount; return rowCount;
} }
@ -573,7 +573,7 @@ LoadColumnBuffers(FILE *tableFile, ColumnBlockSkipNode *blockSkipNodeArray,
ColumnBuffers *columnBuffers = NULL; ColumnBuffers *columnBuffers = NULL;
uint32 blockIndex = 0; uint32 blockIndex = 0;
ColumnBlockBuffers **blockBuffersArray = ColumnBlockBuffers **blockBuffersArray =
palloc0(blockCount * sizeof(ColumnBlockBuffers *)); palloc0(blockCount * sizeof(ColumnBlockBuffers *));
for (blockIndex = 0; blockIndex < blockCount; blockIndex++) for (blockIndex = 0; blockIndex < blockCount; blockIndex++)
{ {
@ -761,7 +761,8 @@ SelectedBlockMask(StripeSkipList *stripeSkipList, List *projectedColumnList,
constraintList = list_make1(baseConstraint); constraintList = list_make1(baseConstraint);
#if (PG_VERSION_NUM >= 100000) #if (PG_VERSION_NUM >= 100000)
predicateRefuted = predicate_refuted_by(constraintList, restrictInfoList, false); predicateRefuted = predicate_refuted_by(constraintList, restrictInfoList,
false);
#else #else
predicateRefuted = predicate_refuted_by(constraintList, restrictInfoList); predicateRefuted = predicate_refuted_by(constraintList, restrictInfoList);
#endif #endif
@ -877,7 +878,7 @@ MakeOpExpression(Var *variable, int16 strategyNumber)
Oid accessMethodId = BTREE_AM_OID; Oid accessMethodId = BTREE_AM_OID;
Oid operatorId = InvalidOid; Oid operatorId = InvalidOid;
Const *constantValue = NULL; Const *constantValue = NULL;
OpExpr *expression = NULL; OpExpr *expression = NULL;
/* Load the operator from system catalogs */ /* Load the operator from system catalogs */
@ -888,7 +889,7 @@ MakeOpExpression(Var *variable, int16 strategyNumber)
/* 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, 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,
(Expr *) constantValue, (Expr *) constantValue,
InvalidOid, collationId); InvalidOid, collationId);
@ -1163,7 +1164,8 @@ DeserializeBlockData(StripeBuffers *stripeBuffers, uint64 blockIndex,
if (columnBuffers != NULL) if (columnBuffers != NULL)
{ {
ColumnBlockBuffers *blockBuffers = columnBuffers->blockBuffersArray[blockIndex]; ColumnBlockBuffers *blockBuffers =
columnBuffers->blockBuffersArray[blockIndex];
StringInfo valueBuffer = NULL; StringInfo valueBuffer = NULL;
/* free previous block's data buffers */ /* free previous block's data buffers */
@ -1214,7 +1216,6 @@ DeserializeBlockData(StripeBuffers *stripeBuffers, uint64 blockIndex,
{ {
memset(blockData->existsArray, false, rowCount); memset(blockData->existsArray, false, rowCount);
} }
} }
} }
} }
@ -1330,8 +1331,6 @@ ReadFromFile(FILE *file, uint64 offset, uint32 size)
} }
/* /*
* ResetUncompressedBlockData iterates over deserialized column block data * ResetUncompressedBlockData iterates over deserialized column block data
* and sets valueBuffer field to empty buffer. This field is allocated in stripe * and sets valueBuffer field to empty buffer. This field is allocated in stripe

7
cstore_version_compat.h
View File

@ -22,7 +22,8 @@
#endif #endif
#if PG_VERSION_NUM < 110000 #if PG_VERSION_NUM < 110000
#define ALLOCSET_DEFAULT_SIZES ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE #define ALLOCSET_DEFAULT_SIZES ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, \
ALLOCSET_DEFAULT_MAXSIZE
#define ACLCHECK_OBJECT_TABLE ACL_KIND_CLASS #define ACLCHECK_OBJECT_TABLE ACL_KIND_CLASS
#else #else
#define ACLCHECK_OBJECT_TABLE OBJECT_TABLE #define ACLCHECK_OBJECT_TABLE OBJECT_TABLE
@ -46,9 +47,9 @@
#endif #endif
#if PG_VERSION_NUM < 120000 #if PG_VERSION_NUM < 120000
#define TTS_EMPTY(slot) ((slot)->tts_isempty) #define TTS_EMPTY(slot) ((slot)->tts_isempty)
#define ExecForceStoreHeapTuple(tuple, slot, shouldFree) \ #define ExecForceStoreHeapTuple(tuple, slot, shouldFree) \
ExecStoreTuple(newTuple, tupleSlot, InvalidBuffer, shouldFree); ExecStoreTuple(newTuple, tupleSlot, InvalidBuffer, shouldFree);
#define TableScanDesc HeapScanDesc #define TableScanDesc HeapScanDesc
#define table_beginscan heap_beginscan #define table_beginscan heap_beginscan
#define table_endscan heap_endscan #define table_endscan heap_endscan

16
cstore_writer.c
View File

@ -149,13 +149,15 @@ CStoreBeginWrite(Oid relationId,
for (columnIndex = 0; columnIndex < columnCount; columnIndex++) for (columnIndex = 0; columnIndex < columnCount; columnIndex++)
{ {
FmgrInfo *comparisonFunction = NULL; FmgrInfo *comparisonFunction = NULL;
FormData_pg_attribute *attributeForm = TupleDescAttr(tupleDescriptor, columnIndex); FormData_pg_attribute *attributeForm = TupleDescAttr(tupleDescriptor,
columnIndex);
if (!attributeForm->attisdropped) if (!attributeForm->attisdropped)
{ {
Oid typeId = attributeForm->atttypid; Oid typeId = attributeForm->atttypid;
comparisonFunction = GetFunctionInfoOrNull(typeId, BTREE_AM_OID, BTORDER_PROC); comparisonFunction = GetFunctionInfoOrNull(typeId, BTREE_AM_OID,
BTORDER_PROC);
} }
comparisonFunctionArray[columnIndex] = comparisonFunction; comparisonFunctionArray[columnIndex] = comparisonFunction;
@ -262,7 +264,7 @@ CStoreWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNul
bool columnTypeByValue = attributeForm->attbyval; bool columnTypeByValue = attributeForm->attbyval;
int columnTypeLength = attributeForm->attlen; int columnTypeLength = attributeForm->attlen;
Oid columnCollation = attributeForm->attcollation; Oid columnCollation = attributeForm->attcollation;
char columnTypeAlign = attributeForm->attalign; char columnTypeAlign = attributeForm->attalign;
blockData->existsArray[blockRowIndex] = true; blockData->existsArray[blockRowIndex] = true;
@ -492,7 +494,7 @@ CreateEmptyStripeSkipList(uint32 stripeMaxRowCount, uint32 blockRowCount,
static StripeMetadata static StripeMetadata
FlushStripe(TableWriteState *writeState) FlushStripe(TableWriteState *writeState)
{ {
StripeMetadata stripeMetadata = {0, 0, 0, 0}; StripeMetadata stripeMetadata = { 0, 0, 0, 0 };
uint64 skipListLength = 0; uint64 skipListLength = 0;
uint64 dataLength = 0; uint64 dataLength = 0;
StringInfo *skipListBufferArray = NULL; StringInfo *skipListBufferArray = NULL;
@ -531,7 +533,7 @@ FlushStripe(TableWriteState *writeState)
for (blockIndex = 0; blockIndex < blockCount; blockIndex++) for (blockIndex = 0; blockIndex < blockCount; blockIndex++)
{ {
ColumnBlockBuffers *blockBuffers = ColumnBlockBuffers *blockBuffers =
columnBuffers->blockBuffersArray[blockIndex]; columnBuffers->blockBuffersArray[blockIndex];
uint64 existsBufferSize = blockBuffers->existsBuffer->len; uint64 existsBufferSize = blockBuffers->existsBuffer->len;
uint64 valueBufferSize = blockBuffers->valueBuffer->len; uint64 valueBufferSize = blockBuffers->valueBuffer->len;
CompressionType valueCompressionType = blockBuffers->valueCompressionType; CompressionType valueCompressionType = blockBuffers->valueCompressionType;
@ -582,7 +584,7 @@ FlushStripe(TableWriteState *writeState)
for (blockIndex = 0; blockIndex < stripeSkipList->blockCount; blockIndex++) for (blockIndex = 0; blockIndex < stripeSkipList->blockCount; blockIndex++)
{ {
ColumnBlockBuffers *blockBuffers = ColumnBlockBuffers *blockBuffers =
columnBuffers->blockBuffersArray[blockIndex]; columnBuffers->blockBuffersArray[blockIndex];
StringInfo existsBuffer = blockBuffers->existsBuffer; StringInfo existsBuffer = blockBuffers->existsBuffer;
WriteToFile(tableFile, existsBuffer->data, existsBuffer->len); WriteToFile(tableFile, existsBuffer->data, existsBuffer->len);
@ -591,7 +593,7 @@ FlushStripe(TableWriteState *writeState)
for (blockIndex = 0; blockIndex < stripeSkipList->blockCount; blockIndex++) for (blockIndex = 0; blockIndex < stripeSkipList->blockCount; blockIndex++)
{ {
ColumnBlockBuffers *blockBuffers = ColumnBlockBuffers *blockBuffers =
columnBuffers->blockBuffersArray[blockIndex]; columnBuffers->blockBuffersArray[blockIndex];
StringInfo valueBuffer = blockBuffers->valueBuffer; StringInfo valueBuffer = blockBuffers->valueBuffer;
WriteToFile(tableFile, valueBuffer->data, valueBuffer->len); WriteToFile(tableFile, valueBuffer->data, valueBuffer->len);

7
mod.c
View File

@ -20,14 +20,15 @@
PG_MODULE_MAGIC; PG_MODULE_MAGIC;
void _PG_init(void) void
_PG_init(void)
{ {
cstore_fdw_init(); cstore_fdw_init();
} }
void _PG_fini(void) void
_PG_fini(void)
{ {
cstore_fdw_finish(); cstore_fdw_finish();
} }