SQLite

Check-in [1945484e6b]
Login

Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Overview
Comment:Add support for LIMIT and OFFSET in a recursive query.
Downloads: Tarball | ZIP archive
Timelines: family | ancestors | descendants | both | cte-via-queue
Files: files | file ages | folders
SHA1: 1945484e6b9769c1943f750f5b09860417fb190a
User & Date: drh 2014-01-22 18:07:04.473
Context
2014-01-22
18:16
Change the WITH RECURSIVE implementation to use a queue instead of a pair of tables. Add support for ORDER BY, LIMIT, and OFFSET on recursive queries. (check-in: b6cea42006 user: drh tags: trunk)
18:07
Add support for LIMIT and OFFSET in a recursive query. (Closed-Leaf check-in: 1945484e6b user: drh tags: cte-via-queue)
17:28
Get ORDER BY working for recursive queries. (check-in: 37b343b018 user: drh tags: cte-via-queue)
Changes
Unified Diff Ignore Whitespace Patch
Changes to src/select.c.
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
}

/*
** Add code to implement the OFFSET
*/
static void codeOffset(
  Vdbe *v,          /* Generate code into this VM */
  Select *p,        /* The SELECT statement being coded */
  int iContinue     /* Jump here to skip the current record */
){
  if( p->iOffset && iContinue!=0 ){
    int addr;
    sqlite3VdbeAddOp2(v, OP_AddImm, p->iOffset, -1);
    addr = sqlite3VdbeAddOp1(v, OP_IfNeg, p->iOffset);
    sqlite3VdbeAddOp2(v, OP_Goto, 0, iContinue);
    VdbeComment((v, "skip OFFSET records"));
    sqlite3VdbeJumpHere(v, addr);
  }
}

/*







|


|

|
|







458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
}

/*
** Add code to implement the OFFSET
*/
static void codeOffset(
  Vdbe *v,          /* Generate code into this VM */
  int iOffset,      /* Register holding the offset counter */
  int iContinue     /* Jump here to skip the current record */
){
  if( iOffset>0 && iContinue!=0 ){
    int addr;
    sqlite3VdbeAddOp2(v, OP_AddImm, iOffset, -1);
    addr = sqlite3VdbeAddOp1(v, OP_IfNeg, iOffset);
    sqlite3VdbeAddOp2(v, OP_Goto, 0, iContinue);
    VdbeComment((v, "skip OFFSET records"));
    sqlite3VdbeJumpHere(v, addr);
  }
}

