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

First formatting attempt 

Skipped csql, ruleutils, readfuncs, and functions obviously copied from
PostgreSQL. Seeing how this looks, then continuing.
pull/327/head
Jason Petersen 2016-02-05 13:18:54 -07:00
parent 334f800016
commit fdb37682b2
No known key found for this signature in database
GPG Key ID: 9F1D3510D110ABA9
55 changed files with 1391 additions and 1314 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/backend/distributed/commands/transmit.c
View File

@ -271,17 +271,27 @@ ReceiveCopyData(StringInfo copyData)
switch (messageType)
{
case 'd': /* CopyData */
{
copyDone = false;
break;
}
case 'c': /* CopyDone */
{
copyDone = true;
break;
}
case 'f': /* CopyFail */
{
ereport(ERROR, (errcode(ERRCODE_QUERY_CANCELED),
errmsg("COPY data failed: %s", pq_getmsgstring(copyData))));
break;
}
case 'H': /* Flush */
case 'S': /* Sync */
{
/*
* Ignore Flush/Sync for the convenience of client libraries (such
* as libpq) that may send those without noticing that the command
@ -289,12 +299,16 @@ ReceiveCopyData(StringInfo copyData)
*/
copyDone = false;
break;
}
default:
{
ereport(ERROR, (errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg("unexpected message type 0x%02X during COPY data",
messageType)));
break;
}
}
return copyDone;
}

1
src/backend/distributed/executor/multi_executor.c
View File

@ -157,7 +157,6 @@ multi_ExecutorStart(QueryDesc *queryDesc, int eflags)
queryDesc->plannedstmt = masterSelectPlan;
eflags |= EXEC_FLAG_CITUS_MASTER_SELECT;
}
}
/* if the execution is not done for router executor, drop into standard executor */

17
src/backend/distributed/executor/multi_real_time_executor.c
View File

@ -287,8 +287,9 @@ ManageTaskExecution(Task *task, TaskExecution *taskExecution)
uint32 currentCount = taskExecution->connectPollCount;
if (currentCount >= maxCount)
{
ereport(WARNING, (errmsg("could not establish asynchronous connection "
"after %u ms", REMOTE_NODE_CONNECT_TIMEOUT)));
ereport(WARNING, (errmsg("could not establish asynchronous "
"connection after %u ms",
REMOTE_NODE_CONNECT_TIMEOUT)));
taskStatusArray[currentIndex] = EXEC_TASK_FAILED;
}
@ -342,7 +343,8 @@ ManageTaskExecution(Task *task, TaskExecution *taskExecution)
{
List *dataFetchTaskList = task->dependedTaskList;
int32 dataFetchTaskIndex = taskExecution->dataFetchTaskIndex;
Task *dataFetchTask = (Task *) list_nth(dataFetchTaskList, dataFetchTaskIndex);
Task *dataFetchTask = (Task *) list_nth(dataFetchTaskList,
dataFetchTaskIndex);
char *dataFetchQuery = dataFetchTask->queryString;
int32 connectionId = connectionIdArray[currentIndex];
@ -411,11 +413,13 @@ ManageTaskExecution(Task *task, TaskExecution *taskExecution)
StringInfo computeTaskQuery = makeStringInfo();
if (BinaryMasterCopyFormat)
{
appendStringInfo(computeTaskQuery, COPY_QUERY_TO_STDOUT_BINARY, queryString);
appendStringInfo(computeTaskQuery, COPY_QUERY_TO_STDOUT_BINARY,
queryString);
}
else
{
appendStringInfo(computeTaskQuery, COPY_QUERY_TO_STDOUT_TEXT, queryString);
appendStringInfo(computeTaskQuery, COPY_QUERY_TO_STDOUT_TEXT,
queryString);
}
querySent = MultiClientSendQuery(connectionId, computeTaskQuery->data);
@ -475,7 +479,8 @@ ManageTaskExecution(Task *task, TaskExecution *taskExecution)
else
{
ereport(WARNING, (errcode_for_file_access(),
errmsg("could not open file \"%s\": %m", filename)));
errmsg("could not open file \"%s\": %m",
filename)));
taskStatusArray[currentIndex] = EXEC_TASK_FAILED;
}

4
src/backend/distributed/executor/multi_router_executor.c
View File

@ -80,6 +80,7 @@ RouterExecutorStart(QueryDesc *queryDesc, int eflags, Task *task)
queryDesc->estate = executorState;
#if (PG_VERSION_NUM < 90500)
/* make sure that upsertQuery is false for versions that UPSERT is not available */
Assert(task->upsertQuery == false);
#endif
@ -219,9 +220,9 @@ RouterExecutorRun(QueryDesc *queryDesc, ScanDirection direction, long count, Tas
}
MemoryContextSwitchTo(oldcontext);
}
/*
* ExecuteDistributedModify is the main entry point for modifying distributed
* tables. A distributed modification is successful if any placement of the
@ -532,6 +533,7 @@ StoreQueryResult(PGconn *connection, TupleDesc tupleDescriptor,
return true;
}
/*
* RouterExecutorFinish cleans up after a distributed execution.
*/

25
src/backend/distributed/executor/multi_task_tracker_executor.c
View File

