/*-------------------------------------------------------------------------

 *

 * createas.c

 *    Execution of CREATE TABLE ... AS, a/k/a SELECT INTO.

 *    Since CREATE MATERIALIZED VIEW shares syntax and most behaviors,

 *    we implement that here, too.

 *

 * We implement this by diverting the query's normal output to a

 * specialized DestReceiver type.

 *

 * Formerly, CTAS 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 QueryCompletion.  (We no

 * longer do that for CTAS ... WITH NO DATA, however.)

 *

 * Portions Copyright (c) 1996-2025, 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/heapam.h"

#include "access/reloptions.h"

#include "access/tableam.h"

#include "access/xact.h"

#include "catalog/namespace.h"

#include "catalog/toasting.h"

#include "commands/createas.h"

#include "commands/matview.h"

#include "commands/prepare.h"

#include "commands/tablecmds.h"

#include "commands/view.h"

#include "executor/execdesc.h"

#include "executor/executor.h"

#include "nodes/makefuncs.h"

#include "nodes/nodeFuncs.h"

#include "nodes/queryjumble.h"

#include "parser/analyze.h"

#include "rewrite/rewriteHandler.h"

#include "tcop/tcopprot.h"

#include "utils/builtins.h"

#include "utils/lsyscache.h"

#include "utils/rls.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 */

  ObjectAddress reladdr;    /* address of rel, for ExecCreateTableAs */

  CommandId output_cid;   /* cmin to insert in output tuples */

  int     ti_options;   /* table_tuple_insert performance options */

  BulkInsertState bistate;  /* bulk insert state */

} DR_intorel;

/* utility functions for CTAS definition creation */

static ObjectAddress create_ctas_internal(List *attrList, IntoClause *into);

static ObjectAddress create_ctas_nodata(List *tlist, IntoClause *into);

/* DestReceiver routines for collecting data */

static void intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);

static bool intorel_receive(TupleTableSlot *slot, DestReceiver *self);

static void intorel_shutdown(DestReceiver *self);

static void intorel_destroy(DestReceiver *self);

/*

 * create_ctas_internal

 *

 * Internal utility used for the creation of the definition of a relation

 * created via CREATE TABLE AS or a materialized view.  Caller needs to

 * provide a list of attributes (ColumnDef nodes).

 */

static ObjectAddress

create_ctas_internal(List *attrList, IntoClause *into)

{

  CreateStmt *create = makeNode(CreateStmt);

  bool    is_matview;

  char    relkind;

  Datum   toast_options;

  const char *const validnsps[] = HEAP_RELOPT_NAMESPACES;

  ObjectAddress intoRelationAddr;

  /* This code supports both CREATE TABLE AS and CREATE MATERIALIZED VIEW */

  is_matview = (into->viewQuery != NULL);

  relkind = is_matview ? RELKIND_MATVIEW : RELKIND_RELATION;

  /*

   * Create the target relation by faking up a CREATE TABLE parsetree and

   * passing it to DefineRelation.

   */

  create->relation = into->rel;

  create->tableElts = attrList;

  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;

  create->accessMethod = into->accessMethod;

  /*

   * Create the relation.  (This will error out if there's an existing view,

   * so we don't need more code to complain if "replace" is false.)

   */

  intoRelationAddr = DefineRelation(create, relkind, InvalidOid, NULL, NULL);

  /*

   * If necessary, create a TOAST table for the target table.  Note that

   * NewRelationCreateToastTable 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);

  NewRelationCreateToastTable(intoRelationAddr.objectId, toast_options);

  /* Create the "view" part of a materialized view. */

  if (is_matview)

  {

    /* StoreViewQuery scribbles on tree, so make a copy */

    Query    *query = copyObject(into->viewQuery);

    StoreViewQuery(intoRelationAddr.objectId, query, false);

    CommandCounterIncrement();

  }

  return intoRelationAddr;

}

/*

 * create_ctas_nodata

 *

 * Create CTAS or materialized view when WITH NO DATA is used, starting from

 * the targetlist of the SELECT or view definition.

 */

