default_partition_v4.patch

diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c
index ab891f6..f5a9f44 100644
--- a/src/backend/catalog/partition.c
+++ b/src/backend/catalog/partition.c
@@ -34,6 +34,7 @@
 #include "nodes/nodeFuncs.h"
 #include "nodes/parsenodes.h"
 #include "optimizer/clauses.h"
+#include "optimizer/prep.h"
 #include "optimizer/planmain.h"
 #include "optimizer/var.h"
 #include "rewrite/rewriteManip.h"
@@ -90,6 +91,10 @@ typedef struct PartitionBoundInfoData
 								 * for range partitioned tables */
 	int			null_index;		/* Index of the null-accepting partition; -1
 								 * for range partitioned tables */
+	bool		has_def;		/* Is there a default partition? Currently false
+								 * for a range partitioned table */
+	int			def_index;		/* Index of the default list partition. -1 for
+								 * range partitioned tables */
 } PartitionBoundInfoData;
 
 /*
@@ -118,7 +123,8 @@ static int32 qsort_partition_list_value_cmp(const void *a, const void *b,
 static int32 qsort_partition_rbound_cmp(const void *a, const void *b,
 						   void *arg);
 
-static List *get_qual_for_list(PartitionKey key, PartitionBoundSpec *spec);
+static List *get_qual_for_list(PartitionKey key, PartitionBoundSpec *spec,
+								bool is_def, List *boundspecs);
 static List *get_qual_for_range(PartitionKey key, PartitionBoundSpec *spec);
 static Oid get_partition_operator(PartitionKey key, int col,
 					   StrategyNumber strategy, bool *need_relabel);
@@ -166,6 +172,8 @@ RelationBuildPartitionDesc(Relation rel)
 	/* List partitioning specific */
 	PartitionListValue **all_values = NULL;
 	bool		found_null = false;
+	bool		found_def = false;
+	int			def_index = -1;
 	int			null_index = -1;
 
 	/* Range partitioning specific */
@@ -249,9 +257,16 @@ RelationBuildPartitionDesc(Relation rel)
 
 				foreach(c, spec->listdatums)
 				{
+					Node *value = lfirst(c);
 					Const	   *val = lfirst(c);
 					PartitionListValue *list_value = NULL;
 
+					if (IsA(value, DefElem))
+					{
+						found_def = true;
+						def_index = i;
+						continue;
+					}
 					if (!val->constisnull)
 					{
 						list_value = (PartitionListValue *)
@@ -459,6 +474,7 @@ RelationBuildPartitionDesc(Relation rel)
 			case PARTITION_STRATEGY_LIST:
 				{
 					boundinfo->has_null = found_null;
+					boundinfo->has_def = found_def;
 					boundinfo->indexes = (int *) palloc(ndatums * sizeof(int));
 
 					/*
@@ -496,6 +512,11 @@ RelationBuildPartitionDesc(Relation rel)
 						if (mapping[null_index] == -1)
 							mapping[null_index] = next_index++;
 					}
+					if (found_def)
+					{
+						if (mapping[def_index] == -1)
+							mapping[def_index] = next_index++;
+					}
 
 					/* All partition must now have a valid mapping */
 					Assert(next_index == nparts);
@@ -504,6 +525,11 @@ RelationBuildPartitionDesc(Relation rel)
 						boundinfo->null_index = mapping[null_index];
 					else
 						boundinfo->null_index = -1;
+
+					if (found_def)
+						boundinfo->def_index = mapping[def_index];
+					else
+						boundinfo->def_index = -1;
 					break;
 				}
 
@@ -671,6 +697,7 @@ check_new_partition_bound(char *relname, Relation parent, Node *bound)
 	PartitionBoundSpec *spec = (PartitionBoundSpec *) bound;
 	PartitionKey key = RelationGetPartitionKey(parent);
 	PartitionDesc partdesc = RelationGetPartitionDesc(parent);
+	PartitionBoundInfo boundinfo;
 	ParseState *pstate = make_parsestate(NULL);
 	int			with = -1;
 	bool		overlap = false;
@@ -679,6 +706,8 @@ check_new_partition_bound(char *relname, Relation parent, Node *bound)
 	{
 		case PARTITION_STRATEGY_LIST:
 			{
+				boundinfo = partdesc->boundinfo;
+
 				Assert(spec->strategy == PARTITION_STRATEGY_LIST);
 
 				if (partdesc->nparts > 0)
@@ -688,7 +717,8 @@ check_new_partition_bound(char *relname, Relation parent, Node *bound)
 
 					Assert(boundinfo &&
 						   boundinfo->strategy == PARTITION_STRATEGY_LIST &&
-						   (boundinfo->ndatums > 0 || boundinfo->has_null));
+						   (boundinfo->ndatums > 0 || boundinfo->has_null
+							|| boundinfo->has_def));
 
 					foreach(cell, spec->listdatums)
 					{
@@ -836,6 +866,76 @@ check_new_partition_bound(char *relname, Relation parent, Node *bound)
 						relname, get_rel_name(partdesc->oids[with])),
 				 parser_errposition(pstate, spec->location)));
 	}
