Rework the fix to ticket #461 so that we do not have to do redundant tests

of WHERE clause terms looking for NULLs.  See also check-in (1103). (CVS 1167)

FossilOrigin-Name: 5fd581787e88173f0303f870fc956ec9285cca4e

2 files changed, 84 insertions, 25 deletions

diff --git a/src/vdbe.c b/src/vdbe.c
index 667591e22..8e01614dd 100644
--- a/src/vdbe.c
+++ b/src/vdbe.c
@@ -43,7 +43,7 @@
 ** in this file for details.  If in doubt, do not deviate from existing
 ** commenting and indentation practices when changing or adding code.
 **
-** $Id: vdbe.c,v 1.248 2004/01/07 19:24:48 drh Exp $
+** $Id: vdbe.c,v 1.249 2004/01/07 20:37:52 drh Exp $
 */
 #include "sqliteInt.h"
 #include "os.h"
@@ -1797,8 +1797,8 @@ case OP_IfNot: {
 /* Opcode: IsNull P1 P2 *
 **
 ** If any of the top abs(P1) values on the stack are NULL, then jump
-** to P2.  The stack is popped P1 times if P1>0.  If P1<0 then all values
-** are left unchanged on the stack.
+** to P2.  Pop the stack P1 times if P1>0.   If P1<0 leave the stack
+** unchanged.
 */
 case OP_IsNull: {
   int i, cnt;
@@ -1817,14 +1817,18 @@ case OP_IsNull: {
 
 /* Opcode: NotNull P1 P2 *
 **
-** Jump to P2 if the top value on the stack is not NULL.  Pop the
-** stack if P1 is greater than zero.  If P1 is less than or equal to
-** zero then leave the value on the stack.
+** Jump to P2 if the top P1 values on the stack are all not NULL.  Pop the
+** stack if P1 times if P1 is greater than zero.  If P1 is less than
+** zero then leave the stack unchanged.
 */
 case OP_NotNull: {
-  VERIFY( if( p->tos<0 ) goto not_enough_stack; )
-  if( (aStack[p->tos].flags & STK_Null)==0 ) pc = pOp->p2-1;
-  if( pOp->p1>0 ){ POPSTACK; }
+  int i, cnt;
+  cnt = pOp->p1;
+  if( cnt<0 ) cnt = -cnt;
+  VERIFY( if( p->tos+1-cnt<0 ) goto not_enough_stack; )
+  for(i=0; i<cnt && (aStack[p->tos-i].flags & STK_Null)==0; i++){}
+  if( i>=cnt ) pc = pOp->p2-1;
+  if( pOp->p1>0 ) sqliteVdbePopStack(p, cnt);
   break;
 }
 
@@ -3027,6 +3031,16 @@ case OP_KeyAsData: {
 ** If the cursor is not pointing to a valid row, a NULL is pushed
 ** onto the stack.
 */
+/* Opcode: RowKey P1 * *
+**
+** Push onto the stack the complete row key for cursor P1.
+** There is no interpretation of the key.  It is just copied
+** onto the stack exactly as it is found in the database file.
+**
+** If the cursor is not pointing to a valid row, a NULL is pushed
+** onto the stack.
+*/
+case OP_RowKey:
 case OP_RowData: {
   int i = pOp->p1;
   int tos = ++p->tos;
@@ -3043,7 +3057,7 @@ case OP_RowData: {
     if( pC->nullRow ){
       aStack[tos].flags = STK_Null;
       break;
-    }else if( pC->keyAsData ){
+    }else if( pC->keyAsData || pOp->opcode==OP_RowKey ){
       sqliteBtreeKeySize(pCrsr, &n);
     }else{
       sqliteBtreeDataSize(pCrsr, &n);
@@ -3058,7 +3072,7 @@ case OP_RowData: {
       aStack[tos].flags = STK_Str | STK_Dyn;
       zStack[tos] = z;
     }
-    if( pC->keyAsData ){
+    if( pC->keyAsData || pOp->opcode==OP_RowKey ){
       sqliteBtreeKey(pCrsr, 0, n, zStack[tos]);
     }else{
       sqliteBtreeData(pCrsr, 0, n, zStack[tos]);
@@ -3557,6 +3571,37 @@ case OP_IdxGE: {
   break;
 }
 
+/* Opcode: IdxIsNull P1 P2 *
+**
+** The top of the stack contains an index entry such as might be generated
+** by the MakeIdxKey opcode.  This routine looks at the first P1 fields of
+** that key.  If any of the first P1 fields are NULL, then a jump is made
+** to address P2.  Otherwise we fall straight through.
+**
+** The index entry is always popped from the stack.
+*/
+case OP_IdxIsNull: {
+  int i = pOp->p1;
+  int tos = p->tos;
+  int k, n;
+  const char *z;
+
+  assert( tos>=0 );
+  assert( aStack[tos].flags & STK_Str );
+  z = zStack[tos];
+  n = aStack[tos].n;
+  for(k=0; k<n && i>0; i--){
+    if( z[k]=='a' ){
+      pc = pOp->p2-1;
+      break;
+    }
+    while( k<n && z[k] ){ k++; }
+    k++;
+  }
+  POPSTACK;
+  break;
+}
+
 /* Opcode: Destroy P1 P2 *
 **
 ** Delete an entire database table or index whose root page in the database
diff --git a/src/where.c b/src/where.c
index e05783b25..b93e2ceb2 100644
--- a/src/where.c
+++ b/src/where.c
@@ -12,7 +12,7 @@
 ** This module contains C code that generates VDBE code used to process
 ** the WHERE clause of SQL statements.
 **
-** $Id: where.c,v 1.85 2003/12/10 01:31:21 drh Exp $
+** $Id: where.c,v 1.86 2004/01/07 20:37:52 drh Exp $
 */
 #include "sqliteInt.h"
 
@@ -764,7 +764,7 @@ WhereInfo *sqliteWhereBegin(
           ){
             if( pX->op==TK_EQ ){
               sqliteExprCode(pParse, pX->pRight);
-              /* aExpr[k].p = 0; // See ticket #461 */
+              aExpr[k].p = 0;
               break;
             }
             if( pX->op==TK_IN && nColumn==1 ){
@@ -781,7 +781,7 @@ WhereInfo *sqliteWhereBegin(
                 pLevel->inOp = OP_Next;
                 pLevel->inP1 = pX->iTable;
               }
-              /* aExpr[k].p = 0; // See ticket #461 */
+              aExpr[k].p = 0;
               break;
             }
           }
@@ -791,13 +791,16 @@ WhereInfo *sqliteWhereBegin(
              && aExpr[k].p->pRight->iColumn==pIdx->aiColumn[j]
           ){
             sqliteExprCode(pParse, aExpr[k].p->pLeft);
-            /* aExpr[k].p = 0; // See ticket #461 */
+            aExpr[k].p = 0;
             break;
           }
         }
       }
       pLevel->iMem = pParse->nMem++;
       cont = pLevel->cont = sqliteVdbeMakeLabel(v);
+      sqliteVdbeAddOp(v, OP_NotNull, -nColumn, sqliteVdbeCurrentAddr(v)+3);
+      sqliteVdbeAddOp(v, OP_Pop, nColumn, 0);
+      sqliteVdbeAddOp(v, OP_Goto, 0, brk);
       sqliteVdbeAddOp(v, OP_MakeKey, nColumn, 0);
       sqliteAddIdxKeyType(v, pIdx);
       if( nColumn==pIdx->nColumn || pLevel->bRev ){
@@ -815,16 +818,17 @@ WhereInfo *sqliteWhereBegin(
         sqliteVdbeAddOp(v, OP_MoveLt, pLevel->iCur, brk);
         start = sqliteVdbeAddOp(v, OP_MemLoad, pLevel->iMem, 0);
         sqliteVdbeAddOp(v, OP_IdxLT, pLevel->iCur, brk);
-        sqliteVdbeAddOp(v, OP_IdxRecno, pLevel->iCur, 0);
         pLevel->op = OP_Prev;
       }else{
         /* Scan in the forward order */
         sqliteVdbeAddOp(v, OP_MoveTo, pLevel->iCur, brk);
         start = sqliteVdbeAddOp(v, OP_MemLoad, pLevel->iMem, 0);
         sqliteVdbeAddOp(v, testOp, pLevel->iCur, brk);
-        sqliteVdbeAddOp(v, OP_IdxRecno, pLevel->iCur, 0);
         pLevel->op = OP_Next;
       }
+      sqliteVdbeAddOp(v, OP_RowKey, pLevel->iCur, 0);
+      sqliteVdbeAddOp(v, OP_IdxIsNull, nColumn, cont);
+      sqliteVdbeAddOp(v, OP_IdxRecno, pLevel->iCur, 0);
       if( i==pTabList->nSrc-1 && pushKey ){
         haveKey = 1;
       }else{
@@ -933,7 +937,7 @@ WhereInfo *sqliteWhereBegin(
              && aExpr[k].p->pLeft->iColumn==pIdx->aiColumn[j]
           ){
             sqliteExprCode(pParse, aExpr[k].p->pRight);
-            /* aExpr[k].p = 0; // See ticket #461 */
+            aExpr[k].p = 0;
             break;
           }
           if( aExpr[k].idxRight==iCur
@@ -942,7 +946,7 @@ WhereInfo *sqliteWhereBegin(
              && aExpr[k].p->pRight->iColumn==pIdx->aiColumn[j]
           ){
             sqliteExprCode(pParse, aExpr[k].p->pLeft);
-            /* aExpr[k].p = 0; // See ticket #461 */
+            aExpr[k].p = 0;
             break;
           }
         }
@@ -979,7 +983,7 @@ WhereInfo *sqliteWhereBegin(
           ){
             sqliteExprCode(pParse, pExpr->pRight);
             leFlag = pExpr->op==TK_LE;
-            /* aExpr[k].p = 0; // See ticket #461 */
+            aExpr[k].p = 0;
             break;
           }
           if( aExpr[k].idxRight==iCur
@@ -989,7 +993,7 @@ WhereInfo *sqliteWhereBegin(
           ){
             sqliteExprCode(pParse, pExpr->pLeft);
             leFlag = pExpr->op==TK_GE;
-            /* aExpr[k].p = 0; // See ticket #461 */
+            aExpr[k].p = 0;
             break;
           }
         }
@@ -999,8 +1003,12 @@ WhereInfo *sqliteWhereBegin(
         leFlag = 1;
       }
       if( testOp!=OP_Noop ){
+        int nCol = nEqColumn + (score & 1);
         pLevel->iMem = pParse->nMem++;
-        sqliteVdbeAddOp(v, OP_MakeKey, nEqColumn + (score & 1), 0);
+        sqliteVdbeAddOp(v, OP_NotNull, -nCol, sqliteVdbeCurrentAddr(v)+3);
+        sqliteVdbeAddOp(v, OP_Pop, nCol, 0);
+        sqliteVdbeAddOp(v, OP_Goto, 0, brk);
+        sqliteVdbeAddOp(v, OP_MakeKey, nCol, 0);
         sqliteAddIdxKeyType(v, pIdx);
         if( leFlag ){
           sqliteVdbeAddOp(v, OP_IncrKey, 0, 0);
@@ -1034,7 +1042,7 @@ WhereInfo *sqliteWhereBegin(
           ){
             sqliteExprCode(pParse, pExpr->pRight);
             geFlag = pExpr->op==TK_GE;
-            /* aExpr[k].p = 0; // See ticket #461 */
+            aExpr[k].p = 0;
             break;
           }
           if( aExpr[k].idxRight==iCur
@@ -1044,7 +1052,7 @@ WhereInfo *sqliteWhereBegin(
           ){
             sqliteExprCode(pParse, pExpr->pLeft);
             geFlag = pExpr->op==TK_LE;
-            /* aExpr[k].p = 0; // See ticket #461 */
+            aExpr[k].p = 0;
             break;
           }
         }
@@ -1052,7 +1060,11 @@ WhereInfo *sqliteWhereBegin(
         geFlag = 1;
       }
       if( nEqColumn>0 || (score&2)!=0 ){
-        sqliteVdbeAddOp(v, OP_MakeKey, nEqColumn + ((score&2)!=0), 0);
+        int nCol = nEqColumn + ((score&2)!=0);
+        sqliteVdbeAddOp(v, OP_NotNull, -nCol, sqliteVdbeCurrentAddr(v)+3);
+        sqliteVdbeAddOp(v, OP_Pop, nCol, 0);
+        sqliteVdbeAddOp(v, OP_Goto, 0, brk);
+        sqliteVdbeAddOp(v, OP_MakeKey, nCol, 0);
         sqliteAddIdxKeyType(v, pIdx);
         if( !geFlag ){
           sqliteVdbeAddOp(v, OP_IncrKey, 0, 0);
@@ -1079,6 +1091,8 @@ WhereInfo *sqliteWhereBegin(
         sqliteVdbeAddOp(v, OP_MemLoad, pLevel->iMem, 0);
         sqliteVdbeAddOp(v, testOp, pLevel->iCur, brk);
       }
+      sqliteVdbeAddOp(v, OP_RowKey, pLevel->iCur, 0);
+      sqliteVdbeAddOp(v, OP_IdxIsNull, nEqColumn + (score & 1), cont);
       sqliteVdbeAddOp(v, OP_IdxRecno, pLevel->iCur, 0);
       if( i==pTabList->nSrc-1 && pushKey ){
         haveKey = 1;