Merge pull request #326 from citusdata/feature/drop_shards_on_drop_table

Drop shards when a distributed table is dropped
pull/382/head
Marco Slot 2016-02-17 23:47:49 +01:00
commit 9e944fbcfc
18 changed files with 230 additions and 130 deletions

View File

@ -148,6 +148,15 @@ CREATE FUNCTION master_append_table_to_shard(bigint, text, text, integer)
COMMENT ON FUNCTION master_append_table_to_shard(bigint, text, text, integer) COMMENT ON FUNCTION master_append_table_to_shard(bigint, text, text, integer)
IS 'append given table to all shard placements and update metadata'; IS 'append given table to all shard placements and update metadata';
CREATE FUNCTION master_drop_all_shards(logicalrelid regclass,
schema_name text,
table_name text)
RETURNS integer
LANGUAGE C STRICT
AS 'MODULE_PATHNAME', $$master_drop_all_shards$$;
COMMENT ON FUNCTION master_drop_all_shards(regclass, text, text)
IS 'drop all shards in a relation and update metadata';
CREATE FUNCTION master_apply_delete_command(text) CREATE FUNCTION master_apply_delete_command(text)
RETURNS integer RETURNS integer
LANGUAGE C STRICT LANGUAGE C STRICT
@ -322,7 +331,7 @@ CREATE OR REPLACE FUNCTION citus_drop_trigger()
DECLARE v_obj record; DECLARE v_obj record;
BEGIN BEGIN
FOR v_obj IN SELECT * FROM pg_event_trigger_dropped_objects() LOOP FOR v_obj IN SELECT * FROM pg_event_trigger_dropped_objects() LOOP
IF v_obj.object_type <> 'table' THEN IF v_obj.object_type NOT IN ('table', 'foreign table') THEN
CONTINUE; CONTINUE;
END IF; END IF;
@ -331,20 +340,11 @@ BEGIN
CONTINUE; CONTINUE;
END IF; END IF;
-- check if there's shards for the table, error out if so -- ensure all shards are dropped
IF EXISTS(SELECT * FROM pg_dist_shard WHERE logicalrelid = v_obj.objid) THEN PERFORM master_drop_all_shards(v_obj.objid, v_obj.schema_name, v_obj.object_name);
RAISE EXCEPTION USING
MESSAGE = 'cannot drop distributed table with existing shards',
HINT = $$Delete shards first using: $$ ||
$$SELECT master_apply_delete_command('DELETE FROM $$ ||
v_obj.object_identity || $$')$$;
END IF;
-- delete partition entry -- delete partition entry
DELETE FROM pg_dist_partition WHERE logicalrelid = v_obj.objid; DELETE FROM pg_dist_partition WHERE logicalrelid = v_obj.objid;
IF NOT FOUND THEN
RAISE EXCEPTION 'could not find previously found pg_dist_partition entry';
END IF;
END LOOP; END LOOP;
END; END;

View File

@ -246,7 +246,7 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
/* finally insert tuple, build index entries & register cache invalidation */ /* finally insert tuple, build index entries & register cache invalidation */
simple_heap_insert(pgDistPartition, newTuple); simple_heap_insert(pgDistPartition, newTuple);
CatalogUpdateIndexes(pgDistPartition, newTuple); CatalogUpdateIndexes(pgDistPartition, newTuple);
CacheInvalidateRelcacheByRelid(distributedRelationId); CitusInvalidateRelcacheByRelid(distributedRelationId);
RecordDistributedRelationDependencies(distributedRelationId, distributionKey); RecordDistributedRelationDependencies(distributedRelationId, distributionKey);

View File

@ -18,20 +18,25 @@
#include "funcapi.h" #include "funcapi.h"
#include "miscadmin.h" #include "miscadmin.h"
#include "access/xact.h"
#include "catalog/namespace.h"
#include "catalog/pg_class.h" #include "catalog/pg_class.h"
#include "commands/dbcommands.h" #include "commands/dbcommands.h"
#include "commands/event_trigger.h"
#include "distributed/master_metadata_utility.h" #include "distributed/master_metadata_utility.h"
#include "distributed/master_protocol.h" #include "distributed/master_protocol.h"
#include "distributed/metadata_cache.h" #include "distributed/metadata_cache.h"
#include "distributed/multi_client_executor.h" #include "distributed/multi_client_executor.h"
#include "distributed/multi_physical_planner.h" #include "distributed/multi_physical_planner.h"
#include "distributed/multi_server_executor.h" #include "distributed/multi_server_executor.h"
#include "distributed/pg_dist_shard.h"
#include "distributed/pg_dist_partition.h" #include "distributed/pg_dist_partition.h"
#include "distributed/worker_protocol.h" #include "distributed/worker_protocol.h"
#include "optimizer/clauses.h" #include "optimizer/clauses.h"
#include "optimizer/predtest.h" #include "optimizer/predtest.h"
#include "optimizer/restrictinfo.h" #include "optimizer/restrictinfo.h"
#include "optimizer/var.h" #include "optimizer/var.h"
#include "nodes/makefuncs.h"
#include "tcop/tcopprot.h" #include "tcop/tcopprot.h"
#include "utils/builtins.h" #include "utils/builtins.h"
#include "utils/datum.h" #include "utils/datum.h"
@ -45,12 +50,15 @@ static void CheckDeleteCriteria(Node *deleteCriteria);
static void CheckPartitionColumn(Oid relationId, Node *whereClause); static void CheckPartitionColumn(Oid relationId, Node *whereClause);
static List * ShardsMatchingDeleteCriteria(Oid relationId, List *shardList, static List * ShardsMatchingDeleteCriteria(Oid relationId, List *shardList,
Node *deleteCriteria); Node *deleteCriteria);
static int DropShards(Oid relationId, char *schemaName, char *relationName,
List *deletableShardIntervalList);
static bool ExecuteRemoteCommand(const char *nodeName, uint32 nodePort, static bool ExecuteRemoteCommand(const char *nodeName, uint32 nodePort,
StringInfo queryString); StringInfo queryString);
/* exports for SQL callable functions */ /* exports for SQL callable functions */
PG_FUNCTION_INFO_V1(master_apply_delete_command); PG_FUNCTION_INFO_V1(master_apply_delete_command);
PG_FUNCTION_INFO_V1(master_drop_all_shards);
/* /*
@ -72,10 +80,9 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
text *queryText = PG_GETARG_TEXT_P(0); text *queryText = PG_GETARG_TEXT_P(0);
char *queryString = text_to_cstring(queryText); char *queryString = text_to_cstring(queryText);
char *relationName = NULL; char *relationName = NULL;
text *relationNameText = NULL; char *schemaName = NULL;
Oid relationId = InvalidOid; Oid relationId = InvalidOid;
List *shardIntervalList = NIL; List *shardIntervalList = NIL;
ListCell *shardIntervalCell = NULL;
List *deletableShardIntervalList = NIL; List *deletableShardIntervalList = NIL;
List *queryTreeList = NIL; List *queryTreeList = NIL;
Query *deleteQuery = NULL; Query *deleteQuery = NULL;
@ -83,11 +90,15 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
Node *deleteCriteria = NULL; Node *deleteCriteria = NULL;
Node *queryTreeNode = NULL; Node *queryTreeNode = NULL;
DeleteStmt *deleteStatement = NULL; DeleteStmt *deleteStatement = NULL;
int32 deleteCriteriaShardCount = 0; int droppedShardCount = 0;
LOCKTAG lockTag; LOCKTAG lockTag;
bool sessionLock = false; bool sessionLock = false;
bool dontWait = false; bool dontWait = false;
char partitionMethod = 0; char partitionMethod = 0;
bool failOK = false;
bool isTopLevel = true;
PreventTransactionChain(isTopLevel, "master_apply_delete_command");
queryTreeNode = ParseTreeNode(queryString); queryTreeNode = ParseTreeNode(queryString);
if (!IsA(queryTreeNode, DeleteStmt)) if (!IsA(queryTreeNode, DeleteStmt))
@ -97,10 +108,10 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
} }
deleteStatement = (DeleteStmt *) queryTreeNode; deleteStatement = (DeleteStmt *) queryTreeNode;
relationName = deleteStatement->relation->relname;
relationNameText = cstring_to_text(relationName);
relationId = ResolveRelationId(relationNameText); schemaName = deleteStatement->relation->schemaname;
relationName = deleteStatement->relation->relname;
relationId = RangeVarGetRelid(deleteStatement->relation, NoLock, failOK);
CheckDistributedTable(relationId); CheckDistributedTable(relationId);
queryTreeList = pg_analyze_and_rewrite(queryTreeNode, queryString, NULL, 0); queryTreeList = pg_analyze_and_rewrite(queryTreeNode, queryString, NULL, 0);
@ -142,6 +153,71 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
deleteCriteria); deleteCriteria);
} }
droppedShardCount = DropShards(relationId, schemaName, relationName,
deletableShardIntervalList);
PG_RETURN_INT32(droppedShardCount);
}
/*
* master_drop_shards attempts to drop all shards for a given relation.
* Unlike master_apply_delete_command, this function can be called even
* if the table has already been dropped.
*/
Datum
master_drop_all_shards(PG_FUNCTION_ARGS)
{
Oid relationId = PG_GETARG_OID(0);
text *schemaNameText = PG_GETARG_TEXT_P(1);
text *relationNameText = PG_GETARG_TEXT_P(2);
char *schemaName = NULL;
char *relationName = NULL;
bool isTopLevel = true;
List *shardIntervalList = NIL;
int droppedShardCount = 0;
PreventTransactionChain(isTopLevel, "DROP distributed table");
relationName = get_rel_name(relationId);
if (relationName != NULL)
{
/* ensure proper values are used if the table exists */
Oid schemaId = get_rel_namespace(relationId);
schemaName = get_namespace_name(schemaId);
}
else
{
/* table has been dropped, rely on user-supplied values */
schemaName = text_to_cstring(schemaNameText);
relationName = text_to_cstring(relationNameText);
}
shardIntervalList = LoadShardIntervalList(relationId);
droppedShardCount = DropShards(relationId, schemaName, relationName,
shardIntervalList);
PG_RETURN_INT32(droppedShardCount);
}
/*
* DropShards drops all given shards in a relation. The id, name and schema
* for the relation are explicitly provided, since this function may be
* called when the table is already dropped.
*
* We mark shard placements that we couldn't drop as to be deleted later, but
* we do delete the shard metadadata.
*/
int
DropShards(Oid relationId, char *schemaName, char *relationName,
List *deletableShardIntervalList)
{
ListCell *shardIntervalCell = NULL;
int droppedShardCount = 0;
foreach(shardIntervalCell, deletableShardIntervalList) foreach(shardIntervalCell, deletableShardIntervalList)
{ {
List *shardPlacementList = NIL; List *shardPlacementList = NIL;
@ -152,17 +228,25 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
ListCell *lingeringPlacementCell = NULL; ListCell *lingeringPlacementCell = NULL;
ShardInterval *shardInterval = (ShardInterval *) lfirst(shardIntervalCell); ShardInterval *shardInterval = (ShardInterval *) lfirst(shardIntervalCell);
uint64 shardId = shardInterval->shardId; uint64 shardId = shardInterval->shardId;
char *shardAlias = NULL;
char *quotedShardName = NULL; char *quotedShardName = NULL;
StringInfo shardName = makeStringInfo();
Assert(shardInterval->relationId == relationId);
/* if shard doesn't have an alias, extend regular table name */ /* if shard doesn't have an alias, extend regular table name */
char *shardName = LoadShardAlias(relationId, shardId); shardAlias = LoadShardAlias(relationId, shardId);
if (shardName == NULL) if (shardAlias == NULL)
{ {
shardName = get_rel_name(relationId); appendStringInfoString(shardName, relationName);
AppendShardIdToName(&shardName, shardId); AppendShardIdToStringInfo(shardName, shardId);
}
else
{
appendStringInfoString(shardName, shardAlias);
} }
quotedShardName = quote_qualified_identifier(NULL, shardName); quotedShardName = quote_qualified_identifier(schemaName, shardName->data);
shardPlacementList = ShardPlacementList(shardId); shardPlacementList = ShardPlacementList(shardId);
foreach(shardPlacementCell, shardPlacementList) foreach(shardPlacementCell, shardPlacementList)
@ -174,13 +258,14 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
bool dropSuccessful = false; bool dropSuccessful = false;
StringInfo workerDropQuery = makeStringInfo(); StringInfo workerDropQuery = makeStringInfo();
char tableType = get_rel_relkind(relationId); char storageType = shardInterval->storageType;
if (tableType == RELKIND_RELATION) if (storageType == SHARD_STORAGE_TABLE)
{ {
appendStringInfo(workerDropQuery, DROP_REGULAR_TABLE_COMMAND, appendStringInfo(workerDropQuery, DROP_REGULAR_TABLE_COMMAND,
quotedShardName); quotedShardName);
} }
else if (tableType == RELKIND_FOREIGN_TABLE) else if (storageType == SHARD_STORAGE_COLUMNAR ||
storageType == SHARD_STORAGE_FOREIGN)
{ {
appendStringInfo(workerDropQuery, DROP_FOREIGN_TABLE_COMMAND, appendStringInfo(workerDropQuery, DROP_FOREIGN_TABLE_COMMAND,
quotedShardName); quotedShardName);
@ -223,7 +308,7 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
workerName, workerPort); workerName, workerPort);
ereport(WARNING, (errmsg("could not delete shard \"%s\" on node " ereport(WARNING, (errmsg("could not delete shard \"%s\" on node "
"\"%s:%u\"", shardName, workerName, workerPort), "\"%s:%u\"", shardName->data, workerName, workerPort),
errdetail("Marking this shard placement for deletion"))); errdetail("Marking this shard placement for deletion")));
} }
@ -239,8 +324,9 @@ master_apply_delete_command(PG_FUNCTION_ARGS)
RESUME_INTERRUPTS(); RESUME_INTERRUPTS();
} }
deleteCriteriaShardCount = list_length(deletableShardIntervalList); droppedShardCount = list_length(deletableShardIntervalList);
PG_RETURN_INT32(deleteCriteriaShardCount);
return droppedShardCount;
} }

