SQLite

Check-in [e34eafd4c5]
Login

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

Overview
Comment:Add more tests for LSM log file recovery. Fix a problem in recovering log files that contain range deletes.
Downloads: Tarball | ZIP archive
Timelines: family | ancestors | descendants | both | trunk
Files: files | file ages | folders
SHA3-256: e34eafd4c5b2bbf2735e136ad69b67bb4288ad4d01a0128d8e107ac46209a182
User & Date: dan 2017-07-03 09:00:18.115
Context
2017-07-03
17:37
Attempt to improve documentation on sqlite3_column_ and sqlite3_value_ interfaces. (check-in: 9111ac69bf user: drh tags: trunk)
09:00
Add more tests for LSM log file recovery. Fix a problem in recovering log files that contain range deletes. (check-in: e34eafd4c5 user: dan tags: trunk)
2017-07-01
20:59
Fix a memory management problem in lsm log recovery code. (check-in: dd55af30b4 user: dan tags: trunk)
Changes
Side-by-Side Diff Ignore Whitespace Patch
Changes to ext/lsm1/lsm-test/lsmtest1.c.
︙
77
78
79
80
81
82
83
84

85
86
87
88
89


90
91
92
93
94
95
96

77
78
79
80
81
82
83

84
85
86
87


88
89
90
91
92
93
94
95
96








-
+



-
-
+
+








  int nIter;                      /* Total number of iterations to run */
};

