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

Overview
Comment:Fix some problems with REPLACE and other conflict handling.
Downloads: Tarball | ZIP archive
Timelines: family | ancestors | descendants | both | primary-keys
Files: files | file ages | folders
SHA1: ff493aaa41b6fc181a59faffd63c22cb7b9516f0
User & Date: dan 2012-04-11 18:48:51.344
Context
2012-04-12
11:49
Fix a problem in kvmemRemoveNode. check-in: f3c424ddf3 user: dan tags: primary-keys
2012-04-11
18:48
Fix some problems with REPLACE and other conflict handling. check-in: ff493aaa41 user: dan tags: primary-keys
15:01
Fix various bugs. check-in: a70fb0629b user: dan tags: primary-keys
Changes
Unified Diff Show Whitespace Changes Patch
Changes to src/build.c.
1957
1958
1959
1960
1961
1962
1963

1964

1965
1966
1967
1968
1969
1970
1971
** Also write code to modify the sqlite_master table and internal schema
** if a root-page of another table is moved by the btree-layer whilst
** erasing iTable (this can happen with an auto-vacuum database).
*/ 
static void destroyRootPage(Parse *pParse, int iTable, int iDb){
  Vdbe *v = sqlite4GetVdbe(pParse);
  sqlite4VdbeAddOp2(v, OP_Clear, iTable, iDb);

  sqlite4MayAbort(pParse);

}

/*
** Write VDBE code to erase table pTab and all associated indices on disk.
** Code to update the sqlite_master tables and internal schema definitions
** in case a root-page belonging to another table is moved by the btree layer
** is also added (this can happen with an auto-vacuum database).







>

>







1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
** Also write code to modify the sqlite_master table and internal schema
** if a root-page of another table is moved by the btree-layer whilst
** erasing iTable (this can happen with an auto-vacuum database).
*/ 
static void destroyRootPage(Parse *pParse, int iTable, int iDb){
  Vdbe *v = sqlite4GetVdbe(pParse);
  sqlite4VdbeAddOp2(v, OP_Clear, iTable, iDb);
#if 0
  sqlite4MayAbort(pParse);
#endif
}

/*
** Write VDBE code to erase table pTab and all associated indices on disk.
** Code to update the sqlite_master tables and internal schema definitions
** in case a root-page belonging to another table is moved by the btree layer
** is also added (this can happen with an auto-vacuum database).
Changes to src/insert.c.
1211
1212
1213
1214
1215
1216
1217




























1218
1219
1220
1221
1222
1223
1224
  int iPk = 0;
  for(p=pTab->pIndex; p && p->eIndexType!=SQLITE_INDEX_PRIMARYKEY; p=p->pNext){
    iPk++;
  }
  if( piPk ) *piPk = iPk;
  return p;
}





























/*
** This function generates code used as part of both INSERT and UPDATE
** statements. The generated code performs two tasks:
**
**   1. Checks all NOT NULL, CHECK and UNIQUE database constraints, 
**      including the implicit NOT NULL and UNIQUE constraints imposed







>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>







1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
  int iPk = 0;
  for(p=pTab->pIndex; p && p->eIndexType!=SQLITE_INDEX_PRIMARYKEY; p=p->pNext){
    iPk++;
  }
  if( piPk ) *piPk = iPk;
  return p;
}

/*
** Index pIdx is a UNIQUE index. This function returns a pointer to a buffer
** containing an error message to tell the user that the UNIQUE constraint
** has failed.
**
** The returned buffer should be freed by the caller using sqlite4DbFree().
*/
static char *notUniqueMessage(
  Parse *pParse,                  /* Parse context */
  Index *pIdx                     /* Index to generate error message for */
){
  const int nCol = pIdx->nColumn; /* Number of columns indexed by pIdx */
  StrAccum errMsg;                /* Buffer to build error message within */
  int iCol;                       /* Used to iterate through indexed columns */

  sqlite4StrAccumInit(&errMsg, 0, 0, 200);
  errMsg.db = pParse->db;
  sqlite4StrAccumAppend(&errMsg, (nCol>1 ? "columns " : "column "), -1);
  for(iCol=0; iCol<pIdx->nColumn; iCol++){
    const char *zCol = indexColumnName(pIdx, iCol);
    sqlite4StrAccumAppend(&errMsg, (iCol==0 ? "" : ", "), -1);
    sqlite4StrAccumAppend(&errMsg, zCol, -1);
  }
  sqlite4StrAccumAppend(&errMsg, (nCol>1 ? " are" : " is"), -1);
  sqlite4StrAccumAppend(&errMsg, " not unique", -1);
  return sqlite4StrAccumFinish(&errMsg);
}