View File

@ -394,7 +394,7 @@ InsertShardRow(Oid relationId, uint64 shardId, char storageType,
/* close relation and invalidate previous cache entry */ /* close relation and invalidate previous cache entry */
heap_close(pgDistShard, RowExclusiveLock); heap_close(pgDistShard, RowExclusiveLock);
CacheInvalidateRelcacheByRelid(relationId); CitusInvalidateRelcacheByRelid(relationId);
} }
@ -479,7 +479,7 @@ DeleteShardRow(uint64 shardId)
heap_close(pgDistShard, RowExclusiveLock); heap_close(pgDistShard, RowExclusiveLock);
/* invalidate previous cache entry */ /* invalidate previous cache entry */
CacheInvalidateRelcacheByRelid(distributedRelationId); CitusInvalidateRelcacheByRelid(distributedRelationId);
} }

View File

@ -520,3 +520,15 @@ AppendShardIdToName(char **name, uint64 shardId)
(*name) = (char *) repalloc((*name), extendedNameLength); (*name) = (char *) repalloc((*name), extendedNameLength);
snprintf((*name), extendedNameLength, "%s", extendedName); snprintf((*name), extendedNameLength, "%s", extendedName);
} }
/*
* AppendShardIdToStringInfo appends shardId to the given name, represented
* by a StringInfo.
*/
void
AppendShardIdToStringInfo(StringInfo name, uint64 shardId)
{
appendStringInfo(name, "%c" UINT64_FORMAT, SHARD_NAME_SEPARATOR, shardId);
}