@ -41,7 +41,6 @@ typedef struct TaskMapKey
TaskType taskType;
uint64 jobId;
uint32 taskId;
} TaskMapKey;
@ -53,7 +52,6 @@ typedef struct TaskMapEntry
{
TaskMapKey key;
Task *task;
} TaskMapEntry;
@ -83,7 +81,8 @@ static TaskTracker * TrackerHashLookup(HTAB *trackerHash, const char *nodeName,
static TaskExecStatus ManageTaskExecution(TaskTracker *taskTracker,
TaskTracker *sourceTaskTracker,
Task *task, TaskExecution *taskExecution);
static TransmitExecStatus ManageTransmitExecution(TaskTracker *transmitTracker, Task *task,
static TransmitExecStatus ManageTransmitExecution(TaskTracker *transmitTracker,
Task *task,
TaskExecution *taskExecution);
static bool TaskExecutionsCompleted(List *taskList);
static StringInfo MapFetchTaskQueryString(Task *mapFetchTask, Task *mapTask);
@ -826,7 +825,8 @@ TrackerConnectPoll(TaskTracker *taskTracker)
uint32 nodePort = taskTracker->workerPort;
char *nodeDatabase = get_database_name(MyDatabaseId);
int32 connectionId = MultiClientConnectStart(nodeName, nodePort, nodeDatabase);
int32 connectionId = MultiClientConnectStart(nodeName, nodePort,
nodeDatabase);
if (connectionId != INVALID_CONNECTION_ID)
{
taskTracker->connectionId = connectionId;
@ -869,8 +869,9 @@ TrackerConnectPoll(TaskTracker *taskTracker)
uint32 currentCount = taskTracker->connectPollCount;
if (currentCount >= maxCount)
{
ereport(WARNING, (errmsg("could not establish asynchronous connection "
"after %u ms", REMOTE_NODE_CONNECT_TIMEOUT)));
ereport(WARNING, (errmsg("could not establish asynchronous "
"connection after %u ms",
REMOTE_NODE_CONNECT_TIMEOUT)));
taskTracker->trackerStatus = TRACKER_CONNECTION_FAILED;
@ -1212,7 +1213,8 @@ ManageTaskExecution(TaskTracker *taskTracker, TaskTracker *sourceTaskTracker,
default:
{
/* we fatal here to avoid leaking client-side resources */
ereport(FATAL, (errmsg("invalid execution status: %d", currentExecutionStatus)));
ereport(FATAL, (errmsg("invalid execution status: %d",
currentExecutionStatus)));
break;
}
}
@ -1327,7 +1329,8 @@ ManageTransmitExecution(TaskTracker *transmitTracker,
else
{
ereport(WARNING, (errcode_for_file_access(),
errmsg("could not open file \"%s\": %m", filename)));
errmsg("could not open file \"%s\": %m",
filename)));
nextTransmitStatus = EXEC_TRANSMIT_TRACKER_RETRY;
}
@ -1463,7 +1466,8 @@ ManageTransmitExecution(TaskTracker *transmitTracker,
default:
{
/* we fatal here to avoid leaking client-side resources */
ereport(FATAL, (errmsg("invalid transmit status: %d", currentTransmitStatus)));
ereport(FATAL, (errmsg("invalid transmit status: %d",
currentTransmitStatus)));
break;
}
}
@ -2833,7 +2837,8 @@ TrackerHashCleanupJob(HTAB *taskTrackerHash, Task *jobCleanupTask)
if (queryStatus == CLIENT_QUERY_DONE)
{
ereport(DEBUG4, (errmsg("completed cleanup query for job " UINT64_FORMAT
" on node \"%s:%u\"", jobId, nodeName, nodePort)));
" on node \"%s:%u\"", jobId, nodeName,
nodePort)));
/* clear connection for future cleanup queries */
taskTracker->connectionBusy = false;

5
src/backend/distributed/executor/multi_utility.c
View File

@ -905,8 +905,9 @@ ExecuteCommandOnWorkerShards(Oid relationId, const char *commandString,
}
else
{
ereport(DEBUG2, (errmsg("applied command on shard " UINT64_FORMAT " on "
"node %s:%d", shardId, workerName, workerPort)));
ereport(DEBUG2, (errmsg("applied command on shard " UINT64_FORMAT
" on node %s:%d", shardId, workerName,
workerPort)));
}
isFirstPlacement = false;

15
src/backend/distributed/master/master_delete_protocol.c
View File

@ -167,7 +167,8 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
shardPlacementList = ShardPlacementList(shardId);
foreach(shardPlacementCell, shardPlacementList)
{
ShardPlacement *shardPlacement = (ShardPlacement *) lfirst(shardPlacementCell);
ShardPlacement *shardPlacement =
(ShardPlacement *) lfirst(shardPlacementCell);
char *workerName = shardPlacement->nodeName;
uint32 workerPort = shardPlacement->nodePort;
bool dropSuccessful = false;
@ -176,14 +177,17 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
char tableType = get_rel_relkind(relationId);
if (tableType == RELKIND_RELATION)
{
appendStringInfo(workerDropQuery, DROP_REGULAR_TABLE_COMMAND, quotedShardName);
appendStringInfo(workerDropQuery, DROP_REGULAR_TABLE_COMMAND,
quotedShardName);
}
else if (tableType == RELKIND_FOREIGN_TABLE)
{
appendStringInfo(workerDropQuery, DROP_FOREIGN_TABLE_COMMAND, quotedShardName);
appendStringInfo(workerDropQuery, DROP_FOREIGN_TABLE_COMMAND,
quotedShardName);
}
dropSuccessful = ExecuteRemoteCommand(workerName, workerPort, workerDropQuery);
dropSuccessful = ExecuteRemoteCommand(workerName, workerPort,
workerDropQuery);
if (dropSuccessful)
{
droppedPlacementList = lappend(droppedPlacementList, shardPlacement);
@ -227,7 +231,8 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
if (QueryCancelPending)
{
ereport(WARNING, (errmsg("cancel requests are ignored during shard deletion")));
ereport(WARNING, (errmsg("cancel requests are ignored during shard "
"deletion")));
QueryCancelPending = false;
}

5
src/backend/distributed/master/master_node_protocol.c
View File

@ -709,11 +709,16 @@ hostname_client_addr(void)
#ifdef HAVE_IPV6
case AF_INET6:
#endif
{
break;
}
default:
{
ereport(ERROR, (errmsg("invalid address family in connection")));
break;
}
}
remoteHost = palloc0(remoteHostLen);

4
src/backend/distributed/planner/modify_planner.c
View File

@ -393,6 +393,7 @@ DistributedModifyTask(Query *query)
query->onConflict = RebuildOnConflict(relationId, query->onConflict);
}
#else
/* always set to false for PG_VERSION_NUM < 90500 */
upsertQuery = false;
#endif
@ -414,6 +415,7 @@ DistributedModifyTask(Query *query)
#if (PG_VERSION_NUM >= 90500)
/*
* RebuildOnConflict rebuilds OnConflictExpr for correct deparsing. The function
* makes WHERE clause elements explicit and filters dropped columns
@ -468,6 +470,8 @@ RebuildOnConflict(Oid relationId, OnConflictExpr *originalOnConflict)
return updatedOnConflict;
}
#endif

39
src/backend/distributed/planner/multi_join_order.c
View File

@ -54,7 +54,8 @@ static bool JoinExprListWalker(Node *node, List **joinList);
static bool ExtractLeftMostRangeTableIndex(Node *node, int *rangeTableIndex);
static List * MergeShardIntervals(List *leftShardIntervalList,
List *rightShardIntervalList, JoinType joinType);
static bool ShardIntervalsMatch(List *leftShardIntervalList, List *rightShardIntervalList);
static bool ShardIntervalsMatch(List *leftShardIntervalList,
List *rightShardIntervalList);
static List * LoadSortedShardIntervalList(Oid relationId);
static List * JoinOrderForTable(TableEntry *firstTable, List *tableEntryList,
List *joinClauseList);
@ -68,31 +69,41 @@ static List * TableEntryListDifference(List *lhsTableList, List *rhsTableList);
static TableEntry * FindTableEntry(List *tableEntryList, uint32 tableId);
/* Local functions forward declarations for join evaluations */
static JoinOrderNode * EvaluateJoinRules(List *joinedTableList, JoinOrderNode *currentJoinNode,
TableEntry *candidateTable, List *candidateShardList,
static JoinOrderNode * EvaluateJoinRules(List *joinedTableList,
JoinOrderNode *currentJoinNode,
TableEntry *candidateTable,
List *candidateShardList,
List *joinClauseList, JoinType joinType);
static List * RangeTableIdList(List *tableList);
static RuleEvalFunction JoinRuleEvalFunction(JoinRuleType ruleType);
static char * JoinRuleName(JoinRuleType ruleType);
static JoinOrderNode * BroadcastJoin(JoinOrderNode *joinNode, TableEntry *candidateTable,
List *candidateShardList, List *applicableJoinClauses,
List *candidateShardList,
List *applicableJoinClauses,
JoinType joinType);
static JoinOrderNode * LocalJoin(JoinOrderNode *joinNode, TableEntry *candidateTable,
List *candidateShardList, List *applicableJoinClauses,
JoinType joinType);
static bool JoinOnColumns(Var *currentPartitioncolumn, Var *candidatePartitionColumn,
List *joinClauseList);
static JoinOrderNode * SinglePartitionJoin(JoinOrderNode *joinNode, TableEntry *candidateTable,
List *candidateShardList, List *applicableJoinClauses,
static JoinOrderNode * SinglePartitionJoin(JoinOrderNode *joinNode,
TableEntry *candidateTable,
List *candidateShardList,
List *applicableJoinClauses,
JoinType joinType);
static JoinOrderNode * DualPartitionJoin(JoinOrderNode *joinNode, TableEntry *candidateTable,
List *candidateShardList, List *applicableJoinClauses,
static JoinOrderNode * DualPartitionJoin(JoinOrderNode *joinNode,
TableEntry *candidateTable,
List *candidateShardList,
List *applicableJoinClauses,
JoinType joinType);
static JoinOrderNode * CartesianProduct(JoinOrderNode *joinNode, TableEntry *candidateTable,
List *candidateShardList, List *applicableJoinClauses,
static JoinOrderNode * CartesianProduct(JoinOrderNode *joinNode,
TableEntry *candidateTable,
List *candidateShardList,
List *applicableJoinClauses,
JoinType joinType);
static JoinOrderNode * MakeJoinOrderNode(TableEntry *tableEntry, JoinRuleType joinRuleType,
Var *partitionColumn, char partitionMethod);
static JoinOrderNode * MakeJoinOrderNode(TableEntry *tableEntry, JoinRuleType
joinRuleType, Var *partitionColumn,
char partitionMethod);
/*
@ -199,7 +210,6 @@ FixedJoinOrderList(FromExpr *fromExpr, List *tableEntryList)
"query"),
errdetail("Shards of relations in outer join queries "
"must have 1-to-1 shard partitioning")));
}
}
else
@ -586,7 +596,8 @@ ShardIntervalsMatch(List *leftShardIntervalList, List *rightShardIntervalList)
nextRightIntervalCell = lnext(rightShardIntervalCell);
if (nextRightIntervalCell != NULL)
{
ShardInterval *nextRightInterval = (ShardInterval *) lfirst(nextRightIntervalCell);
ShardInterval *nextRightInterval =
(ShardInterval *) lfirst(nextRightIntervalCell);
shardIntervalsIntersect = ShardIntervalsOverlap(leftInterval,
nextRightInterval);
if (shardIntervalsIntersect)

15
src/backend/distributed/planner/multi_logical_optimizer.c
View File

@ -91,7 +91,8 @@ static void ParentSetNewChild(MultiNode *parentNode, MultiNode *oldChildNode,
/* Local functions forward declarations for aggregate expressions */
static void ApplyExtendedOpNodes(MultiExtendedOp *originalNode,
MultiExtendedOp *masterNode, MultiExtendedOp *workerNode);
MultiExtendedOp *masterNode,
MultiExtendedOp *workerNode);
static void TransformSubqueryNode(MultiTable *subqueryNode);
static MultiExtendedOp * MasterExtendedOpNode(MultiExtendedOp *originalOpNode);
static Node * MasterAggregateMutator(Node *originalNode, AttrNumber *columnId);
@ -117,7 +118,8 @@ static void ErrorIfUnsupportedArrayAggregate(Aggref *arrayAggregateExpression);
static void ErrorIfUnsupportedAggregateDistinct(Aggref *aggregateExpression,
MultiNode *logicalPlanNode);
static Var * AggregateDistinctColumn(Aggref *aggregateExpression);
static bool TablePartitioningSupportsDistinct(List *tableNodeList, MultiExtendedOp *opNode,
static bool TablePartitioningSupportsDistinct(List *tableNodeList,
MultiExtendedOp *opNode,
Var *distinctColumn);
static bool GroupedByColumn(List *groupClauseList, List *targetList, Var *column);
@ -1980,23 +1982,31 @@ CountDistinctHashFunctionName(Oid argumentType)
switch (argumentType)
{
case INT4OID:
{
hashFunctionName = pstrdup(HLL_HASH_INTEGER_FUNC_NAME);
break;
}
case INT8OID:
{
hashFunctionName = pstrdup(HLL_HASH_BIGINT_FUNC_NAME);
break;
}
case TEXTOID:
case BPCHAROID:
case VARCHAROID:
{
hashFunctionName = pstrdup(HLL_HASH_TEXT_FUNC_NAME);
break;
}
default:
{
hashFunctionName = pstrdup(HLL_HASH_ANY_FUNC_NAME);
break;
}
}
return hashFunctionName;
}
@ -3229,6 +3239,7 @@ SupportedLateralQuery(Query *parentQuery, Query *lateralQuery)
CompositeFieldRecursive(outerQueryExpression, parentQuery);
FieldSelect *localCompositeField =
CompositeFieldRecursive(localQueryExpression, lateralQuery);
/*
* If partition colums are composite fields, add them to list to
* check later if all composite fields are used.

8
src/backend/distributed/planner/multi_logical_planner.c
View File

@ -285,6 +285,7 @@ MultiPlanTree(Query *queryTree)
else
{
bool hasOuterJoin = false;
/*
* We calculate the join order using the list of tables in the query and
* the join clauses between them. Note that this function owns the table
@ -465,6 +466,7 @@ ErrorIfQueryNotSupported(Query *queryTree)
#if (PG_VERSION_NUM >= 90500)
/* HasTablesample returns tree if the query contains tablesample */
static bool
HasTablesample(Query *queryTree)
@ -485,6 +487,8 @@ HasTablesample(Query *queryTree)
return hasTablesample;
}
#endif
@ -529,7 +533,8 @@ HasUnsupportedJoinWalker(Node *node, void *context)
* ErrorIfSubqueryNotSupported checks that we can perform distributed planning for
* the given subquery.
*/
static void ErrorIfSubqueryNotSupported(Query *subqueryTree)
static void
ErrorIfSubqueryNotSupported(Query *subqueryTree)
{
char *errorDetail = NULL;
bool preconditionsSatisfied = true;
@ -587,7 +592,6 @@ HasOuterJoin(Query *queryTree)
static bool
HasOuterJoinWalker(Node *node, void *context)
{
bool hasOuterJoin = false;
if (node == NULL)
{

16
src/backend/distributed/planner/multi_physical_planner.c
View File

@ -155,7 +155,8 @@ static StringInfo DatumArrayString(Datum *datumArray, uint32 datumCount, Oid dat
static Task * CreateBasicTask(uint64 jobId, uint32 taskId, TaskType taskType,
char *queryString);
static void UpdateRangeTableAlias(List *rangeTableList, List *fragmentList);
static Alias * FragmentAlias(RangeTblEntry *rangeTableEntry, RangeTableFragment *fragment);
static Alias * FragmentAlias(RangeTblEntry *rangeTableEntry,
RangeTableFragment *fragment);
static uint64 AnchorShardId(List *fragmentList, uint32 anchorRangeTableId);
static List * PruneSqlTaskDependencies(List *sqlTaskList);
static List * AssignTaskList(List *sqlTaskList);
@ -309,6 +310,7 @@ BuildJobTree(MultiTreeRoot *multiTree)
partitionKey, partitionType,
baseRelationId,
JOIN_MAP_MERGE_JOB);
/* reset depended job list */
loopDependedJobList = NIL;
loopDependedJobList = list_make1(mapMergeJob);
@ -1093,7 +1095,8 @@ QueryJoinTree(MultiNode *multiNode, List *dependedJobList, List **rangeTableList
uint32 columnCount = (uint32) list_length(dependedTargetList);
List *columnNameList = DerivedColumnNameList(columnCount, dependedJob->jobId);
RangeTblEntry *rangeTableEntry = DerivedRangeTableEntry(multiNode, columnNameList,
RangeTblEntry *rangeTableEntry = DerivedRangeTableEntry(multiNode,
columnNameList,
tableIdList);
RangeTblRef *rangeTableRef = makeNode(RangeTblRef);
@ -1864,6 +1867,7 @@ SplitPointObject(ShardInterval **shardIntervalArray, uint32 shardIntervalCount)
return splitPointObject;
}
/* ------------------------------------------------------------
* Functions that relate to building and assigning tasks follow
* ------------------------------------------------------------
@ -4046,6 +4050,7 @@ FragmentAlias(RangeTblEntry *rangeTableEntry, RangeTableFragment *fragment)
return alias;
}
/*
* AnchorShardId walks over each fragment in the given fragment list, finds the
* fragment that corresponds to the given anchor range tableId, and returns this
@ -4984,6 +4989,7 @@ RoundRobinAssignTaskList(List *taskList)
return taskList;
}
/*
* RoundRobinReorder implements the core of the round-robin assignment policy.
* It takes a task and placement list and rotates a copy of the placement list
@ -5116,7 +5122,8 @@ ActiveShardPlacementLists(List *taskList)
List *activeShardPlacementList = ActivePlacementList(shardPlacementList);
/* sort shard placements by their insertion time */
activeShardPlacementList = SortList(activeShardPlacementList, CompareShardPlacements);
activeShardPlacementList = SortList(activeShardPlacementList,
CompareShardPlacements);
shardPlacementLists = lappend(shardPlacementLists, activeShardPlacementList);
}
@ -5257,7 +5264,8 @@ AssignDualHashTaskList(List *taskList)
uint32 replicaIndex = 0;
for (replicaIndex = 0; replicaIndex < ShardReplicationFactor; replicaIndex++)
{
uint32 assignmentOffset = beginningNodeIndex + assignedTaskIndex + replicaIndex;
uint32 assignmentOffset = beginningNodeIndex + assignedTaskIndex +
replicaIndex;
uint32 assignmentIndex = assignmentOffset % workerNodeCount;
WorkerNode *workerNode = list_nth(workerNodeList, assignmentIndex);

11
src/backend/distributed/relay/relay_event_utility.c
View File

@ -205,20 +205,31 @@ RelayEventExtendNames(Node *parseTree, uint64 shardId)
switch (relationNameListLength)
{
case 1:
{
relationNameValue = linitial(relationNameList);
break;
}
case 2:
{
relationNameValue = lsecond(relationNameList);
break;
}
case 3:
{
relationNameValue = lthird(relationNameList);
break;
}
default:
{
ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
errmsg("improper relation name: \"%s\"",
NameListToString(relationNameList))));
break;
}
}
relationName = &(relationNameValue->val.str);
AppendShardIdToName(relationName, shardId);

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

@ -206,9 +206,10 @@ RegisterCitusConfigVariables(void)
DefineCustomBoolVariable(
"citusdb.expire_cached_shards",
gettext_noop("Enables shard cache expiration if a shard's size on disk has changed. "),
gettext_noop("When appending to an existing shard, old data may still be cached on "
"other workers. This configuration entry activates automatic "
gettext_noop("Enables shard cache expiration if a shard's size on disk has "
"changed."),
gettext_noop("When appending to an existing shard, old data may still be cached "
"on other workers. This configuration entry activates automatic "
"expiration, but should not be used with manual updates to shards."),
&ExpireCachedShards,
false,
@ -488,6 +489,7 @@ RegisterCitusConfigVariables(void)
/* warn about config items in the citusdb namespace that are not registered above */
EmitWarningsOnPlaceholders("citusdb");
/* Also warn about citus namespace, as that's a very likely misspelling */
EmitWarningsOnPlaceholders("citus");
}
@ -515,8 +517,10 @@ NormalizeWorkerListPath(void)
{
absoluteFileName = malloc(strlen(DataDir) + strlen(WORKER_LIST_FILENAME) + 2);
if (absoluteFileName == NULL)
{
ereport(FATAL, (errcode(ERRCODE_OUT_OF_MEMORY),
errmsg("out of memory")));
}
sprintf(absoluteFileName, "%s/%s", DataDir, WORKER_LIST_FILENAME);
}
@ -530,6 +534,7 @@ NormalizeWorkerListPath(void)
"environment variable.\n", progname, ConfigFileName)));
}
SetConfigOption("citusdb.worker_list_file", absoluteFileName, PGC_POSTMASTER, PGC_S_OVERRIDE);
SetConfigOption("citusdb.worker_list_file", absoluteFileName, PGC_POSTMASTER,
PGC_S_OVERRIDE);
free(absoluteFileName);
}

5
src/backend/distributed/utils/citus_nodefuncs.c
View File

@ -273,9 +273,13 @@ GetRangeTblKind(RangeTblEntry *rte)
case RTE_JOIN:
case RTE_VALUES:
case RTE_CTE:
{
rteKind = (CitusRTEKind) rte->rtekind;
break;
}
case RTE_FUNCTION:
{
/*
* Extract extra data - correct even if a plain RTE_FUNCTION, not
* an extended one, ExtractRangeTblExtraData handles that case
@ -284,6 +288,7 @@ GetRangeTblKind(RangeTblEntry *rte)
ExtractRangeTblExtraData(rte, &rteKind, NULL, NULL, NULL);
break;
}
}
return rteKind;
}

14
src/backend/distributed/utils/citus_ruleutils.c
View File

@ -447,22 +447,36 @@ pg_get_tablecolumnoptionsdef_string(Oid tableRelationId)
switch (attributeForm->attstorage)
{
case 'p':
{
storageName = "PLAIN";
break;
}
case 'e':
{
storageName = "EXTERNAL";
break;
}
case 'm':
{
storageName = "MAIN";
break;
}
case 'x':
{
storageName = "EXTENDED";
break;
}
default:
{
ereport(ERROR, (errmsg("unrecognized storage type: %c",
attributeForm->attstorage)));
break;
}
}
appendStringInfo(&statement, "ALTER COLUMN %s ",
quote_identifier(attributeName));

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

@ -87,6 +87,7 @@ IsDistributedTable(Oid relationId)
return cacheEntry->isDistributedTable;
}
/*
* LoadShardInterval reads shard metadata for given shardId from pg_dist_shard,
* and converts min/max values in these metadata to their properly typed datum
@ -139,6 +140,7 @@ LoadShardInterval(uint64 shardId)
return shardInterval;
}
/*
* DistributedTableCacheEntry looks up a pg_dist_partition entry for a
* relation.

16
src/backend/distributed/utils/multi_resowner.c
View File

@ -22,7 +22,8 @@
#include "distributed/multi_resowner.h"
typedef struct JobDirectoryEntry {
typedef struct JobDirectoryEntry
{
ResourceOwner owner;
uint64 jobId;
} JobDirectoryEntry;
@ -91,15 +92,17 @@ ResourceOwnerEnlargeJobDirectories(ResourceOwner owner)
if (RegisteredJobDirectories == NULL)
{
newMax = 16;
RegisteredJobDirectories = (JobDirectoryEntry *)
MemoryContextAlloc(TopMemoryContext, newMax * sizeof(JobDirectoryEntry));
RegisteredJobDirectories =
(JobDirectoryEntry *) MemoryContextAlloc(TopMemoryContext,
newMax * sizeof(JobDirectoryEntry));
NumAllocatedJobDirectories = newMax;
}
else if (NumRegisteredJobDirectories + 1 > NumAllocatedJobDirectories)
{
newMax = NumAllocatedJobDirectories * 2;
RegisteredJobDirectories = (JobDirectoryEntry *)
repalloc(RegisteredJobDirectories, newMax * sizeof(JobDirectoryEntry));
RegisteredJobDirectories =
(JobDirectoryEntry *) repalloc(RegisteredJobDirectories,
newMax * sizeof(JobDirectoryEntry));
NumAllocatedJobDirectories = newMax;
}
}
@ -135,7 +138,8 @@ ResourceOwnerForgetJobDirectory(ResourceOwner owner, uint64 jobId)
/* move all later entries one up */
while (jobIndex < lastJobIndex)
{
RegisteredJobDirectories[jobIndex] = RegisteredJobDirectories[jobIndex + 1];
RegisteredJobDirectories[jobIndex] =
RegisteredJobDirectories[jobIndex + 1];
jobIndex++;
}
NumRegisteredJobDirectories = lastJobIndex;

6
src/backend/distributed/worker/task_tracker.c
View File

@ -494,6 +494,7 @@ TrackerDelayLoop(void)
}
}
/* ------------------------------------------------------------
* Signal handling and shared hash initialization functions follow
* ------------------------------------------------------------
@ -579,8 +580,8 @@ TaskTrackerShmemInit(void)
LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
/* allocate struct containing task tracker related shared state */
WorkerTasksSharedState = (WorkerTasksSharedStateData *)
ShmemInitStruct("Worker Task Control",
WorkerTasksSharedState =
(WorkerTasksSharedStateData *) ShmemInitStruct("Worker Task Control",
sizeof(WorkerTasksSharedStateData),
&alreadyInitialized);
@ -607,6 +608,7 @@ TaskTrackerShmemInit(void)
}
}
/* ------------------------------------------------------------
* Task scheduling and management functions follow
* ------------------------------------------------------------

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

@ -331,7 +331,7 @@ UpdateTask(WorkerTask *workerTask, char *taskCallString)
if (taskStatus == TASK_SUCCEEDED || taskStatus == TASK_CANCEL_REQUESTED ||
taskStatus == TASK_CANCELED)
{
; /* nothing to do */
/* nothing to do */
}
else if (taskStatus == TASK_PERMANENTLY_FAILED)
{

11
src/backend/distributed/worker/worker_data_fetch_protocol.c
View File

@ -53,11 +53,14 @@ static void ReceiveResourceCleanup(int32 connectionId, const char *filename,
static void DeleteFile(const char *filename);
static void FetchTableCommon(text *tableName, uint64 remoteTableSize,
ArrayType *nodeNameObject, ArrayType *nodePortObject,
bool (*FetchTableFunction) (const char *, uint32, StringInfo));
bool (*FetchTableFunction)(const char *, uint32,
StringInfo));
static uint64 LocalTableSize(Oid relationId);
static uint64 ExtractShardId(StringInfo tableName);
static bool FetchRegularTable(const char *nodeName, uint32 nodePort, StringInfo tableName);
static bool FetchForeignTable(const char *nodeName, uint32 nodePort, StringInfo tableName);
static bool FetchRegularTable(const char *nodeName, uint32 nodePort,
StringInfo tableName);
static bool FetchForeignTable(const char *nodeName, uint32 nodePort,
StringInfo tableName);
static List * TableDDLCommandList(const char *nodeName, uint32 nodePort,
StringInfo tableName);
static StringInfo ForeignFilePath(const char *nodeName, uint32 nodePort,
@ -309,7 +312,7 @@ ReceiveRegularFile(const char *nodeName, uint32 nodePort,
}
else if (copyStatus == CLIENT_COPY_MORE)
{
; /* remote node will continue to send more data */
/* remote node will continue to send more data */
}
else
{

4
src/backend/distributed/worker/worker_merge_protocol.c
View File

@ -256,8 +256,8 @@ JobSchemaName(uint64 jobId)
*/
#ifdef HAVE_INTTYPES_H
StringInfo jobSchemaName = makeStringInfo();
appendStringInfo(jobSchemaName, "%s%0*"PRIu64,
JOB_SCHEMA_PREFIX, MIN_JOB_DIRNAME_WIDTH, jobId);
appendStringInfo(jobSchemaName, "%s%0*" PRIu64, JOB_SCHEMA_PREFIX,
MIN_JOB_DIRNAME_WIDTH, jobId);
#else
StringInfo jobSchemaName = makeStringInfo();
appendStringInfo(jobSchemaName, "%s%0*llu",

3
src/include/distributed/citus_ruleutils.h
View File

@ -24,7 +24,8 @@ extern char *pg_get_indexclusterdef_string(Oid indexRelationId);
/* Function declarations for version dependent PostgreSQL ruleutils functions */
extern void pg_get_query_def(Query *query, StringInfo buffer);
extern void deparse_shard_query(Query *query, Oid distrelid, int64 shardid, StringInfo buffer);
extern void deparse_shard_query(Query *query, Oid distrelid, int64 shardid, StringInfo
buffer);
extern char * generate_relation_name(Oid relid, List *namespaces);

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

@ -38,7 +38,6 @@ typedef struct ShardInterval
Datum minValue; /* a shard's typed min value datum */
Datum maxValue; /* a shard's typed max value datum */
uint64 shardId;
} ShardInterval;
@ -52,7 +51,6 @@ typedef struct ShardPlacement
RelayFileState shardState;
char *nodeName;
uint32 nodePort;
} ShardPlacement;

9
src/include/distributed/master_protocol.h
View File

@ -49,10 +49,10 @@
#define SHARDID_SEQUENCE_NAME "pg_dist_shardid_seq"
/* Remote call definitions to help with data staging and deletion */
#define WORKER_APPLY_SHARD_DDL_COMMAND "SELECT worker_apply_shard_ddl_command \
("UINT64_FORMAT", %s)"
#define WORKER_APPEND_TABLE_TO_SHARD "SELECT worker_append_table_to_shard \
(%s, %s, %s, %u)"
#define WORKER_APPLY_SHARD_DDL_COMMAND \
"SELECT worker_apply_shard_ddl_command (" UINT64_FORMAT ", %s)"
#define WORKER_APPEND_TABLE_TO_SHARD \
"SELECT worker_append_table_to_shard (%s, %s, %s, %u)"
#define SHARD_MIN_VALUE_QUERY "SELECT min(%s) FROM %s"
#define SHARD_MAX_VALUE_QUERY "SELECT max(%s) FROM %s"
#define SHARD_TABLE_SIZE_QUERY "SELECT pg_table_size('%s')"
@ -67,7 +67,6 @@ typedef enum
SHARD_PLACEMENT_INVALID_FIRST = 0,
SHARD_PLACEMENT_LOCAL_NODE_FIRST = 1,
SHARD_PLACEMENT_ROUND_ROBIN = 2
} ShardPlacementPolicyType;

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

@ -24,6 +24,7 @@
#define INVALID_TASK_ID 0
#if (PG_VERSION_NUM >= 90500)
/* reserved alias name for UPSERTs */
#define UPSERT_ALIAS "citus_table_alias"
#endif

5
src/include/distributed/multi_client_executor.h
View File

@ -29,7 +29,6 @@ typedef enum
CLIENT_CONNECTION_BAD = 1,
CLIENT_CONNECTION_BUSY = 2,
CLIENT_CONNECTION_READY = 3
} ConnectStatus;
@ -40,7 +39,6 @@ typedef enum
CLIENT_RESULT_UNAVAILABLE = 1,
CLIENT_RESULT_BUSY = 2,
CLIENT_RESULT_READY = 3
} ResultStatus;
@ -51,7 +49,6 @@ typedef enum
CLIENT_QUERY_FAILED = 1,
CLIENT_QUERY_DONE = 2,
CLIENT_QUERY_COPY = 3
} QueryStatus;
@ -62,7 +59,6 @@ typedef enum
CLIENT_COPY_MORE = 1,
CLIENT_COPY_FAILED = 2,
CLIENT_COPY_DONE = 3
} CopyStatus;
@ -73,7 +69,6 @@ typedef enum
CLIENT_BATCH_QUERY_FAILED = 1,
CLIENT_BATCH_QUERY_CONTINUE = 2,
CLIENT_BATCH_QUERY_DONE = 3
} BatchQueryStatus;

