/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

/*-------------------------------------------------------------------------
 *
 * cdbdispatchresult.c
 *    Functions to dispatch commands to QExecutors
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"
#include <pthread.h>

#include "gp-libpq-fe.h"               /* prerequisite for libpq-int.h */
#include "gp-libpq-int.h"              /* PQExpBufferData */

#include "lib/stringinfo.h"         /* StringInfoData */
#include "utils/guc.h"              /* log_min_messages */
#include "utils/faultinjector.h"

#include "cdb/cdbconn.h"            /* SegmentDatabaseDescriptor */
#include "cdb/cdbpartition.h"
#include "cdb/cdbvars.h"
#include "cdb/cdbsreh.h"
#include "cdb/cdbdispatchresult.h"  /* me */
#include "commands/tablecmds.h"


/* This mutex serializes writes by dispatcher threads to the
 * iFirstError and errcode fields of CdbDispatchResults objects.
 */
static pthread_mutex_t  setErrcodeMutex = PTHREAD_MUTEX_INITIALIZER;


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

/* Make sure buffer has no trailing newline, whitespace, etc. */
static void
noTrailingNewline(StringInfo buf)
{
    while (buf->len > 0 &&
           buf->data[buf->len-1] <= ' ' &&
           buf->data[buf->len-1] > '\0')
        buf->data[--buf->len] = '\0';
}                               /* noTrailingNewline */

static void
noTrailingNewlinePQ(PQExpBuffer buf)
{
    while (buf->len > 0 &&
           buf->data[buf->len-1] <= ' ' &&
           buf->data[buf->len-1] > '\0')
        buf->data[--buf->len] = '\0';
}                               /* noTrailingNewlinePQ */


/* If buffer is nonempty, make sure it has exactly one trailing newline. */
static void
oneTrailingNewline(StringInfo buf)
{
    noTrailingNewline(buf);
    if (buf->len > 0)
        appendStringInfoChar(buf, '\n');
}                               /* oneTrailingNewline */

static void
oneTrailingNewlinePQ(PQExpBuffer buf)
{
    noTrailingNewlinePQ(buf);
    if (buf->len > 0)
        appendPQExpBufferChar(buf, '\n');
}                               /* oneTrailingNewlinePQ */


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

/*
 * Create a CdbDispatchResult object, appending it to the
 * resultArray of a given CdbDispatchResults object.
 */
CdbDispatchResult *
cdbdisp_makeResult(struct CdbDispatchResults           *meleeResults,
                   struct SegmentDatabaseDescriptor    *segdbDesc,
                   int                                  sliceIndex)
{
    CdbDispatchResult  *dispatchResult;
    int                 meleeIndex;

    Assert(meleeResults &&
           meleeResults->resultArray &&
           meleeResults->resultCount < meleeResults->resultCapacity);

    /* Allocate a slot for the new CdbDispatchResult object. */
    meleeIndex = meleeResults->resultCount++;
    dispatchResult = &meleeResults->resultArray[meleeIndex];

#ifdef FAULT_INJECTOR
				FaultInjector_InjectFaultIfSet(
											   CreateCdbDispathResultObject,
											   DDLNotSpecified,
											   "",	// databaseName
											   ""); // tableName
#endif

    /* Initialize CdbDispatchResult. */
    dispatchResult->meleeResults = meleeResults;
    dispatchResult->meleeIndex = meleeIndex;
    dispatchResult->segdbDesc = segdbDesc;
    dispatchResult->resultbuf = createPQExpBuffer();
    dispatchResult->error_message = NULL;
	dispatchResult->numrowsrejected = 0;

	if (PQExpBufferBroken(dispatchResult->resultbuf))
	{
		destroyPQExpBuffer(dispatchResult->resultbuf);
		dispatchResult->resultbuf = NULL;
		/* caller is responsible for cleanup -- can't elog(ERROR, ...) from here. */
		return NULL;
	}

    /* Reset summary indicators. */
    cdbdisp_resetResult(dispatchResult);

    /* Update slice map entry. */
    if (sliceIndex >= 0 &&
        sliceIndex < meleeResults->sliceCapacity)
    {
        CdbDispatchResults_SliceInfo *si = &meleeResults->sliceMap[sliceIndex];
        if (si->resultBegin == si->resultEnd)
        {
            si->resultBegin = meleeIndex;
            si->resultEnd = meleeIndex + 1;
        }
        else
        {
            Assert(si->resultBegin <= meleeIndex);
            Assert(si->resultEnd <= meleeIndex);
            si->resultEnd = meleeIndex + 1;
        }
    }

    /* Insert the slice index in error messages related to this QE. */
    if (segdbDesc)
        cdbconn_setSliceIndex(segdbDesc, sliceIndex);

    return dispatchResult;
}                               /* cdbdisp_initResult */


