From d78f68a6a33b9b14f7840751e307bb2b90e7640e Mon Sep 17 00:00:00 2001
From: Jehan-Guillaume de Rorthais <jgdr@dalibo.com>
Date: Tue, 4 Feb 2020 18:05:17 +0100
Subject: [PATCH] Fix TRUNCATE on a partition to apply CASCADE to partitioned
 table.

Previously, when running TRUNCATE CASCADE on a child of a
partitioned table referenced by another partitioned table, the
truncate was not applied to referencing children. This leaves a
FK constraint violation in the referencing partitioned table.

This commit fixes that bug.

Reported-by: Christophe Courtois
Author: Jehan-Guillaume de Rorthais
---
 src/backend/catalog/heap.c             | 78 +++++++++++++++++++++-----
 src/test/regress/expected/truncate.out | 45 +++++++++++++++
 src/test/regress/sql/truncate.sql      | 34 +++++++++++
 3 files changed, 142 insertions(+), 15 deletions(-)

diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 046b3d37ce..65ff126c64 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -3396,39 +3396,87 @@ List *
 heap_truncate_find_FKs(List *relationIds)
 {
 	List	   *result = NIL;
+	List	   *oids = NIL;
+	List	   *parent_cons = NIL;
+	ListCell   *cell;
+	ScanKeyData key;
 	Relation	fkeyRel;
 	SysScanDesc fkeyScan;
 	HeapTuple	tuple;
 
+	oids = list_copy(relationIds);
+
 	/*
 	 * Must scan pg_constraint.  Right now, it is a seqscan because there is
 	 * no available index on confrelid.
 	 */
 	fkeyRel = table_open(ConstraintRelationId, AccessShareLock);
 
-	fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
-								  NULL, 0, NULL);
+	for (;;) {
+		int num_oids = oids->length;
 
-	while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
-	{
-		Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
+		fkeyScan = systable_beginscan(fkeyRel, InvalidOid, false,
+									  NULL, 0, NULL);
 
-		/* Not a foreign key */
-		if (con->contype != CONSTRAINT_FOREIGN)
-			continue;
+		while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
+		{
+			Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
 
-		/* Not referencing one of our list of tables */
-		if (!list_member_oid(relationIds, con->confrelid))
-			continue;
+			/* Not a foreign key */
+			if (con->contype != CONSTRAINT_FOREIGN)
+				continue;
+
+			/* Not referencing one of our list of tables */
+			if (!list_member_oid(oids, con->confrelid))
+				continue;
+
+			/* have a parent constraint */
+			if (OidIsValid(con->conparentid) &&
+				!list_member_oid(parent_cons, con->conparentid))
+				parent_cons = lappend_oid(parent_cons, con->conparentid);
+
+			/* Add referencer to result, unless present in input list */
+			if (!list_member_oid(relationIds, con->conrelid) )
+				result = lappend_oid(result, con->conrelid);
+		}
 
-		/* Add referencer to result, unless present in input list */
-		if (!list_member_oid(relationIds, con->conrelid))
-			result = lappend_oid(result, con->conrelid);
+		systable_endscan(fkeyScan);
+
+		if (parent_cons == NIL)
+			break;
+
+		foreach(cell, parent_cons)
+		{
+			Oid		parent = lfirst_oid(cell);
+
+			ScanKeyInit(&key,
+					Anum_pg_constraint_conparentid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(parent));
+
+			fkeyScan = systable_beginscan(fkeyRel, ConstraintParentIndexId,
+										  true, NULL, 1, &key);
+
+			while (HeapTupleIsValid(tuple = systable_getnext(fkeyScan)))
+			{
+				Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
+				/* Add referencer to result, unless present in input list */
+				if (!list_member_oid(oids, con->confrelid))
+					oids = lappend_oid(oids, con->confrelid);
+			}
+
+			systable_endscan(fkeyScan);
+		}
+
+		if (num_oids == oids->length)
+			break;
 	}
 
-	systable_endscan(fkeyScan);
 	table_close(fkeyRel, AccessShareLock);
 
+	list_free(parent_cons);
+	list_free(oids);
+
 	/* Now sort and de-duplicate the result list */
 	list_sort(result, list_oid_cmp);
 	list_deduplicate_oid(result);
diff --git a/src/test/regress/expected/truncate.out b/src/test/regress/expected/truncate.out
index cc68274dca..b07c429158 100644
--- a/src/test/regress/expected/truncate.out
+++ b/src/test/regress/expected/truncate.out
@@ -542,3 +542,48 @@ SELECT * FROM tp_chk_data();
 
 DROP TABLE truncprim, truncpart;
 DROP FUNCTION tp_ins_data(), tp_chk_data();
+-- test cascade when referencing a partitioned table
+CREATE TABLE trunc_a (a INT PRIMARY KEY) PARTITION BY RANGE (a);
+CREATE TABLE trunc_a1 PARTITION OF trunc_a FOR VALUES FROM (0) TO (10);
+CREATE TABLE trunc_a2 PARTITION OF trunc_a FOR VALUES FROM (10) TO (20);
+CREATE TABLE trunc_a3 PARTITION OF trunc_a FOR VALUES FROM (20) TO (30);
+INSERT INTO trunc_a VALUES (0), (5), (10), (15), (20), (25);
+-- truncate a partition cascading to a table
+CREATE TABLE ref_b (
+    b INT PRIMARY KEY,
+    a INT REFERENCES trunc_a(a) ON DELETE CASCADE
+);
+INSERT INTO ref_b VALUES (10, 0), (50, 5), (100, 10), (150, 15);
+TRUNCATE TABLE trunc_a1 CASCADE;
+NOTICE:  truncate cascades to table "ref_b"
+SELECT a FROM ref_b;
+ a 
+---
+(0 rows)
+
+DROP TABLE ref_b;
+-- truncate a partition cascading to a partitioned table
+CREATE TABLE ref_c (
+    c INT PRIMARY KEY,
+    a INT REFERENCES trunc_a(a) ON DELETE CASCADE
+) PARTITION BY RANGE (c);
+CREATE TABLE ref_c1 PARTITION OF ref_c FOR VALUES FROM (100) TO (200);
+CREATE TABLE ref_c2 PARTITION OF ref_c FOR VALUES FROM (200) TO (300);
+INSERT INTO ref_c VALUES (100, 10), (150, 15), (200, 20), (250, 25);
+TRUNCATE TABLE trunc_a2 CASCADE;
+NOTICE:  truncate cascades to table "ref_c"
+NOTICE:  truncate cascades to table "ref_c1"
+NOTICE:  truncate cascades to table "ref_c2"
+SELECT a FROM ref_c;
+ a 
+---
+(0 rows)
+
+SELECT a FROM trunc_a ORDER BY a;
+ a  
+----
+ 20
+ 25
+(2 rows)
+
+DROP TABLE trunc_a, ref_c;
diff --git a/src/test/regress/sql/truncate.sql b/src/test/regress/sql/truncate.sql
index 28395e82bf..b00311a72f 100644
--- a/src/test/regress/sql/truncate.sql
+++ b/src/test/regress/sql/truncate.sql
@@ -289,3 +289,37 @@ TRUNCATE TABLE truncpart;
 SELECT * FROM tp_chk_data();
 DROP TABLE truncprim, truncpart;
 DROP FUNCTION tp_ins_data(), tp_chk_data();
+
+-- test cascade when referencing a partitioned table
+CREATE TABLE trunc_a (a INT PRIMARY KEY) PARTITION BY RANGE (a);
+CREATE TABLE trunc_a1 PARTITION OF trunc_a FOR VALUES FROM (0) TO (10);
+CREATE TABLE trunc_a2 PARTITION OF trunc_a FOR VALUES FROM (10) TO (20);
+CREATE TABLE trunc_a3 PARTITION OF trunc_a FOR VALUES FROM (20) TO (30);
+INSERT INTO trunc_a VALUES (0), (5), (10), (15), (20), (25);
+
+-- truncate a partition cascading to a table
+CREATE TABLE ref_b (
+    b INT PRIMARY KEY,
+    a INT REFERENCES trunc_a(a) ON DELETE CASCADE
+);
+INSERT INTO ref_b VALUES (10, 0), (50, 5), (100, 10), (150, 15);
+
+TRUNCATE TABLE trunc_a1 CASCADE;
+SELECT a FROM ref_b;
+
+DROP TABLE ref_b;
+
+-- truncate a partition cascading to a partitioned table
+CREATE TABLE ref_c (
+    c INT PRIMARY KEY,
+    a INT REFERENCES trunc_a(a) ON DELETE CASCADE
+) PARTITION BY RANGE (c);
+CREATE TABLE ref_c1 PARTITION OF ref_c FOR VALUES FROM (100) TO (200);
+CREATE TABLE ref_c2 PARTITION OF ref_c FOR VALUES FROM (200) TO (300);
+INSERT INTO ref_c VALUES (100, 10), (150, 15), (200, 20), (250, 25);
+
+TRUNCATE TABLE trunc_a2 CASCADE;
+SELECT a FROM ref_c;
+SELECT a FROM trunc_a ORDER BY a;
+
+DROP TABLE trunc_a, ref_c;
-- 
2.20.1

