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

Address feedback

pull/721/head
Murat Tuncer 2016-09-26 09:16:26 +03:00 committed by Jason Petersen
parent 877694296f
commit 6317bbe9a8
No known key found for this signature in database
GPG Key ID: 9F1D3510D110ABA9
10 changed files with 91 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/backend/distributed/Makefile
View File

@ -55,7 +55,6 @@ $(EXTENSION)--5.2-1.sql: $(EXTENSION)--5.1-8.sql $(EXTENSION)--5.1-8--5.2-1.sql
$(EXTENSION)--5.2-2.sql: $(EXTENSION)--5.2-1.sql $(EXTENSION)--5.2-1--5.2-2.sql
cat $^ > $@
NO_PGXS = 1
SHLIB_LINK = $(libpq)

19
src/backend/distributed/citus--5.2-1--5.2-2.sql
View File

@ -9,15 +9,18 @@ DECLARE
BEGIN
SELECT partmethod INTO partitionType
FROM pg_dist_partition WHERE logicalrelid = TG_RELID;
IF FOUND THEN
IF (partitionType = 'a') THEN
PERFORM master_drop_all_shards(TG_RELID, TG_TABLE_SCHEMA, TG_TABLE_NAME);
ELSE
SELECT format('truncate table %s.%s', TG_TABLE_SCHEMA, TG_TABLE_NAME)
INTO commandText;
PERFORM master_modify_multiple_shards(commandText);
END IF;
IF NOT FOUND THEN
RETURN NEW;
END IF;
IF (partitionType = 'a') THEN
PERFORM master_drop_all_shards(TG_RELID, TG_TABLE_SCHEMA, TG_TABLE_NAME);
ELSE
SELECT format('TRUNCATE TABLE %I.%I CASCADE', TG_TABLE_SCHEMA, TG_TABLE_NAME)
INTO commandText;
PERFORM master_modify_multiple_shards(commandText);
END IF;
RETURN NEW;
END;
$cdbtt$;

30
src/backend/distributed/commands/create_distributed_table.c
View File

@ -54,8 +54,7 @@ static void RecordDistributedRelationDependencies(Oid distributedRelationId,
static Oid SupportFunctionForColumn(Var *partitionColumn, Oid accessMethodId,
int16 supportFunctionNumber);
static bool LocalTableEmpty(Oid tableId);
static void CreateTruncateTrigger(Oid relationId, char *qualifiedRelationName);
static void CreateTruncateTrigger(Oid relationId);
/* exports for SQL callable functions */
PG_FUNCTION_INFO_V1(master_create_distributed_table);
@ -82,9 +81,6 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
Relation distributedRelation = NULL;
TupleDesc relationDesc = NULL;
char *distributedRelationName = NULL;
Oid distributedRelationSchemaOid = InvalidOid;
char *distributedRelationSchema = NULL;
char *qualifiedRelationName = NULL;
char relationKind = '\0';
Relation pgDistPartition = NULL;
@ -109,11 +105,6 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
distributedRelation = relation_open(distributedRelationId, AccessExclusiveLock);
relationDesc = RelationGetDescr(distributedRelation);
distributedRelationName = RelationGetRelationName(distributedRelation);
distributedRelationSchemaOid = RelationGetNamespace(distributedRelation);
distributedRelationSchema = get_namespace_name(distributedRelationSchemaOid);
qualifiedRelationName = quote_qualified_identifier(distributedRelationSchema,
distributedRelationName);
EnsureTableOwner(distributedRelationId);
@ -315,7 +306,7 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
*/
if (relationKind == RELKIND_RELATION)
{
CreateTruncateTrigger(distributedRelationId, qualifiedRelationName);
CreateTruncateTrigger(distributedRelationId);
}
PG_RETURN_VOID();
@ -498,18 +489,18 @@ LocalTableEmpty(Oid tableId)
}
/* CreateTruncateTrigger creates a truncate trigger on table identified by relationId
* and assigns citus_truncate_trigger() as handler. The new trigger is named as
* citus_truncate_trigger_on_ + schemaName.tableName. Trigger name for relation my_table
* from schema my_schema will be citus_truncate_trigger_on_my_schema.my_table to prevent
* name conflicts.
/*
* CreateTruncateTrigger creates a truncate trigger on table identified by relationId
* and assigns citus_truncate_trigger() as handler.
*/
static void
CreateTruncateTrigger(Oid relationId, char *qualifiedRelationName)
CreateTruncateTrigger(Oid relationId)
{
CreateTrigStmt *trigger = NULL;
StringInfo triggerName = makeStringInfo();
appendStringInfo(triggerName, "citus_truncate_trigger_on_%s", qualifiedRelationName);
bool internal = true;
appendStringInfo(triggerName, "truncate_trigger");
trigger = makeNode(CreateTrigStmt);
trigger->trigname = triggerName->data;
@ -523,5 +514,6 @@ CreateTruncateTrigger(Oid relationId, char *qualifiedRelationName)
trigger->whenClause = NULL;
trigger->isconstraint = false;
CreateTrigger(trigger, NULL, relationId, InvalidOid, InvalidOid, InvalidOid, false);
CreateTrigger(trigger, NULL, relationId, InvalidOid, InvalidOid, InvalidOid,
internal);
}

