/*
** 2018 May 08
**
** The author disclaims copyright to this source code.  In place of
** a legal notice, here is a blessing:
**
**    May you do good and not evil.
**    May you find forgiveness for yourself and forgive others.
**    May you share freely, never taking more than you give.
**
*************************************************************************
*/
#include "sqliteInt.h"

#ifndef SQLITE_OMIT_WINDOWFUNC

/*
** SELECT REWRITING
**
**   Any SELECT statement that contains one or more window functions in
**   either the select list or ORDER BY clause (the only two places window
**   functions may be used) is transformed by function sqlite3WindowRewrite()
**   in order to support window function processing. For example, with the
**   schema:
**
**     CREATE TABLE t1(a, b, c, d, e, f, g);
**
**   the statement:
**
**     SELECT a+1, max(b) OVER (PARTITION BY c ORDER BY d) FROM t1 ORDER BY e;
**
**   is transformed to:
**
**     SELECT a+1, max(b) OVER (PARTITION BY c ORDER BY d) FROM (
**         SELECT a, e, c, d, b FROM t1 ORDER BY c, d
**     ) ORDER BY e;
**
**   The flattening optimization is disabled when processing this transformed
**   SELECT statement. This allows the implementation of the window function
**   (in this case max()) to process rows sorted in order of (c, d), which
**   makes things easier for obvious reasons. More generally:
**
**     * FROM, WHERE, GROUP BY and HAVING clauses are all moved to 
**       the sub-query.
**
**     * ORDER BY, LIMIT and OFFSET remain part of the parent query.
**
**     * Terminals from each of the expression trees that make up the 
**       select-list and ORDER BY expressions in the parent query are
**       selected by the sub-query. For the purposes of the transformation,
**       terminals are column references and aggregate functions.
**
**   If there is more than one window function in the SELECT that uses
**   the same window declaration (the OVER bit), then a single scan may
**   be used to process more than one window function. For example:
**
**     SELECT max(b) OVER (PARTITION BY c ORDER BY d), 
**            min(e) OVER (PARTITION BY c ORDER BY d) 
**     FROM t1;
**
**   is transformed in the same way as the example above. However:
**
**     SELECT max(b) OVER (PARTITION BY c ORDER BY d), 
**            min(e) OVER (PARTITION BY a ORDER BY b) 
**     FROM t1;
**
**   Must be transformed to:
**
**     SELECT max(b) OVER (PARTITION BY c ORDER BY d) FROM (
**         SELECT e, min(e) OVER (PARTITION BY a ORDER BY b), c, d, b FROM
**           SELECT a, e, c, d, b FROM t1 ORDER BY a, b
**         ) ORDER BY c, d
**     ) ORDER BY e;
**
**   so that both min() and max() may process rows in the order defined by
**   their respective window declarations.
**
** INTERFACE WITH SELECT.C
**
**   When processing the rewritten SELECT statement, code in select.c calls
**   sqlite3WhereBegin() to begin iterating through the results of the
**   sub-query, which is always implemented as a co-routine. It then calls
**   sqlite3WindowCodeStep() to process rows and finish the scan by calling
**   sqlite3WhereEnd().
**
**   sqlite3WindowCodeStep() generates VM code so that, for each row returned
**   by the sub-query a sub-routine (OP_Gosub) coded by select.c is invoked.
**   When the sub-routine is invoked:
**
**     * The results of all window-functions for the row are stored
**       in the associated Window.regResult registers.
**
**     * The required terminal values are stored in the current row of
**       temp table Window.iEphCsr.
**
**   In some cases, depending on the window frame and the specific window
**   functions invoked, sqlite3WindowCodeStep() caches each entire partition
**   in a temp table before returning any rows. In other cases it does not.
**   This detail is encapsulated within this file, the code generated by
**   select.c is the same in either case.
**
** BUILT-IN WINDOW FUNCTIONS
**
**   This implementation features the following built-in window functions:
**
**     row_number()
**     rank()
**     dense_rank()
**     percent_rank()
**     cume_dist()
**     ntile(N)
**     lead(expr [, offset [, default]])
**     lag(expr [, offset [, default]])
**     first_value(expr)
**     last_value(expr)
**     nth_value(expr, N)
**   
**   These are the same built-in window functions supported by Postgres. 
**   Although the behaviour of aggregate window functions (functions that
**   can be used as either aggregates or window funtions) allows them to
**   be implemented using an API, built-in window functions are much more
**   esoteric. Additionally, some window functions (e.g. nth_value()) 
**   may only be implemented by caching the entire partition in memory.
**   As such, some built-in window functions use the same API as aggregate
**   window functions and some are implemented directly using VDBE 
**   instructions. Additionally, for those functions that use the API, the
**   window frame is sometimes modified before the SELECT statement is
**   rewritten. For example, regardless of the specified window frame, the
**   row_number() function always uses:
**
**     ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
**
**   See sqlite3WindowUpdate() for details.
**
**   As well as some of the built-in window functions, aggregate window
**   functions min() and max() are implemented using VDBE instructions if
**   the start of the window frame is declared as anything other than 
**   UNBOUNDED PRECEDING.
*/

/*
** Implementation of built-in window function row_number(). Assumes that the
** window frame has been coerced to:
**
**   ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
*/
static void row_numberStepFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  i64 *p = (i64*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ) (*p)++;
  UNUSED_PARAMETER(nArg);
  UNUSED_PARAMETER(apArg);
}
static void row_numberValueFunc(sqlite3_context *pCtx){
  i64 *p = (i64*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  sqlite3_result_int64(pCtx, (p ? *p : 0));
}

/*
** Context object type used by rank(), dense_rank(), percent_rank() and
** cume_dist().
*/
struct CallCount {
  i64 nValue;
  i64 nStep;
  i64 nTotal;
};

/*
** Implementation of built-in window function dense_rank(). Assumes that
** the window frame has been set to:
**
**   RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW 
*/
static void dense_rankStepFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  struct CallCount *p;
  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ) p->nStep = 1;
  UNUSED_PARAMETER(nArg);
  UNUSED_PARAMETER(apArg);
}
static void dense_rankValueFunc(sqlite3_context *pCtx){
  struct CallCount *p;
  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    if( p->nStep ){
      p->nValue++;
      p->nStep = 0;
    }
    sqlite3_result_int64(pCtx, p->nValue);
  }
}

/*
** Implementation of built-in window function rank(). Assumes that
** the window frame has been set to:
**
**   RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW 
*/
static void rankStepFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  struct CallCount *p;
  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    p->nStep++;
    if( p->nValue==0 ){
      p->nValue = p->nStep;
    }
  }
  UNUSED_PARAMETER(nArg);
  UNUSED_PARAMETER(apArg);
}
static void rankValueFunc(sqlite3_context *pCtx){
  struct CallCount *p;
  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    sqlite3_result_int64(pCtx, p->nValue);
    p->nValue = 0;
  }
}

/*
** Implementation of built-in window function percent_rank(). Assumes that
** the window frame has been set to:
**
**   RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW 
*/
static void percent_rankStepFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  struct CallCount *p;
  UNUSED_PARAMETER(nArg); assert( nArg==1 );

  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    if( p->nTotal==0 ){
      p->nTotal = sqlite3_value_int64(apArg[0]);
    }
    p->nStep++;
    if( p->nValue==0 ){
      p->nValue = p->nStep;
    }
  }
}
static void percent_rankValueFunc(sqlite3_context *pCtx){
  struct CallCount *p;
  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    if( p->nTotal>1 ){
      double r = (double)(p->nValue-1) / (double)(p->nTotal-1);
      sqlite3_result_double(pCtx, r);
    }else{
      sqlite3_result_double(pCtx, 0.0);
    }
    p->nValue = 0;
  }
}

/*
** Implementation of built-in window function cume_dist(). Assumes that
** the window frame has been set to:
**
**   RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW 
*/
static void cume_distStepFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  struct CallCount *p;
  assert( nArg==1 ); UNUSED_PARAMETER(nArg);

  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    if( p->nTotal==0 ){
      p->nTotal = sqlite3_value_int64(apArg[0]);
    }
    p->nStep++;
  }
}
static void cume_distValueFunc(sqlite3_context *pCtx){
  struct CallCount *p;
  p = (struct CallCount*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p && p->nTotal ){
    double r = (double)(p->nStep) / (double)(p->nTotal);
    sqlite3_result_double(pCtx, r);
  }
}

/*
** Context object for ntile() window function.
*/
struct NtileCtx {
  i64 nTotal;                     /* Total rows in partition */
  i64 nParam;                     /* Parameter passed to ntile(N) */
  i64 iRow;                       /* Current row */
};

/*
** Implementation of ntile(). This assumes that the window frame has
** been coerced to:
**
**   ROWS UNBOUNDED PRECEDING AND CURRENT ROW
*/
static void ntileStepFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  struct NtileCtx *p;
  assert( nArg==2 ); UNUSED_PARAMETER(nArg);
  p = (struct NtileCtx*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    if( p->nTotal==0 ){
      p->nParam = sqlite3_value_int64(apArg[0]);
      p->nTotal = sqlite3_value_int64(apArg[1]);
      if( p->nParam<=0 ){
        sqlite3_result_error(
            pCtx, "argument of ntile must be a positive integer", -1
        );
      }
    }
    p->iRow++;
  }
}
static void ntileValueFunc(sqlite3_context *pCtx){
  struct NtileCtx *p;
  p = (struct NtileCtx*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p && p->nParam>0 ){
    int nSize = (p->nTotal / p->nParam);
    if( nSize==0 ){
      sqlite3_result_int64(pCtx, p->iRow);
    }else{
      i64 nLarge = p->nTotal - p->nParam*nSize;
      i64 iSmall = nLarge*(nSize+1);
      i64 iRow = p->iRow-1;

      assert( (nLarge*(nSize+1) + (p->nParam-nLarge)*nSize)==p->nTotal );

      if( iRow<iSmall ){
        sqlite3_result_int64(pCtx, 1 + iRow/(nSize+1));
      }else{
        sqlite3_result_int64(pCtx, 1 + nLarge + (iRow-iSmall)/nSize);
      }
    }
  }
}

/*
** Context object for last_value() window function.
*/
struct LastValueCtx {
  sqlite3_value *pVal;
  int nVal;
};

