Improve updatability checking for views and foreign tables.

Extend the FDW API (which we already changed for 9.3) so that an FDW can
report whether specific foreign tables are insertable/updatable/deletable.
The default assumption continues to be that they're updatable if the
relevant executor callback function is supplied by the FDW, but finer
granularity is now possible.  As a test case, add an "updatable" option to
contrib/postgres_fdw.

This patch also fixes the information_schema views, which previously did
not think that foreign tables were ever updatable, and fixes
view_is_auto_updatable() so that a view on a foreign table can be
auto-updatable.

initdb forced due to changes in information_schema views and the functions
they rely on.  This is a bit unfortunate to do post-beta1, but if we don't
change this now then we'll have another API break for FDWs when we do
change it.

Dean Rasheed, somewhat editorialized on by Tom Lane

9 files changed, 153 insertions, 52 deletions

diff --git a/src/backend/catalog/information_schema.sql b/src/backend/catalog/information_schema.sql
index 230758654cc..e1f8e7f4b1c 100644
--- a/src/backend/catalog/information_schema.sql
+++ b/src/backend/catalog/information_schema.sql
@@ -731,7 +731,8 @@ CREATE VIEW columns AS
            CAST(null AS character_data) AS generation_expression,
 
            CAST(CASE WHEN c.relkind = 'r' OR
-                          (c.relkind = 'v' AND pg_view_is_updatable(c.oid))
+                          (c.relkind IN ('v', 'f') AND
+                           pg_column_is_updatable(c.oid, a.attnum, false))
                 THEN 'YES' ELSE 'NO' END AS yes_or_no) AS is_updatable
 
     FROM (pg_attribute a LEFT JOIN pg_attrdef ad ON attrelid = adrelid AND attnum = adnum)
@@ -1895,7 +1896,9 @@ CREATE VIEW tables AS
            CAST(t.typname AS sql_identifier) AS user_defined_type_name,
 
            CAST(CASE WHEN c.relkind = 'r' OR
-                          (c.relkind = 'v' AND pg_view_is_insertable(c.oid))
+                          (c.relkind IN ('v', 'f') AND
+                           -- 1 << CMD_INSERT
+                           pg_relation_is_updatable(c.oid, false) & 8 = 8)
                 THEN 'YES' ELSE 'NO' END AS yes_or_no) AS is_insertable_into,
 
            CAST(CASE WHEN t.typname IS NOT NULL THEN 'YES' ELSE 'NO' END AS yes_or_no) AS is_typed,
@@ -2494,11 +2497,15 @@ CREATE VIEW views AS
            CAST('NONE' AS character_data) AS check_option,
 
            CAST(
-             CASE WHEN pg_view_is_updatable(c.oid) THEN 'YES' ELSE 'NO' END
+             -- (1 << CMD_UPDATE) + (1 << CMD_DELETE)
+             CASE WHEN pg_relation_is_updatable(c.oid, false) & 20 = 20
+                  THEN 'YES' ELSE 'NO' END
              AS yes_or_no) AS is_updatable,
 
            CAST(
-             CASE WHEN pg_view_is_insertable(c.oid) THEN 'YES' ELSE 'NO' END
+             -- 1 << CMD_INSERT
+             CASE WHEN pg_relation_is_updatable(c.oid, false) & 8 = 8
+                  THEN 'YES' ELSE 'NO' END
              AS yes_or_no) AS is_insertable_into,
 
            CAST(
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 9b0cd8c2070..3b664d09265 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1015,6 +1015,12 @@ CheckValidResultRel(Relation resultRel, CmdType operation)
 								(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 							errmsg("cannot insert into foreign table \"%s\"",
 								   RelationGetRelationName(resultRel))));
+					if (fdwroutine->IsForeignRelUpdatable != NULL &&
+						(fdwroutine->IsForeignRelUpdatable(resultRel) & (1 << CMD_INSERT)) == 0)
+						ereport(ERROR,
+						  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						errmsg("foreign table \"%s\" does not allow inserts",
+							   RelationGetRelationName(resultRel))));
 					break;
 				case CMD_UPDATE:
 					if (fdwroutine->ExecForeignUpdate == NULL)
