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

Merge pull request #4557 from citusdata/columnar_rename_funcs 

Don't use 'cstore' in symbols
pull/4552/head
Hadi Moshayedi 2021-01-21 21:11:27 -08:00 committed by GitHub
parent 0b5551faaf 222fb4d589
commit 9051bf485f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 524 additions and 518 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/backend/columnar/cstore.c
View File

@ -35,12 +35,12 @@
#define DEFAULT_COMPRESSION_TYPE COMPRESSION_PG_LZ
#endif
int cstore_compression = DEFAULT_COMPRESSION_TYPE;
int cstore_stripe_row_count = DEFAULT_STRIPE_ROW_COUNT;
int cstore_chunk_row_count = DEFAULT_CHUNK_ROW_COUNT;
int columnar_compression = DEFAULT_COMPRESSION_TYPE;
int columnar_stripe_row_count = DEFAULT_STRIPE_ROW_COUNT;
int columnar_chunk_row_count = DEFAULT_CHUNK_ROW_COUNT;
int columnar_compression_level = 3;
static const struct config_enum_entry cstore_compression_options[] =
static const struct config_enum_entry columnar_compression_options[] =
{
{ "none", COMPRESSION_NONE, false },
{ "pglz", COMPRESSION_PG_LZ, false },
@ -54,14 +54,14 @@ static const struct config_enum_entry cstore_compression_options[] =
};
void
cstore_init()
columnar_init_gucs()
{
DefineCustomEnumVariable("columnar.compression",
"Compression type for cstore.",
NULL,
&cstore_compression,
&columnar_compression,
DEFAULT_COMPRESSION_TYPE,
cstore_compression_options,
columnar_compression_options,
PGC_USERSET,
0,
NULL,
@ -84,7 +84,7 @@ cstore_init()
DefineCustomIntVariable("columnar.stripe_row_count",
"Maximum number of tuples per stripe.",
NULL,
&cstore_stripe_row_count,
&columnar_stripe_row_count,
DEFAULT_STRIPE_ROW_COUNT,
STRIPE_ROW_COUNT_MINIMUM,
STRIPE_ROW_COUNT_MAXIMUM,
@ -97,7 +97,7 @@ cstore_init()
DefineCustomIntVariable("columnar.chunk_row_count",
"Maximum number of rows per chunk.",
NULL,
&cstore_chunk_row_count,
&columnar_chunk_row_count,
DEFAULT_CHUNK_ROW_COUNT,
CHUNK_ROW_COUNT_MINIMUM,
CHUNK_ROW_COUNT_MAXIMUM,
@ -120,13 +120,13 @@ ParseCompressionType(const char *compressionTypeString)
Assert(compressionTypeString != NULL);
for (int compressionIndex = 0;
cstore_compression_options[compressionIndex].name != NULL;
columnar_compression_options[compressionIndex].name != NULL;
compressionIndex++)
{
const char *compressionName = cstore_compression_options[compressionIndex].name;
const char *compressionName = columnar_compression_options[compressionIndex].name;
if (strncmp(compressionTypeString, compressionName, NAMEDATALEN) == 0)
{
return cstore_compression_options[compressionIndex].val;
return columnar_compression_options[compressionIndex].val;
}
}
@ -143,14 +143,14 @@ const char *
CompressionTypeStr(CompressionType requestedType)
{
for (int compressionIndex = 0;
cstore_compression_options[compressionIndex].name != NULL;
columnar_compression_options[compressionIndex].name != NULL;
compressionIndex++)
{
CompressionType compressionType =
cstore_compression_options[compressionIndex].val;
columnar_compression_options[compressionIndex].val;
if (compressionType == requestedType)
{
return cstore_compression_options[compressionIndex].name;
return columnar_compression_options[compressionIndex].name;
}
}

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

@ -1,6 +1,6 @@
/*-------------------------------------------------------------------------
*
* cstore_customscan.c
* columnar_customscan.c
*
* This file contains the implementation of a postgres custom scan that
* we use to push down the projections into the table access methods.
@ -29,85 +29,86 @@
#include "columnar/cstore_customscan.h"
#include "columnar/cstore_tableam.h"
typedef struct CStoreScanPath
typedef struct ColumnarScanPath
{
CustomPath custom_path;
/* place for local state during planning */
} CStoreScanPath;
} ColumnarScanPath;
typedef struct CStoreScanScan
typedef struct ColumnarScanScan
{
CustomScan custom_scan;
/* place for local state during execution */
} CStoreScanScan;
} ColumnarScanScan;
typedef struct CStoreScanState
typedef struct ColumnarScanState
{
CustomScanState custom_scanstate;
List *qual;
} CStoreScanState;
} ColumnarScanState;
static void CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
static void ColumnarSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
RangeTblEntry *rte);
static Path * CreateCStoreScanPath(RelOptInfo *rel, RangeTblEntry *rte);
static Cost CStoreScanCost(RangeTblEntry *rte);
static Plan * CStoreScanPath_PlanCustomPath(PlannerInfo *root,
static Path * CreateColumnarScanPath(RelOptInfo *rel, RangeTblEntry *rte);
static Cost ColumnarScanCost(RangeTblEntry *rte);
static Plan * ColumnarScanPath_PlanCustomPath(PlannerInfo *root,
RelOptInfo *rel,
struct CustomPath *best_path,
List *tlist,
List *clauses,
List *custom_plans);
static Node * CStoreScan_CreateCustomScanState(CustomScan *cscan);
static Node * ColumnarScan_CreateCustomScanState(CustomScan *cscan);
static void CStoreScan_BeginCustomScan(CustomScanState *node, EState *estate, int eflags);
static TupleTableSlot * CStoreScan_ExecCustomScan(CustomScanState *node);
static void CStoreScan_EndCustomScan(CustomScanState *node);
static void CStoreScan_ReScanCustomScan(CustomScanState *node);
static void CStoreScan_ExplainCustomScan(CustomScanState *node, List *ancestors,
static void ColumnarScan_BeginCustomScan(CustomScanState *node, EState *estate, int
eflags);
static TupleTableSlot * ColumnarScan_ExecCustomScan(CustomScanState *node);
static void ColumnarScan_EndCustomScan(CustomScanState *node);
static void ColumnarScan_ReScanCustomScan(CustomScanState *node);
static void ColumnarScan_ExplainCustomScan(CustomScanState *node, List *ancestors,
ExplainState *es);
/* saved hook value in case of unload */
static set_rel_pathlist_hook_type PreviousSetRelPathlistHook = NULL;
static bool EnableCStoreCustomScan = true;
static bool EnableColumnarCustomScan = true;
const struct CustomPathMethods CStoreScanPathMethods = {
const struct CustomPathMethods ColumnarScanPathMethods = {
.CustomName = "ColumnarScan",
.PlanCustomPath = CStoreScanPath_PlanCustomPath,
.PlanCustomPath = ColumnarScanPath_PlanCustomPath,
};
const struct CustomScanMethods CStoreScanScanMethods = {
const struct CustomScanMethods ColumnarScanScanMethods = {
.CustomName = "ColumnarScan",
.CreateCustomScanState = CStoreScan_CreateCustomScanState,
.CreateCustomScanState = ColumnarScan_CreateCustomScanState,
};
const struct CustomExecMethods CStoreExecuteMethods = {
const struct CustomExecMethods ColumnarExecuteMethods = {
.CustomName = "ColumnarScan",
.BeginCustomScan = CStoreScan_BeginCustomScan,
.ExecCustomScan = CStoreScan_ExecCustomScan,
.EndCustomScan = CStoreScan_EndCustomScan,
.ReScanCustomScan = CStoreScan_ReScanCustomScan,
.BeginCustomScan = ColumnarScan_BeginCustomScan,
.ExecCustomScan = ColumnarScan_ExecCustomScan,
.EndCustomScan = ColumnarScan_EndCustomScan,
.ReScanCustomScan = ColumnarScan_ReScanCustomScan,
.ExplainCustomScan = CStoreScan_ExplainCustomScan,
.ExplainCustomScan = ColumnarScan_ExplainCustomScan,
};
/*
* cstore_customscan_init installs the hook required to intercept the postgres planner and
* columnar_customscan_init installs the hook required to intercept the postgres planner and
* provide extra paths for cstore tables
*/
void
cstore_customscan_init()
columnar_customscan_init()
{
PreviousSetRelPathlistHook = set_rel_pathlist_hook;
set_rel_pathlist_hook = CStoreSetRelPathlistHook;
set_rel_pathlist_hook = ColumnarSetRelPathlistHook;
/* register customscan specific GUC's */
DefineCustomBoolVariable(
@ -115,7 +116,7 @@ cstore_customscan_init()
gettext_noop("Enables the use of a custom scan to push projections and quals "
"into the storage layer"),
NULL,
&EnableCStoreCustomScan,
&EnableColumnarCustomScan,
true,
PGC_USERSET,
GUC_NO_SHOW_ALL,
@ -135,7 +136,7 @@ clear_paths(RelOptInfo *rel)
static void
CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
ColumnarSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
RangeTblEntry *rte)
{
/* call into previous hook if assigned */
@ -144,7 +145,7 @@ CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
PreviousSetRelPathlistHook(root, rel, rti, rte);
}
if (!EnableCStoreCustomScan)
if (!EnableColumnarCustomScan)
{
/* custon scans are disabled, use normal table access method api instead */
return;
@ -170,7 +171,7 @@ CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
errmsg("sample scans not supported on columnar tables")));
}
Path *customPath = CreateCStoreScanPath(rel, rte);
Path *customPath = CreateColumnarScanPath(rel, rte);
ereport(DEBUG1, (errmsg("pathlist hook for cstore table am")));
@ -183,16 +184,16 @@ CStoreSetRelPathlistHook(PlannerInfo *root, RelOptInfo *rel, Index rti,
static Path *
CreateCStoreScanPath(RelOptInfo *rel, RangeTblEntry *rte)
CreateColumnarScanPath(RelOptInfo *rel, RangeTblEntry *rte)
{
CStoreScanPath *cspath = (CStoreScanPath *) newNode(sizeof(CStoreScanPath),
ColumnarScanPath *cspath = (ColumnarScanPath *) newNode(sizeof(ColumnarScanPath),
T_CustomPath);
/*
* popuate custom path information
*/
CustomPath *cpath = &cspath->custom_path;
cpath->methods = &CStoreScanPathMethods;
cpath->methods = &ColumnarScanPathMethods;
/*
* populate generic path information
@ -208,19 +209,19 @@ CreateCStoreScanPath(RelOptInfo *rel, RangeTblEntry *rte)
*/
path->rows = rel->rows;
path->startup_cost = 0;
path->total_cost = path->startup_cost + CStoreScanCost(rte);
path->total_cost = path->startup_cost + ColumnarScanCost(rte);
return (Path *) cspath;
}
/*
* CStoreScanCost calculates the cost of scanning the cstore table. The cost is estimated
* ColumnarScanCost calculates the cost of scanning the cstore table. The cost is estimated
* by using all stripe metadata to estimate based on the columns to read how many pages
* need to be read.
*/
static Cost
CStoreScanCost(RangeTblEntry *rte)
ColumnarScanCost(RangeTblEntry *rte)
{
Relation rel = RelationIdGetRelation(rte->relid);
List *stripeList = StripesForRelfilenode(rel->rd_node);
@ -249,18 +250,18 @@ CStoreScanCost(RangeTblEntry *rte)
static Plan *
CStoreScanPath_PlanCustomPath(PlannerInfo *root,
ColumnarScanPath_PlanCustomPath(PlannerInfo *root,
RelOptInfo *rel,
struct CustomPath *best_path,
List *tlist,
List *clauses,
List *custom_plans)
{
CStoreScanScan *plan = (CStoreScanScan *) newNode(sizeof(CStoreScanScan),
ColumnarScanScan *plan = (ColumnarScanScan *) newNode(sizeof(ColumnarScanScan),
T_CustomScan);
CustomScan *cscan = &plan->custom_scan;
cscan->methods = &CStoreScanScanMethods;
cscan->methods = &ColumnarScanScanMethods;
/* Reduce RestrictInfo list to bare expressions; ignore pseudoconstants */
clauses = extract_actual_clauses(clauses, false);
@ -274,13 +275,13 @@ CStoreScanPath_PlanCustomPath(PlannerInfo *root,
static Node *
CStoreScan_CreateCustomScanState(CustomScan *cscan)
ColumnarScan_CreateCustomScanState(CustomScan *cscan)
{
CStoreScanState *cstorescanstate = (CStoreScanState *) newNode(
sizeof(CStoreScanState), T_CustomScanState);
ColumnarScanState *cstorescanstate = (ColumnarScanState *) newNode(
sizeof(ColumnarScanState), T_CustomScanState);
CustomScanState *cscanstate = &cstorescanstate->custom_scanstate;
cscanstate->methods = &CStoreExecuteMethods;
cscanstate->methods = &ColumnarExecuteMethods;
cstorescanstate->qual = cscan->scan.plan.qual;
@ -289,14 +290,14 @@ CStoreScan_CreateCustomScanState(CustomScan *cscan)
static void
CStoreScan_BeginCustomScan(CustomScanState *cscanstate, EState *estate, int eflags)
ColumnarScan_BeginCustomScan(CustomScanState *cscanstate, EState *estate, int eflags)
{
/* scan slot is already initialized */
}
static Bitmapset *
CStoreAttrNeeded(ScanState *ss)
ColumnarAttrNeeded(ScanState *ss)
{
TupleTableSlot *slot = ss->ss_ScanTupleSlot;
int natts = slot->tts_tupleDescriptor->natts;
@ -337,7 +338,7 @@ CStoreAttrNeeded(ScanState *ss)
static TupleTableSlot *
CStoreScanNext(CStoreScanState *cstorescanstate)
ColumnarScanNext(ColumnarScanState *cstorescanstate)
{
CustomScanState *node = (CustomScanState *) cstorescanstate;
@ -353,13 +354,13 @@ CStoreScanNext(CStoreScanState *cstorescanstate)
{
/* the cstore access method does not use the flags, they are specific to heap */
uint32 flags = 0;
Bitmapset *attr_needed = CStoreAttrNeeded(&node->ss);
Bitmapset *attr_needed = ColumnarAttrNeeded(&node->ss);
/*
* We reach here if the scan is not parallel, or if we're serially
* executing a scan that was planned to be parallel.
*/
scandesc = cstore_beginscan_extended(node->ss.ss_currentRelation,
scandesc = columnar_beginscan_extended(node->ss.ss_currentRelation,
estate->es_snapshot,
0, NULL, NULL, flags, attr_needed,
cstorescanstate->qual);
@ -383,23 +384,23 @@ CStoreScanNext(CStoreScanState *cstorescanstate)
* SeqRecheck -- access method routine to recheck a tuple in EvalPlanQual
*/
static bool
CStoreScanRecheck(CStoreScanState *node, TupleTableSlot *slot)
ColumnarScanRecheck(ColumnarScanState *node, TupleTableSlot *slot)
{
return true;
}
static TupleTableSlot *
CStoreScan_ExecCustomScan(CustomScanState *node)
ColumnarScan_ExecCustomScan(CustomScanState *node)
{
return ExecScan(&node->ss,
(ExecScanAccessMtd) CStoreScanNext,
(ExecScanRecheckMtd) CStoreScanRecheck);
(ExecScanAccessMtd) ColumnarScanNext,
(ExecScanRecheckMtd) ColumnarScanRecheck);
}
static void
CStoreScan_EndCustomScan(CustomScanState *node)
ColumnarScan_EndCustomScan(CustomScanState *node)
{
/*
* get information from node
@ -431,7 +432,7 @@ CStoreScan_EndCustomScan(CustomScanState *node)
static void
CStoreScan_ReScanCustomScan(CustomScanState *node)
ColumnarScan_ReScanCustomScan(CustomScanState *node)
{
TableScanDesc scanDesc = node->ss.ss_currentScanDesc;
if (scanDesc != NULL)
@ -442,7 +443,7 @@ CStoreScan_ReScanCustomScan(CustomScanState *node)
static void
CStoreScan_ExplainCustomScan(CustomScanState *node, List *ancestors,
ColumnarScan_ExplainCustomScan(CustomScanState *node, List *ancestors,
ExplainState *es)
{
TableScanDesc scanDesc = node->ss.ss_currentScanDesc;

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

@ -1,6 +1,6 @@
/*-------------------------------------------------------------------------
*
* cstore_metadata_tables.c
* columnar_metadata_tables.c
*
* Copyright (c), Citus Data, Inc.
*
@ -79,13 +79,13 @@ static void GetHighestUsedAddressAndId(uint64 storageId,
uint64 *highestUsedAddress,
uint64 *highestUsedId);
static List * ReadDataFileStripeList(uint64 storageId, Snapshot snapshot);
static Oid CStoreStripesRelationId(void);
static Oid CStoreStripesIndexRelationId(void);
static Oid ColumnarStripesRelationId(void);
static Oid ColumnarStripesIndexRelationId(void);
static Oid ColumnarOptionsRelationId(void);
static Oid ColumnarOptionsIndexRegclass(void);
static Oid CStoreSkipNodesRelationId(void);
static Oid CStoreSkipNodesIndexRelationId(void);
static Oid CStoreNamespaceId(void);
static Oid ColumnarSkipNodesRelationId(void);
static Oid ColumnarSkipNodesIndexRelationId(void);
static Oid ColumnarNamespaceId(void);
static ModifyState * StartModifyRelation(Relation rel);
static void InsertTupleAndEnforceConstraints(ModifyState *state, Datum *values,
bool *nulls);
@ -102,18 +102,18 @@ static bool WriteColumnarOptions(Oid regclass, ColumnarOptions *options, bool ov
PG_FUNCTION_INFO_V1(columnar_relation_storageid);
/* constants for columnar.options */
#define Natts_cstore_options 5
#define Anum_cstore_options_regclass 1
#define Anum_cstore_options_chunk_row_count 2
#define Anum_cstore_options_stripe_row_count 3
#define Anum_cstore_options_compression_level 4
#define Anum_cstore_options_compression 5
#define Natts_columnar_options 5
#define Anum_columnar_options_regclass 1
#define Anum_columnar_options_chunk_row_count 2
#define Anum_columnar_options_stripe_row_count 3
#define Anum_columnar_options_compression_level 4
#define Anum_columnar_options_compression 5
/* ----------------
* columnar.options definition.
* ----------------
*/
typedef struct FormData_cstore_options
typedef struct FormData_columnar_options
{
Oid regclass;
int32 chunk_row_count;
@ -123,37 +123,37 @@ typedef struct FormData_cstore_options
#ifdef CATALOG_VARLEN /* variable-length fields start here */
#endif
} FormData_cstore_options;
typedef FormData_cstore_options *Form_cstore_options;
} FormData_columnar_options;
typedef FormData_columnar_options *Form_columnar_options;
/* constants for cstore_stripe */
#define Natts_cstore_stripes 8
#define Anum_cstore_stripes_storageid 1
#define Anum_cstore_stripes_stripe 2
#define Anum_cstore_stripes_file_offset 3
#define Anum_cstore_stripes_data_length 4
#define Anum_cstore_stripes_column_count 5
#define Anum_cstore_stripes_chunk_count 6
#define Anum_cstore_stripes_chunk_row_count 7
#define Anum_cstore_stripes_row_count 8
/* constants for columnar_stripe */
#define Natts_columnar_stripes 8
#define Anum_columnar_stripes_storageid 1
#define Anum_columnar_stripes_stripe 2
#define Anum_columnar_stripes_file_offset 3
#define Anum_columnar_stripes_data_length 4
#define Anum_columnar_stripes_column_count 5
#define Anum_columnar_stripes_chunk_count 6
#define Anum_columnar_stripes_chunk_row_count 7
#define Anum_columnar_stripes_row_count 8
/* constants for cstore_skipnodes */
#define Natts_cstore_skipnodes 14
#define Anum_cstore_skipnodes_storageid 1
#define Anum_cstore_skipnodes_stripe 2
#define Anum_cstore_skipnodes_attr 3
#define Anum_cstore_skipnodes_chunk 4
#define Anum_cstore_skipnodes_row_count 5
#define Anum_cstore_skipnodes_minimum_value 6
#define Anum_cstore_skipnodes_maximum_value 7
#define Anum_cstore_skipnodes_value_stream_offset 8
#define Anum_cstore_skipnodes_value_stream_length 9
#define Anum_cstore_skipnodes_exists_stream_offset 10
#define Anum_cstore_skipnodes_exists_stream_length 11
#define Anum_cstore_skipnodes_value_compression_type 12
#define Anum_cstore_skipnodes_value_compression_level 13
#define Anum_cstore_skipnodes_value_decompressed_size 14
/* constants for columnar_skipnodes */
#define Natts_columnar_skipnodes 14
#define Anum_columnar_skipnodes_storageid 1
#define Anum_columnar_skipnodes_stripe 2
#define Anum_columnar_skipnodes_attr 3
#define Anum_columnar_skipnodes_chunk 4
#define Anum_columnar_skipnodes_row_count 5
#define Anum_columnar_skipnodes_minimum_value 6
#define Anum_columnar_skipnodes_maximum_value 7
#define Anum_columnar_skipnodes_value_stream_offset 8
#define Anum_columnar_skipnodes_value_stream_length 9
#define Anum_columnar_skipnodes_exists_stream_offset 10
#define Anum_columnar_skipnodes_exists_stream_length 11
#define Anum_columnar_skipnodes_value_compression_type 12
#define Anum_columnar_skipnodes_value_compression_level 13
#define Anum_columnar_skipnodes_value_decompressed_size 14
/*
@ -173,9 +173,9 @@ InitColumnarOptions(Oid regclass)
}
ColumnarOptions defaultOptions = {
.chunkRowCount = cstore_chunk_row_count,
.stripeRowCount = cstore_stripe_row_count,
.compressionType = cstore_compression,
.chunkRowCount = columnar_chunk_row_count,
.stripeRowCount = columnar_stripe_row_count,
.compressionType = columnar_compression,
.compressionLevel = columnar_compression_level
};
@ -214,8 +214,8 @@ WriteColumnarOptions(Oid regclass, ColumnarOptions *options, bool overwrite)
bool written = false;
bool nulls[Natts_cstore_options] = { 0 };
Datum values[Natts_cstore_options] = {
bool nulls[Natts_columnar_options] = { 0 };
Datum values[Natts_columnar_options] = {
ObjectIdGetDatum(regclass),
Int32GetDatum(options->chunkRowCount),
Int32GetDatum(options->stripeRowCount),
@ -225,7 +225,7 @@ WriteColumnarOptions(Oid regclass, ColumnarOptions *options, bool overwrite)
NameData compressionName = { 0 };
namestrcpy(&compressionName, CompressionTypeStr(options->compressionType));
values[Anum_cstore_options_compression - 1] = NameGetDatum(&compressionName);
values[Anum_columnar_options_compression - 1] = NameGetDatum(&compressionName);
/* create heap tuple and insert into catalog table */
Relation columnarOptions = relation_open(ColumnarOptionsRelationId(),
@ -234,7 +234,8 @@ WriteColumnarOptions(Oid regclass, ColumnarOptions *options, bool overwrite)
/* find existing item to perform update if exist */
ScanKeyData scanKey[1] = { 0 };
ScanKeyInit(&scanKey[0], Anum_cstore_options_regclass, BTEqualStrategyNumber, F_OIDEQ,
ScanKeyInit(&scanKey[0], Anum_columnar_options_regclass, BTEqualStrategyNumber,
F_OIDEQ,
ObjectIdGetDatum(regclass));
Relation index = index_open(ColumnarOptionsIndexRegclass(), AccessShareLock);
@ -248,11 +249,11 @@ WriteColumnarOptions(Oid regclass, ColumnarOptions *options, bool overwrite)
{
/* TODO check if the options are actually different, skip if not changed */
/* update existing record */
bool update[Natts_cstore_options] = { 0 };
update[Anum_cstore_options_chunk_row_count - 1] = true;
update[Anum_cstore_options_stripe_row_count - 1] = true;
update[Anum_cstore_options_compression_level - 1] = true;
update[Anum_cstore_options_compression - 1] = true;
bool update[Natts_columnar_options] = { 0 };
update[Anum_columnar_options_chunk_row_count - 1] = true;
update[Anum_columnar_options_stripe_row_count - 1] = true;
update[Anum_columnar_options_compression_level - 1] = true;
update[Anum_columnar_options_compression - 1] = true;
HeapTuple tuple = heap_modify_tuple(heapTuple, tupleDescriptor,
values, nulls, update);
@ -303,7 +304,8 @@ DeleteColumnarTableOptions(Oid regclass, bool missingOk)
/* find existing item to remove */
ScanKeyData scanKey[1] = { 0 };
ScanKeyInit(&scanKey[0], Anum_cstore_options_regclass, BTEqualStrategyNumber, F_OIDEQ,
ScanKeyInit(&scanKey[0], Anum_columnar_options_regclass, BTEqualStrategyNumber,
F_OIDEQ,
ObjectIdGetDatum(regclass));
Relation index = index_open(ColumnarOptionsIndexRegclass(), AccessShareLock);
@ -336,7 +338,8 @@ ReadColumnarOptions(Oid regclass, ColumnarOptions *options)
{
ScanKeyData scanKey[1];
ScanKeyInit(&scanKey[0], Anum_cstore_options_regclass, BTEqualStrategyNumber, F_OIDEQ,
ScanKeyInit(&scanKey[0], Anum_columnar_options_regclass, BTEqualStrategyNumber,
F_OIDEQ,
ObjectIdGetDatum(regclass));
Oid columnarOptionsOid = ColumnarOptionsRelationId();
@ -365,7 +368,7 @@ ReadColumnarOptions(Oid regclass, ColumnarOptions *options)
HeapTuple heapTuple = systable_getnext(scanDescriptor);
if (HeapTupleIsValid(heapTuple))
{
Form_cstore_options tupOptions = (Form_cstore_options) GETSTRUCT(heapTuple);
Form_columnar_options tupOptions = (Form_columnar_options) GETSTRUCT(heapTuple);
options->chunkRowCount = tupOptions->chunk_row_count;
options->stripeRowCount = tupOptions->stripe_row_count;
@ -375,9 +378,9 @@ ReadColumnarOptions(Oid regclass, ColumnarOptions *options)
else
{
/* populate options with system defaults */
options->compressionType = cstore_compression;
options->stripeRowCount = cstore_stripe_row_count;
options->chunkRowCount = cstore_chunk_row_count;
options->compressionType = columnar_compression;
options->stripeRowCount = columnar_stripe_row_count;
options->chunkRowCount = columnar_chunk_row_count;
options->compressionLevel = columnar_compression_level;
}
@ -391,7 +394,7 @@ ReadColumnarOptions(Oid regclass, ColumnarOptions *options)
/*
* SaveStripeSkipList saves StripeSkipList for a given stripe as rows
* of cstore_skipnodes.
* of columnar_skipnodes.
*/
void
SaveStripeSkipList(RelFileNode relfilenode, uint64 stripe, StripeSkipList *stripeSkipList,
@ -402,7 +405,7 @@ SaveStripeSkipList(RelFileNode relfilenode, uint64 stripe, StripeSkipList *strip
uint32 columnCount = stripeSkipList->columnCount;
ColumnarMetapage *metapage = ReadMetapage(relfilenode, false);
Oid cstoreSkipNodesOid = CStoreSkipNodesRelationId();
Oid cstoreSkipNodesOid = ColumnarSkipNodesRelationId();
Relation cstoreSkipNodes = table_open(cstoreSkipNodesOid, RowExclusiveLock);
ModifyState *modifyState = StartModifyRelation(cstoreSkipNodes);
@ -413,7 +416,7 @@ SaveStripeSkipList(RelFileNode relfilenode, uint64 stripe, StripeSkipList *strip
ColumnChunkSkipNode *skipNode =
&stripeSkipList->chunkSkipNodeArray[columnIndex][chunkIndex];
Datum values[Natts_cstore_skipnodes] = {
Datum values[Natts_columnar_skipnodes] = {
UInt64GetDatum(metapage->storageId),
Int64GetDatum(stripe),
Int32GetDatum(columnIndex + 1),
@ -430,21 +433,21 @@ SaveStripeSkipList(RelFileNode relfilenode, uint64 stripe, StripeSkipList *strip
Int64GetDatum(skipNode->decompressedValueSize)
};
bool nulls[Natts_cstore_skipnodes] = { false };
bool nulls[Natts_columnar_skipnodes] = { false };
if (skipNode->hasMinMax)
{
values[Anum_cstore_skipnodes_minimum_value - 1] =
values[Anum_columnar_skipnodes_minimum_value - 1] =
PointerGetDatum(DatumToBytea(skipNode->minimumValue,
&tupleDescriptor->attrs[columnIndex]));
values[Anum_cstore_skipnodes_maximum_value - 1] =
values[Anum_columnar_skipnodes_maximum_value - 1] =
PointerGetDatum(DatumToBytea(skipNode->maximumValue,
&tupleDescriptor->attrs[columnIndex]));
}
else
{
nulls[Anum_cstore_skipnodes_minimum_value - 1] = true;
nulls[Anum_cstore_skipnodes_maximum_value - 1] = true;
nulls[Anum_columnar_skipnodes_minimum_value - 1] = true;
nulls[Anum_columnar_skipnodes_maximum_value - 1] = true;
}
InsertTupleAndEnforceConstraints(modifyState, values, nulls);
@ -472,13 +475,13 @@ ReadStripeSkipList(RelFileNode relfilenode, uint64 stripe, TupleDesc tupleDescri
ColumnarMetapage *metapage = ReadMetapage(relfilenode, false);
Oid cstoreSkipNodesOid = CStoreSkipNodesRelationId();
Oid cstoreSkipNodesOid = ColumnarSkipNodesRelationId();
Relation cstoreSkipNodes = table_open(cstoreSkipNodesOid, AccessShareLock);
Relation index = index_open(CStoreSkipNodesIndexRelationId(), AccessShareLock);
Relation index = index_open(ColumnarSkipNodesIndexRelationId(), AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_cstore_skipnodes_storageid,
ScanKeyInit(&scanKey[0], Anum_columnar_skipnodes_storageid,
BTEqualStrategyNumber, F_OIDEQ, UInt64GetDatum(metapage->storageId));
ScanKeyInit(&scanKey[1], Anum_cstore_skipnodes_stripe,
ScanKeyInit(&scanKey[1], Anum_columnar_skipnodes_stripe,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(stripe));
SysScanDesc scanDescriptor = systable_beginscan_ordered(cstoreSkipNodes, index, NULL,
@ -496,14 +499,14 @@ ReadStripeSkipList(RelFileNode relfilenode, uint64 stripe, TupleDesc tupleDescri
while (HeapTupleIsValid(heapTuple = systable_getnext(scanDescriptor)))
{
Datum datumArray[Natts_cstore_skipnodes];
bool isNullArray[Natts_cstore_skipnodes];
Datum datumArray[Natts_columnar_skipnodes];
bool isNullArray[Natts_columnar_skipnodes];
heap_deform_tuple(heapTuple, RelationGetDescr(cstoreSkipNodes), datumArray,
isNullArray);
int32 attr = DatumGetInt32(datumArray[Anum_cstore_skipnodes_attr - 1]);
int32 chunkIndex = DatumGetInt32(datumArray[Anum_cstore_skipnodes_chunk - 1]);
int32 attr = DatumGetInt32(datumArray[Anum_columnar_skipnodes_attr - 1]);
int32 chunkIndex = DatumGetInt32(datumArray[Anum_columnar_skipnodes_chunk - 1]);
if (attr <= 0 || attr > columnCount)
{
@ -521,34 +524,36 @@ ReadStripeSkipList(RelFileNode relfilenode, uint64 stripe, TupleDesc tupleDescri
ColumnChunkSkipNode *skipNode =
&skipList->chunkSkipNodeArray[columnIndex][chunkIndex];
skipNode->rowCount = DatumGetInt64(datumArray[Anum_cstore_skipnodes_row_count -
skipNode->rowCount = DatumGetInt64(datumArray[Anum_columnar_skipnodes_row_count -
1]);
skipNode->valueChunkOffset =
DatumGetInt64(datumArray[Anum_cstore_skipnodes_value_stream_offset - 1]);
DatumGetInt64(datumArray[Anum_columnar_skipnodes_value_stream_offset - 1]);
skipNode->valueLength =
DatumGetInt64(datumArray[Anum_cstore_skipnodes_value_stream_length - 1]);
DatumGetInt64(datumArray[Anum_columnar_skipnodes_value_stream_length - 1]);
skipNode->existsChunkOffset =
DatumGetInt64(datumArray[Anum_cstore_skipnodes_exists_stream_offset - 1]);
DatumGetInt64(datumArray[Anum_columnar_skipnodes_exists_stream_offset - 1]);
skipNode->existsLength =
DatumGetInt64(datumArray[Anum_cstore_skipnodes_exists_stream_length - 1]);
DatumGetInt64(datumArray[Anum_columnar_skipnodes_exists_stream_length - 1]);
skipNode->valueCompressionType =
DatumGetInt32(datumArray[Anum_cstore_skipnodes_value_compression_type - 1]);
DatumGetInt32(datumArray[Anum_columnar_skipnodes_value_compression_type - 1]);
skipNode->valueCompressionLevel =
DatumGetInt32(datumArray[Anum_cstore_skipnodes_value_compression_level - 1]);
DatumGetInt32(datumArray[Anum_columnar_skipnodes_value_compression_level -
1]);
skipNode->decompressedValueSize =
DatumGetInt64(datumArray[Anum_cstore_skipnodes_value_decompressed_size - 1]);
DatumGetInt64(datumArray[Anum_columnar_skipnodes_value_decompressed_size -
1]);
if (isNullArray[Anum_cstore_skipnodes_minimum_value - 1] ||
isNullArray[Anum_cstore_skipnodes_maximum_value - 1])
if (isNullArray[Anum_columnar_skipnodes_minimum_value - 1] ||
isNullArray[Anum_columnar_skipnodes_maximum_value - 1])
{
skipNode->hasMinMax = false;
}
else
{
bytea *minValue = DatumGetByteaP(
datumArray[Anum_cstore_skipnodes_minimum_value - 1]);
datumArray[Anum_columnar_skipnodes_minimum_value - 1]);
bytea *maxValue = DatumGetByteaP(
datumArray[Anum_cstore_skipnodes_maximum_value - 1]);
datumArray[Anum_columnar_skipnodes_maximum_value - 1]);
skipNode->minimumValue =
ByteaToDatum(minValue, &tupleDescriptor->attrs[columnIndex]);
@ -568,13 +573,13 @@ ReadStripeSkipList(RelFileNode relfilenode, uint64 stripe, TupleDesc tupleDescri
/*
* InsertStripeMetadataRow adds a row to cstore_stripes.
* InsertStripeMetadataRow adds a row to columnar_stripes.
*/
static void
InsertStripeMetadataRow(uint64 storageId, StripeMetadata *stripe)
{
bool nulls[Natts_cstore_stripes] = { 0 };
Datum values[Natts_cstore_stripes] = {
bool nulls[Natts_columnar_stripes] = { 0 };
Datum values[Natts_columnar_stripes] = {
UInt64GetDatum(storageId),
Int64GetDatum(stripe->id),
Int64GetDatum(stripe->fileOffset),
@ -585,7 +590,7 @@ InsertStripeMetadataRow(uint64 storageId, StripeMetadata *stripe)
Int64GetDatum(stripe->rowCount)
};
Oid cstoreStripesOid = CStoreStripesRelationId();
Oid cstoreStripesOid = ColumnarStripesRelationId();
Relation cstoreStripes = table_open(cstoreStripesOid, RowExclusiveLock);
ModifyState *modifyState = StartModifyRelation(cstoreStripes);
@ -679,7 +684,7 @@ GetHighestUsedAddressAndId(uint64 storageId,
/*
* ReserveStripe reserves and stripe of given size for the given relation,
* and inserts it into cstore_stripes. It is guaranteed that concurrent
* and inserts it into columnar_stripes. It is guaranteed that concurrent
* writes won't overwrite the returned stripe.
*/
StripeMetadata
@ -760,13 +765,13 @@ ReadDataFileStripeList(uint64 storageId, Snapshot snapshot)
ScanKeyData scanKey[1];
HeapTuple heapTuple;
ScanKeyInit(&scanKey[0], Anum_cstore_stripes_storageid,
ScanKeyInit(&scanKey[0], Anum_columnar_stripes_storageid,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(storageId));
Oid cstoreStripesOid = CStoreStripesRelationId();
Oid cstoreStripesOid = ColumnarStripesRelationId();
Relation cstoreStripes = table_open(cstoreStripesOid, AccessShareLock);
Relation index = index_open(CStoreStripesIndexRelationId(), AccessShareLock);
Relation index = index_open(ColumnarStripesIndexRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(cstoreStripes);
SysScanDesc scanDescriptor = systable_beginscan_ordered(cstoreStripes, index,
@ -775,25 +780,25 @@ ReadDataFileStripeList(uint64 storageId, Snapshot snapshot)
while (HeapTupleIsValid(heapTuple = systable_getnext(scanDescriptor)))
{
Datum datumArray[Natts_cstore_stripes];
bool isNullArray[Natts_cstore_stripes];
Datum datumArray[Natts_columnar_stripes];
bool isNullArray[Natts_columnar_stripes];
heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray);
StripeMetadata *stripeMetadata = palloc0(sizeof(StripeMetadata));
stripeMetadata->id = DatumGetInt64(datumArray[Anum_cstore_stripes_stripe - 1]);
stripeMetadata->id = DatumGetInt64(datumArray[Anum_columnar_stripes_stripe - 1]);
stripeMetadata->fileOffset = DatumGetInt64(
datumArray[Anum_cstore_stripes_file_offset - 1]);
datumArray[Anum_columnar_stripes_file_offset - 1]);
stripeMetadata->dataLength = DatumGetInt64(
datumArray[Anum_cstore_stripes_data_length - 1]);
datumArray[Anum_columnar_stripes_data_length - 1]);
stripeMetadata->columnCount = DatumGetInt32(
datumArray[Anum_cstore_stripes_column_count - 1]);
datumArray[Anum_columnar_stripes_column_count - 1]);
stripeMetadata->chunkCount = DatumGetInt32(
datumArray[Anum_cstore_stripes_chunk_count - 1]);
datumArray[Anum_columnar_stripes_chunk_count - 1]);
stripeMetadata->chunkRowCount = DatumGetInt32(
datumArray[Anum_cstore_stripes_chunk_row_count - 1]);
datumArray[Anum_columnar_stripes_chunk_row_count - 1]);
stripeMetadata->rowCount = DatumGetInt64(
datumArray[Anum_cstore_stripes_row_count - 1]);
datumArray[Anum_columnar_stripes_row_count - 1]);
stripeMetadataList = lappend(stripeMetadataList, stripeMetadata);
}
@ -807,7 +812,7 @@ ReadDataFileStripeList(uint64 storageId, Snapshot snapshot)
/*
* DeleteMetadataRows removes the rows with given relfilenode from cstore_stripes.
* DeleteMetadataRows removes the rows with given relfilenode from columnar_stripes.
*/
void
DeleteMetadataRows(RelFileNode relfilenode)
@ -833,10 +838,10 @@ DeleteMetadataRows(RelFileNode relfilenode)
return;
}
ScanKeyInit(&scanKey[0], Anum_cstore_stripes_storageid,
ScanKeyInit(&scanKey[0], Anum_columnar_stripes_storageid,
BTEqualStrategyNumber, F_INT8EQ, UInt64GetDatum(metapage->storageId));
Oid cstoreStripesOid = CStoreStripesRelationId();
Oid cstoreStripesOid = ColumnarStripesRelationId();
Relation cstoreStripes = try_relation_open(cstoreStripesOid, AccessShareLock);
if (cstoreStripes == NULL)
{
@ -844,7 +849,7 @@ DeleteMetadataRows(RelFileNode relfilenode)
return;
}
Relation index = index_open(CStoreStripesIndexRelationId(), AccessShareLock);
Relation index = index_open(ColumnarStripesIndexRelationId(), AccessShareLock);
SysScanDesc scanDescriptor = systable_beginscan_ordered(cstoreStripes, index, NULL,
1, scanKey);
@ -1046,24 +1051,24 @@ ByteaToDatum(bytea *bytes, Form_pg_attribute attrForm)
/*
* CStoreStripesRelationId returns relation id of cstore_stripes.
* ColumnarStripesRelationId returns relation id of columnar_stripes.
* TODO: should we cache this similar to citus?
*/
static Oid
CStoreStripesRelationId(void)
ColumnarStripesRelationId(void)
{
return get_relname_relid("columnar_stripes", CStoreNamespaceId());
return get_relname_relid("columnar_stripes", ColumnarNamespaceId());
}
/*
* CStoreStripesIndexRelationId returns relation id of cstore_stripes_idx.
* ColumnarStripesIndexRelationId returns relation id of columnar_stripes_idx.
* TODO: should we cache this similar to citus?
*/
static Oid
CStoreStripesIndexRelationId(void)
ColumnarStripesIndexRelationId(void)
{
return get_relname_relid("columnar_stripes_pkey", CStoreNamespaceId());
return get_relname_relid("columnar_stripes_pkey", ColumnarNamespaceId());
}
@ -1073,7 +1078,7 @@ CStoreStripesIndexRelationId(void)
static Oid
ColumnarOptionsRelationId(void)
{
return get_relname_relid("options", CStoreNamespaceId());
return get_relname_relid("options", ColumnarNamespaceId());
}
@ -1083,38 +1088,38 @@ ColumnarOptionsRelationId(void)
static Oid
ColumnarOptionsIndexRegclass(void)
{
return get_relname_relid("options_pkey", CStoreNamespaceId());
return get_relname_relid("options_pkey", ColumnarNamespaceId());
}
/*
* CStoreSkipNodesRelationId returns relation id of cstore_skipnodes.
* ColumnarSkipNodesRelationId returns relation id of columnar_skipnodes.
* TODO: should we cache this similar to citus?
*/
static Oid
CStoreSkipNodesRelationId(void)
ColumnarSkipNodesRelationId(void)
{
return get_relname_relid("columnar_skipnodes", CStoreNamespaceId());
return get_relname_relid("columnar_skipnodes", ColumnarNamespaceId());
}
/*
* CStoreSkipNodesIndexRelationId returns relation id of cstore_skipnodes_pkey.
* ColumnarSkipNodesIndexRelationId returns relation id of columnar_skipnodes_pkey.
* TODO: should we cache this similar to citus?
*/
static Oid
CStoreSkipNodesIndexRelationId(void)
ColumnarSkipNodesIndexRelationId(void)
{
return get_relname_relid("columnar_skipnodes_pkey", CStoreNamespaceId());
return get_relname_relid("columnar_skipnodes_pkey", ColumnarNamespaceId());
}
/*
* CStoreNamespaceId returns namespace id of the schema we store cstore
* ColumnarNamespaceId returns namespace id of the schema we store cstore
* related tables.
*/
static Oid
CStoreNamespaceId(void)
ColumnarNamespaceId(void)
{
return get_namespace_oid("columnar", false);
}
@ -1201,7 +1206,7 @@ GetNextStorageId(void)
{
Oid savedUserId = InvalidOid;
int savedSecurityContext = 0;
Oid sequenceId = get_relname_relid("storageid_seq", CStoreNamespaceId());
Oid sequenceId = get_relname_relid("storageid_seq", ColumnarNamespaceId());
Datum sequenceIdDatum = ObjectIdGetDatum(sequenceId);
/*
@ -1236,7 +1241,7 @@ columnar_relation_storageid(PG_FUNCTION_ARGS)
#if HAS_TABLEAM
Oid relationId = PG_GETARG_OID(0);
Relation relation = relation_open(relationId, AccessShareLock);
if (IsCStoreTableAmTable(relationId))
if (IsColumnarTableAmTable(relationId))
{
ColumnarMetapage *metadata = ReadMetapage(relation->rd_node, true);
if (metadata != NULL)

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

@ -80,11 +80,11 @@ static Datum ColumnDefaultValue(TupleConstr *tupleConstraints,
Form_pg_attribute attributeForm);
/*
* CStoreBeginRead initializes a cstore read operation. This function returns a
* ColumnarBeginRead initializes a cstore read operation. This function returns a
* read handle that's used during reading rows and finishing the read operation.
*/
TableReadState *
CStoreBeginRead(Relation relation, TupleDesc tupleDescriptor,
ColumnarBeginRead(Relation relation, TupleDesc tupleDescriptor,
List *projectedColumnList, List *whereClauseList)
{
List *stripeList = StripesForRelfilenode(relation->rd_node);
@ -117,12 +117,12 @@ CStoreBeginRead(Relation relation, TupleDesc tupleDescriptor,
/*
* CStoreReadNextRow tries to read a row from the cstore file. On success, it sets
* ColumnarReadNextRow tries to read a row from the cstore file. On success, it sets
* column values and nulls, and returns true. If there are no more rows to read,
* the function returns false.
*/
bool
CStoreReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNulls)
ColumnarReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNulls)
{
StripeMetadata *stripeMetadata = readState->currentStripeMetadata;
MemoryContext oldContext = NULL;
@ -223,11 +223,11 @@ CStoreReadNextRow(TableReadState *readState, Datum *columnValues, bool *columnNu
/*
* CStoreRescan clears the position where we were scanning so that the next read starts at
* ColumnarRescan clears the position where we were scanning so that the next read starts at
* the beginning again
*/
void
CStoreRescan(TableReadState *readState)
ColumnarRescan(TableReadState *readState)
{
readState->stripeBuffers = NULL;
readState->readStripeCount = 0;
@ -237,7 +237,7 @@ CStoreRescan(TableReadState *readState)
/* Finishes a cstore read operation. */
void
CStoreEndRead(TableReadState *readState)
ColumnarEndRead(TableReadState *readState)
{
MemoryContextDelete(readState->stripeReadContext);
list_free_deep(readState->stripeList);
@ -311,9 +311,9 @@ FreeChunkData(ChunkData *chunkData)
}
/* CStoreTableRowCount returns the exact row count of a table using skiplists */
/* ColumnarTableRowCount returns the exact row count of a table using skiplists */
uint64
CStoreTableRowCount(Relation relation)
ColumnarTableRowCount(Relation relation)
{
ListCell *stripeMetadataCell = NULL;
uint64 totalRowCount = 0;

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

@ -69,10 +69,10 @@
#define VACUUM_TRUNCATE_LOCK_TIMEOUT 4500 /* ms */
/*
* CStoreScanDescData is the scan state passed between beginscan(),
* ColumnarScanDescData is the scan state passed between beginscan(),
* getnextslot(), rescan(), and endscan() calls.
*/
typedef struct CStoreScanDescData
typedef struct ColumnarScanDescData
{
TableScanDescData cs_base;
TableReadState *cs_readState;
@ -90,26 +90,26 @@ typedef struct CStoreScanDescData
* number so we can construct an item pointer based on that.
*/
int rowNumber;
} CStoreScanDescData;
} ColumnarScanDescData;
typedef struct CStoreScanDescData *CStoreScanDesc;
typedef struct ColumnarScanDescData *ColumnarScanDesc;
static object_access_hook_type PrevObjectAccessHook = NULL;
/* forward declaration for static functions */
static void CStoreTableDropHook(Oid tgid);
static void CStoreTriggerCreateHook(Oid tgid);
static void CStoreTableAMObjectAccessHook(ObjectAccessType access, Oid classId,
static void ColumnarTableDropHook(Oid tgid);
static void ColumnarTriggerCreateHook(Oid tgid);
static void ColumnarTableAMObjectAccessHook(ObjectAccessType access, Oid classId,
Oid objectId, int subId,
void *arg);
static bool ConditionalLockRelationWithTimeout(Relation rel, LOCKMODE lockMode,
int timeout, int retryInterval);
static void LogRelationStats(Relation rel, int elevel);
static void TruncateCStore(Relation rel, int elevel);
static void TruncateColumnar(Relation rel, int elevel);
static HeapTuple ColumnarSlotCopyHeapTuple(TupleTableSlot *slot);
static void ColumnarCheckLogicalReplication(Relation rel);
/* Custom tuple slot ops used for columnar. Initialized in cstore_tableam_init(). */
/* Custom tuple slot ops used for columnar. Initialized in columnar_tableam_init(). */
TupleTableSlotOps TTSOpsColumnar;
static List *
@ -142,14 +142,14 @@ RelationColumnList(Relation rel)
static const TupleTableSlotOps *
cstore_slot_callbacks(Relation relation)
columnar_slot_callbacks(Relation relation)
{
return &TTSOpsColumnar;
}
static TableScanDesc
cstore_beginscan(Relation relation, Snapshot snapshot,
columnar_beginscan(Relation relation, Snapshot snapshot,
int nkeys, ScanKey key,
ParallelTableScanDesc parallel_scan,
uint32 flags)
@ -162,7 +162,7 @@ cstore_beginscan(Relation relation, Snapshot snapshot,
/* the cstore access method does not use the flags, they are specific to heap */
flags = 0;
TableScanDesc scandesc = cstore_beginscan_extended(relation, snapshot, nkeys, key,
TableScanDesc scandesc = columnar_beginscan_extended(relation, snapshot, nkeys, key,
parallel_scan,
flags, attr_needed, NULL);
@ -173,7 +173,7 @@ cstore_beginscan(Relation relation, Snapshot snapshot,
TableScanDesc
cstore_beginscan_extended(Relation relation, Snapshot snapshot,
columnar_beginscan_extended(Relation relation, Snapshot snapshot,
int nkeys, ScanKey key,
ParallelTableScanDesc parallel_scan,
uint32 flags, Bitmapset *attr_needed, List *scanQual)
@ -193,7 +193,7 @@ cstore_beginscan_extended(Relation relation, Snapshot snapshot,
MemoryContext oldContext = MemoryContextSwitchTo(scanContext);
CStoreScanDesc scan = palloc(sizeof(CStoreScanDescData));
ColumnarScanDesc scan = palloc(sizeof(ColumnarScanDescData));
scan->cs_base.rs_rd = relation;
scan->cs_base.rs_snapshot = snapshot;
scan->cs_base.rs_nkeys = nkeys;
@ -228,11 +228,11 @@ cstore_beginscan_extended(Relation relation, Snapshot snapshot,
/*
* init_cstore_read_state initializes a column store table read and returns the
* init_columnar_read_state initializes a column store table read and returns the
* state.
*/
static TableReadState *
init_cstore_read_state(Relation relation, TupleDesc tupdesc, Bitmapset *attr_needed,
init_columnar_read_state(Relation relation, TupleDesc tupdesc, Bitmapset *attr_needed,
List *scanQual)
{
List *columnList = RelationColumnList(relation);
@ -250,7 +250,7 @@ init_cstore_read_state(Relation relation, TupleDesc tupdesc, Bitmapset *attr_nee
}
}
TableReadState *readState = CStoreBeginRead(relation, tupdesc, neededColumnList,
TableReadState *readState = ColumnarBeginRead(relation, tupdesc, neededColumnList,
scanQual);
return readState;
@ -258,33 +258,33 @@ init_cstore_read_state(Relation relation, TupleDesc tupdesc, Bitmapset *attr_nee
static void
cstore_endscan(TableScanDesc sscan)
columnar_endscan(TableScanDesc sscan)
{
CStoreScanDesc scan = (CStoreScanDesc) sscan;
ColumnarScanDesc scan = (ColumnarScanDesc) sscan;
if (scan->cs_readState != NULL)
{
CStoreEndRead(scan->cs_readState);
ColumnarEndRead(scan->cs_readState);
scan->cs_readState = NULL;
}
}
static void
cstore_rescan(TableScanDesc sscan, ScanKey key, bool set_params,
columnar_rescan(TableScanDesc sscan, ScanKey key, bool set_params,
bool allow_strat, bool allow_sync, bool allow_pagemode)
{
CStoreScanDesc scan = (CStoreScanDesc) sscan;
ColumnarScanDesc scan = (ColumnarScanDesc) sscan;
if (scan->cs_readState != NULL)
{
CStoreRescan(scan->cs_readState);
ColumnarRescan(scan->cs_readState);
}
}
static bool
cstore_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot)
columnar_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot *slot)
{
CStoreScanDesc scan = (CStoreScanDesc) sscan;
ColumnarScanDesc scan = (ColumnarScanDesc) sscan;
/*
* if this is the first row, initialize read state.
@ -293,14 +293,14 @@ cstore_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot
{
MemoryContext oldContext = MemoryContextSwitchTo(scan->scanContext);
scan->cs_readState =
init_cstore_read_state(scan->cs_base.rs_rd, slot->tts_tupleDescriptor,
init_columnar_read_state(scan->cs_base.rs_rd, slot->tts_tupleDescriptor,
scan->attr_needed, scan->scanQual);
MemoryContextSwitchTo(oldContext);
}
ExecClearTuple(slot);
bool nextRowFound = CStoreReadNextRow(scan->cs_readState, slot->tts_values,
bool nextRowFound = ColumnarReadNextRow(scan->cs_readState, slot->tts_values,
slot->tts_isnull);
if (!nextRowFound)
@ -328,28 +328,28 @@ cstore_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableSlot
static Size
cstore_parallelscan_estimate(Relation rel)
columnar_parallelscan_estimate(Relation rel)
{
elog(ERROR, "columnar_parallelscan_estimate not implemented");
}
static Size
cstore_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan)
columnar_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan)
{
elog(ERROR, "columnar_parallelscan_initialize not implemented");
}
static void
cstore_parallelscan_reinitialize(Relation rel, ParallelTableScanDesc pscan)
columnar_parallelscan_reinitialize(Relation rel, ParallelTableScanDesc pscan)
{
elog(ERROR, "columnar_parallelscan_reinitialize not implemented");
}
static IndexFetchTableData *
cstore_index_fetch_begin(Relation rel)
columnar_index_fetch_begin(Relation rel)
{
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("indexes not supported for columnar tables")));
@ -357,7 +357,7 @@ cstore_index_fetch_begin(Relation rel)
static void
cstore_index_fetch_reset(IndexFetchTableData *scan)
columnar_index_fetch_reset(IndexFetchTableData *scan)
{
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("indexes not supported for columnar tables")));
@ -365,7 +365,7 @@ cstore_index_fetch_reset(IndexFetchTableData *scan)
static void
cstore_index_fetch_end(IndexFetchTableData *scan)
columnar_index_fetch_end(IndexFetchTableData *scan)
{
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("indexes not supported for columnar tables")));
@ -373,7 +373,7 @@ cstore_index_fetch_end(IndexFetchTableData *scan)
static bool
cstore_index_fetch_tuple(struct IndexFetchTableData *scan,
columnar_index_fetch_tuple(struct IndexFetchTableData *scan,
ItemPointer tid,
Snapshot snapshot,
TupleTableSlot *slot,
@ -385,7 +385,7 @@ cstore_index_fetch_tuple(struct IndexFetchTableData *scan,
static bool
cstore_fetch_row_version(Relation relation,
columnar_fetch_row_version(Relation relation,
ItemPointer tid,
Snapshot snapshot,
TupleTableSlot *slot)
@ -395,7 +395,7 @@ cstore_fetch_row_version(Relation relation,
static void
cstore_get_latest_tid(TableScanDesc sscan,
columnar_get_latest_tid(TableScanDesc sscan,
ItemPointer tid)
{
elog(ERROR, "columnar_get_latest_tid not implemented");
@ -403,14 +403,14 @@ cstore_get_latest_tid(TableScanDesc sscan,
static bool
cstore_tuple_tid_valid(TableScanDesc scan, ItemPointer tid)
columnar_tuple_tid_valid(TableScanDesc scan, ItemPointer tid)
{
elog(ERROR, "columnar_tuple_tid_valid not implemented");
}
static bool
cstore_tuple_satisfies_snapshot(Relation rel, TupleTableSlot *slot,
columnar_tuple_satisfies_snapshot(Relation rel, TupleTableSlot *slot,
Snapshot snapshot)
{
return true;
@ -418,7 +418,7 @@ cstore_tuple_satisfies_snapshot(Relation rel, TupleTableSlot *slot,
static TransactionId
cstore_compute_xid_horizon_for_tuples(Relation rel,
columnar_compute_xid_horizon_for_tuples(Relation rel,
ItemPointerData *tids,
int nitems)
{
@ -427,14 +427,14 @@ cstore_compute_xid_horizon_for_tuples(Relation rel,
static void
cstore_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
columnar_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
int options, BulkInsertState bistate)
{
/*
* cstore_init_write_state allocates the write state in a longer
* columnar_init_write_state allocates the write state in a longer
* lasting context, so no need to worry about it.
*/
TableWriteState *writeState = cstore_init_write_state(relation,
TableWriteState *writeState = columnar_init_write_state(relation,
RelationGetDescr(relation),
GetCurrentSubTransactionId());
@ -454,7 +454,7 @@ cstore_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
slot_getallattrs(slot);
CStoreWriteRow(writeState, slot->tts_values, slot->tts_isnull);
ColumnarWriteRow(writeState, slot->tts_values, slot->tts_isnull);
MemoryContextSwitchTo(oldContext);
MemoryContextReset(writeState->perTupleContext);
@ -462,7 +462,7 @@ cstore_tuple_insert(Relation relation, TupleTableSlot *slot, CommandId cid,
static void
cstore_tuple_insert_speculative(Relation relation, TupleTableSlot *slot,
columnar_tuple_insert_speculative(Relation relation, TupleTableSlot *slot,
CommandId cid, int options,
BulkInsertState bistate, uint32 specToken)
{
@ -471,7 +471,7 @@ cstore_tuple_insert_speculative(Relation relation, TupleTableSlot *slot,
static void
cstore_tuple_complete_speculative(Relation relation, TupleTableSlot *slot,
columnar_tuple_complete_speculative(Relation relation, TupleTableSlot *slot,
uint32 specToken, bool succeeded)
{
elog(ERROR, "columnar_tuple_complete_speculative not implemented");
@ -479,10 +479,10 @@ cstore_tuple_complete_speculative(Relation relation, TupleTableSlot *slot,
static void
cstore_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
columnar_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
CommandId cid, int options, BulkInsertState bistate)
{
TableWriteState *writeState = cstore_init_write_state(relation,
TableWriteState *writeState = columnar_init_write_state(relation,
RelationGetDescr(relation),
GetCurrentSubTransactionId());
@ -504,7 +504,7 @@ cstore_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
slot_getallattrs(tupleSlot);
CStoreWriteRow(writeState, tupleSlot->tts_values, tupleSlot->tts_isnull);
ColumnarWriteRow(writeState, tupleSlot->tts_values, tupleSlot->tts_isnull);
MemoryContextSwitchTo(oldContext);
}
@ -513,7 +513,7 @@ cstore_multi_insert(Relation relation, TupleTableSlot **slots, int ntuples,
static TM_Result
cstore_tuple_delete(Relation relation, ItemPointer tid, CommandId cid,
columnar_tuple_delete(Relation relation, ItemPointer tid, CommandId cid,
Snapshot snapshot, Snapshot crosscheck, bool wait,
TM_FailureData *tmfd, bool changingPart)
{
@ -522,7 +522,7 @@ cstore_tuple_delete(Relation relation, ItemPointer tid, CommandId cid,
static TM_Result
cstore_tuple_update(Relation relation, ItemPointer otid, TupleTableSlot *slot,
columnar_tuple_update(Relation relation, ItemPointer otid, TupleTableSlot *slot,
CommandId cid, Snapshot snapshot, Snapshot crosscheck,
bool wait, TM_FailureData *tmfd,
LockTupleMode *lockmode, bool *update_indexes)
@ -532,7 +532,7 @@ cstore_tuple_update(Relation relation, ItemPointer otid, TupleTableSlot *slot,
static TM_Result
cstore_tuple_lock(Relation relation, ItemPointer tid, Snapshot snapshot,
columnar_tuple_lock(Relation relation, ItemPointer tid, Snapshot snapshot,
TupleTableSlot *slot, CommandId cid, LockTupleMode mode,
LockWaitPolicy wait_policy, uint8 flags,
TM_FailureData *tmfd)
@ -542,7 +542,7 @@ cstore_tuple_lock(Relation relation, ItemPointer tid, Snapshot snapshot,
static void
cstore_finish_bulk_insert(Relation relation, int options)
columnar_finish_bulk_insert(Relation relation, int options)
{
/*
* Nothing to do here. We keep write states live until transaction end.
@ -551,7 +551,7 @@ cstore_finish_bulk_insert(Relation relation, int options)
static void
cstore_relation_set_new_filenode(Relation rel,
columnar_relation_set_new_filenode(Relation rel,
const RelFileNode *newrnode,
char persistence,
TransactionId *freezeXid,
@ -584,7 +584,7 @@ cstore_relation_set_new_filenode(Relation rel,
static void
cstore_relation_nontransactional_truncate(Relation rel)
columnar_relation_nontransactional_truncate(Relation rel)
{
RelFileNode relfilenode = rel->rd_node;
@ -607,21 +607,21 @@ cstore_relation_nontransactional_truncate(Relation rel)
static void
cstore_relation_copy_data(Relation rel, const RelFileNode *newrnode)
columnar_relation_copy_data(Relation rel, const RelFileNode *newrnode)
{
elog(ERROR, "columnar_relation_copy_data not implemented");
}
/*
* cstore_relation_copy_for_cluster is called on VACUUM FULL, at which
* columnar_relation_copy_for_cluster is called on VACUUM FULL, at which
* we should copy data from OldHeap to NewHeap.
*
* In general TableAM case this can also be called for the CLUSTER command
* which is not applicable for cstore since it doesn't support indexes.
*/
static void
cstore_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
columnar_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
Relation OldIndex, bool use_sort,
TransactionId OldestXmin,
TransactionId *xid_cutoff,
@ -650,11 +650,11 @@ cstore_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
ColumnarOptions cstoreOptions = { 0 };
ReadColumnarOptions(OldHeap->rd_id, &cstoreOptions);
TableWriteState *writeState = CStoreBeginWrite(NewHeap->rd_node,
TableWriteState *writeState = ColumnarBeginWrite(NewHeap->rd_node,
cstoreOptions,
targetDesc);
TableReadState *readState = CStoreBeginRead(OldHeap, sourceDesc,
TableReadState *readState = ColumnarBeginRead(OldHeap, sourceDesc,
RelationColumnList(OldHeap), NULL);
Datum *values = palloc0(sourceDesc->natts * sizeof(Datum));
@ -662,24 +662,24 @@ cstore_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
*num_tuples = 0;
while (CStoreReadNextRow(readState, values, nulls))
while (ColumnarReadNextRow(readState, values, nulls))
{
CStoreWriteRow(writeState, values, nulls);
ColumnarWriteRow(writeState, values, nulls);
(*num_tuples)++;
}
*tups_vacuumed = 0;
CStoreEndWrite(writeState);
CStoreEndRead(readState);
ColumnarEndWrite(writeState);
ColumnarEndRead(readState);
}
/*
* cstore_vacuum_rel implements VACUUM without FULL option.
* columnar_vacuum_rel implements VACUUM without FULL option.
*/
static void
cstore_vacuum_rel(Relation rel, VacuumParams *params,
columnar_vacuum_rel(Relation rel, VacuumParams *params,
BufferAccessStrategy bstrategy)
{
int elevel = (params->options & VACOPT_VERBOSE) ? INFO : DEBUG2;
@ -695,7 +695,7 @@ cstore_vacuum_rel(Relation rel, VacuumParams *params,
*/
if (params->truncate == VACOPT_TERNARY_ENABLED)
{
TruncateCStore(rel, elevel);
TruncateColumnar(rel, elevel);
}
}
@ -813,14 +813,14 @@ LogRelationStats(Relation rel, int elevel)
/*
* TruncateCStore truncates the unused space at the end of main fork for
* TruncateColumnar truncates the unused space at the end of main fork for
* a cstore table. This unused space can be created by aborted transactions.
*
* This implementation is based on heap_vacuum_rel in vacuumlazy.c with some
* changes so it suits columnar store relations.
*/
static void
TruncateCStore(Relation rel, int elevel)
TruncateColumnar(Relation rel, int elevel)
{
PGRUsage ru0;
@ -939,21 +939,21 @@ ConditionalLockRelationWithTimeout(Relation rel, LOCKMODE lockMode, int timeout,
static bool
cstore_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
columnar_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
BufferAccessStrategy bstrategy)
{
/*
* Our access method is not pages based, i.e. tuples are not confined
* to pages boundaries. So not much to do here. We return true anyway
* so acquire_sample_rows() in analyze.c would call our
* cstore_scan_analyze_next_tuple() callback.
* columnar_scan_analyze_next_tuple() callback.
*/
return true;
}
static bool
cstore_scan_analyze_next_tuple(TableScanDesc scan, TransactionId OldestXmin,
columnar_scan_analyze_next_tuple(TableScanDesc scan, TransactionId OldestXmin,
double *liverows, double *deadrows,
TupleTableSlot *slot)
{
@ -964,11 +964,11 @@ cstore_scan_analyze_next_tuple(TableScanDesc scan, TransactionId OldestXmin,
* tuples from those pages.
*
* We could do something like that here by choosing sample stripes or chunks,
* but getting that correct might need quite some work. Since cstore_fdw's
* but getting that correct might need quite some work. Since columnar_fdw's
* ANALYZE scanned all rows, as a starter we do the same here and scan all
* rows.
*/
if (cstore_getnextslot(scan, ForwardScanDirection, slot))
if (columnar_getnextslot(scan, ForwardScanDirection, slot))
{
(*liverows)++;
return true;
@ -979,7 +979,7 @@ cstore_scan_analyze_next_tuple(TableScanDesc scan, TransactionId OldestXmin,
static double
cstore_index_build_range_scan(Relation heapRelation,
columnar_index_build_range_scan(Relation heapRelation,
Relation indexRelation,
IndexInfo *indexInfo,
bool allow_sync,
@ -997,7 +997,7 @@ cstore_index_build_range_scan(Relation heapRelation,
static void
cstore_index_validate_scan(Relation heapRelation,
columnar_index_validate_scan(Relation heapRelation,
Relation indexRelation,
IndexInfo *indexInfo,
Snapshot snapshot,
@ -1009,7 +1009,7 @@ cstore_index_validate_scan(Relation heapRelation,
static uint64
cstore_relation_size(Relation rel, ForkNumber forkNumber)
columnar_relation_size(Relation rel, ForkNumber forkNumber)
{
uint64 nblocks = 0;
@ -1034,20 +1034,20 @@ cstore_relation_size(Relation rel, ForkNumber forkNumber)
static bool
cstore_relation_needs_toast_table(Relation rel)
columnar_relation_needs_toast_table(Relation rel)
{
return false;
}
static void
cstore_estimate_rel_size(Relation rel, int32 *attr_widths,
columnar_estimate_rel_size(Relation rel, int32 *attr_widths,
BlockNumber *pages, double *tuples,
double *allvisfrac)
{
RelationOpenSmgr(rel);
*pages = smgrnblocks(rel->rd_smgr, MAIN_FORKNUM);
*tuples = CStoreTableRowCount(rel);
*tuples = ColumnarTableRowCount(rel);
/*
* Append-only, so everything is visible except in-progress or rolled-back
@ -1060,14 +1060,14 @@ cstore_estimate_rel_size(Relation rel, int32 *attr_widths,
static bool
cstore_scan_sample_next_block(TableScanDesc scan, SampleScanState *scanstate)
columnar_scan_sample_next_block(TableScanDesc scan, SampleScanState *scanstate)
{
elog(ERROR, "columnar_scan_sample_next_block not implemented");
}
static bool
cstore_scan_sample_next_tuple(TableScanDesc scan, SampleScanState *scanstate,
columnar_scan_sample_next_tuple(TableScanDesc scan, SampleScanState *scanstate,
TupleTableSlot *slot)
{
elog(ERROR, "columnar_scan_sample_next_tuple not implemented");
@ -1075,7 +1075,7 @@ cstore_scan_sample_next_tuple(TableScanDesc scan, SampleScanState *scanstate,
static void
CStoreXactCallback(XactEvent event, void *arg)
ColumnarXactCallback(XactEvent event, void *arg)
{
switch (event)
{
@ -1106,7 +1106,7 @@ CStoreXactCallback(XactEvent event, void *arg)
static void
CStoreSubXactCallback(SubXactEvent event, SubTransactionId mySubid,
ColumnarSubXactCallback(SubXactEvent event, SubTransactionId mySubid,
SubTransactionId parentSubid, void *arg)
{
switch (event)
@ -1134,15 +1134,15 @@ CStoreSubXactCallback(SubXactEvent event, SubTransactionId mySubid,
void
cstore_tableam_init()
columnar_tableam_init()
{
RegisterXactCallback(CStoreXactCallback, NULL);
RegisterSubXactCallback(CStoreSubXactCallback, NULL);
RegisterXactCallback(ColumnarXactCallback, NULL);
RegisterSubXactCallback(ColumnarSubXactCallback, NULL);
PrevObjectAccessHook = object_access_hook;
object_access_hook = CStoreTableAMObjectAccessHook;
object_access_hook = ColumnarTableAMObjectAccessHook;
cstore_customscan_init();
columnar_customscan_init();
TTSOpsColumnar = TTSOpsVirtual;
TTSOpsColumnar.copy_heap_tuple = ColumnarSlotCopyHeapTuple;
@ -1150,7 +1150,7 @@ cstore_tableam_init()
void
cstore_tableam_finish()
columnar_tableam_finish()
{
object_access_hook = PrevObjectAccessHook;
}
@ -1162,7 +1162,7 @@ cstore_tableam_finish()
int64
ColumnarGetChunksFiltered(TableScanDesc scanDesc)
{
CStoreScanDesc cstoreScanDesc = (CStoreScanDesc) scanDesc;
ColumnarScanDesc cstoreScanDesc = (ColumnarScanDesc) scanDesc;
TableReadState *readState = cstoreScanDesc->cs_readState;
if (readState != NULL)
@ -1193,7 +1193,7 @@ ColumnarSlotCopyHeapTuple(TupleTableSlot *slot)
* requires it. See the qsort in acquire_sample_rows() and
* also compare_rows in backend/commands/analyze.c.
*
* slot->tts_tid is filled in cstore_getnextslot.
* slot->tts_tid is filled in columnar_getnextslot.
*/
tuple->t_self = slot->tts_tid;
@ -1202,12 +1202,12 @@ ColumnarSlotCopyHeapTuple(TupleTableSlot *slot)
/*
* CStoreTableDropHook
* ColumnarTableDropHook
*
* Clean-up resources for columnar tables.
*/
static void
CStoreTableDropHook(Oid relid)
ColumnarTableDropHook(Oid relid)
{
/*
* Lock relation to prevent it from being dropped and to avoid
@ -1215,7 +1215,7 @@ CStoreTableDropHook(Oid relid)
*/
LockRelationOid(relid, AccessShareLock);
if (IsCStoreTableAmTable(relid))
if (IsColumnarTableAmTable(relid))
{
/*
* Drop metadata. No need to drop storage here since for
@ -1239,7 +1239,7 @@ CStoreTableDropHook(Oid relid)
* Reject AFTER ... FOR EACH ROW triggers on columnar tables.
*/
static void
CStoreTriggerCreateHook(Oid tgid)
ColumnarTriggerCreateHook(Oid tgid)
{
/*
* Fetch the pg_trigger tuple by the Oid of the trigger
@ -1272,7 +1272,7 @@ CStoreTriggerCreateHook(Oid tgid)
table_close(tgrel, AccessShareLock);
if (TRIGGER_FOR_ROW(tgtype) && TRIGGER_FOR_AFTER(tgtype) &&
IsCStoreTableAmTable(tgrelid))
IsColumnarTableAmTable(tgrelid))
{
ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg(
@ -1286,7 +1286,7 @@ CStoreTriggerCreateHook(Oid tgid)
* Capture create/drop events and dispatch to the proper action.
*/
static void
CStoreTableAMObjectAccessHook(ObjectAccessType access, Oid classId, Oid objectId,
ColumnarTableAMObjectAccessHook(ObjectAccessType access, Oid classId, Oid objectId,
int subId, void *arg)
{
if (PrevObjectAccessHook)
@ -1297,21 +1297,21 @@ CStoreTableAMObjectAccessHook(ObjectAccessType access, Oid classId, Oid objectId
/* dispatch to the proper action */
if (access == OAT_DROP && classId == RelationRelationId && !OidIsValid(subId))
{
CStoreTableDropHook(objectId);
ColumnarTableDropHook(objectId);
}
else if (access == OAT_POST_CREATE && classId == TriggerRelationId)
{
CStoreTriggerCreateHook(objectId);
ColumnarTriggerCreateHook(objectId);
}
}
/*
* IsCStoreTableAmTable returns true if relation has cstore_tableam
* IsColumnarTableAmTable returns true if relation has columnar_tableam
* access method. This can be called before extension creation.
*/
bool
IsCStoreTableAmTable(Oid relationId)
IsColumnarTableAmTable(Oid relationId)
{
if (!OidIsValid(relationId))
{
@ -1330,66 +1330,66 @@ IsCStoreTableAmTable(Oid relationId)
}
static const TableAmRoutine cstore_am_methods = {
static const TableAmRoutine columnar_am_methods = {
.type = T_TableAmRoutine,
.slot_callbacks = cstore_slot_callbacks,
.slot_callbacks = columnar_slot_callbacks,
.scan_begin = cstore_beginscan,
.scan_end = cstore_endscan,
.scan_rescan = cstore_rescan,
.scan_getnextslot = cstore_getnextslot,
.scan_begin = columnar_beginscan,
.scan_end = columnar_endscan,
.scan_rescan = columnar_rescan,
.scan_getnextslot = columnar_getnextslot,
.parallelscan_estimate = cstore_parallelscan_estimate,
.parallelscan_initialize = cstore_parallelscan_initialize,
.parallelscan_reinitialize = cstore_parallelscan_reinitialize,
.parallelscan_estimate = columnar_parallelscan_estimate,
.parallelscan_initialize = columnar_parallelscan_initialize,
.parallelscan_reinitialize = columnar_parallelscan_reinitialize,
.index_fetch_begin = cstore_index_fetch_begin,
.index_fetch_reset = cstore_index_fetch_reset,
.index_fetch_end = cstore_index_fetch_end,
.index_fetch_tuple = cstore_index_fetch_tuple,
.index_fetch_begin = columnar_index_fetch_begin,
.index_fetch_reset = columnar_index_fetch_reset,
.index_fetch_end = columnar_index_fetch_end,
.index_fetch_tuple = columnar_index_fetch_tuple,
.tuple_fetch_row_version = cstore_fetch_row_version,
.tuple_get_latest_tid = cstore_get_latest_tid,
.tuple_tid_valid = cstore_tuple_tid_valid,
.tuple_satisfies_snapshot = cstore_tuple_satisfies_snapshot,
.compute_xid_horizon_for_tuples = cstore_compute_xid_horizon_for_tuples,
.tuple_fetch_row_version = columnar_fetch_row_version,
.tuple_get_latest_tid = columnar_get_latest_tid,
.tuple_tid_valid = columnar_tuple_tid_valid,
.tuple_satisfies_snapshot = columnar_tuple_satisfies_snapshot,
.compute_xid_horizon_for_tuples = columnar_compute_xid_horizon_for_tuples,
.tuple_insert = cstore_tuple_insert,
.tuple_insert_speculative = cstore_tuple_insert_speculative,
.tuple_complete_speculative = cstore_tuple_complete_speculative,
.multi_insert = cstore_multi_insert,
.tuple_delete = cstore_tuple_delete,
.tuple_update = cstore_tuple_update,
.tuple_lock = cstore_tuple_lock,
.finish_bulk_insert = cstore_finish_bulk_insert,
.tuple_insert = columnar_tuple_insert,
.tuple_insert_speculative = columnar_tuple_insert_speculative,
.tuple_complete_speculative = columnar_tuple_complete_speculative,
.multi_insert = columnar_multi_insert,
.tuple_delete = columnar_tuple_delete,
.tuple_update = columnar_tuple_update,
.tuple_lock = columnar_tuple_lock,
.finish_bulk_insert = columnar_finish_bulk_insert,
.relation_set_new_filenode = cstore_relation_set_new_filenode,
.relation_nontransactional_truncate = cstore_relation_nontransactional_truncate,
.relation_copy_data = cstore_relation_copy_data,
.relation_copy_for_cluster = cstore_relation_copy_for_cluster,
.relation_vacuum = cstore_vacuum_rel,
.scan_analyze_next_block = cstore_scan_analyze_next_block,
.scan_analyze_next_tuple = cstore_scan_analyze_next_tuple,
.index_build_range_scan = cstore_index_build_range_scan,
.index_validate_scan = cstore_index_validate_scan,
.relation_set_new_filenode = columnar_relation_set_new_filenode,
.relation_nontransactional_truncate = columnar_relation_nontransactional_truncate,
.relation_copy_data = columnar_relation_copy_data,
.relation_copy_for_cluster = columnar_relation_copy_for_cluster,
.relation_vacuum = columnar_vacuum_rel,
.scan_analyze_next_block = columnar_scan_analyze_next_block,
.scan_analyze_next_tuple = columnar_scan_analyze_next_tuple,
.index_build_range_scan = columnar_index_build_range_scan,
.index_validate_scan = columnar_index_validate_scan,
.relation_size = cstore_relation_size,
.relation_needs_toast_table = cstore_relation_needs_toast_table,
.relation_size = columnar_relation_size,
.relation_needs_toast_table = columnar_relation_needs_toast_table,
.relation_estimate_size = cstore_estimate_rel_size,
.relation_estimate_size = columnar_estimate_rel_size,
.scan_bitmap_next_block = NULL,
.scan_bitmap_next_tuple = NULL,
.scan_sample_next_block = cstore_scan_sample_next_block,
.scan_sample_next_tuple = cstore_scan_sample_next_tuple
.scan_sample_next_block = columnar_scan_sample_next_block,
.scan_sample_next_tuple = columnar_scan_sample_next_tuple
};
const TableAmRoutine *
GetColumnarTableAmRoutine(void)
{
return &cstore_am_methods;
return &columnar_am_methods;
}
@ -1397,7 +1397,7 @@ PG_FUNCTION_INFO_V1(columnar_handler);
Datum
columnar_handler(PG_FUNCTION_ARGS)
{
PG_RETURN_POINTER(&cstore_am_methods);
PG_RETURN_POINTER(&columnar_am_methods);
}
@ -1573,7 +1573,7 @@ alter_columnar_table_set(PG_FUNCTION_ARGS)
Oid relationId = PG_GETARG_OID(0);
Relation rel = table_open(relationId, AccessExclusiveLock); /* ALTER TABLE LOCK */
if (!IsCStoreTableAmTable(relationId))
if (!IsColumnarTableAmTable(relationId))
{
ereport(ERROR, (errmsg("table %s is not a columnar table",
quote_identifier(RelationGetRelationName(rel)))));
@ -1680,7 +1680,7 @@ alter_columnar_table_reset(PG_FUNCTION_ARGS)
Oid relationId = PG_GETARG_OID(0);
Relation rel = table_open(relationId, AccessExclusiveLock); /* ALTER TABLE LOCK */
if (!IsCStoreTableAmTable(relationId))
if (!IsColumnarTableAmTable(relationId))
{
ereport(ERROR, (errmsg("table %s is not a columnar table",
quote_identifier(RelationGetRelationName(rel)))));
@ -1695,7 +1695,7 @@ alter_columnar_table_reset(PG_FUNCTION_ARGS)
/* chunk_row_count => true */
if (!PG_ARGISNULL(1) && PG_GETARG_BOOL(1))
{
options.chunkRowCount = cstore_chunk_row_count;
options.chunkRowCount = columnar_chunk_row_count;
ereport(DEBUG1,
(errmsg("resetting chunk row count to %d", options.chunkRowCount)));
}
@ -1703,7 +1703,7 @@ alter_columnar_table_reset(PG_FUNCTION_ARGS)
/* stripe_row_count => true */
if (!PG_ARGISNULL(2) && PG_GETARG_BOOL(2))
{
options.stripeRowCount = cstore_stripe_row_count;
options.stripeRowCount = columnar_stripe_row_count;
ereport(DEBUG1,
(errmsg("resetting stripe row count to " UINT64_FORMAT,
options.stripeRowCount)));
@ -1712,7 +1712,7 @@ alter_columnar_table_reset(PG_FUNCTION_ARGS)
/* compression => true */
if (!PG_ARGISNULL(3) && PG_GETARG_BOOL(3))
{
options.compressionType = cstore_compression;
options.compressionType = columnar_compression;
ereport(DEBUG1, (errmsg("resetting compression to %s",
CompressionTypeStr(options.compressionType))));
}

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

@ -53,14 +53,14 @@ static Datum DatumCopy(Datum datum, bool datumTypeByValue, int datumTypeLength);
static StringInfo CopyStringInfo(StringInfo sourceString);
/*
* CStoreBeginWrite initializes a cstore data load operation and returns a table
* ColumnarBeginWrite initializes a cstore data load operation and returns a table
* handle. This handle should be used for adding the row values and finishing the
* data load operation. If the cstore footer file already exists, we read the
* footer and then seek to right after the last stripe where the new stripes
* will be added.
*/
TableWriteState *
CStoreBeginWrite(RelFileNode relfilenode,
ColumnarBeginWrite(RelFileNode relfilenode,
ColumnarOptions options,
TupleDesc tupleDescriptor)
{
@ -110,7 +110,7 @@ CStoreBeginWrite(RelFileNode relfilenode,
writeState->chunkData = chunkData;
writeState->compressionBuffer = NULL;
writeState->perTupleContext = AllocSetContextCreate(CurrentMemoryContext,
"CStore per tuple context",
"Columnar per tuple context",
ALLOCSET_DEFAULT_SIZES);
return writeState;
@ -118,7 +118,7 @@ CStoreBeginWrite(RelFileNode relfilenode,
/*
* CStoreWriteRow adds a row to the cstore file. If the stripe is not initialized,
* ColumnarWriteRow adds a row to the cstore file. If the stripe is not initialized,
* we create structures to hold stripe data and skip list. Then, we serialize and
* append data to serialized value buffer for each of the columns and update
* corresponding skip nodes. Then, whole chunk data is compressed at every
@ -126,7 +126,7 @@ CStoreBeginWrite(RelFileNode relfilenode,
* the stripe, and add its metadata to the table footer.
*/
void
CStoreWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNulls)
ColumnarWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNulls)
{
uint32 columnIndex = 0;
StripeBuffers *stripeBuffers = writeState->stripeBuffers;
@ -206,7 +206,7 @@ CStoreWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNul
stripeBuffers->rowCount++;
if (stripeBuffers->rowCount >= options->stripeRowCount)
{
CStoreFlushPendingWrites(writeState);
ColumnarFlushPendingWrites(writeState);
}
MemoryContextSwitchTo(oldContext);
@ -214,15 +214,15 @@ CStoreWriteRow(TableWriteState *writeState, Datum *columnValues, bool *columnNul
/*
* CStoreEndWrite finishes a cstore data load operation. If we have an unflushed
* ColumnarEndWrite finishes a cstore data load operation. If we have an unflushed
* stripe, we flush it. Then, we sync and close the cstore data file. Last, we
* flush the footer to a temporary file, and atomically rename this temporary
* file to the original footer file.
*/
void
CStoreEndWrite(TableWriteState *writeState)
ColumnarEndWrite(TableWriteState *writeState)
{
CStoreFlushPendingWrites(writeState);
ColumnarFlushPendingWrites(writeState);
MemoryContextDelete(writeState->stripeWriteContext);
pfree(writeState->comparisonFunctionArray);
@ -232,7 +232,7 @@ CStoreEndWrite(TableWriteState *writeState)
void
CStoreFlushPendingWrites(TableWriteState *writeState)
ColumnarFlushPendingWrites(TableWriteState *writeState)
{
StripeBuffers *stripeBuffers = writeState->stripeBuffers;
if (stripeBuffers != NULL)

6
src/backend/columnar/mod.c
View File

@ -27,9 +27,9 @@
void
columnar_init(void)
{
cstore_init();
columnar_init_gucs();
#ifdef HAS_TABLEAM
cstore_tableam_init();
columnar_tableam_init();
#endif
}
@ -38,6 +38,6 @@ void
columnar_fini(void)
{
#if HAS_TABLEAM
cstore_tableam_finish();
columnar_tableam_finish();
#endif
}

8
src/backend/columnar/write_state_management.c
View File

@ -116,7 +116,7 @@ CleanupWriteStateMap(void *arg)
TableWriteState *
cstore_init_write_state(Relation relation, TupleDesc tupdesc,
columnar_init_write_state(Relation relation, TupleDesc tupdesc,
SubTransactionId currentSubXid)
{
bool found;
@ -182,7 +182,7 @@ cstore_init_write_state(Relation relation, TupleDesc tupdesc,
ReadColumnarOptions(relation->rd_id, &cstoreOptions);
SubXidWriteState *stackEntry = palloc0(sizeof(SubXidWriteState));
stackEntry->writeState = CStoreBeginWrite(relation->rd_node,
stackEntry->writeState = ColumnarBeginWrite(relation->rd_node,
cstoreOptions,
tupdesc);
stackEntry->subXid = currentSubXid;
@ -215,7 +215,7 @@ FlushWriteStateForRelfilenode(Oid relfilenode, SubTransactionId currentSubXid)
SubXidWriteState *stackEntry = entry->writeStateStack;
if (stackEntry->subXid == currentSubXid)
{
CStoreFlushPendingWrites(stackEntry->writeState);
ColumnarFlushPendingWrites(stackEntry->writeState);
}
}
}
@ -279,7 +279,7 @@ PopWriteStateForAllRels(SubTransactionId currentSubXid, SubTransactionId parentS
{
if (commit)
{
CStoreEndWrite(stackHead->writeState);
ColumnarEndWrite(stackHead->writeState);
}
entry->writeStateStack = stackHead->next;

2
src/backend/distributed/operations/node_protocol.c
View File

@ -660,7 +660,7 @@ GetPreLoadTableCreationCommands(Oid relationId, bool includeSequenceDefaults,
#if PG_VERSION_NUM >= 120000
/* add columnar options for cstore tables */
if (accessMethod == NULL && IsCStoreTableAmTable(relationId))
if (accessMethod == NULL && IsColumnarTableAmTable(relationId))
{
TableDDLCommand *cstoreOptionsDDL = ColumnarGetTableOptionsDDL(relationId);
if (cstoreOptionsDDL != NULL)

42
src/include/columnar/cstore.h
View File

@ -2,7 +2,7 @@
*
* cstore.h
*
* Type and function declarations for CStore
* Type and function declarations for Columnar
*
* Copyright (c) 2016, Citus Data, Inc.
*
@ -41,7 +41,7 @@
#define COMPRESSION_STRING_NONE "none"
#define COMPRESSION_STRING_PG_LZ "pglz"
/* CStore file signature */
/* Columnar file signature */
#define CSTORE_MAGIC_NUMBER "citus_cstore"
#define CSTORE_VERSION_MAJOR 1
#define CSTORE_VERSION_MINOR 7
@ -265,34 +265,34 @@ typedef struct TableWriteState
StringInfo compressionBuffer;
} TableWriteState;
extern int cstore_compression;
extern int cstore_stripe_row_count;
extern int cstore_chunk_row_count;
extern int columnar_compression;
extern int columnar_stripe_row_count;
extern int columnar_chunk_row_count;
extern int columnar_compression_level;
extern void cstore_init(void);
extern void columnar_init_gucs(void);
extern CompressionType ParseCompressionType(const char *compressionTypeString);
/* Function declarations for writing to a cstore file */
extern TableWriteState * CStoreBeginWrite(RelFileNode relfilenode,
extern TableWriteState * ColumnarBeginWrite(RelFileNode relfilenode,
ColumnarOptions options,
TupleDesc tupleDescriptor);
extern void CStoreWriteRow(TableWriteState *state, Datum *columnValues,
extern void ColumnarWriteRow(TableWriteState *state, Datum *columnValues,
bool *columnNulls);
extern void CStoreFlushPendingWrites(TableWriteState *state);
extern void CStoreEndWrite(TableWriteState *state);
extern void ColumnarFlushPendingWrites(TableWriteState *state);
extern void ColumnarEndWrite(TableWriteState *state);
extern bool ContainsPendingWrites(TableWriteState *state);
/* Function declarations for reading from a cstore file */
extern TableReadState * CStoreBeginRead(Relation relation,
extern TableReadState * ColumnarBeginRead(Relation relation,
TupleDesc tupleDescriptor,
List *projectedColumnList, List *qualConditions);
extern bool CStoreReadFinished(TableReadState *state);
extern bool CStoreReadNextRow(TableReadState *state, Datum *columnValues,
List *projectedColumnList,
List *qualConditions);
extern bool ColumnarReadNextRow(TableReadState *state, Datum *columnValues,
bool *columnNulls);
extern void CStoreRescan(TableReadState *readState);
extern void CStoreEndRead(TableReadState *state);
extern void ColumnarRescan(TableReadState *readState);
extern void ColumnarEndRead(TableReadState *state);
/* Function declarations for common functions */
extern FmgrInfo * GetFunctionInfoOrNull(Oid typeId, Oid accessMethodId,
@ -300,7 +300,7 @@ extern FmgrInfo * GetFunctionInfoOrNull(Oid typeId, Oid accessMethodId,
extern ChunkData * CreateEmptyChunkData(uint32 columnCount, bool *columnMask,
uint32 chunkRowCount);
extern void FreeChunkData(ChunkData *chunkData);
extern uint64 CStoreTableRowCount(Relation relation);
extern uint64 ColumnarTableRowCount(Relation relation);
extern bool CompressBuffer(StringInfo inputBuffer,
StringInfo outputBuffer,
CompressionType compressionType,
@ -309,7 +309,7 @@ extern StringInfo DecompressBuffer(StringInfo buffer, CompressionType compressio
uint64 decompressedSize);
extern const char * CompressionTypeStr(CompressionType type);
/* cstore_metadata_tables.c */
/* columnar_metadata_tables.c */
extern void InitColumnarOptions(Oid regclass);
extern void SetColumnarOptions(Oid regclass, ColumnarOptions *options);
extern bool DeleteColumnarTableOptions(Oid regclass, bool missingOk);
@ -317,9 +317,9 @@ extern bool ReadColumnarOptions(Oid regclass, ColumnarOptions *options);
extern void WriteToSmgr(Relation relation, uint64 logicalOffset,
char *data, uint32 dataLength);
extern StringInfo ReadFromSmgr(Relation rel, uint64 offset, uint32 size);
extern bool IsCStoreTableAmTable(Oid relationId);
extern bool IsColumnarTableAmTable(Oid relationId);
/* cstore_metadata_tables.c */
/* columnar_metadata_tables.c */
extern void DeleteMetadataRows(RelFileNode relfilenode);
extern List * StripesForRelfilenode(RelFileNode relfilenode);
extern uint64 GetHighestUsedAddress(RelFileNode relfilenode);
@ -336,7 +336,7 @@ extern Datum columnar_relation_storageid(PG_FUNCTION_ARGS);
/* write_state_management.c */
extern TableWriteState * cstore_init_write_state(Relation relation, TupleDesc
extern TableWriteState * columnar_init_write_state(Relation relation, TupleDesc
tupdesc,
SubTransactionId currentSubXid);
extern void FlushWriteStateForRelfilenode(Oid relfilenode, SubTransactionId

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

@ -13,7 +13,7 @@
#ifndef COLUMNAR_CUSTOMSCAN_H
#define COLUMNAR_CUSTOMSCAN_H
void cstore_customscan_init(void);
void columnar_customscan_init(void);
#endif /* COLUMNAR_CUSTOMSCAN_H */

8
src/include/columnar/cstore_tableam.h
View File

@ -10,16 +10,16 @@
#include "distributed/coordinator_protocol.h"
const TableAmRoutine * GetColumnarTableAmRoutine(void);
extern void cstore_tableam_init(void);
extern void cstore_tableam_finish(void);
extern void columnar_tableam_init(void);
extern void columnar_tableam_finish(void);
extern TableScanDesc cstore_beginscan_extended(Relation relation, Snapshot snapshot,
extern TableScanDesc columnar_beginscan_extended(Relation relation, Snapshot snapshot,
int nkeys, ScanKey key,
ParallelTableScanDesc parallel_scan,
uint32 flags, Bitmapset *attr_needed,
List *scanQual);
extern int64 ColumnarGetChunksFiltered(TableScanDesc scanDesc);
extern bool IsCStoreTableAmTable(Oid relationId);
extern bool IsColumnarTableAmTable(Oid relationId);
extern TableDDLCommand * ColumnarGetTableOptionsDDL(Oid relationId);
extern char * GetShardedTableDDLCommandColumnar(uint64 shardId, void *context);
#endif