/* Destroy a CdbDispatchResult object. */
void
cdbdisp_termResult(CdbDispatchResult  *dispatchResult)
{
	PQExpBuffer trash;

    dispatchResult->segdbDesc = NULL;

    /* Free the PGresult objects. */
    cdbdisp_resetResult(dispatchResult);

    /* Free the error message buffer and result buffer. */
	trash = dispatchResult->resultbuf;
	dispatchResult->resultbuf = NULL;
    destroyPQExpBuffer(trash);

	trash = dispatchResult->error_message;
	dispatchResult->error_message = NULL;
    destroyPQExpBuffer(trash);
}                               /* cdbdisp_termResult */


/* Reset a CdbDispatchResult object for possible reuse. */
void
cdbdisp_resetResult(CdbDispatchResult  *dispatchResult)
{
    PQExpBuffer buf = dispatchResult->resultbuf;
    if (buf != NULL)
    {
        PGresult  **begp = (PGresult **)buf->data;
        PGresult  **endp = (PGresult **)(buf->data + buf->len);
        PGresult  **p;

        /* Free the PGresult objects. */
        for (p = begp; p < endp; ++p)
        {
            Assert(*p != NULL);
            PQclear(*p);
        }
    }

    /* Reset summary indicators. */
    dispatchResult->errcode = 0;
    dispatchResult->errindex = -1;
    dispatchResult->okindex = -1;

    /* Reset progress indicators. */
    dispatchResult->hasDispatched = false;
    dispatchResult->stillRunning = false;
    dispatchResult->requestedCancel = false;
    dispatchResult->wasCanceled = false;

    /* Empty (but don't free) the error message buffer and result buffer. */
    resetPQExpBuffer(dispatchResult->resultbuf);
    resetPQExpBuffer(dispatchResult->error_message);
}                               /* cdbdisp_resetResult */


/* Take note of an error.
 * 'errcode' is the ERRCODE_xxx value for setting the client's SQLSTATE.
 * NB: This can be called from a dispatcher thread, so it must not use
 * palloc/pfree or elog/ereport because they are not thread safe.
 */
void
cdbdisp_seterrcode(int                  errcode,        /* ERRCODE_xxx or 0 */
                   int                  resultIndex,    /* -1 if no PGresult */
                   CdbDispatchResult   *dispatchResult)
{
    CdbDispatchResults *meleeResults = dispatchResult->meleeResults;

    /* Was the command canceled? */
    if (errcode == ERRCODE_GP_OPERATION_CANCELED ||
        errcode == ERRCODE_QUERY_CANCELED)
        dispatchResult->wasCanceled = true;

    /* If this is the first error from this QE, save the error code
     * and the index of the PGresult buffer entry.  We assume the
     * caller has not yet added the item to the PGresult buffer.
     */
    if (!dispatchResult->errcode)
    {
        dispatchResult->errcode =
            (errcode == 0) ? ERRCODE_INTERNAL_ERROR : errcode;
        if (resultIndex >= 0)
            dispatchResult->errindex = resultIndex;
    }

    if (!meleeResults)
        return;

    /*
     * Remember which QE reported an error first among the gangs,
     * but keep quiet about cancellation done at our request.
     *
     * Interconnection errors are given lower precedence because often
     * they are secondary to an earlier and more interesting error.
     */
    if (errcode == ERRCODE_GP_OPERATION_CANCELED &&
        dispatchResult->requestedCancel)
    {}                          /* nop */

    else if (meleeResults->errcode == 0 || 
			 (meleeResults->errcode == ERRCODE_GP_INTERCONNECTION_ERROR &&
			  errcode != ERRCODE_GP_INTERCONNECTION_ERROR))
    {
        pthread_mutex_lock(&setErrcodeMutex);
        if ((errcode != 0) && (meleeResults->errcode == 0 ||
			(meleeResults->errcode == ERRCODE_GP_INTERCONNECTION_ERROR &&
			 errcode != ERRCODE_GP_INTERCONNECTION_ERROR)))
        {
            meleeResults->errcode = errcode;
            meleeResults->iFirstError = dispatchResult->meleeIndex;
        }
        pthread_mutex_unlock(&setErrcodeMutex);
    }
}                               /* cdbdisp_seterrcode */