/*
** This function generates code used as part of both INSERT and UPDATE
** statements. The generated code performs two tasks:
**
**   1. Checks all NOT NULL, CHECK and UNIQUE database constraints, 
**      including the implicit NOT NULL and UNIQUE constraints imposed
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
** aRegIdx:
**   Array sized so that there is one entry for each index (including the
**   PK index) attached to the database table. Entries are in the same order
**   as the linked list of Index structures attached to the table. 
**
**   If an array entry is non-zero, it contains the register that the 
**   corresponding index key should be written to. If an entry is zero, then
**   the corresponding index key is not required by the caller and that any 
**   UNIQUE enforced by the index does not need to be checked.
**
** 
**
** Generate code to do constraint checks prior to an INSERT or an UPDATE.
**
** The input is a range of consecutive registers as follows:
**







|
|







1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
** aRegIdx:
**   Array sized so that there is one entry for each index (including the
**   PK index) attached to the database table. Entries are in the same order
**   as the linked list of Index structures attached to the table. 
**
**   If an array entry is non-zero, it contains the register that the 
**   corresponding index key should be written to. If an entry is zero, then
**   the corresponding index key is not required by the caller. In this case
**   any UNIQUE constraint enforced by the index does not need to be checked.
**
** 
**
** Generate code to do constraint checks prior to an INSERT or an UPDATE.
**
** The input is a range of consecutive registers as follows:
**
1341
1342
1343
1344
1345
1346
1347


1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361

1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387

1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426








1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451

1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
  int rowidChng,      /* True if the rowid might collide with existing entry */
  int isUpdate,       /* True for UPDATE, False for INSERT */

  int overrideError,  /* Override onError to this if not OE_Default */
  int ignoreDest,     /* Jump to this label on an OE_Ignore resolution */
  int *pbMayReplace   /* OUT: Set to true if constraint may cause a replace */
){


  Index *pPk;                     /* Primary key index for table pTab */
  int i;              /* loop counter */
  Vdbe *v;            /* VDBE under constrution */
  int nCol;           /* Number of columns */
  int onError;        /* Conflict resolution strategy */
  int iCur;           /* Table cursor number */
  Index *pIdx;         /* Pointer to one of the indices */
  int seenReplace = 0; /* True if REPLACE is used to resolve INT PK conflict */

  v = sqlite4GetVdbe(pParse);
  assert( v!=0 );
  assert( pTab->pSelect==0 );  /* This table is not a VIEW */
  nCol = pTab->nCol;
  pPk = sqlite4FindPrimaryKey(pTab, 0);


  assert( pPk->eIndexType==SQLITE_INDEX_PRIMARYKEY );

  /* Test all NOT NULL constraints. */
  generateNotNullChecks(pParse, pTab, regContent, overrideError, ignoreDest);

  /* Test all CHECK constraints */
  generateCheckChecks(pParse, pTab, regContent, overrideError, ignoreDest);

  /* Test all UNIQUE constraints by creating entries for each UNIQUE
  ** index and making sure that duplicate entries do not already exist.
  ** Add the new records to the indices as we go.
  */
  for(iCur=0, pIdx=pTab->pIndex; pIdx; pIdx=pIdx->pNext, iCur++){
    int nTmpReg;                  /* Number of temp registers required */
    int regTmp;                   /* First temp register allocated */
    int regShort;                 /* Reg. for number of bytes in short key */

    if( aRegIdx[iCur]==0 ) continue;  /* Skip unused indices */

    /* Create an index key. Primary key indexes consists of just the primary
    ** key values. Other indexes consists of the indexed columns followed by
    ** the primary key values.  */
    nTmpReg = 1 + pIdx->nColumn + (pIdx==pPk ? 0 : pPk->nColumn);
    regTmp = sqlite4GetTempRange(pParse, nTmpReg);
    regShort = regTmp + nTmpReg - 1;

    for(i=0; i<pIdx->nColumn; i++){
      int idx = pIdx->aiColumn[i];
      sqlite4VdbeAddOp2(v, OP_SCopy, regContent+idx, regTmp+i);
    }
    if( pIdx!=pPk ){
      for(i=0; i<pPk->nColumn; i++){
        int idx = pPk->aiColumn[i];
        sqlite4VdbeAddOp2(v, OP_SCopy, regContent+idx, regTmp+i+pIdx->nColumn);
      }
    }
    sqlite4VdbeAddOp3(v, OP_MakeIdxKey, baseCur+iCur, regTmp, aRegIdx[iCur]);
    sqlite4VdbeChangeP4(v, -1, (const char *)regShort, P4_INT32);
    VdbeComment((v, "key for %s", pIdx->zName));

    /* If Index.onError==OE_None, then pIdx is not a UNIQUE or PRIMARY KEY 
    ** index. In this case there is no need to test the index for uniqueness
    ** - all that is required is to populate the aRegIdx[iCur] register. Jump 
    ** to the next iteration of the loop if this is the case.  */
    onError = pIdx->onError;
    if( onError!=OE_None ){
      int iTest;                  /* Address of OP_IsUnique instruction */

      iTest = sqlite4VdbeAddOp3(v, OP_IsUnique, baseCur+iCur, 0, aRegIdx[iCur]);
      sqlite4VdbeChangeP4(v, -1, (const char *)regShort, P4_INT32);

      /* Figure out what to do if a UNIQUE constraint is encountered. 
      **
      ** TODO: If a previous constraint is a REPLACE, why change IGNORE to
      ** REPLACE and FAIL to ABORT here?  */
      if( overrideError!=OE_Default ){
        onError = overrideError;
      }else if( onError==OE_Default ){
        onError = OE_Abort;
      }
      if( seenReplace ){
        if( onError==OE_Ignore ) onError = OE_Replace;
        else if( onError==OE_Fail ) onError = OE_Abort;
      }
      








      switch( onError ){
        case OE_Rollback:
        case OE_Abort:
        case OE_Fail: {
          int j;
          StrAccum errMsg;
          const char *zSep;
          char *zErr;

          sqlite4StrAccumInit(&errMsg, 0, 0, 200);
          errMsg.db = pParse->db;
          zSep = pIdx->nColumn>1 ? "columns " : "column ";
          for(j=0; j<pIdx->nColumn; j++){
            const char *zCol = indexColumnName(pIdx, j);
            sqlite4StrAccumAppend(&errMsg, zSep, -1);
            zSep = ", ";
            sqlite4StrAccumAppend(&errMsg, zCol, -1);
          }
          sqlite4StrAccumAppend(&errMsg,
              pIdx->nColumn>1 ? " are not unique" : " is not unique", -1);
          zErr = sqlite4StrAccumFinish(&errMsg);
          sqlite4HaltConstraint(pParse, onError, zErr, 0);
          sqlite4DbFree(errMsg.db, zErr);
          break;
        }

        case OE_Ignore: {
          assert( seenReplace==0 );
          sqlite4VdbeAddOp2(v, OP_Goto, 0, ignoreDest);
          break;
        }
        default: {
          assert( 0 );
#if 0
          Trigger *pTrigger = 0;
          assert( onError==OE_Replace );
          sqlite4MultiWrite(pParse);
          if( pParse->db->flags&SQLITE_RecTriggers ){
            pTrigger = sqlite4TriggersExist(pParse, pTab, TK_DELETE, 0, 0);
          }
          sqlite4GenerateRowDelete(
              pParse, pTab, baseCur, regR, 0, pTrigger, OE_Replace
          );
          seenReplace = 1;
#endif
          break;
        }
      }

      /* If the OP_IsUnique passes (no constraint violation) jump here */
      sqlite4VdbeJumpHere(v, iTest);
    }







>
>














>
















|








|
>











<









|
<
<















>
>
>
>
>
>
>
>




<
<
<
|
<
<
<
<
<
<
<
<
<
<
<
<
<

|


>






<
<
|


<

<

|


<







1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430

1431
1432
1433
1434
1435
1436
1437
1438
1439
1440


1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467



1468













1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479


1480
1481
1482

1483

1484
1485
1486
1487

1488
1489
1490
1491
1492
1493
1494
  int rowidChng,      /* True if the rowid might collide with existing entry */
  int isUpdate,       /* True for UPDATE, False for INSERT */

  int overrideError,  /* Override onError to this if not OE_Default */
  int ignoreDest,     /* Jump to this label on an OE_Ignore resolution */
  int *pbMayReplace   /* OUT: Set to true if constraint may cause a replace */
){
  u8 aPkRoot[10];                 /* Root page number for pPk as a varint */ 
  int nPkRoot;                    /* Size of aPkRoot in bytes */
  Index *pPk;                     /* Primary key index for table pTab */
  int i;              /* loop counter */
  Vdbe *v;            /* VDBE under constrution */
  int nCol;           /* Number of columns */
  int onError;        /* Conflict resolution strategy */
  int iCur;           /* Table cursor number */
  Index *pIdx;         /* Pointer to one of the indices */
  int seenReplace = 0; /* True if REPLACE is used to resolve INT PK conflict */

  v = sqlite4GetVdbe(pParse);
  assert( v!=0 );
  assert( pTab->pSelect==0 );  /* This table is not a VIEW */
  nCol = pTab->nCol;
  pPk = sqlite4FindPrimaryKey(pTab, 0);
  nPkRoot = sqlite4PutVarint64(aPkRoot, pPk->tnum);

  assert( pPk->eIndexType==SQLITE_INDEX_PRIMARYKEY );

  /* Test all NOT NULL constraints. */
  generateNotNullChecks(pParse, pTab, regContent, overrideError, ignoreDest);

  /* Test all CHECK constraints */
  generateCheckChecks(pParse, pTab, regContent, overrideError, ignoreDest);

  /* Test all UNIQUE constraints by creating entries for each UNIQUE
  ** index and making sure that duplicate entries do not already exist.
  ** Add the new records to the indices as we go.
  */
  for(iCur=0, pIdx=pTab->pIndex; pIdx; pIdx=pIdx->pNext, iCur++){
    int nTmpReg;                  /* Number of temp registers required */
    int regTmp;                   /* First temp register allocated */
    int regPk;                    /* PK of conflicting row (for REPLACE) */

    if( aRegIdx[iCur]==0 ) continue;  /* Skip unused indices */

    /* Create an index key. Primary key indexes consists of just the primary
    ** key values. Other indexes consists of the indexed columns followed by
    ** the primary key values.  */
    nTmpReg = 1 + pIdx->nColumn + (pIdx==pPk ? 0 : pPk->nColumn);
    regTmp = sqlite4GetTempRange(pParse, nTmpReg);
    regPk = regTmp + nTmpReg - 1;

    for(i=0; i<pIdx->nColumn; i++){
      int idx = pIdx->aiColumn[i];
      sqlite4VdbeAddOp2(v, OP_SCopy, regContent+idx, regTmp+i);
    }
    if( pIdx!=pPk ){
      for(i=0; i<pPk->nColumn; i++){
        int idx = pPk->aiColumn[i];
        sqlite4VdbeAddOp2(v, OP_SCopy, regContent+idx, regTmp+i+pIdx->nColumn);
      }
    }
    sqlite4VdbeAddOp3(v, OP_MakeIdxKey, baseCur+iCur, regTmp, aRegIdx[iCur]);

    VdbeComment((v, "key for %s", pIdx->zName));

    /* If Index.onError==OE_None, then pIdx is not a UNIQUE or PRIMARY KEY 
    ** index. In this case there is no need to test the index for uniqueness
    ** - all that is required is to populate the aRegIdx[iCur] register. Jump 
    ** to the next iteration of the loop if this is the case.  */
    onError = pIdx->onError;
    if( onError!=OE_None ){
      int iTest;                  /* Address of OP_IsUnique instruction */
      int regOut = 0;



      /* Figure out what to do if a UNIQUE constraint is encountered. 
      **
      ** TODO: If a previous constraint is a REPLACE, why change IGNORE to
      ** REPLACE and FAIL to ABORT here?  */
      if( overrideError!=OE_Default ){
        onError = overrideError;
      }else if( onError==OE_Default ){
        onError = OE_Abort;
      }
      if( seenReplace ){
        if( onError==OE_Ignore ) onError = OE_Replace;
        else if( onError==OE_Fail ) onError = OE_Abort;
      }
      
      if( onError==OE_Replace ){
        sqlite4VdbeAddOp3(v, OP_Blob, nPkRoot, regPk, 0);
        sqlite4VdbeChangeP4(v, -1, aPkRoot, nPkRoot);
        regOut = regPk;
      }
      iTest = sqlite4VdbeAddOp3(v, OP_IsUnique, baseCur+iCur, 0, aRegIdx[iCur]);
      sqlite4VdbeChangeP4(v, -1, SQLITE_INT_TO_PTR(regOut), P4_INT32);
      
      switch( onError ){
        case OE_Rollback:
        case OE_Abort:
        case OE_Fail: {



          char *zErr = notUniqueMessage(pParse, pIdx);













          sqlite4HaltConstraint(pParse, onError, zErr, 0);
          sqlite4DbFree(pParse->db, zErr);
          break;
        }

        case OE_Ignore: {
          assert( seenReplace==0 );
          sqlite4VdbeAddOp2(v, OP_Goto, 0, ignoreDest);
          break;
        }
        default: {


          Trigger *pTrigger;
          assert( onError==OE_Replace );
          sqlite4MultiWrite(pParse);

            pTrigger = sqlite4TriggersExist(pParse, pTab, TK_DELETE, 0, 0);

          sqlite4GenerateRowDelete(
              pParse, pTab, baseCur, regOut, 0, pTrigger, OE_Replace
          );
          seenReplace = 1;

          break;
        }
      }

      /* If the OP_IsUnique passes (no constraint violation) jump here */
      sqlite4VdbeJumpHere(v, iTest);
    }
Changes to src/vdbe.c.
2982
2983
2984
2985
2986
2987
2988
2989
2990
2991
2992
2993








2994
2995
2996
2997
2998


2999
3000

3001
3002
3003
3004
3005
3006
3007
3008
3009
3010






3011
3012
3013
3014
3015
3016
3017
3018
3019
3020
3021
3022
3023
3024
3025
3026
3027
3028







3029
3030
3031
3032
3033
3034
3035
  break;
}

/* Opcode: IsUnique P1 P2 P3 P4 *
**
** Cursor P1 is open on an index that enforces a UNIQUE constraint. 
** Register P3 contains an encoded key suitable to be inserted into the 
** index.
**
** Jump to instruction P2 if the encoded key can be inserted into the 
** index without violating a unique constraint. Otherwise, fall through
** to the next instruction.








*/
case OP_IsUnique: {        /* jump, in3 */
  VdbeCursor *pC;
  Mem *pShort;
  Mem *pProbe;


  int nShort;
  int dir;


  KVByteArray *aKey;              /* Key read from cursor */
  KVSize nKey;                    /* Size of aKey in bytes */

  assert( pOp->p4type==P4_INT32 );

  pProbe = &aMem[pOp->p3];
  pShort = &aMem[pOp->p4.i];
  nShort = pShort->u.i;
  pC = p->apCsr[pOp->p1];






  assert( nShort<=pProbe->n );
  assert( (nShort==pProbe->n)==(pC->pKeyInfo->nPK==0) );

  dir = (nShort < pProbe->n);
  rc = sqlite4KVCursorSeek(pC->pKVCur, pProbe->z, nShort, dir);

  if( rc==SQLITE_NOTFOUND ){
    rc = SQLITE_OK;
    pc = pOp->p2-1;
  }else if( rc==SQLITE_INEXACT ){
    assert( nShort<pProbe->n );
    rc = sqlite4KVCursorKey(pC->pKVCur, &aKey, &nKey);
    if( rc==SQLITE_OK ){
      if( nKey<nShort 
       || memcmp(pProbe->z, aKey, nShort)
       || (nKey==pProbe->n && 0==memcmp(pProbe->z, aKey, nKey))
      ){
        pc = pOp->p2-1;







      }
    }
  }

#if 0
  u16 ii;
  VdbeCursor *pCx;







<
<
|
|

>
>
>
>
>
>
>
>





>
>


>







<
<

>
>
>
>
>
>


















>
>
>
>
>
>
>







2982
2983
2984
2985
2986
2987
2988


2989
2990
2991
2992
2993
2994
2995
2996
2997
2998
2999
3000
3001
3002
3003
3004
3005
3006
3007
3008
3009
3010
3011
3012
3013
3014
3015
3016


3017
3018
3019
3020
3021
3022
3023
3024
3025
3026
3027
3028
3029
3030
3031
3032
3033
3034
3035
3036
3037
3038
3039
3040
3041
3042
3043
3044
3045
3046
3047
3048
3049
3050
3051
3052
3053
3054
3055
  break;
}

/* Opcode: IsUnique P1 P2 P3 P4 *
**
** Cursor P1 is open on an index that enforces a UNIQUE constraint. 
** Register P3 contains an encoded key suitable to be inserted into the 


** index. If the key can be inserted into the index without violating
** a UNIQUE constraint, jump to instruction P2. Otherwise, fall through
** to the next instruction.
**
** If P4 is a non-zero integer and the jump is not taken, then it is
** a register that currently contains a blob. At the start of the blob
** is a varint that contains the index number for the PRIMARY KEY index
** of the table. The contents of P4 are overwritten with an index key
** composed of the varint from the start of the initial blob content
** and the PRIMARY KEY values from the index entry causing the UNIQUE
** constraint to fail.
*/
case OP_IsUnique: {        /* jump, in3 */
  VdbeCursor *pC;
  Mem *pShort;
  Mem *pProbe;
  Mem *pOut;
  int iOut;
  int nShort;
  int dir;
  u64 dummy;

  KVByteArray *aKey;              /* Key read from cursor */
  KVSize nKey;                    /* Size of aKey in bytes */

  assert( pOp->p4type==P4_INT32 );

  pProbe = &aMem[pOp->p3];


  pC = p->apCsr[pOp->p1];
  pOut = (pOp->p4.i==0 ? 0 : &aMem[pOp->p4.i]);
  assert( pOut==0 || (pOut->flags & MEM_Blob) );

  nShort = sqlite4VdbeShortKey(pProbe->z, pProbe->n, 
      pC->pKeyInfo->nField - pC->pKeyInfo->nPK
  );
  assert( nShort<=pProbe->n );
  assert( (nShort==pProbe->n)==(pC->pKeyInfo->nPK==0) );

  dir = (nShort < pProbe->n);
  rc = sqlite4KVCursorSeek(pC->pKVCur, pProbe->z, nShort, dir);

  if( rc==SQLITE_NOTFOUND ){
    rc = SQLITE_OK;
    pc = pOp->p2-1;
  }else if( rc==SQLITE_INEXACT ){
    assert( nShort<pProbe->n );
    rc = sqlite4KVCursorKey(pC->pKVCur, &aKey, &nKey);
    if( rc==SQLITE_OK ){
      if( nKey<nShort 
       || memcmp(pProbe->z, aKey, nShort)
       || (nKey==pProbe->n && 0==memcmp(pProbe->z, aKey, nKey))
      ){
        pc = pOp->p2-1;
      }else if( pOut ){
        iOut = sqlite4GetVarint64(pOut->z, pOut->n, &dummy);
        rc = sqlite4VdbeMemGrow(pOut, iOut+(pProbe->n - nShort), 1);
        if( rc==SQLITE_OK ){
          memcpy(&pOut->z[iOut], &aKey[nShort], (pProbe->n - nShort));
          pOut->n = iOut + (pProbe->n - nShort);
        }
      }
    }
  }

#if 0
  u16 ii;
  VdbeCursor *pCx;
Changes to src/vdbeInt.h.
386
387
388
389
390
391
392

393
394
395
396
397
398
399
  KeyInfo *pKeyInfo,           /* Collating sequence information */
  u8 **pzOut,                  /* Write the resulting key here */
  int *pnOut,                  /* Number of bytes in the key */
  int *pnShort                 /* Number of bytes omitting primary key */
);
int sqlite4VdbeEncodeIntKey(u8 *aBuf,sqlite4_int64 v);
int sqlite4VdbeDecodeIntKey(const KVByteArray*, KVSize, sqlite4_int64*);

int sqlite4MemCompare(const Mem*, const Mem*, const CollSeq*);
int sqlite4VdbeExec(Vdbe*);
int sqlite4VdbeList(Vdbe*);
int sqlite4VdbeHalt(Vdbe*);
int sqlite4VdbeChangeEncoding(Mem *, int);
int sqlite4VdbeMemTooBig(Mem*);
int sqlite4VdbeMemCopy(Mem*, const Mem*);







>







386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
  KeyInfo *pKeyInfo,           /* Collating sequence information */
  u8 **pzOut,                  /* Write the resulting key here */
  int *pnOut,                  /* Number of bytes in the key */
  int *pnShort                 /* Number of bytes omitting primary key */
);
int sqlite4VdbeEncodeIntKey(u8 *aBuf,sqlite4_int64 v);
int sqlite4VdbeDecodeIntKey(const KVByteArray*, KVSize, sqlite4_int64*);
int sqlite4VdbeShortKey(u8 *, int, int);
int sqlite4MemCompare(const Mem*, const Mem*, const CollSeq*);
int sqlite4VdbeExec(Vdbe*);
int sqlite4VdbeList(Vdbe*);
int sqlite4VdbeHalt(Vdbe*);
int sqlite4VdbeChangeEncoding(Mem *, int);
int sqlite4VdbeMemTooBig(Mem*);
int sqlite4VdbeMemCopy(Mem*, const Mem*);
Changes to src/vdbecodec.c.
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
      p->aOut[p->nOut++] = 0x22;  /* Large positive values */
      e = encodeLargeFloatKey(r, p);
      if( e<=10 ) p->aOut[i] = 0x17+e;
    }
  }else
  if( flags & MEM_Str ){
    if( enlargeEncoderAllocation(p, pMem->n*4 + 2) ) return SQLITE_NOMEM;
    p->aOut[p->nOut++] = 0x0e;   /* Text */
    if( pColl==0 || pColl->xMkKey==0 ){
      memcpy(p->aOut+p->nOut, pMem->z, pMem->n);
      p->nOut += pMem->n;
    }else{
      n = pColl->xMkKey(pColl->pUser, pMem->z, pMem->n, p->aOut+p->nOut,
                        p->nAlloc - p->nOut);
      if( n > p->nAlloc - p->nOut ){







|







526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
      p->aOut[p->nOut++] = 0x22;  /* Large positive values */
      e = encodeLargeFloatKey(r, p);
      if( e<=10 ) p->aOut[i] = 0x17+e;
    }
  }else
  if( flags & MEM_Str ){
    if( enlargeEncoderAllocation(p, pMem->n*4 + 2) ) return SQLITE_NOMEM;
    p->aOut[p->nOut++] = 0x24;   /* Text */
    if( pColl==0 || pColl->xMkKey==0 ){
      memcpy(p->aOut+p->nOut, pMem->z, pMem->n);
      p->nOut += pMem->n;
    }else{
      n = pColl->xMkKey(pColl->pUser, pMem->z, pMem->n, p->aOut+p->nOut,
                        p->nAlloc - p->nOut);
      if( n > p->nAlloc - p->nOut ){
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577

















































578
579
580
581
582
583
584
    unsigned char s, t;
    assert( flags & MEM_Blob );
    n = pMem->n;
    a = (u8*)pMem->z;
    s = 1;
    t = 0;
    if( enlargeEncoderAllocation(p, (n*8+6)/7 + 2) ) return SQLITE_NOMEM;
    p->aOut[p->nOut++] = 0x0f;   /* Blob */
     for(i=0; i<n; i++){
      unsigned char x = a[i];
      p->aOut[p->nOut++] = 0x80 | t | (x>>s);
      if( s<7 ){
        t = x<<(7-s);
        s++;
      }else{
        p->aOut[p->nOut++] = 0x80 | x;
        s = 1;
        t = 0;
      }
    }
    if( s>1 ) p->aOut[p->nOut++] = 0x80 | t;
    p->aOut[p->nOut++] = 0x00;
  }
  if( sortOrder==SQLITE_SO_DESC ){
    for(i=iStart; i<p->nOut; i++) p->aOut[i] ^= 0xff;
  }
  return SQLITE_OK;
}


















































/*
** Generate a record key from one or more data values
**
** Space to hold the key is obtained from sqlite4DbMalloc() and should
** be freed by the caller using sqlite4DbFree() to avoid a memory leak.
*/







|




















>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>







550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
    unsigned char s, t;
    assert( flags & MEM_Blob );
    n = pMem->n;
    a = (u8*)pMem->z;
    s = 1;
    t = 0;
    if( enlargeEncoderAllocation(p, (n*8+6)/7 + 2) ) return SQLITE_NOMEM;
    p->aOut[p->nOut++] = 0x25;   /* Blob */
     for(i=0; i<n; i++){
      unsigned char x = a[i];
      p->aOut[p->nOut++] = 0x80 | t | (x>>s);
      if( s<7 ){
        t = x<<(7-s);
        s++;
      }else{
        p->aOut[p->nOut++] = 0x80 | x;
        s = 1;
        t = 0;
      }
    }
    if( s>1 ) p->aOut[p->nOut++] = 0x80 | t;
    p->aOut[p->nOut++] = 0x00;
  }
  if( sortOrder==SQLITE_SO_DESC ){
    for(i=iStart; i<p->nOut; i++) p->aOut[i] ^= 0xff;
  }
  return SQLITE_OK;
}

