7 files changed, 546 insertions, 76 deletions

diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index 4af7aad00b..9573a0db45 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -13,7 +13,7 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o  \
-	collationcmds.o constraint.o conversioncmds.o copy.o \
+	collationcmds.o constraint.o conversioncmds.o copy.o createas.o \
 	dbcommands.o define.o discard.o dropcmds.o explain.o extension.o \
 	foreigncmds.o functioncmds.o \
 	indexcmds.o lockcmds.o operatorcmds.o opclasscmds.o \
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 517660d373..6b5bcd83c5 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -1210,15 +1210,17 @@ BeginCopy(bool is_from,
 			elog(ERROR, "unexpected rewrite result");
 
 		query = (Query *) linitial(rewritten);
-		Assert(query->commandType == CMD_SELECT);
-		Assert(query->utilityStmt == NULL);
 
-		/* Query mustn't use INTO, either */
-		if (query->intoClause)
+		/* The grammar allows SELECT INTO, but we don't support that */
+		if (query->utilityStmt != NULL &&
+			IsA(query->utilityStmt, CreateTableAsStmt))
 			ereport(ERROR,
 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 					 errmsg("COPY (SELECT INTO) is not supported")));
 
+		Assert(query->commandType == CMD_SELECT);
+		Assert(query->utilityStmt == NULL);
+
 		/* plan the query */
 		plan = planner(query, 0, NULL);
 
diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c
new file mode 100644
index 0000000000..5173f5a308
--- /dev/null
+++ b/src/backend/commands/createas.c
@@ -0,0 +1,423 @@
+/*-------------------------------------------------------------------------
+ *
+ * createas.c
+ *	  Execution of CREATE TABLE ... AS, a/k/a SELECT INTO
+ *
+ * We implement this by diverting the query's normal output to a
+ * specialized DestReceiver type.
+ *
+ * Formerly, this command was implemented as a variant of SELECT, which led
+ * to assorted legacy behaviors that we still try to preserve, notably that
+ * we must return a tuples-processed count in the completionTag.
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/commands/createas.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/reloptions.h"
+#include "access/sysattr.h"
+#include "access/xact.h"
+#include "catalog/toasting.h"
+#include "commands/createas.h"
+#include "commands/prepare.h"
+#include "commands/tablecmds.h"
+#include "parser/parse_clause.h"
+#include "rewrite/rewriteHandler.h"
+#include "storage/smgr.h"
+#include "tcop/tcopprot.h"
+#include "utils/builtins.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/snapmgr.h"
+
+
+typedef struct
+{
+	DestReceiver pub;			/* publicly-known function pointers */
+	IntoClause *into;			/* target relation specification */
+	/* These fields are filled by intorel_startup: */
+	Relation	rel;			/* relation to write to */
+	CommandId	output_cid;		/* cmin to insert in output tuples */
+	int			hi_options;		/* heap_insert performance options */
+	BulkInsertState bistate;	/* bulk insert state */
+} DR_intorel;
+
+static void intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
+static void intorel_receive(TupleTableSlot *slot, DestReceiver *self);
+static void intorel_shutdown(DestReceiver *self);
+static void intorel_destroy(DestReceiver *self);
+
+
+/*
+ * ExecCreateTableAs -- execute a CREATE TABLE AS command
+ */
+void
+ExecCreateTableAs(CreateTableAsStmt *stmt, const char *queryString,
+				  ParamListInfo params, char *completionTag)
+{
+	Query *query = (Query *) stmt->query;
+	IntoClause *into = stmt->into;
+	DestReceiver *dest;
+	List *rewritten;
+	PlannedStmt *plan;
+	QueryDesc *queryDesc;
+	ScanDirection dir;
+
+	/*
+	 * Create the tuple receiver object and insert info it will need
+	 */
+	dest = CreateIntoRelDestReceiver(into);
+
+	/*
+	 * The contained Query could be a SELECT, or an EXECUTE utility command.
+	 * If the latter, we just pass it off to ExecuteQuery.
+	 */
+	Assert(IsA(query, Query));
+	if (query->commandType == CMD_UTILITY &&
+		IsA(query->utilityStmt, ExecuteStmt))
+	{
+		ExecuteStmt *estmt = (ExecuteStmt *) query->utilityStmt;
+
+		ExecuteQuery(estmt, into, queryString, params, dest, completionTag);
+
+		return;
+	}
+	Assert(query->commandType == CMD_SELECT);
+
+	/*
+	 * Parse analysis was done already, but we still have to run the rule
+	 * rewriter.  We do not do AcquireRewriteLocks: we assume the query either
+	 * came straight from the parser, or suitable locks were acquired by
+	 * plancache.c.
+	 *
+	 * Because the rewriter and planner tend to scribble on the input, we make
+	 * a preliminary copy of the source querytree.  This prevents problems in
+	 * the case that CTAS is in a portal or plpgsql function and is executed
+	 * repeatedly.  (See also the same hack in EXPLAIN and PREPARE.)
+	 */
+	rewritten = QueryRewrite((Query *) copyObject(stmt->query));
+
+	/* SELECT should never rewrite to more or less than one SELECT query */
+	if (list_length(rewritten) != 1)
+		elog(ERROR, "unexpected rewrite result for CREATE TABLE AS SELECT");
+	query = (Query *) linitial(rewritten);
+	Assert(query->commandType == CMD_SELECT);
+
+	/* plan the query */
+	plan = pg_plan_query(query, 0, params);
+
+	/*
+	 * Use a snapshot with an updated command ID to ensure this query sees
+	 * results of any previously executed queries.  (This could only matter
+	 * if the planner executed an allegedly-stable function that changed
+	 * the database contents, but let's do it anyway to be parallel to the
+	 * EXPLAIN code path.)
+	 */
+	PushCopiedSnapshot(GetActiveSnapshot());
+	UpdateActiveSnapshotCommandId();
+
+	/* Create a QueryDesc, redirecting output to our tuple receiver */
+	queryDesc = CreateQueryDesc(plan, queryString,
+								GetActiveSnapshot(), InvalidSnapshot,
+								dest, params, 0);
+
+	/* call ExecutorStart to prepare the plan for execution */
+	ExecutorStart(queryDesc, GetIntoRelEFlags(into));
+
+	/*
+	 * Normally, we run the plan to completion; but if skipData is specified,
+	 * just do tuple receiver startup and shutdown.
+	 */
+	if (into->skipData)
+		dir = NoMovementScanDirection;
+	else
+		dir = ForwardScanDirection;
+
+	/* run the plan */
+	ExecutorRun(queryDesc, dir, 0L);
+
+	/* save the rowcount if we're given a completionTag to fill */
+	if (completionTag)
+		snprintf(completionTag, COMPLETION_TAG_BUFSIZE,
+				 "SELECT %u", queryDesc->estate->es_processed);
+
+	/* and clean up */
+	ExecutorFinish(queryDesc);
+	ExecutorEnd(queryDesc);
+
+	FreeQueryDesc(queryDesc);
+
+	PopActiveSnapshot();
+}
+
+/*
+ * GetIntoRelEFlags --- compute executor flags needed for CREATE TABLE AS
+ *
+ * This is exported because EXPLAIN and PREPARE need it too.  (Note: those
+ * callers still need to deal explicitly with the skipData flag; since they
+ * use different methods for suppressing execution, it doesn't seem worth
+ * trying to encapsulate that part.)
+ */
+int
+GetIntoRelEFlags(IntoClause *intoClause)
+{
+	/*
+	 * We need to tell the executor whether it has to produce OIDs or not,
+	 * because it doesn't have enough information to do so itself (since we
+	 * can't build the target relation until after ExecutorStart).
+	 */
+	if (interpretOidsOption(intoClause->options))
+		return EXEC_FLAG_WITH_OIDS;
+	else
+		return EXEC_FLAG_WITHOUT_OIDS;
+}
+
+/*
+ * CreateIntoRelDestReceiver -- create a suitable DestReceiver object
+ *
+ * intoClause will be NULL if called from CreateDestReceiver(), in which
+ * case it has to be provided later.  However, it is convenient to allow
+ * self->into to be filled in immediately for other callers.
+ */
+DestReceiver *
+CreateIntoRelDestReceiver(IntoClause *intoClause)
+{
+	DR_intorel *self = (DR_intorel *) palloc0(sizeof(DR_intorel));
+
+	self->pub.receiveSlot = intorel_receive;
+	self->pub.rStartup = intorel_startup;
+	self->pub.rShutdown = intorel_shutdown;
+	self->pub.rDestroy = intorel_destroy;
+	self->pub.mydest = DestIntoRel;
+	self->into = intoClause;
+	/* other private fields will be set during intorel_startup */
+
+	return (DestReceiver *) self;
+}
+
+/*
+ * intorel_startup --- executor startup
+ */
+static void
+intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
+{
+	DR_intorel *myState = (DR_intorel *) self;
+	IntoClause *into = myState->into;
+	CreateStmt *create;
+	Oid intoRelationId;
+	Relation intoRelationDesc;
+	RangeTblEntry *rte;
+	Datum		toast_options;
+	ListCell *lc;
+	int attnum;
+	static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
+
+	Assert(into != NULL);		/* else somebody forgot to set it */
+
+	/*
+	 * Create the target relation by faking up a CREATE TABLE parsetree and
+	 * passing it to DefineRelation.
+	 */
+	create = makeNode(CreateStmt);
+	create->relation = into->rel;
+	create->tableElts = NIL;	/* will fill below */
+	create->inhRelations = NIL;
+	create->ofTypename = NULL;
+	create->constraints = NIL;
+	create->options = into->options;
+	create->oncommit = into->onCommit;
+	create->tablespacename = into->tableSpaceName;
+	create->if_not_exists = false;
+
+	/*
+	 * Build column definitions using "pre-cooked" type and collation info.
+	 * If a column name list was specified in CREATE TABLE AS, override the
+	 * column names derived from the query.  (Too few column names are OK, too
+	 * many are not.)
+	 */
+	lc = list_head(into->colNames);
+	for (attnum = 0; attnum < typeinfo->natts; attnum++)
+	{
+		Form_pg_attribute attribute = typeinfo->attrs[attnum];
+		ColumnDef *col = makeNode(ColumnDef);
+		TypeName *coltype = makeNode(TypeName);
+
+		if (lc)
+		{
+			col->colname = strVal(lfirst(lc));
+			lc = lnext(lc);
+		}
+		else
+			col->colname = NameStr(attribute->attname);
+		col->typeName = coltype;
+		col->inhcount = 0;
+		col->is_local = true;
+		col->is_not_null = false;
+		col->is_from_type = false;
+		col->storage = 0;
+		col->raw_default = NULL;
+		col->cooked_default = NULL;
+		col->collClause = NULL;
+		col->collOid = attribute->attcollation;
+		col->constraints = NIL;
+		col->fdwoptions = NIL;
+
+		coltype->names = NIL;
+		coltype->typeOid = attribute->atttypid;
+		coltype->setof = false;
+		coltype->pct_type = false;
+		coltype->typmods = NIL;
+		coltype->typemod = attribute->atttypmod;
+		coltype->arrayBounds = NIL;
+		coltype->location = -1;
+
+		/*
+		 * It's possible that the column is of a collatable type but the
+		 * collation could not be resolved, so double-check.  (We must
+		 * check this here because DefineRelation would adopt the type's
+		 * default collation rather than complaining.)
+		 */
+		if (!OidIsValid(col->collOid) &&
+			type_is_collatable(coltype->typeOid))
+			ereport(ERROR,
+					(errcode(ERRCODE_INDETERMINATE_COLLATION),
+					 errmsg("no collation was derived for column \"%s\" with collatable type %s",
+							col->colname, format_type_be(coltype->typeOid)),
+					 errhint("Use the COLLATE clause to set the collation explicitly.")));
+
+		create->tableElts = lappend(create->tableElts, col);
+	}
+
+	if (lc != NULL)
+		ereport(ERROR,
+		        (errcode(ERRCODE_SYNTAX_ERROR),
+		         errmsg("CREATE TABLE AS specifies too many column names")));
+
+	/*
+	 * Actually create the target table
+	 */
+	intoRelationId = DefineRelation(create, RELKIND_RELATION, InvalidOid);
+
+	/*
+	 * If necessary, create a TOAST table for the target table.  Note that
+	 * AlterTableCreateToastTable ends with CommandCounterIncrement(), so that
+	 * the TOAST table will be visible for insertion.
+	 */
+	CommandCounterIncrement();
+
+	/* parse and validate reloptions for the toast table */
+	toast_options = transformRelOptions((Datum) 0,
+										create->options,
+										"toast",
+										validnsps,
+										true, false);
+
+	(void) heap_reloptions(RELKIND_TOASTVALUE, toast_options, true);
+
+	AlterTableCreateToastTable(intoRelationId, toast_options);
+
+	/*
+	 * Finally we can open the target table
+	 */
+	intoRelationDesc = heap_open(intoRelationId, AccessExclusiveLock);
+
+	/*
+	 * Check INSERT permission on the constructed table.
+	 *
+	 * XXX: It would arguably make sense to skip this check if into->skipData
+	 * is true.
+	 */
+	rte = makeNode(RangeTblEntry);
+	rte->rtekind = RTE_RELATION;
+	rte->relid = intoRelationId;
+	rte->relkind = RELKIND_RELATION;
+	rte->requiredPerms = ACL_INSERT;
+
+	for (attnum = 1; attnum <= intoRelationDesc->rd_att->natts; attnum++)
+		rte->modifiedCols = bms_add_member(rte->modifiedCols,
+										   attnum - FirstLowInvalidHeapAttributeNumber);
+
+	ExecCheckRTPerms(list_make1(rte), true);
+
+	/*
+	 * Fill private fields of myState for use by later routines
+	 */
+	myState->rel = intoRelationDesc;
+	myState->output_cid = GetCurrentCommandId(true);
+
+	/*
+	 * We can skip WAL-logging the insertions, unless PITR or streaming
+	 * replication is in use. We can skip the FSM in any case.
+	 */
+	myState->hi_options = HEAP_INSERT_SKIP_FSM |
+		(XLogIsNeeded() ? 0 : HEAP_INSERT_SKIP_WAL);
+	myState->bistate = GetBulkInsertState();
+
+	/* Not using WAL requires smgr_targblock be initially invalid */
+	Assert(RelationGetTargetBlock(intoRelationDesc) == InvalidBlockNumber);
+}
+
+/*
+ * intorel_receive --- receive one tuple
+ */
+static void
+intorel_receive(TupleTableSlot *slot, DestReceiver *self)
+{
+	DR_intorel *myState = (DR_intorel *) self;
+	HeapTuple	tuple;
+
+	/*
+	 * get the heap tuple out of the tuple table slot, making sure we have a
+	 * writable copy
+	 */
+	tuple = ExecMaterializeSlot(slot);
+
+	/*
+	 * force assignment of new OID (see comments in ExecInsert)
+	 */
+	if (myState->rel->rd_rel->relhasoids)
+		HeapTupleSetOid(tuple, InvalidOid);
+
+	heap_insert(myState->rel,
+				tuple,
+				myState->output_cid,
+				myState->hi_options,
+				myState->bistate);
+
+	/* We know this is a newly created relation, so there are no indexes */
+}
+
+/*
+ * intorel_shutdown --- executor end
+ */
+static void
+intorel_shutdown(DestReceiver *self)
+{
+	DR_intorel *myState = (DR_intorel *) self;
+
+	FreeBulkInsertState(myState->bistate);
+
+	/* If we skipped using WAL, must heap_sync before commit */
+	if (myState->hi_options & HEAP_INSERT_SKIP_WAL)
+		heap_sync(myState->rel);
+
+	/* close rel, but keep lock until commit */
+	heap_close(myState->rel, NoLock);
+	myState->rel = NULL;
+}
+
+/*
+ * intorel_destroy --- release DestReceiver object
+ */
+static void
+intorel_destroy(DestReceiver *self)
+{
+	pfree(self);
+}
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 93b1f34ca0..a14cae1442 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -15,6 +15,7 @@
 
 #include "access/xact.h"
 #include "catalog/pg_type.h"
