fix partition creation for tables with dropped columns

Author: funbringer

Makefile

@@ -37,6 +37,7 @@ REGRESS = pathman_array_qual \
 		  pathman_column_type \
 		  pathman_cte \
 		  pathman_domains \
+		  pathman_dropped_cols \
 		  pathman_expressions \
 		  pathman_foreign_keys \
 		  pathman_gaps \

expected/pathman_dropped_cols.out

@@ -0,0 +1,84 @@
+\set VERBOSITY terse
+SET search_path = 'public';
+CREATE EXTENSION pg_pathman;
+CREATE SCHEMA dropped_cols;
+/*
+ * we should be able to manage tables with dropped columns
+ */
+create table test_range(a int, b int, key int not null);
+alter table test_range drop column a;
+select create_range_partitions('test_range', 'key', 1, 10, 2);
+ create_range_partitions 
+-------------------------
+                       2
+(1 row)
+
+alter table test_range drop column b;
+select prepend_range_partition('test_range');
+ prepend_range_partition 
+-------------------------
+ test_range_3
+(1 row)
+
+select * from pathman_partition_list order by parent, partition;
+   parent   |  partition   | parttype | expr | range_min | range_max 
+------------+--------------+----------+------+-----------+-----------
+ test_range | test_range_1 |        2 | key  | 1         | 11
+ test_range | test_range_2 |        2 | key  | 11        | 21
+ test_range | test_range_3 |        2 | key  | -9        | 1
+(3 rows)
+
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_range_1_check';
+     pg_get_constraintdef      
+-------------------------------
+ CHECK (key >= 1 AND key < 11)
+(1 row)
+
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_range_3_check';
+           pg_get_constraintdef           
+------------------------------------------
+ CHECK (key >= '-9'::integer AND key < 1)
+(1 row)
+
+drop table test_range cascade;
+NOTICE:  drop cascades to 4 other objects
+create table test_hash(a int, b int, key int not null);
+alter table test_hash drop column a;
+select create_hash_partitions('test_hash', 'key', 3);
+ create_hash_partitions 
+------------------------
+                      3
+(1 row)
+
+alter table test_hash drop column b;
+create table test_dummy (like test_hash);
+select replace_hash_partition('test_hash_2', 'test_dummy', true);
+ replace_hash_partition 
+------------------------
+ test_dummy
+(1 row)
+
+select * from pathman_partition_list order by parent, partition;
+  parent   |  partition  | parttype | expr | range_min | range_max 
+-----------+-------------+----------+------+-----------+-----------
+ test_hash | test_hash_0 |        1 | key  |           | 
+ test_hash | test_hash_1 |        1 | key  |           | 
+ test_hash | test_dummy  |        1 | key  |           | 
+(3 rows)
+
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_hash_1_check';
+              pg_get_constraintdef               
+-------------------------------------------------
+ CHECK (get_hash_part_idx(hashint4(key), 3) = 1)
+(1 row)
+
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_dummy_check';
+              pg_get_constraintdef               
+-------------------------------------------------
+ CHECK (get_hash_part_idx(hashint4(key), 3) = 2)
+(1 row)
+
+drop table test_hash cascade;
+NOTICE:  drop cascades to 3 other objects
+DROP SCHEMA dropped_cols CASCADE;
+DROP EXTENSION pg_pathman;

sql/pathman_dropped_cols.sql