/*
** Variables aKey/nKey contain an encoded index key. This function returns
** the length (in bytes) of the key with all but the first nField fields
** removed.
*/
int sqlite4VdbeShortKey(
  u8 *aKey,                       /* Buffer containing encoded key */
  int nKey,                       /* Size of buffer aKey[] in bytes */
  int nField                      /* Number of fields */
){
  u8 *p = aKey;
  u8 *pEnd = &aKey[nKey];
  u64 dummy;
  int i;

  p = aKey;
  p += sqlite4GetVarint64(p, pEnd-p, &dummy);

  for(i=0; i<nField; i++){
    switch( *(p++) ){
      case 0x05:                  /* NULL */
      case 0x06:                  /* NaN */
      case 0x07:                  /* -ve infinity */
      case 0x15:                  /* zero */
      case 0x23:                  /* +ve infinity */
        break;

      case 0x24:                  /* Text */
      case 0x25:                  /* Blob */
        while( *(p++) );
        break;

      case 0x22:                  /* Large positive number */
      case 0x16:                  /* Small positive number */
      case 0x14:                  /* Small negative number */
      case 0x08:                  /* Large negative number */
        p += sqlite4GetVarint64(p, pEnd-p, &dummy);
        p += sqlite4GetVarint64(p, pEnd-p, &dummy);
        break;

      default:                    /* Medium sized number */
        p += sqlite4GetVarint64(p, pEnd-p, &dummy);
        break;
    }
  }

  return (p - aKey);
}