@@ -1022,6 +1028,12 @@ CheckValidResultRel(Relation resultRel, CmdType operation)
 								(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 								 errmsg("cannot update foreign table \"%s\"",
 										RelationGetRelationName(resultRel))));
+					if (fdwroutine->IsForeignRelUpdatable != NULL &&
+						(fdwroutine->IsForeignRelUpdatable(resultRel) & (1 << CMD_UPDATE)) == 0)
+						ereport(ERROR,
+						  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						errmsg("foreign table \"%s\" does not allow updates",
+							   RelationGetRelationName(resultRel))));
 					break;
 				case CMD_DELETE:
 					if (fdwroutine->ExecForeignDelete == NULL)
@@ -1029,6 +1041,12 @@ CheckValidResultRel(Relation resultRel, CmdType operation)
 								(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 							errmsg("cannot delete from foreign table \"%s\"",
 								   RelationGetRelationName(resultRel))));
+					if (fdwroutine->IsForeignRelUpdatable != NULL &&
+						(fdwroutine->IsForeignRelUpdatable(resultRel) & (1 << CMD_DELETE)) == 0)
+						ereport(ERROR,
+						  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						errmsg("foreign table \"%s\" does not allow deletes",
+							   RelationGetRelationName(resultRel))));
 					break;
 				default:
 					elog(ERROR, "unrecognized CmdType: %d", (int) operation);
diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c
index 01875fcd45f..a467588e50e 100644
--- a/src/backend/rewrite/rewriteHandler.c
+++ b/src/backend/rewrite/rewriteHandler.c
@@ -2014,6 +2014,7 @@ view_is_auto_updatable(Relation view)
 	base_rte = rt_fetch(rtr->rtindex, viewquery->rtable);
 	if (base_rte->rtekind != RTE_RELATION ||
 		(base_rte->relkind != RELKIND_RELATION &&
+		 base_rte->relkind != RELKIND_FOREIGN_TABLE &&
 		 base_rte->relkind != RELKIND_VIEW))
 		return gettext_noop("Views that do not select from a single table or view are not automatically updatable.");
 
@@ -2058,49 +2059,56 @@ view_is_auto_updatable(Relation view)
 
 
 /*
- * relation_is_updatable - test if the specified relation is updatable.
+ * relation_is_updatable - determine which update events the specified
+ * relation supports.
  *
  * This is used for the information_schema views, which have separate concepts
  * of "updatable" and "trigger updatable".	A relation is "updatable" if it
  * can be updated without the need for triggers (either because it has a
  * suitable RULE, or because it is simple enough to be automatically updated).
- *
  * A relation is "trigger updatable" if it has a suitable INSTEAD OF trigger.
  * The SQL standard regards this as not necessarily updatable, presumably
  * because there is no way of knowing what the trigger will actually do.
- * That's currently handled directly in the information_schema views, so
- * need not be considered here.
- *
- * In the case of an automatically updatable view, the base relation must
- * also be updatable.
+ * The information_schema views therefore call this function with
+ * include_triggers = false.  However, other callers might only care whether
+ * data-modifying SQL will work, so they can pass include_triggers = true
+ * to have trigger updatability included in the result.
  *
- * reloid is the pg_class OID to examine.  req_events is a bitmask of
- * rule event numbers; the relation is considered rule-updatable if it has
- * all the specified rules.  (We do it this way so that we can test for
- * UPDATE plus DELETE rules in a single call.)
+ * The return value is a bitmask of rule event numbers indicating which of
+ * the INSERT, UPDATE and DELETE operations are supported.	(We do it this way
+ * so that we can test for UPDATE plus DELETE support in a single call.)
  */
