Skip to content

Commit ce054a8

Browse files
alvherrealvherre
committed
Fix TRUNCATE .. CASCADE on partitions
When running TRUNCATE CASCADE on a child of a partitioned table referenced by another partitioned table, the truncate was not applied to partitions of the referencing table; this could leave rows violating the constraint in the referencing partitioned table. Repair by walking the pg_constraint chain all the way up to the topmost referencing table. Note: any partitioned tables containing FKs that reference other partitioned tables should be checked for possible violating rows, if TRUNCATE has occurred in partitions of the referenced table. Reported-by: Christophe Courtois Author: Jehan-Guillaume de Rorthais Discussion: https://postgr.es/m/20200204183906.115f693e@firost
1 parent 598b466 commit ce054a8

File tree

4 files changed

+173
-2
lines changed

4 files changed

+173
-2
lines changed

‎doc/src/sgml/ref/truncate.sgml

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,9 @@ TRUNCATE [ TABLE ] [ ONLY ] <replaceable class="parameter">name</replaceable> [
124124
option can be used to automatically include all dependent tables &mdash;
125125
but be very careful when using this option, or else you might lose data you
126126
did not intend to!
127+
Note in particular that when the table to be truncated is a partition,
128+
siblings partitions are left untouched, but cascading occurs to all
129+
referencing tables and all their partitions with no distinction.
127130
</para>
128131

129132
<para>

‎src/backend/catalog/heap.c

Lines changed: 82 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3399,16 +3399,27 @@ List *
33993399
heap_truncate_find_FKs(List *relationIds)
34003400
{
34013401
List *result = NIL;
3402+
List *oids = list_copy(relationIds);
3403+
List *parent_cons;
3404+
ListCell *cell;
3405+
ScanKeyData key;
34023406
Relation fkeyRel;
34033407
SysScanDesc fkeyScan;
34043408
HeapTuple tuple;
3409+
bool restart;
3410+
3411+
oids = list_copy(relationIds);
34053412

34063413
/*
34073414
* Must scan pg_constraint. Right now, it is a seqscan because there is
34083415
* no available index on confrelid.
34093416
*/
34103417
fkeyRel = table_open(ConstraintRelationId, AccessShareLock);
34113418

3419+
restart:
3420+
restart = false;
3421+
parent_cons = NIL;
3422+
34123423
fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
34133424
NULL, 0, NULL);
34143425

@@ -3421,16 +3432,85 @@ heap_truncate_find_FKs(List *relationIds)
34213432
continue;
34223433

34233434
/* Not referencing one of our list of tables */
3424-
if (!list_member_oid(relationIds, con->confrelid))
3435+
if (!list_member_oid(oids, con->confrelid))
34253436
continue;
34263437

3427-
/* Add referencer unless already in input or result list */
3438+
/*
3439+
* If this constraint has a parent constraint which we have not seen
3440+
* yet, keep track of it for the second loop, below. Tracking parent
3441+
* constraints allows us to climb up to the top-level level constraint
3442+
* and look for all possible relations referencing the partitioned
3443+
* table.
3444+
*/
3445+
if (OidIsValid(con->conparentid) &&
3446+
!list_member_oid(parent_cons, con->conparentid))
3447+
parent_cons = lappend_oid(parent_cons, con->conparentid);
3448+
3449+
/*
3450+
* Add referencer to result, unless already present in input or result
3451+
* list.
3452+
*/
34283453
if (!list_member_oid(relationIds, con->conrelid))
34293454
result = insert_ordered_unique_oid(result, con->conrelid);
34303455
}
34313456

