SQLite

#define SQLITE_FUNC_CONSTANT 0x0800 /* Constant inputs give a constant output */
#define SQLITE_FUNC_MINMAX   0x1000 /* True for min() and max() aggregates */
#define SQLITE_FUNC_SLOCHNG  0x2000 /* "Slow Change". Value constant during a
                                    ** single query - might change over time */
#define SQLITE_FUNC_AFFINITY 0x4000 /* Built-in affinity() function */
#define SQLITE_FUNC_OFFSET   0x8000 /* Built-in sqlite_offset() function */
#define SQLITE_FUNC_WINDOW   0x00010000 /* Built-in window-only function */
#define SQLITE_FUNC_WINDOW_SIZE 0x20000 /* Requires partition size as arg. */
#define SQLITE_FUNC_INTERNAL 0x00040000 /* For use by NestedParse() only */

/*
** The following three macros, FUNCTION(), LIKEFUNC() and AGGREGATE() are
** used to create the initializers for the FuncDef structures.
**
**   FUNCTION(zName, nArg, iArg, bNC, xFunc)

  int csrApp;             /* Function cursor (used by min/max) */
  int regApp;             /* Function register (also used by min/max) */
  int regPart;            /* First in a set of registers holding PARTITION BY
                          ** and ORDER BY values for the window */
  Expr *pOwner;           /* Expression object this window is attached to */
  int nBufferCol;         /* Number of columns in buffer table */
  int iArgCol;            /* Offset of first argument for this function */

  int regFirst;
  int regSize;
};

#ifndef SQLITE_OMIT_WINDOWFUNC
void sqlite3WindowDelete(sqlite3*, Window*);
void sqlite3WindowListDelete(sqlite3 *db, Window *p);
Window *sqlite3WindowAlloc(Parse*, int, int, Expr*, int , Expr*);
void sqlite3WindowAttach(Parse*, Expr*, Window*);

  }
}

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

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

  }else
  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

    pMWin->regPart = pParse->nMem+1;
    pParse->nMem += nPart;
    sqlite3VdbeAddOp3(v, OP_Null, 0, pMWin->regPart, pMWin->regPart+nPart-1);
  }

  pMWin->regFirst = ++pParse->nMem;
  sqlite3VdbeAddOp2(v, OP_Integer, 1, pMWin->regFirst);
  pMWin->regSize = ++pParse->nMem;
  sqlite3VdbeAddOp2(v, OP_Integer, 0, pMWin->regSize);

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

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

    }
  }
  regArg = pParse->nMem+1;
  pParse->nMem += nArg;
  return regArg;
}


/* 
** Return true if the entire partition should be cached in the ephemeral
** table before processing any rows.

*/
static int windowCachePartition(Window *pMWin){
  Window *pWin;
  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)
    ){
      return 1;
    }
  }
  return 0;
}