/* Transfer connection error messages to dispatchResult from segdbDesc. */
bool                            /* returns true if segdbDesc had err info */
cdbdisp_mergeConnectionErrors(CdbDispatchResult                *dispatchResult,
                              struct SegmentDatabaseDescriptor *segdbDesc)
{
    if (!segdbDesc)
        return false;
    if (segdbDesc->errcode == 0 &&
        segdbDesc->error_message.len == 0)
        return false;

    /* Error code should always be accompanied by text and vice-versa. */
    Assert(segdbDesc->errcode != 0 && segdbDesc->error_message.len > 0);

    /* Append error message text and save error code. */
    cdbdisp_appendMessage(dispatchResult, 0, segdbDesc->errcode, "%s",
                          segdbDesc->error_message.data);

    /* Reset connection object's error info. */
    segdbDesc->errcode = 0;
    segdbDesc->error_message.len = 0;
    segdbDesc->error_message.data[0] = '\0';

    return true;
}                               /* cdbdisp_mergeConnectionErrors */


/* Format a message, printf-style, and append to the error_message buffer.
 * Also write it to stderr if logging is enabled for messages of the
 * given severity level 'elevel' (for example, DEBUG1; or 0 to suppress).
 * 'errcode' is the ERRCODE_xxx value for setting the client's SQLSTATE.
 * NB: This can be called from a dispatcher thread, so it must not use
 * palloc/pfree or elog/ereport because they are not thread safe.
 */
void
cdbdisp_appendMessage(CdbDispatchResult    *dispatchResult,
                      int                   elevel,
                      int                   errcode,
                      const char           *fmt,
                      ...)
{
    va_list args;
    int     msgoff;

    /* Remember first error. */
    cdbdisp_seterrcode(errcode, -1, dispatchResult);

    /* Allocate buffer if first message.
     * Insert newline between previous message and new one.
     */
    if (!dispatchResult->error_message)
	{
        dispatchResult->error_message = createPQExpBuffer();

		if (PQExpBufferBroken(dispatchResult->error_message))
		{
			destroyPQExpBuffer(dispatchResult->error_message);
			dispatchResult->error_message = NULL;

			write_log("cdbdisp_appendMessage: allocation failed, can't save error-message.");
			return;
		}
	}
    else
        oneTrailingNewlinePQ(dispatchResult->error_message);

    msgoff = dispatchResult->error_message->len;

    /* Format the message and append it to the buffer. */
    va_start(args, fmt);
    appendPQExpBufferVA(dispatchResult->error_message, fmt, args);
    va_end(args);

    /* Display the message on stderr for debugging, if requested.
     * This helps to clarify the actual timing of threaded events.
     */
    if (elevel >= log_min_messages)
    {
        oneTrailingNewlinePQ(dispatchResult->error_message);
        write_log("%s", dispatchResult->error_message->data + msgoff);
    }

    /* In case the caller wants to hand the buffer to ereport(),
     * follow the ereport() convention of not ending with a newline.
     */
    noTrailingNewlinePQ(dispatchResult->error_message);

}                               /* cdbdisp_appendMessage */


/* Store a PGresult object ptr in the result buffer.
 * NB: Caller must not PQclear() the PGresult object.
 */
void
cdbdisp_appendResult(CdbDispatchResult *dispatchResult,
                     struct pg_result  *res)
{
    Assert(dispatchResult && res);

    /* Attach the QE identification string to the PGresult */
    if (dispatchResult->segdbDesc &&
        dispatchResult->segdbDesc->whoami)
        pqSaveMessageField(res, PG_DIAG_GP_PROCESS_TAG,
                           dispatchResult->segdbDesc->whoami);

    appendBinaryPQExpBuffer(dispatchResult->resultbuf, (char *)&res, sizeof(res));
}                               /* cdbdisp_appendResult */


/* Return the i'th PGresult object ptr (if i >= 0), or
 * the n+i'th one (if i < 0), or NULL (if i out of bounds).
 * NB: Caller must not PQclear() the PGresult object.
 */
