4 files changed, 42 insertions, 432 deletions

diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 36dcc8e4b5d..fbb36fa6dc4 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -37,27 +37,20 @@
  */
 #include "postgres.h"
 
-#include "access/reloptions.h"
 #include "access/sysattr.h"
 #include "access/transam.h"
 #include "access/xact.h"
-#include "catalog/heap.h"
 #include "catalog/namespace.h"
-#include "catalog/toasting.h"
-#include "commands/tablespace.h"
 #include "commands/trigger.h"
 #include "executor/execdebug.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
 #include "optimizer/clauses.h"
-#include "parser/parse_clause.h"
 #include "parser/parsetree.h"
 #include "storage/bufmgr.h"
 #include "storage/lmgr.h"
-#include "storage/smgr.h"
 #include "tcop/utility.h"
 #include "utils/acl.h"
-#include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/snapmgr.h"
@@ -90,12 +83,6 @@ static char *ExecBuildSlotValueDescription(TupleTableSlot *slot,
 										   int maxfieldlen);
 static void EvalPlanQualStart(EPQState *epqstate, EState *parentestate,
 				  Plan *planTree);
-static void OpenIntoRel(QueryDesc *queryDesc);
-static void CloseIntoRel(QueryDesc *queryDesc);
-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);
 
 /* end of local decls */
 
@@ -174,11 +161,9 @@ standard_ExecutorStart(QueryDesc *queryDesc, int eflags)
 		case CMD_SELECT:
 
 			/*
-			 * SELECT INTO, SELECT FOR UPDATE/SHARE and modifying CTEs need to
-			 * mark tuples
+			 * SELECT FOR UPDATE/SHARE and modifying CTEs need to mark tuples
 			 */