/*
** Generate a record key from one or more data values
**
** Space to hold the key is obtained from sqlite4DbMalloc() and should
** be freed by the caller using sqlite4DbFree() to avoid a memory leak.
*/
Changes to test/conflict.test.
49
50
51
52
53
54
55
56
57
58
59
60
61


62
63
64
65
66
67
68
  4 REPLACE                 0 4   1  0
  5 {INSERT OR FAIL}        1 {}  1  0
  6 {INSERT OR ABORT}       1 {}  1  0
  7 {INSERT OR ROLLBACK}    1 {}  {} 0
} {
  do_test conflict-1.$i {
    set ::sqlite_opentemp_count 0
    set r0 [catch {execsql [subst {
      DELETE FROM t1;
      DELETE FROM t2;
      INSERT INTO t1 VALUES(1,2,3);
      BEGIN;
      INSERT INTO t2 VALUES(1); 


      $cmd INTO t1 VALUES(1,2,4);
    }]} r1]
    catch {execsql {COMMIT}}
    if {$r0} {set r1 {}} {set r1 [execsql {SELECT c FROM t1}]}
    set r2 [execsql {SELECT x FROM t2}]
    set r3 $::sqlite_opentemp_count
    list $r0 $r1 $r2 $r3







|





>
>







49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
  4 REPLACE                 0 4   1  0
  5 {INSERT OR FAIL}        1 {}  1  0
  6 {INSERT OR ABORT}       1 {}  1  0
  7 {INSERT OR ROLLBACK}    1 {}  {} 0
} {
  do_test conflict-1.$i {
    set ::sqlite_opentemp_count 0
    execsql {
      DELETE FROM t1;
      DELETE FROM t2;
      INSERT INTO t1 VALUES(1,2,3);
      BEGIN;
      INSERT INTO t2 VALUES(1); 
    }
    set r0 [catch {execsql [subst {
      $cmd INTO t1 VALUES(1,2,4);
    }]} r1]
    catch {execsql {COMMIT}}
    if {$r0} {set r1 {}} {set r1 [execsql {SELECT c FROM t1}]}
    set r2 [execsql {SELECT x FROM t2}]
    set r3 $::sqlite_opentemp_count
    list $r0 $r1 $r2 $r3
95
96
97
98
99
100
101
102
103
104
105
106
107


108
109
110
111
112
113
114
  3 {INSERT OR REPLACE}     0 4   1
  4 REPLACE                 0 4   1
  5 {INSERT OR FAIL}        1 {}  1
  6 {INSERT OR ABORT}       1 {}  1
  7 {INSERT OR ROLLBACK}    1 {}  {}
} {
  do_test conflict-2.$i {
    set r0 [catch {execsql [subst {
      DELETE FROM t1;
      DELETE FROM t2;
      INSERT INTO t1 VALUES(1,2,3);
      BEGIN;
      INSERT INTO t2 VALUES(1); 


      $cmd INTO t1 VALUES(1,2,4);
    }]} r1]
    catch {execsql {COMMIT}}
    if {$r0} {set r1 {}} {set r1 [execsql {SELECT c FROM t1}]}
    set r2 [execsql {SELECT x FROM t2}]
    list $r0 $r1 $r2
  } [list $t0 $t1 $t2]







|





>
>







97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
  3 {INSERT OR REPLACE}     0 4   1
  4 REPLACE                 0 4   1
  5 {INSERT OR FAIL}        1 {}  1
  6 {INSERT OR ABORT}       1 {}  1
  7 {INSERT OR ROLLBACK}    1 {}  {}
} {
  do_test conflict-2.$i {
    execsql {
      DELETE FROM t1;
      DELETE FROM t2;
      INSERT INTO t1 VALUES(1,2,3);
      BEGIN;
      INSERT INTO t2 VALUES(1); 
    }
    set r0 [catch {execsql [subst {
      $cmd INTO t1 VALUES(1,2,4);
    }]} r1]
    catch {execsql {COMMIT}}
    if {$r0} {set r1 {}} {set r1 [execsql {SELECT c FROM t1}]}
    set r2 [execsql {SELECT x FROM t2}]
    list $r0 $r1 $r2
  } [list $t0 $t1 $t2]