struct pg_result *
cdbdisp_getPGresult(CdbDispatchResult *dispatchResult, int i)
{
    if (dispatchResult)
    {
        PQExpBuffer buf = dispatchResult->resultbuf;
        PGresult  **begp = (PGresult **)buf->data;
        PGresult  **endp = (PGresult **)(buf->data + buf->len);
        PGresult  **p = (i >= 0) ? &begp[i] : &endp[i];

        if (p >= begp && p < endp)
        {
            Assert(*p != NULL);
            return *p;
        }
    }
    return NULL;
}                               /* cdbdisp_getPGresult */


/* Return the number of PGresult objects in the result buffer. */
int
cdbdisp_numPGresult(CdbDispatchResult  *dispatchResult)
{
    return dispatchResult ? dispatchResult->resultbuf->len / sizeof(PGresult *)
                          : 0;
}                               /* cdbdisp_numPGresult */


/* Remove all of the PGresult ptrs from a CdbDispatchResult object
 * and place them into an array provided by the caller.  The caller
 * becomes responsible for PQclear()ing them.  Returns the number of
 * PGresult ptrs placed in the array.
 */
int
cdbdisp_snatchPGresults(CdbDispatchResult  *dispatchResult,
                        struct pg_result  **pgresultptrs,
                        int                 maxresults)
{
    PQExpBuffer     buf = dispatchResult->resultbuf;
    PGresult      **begp = (PGresult **)buf->data;
    PGresult      **endp = (PGresult **)(buf->data + buf->len);
    PGresult      **p;
    int             nresults = 0;

    /* Snatch the PGresult objects. */
    for (p = begp; p < endp; ++p)
    {
        Assert(*p != NULL);
        Assert(nresults < maxresults);
        pgresultptrs[nresults++] = *p;
        *p = NULL;
    }

    /* Empty our PGresult array. */
    resetPQExpBuffer(buf);
    dispatchResult->errindex = -1;
    dispatchResult->okindex = -1;

    return nresults;
}                               /* cdbdisp_snatchPGresults */


/* Display a CdbDispatchResult in the log for debugging.
 * Call only from main thread, during or after cdbdisp_checkDispatchResults.
 */
void
cdbdisp_debugDispatchResult(CdbDispatchResult  *dispatchResult,
                            int                 elevel_error,
                            int                 elevel_success)
{
    char        esqlstate[8];
    int         ires;
    int         nres;

    /* Skip if user has messages turned off. */
    if (elevel_error < log_min_messages &&
        elevel_success < log_min_messages)
        return;
    
    if (dispatchResult == NULL)
    	return;

    /* PGresult messages */
    nres = cdbdisp_numPGresult(dispatchResult);
    for (ires = 0; ires < nres; ++ires)
    {                           /* for each PGresult */
        PGresult       *pgresult = cdbdisp_getPGresult(dispatchResult, ires);
        ExecStatusType  resultStatus = PQresultStatus(pgresult);
        char   *whoami = PQresultErrorField(pgresult, PG_DIAG_GP_PROCESS_TAG);

        if (!whoami)
            whoami = "no process id";

        /* QE success */
        if (resultStatus == PGRES_COMMAND_OK ||
            resultStatus == PGRES_TUPLES_OK ||
            resultStatus == PGRES_COPY_IN ||
            resultStatus == PGRES_COPY_OUT ||
            resultStatus == PGRES_EMPTY_QUERY)
        {
            char *cmdStatus = PQcmdStatus(pgresult);

            elog(elevel_success, "DispatchResult: ok %s (%s)",
                 cmdStatus ? cmdStatus : "(no cmdStatus)",
                 whoami);
        }

        /* QE error or libpq error */
        else
        {
            char   *sqlstate = PQresultErrorField(pgresult, PG_DIAG_SQLSTATE);
            char   *pri = PQresultErrorField(pgresult, PG_DIAG_MESSAGE_PRIMARY);
            char   *dtl = PQresultErrorField(pgresult, PG_DIAG_MESSAGE_DETAIL);
            int     lenpri = (pri == NULL)? 0 :strlen(pri);

            if (!sqlstate)
                sqlstate = "no SQLSTATE";

            while (lenpri > 0 &&
                   pri[lenpri-1] <= ' ' &&
                   pri[lenpri-1] > '\0')
                lenpri--;

            ereport(elevel_error, (
                 errmsg("DispatchResult: (%s) %s %.*s (%s)",
                        sqlstate,
                        PQresStatus(PQresultStatus(pgresult)),
                        lenpri,
                        pri ? pri : "",
                        whoami),
                 errdetail("(%s:%s) %s",
                           PQresultErrorField(pgresult, PG_DIAG_SOURCE_FILE),
                           PQresultErrorField(pgresult, PG_DIAG_SOURCE_LINE),
                           dtl ? dtl : "") ));
        }
    }                           /* for each PGresult */

    /* Error found on our side of the libpq interface? */
    if (dispatchResult->error_message &&
        dispatchResult->error_message->len > 0)
    {
        cdbdisp_errcode_to_sqlstate(dispatchResult->errcode, esqlstate);
        elog(elevel_error, "DispatchResult: (%s) %s",
             esqlstate,
             dispatchResult->error_message->data);
    }

    /* Connection error? */
    if (dispatchResult->segdbDesc &&
        dispatchResult->segdbDesc->error_message.len > 0)
	{
        cdbdisp_errcode_to_sqlstate(dispatchResult->segdbDesc->errcode, esqlstate);
        elog(elevel_error, "DispatchResult: (%s) %s",
             esqlstate,
             dispatchResult->segdbDesc->error_message.data);
	}

    /* Should have either an error code or an ok result. */
    if (!dispatchResult->errcode &&
        dispatchResult->okindex < 0)
    {
        elog(elevel_error, "DispatchResult: No ending status from %s",
             dispatchResult->segdbDesc ? dispatchResult->segdbDesc->whoami
                                       : "?");
    }
}                               /* cdbdisp_debugDispatchResult */