-			if (queryDesc->plannedstmt->intoClause != NULL ||
-				queryDesc->plannedstmt->rowMarks != NIL ||
+			if (queryDesc->plannedstmt->rowMarks != NIL ||
 				queryDesc->plannedstmt->hasModifyingCTE)
 				estate->es_output_cid = GetCurrentCommandId(true);
 
@@ -310,13 +295,6 @@ standard_ExecutorRun(QueryDesc *queryDesc,
 		(*dest->rStartup) (dest, operation, queryDesc->tupDesc);
 
 	/*
-	 * if it's CREATE TABLE AS ... WITH NO DATA, skip plan execution
-	 */
-	if (estate->es_select_into &&
-		queryDesc->plannedstmt->intoClause->skipData)
-		direction = NoMovementScanDirection;
-
-	/*
 	 * run plan
 	 */
 	if (!ScanDirectionIsNoMovement(direction))
@@ -451,12 +429,6 @@ standard_ExecutorEnd(QueryDesc *queryDesc)
 
 	ExecEndPlan(queryDesc->planstate, estate);
 
-	/*
-	 * Close the SELECT INTO relation if any
-	 */
-	if (estate->es_select_into)
-		CloseIntoRel(queryDesc);
-
 	/* do away with our snapshots */
 	UnregisterSnapshot(estate->es_snapshot);
 	UnregisterSnapshot(estate->es_crosscheck_snapshot);
@@ -706,15 +678,6 @@ ExecCheckXactReadOnly(PlannedStmt *plannedstmt)
 {
 	ListCell   *l;
 
-	/*
-	 * CREATE TABLE AS or SELECT INTO?
-	 *
-	 * XXX should we allow this if the destination is temp?  Considering that
-	 * it would still require catalog changes, probably not.
-	 */
-	if (plannedstmt->intoClause != NULL)
-		PreventCommandIfReadOnly(CreateCommandTag((Node *) plannedstmt));
-
 	/* Fail if write permissions are requested on any non-temp table */
 	foreach(l, plannedstmt->rtable)
 	{
@@ -864,18 +827,6 @@ InitPlan(QueryDesc *queryDesc, int eflags)
 	}
 
 	/*
-	 * Detect whether we're doing SELECT INTO.  If so, set the es_into_oids
-	 * flag appropriately so that the plan tree will be initialized with the
-	 * correct tuple descriptors.  (Other SELECT INTO stuff comes later.)
-	 */
-	estate->es_select_into = false;
-	if (operation == CMD_SELECT && plannedstmt->intoClause != NULL)
-	{
-		estate->es_select_into = true;
-		estate->es_into_oids = interpretOidsOption(plannedstmt->intoClause->options);
-	}
-
-	/*
 	 * Initialize the executor's tuple table to empty.
 	 */
 	estate->es_tupleTable = NIL;
@@ -926,9 +877,7 @@ InitPlan(QueryDesc *queryDesc, int eflags)
 	planstate = ExecInitNode(plan, estate, eflags);
 
 	/*
-	 * Get the tuple descriptor describing the type of tuples to return. (this
-	 * is especially important if we are creating a relation with "SELECT
-	 * INTO")
+	 * Get the tuple descriptor describing the type of tuples to return.
 	 */
 	tupType = ExecGetResultType(planstate);
 
@@ -968,16 +917,6 @@ InitPlan(QueryDesc *queryDesc, int eflags)
 
 	queryDesc->tupDesc = tupType;
 	queryDesc->planstate = planstate;
-
-	/*
-	 * If doing SELECT INTO, initialize the "into" relation.  We must wait
-	 * till now so we have the "clean" result tuple type to create the new
-	 * table from.
-	 *
-	 * If EXPLAIN, skip creating the "into" relation.
-	 */
-	if (estate->es_select_into && !(eflags & EXEC_FLAG_EXPLAIN_ONLY))
-		OpenIntoRel(queryDesc);
 }
 
 /*
@@ -1230,7 +1169,7 @@ ExecGetTriggerResultRel(EState *estate, Oid relid)
 /*
  *		ExecContextForcesOids
  *
- * This is pretty grotty: when doing INSERT, UPDATE, or SELECT INTO,
+ * This is pretty grotty: when doing INSERT, UPDATE, or CREATE TABLE AS,
  * we need to ensure that result tuples have space for an OID iff they are
  * going to be stored into a relation that has OIDs.  In other contexts
  * we are free to choose whether to leave space for OIDs in result tuples
@@ -1255,9 +1194,9 @@ ExecGetTriggerResultRel(EState *estate, Oid relid)
  * the ModifyTable node, so ModifyTable has to set es_result_relation_info
  * while initializing each subplan.
  *
- * SELECT INTO is even uglier, because we don't have the INTO relation's
- * descriptor available when this code runs; we have to look aside at a
- * flag set by InitPlan().
+ * CREATE TABLE AS is even uglier, because we don't have the target relation's
+ * descriptor available when this code runs; we have to look aside at the
+ * flags passed to ExecutorStart().
  */
 bool
 ExecContextForcesOids(PlanState *planstate, bool *hasoids)
@@ -1275,9 +1214,14 @@ ExecContextForcesOids(PlanState *planstate, bool *hasoids)
 		}
 	}
 
-	if (planstate->state->es_select_into)
+	if (planstate->state->es_top_eflags & EXEC_FLAG_WITH_OIDS)
 	{
-		*hasoids = planstate->state->es_into_oids;
+		*hasoids = true;
+		return true;
+	}
+	if (planstate->state->es_top_eflags & EXEC_FLAG_WITHOUT_OIDS)
+	{
+		*hasoids = false;
 		return true;
 	}
 
@@ -2290,8 +2234,6 @@ EvalPlanQualStart(EPQState *epqstate, EState *parentestate, Plan *planTree)
 	estate->es_rowMarks = parentestate->es_rowMarks;
 	estate->es_top_eflags = parentestate->es_top_eflags;
 	estate->es_instrument = parentestate->es_instrument;