/*
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
  int iParm = pDest->iSDParm; /* First argument to disposal method */
  int nResultCol;             /* Number of result columns */

  assert( v );
  assert( pEList!=0 );
  hasDistinct = pDistinct ? pDistinct->eTnctType : WHERE_DISTINCT_NOOP;
  if( pOrderBy==0 && !hasDistinct ){
    codeOffset(v, p, iContinue);
  }

  /* Pull the requested columns.
  */
  nResultCol = pEList->nExpr;
  if( pDest->iSdst==0 ){
    pDest->iSdst = pParse->nMem+1;







|







567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
  int iParm = pDest->iSDParm; /* First argument to disposal method */
  int nResultCol;             /* Number of result columns */

  assert( v );
  assert( pEList!=0 );
  hasDistinct = pDistinct ? pDistinct->eTnctType : WHERE_DISTINCT_NOOP;
  if( pOrderBy==0 && !hasDistinct ){
    codeOffset(v, p->iOffset, iContinue);
  }

  /* Pull the requested columns.
  */
  nResultCol = pEList->nExpr;
  if( pDest->iSdst==0 ){
    pDest->iSdst = pParse->nMem+1;
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
      default: {
        assert( pDistinct->eTnctType==WHERE_DISTINCT_UNORDERED );
        codeDistinct(pParse, pDistinct->tabTnct, iContinue, nResultCol, regResult);
        break;
      }
    }
    if( pOrderBy==0 ){
      codeOffset(v, p, iContinue);
    }
  }

  switch( eDest ){
    /* In this mode, write each query result to the key of the temporary
    ** table iParm.
    */







|







649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
      default: {
        assert( pDistinct->eTnctType==WHERE_DISTINCT_UNORDERED );
        codeDistinct(pParse, pDistinct->tabTnct, iContinue, nResultCol, regResult);
        break;
      }
    }
    if( pOrderBy==0 ){
      codeOffset(v, p->iOffset, iContinue);
    }
  }

  switch( eDest ){
    /* In this mode, write each query result to the key of the temporary
    ** table iParm.
    */
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
    regRowid = sqlite3GetTempReg(pParse);
  }
  if( p->selFlags & SF_UseSorter ){
    int regSortOut = ++pParse->nMem;
    int ptab2 = pParse->nTab++;
    sqlite3VdbeAddOp3(v, OP_OpenPseudo, ptab2, regSortOut, pOrderBy->nExpr+2);
    addr = 1 + sqlite3VdbeAddOp2(v, OP_SorterSort, iTab, addrBreak);
    codeOffset(v, p, addrContinue);
    sqlite3VdbeAddOp2(v, OP_SorterData, iTab, regSortOut);
    sqlite3VdbeAddOp3(v, OP_Column, ptab2, pOrderBy->nExpr+1, regRow);
    sqlite3VdbeChangeP5(v, OPFLAG_CLEARCACHE);
  }else{
    addr = 1 + sqlite3VdbeAddOp2(v, OP_Sort, iTab, addrBreak);
    codeOffset(v, p, addrContinue);
    sqlite3VdbeAddOp3(v, OP_Column, iTab, pOrderBy->nExpr+1, regRow);
  }
  switch( eDest ){
    case SRT_Table:
    case SRT_EphemTab: {
      testcase( eDest==SRT_Table );
      testcase( eDest==SRT_EphemTab );







|





|







1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
    regRowid = sqlite3GetTempReg(pParse);
  }
  if( p->selFlags & SF_UseSorter ){
    int regSortOut = ++pParse->nMem;
    int ptab2 = pParse->nTab++;
    sqlite3VdbeAddOp3(v, OP_OpenPseudo, ptab2, regSortOut, pOrderBy->nExpr+2);
    addr = 1 + sqlite3VdbeAddOp2(v, OP_SorterSort, iTab, addrBreak);
    codeOffset(v, p->iOffset, addrContinue);
    sqlite3VdbeAddOp2(v, OP_SorterData, iTab, regSortOut);
    sqlite3VdbeAddOp3(v, OP_Column, ptab2, pOrderBy->nExpr+1, regRow);
    sqlite3VdbeChangeP5(v, OPFLAG_CLEARCACHE);
  }else{
    addr = 1 + sqlite3VdbeAddOp2(v, OP_Sort, iTab, addrBreak);
    codeOffset(v, p->iOffset, addrContinue);
    sqlite3VdbeAddOp3(v, OP_Column, iTab, pOrderBy->nExpr+1, regRow);
  }
  switch( eDest ){
    case SRT_Table:
    case SRT_EphemTab: {
      testcase( eDest==SRT_Table );
      testcase( eDest==SRT_EphemTab );
1635
1636
1637
1638
1639
1640
1641
1642
1643





1644
1645
1646
1647
1648
1649
1650
** keywords.  Or NULL if those keywords are omitted. iLimit and iOffset 
** are the integer memory register numbers for counters used to compute 
** the limit and offset.  If there is no limit and/or offset, then 
** iLimit and iOffset are negative.
**
** This routine changes the values of iLimit and iOffset only if
** a limit or offset is defined by pLimit and pOffset.  iLimit and
** iOffset should have been preset to appropriate default values
** (usually but not always -1) prior to calling this routine.





** Only if pLimit!=0 or pOffset!=0 do the limit registers get
** redefined.  The UNION ALL operator uses this property to force
** the reuse of the same limit and offset registers across multiple
** SELECT statements.
*/
static void computeLimitRegisters(Parse *pParse, Select *p, int iBreak){
  Vdbe *v = 0;







|
|
>
>
>
>
>







1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
** keywords.  Or NULL if those keywords are omitted. iLimit and iOffset 
** are the integer memory register numbers for counters used to compute 
** the limit and offset.  If there is no limit and/or offset, then 
** iLimit and iOffset are negative.
**
** This routine changes the values of iLimit and iOffset only if
** a limit or offset is defined by pLimit and pOffset.  iLimit and
** iOffset should have been preset to appropriate default values (zero)
** prior to calling this routine.
**
** The iOffset register (if it exists) is initialized to the value
** of the OFFSET.  The iLimit register is initialized to LIMIT.  Register
** iOffset+1 is initialized to LIMIT+OFFSET.
**
** Only if pLimit!=0 or pOffset!=0 do the limit registers get
** redefined.  The UNION ALL operator uses this property to force
** the reuse of the same limit and offset registers across multiple
** SELECT statements.
*/
static void computeLimitRegisters(Parse *pParse, Select *p, int iBreak){
  Vdbe *v = 0;
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
  ** no rows.
  */
  sqlite3ExprCacheClear(pParse);
  assert( p->pOffset==0 || p->pLimit!=0 );
  if( p->pLimit ){
    p->iLimit = iLimit = ++pParse->nMem;
    v = sqlite3GetVdbe(pParse);
    if( NEVER(v==0) ) return;  /* VDBE should have already been allocated */
    if( sqlite3ExprIsInteger(p->pLimit, &n) ){
      sqlite3VdbeAddOp2(v, OP_Integer, n, iLimit);
      VdbeComment((v, "LIMIT counter"));
      if( n==0 ){
        sqlite3VdbeAddOp2(v, OP_Goto, 0, iBreak);
      }else if( n>=0 && p->nSelectRow>(u64)n ){
        p->nSelectRow = n;







|







1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
  ** no rows.
  */
  sqlite3ExprCacheClear(pParse);
  assert( p->pOffset==0 || p->pLimit!=0 );
  if( p->pLimit ){
    p->iLimit = iLimit = ++pParse->nMem;
    v = sqlite3GetVdbe(pParse);
    assert( v!=0 );
    if( sqlite3ExprIsInteger(p->pLimit, &n) ){
      sqlite3VdbeAddOp2(v, OP_Integer, n, iLimit);
      VdbeComment((v, "LIMIT counter"));
      if( n==0 ){
        sqlite3VdbeAddOp2(v, OP_Goto, 0, iBreak);
      }else if( n>=0 && p->nSelectRow>(u64)n ){
        p->nSelectRow = n;
1774
1775
1776
1777
1778
1779
1780


1781
1782
1783


1784
1785
1786
1787
1788

1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
  int iQueue;                   /* The Queue table */
  int iDistinct = 0;            /* To ensure unique results if UNION */
  int eDest = SRT_Table;        /* How to write to Queue */
  SelectDest destQueue;         /* SelectDest targetting the Queue table */
  int i;                        /* Loop counter */
  int rc;                       /* Result code */
  ExprList *pOrderBy;           /* The ORDER BY clause */



  /* Obtain authorization to do a recursive query */
  if( sqlite3AuthCheck(pParse, SQLITE_RECURSIVE, 0, 0, 0) ) return;


  addrBreak = sqlite3VdbeMakeLabel(v);

  /* Check that there is no ORDER BY or LIMIT clause. Neither of these 
  ** are currently supported on recursive queries.
  */

  assert( p->pOffset==0 || p->pLimit );
  if( /*p->pOrderBy ||*/ p->pLimit ){
    sqlite3ErrorMsg(pParse, "%s in a recursive query",
        p->pOrderBy ? "ORDER BY" : "LIMIT"
    );
    return;
  }

  /* Locate the cursor number of the Current table */
  for(i=0; ALWAYS(i<pSrc->nSrc); i++){
    if( pSrc->a[i].isRecursive ){
      iCurrent = pSrc->a[i].iCursor;
      break;
    }







>
>



>
>

|
<
<
<
>
|
|
|
|
<
|
<







1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794



1795
1796
1797
1798
1799

1800

1801
1802
1803
1804
1805
1806
1807
  int iQueue;                   /* The Queue table */
  int iDistinct = 0;            /* To ensure unique results if UNION */
  int eDest = SRT_Table;        /* How to write to Queue */
  SelectDest destQueue;         /* SelectDest targetting the Queue table */
  int i;                        /* Loop counter */
  int rc;                       /* Result code */
  ExprList *pOrderBy;           /* The ORDER BY clause */
  Expr *pLimit, *pOffset;       /* Saved LIMIT and OFFSET */
  int regLimit, regOffset;      /* Registers used by LIMIT and OFFSET */

  /* Obtain authorization to do a recursive query */
  if( sqlite3AuthCheck(pParse, SQLITE_RECURSIVE, 0, 0, 0) ) return;

  /* Process the LIMIT and OFFSET clauses, if they exist */
  addrBreak = sqlite3VdbeMakeLabel(v);
  computeLimitRegisters(pParse, p, addrBreak);



  pLimit = p->pLimit;
  pOffset = p->pOffset;
  regLimit = p->iLimit;
  regOffset = p->iOffset;
  p->pLimit = p->pOffset = 0;

  p->iLimit = p->iOffset = 0;


  /* Locate the cursor number of the Current table */
  for(i=0; ALWAYS(i<pSrc->nSrc); i++){
    if( pSrc->a[i].isRecursive ){
      iCurrent = pSrc->a[i].iCursor;
      break;
    }
1849
1850
1851
1852
1853
1854
1855

1856
1857

1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873


1874
1875
1876
1877
1878
1879
1880
  }else{
    sqlite3VdbeAddOp2(v, OP_RowData, iQueue, regCurrent);
  }
  sqlite3VdbeAddOp1(v, OP_Delete, iQueue);

  /* Output the single row in Current */
  addrCont = sqlite3VdbeMakeLabel(v);

  selectInnerLoop(pParse, p, p->pEList, iCurrent,
      0, 0, pDest, addrCont, addrBreak);

  sqlite3VdbeResolveLabel(v, addrCont);

  /* Execute the recursive SELECT taking the single row in Current as
  ** the value for the recursive-table. Store the results in the Queue.
  */
  p->pPrior = 0;
  sqlite3Select(pParse, p, &destQueue);
  assert( p->pPrior==0 );
  p->pPrior = pSetup;

  /* Keep running the loop until the Queue is empty */
  sqlite3VdbeAddOp2(v, OP_Goto, 0, addrTop);
  sqlite3VdbeResolveLabel(v, addrBreak);

end_of_recursive_query:
  p->pOrderBy = pOrderBy;


  return;
}
#endif

/* Forward references */
static int multiSelectOrderBy(
  Parse *pParse,        /* Parsing context */







>


>
















>
>







1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
  }else{
    sqlite3VdbeAddOp2(v, OP_RowData, iQueue, regCurrent);
  }
  sqlite3VdbeAddOp1(v, OP_Delete, iQueue);

  /* Output the single row in Current */
  addrCont = sqlite3VdbeMakeLabel(v);
  codeOffset(v, regOffset, addrCont);
  selectInnerLoop(pParse, p, p->pEList, iCurrent,
      0, 0, pDest, addrCont, addrBreak);
  if( regLimit ) sqlite3VdbeAddOp3(v, OP_IfZero, regLimit, addrBreak, -1);
  sqlite3VdbeResolveLabel(v, addrCont);

  /* Execute the recursive SELECT taking the single row in Current as
  ** the value for the recursive-table. Store the results in the Queue.
  */
  p->pPrior = 0;
  sqlite3Select(pParse, p, &destQueue);
  assert( p->pPrior==0 );
  p->pPrior = pSetup;

  /* Keep running the loop until the Queue is empty */
  sqlite3VdbeAddOp2(v, OP_Goto, 0, addrTop);
  sqlite3VdbeResolveLabel(v, addrBreak);

end_of_recursive_query:
  p->pOrderBy = pOrderBy;
  p->pLimit = pLimit;
  p->pOffset = pOffset;
  return;
}
#endif

/* Forward references */
static int multiSelectOrderBy(
  Parse *pParse,        /* Parsing context */
2322
2323
2324
2325
2326
2327
2328
2329
2330
2331
2332
2333
2334
2335
2336
    sqlite3VdbeAddOp3(v, OP_Copy, pIn->iSdst, regPrev+1, pIn->nSdst-1);
    sqlite3VdbeAddOp2(v, OP_Integer, 1, regPrev);
  }
  if( pParse->db->mallocFailed ) return 0;

  /* Suppress the first OFFSET entries if there is an OFFSET clause
  */
  codeOffset(v, p, iContinue);

  switch( pDest->eDest ){
    /* Store the result as data using a unique key.
    */
    case SRT_Table:
    case SRT_EphemTab: {
      int r1 = sqlite3GetTempReg(pParse);







|







2331
2332
2333
2334
2335
2336
2337
2338
2339
2340
2341
2342
2343
2344
2345
    sqlite3VdbeAddOp3(v, OP_Copy, pIn->iSdst, regPrev+1, pIn->nSdst-1);
    sqlite3VdbeAddOp2(v, OP_Integer, 1, regPrev);
  }
  if( pParse->db->mallocFailed ) return 0;

  /* Suppress the first OFFSET entries if there is an OFFSET clause
  */
  codeOffset(v, p->iOffset, iContinue);

  switch( pDest->eDest ){
    /* Store the result as data using a unique key.
    */
    case SRT_Table:
    case SRT_EphemTab: {
      int r1 = sqlite3GetTempReg(pParse);
Changes to test/with1.test.
186
187
188
189
190
191
192








193



194
195
196
197
198
199
200
201
202
203
204
       UNION ALL
       SELECT edge.xto, edge.seq FROM edge, ancest
        WHERE edge.xfrom=ancest.id
        ORDER BY 2
      )
  SELECT * FROM ancest;
} {0 0 1 10 2 20 3 30 4 40 4 40 5 50 6 60 7 70 7 70 8 80 8 80 8 80 8 80 9 90 9 90 9 90 9 90}












do_catchsql_test 5.3 {
  WITH i(x) AS ( VALUES(1) UNION ALL SELECT x+1 FROM i LIMIT 10 )
  SELECT x FROM i LIMIT 10;
} {1 {LIMIT in a recursive query}}

do_execsql_test 5.4 {
  WITH i(x) AS ( VALUES(1) UNION ALL SELECT (x+1)%10 FROM i)
  SELECT x FROM i LIMIT 20;
} {1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0}

do_execsql_test 5.5 {







>
>
>
>
>
>
>
>
|
>
>
>

|
|
|







186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
       UNION ALL
       SELECT edge.xto, edge.seq FROM edge, ancest
        WHERE edge.xfrom=ancest.id
        ORDER BY 2
      )
  SELECT * FROM ancest;
} {0 0 1 10 2 20 3 30 4 40 4 40 5 50 6 60 7 70 7 70 8 80 8 80 8 80 8 80 9 90 9 90 9 90 9 90}
do_execsql_test 5.2.3 {
  WITH RECURSIVE
    ancest(id, mtime) AS
      (VALUES(0, 0)
       UNION ALL
       SELECT edge.xto, edge.seq FROM edge, ancest
        WHERE edge.xfrom=ancest.id
        ORDER BY 2 LIMIT 4 OFFSET 2
      )
  SELECT * FROM ancest;
} {2 20 3 30 4 40 4 40}

do_catchsql_test 5.3 {
  WITH i(x) AS ( VALUES(1) UNION ALL SELECT x+1 FROM i LIMIT 5)
  SELECT x FROM i;
} {0 {1 2 3 4 5}}

do_execsql_test 5.4 {
  WITH i(x) AS ( VALUES(1) UNION ALL SELECT (x+1)%10 FROM i)
  SELECT x FROM i LIMIT 20;
} {1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0}

do_execsql_test 5.5 {