View File

@ -479,33 +479,17 @@ master_dist_partition_cache_invalidate(PG_FUNCTION_ARGS)
/* /*
* Invalidate relcache for the relevant relation(s). In theory * Invalidate relcache for the relevant relation(s). In theory
* logicalrelid should never change, but it doesn't hurt to be * logicalrelid should never change, but it doesn't hurt to be
* paranoid. We ignore the case that there's no corresponding pg_class * paranoid.
* entry - that happens if the pg_dist_partition tuple is deleted after
* the relation has been dropped.
*/ */
if (oldLogicalRelationId != InvalidOid && if (oldLogicalRelationId != InvalidOid &&
oldLogicalRelationId != newLogicalRelationId) oldLogicalRelationId != newLogicalRelationId)
{ {
HeapTuple oldClassTuple = CitusInvalidateRelcacheByRelid(oldLogicalRelationId);
SearchSysCache1(RELOID, ObjectIdGetDatum(oldLogicalRelationId));
if (HeapTupleIsValid(oldClassTuple))
{
CacheInvalidateRelcacheByTuple(oldClassTuple);
ReleaseSysCache(oldClassTuple);
}
} }
if (newLogicalRelationId != InvalidOid) if (newLogicalRelationId != InvalidOid)
{ {
HeapTuple newClassTuple = CitusInvalidateRelcacheByRelid(newLogicalRelationId);
SearchSysCache1(RELOID, ObjectIdGetDatum(newLogicalRelationId));
if (HeapTupleIsValid(newClassTuple))
{
CacheInvalidateRelcacheByTuple(newClassTuple);
ReleaseSysCache(newClassTuple);
}
} }
PG_RETURN_DATUM(PointerGetDatum(NULL)); PG_RETURN_DATUM(PointerGetDatum(NULL));
@ -553,33 +537,17 @@ master_dist_shard_cache_invalidate(PG_FUNCTION_ARGS)
/* /*
* Invalidate relcache for the relevant relation(s). In theory * Invalidate relcache for the relevant relation(s). In theory
* logicalrelid should never change, but it doesn't hurt to be * logicalrelid should never change, but it doesn't hurt to be
* paranoid. We ignore the case that there's no corresponding pg_class * paranoid.
* entry - that happens if the pg_dist_shard tuple is deleted after
* the relation has been dropped.
*/ */
if (oldLogicalRelationId != InvalidOid && if (oldLogicalRelationId != InvalidOid &&
oldLogicalRelationId != newLogicalRelationId) oldLogicalRelationId != newLogicalRelationId)
{ {
HeapTuple oldClassTuple = CitusInvalidateRelcacheByRelid(oldLogicalRelationId);
SearchSysCache1(RELOID, ObjectIdGetDatum(oldLogicalRelationId));
if (HeapTupleIsValid(oldClassTuple))
{
CacheInvalidateRelcacheByTuple(oldClassTuple);
ReleaseSysCache(oldClassTuple);
}
} }
if (newLogicalRelationId != InvalidOid) if (newLogicalRelationId != InvalidOid)
{ {
HeapTuple newClassTuple = CitusInvalidateRelcacheByRelid(newLogicalRelationId);
SearchSysCache1(RELOID, ObjectIdGetDatum(newLogicalRelationId));
if (HeapTupleIsValid(newClassTuple))
{
CacheInvalidateRelcacheByTuple(newClassTuple);
ReleaseSysCache(newClassTuple);
}
} }
PG_RETURN_DATUM(PointerGetDatum(NULL)); PG_RETURN_DATUM(PointerGetDatum(NULL));
@ -929,3 +897,25 @@ CachedRelationLookup(const char *relationName, Oid *cachedOid)
} }
} }
} }
/*
* Register a relcache invalidation for a non-shared relation.
*
* We ignore the case that there's no corresponding pg_class entry - that
* happens if we register a relcache invalidation (e.g. for a
* pg_dist_partition deletion) after the relation has been dropped. That's ok,
* because in those cases we're guaranteed to already have registered an
* invalidation for the target relation.
*/
void
CitusInvalidateRelcacheByRelid(Oid relationId)
{
HeapTuple classTuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relationId));
if (HeapTupleIsValid(classTuple))
{
CacheInvalidateRelcacheByTuple(classTuple);
ReleaseSysCache(classTuple);
}
}