-bool
-relation_is_updatable(Oid reloid, int req_events)
+int
+relation_is_updatable(Oid reloid, bool include_triggers)
 {
+	int			events = 0;
 	Relation	rel;
 	RuleLock   *rulelocks;
 
+#define ALL_EVENTS ((1 << CMD_INSERT) | (1 << CMD_UPDATE) | (1 << CMD_DELETE))
+
 	rel = try_relation_open(reloid, AccessShareLock);
 
 	/*
-	 * If the relation doesn't exist, say "false" rather than throwing an
+	 * If the relation doesn't exist, return zero rather than throwing an
 	 * error.  This is helpful since scanning an information_schema view under
 	 * MVCC rules can result in referencing rels that were just deleted
 	 * according to a SnapshotNow probe.
 	 */
 	if (rel == NULL)
-		return false;
+		return 0;
+
+	/* If the relation is a table, it is always updatable */
+	if (rel->rd_rel->relkind == RELKIND_RELATION)
+	{
+		relation_close(rel, AccessShareLock);
+		return ALL_EVENTS;
+	}
 
 	/* Look for unconditional DO INSTEAD rules, and note supported events */
 	rulelocks = rel->rd_rules;
 	if (rulelocks != NULL)
 	{
-		int			events = 0;
 		int			i;
 
 		for (i = 0; i < rulelocks->numLocks; i++)
@@ -2108,16 +2116,61 @@ relation_is_updatable(Oid reloid, int req_events)
 			if (rulelocks->rules[i]->isInstead &&
 				rulelocks->rules[i]->qual == NULL)
 			{
-				events |= 1 << rulelocks->rules[i]->event;
+				events |= ((1 << rulelocks->rules[i]->event) & ALL_EVENTS);
 			}
 		}
 
-		/* If we have all rules needed, say "yes" */
-		if ((events & req_events) == req_events)
+		/* If we have rules for all events, we're done */
+		if (events == ALL_EVENTS)
 		{
 			relation_close(rel, AccessShareLock);
-			return true;
+			return events;
+		}
+	}
+
+	/* Similarly look for INSTEAD OF triggers, if they are to be included */
+	if (include_triggers)
+	{
+		TriggerDesc *trigDesc = rel->trigdesc;
+
+		if (trigDesc)
+		{
+			if (trigDesc->trig_insert_instead_row)
+				events |= (1 << CMD_INSERT);
+			if (trigDesc->trig_update_instead_row)
+				events |= (1 << CMD_UPDATE);
+			if (trigDesc->trig_delete_instead_row)
+				events |= (1 << CMD_DELETE);
+
+			/* If we have triggers for all events, we're done */
+			if (events == ALL_EVENTS)
+			{
+				relation_close(rel, AccessShareLock);
+				return events;
+			}
+		}
+	}
+
+	/* If this is a foreign table, check which update events it supports */
+	if (rel->rd_rel->relkind == RELKIND_FOREIGN_TABLE)
+	{
+		FdwRoutine *fdwroutine = GetFdwRoutineForRelation(rel, false);
+
+		if (fdwroutine->IsForeignRelUpdatable != NULL)
+			events |= fdwroutine->IsForeignRelUpdatable(rel);
+		else
+		{
+			/* Assume presence of executor functions is sufficient */
+			if (fdwroutine->ExecForeignInsert != NULL)
+				events |= (1 << CMD_INSERT);
+			if (fdwroutine->ExecForeignUpdate != NULL)
+				events |= (1 << CMD_UPDATE);
+			if (fdwroutine->ExecForeignDelete != NULL)
+				events |= (1 << CMD_DELETE);
 		}
+
+		relation_close(rel, AccessShareLock);
+		return events;
 	}
 
 	/* Check if this is an automatically updatable view */
@@ -2133,25 +2186,26 @@ relation_is_updatable(Oid reloid, int req_events)
 		viewquery = get_view_query(rel);
 		rtr = (RangeTblRef *) linitial(viewquery->jointree->fromlist);
 		base_rte = rt_fetch(rtr->rtindex, viewquery->rtable);
+		Assert(base_rte->rtekind == RTE_RELATION);
 
 		if (base_rte->relkind == RELKIND_RELATION)
 		{
 			/* Tables are always updatable */
 			relation_close(rel, AccessShareLock);
-			return true;
+			return ALL_EVENTS;
 		}
 		else
 		{
 			/* Do a recursive check for any other kind of base relation */
 			baseoid = base_rte->relid;
 			relation_close(rel, AccessShareLock);
-			return relation_is_updatable(baseoid, req_events);
+			return relation_is_updatable(baseoid, include_triggers);
 		}
 	}
 
