PostgreSQL rows return by ProcessQuery queryDesc->estate->es_processed

5 minute read

背景

群里一位朋友问到的, 关于执行插入SQL后, 有些返回0, 有些返回1的情况.

一般我们执行一条插入, 可以看到两个值, 一个是last oid, 一个是有多少行被插入.

代码如下 :

                                snprintf(completionTag, COMPLETION_TAG_BUFSIZE,  
                                   "INSERT %u %u", lastOid, queryDesc->estate->es_processed);  

例子 :

创建两个表, 分别使用oids和不使用oids.

digoal=# create table tt(id int, info text) with oids;  
CREATE TABLE  
digoal=# create table tt1(id int, info text) without oids;  
CREATE TABLE  

批量插入没有返回oid的信息, 但是被插入的行如实返回.

digoal=# insert into tt values (1,'test'),(2,'test');  
INSERT 0 2  
digoal=# insert into tt1 values (1,'test'),(2,'test');  
INSERT 0 2  

插入单条时, 可以正常返回oid和插入的行.

digoal=# insert into tt1 values (1,'test');  
INSERT 0 1  
digoal=# insert into tt values (1,'test');  
INSERT 4710093 1  
digoal=# select oid,* from tt;  
   oid   | id | info   
---------+----+------  
 4710091 |  1 | test  
 4710092 |  2 | test  
 4710093 |  1 | test  
(3 rows)  

有一种场景是使用触发器来实现数据插入分区表的情况, 例如 :

digoal=# create table a(id int, info text);  
CREATE TABLE  
digoal=# create table b(id int, info text);  
CREATE TABLE  
digoal=# create or replace function tg1() returns trigger as $$  
declare  
begin  
  if NEW.id>10000 then  
    insert into b values (NEW.*);  
  else  
    return NEW;  
  end if;  
  return null;  -- 注意触发器返回null, before触发器, null的话就等于没什么好做的了  
end;  
$$ language plpgsql strict;  
CREATE FUNCTION  
  
digoal=# create trigger tg1 before insert on a for each row execute procedure tg1();  
CREATE TRIGGER  

这条记录被插入到B表了, 所以返回的行为0. 但是实际上已经插入了.

digoal=# insert into a values (10001,'test');  
INSERT 0 0  
digoal=# insert into a values (1,'test');  
INSERT 0 1  
digoal=# select * from a;  
 id | info   
----+------  
  1 | test  
(1 row)  
digoal=# select * from b;  
  id   | info   
-------+------  
 10001 | test  
(1 row)  
digoal=# insert into a values (10001,'test'),(1,'test'),(10001,'test'),(1,'test');  
INSERT 0 2  

这条SQL实际插入了4行, 但是对于入口表a来说, 只插入了2行. 还有2行通过触发器插入了b表.

另外, 如果使用LIBPQ的话, 可以使用PQcmdTuples来获得类似的信息.

PQcmdTuples  
Returns the number of rows affected by the SQL command.  
  
char *PQcmdTuples(PGresult *res);  
This function returns a string containing the number of rows affected by the SQL statement that generated the PGresult.   
  
This function can only be used following the execution of aSELECT, CREATE TABLE AS, INSERT, UPDATE, DELETE, MOVE, FETCH,   
or COPY statement, or an EXECUTE of a prepared query that contains an INSERT, UPDATE, or DELETE statement.   
  
If the command that generated the PGresult was anything else, PQcmdTuples returns an empty string.   
  
The caller should not free the return value directly. It will be freed when the associated PGresult handle is passed to PQclear.  

参考

1.

src/include/executor/execdesc.h

/* ----------------  
 *              query descriptor:  
 *  
 *      a QueryDesc encapsulates everything that the executor  
 *      needs to execute the query.  
 *  
 *      For the convenience of SQL-language functions, we also support QueryDescs  
 *      containing utility statements; these must not be passed to the executor  
 *      however.  
 * ---------------------  
 */  
typedef struct QueryDesc  
{  
        /* These fields are provided by CreateQueryDesc */  
        CmdType         operation;              /* CMD_SELECT, CMD_UPDATE, etc. */  
        PlannedStmt *plannedstmt;       /* planner's output, or null if utility */  
        Node       *utilitystmt;        /* utility statement, or null */  
        const char *sourceText;         /* source text of the query */  
        Snapshot        snapshot;               /* snapshot to use for query */  
        Snapshot        crosscheck_snapshot;    /* crosscheck for RI update/delete */  
        DestReceiver *dest;                     /* the destination for tuple output */  
        ParamListInfo params;           /* param values being passed in */  
        int                     instrument_options;             /* OR of InstrumentOption flags */  
  
        /* These fields are set by ExecutorStart */  
        TupleDesc       tupDesc;                /* descriptor for result tuples */  
        EState     *estate;                     /* executor's query-wide state */  
        PlanState  *planstate;          /* tree of per-plan-node state */  
  
        /* This is always set NULL by the core system, but plugins can change it */  
        struct Instrumentation *totaltime;      /* total time spent in ExecutorRun */  
} QueryDesc;  

2.

src/include/nodes/execnodes.h