View File

@ -98,6 +98,7 @@ extern Datum master_get_active_worker_nodes(PG_FUNCTION_ARGS);
extern Datum master_create_empty_shard(PG_FUNCTION_ARGS); extern Datum master_create_empty_shard(PG_FUNCTION_ARGS);
extern Datum master_append_table_to_shard(PG_FUNCTION_ARGS); extern Datum master_append_table_to_shard(PG_FUNCTION_ARGS);
extern Datum master_apply_delete_command(PG_FUNCTION_ARGS); extern Datum master_apply_delete_command(PG_FUNCTION_ARGS);
extern Datum master_drop_all_shards(PG_FUNCTION_ARGS);
/* function declarations for shard creation functionality */ /* function declarations for shard creation functionality */
extern Datum master_create_worker_shards(PG_FUNCTION_ARGS); extern Datum master_create_worker_shards(PG_FUNCTION_ARGS);

View File

@ -45,6 +45,7 @@ typedef struct
extern bool IsDistributedTable(Oid relationId); extern bool IsDistributedTable(Oid relationId);
extern ShardInterval * LoadShardInterval(uint64 shardId); extern ShardInterval * LoadShardInterval(uint64 shardId);
extern DistTableCacheEntry * DistributedTableCacheEntry(Oid distributedRelationId); extern DistTableCacheEntry * DistributedTableCacheEntry(Oid distributedRelationId);
extern void CitusInvalidateRelcacheByRelid(Oid relationId);
extern bool CitusHasBeenLoaded(void); extern bool CitusHasBeenLoaded(void);