/*
 * Format a CdbDispatchResult into a StringInfo buffer provided by caller.
 * If verbose = true, reports all info; else reports at most one error.
 */
void
cdbdisp_dumpDispatchResult(CdbDispatchResult       *dispatchResult,
                           bool                     verbose,
                           struct StringInfoData   *buf)
{
    SegmentDatabaseDescriptor  *segdbDesc = dispatchResult->segdbDesc;
    int     ires;
    int     nres;

    if (!dispatchResult ||
        !buf)
        return;

    /* Format PGresult messages */
    nres = cdbdisp_numPGresult(dispatchResult);
    for (ires = 0; ires < nres; ++ires)
    {                           /* for each PGresult */
        PGresult       *pgresult = cdbdisp_getPGresult(dispatchResult, ires);
        ExecStatusType  resultStatus = PQresultStatus(pgresult);
        char   *whoami = PQresultErrorField(pgresult, PG_DIAG_GP_PROCESS_TAG);

        /* QE success */
        if (resultStatus == PGRES_COMMAND_OK ||
            resultStatus == PGRES_TUPLES_OK ||
            resultStatus == PGRES_COPY_IN ||
            resultStatus == PGRES_COPY_OUT ||
            resultStatus == PGRES_EMPTY_QUERY)
        {
            if (verbose)
            {
                char *cmdStatus = PQcmdStatus(pgresult);

                oneTrailingNewline(buf);
                appendStringInfo(buf, "ok: %s", cmdStatus ? cmdStatus : "");
                if (whoami)
                    appendStringInfo(buf," (%s)", whoami);
            }
        }

        /* QE error or libpq error */
        else
        {
            char   *pri = PQresultErrorField(pgresult, PG_DIAG_MESSAGE_PRIMARY);
            char   *dtl = PQresultErrorField(pgresult, PG_DIAG_MESSAGE_DETAIL);
            char   *ctx = PQresultErrorField(pgresult, PG_DIAG_CONTEXT);

            oneTrailingNewline(buf);
            if (pri)
            {
            	appendStringInfoString(buf, pri);
            }
            else
            {
            	elog(LOG,"No primary message?");
            	appendStringInfoString(buf,PQresultErrorMessage(pgresult));
            }

            if (whoami)
            {
                noTrailingNewline(buf);
                appendStringInfo(buf,"  (%s)", whoami);
            }

            if (dtl)
            {
                oneTrailingNewline(buf);
                appendStringInfo(buf, "%s", dtl);
            }
			
            if (ctx)
            {
                oneTrailingNewline(buf);
                appendStringInfo(buf, "%s", ctx);
            }

            if (!verbose)
                goto done;
        }
    }                           /* for each PGresult */