3
src/include/distributed/multi_join_order.h
View File

@ -40,7 +40,6 @@ typedef enum JoinRuleType
* RuleNameArray.
*/
JOIN_RULE_LAST
} JoinRuleType;
@ -53,7 +52,6 @@ typedef struct TableEntry
{
Oid relationId;
uint32 rangeTableId;
} TableEntry;
@ -72,7 +70,6 @@ typedef struct JoinOrderNode
char partitionMethod;
List *joinClauseList; /* not relevant for the first table */
List *shardIntervalList;
} JoinOrderNode;

9
src/include/distributed/multi_logical_optimizer.h
View File

@ -55,7 +55,6 @@ typedef enum
AGGREGATE_SUM = 4,
AGGREGATE_COUNT = 5,
AGGREGATE_ARRAY_AGG = 6
} AggregateType;
@ -69,7 +68,6 @@ typedef enum
PUSH_DOWN_VALID = 1,
PUSH_DOWN_NOT_VALID = 2,
PUSH_DOWN_SPECIAL_CONDITIONS = 3
} PushDownStatus;
@ -82,7 +80,6 @@ typedef enum
PULL_UP_INVALID_FIRST = 0,
PULL_UP_VALID = 1,
PULL_UP_NOT_VALID = 2
} PullUpStatus;
@ -97,8 +94,10 @@ typedef enum
* Please note that the order of elements in this array is tied to the order of
* values in the preceding AggregateType enum. This order needs to be preserved.
*/
static const char * const AggregateNames[] = { "invalid", "avg", "min", "max",
"sum", "count", "array_agg" };
static const char *const AggregateNames[] = {
"invalid", "avg", "min", "max", "sum",
"count", "array_agg"
};
/* Config variable managed via guc.c */