34323457
systable_endscan(fkeyScan);
3458+
3459+
/*
3460+
* Process each parent constraint we found to add the list of referenced
3461+
* relations by them to the oids list. If we do add any new such
3462+
* relations, redo the first loop above. Also, if we see that the parent
3463+
* constraint in turn has a parent, add that so that we process all
3464+
* relations in a single additional pass.
3465+
*/
3466+
foreach(cell, parent_cons)
3467+
{
3468+
Oid parent = lfirst_oid(cell);
3469+
3470+
ScanKeyInit(&key,
3471+
Anum_pg_constraint_oid,
3472+
BTEqualStrategyNumber, F_OIDEQ,
3473+
ObjectIdGetDatum(parent));
3474+
3475+
fkeyScan = systable_beginscan(fkeyRel, ConstraintOidIndexId,
3476+
true, NULL, 1, &key);
3477+
3478+
tuple = systable_getnext(fkeyScan);
3479+
if (HeapTupleIsValid(tuple))
3480+
{
3481+
Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
3482+
3483+
/*
3484+
* pg_constraint rows always appear for partitioned hierarchies
3485+
* this way: on the each side of the constraint, one row appears
3486+
* for each partition that points to the top-most table on the
3487+
* other side.
3488+
*
3489+
* Because of this arrangement, we can correctly catch all
3490+
* relevant relations by adding to 'parent_cons' all rows with
3491+
* valid conparentid, and to the 'oids' list all rows with a
3492+
* zero conparentid. If any oids are added to 'oids', redo the
3493+
* first loop above by setting 'restart'.
3494+
*/
3495+
if (OidIsValid(con->conparentid))
3496+
parent_cons = list_append_unique_oid(parent_cons,
3497+
con->conparentid);
3498+
else if (!list_member_oid(oids, con->confrelid))
3499+
{
3500+
oids = lappend_oid(oids, con->confrelid);
3501+
restart = true;
3502+
}
3503+
}
3504+
3505+
systable_endscan(fkeyScan);
3506+
}
3507+
3508+
list_free(parent_cons);
3509+
if (restart)
3510+
goto restart;
3511+
34333512
table_close(fkeyRel, AccessShareLock);
3513+
list_free(oids);
34343514

34353515
return result;
34363516
}

‎src/test/regress/expected/truncate.out

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -542,3 +542,53 @@ SELECT * FROM tp_chk_data();
542542

543543
DROP TABLE truncprim, truncpart;
544544
DROP FUNCTION tp_ins_data(), tp_chk_data();
545+
-- test cascade when referencing a partitioned table
546+
CREATE TABLE trunc_a (a INT PRIMARY KEY) PARTITION BY RANGE (a);
547+
CREATE TABLE trunc_a1 PARTITION OF trunc_a FOR VALUES FROM (0) TO (10);
548+
CREATE TABLE trunc_a2 PARTITION OF trunc_a FOR VALUES FROM (10) TO (20)
549+
PARTITION BY RANGE (a);
550+
CREATE TABLE trunc_a21 PARTITION OF trunc_a2 FOR VALUES FROM (10) TO (12);
551+
CREATE TABLE trunc_a22 PARTITION OF trunc_a2 FOR VALUES FROM (12) TO (16);
552+
CREATE TABLE trunc_a2d PARTITION OF trunc_a2 DEFAULT;
553+
CREATE TABLE trunc_a3 PARTITION OF trunc_a FOR VALUES FROM (20) TO (30);
554+
INSERT INTO trunc_a VALUES (0), (5), (10), (15), (20), (25);
555+
-- truncate a partition cascading to a table
556+
CREATE TABLE ref_b (
557+
b INT PRIMARY KEY,
558+
a INT REFERENCES trunc_a(a) ON DELETE CASCADE
559+
);
560+
INSERT INTO ref_b VALUES (10, 0), (50, 5), (100, 10), (150, 15);
561+
TRUNCATE TABLE trunc_a1 CASCADE;
562+
NOTICE: truncate cascades to table "ref_b"
563+
SELECT a FROM ref_b;
564+
a
565+
---
566+
(0 rows)
567+
568+
DROP TABLE ref_b;
569+
-- truncate a partition cascading to a partitioned table
570+
CREATE TABLE ref_c (
571+
c INT PRIMARY KEY,
572+
a INT REFERENCES trunc_a(a) ON DELETE CASCADE
573+
) PARTITION BY RANGE (c);
574+
CREATE TABLE ref_c1 PARTITION OF ref_c FOR VALUES FROM (100) TO (200);
575+
CREATE TABLE ref_c2 PARTITION OF ref_c FOR VALUES FROM (200) TO (300);
576+
INSERT INTO ref_c VALUES (100, 10), (150, 15), (200, 20), (250, 25);
577+
TRUNCATE TABLE trunc_a21 CASCADE;
578+
NOTICE: truncate cascades to table "ref_c"
579+
NOTICE: truncate cascades to table "ref_c1"
580+
NOTICE: truncate cascades to table "ref_c2"
581+
SELECT a as "from table ref_c" FROM ref_c;
582+
from table ref_c
583+
------------------
584+
(0 rows)
585+
586+
SELECT a as "from table trunc_a" FROM trunc_a ORDER BY a;
587+
from table trunc_a
588+
--------------------
589+
15
590+
20
591+
25
592+
(3 rows)
593+
594+
DROP TABLE trunc_a, ref_c;