View File

@ -15,6 +15,7 @@
#ifndef RELAY_UTILITY_H #ifndef RELAY_UTILITY_H
#define RELAY_UTILITY_H #define RELAY_UTILITY_H
#include "lib/stringinfo.h"
#include "nodes/nodes.h" #include "nodes/nodes.h"
@ -41,6 +42,7 @@ typedef enum
/* Function declarations to extend names in DDL commands */ /* Function declarations to extend names in DDL commands */
extern void RelayEventExtendNames(Node *parseTree, uint64 shardId); extern void RelayEventExtendNames(Node *parseTree, uint64 shardId);
extern void AppendShardIdToName(char **name, uint64 shardId); extern void AppendShardIdToName(char **name, uint64 shardId);
extern void AppendShardIdToStringInfo(StringInfo name, uint64 shardId);
#endif /* RELAY_UTILITY_H */ #endif /* RELAY_UTILITY_H */

View File

@ -136,7 +136,7 @@ DEBUG: applied command on shard 102010 on node localhost:57638
DEBUG: applied command on shard 102010 on node localhost:57637 DEBUG: applied command on shard 102010 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57637 DEBUG: applied command on shard 102009 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57638 DEBUG: applied command on shard 102009 on node localhost:57638
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
DROP INDEX lineitem_partkey_desc_index; DROP INDEX lineitem_partkey_desc_index;
DEBUG: applied command on shard 102014 on node localhost:57638 DEBUG: applied command on shard 102014 on node localhost:57638
DEBUG: applied command on shard 102014 on node localhost:57637 DEBUG: applied command on shard 102014 on node localhost:57637
@ -150,7 +150,7 @@ DEBUG: applied command on shard 102010 on node localhost:57638
DEBUG: applied command on shard 102010 on node localhost:57637 DEBUG: applied command on shard 102010 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57637 DEBUG: applied command on shard 102009 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57638 DEBUG: applied command on shard 102009 on node localhost:57638
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
DROP INDEX lineitem_partial_index; DROP INDEX lineitem_partial_index;
DEBUG: applied command on shard 102014 on node localhost:57638 DEBUG: applied command on shard 102014 on node localhost:57638
DEBUG: applied command on shard 102014 on node localhost:57637 DEBUG: applied command on shard 102014 on node localhost:57637
@ -164,7 +164,7 @@ DEBUG: applied command on shard 102010 on node localhost:57638
DEBUG: applied command on shard 102010 on node localhost:57637 DEBUG: applied command on shard 102010 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57637 DEBUG: applied command on shard 102009 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57638 DEBUG: applied command on shard 102009 on node localhost:57638
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
-- Verify that we handle if exists statements correctly -- Verify that we handle if exists statements correctly
DROP INDEX non_existent_index; DROP INDEX non_existent_index;
ERROR: index "non_existent_index" does not exist ERROR: index "non_existent_index" does not exist
@ -183,7 +183,7 @@ DEBUG: applied command on shard 102010 on node localhost:57638
DEBUG: applied command on shard 102010 on node localhost:57637 DEBUG: applied command on shard 102010 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57637 DEBUG: applied command on shard 102009 on node localhost:57637
DEBUG: applied command on shard 102009 on node localhost:57638 DEBUG: applied command on shard 102009 on node localhost:57638
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
DROP INDEX lineitem_orderkey_hash_index; DROP INDEX lineitem_orderkey_hash_index;
ERROR: index "lineitem_orderkey_hash_index" does not exist ERROR: index "lineitem_orderkey_hash_index" does not exist
-- Verify that all the indexes are also dropped from the master node -- Verify that all the indexes are also dropped from the master node