/*
** Implementation of last_value().
*/
static void last_valueStepFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  struct LastValueCtx *p;
  UNUSED_PARAMETER(nArg);
  p = (struct LastValueCtx*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p ){
    sqlite3_value_free(p->pVal);
    p->pVal = sqlite3_value_dup(apArg[0]);
    if( p->pVal==0 ){
      sqlite3_result_error_nomem(pCtx);
    }else{
      p->nVal++;
    }
  }
}
static void last_valueInvFunc(
  sqlite3_context *pCtx, 
  int nArg,
  sqlite3_value **apArg
){
  struct LastValueCtx *p;
  UNUSED_PARAMETER(nArg);
  UNUSED_PARAMETER(apArg);
  p = (struct LastValueCtx*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( ALWAYS(p) ){
    p->nVal--;
    if( p->nVal==0 ){
      sqlite3_value_free(p->pVal);
      p->pVal = 0;
    }
  }
}
static void last_valueValueFunc(sqlite3_context *pCtx){
  struct LastValueCtx *p;
  p = (struct LastValueCtx*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p && p->pVal ){
    sqlite3_result_value(pCtx, p->pVal);
  }
}
static void last_valueFinalizeFunc(sqlite3_context *pCtx){
  struct LastValueCtx *p;
  p = (struct LastValueCtx*)sqlite3_aggregate_context(pCtx, sizeof(*p));
  if( p && p->pVal ){
    sqlite3_result_value(pCtx, p->pVal);
    sqlite3_value_free(p->pVal);
    p->pVal = 0;
  }
}

/*
** Static names for the built-in window function names.  These static
** names are used, rather than string literals, so that FuncDef objects
** can be associated with a particular window function by direct
** comparison of the zName pointer.  Example:
**
**       if( pFuncDef->zName==row_valueName ){ ... }
*/
static const char row_numberName[] =   "row_number";
static const char dense_rankName[] =   "dense_rank";
static const char rankName[] =         "rank";
static const char percent_rankName[] = "percent_rank";
static const char cume_distName[] =    "cume_dist";
static const char ntileName[] =        "ntile";
static const char last_valueName[] =   "last_value";
static const char nth_valueName[] =    "nth_value";
static const char first_valueName[] =  "first_value";
static const char leadName[] =         "lead";
static const char lagName[] =          "lag";

/*
** No-op implementations of xStep() and xFinalize().  Used as place-holders
** for built-in window functions that never call those interfaces.
**
** The noopValueFunc() is called but is expected to do nothing.  The
** noopStepFunc() is never called, and so it is marked with NO_TEST to
** let the test coverage routine know not to expect this function to be
** invoked.
*/
static void noopStepFunc(    /*NO_TEST*/
  sqlite3_context *p,        /*NO_TEST*/
  int n,                     /*NO_TEST*/
  sqlite3_value **a          /*NO_TEST*/
){                           /*NO_TEST*/
  UNUSED_PARAMETER(p);       /*NO_TEST*/
  UNUSED_PARAMETER(n);       /*NO_TEST*/
  UNUSED_PARAMETER(a);       /*NO_TEST*/
  assert(0);                 /*NO_TEST*/
}                            /*NO_TEST*/
static void noopValueFunc(sqlite3_context *p){ UNUSED_PARAMETER(p); /*no-op*/ }

/* Window functions that use all window interfaces: xStep, xFinal,
** xValue, and xInverse */
#define WINDOWFUNCALL(name,nArg,extra) {                                   \
  nArg, (SQLITE_UTF8|SQLITE_FUNC_WINDOW|extra), 0, 0,                      \
  name ## StepFunc, name ## FinalizeFunc, name ## ValueFunc,               \
  name ## InvFunc, name ## Name, {0}                                       \
}

/* Window functions that are implemented using bytecode and thus have
** no-op routines for their methods */
#define WINDOWFUNCNOOP(name,nArg,extra) {                                  \
  nArg, (SQLITE_UTF8|SQLITE_FUNC_WINDOW|extra), 0, 0,                      \
  noopStepFunc, noopValueFunc, noopValueFunc,                              \
  noopStepFunc, name ## Name, {0}                                          \
}

/* Window functions that use all window interfaces: xStep, the
** same routine for xFinalize and xValue and which never call
** xInverse. */
#define WINDOWFUNCX(name,nArg,extra) {                                     \
  nArg, (SQLITE_UTF8|SQLITE_FUNC_WINDOW|extra), 0, 0,                      \
  name ## StepFunc, name ## ValueFunc, name ## ValueFunc,                  \
  noopStepFunc, name ## Name, {0}                                          \
}


/*
** Register those built-in window functions that are not also aggregates.
*/
void sqlite3WindowFunctions(void){
  static FuncDef aWindowFuncs[] = {
    WINDOWFUNCX(row_number, 0, 0),
    WINDOWFUNCX(dense_rank, 0, 0),
    WINDOWFUNCX(rank, 0, 0),
    WINDOWFUNCX(percent_rank, 0, SQLITE_FUNC_WINDOW_SIZE),
    WINDOWFUNCX(cume_dist, 0, SQLITE_FUNC_WINDOW_SIZE),
    WINDOWFUNCX(ntile, 1, SQLITE_FUNC_WINDOW_SIZE),
    WINDOWFUNCALL(last_value, 1, 0),
    WINDOWFUNCNOOP(nth_value, 2, 0),
    WINDOWFUNCNOOP(first_value, 1, 0),
    WINDOWFUNCNOOP(lead, 1, 0),
    WINDOWFUNCNOOP(lead, 2, 0),
    WINDOWFUNCNOOP(lead, 3, 0),
    WINDOWFUNCNOOP(lag, 1, 0),
    WINDOWFUNCNOOP(lag, 2, 0),
    WINDOWFUNCNOOP(lag, 3, 0),
  };
  sqlite3InsertBuiltinFuncs(aWindowFuncs, ArraySize(aWindowFuncs));
}

/*
** This function is called immediately after resolving the function name
** for a window function within a SELECT statement. Argument pList is a
** linked list of WINDOW definitions for the current SELECT statement.
** Argument pFunc is the function definition just resolved and pWin
** is the Window object representing the associated OVER clause. This
** function updates the contents of pWin as follows:
**
**   * If the OVER clause refered to a named window (as in "max(x) OVER win"),
**     search list pList for a matching WINDOW definition, and update pWin
**     accordingly. If no such WINDOW clause can be found, leave an error
**     in pParse.
**
**   * If the function is a built-in window function that requires the
**     window to be coerced (see "BUILT-IN WINDOW FUNCTIONS" at the top
**     of this file), pWin is updated here.
*/
void sqlite3WindowUpdate(
  Parse *pParse, 
  Window *pList,                  /* List of named windows for this SELECT */
  Window *pWin,                   /* Window frame to update */
  FuncDef *pFunc                  /* Window function definition */
){
  if( pWin->zName && pWin->eType==0 ){
    Window *p;
    for(p=pList; p; p=p->pNextWin){
      if( sqlite3StrICmp(p->zName, pWin->zName)==0 ) break;
    }
    if( p==0 ){
      sqlite3ErrorMsg(pParse, "no such window: %s", pWin->zName);
      return;
    }
    pWin->pPartition = sqlite3ExprListDup(pParse->db, p->pPartition, 0);
    pWin->pOrderBy = sqlite3ExprListDup(pParse->db, p->pOrderBy, 0);
    pWin->pStart = sqlite3ExprDup(pParse->db, p->pStart, 0);
    pWin->pEnd = sqlite3ExprDup(pParse->db, p->pEnd, 0);
    pWin->eStart = p->eStart;
    pWin->eEnd = p->eEnd;
    pWin->eType = p->eType;
  }
  if( pFunc->funcFlags & SQLITE_FUNC_WINDOW ){
    sqlite3 *db = pParse->db;
    if( pWin->pFilter ){
      sqlite3ErrorMsg(pParse, 
          "FILTER clause may only be used with aggregate window functions"
      );
    }else
    if( pFunc->zName==row_numberName || pFunc->zName==ntileName ){
      sqlite3ExprDelete(db, pWin->pStart);
      sqlite3ExprDelete(db, pWin->pEnd);
      pWin->pStart = pWin->pEnd = 0;
      pWin->eType = TK_ROWS;
      pWin->eStart = TK_UNBOUNDED;
      pWin->eEnd = TK_CURRENT;
    }else

    if( pFunc->zName==dense_rankName || pFunc->zName==rankName
     || pFunc->zName==percent_rankName || pFunc->zName==cume_distName
    ){
      sqlite3ExprDelete(db, pWin->pStart);
      sqlite3ExprDelete(db, pWin->pEnd);
      pWin->pStart = pWin->pEnd = 0;
      pWin->eType = TK_RANGE;
      pWin->eStart = TK_UNBOUNDED;
      pWin->eEnd = TK_CURRENT;
    }
  }
  pWin->pFunc = pFunc;
}

/*
** Context object passed through sqlite3WalkExprList() to
** selectWindowRewriteExprCb() by selectWindowRewriteEList().
*/
typedef struct WindowRewrite WindowRewrite;
struct WindowRewrite {
  Window *pWin;
  SrcList *pSrc;
  ExprList *pSub;
  Select *pSubSelect;             /* Current sub-select, if any */
};

/*
** Callback function used by selectWindowRewriteEList(). If necessary,
** this function appends to the output expression-list and updates 
** expression (*ppExpr) in place.
*/
static int selectWindowRewriteExprCb(Walker *pWalker, Expr *pExpr){
  struct WindowRewrite *p = pWalker->u.pRewrite;
  Parse *pParse = pWalker->pParse;

  /* If this function is being called from within a scalar sub-select
  ** that used by the SELECT statement being processed, only process
  ** TK_COLUMN expressions that refer to it (the outer SELECT). Do
  ** not process aggregates or window functions at all, as they belong
  ** to the scalar sub-select.  */
  if( p->pSubSelect ){
    if( pExpr->op!=TK_COLUMN ){
      return WRC_Continue;
    }else{
      int nSrc = p->pSrc->nSrc;
      int i;
      for(i=0; i<nSrc; i++){
        if( pExpr->iTable==p->pSrc->a[i].iCursor ) break;
      }
      if( i==nSrc ) return WRC_Continue;
    }
  }

  switch( pExpr->op ){

    case TK_FUNCTION:
      if( !ExprHasProperty(pExpr, EP_WinFunc) ){
        break;
      }else{
        Window *pWin;
        for(pWin=p->pWin; pWin; pWin=pWin->pNextWin){
          if( pExpr->y.pWin==pWin ){
            assert( pWin->pOwner==pExpr );
            return WRC_Prune;
          }
        }
      }
      /* Fall through.  */

    case TK_AGG_FUNCTION:
    case TK_COLUMN: {
      Expr *pDup = sqlite3ExprDup(pParse->db, pExpr, 0);
      p->pSub = sqlite3ExprListAppend(pParse, p->pSub, pDup);
      if( p->pSub ){
        assert( ExprHasProperty(pExpr, EP_Static)==0 );
        ExprSetProperty(pExpr, EP_Static);
        sqlite3ExprDelete(pParse->db, pExpr);
        ExprClearProperty(pExpr, EP_Static);
        memset(pExpr, 0, sizeof(Expr));

        pExpr->op = TK_COLUMN;
        pExpr->iColumn = p->pSub->nExpr-1;
        pExpr->iTable = p->pWin->iEphCsr;
      }

      break;
    }

    default: /* no-op */
      break;
  }

  return WRC_Continue;
}
static int selectWindowRewriteSelectCb(Walker *pWalker, Select *pSelect){
  struct WindowRewrite *p = pWalker->u.pRewrite;
  Select *pSave = p->pSubSelect;
  if( pSave==pSelect ){
    return WRC_Continue;
  }else{
    p->pSubSelect = pSelect;
    sqlite3WalkSelect(pWalker, pSelect);
    p->pSubSelect = pSave;
  }
  return WRC_Prune;
}