    /* segdbDesc error message shouldn't be possible here, but check anyway.
     *   Ordinarily dispatchResult->segdbDesc is NULL here because we are
     *   called after it has been cleared by CdbCheckDispatchResult().
     */
    if (dispatchResult->segdbDesc)
        cdbdisp_mergeConnectionErrors(dispatchResult, segdbDesc);

    /* Error found on our side of the libpq interface? */
    if (dispatchResult->error_message &&
        dispatchResult->error_message->len > 0)
    {
        oneTrailingNewline(buf);
        appendStringInfoString(buf, dispatchResult->error_message->data);
        if (!verbose)
            goto done;
    }

done:
    noTrailingNewline(buf);
}                               /* cdbdisp_dumpDispatchResult */


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


/*
 * cdbdisp_makeDispatchResults:
 * Allocates a CdbDispatchResults object in the current memory context.
 * The caller is responsible for calling DestroyCdbDispatchResults on the returned
 * pointer when done using it.
 */
CdbDispatchResults *
cdbdisp_makeDispatchResults(int     resultCapacity,
                            int     sliceCapacity,
                            bool    cancelOnError)
{
    CdbDispatchResults *results = (CdbDispatchResults*) palloc0(sizeof(CdbDispatchResults));
    int     nbytes = resultCapacity * sizeof(CdbDispatchResult);

    results->resultArray = (CdbDispatchResult*) palloc0(nbytes);

    results->resultCapacity = resultCapacity;
    results->resultCount = 0;
    results->iFirstError = -1;
    results->errcode = 0;
    results->cancelOnError = cancelOnError;

    results->sliceMap = NULL;
    results->sliceCapacity = sliceCapacity;
    if (sliceCapacity > 0)
    {
        nbytes = sliceCapacity * sizeof(CdbDispatchResults_SliceInfo);
        results->sliceMap = (CdbDispatchResults_SliceInfo*) palloc0(nbytes);
    }

    return results;
}                               /* cdbdisp_makeDispatchResults */

/*
 * cdbdisp_destroyDispatchResults:
 * Frees all memory allocated in CdbDispatchResults struct.
 */
void
cdbdisp_destroyDispatchResults(CdbDispatchResults * results)
{
    if (!results)
        return;

    if (results->resultArray)
    {
        int i;
        for (i = 0; i < results->resultCount; i++)
        {
            cdbdisp_termResult(&results->resultArray[i]);
        }
        pfree(results->resultArray);
		results->resultArray = NULL;
    }

    if (results->sliceMap)
    {
        pfree(results->sliceMap);
        results->sliceMap = NULL;
    }

    pfree(results);
}                               /* cdbdisp_destroyDispatchResults */


/* Format a CdbDispatchResults object.
 * Appends error messages to caller's StringInfo buffer.
 * Returns ERRCODE_xxx if some error was found, or 0 if no errors.
 * Before calling this function, you must call CdbCheckDispatchResult().
 */
int
cdbdisp_dumpDispatchResults(struct CdbDispatchResults  *meleeResults,
                            struct StringInfoData      *buffer,
                            bool                        verbose)
{
    CdbDispatchResult  *dispatchResult;

    /* Quick exit if no error (not counting ERRCODE_GP_OPERATION_CANCELED). */
    if (!meleeResults ||
        !meleeResults->errcode)
        return 0;

    /* Find the CdbDispatchResult of the first QE that got an error. */
    Assert(meleeResults->iFirstError >= 0 &&
           meleeResults->iFirstError < meleeResults->resultCount);

    dispatchResult = &meleeResults->resultArray[meleeResults->iFirstError];

    Assert(dispatchResult->meleeResults == meleeResults &&
           dispatchResult->errcode != 0);

    /* Format one QE's result. */
    cdbdisp_dumpDispatchResult(dispatchResult, verbose, buffer);

    /* Optionally, format results from the rest of the QEs that got errors. */
    if (verbose)
    {
        int i;
        for (i = 0; i < meleeResults->resultCount; ++i)
        {
            dispatchResult = &meleeResults->resultArray[i];
            if (i == meleeResults->iFirstError)
                continue;
            if (dispatchResult->errcode)
                cdbdisp_dumpDispatchResult(dispatchResult, verbose, buffer);
        }
    }

    return meleeResults->errcode;
}                               /* cdbdisp_dumpDispatchResults */


/* Return sum of the cmdTuples values from CdbDispatchResult
 * entries that have a successful PGresult.  If sliceIndex >= 0,
 * uses only the results belonging to the specified slice.
 */