49
src/backend/distributed/master/master_modify_multiple_shards.c
View File

@ -61,8 +61,6 @@ static int SendQueryToShards(Query *query, List *shardIntervalList, Oid relation
static int SendQueryToPlacements(char *shardQueryString,
ShardConnections *shardConnections,
bool returnTupleCount);
static void deparse_truncate_query(Query *query, Oid distrelid, int64 shardid, StringInfo
buffer);
PG_FUNCTION_INFO_V1(master_modify_multiple_shards);
@ -91,7 +89,6 @@ master_modify_multiple_shards(PG_FUNCTION_ARGS)
List *shardIntervalList = NIL;
List *prunedShardIntervalList = NIL;
int32 affectedTupleCount = 0;
bool validateModifyQuery = true;
PreventTransactionChain(isTopLevel, "master_modify_multiple_shards");
@ -124,8 +121,14 @@ master_modify_multiple_shards(PG_FUNCTION_ARGS)
rangeVar = (RangeVar *) linitial(relationList);
relationId = RangeVarGetRelid(rangeVar, NoLock, failOK);
if (rangeVar->schemaname == NULL)
{
Oid schemaOid = get_rel_namespace(relationId);
char *schemaName = get_namespace_name(schemaOid);
rangeVar->schemaname = schemaName;
}
EnsureTablePermissions(relationId, ACL_TRUNCATE);
validateModifyQuery = false;
}
else
{
@ -138,7 +141,7 @@ master_modify_multiple_shards(PG_FUNCTION_ARGS)
queryTreeList = pg_analyze_and_rewrite(queryTreeNode, queryString, NULL, 0);
modifyQuery = (Query *) linitial(queryTreeList);
if (validateModifyQuery)
if (modifyQuery->commandType != CMD_UTILITY)
{
ErrorIfModifyQueryNotSupported(modifyQuery);
}
@ -267,14 +270,7 @@ SendQueryToShards(Query *query, List *shardIntervalList, Oid relationId)
shardConnections = GetShardConnections(shardId, &shardConnectionsFound);
Assert(shardConnectionsFound);
if (truncateCommand)
{
deparse_truncate_query(query, relationId, shardId, shardQueryString);
}
else
{
deparse_shard_query(query, relationId, shardId, shardQueryString);
}
deparse_shard_query(query, relationId, shardId, shardQueryString);
shardQueryStringData = shardQueryString->data;
shardAffectedTupleCount = SendQueryToPlacements(shardQueryStringData,
@ -290,33 +286,6 @@ SendQueryToShards(Query *query, List *shardIntervalList, Oid relationId)
}
/*
* deparse_truncate_query creates sql representation of a truncate statement. The
* function only generated basic truncate statement of the form
* 'truncate table <table_name>' it ignores all options. It also assumes that
* there is only one relation in the relation list.
*/
void
deparse_truncate_query(Query *query, Oid distrelid, int64 shardid, StringInfo buffer)
{
TruncateStmt *truncateStatement = NULL;
RangeVar *relation = NULL;
char *qualifiedName = NULL;
Assert(query->commandType == CMD_UTILITY);
Assert(IsA(query->utilityStmt, TruncateStmt));
truncateStatement = (TruncateStmt *) query->utilityStmt;
Assert(list_length(truncateStatement->relations) == 1);
relation = (RangeVar *) linitial(truncateStatement->relations);
qualifiedName = quote_qualified_identifier(relation->schemaname,
relation->relname);
appendStringInfo(buffer, "TRUNCATE TABLE %s_" UINT64_FORMAT, qualifiedName, shardid);
}
/*
* SendQueryToPlacements sends the given query string to all given placement
* connections of a shard. CommitRemoteTransactions or AbortRemoteTransactions

36
src/backend/distributed/utils/ruleutils_95.c
View File

@ -3398,6 +3398,42 @@ get_utility_query_def(Query *query, deparse_context *context)
simple_quote_literal(buf, stmt->payload);
}
}
else if (query->utilityStmt && IsA(query->utilityStmt, TruncateStmt))
{
TruncateStmt *stmt = (TruncateStmt *) query->utilityStmt;
List *relationList = stmt->relations;
ListCell *relationCell = NULL;
appendContextKeyword(context, "",
0, PRETTYINDENT_STD, 1);
appendStringInfo(buf, "TRUNCATE TABLE");
foreach(relationCell, relationList)
{
RangeVar *relationVar = (RangeVar *) lfirst(relationCell);
Oid relationId = RangeVarGetRelid(relationVar, NoLock, false);
char *relationName = generate_relation_or_shard_name(relationId,
context->distrelid,
context->shardid, NIL);
appendStringInfo(buf, " %s", relationName);
if (lnext(relationCell) != NULL)
{
appendStringInfo(buf, ",");
}
}
if (stmt->restart_seqs)
{
appendStringInfo(buf, " RESTART IDENTITY");
}
if (stmt->behavior == DROP_CASCADE)
{
appendStringInfo(buf, " CASCADE");
}
}
else
{
/* Currently only NOTIFY utility commands can appear in rules */

6
src/test/regress/expected/multi_modifying_xacts.out
View File

@ -172,8 +172,6 @@ COMMIT;
--------+--------+-----------
id | bigint | not null
name | text | not null
Triggers:
"citus_truncate_trigger_on_public.labs" BEFORE TRUNCATE ON labs FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
SELECT * FROM labs WHERE id = 6;
id | name
@ -225,8 +223,6 @@ COMMIT;
--------+--------+-----------
id | bigint | not null
name | text | not null
Triggers:
"citus_truncate_trigger_on_public.labs" BEFORE TRUNCATE ON labs FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
SELECT * FROM labs WHERE id = 12;
id | name
@ -246,8 +242,6 @@ COMMIT;
--------+--------+-----------
id | bigint | not null
name | text | not null
Triggers:
"citus_truncate_trigger_on_public.labs" BEFORE TRUNCATE ON labs FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
SELECT * FROM labs WHERE id = 12;
id | name

16
src/test/regress/expected/multi_schema_support.out
View File

@ -572,8 +572,6 @@ HINT: You can enable two-phase commit for extra safety with: SET citus.multi_sh
n_regionkey | integer | not null
n_comment | character varying(152) |
new_col | integer |
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON test_schema_support.nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;
@ -601,8 +599,6 @@ ALTER TABLE test_schema_support.nation_hash DROP COLUMN IF EXISTS new_col;
n_name | character(25) | not null
n_regionkey | integer | not null
n_comment | character varying(152) |
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON test_schema_support.nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;
@ -630,8 +626,6 @@ HINT: You can enable two-phase commit for extra safety with: SET citus.multi_sh
n_regionkey | integer | not null
n_comment | character varying(152) |
new_col | integer |
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;
@ -660,8 +654,6 @@ ALTER TABLE nation_hash DROP COLUMN IF EXISTS new_col;
n_name | character(25) | not null
n_regionkey | integer | not null
n_comment | character varying(152) |
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;
@ -691,8 +683,6 @@ HINT: You can enable two-phase commit for extra safety with: SET citus.multi_sh
n_comment | character varying(152) |
Indexes:
"index1" btree (n_name)
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON test_schema_support.nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;
@ -720,8 +710,6 @@ HINT: You can enable two-phase commit for extra safety with: SET citus.multi_sh
n_name | character(25) | not null
n_regionkey | integer | not null
n_comment | character varying(152) |
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON test_schema_support.nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;
@ -751,8 +739,6 @@ HINT: You can enable two-phase commit for extra safety with: SET citus.multi_sh
n_comment | character varying(152) |
Indexes:
"index1" btree (n_name)
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;
@ -781,8 +767,6 @@ HINT: You can enable two-phase commit for extra safety with: SET citus.multi_sh
n_name | character(25) | not null
n_regionkey | integer | not null
n_comment | character varying(152) |
Triggers:
"citus_truncate_trigger_on_test_schema_support.nation_hash" BEFORE TRUNCATE ON nation_hash FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
\c - - - :worker_1_port
\d test_schema_support.nation_hash_1190003;

15
src/test/regress/expected/multi_truncate.out
View File

@ -68,6 +68,11 @@ SELECT shardid FROM pg_dist_shard where logicalrelid = 'test_truncate_append'::r
---------
(0 rows)
-- command can not be run inside transaction
BEGIN; TRUNCATE TABLE test_truncate_append; COMMIT;
ERROR: DROP distributed table cannot run inside a transaction block
CONTEXT: SQL statement "SELECT master_drop_all_shards(TG_RELID, TG_TABLE_SCHEMA, TG_TABLE_NAME)"
PL/pgSQL function citus_truncate_trigger() line 13 at PERFORM
DROP TABLE test_truncate_append;
--
-- truncate for range distribution
@ -133,6 +138,11 @@ SELECT shardid FROM pg_dist_shard where logicalrelid = 'test_truncate_range'::re
1210005
(3 rows)
-- command can not be run inside transaction
BEGIN; TRUNCATE TABLE test_truncate_range; COMMIT;
ERROR: master_modify_multiple_shards cannot run inside a transaction block
CONTEXT: SQL statement "SELECT master_modify_multiple_shards(commandText)"
PL/pgSQL function citus_truncate_trigger() line 17 at PERFORM
DROP TABLE test_truncate_range;
--
-- truncate for hash distribution.
@ -216,4 +226,9 @@ SELECT shardid FROM pg_dist_shard where logicalrelid = 'test_truncate_hash'::reg
1210009
(4 rows)
-- command can not be run inside transaction
BEGIN; TRUNCATE TABLE test_truncate_hash; COMMIT;
ERROR: master_modify_multiple_shards cannot run inside a transaction block
CONTEXT: SQL statement "SELECT master_modify_multiple_shards(commandText)"
PL/pgSQL function citus_truncate_trigger() line 17 at PERFORM
DROP TABLE test_truncate_hash;

18
src/test/regress/output/multi_alter_table_statements.source
View File

@ -103,8 +103,6 @@ ORDER BY attnum;
int_column1 | integer | default 1
int_column2 | integer | default 2
null_column | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
SELECT float_column, count(*) FROM lineitem_alter GROUP BY float_column;
float_column | count
@ -166,8 +164,6 @@ ALTER TABLE lineitem_alter ALTER COLUMN int_column2 SET NOT NULL;
int_column1 | integer |
int_column2 | integer | not null default 2
null_column | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
-- Drop default so that NULLs will be inserted for this column
ALTER TABLE lineitem_alter ALTER COLUMN int_column2 DROP DEFAULT;
@ -203,8 +199,6 @@ ALTER TABLE lineitem_alter ALTER COLUMN int_column2 DROP NOT NULL;
int_column1 | integer |
int_column2 | integer |
null_column | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
-- \copy should succeed now
\copy 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 '|'
@ -248,8 +242,6 @@ ALTER TABLE lineitem_alter ALTER COLUMN int_column2 SET DATA TYPE FLOAT;
int_column1 | integer |
int_column2 | double precision |
null_column | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
SELECT int_column2, pg_typeof(int_column2), count(*) from lineitem_alter GROUP BY int_column2;
int_column2 | pg_typeof | count
@ -296,8 +288,6 @@ ALTER TABLE lineitem_alter DROP COLUMN IF EXISTS int_column2;
l_shipmode | character(10) | not null
l_comment | character varying(44) | not null
null_column | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
-- Verify that we can execute commands with multiple subcommands
ALTER TABLE lineitem_alter ADD COLUMN int_column1 INTEGER,
@ -325,8 +315,6 @@ ALTER TABLE lineitem_alter ADD COLUMN int_column1 INTEGER,
null_column | integer |
int_column1 | integer |
int_column2 | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
ALTER TABLE lineitem_alter ADD COLUMN int_column3 INTEGER,
ALTER COLUMN int_column1 SET STATISTICS 10;
@ -354,8 +342,6 @@ ALTER TABLE lineitem_alter DROP COLUMN int_column1, DROP COLUMN int_column2;
l_shipmode | character(10) | not null
l_comment | character varying(44) | not null
null_column | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
-- Verify that we cannot execute alter commands on the distribution column
ALTER TABLE lineitem_alter ALTER COLUMN l_orderkey DROP NOT NULL;
@ -425,8 +411,6 @@ ERROR: renaming distributed tables or their objects is currently unsupported
l_shipmode | character(10) | not null
l_comment | character varying(44) | not null
null_column | integer |
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
-- verify that non-propagated ddl commands are allowed inside a transaction block
SET citus.enable_ddl_propagation to false;
@ -481,8 +465,6 @@ COMMIT;
first | integer |
Indexes:
"temp_index_2" btree (l_orderkey)
Triggers:
"citus_truncate_trigger_on_public.lineitem_alter" BEFORE TRUNCATE ON lineitem_alter FOR EACH STATEMENT EXECUTE PROCEDURE citus_truncate_trigger()
ALTER TABLE lineitem_alter DROP COLUMN first;
DROP INDEX temp_index_2;

9
src/test/regress/sql/multi_truncate.sql
View File

@ -41,6 +41,9 @@ SELECT count(*) FROM test_truncate_append;
-- verify no shard exists anymore
SELECT shardid FROM pg_dist_shard where logicalrelid = 'test_truncate_append'::regclass;
-- command can not be run inside transaction
BEGIN; TRUNCATE TABLE test_truncate_append; COMMIT;
DROP TABLE test_truncate_append;
--
@ -86,6 +89,9 @@ SELECT count(*) FROM test_truncate_range;
-- verify 3 shards are still present
SELECT shardid FROM pg_dist_shard where logicalrelid = 'test_truncate_range'::regclass;
-- command can not be run inside transaction
BEGIN; TRUNCATE TABLE test_truncate_range; COMMIT;
DROP TABLE test_truncate_range;
@ -130,4 +136,7 @@ SELECT count(*) FROM test_truncate_hash;
-- verify 4 shards are still presents
SELECT shardid FROM pg_dist_shard where logicalrelid = 'test_truncate_hash'::regclass;
-- command can not be run inside transaction
BEGIN; TRUNCATE TABLE test_truncate_hash; COMMIT;
DROP TABLE test_truncate_hash;