/*
** Iterate through each expression in expression-list pEList. For each:
**
**   * TK_COLUMN,
**   * aggregate function, or
**   * window function with a Window object that is not a member of the 
**     Window list passed as the second argument (pWin).
**
** Append the node to output expression-list (*ppSub). And replace it
** with a TK_COLUMN that reads the (N-1)th element of table 
** pWin->iEphCsr, where N is the number of elements in (*ppSub) after
** appending the new one.
*/
static void selectWindowRewriteEList(
  Parse *pParse, 
  Window *pWin,
  SrcList *pSrc,
  ExprList *pEList,               /* Rewrite expressions in this list */
  ExprList **ppSub                /* IN/OUT: Sub-select expression-list */
){
  Walker sWalker;
  WindowRewrite sRewrite;

  memset(&sWalker, 0, sizeof(Walker));
  memset(&sRewrite, 0, sizeof(WindowRewrite));

  sRewrite.pSub = *ppSub;
  sRewrite.pWin = pWin;
  sRewrite.pSrc = pSrc;

  sWalker.pParse = pParse;
  sWalker.xExprCallback = selectWindowRewriteExprCb;
  sWalker.xSelectCallback = selectWindowRewriteSelectCb;
  sWalker.u.pRewrite = &sRewrite;

  (void)sqlite3WalkExprList(&sWalker, pEList);

  *ppSub = sRewrite.pSub;
}

/*
** Append a copy of each expression in expression-list pAppend to
** expression list pList. Return a pointer to the result list.
*/
static ExprList *exprListAppendList(
  Parse *pParse,          /* Parsing context */
  ExprList *pList,        /* List to which to append. Might be NULL */
  ExprList *pAppend       /* List of values to append. Might be NULL */
){
  if( pAppend ){
    int i;
    int nInit = pList ? pList->nExpr : 0;
    for(i=0; i<pAppend->nExpr; i++){
      Expr *pDup = sqlite3ExprDup(pParse->db, pAppend->a[i].pExpr, 0);
      pList = sqlite3ExprListAppend(pParse, pList, pDup);
      if( pList ) pList->a[nInit+i].sortOrder = pAppend->a[i].sortOrder;
    }
  }
  return pList;
}

/*
** If the SELECT statement passed as the second argument does not invoke
** any SQL window functions, this function is a no-op. Otherwise, it 
** rewrites the SELECT statement so that window function xStep functions
** are invoked in the correct order as described under "SELECT REWRITING"
** at the top of this file.
*/
int sqlite3WindowRewrite(Parse *pParse, Select *p){
  int rc = SQLITE_OK;
  if( p->pWin ){
    Vdbe *v = sqlite3GetVdbe(pParse);
    sqlite3 *db = pParse->db;
    Select *pSub = 0;             /* The subquery */
    SrcList *pSrc = p->pSrc;
    Expr *pWhere = p->pWhere;
    ExprList *pGroupBy = p->pGroupBy;
    Expr *pHaving = p->pHaving;
    ExprList *pSort = 0;

    ExprList *pSublist = 0;       /* Expression list for sub-query */
    Window *pMWin = p->pWin;      /* Master window object */
    Window *pWin;                 /* Window object iterator */

    p->pSrc = 0;
    p->pWhere = 0;
    p->pGroupBy = 0;
    p->pHaving = 0;

    /* Create the ORDER BY clause for the sub-select. This is the concatenation
    ** of the window PARTITION and ORDER BY clauses. Then, if this makes it
    ** redundant, remove the ORDER BY from the parent SELECT.  */
    pSort = sqlite3ExprListDup(db, pMWin->pPartition, 0);
    pSort = exprListAppendList(pParse, pSort, pMWin->pOrderBy);
    if( pSort && p->pOrderBy ){
      if( sqlite3ExprListCompare(pSort, p->pOrderBy, -1)==0 ){
        sqlite3ExprListDelete(db, p->pOrderBy);
        p->pOrderBy = 0;
      }
    }

    /* Assign a cursor number for the ephemeral table used to buffer rows.
    ** The OpenEphemeral instruction is coded later, after it is known how
    ** many columns the table will have.  */
    pMWin->iEphCsr = pParse->nTab++;

    selectWindowRewriteEList(pParse, pMWin, pSrc, p->pEList, &pSublist);
    selectWindowRewriteEList(pParse, pMWin, pSrc, p->pOrderBy, &pSublist);
    pMWin->nBufferCol = (pSublist ? pSublist->nExpr : 0);

    /* Append the PARTITION BY and ORDER BY expressions to the to the 
    ** sub-select expression list. They are required to figure out where 
    ** boundaries for partitions and sets of peer rows lie.  */
    pSublist = exprListAppendList(pParse, pSublist, pMWin->pPartition);
    pSublist = exprListAppendList(pParse, pSublist, pMWin->pOrderBy);

    /* Append the arguments passed to each window function to the
    ** sub-select expression list. Also allocate two registers for each
    ** window function - one for the accumulator, another for interim
    ** results.  */
    for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
      pWin->iArgCol = (pSublist ? pSublist->nExpr : 0);
      pSublist = exprListAppendList(pParse, pSublist, pWin->pOwner->x.pList);
      if( pWin->pFilter ){
        Expr *pFilter = sqlite3ExprDup(db, pWin->pFilter, 0);
        pSublist = sqlite3ExprListAppend(pParse, pSublist, pFilter);
      }
      pWin->regAccum = ++pParse->nMem;
      pWin->regResult = ++pParse->nMem;
      sqlite3VdbeAddOp2(v, OP_Null, 0, pWin->regAccum);
    }

    /* If there is no ORDER BY or PARTITION BY clause, and the window
    ** function accepts zero arguments, and there are no other columns
    ** selected (e.g. "SELECT row_number() OVER () FROM t1"), it is possible
    ** that pSublist is still NULL here. Add a constant expression here to 
    ** keep everything legal in this case. 
    */
    if( pSublist==0 ){
      pSublist = sqlite3ExprListAppend(pParse, 0, 
          sqlite3ExprAlloc(db, TK_INTEGER, &sqlite3IntTokens[0], 0)
      );
    }

    pSub = sqlite3SelectNew(
        pParse, pSublist, pSrc, pWhere, pGroupBy, pHaving, pSort, 0, 0
    );
    p->pSrc = sqlite3SrcListAppend(db, 0, 0, 0);
    assert( p->pSrc || db->mallocFailed );
    if( p->pSrc ){
      p->pSrc->a[0].pSelect = pSub;
      sqlite3SrcListAssignCursors(pParse, p->pSrc);
      if( sqlite3ExpandSubquery(pParse, &p->pSrc->a[0]) ){
        rc = SQLITE_NOMEM;
      }else{
        pSub->selFlags |= SF_Expanded;
        p->selFlags &= ~SF_Aggregate;
        sqlite3SelectPrep(pParse, pSub, 0);
      }

      sqlite3VdbeAddOp2(v, OP_OpenEphemeral, pMWin->iEphCsr, pSublist->nExpr);
    }else{
      sqlite3SelectDelete(db, pSub);
    }
    if( db->mallocFailed ) rc = SQLITE_NOMEM;
  }

  return rc;
}

/*
** Free the Window object passed as the second argument.
*/
void sqlite3WindowDelete(sqlite3 *db, Window *p){
  if( p ){
    sqlite3ExprDelete(db, p->pFilter);
    sqlite3ExprListDelete(db, p->pPartition);
    sqlite3ExprListDelete(db, p->pOrderBy);
    sqlite3ExprDelete(db, p->pEnd);
    sqlite3ExprDelete(db, p->pStart);
    sqlite3DbFree(db, p->zName);
    sqlite3DbFree(db, p);
  }
}

/*
** Free the linked list of Window objects starting at the second argument.
*/
void sqlite3WindowListDelete(sqlite3 *db, Window *p){
  while( p ){
    Window *pNext = p->pNextWin;
    sqlite3WindowDelete(db, p);
    p = pNext;
  }
}

/*
** The argument expression is an PRECEDING or FOLLOWING offset.  The
** value should be a non-negative integer.  If the value is not a
** constant, change it to NULL.  The fact that it is then a non-negative
** integer will be caught later.  But it is important not to leave
** variable values in the expression tree.
*/
static Expr *sqlite3WindowOffsetExpr(Parse *pParse, Expr *pExpr){
  if( 0==sqlite3ExprIsConstant(pExpr) ){
    sqlite3ExprDelete(pParse->db, pExpr);
    pExpr = sqlite3ExprAlloc(pParse->db, TK_NULL, 0, 0);
  }
  return pExpr;
}

/*
** Allocate and return a new Window object describing a Window Definition.
*/
Window *sqlite3WindowAlloc(
  Parse *pParse,    /* Parsing context */
  int eType,        /* Frame type. TK_RANGE or TK_ROWS */
  int eStart,       /* Start type: CURRENT, PRECEDING, FOLLOWING, UNBOUNDED */
  Expr *pStart,     /* Start window size if TK_PRECEDING or FOLLOWING */
  int eEnd,         /* End type: CURRENT, FOLLOWING, TK_UNBOUNDED, PRECEDING */
  Expr *pEnd        /* End window size if TK_FOLLOWING or PRECEDING */
){
  Window *pWin = 0;

  /* Parser assures the following: */
  assert( eType==TK_RANGE || eType==TK_ROWS );
  assert( eStart==TK_CURRENT || eStart==TK_PRECEDING
           || eStart==TK_UNBOUNDED || eStart==TK_FOLLOWING );
  assert( eEnd==TK_CURRENT || eEnd==TK_FOLLOWING
           || eEnd==TK_UNBOUNDED || eEnd==TK_PRECEDING );
  assert( (eStart==TK_PRECEDING || eStart==TK_FOLLOWING)==(pStart!=0) );
  assert( (eEnd==TK_FOLLOWING || eEnd==TK_PRECEDING)==(pEnd!=0) );


  /* If a frame is declared "RANGE" (not "ROWS"), then it may not use
  ** either "<expr> PRECEDING" or "<expr> FOLLOWING".
  */
  if( eType==TK_RANGE && (pStart!=0 || pEnd!=0) ){
    sqlite3ErrorMsg(pParse, "RANGE must use only UNBOUNDED or CURRENT ROW");
    goto windowAllocErr;
  }

  /* Additionally, the
  ** starting boundary type may not occur earlier in the following list than
  ** the ending boundary type:
  **
  **   UNBOUNDED PRECEDING
  **   <expr> PRECEDING
  **   CURRENT ROW
  **   <expr> FOLLOWING
  **   UNBOUNDED FOLLOWING
  **
  ** The parser ensures that "UNBOUNDED PRECEDING" cannot be used as an ending
  ** boundary, and than "UNBOUNDED FOLLOWING" cannot be used as a starting
  ** frame boundary.
  */
  if( (eStart==TK_CURRENT && eEnd==TK_PRECEDING)
   || (eStart==TK_FOLLOWING && (eEnd==TK_PRECEDING || eEnd==TK_CURRENT))
  ){
    sqlite3ErrorMsg(pParse, "unsupported frame delimiter for ROWS");
    goto windowAllocErr;
  }

  pWin = (Window*)sqlite3DbMallocZero(pParse->db, sizeof(Window));
  if( pWin==0 ) goto windowAllocErr;
  pWin->eType = eType;
  pWin->eStart = eStart;
  pWin->eEnd = eEnd;
  pWin->pEnd = sqlite3WindowOffsetExpr(pParse, pEnd);
  pWin->pStart = sqlite3WindowOffsetExpr(pParse, pStart);
  return pWin;

windowAllocErr:
  sqlite3ExprDelete(pParse->db, pEnd);
  sqlite3ExprDelete(pParse->db, pStart);
  return 0;
}