-	estate->es_select_into = parentestate->es_select_into;
-	estate->es_into_oids = parentestate->es_into_oids;
 	/* es_auxmodifytables must NOT be copied */
 
 	/*
@@ -2423,351 +2365,3 @@ EvalPlanQualEnd(EPQState *epqstate)
 	epqstate->planstate = NULL;
 	epqstate->origslot = NULL;
 }
-
-
-/*
- * Support for SELECT INTO (a/k/a CREATE TABLE AS)
- *
- * We implement SELECT INTO by diverting SELECT's normal output with
- * a specialized DestReceiver type.
- */
-
-typedef struct
-{
-	DestReceiver pub;			/* publicly-known function pointers */
-	EState	   *estate;			/* EState we are working with */
-	DestReceiver *origdest;		/* QueryDesc's original receiver */
-	Relation	rel;			/* Relation to write to */
-	int			hi_options;		/* heap_insert performance options */
-	BulkInsertState bistate;	/* bulk insert state */
-} DR_intorel;
-
-/*
- * OpenIntoRel --- actually create the SELECT INTO target relation
- *
- * This also replaces QueryDesc->dest with the special DestReceiver for
- * SELECT INTO.  We assume that the correct result tuple type has already
- * been placed in queryDesc->tupDesc.
- */
-static void
-OpenIntoRel(QueryDesc *queryDesc)
-{
-	IntoClause *into = queryDesc->plannedstmt->intoClause;
-	EState	   *estate = queryDesc->estate;
-	TupleDesc	intoTupDesc = queryDesc->tupDesc;
-	Relation	intoRelationDesc;
-	char	   *intoName;
-	Oid			namespaceId;
-	Oid			tablespaceId;
-	Datum		reloptions;
-	Oid			intoRelationId;
-	DR_intorel *myState;
-	RangeTblEntry  *rte;
-	AttrNumber		attnum;
-	static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
-
-	Assert(into);
-
-	/*
-	 * XXX This code needs to be kept in sync with DefineRelation(). Maybe we
-	 * should try to use that function instead.
-	 */
-
-	/*
-	 * Check consistency of arguments
-	 */
-	if (into->onCommit != ONCOMMIT_NOOP
-		&& into->rel->relpersistence != RELPERSISTENCE_TEMP)
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
-				 errmsg("ON COMMIT can only be used on temporary tables")));
-
-	{
-		AclResult aclresult;
-		int i;
-
-		for (i = 0; i < intoTupDesc->natts; i++)
-		{
-			Oid atttypid = intoTupDesc->attrs[i]->atttypid;
-
-			aclresult = pg_type_aclcheck(atttypid, GetUserId(), ACL_USAGE);
-			if (aclresult != ACLCHECK_OK)
-				aclcheck_error(aclresult, ACL_KIND_TYPE,
-							   format_type_be(atttypid));
-		}
-	}
-
-	/*
-	 * 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.)  It would probably be all right to scribble directly on
-	 * the query's result tupdesc, but let's be safe and make a copy.
-	 */
-	if (into->colNames)
-	{
-		ListCell   *lc;
-
-		intoTupDesc = CreateTupleDescCopy(intoTupDesc);
-		attnum = 1;
-		foreach(lc, into->colNames)
-		{
-			char	   *colname = strVal(lfirst(lc));
-
-			if (attnum > intoTupDesc->natts)
-				ereport(ERROR,
-						(errcode(ERRCODE_SYNTAX_ERROR),
-						 errmsg("CREATE TABLE AS specifies too many column names")));
-			namestrcpy(&(intoTupDesc->attrs[attnum - 1]->attname), colname);
-			attnum++;
-		}
-	}
-
-	/*
-	 * Find namespace to create in, check its permissions, lock it against
-	 * concurrent drop, and mark into->rel as RELPERSISTENCE_TEMP if the
-	 * selected namespace is temporary.
-	 */
-	intoName = into->rel->relname;
-	namespaceId = RangeVarGetAndCheckCreationNamespace(into->rel, NoLock,
-													   NULL);
-
-	/*
-	 * Security check: disallow creating temp tables from security-restricted
-	 * code.  This is needed because calling code might not expect untrusted
-	 * tables to appear in pg_temp at the front of its search path.
-	 */
-	if (into->rel->relpersistence == RELPERSISTENCE_TEMP
-		&& InSecurityRestrictedOperation())
-		ereport(ERROR,
-				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("cannot create temporary table within security-restricted operation")));
-
-	/*
-	 * Select tablespace to use.  If not specified, use default tablespace
-	 * (which may in turn default to database's default).
-	 */
-	if (into->tableSpaceName)
-	{
-		tablespaceId = get_tablespace_oid(into->tableSpaceName, false);
-	}
-	else
-	{
-		tablespaceId = GetDefaultTablespace(into->rel->relpersistence);
-		/* note InvalidOid is OK in this case */
-	}
-
-	/* Check permissions except when using the database's default space */
-	if (OidIsValid(tablespaceId) && tablespaceId != MyDatabaseTableSpace)
-	{
-		AclResult	aclresult;
-
-		aclresult = pg_tablespace_aclcheck(tablespaceId, GetUserId(),
-										   ACL_CREATE);
-
-		if (aclresult != ACLCHECK_OK)
-			aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
-						   get_tablespace_name(tablespaceId));
-	}
-
-	/* Parse and validate any reloptions */
-	reloptions = transformRelOptions((Datum) 0,
-									 into->options,
-									 NULL,
-									 validnsps,
-									 true,
-									 false);
-	(void) heap_reloptions(RELKIND_RELATION, reloptions, true);
-
-	/* Now we can actually create the new relation */
-	intoRelationId = heap_create_with_catalog(intoName,
-											  namespaceId,
-											  tablespaceId,
-											  InvalidOid,
-											  InvalidOid,
-											  InvalidOid,
-											  GetUserId(),
-											  intoTupDesc,
-											  NIL,
-											  RELKIND_RELATION,
-											  into->rel->relpersistence,
-											  false,
-											  false,
-											  true,
-											  0,
-											  into->onCommit,
-											  reloptions,
-											  true,
-											  allowSystemTableMods);
-	Assert(intoRelationId != InvalidOid);
-
-	/*
-	 * Advance command counter so that the newly-created relation's catalog
-	 * tuples will be visible to heap_open.
-	 */
-	CommandCounterIncrement();
-
-	/*
-	 * If necessary, create a TOAST table for the INTO relation. Note that
-	 * AlterTableCreateToastTable ends with CommandCounterIncrement(), so that
-	 * the TOAST table will be visible for insertion.
-	 */
-	reloptions = transformRelOptions((Datum) 0,
-									 into->options,
-									 "toast",
-									 validnsps,
-									 true,
-									 false);
-
-	(void) heap_reloptions(RELKIND_TOASTVALUE, reloptions, true);
-
-	AlterTableCreateToastTable(intoRelationId, reloptions);
-
-	/*
-	 * And open the constructed table for writing.
-	 */
-	intoRelationDesc = heap_open(intoRelationId, AccessExclusiveLock);
-
-	/*
-	 * Check INSERT permission on the constructed table.
-	 */
-	rte = makeNode(RangeTblEntry);
-	rte->rtekind = RTE_RELATION;
-	rte->relid = intoRelationId;
-	rte->relkind = RELKIND_RELATION;
-	rte->requiredPerms = ACL_INSERT;
-
-	for (attnum = 1; attnum <= intoTupDesc->natts; attnum++)
-		rte->modifiedCols = bms_add_member(rte->modifiedCols,
-				attnum - FirstLowInvalidHeapAttributeNumber);
-
-	ExecCheckRTPerms(list_make1(rte), true);
-
-	/*
-	 * Now replace the query's DestReceiver with one for SELECT INTO
-	 */
-	myState = (DR_intorel *) CreateDestReceiver(DestIntoRel);
-	Assert(myState->pub.mydest == DestIntoRel);
-	myState->estate = estate;
-	myState->origdest = queryDesc->dest;
-	myState->rel = intoRelationDesc;
-
-	queryDesc->dest = (DestReceiver *) myState;
-
-	/*
-	 * 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);
-}
-
-/*
- * CloseIntoRel --- clean up SELECT INTO at ExecutorEnd time
- */
-static void
-CloseIntoRel(QueryDesc *queryDesc)
-{
-	DR_intorel *myState = (DR_intorel *) queryDesc->dest;
-
-	/*
-	 * OpenIntoRel might never have gotten called, and we also want to guard
-	 * against double destruction.
-	 */
-	if (myState && myState->pub.mydest == DestIntoRel)
-	{
-		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);
-
-		/* restore the receiver belonging to executor's caller */
-		queryDesc->dest = myState->origdest;
-
-		/* might as well invoke my destructor */
-		intorel_destroy((DestReceiver *) myState);
-	}
-}
-
-/*
- * CreateIntoRelDestReceiver -- create a suitable DestReceiver object
- */
-DestReceiver *
-CreateIntoRelDestReceiver(void)
-{
-	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;
-
-	/* private fields will be set by OpenIntoRel */
-
-	return (DestReceiver *) self;
-}
-
-/*
- * intorel_startup --- executor startup
- */
-static void
-intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
-{
-	/* no-op */
-}
-
-/*
- * 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->estate->es_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)
-{
-	/* no-op */
-}
-
-/*
- * intorel_destroy --- release DestReceiver object
- */
-static void
-intorel_destroy(DestReceiver *self)
-{
-	pfree(self);
-}
diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
index 6db42e7b970..40cd5ce5d19 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -137,8 +137,6 @@ CreateExecutorState(void)
 
 	estate->es_top_eflags = 0;
 	estate->es_instrument = 0;
