Bug fixes in the INSERT constraint checker. Refactor the Rowid handling logic

for ANALYZE with STAT3/4.

FossilOrigin-Name: 1ea43c0f236792a3bc13e1cb330f5ff3402c2851

3 files changed, 92 insertions, 64 deletions

diff --git a/src/analyze.c b/src/analyze.c
index a71231eb5..0e5869427 100644
--- a/src/analyze.c
+++ b/src/analyze.c
@@ -268,9 +268,11 @@ struct Stat4Sample {
   tRowcnt *anDLt;                 /* sqlite_stat4.nDLt */
 #ifdef SQLITE_ENABLE_STAT3_OR_STAT4
   tRowcnt *anLt;                  /* sqlite_stat4.nLt */
-  i64 iRowid;                     /* Rowid in main table of the key */
+  union {
+    i64 iRowid;                     /* Rowid in main table of the key */
+    u8 *aRowid;                     /* Key for WITHOUT ROWID tables */
+  } u;
   u32 nRowid;                     /* Sizeof aRowid[] */
-  u8 *aRowid;                     /* Key for WITHOUT ROWID tables */
   u8 isPSample;                   /* True if a periodic sample */
   int iCol;                       /* If !isPSample, the reason for inclusion */
   u32 iHash;                      /* Tiebreaker hash */
@@ -295,22 +297,57 @@ struct Stat4Accum {
 */
 #ifdef SQLITE_ENABLE_STAT3_OR_STAT4
 static void sampleClear(sqlite3 *db, Stat4Sample *p){
-  sqlite3DbFree(db, p->aRowid);
-  p->aRowid = 0;
+  assert( db!=0 );
+  if( p->nRowid ){
+    sqlite3DbFree(db, p->u.aRowid);
+    p->nRowid = 0;
+  }
+}
+#endif
+
+/* Initialize the BLOB value of a ROWID
+*/
+#ifdef SQLITE_ENABLE_STAT3_OR_STAT4
+static void sampleSetRowid(sqlite3 *db, Stat4Sample *p, int n, const u8 *pData){
+  assert( db!=0 );
+  if( p->nRowid ) sqlite3DbFree(db, p->u.aRowid);
+  p->u.aRowid = sqlite3DbMallocRaw(db, n);
+  if( p->u.aRowid ){
+    p->nRowid = n;
+    memcpy(p->u.aRowid, pData, n);
+  }else{
+    p->nRowid = 0;
+  }
+}
+#endif
+
+/* Initialize the INTEGER value of a ROWID.
+*/
+#ifdef SQLITE_ENABLE_STAT3_OR_STAT4
+static void sampleSetRowidInt64(sqlite3 *db, Stat4Sample *p, i64 iRowid){
+  assert( db!=0 );
+  if( p->nRowid ) sqlite3DbFree(db, p->u.aRowid);
   p->nRowid = 0;
+  p->u.iRowid = iRowid;
 }
 #endif
 
-/* Make a copy of the Stat4Sample.aRowid field.
+
+/*
+** Copy the contents of object (*pFrom) into (*pTo).
 */
 #ifdef SQLITE_ENABLE_STAT3_OR_STAT4
-static void sampleDup(sqlite3 *db, Stat4Sample *p){
-  if( p->aRowid ){
-    u8 *aCopy = sqlite3DbMallocRaw(db, p->nRowid);
-    if( aCopy ){
-      memcpy(aCopy, p->aRowid, p->nRowid);
-      p->aRowid = aCopy;
-    }
+static void sampleCopy(Stat4Accum *p, Stat4Sample *pTo, Stat4Sample *pFrom){
+  pTo->isPSample = pFrom->isPSample;
+  pTo->iCol = pFrom->iCol;
+  pTo->iHash = pFrom->iHash;
+  memcpy(pTo->anEq, pFrom->anEq, sizeof(tRowcnt)*p->nCol);
+  memcpy(pTo->anLt, pFrom->anLt, sizeof(tRowcnt)*p->nCol);
+  memcpy(pTo->anDLt, pFrom->anDLt, sizeof(tRowcnt)*p->nCol);
+  if( pFrom->nRowid ){
+    sampleSetRowid(p->db, pTo, pFrom->nRowid, pFrom->u.aRowid);
+  }else{
+    sampleSetRowidInt64(p->db, pTo, pFrom->u.iRowid);
   }
 }
 #endif
@@ -324,6 +361,7 @@ static void stat4Destructor(void *pOld){
   int i;
   for(i=0; i<p->nCol; i++) sampleClear(p->db, p->aBest+i);
   for(i=0; i<p->mxSample; i++) sampleClear(p->db, p->a+i);
+  sampleClear(p->db, &p->current);
 #endif
   sqlite3DbFree(p->db, p);
 }
@@ -484,23 +522,6 @@ static int sampleIsBetter(
 }
 
 /*
-** Copy the contents of object (*pFrom) into (*pTo).
-*/
-static void sampleCopy(Stat4Accum *p, Stat4Sample *pTo, Stat4Sample *pFrom){
-  pTo->iRowid = pFrom->iRowid;
-  pTo->isPSample = pFrom->isPSample;
-  pTo->iCol = pFrom->iCol;
-  pTo->iHash = pFrom->iHash;
-  sampleClear(p->db, pTo);
-  pTo->nRowid = pFrom->nRowid;
-  pTo->aRowid = pFrom->aRowid;
-  sampleDup(p->db, pTo);
-  memcpy(pTo->anEq, pFrom->anEq, sizeof(tRowcnt)*p->nCol);
-  memcpy(pTo->anLt, pFrom->anLt, sizeof(tRowcnt)*p->nCol);
-  memcpy(pTo->anDLt, pFrom->anDLt, sizeof(tRowcnt)*p->nCol);
-}
-
-/*
 ** Copy the contents of sample *pNew into the p->a[] array. If necessary,
 ** remove the least desirable sample from p->a[] to make room.
 */
@@ -545,6 +566,7 @@ static void sampleInsert(Stat4Accum *p, Stat4Sample *pNew, int nEqZero){
     tRowcnt *anEq = pMin->anEq;
     tRowcnt *anLt = pMin->anLt;
     tRowcnt *anDLt = pMin->anDLt;
+    sampleClear(p->db, pMin);
     memmove(pMin, &pMin[1], sizeof(p->a[0])*(p->nSample-p->iMin-1));
     pSample = &p->a[p->nSample-1];
     pSample->anEq = anEq;
@@ -694,13 +716,10 @@ static void statPush(
   p->nRow++;
 #ifdef SQLITE_ENABLE_STAT3_OR_STAT4
   if( sqlite3_value_type(argv[2])==SQLITE_INTEGER ){
-    p->current.iRowid = sqlite3_value_int64(argv[2]);
-    p->current.aRowid = 0;
-    p->current.nRowid = 0;
+    sampleSetRowidInt64(p->db, &p->current, sqlite3_value_int64(argv[2]));
   }else{
-    p->current.iRowid = 0;
-    p->current.nRowid = sqlite3_value_bytes(argv[2]);
-    p->current.aRowid = (u8*)sqlite3_value_blob(argv[2]);
+    sampleSetRowid(p->db, &p->current, sqlite3_value_bytes(argv[2]),
+                                       sqlite3_value_blob(argv[2]));
   }
   p->current.iHash = p->iPrn = p->iPrn*1103515245 + 12345;
 #endif
@@ -827,9 +846,10 @@ static void statGet(
     if( p->iGet<p->nSample ){
       Stat4Sample *pS = p->a + p->iGet;
       if( pS->nRowid==0 ){
-        sqlite3_result_int64(context, pS->iRowid);
+        sqlite3_result_int64(context, pS->u.iRowid);
       }else{
-        sqlite3_result_blob(context, pS->aRowid, pS->nRowid, SQLITE_STATIC);
+        sqlite3_result_blob(context, pS->u.aRowid, pS->nRowid,
+                            SQLITE_TRANSIENT);
       }
     }
   }else{
diff --git a/src/insert.c b/src/insert.c
index 3cc88f96f..a6cab3e89 100644
--- a/src/insert.c
+++ b/src/insert.c
@@ -1463,7 +1463,10 @@ void sqlite3GenerateConstraintChecks(
     ** of a WITHOUT ROWID table and there has been no change the
     ** primary key, then no collision is possible.  The collision detection
     ** logic below can all be skipped. */
-    if( isUpdate && pPk && pkChng==0 ) continue;
+    if( isUpdate && pPk && pkChng==0 ){
+      sqlite3VdbeResolveLabel(v, addrUniqueOk);
+      continue;
+    }
 
     /* Find out what action to take in case there is a uniqueness conflict */
     onError = pIdx->onError;
@@ -1498,38 +1501,40 @@ void sqlite3GenerateConstraintChecks(
     }else{
       int x;
       /* Extract the PRIMARY KEY from the end of the index entry and
-      ** store it in register regR..regR+nPk-1 */
-      for(i=0; i<pPk->nKeyCol; i++){
-        x = sqlite3ColumnOfIndex(pIdx, pPk->aiColumn[i]);
-        sqlite3VdbeAddOp3(v, OP_Column, iThisCur, x, regR+i);
-        VdbeComment((v, "%s.%s", pTab->zName,
-                     pTab->aCol[pPk->aiColumn[i]].zName));
+      ** store it in registers regR..regR+nPk-1 */
+      if( isUpdate || onError==OE_Replace ){
+        for(i=0; i<pPk->nKeyCol; i++){
+          x = sqlite3ColumnOfIndex(pIdx, pPk->aiColumn[i]);
+          sqlite3VdbeAddOp3(v, OP_Column, iThisCur, x, regR+i);
+          VdbeComment((v, "%s.%s", pTab->zName,
+                       pTab->aCol[pPk->aiColumn[i]].zName));
+        }
       }
-      if( pIdx->autoIndex==2 ){
-        /* For a PRIMARY KEY index on a WITHOUT ROWID table, always conflict
-        ** on an INSERT.  On an UPDATE, only conflict if the PRIMARY KEY
-        ** has changed. */
-        if( isUpdate ){
+      if( isUpdate ){
+        if( pIdx->autoIndex==2 ){
+          /* For a PRIMARY KEY index on a WITHOUT ROWID table, always conflict
+          ** on an INSERT.  On an UPDATE, only conflict if the PRIMARY KEY
+          ** has changed. */
           int addrPkConflict = sqlite3VdbeCurrentAddr(v)+pPk->nKeyCol;
           for(i=0; i<pPk->nKeyCol-1; i++){
             x = pPk->aiColumn[i];
             sqlite3VdbeAddOp3(v, OP_Ne, regOldData+1+x,
-                              addrPkConflict, regIdx+x);
+                              addrPkConflict, regIdx+i);
           }
           x = pPk->aiColumn[i];
           sqlite3VdbeAddOp3(v, OP_Eq, regOldData+1+x, addrUniqueOk, regIdx+i);
+        }else{
+          /* For a UNIQUE index on a WITHOUT ROWID table, conflict only if the
+          ** PRIMARY KEY value of the match is different from the old
+          ** PRIMARY KEY value from before the update. */
+          int addrConflict = sqlite3VdbeCurrentAddr(v)+pPk->nKeyCol;
+          for(i=0; i<pPk->nKeyCol-1; i++){
+            sqlite3VdbeAddOp3(v, OP_Ne, regOldData+pPk->aiColumn[i]+1,
+                              addrConflict, regR+i);
+          }
+          sqlite3VdbeAddOp3(v, OP_Eq, regOldData+pPk->aiColumn[i]+1,
+                            addrUniqueOk, regR+i);
         }
-      }else{
-        /* For a UNIQUE index on a WITHOUT ROWID table, conflict only if the
-        ** PRIMARY KEY value of the match is different from the old PRIMARY KEY
-        ** value from before the update. */
-        int addrConflict = sqlite3VdbeCurrentAddr(v)+pPk->nKeyCol;
-        for(i=0; i<pPk->nKeyCol-1; i++){
-          sqlite3VdbeAddOp3(v, OP_Ne,
-                           regOldData+pPk->aiColumn[i]+1, addrConflict, regR+i);
-        }
-        sqlite3VdbeAddOp3(v, OP_Eq,
-                          regOldData+pPk->aiColumn[i]+1, addrUniqueOk, regR+i);
       }
     }
     sqlite3ReleaseTempRange(pParse, regIdx, pIdx->nColumn);
@@ -1827,8 +1832,8 @@ static int xferOptimization(
   int iDbSrc;                      /* The database of pSrc */
   int iSrc, iDest;                 /* Cursors from source and destination */
   int addr1, addr2;                /* Loop addresses */
-  int emptyDestTest;               /* Address of test for empty pDest */
-  int emptySrcTest;                /* Address of test for empty pSrc */
+  int emptyDestTest = 0;           /* Address of test for empty pDest */
+  int emptySrcTest = 0;            /* Address of test for empty pSrc */
   Vdbe *v;                         /* The VDBE we are building */
   KeyInfo *pKey;                   /* Key information for an index */
   int regAutoinc;                  /* Memory register used by AUTOINC */
@@ -2026,6 +2031,9 @@ static int xferOptimization(
     sqlite3VdbeAddOp2(v, OP_Next, iSrc, addr1);
     sqlite3VdbeAddOp2(v, OP_Close, iSrc, 0);
     sqlite3VdbeAddOp2(v, OP_Close, iDest, 0);
+  }else{
+    sqlite3TableLock(pParse, iDbDest, pDest->tnum, 1, pDest->zName);
+    sqlite3TableLock(pParse, iDbSrc, pSrc->tnum, 0, pSrc->zName);
   }
   for(pDestIdx=pDest->pIndex; pDestIdx; pDestIdx=pDestIdx->pNext){
     for(pSrcIdx=pSrc->pIndex; ALWAYS(pSrcIdx); pSrcIdx=pSrcIdx->pNext){
diff --git a/src/pragma.c b/src/pragma.c
index 3871ad02a..948d96056 100644
--- a/src/pragma.c
+++ b/src/pragma.c
@@ -1903,7 +1903,7 @@ void sqlite3Pragma(
           r1 = sqlite3GenerateIndexKey(pParse, pIdx, iDataCur, 0, 0, &jmp3);
           sqlite3VdbeAddOp2(v, OP_AddImm, 8+j, 1);  /* increment entry count */
           jmp2 = sqlite3VdbeAddOp4Int(v, OP_Found, iIdxCur+j, 0, r1,
-                                      pIdx->nKeyCol+1);
+                                      pIdx->nColumn);
           sqlite3VdbeAddOp2(v, OP_AddImm, 1, -1); /* Decrement error limit */
           sqlite3VdbeAddOp4(v, OP_String8, 0, 3, 0, "row ", P4_STATIC);
           sqlite3VdbeAddOp3(v, OP_Concat, 7, 3, 3);