/*
** Attach window object pWin to expression p.
*/
void sqlite3WindowAttach(Parse *pParse, Expr *p, Window *pWin){
  if( p ){
    assert( p->op==TK_FUNCTION );
    /* This routine is only called for the parser.  If pWin was not
    ** allocated due to an OOM, then the parser would fail before ever
    ** invoking this routine */
    if( ALWAYS(pWin) ){
      p->y.pWin = pWin;
      ExprSetProperty(p, EP_WinFunc);
      pWin->pOwner = p;
      if( p->flags & EP_Distinct ){
        sqlite3ErrorMsg(pParse,
           "DISTINCT is not supported for window functions");
      }
    }
  }else{
    sqlite3WindowDelete(pParse->db, pWin);
  }
}

/*
** Return 0 if the two window objects are identical, or non-zero otherwise.
** Identical window objects can be processed in a single scan.
*/
int sqlite3WindowCompare(Parse *pParse, Window *p1, Window *p2){
  if( p1->eType!=p2->eType ) return 1;
  if( p1->eStart!=p2->eStart ) return 1;
  if( p1->eEnd!=p2->eEnd ) return 1;
  if( sqlite3ExprCompare(pParse, p1->pStart, p2->pStart, -1) ) return 1;
  if( sqlite3ExprCompare(pParse, p1->pEnd, p2->pEnd, -1) ) return 1;
  if( sqlite3ExprListCompare(p1->pPartition, p2->pPartition, -1) ) return 1;
  if( sqlite3ExprListCompare(p1->pOrderBy, p2->pOrderBy, -1) ) return 1;
  return 0;
}


/*
** This is called by code in select.c before it calls sqlite3WhereBegin()
** to begin iterating through the sub-query results. It is used to allocate
** and initialize registers and cursors used by sqlite3WindowCodeStep().
*/
void sqlite3WindowCodeInit(Parse *pParse, Window *pMWin){
  Window *pWin;
  Vdbe *v = sqlite3GetVdbe(pParse);
  int nPart = (pMWin->pPartition ? pMWin->pPartition->nExpr : 0);
  nPart += (pMWin->pOrderBy ? pMWin->pOrderBy->nExpr : 0);
  if( nPart ){
    pMWin->regPart = pParse->nMem+1;
    pParse->nMem += nPart;
    sqlite3VdbeAddOp3(v, OP_Null, 0, pMWin->regPart, pMWin->regPart+nPart-1);
  }

  for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
    FuncDef *p = pWin->pFunc;
    if( (p->funcFlags & SQLITE_FUNC_MINMAX) && pWin->eStart!=TK_UNBOUNDED ){
      /* The inline versions of min() and max() require a single ephemeral
      ** table and 3 registers. The registers are used as follows:
      **
      **   regApp+0: slot to copy min()/max() argument to for MakeRecord
      **   regApp+1: integer value used to ensure keys are unique
      **   regApp+2: output of MakeRecord
      */
      ExprList *pList = pWin->pOwner->x.pList;
      KeyInfo *pKeyInfo = sqlite3KeyInfoFromExprList(pParse, pList, 0, 0);
      pWin->csrApp = pParse->nTab++;
      pWin->regApp = pParse->nMem+1;
      pParse->nMem += 3;
      if( pKeyInfo && pWin->pFunc->zName[1]=='i' ){
        assert( pKeyInfo->aSortOrder[0]==0 );
        pKeyInfo->aSortOrder[0] = 1;
      }
      sqlite3VdbeAddOp2(v, OP_OpenEphemeral, pWin->csrApp, 2);
      sqlite3VdbeAppendP4(v, pKeyInfo, P4_KEYINFO);
      sqlite3VdbeAddOp2(v, OP_Integer, 0, pWin->regApp+1);
    }
    else if( p->zName==nth_valueName || p->zName==first_valueName ){
      /* Allocate two registers at pWin->regApp. These will be used to
      ** store the start and end index of the current frame.  */
      assert( pMWin->iEphCsr );
      pWin->regApp = pParse->nMem+1;
      pWin->csrApp = pParse->nTab++;
      pParse->nMem += 2;
      sqlite3VdbeAddOp2(v, OP_OpenDup, pWin->csrApp, pMWin->iEphCsr);
    }
    else if( p->zName==leadName || p->zName==lagName ){
      assert( pMWin->iEphCsr );
      pWin->csrApp = pParse->nTab++;
      sqlite3VdbeAddOp2(v, OP_OpenDup, pWin->csrApp, pMWin->iEphCsr);
    }
  }
}

/*
** A "PRECEDING <expr>" (eCond==0) or "FOLLOWING <expr>" (eCond==1) or the
** value of the second argument to nth_value() (eCond==2) has just been
** evaluated and the result left in register reg. This function generates VM
** code to check that the value is a non-negative integer and throws an
** exception if it is not.
*/
static void windowCheckIntValue(Parse *pParse, int reg, int eCond){
  static const char *azErr[] = {
    "frame starting offset must be a non-negative integer",
    "frame ending offset must be a non-negative integer",
    "second argument to nth_value must be a positive integer"
  };
  static int aOp[] = { OP_Ge, OP_Ge, OP_Gt };
  Vdbe *v = sqlite3GetVdbe(pParse);
  int regZero = sqlite3GetTempReg(pParse);
  assert( eCond==0 || eCond==1 || eCond==2 );
  sqlite3VdbeAddOp2(v, OP_Integer, 0, regZero);
  sqlite3VdbeAddOp2(v, OP_MustBeInt, reg, sqlite3VdbeCurrentAddr(v)+2);
  VdbeCoverageIf(v, eCond==0);
  VdbeCoverageIf(v, eCond==1);
  VdbeCoverageIf(v, eCond==2);
  sqlite3VdbeAddOp3(v, aOp[eCond], regZero, sqlite3VdbeCurrentAddr(v)+2, reg);
  VdbeCoverageNeverNullIf(v, eCond==0);
  VdbeCoverageNeverNullIf(v, eCond==1);
  VdbeCoverageNeverNullIf(v, eCond==2);
  sqlite3VdbeAddOp2(v, OP_Halt, SQLITE_ERROR, OE_Abort);
  sqlite3VdbeAppendP4(v, (void*)azErr[eCond], P4_STATIC);
  sqlite3ReleaseTempReg(pParse, regZero);
}

/*
** Return the number of arguments passed to the window-function associated
** with the object passed as the only argument to this function.
*/
static int windowArgCount(Window *pWin){
  ExprList *pList = pWin->pOwner->x.pList;
  return (pList ? pList->nExpr : 0);
}

/*
** Generate VM code to invoke either xStep() (if bInverse is 0) or 
** xInverse (if bInverse is non-zero) for each window function in the 
** linked list starting at pMWin. Or, for built-in window functions
** that do not use the standard function API, generate the required
** inline VM code.
**
** If argument csr is greater than or equal to 0, then argument reg is
** the first register in an array of registers guaranteed to be large
** enough to hold the array of arguments for each function. In this case
** the arguments are extracted from the current row of csr into the
** array of registers before invoking OP_AggStep or OP_AggInverse
**
** Or, if csr is less than zero, then the array of registers at reg is
** already populated with all columns from the current row of the sub-query.
**
** If argument regPartSize is non-zero, then it is a register containing the
** number of rows in the current partition.
*/
static void windowAggStep(
  Parse *pParse, 
  Window *pMWin,                  /* Linked list of window functions */
  int csr,                        /* Read arguments from this cursor */
  int bInverse,                   /* True to invoke xInverse instead of xStep */
  int reg,                        /* Array of registers */
  int regPartSize                 /* Register containing size of partition */
){
  Vdbe *v = sqlite3GetVdbe(pParse);
  Window *pWin;
  for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
    int flags = pWin->pFunc->funcFlags;
    int regArg;
    int nArg = windowArgCount(pWin);

    if( csr>=0 ){
      int i;
      for(i=0; i<nArg; i++){
        sqlite3VdbeAddOp3(v, OP_Column, csr, pWin->iArgCol+i, reg+i);
      }
      regArg = reg;
      if( flags & SQLITE_FUNC_WINDOW_SIZE ){
        if( nArg==0 ){
          regArg = regPartSize;
        }else{
          sqlite3VdbeAddOp2(v, OP_SCopy, regPartSize, reg+nArg);
        }
        nArg++;
      }
    }else{
      assert( !(flags & SQLITE_FUNC_WINDOW_SIZE) );
      regArg = reg + pWin->iArgCol;
    }

    if( (pWin->pFunc->funcFlags & SQLITE_FUNC_MINMAX) 
      && pWin->eStart!=TK_UNBOUNDED 
    ){
      int addrIsNull = sqlite3VdbeAddOp1(v, OP_IsNull, regArg);
      VdbeCoverage(v);
      if( bInverse==0 ){
        sqlite3VdbeAddOp2(v, OP_AddImm, pWin->regApp+1, 1);
        sqlite3VdbeAddOp2(v, OP_SCopy, regArg, pWin->regApp);
        sqlite3VdbeAddOp3(v, OP_MakeRecord, pWin->regApp, 2, pWin->regApp+2);
        sqlite3VdbeAddOp2(v, OP_IdxInsert, pWin->csrApp, pWin->regApp+2);
      }else{
        sqlite3VdbeAddOp4Int(v, OP_SeekGE, pWin->csrApp, 0, regArg, 1);
        VdbeCoverageNeverTaken(v);
        sqlite3VdbeAddOp1(v, OP_Delete, pWin->csrApp);
        sqlite3VdbeJumpHere(v, sqlite3VdbeCurrentAddr(v)-2);
      }
      sqlite3VdbeJumpHere(v, addrIsNull);
    }else if( pWin->regApp ){
      assert( pWin->pFunc->zName==nth_valueName
           || pWin->pFunc->zName==first_valueName
      );
      assert( bInverse==0 || bInverse==1 );
      sqlite3VdbeAddOp2(v, OP_AddImm, pWin->regApp+1-bInverse, 1);
    }else if( pWin->pFunc->zName==leadName
           || pWin->pFunc->zName==lagName
    ){
      /* no-op */
    }else{
      int addrIf = 0;
      if( pWin->pFilter ){
        int regTmp;
        assert( nArg==0 || nArg==pWin->pOwner->x.pList->nExpr );
        assert( nArg || pWin->pOwner->x.pList==0 );
        if( csr>0 ){
          regTmp = sqlite3GetTempReg(pParse);
          sqlite3VdbeAddOp3(v, OP_Column, csr, pWin->iArgCol+nArg,regTmp);
        }else{
          regTmp = regArg + nArg;
        }
        addrIf = sqlite3VdbeAddOp3(v, OP_IfNot, regTmp, 0, 1);
        VdbeCoverage(v);
        if( csr>0 ){
          sqlite3ReleaseTempReg(pParse, regTmp);
        }
      }
      if( pWin->pFunc->funcFlags & SQLITE_FUNC_NEEDCOLL ){
        CollSeq *pColl;
        assert( nArg>0 );
        pColl = sqlite3ExprNNCollSeq(pParse, pWin->pOwner->x.pList->a[0].pExpr);
        sqlite3VdbeAddOp4(v, OP_CollSeq, 0,0,0, (const char*)pColl, P4_COLLSEQ);
      }
      sqlite3VdbeAddOp3(v, bInverse? OP_AggInverse : OP_AggStep, 
                        bInverse, regArg, pWin->regAccum);
      sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
      sqlite3VdbeChangeP5(v, (u8)nArg);
      if( addrIf ) sqlite3VdbeJumpHere(v, addrIf);
    }
  }
}