View File

@ -22,27 +22,29 @@ ERROR: cannot execute ALTER TABLE command involving partition column
-- verify that the distribution column can't be dropped -- verify that the distribution column can't be dropped
ALTER TABLE testtableddl DROP COLUMN distributecol; ALTER TABLE testtableddl DROP COLUMN distributecol;
ERROR: cannot execute ALTER TABLE command involving partition column ERROR: cannot execute ALTER TABLE command involving partition column
-- verify that the table cannot be dropped while shards exist -- verify that the table cannot be dropped in a transaction block
BEGIN;
DROP TABLE testtableddl;
ERROR: DROP distributed table cannot run inside a transaction block
CONTEXT: SQL statement "SELECT master_drop_all_shards(v_obj.objid, v_obj.schema_name, v_obj.object_name)"
PL/pgSQL function citus_drop_trigger() line 15 at PERFORM
ROLLBACK;
-- verify that the table can be dropped
DROP TABLE testtableddl;
-- verify that the table can dropped even if shards exist
CREATE TABLE testtableddl(somecol int, distributecol text NOT NULL);
SELECT master_create_distributed_table('testtableddl', 'distributecol', 'append');
master_create_distributed_table
---------------------------------
(1 row)
SELECT 1 FROM master_create_empty_shard('testtableddl'); SELECT 1 FROM master_create_empty_shard('testtableddl');
?column? ?column?
---------- ----------
1 1
(1 row) (1 row)
DROP TABLE testtableddl;
ERROR: cannot drop distributed table with existing shards
HINT: Delete shards first using: SELECT master_apply_delete_command('DELETE FROM public.testtableddl')
-- not even with cascade
DROP TABLE testtableddl CASCADE;
ERROR: cannot drop distributed table with existing shards
HINT: Delete shards first using: SELECT master_apply_delete_command('DELETE FROM public.testtableddl')
-- but it can be dropped after dropping the shards
SELECT master_apply_delete_command('DELETE FROM testtableddl');
master_apply_delete_command
-----------------------------
1
(1 row)
DROP TABLE testtableddl; DROP TABLE testtableddl;
-- ensure no metadata of distributed tables are remaining -- ensure no metadata of distributed tables are remaining
SELECT * FROM pg_dist_partition; SELECT * FROM pg_dist_partition;

View File

@ -65,10 +65,6 @@ EXECUTE sharded_query;
------ ------
(0 rows) (0 rows)
-- try to drop table
DROP TABLE sharded_table;
ERROR: cannot drop distributed table with existing shards
HINT: Delete shards first using: SELECT master_apply_delete_command('DELETE FROM public.sharded_table')
-- try to drop shards with where clause -- try to drop shards with where clause
SELECT master_apply_delete_command('DELETE FROM sharded_table WHERE id > 0'); SELECT master_apply_delete_command('DELETE FROM sharded_table WHERE id > 0');
ERROR: cannot delete from distributed table ERROR: cannot delete from distributed table

View File

@ -46,3 +46,8 @@ SELECT master_create_empty_shard('customer_delete_protocol');
SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol
WHERE c_custkey > 1000'); WHERE c_custkey > 1000');
SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol'); SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol');
-- Verify that master_apply_delete_command cannot be called in a transaction block
BEGIN;
SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol');
ROLLBACK;

View File

@ -55,8 +55,8 @@ DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17666 DEBUG: drop auto-cascades to type pg_temp_17675
DEBUG: drop auto-cascades to type pg_temp_17666[] DEBUG: drop auto-cascades to type pg_temp_17675[]
ALTER TABLE lineitem_alter ADD COLUMN int_column2 INTEGER DEFAULT 2; ALTER TABLE lineitem_alter ADD COLUMN int_column2 INTEGER DEFAULT 2;
DEBUG: applied command on shard 103002 on node localhost:57638 DEBUG: applied command on shard 103002 on node localhost:57638
DEBUG: applied command on shard 103002 on node localhost:57637 DEBUG: applied command on shard 103002 on node localhost:57637
@ -65,8 +65,8 @@ DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17666 DEBUG: drop auto-cascades to type pg_temp_17675
DEBUG: drop auto-cascades to type pg_temp_17666[] DEBUG: drop auto-cascades to type pg_temp_17675[]
ALTER TABLE lineitem_alter ADD COLUMN null_column INTEGER; ALTER TABLE lineitem_alter ADD COLUMN null_column INTEGER;
DEBUG: applied command on shard 103002 on node localhost:57638 DEBUG: applied command on shard 103002 on node localhost:57638
DEBUG: applied command on shard 103002 on node localhost:57637 DEBUG: applied command on shard 103002 on node localhost:57637
@ -359,8 +359,8 @@ DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17666 DEBUG: drop auto-cascades to type pg_temp_17675
DEBUG: drop auto-cascades to type pg_temp_17666[] DEBUG: drop auto-cascades to type pg_temp_17675[]
\d lineitem_alter \d lineitem_alter
Table "public.lineitem_alter" Table "public.lineitem_alter"
Column | Type | Modifiers Column | Type | Modifiers
@ -478,8 +478,8 @@ DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17666 DEBUG: drop auto-cascades to type pg_temp_17675
DEBUG: drop auto-cascades to type pg_temp_17666[] DEBUG: drop auto-cascades to type pg_temp_17675[]
ALTER TABLE lineitem_alter DROP COLUMN non_existent_column; ALTER TABLE lineitem_alter DROP COLUMN non_existent_column;
WARNING: could not receive query results from localhost:57637 WARNING: could not receive query results from localhost:57637
DETAIL: Client error: column "non_existent_column" of relation "lineitem_alter_103009" does not exist DETAIL: Client error: column "non_existent_column" of relation "lineitem_alter_103009" does not exist