13
src/include/distributed/multi_logical_planner.h
View File

@ -40,8 +40,8 @@ typedef struct MultiNode
CitusNodeTag type;
struct MultiNode *parentNode;
/* child node(s) are defined in unary and binary nodes */
/* child node(s) are defined in unary and binary nodes */
} MultiNode;
@ -51,7 +51,6 @@ typedef struct MultiUnaryNode
MultiNode node;
struct MultiNode *childNode;
} MultiUnaryNode;
@ -62,7 +61,6 @@ typedef struct MultiBinaryNode
struct MultiNode *leftChildNode;
struct MultiNode *rightChildNode;
} MultiBinaryNode;
@ -73,7 +71,6 @@ typedef struct MultiBinaryNode
typedef struct MultiTreeRoot
{
MultiUnaryNode unaryNode;
} MultiTreeRoot;
@ -91,7 +88,6 @@ typedef struct MultiTable
Alias *alias;
Alias *referenceNames;
Query *subquery; /* this field is only valid for non-relation subquery types */
} MultiTable;
@ -100,7 +96,6 @@ typedef struct MultiProject
{
MultiUnaryNode unaryNode;
List *columnList;
} MultiProject;
@ -112,7 +107,6 @@ typedef struct MultiProject
typedef struct MultiCollect
{
MultiUnaryNode unaryNode;
} MultiCollect;
@ -125,7 +119,6 @@ typedef struct MultiSelect
{
MultiUnaryNode unaryNode;
List *selectClauseList;
} MultiSelect;
@ -140,7 +133,6 @@ typedef struct MultiJoin
List *joinClauseList;
JoinRuleType joinRuleType;
JoinType joinType;
} MultiJoin;
@ -150,7 +142,6 @@ typedef struct MultiPartition
MultiUnaryNode unaryNode;
Var *partitionColumn;
uint32 splitPointTableId;
} MultiPartition;
@ -158,7 +149,6 @@ typedef struct MultiPartition
typedef struct MultiCartesianProduct
{
MultiBinaryNode binaryNode;
} MultiCartesianProduct;
@ -183,7 +173,6 @@ typedef struct MultiExtendedOp
List *sortClauseList;
Node *limitCount;
Node *limitOffset;
} MultiExtendedOp;