/*
** Generate VM code to invoke either xValue() (bFinal==0) or xFinalize()
** (bFinal==1) for each window function in the linked list starting at
** pMWin. Or, for built-in window-functions that do not use the standard
** API, generate the equivalent VM code.
*/
static void windowAggFinal(Parse *pParse, Window *pMWin, int bFinal){
  Vdbe *v = sqlite3GetVdbe(pParse);
  Window *pWin;

  for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
    if( (pWin->pFunc->funcFlags & SQLITE_FUNC_MINMAX) 
     && pWin->eStart!=TK_UNBOUNDED 
    ){
      sqlite3VdbeAddOp2(v, OP_Null, 0, pWin->regResult);
      sqlite3VdbeAddOp1(v, OP_Last, pWin->csrApp);
      VdbeCoverage(v);
      sqlite3VdbeAddOp3(v, OP_Column, pWin->csrApp, 0, pWin->regResult);
      sqlite3VdbeJumpHere(v, sqlite3VdbeCurrentAddr(v)-2);
      if( bFinal ){
        sqlite3VdbeAddOp1(v, OP_ResetSorter, pWin->csrApp);
      }
    }else if( pWin->regApp ){
    }else{
      if( bFinal ){
        sqlite3VdbeAddOp2(v, OP_AggFinal, pWin->regAccum, windowArgCount(pWin));
        sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
        sqlite3VdbeAddOp2(v, OP_Copy, pWin->regAccum, pWin->regResult);
        sqlite3VdbeAddOp2(v, OP_Null, 0, pWin->regAccum);
      }else{
        sqlite3VdbeAddOp3(v, OP_AggValue, pWin->regAccum, windowArgCount(pWin),
                             pWin->regResult);
        sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
      }
    }
  }
}

/*
** This function generates VM code to invoke the sub-routine at address
** lblFlushPart once for each partition with the entire partition cached in
** the Window.iEphCsr temp table.
*/
static void windowPartitionCache(
  Parse *pParse,
  Select *p,                      /* The rewritten SELECT statement */
  WhereInfo *pWInfo,              /* WhereInfo to call WhereEnd() on */
  int regFlushPart,               /* Register to use with Gosub lblFlushPart */
  int lblFlushPart,               /* Subroutine to Gosub to */
  int *pRegSize                   /* OUT: Register containing partition size */
){
  Window *pMWin = p->pWin;
  Vdbe *v = sqlite3GetVdbe(pParse);
  int iSubCsr = p->pSrc->a[0].iCursor;
  int nSub = p->pSrc->a[0].pTab->nCol;
  int k;

  int reg = pParse->nMem+1;
  int regRecord = reg+nSub;
  int regRowid = regRecord+1;

  *pRegSize = regRowid;
  pParse->nMem += nSub + 2;

  /* Load the column values for the row returned by the sub-select
  ** into an array of registers starting at reg. */
  for(k=0; k<nSub; k++){
    sqlite3VdbeAddOp3(v, OP_Column, iSubCsr, k, reg+k);
  }
  sqlite3VdbeAddOp3(v, OP_MakeRecord, reg, nSub, regRecord);

  /* Check if this is the start of a new partition. If so, call the
  ** flush_partition sub-routine.  */
  if( pMWin->pPartition ){
    int addr;
    ExprList *pPart = pMWin->pPartition;
    int nPart = pPart->nExpr;
    int regNewPart = reg + pMWin->nBufferCol;
    KeyInfo *pKeyInfo = sqlite3KeyInfoFromExprList(pParse, pPart, 0, 0);

    addr = sqlite3VdbeAddOp3(v, OP_Compare, regNewPart, pMWin->regPart,nPart);
    sqlite3VdbeAppendP4(v, (void*)pKeyInfo, P4_KEYINFO);
    sqlite3VdbeAddOp3(v, OP_Jump, addr+2, addr+4, addr+2);
    VdbeCoverageEqNe(v);
    sqlite3VdbeAddOp3(v, OP_Copy, regNewPart, pMWin->regPart, nPart-1);
    sqlite3VdbeAddOp2(v, OP_Gosub, regFlushPart, lblFlushPart);
    VdbeComment((v, "call flush_partition"));
  }

  /* Buffer the current row in the ephemeral table. */
  sqlite3VdbeAddOp2(v, OP_NewRowid, pMWin->iEphCsr, regRowid);
  sqlite3VdbeAddOp3(v, OP_Insert, pMWin->iEphCsr, regRecord, regRowid);

  /* End of the input loop */
  sqlite3WhereEnd(pWInfo);

  /* Invoke "flush_partition" to deal with the final (or only) partition */
  sqlite3VdbeAddOp2(v, OP_Gosub, regFlushPart, lblFlushPart);
  VdbeComment((v, "call flush_partition"));
}

/*
** Invoke the sub-routine at regGosub (generated by code in select.c) to
** return the current row of Window.iEphCsr. If all window functions are
** aggregate window functions that use the standard API, a single
** OP_Gosub instruction is all that this routine generates. Extra VM code
** for per-row processing is only generated for the following built-in window
** functions:
**
**   nth_value()
**   first_value()
**   lag()
**   lead()
*/
static void windowReturnOneRow(
  Parse *pParse,
  Window *pMWin,
  int regGosub,
  int addrGosub
){
  Vdbe *v = sqlite3GetVdbe(pParse);
  Window *pWin;
  for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
    FuncDef *pFunc = pWin->pFunc;
    if( pFunc->zName==nth_valueName
     || pFunc->zName==first_valueName
    ){
      int csr = pWin->csrApp;
      int lbl = sqlite3VdbeMakeLabel(v);
      int tmpReg = sqlite3GetTempReg(pParse);
      sqlite3VdbeAddOp2(v, OP_Null, 0, pWin->regResult);

      if( pFunc->zName==nth_valueName ){
        sqlite3VdbeAddOp3(v, OP_Column, pMWin->iEphCsr, pWin->iArgCol+1,tmpReg);
        windowCheckIntValue(pParse, tmpReg, 2);
      }else{
        sqlite3VdbeAddOp2(v, OP_Integer, 1, tmpReg);
      }
      sqlite3VdbeAddOp3(v, OP_Add, tmpReg, pWin->regApp, tmpReg);
      sqlite3VdbeAddOp3(v, OP_Gt, pWin->regApp+1, lbl, tmpReg);
      VdbeCoverageNeverNull(v);
      sqlite3VdbeAddOp3(v, OP_SeekRowid, csr, 0, tmpReg);
      VdbeCoverageNeverTaken(v);
      sqlite3VdbeAddOp3(v, OP_Column, csr, pWin->iArgCol, pWin->regResult);
      sqlite3VdbeResolveLabel(v, lbl);
      sqlite3ReleaseTempReg(pParse, tmpReg);
    }
    else if( pFunc->zName==leadName || pFunc->zName==lagName ){
      int nArg = pWin->pOwner->x.pList->nExpr;
      int iEph = pMWin->iEphCsr;
      int csr = pWin->csrApp;
      int lbl = sqlite3VdbeMakeLabel(v);
      int tmpReg = sqlite3GetTempReg(pParse);

      if( nArg<3 ){
        sqlite3VdbeAddOp2(v, OP_Null, 0, pWin->regResult);
      }else{
        sqlite3VdbeAddOp3(v, OP_Column, iEph, pWin->iArgCol+2, pWin->regResult);
      }
      sqlite3VdbeAddOp2(v, OP_Rowid, iEph, tmpReg);
      if( nArg<2 ){
        int val = (pFunc->zName==leadName ? 1 : -1);
        sqlite3VdbeAddOp2(v, OP_AddImm, tmpReg, val);
      }else{
        int op = (pFunc->zName==leadName ? OP_Add : OP_Subtract);
        int tmpReg2 = sqlite3GetTempReg(pParse);
        sqlite3VdbeAddOp3(v, OP_Column, iEph, pWin->iArgCol+1, tmpReg2);
        sqlite3VdbeAddOp3(v, op, tmpReg2, tmpReg, tmpReg);
        sqlite3ReleaseTempReg(pParse, tmpReg2);
      }

      sqlite3VdbeAddOp3(v, OP_SeekRowid, csr, lbl, tmpReg);
      VdbeCoverage(v);
      sqlite3VdbeAddOp3(v, OP_Column, csr, pWin->iArgCol, pWin->regResult);
      sqlite3VdbeResolveLabel(v, lbl);
      sqlite3ReleaseTempReg(pParse, tmpReg);
    }
  }
  sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
}

/*
** Invoke the code generated by windowReturnOneRow() and, optionally, the
** xInverse() function for each window function, for one or more rows
** from the Window.iEphCsr temp table. This routine generates VM code
** similar to:
**
**   while( regCtr>0 ){
**     regCtr--;
**     windowReturnOneRow()
**     if( bInverse ){
**       AggInverse
**     }
**     Next (Window.iEphCsr)
**   }
*/
static void windowReturnRows(
  Parse *pParse,
  Window *pMWin,                  /* List of window functions */
  int regCtr,                     /* Register containing number of rows */
  int regGosub,                   /* Register for Gosub addrGosub */
  int addrGosub,                  /* Address of sub-routine for ReturnOneRow */
  int regInvArg,                  /* Array of registers for xInverse args */
  int regInvSize                  /* Register containing size of partition */
){
  int addr;
  Vdbe *v = sqlite3GetVdbe(pParse);
  windowAggFinal(pParse, pMWin, 0);
  addr = sqlite3VdbeAddOp3(v, OP_IfPos, regCtr, sqlite3VdbeCurrentAddr(v)+2 ,1);
  VdbeCoverage(v);
  sqlite3VdbeAddOp2(v, OP_Goto, 0, 0);
  windowReturnOneRow(pParse, pMWin, regGosub, addrGosub);
  if( regInvArg ){
    windowAggStep(pParse, pMWin, pMWin->iEphCsr, 1, regInvArg, regInvSize);
  }
  sqlite3VdbeAddOp2(v, OP_Next, pMWin->iEphCsr, addr);
  VdbeCoverage(v);
  sqlite3VdbeJumpHere(v, addr+1);   /* The OP_Goto */
}