View File

@ -54,10 +54,10 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerTableRewrite(17667) DEBUG: EventTriggerTableRewrite(17676)
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17667 DEBUG: drop auto-cascades to type pg_temp_17676
DEBUG: drop auto-cascades to type pg_temp_17667[] DEBUG: drop auto-cascades to type pg_temp_17676[]
ALTER TABLE lineitem_alter ADD COLUMN int_column2 INTEGER DEFAULT 2; ALTER TABLE lineitem_alter ADD COLUMN int_column2 INTEGER DEFAULT 2;
DEBUG: applied command on shard 103002 on node localhost:57638 DEBUG: applied command on shard 103002 on node localhost:57638
DEBUG: applied command on shard 103002 on node localhost:57637 DEBUG: applied command on shard 103002 on node localhost:57637
@ -65,10 +65,10 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerTableRewrite(17667) DEBUG: EventTriggerTableRewrite(17676)
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17667 DEBUG: drop auto-cascades to type pg_temp_17676
DEBUG: drop auto-cascades to type pg_temp_17667[] DEBUG: drop auto-cascades to type pg_temp_17676[]
ALTER TABLE lineitem_alter ADD COLUMN null_column INTEGER; ALTER TABLE lineitem_alter ADD COLUMN null_column INTEGER;
DEBUG: applied command on shard 103002 on node localhost:57638 DEBUG: applied command on shard 103002 on node localhost:57638
DEBUG: applied command on shard 103002 on node localhost:57637 DEBUG: applied command on shard 103002 on node localhost:57637
@ -129,7 +129,7 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
-- \stage to verify that default values take effect -- \stage to verify that default values take effect
\STAGE lineitem_alter (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate, l_receiptdate, l_shipinstruct, l_shipmode, l_comment) FROM '@abs_srcdir@/data/lineitem.1.data' with delimiter '|' \STAGE lineitem_alter (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate, l_receiptdate, l_shipinstruct, l_shipmode, l_comment) FROM '@abs_srcdir@/data/lineitem.1.data' with delimiter '|'
DEBUG: parse <unnamed>: SELECT * FROM master_get_table_metadata($1::text) DEBUG: parse <unnamed>: SELECT * FROM master_get_table_metadata($1::text)
@ -235,7 +235,7 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
-- \stage should fail because it will try to insert NULLs for a NOT NULL column -- \stage should fail because it will try to insert NULLs for a NOT NULL column
\STAGE lineitem_alter (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate, l_receiptdate, l_shipinstruct, l_shipmode, l_comment) FROM '@abs_srcdir@/data/lineitem.1.data' with delimiter '|' \STAGE lineitem_alter (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate, l_receiptdate, l_shipinstruct, l_shipmode, l_comment) FROM '@abs_srcdir@/data/lineitem.1.data' with delimiter '|'
DEBUG: parse <unnamed>: SELECT * FROM master_get_table_metadata($1::text) DEBUG: parse <unnamed>: SELECT * FROM master_get_table_metadata($1::text)
@ -362,10 +362,10 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerTableRewrite(17667) DEBUG: EventTriggerTableRewrite(17676)
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17667 DEBUG: drop auto-cascades to type pg_temp_17676
DEBUG: drop auto-cascades to type pg_temp_17667[] DEBUG: drop auto-cascades to type pg_temp_17676[]
\d lineitem_alter \d lineitem_alter
Table "public.lineitem_alter" Table "public.lineitem_alter"
Column | Type | Modifiers Column | Type | Modifiers
@ -419,7 +419,7 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
ALTER TABLE lineitem_alter DROP COLUMN float_column; ALTER TABLE lineitem_alter DROP COLUMN float_column;
DEBUG: applied command on shard 103009 on node localhost:57637 DEBUG: applied command on shard 103009 on node localhost:57637
DEBUG: applied command on shard 103009 on node localhost:57638 DEBUG: applied command on shard 103009 on node localhost:57638
@ -440,7 +440,7 @@ DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: drop auto-cascades to default for table lineitem_alter column float_column DEBUG: drop auto-cascades to default for table lineitem_alter column float_column
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
ALTER TABLE lineitem_alter DROP COLUMN date_column; ALTER TABLE lineitem_alter DROP COLUMN date_column;
DEBUG: applied command on shard 103009 on node localhost:57637 DEBUG: applied command on shard 103009 on node localhost:57637
DEBUG: applied command on shard 103009 on node localhost:57638 DEBUG: applied command on shard 103009 on node localhost:57638
@ -460,7 +460,7 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
-- Verify that IF EXISTS works as expected -- Verify that IF EXISTS works as expected
ALTER TABLE non_existent_table ADD COLUMN new_column INTEGER; ALTER TABLE non_existent_table ADD COLUMN new_column INTEGER;
ERROR: relation "non_existent_table" does not exist ERROR: relation "non_existent_table" does not exist
@ -485,10 +485,10 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerTableRewrite(17667) DEBUG: EventTriggerTableRewrite(17676)
DEBUG: rewriting table "lineitem_alter" DEBUG: rewriting table "lineitem_alter"
DEBUG: drop auto-cascades to type pg_temp_17667 DEBUG: drop auto-cascades to type pg_temp_17676
DEBUG: drop auto-cascades to type pg_temp_17667[] DEBUG: drop auto-cascades to type pg_temp_17676[]
ALTER TABLE lineitem_alter DROP COLUMN non_existent_column; ALTER TABLE lineitem_alter DROP COLUMN non_existent_column;
WARNING: could not receive query results from localhost:57637 WARNING: could not receive query results from localhost:57637
DETAIL: Client error: column "non_existent_column" of relation "lineitem_alter_103009" does not exist DETAIL: Client error: column "non_existent_column" of relation "lineitem_alter_103009" does not exist
@ -532,7 +532,7 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
\d lineitem_alter \d lineitem_alter
Table "public.lineitem_alter" Table "public.lineitem_alter"
Column | Type | Modifiers Column | Type | Modifiers
@ -623,7 +623,7 @@ DEBUG: applied command on shard 103001 on node localhost:57637
DEBUG: applied command on shard 103001 on node localhost:57638 DEBUG: applied command on shard 103001 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57638 DEBUG: applied command on shard 103000 on node localhost:57638
DEBUG: applied command on shard 103000 on node localhost:57637 DEBUG: applied command on shard 103000 on node localhost:57637
DEBUG: EventTriggerInvoke 16532 DEBUG: EventTriggerInvoke 16541
\d lineitem_alter \d lineitem_alter
Table "public.lineitem_alter" Table "public.lineitem_alter"
Column | Type | Modifiers Column | Type | Modifiers