/*
** Generate a unique name for the test case pTest with database system
** zSystem.
*/
static char *getName(const char *zSystem, Datatest1 *pTest){
static char *getName(const char *zSystem, int bRecover, Datatest1 *pTest){
  char *zRet;
  char *zData;
  zData = testDatasourceName(&pTest->defn);
  zRet = testMallocPrintf("data.%s.%s.%d.%d", 
      zSystem, zData, pTest->nRow, pTest->nVerify
  zRet = testMallocPrintf("data.%s.%s.rec=%d.%d.%d", 
      zSystem, zData, bRecover, pTest->nRow, pTest->nVerify
  );
  testFree(zData);
  return zRet;
}

int testControlDb(TestDb **ppDb){
#ifdef HAVE_KYOTOCABINET
︙
246
247
248
249
250
251
252












253
254
255

256
257
258
259
260
261
262

246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275








+
+
+
+
+
+
+
+
+
+
+
+



+








static void printScanCb(
    void *pCtx, void *pKey, int nKey, void *pVal, int nVal
){
  printf("%s\n", (char *)pKey);
  fflush(stdout);
}
#endif

void testReopenRecover(TestDb **ppDb, int *pRc){
  if( *pRc==0 ){
    const char *zLib = tdb_library_name(*ppDb);
    const char *zDflt = tdb_default_db(zLib);
    testCopyLsmdb(zDflt, "bak.db");
    testClose(ppDb);
    testCopyLsmdb("bak.db", zDflt);
    *pRc = tdb_open(zLib, 0, 0, ppDb);
  }
}


static void doDataTest1(
  const char *zSystem,            /* Database system to test */
  int bRecover,
  Datatest1 *p,                   /* Structure containing test parameters */
  int *pRc                        /* OUT: Error code */
){
  int i;
  int iDot;
  int rc = LSM_OK;
  Datasource *pData;
︙
273
274
275
276
277
278
279

280
281




282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301



302


303
304
305
306
307
308
309

286
287
288
289
290
291
292
293


294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320

321
322
323
324
325
326
327
328
329








+
-
-
+
+
+
+




















+
+
+
-
+
+








    /* Insert some data */
    testWriteDatasourceRange(pDb, pData, i, p->nVerify, &rc);
    i += p->nVerify;

    /* Check that the db content is correct. */
    testDbContents(pDb, pData, p->nRow, 0, i-1, p->nTest, p->bTestScan, &rc);

    if( bRecover ){
    /* Close and reopen the database. */
    testReopen(&pDb, &rc);
      testReopenRecover(&pDb, &rc);
    }else{
      testReopen(&pDb, &rc);
    }

    /* Check that the db content is still correct. */
    testDbContents(pDb, pData, p->nRow, 0, i-1, p->nTest, p->bTestScan, &rc);

    /* Update the progress dots... */
    testCaseProgress(i, p->nRow, testCaseNDot()/2, &iDot);
  }

  i = 0;
  iDot = 0;
  while( rc==LSM_OK && i<p->nRow ){

    /* Delete some entries */
    testDeleteDatasourceRange(pDb, pData, i, p->nVerify, &rc);
    i += p->nVerify;

    /* Check that the db content is correct. */
    testDbContents(pDb, pData, p->nRow, i, p->nRow-1,p->nTest,p->bTestScan,&rc);

    /* Close and reopen the database. */
    if( bRecover ){
      testReopenRecover(&pDb, &rc);
    }else{
    testReopen(&pDb, &rc);
      testReopen(&pDb, &rc);
    }

    /* Check that the db content is still correct. */
    testDbContents(pDb, pData, p->nRow, i, p->nRow-1,p->nTest,p->bTestScan,&rc);

    /* Update the progress dots... */
    testCaseProgress(i, p->nRow, testCaseNDot()/2, &iDot);
  }
︙
335
336
337
338
339
340
341

342


343
344
345
346
347
348







349
350
351
352
353
354
355

355
356
357
358
359
360
361
362
363
364
365






366
367
368
369
370
371
372
373
374
375
376
377
378
379








+

+
+
-
-
-
-
-
-
+
+
+
+
+
+
+








    { {DATA_SEQUENTIAL, 5,10,      1000,2000},     1000,  250, 1000, 1},
    { {DATA_SEQUENTIAL, 5,100,     10000,20000},    100,   25,  100, 1},
    { {DATA_RANDOM,     10,10,     100,100},     100000, 1000,  100, 0},
    { {DATA_SEQUENTIAL, 10,10,     100,100},     100000, 1000,  100, 0},
  };

  int i;
  int bRecover;

  for(bRecover=0; bRecover<2; bRecover++){
    if( bRecover==1 && memcmp(zSystem, "lsm", 3) ) break;
  for(i=0; *pRc==LSM_OK && i<ArraySize(aTest); i++){
    char *zName = getName(zSystem, &aTest[i]);
    if( testCaseBegin(pRc, zPattern, "%s", zName) ){
      doDataTest1(zSystem, &aTest[i], pRc);
    }
    testFree(zName);
    for(i=0; *pRc==LSM_OK && i<ArraySize(aTest); i++){
      char *zName = getName(zSystem, bRecover, &aTest[i]);
      if( testCaseBegin(pRc, zPattern, "%s", zName) ){
        doDataTest1(zSystem, bRecover, &aTest[i], pRc);
      }
      testFree(zName);
    }
  }
}

void testCompareDb(
  Datasource *pData,
  int nData,
  int iSeed,
︙
392
393
394
395
396
397
398

399
400
401
402
403
404
405

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430








+








    testDatasourceEntry(pData, i, &pKey, &nKey, 0, 0);
    testFetchCompare(pControl, pDb, pKey, nKey, pRc);
  }
}

static void doDataTest2(
  const char *zSystem,            /* Database system to test */
  int bRecover,
  Datatest2 *p,                   /* Structure containing test parameters */
  int *pRc                        /* OUT: Error code */
){
  TestDb *pDb;
  TestDb *pControl;
  Datasource *pData;
  int i;
︙
433
434
435
436
437
438
439



440


441
442
443
444
445
446
447
448
449
450
451
452
453
454

455
456
457
458
459


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

479


480
481
482
483
484
485







486
487
488
489
490
491
492

458
459
460
461
462
463
464
465
466
467

468
469
470
471
472
473
474
475
476
477
478
479
480
481
482

483
484
485
486


487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511






512
513
514
515
516
517
518
519
520
521
522
523
524
525








+
+
+
-
+
+













-
+



-
-
+
+



















+

+
+
-
-
-
-
-
-
+
+
+
+
+
+
+








    testDatasourceEntry(pData, i+2000000, &pKey2, &nKey2, 0, 0);

    testDeleteRange(pDb, pKey1, nKey1, pKey2, nKey2, &rc);
    testDeleteRange(pControl, pKey1, nKey1, pKey2, nKey2, &rc);
    testFree(pKey1);

    testCompareDb(pData, nRange, i, pControl, pDb, &rc);
    if( bRecover ){
      testReopenRecover(&pDb, &rc);
    }else{
    testReopen(&pDb, &rc);
      testReopen(&pDb, &rc);
    }
    testCompareDb(pData, nRange, i, pControl, pDb, &rc);

    /* Update the progress dots... */
    testCaseProgress(i, p->nIter, testCaseNDot(), &iDot);
  }

  testClose(&pDb);
  testClose(&pControl);
  testDatasourceFree(pData);
  testCaseFinish(rc);
  *pRc = rc;
}