+
+	/*
+	 * When adding a list partition after default partition, scan the
+	 * default partiton for rows satisfying the new partition
+	 * constraint. If found dont allow addition of a new partition.
+	 * Otherwise continue with the creation of new  partition.
+	 */
+	if (spec->strategy == PARTITION_STRATEGY_LIST && partdesc->nparts > 0
+		&& boundinfo->has_def)
+	{
+		List	       *partConstraint = NIL;
+		ExprContext    *econtext;
+		EState	       *estate;
+		Relation	    defrel;
+		HeapScanDesc    scan;
+		HeapTuple	    tuple;
+		ExprState	   *partqualstate = NULL;
+		Snapshot	    snapshot;
+		Oid			    defid;
+		MemoryContext   oldCxt;
+		TupleTableSlot *tupslot;
+		TupleDesc	    tupdesc;
+
+		partConstraint = generate_qual_for_defaultpart(parent, bound, &defid);
+		partConstraint = (List *) eval_const_expressions(NULL,
+													(Node *) partConstraint);
+		partConstraint = (List *) canonicalize_qual((Expr *) partConstraint);
+		partConstraint = list_make1(make_ands_explicit(partConstraint));
+
+		/*
+		 * Generate the constraint and default execution states
+		 */
+		estate = CreateExecutorState();
+
+		defrel = heap_open(defid, AccessShareLock);
+		tupdesc = CreateTupleDescCopy(RelationGetDescr(defrel));
+
+		/* Build expression execution states for partition check quals */
+		partqualstate = ExecPrepareExpr((Expr *) partConstraint,
+						estate);
+
+		econtext = GetPerTupleExprContext(estate);
+		snapshot = RegisterSnapshot(GetLatestSnapshot());
+		scan = heap_beginscan(defrel, snapshot, 0, NULL);
+		tupslot = MakeSingleTupleTableSlot(tupdesc);
+
+		/*
+		 * Switch to per-tuple memory context and reset it for each tuple
+		 * produced, so we don't leak memory.
+		 */
+		oldCxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+
+		while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+		{
+			ExecStoreTuple(tuple, tupslot, InvalidBuffer, false);
+			econtext->ecxt_scantuple = tupslot;
+			if (partqualstate && !ExecCheck(partqualstate, econtext))
+				ereport(ERROR,
+						(errcode(ERRCODE_CHECK_VIOLATION),
+				errmsg("new default partition constraint is violated by some row")));
+			ResetExprContext(econtext);
+			CHECK_FOR_INTERRUPTS();
+		}
+		MemoryContextSwitchTo(oldCxt);
+		heap_endscan(scan);
+		UnregisterSnapshot(snapshot);
+		heap_close(defrel, AccessShareLock);
+		ExecDropSingleTupleTableSlot(tupslot);
+	}
+
 }
 
 /*
@@ -884,6 +984,90 @@ get_partition_parent(Oid relid)
 }
 
 /*
+ * Return a list of executable expressions as new partition constraint
+ * for default partition while adding a new partition after default
+ */
+List *
+generate_qual_for_defaultpart(Relation parent, Node *bound, Oid * defid)
+{
+	PartitionKey key = RelationGetPartitionKey(parent);
+	PartitionBoundSpec *spec;
+	List	   *partConstraint = NIL;
+	List	   *inhoids;
+	ListCell   *cell2;
+	ListCell   *cell4;
+	List       *boundspecs = NIL;
+	bool		is_def = true;
+
+	spec = (PartitionBoundSpec *) bound;
+
+	inhoids = find_inheritance_children(RelationGetRelid(parent), NoLock);
+
+	foreach(cell2, inhoids)
+	{
+		PartitionBoundSpec *bspec;
+		Oid			inhrelid = lfirst_oid(cell2);
+		HeapTuple	tuple;
+		Datum		datum;
+		bool		isnull;
+		Node	   *boundspec;
+		bool		def_elem = false;
+		ListCell   *cell1;
+		ListCell   *cell3;
+
+		tuple = SearchSysCache1(RELOID, inhrelid);
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for relation %u", inhrelid);
+
+		/*
+		 * It is possible that the pg_class tuple of a partition has not been
+		 * updated yet to set its relpartbound field.  The only case where
+		 * this happens is when we open the parent relation to check using its
+		 * partition descriptor that a new partition's bound does not overlap
+		 * some existing partition.
+		 */
+		if (!((Form_pg_class) GETSTRUCT(tuple))->relispartition)
+		{
+			ReleaseSysCache(tuple);
+			continue;
+		}
+
+		datum = SysCacheGetAttr(RELOID, tuple,
+								Anum_pg_class_relpartbound,
+								&isnull);
+		Assert(!isnull);
+		boundspec = (Node *) stringToNode(TextDatumGetCString(datum));
+		bspec = (PartitionBoundSpec *)boundspec;
+		foreach(cell1, bspec->listdatums)
+		{
+			Node *value = lfirst(cell1);
+			if (IsA(value, DefElem))
+			{
+				def_elem = true;
+				*defid = inhrelid;
+			}
+		}
+		if (def_elem)
+		{
+			ReleaseSysCache(tuple);
+			continue;
+		}
+		foreach(cell3, bspec->listdatums)
+		{
+			Node *value = lfirst(cell3);
+			boundspecs = lappend(boundspecs, value);
+		}
+		ReleaseSysCache(tuple);
+	}
+	foreach(cell4, spec->listdatums)
+	{
+		Node *value = lfirst(cell4);
+		boundspecs = lappend(boundspecs, value);
+	}
+	partConstraint = get_qual_for_list(key, spec, is_def, boundspecs);
+	return partConstraint;
+}
+/*
  * get_qual_from_partbound
  *		Given a parser node for partition bound, return the list of executable
  *		expressions as partition constraint
@@ -894,6 +1078,12 @@ get_qual_from_partbound(Relation rel, Relation parent, Node *bound)
 	PartitionBoundSpec *spec = (PartitionBoundSpec *) bound;
 	PartitionKey key = RelationGetPartitionKey(parent);
 	List	   *my_qual = NIL;
+	List	   *inhoids,
+			   *partoids;
+	List	   *boundspecs = NIL;
+	bool		is_def = false;
+	ListCell   *cell;
+	ListCell   *cell2;
 
 	Assert(key != NULL);
 
@@ -901,9 +1091,78 @@ get_qual_from_partbound(Relation rel, Relation parent, Node *bound)
 	{
 		case PARTITION_STRATEGY_LIST:
 			Assert(spec->strategy == PARTITION_STRATEGY_LIST);
-			my_qual = get_qual_for_list(key, spec);
-			break;
+			foreach(cell, spec->listdatums)
+			{
+				Node *value = lfirst(cell);
+				if (IsA(value, DefElem))
+					is_def = true;
+			}
+			if (is_def)
+			{
+				/* Collect bound spec nodes in a list. This is done if the partition is
+				 * a default partition. In case of default partition, constraint is formed
+				 * by performing <> operation over the partition constraints of the
+				 * existing partitions.
+				 */
+				inhoids = find_inheritance_children(RelationGetRelid(parent), NoLock);
+				partoids = NIL;
+				foreach(cell2, inhoids)
+				{
+					Oid			inhrelid = lfirst_oid(cell2);
+					HeapTuple	tuple;
+					Datum		datum;
+					bool		isnull;
+					Node	   *boundspec;
+					bool		def_elem = false;
+					PartitionBoundSpec *bspec;
+					ListCell *cell1;
+					ListCell *cell3;
+
+					tuple = SearchSysCache1(RELOID, inhrelid);
+					if (!HeapTupleIsValid(tuple))
+						elog(ERROR, "cache lookup failed for relation %u", inhrelid);
+
+					/*
+					 * It is possible that the pg_class tuple of a partition has not been
+					 * updated yet to set its relpartbound field.  The only case where
+					 * this happens is when we open the parent relation to check using its
+					 * partition descriptor that a new partition's bound does not overlap
+					 * some existing partition.
+					 */
+					if (!((Form_pg_class) GETSTRUCT(tuple))->relispartition)
+					{
+						ReleaseSysCache(tuple);
+						continue;
+					}
 
+					datum = SysCacheGetAttr(RELOID, tuple,
+											Anum_pg_class_relpartbound,
+											&isnull);
+					Assert(!isnull);
+					boundspec = (Node *) stringToNode(TextDatumGetCString(datum));
+					bspec = (PartitionBoundSpec *)boundspec;
+					foreach(cell1, bspec->listdatums)
+					{
+						Node *value = lfirst(cell1);
+						if (IsA(value, DefElem))
+							def_elem = true;
+					}
+					if (def_elem)
+					{
+						ReleaseSysCache(tuple);
+						continue;
+					}
+					foreach(cell3, bspec->listdatums)
+					{
+						Node *value = lfirst(cell3);
+						boundspecs = lappend(boundspecs, value);
+					}
+					partoids = lappend_oid(partoids, inhrelid);
+					ReleaseSysCache(tuple);
+				}
+			}
+			my_qual = get_qual_for_list(key, spec, is_def, boundspecs);
+			break;
 		case PARTITION_STRATEGY_RANGE:
 			Assert(spec->strategy == PARTITION_STRATEGY_RANGE);
 			my_qual = get_qual_for_range(key, spec);