View File

@ -86,3 +86,8 @@ SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol');
1 1
(1 row) (1 row)
-- Verify that master_apply_delete_command cannot be called in a transaction block
BEGIN;
SELECT master_apply_delete_command('DELETE FROM customer_delete_protocol');
ERROR: master_apply_delete_command cannot run inside a transaction block
ROLLBACK;

View File

@ -15,15 +15,18 @@ ALTER TABLE testtableddl ALTER COLUMN distributecol TYPE text;
-- verify that the distribution column can't be dropped -- verify that the distribution column can't be dropped
ALTER TABLE testtableddl DROP COLUMN distributecol; ALTER TABLE testtableddl DROP COLUMN distributecol;
-- verify that the table cannot be dropped while shards exist -- verify that the table cannot be dropped in a transaction block
SELECT 1 FROM master_create_empty_shard('testtableddl'); BEGIN;
DROP TABLE testtableddl;
ROLLBACK;
-- verify that the table can be dropped
DROP TABLE testtableddl; DROP TABLE testtableddl;
-- not even with cascade -- verify that the table can dropped even if shards exist
DROP TABLE testtableddl CASCADE; CREATE TABLE testtableddl(somecol int, distributecol text NOT NULL);
SELECT master_create_distributed_table('testtableddl', 'distributecol', 'append');
-- but it can be dropped after dropping the shards SELECT 1 FROM master_create_empty_shard('testtableddl');
SELECT master_apply_delete_command('DELETE FROM testtableddl');
DROP TABLE testtableddl; DROP TABLE testtableddl;
-- ensure no metadata of distributed tables are remaining -- ensure no metadata of distributed tables are remaining

View File

@ -36,9 +36,6 @@ EXECUTE sharded_query;
EXECUTE sharded_delete; EXECUTE sharded_delete;
EXECUTE sharded_query; EXECUTE sharded_query;
-- try to drop table
DROP TABLE sharded_table;
-- try to drop shards with where clause -- try to drop shards with where clause
SELECT master_apply_delete_command('DELETE FROM sharded_table WHERE id > 0'); SELECT master_apply_delete_command('DELETE FROM sharded_table WHERE id > 0');