/*
** Generate code to set the accumulator register for each window function
** in the linked list passed as the second argument to NULL. And perform
** any equivalent initialization required by any built-in window functions
** in the list.
*/
static int windowInitAccum(Parse *pParse, Window *pMWin){
  Vdbe *v = sqlite3GetVdbe(pParse);
  int regArg;
  int nArg = 0;
  Window *pWin;
  for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
    FuncDef *pFunc = pWin->pFunc;
    sqlite3VdbeAddOp2(v, OP_Null, 0, pWin->regAccum);
    nArg = MAX(nArg, windowArgCount(pWin));
    if( pFunc->zName==nth_valueName
     || pFunc->zName==first_valueName
    ){
      sqlite3VdbeAddOp2(v, OP_Integer, 0, pWin->regApp);
      sqlite3VdbeAddOp2(v, OP_Integer, 0, pWin->regApp+1);
    }

    if( (pFunc->funcFlags & SQLITE_FUNC_MINMAX) && pWin->csrApp ){
      assert( pWin->eStart!=TK_UNBOUNDED );
      sqlite3VdbeAddOp1(v, OP_ResetSorter, pWin->csrApp);
      sqlite3VdbeAddOp2(v, OP_Integer, 0, pWin->regApp+1);
    }
  }
  regArg = pParse->nMem+1;
  pParse->nMem += nArg;
  return regArg;
}


/*
** This function does the work of sqlite3WindowCodeStep() for all "ROWS"
** window frame types except for "BETWEEN UNBOUNDED PRECEDING AND CURRENT
** ROW". Pseudo-code for each follows.
**
** ROWS BETWEEN <expr1> PRECEDING AND <expr2> FOLLOWING
**
**     ...
**       if( new partition ){
**         Gosub flush_partition
**       }
**       Insert (record in eph-table)
**     sqlite3WhereEnd()
**     Gosub flush_partition
**  
**   flush_partition:
**     Once {
**       OpenDup (iEphCsr -> csrStart)
**       OpenDup (iEphCsr -> csrEnd)
**     }
**     regStart = <expr1>                // PRECEDING expression
**     regEnd = <expr2>                  // FOLLOWING expression
**     if( regStart<0 || regEnd<0 ){ error! }
**     Rewind (csr,csrStart,csrEnd)      // if EOF goto flush_partition_done
**       Next(csrEnd)                    // if EOF skip Aggstep
**       Aggstep (csrEnd)
**       if( (regEnd--)<=0 ){
**         AggFinal (xValue)
**         Gosub addrGosub
**         Next(csr)                // if EOF goto flush_partition_done
**         if( (regStart--)<=0 ){
**           AggInverse (csrStart)
**           Next(csrStart)
**         }
**       }
**   flush_partition_done:
**     ResetSorter (csr)
**     Return
**
** ROWS BETWEEN <expr> PRECEDING    AND CURRENT ROW
** ROWS BETWEEN CURRENT ROW         AND <expr> FOLLOWING
** ROWS BETWEEN UNBOUNDED PRECEDING AND <expr> FOLLOWING
**
**   These are similar to the above. For "CURRENT ROW", intialize the
**   register to 0. For "UNBOUNDED PRECEDING" to infinity.
**
** ROWS BETWEEN <expr> PRECEDING    AND UNBOUNDED FOLLOWING
** ROWS BETWEEN CURRENT ROW         AND UNBOUNDED FOLLOWING
**
**     Rewind (csr,csrStart,csrEnd)    // if EOF goto flush_partition_done
**     while( 1 ){
**       Next(csrEnd)                  // Exit while(1) at EOF
**       Aggstep (csrEnd)
**     }
**     while( 1 ){
**       AggFinal (xValue)
**       Gosub addrGosub
**       Next(csr)                     // if EOF goto flush_partition_done
**       if( (regStart--)<=0 ){
**         AggInverse (csrStart)
**         Next(csrStart)
**       }
**     }
**
**   For the "CURRENT ROW AND UNBOUNDED FOLLOWING" case, the final if() 
**   condition is always true (as if regStart were initialized to 0).
**
** RANGE BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING
** 
**   This is the only RANGE case handled by this routine. It modifies the
**   second while( 1 ) loop in "ROWS BETWEEN CURRENT ... UNBOUNDED..." to
**   be:
**
**     while( 1 ){
**       AggFinal (xValue)
**       while( 1 ){
**         regPeer++
**         Gosub addrGosub
**         Next(csr)                     // if EOF goto flush_partition_done
**         if( new peer ) break;
**       }
**       while( (regPeer--)>0 ){
**         AggInverse (csrStart)
**         Next(csrStart)
**       }
**     }
**
** ROWS BETWEEN <expr> FOLLOWING    AND <expr> FOLLOWING
**
**   regEnd = regEnd - regStart
**   Rewind (csr,csrStart,csrEnd)   // if EOF goto flush_partition_done
**     Aggstep (csrEnd)
**     Next(csrEnd)                 // if EOF fall-through
**     if( (regEnd--)<=0 ){
**       if( (regStart--)<=0 ){
**         AggFinal (xValue)
**         Gosub addrGosub
**         Next(csr)              // if EOF goto flush_partition_done
**       }
**       AggInverse (csrStart)
**       Next (csrStart)
**     }
**
** ROWS BETWEEN <expr> PRECEDING    AND <expr> PRECEDING
**
**   Replace the bit after "Rewind" in the above with:
**
**     if( (regEnd--)<=0 ){
**       AggStep (csrEnd)
**       Next (csrEnd)
**     }
**     AggFinal (xValue)
**     Gosub addrGosub
**     Next(csr)                  // if EOF goto flush_partition_done
**     if( (regStart--)<=0 ){
**       AggInverse (csr2)
**       Next (csr2)
**     }
**
*/
static void windowCodeRowExprStep(
  Parse *pParse, 
  Select *p,
  WhereInfo *pWInfo,
  int regGosub, 
  int addrGosub
){
  Window *pMWin = p->pWin;
  Vdbe *v = sqlite3GetVdbe(pParse);
  int regFlushPart;               /* Register for "Gosub flush_partition" */
  int lblFlushPart;               /* Label for "Gosub flush_partition" */
  int lblFlushDone;               /* Label for "Gosub flush_partition_done" */

  int regArg;
  int addr;
  int csrStart = pParse->nTab++;
  int csrEnd = pParse->nTab++;
  int regStart;                    /* Value of <expr> PRECEDING */
  int regEnd;                      /* Value of <expr> FOLLOWING */
  int addrGoto;
  int addrTop;
  int addrIfPos1 = 0;
  int addrIfPos2 = 0;
  int regSize = 0;

  assert( pMWin->eStart==TK_PRECEDING 
       || pMWin->eStart==TK_CURRENT 
       || pMWin->eStart==TK_FOLLOWING 
       || pMWin->eStart==TK_UNBOUNDED 
  );
  assert( pMWin->eEnd==TK_FOLLOWING 
       || pMWin->eEnd==TK_CURRENT 
       || pMWin->eEnd==TK_UNBOUNDED 
       || pMWin->eEnd==TK_PRECEDING 
  );

  /* Allocate register and label for the "flush_partition" sub-routine. */
  regFlushPart = ++pParse->nMem;
  lblFlushPart = sqlite3VdbeMakeLabel(v);
  lblFlushDone = sqlite3VdbeMakeLabel(v);

  regStart = ++pParse->nMem;
  regEnd = ++pParse->nMem;

  windowPartitionCache(pParse, p, pWInfo, regFlushPart, lblFlushPart, &regSize);

  addrGoto = sqlite3VdbeAddOp0(v, OP_Goto);

  /* Start of "flush_partition" */
  sqlite3VdbeResolveLabel(v, lblFlushPart);
  sqlite3VdbeAddOp2(v, OP_Once, 0, sqlite3VdbeCurrentAddr(v)+3);
  VdbeCoverage(v);
  VdbeComment((v, "Flush_partition subroutine"));
  sqlite3VdbeAddOp2(v, OP_OpenDup, csrStart, pMWin->iEphCsr);
  sqlite3VdbeAddOp2(v, OP_OpenDup, csrEnd, pMWin->iEphCsr);

  /* If either regStart or regEnd are not non-negative integers, throw 
  ** an exception.  */
  if( pMWin->pStart ){
    sqlite3ExprCode(pParse, pMWin->pStart, regStart);
    windowCheckIntValue(pParse, regStart, 0);
  }
  if( pMWin->pEnd ){
    sqlite3ExprCode(pParse, pMWin->pEnd, regEnd);
    windowCheckIntValue(pParse, regEnd, 1);
  }

  /* If this is "ROWS <expr1> FOLLOWING AND ROWS <expr2> FOLLOWING", do:
  **
  **   if( regEnd<regStart ){
  **     // The frame always consists of 0 rows
  **     regStart = regSize;
  **   }
  **   regEnd = regEnd - regStart;
  */
  if( pMWin->pEnd && pMWin->eStart==TK_FOLLOWING ){
    assert( pMWin->pStart!=0 );
    assert( pMWin->eEnd==TK_FOLLOWING );
    sqlite3VdbeAddOp3(v, OP_Ge, regStart, sqlite3VdbeCurrentAddr(v)+2, regEnd);
    VdbeCoverageNeverNull(v);
    sqlite3VdbeAddOp2(v, OP_Copy, regSize, regStart);
    sqlite3VdbeAddOp3(v, OP_Subtract, regStart, regEnd, regEnd);
  }

  if( pMWin->pStart && pMWin->eEnd==TK_PRECEDING ){
    assert( pMWin->pEnd!=0 );
    assert( pMWin->eStart==TK_PRECEDING );
    sqlite3VdbeAddOp3(v, OP_Le, regStart, sqlite3VdbeCurrentAddr(v)+3, regEnd);
    VdbeCoverageNeverNull(v);
    sqlite3VdbeAddOp2(v, OP_Copy, regSize, regStart);
    sqlite3VdbeAddOp2(v, OP_Copy, regSize, regEnd);
  }

  /* Initialize the accumulator register for each window function to NULL */
  regArg = windowInitAccum(pParse, pMWin);

  sqlite3VdbeAddOp2(v, OP_Rewind, pMWin->iEphCsr, lblFlushDone);
  VdbeCoverage(v);
  sqlite3VdbeAddOp2(v, OP_Rewind, csrStart, lblFlushDone);
  VdbeCoverageNeverTaken(v);
  sqlite3VdbeChangeP5(v, 1);
  sqlite3VdbeAddOp2(v, OP_Rewind, csrEnd, lblFlushDone);
  VdbeCoverageNeverTaken(v);
  sqlite3VdbeChangeP5(v, 1);

  /* Invoke AggStep function for each window function using the row that
  ** csrEnd currently points to. Or, if csrEnd is already at EOF,
  ** do nothing.  */
  addrTop = sqlite3VdbeCurrentAddr(v);
  if( pMWin->eEnd==TK_PRECEDING ){
    addrIfPos1 = sqlite3VdbeAddOp3(v, OP_IfPos, regEnd, 0 , 1);
    VdbeCoverage(v);
  }
  sqlite3VdbeAddOp2(v, OP_Next, csrEnd, sqlite3VdbeCurrentAddr(v)+2);
  VdbeCoverage(v);
  addr = sqlite3VdbeAddOp0(v, OP_Goto);
  windowAggStep(pParse, pMWin, csrEnd, 0, regArg, regSize);
  if( pMWin->eEnd==TK_UNBOUNDED ){
    sqlite3VdbeAddOp2(v, OP_Goto, 0, addrTop);
    sqlite3VdbeJumpHere(v, addr);
    addrTop = sqlite3VdbeCurrentAddr(v);
  }else{
    sqlite3VdbeJumpHere(v, addr);
    if( pMWin->eEnd==TK_PRECEDING ){
      sqlite3VdbeJumpHere(v, addrIfPos1);
    }
  }

  if( pMWin->eEnd==TK_FOLLOWING ){
    addrIfPos1 = sqlite3VdbeAddOp3(v, OP_IfPos, regEnd, 0 , 1);
    VdbeCoverage(v);
  }
  if( pMWin->eStart==TK_FOLLOWING ){
    addrIfPos2 = sqlite3VdbeAddOp3(v, OP_IfPos, regStart, 0 , 1);
    VdbeCoverage(v);
  }
  windowAggFinal(pParse, pMWin, 0);
  windowReturnOneRow(pParse, pMWin, regGosub, addrGosub);
  sqlite3VdbeAddOp2(v, OP_Next, pMWin->iEphCsr, sqlite3VdbeCurrentAddr(v)+2);
  VdbeCoverage(v);
  sqlite3VdbeAddOp2(v, OP_Goto, 0, lblFlushDone);
  if( pMWin->eStart==TK_FOLLOWING ){
    sqlite3VdbeJumpHere(v, addrIfPos2);
  }

  if( pMWin->eStart==TK_CURRENT 
   || pMWin->eStart==TK_PRECEDING 
   || pMWin->eStart==TK_FOLLOWING 
  ){
    int lblSkipInverse = sqlite3VdbeMakeLabel(v);;
    if( pMWin->eStart==TK_PRECEDING ){
      sqlite3VdbeAddOp3(v, OP_IfPos, regStart, lblSkipInverse, 1);
      VdbeCoverage(v);
    }
    if( pMWin->eStart==TK_FOLLOWING ){
      sqlite3VdbeAddOp2(v, OP_Next, csrStart, sqlite3VdbeCurrentAddr(v)+2);
      VdbeCoverage(v);
      sqlite3VdbeAddOp2(v, OP_Goto, 0, lblSkipInverse);
    }else{
      sqlite3VdbeAddOp2(v, OP_Next, csrStart, sqlite3VdbeCurrentAddr(v)+1);
      VdbeCoverageAlwaysTaken(v);
    }
    windowAggStep(pParse, pMWin, csrStart, 1, regArg, regSize);
    sqlite3VdbeResolveLabel(v, lblSkipInverse);
  }
  if( pMWin->eEnd==TK_FOLLOWING ){
    sqlite3VdbeJumpHere(v, addrIfPos1);
  }
  sqlite3VdbeAddOp2(v, OP_Goto, 0, addrTop);

  /* flush_partition_done: */
  sqlite3VdbeResolveLabel(v, lblFlushDone);
  sqlite3VdbeAddOp1(v, OP_ResetSorter, pMWin->iEphCsr);
  sqlite3VdbeAddOp1(v, OP_Return, regFlushPart);
  VdbeComment((v, "end flush_partition subroutine"));

  /* Jump to here to skip over flush_partition */
  sqlite3VdbeJumpHere(v, addrGoto);
}