/*
** regOld and regNew are each the first register in an array of size
** pOrderBy->nExpr. This function generates code to compare the two
** arrays of registers using the collation sequences and other comparison
** parameters specified by pOrderBy. 
**

      reg = p->current.reg;
      windowReturnOneRow(pParse, pMWin, p->regGosub, p->addrGosub);
      break;

    case WINDOW_AGGINVERSE:
      csr = p->start.csr;
      reg = p->start.reg;
      windowAggStep(pParse, pMWin, csr, 1, p->regArg, pMWin->regSize);
      break;

    case WINDOW_AGGSTEP:
      csr = p->end.csr;
      reg = p->end.reg;
      windowAggStep(pParse, pMWin, csr, 0, p->regArg, pMWin->regSize);
      break;
  }

  if( jumpOnEof ){
    sqlite3VdbeAddOp2(v, OP_Next, csr, sqlite3VdbeCurrentAddr(v)+2);
    ret = sqlite3VdbeAddOp0(v, OP_Goto);
  }else{

**         if( eof ) break
**       }
**       RETURN_ROW
**     }
**     while( !eof csrCurrent ){
**       RETURN_ROW
**     }
**
** Sometimes, this function generates code to run in "cache mode" - meaning
** the entire partition is cached in the ephemeral table before any of its
** rows are processed, instead of processing rows as the sub-select delivers
** them. This is required by certain built-in window functions, for example
** percent_rank() or lead(). In that case, the relevant pseudo-code above
** is modified to:
**
**     ... loop started by sqlite3WhereBegin() ...
**     if( new partition ){
**       Gosub flush
**     }
**     Insert new row into eph table.
**   }
**   flush:
**     for each row in eph table {
**
** followed immediately by the code that usually follows the "Insert new row
** into eph table." line.
**
*/
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;
  ExprList *pOrderBy = pMWin->pOrderBy;
  Vdbe *v = sqlite3GetVdbe(pParse);
  int bCache;                     /* True if generating "cache-mode" code */
  int regFlushPart;               /* Register for "Gosub flush_partition" */
  int csrWrite;                   /* Cursor used to write to eph. table */
  int csrInput = p->pSrc->a[0].iCursor;     /* Cursor of sub-select */
  int nInput = p->pSrc->a[0].pTab->nCol;    /* Number of cols returned by sub */
  int iInput;                               /* To iterate through sub cols */
  int addrGoto;                   /* Address of OP_Goto */
  int addrIfNot;                  /* Address of OP_IfNot */
  int addrGosubFlush;             /* Address of OP_Gosub to flush: */
  int addrInteger;                /* Address of OP_Integer */
  int addrCacheRewind;            /* Address of OP_Rewind used in cache-mode */
  int addrCacheNext;              /* Jump here for next row in cache-mode */
  int addrShortcut = 0;
  int addrEmpty = 0;              /* Address of OP_Rewind in flush: */
  int addrPeerJump = 0;           /* Address of jump taken if not new peer */
  int regStart = 0;               /* Value of <expr> PRECEDING */
  int regEnd = 0;                 /* Value of <expr> FOLLOWING */
  int regNew;                     /* Array of registers holding new input row */
  int regRecord;                  /* regNew array in record form */

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

  /* Determine whether or not each partition will be cached before beginning
  ** to process rows within it.  */
  bCache = windowCachePartition(pMWin);

  /* Fill in the context object */
  memset(&s, 0, sizeof(WindowCodeArg));
  s.pParse = pParse;
  s.pMWin = pMWin;
  s.pVdbe = v;
  s.regGosub = regGosub;

    VdbeComment((v, "call flush_partition"));
    sqlite3VdbeAddOp3(v, OP_Copy, regNewPart, pMWin->regPart, nPart-1);
  }

  /* Insert the new row into the ephemeral table */
  sqlite3VdbeAddOp2(v, OP_NewRowid, csrWrite, regRowid);
  sqlite3VdbeAddOp3(v, OP_Insert, csrWrite, regRecord, regRowid);
  sqlite3VdbeAddOp2(v, OP_AddImm, pMWin->regSize, 1);

  if( bCache ){
    sqlite3VdbeAddOp2(v, OP_Integer, 0, pMWin->regFirst);
    sqlite3WhereEnd(pWInfo);
    addrInteger = sqlite3VdbeAddOp2(v, OP_Integer, 0, regFlushPart);
    if( pMWin->pPartition ){
      sqlite3VdbeJumpHere(v, addrGosubFlush);
    }
    addrCacheRewind = sqlite3VdbeAddOp1(v, OP_Rewind, csrWrite);
  }else{
    addrIfNot = sqlite3VdbeAddOp1(v, OP_IfNot, pMWin->regFirst);
  }

  /* This block is run for the first row of each partition */
  s.regArg = windowInitAccum(pParse, pMWin);

  if( regStart ){
    sqlite3ExprCode(pParse, pMWin->pStart, regStart);
    windowCheckIntValue(pParse, regStart, 0);
  }
  if( regEnd ){
    sqlite3ExprCode(pParse, pMWin->pEnd, regEnd);
    windowCheckIntValue(pParse, regEnd, 1);
  }

  if( pMWin->eStart==pMWin->eEnd && regStart && regEnd ){
    int op = ((pMWin->eStart==TK_FOLLOWING) ? OP_Ge : OP_Le);
    int addrGe = sqlite3VdbeAddOp3(v, op, regStart, 0, regEnd);
    windowAggFinal(pParse, pMWin, 0);
    if( bCache ){
      sqlite3VdbeAddOp2(v, OP_Rowid, csrWrite, regRowid);
      sqlite3VdbeAddOp3(v, OP_NotExists, s.current.csr, 0, regRowid);
      windowReturnOneRow(pParse, pMWin, regGosub, addrGosub);
      sqlite3VdbeAddOp2(v, OP_Next, csrWrite, addrCacheRewind+1);
    }else{
      sqlite3VdbeAddOp2(v, OP_Rewind, s.current.csr, 1);
      windowReturnOneRow(pParse, pMWin, regGosub, addrGosub);
      sqlite3VdbeAddOp1(v, OP_ResetSorter, s.current.csr);
    }
    addrShortcut = sqlite3VdbeAddOp0(v, OP_Goto);
    sqlite3VdbeJumpHere(v, addrGe);
  }
  if( pMWin->eStart==TK_FOLLOWING && pMWin->eType!=TK_RANGE && regEnd ){
    assert( pMWin->eEnd==TK_FOLLOWING );
    sqlite3VdbeAddOp3(v, OP_Subtract, regStart, regEnd, regStart);
  }

  if( pMWin->eStart!=TK_UNBOUNDED ){
    sqlite3VdbeAddOp2(v, OP_Rewind, s.start.csr, 1);
  }
  sqlite3VdbeAddOp2(v, OP_Rewind, s.current.csr, 1);
  sqlite3VdbeAddOp2(v, OP_Rewind, s.end.csr, 1);
  if( regPeer && pOrderBy ){
    if( bCache ){
      windowReadPeerValues(&s, csrWrite, regPeer);
    }else{
      sqlite3VdbeAddOp3(v, OP_Copy, regNewPeer, regPeer, pOrderBy->nExpr-1);
    }
    sqlite3VdbeAddOp3(v, OP_Copy, regPeer, s.start.reg, pOrderBy->nExpr-1);
    sqlite3VdbeAddOp3(v, OP_Copy, regPeer, s.current.reg, pOrderBy->nExpr-1);
    sqlite3VdbeAddOp3(v, OP_Copy, regPeer, s.end.reg, pOrderBy->nExpr-1);
  }

  sqlite3VdbeAddOp2(v, OP_Integer, 0, pMWin->regFirst);
  addrGoto = sqlite3VdbeAddOp0(v, OP_Goto);

  /* Begin generating SECOND_ROW_CODE */
  VdbeModuleComment((pParse->pVdbe, "Begin WindowCodeStep.SECOND_ROW"));
  if( bCache ){
    addrCacheNext = sqlite3VdbeCurrentAddr(v);
    if( pMWin->eType!=TK_ROWS ){
      windowReadPeerValues(&s, csrWrite, regNewPeer);
    }
  }else{
    sqlite3VdbeJumpHere(v, addrIfNot);
  }
  if( regPeer ){
    addrPeerJump = windowIfNewPeer(pParse, pOrderBy, regNewPeer, regPeer);
  }
  if( pMWin->eStart==TK_FOLLOWING ){
    windowCodeOp(&s, WINDOW_AGGSTEP, 0, 0);
    if( pMWin->eEnd!=TK_UNBOUNDED ){
      if( pMWin->eType==TK_RANGE ){

  if( addrPeerJump ){
    sqlite3VdbeJumpHere(v, addrPeerJump);
  }
  VdbeModuleComment((pParse->pVdbe, "End WindowCodeStep.SECOND_ROW"));

  /* End of the main input loop */
  sqlite3VdbeJumpHere(v, addrGoto);
  if( bCache ){
    sqlite3VdbeAddOp2(v, OP_Next, csrWrite, addrCacheNext);
    sqlite3VdbeJumpHere(v, addrCacheRewind); 
  }else{
    if( addrShortcut>0 ) sqlite3VdbeJumpHere(v, addrShortcut);
    sqlite3WhereEnd(pWInfo);
  }

  /* Fall through */
  if( pMWin->pPartition && bCache==0 ){
    addrInteger = sqlite3VdbeAddOp2(v, OP_Integer, 0, regFlushPart);
    sqlite3VdbeJumpHere(v, addrGosubFlush);
  }

  VdbeModuleComment((pParse->pVdbe, "Begin WindowCodeStep.FLUSH"));
  addrEmpty = sqlite3VdbeAddOp1(v, OP_Rewind, csrWrite);
  if( pMWin->eEnd==TK_PRECEDING ){

    windowCodeOp(&s, WINDOW_AGGINVERSE, regStart, 0);
    sqlite3VdbeAddOp2(v, OP_Goto, 0, addrStart);
    sqlite3VdbeJumpHere(v, addrBreak);
  }

  sqlite3VdbeJumpHere(v, addrEmpty);

  if( bCache && addrShortcut>0 ) sqlite3VdbeJumpHere(v, addrShortcut);
  sqlite3VdbeAddOp1(v, OP_ResetSorter, s.current.csr);
  sqlite3VdbeAddOp2(v, OP_Integer, 0, pMWin->regSize);
  if( bCache==0 ) sqlite3VdbeAddOp2(v, OP_Integer, 1, pMWin->regFirst);
  VdbeModuleComment((pParse->pVdbe, "End WindowCodeStep.FLUSH"));
  if( pMWin->pPartition ){
    sqlite3VdbeChangeP1(v, addrInteger, sqlite3VdbeCurrentAddr(v));
    sqlite3VdbeAddOp1(v, OP_Return, regFlushPart);
  }
}

#endif /* SQLITE_OMIT_WINDOWFUNC */