int64
cdbdisp_sumCmdTuples(CdbDispatchResults *results, int sliceIndex)
{
    CdbDispatchResult  *dispatchResult;
    CdbDispatchResult  *resultEnd = cdbdisp_resultEnd(results, sliceIndex);
    PGresult           *pgresult;
    int64              sum = 0;

    for (dispatchResult = cdbdisp_resultBegin(results, sliceIndex);
         dispatchResult < resultEnd;
         ++dispatchResult)
    {
        pgresult = cdbdisp_getPGresult(dispatchResult, dispatchResult->okindex);
        if (pgresult &&
            !dispatchResult->errcode)
        {
            char   *cmdTuples = PQcmdTuples(pgresult);
            if (cmdTuples)
                sum += atoll(cmdTuples);
        }
    }
    return sum;
}                               /* cdbdisp_sumCmdTuples */

/*
 * If several tuples were eliminated/rejected from the result because of
 * bad data formatting (this is currenly only possible in external tables
 * with single row error handling) - sum up the total rows rejected from
 * all QE's and notify the client.
 */
void
cdbdisp_sumRejectedRows(CdbDispatchResults *results)
{
    CdbDispatchResult  *dispatchResult;
    CdbDispatchResult  *resultEnd = cdbdisp_resultEnd(results, -1);
    PGresult           *pgresult;
    int					totalRejected = 0;
	
    for (dispatchResult = cdbdisp_resultBegin(results, -1);
         dispatchResult < resultEnd;
         ++dispatchResult)
    {
        pgresult = cdbdisp_getPGresult(dispatchResult, dispatchResult->okindex);
        if (pgresult &&
            !dispatchResult->errcode)
        {
			/* add num rows rejected from this QE to the total */
			totalRejected += dispatchResult->numrowsrejected;
        }
    }
    
	if(totalRejected > 0)
		ReportSrehResults(NULL, totalRejected);

}                               /* cdbdisp_sumRejectedRows */

HTAB *
process_aotupcounts(PartitionNode *parts, HTAB *ht,
					void *aotupcounts, int naotupcounts)
{
	PQaoRelTupCount *ao = (PQaoRelTupCount *)aotupcounts;
	
	if (Debug_appendonly_print_insert)
		ereport(LOG,(errmsg("found %d AO tuple counts to process",
							naotupcounts)));

	if (naotupcounts)
	{
		int j;

		for (j = 0; j < naotupcounts; j++)
		{
			if (OidIsValid(ao->aorelid))
			{
				bool found;
				PQaoRelTupCount *entry;

				if (!ht)
				{
					HASHCTL ctl;
					/* reasonable assumption? */
					long num_buckets = 
						list_length(all_partition_relids(parts));
					num_buckets /= num_partition_levels(parts);

					ctl.keysize = sizeof(Oid);
					ctl.entrysize = sizeof(*entry);
					ht = hash_create("AO hash map",
									 num_buckets,
									 &ctl,
									 HASH_ELEM);
				}

				entry = hash_search(ht,
									&(ao->aorelid),
									HASH_ENTER,
									&found);
				
				if (found)
					entry->tupcount += ao->tupcount;
				else
					entry->tupcount = ao->tupcount;
								
				if (Debug_appendonly_print_insert)
					ereport(LOG,(errmsg("processed AO tuple counts for partitioned "
										"relation %d. found total " INT64_FORMAT 
										"tuples", ao->aorelid, entry->tupcount)));
			}
			ao++;
		}
	}


	return ht;
}

void
cdbdisp_handleModifiedCatalogOnSegments(CdbDispatchResults *results,
		void (*handler)(QueryContextDispatchingSendBack sendback))
{
	int i;
	for (i = 0; i < results->resultCount; ++i)
	{
		CdbDispatchResult *dispatchResult = &results->resultArray[i];
		int nres = cdbdisp_numPGresult(dispatchResult);
		int ires;
		for (ires = 0; ires < nres; ++ires)
		{
			/* for each PGresult */
			PGresult *pgresult = cdbdisp_getPGresult(dispatchResult, ires);
			if (handler && pgresult && pgresult->sendback)
			{
				int j;
				for (j = 0 ; j < pgresult->numSendback ; ++j)
				{
					handler(&pgresult->sendback[j]);
				}
			}
		}
	}
}