/*
** This function does the work of sqlite3WindowCodeStep() for cases that
** would normally be handled by windowCodeDefaultStep() when there are
** one or more built-in window-functions that require the entire partition
** to be cached in a temp table before any rows can be returned. Additionally.
** "RANGE BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING" is always handled by
** this function.
**
** Pseudo-code corresponding to the VM code generated by this function
** for each type of window follows.
**
** RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
**
**   flush_partition:
**     Once {
**       OpenDup (iEphCsr -> csrLead)
**     }
**     Integer ctr 0
**     foreach row (csrLead){
**       if( new peer ){
**         AggFinal (xValue)
**         for(i=0; i<ctr; i++){
**           Gosub addrGosub
**           Next iEphCsr
**         }
**         Integer ctr 0
**       }
**       AggStep (csrLead)
**       Incr ctr
**     }
**
**     AggFinal (xFinalize)
**     for(i=0; i<ctr; i++){
**       Gosub addrGosub
**       Next iEphCsr
**     }
**
**     ResetSorter (csr)
**     Return
**
** ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
**
**   As above, except that the "if( new peer )" branch is always taken.
**
** RANGE BETWEEN CURRENT ROW AND CURRENT ROW 
**
**   As above, except that each of the for() loops becomes:
**
**         for(i=0; i<ctr; i++){
**           Gosub addrGosub
**           AggInverse (iEphCsr)
**           Next iEphCsr
**         }
**
** RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
**
**   flush_partition:
**     Once {
**       OpenDup (iEphCsr -> csrLead)
**     }
**     foreach row (csrLead) {
**       AggStep (csrLead)
**     }
**     foreach row (iEphCsr) {
**       Gosub addrGosub
**     }
** 
** RANGE BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING
**
**   flush_partition:
**     Once {
**       OpenDup (iEphCsr -> csrLead)
**     }
**     foreach row (csrLead){
**       AggStep (csrLead)
**     }
**     Rewind (csrLead)
**     Integer ctr 0
**     foreach row (csrLead){
**       if( new peer ){
**         AggFinal (xValue)
**         for(i=0; i<ctr; i++){
**           Gosub addrGosub
**           AggInverse (iEphCsr)
**           Next iEphCsr
**         }
**         Integer ctr 0
**       }
**       Incr ctr
**     }
**
**     AggFinal (xFinalize)
**     for(i=0; i<ctr; i++){
**       Gosub addrGosub
**       Next iEphCsr
**     }
**
**     ResetSorter (csr)
**     Return
*/
static void windowCodeCacheStep(
  Parse *pParse, 
  Select *p,
  WhereInfo *pWInfo,
  int regGosub, 
  int addrGosub
){
  Window *pMWin = p->pWin;
  Vdbe *v = sqlite3GetVdbe(pParse);
  int k;
  int addr;
  ExprList *pPart = pMWin->pPartition;
  ExprList *pOrderBy = pMWin->pOrderBy;
  int nPeer = pOrderBy ? pOrderBy->nExpr : 0;
  int regNewPeer;

  int addrGoto;                   /* Address of Goto used to jump flush_par.. */
  int addrNext;                   /* Jump here for next iteration of loop */
  int regFlushPart;
  int lblFlushPart;
  int csrLead;
  int regCtr;
  int regArg;                     /* Register array to martial function args */
  int regSize;
  int lblEmpty;
  int bReverse = pMWin->pOrderBy && pMWin->eStart==TK_CURRENT 
          && pMWin->eEnd==TK_UNBOUNDED;

  assert( (pMWin->eStart==TK_UNBOUNDED && pMWin->eEnd==TK_CURRENT) 
       || (pMWin->eStart==TK_UNBOUNDED && pMWin->eEnd==TK_UNBOUNDED) 
       || (pMWin->eStart==TK_CURRENT && pMWin->eEnd==TK_CURRENT) 
       || (pMWin->eStart==TK_CURRENT && pMWin->eEnd==TK_UNBOUNDED) 
  );

  lblEmpty = sqlite3VdbeMakeLabel(v);
  regNewPeer = pParse->nMem+1;
  pParse->nMem += nPeer;

  /* Allocate register and label for the "flush_partition" sub-routine. */
  regFlushPart = ++pParse->nMem;
  lblFlushPart = sqlite3VdbeMakeLabel(v);

  csrLead = pParse->nTab++;
  regCtr = ++pParse->nMem;

  windowPartitionCache(pParse, p, pWInfo, regFlushPart, lblFlushPart, &regSize);
  addrGoto = sqlite3VdbeAddOp0(v, OP_Goto);

  /* Start of "flush_partition" */
  sqlite3VdbeResolveLabel(v, lblFlushPart);
  sqlite3VdbeAddOp2(v, OP_Once, 0, sqlite3VdbeCurrentAddr(v)+2);
  VdbeCoverage(v);
  sqlite3VdbeAddOp2(v, OP_OpenDup, csrLead, pMWin->iEphCsr);

  /* Initialize the accumulator register for each window function to NULL */
  regArg = windowInitAccum(pParse, pMWin);

  sqlite3VdbeAddOp2(v, OP_Integer, 0, regCtr);
  sqlite3VdbeAddOp2(v, OP_Rewind, csrLead, lblEmpty);
  VdbeCoverage(v);
  sqlite3VdbeAddOp2(v, OP_Rewind, pMWin->iEphCsr, lblEmpty);
  VdbeCoverageNeverTaken(v);

  if( bReverse ){
    int addr2 = sqlite3VdbeCurrentAddr(v);
    windowAggStep(pParse, pMWin, csrLead, 0, regArg, regSize);
    sqlite3VdbeAddOp2(v, OP_Next, csrLead, addr2);
    VdbeCoverage(v);
    sqlite3VdbeAddOp2(v, OP_Rewind, csrLead, lblEmpty);
    VdbeCoverageNeverTaken(v);
  }
  addrNext = sqlite3VdbeCurrentAddr(v);

  if( pOrderBy && (pMWin->eEnd==TK_CURRENT || pMWin->eStart==TK_CURRENT) ){
    int bCurrent = (pMWin->eStart==TK_CURRENT);
    int addrJump = 0;             /* Address of OP_Jump below */
    if( pMWin->eType==TK_RANGE ){
      int iOff = pMWin->nBufferCol + (pPart ? pPart->nExpr : 0);
      int regPeer = pMWin->regPart + (pPart ? pPart->nExpr : 0);
      KeyInfo *pKeyInfo = sqlite3KeyInfoFromExprList(pParse, pOrderBy, 0, 0);
      for(k=0; k<nPeer; k++){
        sqlite3VdbeAddOp3(v, OP_Column, csrLead, iOff+k, regNewPeer+k);
      }
      addr = sqlite3VdbeAddOp3(v, OP_Compare, regNewPeer, regPeer, nPeer);
      sqlite3VdbeAppendP4(v, (void*)pKeyInfo, P4_KEYINFO);
      addrJump = sqlite3VdbeAddOp3(v, OP_Jump, addr+2, 0, addr+2);
      VdbeCoverage(v);
      sqlite3VdbeAddOp3(v, OP_Copy, regNewPeer, regPeer, nPeer-1);
    }

    windowReturnRows(pParse, pMWin, regCtr, regGosub, addrGosub, 
        (bCurrent ? regArg : 0), (bCurrent ? regSize : 0)
    );
    if( addrJump ) sqlite3VdbeJumpHere(v, addrJump);
  }

  if( bReverse==0 ){
    windowAggStep(pParse, pMWin, csrLead, 0, regArg, regSize);
  }
  sqlite3VdbeAddOp2(v, OP_AddImm, regCtr, 1);
  sqlite3VdbeAddOp2(v, OP_Next, csrLead, addrNext);
  VdbeCoverage(v);

  windowReturnRows(pParse, pMWin, regCtr, regGosub, addrGosub, 0, 0);

  sqlite3VdbeResolveLabel(v, lblEmpty);
  sqlite3VdbeAddOp1(v, OP_ResetSorter, pMWin->iEphCsr);
  sqlite3VdbeAddOp1(v, OP_Return, regFlushPart);

  /* Jump to here to skip over flush_partition */
  sqlite3VdbeJumpHere(v, addrGoto);
}


