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

Merge branch 'main' into packaging_validations

pull/7054/head
Gürkan İndibay 2023-07-21 14:47:23 +03:00 committed by GitHub
parent e5fbfadee6 42d956888d
commit 47f7f4751c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
50 changed files with 703 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/backend/columnar/columnar_compression.c
View File

@ -18,11 +18,16 @@
#include "lib/stringinfo.h"
#include "columnar/columnar_compression.h"
#include "distributed/pg_version_constants.h"
#if HAVE_CITUS_LIBLZ4
#include <lz4.h>
#endif
#if PG_VERSION_NUM >= PG_VERSION_16
#include "varatt.h"
#endif
#if HAVE_LIBZSTD
#include <zstd.h>
#endif

120
src/backend/columnar/columnar_customscan.c
View File

@ -33,6 +33,10 @@
#include "optimizer/paths.h"
#include "optimizer/plancat.h"
#include "optimizer/restrictinfo.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "parser/parse_relation.h"
#include "parser/parsetree.h"
#endif
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/relcache.h"
@ -127,6 +131,9 @@ static List * set_deparse_context_planstate(List *dpcontext, Node *node,
/* other helpers */
static List * ColumnarVarNeeded(ColumnarScanState *columnarScanState);
static Bitmapset * ColumnarAttrNeeded(ScanState *ss);
#if PG_VERSION_NUM >= PG_VERSION_16
static Bitmapset * fixup_inherited_columns(Oid parentId, Oid childId, Bitmapset *columns);
#endif
/* saved hook value in case of unload */
static set_rel_pathlist_hook_type PreviousSetRelPathlistHook = NULL;
@ -535,7 +542,7 @@ ColumnarIndexScanAdditionalCost(PlannerInfo *root, RelOptInfo *rel,
* "anti-correlated" (-1) since both help us avoiding from reading the
* same stripe again and again.
*/
double absIndexCorrelation = Abs(indexCorrelation);
double absIndexCorrelation = float_abs(indexCorrelation);
/*
* To estimate the number of stripes that we need to read, we do linear
@ -654,7 +661,7 @@ CheckVarStats(PlannerInfo *root, Var *var, Oid sortop, float4 *absVarCorrelation
* If the Var is not highly correlated, then the chunk's min/max bounds
* will be nearly useless.
*/
if (Abs(varCorrelation) < ColumnarQualPushdownCorrelationThreshold)
if (float_abs(varCorrelation) < ColumnarQualPushdownCorrelationThreshold)
{
if (absVarCorrelation)
{
@ -662,7 +669,7 @@ CheckVarStats(PlannerInfo *root, Var *var, Oid sortop, float4 *absVarCorrelation
* Report absVarCorrelation if caller wants to know why given
* var is rejected.
*/
*absVarCorrelation = Abs(varCorrelation);
*absVarCorrelation = float_abs(varCorrelation);
}
return false;
}
@ -1371,7 +1378,43 @@ AddColumnarScanPath(PlannerInfo *root, RelOptInfo *rel, RangeTblEntry *rte,
cpath->custom_private = list_make2(NIL, NIL);
}
int numberOfColumnsRead = bms_num_members(rte->selectedCols);
int numberOfColumnsRead = 0;
#if PG_VERSION_NUM >= PG_VERSION_16
if (rte->perminfoindex > 0)
{
/*
* If perminfoindex > 0, that means that this relation's permission info
* is directly found in the list of rteperminfos of the Query(root->parse)
* So, all we have to do here is retrieve that info.
*/
RTEPermissionInfo *perminfo = getRTEPermissionInfo(root->parse->rteperminfos,
rte);
numberOfColumnsRead = bms_num_members(perminfo->selectedCols);
}
else
{
/*
* If perminfoindex = 0, that means we are skipping the check for permission info
* for this relation, which means that it's either a partition or an inheritance child.
* In these cases, we need to access the permission info of the top parent of this relation.
* After thorough checking, we found that the index of the top parent pointing to the correct
* range table entry in Query's range tables (root->parse->rtable) is found under
* RelOptInfo rel->top_parent->relid.
* For reference, check expand_partitioned_rtentry and expand_inherited_rtentry PG functions
*/
Assert(rel->top_parent);
RangeTblEntry *parent_rte = rt_fetch(rel->top_parent->relid, root->parse->rtable);
RTEPermissionInfo *perminfo = getRTEPermissionInfo(root->parse->rteperminfos,
parent_rte);
numberOfColumnsRead = bms_num_members(fixup_inherited_columns(perminfo->relid,
rte->relid,
perminfo->
selectedCols));
}
#else
numberOfColumnsRead = bms_num_members(rte->selectedCols);
#endif
int numberOfClausesPushed = list_length(allClauses);
CostColumnarScan(root, rel, rte->relid, cpath, numberOfColumnsRead,
@ -1391,6 +1434,69 @@ AddColumnarScanPath(PlannerInfo *root, RelOptInfo *rel, RangeTblEntry *rte,
}
#if PG_VERSION_NUM >= PG_VERSION_16
/*
* fixup_inherited_columns
*
* Exact function Copied from PG16 as it's static.
*
* When user is querying on a table with children, it implicitly accesses
* child tables also. So, we also need to check security label of child
* tables and columns, but there is no guarantee attribute numbers are
* same between the parent and children.
* It returns a bitmapset which contains attribute number of the child
* table based on the given bitmapset of the parent.
*/
static Bitmapset *
fixup_inherited_columns(Oid parentId, Oid childId, Bitmapset *columns)
{
Bitmapset *result = NULL;
/*
* obviously, no need to do anything here
*/
if (parentId == childId)
{
return columns;
}
int index = -1;
while ((index = bms_next_member(columns, index)) >= 0)
{
/* bit numbers are offset by FirstLowInvalidHeapAttributeNumber */
AttrNumber attno = index + FirstLowInvalidHeapAttributeNumber;
/*
* whole-row-reference shall be fixed-up later
*/
if (attno == InvalidAttrNumber)
{
result = bms_add_member(result, index);
continue;
}
char *attname = get_attname(parentId, attno, false);
attno = get_attnum(childId, attname);
if (attno == InvalidAttrNumber)
{
elog(ERROR, "cache lookup failed for attribute %s of relation %u",
attname, childId);
}
result = bms_add_member(result,
attno - FirstLowInvalidHeapAttributeNumber);
pfree(attname);
}
return result;
}
#endif
/*
* CostColumnarScan calculates the cost of scanning the columnar table. The
* cost is estimated by using all stripe metadata to estimate based on the
@ -1435,7 +1541,8 @@ ColumnarPerStripeScanCost(RelOptInfo *rel, Oid relationId, int numberOfColumnsRe
ereport(ERROR, (errmsg("could not open relation with OID %u", relationId)));
}
List *stripeList = StripesForRelfilenode(relation->rd_node);
List *stripeList = StripesForRelfilelocator(RelationPhysicalIdentifier_compat(
relation));
RelationClose(relation);
uint32 maxColumnCount = 0;
@ -1492,7 +1599,8 @@ ColumnarTableStripeCount(Oid relationId)
ereport(ERROR, (errmsg("could not open relation with OID %u", relationId)));
}
List *stripeList = StripesForRelfilenode(relation->rd_node);
List *stripeList = StripesForRelfilelocator(RelationPhysicalIdentifier_compat(
relation));
int stripeCount = list_length(stripeList);
RelationClose(relation);

76
src/backend/columnar/columnar_metadata.c
View File

@ -47,6 +47,9 @@
#include "miscadmin.h"
#include "nodes/execnodes.h"
#include "lib/stringinfo.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "parser/parse_relation.h"
#endif
#include "port.h"
#include "storage/fd.h"
#include "storage/lmgr.h"
@ -57,7 +60,12 @@
#include "utils/memutils.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "storage/relfilelocator.h"
#include "utils/relfilenumbermap.h"
#else
#include "utils/relfilenodemap.h"
#endif
#define COLUMNAR_RELOPTION_NAMESPACE "columnar"
#define SLOW_METADATA_ACCESS_WARNING \
@ -112,7 +120,7 @@ static Oid ColumnarChunkGroupRelationId(void);
static Oid ColumnarChunkIndexRelationId(void);
static Oid ColumnarChunkGroupIndexRelationId(void);
static Oid ColumnarNamespaceId(void);
static uint64 LookupStorageId(RelFileNode relfilenode);
static uint64 LookupStorageId(RelFileLocator relfilelocator);
static uint64 GetHighestUsedRowNumber(uint64 storageId);
static void DeleteStorageFromColumnarMetadataTable(Oid metadataTableId,
AttrNumber storageIdAtrrNumber,
@ -591,14 +599,15 @@ ReadColumnarOptions(Oid regclass, ColumnarOptions *options)
* of columnar.chunk.
*/
void
SaveStripeSkipList(RelFileNode relfilenode, uint64 stripe, StripeSkipList *chunkList,
SaveStripeSkipList(RelFileLocator relfilelocator, uint64 stripe,
StripeSkipList *chunkList,
TupleDesc tupleDescriptor)
{
uint32 columnIndex = 0;
uint32 chunkIndex = 0;
uint32 columnCount = chunkList->columnCount;
uint64 storageId = LookupStorageId(relfilenode);
uint64 storageId = LookupStorageId(relfilelocator);
Oid columnarChunkOid = ColumnarChunkRelationId();
Relation columnarChunk = table_open(columnarChunkOid, RowExclusiveLock);
ModifyState *modifyState = StartModifyRelation(columnarChunk);
@ -657,10 +666,10 @@ SaveStripeSkipList(RelFileNode relfilenode, uint64 stripe, StripeSkipList *chunk
* SaveChunkGroups saves the metadata for given chunk groups in columnar.chunk_group.
*/
void
SaveChunkGroups(RelFileNode relfilenode, uint64 stripe,
SaveChunkGroups(RelFileLocator relfilelocator, uint64 stripe,
List *chunkGroupRowCounts)
{
uint64 storageId = LookupStorageId(relfilenode);
uint64 storageId = LookupStorageId(relfilelocator);
Oid columnarChunkGroupOid = ColumnarChunkGroupRelationId();
Relation columnarChunkGroup = table_open(columnarChunkGroupOid, RowExclusiveLock);
ModifyState *modifyState = StartModifyRelation(columnarChunkGroup);
@ -693,7 +702,8 @@ SaveChunkGroups(RelFileNode relfilenode, uint64 stripe,
* ReadStripeSkipList fetches chunk metadata for a given stripe.
*/
StripeSkipList *
ReadStripeSkipList(RelFileNode relfilenode, uint64 stripe, TupleDesc tupleDescriptor,
ReadStripeSkipList(RelFileLocator relfilelocator, uint64 stripe,
TupleDesc tupleDescriptor,
uint32 chunkCount, Snapshot snapshot)
{
int32 columnIndex = 0;
@ -701,15 +711,15 @@ ReadStripeSkipList(RelFileNode relfilenode, uint64 stripe, TupleDesc tupleDescri
uint32 columnCount = tupleDescriptor->natts;
ScanKeyData scanKey[2];
uint64 storageId = LookupStorageId(relfilenode);
uint64 storageId = LookupStorageId(relfilelocator);
Oid columnarChunkOid = ColumnarChunkRelationId();
Relation columnarChunk = table_open(columnarChunkOid, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_columnar_chunk_storageid,
BTEqualStrategyNumber, F_OIDEQ, UInt64GetDatum(storageId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(storageId));
ScanKeyInit(&scanKey[1], Anum_columnar_chunk_stripe,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(stripe));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(stripe));
Oid indexId = ColumnarChunkIndexRelationId();
bool indexOk = OidIsValid(indexId);
@ -915,7 +925,7 @@ StripeMetadataLookupRowNumber(Relation relation, uint64 rowNumber, Snapshot snap
uint64 storageId = ColumnarStorageGetStorageId(relation, false);
ScanKeyData scanKey[2];
ScanKeyInit(&scanKey[0], Anum_columnar_stripe_storageid,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(storageId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(storageId));
StrategyNumber strategyNumber = InvalidStrategy;
RegProcedure procedure = InvalidOid;
@ -930,7 +940,7 @@ StripeMetadataLookupRowNumber(Relation relation, uint64 rowNumber, Snapshot snap
procedure = F_INT8GT;
}
ScanKeyInit(&scanKey[1], Anum_columnar_stripe_first_row_number,
strategyNumber, procedure, UInt64GetDatum(rowNumber));
strategyNumber, procedure, Int64GetDatum(rowNumber));
Relation columnarStripes = table_open(ColumnarStripeRelationId(), AccessShareLock);
@ -1081,7 +1091,7 @@ FindStripeWithHighestRowNumber(Relation relation, Snapshot snapshot)
uint64 storageId = ColumnarStorageGetStorageId(relation, false);
ScanKeyData scanKey[1];
ScanKeyInit(&scanKey[0], Anum_columnar_stripe_storageid,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(storageId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(storageId));
Relation columnarStripes = table_open(ColumnarStripeRelationId(), AccessShareLock);
@ -1143,9 +1153,9 @@ ReadChunkGroupRowCounts(uint64 storageId, uint64 stripe, uint32 chunkGroupCount,
ScanKeyData scanKey[2];
ScanKeyInit(&scanKey[0], Anum_columnar_chunkgroup_storageid,
BTEqualStrategyNumber, F_OIDEQ, UInt64GetDatum(storageId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(storageId));
ScanKeyInit(&scanKey[1], Anum_columnar_chunkgroup_stripe,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(stripe));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(stripe));
Oid indexId = ColumnarChunkGroupIndexRelationId();
bool indexOk = OidIsValid(indexId);
@ -1235,13 +1245,13 @@ InsertEmptyStripeMetadataRow(uint64 storageId, uint64 stripeId, uint32 columnCou
/*
* StripesForRelfilenode returns a list of StripeMetadata for stripes
* StripesForRelfilelocator returns a list of StripeMetadata for stripes
* of the given relfilenode.
*/
List *
StripesForRelfilenode(RelFileNode relfilenode)
StripesForRelfilelocator(RelFileLocator relfilelocator)
{
uint64 storageId = LookupStorageId(relfilenode);
uint64 storageId = LookupStorageId(relfilelocator);
return ReadDataFileStripeList(storageId, GetTransactionSnapshot());
}
@ -1256,9 +1266,9 @@ StripesForRelfilenode(RelFileNode relfilenode)
* returns 0.
*/
uint64
GetHighestUsedAddress(RelFileNode relfilenode)
GetHighestUsedAddress(RelFileLocator relfilelocator)
{
uint64 storageId = LookupStorageId(relfilenode);
uint64 storageId = LookupStorageId(relfilelocator);
uint64 highestUsedAddress = 0;
uint64 highestUsedId = 0;
@ -1372,9 +1382,9 @@ UpdateStripeMetadataRow(uint64 storageId, uint64 stripeId, bool *update,
ScanKeyData scanKey[2];
ScanKeyInit(&scanKey[0], Anum_columnar_stripe_storageid,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(storageId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(storageId));
ScanKeyInit(&scanKey[1], Anum_columnar_stripe_stripe,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(stripeId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(stripeId));
Oid columnarStripesOid = ColumnarStripeRelationId();
@ -1451,7 +1461,7 @@ ReadDataFileStripeList(uint64 storageId, Snapshot snapshot)
HeapTuple heapTuple;
ScanKeyInit(&scanKey[0], Anum_columnar_stripe_storageid,
BTEqualStrategyNumber, F_OIDEQ, Int32GetDatum(storageId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(storageId));
Oid columnarStripesOid = ColumnarStripeRelationId();
@ -1539,7 +1549,7 @@ BuildStripeMetadata(Relation columnarStripes, HeapTuple heapTuple)
* metadata tables.
*/
void
DeleteMetadataRows(RelFileNode relfilenode)
DeleteMetadataRows(RelFileLocator relfilelocator)
{
/*
* During a restore for binary upgrade, metadata tables and indexes may or
@ -1550,7 +1560,7 @@ DeleteMetadataRows(RelFileNode relfilenode)
return;
}
uint64 storageId = LookupStorageId(relfilenode);
uint64 storageId = LookupStorageId(relfilelocator);
DeleteStorageFromColumnarMetadataTable(ColumnarStripeRelationId(),
Anum_columnar_stripe_storageid,
@ -1578,7 +1588,7 @@ DeleteStorageFromColumnarMetadataTable(Oid metadataTableId,
{
ScanKeyData scanKey[1];
ScanKeyInit(&scanKey[0], storageIdAtrrNumber, BTEqualStrategyNumber,
F_INT8EQ, UInt64GetDatum(storageId));
F_INT8EQ, Int64GetDatum(storageId));
Relation metadataTable = try_relation_open(metadataTableId, AccessShareLock);
if (metadataTable == NULL)
@ -1713,7 +1723,14 @@ create_estate_for_relation(Relation rel)
rte->relid = RelationGetRelid(rel);
rte->relkind = rel->rd_rel->relkind;
rte->rellockmode = AccessShareLock;
#if PG_VERSION_NUM >= PG_VERSION_16
List *perminfos = NIL;
addRTEPermissionInfo(&perminfos, rte);
ExecInitRangeTable(estate, list_make1(rte), perminfos);
#else
ExecInitRangeTable(estate, list_make1(rte));
#endif
estate->es_output_cid = GetCurrentCommandId(true);
@ -1917,10 +1934,11 @@ ColumnarNamespaceId(void)
* false if the relation doesn't have a meta page yet.
*/
static uint64
LookupStorageId(RelFileNode relfilenode)
LookupStorageId(RelFileLocator relfilelocator)
{
Oid relationId = RelidByRelfilenode(relfilenode.spcNode,
relfilenode.relNode);
Oid relationId = RelidByRelfilenumber(RelationTablespace_compat(relfilelocator),
RelationPhysicalIdentifierNumber_compat(
relfilelocator));
Relation relation = relation_open(relationId, AccessShareLock);
uint64 storageId = ColumnarStorageGetStorageId(relation, false);
@ -1951,7 +1969,7 @@ columnar_relation_storageid(PG_FUNCTION_ARGS)
Oid relationId = PG_GETARG_OID(0);
Relation relation = relation_open(relationId, AccessShareLock);
if (!pg_class_ownercheck(relationId, GetUserId()))
if (!object_ownercheck(RelationRelationId, relationId, GetUserId()))
{
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLE,
get_rel_name(relationId));

11
src/backend/columnar/columnar_reader.c
View File

@ -254,8 +254,9 @@ ColumnarReadFlushPendingWrites(ColumnarReadState *readState)
{
Assert(!readState->snapshotRegisteredByUs);
Oid relfilenode = readState->relation->rd_node.relNode;
FlushWriteStateForRelfilenode(relfilenode, GetCurrentSubTransactionId());
RelFileNumber relfilenumber = RelationPhysicalIdentifierNumber_compat(
RelationPhysicalIdentifier_compat(readState->relation));
FlushWriteStateForRelfilenumber(relfilenumber, GetCurrentSubTransactionId());
if (readState->snapshot == InvalidSnapshot || !IsMVCCSnapshot(readState->snapshot))
{
@ -984,7 +985,8 @@ ColumnarTableRowCount(Relation relation)
{
ListCell *stripeMetadataCell = NULL;
uint64 totalRowCount = 0;
List *stripeList = StripesForRelfilenode(relation->rd_node);
List *stripeList = StripesForRelfilelocator(RelationPhysicalIdentifier_compat(
relation));
foreach(stripeMetadataCell, stripeList)
{
@ -1012,7 +1014,8 @@ LoadFilteredStripeBuffers(Relation relation, StripeMetadata *stripeMetadata,
bool *projectedColumnMask = ProjectedColumnMask(columnCount, projectedColumnList);
StripeSkipList *stripeSkipList = ReadStripeSkipList(relation->rd_node,
StripeSkipList *stripeSkipList = ReadStripeSkipList(RelationPhysicalIdentifier_compat(
relation),
stripeMetadata->id,
tupleDescriptor,
stripeMetadata->chunkCount,

4
src/backend/columnar/columnar_storage.c
View File

@ -188,7 +188,7 @@ ColumnarStorageInit(SMgrRelation srel, uint64 storageId)
(char *) &metapage, sizeof(ColumnarMetapage));
phdr->pd_lower += sizeof(ColumnarMetapage);
log_newpage(&srel->smgr_rnode.node, MAIN_FORKNUM,
log_newpage(RelationPhysicalIdentifierBackend_compat(&srel), MAIN_FORKNUM,
COLUMNAR_METAPAGE_BLOCKNO, page, true);
PageSetChecksumInplace(page, COLUMNAR_METAPAGE_BLOCKNO);
smgrextend(srel, MAIN_FORKNUM, COLUMNAR_METAPAGE_BLOCKNO, page, true);
@ -196,7 +196,7 @@ ColumnarStorageInit(SMgrRelation srel, uint64 storageId)
/* write empty page */
PageInit(page, BLCKSZ, 0);
log_newpage(&srel->smgr_rnode.node, MAIN_FORKNUM,
log_newpage(RelationPhysicalIdentifierBackend_compat(&srel), MAIN_FORKNUM,
COLUMNAR_EMPTY_BLOCKNO, page, true);
PageSetChecksumInplace(page, COLUMNAR_EMPTY_BLOCKNO);
smgrextend(srel, MAIN_FORKNUM, COLUMNAR_EMPTY_BLOCKNO, page, true);

101
src/backend/columnar/columnar_tableam.c
View File

@ -31,6 +31,7 @@
#include "executor/executor.h"
#include "nodes/makefuncs.h"
#include "optimizer/plancat.h"
#include "pg_version_compat.h"
#include "pgstat.h"
#include "safe_lib.h"
#include "storage/bufmgr.h"
@ -206,7 +207,8 @@ columnar_beginscan_extended(Relation relation, Snapshot snapshot,
uint32 flags, Bitmapset *attr_needed, List *scanQual)
{
CheckCitusColumnarVersion(ERROR);
Oid relfilenode = relation->rd_node.relNode;
RelFileNumber relfilenumber = RelationPhysicalIdentifierNumber_compat(
RelationPhysicalIdentifier_compat(relation));
/*
* A memory context to use for scan-wide data, including the lazily
@ -236,7 +238,7 @@ columnar_beginscan_extended(Relation relation, Snapshot snapshot,
scan->scanQual = copyObject(scanQual);
scan->scanContext = scanContext;
if (PendingWritesInUpperTransactions(relfilenode, GetCurrentSubTransactionId()))
if (PendingWritesInUpperTransactions(relfilenumber, GetCurrentSubTransactionId()))
{
elog(ERROR,
"cannot read from table when there is unflushed data in upper transactions");
@ -432,8 +434,9 @@ columnar_index_fetch_begin(Relation rel)
{
CheckCitusColumnarVersion(ERROR);
Oid relfilenode = rel->rd_node.relNode;
if (PendingWritesInUpperTransactions(relfilenode, GetCurrentSubTransactionId()))
RelFileNumber relfilenumber = RelationPhysicalIdentifierNumber_compat(
RelationPhysicalIdentifier_compat(rel));
if (PendingWritesInUpperTransactions(relfilenumber, GetCurrentSubTransactionId()))
{
/* XXX: maybe we can just flush the data and continue */
elog(ERROR, "cannot read from index when there is unflushed data in "
@ -815,7 +818,7 @@ static TM_Result
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)
LockTupleMode *lockmode, TU_UpdateIndexes *update_indexes)
{
elog(ERROR, "columnar_tuple_update not implemented");
}
@ -841,8 +844,8 @@ columnar_finish_bulk_insert(Relation relation, int options)
static void
columnar_relation_set_new_filenode(Relation rel,
const RelFileNode *newrnode,
columnar_relation_set_new_filelocator(Relation rel,
const RelFileLocator *newrlocator,
char persistence,
TransactionId *freezeXid,
MultiXactId *minmulti)
@ -861,16 +864,19 @@ columnar_relation_set_new_filenode(Relation rel,
* state. If they are equal, this is a new relation object and we don't
* need to clean anything.
*/
if (rel->rd_node.relNode != newrnode->relNode)
if (RelationPhysicalIdentifierNumber_compat(RelationPhysicalIdentifier_compat(rel)) !=
RelationPhysicalIdentifierNumberPtr_compat(newrlocator))
{
MarkRelfilenodeDropped(rel->rd_node.relNode, GetCurrentSubTransactionId());
MarkRelfilenumberDropped(RelationPhysicalIdentifierNumber_compat(
RelationPhysicalIdentifier_compat(rel)),
GetCurrentSubTransactionId());
DeleteMetadataRows(rel->rd_node);
DeleteMetadataRows(RelationPhysicalIdentifier_compat(rel));
}
*freezeXid = RecentXmin;
*minmulti = GetOldestMultiXactId();
SMgrRelation srel = RelationCreateStorage_compat(*newrnode, persistence, true);
SMgrRelation srel = RelationCreateStorage_compat(*newrlocator, persistence, true);
ColumnarStorageInit(srel, ColumnarMetadataNewStorageId());
InitColumnarOptions(rel->rd_id);
@ -885,12 +891,12 @@ static void
columnar_relation_nontransactional_truncate(Relation rel)
{
CheckCitusColumnarVersion(ERROR);
RelFileNode relfilenode = rel->rd_node;
RelFileLocator relfilelocator = RelationPhysicalIdentifier_compat(rel);
NonTransactionDropWriteState(relfilenode.relNode);
NonTransactionDropWriteState(RelationPhysicalIdentifierNumber_compat(relfilelocator));
/* Delete old relfilenode metadata */
DeleteMetadataRows(relfilenode);
DeleteMetadataRows(relfilelocator);
/*
* No need to set new relfilenode, since the table was created in this
@ -907,7 +913,7 @@ columnar_relation_nontransactional_truncate(Relation rel)
static void
columnar_relation_copy_data(Relation rel, const RelFileNode *newrnode)
columnar_relation_copy_data(Relation rel, const RelFileLocator *newrnode)
{
elog(ERROR, "columnar_relation_copy_data not implemented");
}
@ -953,7 +959,8 @@ columnar_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
ColumnarOptions columnarOptions = { 0 };
ReadColumnarOptions(OldHeap->rd_id, &columnarOptions);
ColumnarWriteState *writeState = ColumnarBeginWrite(NewHeap->rd_node,
ColumnarWriteState *writeState = ColumnarBeginWrite(RelationPhysicalIdentifier_compat(
NewHeap),
columnarOptions,
targetDesc);
@ -1028,7 +1035,8 @@ NeededColumnsList(TupleDesc tupdesc, Bitmapset *attr_needed)
static uint64
ColumnarTableTupleCount(Relation relation)
{
List *stripeList = StripesForRelfilenode(relation->rd_node);
List *stripeList = StripesForRelfilelocator(RelationPhysicalIdentifier_compat(
relation));
uint64 tupleCount = 0;
ListCell *lc = NULL;
@ -1091,12 +1099,38 @@ columnar_vacuum_rel(Relation rel, VacuumParams *params,
List *indexList = RelationGetIndexList(rel);
int nindexes = list_length(indexList);
#if PG_VERSION_NUM >= PG_VERSION_16
struct VacuumCutoffs cutoffs;
vacuum_get_cutoffs(rel, params, &cutoffs);
Assert(MultiXactIdPrecedesOrEquals(cutoffs.MultiXactCutoff, cutoffs.OldestMxact));
Assert(TransactionIdPrecedesOrEquals(cutoffs.FreezeLimit, cutoffs.OldestXmin));
/*
* Columnar storage doesn't hold any transaction IDs, so we can always
* just advance to the most aggressive value.
*/
TransactionId newRelFrozenXid = cutoffs.OldestXmin;
MultiXactId newRelminMxid = cutoffs.OldestMxact;
double new_live_tuples = ColumnarTableTupleCount(rel);
/* all visible pages are always 0 */
BlockNumber new_rel_allvisible = 0;
bool frozenxid_updated;
bool minmulti_updated;
vac_update_relstats(rel, new_rel_pages, new_live_tuples,
new_rel_allvisible, nindexes > 0,
newRelFrozenXid, newRelminMxid,
&frozenxid_updated, &minmulti_updated, false);
#else
TransactionId oldestXmin;
TransactionId freezeLimit;
MultiXactId multiXactCutoff;
/* initialize xids */
#if PG_VERSION_NUM >= PG_VERSION_15
#if (PG_VERSION_NUM >= PG_VERSION_15) && (PG_VERSION_NUM < PG_VERSION_16)
MultiXactId oldestMxact;
vacuum_set_xid_limits(rel,
params->freeze_min_age,
@ -1126,7 +1160,7 @@ columnar_vacuum_rel(Relation rel, VacuumParams *params,
* just advance to the most aggressive value.
*/
TransactionId newRelFrozenXid = oldestXmin;
#if PG_VERSION_NUM >= PG_VERSION_15
#if (PG_VERSION_NUM >= PG_VERSION_15) && (PG_VERSION_NUM < PG_VERSION_16)
MultiXactId newRelminMxid = oldestMxact;
#else
MultiXactId newRelminMxid = multiXactCutoff;
@ -1137,7 +1171,7 @@ columnar_vacuum_rel(Relation rel, VacuumParams *params,
/* all visible pages are always 0 */
BlockNumber new_rel_allvisible = 0;
#if PG_VERSION_NUM >= PG_VERSION_15
#if (PG_VERSION_NUM >= PG_VERSION_15) && (PG_VERSION_NUM < PG_VERSION_16)
bool frozenxid_updated;
bool minmulti_updated;
@ -1149,6 +1183,7 @@ columnar_vacuum_rel(Relation rel, VacuumParams *params,
vac_update_relstats(rel, new_rel_pages, new_live_tuples,
new_rel_allvisible, nindexes > 0,
newRelFrozenXid, newRelminMxid, false);
#endif
#endif
pgstat_report_vacuum(RelationGetRelid(rel),
@ -1166,7 +1201,7 @@ static void
LogRelationStats(Relation rel, int elevel)
{
ListCell *stripeMetadataCell = NULL;
RelFileNode relfilenode = rel->rd_node;
RelFileLocator relfilelocator = RelationPhysicalIdentifier_compat(rel);
StringInfo infoBuf = makeStringInfo();
int compressionStats[COMPRESSION_COUNT] = { 0 };
@ -1177,13 +1212,13 @@ LogRelationStats(Relation rel, int elevel)
uint64 droppedChunksWithData = 0;
uint64 totalDecompressedLength = 0;
List *stripeList = StripesForRelfilenode(relfilenode);
List *stripeList = StripesForRelfilelocator(relfilelocator);
int stripeCount = list_length(stripeList);
foreach(stripeMetadataCell, stripeList)
{
StripeMetadata *stripe = lfirst(stripeMetadataCell);
StripeSkipList *skiplist = ReadStripeSkipList(relfilenode, stripe->id,
StripeSkipList *skiplist = ReadStripeSkipList(relfilelocator, stripe->id,
RelationGetDescr(rel),
stripe->chunkCount,
GetTransactionSnapshot());
@ -1319,7 +1354,8 @@ TruncateColumnar(Relation rel, int elevel)
* new stripes be added beyond highestPhysicalAddress while
* we're truncating.
*/
uint64 newDataReservation = Max(GetHighestUsedAddress(rel->rd_node) + 1,
uint64 newDataReservation = Max(GetHighestUsedAddress(
RelationPhysicalIdentifier_compat(rel)) + 1,
ColumnarFirstLogicalOffset);
BlockNumber old_rel_pages = smgrnblocks(RelationGetSmgr(rel), MAIN_FORKNUM);
@ -1826,8 +1862,8 @@ TupleSortSkipSmallerItemPointers(Tuplesortstate *tupleSort, ItemPointer targetIt
Datum *abbrev = NULL;
Datum tsDatum;
bool tsDatumIsNull;
if (!tuplesort_getdatum(tupleSort, forwardDirection, &tsDatum,
&tsDatumIsNull, abbrev))
if (!tuplesort_getdatum_compat(tupleSort, forwardDirection, false,
&tsDatum, &tsDatumIsNull, abbrev))
{
ItemPointerSetInvalid(&tsItemPointerData);
break;
@ -2068,12 +2104,13 @@ ColumnarTableDropHook(Oid relid)
* tableam tables storage is managed by postgres.
*/
Relation rel = table_open(relid, AccessExclusiveLock);
RelFileNode relfilenode = rel->rd_node;
RelFileLocator relfilelocator = RelationPhysicalIdentifier_compat(rel);
DeleteMetadataRows(relfilenode);
DeleteMetadataRows(relfilelocator);
DeleteColumnarTableOptions(rel->rd_id, true);
MarkRelfilenodeDropped(relfilenode.relNode, GetCurrentSubTransactionId());
MarkRelfilenumberDropped(RelationPhysicalIdentifierNumber_compat(relfilelocator),
GetCurrentSubTransactionId());
/* keep the lock since we did physical changes to the relation */
table_close(rel, NoLock);
@ -2490,7 +2527,11 @@ static const TableAmRoutine columnar_am_methods = {
.tuple_lock = columnar_tuple_lock,
.finish_bulk_insert = columnar_finish_bulk_insert,
.relation_set_new_filenode = columnar_relation_set_new_filenode,
#if PG_VERSION_NUM >= PG_VERSION_16
.relation_set_new_filelocator = columnar_relation_set_new_filelocator,
#else
.relation_set_new_filenode = columnar_relation_set_new_filelocator,
#endif
.relation_nontransactional_truncate = columnar_relation_nontransactional_truncate,
.relation_copy_data = columnar_relation_copy_data,
.relation_copy_for_cluster = columnar_relation_copy_for_cluster,

28
src/backend/columnar/columnar_writer.c
View File

@ -22,12 +22,18 @@
#include "access/nbtree.h"
#include "catalog/pg_am.h"
#include "miscadmin.h"
#include "pg_version_compat.h"
#include "storage/fd.h"
#include "storage/smgr.h"
#include "utils/guc.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "storage/relfilelocator.h"
#include "utils/relfilenumbermap.h"
#else
#include "utils/relfilenodemap.h"
#endif
#include "columnar/columnar.h"
#include "columnar/columnar_storage.h"
@ -37,7 +43,7 @@ struct ColumnarWriteState
{
TupleDesc tupleDescriptor;
FmgrInfo **comparisonFunctionArray;
RelFileNode relfilenode;
RelFileLocator relfilelocator;
MemoryContext stripeWriteContext;
MemoryContext perTupleContext;
@ -84,7 +90,7 @@ static StringInfo CopyStringInfo(StringInfo sourceString);
* data load operation.
*/
ColumnarWriteState *
ColumnarBeginWrite(RelFileNode relfilenode,
ColumnarBeginWrite(RelFileLocator relfilelocator,
ColumnarOptions options,
TupleDesc tupleDescriptor)
{
@ -124,7 +130,7 @@ ColumnarBeginWrite(RelFileNode relfilenode,
options.chunkRowCount);
ColumnarWriteState *writeState = palloc0(sizeof(ColumnarWriteState));
writeState->relfilenode = relfilenode;
writeState->relfilelocator = relfilelocator;
writeState->options = options;
writeState->tupleDescriptor = CreateTupleDescCopy(tupleDescriptor);
writeState->comparisonFunctionArray = comparisonFunctionArray;
@ -174,8 +180,10 @@ ColumnarWriteRow(ColumnarWriteState *writeState, Datum *columnValues, bool *colu
writeState->stripeSkipList = stripeSkipList;
writeState->compressionBuffer = makeStringInfo();
Oid relationId = RelidByRelfilenode(writeState->relfilenode.spcNode,
writeState->relfilenode.relNode);
Oid relationId = RelidByRelfilenumber(RelationTablespace_compat(
writeState->relfilelocator),
RelationPhysicalIdentifierNumber_compat(
writeState->relfilelocator));
Relation relation = relation_open(relationId, NoLock);
writeState->emptyStripeReservation =
ReserveEmptyStripe(relation, columnCount, chunkRowCount,
@ -393,8 +401,10 @@ FlushStripe(ColumnarWriteState *writeState)
elog(DEBUG1, "Flushing Stripe of size %d", stripeBuffers->rowCount);
Oid relationId = RelidByRelfilenode(writeState->relfilenode.spcNode,
writeState->relfilenode.relNode);
Oid relationId = RelidByRelfilenumber(RelationTablespace_compat(
writeState->relfilelocator),
RelationPhysicalIdentifierNumber_compat(
writeState->relfilelocator));
Relation relation = relation_open(relationId, NoLock);
/*
@ -486,10 +496,10 @@ FlushStripe(ColumnarWriteState *writeState)
}
}
SaveChunkGroups(writeState->relfilenode,
SaveChunkGroups(writeState->relfilelocator,
stripeMetadata->id,
writeState->chunkGroupRowCounts);
SaveStripeSkipList(writeState->relfilenode,
SaveStripeSkipList(writeState->relfilelocator,
stripeMetadata->id,
stripeSkipList, tupleDescriptor);

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

@ -29,6 +29,7 @@
#include "executor/executor.h"
#include "nodes/makefuncs.h"
#include "optimizer/plancat.h"
#include "pg_version_compat.h"
#include "pgstat.h"
#include "storage/bufmgr.h"
#include "storage/bufpage.h"
@ -77,7 +78,7 @@ typedef struct SubXidWriteState
typedef struct WriteStateMapEntry
{
/* key of the entry */
Oid relfilenode;
RelFileNumber relfilenumber;
/*
* If a table is dropped, we set dropped to true and set dropSubXid to the
@ -132,7 +133,7 @@ columnar_init_write_state(Relation relation, TupleDesc tupdesc,
HASHCTL info;
uint32 hashFlags = (HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
memset(&info, 0, sizeof(info));
info.keysize = sizeof(Oid);
info.keysize = sizeof(RelFileNumber);
info.hash = oid_hash;
info.entrysize = sizeof(WriteStateMapEntry);
info.hcxt = WriteStateContext;
@ -146,7 +147,10 @@ columnar_init_write_state(Relation relation, TupleDesc tupdesc,
MemoryContextRegisterResetCallback(WriteStateContext, &cleanupCallback);
}
WriteStateMapEntry *hashEntry = hash_search(WriteStateMap, &relation->rd_node.relNode,
WriteStateMapEntry *hashEntry = hash_search(WriteStateMap,
&RelationPhysicalIdentifierNumber_compat(
RelationPhysicalIdentifier_compat(
relation)),
HASH_ENTER, &found);
if (!found)
{
@ -189,7 +193,8 @@ columnar_init_write_state(Relation relation, TupleDesc tupdesc,
ReadColumnarOptions(tupSlotRelationId, &columnarOptions);
SubXidWriteState *stackEntry = palloc0(sizeof(SubXidWriteState));
stackEntry->writeState = ColumnarBeginWrite(relation->rd_node,
stackEntry->writeState = ColumnarBeginWrite(RelationPhysicalIdentifier_compat(
relation),
columnarOptions,
tupdesc);
stackEntry->subXid = currentSubXid;
@ -206,14 +211,16 @@ columnar_init_write_state(Relation relation, TupleDesc tupdesc,
* Flushes pending writes for given relfilenode in the given subtransaction.
*/
void
FlushWriteStateForRelfilenode(Oid relfilenode, SubTransactionId currentSubXid)
FlushWriteStateForRelfilenumber(RelFileNumber relfilenumber,
SubTransactionId currentSubXid)
{
if (WriteStateMap == NULL)
{
return;
}
WriteStateMapEntry *entry = hash_search(WriteStateMap, &relfilenode, HASH_FIND, NULL);
WriteStateMapEntry *entry = hash_search(WriteStateMap, &relfilenumber, HASH_FIND,
NULL);
Assert(!entry || !entry->dropped);
@ -320,14 +327,14 @@ DiscardWriteStateForAllRels(SubTransactionId currentSubXid, SubTransactionId par
* Called when the given relfilenode is dropped.
*/
void
MarkRelfilenodeDropped(Oid relfilenode, SubTransactionId currentSubXid)
MarkRelfilenumberDropped(RelFileNumber relfilenumber, SubTransactionId currentSubXid)
{
if (WriteStateMap == NULL)
{
return;
}
WriteStateMapEntry *entry = hash_search(WriteStateMap, &relfilenode, HASH_FIND,
WriteStateMapEntry *entry = hash_search(WriteStateMap, &relfilenumber, HASH_FIND,
NULL);
if (!entry || entry->dropped)
{
@ -343,11 +350,11 @@ MarkRelfilenodeDropped(Oid relfilenode, SubTransactionId currentSubXid)
* Called when the given relfilenode is dropped in non-transactional TRUNCATE.
*/
void
NonTransactionDropWriteState(Oid relfilenode)
NonTransactionDropWriteState(RelFileNumber relfilenumber)
{
if (WriteStateMap)
{
hash_search(WriteStateMap, &relfilenode, HASH_REMOVE, false);
hash_search(WriteStateMap, &relfilenumber, HASH_REMOVE, false);
}
}
@ -356,14 +363,16 @@ NonTransactionDropWriteState(Oid relfilenode)
* Returns true if there are any pending writes in upper transactions.
*/
bool
PendingWritesInUpperTransactions(Oid relfilenode, SubTransactionId currentSubXid)
PendingWritesInUpperTransactions(RelFileNumber relfilenumber,
SubTransactionId currentSubXid)
{
if (WriteStateMap == NULL)
{
return false;
}
WriteStateMapEntry *entry = hash_search(WriteStateMap, &relfilenode, HASH_FIND, NULL);
WriteStateMapEntry *entry = hash_search(WriteStateMap, &relfilenumber, HASH_FIND,
NULL);
if (entry && entry->writeStateStack != NULL)
{

2
src/backend/distributed/commands/citus_add_local_table_to_metadata.c
View File

@ -892,7 +892,7 @@ GetConstraintNameList(Oid relationId)
Relation pgConstraint = table_open(ConstraintRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_constraint_conrelid,
BTEqualStrategyNumber, F_OIDEQ, relationId);
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId));
bool useIndex = true;
SysScanDesc scanDescriptor = systable_beginscan(pgConstraint,

3
src/backend/distributed/commands/domain.c
View File

@ -64,7 +64,8 @@ CreateDomainStmt *
RecreateDomainStmt(Oid domainOid)
{
CreateDomainStmt *stmt = makeNode(CreateDomainStmt);
stmt->domainname = stringToQualifiedNameList(format_type_be_qualified(domainOid));
stmt->domainname = stringToQualifiedNameList_compat(format_type_be_qualified(
domainOid));
HeapTuple tup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(domainOid));
if (!HeapTupleIsValid(tup))

2
src/backend/distributed/commands/foreign_constraint.c
View File

@ -1227,7 +1227,7 @@ GetForeignKeyOids(Oid relationId, int flags)
Relation pgConstraint = table_open(ConstraintRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], pgConstraintTargetAttrNumber,
BTEqualStrategyNumber, F_OIDEQ, relationId);
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId));
SysScanDesc scanDescriptor = systable_beginscan(pgConstraint, indexOid, useIndex,
NULL, scanKeyCount, scanKey);

2
src/backend/distributed/commands/function.c
View File

@ -780,7 +780,7 @@ UpdateFunctionDistributionInfo(const ObjectAddress *distAddress,
ScanKeyInit(&scanKey[1], Anum_pg_dist_object_objid, BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(distAddress->objectId));
ScanKeyInit(&scanKey[2], Anum_pg_dist_object_objsubid, BTEqualStrategyNumber,
F_INT4EQ, ObjectIdGetDatum(distAddress->objectSubId));
F_INT4EQ, Int32GetDatum(distAddress->objectSubId));
SysScanDesc scanDescriptor = systable_beginscan(pgDistObjectRel,
DistObjectPrimaryKeyIndexId(),

11
src/backend/distributed/commands/index.c
View File

@ -18,6 +18,9 @@
#include "catalog/index.h"
#include "catalog/namespace.h"
#include "catalog/pg_class.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "catalog/pg_namespace.h"
#endif
#include "commands/defrem.h"
#include "commands/tablecmds.h"
#include "distributed/citus_ruleutils.h"
@ -558,7 +561,7 @@ ReindexStmtFindRelationOid(ReindexStmt *reindexStmt, bool missingOk)
{
relationId = RangeVarGetRelidExtended(reindexStmt->relation, lockmode,
(missingOk) ? RVR_MISSING_OK : 0,
RangeVarCallbackOwnsTable, NULL);
RANGE_VAR_TABLE_CALLBACK, NULL);
}
return relationId;
@ -1055,8 +1058,8 @@ RangeVarCallbackForDropIndex(const RangeVar *rel, Oid relOid, Oid oldRelOid, voi
errmsg("\"%s\" is not an index", rel->relname)));
/* Allow DROP to either table owner or schema owner */
if (!pg_class_ownercheck(relOid, GetUserId()) &&
!pg_namespace_ownercheck(classform->relnamespace, GetUserId()))
if (!object_ownercheck(RelationRelationId, relOid, GetUserId()) &&
!object_ownercheck(NamespaceRelationId, classform->relnamespace, GetUserId()))
{
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX, rel->relname);
}
@ -1140,7 +1143,7 @@ RangeVarCallbackForReindexIndex(const RangeVar *relation, Oid relId, Oid oldRelI
errmsg("\"%s\" is not an index", relation->relname)));
/* Check permissions */
if (!pg_class_ownercheck(relId, GetUserId()))
if (!object_ownercheck(RelationRelationId, relId, GetUserId()))
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX, relation->relname);
/* Lock heap before index to avoid deadlock. */

29
src/backend/distributed/commands/multi_copy.c
View File

@ -83,6 +83,9 @@
#include "distributed/locally_reserved_shared_connections.h"
#include "distributed/placement_connection.h"
#include "distributed/relation_access_tracking.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "distributed/relation_utils.h"
#endif
#include "distributed/remote_commands.h"
#include "distributed/remote_transaction.h"
#include "distributed/replication_origin_session_utils.h"
@ -3149,10 +3152,17 @@ CheckCopyPermissions(CopyStmt *copyStatement)
rel = table_openrv(copyStatement->relation,
is_from ? RowExclusiveLock : AccessShareLock);
range_table = CreateRangeTable(rel, required_access);
range_table = CreateRangeTable(rel);
RangeTblEntry *rte = (RangeTblEntry*) linitial(range_table);
tupDesc = RelationGetDescr(rel);
#if PG_VERSION_NUM >= PG_VERSION_16
/* create permission info for rte */
RTEPermissionInfo *perminfo = GetFilledPermissionInfo(rel->rd_id, rte->inh, required_access);
#else
rte->requiredPerms = required_access;
#endif
attnums = CopyGetAttnums(tupDesc, rel, copyStatement->attlist);
foreach(cur, attnums)
{
@ -3160,15 +3170,29 @@ CheckCopyPermissions(CopyStmt *copyStatement)
if (is_from)
{
#if PG_VERSION_NUM >= PG_VERSION_16
perminfo->insertedCols = bms_add_member(perminfo->insertedCols, attno);
#else
rte->insertedCols = bms_add_member(rte->insertedCols, attno);
#endif
}
else
{
#if PG_VERSION_NUM >= PG_VERSION_16
perminfo->selectedCols = bms_add_member(perminfo->selectedCols, attno);
#else
rte->selectedCols = bms_add_member(rte->selectedCols, attno);
#endif
}
}
#if PG_VERSION_NUM >= PG_VERSION_16
/* link rte to its permission info then check permissions */
rte->perminfoindex = 1;
ExecCheckPermissions(list_make1(rte), list_make1(perminfo), true);
#else
ExecCheckRTPerms(range_table, true);
#endif
/* TODO: Perform RLS checks once supported */
@ -3181,13 +3205,12 @@ CheckCopyPermissions(CopyStmt *copyStatement)
* CreateRangeTable creates a range table with the given relation.
*/
List *
CreateRangeTable(Relation rel, AclMode requiredAccess)
CreateRangeTable(Relation rel)
{
RangeTblEntry *rte = makeNode(RangeTblEntry);
rte->rtekind = RTE_RELATION;
rte->relid = rel->rd_id;
rte->relkind = rel->rd_rel->relkind;
rte->requiredPerms = requiredAccess;
return list_make1(rte);
}

11
src/backend/distributed/commands/role.c
View File

@ -703,12 +703,13 @@ MakeSetStatementArguments(char *configurationName, char *configurationValue)
* is no other way to determine allowed units, and value types other than
* using this function
*/
struct config_generic **gucVariables = get_guc_variables();
int numOpts = GetNumConfigOptions();
int gucCount = 0;
struct config_generic **gucVariables = get_guc_variables_compat(&gucCount);
struct config_generic **matchingConfig =
(struct config_generic **) SafeBsearch((void *) &key,
(void *) gucVariables,
numOpts,
gucCount,
sizeof(struct config_generic *),
ConfigGenericNameCompare);
@ -818,10 +819,12 @@ GenerateGrantRoleStmtsFromOptions(RoleSpec *roleSpec, List *options)
grantRoleStmt->grantee_roles = list_make1(roleSpec);
}
#if PG_VERSION_NUM < PG_VERSION_16
if (strcmp(option->defname, "adminmembers") == 0)
{
grantRoleStmt->admin_opt = true;
}
#endif
stmts = lappend(stmts, grantRoleStmt);
}
@ -868,7 +871,9 @@ GenerateGrantRoleStmtsOfRole(Oid roleid)
grantRoleStmt->grantor = NULL;
#if PG_VERSION_NUM < PG_VERSION_16
grantRoleStmt->admin_opt = membership->admin_option;
#endif
stmts = lappend(stmts, grantRoleStmt);
}

2
src/backend/distributed/commands/schema.c
View File

@ -369,7 +369,7 @@ SchemaHasDistributedTableWithFKey(char *schemaName)
Relation pgClass = table_open(RelationRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_class_relnamespace, BTEqualStrategyNumber,
F_OIDEQ, namespaceOid);
F_OIDEQ, ObjectIdGetDatum(namespaceOid));
SysScanDesc scanDescriptor = systable_beginscan(pgClass, scanIndexId, useIndex, NULL,
scanKeyCount, scanKey);

2
src/backend/distributed/commands/trigger.c
View File

@ -249,7 +249,7 @@ GetExplicitTriggerIdList(Oid relationId)
ScanKeyData scanKey[1];
ScanKeyInit(&scanKey[0], Anum_pg_trigger_tgrelid,
BTEqualStrategyNumber, F_OIDEQ, relationId);
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId));
bool useIndex = true;
SysScanDesc scanDescriptor = systable_beginscan(pgTrigger, TriggerRelidNameIndexId,

2
src/backend/distributed/commands/truncate.c
View File

@ -182,7 +182,7 @@ truncate_local_data_after_distributing_table(PG_FUNCTION_ARGS)
TruncateStmt *truncateStmt = makeNode(TruncateStmt);
char *relationName = generate_qualified_relation_name(relationId);
List *names = stringToQualifiedNameList(relationName);
List *names = stringToQualifiedNameList_compat(relationName);
truncateStmt->relations = list_make1(makeRangeVarFromNameList(names));
truncateStmt->restart_seqs = false;
truncateStmt->behavior = DROP_CASCADE;

7
src/backend/distributed/commands/type.c
View File

@ -187,7 +187,7 @@ RecreateCompositeTypeStmt(Oid typeOid)
Assert(get_typtype(typeOid) == TYPTYPE_COMPOSITE);
CompositeTypeStmt *stmt = makeNode(CompositeTypeStmt);
List *names = stringToQualifiedNameList(format_type_be_qualified(typeOid));
List *names = stringToQualifiedNameList_compat(format_type_be_qualified(typeOid));
stmt->typevar = makeRangeVarFromNameList(names);
stmt->coldeflist = CompositeTypeColumnDefList(typeOid);
@ -252,7 +252,7 @@ RecreateEnumStmt(Oid typeOid)
Assert(get_typtype(typeOid) == TYPTYPE_ENUM);
CreateEnumStmt *stmt = makeNode(CreateEnumStmt);
stmt->typeName = stringToQualifiedNameList(format_type_be_qualified(typeOid));
stmt->typeName = stringToQualifiedNameList_compat(format_type_be_qualified(typeOid));
stmt->vals = EnumValsList(typeOid);
return stmt;
@ -565,7 +565,8 @@ CreateTypeDDLCommandsIdempotent(const ObjectAddress *typeAddress)
char *
GenerateBackupNameForTypeCollision(const ObjectAddress *address)
{
List *names = stringToQualifiedNameList(format_type_be_qualified(address->objectId));
List *names = stringToQualifiedNameList_compat(format_type_be_qualified(
address->objectId));
RangeVar *rel = makeRangeVarFromNameList(names);
char *newName = palloc0(NAMEDATALEN);

4
src/backend/distributed/deparser/deparse_role_stmts.c
View File

@ -349,10 +349,12 @@ AppendGrantRoleStmt(StringInfo buf, GrantRoleStmt *stmt)
{
appendStringInfo(buf, "%s ", stmt->is_grant ? "GRANT" : "REVOKE");
#if PG_VERSION_NUM < PG_VERSION_16
if (!stmt->is_grant && stmt->admin_opt)
{
appendStringInfo(buf, "ADMIN OPTION FOR ");
}
#endif
AppendRoleList(buf, stmt->granted_roles);
@ -362,10 +364,12 @@ AppendGrantRoleStmt(StringInfo buf, GrantRoleStmt *stmt)
if (stmt->is_grant)
{
#if PG_VERSION_NUM < PG_VERSION_16
if (stmt->admin_opt)
{
appendStringInfo(buf, " WITH ADMIN OPTION");
}
#endif
if (stmt->grantor)
{

2
src/backend/distributed/metadata/dependency.c
View File

@ -1200,7 +1200,7 @@ FirstExtensionWithSchema(Oid schemaId)
ScanKeyData entry[1];
ScanKeyInit(&entry[0], Anum_pg_extension_extnamespace, BTEqualStrategyNumber,
F_INT4EQ, schemaId);
F_OIDEQ, ObjectIdGetDatum(schemaId));
SysScanDesc scan = systable_beginscan(relation, InvalidOid, false, NULL, 1, entry);
HeapTuple extensionTuple = systable_getnext(scan);

2
src/backend/distributed/metadata/distobject.c
View File

@ -510,7 +510,7 @@ UpdateDistributedObjectColocationId(uint32 oldColocationId,
/* scan pg_dist_object for colocationId equal to old colocationId */
ScanKeyInit(&scanKey[0], Anum_pg_dist_object_colocationid,
BTEqualStrategyNumber,
F_INT4EQ, UInt32GetDatum(oldColocationId));
F_INT4EQ, Int32GetDatum(oldColocationId));
SysScanDesc scanDescriptor = systable_beginscan(pgDistObjectRel,
InvalidOid,

4
src/backend/distributed/metadata/metadata_cache.c
View File

@ -83,7 +83,9 @@
#include "utils/memutils.h"
#include "utils/palloc.h"
#include "utils/rel.h"
#if PG_VERSION_NUM < PG_VERSION_16
#include "utils/relfilenodemap.h"
#endif
#include "utils/relmapper.h"
#include "utils/resowner.h"
#include "utils/syscache.h"
@ -5600,7 +5602,7 @@ role_exists(PG_FUNCTION_ARGS)
* Otherwise, this function returns NULL.
*/
char *
GetPoolinfoViaCatalog(int64 nodeId)
GetPoolinfoViaCatalog(int32 nodeId)
{
ScanKeyData scanKey[1];
const int scanKeyCount = 1;

4
src/backend/distributed/metadata/metadata_sync.c
View File

@ -1757,8 +1757,8 @@ GetFunctionDependenciesForObjects(ObjectAddress *objectAddress)
ObjectIdGetDatum(objectAddress->objectId));
ScanKeyInit(&key[2],
Anum_pg_depend_objsubid,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(objectAddress->objectSubId));
BTEqualStrategyNumber, F_INT4EQ,
Int32GetDatum(objectAddress->objectSubId));
SysScanDesc scan = systable_beginscan(depRel, DependDependerIndexId, true,
NULL, 3, key);

13
src/backend/distributed/metadata/metadata_utility.c
View File

@ -29,6 +29,9 @@
#include "catalog/pg_constraint.h"
#include "catalog/pg_extension.h"
#include "catalog/pg_namespace.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "catalog/pg_proc_d.h"
#endif
#include "catalog/pg_type.h"
#include "commands/extension.h"
#include "commands/sequence.h"
@ -2263,7 +2266,7 @@ EnsureTablePermissions(Oid relationId, AclMode mode)
void
EnsureTableOwner(Oid relationId)
{
if (!pg_class_ownercheck(relationId, GetUserId()))
if (!object_ownercheck(RelationRelationId, relationId, GetUserId()))
{
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLE,
get_rel_name(relationId));
@ -2278,7 +2281,7 @@ EnsureTableOwner(Oid relationId)
void
EnsureSchemaOwner(Oid schemaId)
{
if (!pg_namespace_ownercheck(schemaId, GetUserId()))
if (!object_ownercheck(NamespaceRelationId, schemaId, GetUserId()))
{
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
get_namespace_name(schemaId));
@ -2294,7 +2297,7 @@ EnsureSchemaOwner(Oid schemaId)
void
EnsureFunctionOwner(Oid functionId)
{
if (!pg_proc_ownercheck(functionId, GetUserId()))
if (!object_ownercheck(ProcedureRelationId, functionId, GetUserId()))
{
aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_FUNCTION,
get_func_name(functionId));
@ -3286,11 +3289,11 @@ BackgroundTaskHasUmnetDependencies(int64 jobId, int64 taskId)
/* pg_catalog.pg_dist_background_task_depend.job_id = jobId */
ScanKeyInit(&scanKey[0], Anum_pg_dist_background_task_depend_job_id,
BTEqualStrategyNumber, F_INT8EQ, jobId);
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(jobId));
/* pg_catalog.pg_dist_background_task_depend.task_id = $taskId */
ScanKeyInit(&scanKey[1], Anum_pg_dist_background_task_depend_task_id,
BTEqualStrategyNumber, F_INT8EQ, taskId);
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(taskId));
SysScanDesc scanDescriptor =
systable_beginscan(pgDistBackgroundTasksDepend,

5
src/backend/distributed/metadata/pg_get_object_address_13_14_15.c
View File

@ -93,7 +93,7 @@ PgGetObjectAddress(char *ttype, ArrayType *namearr, ArrayType *argsarr)
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("name or argument lists may not contain nulls")));
}
typename = typeStringToTypeName(TextDatumGetCString(elems[0]));
typename = typeStringToTypeName_compat(TextDatumGetCString(elems[0]), NULL);
}
else if (type == OBJECT_LARGEOBJECT)
{
@ -160,7 +160,8 @@ PgGetObjectAddress(char *ttype, ArrayType *namearr, ArrayType *argsarr)
errmsg("name or argument lists may not contain nulls")));
}
args = lappend(args,
typeStringToTypeName(TextDatumGetCString(elems[i])));
typeStringToTypeName_compat(TextDatumGetCString(elems[i]),
NULL));
}
}
else

6
src/backend/distributed/operations/shard_cleaner.c
View File

@ -1005,7 +1005,7 @@ ListCleanupRecordsForCurrentOperation(void)
ScanKeyData scanKey[1];
ScanKeyInit(&scanKey[0], Anum_pg_dist_cleanup_operation_id, BTEqualStrategyNumber,
F_INT8EQ, UInt64GetDatum(CurrentOperationId));
F_INT8EQ, Int64GetDatum(CurrentOperationId));
int scanKeyCount = 1;
Oid scanIndexId = InvalidOid;
@ -1119,7 +1119,7 @@ CleanupRecordExists(uint64 recordId)
bool indexOK = true;
ScanKeyInit(&scanKey[0], Anum_pg_dist_cleanup_record_id,
BTEqualStrategyNumber, F_INT8EQ, UInt64GetDatum(recordId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(recordId));
SysScanDesc scanDescriptor = systable_beginscan(pgDistCleanup,
DistCleanupPrimaryKeyIndexId(),
@ -1152,7 +1152,7 @@ DeleteCleanupRecordByRecordId(uint64 recordId)
bool indexOK = true;
ScanKeyInit(&scanKey[0], Anum_pg_dist_cleanup_record_id,
BTEqualStrategyNumber, F_INT8EQ, UInt64GetDatum(recordId));
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(recordId));
SysScanDesc scanDescriptor = systable_beginscan(pgDistCleanup,
DistCleanupPrimaryKeyIndexId(),

4
src/backend/distributed/operations/shard_rebalancer.c
View File

@ -2373,8 +2373,8 @@ GetSetCommandListForNewConnections(void)
{
List *commandList = NIL;
struct config_generic **guc_vars = get_guc_variables();
int gucCount = GetNumConfigOptions();
int gucCount = 0;
struct config_generic **guc_vars = get_guc_variables_compat(&gucCount);
for (int gucIndex = 0; gucIndex < gucCount; gucIndex++)
{

9
src/backend/distributed/planner/fast_path_router_planner.c
View File

@ -103,15 +103,24 @@ PlannedStmt *
GeneratePlaceHolderPlannedStmt(Query *parse)
{
PlannedStmt *result = makeNode(PlannedStmt);
#if PG_VERSION_NUM >= PG_VERSION_16
SeqScan *scanNode = makeNode(SeqScan);
Plan *plan = &(scanNode->scan.plan);
#else
Scan *scanNode = makeNode(Scan);
Plan *plan = &scanNode->plan;
#endif
Node *distKey PG_USED_FOR_ASSERTS_ONLY = NULL;
Assert(FastPathRouterQuery(parse, &distKey));
/* there is only a single relation rte */
#if PG_VERSION_NUM >= PG_VERSION_16
scanNode->scan.scanrelid = 1;
#else
scanNode->scanrelid = 1;
#endif
plan->targetlist =
copyObject(FetchStatementTargetList((Node *) parse));

2
src/backend/distributed/planner/query_colocation_checker.c
View File

@ -126,7 +126,7 @@ static RangeTblEntry *
AnchorRte(Query *subquery)
{
FromExpr *joinTree = subquery->jointree;
Relids joinRelIds = get_relids_in_jointree((Node *) joinTree, false);
Relids joinRelIds = get_relids_in_jointree_compat((Node *) joinTree, false, false);
int currentRTEIndex = -1;
RangeTblEntry *anchorRangeTblEntry = NULL;

6
src/backend/distributed/shared_library_init.c
View File

@ -2650,8 +2650,8 @@ RegisterCitusConfigVariables(void)
static void
OverridePostgresConfigProperties(void)
{
struct config_generic **guc_vars = get_guc_variables();
int gucCount = GetNumConfigOptions();
int gucCount = 0;
struct config_generic **guc_vars = get_guc_variables_compat(&gucCount);
for (int gucIndex = 0; gucIndex < gucCount; gucIndex++)
{
@ -2810,7 +2810,7 @@ ShowShardsForAppNamePrefixesCheckHook(char **newval, void **extra, GucSource sou
}
char *prefixAscii = pstrdup(appNamePrefix);
pg_clean_ascii(prefixAscii);
pg_clean_ascii_compat(prefixAscii, 0);
if (strcmp(prefixAscii, appNamePrefix) != 0)
{

10
src/backend/distributed/test/fake_am.c
View File

@ -254,7 +254,7 @@ fake_tuple_update(Relation relation, ItemPointer otid,
TupleTableSlot *slot, CommandId cid,
Snapshot snapshot, Snapshot crosscheck,
bool wait, TM_FailureData *tmfd,
LockTupleMode *lockmode, bool *update_indexes)
LockTupleMode *lockmode, TU_UpdateIndexes *update_indexes)
{
elog(ERROR, "fake_tuple_update not implemented");
}
@ -283,7 +283,7 @@ fake_finish_bulk_insert(Relation relation, int options)
*/
static void
fake_relation_set_new_filenode(Relation rel,
const RelFileNode *newrnode,
const RelFileLocator *newrnode,
char persistence,
TransactionId *freezeXid,
MultiXactId *minmulti)
@ -344,7 +344,7 @@ fake_relation_nontransactional_truncate(Relation rel)
static void
fake_copy_data(Relation rel, const RelFileNode *newrnode)
fake_copy_data(Relation rel, const RelFileLocator *newrnode)
{
elog(ERROR, "fake_copy_data not implemented");
}
@ -555,7 +555,11 @@ static const TableAmRoutine fake_methods = {
.tuple_satisfies_snapshot = fake_tuple_satisfies_snapshot,
.index_delete_tuples = fake_index_delete_tuples,
#if PG_VERSION_NUM >= PG_VERSION_16
.relation_set_new_filelocator = fake_relation_set_new_filenode,
#else
.relation_set_new_filenode = fake_relation_set_new_filenode,
#endif
.relation_nontransactional_truncate = fake_relation_nontransactional_truncate,
.relation_copy_data = fake_copy_data,
.relation_copy_for_cluster = fake_copy_for_cluster,

100
src/backend/distributed/transaction/lock_graph.c
View File

@ -725,7 +725,100 @@ UnlockLockData(void)
* which also contains entries for locks which have not been granted yet, but
* it does not reflect the order of the wait queue. We therefore handle the
* wait queue separately.
*
* We have separate blocks for PG16 and <PG16 because SHM_QUEUE is completely
* removed from PG16
*/
/*
* AddEdgesForWaitQueue adds an edge to the wait graph for processes in front of
* waitingProc in the wait queue that are trying to acquire a conflicting lock.
*
* We have separate blocks for PG16 and <PG16 because SHM_QUEUE is completely
* removed from PG16
*/
#if PG_VERSION_NUM >= PG_VERSION_16
static void
AddEdgesForLockWaits(WaitGraph *waitGraph, PGPROC *waitingProc, PROCStack *remaining)
{
/* the lock for which this process is waiting */
LOCK *waitLock = waitingProc->waitLock;
/* determine the conflict mask for the lock level used by the process */
LockMethod lockMethodTable = GetLocksMethodTable(waitLock);
int conflictMask = lockMethodTable->conflictTab[waitingProc->waitLockMode];
/* iterate through the queue of processes holding the lock */
dlist_head *procLocks = &waitLock->procLocks;
dlist_iter iter;
dlist_foreach(iter, procLocks)
{
PROCLOCK *procLock = dlist_container(PROCLOCK, lockLink, iter.cur);
PGPROC *currentProc = procLock->tag.myProc;
/*
* Skip processes from the same lock group, processes that don't conflict,
* and processes that are waiting on safe operations.
*/
if (!IsSameLockGroup(waitingProc, currentProc) &&
IsConflictingLockMask(procLock->holdMask, conflictMask) &&
!IsProcessWaitingForSafeOperations(currentProc))
{
AddWaitEdge(waitGraph, waitingProc, currentProc, remaining);
}
}
}
static void
AddEdgesForWaitQueue(WaitGraph *waitGraph, PGPROC *waitingProc, PROCStack *remaining)
{
/* the lock for which this process is waiting */
LOCK *waitLock = waitingProc->waitLock;
/* determine the conflict mask for the lock level used by the process */
LockMethod lockMethodTable = GetLocksMethodTable(waitLock);
int conflictMask = lockMethodTable->conflictTab[waitingProc->waitLockMode];
/* iterate through the wait queue */
dclist_head *waitQueue = &waitLock->waitProcs;
dlist_iter iter;
dclist_foreach(iter, waitQueue)
{
PGPROC *currentProc = dlist_container(PGPROC, links, iter.cur);
if (currentProc == waitingProc)
{
/*
* Iterate through the queue from the start until we encounter waitingProc,
* since we only care about processes in front of waitingProc in the queue.
*/
break;
}
int awaitMask = LOCKBIT_ON(currentProc->waitLockMode);
/*
* Skip processes from the same lock group, processes that don't conflict,
* and processes that are waiting on safe operations.
*/
if (!IsSameLockGroup(waitingProc, currentProc) &&
IsConflictingLockMask(awaitMask, conflictMask) &&
!IsProcessWaitingForSafeOperations(currentProc))
{
AddWaitEdge(waitGraph, waitingProc, currentProc, remaining);
}
currentProc = (PGPROC *) currentProc->links.next;
}
}
#else
static void
AddEdgesForLockWaits(WaitGraph *waitGraph, PGPROC *waitingProc, PROCStack *remaining)
{
@ -762,10 +855,6 @@ AddEdgesForLockWaits(WaitGraph *waitGraph, PGPROC *waitingProc, PROCStack *remai
}
/*
* AddEdgesForWaitQueue adds an edge to the wait graph for processes in front of
* waitingProc in the wait queue that are trying to acquire a conflicting lock.
*/
static void
AddEdgesForWaitQueue(WaitGraph *waitGraph, PGPROC *waitingProc, PROCStack *remaining)
{
@ -805,6 +894,9 @@ AddEdgesForWaitQueue(WaitGraph *waitGraph, PGPROC *waitingProc, PROCStack *remai
}
#endif
/*
* AddWaitEdge adds a new wait edge to a wait graph. The nodes in the graph are
* transactions and an edge indicates the "waiting" process is blocked on a lock

3
src/backend/distributed/utils/aggregate_utils.c
View File

@ -168,7 +168,8 @@ aclcheckAggregate(ObjectType objectType, Oid userOid, Oid funcOid)
AclResult aclresult;
if (funcOid != InvalidOid)
{
aclresult = pg_proc_aclcheck(funcOid, userOid, ACL_EXECUTE);
aclresult = object_aclcheck(ProcedureRelationId, funcOid, userOid,
ACL_EXECUTE);
if (aclresult != ACLCHECK_OK)
{
aclcheck_error(aclresult, objectType, get_func_name(funcOid));

10
src/backend/distributed/utils/colocation_utils.c
View File

@ -532,7 +532,7 @@ ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType,
ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_distributioncolumntype,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(distributionColumnType));
ScanKeyInit(&scanKey[1], Anum_pg_dist_colocation_shardcount,
BTEqualStrategyNumber, F_INT4EQ, UInt32GetDatum(shardCount));
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(shardCount));
ScanKeyInit(&scanKey[2], Anum_pg_dist_colocation_replicationfactor,
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(replicationFactor));
ScanKeyInit(&scanKey[3], Anum_pg_dist_colocation_distributioncolumncollation,
@ -989,7 +989,7 @@ ColocationGroupTableList(uint32 colocationId, uint32 count)
}
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_colocationid,
BTEqualStrategyNumber, F_INT4EQ, UInt32GetDatum(colocationId));
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(colocationId));
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
@ -1166,7 +1166,7 @@ ColocatedNonPartitionShardIntervalList(ShardInterval *shardInterval)
* guarantee that the table isn't dropped for the remainder of the transaction.
*/
Oid
ColocatedTableId(Oid colocationId)
ColocatedTableId(int32 colocationId)
{
Oid colocatedTableId = InvalidOid;
bool indexOK = true;
@ -1183,7 +1183,7 @@ ColocatedTableId(Oid colocationId)
}
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_colocationid,
BTEqualStrategyNumber, F_INT4EQ, ObjectIdGetDatum(colocationId));
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(colocationId));
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
@ -1292,7 +1292,7 @@ DeleteColocationGroupLocally(uint32 colocationId)
Relation pgDistColocation = table_open(DistColocationRelationId(), RowExclusiveLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_colocationid,
BTEqualStrategyNumber, F_INT4EQ, UInt32GetDatum(colocationId));
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(colocationId));
SysScanDesc scanDescriptor = systable_beginscan(pgDistColocation, InvalidOid, indexOK,
NULL, scanKeyCount, scanKey);

3
src/backend/distributed/utils/function_utils.c
View File

@ -41,7 +41,8 @@ FunctionOidExtended(const char *schemaName, const char *functionName, int argume
bool missingOK)
{
char *qualifiedFunctionName = quote_qualified_identifier(schemaName, functionName);
List *qualifiedFunctionNameList = stringToQualifiedNameList(qualifiedFunctionName);
List *qualifiedFunctionNameList = stringToQualifiedNameList_compat(
qualifiedFunctionName);
List *argumentList = NIL;
const bool findVariadics = false;
const bool findDefaults = false;

4
src/backend/distributed/utils/multi_partitioning_utils.c
View File

@ -411,9 +411,9 @@ CheckConstraintNameListForRelation(Oid relationId)
Relation pgConstraint = table_open(ConstraintRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_constraint_conrelid,
BTEqualStrategyNumber, F_OIDEQ, relationId);
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId));
ScanKeyInit(&scanKey[1], Anum_pg_constraint_contype,
BTEqualStrategyNumber, F_CHAREQ, CONSTRAINT_CHECK);
BTEqualStrategyNumber, F_CHAREQ, CharGetDatum(CONSTRAINT_CHECK));
bool useIndex = false;
SysScanDesc scanDescriptor = systable_beginscan(pgConstraint, InvalidOid, useIndex,

37
src/backend/distributed/utils/relation_utils.c
View File

@ -14,6 +14,9 @@
#include "distributed/relation_utils.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "miscadmin.h"
#endif
#include "utils/lsyscache.h"
#include "utils/rel.h"
@ -28,3 +31,37 @@ RelationGetNamespaceName(Relation relation)
char *namespaceName = get_namespace_name(namespaceId);
return namespaceName;
}
#if PG_VERSION_NUM >= PG_VERSION_16
/*
* GetFilledPermissionInfo creates RTEPermissionInfo for a given RTE
* and fills it with given data and returns this RTEPermissionInfo object.
* Added this function since Postgres's addRTEPermissionInfo doesn't fill the data.
*
* Given data consists of relid, inh and requiredPerms
* Took a quick look around Postgres, unless specified otherwise,
* we are dealing with GetUserId().
* Currently the following entries are filled like this:
* perminfo->checkAsUser = GetUserId();
* perminfo->selectedCols = NULL;
* perminfo->insertedCols = NULL;
* perminfo->updatedCols = NULL;
*/
RTEPermissionInfo *
GetFilledPermissionInfo(Oid relid, bool inh, AclMode requiredPerms)
{
RTEPermissionInfo *perminfo = makeNode(RTEPermissionInfo);
perminfo->relid = relid;
perminfo->inh = inh;
perminfo->requiredPerms = requiredPerms;
perminfo->checkAsUser = GetUserId();
perminfo->selectedCols = NULL;
perminfo->insertedCols = NULL;
perminfo->updatedCols = NULL;
return perminfo;
}
#endif

2
src/backend/distributed/utils/tenant_schema_metadata.c
View File

@ -134,7 +134,7 @@ ColocationIdGetTenantSchemaId(uint32 colocationId)
AccessShareLock);
ScanKeyData scanKey[1];
ScanKeyInit(&scanKey[0], Anum_pg_dist_schema_colocationid,
BTEqualStrategyNumber, F_INT4EQ, UInt32GetDatum(colocationId));
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(colocationId));
bool indexOk = true;
SysScanDesc scanDescriptor = systable_beginscan(pgDistTenantSchema,

2
src/backend/distributed/worker/worker_create_or_replace.c
View File

@ -525,7 +525,7 @@ CreateRenameTypeStmt(const ObjectAddress *address, char *newName)
RenameStmt *stmt = makeNode(RenameStmt);
stmt->renameType = OBJECT_TYPE;
stmt->object = (Node *) stringToQualifiedNameList(format_type_be_qualified(
stmt->object = (Node *) stringToQualifiedNameList_compat(format_type_be_qualified(
address->objectId));
stmt->newname = newName;

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

@ -16,9 +16,14 @@
#include "fmgr.h"
#include "lib/stringinfo.h"
#include "nodes/parsenodes.h"
#include "pg_version_compat.h"
#include "storage/bufpage.h"
#include "storage/lockdefs.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "storage/relfilelocator.h"
#else
#include "storage/relfilenode.h"
#endif
#include "utils/relcache.h"
#include "utils/snapmgr.h"
@ -224,7 +229,7 @@ extern void columnar_init_gucs(void);
extern CompressionType ParseCompressionType(const char *compressionTypeString);
/* Function declarations for writing to a columnar table */
extern ColumnarWriteState * ColumnarBeginWrite(RelFileNode relfilenode,
extern ColumnarWriteState * ColumnarBeginWrite(RelFileLocator relfilelocator,
ColumnarOptions options,
TupleDesc tupleDescriptor);
extern uint64 ColumnarWriteRow(ColumnarWriteState *state, Datum *columnValues,
@ -279,21 +284,21 @@ extern bool ReadColumnarOptions(Oid regclass, ColumnarOptions *options);
extern bool IsColumnarTableAmTable(Oid relationId);
/* columnar_metadata_tables.c */
extern void DeleteMetadataRows(RelFileNode relfilenode);
extern void DeleteMetadataRows(RelFileLocator relfilelocator);
extern uint64 ColumnarMetadataNewStorageId(void);
extern uint64 GetHighestUsedAddress(RelFileNode relfilenode);
extern uint64 GetHighestUsedAddress(RelFileLocator relfilelocator);
extern EmptyStripeReservation * ReserveEmptyStripe(Relation rel, uint64 columnCount,
uint64 chunkGroupRowCount,
uint64 stripeRowCount);
extern StripeMetadata * CompleteStripeReservation(Relation rel, uint64 stripeId,
uint64 sizeBytes, uint64 rowCount,
uint64 chunkCount);
extern void SaveStripeSkipList(RelFileNode relfilenode, uint64 stripe,
extern void SaveStripeSkipList(RelFileLocator relfilelocator, uint64 stripe,
StripeSkipList *stripeSkipList,
TupleDesc tupleDescriptor);
extern void SaveChunkGroups(RelFileNode relfilenode, uint64 stripe,
extern void SaveChunkGroups(RelFileLocator relfilelocator, uint64 stripe,
List *chunkGroupRowCounts);
extern StripeSkipList * ReadStripeSkipList(RelFileNode relfilenode, uint64 stripe,
extern StripeSkipList * ReadStripeSkipList(RelFileLocator relfilelocator, uint64 stripe,
TupleDesc tupleDescriptor,
uint32 chunkCount,
Snapshot snapshot);
@ -316,15 +321,16 @@ extern ColumnarWriteState * columnar_init_write_state(Relation relation, TupleDe
tupdesc,
Oid tupSlotRelationId,
SubTransactionId currentSubXid);
extern void FlushWriteStateForRelfilenode(Oid relfilenode, SubTransactionId
currentSubXid);
extern void FlushWriteStateForRelfilenumber(RelFileNumber relfilenumber,
SubTransactionId currentSubXid);
extern void FlushWriteStateForAllRels(SubTransactionId currentSubXid, SubTransactionId
parentSubXid);
extern void DiscardWriteStateForAllRels(SubTransactionId currentSubXid, SubTransactionId
parentSubXid);
extern void MarkRelfilenodeDropped(Oid relfilenode, SubTransactionId currentSubXid);
extern void NonTransactionDropWriteState(Oid relfilenode);
extern bool PendingWritesInUpperTransactions(Oid relfilenode,
extern void MarkRelfilenumberDropped(RelFileNumber relfilenumber,
SubTransactionId currentSubXid);
extern void NonTransactionDropWriteState(RelFileNumber relfilenumber);
extern bool PendingWritesInUpperTransactions(RelFileNumber relfilenumber,
SubTransactionId currentSubXid);
extern MemoryContext GetWriteContextForDebug(void);

4
src/include/columnar/columnar_metadata.h
View File

@ -12,6 +12,8 @@
#ifndef COLUMNAR_METADATA_H
#define COLUMNAR_METADATA_H
#include "pg_version_compat.h"
/*
* StripeMetadata represents information about a stripe. This information is
* stored in the metadata table "columnar.stripe".
@ -49,7 +51,7 @@ typedef struct EmptyStripeReservation
uint64 stripeFirstRowNumber;
} EmptyStripeReservation;
extern List * StripesForRelfilenode(RelFileNode relfilenode);
extern List * StripesForRelfilelocator(RelFileLocator relfilelocator);
extern void ColumnarStorageUpdateIfNeeded(Relation rel, bool isUpgrade);
extern List * ExtractColumnarRelOptions(List *inOptions, List **outColumnarOptions);
extern void SetColumnarRelOptions(RangeVar *rv, List *reloptions);

2
src/include/distributed/colocation_utils.h
View File

@ -24,7 +24,7 @@ extern bool ShardsColocated(ShardInterval *leftShardInterval,
extern List * ColocatedTableList(Oid distributedTableId);
extern List * ColocatedShardIntervalList(ShardInterval *shardInterval);
extern List * ColocatedNonPartitionShardIntervalList(ShardInterval *shardInterval);
extern Oid ColocatedTableId(Oid colocationId);
extern Oid ColocatedTableId(int32 colocationId);
extern uint64 ColocatedShardIdInRelation(Oid relationId, int shardIndex);
uint32 ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType,
Oid distributionColumnCollation);

2
src/include/distributed/commands/multi_copy.h
View File

@ -182,7 +182,7 @@ extern void AppendCopyRowData(Datum *valueArray, bool *isNullArray,
extern void AppendCopyBinaryHeaders(CopyOutState headerOutputState);
extern void AppendCopyBinaryFooters(CopyOutState footerOutputState);
extern void EndRemoteCopy(int64 shardId, List *connectionList);
extern List * CreateRangeTable(Relation rel, AclMode requiredAccess);
extern List * CreateRangeTable(Relation rel);
extern Node * ProcessCopyStmt(CopyStmt *copyStatement,
QueryCompletion *completionTag,
const char *queryString);

1
src/include/distributed/errormessage.h
View File

@ -12,6 +12,7 @@
#include "c.h"
#include "distributed/citus_nodes.h"
#include "pg_version_compat.h"
typedef struct DeferredErrorMessage

2
src/include/distributed/metadata_cache.h
View File

@ -320,6 +320,6 @@ extern const char * CurrentDatabaseName(void);
/* connection-related functions */
extern char * GetAuthinfoViaCatalog(const char *roleName, int64 nodeId);
extern char * GetPoolinfoViaCatalog(int64 nodeId);
extern char * GetPoolinfoViaCatalog(int32 nodeId);
#endif /* METADATA_CACHE_H */

8
src/include/distributed/relation_utils.h
View File

@ -13,8 +13,16 @@
#include "postgres.h"
#include "distributed/pg_version_constants.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "parser/parse_relation.h"
#endif
#include "utils/relcache.h"
extern char * RelationGetNamespaceName(Relation relation);
#if PG_VERSION_NUM >= PG_VERSION_16
extern RTEPermissionInfo * GetFilledPermissionInfo(Oid relid, bool inh,
AclMode requiredPerms);
#endif
#endif /* RELATION_UTILS_H */

133
src/include/pg_version_compat.h
View File

@ -13,6 +13,139 @@
#include "distributed/pg_version_constants.h"
#if PG_VERSION_NUM >= PG_VERSION_16
#include "utils/guc_tables.h"
#define pg_clean_ascii_compat(a, b) pg_clean_ascii(a, b)
#define RelationPhysicalIdentifier_compat(a) ((a)->rd_locator)
#define RelationTablespace_compat(a) (a.spcOid)
#define RelationPhysicalIdentifierNumber_compat(a) (a.relNumber)
#define RelationPhysicalIdentifierNumberPtr_compat(a) (a->relNumber)
#define RelationPhysicalIdentifierBackend_compat(a) (a->smgr_rlocator.locator)
#define float_abs(a) fabs(a)
#define RANGE_VAR_TABLE_CALLBACK RangeVarCallbackMaintainsTable
#define tuplesort_getdatum_compat(a, b, c, d, e, f) tuplesort_getdatum(a, b, c, d, e, f)
static inline struct config_generic **
get_guc_variables_compat(int *gucCount)
{
return get_guc_variables(gucCount);
}
#define PG_FUNCNAME_MACRO __func__
#define stringToQualifiedNameList_compat(a) stringToQualifiedNameList(a, NULL)
#define typeStringToTypeName_compat(a, b) typeStringToTypeName(a, b)
#define get_relids_in_jointree_compat(a, b, c) get_relids_in_jointree(a, b, c)
#define object_ownercheck(a, b, c) object_ownercheck(a, b, c)
#define object_aclcheck(a, b, c, d) object_aclcheck(a, b, c, d)
#else
#include "catalog/pg_class_d.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_proc_d.h"
#include "storage/relfilenode.h"
#include "utils/guc.h"
#include "utils/guc_tables.h"
#define RANGE_VAR_TABLE_CALLBACK RangeVarCallbackOwnsTable
#define pg_clean_ascii_compat(a, b) pg_clean_ascii(a)
#define RelationPhysicalIdentifier_compat(a) ((a)->rd_node)
#define RelationTablespace_compat(a) (a.spcNode)
#define RelationPhysicalIdentifierNumber_compat(a) (a.relNode)
#define RelationPhysicalIdentifierNumberPtr_compat(a) (a->relNode)
#define RelationPhysicalIdentifierBackend_compat(a) (a->smgr_rnode.node)
typedef RelFileNode RelFileLocator;
typedef Oid RelFileNumber;
#define RelidByRelfilenumber(a, b) RelidByRelfilenode(a, b)
#define float_abs(a) Abs(a)
#define tuplesort_getdatum_compat(a, b, c, d, e, f) tuplesort_getdatum(a, b, d, e, f)
static inline struct config_generic **
get_guc_variables_compat(int *gucCount)
{
*gucCount = GetNumConfigOptions();
return get_guc_variables();
}
#define stringToQualifiedNameList_compat(a) stringToQualifiedNameList(a)
#define typeStringToTypeName_compat(a, b) typeStringToTypeName(a)
#define get_relids_in_jointree_compat(a, b, c) get_relids_in_jointree(a, b)
static inline bool
object_ownercheck(Oid classid, Oid objectid, Oid roleid)
{
switch (classid)
{
case RelationRelationId:
{
return pg_class_ownercheck(objectid, roleid);
}
case NamespaceRelationId:
{
return pg_namespace_ownercheck(objectid, roleid);
}
case ProcedureRelationId:
{
return pg_proc_ownercheck(objectid, roleid);
}
default:
{
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Missing classid:%d",
classid)));
}
}
}
static inline AclResult
object_aclcheck(Oid classid, Oid objectid, Oid roleid, AclMode mode)
{
switch (classid)
{
case NamespaceRelationId:
{
return pg_namespace_aclcheck(objectid, roleid, mode);
}
case ProcedureRelationId:
{
return pg_proc_aclcheck(objectid, roleid, mode);
}
default:
{
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("Missing classid:%d",
classid)));
}
}
}
typedef bool TU_UpdateIndexes;
#endif
#if PG_VERSION_NUM >= PG_VERSION_15
#define ProcessCompletedNotifies()
#define RelationCreateStorage_compat(a, b, c) RelationCreateStorage(a, b, c)

2
src/test/regress/expected/columnar_create.out
View File

@ -54,7 +54,7 @@ CREATE MATERIALIZED VIEW columnar_table_1_mv USING columnar
AS SELECT * FROM columnar_table_1;
SELECT columnar.get_storage_id(oid) AS columnar_table_1_mv_storage_id
FROM pg_class WHERE relname='columnar_table_1_mv' \gset
-- test columnar_relation_set_new_filenode
-- test columnar_relation_set_new_filelocator
REFRESH MATERIALIZED VIEW columnar_table_1_mv;
SELECT columnar_test_helpers.columnar_metadata_has_storage_id(:columnar_table_1_mv_storage_id);
columnar_metadata_has_storage_id

2
src/test/regress/sql/columnar_create.sql
View File

@ -56,7 +56,7 @@ AS SELECT * FROM columnar_table_1;
SELECT columnar.get_storage_id(oid) AS columnar_table_1_mv_storage_id
FROM pg_class WHERE relname='columnar_table_1_mv' \gset
-- test columnar_relation_set_new_filenode
-- test columnar_relation_set_new_filelocator
REFRESH MATERIALIZED VIEW columnar_table_1_mv;
SELECT columnar_test_helpers.columnar_metadata_has_storage_id(:columnar_table_1_mv_storage_id);