12
src/include/distributed/multi_physical_planner.h
View File

@ -40,7 +40,8 @@
(" UINT64_FORMAT ", %d, %s, '%s', %d, %d)"
#define MERGE_FILES_INTO_TABLE_COMMAND "SELECT worker_merge_files_into_table \
(" UINT64_FORMAT ", %d, '%s', '%s')"
#define MERGE_FILES_AND_RUN_QUERY_COMMAND "SELECT worker_merge_files_and_run_query(" UINT64_FORMAT ", %d, '%s', '%s')"
#define MERGE_FILES_AND_RUN_QUERY_COMMAND \
"SELECT worker_merge_files_and_run_query(" UINT64_FORMAT ", %d, '%s', '%s')"
typedef enum CitusRTEKind
@ -62,7 +63,6 @@ typedef enum
PARTITION_INVALID_FIRST = 0,
RANGE_PARTITION_TYPE = 1,
HASH_PARTITION_TYPE = 2
} PartitionType;
@ -77,7 +77,6 @@ typedef enum
MAP_OUTPUT_FETCH_TASK = 5,
MERGE_FETCH_TASK = 6,
MODIFY_TASK = 7
} TaskType;
@ -88,7 +87,6 @@ typedef enum
TASK_ASSIGNMENT_GREEDY = 1,
TASK_ASSIGNMENT_ROUND_ROBIN = 2,
TASK_ASSIGNMENT_FIRST_REPLICA = 3
} TaskAssignmentPolicyType;
@ -99,7 +97,6 @@ typedef enum
JOIN_MAP_MERGE_JOB = 1,
SUBQUERY_MAP_MERGE_JOB = 2,
TOP_LEVEL_WORKER_JOB = 3
} BoundaryNodeJobType;
@ -133,7 +130,6 @@ typedef struct MapMergeJob
ShardInterval **sortedShardIntervalArray; /* only applies to range partitioning */
List *mapTaskList;
List *mergeTaskList;
} MapMergeJob;
@ -164,7 +160,6 @@ typedef struct Task
uint64 shardId; /* only applies to shard fetch tasks */
TaskExecution *taskExecution; /* used by task tracker executor */
bool upsertQuery; /* only applies to modify tasks */
} Task;
@ -177,7 +172,6 @@ typedef struct RangeTableFragment
CitusRTEKind fragmentType;
void *fragmentReference;
uint32 rangeTableId;
} RangeTableFragment;
@ -190,7 +184,6 @@ typedef struct JoinSequenceNode
{
uint32 rangeTableId;
int32 joiningRangeTableId;
} JoinSequenceNode;
@ -203,7 +196,6 @@ typedef struct MultiPlan
Job *workerJob;
Query *masterQuery;
char *masterTableName;
} MultiPlan;