+#include "commands/createas.h"
 #include "commands/defrem.h"
 #include "commands/prepare.h"
 #include "executor/hashjoin.h"
@@ -45,7 +46,7 @@ explain_get_index_name_hook_type explain_get_index_name_hook = NULL;
 #define X_CLOSE_IMMEDIATE 2
 #define X_NOWHITESPACE 4
 
-static void ExplainOneQuery(Query *query, ExplainState *es,
+static void ExplainOneQuery(Query *query, IntoClause *into, ExplainState *es,
 				const char *queryString, ParamListInfo params);
 static void report_triggers(ResultRelInfo *rInfo, bool show_relname,
 				ExplainState *es);
@@ -212,7 +213,8 @@ ExplainQuery(ExplainStmt *stmt, const char *queryString,
 		/* Explain every plan */
 		foreach(l, rewritten)
 		{
-			ExplainOneQuery((Query *) lfirst(l), &es, queryString, params);
+			ExplainOneQuery((Query *) lfirst(l), NULL, &es,
+							queryString, params);
 
 			/* Separate plans with an appropriate separator */
 			if (lnext(l) != NULL)
@@ -288,21 +290,23 @@ ExplainResultDesc(ExplainStmt *stmt)
 /*
  * ExplainOneQuery -
  *	  print out the execution plan for one Query
+ *
+ * "into" is NULL unless we are explaining the contents of a CreateTableAsStmt.
  */
 static void
-ExplainOneQuery(Query *query, ExplainState *es,
+ExplainOneQuery(Query *query, IntoClause *into, ExplainState *es,
 				const char *queryString, ParamListInfo params)
 {
 	/* planner will not cope with utility statements */
 	if (query->commandType == CMD_UTILITY)
 	{
-		ExplainOneUtility(query->utilityStmt, es, queryString, params);
+		ExplainOneUtility(query->utilityStmt, into, es, queryString, params);
 		return;
 	}
 
 	/* if an advisor plugin is present, let it manage things */
 	if (ExplainOneQuery_hook)
-		(*ExplainOneQuery_hook) (query, es, queryString, params);
+		(*ExplainOneQuery_hook) (query, into, es, queryString, params);
 	else
 	{
 		PlannedStmt *plan;
@@ -311,7 +315,7 @@ ExplainOneQuery(Query *query, ExplainState *es,
 		plan = pg_plan_query(query, 0, params);
 
 		/* run it (if needed) and produce output */
-		ExplainOnePlan(plan, es, queryString, params);
+		ExplainOnePlan(plan, into, es, queryString, params);
 	}
 }
 
@@ -321,18 +325,36 @@ ExplainOneQuery(Query *query, ExplainState *es,
  *	  (In general, utility statements don't have plans, but there are some
  *	  we treat as special cases)
  *
+ * "into" is NULL unless we are explaining the contents of a CreateTableAsStmt.
+ *
  * This is exported because it's called back from prepare.c in the
- * EXPLAIN EXECUTE case
+ * EXPLAIN EXECUTE case.
  */
 void
-ExplainOneUtility(Node *utilityStmt, ExplainState *es,
+ExplainOneUtility(Node *utilityStmt, IntoClause *into, ExplainState *es,
 				  const char *queryString, ParamListInfo params)
 {
 	if (utilityStmt == NULL)
 		return;
 
-	if (IsA(utilityStmt, ExecuteStmt))
-		ExplainExecuteQuery((ExecuteStmt *) utilityStmt, es,
+	if (IsA(utilityStmt, CreateTableAsStmt))
+	{
+		/*
+		 * We have to rewrite the contained SELECT and then pass it back
+		 * to ExplainOneQuery.  It's probably not really necessary to copy
+		 * the contained parsetree another time, but let's be safe.
+		 */
+		CreateTableAsStmt *ctas = (CreateTableAsStmt *) utilityStmt;
+		List	   *rewritten;
+
+		Assert(IsA(ctas->query, Query));
+		rewritten = QueryRewrite((Query *) copyObject(ctas->query));
+		Assert(list_length(rewritten) == 1);
+		ExplainOneQuery((Query *) linitial(rewritten), ctas->into, es,
+						queryString, params);
+	}
+	else if (IsA(utilityStmt, ExecuteStmt))
+		ExplainExecuteQuery((ExecuteStmt *) utilityStmt, into, es,
 							queryString, params);
 	else if (IsA(utilityStmt, NotifyStmt))
 	{
@@ -356,6 +378,9 @@ ExplainOneUtility(Node *utilityStmt, ExplainState *es,
  *		given a planned query, execute it if needed, and then print
  *		EXPLAIN output
  *
+ * "into" is NULL unless we are explaining the contents of a CreateTableAsStmt,
+ * in which case executing the query should result in creating that table.
+ *
  * Since we ignore any DeclareCursorStmt that might be attached to the query,
  * if you say EXPLAIN ANALYZE DECLARE CURSOR then we'll actually run the
  * query.  This is different from pre-8.3 behavior but seems more useful than
@@ -366,9 +391,10 @@ ExplainOneUtility(Node *utilityStmt, ExplainState *es,
  * to call it.
  */
 void
-ExplainOnePlan(PlannedStmt *plannedstmt, ExplainState *es,
+ExplainOnePlan(PlannedStmt *plannedstmt, IntoClause *into, ExplainState *es,
 			   const char *queryString, ParamListInfo params)
 {
+	DestReceiver *dest;
 	QueryDesc  *queryDesc;
 	instr_time	starttime;
 	double		totaltime = 0;
@@ -392,16 +418,27 @@ ExplainOnePlan(PlannedStmt *plannedstmt, ExplainState *es,
 	PushCopiedSnapshot(GetActiveSnapshot());
 	UpdateActiveSnapshotCommandId();
 
-	/* Create a QueryDesc requesting no output */
+	/*
+	 * Normally we discard the query's output, but if explaining CREATE TABLE
+	 * AS, we'd better use the appropriate tuple receiver.
+	 */
+	if (into)
+		dest = CreateIntoRelDestReceiver(into);
+	else
+		dest = None_Receiver;
+
+	/* Create a QueryDesc for the query */
 	queryDesc = CreateQueryDesc(plannedstmt, queryString,
 								GetActiveSnapshot(), InvalidSnapshot,
-								None_Receiver, params, instrument_option);
+								dest, params, instrument_option);
 
 	/* Select execution options */
 	if (es->analyze)
 		eflags = 0;				/* default run-to-completion flags */
 	else
 		eflags = EXEC_FLAG_EXPLAIN_ONLY;
+	if (into)
+		eflags |= GetIntoRelEFlags(into);
 
 	/* call ExecutorStart to prepare the plan for execution */
 	ExecutorStart(queryDesc, eflags);
@@ -409,8 +446,16 @@ ExplainOnePlan(PlannedStmt *plannedstmt, ExplainState *es,
 	/* Execute the plan for statistics if asked for */
 	if (es->analyze)
 	{
+		ScanDirection dir;
+
+		/* EXPLAIN ANALYZE CREATE TABLE AS WITH NO DATA is weird */
+		if (into && into->skipData)
+			dir = NoMovementScanDirection;
+		else
+			dir = ForwardScanDirection;
+
 		/* run the plan */
-		ExecutorRun(queryDesc, ForwardScanDirection, 0L);
+		ExecutorRun(queryDesc, dir, 0L);
 
 		/* run cleanup too */
 		ExecutorFinish(queryDesc);
diff --git a/src/backend/commands/portalcmds.c b/src/backend/commands/portalcmds.c
index 1c7a1c3a33..e402042332 100644
--- a/src/backend/commands/portalcmds.c
+++ b/src/backend/commands/portalcmds.c
@@ -121,7 +121,7 @@ PerformCursorOpen(PlannedStmt *stmt, ParamListInfo params,
 	/*
 	 * Start execution, inserting parameters if any.
 	 */
-	PortalStart(portal, params, true);
+	PortalStart(portal, params, 0, true);
 
 	Assert(portal->strategy == PORTAL_ONE_SELECT);
 
diff --git a/src/backend/commands/prepare.c b/src/backend/commands/prepare.c
index 4883abe470..edd646e7c3 100644
--- a/src/backend/commands/prepare.c
+++ b/src/backend/commands/prepare.c
@@ -18,6 +18,7 @@
 
 #include "access/xact.h"
 #include "catalog/pg_type.h"
+#include "commands/createas.h"
 #include "commands/prepare.h"
 #include "miscadmin.h"
 #include "nodes/nodeFuncs.h"
@@ -170,7 +171,12 @@ PrepareQuery(PrepareStmt *stmt, const char *queryString)
 }
 
 /*
- * Implements the 'EXECUTE' utility statement.
+ * ExecuteQuery --- implement the 'EXECUTE' utility statement.
+ *
+ * This code also supports CREATE TABLE ... AS EXECUTE.  That case is
+ * indicated by passing a non-null intoClause.  The DestReceiver is already
+ * set up correctly for CREATE TABLE AS, but we still have to make a few
+ * other adjustments here.
  *
  * Note: this is one of very few places in the code that needs to deal with
  * two query strings at once.  The passed-in queryString is that of the
@@ -179,8 +185,8 @@ PrepareQuery(PrepareStmt *stmt, const char *queryString)
  * source is that of the original PREPARE.
  */
 void
-ExecuteQuery(ExecuteStmt *stmt, const char *queryString,
-			 ParamListInfo params,
+ExecuteQuery(ExecuteStmt *stmt, IntoClause *intoClause,
+			 const char *queryString, ParamListInfo params,
 			 DestReceiver *dest, char *completionTag)
 {
 	PreparedStatement *entry;
@@ -190,6 +196,8 @@ ExecuteQuery(ExecuteStmt *stmt, const char *queryString,
 	EState	   *estate = NULL;
 	Portal		portal;
 	char	   *query_string;
+	int			eflags;
+	long		count;
 
 	/* Look it up in the hash table */
 	entry = FetchPreparedStatement(stmt->name, true);
@@ -222,25 +230,27 @@ ExecuteQuery(ExecuteStmt *stmt, const char *queryString,
 	query_string = MemoryContextStrdup(PortalGetHeapMemory(portal),
 									   entry->plansource->query_string);
 
+	/* Replan if needed, and increment plan refcount for portal */
+	cplan = GetCachedPlan(entry->plansource, paramLI, false);
+	plan_list = cplan->stmt_list;
+
 	/*
-	 * For CREATE TABLE / AS EXECUTE, we must make a copy of the stored query
-	 * so that we can modify its destination (yech, but this has always been
-	 * ugly).  For regular EXECUTE we can just use the cached query, since the
-	 * executor is read-only.
+	 * For CREATE TABLE ... AS EXECUTE, we must verify that the prepared
+	 * statement is one that produces tuples.  Currently we insist that it be
+	 * a plain old SELECT.  In future we might consider supporting other
+	 * things such as INSERT ... RETURNING, but there are a couple of issues
+	 * to be settled first, notably how WITH NO DATA should be handled in such
+	 * a case (do we really want to suppress execution?) and how to pass down
+	 * the OID-determining eflags (PortalStart won't handle them in such a
+	 * case, and for that matter it's not clear the executor will either).
+	 *
+	 * For CREATE TABLE ... AS EXECUTE, we also have to ensure that the
+	 * proper eflags and fetch count are passed to PortalStart/PortalRun.
 	 */
-	if (stmt->into)
+	if (intoClause)
 	{
-		MemoryContext oldContext;
 		PlannedStmt *pstmt;
 
-		/* Replan if needed, and increment plan refcount transiently */
-		cplan = GetCachedPlan(entry->plansource, paramLI, true);
-
-		/* Copy plan into portal's context, and modify */
-		oldContext = MemoryContextSwitchTo(PortalGetHeapMemory(portal));
-
-		plan_list = copyObject(cplan->stmt_list);
-
 		if (list_length(plan_list) != 1)
 			ereport(ERROR,
 					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
@@ -252,20 +262,21 @@ ExecuteQuery(ExecuteStmt *stmt, const char *queryString,
 			ereport(ERROR,
 					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 					 errmsg("prepared statement is not a SELECT")));
-		pstmt->intoClause = copyObject(stmt->into);
 
-		MemoryContextSwitchTo(oldContext);
+		/* Set appropriate eflags */
+		eflags = GetIntoRelEFlags(intoClause);
 
-		/* We no longer need the cached plan refcount ... */
-		ReleaseCachedPlan(cplan, true);
-		/* ... and we don't want the portal to depend on it, either */
-		cplan = NULL;
+		/* And tell PortalRun whether to run to completion or not */
+		if (intoClause->skipData)
+			count = 0;
+		else
+			count = FETCH_ALL;
 	}
 	else
 	{
-		/* Replan if needed, and increment plan refcount for portal */
-		cplan = GetCachedPlan(entry->plansource, paramLI, false);
-		plan_list = cplan->stmt_list;
+		/* Plain old EXECUTE */
+		eflags = 0;
+		count = FETCH_ALL;
 	}
 
 	PortalDefineQuery(portal,
@@ -276,11 +287,11 @@ ExecuteQuery(ExecuteStmt *stmt, const char *queryString,
 					  cplan);
 
 	/*
-	 * Run the portal to completion.
+	 * Run the portal as appropriate.
 	 */
-	PortalStart(portal, paramLI, true);
+	PortalStart(portal, paramLI, eflags, true);
 
-	(void) PortalRun(portal, FETCH_ALL, false, dest, dest, completionTag);
+	(void) PortalRun(portal, count, false, dest, dest, completionTag);
 
 	PortalDrop(portal, false);
 
@@ -615,11 +626,14 @@ DropAllPreparedStatements(void)
 /*
  * Implements the 'EXPLAIN EXECUTE' utility statement.
  *
+ * "into" is NULL unless we are doing EXPLAIN CREATE TABLE AS EXECUTE,
+ * in which case executing the query should result in creating that table.
+ *
  * Note: the passed-in queryString is that of the EXPLAIN EXECUTE,
  * not the original PREPARE; we get the latter string from the plancache.
  */
 void
-ExplainExecuteQuery(ExecuteStmt *execstmt, ExplainState *es,
+ExplainExecuteQuery(ExecuteStmt *execstmt, IntoClause *into, ExplainState *es,
 					const char *queryString, ParamListInfo params)
 {
 	PreparedStatement *entry;
@@ -665,27 +679,9 @@ ExplainExecuteQuery(ExecuteStmt *execstmt, ExplainState *es,
 		PlannedStmt *pstmt = (PlannedStmt *) lfirst(p);
 
 		if (IsA(pstmt, PlannedStmt))
-		{
-			if (execstmt->into)
-			{
-				if (pstmt->commandType != CMD_SELECT ||
-					pstmt->utilityStmt != NULL)
-					ereport(ERROR,
-							(errcode(ERRCODE_WRONG_OBJECT_TYPE),
-							 errmsg("prepared statement is not a SELECT")));
-
-				/* Copy the stmt so we can modify it */
-				pstmt = copyObject(pstmt);
-
-				pstmt->intoClause = execstmt->into;
-			}
-
-			ExplainOnePlan(pstmt, es, query_string, paramLI);
-		}
+			ExplainOnePlan(pstmt, into, es, query_string, paramLI);
 		else
-		{
-			ExplainOneUtility((Node *) pstmt, es, query_string, paramLI);
-		}
+			ExplainOneUtility((Node *) pstmt, into, es, query_string, paramLI);
 
 		/* No need for CommandCounterIncrement, as ExplainOnePlan did it */
 
diff --git a/src/backend/commands/view.c b/src/backend/commands/view.c
index 99fb7dbb8f..c887961bc9 100644
--- a/src/backend/commands/view.c
+++ b/src/backend/commands/view.c
@@ -439,9 +439,17 @@ DefineView(ViewStmt *stmt, const char *queryString)
 
 	/*
 	 * The grammar should ensure that the result is a single SELECT Query.
+	 * However, it doesn't forbid SELECT INTO, so we have to check for that.
 	 */
-	if (!IsA(viewParse, Query) ||
-		viewParse->commandType != CMD_SELECT)
+	if (!IsA(viewParse, Query))
+		elog(ERROR, "unexpected parse analysis result");
+	if (viewParse->utilityStmt != NULL &&
+		IsA(viewParse->utilityStmt, CreateTableAsStmt))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("views must not contain SELECT INTO")));
+	if (viewParse->commandType != CMD_SELECT ||
+		viewParse->utilityStmt != NULL)
 		elog(ERROR, "unexpected parse analysis result");
 
 	/*
@@ -449,10 +457,6 @@ DefineView(ViewStmt *stmt, const char *queryString)
 	 * DefineQueryRewrite(), but that function will complain about a bogus ON
 	 * SELECT rule, and we'd rather the message complain about a view.
 	 */
-	if (viewParse->intoClause != NULL)
-		ereport(ERROR,
-				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-				 errmsg("views must not contain SELECT INTO")));
 	if (viewParse->hasModifyingCTE)
 		ereport(ERROR,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),