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

Real-time executor marks multi shard relation accesses before opening connections

pull/2220/head
Onder Kalaci 2018-06-20 15:23:03 +03:00
parent 2890154420
commit 8ccb8b679e
12 changed files with 694 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -106,8 +106,15 @@ MultiRealTimeExecute(Job *job)
foreach(taskCell, taskList)
{
Task *task = (Task *) lfirst(taskCell);
TaskExecution *taskExecution = NULL;
TaskExecution *taskExecution = InitTaskExecution(task, EXEC_TASK_CONNECT_START);
/* keep track of multi shard accesses before opening the connections */
if (MultiShardConnectionType == PARALLEL_CONNECTION)
{
RecordRelationParallelSelectAccessForTask(task);
}
taskExecution = InitTaskExecution(task, EXEC_TASK_CONNECT_START);
taskExecutionList = lappend(taskExecutionList, taskExecution);
}
@ -149,13 +156,6 @@ MultiRealTimeExecute(Job *job)
/* update the connection counter for throttling */
UpdateConnectionCounter(workerNodeState, connectAction);
/* keep track of multi shard select accesses */
if (MultiShardConnectionType == PARALLEL_CONNECTION &&
connectAction == CONNECT_ACTION_OPENED)
{
RecordRelationMultiShardSelectAccessForTask(task);
}
/*
* If this task failed, we need to iterate over task executions, and
* manually clean out their client-side resources. Hence, we record

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

@ -1333,6 +1333,32 @@ ExecuteModifyTasks(List *taskList, bool expectResults, ParamListInfo paramListIn
CoordinatedTransactionUse2PC();
}
/*
* With a similar rationale as above, where we expect all tasks to operate on
* the same relations, we prefer to record relation accesses for the first
* task only.
*/
if (firstTask->taskType == MODIFY_TASK)
{
RecordRelationParallelModifyAccessForTask(firstTask);
/*
* We prefer to mark with SELECT access as well because for multi shard
* modification queries, the placement access list is always marked with both
* DML and SELECT accesses.
*/
RecordRelationParallelSelectAccessForTask(firstTask);
}
else if (firstTask->taskType == DDL_TASK &&
PartitionMethod(firstShardInterval->relationId) != DISTRIBUTE_BY_NONE)
{
/*
* Even single task DDLs hit here, so we'd prefer
* not to record for reference tables.
*/
RecordRelationParallelDDLAccessForTask(firstTask);
}
if (firstTask->taskType == DDL_TASK || firstTask->taskType == VACUUM_ANALYZE_TASK)
{
connectionFlags = FOR_DDL;
@ -1370,20 +1396,6 @@ ExecuteModifyTasks(List *taskList, bool expectResults, ParamListInfo paramListIn
&shardConnectionsFound);
connectionList = shardConnections->connectionList;
if (task->taskType == MODIFY_TASK)
{
RecordRelationMultiShardModifyAccessForTask(task);
}
else if (task->taskType == DDL_TASK &&
PartitionMethod(RelationIdForShard(shardId)) != DISTRIBUTE_BY_NONE)
{
/*
* Even single task DDLs hit here, so we'd prefer
* not to record for reference tables.
*/
RecordRelationMultiShardDDLAccessForTask(task);
}
if (placementIndex >= list_length(connectionList))
{
/* no more active placements for this task */

13
src/backend/distributed/master/master_create_shards.c
View File

@ -34,7 +34,6 @@
#include "distributed/pg_dist_partition.h"
#include "distributed/pg_dist_shard.h"
#include "distributed/reference_table_utils.h"
#include "distributed/relation_access_tracking.h"
#include "distributed/resource_lock.h"
#include "distributed/shardinterval_utils.h"
#include "distributed/transaction_management.h"
@ -337,18 +336,6 @@ CreateColocatedShards(Oid targetRelationId, Oid sourceRelationId, bool
CreateShardsOnWorkers(targetRelationId, insertedShardPlacements,
useExclusiveConnections, colocatedShard);
if (useExclusiveConnections)
{
RecordParallelDDLAccess(targetRelationId);
/* we should mark the parent as well */
if (PartitionTable(targetRelationId))
{
Oid parentRelationId = PartitionParentOid(targetRelationId);
RecordParallelDDLAccess(parentRelationId);
}
}
}

16
src/backend/distributed/master/master_stage_protocol.c
View File

@ -42,6 +42,7 @@
#include "distributed/pg_dist_partition.h"
#include "distributed/pg_dist_shard.h"
#include "distributed/placement_connection.h"
#include "distributed/relation_access_tracking.h"
#include "distributed/remote_commands.h"
#include "distributed/resource_lock.h"
#include "distributed/transaction_management.h"
@ -520,6 +521,19 @@ CreateShardsOnWorkers(Oid distributedRelationId, List *shardPlacements,
CoordinatedTransactionUse2PC();
}
/* mark parallel relation accesses before opening connections */
if (ShouldRecordRelationAccess() && useExclusiveConnection)
{
RecordParallelDDLAccess(distributedRelationId);
/* we should mark the parent as well */
if (alterTableAttachPartitionCommand != NULL)
{
Oid parentRelationId = PartitionParentOid(distributedRelationId);
RecordParallelDDLAccess(parentRelationId);
}
}
foreach(shardPlacementCell, shardPlacements)
{
ShardPlacement *shardPlacement = (ShardPlacement *) lfirst(shardPlacementCell);
@ -538,7 +552,7 @@ CreateShardsOnWorkers(Oid distributedRelationId, List *shardPlacements,
* with DDL. This is only important for parallel relation access in transaction
* blocks, thus check useExclusiveConnection and transaction block as well.
*/
if ((IsTransactionBlock() && useExclusiveConnection) &&
if ((ShouldRecordRelationAccess() && useExclusiveConnection) &&
alterTableAttachPartitionCommand != NULL)
{
RelationShard *parentRelationShard = CitusMakeNode(RelationShard);

19
src/backend/distributed/transaction/multi_shard_transaction.c
View File

@ -91,7 +91,6 @@ OpenTransactionsForAllTasks(List *taskList, int connectionFlags)
ShardPlacement *shardPlacement = (ShardPlacement *) lfirst(placementCell);
ShardPlacementAccess placementModification;
List *placementAccessList = NIL;
List *placementSelectList = NIL;
MultiConnection *connection = NULL;
WorkerNode *workerNode = FindWorkerNode(shardPlacement->nodeName,
@ -111,21 +110,25 @@ OpenTransactionsForAllTasks(List *taskList, int connectionFlags)
if (accessType == PLACEMENT_ACCESS_DDL)
{
List *placementDDLList = BuildPlacementDDLList(shardPlacement->groupId,
task->relationShardList);
/*
* All relations appearing inter-shard DDL commands should be marked
* with DDL access.
*/
placementSelectList = BuildPlacementDDLList(shardPlacement->groupId,
task->relationShardList);
placementAccessList = list_concat(placementAccessList, placementDDLList);
}
else
{
/* add additional placement accesses for subselects (e.g. INSERT .. SELECT) */
placementSelectList = BuildPlacementSelectList(shardPlacement->groupId,
task->relationShardList);
}
List *placementSelectList =
BuildPlacementSelectList(shardPlacement->groupId,
task->relationShardList);
placementAccessList = list_concat(placementAccessList, placementSelectList);
/* add additional placement accesses for subselects (e.g. INSERT .. SELECT) */
placementAccessList =
list_concat(placementAccessList, placementSelectList);
}
/*
* Find a connection that sees preceding writes and cannot self-deadlock,

201
src/backend/distributed/transaction/relation_access_tracking.c
View File

@ -18,8 +18,10 @@
#include "miscadmin.h"
#include "access/xact.h"
#include "distributed/colocation_utils.h"
#include "distributed/hash_helpers.h"
#include "distributed/multi_join_order.h"
#include "distributed/multi_partitioning_utils.h"
#include "distributed/metadata_cache.h"
#include "distributed/relation_access_tracking.h"
#include "utils/hsearch.h"
@ -63,10 +65,15 @@ typedef struct RelationAccessHashEntry
static HTAB *RelationAccessHash;
static void RecordPlacementAccessToCache(Oid relationId,
ShardPlacementAccessType accessType);
static RelationAccessMode GetRelationAccessMode(Oid relationId,
ShardPlacementAccessType accessType);
static void RecordParallelRelationAccess(Oid relationId, ShardPlacementAccessType
placementAccess);
static void RecordParallelRelationAccessToCache(Oid relationId,
ShardPlacementAccessType placementAccess);
/*
@ -83,7 +90,7 @@ ResetRelationAccessHash()
* Allocate RelationAccessHash.
*/
void
AllocateRelationAccessHash()
AllocateRelationAccessHash(void)
{
HASHCTL info;
uint32 hashFlags = 0;
@ -108,8 +115,81 @@ void
AssociatePlacementAccessWithRelation(ShardPlacement *placement,
ShardPlacementAccessType accessType)
{
uint64 shardId = placement->shardId;
Oid relationId = RelationIdForShard(shardId);
uint64 shardId = INVALID_SHARD_ID;
Oid relationId = InvalidOid;
if (!ShouldRecordRelationAccess())
{
return;
}
shardId = placement->shardId;
relationId = RelationIdForShard(shardId);
/*
* If a relation is partitioned, record accesses to all of its partitions as well.
* We prefer to use PartitionedTableNoLock() because at this point the necessary
* locks on the relation has already been acquired.
*/
if (PartitionedTableNoLock(relationId))
{
List *partitionList = PartitionList(relationId);
ShardInterval *shardInterval = LoadShardInterval(shardId);
int shardIndex = shardInterval->shardIndex;
ListCell *partitionCell = NULL;
foreach(partitionCell, partitionList)
{
Oid partitionOid = lfirst_oid(partitionCell);
uint64 partitionShardId = INVALID_SHARD_ID;
List *partitionShardPlacementList = NIL;
ShardPlacement *partitionInitialShardPlacement = NULL;
/*
* During create_distributed_table, the partitions may not
* have been created yet and so there are no placements yet.
* We're already going to register them when we distribute
* the partitions.
*/
if (!IsDistributedTable(partitionOid))
{
continue;
}
partitionShardId =
ColocatedShardIdInRelation(partitionOid, shardIndex);
partitionShardPlacementList = ShardPlacementList(partitionShardId);
partitionInitialShardPlacement =
(ShardPlacement *) linitial(partitionShardPlacementList);
/*
* Recursively record all relation accesses of its partitions. Note that
* we prefer to recursively call AssociatePlacementAccessWithRelation()
* to support multi-level partitioned tables.
*/
AssociatePlacementAccessWithRelation(partitionInitialShardPlacement,
accessType);
}
}
else if (PartitionTableNoLock(relationId))
{
Oid parentOid = PartitionParentOid(relationId);
/* only record the parent */
RecordPlacementAccessToCache(parentOid, accessType);
}
RecordPlacementAccessToCache(relationId, accessType);
}
/*
* RecordPlacementAccessToCache is a utility function which saves the given
* relation id's access to the RelationAccessHash.
*/
static void
RecordPlacementAccessToCache(Oid relationId, ShardPlacementAccessType accessType)
{
RelationAccessHashKey hashKey;
RelationAccessHashEntry *hashEntry;
bool found = false;
@ -128,18 +208,18 @@ AssociatePlacementAccessWithRelation(ShardPlacement *placement,
/*
* RecordRelationMultiShardSelectAccessForTask goes over all the relations
* RecordRelationParallelSelectAccessForTask goes over all the relations
* in the relationShardList and records the select access per each table.
*/
void
RecordRelationMultiShardSelectAccessForTask(Task *task)
RecordRelationParallelSelectAccessForTask(Task *task)
{
List *relationShardList = NIL;
ListCell *relationShardCell = NULL;
Oid lastRelationId = InvalidOid;
/* no point in recoding accesses in non-transaction blocks, skip the loop */
if (!IsTransactionBlock())
if (!ShouldRecordRelationAccess())
{
return;
}
@ -168,20 +248,20 @@ RecordRelationMultiShardSelectAccessForTask(Task *task)
/*
* RecordRelationMultiShardModifyAccessForTask gets a task and records
* RecordRelationParallelModifyAccessForTask gets a task and records
* the accesses. Note that the target relation is recorded with modify access
* where as the subqueries inside the modify query is recorded with select
* access.
*/
void
RecordRelationMultiShardModifyAccessForTask(Task *task)
RecordRelationParallelModifyAccessForTask(Task *task)
{
List *relationShardList = NULL;
ListCell *relationShardCell = NULL;
Oid lastRelationId = InvalidOid;
/* no point in recoding accesses in non-transaction blocks, skip the loop */
if (!IsTransactionBlock())
if (!ShouldRecordRelationAccess())
{
return;
}
@ -215,22 +295,34 @@ RecordRelationMultiShardModifyAccessForTask(Task *task)
/*
* RecordRelationMultiShardDDLAccessForTask marks all the relationShards
* RecordRelationParallelDDLAccessForTask marks all the relationShards
* with parallel DDL access if exists. That case is valid for inter-shard
* DDL commands such as foreign key creation. The function also records
* the relation that anchorShardId belongs to.
*/
void
RecordRelationMultiShardDDLAccessForTask(Task *task)
RecordRelationParallelDDLAccessForTask(Task *task)
{
List *relationShardList = task->relationShardList;
ListCell *relationShardCell = NULL;
Oid lastRelationId = InvalidOid;
foreach(relationShardCell, relationShardList)
{
RelationShard *relationShard = (RelationShard *) lfirst(relationShardCell);
Oid currentRelationId = relationShard->relationId;
RecordParallelDDLAccess(relationShard->relationId);
/*
* An optimization, skip going to hash table if we've already
* recorded the relation.
*/
if (currentRelationId == lastRelationId)
{
continue;
}
RecordParallelDDLAccess(currentRelationId);
lastRelationId = currentRelationId;
}
RecordParallelDDLAccess(RelationIdForShard(task->anchorShardId));
@ -271,22 +363,59 @@ RecordParallelDDLAccess(Oid relationId)
* RecordParallelRelationAccess records the relation access mode as parallel
* for the given access type (e.g., select, dml or ddl) in the RelationAccessHash.
*
* The function becomes no-op for non-transaction blocks
* The function also takes partitions and partitioned tables into account.
*/
static void
RecordParallelRelationAccess(Oid relationId, ShardPlacementAccessType placementAccess)
{
RelationAccessHashKey hashKey;
RelationAccessHashEntry *hashEntry;
bool found = false;
int multiShardAccessBit = 0;
/* no point in recoding accesses in non-transaction blocks */
if (!IsTransactionBlock())
if (!ShouldRecordRelationAccess())
{
return;
}
/*
* If a relation is partitioned, record accesses to all of its partitions as well.
* We prefer to use PartitionedTableNoLock() because at this point the necessary
* locks on the relation has already been acquired.
*/
if (PartitionedTableNoLock(relationId))
{
List *partitionList = PartitionList(relationId);
ListCell *partitionCell = NULL;
foreach(partitionCell, partitionList)
{
Oid partitionOid = lfirst_oid(partitionCell);
/* recursively record all relation accesses of its partitions */
RecordParallelRelationAccess(partitionOid, placementAccess);
}
}
else if (PartitionTableNoLock(relationId))
{
Oid parentOid = PartitionParentOid(relationId);
/* only record the parent */
RecordParallelRelationAccessToCache(parentOid, placementAccess);
}
RecordParallelRelationAccessToCache(relationId, placementAccess);
}
/*
* RecordParallelRelationAccessToCache is a utility function which saves the given
* relation id's access to the RelationAccessHash.
*/
static void
RecordParallelRelationAccessToCache(Oid relationId,
ShardPlacementAccessType placementAccess)
{
RelationAccessHashKey hashKey;
RelationAccessHashEntry *hashEntry;
bool found = false;
int parallelRelationAccessBit = 0;
hashKey.relationId = relationId;
hashEntry = hash_search(RelationAccessHash, &hashKey, HASH_ENTER, &found);
@ -299,8 +428,8 @@ RecordParallelRelationAccess(Oid relationId, ShardPlacementAccessType placementA
hashEntry->relationAccessMode |= (1 << (placementAccess));
/* set the bit representing access mode */
multiShardAccessBit = placementAccess + PARALLEL_MODE_FLAG_OFFSET;
hashEntry->relationAccessMode |= (1 << multiShardAccessBit);
parallelRelationAccessBit = placementAccess + PARALLEL_MODE_FLAG_OFFSET;
hashEntry->relationAccessMode |= (1 << parallelRelationAccessBit);
}
@ -345,10 +474,10 @@ GetRelationAccessMode(Oid relationId, ShardPlacementAccessType accessType)
RelationAccessHashEntry *hashEntry;
int relationAcessMode = 0;
bool found = false;
int multiShardAccessBit = accessType + PARALLEL_MODE_FLAG_OFFSET;
int parallelRelationAccessBit = accessType + PARALLEL_MODE_FLAG_OFFSET;
/* no point in getting the mode when not inside a transaction block */
if (!IsTransactionBlock())
if (!ShouldRecordRelationAccess())
{
return RELATION_NOT_ACCESSED;
}
@ -370,7 +499,7 @@ GetRelationAccessMode(Oid relationId, ShardPlacementAccessType accessType)
return RELATION_NOT_ACCESSED;
}
if (relationAcessMode & (1 << multiShardAccessBit))
if (relationAcessMode & (1 << parallelRelationAccessBit))
{
return RELATION_PARALLEL_ACCESSED;
}
@ -379,3 +508,25 @@ GetRelationAccessMode(Oid relationId, ShardPlacementAccessType accessType)
return RELATION_SEQUENTIAL_ACCESSED;
}
}
/*
* ShouldRecordRelationAccess returns true when we should keep track
* of the relation accesses.
*
* In many cases, we'd only need IsTransactionBlock(), however, for some cases such as
* CTEs, where Citus uses the same connections accross multiple queries, we should
* still record the relation accesses even not inside an explicit transaction block.
* Thus, keeping track of the relation accesses inside coordinated transactions is
* also required.
*/
bool
ShouldRecordRelationAccess()
{
if (IsTransactionBlock() || InCoordinatedTransaction())
{
return true;
}
return false;
}

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

@ -57,6 +57,37 @@ PartitionedTable(Oid relationId)
}
/*
* Returns true if the given relation is a partitioned table. The function
* doesn't acquire any locks on the input relation, thus the caller is
* reponsible for holding the appropriate locks.
*/
bool
PartitionedTableNoLock(Oid relationId)
{
Relation rel = try_relation_open(relationId, NoLock);
bool partitionedTable = false;
/* don't error out for tables that are dropped */
if (rel == NULL)
{
return false;
}
#if (PG_VERSION_NUM >= 100000)
if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
partitionedTable = true;
}
#endif
/* keep the lock */
heap_close(rel, NoLock);
return partitionedTable;
}
/*
* Returns true if the given relation is a partition.
*/
@ -77,6 +108,34 @@ PartitionTable(Oid relationId)
}
/*
* Returns true if the given relation is a partition. The function
* doesn't acquire any locks on the input relation, thus the caller is
* reponsible for holding the appropriate locks.
*/
bool
PartitionTableNoLock(Oid relationId)
{
Relation rel = try_relation_open(relationId, NoLock);
bool partitionTable = false;
/* don't error out for tables that are dropped */
if (rel == NULL)
{
return false;
}
#if (PG_VERSION_NUM >= 100000)
partitionTable = rel->rd_rel->relispartition;
#endif
/* keep the lock */
heap_close(rel, NoLock);
return partitionTable;
}
/*
* IsChildTable returns true if the table is inherited. Note that
* partition tables inherites by default. However, this function

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

@ -12,7 +12,9 @@
extern bool PartitionedTable(Oid relationId);
extern bool PartitionedTableNoLock(Oid relationId);
extern bool PartitionTable(Oid relationId);
extern bool PartitionTableNoLock(Oid relationId);
extern bool IsChildTable(Oid relationId);
extern bool IsParentTable(Oid relationId);
extern Oid PartitionParentOid(Oid partitionOid);

7
src/include/distributed/relation_access_tracking.h
View File

@ -28,14 +28,15 @@ extern void ResetRelationAccessHash(void);
extern void AssociatePlacementAccessWithRelation(ShardPlacement *placement,
ShardPlacementAccessType accessType);
extern void RecordParallelSelectAccess(Oid relationId);
extern void RecordRelationMultiShardSelectAccessForTask(Task *task);
extern void RecordRelationMultiShardModifyAccessForTask(Task *task);
extern void RecordRelationParallelSelectAccessForTask(Task *task);
extern void RecordRelationParallelModifyAccessForTask(Task *task);
extern void RecordParallelModifyAccess(Oid relationId);
extern void RecordParallelDDLAccess(Oid relationId);
extern void RecordRelationMultiShardDDLAccessForTask(Task *task);
extern void RecordRelationParallelDDLAccessForTask(Task *task);
extern RelationAccessMode GetRelationDDLAccessMode(Oid relationId);
extern RelationAccessMode GetRelationDMLAccessMode(Oid relationId);
extern RelationAccessMode GetRelationSelectAccessMode(Oid relationId);
extern bool ShouldRecordRelationAccess(void);
#endif /* RELATION_ACCESS_TRACKING_H_ */

208
src/test/regress/expected/relation_access_tracking.out
View File

@ -338,9 +338,9 @@ ROLLBACK;
BEGIN;
UPDATE table_1 SET value = 15;
SELECT * FROM relation_acesses WHERE table_name = 'table_1';
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | parallel_access | not_accessed
(1 row)
SET LOCAL citus.multi_shard_modify_mode = 'sequential';
@ -348,7 +348,7 @@ BEGIN;
SELECT * FROM relation_acesses WHERE table_name IN ('table_1', 'table_2') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-------------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_1 | parallel_access | parallel_access | not_accessed
table_2 | sequential_access | sequential_access | not_accessed
(2 rows)
@ -497,12 +497,12 @@ BEGIN;
) as foo
) AND value IN (SELECT key FROM table_4);
SELECT * FROM relation_acesses WHERE table_name IN ('table_1', 'table_2', 'table_3', 'table_4') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | parallel_access | not_accessed | not_accessed
table_2 | parallel_access | not_accessed | not_accessed
table_3 | sequential_access | parallel_access | not_accessed
table_4 | parallel_access | not_accessed | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | not_accessed | not_accessed
table_2 | parallel_access | not_accessed | not_accessed
table_3 | parallel_access | parallel_access | not_accessed
table_4 | parallel_access | not_accessed | not_accessed
(4 rows)
ROLLBACK;
@ -669,7 +669,7 @@ BEGIN;
partitioning_test_2009 | not_accessed | not_accessed | parallel_access
(2 rows)
ROLLBACK;
COMMIT;
-- Adding partition tables via ATTACH PARTITION on distributed tables should have DDL access the partitioned table as well
CREATE TABLE partitioning_test_2010 AS SELECT * FROM partitioning_test;
SELECT create_distributed_table('partitioning_test_2010', 'id');
@ -687,6 +687,176 @@ BEGIN;
partitioning_test_2010 | not_accessed | not_accessed | parallel_access
(2 rows)
COMMIT;
-- reading from partitioned table marks all of its partitions
BEGIN;
SELECT count(*) FROM partitioning_test;
count
-------
0
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-----------------+--------------+--------------
partitioning_test | parallel_access | not_accessed | not_accessed
partitioning_test_2009 | parallel_access | not_accessed | not_accessed
partitioning_test_2010 | parallel_access | not_accessed | not_accessed
(3 rows)
COMMIT;
-- reading from partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
SELECT count(*) FROM partitioning_test;
count
-------
0
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-------------------+--------------+--------------
partitioning_test | sequential_access | not_accessed | not_accessed
partitioning_test_2009 | sequential_access | not_accessed | not_accessed
partitioning_test_2010 | sequential_access | not_accessed | not_accessed
(3 rows)
COMMIT;
-- updating partitioned table marks all of its partitions
BEGIN;
UPDATE partitioning_test SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-----------------+-----------------+--------------
partitioning_test | parallel_access | parallel_access | not_accessed
partitioning_test_2009 | parallel_access | parallel_access | not_accessed
partitioning_test_2010 | parallel_access | parallel_access | not_accessed
(3 rows)
COMMIT;
-- updating partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
UPDATE partitioning_test SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-------------------+-------------------+--------------
partitioning_test | sequential_access | sequential_access | not_accessed
partitioning_test_2009 | sequential_access | sequential_access | not_accessed
partitioning_test_2010 | sequential_access | sequential_access | not_accessed
(3 rows)
COMMIT;
-- DDLs on partitioned table marks all of its partitions
BEGIN;
ALTER TABLE partitioning_test ADD COLUMN X INT;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+---------------+--------------+-----------------
partitioning_test | not_accessed | not_accessed | parallel_access
partitioning_test_2009 | not_accessed | not_accessed | parallel_access
partitioning_test_2010 | not_accessed | not_accessed | parallel_access
(3 rows)
ROLLBACK;
-- DDLs on partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
ALTER TABLE partitioning_test ADD COLUMN X INT;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+---------------+--------------+-------------------
partitioning_test | not_accessed | not_accessed | sequential_access
partitioning_test_2009 | not_accessed | not_accessed | sequential_access
partitioning_test_2010 | not_accessed | not_accessed | sequential_access
(3 rows)
ROLLBACK;
-- reading from partition table marks its parent
BEGIN;
SELECT count(*) FROM partitioning_test_2009;
count
-------
0
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-----------------+--------------+--------------
partitioning_test | parallel_access | not_accessed | not_accessed
partitioning_test_2009 | parallel_access | not_accessed | not_accessed
partitioning_test_2010 | not_accessed | not_accessed | not_accessed
(3 rows)
COMMIT;
-- rreading from partition table marks its parent with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
SELECT count(*) FROM partitioning_test_2009;
count
-------
0
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-------------------+--------------+--------------
partitioning_test | sequential_access | not_accessed | not_accessed
partitioning_test_2009 | sequential_access | not_accessed | not_accessed
partitioning_test_2010 | not_accessed | not_accessed | not_accessed
(3 rows)
COMMIT;
-- updating from partition table marks its parent
BEGIN;
UPDATE partitioning_test_2009 SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-----------------+-----------------+--------------
partitioning_test | parallel_access | parallel_access | not_accessed
partitioning_test_2009 | parallel_access | parallel_access | not_accessed
partitioning_test_2010 | not_accessed | not_accessed | not_accessed
(3 rows)
COMMIT;
-- updating from partition table marks its parent sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
UPDATE partitioning_test_2009 SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+-------------------+-------------------+--------------
partitioning_test | sequential_access | sequential_access | not_accessed
partitioning_test_2009 | sequential_access | sequential_access | not_accessed
partitioning_test_2010 | not_accessed | not_accessed | not_accessed
(3 rows)
COMMIT;
-- DDLs on partition table marks its parent
BEGIN;
CREATE INDEX i1000000 ON partitioning_test_2009 (id);
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+---------------+--------------+-----------------
partitioning_test | not_accessed | not_accessed | parallel_access
partitioning_test_2009 | not_accessed | not_accessed | parallel_access
partitioning_test_2010 | not_accessed | not_accessed | not_accessed
(3 rows)
ROLLBACK;
-- DDLs on partition table marks its parent in sequential mode
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
CREATE INDEX i1000000 ON partitioning_test_2009 (id);
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------------------+---------------+--------------+-------------------
partitioning_test | not_accessed | not_accessed | sequential_access
partitioning_test_2009 | not_accessed | not_accessed | sequential_access
partitioning_test_2010 | not_accessed | not_accessed | not_accessed
(3 rows)
ROLLBACK;
-- TRUNCATE CASCADE works fine
ALTER TABLE table_2 ADD CONSTRAINT table_2_u FOREIGN KEY (key) REFERENCES table_1(key);
@ -765,9 +935,9 @@ BEGIN;
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('table_1') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | parallel_access | not_accessed
(1 row)
ROLLBACK;
@ -782,9 +952,9 @@ BEGIN;
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('table_1') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | parallel_access | not_accessed
(1 row)
ROLLBACK;
@ -810,7 +980,7 @@ NOTICE: Copying data from local table...
COMMIT;
SET search_path TO 'public';
DROP SCHEMA access_tracking CASCADE;
NOTICE: drop cascades to 15 other objects
NOTICE: drop cascades to 13 other objects
DETAIL: drop cascades to function access_tracking.relation_select_access_mode(oid)
drop cascades to function access_tracking.relation_dml_access_mode(oid)
drop cascades to function access_tracking.relation_ddl_access_mode(oid)
@ -823,6 +993,4 @@ drop cascades to table access_tracking.table_5
drop cascades to table access_tracking.table_6
drop cascades to table access_tracking.table_7
drop cascades to table access_tracking.partitioning_test
drop cascades to table access_tracking.partitioning_test_2009
drop cascades to table access_tracking.partitioning_test_2010
drop cascades to table access_tracking.table_3

140
src/test/regress/expected/relation_access_tracking_0.out
View File

@ -338,9 +338,9 @@ ROLLBACK;
BEGIN;
UPDATE table_1 SET value = 15;
SELECT * FROM relation_acesses WHERE table_name = 'table_1';
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | parallel_access | not_accessed
(1 row)
SET LOCAL citus.multi_shard_modify_mode = 'sequential';
@ -348,7 +348,7 @@ BEGIN;
SELECT * FROM relation_acesses WHERE table_name IN ('table_1', 'table_2') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-------------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_1 | parallel_access | parallel_access | not_accessed
table_2 | sequential_access | sequential_access | not_accessed
(2 rows)
@ -497,12 +497,12 @@ BEGIN;
) as foo
) AND value IN (SELECT key FROM table_4);
SELECT * FROM relation_acesses WHERE table_name IN ('table_1', 'table_2', 'table_3', 'table_4') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | parallel_access | not_accessed | not_accessed
table_2 | parallel_access | not_accessed | not_accessed
table_3 | sequential_access | parallel_access | not_accessed
table_4 | parallel_access | not_accessed | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | not_accessed | not_accessed
table_2 | parallel_access | not_accessed | not_accessed
table_3 | parallel_access | parallel_access | not_accessed
table_4 | parallel_access | not_accessed | not_accessed
(4 rows)
ROLLBACK;
@ -669,7 +669,7 @@ LINE 1: ALTER TABLE partitioning_test ATTACH PARTITION partitioning_...
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
ROLLBACK;
COMMIT;
-- Adding partition tables via ATTACH PARTITION on distributed tables should have DDL access the partitioned table as well
CREATE TABLE partitioning_test_2010 AS SELECT * FROM partitioning_test;
ERROR: relation "partitioning_test" does not exist
@ -686,6 +686,112 @@ LINE 1: ALTER TABLE partitioning_test ATTACH PARTITION partitioning_...
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- reading from partitioned table marks all of its partitions
BEGIN;
SELECT count(*) FROM partitioning_test;
ERROR: relation "partitioning_test" does not exist
LINE 1: SELECT count(*) FROM partitioning_test;
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- reading from partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
SELECT count(*) FROM partitioning_test;
ERROR: relation "partitioning_test" does not exist
LINE 1: SELECT count(*) FROM partitioning_test;
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- updating partitioned table marks all of its partitions
BEGIN;
UPDATE partitioning_test SET time = now();
ERROR: relation "partitioning_test" does not exist
LINE 1: UPDATE partitioning_test SET time = now();
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- updating partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
UPDATE partitioning_test SET time = now();
ERROR: relation "partitioning_test" does not exist
LINE 1: UPDATE partitioning_test SET time = now();
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- DDLs on partitioned table marks all of its partitions
BEGIN;
ALTER TABLE partitioning_test ADD COLUMN X INT;
ERROR: relation "partitioning_test" does not exist
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
ROLLBACK;
-- DDLs on partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
ALTER TABLE partitioning_test ADD COLUMN X INT;
ERROR: relation "partitioning_test" does not exist
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
ROLLBACK;
-- reading from partition table marks its parent
BEGIN;
SELECT count(*) FROM partitioning_test_2009;
ERROR: relation "partitioning_test_2009" does not exist
LINE 1: SELECT count(*) FROM partitioning_test_2009;
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- rreading from partition table marks its parent with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
SELECT count(*) FROM partitioning_test_2009;
ERROR: relation "partitioning_test_2009" does not exist
LINE 1: SELECT count(*) FROM partitioning_test_2009;
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- updating from partition table marks its parent
BEGIN;
UPDATE partitioning_test_2009 SET time = now();
ERROR: relation "partitioning_test_2009" does not exist
LINE 1: UPDATE partitioning_test_2009 SET time = now();
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- updating from partition table marks its parent sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
UPDATE partitioning_test_2009 SET time = now();
ERROR: relation "partitioning_test_2009" does not exist
LINE 1: UPDATE partitioning_test_2009 SET time = now();
^
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
COMMIT;
-- DDLs on partition table marks its parent
BEGIN;
CREATE INDEX i1000000 ON partitioning_test_2009 (id);
ERROR: relation "partitioning_test_2009" does not exist
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
ROLLBACK;
-- DDLs on partition table marks its parent in sequential mode
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
CREATE INDEX i1000000 ON partitioning_test_2009 (id);
ERROR: relation "partitioning_test_2009" does not exist
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ERROR: current transaction is aborted, commands ignored until end of transaction block
ROLLBACK;
-- TRUNCATE CASCADE works fine
ALTER TABLE table_2 ADD CONSTRAINT table_2_u FOREIGN KEY (key) REFERENCES table_1(key);
@ -764,9 +870,9 @@ BEGIN;
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('table_1') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | parallel_access | not_accessed
(1 row)
ROLLBACK;
@ -781,9 +887,9 @@ BEGIN;
(1 row)
SELECT * FROM relation_acesses WHERE table_name IN ('table_1') ORDER BY 1;
table_name | select_access | dml_access | ddl_access
------------+-------------------+-----------------+--------------
table_1 | sequential_access | parallel_access | not_accessed
table_name | select_access | dml_access | ddl_access
------------+-----------------+-----------------+--------------
table_1 | parallel_access | parallel_access | not_accessed
(1 row)
ROLLBACK;

83
src/test/regress/sql/relation_access_tracking.sql
View File

@ -400,7 +400,7 @@ CREATE TABLE partitioning_test_2009 AS SELECT * FROM partitioning_test;
BEGIN;
ALTER TABLE partitioning_test ATTACH PARTITION partitioning_test_2009 FOR VALUES FROM ('2009-01-01') TO ('2010-01-01');
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009') ORDER BY 1;
ROLLBACK;
COMMIT;
-- Adding partition tables via ATTACH PARTITION on distributed tables should have DDL access the partitioned table as well
CREATE TABLE partitioning_test_2010 AS SELECT * FROM partitioning_test;
@ -408,6 +408,87 @@ SELECT create_distributed_table('partitioning_test_2010', 'id');
BEGIN;
ALTER TABLE partitioning_test ATTACH PARTITION partitioning_test_2010 FOR VALUES FROM ('2010-01-01') TO ('2011-01-01');
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- reading from partitioned table marks all of its partitions
BEGIN;
SELECT count(*) FROM partitioning_test;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- reading from partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
SELECT count(*) FROM partitioning_test;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- updating partitioned table marks all of its partitions
BEGIN;
UPDATE partitioning_test SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- updating partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
UPDATE partitioning_test SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- DDLs on partitioned table marks all of its partitions
BEGIN;
ALTER TABLE partitioning_test ADD COLUMN X INT;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ROLLBACK;
-- DDLs on partitioned table sequentially marks all of its partitions with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
ALTER TABLE partitioning_test ADD COLUMN X INT;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ROLLBACK;
-- reading from partition table marks its parent
BEGIN;
SELECT count(*) FROM partitioning_test_2009;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- rreading from partition table marks its parent with sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
SELECT count(*) FROM partitioning_test_2009;
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- updating from partition table marks its parent
BEGIN;
UPDATE partitioning_test_2009 SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- updating from partition table marks its parent sequential accesses
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
UPDATE partitioning_test_2009 SET time = now();
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
COMMIT;
-- DDLs on partition table marks its parent
BEGIN;
CREATE INDEX i1000000 ON partitioning_test_2009 (id);
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ROLLBACK;
-- DDLs on partition table marks its parent in sequential mode
BEGIN;
SET LOCAL citus.multi_shard_modify_mode TO 'sequential';
CREATE INDEX i1000000 ON partitioning_test_2009 (id);
SELECT * FROM relation_acesses WHERE table_name IN ('partitioning_test', 'partitioning_test_2009', 'partitioning_test_2010') ORDER BY 1;
ROLLBACK;
-- TRUNCATE CASCADE works fine