9
src/include/distributed/multi_server_executor.h
View File

@ -60,7 +60,6 @@ typedef enum
EXEC_TASK_TRACKER_FAILED = 14,
EXEC_SOURCE_TASK_TRACKER_RETRY = 15,
EXEC_SOURCE_TASK_TRACKER_FAILED = 16
} TaskExecStatus;
@ -74,7 +73,6 @@ typedef enum
EXEC_TRANSMIT_TRACKER_RETRY = 4,
EXEC_TRANSMIT_TRACKER_FAILED = 5,
EXEC_TRANSMIT_DONE = 6
} TransmitExecStatus;
@ -86,7 +84,6 @@ typedef enum
TRACKER_CONNECT_POLL = 2,
TRACKER_CONNECTED = 3,
TRACKER_CONNECTION_FAILED = 4
} TrackerStatus;
@ -97,7 +94,6 @@ typedef enum
MULTI_EXECUTOR_REAL_TIME = 1,
MULTI_EXECUTOR_TASK_TRACKER = 2,
MULTI_EXECUTOR_ROUTER = 3
} MultiExecutorType;
@ -107,7 +103,6 @@ typedef enum
CONNECT_ACTION_NONE = 0,
CONNECT_ACTION_OPENED = 1,
CONNECT_ACTION_CLOSED = 2
} ConnectAction;
@ -132,7 +127,6 @@ struct TaskExecution
uint32 querySourceNodeIndex; /* only applies to map fetch tasks */
int32 dataFetchTaskIndex;
uint32 failureCount;
};
@ -147,7 +141,6 @@ typedef struct TrackerTaskState
uint32 taskId;
TaskStatus status;
StringInfo taskAssignmentQuery;
} TrackerTaskState;
@ -171,7 +164,6 @@ typedef struct TaskTracker
int32 currentTaskIndex;
bool connectionBusy;
TrackerTaskState *connectionBusyOnTask;
} TaskTracker;
@ -184,7 +176,6 @@ typedef struct WorkerNodeState
uint32 workerPort;
char workerName[WORKER_LENGTH];
uint32 openConnectionCount;
} WorkerNodeState;

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

