1
0
mirror of https://github.com/sqlite/sqlite.git synced 2025-08-07 02:42:48 +03:00

Add support for "ROWS BETWEEN <expr> PRECEDING AND <expr> FOLLOWING" window

frames.

FossilOrigin-Name: 3a203660f1e4da3b8d2d605c494f4843f6e00752f28042b49e11d7d6550dd406
This commit is contained in:
dan
2018-05-23 20:55:37 +00:00
parent 72779425b5
commit c3a20c19a8
11 changed files with 452 additions and 137 deletions

View File

@@ -25,8 +25,8 @@ void sqlite3WindowDelete(sqlite3 *db, Window *p){
Window *sqlite3WindowAlloc(
Parse *pParse,
int eType,
int eEnd, Expr *pEnd,
int eStart, Expr *pStart
int eStart, Expr *pStart,
int eEnd, Expr *pEnd
){
Window *pWin = (Window*)sqlite3DbMallocZero(pParse->db, sizeof(Window));
@@ -77,6 +77,279 @@ void sqlite3WindowCodeInit(Parse *pParse, Window *pWin){
}
}
static void windowCheckFrameValue(Parse *pParse, int reg, int bEnd){
static const char *azErr[] = {
"frame starting offset must be a non-negative integer",
"frame ending offset must be a non-negative integer"
};
Vdbe *v = sqlite3GetVdbe(pParse);
int regZero = ++pParse->nMem;
sqlite3VdbeAddOp2(v, OP_Integer, 0, regZero);
sqlite3VdbeAddOp2(v, OP_MustBeInt, reg, sqlite3VdbeCurrentAddr(v)+2);
sqlite3VdbeAddOp3(v, OP_Ge, regZero, sqlite3VdbeCurrentAddr(v)+2, reg);
sqlite3VdbeAddOp2(v, OP_Halt, SQLITE_ERROR, OE_Abort);
sqlite3VdbeAppendP4(v, (void*)azErr[bEnd], P4_STATIC);
}
static void windowCodeRowExprStep(
Parse *pParse,
Select *p,
WhereInfo *pWInfo,
int regGosub,
int addrGosub
){
Window *pMWin = p->pWin;
Vdbe *v = sqlite3GetVdbe(pParse);
Window *pWin;
int k;
int iSubCsr = p->pSrc->a[0].iCursor;
int nSub = p->pSrc->a[0].pTab->nCol;
int regFlushPart; /* Register for "Gosub flush_partition" */
int addrFlushPart; /* Label for "Gosub flush_partition" */
int addrDone; /* Label for "Gosub flush_partition_done" */
int reg = pParse->nMem+1;
int regRecord = reg+nSub;
int regRowid = regRecord+1;
int addr;
int csrPrec = pParse->nTab++;
int csrFollow = pParse->nTab++;
int regPrec; /* Value of <expr> PRECEDING */
int regFollow; /* Value of <expr> FOLLOWING */
int addrNext;
int addrGoto;
int addrIfPos1;
int addrIfPos2;
pParse->nMem += nSub + 2;
/* Allocate register and label for the "flush_partition" sub-routine. */
regFlushPart = ++pParse->nMem;
addrFlushPart = sqlite3VdbeMakeLabel(v);
addrDone = sqlite3VdbeMakeLabel(v);
regPrec = ++pParse->nMem;
regFollow = ++pParse->nMem;
/* Martial 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);
}
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 ){
ExprList *pPart = pMWin->pPartition;
int nPart = (pPart ? pPart->nExpr : 0);
int addrJump = 0;
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, addr+4, addr+2);
sqlite3VdbeAddOp2(v, OP_Gosub, regFlushPart, addrFlushPart);
sqlite3VdbeAddOp3(v, OP_Copy, regNewPart, pMWin->regPart, nPart);
}
/* 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, addrFlushPart);
addrGoto = sqlite3VdbeAddOp0(v, OP_Goto);
/* flush_partition: */
sqlite3VdbeResolveLabel(v, addrFlushPart);
sqlite3VdbeAddOp2(v, OP_Once, 0, sqlite3VdbeCurrentAddr(v)+3);
sqlite3VdbeAddOp2(v, OP_OpenDup, csrPrec, pMWin->iEphCsr);
sqlite3VdbeAddOp2(v, OP_OpenDup, csrFollow, pMWin->iEphCsr);
sqlite3ExprCode(pParse, pMWin->pStart, regPrec);
sqlite3ExprCode(pParse, pMWin->pEnd, regFollow);
sqlite3VdbeAddOp2(v, OP_Null, 0, pMWin->regResult);
sqlite3VdbeAddOp2(v, OP_Null, 0, pMWin->regAccum);
/* If either regPrec or regFollow are not non-negative integers, throw an
** exception. */
windowCheckFrameValue(pParse, regPrec, 0);
windowCheckFrameValue(pParse, regFollow, 1);
sqlite3VdbeAddOp2(v, OP_Rewind, pMWin->iEphCsr, addrDone);
sqlite3VdbeAddOp2(v, OP_Rewind, csrPrec, addrDone);
sqlite3VdbeChangeP5(v, 1);
sqlite3VdbeAddOp2(v, OP_Rewind, csrFollow, addrDone);
sqlite3VdbeChangeP5(v, 1);
/* Invoke AggStep function for each window function using the row that
** csrFollow currently points to. Or, if csrFollow is already at EOF,
** do nothing. */
addrNext = sqlite3VdbeCurrentAddr(v);
sqlite3VdbeAddOp2(v, OP_Next, csrFollow, addrNext+2);
sqlite3VdbeAddOp0(v, OP_Goto);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
int i;
for(i=0; i<pWin->nArg; i++){
sqlite3VdbeAddOp3(v, OP_Column, csrFollow, pWin->iArgCol+i, reg+i);
}
sqlite3VdbeAddOp3(v, OP_AggStep0, 0, reg, pWin->regAccum);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeChangeP5(v, (u8)pWin->nArg);
}
sqlite3VdbeJumpHere(v, addrNext+1);
addrIfPos1 = sqlite3VdbeAddOp3(v, OP_IfPos, regFollow, 0 , 1);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp3(v,
OP_AggFinal, pWin->regAccum, pWin->nArg, pWin->regResult
);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
}
sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
sqlite3VdbeAddOp2(v, OP_Next, pMWin->iEphCsr, sqlite3VdbeCurrentAddr(v)+2);
sqlite3VdbeAddOp2(v, OP_Goto, 0, addrDone);
addrIfPos2 = sqlite3VdbeAddOp3(v, OP_IfPos, regPrec, 0 , 1);
sqlite3VdbeAddOp2(v, OP_Next, csrPrec, sqlite3VdbeCurrentAddr(v)+1);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
int i;
for(i=0; i<pWin->nArg; i++){
sqlite3VdbeAddOp3(v, OP_Column, csrPrec, pWin->iArgCol+i, reg+i);
}
sqlite3VdbeAddOp3(v, OP_AggStep0, 1, reg, pWin->regAccum);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeChangeP5(v, (u8)pWin->nArg);
}
sqlite3VdbeJumpHere(v, addrIfPos2);
sqlite3VdbeJumpHere(v, addrIfPos1);
sqlite3VdbeAddOp2(v, OP_Goto, 0, addrNext);
/* flush_partition_done: */
sqlite3VdbeResolveLabel(v, addrDone);
sqlite3VdbeAddOp1(v, OP_ResetSorter, pMWin->iEphCsr);
sqlite3VdbeAddOp1(v, OP_Return, regFlushPart);
/* Jump to here to skip over flush_partition */
sqlite3VdbeJumpHere(v, addrGoto);
}
static void windowCodeDefaultStep(
Parse *pParse,
Select *p,
WhereInfo *pWInfo,
int regGosub,
int addrGosub
){
Window *pMWin = p->pWin;
Vdbe *v = sqlite3GetVdbe(pParse);
Window *pWin;
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;
pParse->nMem += nSub + 2;
/* Martial 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( pMWin->regPart ){
ExprList *pPart = pMWin->pPartition;
int nPart = (pPart ? pPart->nExpr : 0);
ExprList *pOrderBy = pMWin->pOrderBy;
int nPeer = (pOrderBy ? pOrderBy->nExpr : 0);
int addrGoto = 0;
int addrJump = 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);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp2(v, OP_AggFinal, pWin->regAccum, pWin->nArg);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeAddOp2(v, OP_Copy, pWin->regAccum, pWin->regResult);
}
if( pOrderBy ){
addrGoto = sqlite3VdbeAddOp0(v, OP_Goto);
}
}
if( pOrderBy ){
int regNewPeer = reg + pMWin->nBufferCol + nPart;
int regPeer = pMWin->regPart + nPart;
KeyInfo *pKeyInfo = sqlite3KeyInfoFromExprList(pParse, pOrderBy, 0, 0);
if( addrJump ) sqlite3VdbeJumpHere(v, addrJump);
addr = sqlite3VdbeAddOp3(v, OP_Compare, regNewPeer, regPeer, nPeer);
sqlite3VdbeAppendP4(v, (void*)pKeyInfo, P4_KEYINFO);
addrJump = sqlite3VdbeAddOp3(v, OP_Jump, addr+2, 0, addr+2);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp3(v,
OP_AggFinal, pWin->regAccum, pWin->nArg, pWin->regResult
);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
}
if( addrGoto ) sqlite3VdbeJumpHere(v, addrGoto);
}
sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
sqlite3VdbeAddOp1(v, OP_ResetSorter, pMWin->iEphCsr);
sqlite3VdbeAddOp3(
v, OP_Copy, reg+pMWin->nBufferCol, pMWin->regPart, nPart+nPeer-1
);
sqlite3VdbeJumpHere(v, addrJump);
}
/* Invoke step function for window functions */
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp3(v, OP_AggStep0, 0, reg+pWin->iArgCol, pWin->regAccum);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeChangeP5(v, (u8)pWin->nArg);
}
/* 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);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp2(v, OP_AggFinal, pWin->regAccum, pWin->nArg);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeAddOp2(v, OP_Copy, pWin->regAccum, pWin->regResult);
}
sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
}
/*
** RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
**
@@ -243,105 +516,22 @@ void sqlite3WindowCodeStep(
Select *p,
WhereInfo *pWInfo,
int regGosub,
int addrGosub
int addrGosub,
int *pbLoop
){
Vdbe *v = sqlite3GetVdbe(pParse);
Window *pWin;
Window *pMWin = p->pWin;
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;
pParse->nMem += nSub + 2;
/* Martial 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);
if( pMWin->eType==TK_ROWS
&& pMWin->eStart==TK_PRECEDING
&& pMWin->eEnd==TK_FOLLOWING
){
*pbLoop = 0;
windowCodeRowExprStep(pParse, p, pWInfo, regGosub, addrGosub);
return;
}
/* Check if this is the start of a new partition or peer group. */
if( pMWin->regPart ){
ExprList *pPart = pMWin->pPartition;
int nPart = (pPart ? pPart->nExpr : 0);
ExprList *pOrderBy = pMWin->pOrderBy;
int nPeer = (pOrderBy ? pOrderBy->nExpr : 0);
int addrGoto = 0;
int addrJump = 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);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp2(v, OP_AggFinal, pWin->regAccum, pWin->nArg);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeAddOp2(v, OP_Copy, pWin->regAccum, pWin->regResult);
}
if( pOrderBy ){
addrGoto = sqlite3VdbeAddOp0(v, OP_Goto);
}
}
if( pOrderBy ){
int regNewPeer = reg + pMWin->nBufferCol + nPart;
int regPeer = pMWin->regPart + nPart;
KeyInfo *pKeyInfo = sqlite3KeyInfoFromExprList(pParse, pOrderBy, 0, 0);
if( addrJump ) sqlite3VdbeJumpHere(v, addrJump);
addr = sqlite3VdbeAddOp3(v, OP_Compare, regNewPeer, regPeer, nPeer);
sqlite3VdbeAppendP4(v, (void*)pKeyInfo, P4_KEYINFO);
addrJump = sqlite3VdbeAddOp3(v, OP_Jump, addr+2, 0, addr+2);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp3(v,
OP_AggFinal, pWin->regAccum, pWin->nArg, pWin->regResult
);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
}
if( addrGoto ) sqlite3VdbeJumpHere(v, addrGoto);
}
sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
sqlite3VdbeAddOp1(v, OP_ResetSorter, pMWin->iEphCsr);
sqlite3VdbeAddOp3(
v, OP_Copy, reg+pMWin->nBufferCol, pMWin->regPart, nPart+nPeer-1
);
sqlite3VdbeJumpHere(v, addrJump);
}
/* Invoke step function for window functions */
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp3(v, OP_AggStep0, 0, reg+pWin->iArgCol, pWin->regAccum);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeChangeP5(v, (u8)pWin->nArg);
}
/* 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);
for(pWin=pMWin; pWin; pWin=pWin->pNextWin){
sqlite3VdbeAddOp2(v, OP_AggFinal, pWin->regAccum, pWin->nArg);
sqlite3VdbeAppendP4(v, pWin->pFunc, P4_FUNCDEF);
sqlite3VdbeAddOp2(v, OP_Copy, pWin->regAccum, pWin->regResult);
}
sqlite3VdbeAddOp2(v, OP_Gosub, regGosub, addrGosub);
*pbLoop = 1;
windowCodeDefaultStep(pParse, p, pWInfo, regGosub, addrGosub);
}