@@ -1151,7 +1410,8 @@ RelationGetPartitionDispatchInfo(Relation rel, int lockmode,
  * Returns a list of expressions to use as a list partition's constraint.
  */
 static List *
-get_qual_for_list(PartitionKey key, PartitionBoundSpec *spec)
+get_qual_for_list(PartitionKey key, PartitionBoundSpec *spec, bool is_def,
+					List *boundspecs)
 {
 	List	   *result;
 	ArrayExpr  *arr;
@@ -1229,7 +1489,10 @@ get_qual_for_list(PartitionKey key, PartitionBoundSpec *spec)
 		: key->parttypid[0];
 	arr->array_collid = key->parttypcoll[0];
 	arr->element_typeid = key->parttypid[0];
-	arr->elements = spec->listdatums;
+	if (is_def)
+		arr->elements = boundspecs;
+	else
+		arr->elements = spec->listdatums;
 	arr->multidims = false;
 	arr->location = -1;
 
@@ -1243,15 +1506,28 @@ get_qual_for_list(PartitionKey key, PartitionBoundSpec *spec)
 										  key->partcollation[0],
 										  COERCE_EXPLICIT_CAST);
 
+	/* Build leftop <> ALL(rightop). This is for default partition */
+	if (is_def)
+	{
+		opexpr = makeNode(ScalarArrayOpExpr);
+		opexpr->opno = 518;
+		opexpr->opfuncid = get_opcode(opexpr->opno);
+		opexpr->useOr = false;
+		opexpr->inputcollid = key->partcollation[0];
+		opexpr->args = list_make2(keyCol, arr);
+		opexpr->location = -1;
+	}
 	/* Build leftop = ANY (rightop) */
-	opexpr = makeNode(ScalarArrayOpExpr);
-	opexpr->opno = operoid;
-	opexpr->opfuncid = get_opcode(operoid);
-	opexpr->useOr = true;
-	opexpr->inputcollid = key->partcollation[0];
-	opexpr->args = list_make2(keyCol, arr);
-	opexpr->location = -1;
-
+	else
+	{
+		opexpr = makeNode(ScalarArrayOpExpr);
+		opexpr->opno = operoid;
+		opexpr->opfuncid = get_opcode(operoid);
+		opexpr->useOr = true;
+		opexpr->inputcollid = key->partcollation[0];
+		opexpr->args = list_make2(keyCol, arr);
+		opexpr->location = -1;
+	}
 	if (nulltest1)
 		result = list_make2(nulltest1, opexpr);
 	else if (nulltest2)
@@ -1778,6 +2054,13 @@ get_partition_for_tuple(PartitionDispatch *pd,
 			result = -1;
 			*failed_at = parent;
 			*failed_slot = slot;
+			/*
+			 * If partitioned table has a default partition, return
+			 * its sequence number
+			 */
+			if (partdesc->boundinfo->has_def && key->strategy
+				== PARTITION_STRATEGY_LIST)
+				result = parent->indexes[partdesc->boundinfo->def_index];
 			break;
 		}
 		else if (parent->indexes[cur_index] >= 0)
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index d418d56..69268b1 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -750,6 +750,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		ParseState *pstate;
 		Oid			parentId = linitial_oid(inheritOids);
 		Relation	parent;
+		PartitionDesc partdesc;
 
 		/* Already have strong enough lock on the parent */
 		parent = heap_open(parentId, NoLock);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 9d53a29..7114d7f 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -2596,6 +2596,7 @@ partbound_datum:
 			Sconst			{ $$ = makeStringConst($1, @1); }
 			| NumericOnly	{ $$ = makeAConst($1, @1); }
 			| NULL_P		{ $$ = makeNullAConst(@1); }
+			| DEFAULT  { $$ = (Node *)makeDefElem("DEFAULT", NULL, @1); }
 		;
 
 partbound_datum_list:
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 1ae43dc..17605c3 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -3088,47 +3088,50 @@ transformPartitionBound(ParseState *pstate, Relation parent, Node *bound)
 		result_spec->listdatums = NIL;
 		foreach(cell, spec->listdatums)
 		{
-			A_Const    *con = (A_Const *) lfirst(cell);
-			Node	   *value;
-			ListCell   *cell2;
-			bool		duplicate;
-
-			value = (Node *) make_const(pstate, &con->val, con->location);
-			value = coerce_to_target_type(pstate,
-										  value, exprType(value),
-										  get_partition_col_typid(key, 0),
-										  get_partition_col_typmod(key, 0),
-										  COERCION_ASSIGNMENT,
-										  COERCE_IMPLICIT_CAST,
-										  -1);
-
-			if (value == NULL)
-				ereport(ERROR,
-						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg("specified value cannot be cast to type \"%s\" of column \"%s\"",
-							 format_type_be(get_partition_col_typid(key, 0)),
-								colname),
-						 parser_errposition(pstate,
-											exprLocation((Node *) con))));
-
-			/* Simplify the expression */
-			value = (Node *) expression_planner((Expr *) value);
-
-			/* Don't add to the result if the value is a duplicate */
-			duplicate = false;
-			foreach(cell2, result_spec->listdatums)
+			Node *value = lfirst(cell);
+			/* Perform the transformation only for non default partition */
+			if (!IsA(value, DefElem))
 			{
-				Const	   *value2 = (Const *) lfirst(cell2);
+				A_Const    *con = (A_Const *) lfirst(cell);
+				ListCell   *cell2;
+				bool		duplicate;
+
+				value = (Node *) make_const(pstate, &con->val, con->location);
+				value = coerce_to_target_type(pstate,
+											value, exprType(value),
+											get_partition_col_typid(key, 0),
+											get_partition_col_typmod(key, 0),
+											COERCION_ASSIGNMENT,
+											COERCE_IMPLICIT_CAST,
+											-1);
+
+				if (value == NULL)
+					ereport(ERROR,
+							(errcode(ERRCODE_DATATYPE_MISMATCH),
+							errmsg("specified value cannot be cast to type \"%s\" of column \"%s\"",
+								format_type_be(get_partition_col_typid(key, 0)),
+									colname),
+							parser_errposition(pstate,
+												exprLocation((Node *) con))));
 
-				if (equal(value, value2))
+				/* Simplify the expression */
+				value = (Node *) expression_planner((Expr *) value);
+
+				/* Don't add to the result if the value is a duplicate */
+				duplicate = false;
+				foreach(cell2, result_spec->listdatums)
 				{
-					duplicate = true;
-					break;
+					Const	   *value2 = (Const *) lfirst(cell2);
+
+					if (equal(value, value2))
+					{
+						duplicate = true;
+						break;
+					}
 				}
+				if (duplicate)
+					continue;
 			}
-			if (duplicate)
-				continue;
-
 			result_spec->listdatums = lappend(result_spec->listdatums,
 											  value);
 		}
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 0c1a201..5732bbc 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -8559,8 +8559,14 @@ get_rule_expr(Node *node, deparse_context *context,
 						sep = "";
 						foreach(cell, spec->listdatums)
 						{
+							Node *value = lfirst(cell);
 							Const	   *val = lfirst(cell);
 
+							if (IsA(value, DefElem))
+							{
+								appendStringInfoString(buf, "DEFAULT");
+								continue;
+							}
 							appendStringInfoString(buf, sep);
 							get_const_expr(val, context, -1);
 							sep = ", ";
diff --git a/src/include/catalog/partition.h b/src/include/catalog/partition.h
index 421644c..9c92a8a 100644
--- a/src/include/catalog/partition.h
+++ b/src/include/catalog/partition.h
@@ -77,6 +77,7 @@ extern bool partition_bounds_equal(PartitionKey key,
 extern void check_new_partition_bound(char *relname, Relation parent, Node *bound);
 extern Oid	get_partition_parent(Oid relid);
 extern List *get_qual_from_partbound(Relation rel, Relation parent, Node *bound);
+extern List *generate_qual_for_defaultpart(Relation parent, Node *bound ,Oid *defid);
 extern List *map_partition_varattnos(List *expr, int target_varno,
 						Relation partrel, Relation parent);
 extern List *RelationGetPartitionQual(Relation rel);