-	/* If we reach here, the relation is not updatable */
+	/* If we reach here, the relation may support some update commands */
 	relation_close(rel, AccessShareLock);
-	return false;
+	return events;
 }
 
 
diff --git a/src/backend/utils/adt/misc.c b/src/backend/utils/adt/misc.c
index 829ce59888c..bf06ec048ff 100644
--- a/src/backend/utils/adt/misc.c
+++ b/src/backend/utils/adt/misc.c
@@ -528,30 +528,49 @@ pg_collation_for(PG_FUNCTION_ARGS)
 
 
 /*
- * information_schema support functions
+ * pg_relation_is_updatable - determine which update events the specified
+ * relation supports.
  *
- * Test whether a view (identified by pg_class OID) is insertable-into or
- * updatable.  The latter requires delete capability too.  This is an
- * artifact of the way the SQL standard defines the information_schema views:
- * if we defined separate functions for update and delete, we'd double the
- * work required to compute the view columns.
- *
- * These rely on relation_is_updatable(), which is in rewriteHandler.c.
+ * This relies on relation_is_updatable() in rewriteHandler.c, which see
+ * for additional information.
  */
 Datum
-pg_view_is_insertable(PG_FUNCTION_ARGS)
+pg_relation_is_updatable(PG_FUNCTION_ARGS)
 {
-	Oid			viewoid = PG_GETARG_OID(0);
-	int			req_events = (1 << CMD_INSERT);
+	Oid			reloid = PG_GETARG_OID(0);
+	bool		include_triggers = PG_GETARG_BOOL(1);
 
-	PG_RETURN_BOOL(relation_is_updatable(viewoid, req_events));
+	PG_RETURN_INT32(relation_is_updatable(reloid, include_triggers));
 }
 
+/*
+ * pg_column_is_updatable - determine whether a column is updatable
+ *
+ * Currently we just check whether the column's relation is updatable.
+ * Eventually we might allow views to have some updatable and some
+ * non-updatable columns.
+ *
+ * Also, this function encapsulates the decision about just what
+ * information_schema.columns.is_updatable actually means.	It's not clear
+ * whether deletability of the column's relation should be required, so
+ * we want that decision in C code where we could change it without initdb.
+ */
 Datum
-pg_view_is_updatable(PG_FUNCTION_ARGS)
+pg_column_is_updatable(PG_FUNCTION_ARGS)
 {
-	Oid			viewoid = PG_GETARG_OID(0);
-	int			req_events = (1 << CMD_UPDATE) | (1 << CMD_DELETE);
+	Oid			reloid = PG_GETARG_OID(0);
+	AttrNumber	attnum = PG_GETARG_INT16(1);
+	bool		include_triggers = PG_GETARG_BOOL(2);
+	int			events;
+
+	/* System columns are never updatable */
+	if (attnum <= 0)
+		PG_RETURN_BOOL(false);
+
+	events = relation_is_updatable(reloid, include_triggers);
+
+	/* We require both updatability and deletability of the relation */
+#define REQ_EVENTS ((1 << CMD_UPDATE) | (1 << CMD_DELETE))
 
-	PG_RETURN_BOOL(relation_is_updatable(viewoid, req_events));
+	PG_RETURN_BOOL((events & REQ_EVENTS) == REQ_EVENTS);
 }
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 392649c37e6..d46fe9ede37 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201305061
+#define CATALOG_VERSION_NO	201306121
 
 #endif
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 4102deca694..b5be075ee16 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -1976,10 +1976,10 @@ DESCR("type of the argument");
 DATA(insert OID = 3162 (  pg_collation_for		PGNSP PGUID 12 1 0 0 0 f f f f f f s 1 0   25 "2276" _null_ _null_ _null_ _null_  pg_collation_for _null_ _null_ _null_ ));
 DESCR("collation of the argument; implementation of the COLLATION FOR expression");
 