static char *getName2(const char *zSystem, Datatest2 *pTest){
static char *getName2(const char *zSystem, int bRecover, Datatest2 *pTest){
  char *zRet;
  char *zData;
  zData = testDatasourceName(&pTest->defn);
  zRet = testMallocPrintf("data2.%s.%s.%d.%d.%d", 
      zSystem, zData, pTest->nRange, pTest->nWrite, pTest->nIter
  zRet = testMallocPrintf("data2.%s.%s.rec=%d.%d.%d.%d", 
      zSystem, zData, bRecover, pTest->nRange, pTest->nWrite, pTest->nIter
  );
  testFree(zData);
  return zRet;
}

void test_data_2(
  const char *zSystem,            /* Database system name */
  const char *zPattern,           /* Run test cases that match this pattern */
  int *pRc                        /* IN/OUT: Error code */
){
  Datatest2 aTest[] = {
      /* defn,                                 nRange, nWrite, nIter */
    { {DATA_RANDOM,     20,25,     100,200},   10000,  10,     50   },
    { {DATA_RANDOM,     20,25,     100,200},   10000,  200,    50   },
    { {DATA_RANDOM,     20,25,     100,200},   100,    10,     1000 },
    { {DATA_RANDOM,     20,25,     100,200},   100,    200,    50   },
  };

  int i;
  int bRecover;

  for(bRecover=0; bRecover<2; bRecover++){
    if( bRecover==1 && memcmp(zSystem, "lsm", 3) ) break;
  for(i=0; *pRc==LSM_OK && i<ArraySize(aTest); i++){
    char *zName = getName2(zSystem, &aTest[i]);
    if( testCaseBegin(pRc, zPattern, "%s", zName) ){
      doDataTest2(zSystem, &aTest[i], pRc);
    }
    testFree(zName);
    for(i=0; *pRc==LSM_OK && i<ArraySize(aTest); i++){
      char *zName = getName2(zSystem, bRecover, &aTest[i]);
      if( testCaseBegin(pRc, zPattern, "%s", zName) ){
        doDataTest2(zSystem, bRecover, &aTest[i], pRc);
      }
      testFree(zName);
    }
  }
}

/*************************************************************************
** Test case data3.*
*/
︙
Changes to ext/lsm1/lsm-test/lsmtest_main.c.
︙
22
23
24
25
26
27
28
29

30
31
32
33
34
35
36

22
23
24
25
26
27
28

29
30
31
32
33
34
35
36








-
+








  return; 
}

#define testSetError(rc) testSetErrorFunc(rc, pRc, __FILE__, __LINE__)
static void testSetErrorFunc(int rc, int *pRc, const char *zFile, int iLine){
  if( rc ){
    *pRc = rc;
    printf("FAILED (%s:%d) rc=%d ", zFile, iLine, rc);
    fprintf(stderr, "FAILED (%s:%d) rc=%d ", zFile, iLine, rc);
    test_failed();
  }
}

static int lsm_memcmp(u8 *a, u8 *b, int c){
  int i;
  for(i=0; i<c; i++){
︙
Changes to ext/lsm1/lsm-test/lsmtest_tdb.c.
︙
740
741
742
743
744
745
746








747
748
749
750
751
752
753

740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761








+
+
+
+
+
+
+
+








#endif
};

const char *tdb_system_name(int i){
  if( i<0 || i>=ArraySize(aLib) ) return 0;
  return aLib[i].zName;
}

const char *tdb_default_db(const char *zSys){
  int i;
  for(i=0; i<ArraySize(aLib); i++){
    if( strcmp(aLib[i].zName, zSys)==0 ) return aLib[i].zDefaultDb;
  }
  return 0;
}

int tdb_open(const char *zLib, const char *zDb, int bClear, TestDb **ppDb){
  int i;
  int rc = 1;
  const char *zSpec = 0;

  int nLib = 0;
︙
Changes to ext/lsm1/lsm-test/lsmtest_tdb.h.
︙
121
122
123
124
125
126
127

128
129
130
131
132
133
134

121
122
123
124
125
126
127
128
129
130
131
132
133
134
135








+








  int bReverse,                   /* True to scan in reverse order */
  void *pKey1, int nKey1,         /* Start of search */
  void *pKey2, int nKey2,         /* End of search */
  void (*xCallback)(void *pCtx, void *pKey, int nKey, void *pVal, int nVal)
);

const char *tdb_system_name(int i);
const char *tdb_default_db(const char *zSys);

int tdb_lsm_open(const char *zCfg, const char *zDb, int bClear, TestDb **ppDb);

/*
** If the TestDb handle passed as an argument is a wrapper around an LSM
** database, return the LSM handle. Otherwise, if the argument is some other
** database system, return NULL.
︙
Changes to ext/lsm1/lsmInt.h.
︙
860
861
862
863
864
865
866
867

868
869
870
871
872
873
874
875
876




877
878
879
880
881
882
883

860
861
862
863
864
865
866

867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887








-
+









+
+
+
+








void lsmLogMessage(lsm_db *, int, const char *, ...);
int lsmInfoFreelist(lsm_db *pDb, char **pzOut);

/*
** Functions from file "lsm_log.c".
*/
int lsmLogBegin(lsm_db *pDb);
int lsmLogWrite(lsm_db *, void *, int, void *, int);
int lsmLogWrite(lsm_db *, int, void *, int, void *, int);
int lsmLogCommit(lsm_db *);
void lsmLogEnd(lsm_db *pDb, int bCommit);
void lsmLogTell(lsm_db *, LogMark *);
void lsmLogSeek(lsm_db *, LogMark *);
void lsmLogClose(lsm_db *);

int lsmLogRecover(lsm_db *);
int lsmInfoLogStructure(lsm_db *pDb, char **pzVal);

/* Valid values for the second argument to lsmLogWrite(). */
#define LSM_WRITE        0x06
#define LSM_DELETE       0x08
#define LSM_DRANGE       0x0A

/**************************************************************************
** Functions from file "lsm_shared.c".
*/

int lsmDbDatabaseConnect(lsm_db*, const char *);
void lsmDbDatabaseRelease(lsm_db *);
︙
Changes to ext/lsm1/lsm_log.c.
︙
195
196
197
198
199
200
201

202
203



204
205
206
207
208
209
210

195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214








+


+
+
+








#define LSM_LOG_PAD1         0x01
#define LSM_LOG_PAD2         0x02
#define LSM_LOG_COMMIT       0x03
#define LSM_LOG_JUMP         0x04

#define LSM_LOG_WRITE        0x06
#define LSM_LOG_WRITE_CKSUM  0x07

#define LSM_LOG_DELETE       0x08
#define LSM_LOG_DELETE_CKSUM 0x09

#define LSM_LOG_DRANGE       0x0A
#define LSM_LOG_DRANGE_CKSUM 0x0B

/* Require a checksum every 32KB. */
#define LSM_CKSUM_MAXDATA (32*1024)

/* Do not wrap a log file smaller than this in bytes. */
#define LSM_MIN_LOGWRAP      (128*1024)
︙
649
650
651
652
653
654
655

656
657
658
659
660
661
662
663






664
665
666
667
668
669
670

671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689

690

691
692

693
694
695
696
697
698
699
700
701
702

703
704
705
706
707
708
709

653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680

681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701

702
703

704
705
706
707
708
709
710
711
712
713

714
715
716
717
718
719
720
721








+








+
+
+
+
+
+






-
+



















+
-
+

-
+









-
+









/*
** Append an LSM_LOG_WRITE (if nVal>=0) or LSM_LOG_DELETE (if nVal<0) 
** record to the database log.
*/
int lsmLogWrite(
  lsm_db *pDb,                    /* Database handle */
  int eType,
  void *pKey, int nKey,           /* Database key to write to log */
  void *pVal, int nVal            /* Database value (or nVal<0) to write */
){
  int rc = LSM_OK;
  LogWriter *pLog;                /* Log object to write to */
  int nReq;                       /* Bytes of space required in log */
  int bCksum = 0;                 /* True to embed a checksum in this record */

  assert( eType==LSM_WRITE || eType==LSM_DELETE || eType==LSM_DRANGE );
  assert( LSM_LOG_WRITE==LSM_WRITE );
  assert( LSM_LOG_DELETE==LSM_DELETE );
  assert( LSM_LOG_DRANGE==LSM_DRANGE );
  assert( (eType==LSM_LOG_DELETE)==(nVal<0) );

  if( pDb->bUseLog==0 ) return LSM_OK;
  pLog = pDb->pLogWriter;

  /* Determine how many bytes of space are required, assuming that a checksum
  ** will be embedded in this record (even though it may not be).  */
  nReq = 1 + lsmVarintLen32(nKey) + 8 + nKey;
  if( nVal>=0 ) nReq += lsmVarintLen32(nVal) + nVal;
  if( eType!=LSM_LOG_DELETE ) nReq += lsmVarintLen32(nVal) + nVal;

  /* Jump over the jump region if required. Set bCksum to true to tell the
  ** code below to include a checksum in the record if either (a) writing
  ** this record would mean that more than LSM_CKSUM_MAXDATA bytes of data
  ** have been written to the log since the last checksum, or (b) the jump
  ** is taken.  */
  rc = jumpIfRequired(pDb, pLog, nReq, &bCksum);
  if( (pLog->buf.n+nReq) > LSM_CKSUM_MAXDATA ) bCksum = 1;

  if( rc==LSM_OK ){
    rc = lsmStringExtend(&pLog->buf, nReq);
  }
  if( rc==LSM_OK ){
    u8 *a = (u8 *)&pLog->buf.z[pLog->buf.n];
    
    /* Write the record header - the type byte followed by either 1 (for
    ** DELETE) or 2 (for WRITE) varints.  */
    assert( LSM_LOG_WRITE_CKSUM == (LSM_LOG_WRITE | 0x0001) );
    assert( LSM_LOG_DELETE_CKSUM == (LSM_LOG_DELETE | 0x0001) );
    assert( LSM_LOG_DRANGE_CKSUM == (LSM_LOG_DRANGE | 0x0001) );
    *(a++) = (nVal>=0 ? LSM_LOG_WRITE : LSM_LOG_DELETE) | (u8)bCksum;
    *(a++) = (u8)eType | (u8)bCksum;
    a += lsmVarintPut32(a, nKey);
    if( nVal>=0 ) a += lsmVarintPut32(a, nVal);
    if( eType!=LSM_LOG_DELETE ) a += lsmVarintPut32(a, nVal);

    if( bCksum ){
      pLog->buf.n = (a - (u8 *)pLog->buf.z);
      rc = logCksumAndFlush(pDb);
      a = (u8 *)&pLog->buf.z[pLog->buf.n];
    }

    memcpy(a, pKey, nKey);
    a += nKey;
    if( nVal>=0 ){
    if( eType!=LSM_LOG_DELETE ){
      memcpy(a, pVal, nVal);
      a += nVal;
    }
    pLog->buf.n = a - (u8 *)pLog->buf.z;
    assert( pLog->buf.n<=pLog->buf.nAlloc );
  }
︙
1001
1002
1003
1004
1005
1006
1007


1008
1009
1010
1011
1012
1013
1014
1015
1016

1017
1018
1019
1020
1021
1022
1023
1024
1025

1026




1027
1028
1029
1030
1031
1032
1033

1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029

1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040

1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051








+
+








-
+









+
-
+
+
+
+








          case LSM_LOG_PAD2: {
            int nPad;
            logReaderVarint(&reader, &buf1, &nPad, &rc);
            logReaderBlob(&reader, &buf1, nPad, 0, &rc);
            break;
          }

          case LSM_LOG_DRANGE:
          case LSM_LOG_DRANGE_CKSUM:
          case LSM_LOG_WRITE:
          case LSM_LOG_WRITE_CKSUM: {
            int nKey;
            int nVal;
            u8 *aVal;
            logReaderVarint(&reader, &buf1, &nKey, &rc);
            logReaderVarint(&reader, &buf2, &nVal, &rc);

            if( eType==LSM_LOG_WRITE_CKSUM ){
            if( eType==LSM_LOG_WRITE_CKSUM || eType==LSM_LOG_DRANGE_CKSUM ){
              logReaderCksum(&reader, &buf1, &bEof, &rc);
            }else{
              bEof = logRequireCksum(&reader, nKey+nVal);
            }
            if( bEof ) break;

            logReaderBlob(&reader, &buf1, nKey, 0, &rc);
            logReaderBlob(&reader, &buf2, nVal, &aVal, &rc);
            if( iPass==1 && rc==LSM_OK ){ 
              if( eType==LSM_LOG_WRITE || eType==LSM_LOG_WRITE_CKSUM ){
              rc = lsmTreeInsert(pDb, (u8 *)buf1.z, nKey, aVal, nVal);
                rc = lsmTreeInsert(pDb, (u8 *)buf1.z, nKey, aVal, nVal);
              }else{
                rc = lsmTreeDelete(pDb, (u8 *)buf1.z, nKey, aVal, nVal);
              }
            }
            break;
          }

          case LSM_LOG_DELETE:
          case LSM_LOG_DELETE_CKSUM: {
            int nKey; u8 *aKey;
︙
Changes to ext/lsm1/lsm_main.c.
︙
665
666
667
668
669
670
671
672
673


674
675
676
677
678
679
680
681
682
683

665
666
667
668
669
670
671


672
673



674
675
676
677
678
679
680








-
-
+
+
-
-
-









  if( pDb->nTransOpen==0 ){
    bCommit = 1;
    rc = lsm_begin(pDb, 1);
  }

  if( rc==LSM_OK ){
    if( bDeleteRange==0 ){
      rc = lsmLogWrite(pDb, (void *)pKey, nKey, (void *)pVal, nVal);
    int eType = (bDeleteRange ? LSM_DRANGE : (nVal>=0?LSM_WRITE:LSM_DELETE));
    rc = lsmLogWrite(pDb, eType, (void *)pKey, nKey, (void *)pVal, nVal);
    }else{
      /* TODO */
    }
  }

  lsmSortedSaveTreeCursors(pDb);

  if( rc==LSM_OK ){
    int pgsz = lsmFsPageSize(pDb->pFS);
    int nQuant = LSM_AUTOWORK_QUANT * pgsz;
︙