/*
** RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
**
**   ...
**     if( new partition ){
**       AggFinal (xFinalize)
**       Gosub addrGosub
**       ResetSorter eph-table
**     }
**     else if( new peer ){
**       AggFinal (xValue)
**       Gosub addrGosub
**       ResetSorter eph-table
**     }
**     AggStep
**     Insert (record into eph-table)
**   sqlite3WhereEnd()
**   AggFinal (xFinalize)
**   Gosub addrGosub
**
** RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
**
**   As above, except take no action for a "new peer". Invoke
**   the sub-routine once only for each partition.
**
** RANGE BETWEEN CURRENT ROW AND CURRENT ROW
**
**   As above, except that the "new peer" condition is handled in the
**   same way as "new partition" (so there is no "else if" block).
**
** ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
** 
**   As above, except assume every row is a "new peer".
*/
static void windowCodeDefaultStep(
  Parse *pParse, 
  Select *p,
  WhereInfo *pWInfo,
  int regGosub, 
  int addrGosub
){
  Window *pMWin = p->pWin;
  Vdbe *v = sqlite3GetVdbe(pParse);
  int k;
  int iSubCsr = p->pSrc->a[0].iCursor;
  int nSub = p->pSrc->a[0].pTab->nCol;
  int reg = pParse->nMem+1;
  int regRecord = reg+nSub;
  int regRowid = regRecord+1;
  int addr;
  ExprList *pPart = pMWin->pPartition;
  ExprList *pOrderBy = pMWin->pOrderBy;

  assert( pMWin->eType==TK_RANGE 
      || (pMWin->eStart==TK_UNBOUNDED && pMWin->eEnd==TK_CURRENT)
  );

  assert( (pMWin->eStart==TK_UNBOUNDED && pMWin->eEnd==TK_CURRENT)
       || (pMWin->eStart==TK_UNBOUNDED && pMWin->eEnd==TK_UNBOUNDED)
       || (pMWin->eStart==TK_CURRENT && pMWin->eEnd==TK_CURRENT)
       || (pMWin->eStart==TK_CURRENT && pMWin->eEnd==TK_UNBOUNDED && !pOrderBy)
  );

  if( pMWin->eEnd==TK_UNBOUNDED ){
    pOrderBy = 0;
  }

  pParse->nMem += nSub + 2;

  /* Load the individual column values of the row returned by
  ** the sub-select into an array of registers. */
  for(k=0; k<nSub; k++){
    sqlite3VdbeAddOp3(v, OP_Column, iSubCsr, k, reg+k);
  }

  /* Check if this is the start of a new partition or peer group. */
  if( pPart || pOrderBy ){
    int nPart = (pPart ? pPart->nExpr : 0);
    int addrGoto = 0;
    int addrJump = 0;
    int nPeer = (pOrderBy ? pOrderBy->nExpr : 0);

    if( pPart ){
      int regNewPart = reg + pMWin->nBufferCol;
      KeyInfo *pKeyInfo = sqlite3KeyInfoFromExprList(pParse, pPart, 0, 0);
      addr = sqlite3VdbeAddOp3(v, OP_Compare, regNewPart, pMWin->regPart,nPart);
      sqlite3VdbeAppendP4(v, (void*)pKeyInfo, P4_KEYINFO);
      addrJump = sqlite3VdbeAddOp3(v, OP_Jump, addr+2, 0, addr+2);
      VdbeCoverageEqNe(v);
      windowAggFinal(pParse, pMWin, 1);
      if( pOrderBy ){
        addrGoto = sqlite3VdbeAddOp0(v, OP_Goto);
      }
    }

    if( pOrderBy ){
      int regNewPeer = reg + pMWin->nBufferCol + nPart;
      int regPeer = pMWin->regPart + nPart;

      if( addrJump ) sqlite3VdbeJumpHere(v, addrJump);
      if( pMWin->eType==TK_RANGE ){
        KeyInfo *pKeyInfo = sqlite3KeyInfoFromExprList(pParse, pOrderBy, 0, 0);
        addr = sqlite3VdbeAddOp3(v, OP_Compare, regNewPeer, regPeer, nPeer);
        sqlite3VdbeAppendP4(v, (void*)pKeyInfo, P4_KEYINFO);
        addrJump = sqlite3VdbeAddOp3(v, OP_Jump, addr+2, 0, addr+2);
        VdbeCoverage(v);
      }else{
        addrJump = 0;
      }
      windowAggFinal(pParse, pMWin, pMWin->eStart==TK_CURRENT);
      if( addrGoto ) sqlite3VdbeJumpHere(v, addrGoto);
    }

    sqlite3VdbeAddOp2(v, OP_Rewind, pMWin->iEphCsr,sqlite3VdbeCurrentAddr(v)+3);
    VdbeCoverage(v);
    sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
    sqlite3VdbeAddOp2(v, OP_Next, pMWin->iEphCsr, sqlite3VdbeCurrentAddr(v)-1);
    VdbeCoverage(v);

    sqlite3VdbeAddOp1(v, OP_ResetSorter, pMWin->iEphCsr);
    sqlite3VdbeAddOp3(
        v, OP_Copy, reg+pMWin->nBufferCol, pMWin->regPart, nPart+nPeer-1
    );

    if( addrJump ) sqlite3VdbeJumpHere(v, addrJump);
  }

  /* Invoke step function for window functions */
  windowAggStep(pParse, pMWin, -1, 0, reg, 0);

  /* Buffer the current row in the ephemeral table. */
  if( pMWin->nBufferCol>0 ){
    sqlite3VdbeAddOp3(v, OP_MakeRecord, reg, pMWin->nBufferCol, regRecord);
  }else{
    sqlite3VdbeAddOp2(v, OP_Blob, 0, regRecord);
    sqlite3VdbeAppendP4(v, (void*)"", 0);
  }
  sqlite3VdbeAddOp2(v, OP_NewRowid, pMWin->iEphCsr, regRowid);
  sqlite3VdbeAddOp3(v, OP_Insert, pMWin->iEphCsr, regRecord, regRowid);

  /* End the database scan loop. */
  sqlite3WhereEnd(pWInfo);

  windowAggFinal(pParse, pMWin, 1);
  sqlite3VdbeAddOp2(v, OP_Rewind, pMWin->iEphCsr,sqlite3VdbeCurrentAddr(v)+3);
  VdbeCoverage(v);
  sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
  sqlite3VdbeAddOp2(v, OP_Next, pMWin->iEphCsr, sqlite3VdbeCurrentAddr(v)-1);
  VdbeCoverage(v);
}

/*
** Allocate and return a duplicate of the Window object indicated by the
** third argument. Set the Window.pOwner field of the new object to
** pOwner.
*/
Window *sqlite3WindowDup(sqlite3 *db, Expr *pOwner, Window *p){
  Window *pNew = 0;
  if( ALWAYS(p) ){
    pNew = sqlite3DbMallocZero(db, sizeof(Window));
    if( pNew ){
      pNew->zName = sqlite3DbStrDup(db, p->zName);
      pNew->pFilter = sqlite3ExprDup(db, p->pFilter, 0);
      pNew->pPartition = sqlite3ExprListDup(db, p->pPartition, 0);
      pNew->pOrderBy = sqlite3ExprListDup(db, p->pOrderBy, 0);
      pNew->eType = p->eType;
      pNew->eEnd = p->eEnd;
      pNew->eStart = p->eStart;
      pNew->pStart = sqlite3ExprDup(db, p->pStart, 0);
      pNew->pEnd = sqlite3ExprDup(db, p->pEnd, 0);
      pNew->pOwner = pOwner;
    }
  }
  return pNew;
}

/*
** Return a copy of the linked list of Window objects passed as the
** second argument.
*/
Window *sqlite3WindowListDup(sqlite3 *db, Window *p){
  Window *pWin;
  Window *pRet = 0;
  Window **pp = &pRet;

  for(pWin=p; pWin; pWin=pWin->pNextWin){
    *pp = sqlite3WindowDup(db, 0, pWin);
    if( *pp==0 ) break;
    pp = &((*pp)->pNextWin);
  }

  return pRet;
}

/*
** sqlite3WhereBegin() has already been called for the SELECT statement 
** passed as the second argument when this function is invoked. It generates
** code to populate the Window.regResult register for each window function and
** invoke the sub-routine at instruction addrGosub once for each row.
** This function calls sqlite3WhereEnd() before returning. 
*/
void sqlite3WindowCodeStep(
  Parse *pParse,                  /* Parse context */
  Select *p,                      /* Rewritten SELECT statement */
  WhereInfo *pWInfo,              /* Context returned by sqlite3WhereBegin() */
  int regGosub,                   /* Register for OP_Gosub */
  int addrGosub                   /* OP_Gosub here to return each row */
){
  Window *pMWin = p->pWin;

  /* There are three different functions that may be used to do the work
  ** of this one, depending on the window frame and the specific built-in
  ** window functions used (if any).
  **
  ** windowCodeRowExprStep() handles all "ROWS" window frames, except for:
  **
  **   ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
  **
  ** The exception is because windowCodeRowExprStep() implements all window
  ** frame types by caching the entire partition in a temp table, and
  ** "ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW" is easy enough to
  ** implement without such a cache.
  **
  ** windowCodeCacheStep() is used for:
  **
  **   RANGE BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING
  **
  ** It is also used for anything not handled by windowCodeRowExprStep() 
  ** that invokes a built-in window function that requires the entire 
  ** partition to be cached in a temp table before any rows are returned
  ** (e.g. nth_value() or percent_rank()).
  **
  ** Finally, assuming there is no built-in window function that requires
  ** the partition to be cached, windowCodeDefaultStep() is used for:
  **
  **   RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW 
  **   RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
  **   RANGE BETWEEN CURRENT ROW AND CURRENT ROW 
  **   ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
  **
  ** windowCodeDefaultStep() is the only one of the three functions that
  ** does not cache each partition in a temp table before beginning to
  ** return rows.
  */
  if( pMWin->eType==TK_ROWS 
   && (pMWin->eStart!=TK_UNBOUNDED||pMWin->eEnd!=TK_CURRENT||!pMWin->pOrderBy)
  ){
    VdbeModuleComment((pParse->pVdbe, "Begin RowExprStep()"));
    windowCodeRowExprStep(pParse, p, pWInfo, regGosub, addrGosub);
  }else{
    Window *pWin;
    int bCache = 0;               /* True to use CacheStep() */

    if( pMWin->eStart==TK_CURRENT && pMWin->eEnd==TK_UNBOUNDED ){
      bCache = 1;
    }else{
      for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
        FuncDef *pFunc = pWin->pFunc;
        if( (pFunc->funcFlags & SQLITE_FUNC_WINDOW_SIZE)
         || (pFunc->zName==nth_valueName)
         || (pFunc->zName==first_valueName)
         || (pFunc->zName==leadName)
         || (pFunc->zName==lagName)
        ){
          bCache = 1;
          break;
        }
      }
    }

    /* Otherwise, call windowCodeDefaultStep().  */
    if( bCache ){
      VdbeModuleComment((pParse->pVdbe, "Begin CacheStep()"));
      windowCodeCacheStep(pParse, p, pWInfo, regGosub, addrGosub);
    }else{
      VdbeModuleComment((pParse->pVdbe, "Begin DefaultStep()"));
      windowCodeDefaultStep(pParse, p, pWInfo, regGosub, addrGosub);
    }
  }
}

#endif /* SQLITE_OMIT_WINDOWFUNC */