static ObjectAddress

create_ctas_nodata(List *tlist, IntoClause *into)

{

  List     *attrList;

  ListCell   *t,

         *lc;

  /*

   * Build list of ColumnDefs from non-junk elements of the tlist.  If a

   * column name list was specified in CREATE TABLE AS, override the column

   * names in the query.  (Too few column names are OK, too many are not.)

   */

  attrList = NIL;

  lc = list_head(into->colNames);

  foreach(t, tlist)

  {

    TargetEntry *tle = (TargetEntry *) lfirst(t);

    if (!tle->resjunk)

    {

      ColumnDef  *col;

      char     *colname;

      if (lc)

      {

        colname = strVal(lfirst(lc));

        lc = lnext(into->colNames, lc);

      }

      else

        colname = tle->resname;

      col = makeColumnDef(colname,

                exprType((Node *) tle->expr),

                exprTypmod((Node *) tle->expr),

                exprCollation((Node *) tle->expr));

      /*

       * 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(col->typeName->typeOid))

        ereport(ERROR,

            (errcode(ERRCODE_INDETERMINATE_COLLATION),

             errmsg("no collation was derived for column \"%s\" with collatable type %s",

                col->colname,

                format_type_be(col->typeName->typeOid)),

             errhint("Use the COLLATE clause to set the collation explicitly.")));

      attrList = lappend(attrList, col);

    }

  }

  if (lc != NULL)

    ereport(ERROR,

        (errcode(ERRCODE_SYNTAX_ERROR),

         errmsg("too many column names were specified")));

  /* Create the relation definition using the ColumnDef list */

  return create_ctas_internal(attrList, into);

}

/*

 * ExecCreateTableAs -- execute a CREATE TABLE AS command

 */

ObjectAddress

ExecCreateTableAs(ParseState *pstate, CreateTableAsStmt *stmt,

          ParamListInfo params, QueryEnvironment *queryEnv,

          QueryCompletion *qc)

{

  Query    *query = castNode(Query, stmt->query);

  IntoClause *into = stmt->into;

  JumbleState *jstate = NULL;

  bool    is_matview = (into->viewQuery != NULL);

  bool    do_refresh = false;

  DestReceiver *dest;

  ObjectAddress address;

  /* Check if the relation exists or not */

  if (CreateTableAsRelExists(stmt))

    return InvalidObjectAddress;

  /*

   * Create the tuple receiver object and insert info it will need

   */

  dest = CreateIntoRelDestReceiver(into);

  /* Query contained by CTAS needs to be jumbled if requested */

  if (IsQueryIdEnabled())

    jstate = JumbleQuery(query);

  if (post_parse_analyze_hook)

    (*post_parse_analyze_hook) (pstate, query, jstate);

  /*

   * The contained Query could be a SELECT, or an EXECUTE utility command.

   * If the latter, we just pass it off to ExecuteQuery.

   */

  if (query->commandType == CMD_UTILITY &&

    IsA(query->utilityStmt, ExecuteStmt))

  {

    ExecuteStmt *estmt = castNode(ExecuteStmt, query->utilityStmt);

    Assert(!is_matview);  /* excluded by syntax */

    ExecuteQuery(pstate, estmt, into, params, dest, qc);

    /* get object address that intorel_startup saved for us */

    address = ((DR_intorel *) dest)->reladdr;

    return address;

  }

  Assert(query->commandType == CMD_SELECT);

  /*

   * For materialized views, always skip data during table creation, and use

   * REFRESH instead (see below).

   */

  if (is_matview)

  {

    do_refresh = !into->skipData;

    into->skipData = true;

  }

  if (into->skipData)

  {

    /*

     * If WITH NO DATA was specified, do not go through the rewriter,

     * planner and executor.  Just define the relation using a code path

     * similar to CREATE VIEW.  This avoids dump/restore problems stemming

     * from running the planner before all dependencies are set up.

     */

    address = create_ctas_nodata(query->targetList, into);

    /*

     * For materialized views, reuse the REFRESH logic, which locks down

     * security-restricted operations and restricts the search_path.  This

     * reduces the chance that a subsequent refresh will fail.

     */

    if (do_refresh)

      RefreshMatViewByOid(address.objectId, true, false, false,

                pstate->p_sourcetext, qc);

  }

  else

  {

    List     *rewritten;

    PlannedStmt *plan;

    QueryDesc  *queryDesc;

    Assert(!is_matview);

    /*

     * 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.

     */

    rewritten = QueryRewrite(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 = linitial_node(Query, rewritten);

    Assert(query->commandType == CMD_SELECT);

    /* plan the query */

    plan = pg_plan_query(query, pstate->p_sourcetext,

               CURSOR_OPT_PARALLEL_OK, 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, pstate->p_sourcetext,

                  GetActiveSnapshot(), InvalidSnapshot,

                  dest, params, queryEnv, 0);

    /* call ExecutorStart to prepare the plan for execution */

    ExecutorStart(queryDesc, GetIntoRelEFlags(into));

    /* run the plan to completion */

    ExecutorRun(queryDesc, ForwardScanDirection, 0);

    /* save the rowcount if we're given a qc to fill */

    if (qc)

      SetQueryCompletion(qc, CMDTAG_SELECT, queryDesc->estate->es_processed);

    /* get object address that intorel_startup saved for us */

    address = ((DR_intorel *) dest)->reladdr;

    /* and clean up */

    ExecutorFinish(queryDesc);

    ExecutorEnd(queryDesc);

    FreeQueryDesc(queryDesc);

    PopActiveSnapshot();

  }

  return address;

}

/*

 * 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)

{

  int     flags = 0;

  if (intoClause->skipData)

    flags |= EXEC_FLAG_WITH_NO_DATA;

  return flags;

}

/*

 * CreateTableAsRelExists --- check existence of relation for CreateTableAsStmt

 *

 * Utility wrapper checking if the relation pending for creation in this

 * CreateTableAsStmt query already exists or not.  Returns true if the

 * relation exists, otherwise false.

 */

bool

CreateTableAsRelExists(CreateTableAsStmt *ctas)

{

  Oid     nspid;

  Oid     oldrelid;

  ObjectAddress address;

  IntoClause *into = ctas->into;

  nspid = RangeVarGetCreationNamespace(into->rel);

  oldrelid = get_relname_relid(into->rel->relname, nspid);

  if (OidIsValid(oldrelid))

  {

    if (!ctas->if_not_exists)

      ereport(ERROR,

          (errcode(ERRCODE_DUPLICATE_TABLE),

           errmsg("relation \"%s\" already exists",

              into->rel->relname)));

    /*

     * The relation exists and IF NOT EXISTS has been specified.

     *

     * If we are in an extension script, insist that the pre-existing

     * object be a member of the extension, to avoid security risks.

     */

    ObjectAddressSet(address, RelationRelationId, oldrelid);

    checkMembershipInCurrentExtension(&address);

    /* OK to skip */

    ereport(NOTICE,

        (errcode(ERRCODE_DUPLICATE_TABLE),

         errmsg("relation \"%s\" already exists, skipping",

            into->rel->relname)));

    return true;

  }

  /* Relation does not exist, it can be created */

  return false;

}

/*

 * 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;

  bool    is_matview;

  List     *attrList;

  ObjectAddress intoRelationAddr;

  Relation  intoRelationDesc;

  ListCell   *lc;

  int     attnum;

  Assert(into != NULL);   /* else somebody forgot to set it */

  /* This code supports both CREATE TABLE AS and CREATE MATERIALIZED VIEW */

  is_matview = (into->viewQuery != NULL);

  /*

   * 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.)

   */

  attrList = NIL;

  lc = list_head(into->colNames);

  for (attnum = 0; attnum < typeinfo->natts; attnum++)

  {

    Form_pg_attribute attribute = TupleDescAttr(typeinfo, attnum);

    ColumnDef  *col;

    char     *colname;

    if (lc)

    {

      colname = strVal(lfirst(lc));

      lc = lnext(into->colNames, lc);

    }

    else

      colname = NameStr(attribute->attname);

    col = makeColumnDef(colname,

              attribute->atttypid,

              attribute->atttypmod,

              attribute->attcollation);

    /*

     * 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(col->typeName->typeOid))

      ereport(ERROR,

          (errcode(ERRCODE_INDETERMINATE_COLLATION),

           errmsg("no collation was derived for column \"%s\" with collatable type %s",

              col->colname,

              format_type_be(col->typeName->typeOid)),

           errhint("Use the COLLATE clause to set the collation explicitly.")));

    attrList = lappend(attrList, col);

  }

  if (lc != NULL)

    ereport(ERROR,

        (errcode(ERRCODE_SYNTAX_ERROR),

         errmsg("too many column names were specified")));

  /*

   * Actually create the target table

   */

  intoRelationAddr = create_ctas_internal(attrList, into);

  /*

   * Finally we can open the target table

   */

  intoRelationDesc = table_open(intoRelationAddr.objectId, AccessExclusiveLock);

  /*

   * Make sure the constructed table does not have RLS enabled.

   *

   * check_enable_rls() will ereport(ERROR) itself if the user has requested

   * something invalid, and otherwise will return RLS_ENABLED if RLS should

   * be enabled here.  We don't actually support that currently, so throw

   * our own ereport(ERROR) if that happens.

   */

  if (check_enable_rls(intoRelationAddr.objectId, InvalidOid, false) == RLS_ENABLED)

    ereport(ERROR,

        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),

         errmsg("policies not yet implemented for this command")));

  /*

   * Tentatively mark the target as populated, if it's a matview and we're

   * going to fill it; otherwise, no change needed.

   */

  if (is_matview && !into->skipData)

    SetMatViewPopulatedState(intoRelationDesc, true);

  /*

   * Fill private fields of myState for use by later routines

   */

  myState->rel = intoRelationDesc;

  myState->reladdr = intoRelationAddr;

  myState->output_cid = GetCurrentCommandId(true);

  myState->ti_options = TABLE_INSERT_SKIP_FSM;

  /*

   * If WITH NO DATA is specified, there is no need to set up the state for

   * bulk inserts as there are no tuples to insert.

   */

  if (!into->skipData)

    myState->bistate = GetBulkInsertState();

  else

    myState->bistate = NULL;

  /*

   * Valid smgr_targblock implies something already wrote to the relation.

   * This may be harmless, but this function hasn't planned for it.

   */

  Assert(RelationGetTargetBlock(intoRelationDesc) == InvalidBlockNumber);

}

/*

 * intorel_receive --- receive one tuple

 */

static bool

intorel_receive(TupleTableSlot *slot, DestReceiver *self)

{

  DR_intorel *myState = (DR_intorel *) self;

  /* Nothing to insert if WITH NO DATA is specified. */

  if (!myState->into->skipData)

  {

    /*

     * Note that the input slot might not be of the type of the target

     * relation. That's supported by table_tuple_insert(), but slightly

     * less efficient than inserting with the right slot - but the

     * alternative would be to copy into a slot of the right type, which

     * would not be cheap either. This also doesn't allow accessing per-AM

     * data (say a tuple's xmin), but since we don't do that here...

     */

    table_tuple_insert(myState->rel,

               slot,

               myState->output_cid,

               myState->ti_options,

               myState->bistate);

  }

  /* We know this is a newly created relation, so there are no indexes */

  return true;

}

/*

 * intorel_shutdown --- executor end

 */

static void

intorel_shutdown(DestReceiver *self)

{

  DR_intorel *myState = (DR_intorel *) self;

  IntoClause *into = myState->into;

  if (!into->skipData)

  {

    FreeBulkInsertState(myState->bistate);

    table_finish_bulk_insert(myState->rel, myState->ti_options);

  }

  /* close rel, but keep lock until commit */

  table_close(myState->rel, NoLock);

  myState->rel = NULL;

}

/*

 * intorel_destroy --- release DestReceiver object

 */

static void

intorel_destroy(DestReceiver *self)

{

  pfree(self);

}