-DATA(insert OID = 3842 (  pg_view_is_insertable PGNSP PGUID 12 10 0 0 0 f f f f t f s 1 0 16 "26" _null_ _null_ _null_ _null_ pg_view_is_insertable _null_ _null_ _null_ ));
-DESCR("is a view insertable-into");
-DATA(insert OID = 3843 (  pg_view_is_updatable	PGNSP PGUID 12 10 0 0 0 f f f f t f s 1 0 16 "26" _null_ _null_ _null_ _null_ pg_view_is_updatable _null_ _null_ _null_ ));
-DESCR("is a view updatable");
+DATA(insert OID = 3842 (  pg_relation_is_updatable	PGNSP PGUID 12 10 0 0 0 f f f f t f s 2 0 23 "2205 16" _null_ _null_ _null_ _null_ pg_relation_is_updatable _null_ _null_ _null_ ));
+DESCR("is a relation insertable/updatable/deletable");
+DATA(insert OID = 3843 (  pg_column_is_updatable	PGNSP PGUID 12 10 0 0 0 f f f f t f s 3 0 16 "2205 21 16" _null_ _null_ _null_ _null_ pg_column_is_updatable _null_ _null_ _null_ ));
+DESCR("is a column updatable");
 
 /* Deferrable unique constraint trigger */
 DATA(insert OID = 1250 (  unique_key_recheck	PGNSP PGUID 12 1 0 0 0 f f f f t f v 0 0 2279 "" _null_ _null_ _null_ _null_ unique_key_recheck _null_ _null_ _null_ ));
diff --git a/src/include/foreign/fdwapi.h b/src/include/foreign/fdwapi.h
index 485eee320f8..e8326652693 100644
--- a/src/include/foreign/fdwapi.h
+++ b/src/include/foreign/fdwapi.h
@@ -80,6 +80,8 @@ typedef TupleTableSlot *(*ExecForeignDelete_function) (EState *estate,
 typedef void (*EndForeignModify_function) (EState *estate,
 													   ResultRelInfo *rinfo);
 
+typedef int (*IsForeignRelUpdatable_function) (Relation rel);
+
 typedef void (*ExplainForeignScan_function) (ForeignScanState *node,
 													struct ExplainState *es);
 
@@ -134,6 +136,7 @@ typedef struct FdwRoutine
 	ExecForeignUpdate_function ExecForeignUpdate;
 	ExecForeignDelete_function ExecForeignDelete;
 	EndForeignModify_function EndForeignModify;
+	IsForeignRelUpdatable_function IsForeignRelUpdatable;
 
 	/* Support functions for EXPLAIN */
 	ExplainForeignScan_function ExplainForeignScan;
diff --git a/src/include/rewrite/rewriteHandler.h b/src/include/rewrite/rewriteHandler.h
index 59833158031..1831de46406 100644
--- a/src/include/rewrite/rewriteHandler.h
+++ b/src/include/rewrite/rewriteHandler.h
@@ -21,6 +21,6 @@ extern List *QueryRewrite(Query *parsetree);
 extern void AcquireRewriteLocks(Query *parsetree, bool forUpdatePushedDown);
 
 extern Node *build_column_default(Relation rel, int attrno);
-extern bool relation_is_updatable(Oid reloid, int req_events);
+extern int	relation_is_updatable(Oid reloid, bool include_triggers);
 
 #endif   /* REWRITEHANDLER_H */
diff --git a/src/include/utils/builtins.h b/src/include/utils/builtins.h
index 8acdcaaf987..667c58b5d0c 100644
--- a/src/include/utils/builtins.h
+++ b/src/include/utils/builtins.h
@@ -485,8 +485,8 @@ extern Datum pg_sleep(PG_FUNCTION_ARGS);
 extern Datum pg_get_keywords(PG_FUNCTION_ARGS);
 extern Datum pg_typeof(PG_FUNCTION_ARGS);
 extern Datum pg_collation_for(PG_FUNCTION_ARGS);
-extern Datum pg_view_is_insertable(PG_FUNCTION_ARGS);
-extern Datum pg_view_is_updatable(PG_FUNCTION_ARGS);
+extern Datum pg_relation_is_updatable(PG_FUNCTION_ARGS);
+extern Datum pg_column_is_updatable(PG_FUNCTION_ARGS);
 
 /* oid.c */
 extern Datum oidin(PG_FUNCTION_ARGS);