void
cdbdisp_iterate_results_sendback(PGresult **results, int numresults,
			void (*handler)(QueryContextDispatchingSendBack sendback))
{
	int i, j;

	if (handler == NULL)
		return;

	for (i = 0; i < numresults; i++)
	{
		if (results[i] == NULL || results[i]->sendback == 0)
			continue;

		for (j = 0; j < results[i]->numSendback; j++)
			handler(&results[i]->sendback[j]);		
	}
}

/*
 * sum tuple counts that were added into a partitioned AO table
 */
HTAB *
cdbdisp_sumAoPartTupCount(PartitionNode *parts,
						  CdbDispatchResults *results)
{
	int i;
	HTAB *ht = NULL;

	if (!parts)
		return NULL;


	for (i = 0; i < results->resultCount; ++i)
	{
		CdbDispatchResult  *dispatchResult = &results->resultArray[i];
		int nres = cdbdisp_numPGresult(dispatchResult);
		int ires;
		for (ires = 0; ires < nres; ++ires)
		{						   /* for each PGresult */
			PGresult *pgresult = cdbdisp_getPGresult(dispatchResult, ires);

			ht = process_aotupcounts(parts, ht, (void *)pgresult->aotupcounts,
									 pgresult->naotupcounts);
		}
	}

	return ht;
}
/*
 * Find the max of the lastOid values returned from the QEs
 */
Oid
cdbdisp_maxLastOid(CdbDispatchResults *results, int sliceIndex)
{
    CdbDispatchResult  *dispatchResult;
    CdbDispatchResult  *resultEnd = cdbdisp_resultEnd(results, sliceIndex);
    PGresult           *pgresult;
    Oid                 oid = InvalidOid;

    for (dispatchResult = cdbdisp_resultBegin(results, sliceIndex);
         dispatchResult < resultEnd;
         ++dispatchResult)
    {
        pgresult = cdbdisp_getPGresult(dispatchResult, dispatchResult->okindex);
        if (pgresult &&
            !dispatchResult->errcode)
        {
            Oid tmpoid = PQoidValue(pgresult);
            if (tmpoid >oid)
                oid = tmpoid;
        }
    }

    return oid;
}                               /* cdbdisp_maxLastOid */

/* Return ptr to first resultArray entry for a given sliceIndex. */
CdbDispatchResult *
cdbdisp_resultBegin(CdbDispatchResults *results, int sliceIndex)
{
    CdbDispatchResults_SliceInfo   *si;

    if (!results)
        return NULL;

    if (sliceIndex < 0)
        return &results->resultArray[0];
    
    si = &results->sliceMap[sliceIndex];

    Assert(sliceIndex < results->sliceCapacity &&
           si->resultBegin >= 0 &&
           si->resultBegin <= si->resultEnd &&
           si->resultEnd <= results->resultCount);

    return &results->resultArray[si->resultBegin];
}                               /* cdbdisp_resultBegin */


/* Return ptr to last+1 resultArray entry for a given sliceIndex. */
CdbDispatchResult *
cdbdisp_resultEnd(CdbDispatchResults *results, int sliceIndex)
{
    CdbDispatchResults_SliceInfo   *si;

    if (!results)
        return NULL;

    if (sliceIndex < 0)
        return &results->resultArray[results->resultCount];
    
    si = &results->sliceMap[sliceIndex];

    return &results->resultArray[si->resultEnd];
}                               /* cdbdisp_resultEnd */


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


/* Convert compact error code (ERRCODE_xxx) to 5-char SQLSTATE string,
 * and put it into a 6-char buffer provided by caller.
 */
char *                          /* returns outbuf+5 */
cdbdisp_errcode_to_sqlstate(int errcode, char outbuf[6])
{
    int i;
    for (i = 0; i < 5; ++i)
    {
        outbuf[i] = PGUNSIXBIT(errcode);
        errcode >>= 6;
    }
    outbuf[5] = '\0';
    return &outbuf[5];
}                               /* cdbdisp_errcode_to_sqlstate */


/* Convert SQLSTATE string to compact error code (ERRCODE_xxx). */
int
cdbdisp_sqlstate_to_errcode(const char *sqlstate)
{
    return MAKE_SQLSTATE(sqlstate[0], sqlstate[1], sqlstate[2],
        sqlstate[3], sqlstate[4]);
}                               /* cdbdisp_sqlstate_to_errcode */