@ -35,7 +35,6 @@ typedef enum
FILE_CACHED = 2,
FILE_INACTIVE = 3,
FILE_TO_DELETE = 4
} RelayFileState;

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

@ -29,6 +29,7 @@ typedef enum AdvisoryLocktagClass
/* values defined in postgres' lockfuncs.c */
ADV_LOCKTAG_CLASS_INT64 = 1,
ADV_LOCKTAG_CLASS_INT32 = 2,
/* CitusDB lock types */
ADV_LOCKTAG_CLASS_CITUS_SHARD_METADATA = 4,
ADV_LOCKTAG_CLASS_CITUS_SHARD = 5,

3
src/include/distributed/task_tracker.h
View File

@ -63,7 +63,6 @@ typedef enum
* TASK_STATUS_LAST, should never have their numbers changed.
*/
TASK_STATUS_LAST
} TaskStatus;
@ -85,7 +84,6 @@ typedef struct WorkerTask
char databaseName[NAMEDATALEN]; /* name to use for local backend connection */
int32 connectionId; /* connection id to local backend */
uint32 failureCount; /* number of task failures */
} WorkerTask;
@ -97,6 +95,7 @@ typedef struct WorkerTasksSharedStateData
{
/* Hash table shared by the task tracker and task tracker protocol functions */
HTAB *taskHash;
/* Lock protecting workerNodesHash */
LWLock *taskHashLock;
} WorkerTasksSharedStateData;

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

@ -48,7 +48,6 @@ typedef struct WorkerNode
char workerRack[WORKER_LENGTH]; /* node's network location */
bool inWorkerFile; /* is node in current membership file? */
} WorkerNode;

3
src/include/distributed/worker_protocol.h
View File

@ -65,7 +65,6 @@ typedef struct RangePartitionContext
FmgrInfo *comparisonFunction;
Datum *splitPointArray;
int32 splitPointCount;
} RangePartitionContext;
@ -77,7 +76,6 @@ typedef struct HashPartitionContext
{
FmgrInfo *hashFunction;
uint32 partitionCount;
} HashPartitionContext;
@ -114,7 +112,6 @@ typedef struct FileOutputStream
File fileDescriptor;
StringInfo fileBuffer;
StringInfo filePath;
} FileOutputStream;