Set range table for CopyFrom() in tablesync

CopyFrom() needs a range table for formatting certain errors for
constraint violations.

This changes the mechanism of how the range table is passed to the
CopyFrom() executor state.  We used to generate the range table and one
entry for the relation manually inside DoCopy().  Now we use
addRangeTableEntryForRelation() to setup the range table and relation
entry for the ParseState, which is then passed down by BeginCopyFrom().

Author: Petr Jelinek <[email protected]>
Reported-by: Euler Taveira <[email protected]>

diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index b5af2be39bd735048cfee71122a254e479278525..bcaa58cae0ea5f4043f2860173877b1b274328ea 100644 (file)
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -37,6 +37,7 @@
 #include "optimizer/clauses.h"
 #include "optimizer/planner.h"
 #include "nodes/makefuncs.h"
+#include "parser/parse_relation.h"
 #include "rewrite/rewriteHandler.h"
 #include "storage/fd.h"
 #include "tcop/tcopprot.h"
@@ -787,7 +788,6 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt,
    Relation    rel;
    Oid         relid;
    RawStmt    *query = NULL;
-   List       *range_table = NIL;
 
    /* Disallow COPY to/from file or program except to superusers. */
    if (!pipe && !superuser())
@@ -809,7 +809,6 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt,
    if (stmt->relation)
    {
        TupleDesc   tupDesc;
-       AclMode     required_access = (is_from ? ACL_INSERT : ACL_SELECT);
        List       *attnums;
        ListCell   *cur;
        RangeTblEntry *rte;
@@ -822,12 +821,8 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt,
 
        relid = RelationGetRelid(rel);
 
-       rte = makeNode(RangeTblEntry);
-       rte->rtekind = RTE_RELATION;
-       rte->relid = RelationGetRelid(rel);
-       rte->relkind = rel->rd_rel->relkind;
-       rte->requiredPerms = required_access;
-       range_table = list_make1(rte);
+       rte = addRangeTableEntryForRelation(pstate, rel, NULL, false, false);
+       rte->requiredPerms = (is_from ? ACL_INSERT : ACL_SELECT);
 
        tupDesc = RelationGetDescr(rel);
        attnums = CopyGetAttnums(tupDesc, rel, stmt->attlist);
@@ -841,7 +836,7 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt,
            else
                rte->selectedCols = bms_add_member(rte->selectedCols, attno);
        }
-       ExecCheckRTPerms(range_table, true);
+       ExecCheckRTPerms(pstate->p_rtable, true);
 
        /*
         * Permission check for row security policies.
@@ -977,7 +972,6 @@ DoCopy(ParseState *pstate, const CopyStmt *stmt,
 
        cstate = BeginCopyFrom(pstate, rel, stmt->filename, stmt->is_program,
                               NULL, stmt->attlist, stmt->options);
-       cstate->range_table = range_table;
        *processed = CopyFrom(cstate);  /* copy from file to database */
        EndCopyFrom(cstate);
    }
@@ -2921,6 +2915,10 @@ BeginCopyFrom(ParseState *pstate,
    cstate->raw_buf = (char *) palloc(RAW_BUF_SIZE + 1);
    cstate->raw_buf_index = cstate->raw_buf_len = 0;
 
+   /* Assign range table, we'll need it in CopyFrom. */
+   if (pstate)
+       cstate->range_table = pstate->p_rtable;
+
    tupDesc = RelationGetDescr(cstate->rel);
    attr = tupDesc->attrs;
    num_phys_attrs = tupDesc->natts;

diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index d287e95df1ded4eadf9770bef6974314d1b0c69c..108326bef1c4100b3c8e1a3f70436b01e881c0e3 100644 (file)
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -93,6 +93,8 @@
 
 #include "commands/copy.h"
 
+#include "parser/parse_relation.h"
+
 #include "replication/logicallauncher.h"
 #include "replication/logicalrelation.h"
 #include "replication/walreceiver.h"
@@ -654,6 +656,7 @@ copy_table(Relation rel)
    StringInfoData      cmd;
    CopyState   cstate;
    List       *attnamelist;
+   ParseState *pstate;
 
    /* Get the publisher relation info. */
    fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
@@ -680,9 +683,11 @@ copy_table(Relation rel)
 
    copybuf = makeStringInfo();
 
-   /* Create CopyState for ingestion of the data from publisher. */
+   pstate = make_parsestate(NULL);
+   addRangeTableEntryForRelation(pstate, rel, NULL, false, false);
+
    attnamelist = make_copy_attnamelist(relmapentry);
-   cstate = BeginCopyFrom(NULL, rel, NULL, false, copy_read_data, attnamelist, NIL);
+   cstate = BeginCopyFrom(pstate, rel, NULL, false, copy_read_data, attnamelist, NIL);
 
    /* Do the copy */
    (void) CopyFrom(cstate);