‎src/test/regress/sql/truncate.sql

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -289,3 +289,41 @@ TRUNCATE TABLE truncpart;
289289
SELECT * FROM tp_chk_data();
290290
DROP TABLE truncprim, truncpart;
291291
DROP FUNCTION tp_ins_data(), tp_chk_data();
292+
293+
-- test cascade when referencing a partitioned table
294+
CREATE TABLE trunc_a (a INT PRIMARY KEY) PARTITION BY RANGE (a);
295+
CREATE TABLE trunc_a1 PARTITION OF trunc_a FOR VALUES FROM (0) TO (10);
296+
CREATE TABLE trunc_a2 PARTITION OF trunc_a FOR VALUES FROM (10) TO (20)
297+
PARTITION BY RANGE (a);
298+
CREATE TABLE trunc_a21 PARTITION OF trunc_a2 FOR VALUES FROM (10) TO (12);
299+
CREATE TABLE trunc_a22 PARTITION OF trunc_a2 FOR VALUES FROM (12) TO (16);
300+
CREATE TABLE trunc_a2d PARTITION OF trunc_a2 DEFAULT;
301+
CREATE TABLE trunc_a3 PARTITION OF trunc_a FOR VALUES FROM (20) TO (30);
302+
INSERT INTO trunc_a VALUES (0), (5), (10), (15), (20), (25);
303+
304+
-- truncate a partition cascading to a table
305+
CREATE TABLE ref_b (
306+
b INT PRIMARY KEY,
307+
a INT REFERENCES trunc_a(a) ON DELETE CASCADE
308+
);
309+
INSERT INTO ref_b VALUES (10, 0), (50, 5), (100, 10), (150, 15);
310+
311+
TRUNCATE TABLE trunc_a1 CASCADE;
312+
SELECT a FROM ref_b;
313+
314+
DROP TABLE ref_b;
315+
316+
-- truncate a partition cascading to a partitioned table
317+
CREATE TABLE ref_c (
318+
c INT PRIMARY KEY,
319+
a INT REFERENCES trunc_a(a) ON DELETE CASCADE
320+
) PARTITION BY RANGE (c);
321+
CREATE TABLE ref_c1 PARTITION OF ref_c FOR VALUES FROM (100) TO (200);
322+
CREATE TABLE ref_c2 PARTITION OF ref_c FOR VALUES FROM (200) TO (300);
323+
INSERT INTO ref_c VALUES (100, 10), (150, 15), (200, 20), (250, 25);
324+
325+
TRUNCATE TABLE trunc_a21 CASCADE;
326+
SELECT a as "from table ref_c" FROM ref_c;
327+
SELECT a as "from table trunc_a" FROM trunc_a ORDER BY a;
328+
329+
DROP TABLE trunc_a, ref_c;

0 commit comments

Comments
 (0)