/* ----------------  
 *        ExprContext  
 *  
 *              This class holds the "current context" information  
 *              needed to evaluate expressions for doing tuple qualifications  
 *              and tuple projections.  For example, if an expression refers  
 *              to an attribute in the current inner tuple then we need to know  
 *              what the current inner tuple is and so we look at the expression  
 *              context.  
 *  
 *      There are two memory contexts associated with an ExprContext:  
 *      * ecxt_per_query_memory is a query-lifespan context, typically the same  
 *        context the ExprContext node itself is allocated in.  This context  
 *        can be used for purposes such as storing function call cache info.  
 *      * ecxt_per_tuple_memory is a short-term context for expression results.  
 *        As the name suggests, it will typically be reset once per tuple,  
 *        before we begin to evaluate expressions for that tuple.  Each  
 *        ExprContext normally has its very own per-tuple memory context.  
 *  
 *      CurrentMemoryContext should be set to ecxt_per_tuple_memory before  
 *      calling ExecEvalExpr() --- see ExecEvalExprSwitchContext().  
 * ----------------  
 */  
typedef struct ExprContext  
{  
        NodeTag         type;  
  
        /* Tuples that Var nodes in expression may refer to */  
        TupleTableSlot *ecxt_scantuple;  
        TupleTableSlot *ecxt_innertuple;  
        TupleTableSlot *ecxt_outertuple;  
  
        /* Memory contexts for expression evaluation --- see notes above */  
        MemoryContext ecxt_per_query_memory;  
        MemoryContext ecxt_per_tuple_memory;  
  
        /* Values to substitute for Param nodes in expression */  
        ParamExecData *ecxt_param_exec_vals;            /* for PARAM_EXEC params */  
        ParamListInfo ecxt_param_list_info; /* for other param types */  
  
        /*  
         * Values to substitute for Aggref nodes in the expressions of an Agg  
         * node, or for WindowFunc nodes within a WindowAgg node.  
         */  
        Datum      *ecxt_aggvalues; /* precomputed values for aggs/windowfuncs */  
        bool       *ecxt_aggnulls;      /* null flags for aggs/windowfuncs */  
  
        /* Value to substitute for CaseTestExpr nodes in expression */  
        Datum           caseValue_datum;  
        bool            caseValue_isNull;  
  
        /* Value to substitute for CoerceToDomainValue nodes in expression */  
        Datum           domainValue_datum;  
        bool            domainValue_isNull;  
  
        /* Link to containing EState (NULL if a standalone ExprContext) */  
        struct EState *ecxt_estate;  
  
        /* Functions to call back when ExprContext is shut down */  
        ExprContext_CB *ecxt_callbacks;  
} ExprContext;  

3.

src/backend/tcop/pquery.c

/*  
 * ProcessQuery  
 *              Execute a single plannable query within a PORTAL_MULTI_QUERY,  
 *              PORTAL_ONE_RETURNING, or PORTAL_ONE_MOD_WITH portal  
 *  
 *      plan: the plan tree for the query  
 *      sourceText: the source text of the query  
 *      params: any parameters needed  
 *      dest: where to send results  
 *      completionTag: points to a buffer of size COMPLETION_TAG_BUFSIZE  
 *              in which to store a command completion status string.  
 *  
 * completionTag may be NULL if caller doesn't want a status string.  
 *  
 * Must be called in a memory context that will be reset or deleted on  
 * error; otherwise the executor's memory usage will be leaked.  
 */  
static void  
ProcessQuery(PlannedStmt *plan,  
                         const char *sourceText,  
                         ParamListInfo params,  
                         DestReceiver *dest,  
                         char *completionTag)  
{  
        QueryDesc  *queryDesc;  
  
        elog(DEBUG3, "ProcessQuery");  
  
        /*  
         * Create the QueryDesc object  
         */  
        queryDesc = CreateQueryDesc(plan, sourceText,  
                                                                GetActiveSnapshot(), InvalidSnapshot,  
                                                                dest, params, 0);  
  
        /*  
         * Call ExecutorStart to prepare the plan for execution  
         */  
        ExecutorStart(queryDesc, 0);  
  
        /*  
         * Run the plan to completion.  
         */  
        ExecutorRun(queryDesc, ForwardScanDirection, 0L);  
  
        /*  
         * Build command completion status string, if caller wants one.  
         */  
        if (completionTag)  
        {  
                Oid                     lastOid;  
  
                switch (queryDesc->operation)  
                {  
                        case CMD_SELECT:  
                                snprintf(completionTag, COMPLETION_TAG_BUFSIZE,  
                                                 "SELECT %u", queryDesc->estate->es_processed);  
                                break;  
                        case CMD_INSERT:  
                                if (queryDesc->estate->es_processed == 1)  
                                        lastOid = queryDesc->estate->es_lastoid;  
                                else  
                                        lastOid = InvalidOid;  
                                snprintf(completionTag, COMPLETION_TAG_BUFSIZE,  
                                   "INSERT %u %u", lastOid, queryDesc->estate->es_processed);  
                                break;  
                        case CMD_UPDATE:  
                                snprintf(completionTag, COMPLETION_TAG_BUFSIZE,  
                                                 "UPDATE %u", queryDesc->estate->es_processed);  
                                break;  
                        case CMD_DELETE:  
                                snprintf(completionTag, COMPLETION_TAG_BUFSIZE,  
                                                 "DELETE %u", queryDesc->estate->es_processed);  
                                break;  
                        default:  
                                strcpy(completionTag, "???");  
                                break;  
                }  
        }  
  
        /*  
         * Now, we close down all the scans and free allocated resources.  
         */  
        ExecutorFinish(queryDesc);  
        ExecutorEnd(queryDesc);  
  
        FreeQueryDesc(queryDesc);  
}  

4. http://www.postgresql.org/docs/9.4/static/libpq-exec.html

Flag Counter

digoal’s 大量PostgreSQL文章入口