Add fallback relation for COPY

Author: ildus

expected/pathman_utility_stmt.out

@@ -279,8 +279,30 @@ SELECT COUNT(*) FROM copy_stmt_hooking.test2;
      1
 (1 row)
 
+/* check fallback relation for NULLs */
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+ERROR:  partitioning expression's value should not be NULL
+SET pg_pathman.enable_fallback_relation=ON;
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+ERROR:  null value in column "t" violates not-null constraint
+ALTER TABLE copy_stmt_hooking.test2 ALTER COLUMN t DROP NOT NULL;
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+SELECT * FROM copy_stmt_hooking.pathman_test2_fallback;
+ b | t 
+---+---
+   | 
+(1 row)
+
+SET pg_pathman.fallback_relation_name='fallbacks';
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+SELECT * FROM copy_stmt_hooking.fallbacks;
+ b | t 
+---+---
+   | 
+(1 row)
+
 DROP SCHEMA copy_stmt_hooking CASCADE;
-NOTICE:  drop cascades to 797 other objects
+NOTICE:  drop cascades to 823 other objects
 /*
  * Test auto check constraint renaming
  */

sql/pathman_utility_stmt.sql

@@ -158,6 +158,33 @@ COPY copy_stmt_hooking.test2(t) FROM stdin;
 \.
 SELECT COUNT(*) FROM copy_stmt_hooking.test2;
 
+/* check fallback relation for NULLs */
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+2017-02-02 20:00:00
+\N
+2017-02-02 20:00:00
+\.
+SET pg_pathman.enable_fallback_relation=ON;
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+2017-02-02 20:00:00
+\N
+2017-02-02 20:00:00
+\.
+ALTER TABLE copy_stmt_hooking.test2 ALTER COLUMN t DROP NOT NULL;
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+2017-02-02 20:00:00
+\N
+2017-02-02 20:00:00
+\.
+SELECT * FROM copy_stmt_hooking.pathman_test2_fallback;
+SET pg_pathman.fallback_relation_name='fallbacks';
+COPY copy_stmt_hooking.test2(t) FROM stdin;
+2017-02-03 20:00:00
+\N
+2017-02-03 20:00:00
+\.
+SELECT * FROM copy_stmt_hooking.fallbacks;
+
 DROP SCHEMA copy_stmt_hooking CASCADE;
 
 

src/include/partition_creation.h

@@ -43,6 +43,14 @@ Oid create_single_hash_partition_internal(Oid parent_relid,
 										  RangeVar *partition_rv,
 										  char *tablespace);
 