-	estate->es_select_into = false;
-	estate->es_into_oids = false;
 	estate->es_finished = false;
 
 	estate->es_exprcontexts = NIL;
diff --git a/src/backend/executor/functions.c b/src/backend/executor/functions.c
index 61f462254ff..ae8d374db21 100644
--- a/src/backend/executor/functions.c
+++ b/src/backend/executor/functions.c
@@ -535,7 +535,6 @@ init_execution_state(List *queryTree_list,
 
 			if (ps->commandType == CMD_SELECT &&
 				ps->utilityStmt == NULL &&
-				ps->intoClause == NULL &&
 				!ps->hasModifyingCTE)
 				fcache->lazyEval = lasttages->lazyEval = true;
 		}
@@ -1493,8 +1492,7 @@ check_sql_fn_retval(Oid func_id, Oid rettype, List *queryTreeList,
 	 */
 	if (parse &&
 		parse->commandType == CMD_SELECT &&
-		parse->utilityStmt == NULL &&
-		parse->intoClause == NULL)
+		parse->utilityStmt == NULL)
 	{
 		tlist_ptr = &parse->targetList;
 		tlist = parse->targetList;
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 81f284ca044..5e4ae426b1b 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -1284,11 +1284,11 @@ SPI_cursor_open_internal(const char *name, SPIPlanPtr plan,
 	 * Start portal execution.
 	 */
 	if (read_only)
-		PortalStart(portal, paramLI, true);
+		PortalStart(portal, paramLI, 0, true);
 	else
 	{
 		CommandCounterIncrement();
-		PortalStart(portal, paramLI, false);
+		PortalStart(portal, paramLI, 0, false);
 	}
 
 	Assert(portal->strategy != PORTAL_MULTI_QUERY);
@@ -1907,17 +1907,39 @@ _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI,
 			}
 			else
 			{
+				char	completionTag[COMPLETION_TAG_BUFSIZE];
+
 				ProcessUtility(stmt,
 							   plansource->query_string,
 							   paramLI,
 							   false,	/* not top level */
 							   dest,
-							   NULL);
+							   completionTag);
+
 				/* Update "processed" if stmt returned tuples */
 				if (_SPI_current->tuptable)
 					_SPI_current->processed = _SPI_current->tuptable->alloced -
 						_SPI_current->tuptable->free;
-				res = SPI_OK_UTILITY;
+
+				/*
+				 * CREATE TABLE AS is a messy special case for historical
+				 * reasons.  We must set _SPI_current->processed even though
+				 * the tuples weren't returned to the caller, and we must
+				 * return a special result code if the statement was spelled
+				 * SELECT INTO.
+				 */
+				if (IsA(stmt, CreateTableAsStmt))
+				{
+					Assert(strncmp(completionTag, "SELECT ", 7) == 0);
+					_SPI_current->processed = strtoul(completionTag + 7,
+													  NULL, 10);
+					if (((CreateTableAsStmt *) stmt)->is_select_into)
+						res = SPI_OK_SELINTO;
+					else
+						res = SPI_OK_UTILITY;
+				}
+				else
+					res = SPI_OK_UTILITY;
 			}
 
 			/*
@@ -2042,9 +2064,7 @@ _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, long tcount)
 	{
 		case CMD_SELECT:
 			Assert(queryDesc->plannedstmt->utilityStmt == NULL);
-			if (queryDesc->plannedstmt->intoClause)		/* select into table? */
-				res = SPI_OK_SELINTO;
-			else if (queryDesc->dest->mydest != DestSPI)
+			if (queryDesc->dest->mydest != DestSPI)
 			{
 				/* Don't return SPI_OK_SELECT if we're discarding result */
 				res = SPI_OK_UTILITY;