@@ -0,0 +1,43 @@
+\set VERBOSITY terse
+
+SET search_path = 'public';
+CREATE EXTENSION pg_pathman;
+CREATE SCHEMA dropped_cols;
+
+
+/*
+ * we should be able to manage tables with dropped columns
+ */
+
+create table test_range(a int, b int, key int not null);
+
+alter table test_range drop column a;
+select create_range_partitions('test_range', 'key', 1, 10, 2);
+
+alter table test_range drop column b;
+select prepend_range_partition('test_range');
+
+select * from pathman_partition_list order by parent, partition;
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_range_1_check';
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_range_3_check';
+
+drop table test_range cascade;
+
+
+create table test_hash(a int, b int, key int not null);
+
+alter table test_hash drop column a;
+select create_hash_partitions('test_hash', 'key', 3);
+
+alter table test_hash drop column b;
+create table test_dummy (like test_hash);
+select replace_hash_partition('test_hash_2', 'test_dummy', true);
+
+select * from pathman_partition_list order by parent, partition;
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_hash_1_check';
+select pg_get_constraintdef(oid, true) from pg_constraint where conname = 'pathman_test_dummy_check';
+drop table test_hash cascade;
+
+
+DROP SCHEMA dropped_cols CASCADE;
+DROP EXTENSION pg_pathman;

src/include/utils.h

@@ -24,7 +24,7 @@
 bool clause_contains_params(Node *clause);
 bool is_date_type_internal(Oid typid);
 bool check_security_policy_internal(Oid relid, Oid role);
-bool match_expr_to_operand(Node *expr, Node *operand);
+bool match_expr_to_operand(const Node *expr, const Node *operand);
 
 /*
  * Misc.

src/init.c

@@ -1165,7 +1165,6 @@ validate_hash_constraint(const Expr *expr,
 		Node   *first = linitial(get_hash_expr->args);	/* arg #1: TYPE_HASH_PROC(EXPRESSION) */
 		Node   *second = lsecond(get_hash_expr->args);	/* arg #2: PARTITIONS_COUNT */
 		Const  *cur_partition_idx;						/* hash value for this partition */
-		Node   *hash_arg;
 
 		if (!IsA(first, FuncExpr) || !IsA(second, Const))
 			return false;
@@ -1180,13 +1179,6 @@ validate_hash_constraint(const Expr *expr,
 		if (list_length(type_hash_proc_expr->args) != 1)
 			return false;
 
-		/* Extract arg of TYPE_HASH_PROC() */
-		hash_arg = (Node *) linitial(type_hash_proc_expr->args);
-
-		/* Check arg of TYPE_HASH_PROC() */
-		if (!match_expr_to_operand(prel->expr, hash_arg))
-			return false;
-
 		/* Check that PARTITIONS_COUNT is equal to total amount of partitions */
 		if (DatumGetUInt32(((Const *) second)->constvalue) != PrelChildrenCount(prel))
 			return false;

src/relation_info.c

@@ -976,7 +976,7 @@ delay_invalidation_vague_rel(Oid vague_rel)
 /* Finish all pending invalidation jobs if possible */
 void
 finish_delayed_invalidation(void)
-{	
+{
 	/* Exit early if there's nothing to do */
 	if (delayed_invalidation_whole_cache  == false &&
 		delayed_invalidation_parent_rels == NIL &&

src/utils.c

@@ -37,6 +37,21 @@
 #include "utils/regproc.h"
 #endif
 
+static const Node *
+drop_irrelevant_expr_wrappers(const Node *expr)
+{
+	switch (nodeTag(expr))
+	{
+		/* Strip relabeling */
+		case T_RelabelType:
+			return (const Node *) ((const RelabelType *) expr)->arg;
+
+		/* no special actions required */
+		default:
+			return expr;
+	}
+}
+
 static bool
 clause_contains_params_walker(Node *node, void *context)
 {
@@ -110,14 +125,10 @@ check_security_policy_internal(Oid relid, Oid role)
 
 /* Compare clause operand with expression */
 bool
-match_expr_to_operand(Node *expr, Node *operand)
+match_expr_to_operand(const Node *expr, const Node *operand)
 {
-	/* Strip relabeling for both operand and expr */
-	if (operand && IsA(operand, RelabelType))
-		operand = (Node *) ((RelabelType *) operand)->arg;
-
-	if (expr && IsA(expr, RelabelType))
-		expr = (Node *) ((RelabelType *) expr)->arg;
+	expr = drop_irrelevant_expr_wrappers(expr);
+	operand = drop_irrelevant_expr_wrappers(operand);
 
 	/* compare expressions and return result right away */
 	return equal(expr, operand);