+/*
+ * Create one partition. Also used to created parent like tables
+ * without any inheritance.
+ */
+Oid create_single_partition_internal(Oid parent_relid,
+								 RangeVar *partition_rv,
+								 char *tablespace,
+								 bool inherit_parent);
 
 /* RANGE constraints */
 Constraint * build_range_check_constraint(Oid child_relid,

src/include/partition_filter.h

@@ -103,6 +103,7 @@ typedef struct
 
 extern bool					pg_pathman_enable_partition_filter;
 extern int					pg_pathman_insert_into_fdw;
+extern bool					pg_pathman_enable_fallback_relation;
 
 extern CustomScanMethods	partition_filter_plan_methods;
 extern CustomExecMethods	partition_filter_exec_methods;
@@ -138,6 +139,9 @@ ResultRelInfoHolder * select_partition_for_insert(Datum value, Oid value_type,
 												  ResultPartsStorage *parts_storage,
 												  EState *estate);
 
+ResultRelInfoHolder * get_relation_for_fallback(const PartRelationInfo *prel,
+												ResultPartsStorage *parts_storage,
+												EState *estate);
 
 Plan * make_partition_filter(Plan *subplan,
 							 Oid parent_relid,

src/include/pathman.h

@@ -115,9 +115,10 @@ Index append_child_relation(PlannerInfo *root,
 
 
 /*
- * Copied from PostgreSQL (prepunion.c)
+ * Copied from PostgreSQL (prepunion.c, make_inh_translation_list)
+ * 'attinhcount' checks were removed
  */
-void make_inh_translation_list(Relation oldrelation, Relation newrelation,
+void make_translation_list(Relation oldrelation, Relation newrelation,
 							   Index newvarno, List **translated_vars);
 
 Bitmapset *translate_col_privs(const Bitmapset *parent_privs,

src/include/utility_stmt_hooking.h

@@ -19,7 +19,6 @@
 #include "commands/copy.h"
 #include "nodes/nodes.h"
 
-
 /* Various traits */
 bool is_pathman_related_copy(Node *parsetree);
 bool is_pathman_related_table_rename(Node *parsetree,

src/partition_creation.c

@@ -63,10 +63,6 @@ static void create_single_partition_common(Oid parent_relid,
 										   init_callback_params *callback_params,
 										   List *trigger_columns);
 
-static Oid create_single_partition_internal(Oid parent_relid,
-											RangeVar *partition_rv,
-											char *tablespace);
-
 static char *choose_range_partition_name(Oid parent_relid, Oid parent_nsp);
 static char *choose_hash_partition_name(Oid parent_relid, uint32 part_idx);
 
@@ -123,10 +119,11 @@ create_single_range_partition_internal(Oid parent_relid,
 	/* Check pathman config anld fill variables */
 	expr = build_partitioning_expression(parent_relid, NULL, &trigger_columns);
 
-	/* Create a partition & get 'partitioning expression' */
+	/* Create a partition */
 	partition_relid = create_single_partition_internal(parent_relid,
 													   partition_rv,
-													   tablespace);
+													   tablespace,
+													   true);
 
 	/* Build check constraint for RANGE partition */
 	check_constr = build_range_check_constraint(partition_relid,
@@ -179,12 +176,13 @@ create_single_hash_partition_internal(Oid parent_relid,
 		partition_rv = makeRangeVar(parent_nsp_name, partition_name, -1);
 	}
 
-	/* Create a partition & get 'partitionining expression' */
+	/* Create a partition */
 	partition_relid = create_single_partition_internal(parent_relid,
 													   partition_rv,
-													   tablespace);
+													   tablespace,
+													   true);
 
-	/* check pathman config and fill variables */
+	/* Build an expression and get column used in expression for triggers */
 	expr = build_partitioning_expression(parent_relid, &expr_type, &trigger_columns);
 
 	/* Build check constraint for HASH partition */
@@ -667,10 +665,11 @@ choose_hash_partition_name(Oid parent_relid, uint32 part_idx)
 }
 
 /* Create a partition-like table (no constraints yet) */
-static Oid
+Oid
 create_single_partition_internal(Oid parent_relid,
 								 RangeVar *partition_rv,
-								 char *tablespace)
+								 char *tablespace,
+								 bool inherit_parent)
 {
 	/* Value to be returned */
 	Oid					partition_relid = InvalidOid; /* safety */
@@ -739,14 +738,15 @@ create_single_partition_internal(Oid parent_relid,
 	NodeSetTag(&like_clause, T_TableLikeClause);
 	like_clause.relation		= copyObject(parent_rv);
 	like_clause.options			= CREATE_TABLE_LIKE_DEFAULTS |
-								  CREATE_TABLE_LIKE_INDEXES |
 								  CREATE_TABLE_LIKE_STORAGE;
+	if (inherit_parent)
+		like_clause.options |= CREATE_TABLE_LIKE_INDEXES;
 
 	/* Initialize CreateStmt structure */
 	NodeSetTag(&create_stmt, T_CreateStmt);
 	create_stmt.relation		= copyObject(partition_rv);
 	create_stmt.tableElts		= list_make1(copyObject(&like_clause));
-	create_stmt.inhRelations	= list_make1(copyObject(parent_rv));
+	create_stmt.inhRelations	= inherit_parent? list_make1(copyObject(parent_rv)) : NIL;
 	create_stmt.ofTypename		= NULL;
 	create_stmt.constraints		= NIL;
 	create_stmt.options			= NIL;
@@ -898,7 +898,7 @@ postprocess_child_table_and_atts(Oid parent_relid, Oid partition_relid)
 	parent_rel = heap_open(parent_relid, NoLock);
 	partition_rel = heap_open(partition_relid, NoLock);
 
-	make_inh_translation_list(parent_rel, partition_rel, 0, &translated_vars);
+	make_translation_list(parent_rel, partition_rel, 0, &translated_vars);
 
 	heap_close(parent_rel, NoLock);
 	heap_close(partition_rel, NoLock);

src/partition_filter.c

@@ -27,6 +27,7 @@
 
 
 #define ALLOC_EXP	2
+#define PARENT_NAME_FMT "{parent_table}"
 
 
 /*
@@ -64,6 +65,9 @@ static const struct config_enum_entry pg_pathman_insert_into_fdw_options[] = {
 bool				pg_pathman_enable_partition_filter = true;
 int					pg_pathman_insert_into_fdw = PF_FDW_INSERT_POSTGRES;
 
+bool				pg_pathman_enable_fallback_relation = true;
+char			   *pg_pathman_fallback_relation_name;
+
 CustomScanMethods	partition_filter_plan_methods;
 CustomExecMethods	partition_filter_exec_methods;
 
@@ -128,6 +132,25 @@ init_partition_filter_static_data(void)
 							 NULL,
 							 NULL,
 							 NULL);
+
+	DefineCustomBoolVariable("pg_pathman.enable_fallback_relation",
+							 "Enables fallback relation for COPY when partitioning key returns NULL",
+							 NULL,
+							 &pg_pathman_enable_fallback_relation,
+							 false,
+							 PGC_USERSET,
+							 0,
+							 NULL,
+							 NULL,
+							 NULL);
+
+	DefineCustomStringVariable("pg_pathman.fallback_relation_name",
+							   "Partition name for NULL tuples in COPY command",
+							   NULL,
+							   &pg_pathman_fallback_relation_name,
+							   "pathman_" PARENT_NAME_FMT "_fallback",
+							   PGC_SUSET, 0,
+							   NULL, NULL, NULL);
 }
 
 
@@ -270,9 +293,6 @@ scan_result_parts_storage(Oid partid, ResultPartsStorage *parts_storage)
 		child_rel = heap_open(partid, NoLock);
 		CheckValidResultRel(child_rel, parts_storage->command_type);
 
-		/* Build Var translation list for 'inserted_cols' */
-		make_inh_translation_list(parent_rel, child_rel, 0, &translated_vars);
-
 		/* Create RangeTblEntry for partition */
 		child_rte = makeNode(RangeTblEntry);
 		child_rte->rtekind			= RTE_RELATION;
@@ -281,6 +301,9 @@ scan_result_parts_storage(Oid partid, ResultPartsStorage *parts_storage)
 		child_rte->eref				= parent_rte->eref;
 		child_rte->requiredPerms	= parent_rte->requiredPerms;
 		child_rte->checkAsUser		= parent_rte->checkAsUser;
+
+		/* Build Var translation list for 'inserted_cols' */
+		make_translation_list(parent_rel, child_rel, 0, &translated_vars);
 		child_rte->insertedCols		= translate_col_privs(parent_rte->insertedCols,
 														  translated_vars);
 		child_rte->updatedCols		= translate_col_privs(parent_rte->updatedCols,
@@ -473,6 +496,76 @@ select_partition_for_insert(Datum value, Oid value_type,
 	return rri_holder;
 }
 
+static char *
+get_fallback_relation_name(Oid parent_relid)
+{
+	char		   *parent_name = get_rel_name(parent_relid),
+				   *s = pstrdup(pg_pathman_fallback_relation_name);
+	StringInfoData	str;
+	char		   *pos;
+
+	initStringInfo(&str);
+
+	if (strstr(s, PARENT_NAME_FMT) != NULL)
+	{
+		while ((pos = strstr(s, PARENT_NAME_FMT)) != NULL)
+		{
+			int tmpl_len = strlen(PARENT_NAME_FMT);
+
+			appendBinaryStringInfo(&str, s, pos - s);
+			appendBinaryStringInfo(&str, parent_name, strlen(parent_name));
+			appendBinaryStringInfo(&str, pos + tmpl_len, strlen(s) - (pos - s) - tmpl_len);
+			memset(s, 0, tmpl_len);
+		}
+	}
+	else appendStringInfoString(&str, s);
+	pfree(s);
+
+	return str.data;
+}
+
+static Oid
+get_fallback_relation(Oid parent_relid)
+{
+	char       *name = get_fallback_relation_name(parent_relid);
+	Oid			result,
+				nsp;
+
+	nsp = get_rel_namespace(parent_relid);
+	result = get_relname_relid(name, nsp);
+	if (result == InvalidOid)
+	{
+		RangeVar	*rv;
+		char		*nspname = get_namespace_name(nsp);
+
+		rv = makeRangeVar(nspname, name, -1);
+		result = create_single_partition_internal(parent_relid, rv, NULL, false);
+	}
+
+	return result;
+}
+
+/*
+ * Return holder for fallback partition
+ */
+ResultRelInfoHolder *
+get_relation_for_fallback(const PartRelationInfo *prel,
+							ResultPartsStorage *parts_storage,
+							EState *estate)
+{
+	MemoryContext			old_mcxt;
+	ResultRelInfoHolder	   *rri_holder;
+	Oid						parent_relid = PrelParentRelid(prel),
+							fallback_relid;
+
+	/* Replace parent table with a suitable partition */
+	old_mcxt = MemoryContextSwitchTo(estate->es_query_cxt);
+	fallback_relid = get_fallback_relation(parent_relid);
+	rri_holder = scan_result_parts_storage(fallback_relid, parts_storage);
+	MemoryContextSwitchTo(old_mcxt);
+
+	return rri_holder;
+}
 
 /*
  * --------------------------------

src/pg_pathman.c

@@ -470,7 +470,7 @@ append_child_relation(PlannerInfo *root,
 	appinfo->parent_reltype = RelationGetDescr(parent_relation)->tdtypeid;
 	appinfo->child_reltype  = RelationGetDescr(child_relation)->tdtypeid;
 
-	make_inh_translation_list(parent_relation, child_relation, childRTindex,
+	make_translation_list(parent_relation, child_relation, childRTindex,
 							  &appinfo->translated_vars);
 
 	/* Now append 'appinfo' to 'root->append_rel_list' */
@@ -1747,14 +1747,14 @@ translate_col_privs(const Bitmapset *parent_privs,
 
 
 /*
- * make_inh_translation_list
+ * make_translation_list
  *	  Build the list of translations from parent Vars to child Vars for
  *	  an inheritance child.
  *
  * For paranoia's sake, we match type/collation as well as attribute name.
  */
 void
-make_inh_translation_list(Relation oldrelation, Relation newrelation,
+make_translation_list(Relation oldrelation, Relation newrelation,
 						  Index newvarno, List **translated_vars)
 {
 	List	   *vars = NIL;
@@ -1812,7 +1812,7 @@ make_inh_translation_list(Relation oldrelation, Relation newrelation,
 		 */
 		if (old_attno < newnatts &&
 			(att = new_tupdesc->attrs[old_attno]) != NULL &&
-			!att->attisdropped && att->attinhcount != 0 &&
+			!att->attisdropped &&
 			strcmp(attname, NameStr(att->attname)) == 0)
 			new_attno = old_attno;
 		else
@@ -1831,7 +1831,7 @@ make_inh_translation_list(Relation oldrelation, Relation newrelation,
 					elog(ERROR, "error in function "
 								CppAsString(make_inh_translation_list));
 
-				if (!att->attisdropped && att->attinhcount != 0 &&
+				if (!att->attisdropped &&
 					strcmp(attname, NameStr(att->attname)) == 0)
 					break;
 			}