1/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
2/*
3 *     Copyright 2010 Couchbase, Inc
4 *
5 *   Licensed under the Apache License, Version 2.0 (the "License");
6 *   you may not use this file except in compliance with the License.
7 *   You may obtain a copy of the License at
8 *
9 *       http://www.apache.org/licenses/LICENSE-2.0
10 *
11 *   Unless required by applicable law or agreed to in writing, software
12 *   distributed under the License is distributed on an "AS IS" BASIS,
13 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 *   See the License for the specific language governing permissions and
15 *   limitations under the License.
16 */
17
18#include <stdio.h>
19#include <stdlib.h>
20#include <string.h>
21#include <stdint.h>
22#include <time.h>
23#if !defined(WIN32) && !defined(_WIN32)
24#include <unistd.h>
25#include <sys/types.h>
26#include <sys/stat.h>
27#endif
28
29#include "libforestdb/forestdb.h"
30#include "test.h"
31
32#include "internal_types.h"
33#include "wal.h"
34#include "functional_util.h"
35
36struct cb_args {
37    int n_moved_docs;
38    int n_batch_move;
39    bool begin;
40    bool end;
41    bool wal_flush;
42    fdb_kvs_handle *handle;
43};
44
45static fdb_compact_decision compaction_cb(fdb_file_handle *fhandle,
46                            fdb_compaction_status status, const char *kv_name,
47                            fdb_doc *doc, uint64_t old_offset,
48                            uint64_t new_offset,
49                            void *ctx)
50{
51    TEST_INIT();
52    fdb_doc *rdoc;
53    fdb_status s;
54    fdb_compact_decision ret = FDB_CS_KEEP_DOC;
55    struct cb_args *args = (struct cb_args *)ctx;
56
57    (void) doc;
58    (void) new_offset;
59
60    if (status == FDB_CS_BEGIN) {
61        args->begin = true;
62    } else if (status == FDB_CS_END) {
63        args->end = true;
64    } else if (status == FDB_CS_FLUSH_WAL) {
65        args->wal_flush = true;
66    } else if (status == FDB_CS_MOVE_DOC) {
67        if (doc->deleted) {
68            ret = FDB_CS_DROP_DOC;
69        }
70
71        args->n_moved_docs++;
72        fdb_doc_create(&rdoc, NULL, 0, NULL, 0, NULL, 0);
73        rdoc->offset = old_offset;
74        s = fdb_get_byoffset(args->handle, rdoc);
75        TEST_CHK(s == FDB_RESULT_SUCCESS);
76        fdb_doc_free(rdoc);
77
78        if (fhandle->root->config.multi_kv_instances) {
79            TEST_CMP(kv_name, "db", 2);
80        } else {
81            TEST_CMP(kv_name, "default", 7);
82        }
83    } else { // FDB_CS_BATCH_MOVE
84        args->n_batch_move++;
85        fdb_doc_create(&rdoc, NULL, 0, NULL, 0, NULL, 0);
86        rdoc->offset = old_offset;
87        s = fdb_get_byoffset(args->handle, rdoc);
88        TEST_CHK (s == FDB_RESULT_SUCCESS);
89        fdb_doc_free(rdoc);
90    }
91    return ret;
92}
93
94void compaction_callback_test(bool multi_kv)
95{
96    TEST_INIT();
97    memleak_start();
98
99    int i, r;
100    int n = 1000;
101    char keybuf[256], bodybuf[256];
102    fdb_file_handle *dbfile;
103    fdb_kvs_handle *db;
104    fdb_status s;
105    fdb_config fconfig = fdb_get_default_config();
106    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
107    struct cb_args cb_args;
108
109    memset(&cb_args, 0x0, sizeof(struct cb_args));
110    fconfig.wal_threshold = 1024;
111    fconfig.flags = FDB_OPEN_FLAG_CREATE;
112    fconfig.compaction_cb = compaction_cb;
113    fconfig.compaction_cb_ctx = &cb_args;
114    fconfig.compaction_cb_mask = FDB_CS_BEGIN |
115                                 FDB_CS_MOVE_DOC |
116                                 FDB_CS_FLUSH_WAL |
117                                 FDB_CS_END;
118    fconfig.multi_kv_instances = multi_kv;
119
120    // remove all previous compact_test files
121    r = system(SHELL_DEL" compact_test* > errorlog.txt");
122    (void)r;
123
124    // open db
125    fdb_open(&dbfile, "./compact_test1", &fconfig);
126    if (multi_kv) {
127        fdb_kvs_open(dbfile, &db, "db", &kvs_config);
128    } else {
129        fdb_kvs_open_default(dbfile, &db, &kvs_config);
130    }
131    cb_args.handle = db;
132
133    // write docs
134    for (i=0;i<n;++i){
135        sprintf(keybuf, "key%04d", i);
136        sprintf(bodybuf, "body%04d", i);
137        s = fdb_set_kv(db, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
138        TEST_CHK(s == FDB_RESULT_SUCCESS);
139    }
140    s = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
141    TEST_CHK(s == FDB_RESULT_SUCCESS);
142    s = fdb_compact(dbfile, "./compact_test2");
143    TEST_CHK(s == FDB_RESULT_SUCCESS);
144
145    TEST_CHK(cb_args.n_moved_docs == n);
146    TEST_CHK(cb_args.begin);
147    TEST_CHK(cb_args.end);
148    TEST_CHK(cb_args.wal_flush);
149    fdb_close(dbfile);
150
151    // open db without move doc
152    memset(&cb_args, 0x0, sizeof(struct cb_args));
153    fconfig.compaction_cb_mask = FDB_CS_BEGIN |
154                                 FDB_CS_FLUSH_WAL |
155                                 FDB_CS_END;
156    fdb_open(&dbfile, "./compact_test2", &fconfig);
157
158    if (multi_kv) {
159        fdb_kvs_open(dbfile, &db, "db", &kvs_config);
160    } else {
161        fdb_kvs_open_default(dbfile, &db, &kvs_config);
162    }
163    cb_args.handle = db;
164    s = fdb_compact(dbfile, "./compact_test3");
165    TEST_CHK(s == FDB_RESULT_SUCCESS);
166    TEST_CHK(cb_args.n_moved_docs == 0);
167    TEST_CHK(cb_args.begin);
168    TEST_CHK(cb_args.end);
169    TEST_CHK(cb_args.wal_flush);
170    fdb_close(dbfile);
171
172    // open db without wal_flush
173    memset(&cb_args, 0x0, sizeof(struct cb_args));
174    fconfig.compaction_cb_mask = FDB_CS_BEGIN |
175                                 FDB_CS_MOVE_DOC |
176                                 FDB_CS_END;
177    fdb_open(&dbfile, "./compact_test3", &fconfig);
178    if (multi_kv) {
179        fdb_kvs_open(dbfile, &db, "db", &kvs_config);
180    } else {
181        fdb_kvs_open_default(dbfile, &db, &kvs_config);
182    }
183    cb_args.handle = db;
184    s = fdb_compact(dbfile, "./compact_test4");
185    TEST_CHK(s == FDB_RESULT_SUCCESS);
186    TEST_CHK(cb_args.n_moved_docs == n);
187    TEST_CHK(cb_args.begin);
188    TEST_CHK(cb_args.end);
189    TEST_CHK(!cb_args.wal_flush);
190    fdb_close(dbfile);
191
192    // open db without begin/end
193    memset(&cb_args, 0x0, sizeof(struct cb_args));
194    fconfig.compaction_cb_mask = FDB_CS_MOVE_DOC |
195                                 FDB_CS_FLUSH_WAL;
196    fdb_open(&dbfile, "./compact_test4", &fconfig);
197    if (multi_kv) {
198        fdb_kvs_open(dbfile, &db, "db", &kvs_config);
199    } else {
200        fdb_kvs_open_default(dbfile, &db, &kvs_config);
201    }
202    cb_args.handle = db;
203    s = fdb_compact(dbfile, "./compact_test5");
204    TEST_CHK(s == FDB_RESULT_SUCCESS);
205    TEST_CHK(cb_args.n_moved_docs == n);
206    TEST_CHK(!cb_args.begin);
207    TEST_CHK(!cb_args.end);
208    TEST_CHK(cb_args.wal_flush);
209    fdb_close(dbfile);
210
211    // open db with batch move
212    memset(&cb_args, 0x0, sizeof(struct cb_args));
213    fconfig.compaction_cb_mask = FDB_CS_BATCH_MOVE;
214    fdb_open(&dbfile, "./compact_test5", &fconfig);
215    if (multi_kv) {
216        fdb_kvs_open(dbfile, &db, "db", &kvs_config);
217    } else {
218        fdb_kvs_open_default(dbfile, &db, &kvs_config);
219    }
220    cb_args.handle = db;
221    s = fdb_compact(dbfile, "./compact_test6");
222    TEST_CHK(s == FDB_RESULT_SUCCESS);
223    TEST_CHK(cb_args.n_moved_docs == 0);
224    TEST_CHK(cb_args.n_batch_move && cb_args.n_batch_move <= n);
225    TEST_CHK(!cb_args.begin);
226    TEST_CHK(!cb_args.end);
227    TEST_CHK(!cb_args.wal_flush);
228    fdb_close(dbfile);
229
230    fdb_shutdown();
231
232    memleak_end();
233    if (multi_kv) {
234        TEST_RESULT("compaction callback function multi kv mode test");
235    } else {
236        TEST_RESULT("compaction callback function single kv mode test");
237    }
238}
239
240void compact_wo_reopen_test()
241{
242    TEST_INIT();
243
244    memleak_start();
245
246    int i, r;
247    int n = 3;
248    fdb_file_handle *dbfile, *dbfile_new;
249    fdb_kvs_handle *db;
250    fdb_kvs_handle *db_new;
251    fdb_doc **doc = alca(fdb_doc*, n);
252    fdb_doc *rdoc;
253    fdb_status status;
254
255    char keybuf[256], metabuf[256], bodybuf[256];
256
257    // remove previous compact_test files
258    r = system(SHELL_DEL" compact_test* > errorlog.txt");
259    (void)r;
260
261    fdb_config fconfig = fdb_get_default_config();
262    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
263    fconfig.buffercache_size = 16777216;
264    fconfig.wal_threshold = 1024;
265    fconfig.flags = FDB_OPEN_FLAG_CREATE;
266    fconfig.compaction_threshold = 0;
267
268    // open db
269    fdb_open(&dbfile, "./compact_test1", &fconfig);
270    fdb_kvs_open_default(dbfile, &db, &kvs_config);
271    status = fdb_set_log_callback(db, logCallbackFunc,
272                                  (void *) "compact_wo_reopen_test");
273    TEST_CHK(status == FDB_RESULT_SUCCESS);
274    fdb_open(&dbfile_new, "./compact_test1", &fconfig);
275    fdb_kvs_open_default(dbfile_new, &db_new, &kvs_config);
276    status = fdb_set_log_callback(db_new, logCallbackFunc,
277                                  (void *) "compact_wo_reopen_test");
278    TEST_CHK(status == FDB_RESULT_SUCCESS);
279
280    // insert documents
281    for (i=0;i<n;++i){
282        sprintf(keybuf, "key%d", i);
283        sprintf(metabuf, "meta%d", i);
284        sprintf(bodybuf, "body%d", i);
285        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
286            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
287        fdb_set(db, doc[i]);
288    }
289
290    // remove doc
291    fdb_doc_create(&rdoc, doc[1]->key, doc[1]->keylen, doc[1]->meta, doc[1]->metalen, NULL, 0);
292    rdoc->deleted = true;
293    fdb_set(db, rdoc);
294    fdb_doc_free(rdoc);
295
296    // commit
297    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
298
299    // perform compaction using one handle
300    fdb_compact(dbfile, (char *) "./compact_test2");
301
302    // retrieve documents using the other handle without close/re-open
303    for (i=0;i<n;++i){
304        // search by key
305        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
306        status = fdb_get(db_new, rdoc);
307
308        if (i != 1) {
309            TEST_CHK(status == FDB_RESULT_SUCCESS);
310            TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
311            TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
312        }else{
313            TEST_CHK(status == FDB_RESULT_KEY_NOT_FOUND);
314        }
315
316        // free result document
317        fdb_doc_free(rdoc);
318    }
319    // check the other handle's filename
320    fdb_file_info info;
321    fdb_get_file_info(dbfile_new, &info);
322    TEST_CHK(!strcmp("./compact_test2", info.filename));
323
324    // free all documents
325    for (i=0;i<n;++i){
326        fdb_doc_free(doc[i]);
327    }
328
329    // close db file
330    fdb_kvs_close(db);
331    fdb_kvs_close(db_new);
332    fdb_close(dbfile);
333    fdb_close(dbfile_new);
334
335    // free all resources
336    fdb_shutdown();
337
338    memleak_end();
339
340    TEST_RESULT("compaction without reopen test");
341}
342
343void compact_with_reopen_test()
344{
345    TEST_INIT();
346
347    memleak_start();
348
349    int i, r;
350    int n = 100;
351    fdb_file_handle *dbfile;
352    fdb_kvs_handle *db;
353    fdb_doc **doc = alca(fdb_doc*, n);
354    fdb_doc *rdoc;
355    fdb_status status;
356
357    char keybuf[256], metabuf[256], bodybuf[256], temp[256];
358
359    // remove previous compact_test files
360    r = system(SHELL_DEL" compact_test* > errorlog.txt");
361    (void)r;
362
363    fdb_config fconfig = fdb_get_default_config();
364    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
365    fconfig.buffercache_size = 16777216;
366    fconfig.wal_threshold = 1024;
367    fconfig.flags = FDB_OPEN_FLAG_CREATE;
368    fconfig.compaction_threshold = 0;
369
370    // open db
371    fdb_open(&dbfile, "./compact_test1", &fconfig);
372    fdb_kvs_open_default(dbfile, &db, &kvs_config);
373    status = fdb_set_log_callback(db, logCallbackFunc,
374                                  (void *) "compact_with_reopen_test");
375    TEST_CHK(status == FDB_RESULT_SUCCESS);
376
377    // insert documents
378    for (i=0;i<n;++i){
379        sprintf(keybuf, "key%d", i);
380        sprintf(metabuf, "meta%d", i);
381        sprintf(bodybuf, "body%d", i);
382        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
383            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
384        fdb_set(db, doc[i]);
385    }
386
387    // remove doc
388    fdb_doc_create(&rdoc, doc[1]->key, doc[1]->keylen, doc[1]->meta, doc[1]->metalen, NULL, 0);
389    rdoc->deleted = true;
390    fdb_set(db, rdoc);
391    fdb_doc_free(rdoc);
392
393    // commit
394    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
395
396    // perform compaction using one handle
397    fdb_compact(dbfile, (char *) "./compact_test2");
398
399    // close db file
400    fdb_kvs_close(db);
401    fdb_close(dbfile);
402
403    r = system(SHELL_MOVE " compact_test2 compact_test1 > errorlog.txt");
404    (void)r;
405    fdb_open(&dbfile, "./compact_test1", &fconfig);
406    fdb_kvs_open_default(dbfile, &db, &kvs_config);
407    status = fdb_set_log_callback(db, logCallbackFunc,
408                                  (void *) "compact_with_reopen_test");
409    TEST_CHK(status == FDB_RESULT_SUCCESS);
410
411    // retrieve documents using the other handle without close/re-open
412    for (i=0;i<n;++i){
413        // search by key
414        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
415        status = fdb_get(db, rdoc);
416
417        if (i != 1) {
418            TEST_CHK(status == FDB_RESULT_SUCCESS);
419            TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
420            TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
421        }else{
422            TEST_CHK(status == FDB_RESULT_KEY_NOT_FOUND);
423        }
424
425        // free result document
426        fdb_doc_free(rdoc);
427    }
428    // check the other handle's filename
429    fdb_file_info info;
430    fdb_get_file_info(dbfile, &info);
431    TEST_CHK(!strcmp("./compact_test1", info.filename));
432
433    // update documents
434    for (i=0;i<n;++i){
435        sprintf(metabuf, "newmeta%d", i);
436        sprintf(bodybuf, "newbody%d_%s", i, temp);
437        fdb_doc_update(&doc[i], (void *)metabuf, strlen(metabuf),
438                       (void *)bodybuf, strlen(bodybuf));
439        fdb_set(db, doc[i]);
440    }
441
442    // Open the database with another handle.
443    fdb_file_handle *second_dbfile;
444    fdb_kvs_handle *second_dbh;
445    fdb_open(&second_dbfile, "./compact_test1", &fconfig);
446    fdb_kvs_open_default(second_dbfile, &second_dbh, &kvs_config);
447    status = fdb_set_log_callback(second_dbh, logCallbackFunc,
448                                  (void *) "compact_with_reopen_test");
449    TEST_CHK(status == FDB_RESULT_SUCCESS);
450    // In-place compactions with a handle still open on the first old file
451    status = fdb_compact(dbfile, NULL);
452    TEST_CHK(status == FDB_RESULT_SUCCESS);
453
454    status = fdb_compact(dbfile, NULL);
455    TEST_CHK(status == FDB_RESULT_SUCCESS);
456
457    // MB-12977: retest compaction again..
458    status = fdb_compact(dbfile, NULL);
459    TEST_CHK(status == FDB_RESULT_SUCCESS);
460
461    fdb_kvs_close(db);
462    fdb_close(dbfile);
463
464    for (i=0;i<n;++i){
465        // search by key
466        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
467        status = fdb_get(second_dbh, rdoc);
468        TEST_CHK(status == FDB_RESULT_SUCCESS);
469        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
470        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
471        // free result document
472        fdb_doc_free(rdoc);
473    }
474
475    // Open database with an original name.
476    status = fdb_open(&dbfile, "./compact_test1", &fconfig);
477    TEST_CHK(status == FDB_RESULT_SUCCESS);
478    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
479    TEST_CHK(status == FDB_RESULT_SUCCESS);
480    status = fdb_set_log_callback(db, logCallbackFunc,
481                                  (void *) "compact_with_reopen_test");
482    TEST_CHK(status == FDB_RESULT_SUCCESS);
483    fdb_get_file_info(dbfile, &info);
484    // The actual file name should be a compacted one.
485    TEST_CHK(!strcmp("./compact_test1.3", info.filename));
486
487    fdb_kvs_close(second_dbh);
488    fdb_close(second_dbfile);
489
490    for (i=0;i<n;++i){
491        // search by key
492        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
493        status = fdb_get(db, rdoc);
494        TEST_CHK(status == FDB_RESULT_SUCCESS);
495        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
496        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
497        // free result document
498        fdb_doc_free(rdoc);
499    }
500
501    fdb_compact(dbfile, NULL);
502    fdb_kvs_close(db);
503    fdb_close(dbfile);
504
505    r = system(SHELL_MOVE " compact_test1 compact_test.fdb > errorlog.txt");
506    (void)r;
507    fdb_open(&dbfile, "./compact_test.fdb", &fconfig);
508    fdb_kvs_open_default(dbfile, &db, &kvs_config);
509    status = fdb_set_log_callback(db, logCallbackFunc,
510                                  (void *) "compact_with_reopen_test");
511    TEST_CHK(status == FDB_RESULT_SUCCESS);
512    // In-place compaction
513    fdb_compact(dbfile, NULL);
514    fdb_kvs_close(db);
515    fdb_close(dbfile);
516    // Open database with an original name.
517    status = fdb_open(&dbfile, "./compact_test.fdb", &fconfig);
518    TEST_CHK(status == FDB_RESULT_SUCCESS);
519    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
520    TEST_CHK(status == FDB_RESULT_SUCCESS);
521    status = fdb_set_log_callback(db, logCallbackFunc,
522                                  (void *) "compact_with_reopen_test");
523    TEST_CHK(status == FDB_RESULT_SUCCESS);
524    fdb_get_file_info(dbfile, &info);
525    TEST_CHK(!strcmp("./compact_test.fdb", info.filename));
526    TEST_CHK(info.doc_count == 100);
527
528    // free all documents
529    for (i=0;i<n;++i){
530        fdb_doc_free(doc[i]);
531    }
532
533    fdb_kvs_close(db);
534    fdb_close(dbfile);
535
536    // free all resources
537    fdb_shutdown();
538
539    memleak_end();
540
541    TEST_RESULT("compaction with reopen test");
542}
543
544void compact_reopen_named_kvs()
545{
546    TEST_INIT();
547
548    memleak_start();
549
550    int i, r;
551    int n = 100;
552    int nkvdocs;
553    fdb_file_handle *dbfile;
554    fdb_kvs_handle *db;
555    fdb_doc **doc = alca(fdb_doc*, n);
556    fdb_status status;
557    fdb_kvs_info kvs_info;
558
559    char keybuf[256], metabuf[256], bodybuf[256];
560
561    // remove previous compact_test files
562    r = system(SHELL_DEL" compact_test* > errorlog.txt");
563    (void)r;
564
565    fdb_config fconfig = fdb_get_default_config();
566    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
567    fconfig.wal_threshold = 1024;
568    fconfig.flags = FDB_OPEN_FLAG_CREATE;
569    fconfig.compaction_threshold = 0;
570
571    // open db
572    fdb_open(&dbfile, "./compact_test1", &fconfig);
573    fdb_kvs_open(dbfile, &db, "db",  &kvs_config);
574
575    status = fdb_set_log_callback(db, logCallbackFunc,
576                                  (void *) "compact_reopen_named_kvs");
577    TEST_CHK(status == FDB_RESULT_SUCCESS);
578
579    for (i=0;i<n;++i){
580        sprintf(keybuf, "key%d", i);
581        sprintf(metabuf, "meta%d", i);
582        sprintf(bodybuf, "body%d", i);
583        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
584            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
585        fdb_set(db, doc[i]);
586    }
587
588    // commit
589    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
590
591    // compact
592    fdb_compact(dbfile, NULL);
593
594    // save ndocs
595    fdb_get_kvs_info(db, &kvs_info);
596    nkvdocs = kvs_info.doc_count;
597
598    // close db file
599    fdb_kvs_close(db);
600    fdb_close(dbfile);
601
602    // reopen
603    fdb_open(&dbfile, "./compact_test1", &fconfig);
604    fdb_kvs_open(dbfile, &db, "db",  &kvs_config);
605
606    // verify kvs stats
607    fdb_get_kvs_info(db, &kvs_info);
608    TEST_CHK((uint64_t)nkvdocs == kvs_info.doc_count);
609
610    fdb_kvs_close(db);
611    fdb_close(dbfile);
612
613    // free all documents
614    for (i=0;i<n;++i){
615        fdb_doc_free(doc[i]);
616    }
617    fdb_shutdown();
618
619    memleak_end();
620
621    TEST_RESULT("compact reopen named kvs");
622}
623
624void compact_reopen_with_iterator()
625{
626    TEST_INIT();
627
628    memleak_start();
629
630    int i, r;
631    int n = 9;
632    int count = 0;
633    int nkvdocs;
634    fdb_file_handle *dbfile, *compact_file;
635    fdb_kvs_handle *db;
636    fdb_doc **doc = alca(fdb_doc*, n);
637    fdb_doc *rdoc = NULL;
638    fdb_status status;
639    fdb_kvs_info kvs_info;
640    fdb_iterator *iterator;
641
642    char keybuf[256], metabuf[256], bodybuf[256];
643
644    // remove previous compact_test files
645    r = system(SHELL_DEL" compact_test* > errorlog.txt");
646    (void)r;
647
648    fdb_config fconfig = fdb_get_default_config();
649    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
650    fconfig.wal_threshold = 1024;
651    fconfig.flags = FDB_OPEN_FLAG_CREATE;
652    fconfig.compaction_threshold = 0;
653
654    // open db
655    fdb_open(&dbfile, "./compact_test1", &fconfig);
656    fdb_kvs_open(dbfile, &db, "db",  &kvs_config);
657
658    status = fdb_set_log_callback(db, logCallbackFunc,
659                                  (void *) "compact_reopen_with_iterator");
660    TEST_CHK(status == FDB_RESULT_SUCCESS);
661
662    for (i=0;i<n;++i){
663        sprintf(keybuf, "key%d", i);
664        sprintf(metabuf, "meta%d", i);
665        sprintf(bodybuf, "body%d", i);
666        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
667            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
668        fdb_set(db, doc[i]);
669    }
670
671    // commit
672    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
673
674    // MB-13859: compact the file using a separate handle..
675    fdb_open(&compact_file, "./compact_test1", &fconfig);
676    // compact
677    status = fdb_compact(compact_file, "./compact_test2");
678    TEST_CHK(status == FDB_RESULT_SUCCESS);
679    // close file after compaction to make the new_file's ref count 0
680    fdb_close(compact_file);
681
682    i = 0;
683    status = fdb_iterator_init(db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
684    TEST_CHK(status == FDB_RESULT_SUCCESS);
685    do {
686        status = fdb_iterator_get(iterator, &rdoc);
687        TEST_CHK(status == FDB_RESULT_SUCCESS);
688
689        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
690        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
691        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
692
693        fdb_doc_free(rdoc);
694        rdoc = NULL;
695        i++;
696        count++;
697    } while (fdb_iterator_next(iterator) != FDB_RESULT_ITERATOR_FAIL);
698
699    TEST_CHK(count==n);
700
701
702    status = fdb_iterator_close(iterator);
703    TEST_CHK(status == FDB_RESULT_SUCCESS);
704
705    // save ndocs
706    fdb_get_kvs_info(db, &kvs_info);
707    nkvdocs = kvs_info.doc_count;
708
709    // close db file
710    fdb_kvs_close(db);
711    fdb_close(dbfile);
712
713    // reopen
714    fdb_open(&dbfile, "./compact_test2", &fconfig);
715    fdb_kvs_open(dbfile, &db, "db",  &kvs_config);
716
717    // verify kvs stats
718    fdb_get_kvs_info(db, &kvs_info);
719    TEST_CHK((uint64_t)nkvdocs == kvs_info.doc_count);
720
721    fdb_kvs_close(db);
722    fdb_close(dbfile);
723
724    // free all documents
725    for (i=0;i<n;++i){
726        fdb_doc_free(doc[i]);
727    }
728    fdb_shutdown();
729
730    memleak_end();
731
732    TEST_RESULT("compact reopen with iterator");
733}
734
735void estimate_space_upto_test(bool multi_kv)
736{
737    TEST_INIT();
738
739    memleak_start();
740
741    int i, j, r;
742    int n = 300;
743    int num_kvs = 4; // keep this the same as number of fdb_commit() calls
744    fdb_file_handle *dbfile;
745    fdb_kvs_handle **db = alca(fdb_kvs_handle *, num_kvs);
746    fdb_doc **doc = alca(fdb_doc*, n);
747    fdb_status status;
748    fdb_snapshot_info_t *markers;
749    uint64_t num_markers;
750    size_t space_used;
751
752    char keybuf[256], metabuf[256], bodybuf[256];
753    char kv_name[8];
754
755    fdb_config fconfig = fdb_get_default_config();
756    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
757    fconfig.buffercache_size = 0;
758    fconfig.wal_threshold = 50;
759    fconfig.flags = FDB_OPEN_FLAG_CREATE;
760    fconfig.compaction_threshold = 0;
761    fconfig.multi_kv_instances = multi_kv;
762
763    // remove previous compact_test files
764    r = system(SHELL_DEL" compact_test* > errorlog.txt");
765    (void)r;
766
767    // open db
768    fdb_open(&dbfile, "./compact_test1", &fconfig);
769    if (multi_kv) {
770        for (r = 0; r < num_kvs; ++r) {
771            sprintf(kv_name, "kv%d", r);
772            fdb_kvs_open(dbfile, &db[r], kv_name, &kvs_config);
773        }
774    } else {
775        num_kvs = 1;
776        fdb_kvs_open_default(dbfile, &db[0], &kvs_config);
777    }
778
779   // ------- Setup test ----------------------------------
780   // insert first quarter of documents
781    for (i=0; i<n/4; i++){
782        sprintf(keybuf, "key%d", i);
783        sprintf(metabuf, "meta%d", i);
784        sprintf(bodybuf, "body%d", i);
785        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
786            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
787        for (r = 0; r < num_kvs; ++r) {
788            fdb_set(db[r], doc[i]);
789        }
790    }
791
792    // commit with a manual WAL flush (these docs go into HB-trie)
793    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
794
795    // insert second quarter of documents
796    for (; i < n/2; i++){
797        sprintf(keybuf, "key%d", i);
798        sprintf(metabuf, "meta%d", i);
799        sprintf(bodybuf, "body%d", i);
800        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
801            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
802        for (r = 0; r < num_kvs; ++r) {
803            fdb_set(db[r], doc[i]);
804        }
805    }
806    // Update first quarter of documents again..
807    for (j = 0; j < n/4; j++){
808        for (r = 0; r < num_kvs; ++r) {
809            fdb_set(db[r], doc[j]);
810        }
811    }
812    // Update first quarter of documents again overwriting previous update..
813    for (j = 0; j < n/4; j++){
814        for (r = 0; r < num_kvs; ++r) {
815            fdb_set(db[r], doc[j]);
816        }
817    }
818
819    // commit again without a WAL flush (some of these docs remain in WAL)
820    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
821
822    // insert third quarter of documents
823    for (; i < (n/2 + n/4); i++){
824        sprintf(keybuf, "key%d", i);
825        sprintf(metabuf, "meta%d", i);
826        sprintf(bodybuf, "body%d", i);
827        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
828            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
829        for (r = 0; r < num_kvs; ++r) {
830            fdb_set(db[r], doc[i]);
831        }
832    }
833    // manually flush WAL & commit (these docs go into HB-trie)
834    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
835
836    // insert fourth quarter of documents
837    for (; i < n; i++){
838        sprintf(keybuf, "key%d", i);
839        sprintf(metabuf, "meta%d", i);
840        sprintf(bodybuf, "body%d", i);
841        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
842            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
843        for (r = 0; r < num_kvs; ++r) {
844            fdb_set(db[r], doc[i]);
845        }
846    }
847    // commit without a WAL flush (some of these docs remain in the WAL)
848    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
849
850    for (r = 0; r < num_kvs; ++r) {
851        status = fdb_set_log_callback(db[r], logCallbackFunc,
852                                      (void *) "estimate_space_upto_test");
853        TEST_CHK(status == FDB_RESULT_SUCCESS);
854    }
855
856    status = fdb_get_all_snap_markers(dbfile, &markers, &num_markers);
857    TEST_CHK(status == FDB_RESULT_SUCCESS);
858
859    if (!multi_kv) {
860        size_t space_used2;
861        TEST_CHK(num_markers == 4);
862        space_used = fdb_estimate_space_used_from(dbfile, markers[1].marker);
863        space_used2 = fdb_estimate_space_used_from(dbfile, markers[2].marker);
864        TEST_CHK(space_used2 > space_used); // greater than space used by just 1
865    } else {
866        size_t space_used2;
867        TEST_CHK(num_markers == 8);
868        space_used = fdb_estimate_space_used_from(dbfile, markers[1].marker);
869        space_used2 = fdb_estimate_space_used_from(dbfile, markers[2].marker);
870        TEST_CHK(space_used2 > space_used); // greater than space used by just 1
871    }
872
873    status = fdb_free_snap_markers(markers, num_markers);
874    TEST_CHK(status == FDB_RESULT_SUCCESS);
875
876    // close db file
877    fdb_close(dbfile);
878
879    // free all documents
880    for (i=0;i<n;++i){
881        fdb_doc_free(doc[i]);
882    }
883
884    // free all resources
885    fdb_shutdown();
886
887    memleak_end();
888
889    sprintf(bodybuf, "estimate space upto marker in file test %s", multi_kv ?
890                                                           "multiple kv mode:"
891                                                         : "single kv mode:");
892    TEST_RESULT(bodybuf);
893}
894
895void compact_upto_test(bool multi_kv)
896{
897    TEST_INIT();
898
899    memleak_start();
900
901    int i, r;
902    int n = 20;
903    int num_kvs = 4; // keep this the same as number of fdb_commit() calls
904    fdb_file_handle *dbfile;
905    fdb_kvs_handle **db = alca(fdb_kvs_handle *, num_kvs);
906    fdb_doc **doc = alca(fdb_doc*, n);
907    fdb_status status;
908    fdb_snapshot_info_t *markers;
909    fdb_kvs_handle *snapshot;
910    uint64_t num_markers;
911
912    char keybuf[256], metabuf[256], bodybuf[256];
913    char kv_name[8];
914    char compact_filename[32];
915
916    fdb_config fconfig = fdb_get_default_config();
917    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
918    fconfig.buffercache_size = 0;
919    fconfig.wal_threshold = 1024;
920    fconfig.flags = FDB_OPEN_FLAG_CREATE;
921    fconfig.compaction_threshold = 0;
922    fconfig.multi_kv_instances = multi_kv;
923    // since this test requires static number of markers,
924    // disable block reusing
925    fconfig.block_reusing_threshold = 0;
926
927    // remove previous compact_test files
928    r = system(SHELL_DEL" compact_test* > errorlog.txt");
929    (void)r;
930
931    // open db
932    fdb_open(&dbfile, "./compact_test1", &fconfig);
933    if (multi_kv) {
934        for (r = 0; r < num_kvs; ++r) {
935            sprintf(kv_name, "kv%d", r);
936            fdb_kvs_open(dbfile, &db[r], kv_name, &kvs_config);
937        }
938    } else {
939        num_kvs = 1;
940        fdb_kvs_open_default(dbfile, &db[0], &kvs_config);
941    }
942
943   // ------- Setup test ----------------------------------
944   // insert documents of 0-4
945    for (i=0; i<n/4; i++){
946        sprintf(keybuf, "key%d", i);
947        sprintf(metabuf, "meta%d", i);
948        sprintf(bodybuf, "body%d", i);
949        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
950            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
951        for (r = 0; r < num_kvs; ++r) {
952            fdb_set(db[r], doc[i]);
953        }
954    }
955
956    // commit with a manual WAL flush (these docs go into HB-trie)
957    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
958
959    // insert documents from 5 - 9
960    for (; i < n/2; i++){
961        sprintf(keybuf, "key%d", i);
962        sprintf(metabuf, "meta%d", i);
963        sprintf(bodybuf, "body%d", i);
964        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
965            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
966        for (r = 0; r < num_kvs; ++r) {
967            fdb_set(db[r], doc[i]);
968        }
969    }
970
971    // commit again without a WAL flush
972    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
973
974    // insert documents from 10-14 into HB-trie
975    for (; i < (n/2 + n/4); i++){
976        sprintf(keybuf, "key%d", i);
977        sprintf(metabuf, "meta%d", i);
978        sprintf(bodybuf, "body%d", i);
979        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
980            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
981        for (r = 0; r < num_kvs; ++r) {
982            fdb_set(db[r], doc[i]);
983        }
984    }
985    // manually flush WAL & commit
986    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
987
988    // insert documents from 15 - 19 on file into the WAL
989    for (; i < n; i++){
990        sprintf(keybuf, "key%d", i);
991        sprintf(metabuf, "meta%d", i);
992        sprintf(bodybuf, "body%d", i);
993        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
994            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
995        for (r = 0; r < num_kvs; ++r) {
996            fdb_set(db[r], doc[i]);
997        }
998    }
999    // commit without a WAL flush
1000    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1001
1002    for (r = 0; r < num_kvs; ++r) {
1003        status = fdb_set_log_callback(db[r], logCallbackFunc,
1004                                      (void *) "compact_upto_test");
1005        TEST_CHK(status == FDB_RESULT_SUCCESS);
1006    }
1007
1008    status = fdb_get_all_snap_markers(dbfile, &markers, &num_markers);
1009    TEST_CHK(status == FDB_RESULT_SUCCESS);
1010
1011    if (!multi_kv) {
1012        TEST_CHK(num_markers == 4);
1013        for (r = 0; (uint64_t)r < num_markers; ++r) {
1014            TEST_CHK(markers[r].num_kvs_markers == 1);
1015            TEST_CHK(markers[r].kvs_markers[0].seqnum ==
1016                     (fdb_seqnum_t)(n - r*5));
1017        }
1018        r = 1; // Test compacting upto sequence number 15
1019        sprintf(compact_filename, "compact_test_compact%d", r);
1020        status = fdb_compact_upto(dbfile, compact_filename,
1021                                  markers[r].marker);
1022        TEST_CHK(status == FDB_RESULT_SUCCESS);
1023        // create a snapshot
1024        status = fdb_snapshot_open(db[0], &snapshot,
1025                                   markers[r].kvs_markers[0].seqnum);
1026        TEST_CHK(status == FDB_RESULT_SUCCESS);
1027        // close snapshot
1028        fdb_kvs_close(snapshot);
1029    } else {
1030        TEST_CHK(num_markers == 8);
1031        for (r = 0; r < num_kvs; ++r) {
1032            TEST_CHK(markers[r].num_kvs_markers == num_kvs);
1033            for (i = 0; i < num_kvs; ++i) {
1034                TEST_CHK(markers[r].kvs_markers[i].seqnum
1035                         == (fdb_seqnum_t)(n - r*5));
1036                sprintf(kv_name, "kv%d", i);
1037                TEST_CMP(markers[r].kvs_markers[i].kv_store_name, kv_name, 3);
1038            }
1039        }
1040        i = r = 1;
1041        sprintf(compact_filename, "compact_test_compact%d", i);
1042        status = fdb_compact_upto(dbfile, compact_filename,
1043                markers[i].marker);
1044        TEST_CHK(status == FDB_RESULT_SUCCESS);
1045        // create a snapshot
1046        status = fdb_snapshot_open(db[r], &snapshot,
1047                markers[i].kvs_markers[r].seqnum);
1048        TEST_CHK(status == FDB_RESULT_SUCCESS);
1049        // close snapshot
1050        fdb_kvs_close(snapshot);
1051    }
1052
1053    status = fdb_free_snap_markers(markers, num_markers);
1054    TEST_CHK(status == FDB_RESULT_SUCCESS);
1055
1056    // close db file
1057    fdb_close(dbfile);
1058
1059    // free all documents
1060    for (i=0;i<n;++i){
1061        fdb_doc_free(doc[i]);
1062    }
1063
1064    // free all resources
1065    fdb_shutdown();
1066
1067    memleak_end();
1068
1069    sprintf(bodybuf, "compact upto marker in file test %s", multi_kv ?
1070                                                           "multiple kv mode:"
1071                                                         : "single kv mode:");
1072    TEST_RESULT(bodybuf);
1073}
1074
1075void auto_recover_compact_ok_test()
1076{
1077    TEST_INIT();
1078
1079    memleak_start();
1080
1081    int i, r;
1082    int n = 3;
1083    fdb_file_handle *dbfile, *dbfile_new;
1084    fdb_kvs_handle *db;
1085    fdb_kvs_handle *db_new;
1086    fdb_doc **doc = alca(fdb_doc *, n);
1087    fdb_doc *rdoc;
1088    fdb_status status;
1089
1090    char keybuf[256], metabuf[256], bodybuf[256];
1091
1092    // remove previous compact_test files
1093    r = system(SHELL_DEL " compact_test* > errorlog.txt");
1094    (void)r;
1095
1096    fdb_config fconfig = fdb_get_default_config();
1097    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
1098    fconfig.buffercache_size = 16777216;
1099    fconfig.wal_threshold = 1024;
1100    fconfig.flags = FDB_OPEN_FLAG_CREATE;
1101    fconfig.compaction_threshold = 0;
1102
1103    // open db
1104    fdb_open(&dbfile, "./compact_test1", &fconfig);
1105    fdb_kvs_open_default(dbfile, &db, &kvs_config);
1106    status = fdb_set_log_callback(db, logCallbackFunc,
1107                                  (void *) "auto_recover_compact_ok_test");
1108    TEST_CHK(status == FDB_RESULT_SUCCESS);
1109    fdb_open(&dbfile_new, "./compact_test1", &fconfig);
1110    fdb_kvs_open_default(dbfile_new, &db_new, &kvs_config);
1111    status = fdb_set_log_callback(db_new, logCallbackFunc,
1112                                  (void *) "auto_recover_compact_ok_test");
1113    TEST_CHK(status == FDB_RESULT_SUCCESS);
1114
1115    // insert first two documents
1116    for (i=0;i<2;++i){
1117        sprintf(keybuf, "key%d", i);
1118        sprintf(metabuf, "meta%d", i);
1119        sprintf(bodybuf, "body%d", i);
1120        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1121            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1122        fdb_set(db, doc[i]);
1123    }
1124
1125    // remove second doc
1126    fdb_doc_create(&rdoc, doc[1]->key, doc[1]->keylen, doc[1]->meta, doc[1]->metalen, NULL, 0);
1127    rdoc->deleted = true;
1128    fdb_set(db, rdoc);
1129    fdb_doc_free(rdoc);
1130
1131    // commit
1132    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
1133
1134    // perform compaction using one handle
1135    fdb_compact(dbfile, (char *) "./compact_test2");
1136
1137    // save the old file after compaction is done ..
1138    r = system(SHELL_COPY " compact_test1 compact_test11 > errorlog.txt");
1139    (void)r;
1140
1141    // now insert third doc: it should go to the newly compacted file.
1142    sprintf(keybuf, "key%d", i);
1143    sprintf(metabuf, "meta%d", i);
1144    sprintf(bodybuf, "body%d", i);
1145    fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1146        (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1147    fdb_set(db, doc[i]);
1148
1149    // commit
1150    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
1151
1152    // close both the db files ...
1153    fdb_kvs_close(db);
1154    fdb_kvs_close(db_new);
1155    fdb_close(dbfile);
1156    fdb_close(dbfile_new);
1157
1158    // restore the old file after close is done ..
1159    r = system(SHELL_MOVE " compact_test11 compact_test1 > errorlog.txt");
1160    (void)r;
1161
1162    // now open the old saved compacted file, it should automatically recover
1163    // and use the new file since compaction was done successfully
1164    fdb_open(&dbfile_new, "./compact_test1", &fconfig);
1165    fdb_kvs_open_default(dbfile_new, &db_new, &kvs_config);
1166    status = fdb_set_log_callback(db_new, logCallbackFunc,
1167                                  (void *) "auto_recover_compact_ok_test");
1168    TEST_CHK(status == FDB_RESULT_SUCCESS);
1169
1170    // retrieve documents using the old handle and expect all 3 docs
1171    for (i=0;i<n;++i){
1172        // search by key
1173        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
1174        status = fdb_get(db_new, rdoc);
1175
1176        if (i != 1) {
1177            TEST_CHK(status == FDB_RESULT_SUCCESS);
1178            TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
1179            TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1180        }else{
1181            TEST_CHK(status == FDB_RESULT_KEY_NOT_FOUND);
1182        }
1183
1184        // free result document
1185        fdb_doc_free(rdoc);
1186    }
1187    // check this handle's filename it should point to newly compacted file
1188    fdb_file_info info;
1189    fdb_get_file_info(dbfile_new, &info);
1190    TEST_CHK(!strcmp("./compact_test2", info.filename));
1191
1192    // close the file
1193    fdb_kvs_close(db_new);
1194    fdb_close(dbfile_new);
1195
1196    // free all documents
1197    for (i=0;i<n;++i){
1198        fdb_doc_free(doc[i]);
1199    }
1200
1201    // free all resources
1202    fdb_shutdown();
1203
1204    memleak_end();
1205
1206    TEST_RESULT("auto recovery after compaction test");
1207}
1208
1209#if !defined(WIN32) && !defined(_WIN32)
1210static bool does_file_exist(const char *filename) {
1211    struct stat st;
1212    int result = stat(filename, &st);
1213    return result == 0;
1214}
1215#else
1216static bool does_file_exist(const char *filename) {
1217    return GetFileAttributes(filename) != INVALID_FILE_ATTRIBUTES;
1218}
1219#endif
1220
1221void unlink_after_compaction_test()
1222{
1223    TEST_INIT();
1224    memleak_start();
1225
1226    int i, r;
1227    int n = 10;
1228    fdb_file_handle *dbfile;
1229    fdb_kvs_handle *db, *snap;
1230    fdb_doc **doc = alca(fdb_doc *, n);
1231    fdb_doc *rdoc;
1232    fdb_status s;
1233
1234    char keybuf[256], metabuf[256], bodybuf[256];
1235
1236    // remove previous compact_test files
1237    r = system(SHELL_DEL " compact_test* > errorlog.txt");
1238    (void)r;
1239
1240    fdb_config fconfig = fdb_get_default_config();
1241    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
1242    fconfig.buffercache_size = 16777216;
1243    fconfig.wal_threshold = 1024;
1244    fconfig.flags = FDB_OPEN_FLAG_CREATE;
1245    fconfig.compaction_threshold = 0;
1246
1247    // open db
1248    fdb_open(&dbfile, "./compact_test1", &fconfig);
1249    fdb_kvs_open_default(dbfile, &db, &kvs_config);
1250    s = fdb_set_log_callback(db, logCallbackFunc,
1251                             (void *) "unlink_after_compaction_test");
1252    TEST_CHK(s == FDB_RESULT_SUCCESS);
1253
1254    // set docs
1255    for (i=0;i<n;++i){
1256        sprintf(keybuf, "key%d", i);
1257        sprintf(metabuf, "meta%d", i);
1258        sprintf(bodybuf, "body%d", i);
1259        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1260            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1261        fdb_set(db, doc[i]);
1262    }
1263
1264    // commit
1265    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
1266
1267    // create a snapshot
1268    s = fdb_snapshot_open(db, &snap, n);
1269    TEST_CHK(s == FDB_RESULT_SUCCESS);
1270
1271    // compaction
1272    s = fdb_compact(dbfile, "./compact_test2");
1273    TEST_CHK(s == FDB_RESULT_SUCCESS);
1274
1275    // retrieve check on the new file
1276    for (i=0;i<n;++i){
1277        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
1278        s = fdb_get(db, rdoc);
1279        TEST_CHK(s == FDB_RESULT_SUCCESS);
1280        fdb_doc_free(rdoc);
1281    }
1282
1283    // the old file should not be seen by user level application
1284#if !defined(WIN32) && !defined(_WIN32)
1285#ifndef _MSC_VER
1286    TEST_CHK(!does_file_exist("./compact_test1"));
1287#endif
1288#endif
1289
1290    // retrieve check on the old file (snapshot)
1291    for (i=0;i<n;++i){
1292        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
1293        s = fdb_get(snap, rdoc);
1294        TEST_CHK(s == FDB_RESULT_SUCCESS);
1295        fdb_doc_free(rdoc);
1296    }
1297
1298    fdb_close(dbfile);
1299
1300    fdb_shutdown();
1301    for (i=0; i<n; ++i){
1302        fdb_doc_free(doc[i]);
1303    }
1304
1305    memleak_end();
1306    TEST_RESULT("unlink after compaction test");
1307}
1308
1309void db_compact_overwrite()
1310{
1311    TEST_INIT();
1312    memleak_start();
1313
1314    int i, r;
1315    int n = 30;
1316    fdb_file_handle *dbfile, *dbfile2;
1317    fdb_kvs_handle *db, *db2;
1318    fdb_doc **doc = alca(fdb_doc *, n);
1319    fdb_doc **doc2 = alca(fdb_doc *, 2*n);
1320    fdb_doc *rdoc;
1321    fdb_status status;
1322    fdb_config fconfig;
1323    fdb_kvs_info kvs_info;
1324    fdb_kvs_config kvs_config;
1325
1326    char keybuf[256], metabuf[256], bodybuf[256];
1327
1328    // remove previous compact_test files
1329    r = system(SHELL_DEL " compact_test* > errorlog.txt");
1330    (void)r;
1331
1332    fconfig = fdb_get_default_config();
1333    kvs_config = fdb_get_default_kvs_config();
1334    fconfig.buffercache_size = 16777216;
1335    fconfig.wal_threshold = 1024;
1336    fconfig.flags = FDB_OPEN_FLAG_CREATE;
1337    fconfig.compaction_threshold = 0;
1338
1339    // write to db1
1340    fdb_open(&dbfile, "./compact_test1", &fconfig);
1341    fdb_kvs_open(dbfile, &db, NULL, &kvs_config);
1342    status = fdb_set_log_callback(db, logCallbackFunc,
1343                                  (void *) "db_destroy_test");
1344    TEST_CHK(status == FDB_RESULT_SUCCESS);
1345    for (i=0;i<n;++i){
1346        sprintf(keybuf, "key%d", i);
1347        sprintf(metabuf, "meta%d", i);
1348        sprintf(bodybuf, "body%d", i);
1349        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1350            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1351        fdb_set(db, doc[i]);
1352    }
1353    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1354
1355    // Open the empty db with future compact name
1356    fdb_open(&dbfile2, "./compact_test1.1", &fconfig);
1357    fdb_kvs_open(dbfile2, &db2, NULL, &kvs_config);
1358    status = fdb_set_log_callback(db2, logCallbackFunc,
1359                                  (void *) "db_destroy_test");
1360    TEST_CHK(status == FDB_RESULT_SUCCESS);
1361    // write to db2
1362    for (i=0;i < 2*n;++i){
1363        sprintf(keybuf, "k2ey%d", i);
1364        sprintf(metabuf, "m2eta%d", i);
1365        sprintf(bodybuf, "b2ody%d", i);
1366        fdb_doc_create(&doc2[i], (void*)keybuf, strlen(keybuf),
1367            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1368        fdb_set(db2, doc2[i]);
1369    }
1370    fdb_commit(dbfile2, FDB_COMMIT_NORMAL);
1371
1372
1373    // verify db2 seqnum and close
1374    fdb_get_kvs_info(db2, &kvs_info);
1375    TEST_CHK(kvs_info.last_seqnum = 2*n);
1376    fdb_kvs_close(db2);
1377    fdb_close(dbfile2);
1378
1379    // compact db1
1380    status = fdb_compact(dbfile, NULL);
1381    TEST_CHK(status == FDB_RESULT_SUCCESS);
1382
1383    // close db1
1384    fdb_kvs_close(db);
1385    fdb_close(dbfile);
1386
1387    // reopen db1
1388    fdb_open(&dbfile, "./compact_test1", &fconfig);
1389    fdb_kvs_open(dbfile, &db, NULL, &kvs_config);
1390    status = fdb_set_log_callback(db, logCallbackFunc,
1391                                  (void *) "db_destroy_test");
1392    TEST_CHK(status == FDB_RESULT_SUCCESS);
1393    // read db1
1394    for (i=0;i<n;++i){
1395        // search by key
1396        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
1397        status = fdb_get(db, rdoc);
1398        TEST_CHK(status == FDB_RESULT_SUCCESS);
1399        TEST_CHK(!memcmp(rdoc->meta, doc[i]->meta, rdoc->metalen));
1400        TEST_CHK(!memcmp(rdoc->body, doc[i]->body, rdoc->bodylen));
1401        // free result document
1402        fdb_doc_free(rdoc);
1403    }
1404
1405    // reopen db2
1406    fdb_open(&dbfile2, "./compact_test1.1", &fconfig);
1407    fdb_kvs_open(dbfile2, &db2, NULL, &kvs_config);
1408    status = fdb_set_log_callback(db2, logCallbackFunc,
1409                                  (void *) "db_destroy_test");
1410    TEST_CHK(status == FDB_RESULT_SUCCESS);
1411
1412    fdb_get_kvs_info(db2, &kvs_info);
1413    TEST_CHK(kvs_info.last_seqnum = 2*n);
1414
1415    // read db2
1416    for (i=0;i<2*n;++i){
1417        // search by key
1418        fdb_doc_create(&rdoc, doc2[i]->key, doc2[i]->keylen, NULL, 0, NULL, 0);
1419        status = fdb_get(db2, rdoc);
1420        TEST_CHK(status == FDB_RESULT_SUCCESS);
1421        TEST_CHK(!memcmp(rdoc->meta, doc2[i]->meta, rdoc->metalen));
1422        TEST_CHK(!memcmp(rdoc->body, doc2[i]->body, rdoc->bodylen));
1423        // free result document
1424        fdb_doc_free(rdoc);
1425    }
1426
1427    // free all documents
1428    for (i=0;i<n;++i){
1429        fdb_doc_free(doc[i]);
1430        fdb_doc_free(doc2[i]);
1431    }
1432    for (i=n;i<2*n;++i){
1433        fdb_doc_free(doc2[i]);
1434    }
1435
1436    fdb_kvs_close(db);
1437    fdb_close(dbfile);
1438    fdb_kvs_close(db2);
1439    fdb_close(dbfile2);
1440
1441
1442    fdb_shutdown();
1443    memleak_end();
1444    TEST_RESULT("compact overwrite");
1445}
1446
1447void *db_compact_during_doc_delete(void *args)
1448{
1449
1450    TEST_INIT();
1451    memleak_start();
1452
1453    int i, r;
1454    int n = 100;
1455    fdb_file_handle *dbfile;
1456    fdb_kvs_handle *db;
1457    fdb_status status;
1458    fdb_config fconfig;
1459    fdb_kvs_config kvs_config;
1460    fdb_kvs_info kvs_info;
1461    thread_t tid;
1462    void *thread_ret;
1463    fdb_doc **doc = alca(fdb_doc*, n);
1464    char keybuf[256], metabuf[256], bodybuf[256];
1465
1466    // init dbfile
1467    kvs_config = fdb_get_default_kvs_config();
1468    fconfig = fdb_get_default_config();
1469    fconfig.buffercache_size = 0;
1470    fconfig.wal_threshold = 1024;
1471    fconfig.compaction_threshold = 0;
1472
1473    if (args == NULL)
1474    { // parent
1475
1476        r = system(SHELL_DEL" compact_test* > errorlog.txt");
1477        (void)r;
1478
1479        status = fdb_open(&dbfile, "./compact_test1", &fconfig);
1480        TEST_CHK(status == FDB_RESULT_SUCCESS);
1481        status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
1482        TEST_CHK(status == FDB_RESULT_SUCCESS);
1483
1484        // insert documents
1485        for (i=0;i<n;++i){
1486            sprintf(keybuf, "key%d", i);
1487            sprintf(metabuf, "meta%d", i);
1488            sprintf(bodybuf, "body%d", i);
1489            fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1490                (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1491            fdb_set(db, doc[i]);
1492        }
1493
1494        fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1495        // verify no docs remaining
1496        fdb_get_kvs_info(db, &kvs_info);
1497        TEST_CHK(kvs_info.doc_count == (uint64_t)n);
1498
1499        // start deleting docs
1500        for (i=0;i<n;++i){
1501            fdb_del(db, doc[i]);
1502            if (i == n/2){
1503                // compact half-way
1504                thread_create(&tid, db_compact_during_doc_delete, (void *)dbfile);
1505            }
1506        }
1507
1508        // join compactor
1509        thread_join(tid, &thread_ret);
1510
1511        fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
1512
1513        // verify no docs remaining
1514        fdb_get_kvs_info(db, &kvs_info);
1515        TEST_CHK(kvs_info.doc_count == 0);
1516
1517        // reopen
1518        fdb_kvs_close(db);
1519        fdb_close(dbfile);
1520        status = fdb_open(&dbfile, "./compact_test1", &fconfig);
1521        TEST_CHK(status == FDB_RESULT_SUCCESS);
1522        status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
1523        TEST_CHK(status == FDB_RESULT_SUCCESS);
1524
1525        fdb_get_kvs_info(db, &kvs_info);
1526        TEST_CHK(kvs_info.doc_count == 0);
1527
1528        // cleanup
1529        for (i=0;i<n;++i){
1530            fdb_doc_free(doc[i]);
1531        }
1532        fdb_kvs_close(db);
1533        fdb_close(dbfile);
1534        fdb_shutdown();
1535
1536        memleak_end();
1537        TEST_RESULT("multi thread client shutdown");
1538        return NULL;
1539    }
1540
1541    // compaction thread enters here //
1542    status = fdb_open(&dbfile, "./compact_test1", &fconfig);
1543    TEST_CHK(status == FDB_RESULT_SUCCESS);
1544    status = fdb_compact(dbfile, NULL);
1545    TEST_CHK(status == FDB_RESULT_SUCCESS);
1546    fdb_close(dbfile);
1547
1548    // shutdown
1549    thread_exit(0);
1550    return NULL;
1551}
1552
1553void compaction_daemon_test(size_t time_sec)
1554{
1555    TEST_INIT();
1556
1557    memleak_start();
1558
1559    int i, r;
1560    int n = 10000;
1561    int compaction_threshold = 30;
1562    int escape = 0;
1563    fdb_file_handle *dbfile, *dbfile_less, *dbfile_non, *dbfile_manual, *dbfile_new;
1564    fdb_kvs_handle *db, *db_less, *db_non, *db_manual;
1565    fdb_kvs_handle *snapshot;
1566    fdb_doc **doc = alca(fdb_doc*, n);
1567    fdb_doc *rdoc;
1568    fdb_file_info info;
1569    fdb_status status;
1570    struct timeval ts_begin, ts_cur, ts_gap;
1571
1572    char keybuf[256], metabuf[256], bodybuf[256];
1573
1574    // remove previous compact_test files
1575    r = system(SHELL_DEL" compact_test* > errorlog.txt");
1576    (void)r;
1577
1578    fdb_config fconfig = fdb_get_default_config();
1579    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
1580    fconfig.buffercache_size = 0;
1581    fconfig.wal_threshold = 1024;
1582    fconfig.flags = FDB_OPEN_FLAG_CREATE;
1583    fconfig.compaction_mode = FDB_COMPACTION_AUTO;
1584    fconfig.compaction_threshold = compaction_threshold;
1585    fconfig.compactor_sleep_duration = 1; // for quick test
1586
1587    fconfig.num_compactor_threads = 0;
1588    status = fdb_open(&dbfile, "compact_test", &fconfig);
1589    TEST_CHK(status == FDB_RESULT_INVALID_CONFIG);
1590
1591    fconfig.num_compactor_threads = DEFAULT_NUM_COMPACTOR_THREADS;
1592    // open db
1593    fdb_open(&dbfile, "compact_test", &fconfig);
1594    fdb_kvs_open_default(dbfile, &db, &kvs_config);
1595    status = fdb_set_log_callback(db, logCallbackFunc,
1596                                  (void *) "compaction_daemon_test");
1597    TEST_CHK(status == FDB_RESULT_SUCCESS);
1598    // insert documents
1599    printf("Initialize..\n");
1600    for (i=0;i<n;++i){
1601        sprintf(keybuf, "key%04d", i);
1602        sprintf(metabuf, "meta%04d", i);
1603        sprintf(bodybuf, "body%04d", i);
1604        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf)+1,
1605            (void*)metabuf, strlen(metabuf)+1, (void*)bodybuf, strlen(bodybuf)+1);
1606        fdb_set(db, doc[i]);
1607    }
1608    // commit
1609    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1610    // close db file
1611    fdb_close(dbfile);
1612
1613    // ---- basic retrieve test ------------------------
1614    // reopen db file
1615    status = fdb_open(&dbfile, "compact_test", &fconfig);
1616    TEST_CHK(status == FDB_RESULT_SUCCESS);
1617    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
1618    TEST_CHK(status == FDB_RESULT_SUCCESS);
1619    status = fdb_set_log_callback(db, logCallbackFunc,
1620                                  (void *) "compaction_daemon_test");
1621    TEST_CHK(status == FDB_RESULT_SUCCESS);
1622    // check db filename
1623    fdb_get_file_info(dbfile, &info);
1624    TEST_CHK(!strcmp(info.filename, "compact_test"));
1625
1626    // retrieve documents
1627    for (i=0;i<n;++i){
1628        sprintf(keybuf, "key%04d", i);
1629        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf)+1,
1630            NULL, 0, NULL, 0);
1631        status = fdb_get(db, rdoc);
1632        //printf("%s %s\n", rdoc->key, rdoc->body);
1633        TEST_CHK(status == FDB_RESULT_SUCCESS);
1634        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1635        fdb_doc_free(rdoc);
1636    }
1637
1638    // create a snapshot
1639    status = fdb_snapshot_open(db, &snapshot, n);
1640    TEST_CHK(status == FDB_RESULT_SUCCESS);
1641    // close snapshot
1642    fdb_kvs_close(snapshot);
1643
1644    // close db file
1645    fdb_close(dbfile);
1646
1647    // ---- handling when metafile is removed ------------
1648    // remove meta file
1649    r = system(SHELL_DEL" compact_test.meta > errorlog.txt");
1650    (void)r;
1651    // reopen db file
1652    status = fdb_open(&dbfile, "compact_test", &fconfig);
1653    TEST_CHK(status == FDB_RESULT_SUCCESS);
1654    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
1655    TEST_CHK(status == FDB_RESULT_SUCCESS);
1656    status = fdb_set_log_callback(db, logCallbackFunc,
1657                                  (void *) "compaction_daemon_test");
1658    TEST_CHK(status == FDB_RESULT_SUCCESS);
1659    // retrieve documents
1660    for (i=0;i<n;++i){
1661        sprintf(keybuf, "key%04d", i);
1662        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf)+1,
1663            NULL, 0, NULL, 0);
1664        status = fdb_get(db, rdoc);
1665        //printf("%s %s\n", rdoc->key, rdoc->body);
1666        TEST_CHK(status == FDB_RESULT_SUCCESS);
1667        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1668        fdb_doc_free(rdoc);
1669    }
1670    // close db file
1671    fdb_close(dbfile);
1672
1673    // ---- handling when metafile points to non-exist file ------------
1674    // remove meta file
1675    r = system(SHELL_MOVE" compact_test.0 compact_test.23 > errorlog.txt");
1676    (void)r;
1677    // reopen db file
1678    status = fdb_open(&dbfile, "compact_test", &fconfig);
1679    TEST_CHK(status == FDB_RESULT_SUCCESS);
1680    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
1681    TEST_CHK(status == FDB_RESULT_SUCCESS);
1682    status = fdb_set_log_callback(db, logCallbackFunc,
1683                                  (void *) "compaction_daemon_test");
1684    TEST_CHK(status == FDB_RESULT_SUCCESS);
1685    // retrieve documents
1686    for (i=0;i<n;++i){
1687        sprintf(keybuf, "key%04d", i);
1688        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf)+1,
1689            NULL, 0, NULL, 0);
1690        status = fdb_get(db, rdoc);
1691        //printf("%s %s\n", rdoc->key, rdoc->body);
1692        TEST_CHK(status == FDB_RESULT_SUCCESS);
1693        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1694        fdb_doc_free(rdoc);
1695    }
1696    // close db file
1697    fdb_close(dbfile);
1698
1699    // ---- compaction daemon test -------------------
1700    // db: DB instance to be compacted
1701    // db_less: DB instance to be compacted but with much lower update throughput
1702    // db_non: DB instance not to be compacted (auto compaction with threshold = 0)
1703    // db_manual: DB instance not to be compacted (manual compaction)
1704
1705    // open & create db_less, db_non and db_manual
1706    status = fdb_open(&dbfile_less, "compact_test_less", &fconfig);
1707    TEST_CHK(status == FDB_RESULT_SUCCESS);
1708    status = fdb_kvs_open_default(dbfile_less, &db_less, &kvs_config);
1709    TEST_CHK(status == FDB_RESULT_SUCCESS);
1710
1711    fconfig.compaction_threshold = 0;
1712    status = fdb_open(&dbfile_non, "compact_test_non", &fconfig);
1713    TEST_CHK(status == FDB_RESULT_SUCCESS);
1714    status = fdb_kvs_open_default(dbfile_non, &db_non, &kvs_config);
1715    TEST_CHK(status == FDB_RESULT_SUCCESS);
1716
1717    fconfig.compaction_mode = FDB_COMPACTION_MANUAL;
1718    status = fdb_open(&dbfile_manual, "compact_test_manual", &fconfig);
1719    TEST_CHK(status == FDB_RESULT_SUCCESS);
1720    status = fdb_kvs_open_default(dbfile_manual, &db_manual, &kvs_config);
1721    TEST_CHK(status == FDB_RESULT_SUCCESS);
1722
1723    // reopen db file
1724    fconfig.compaction_threshold = 30;
1725    fconfig.compaction_mode = FDB_COMPACTION_AUTO;
1726    status = fdb_open(&dbfile, "compact_test", &fconfig);
1727    TEST_CHK(status == FDB_RESULT_SUCCESS);
1728    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
1729    TEST_CHK(status == FDB_RESULT_SUCCESS);
1730    status = fdb_set_log_callback(db, logCallbackFunc,
1731                                  (void *) "compaction_daemon_test");
1732    TEST_CHK(status == FDB_RESULT_SUCCESS);
1733    // continuously update documents
1734    printf("wait for %d seconds..\n", (int)time_sec);
1735    gettimeofday(&ts_begin, NULL);
1736    while (!escape) {
1737        for (i=0;i<n;++i){
1738            // update db
1739            fdb_set(db, doc[i]);
1740            fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1741
1742            // update db_less (1/100 throughput)
1743            if (i%100 == 0){
1744                fdb_set(db_less, doc[i]);
1745                fdb_commit(dbfile_less, FDB_COMMIT_NORMAL);
1746            }
1747
1748            // update db_non
1749            fdb_set(db_non, doc[i]);
1750            fdb_commit(dbfile_non, FDB_COMMIT_NORMAL);
1751
1752            // update db_manual
1753            fdb_set(db_manual, doc[i]);
1754            fdb_commit(dbfile_manual, FDB_COMMIT_NORMAL);
1755
1756            gettimeofday(&ts_cur, NULL);
1757            ts_gap = _utime_gap(ts_begin, ts_cur);
1758            if ((size_t)ts_gap.tv_sec >= time_sec) {
1759                escape = 1;
1760                break;
1761            }
1762        }
1763    }
1764
1765    // Change the compaction interval to 60 secs.
1766    status = fdb_set_daemon_compaction_interval(dbfile, 60);
1767    TEST_CHK(status == FDB_RESULT_SUCCESS);
1768    // Change the compaction interval back to 1 sec.
1769    status = fdb_set_daemon_compaction_interval(dbfile, 1);
1770    TEST_CHK(status == FDB_RESULT_SUCCESS);
1771
1772    // perform manual compaction of auto-compact file
1773    status = fdb_compact(dbfile_non, NULL);
1774    TEST_CHK(status == FDB_RESULT_SUCCESS);
1775
1776    // perform manual compaction of manual-compact file
1777    status = fdb_compact(dbfile_manual, "compact_test_manual_compacted");
1778    TEST_CHK(status == FDB_RESULT_SUCCESS);
1779
1780    // open compact_test_manual_compacted using new db handle
1781    fconfig.compaction_mode = FDB_COMPACTION_MANUAL;
1782    status = fdb_open(&dbfile_new, "compact_test_manual_compacted", &fconfig);
1783    TEST_CHK(status == FDB_RESULT_SUCCESS);
1784
1785    // try to switch compaction mode
1786    status = fdb_switch_compaction_mode(dbfile_manual, FDB_COMPACTION_AUTO, 30);
1787    TEST_CHK(status == FDB_RESULT_FILE_IS_BUSY);
1788
1789    // close db_new
1790    status = fdb_close(dbfile_new);
1791    TEST_CHK(status == FDB_RESULT_SUCCESS);
1792
1793    // switch compaction mode of 'db_manual' from MANUAL to AUTO
1794    status = fdb_switch_compaction_mode(dbfile_manual, FDB_COMPACTION_AUTO, 10);
1795    TEST_CHK(status == FDB_RESULT_SUCCESS);
1796
1797    // change compaction value
1798    status = fdb_switch_compaction_mode(dbfile_manual, FDB_COMPACTION_AUTO, 30);
1799    TEST_CHK(status == FDB_RESULT_SUCCESS);
1800
1801    // close and open with auto-compact option
1802    status = fdb_close(dbfile_manual);
1803    TEST_CHK(status == FDB_RESULT_SUCCESS);
1804    fconfig.compaction_mode = FDB_COMPACTION_AUTO;
1805    status = fdb_open(&dbfile_manual, "compact_test_manual_compacted", &fconfig);
1806    TEST_CHK(status == FDB_RESULT_SUCCESS);
1807
1808    // switch compaction mode of 'db_non' from AUTO to MANUAL
1809    status = fdb_switch_compaction_mode(dbfile_non, FDB_COMPACTION_MANUAL, 0);
1810    TEST_CHK(status == FDB_RESULT_SUCCESS);
1811
1812    // close and open with manual-compact option
1813    status = fdb_close(dbfile_non);
1814    TEST_CHK(status == FDB_RESULT_SUCCESS);
1815    fconfig.compaction_mode = FDB_COMPACTION_MANUAL;
1816    status = fdb_open(&dbfile_non, "compact_test_non", &fconfig);
1817    TEST_CHK(status == FDB_RESULT_SUCCESS);
1818
1819    // Now perform one manual compaction on compact_test_non
1820    fdb_compact(dbfile_non, "compact_test_non.manual");
1821
1822    // close all db files except compact_test_non
1823    fdb_close(dbfile);
1824    fdb_close(dbfile_less);
1825    fdb_close(dbfile_manual);
1826
1827    // open manual compact file (compact_test_non) using auto compact mode
1828    fconfig.compaction_mode = FDB_COMPACTION_AUTO;
1829    status = fdb_open(&dbfile, "compact_test_non.manual", &fconfig);
1830    TEST_CHK(status == FDB_RESULT_INVALID_COMPACTION_MODE);
1831
1832    // Attempt to destroy manual compact file using auto compact mode
1833    status = fdb_destroy("compact_test_non.manual", &fconfig);
1834    TEST_CHK(status == FDB_RESULT_INVALID_COMPACTION_MODE);
1835
1836    // open auto copmact file (compact_test_manual_compacted) using manual compact mode
1837    fconfig.compaction_mode = FDB_COMPACTION_MANUAL;
1838    status = fdb_open(&dbfile, "compact_test_manual_compacted", &fconfig);
1839    TEST_CHK(status == FDB_RESULT_INVALID_COMPACTION_MODE);
1840
1841    // Attempt to destroy auto copmact file using manual compact mode
1842    status = fdb_destroy("compact_test_manual_compacted", &fconfig);
1843    TEST_CHK(status == FDB_RESULT_INVALID_COMPACTION_MODE);
1844
1845    // DESTROY auto copmact file with correct mode
1846    fconfig.compaction_mode = FDB_COMPACTION_AUTO;
1847    status = fdb_destroy("compact_test_manual_compacted", &fconfig);
1848    TEST_CHK(status == FDB_RESULT_SUCCESS);
1849
1850    // DESTROY manual compacted file with past version open!
1851    fconfig.compaction_mode = FDB_COMPACTION_MANUAL;
1852    status = fdb_destroy("compact_test_non.manual", &fconfig);
1853    TEST_CHK(status == FDB_RESULT_FILE_IS_BUSY);
1854    fdb_close(dbfile_non);
1855
1856    // Simulate a database crash by doing a premature shutdown
1857    // Note that db_non was never closed properly
1858    status = fdb_shutdown();
1859    TEST_CHK(status == FDB_RESULT_SUCCESS);
1860
1861    status = fdb_destroy("compact_test_non.manual", &fconfig);
1862    TEST_CHK(status == FDB_RESULT_SUCCESS);
1863
1864    // Attempt to read-only auto compacted and destroyed file
1865    fconfig.flags = FDB_OPEN_FLAG_RDONLY;
1866    status = fdb_open(&dbfile, "./compact_test_manual_compacted", &fconfig);
1867    TEST_CHK(status == FDB_RESULT_NO_SUCH_FILE);
1868
1869    status = fdb_open(&dbfile, "./compact_test_manual_compacted.meta", &fconfig);
1870    TEST_CHK(status == FDB_RESULT_NO_SUCH_FILE);
1871
1872    // Attempt to read-only past version of manually compacted destroyed file
1873    status = fdb_open(&dbfile, "compact_test_non", &fconfig);
1874    TEST_CHK(status == FDB_RESULT_NO_SUCH_FILE);
1875
1876    // Attempt to read-only current version of manually compacted destroyed file
1877    status = fdb_open(&dbfile, "compact_test_non.manual", &fconfig);
1878    TEST_CHK(status == FDB_RESULT_NO_SUCH_FILE);
1879
1880    // free all documents
1881    for (i=0;i<n;++i){
1882        fdb_doc_free(doc[i]);
1883    }
1884
1885    // free all resources
1886    fdb_shutdown();
1887
1888    memleak_end();
1889
1890    TEST_RESULT("compaction daemon test");
1891}
1892
1893// MB-13117
1894void auto_compaction_with_concurrent_insert_test(size_t t_limit)
1895{
1896    TEST_INIT();
1897
1898    memleak_start();
1899
1900    int i, r;
1901    fdb_file_handle *file;
1902    fdb_kvs_handle *kvs;
1903    fdb_status status;
1904    fdb_config config;
1905    fdb_kvs_config kvs_config;
1906    struct timeval ts_begin, ts_cur, ts_gap;
1907
1908    r = system(SHELL_DEL" compact_test* > errorlog.txt");
1909    (void)r;
1910
1911    // Open Database File
1912    config = fdb_get_default_config();
1913    config.compaction_mode=FDB_COMPACTION_AUTO;
1914    config.compactor_sleep_duration = 1;
1915    status = fdb_open(&file, "compact_test", &config);
1916    TEST_CHK(status == FDB_RESULT_SUCCESS);
1917
1918    // Open KV Store
1919    kvs_config = fdb_get_default_kvs_config();
1920    status = fdb_kvs_open_default(file, &kvs, &kvs_config);
1921    TEST_CHK(status == FDB_RESULT_SUCCESS);
1922
1923    gettimeofday(&ts_begin, NULL);
1924    printf("wait for %d seconds..\n", (int)t_limit);
1925
1926    // Several kv pairs
1927    for(i=0;i<100000;i++) {
1928        char str[15];
1929        sprintf(str, "%d", i);
1930        status = fdb_set_kv(kvs, str, strlen(str), (void*)"value", 5);
1931        TEST_CHK(status == FDB_RESULT_SUCCESS);
1932
1933        // Commit
1934        status = fdb_commit(file, FDB_COMMIT_NORMAL);
1935        TEST_CHK(status == FDB_RESULT_SUCCESS);
1936
1937        gettimeofday(&ts_cur, NULL);
1938        ts_gap = _utime_gap(ts_begin, ts_cur);
1939        if ((size_t)ts_gap.tv_sec >= t_limit) {
1940            break;
1941        }
1942    }
1943
1944    status = fdb_close(file);
1945    TEST_CHK(status == FDB_RESULT_SUCCESS);
1946    status = fdb_shutdown();
1947    TEST_CHK(status == FDB_RESULT_SUCCESS);
1948
1949    memleak_end();
1950
1951    TEST_RESULT("auto compaction with concurrent insert test");
1952}
1953
1954// lexicographically compares two variable-length binary streams
1955#define MIN(a,b) (((a)<(b))?(a):(b))
1956static int _compact_test_keycmp(void *key1, size_t keylen1,
1957                                void *key2, size_t keylen2)
1958{
1959    if (keylen1 == keylen2) {
1960        return memcmp(key1, key2, keylen1);
1961    }else {
1962        size_t len = MIN(keylen1, keylen2);
1963        int cmp = memcmp(key1, key2, len);
1964        if (cmp != 0) return cmp;
1965        else {
1966            return (int)((int)keylen1 - (int)keylen2);
1967        }
1968    }
1969}
1970
1971void auto_compaction_with_custom_cmp_function()
1972{
1973    TEST_INIT();
1974
1975    memleak_start();
1976
1977    int i, r, n=10000;
1978    char keybuf[256], bodybuf[256];
1979    uint64_t max_filesize = 0;
1980    fdb_file_handle *file;
1981    fdb_kvs_handle *db1, *db2, *db3;
1982    fdb_status status;
1983    fdb_config config;
1984    fdb_kvs_config kvs_config;
1985    fdb_file_info file_info;
1986    char *kvs_names[] = {NULL};
1987    fdb_custom_cmp_variable functions[] = {_compact_test_keycmp};
1988
1989    r = system(SHELL_DEL" compact_test* > errorlog.txt");
1990    (void)r;
1991
1992    // Open Database File
1993    config = fdb_get_default_config();
1994    config.wal_threshold = 4096; // reset WAL threshold for correct file size estimation
1995    config.compaction_mode=FDB_COMPACTION_AUTO;
1996    config.compactor_sleep_duration = 1;
1997    config.compaction_threshold = 10;
1998    status = fdb_open_custom_cmp(&file, "compact_test", &config,
1999                                 1, kvs_names, functions);
2000    TEST_CHK(status == FDB_RESULT_SUCCESS);
2001
2002    // Open 2 KV Stores
2003    kvs_config = fdb_get_default_kvs_config();
2004    status = fdb_kvs_open_default(file, &db1, &kvs_config);
2005    TEST_CHK(status == FDB_RESULT_SUCCESS);
2006    status = fdb_kvs_open(file, &db2, "db", &kvs_config);
2007    TEST_CHK(status == FDB_RESULT_SUCCESS);
2008
2009    // initial load
2010    for(i=0;i<n;i++) {
2011        sprintf(keybuf, "key%06d", i);
2012        sprintf(bodybuf, "body%06d", i);
2013        status = fdb_set_kv(db1, keybuf, strlen(keybuf),
2014                                 bodybuf, strlen(bodybuf));
2015        TEST_CHK(status == FDB_RESULT_SUCCESS);
2016        status = fdb_set_kv(db2, keybuf, strlen(keybuf),
2017                                 bodybuf, strlen(bodybuf));
2018        TEST_CHK(status == FDB_RESULT_SUCCESS);
2019    }
2020
2021    // create one more KVS using custom cmp
2022    // it doesn't exist on the initial cmp_func list
2023    kvs_config.custom_cmp = _compact_test_keycmp;
2024    status = fdb_kvs_open(file, &db3, "db_custom", &kvs_config);
2025    TEST_CHK(status == FDB_RESULT_SUCCESS);
2026    i = 0;
2027    sprintf(keybuf, "key%06d", i);
2028    sprintf(bodybuf, "body%06d", i);
2029    status = fdb_set_kv(db3, keybuf, strlen(keybuf),
2030                             bodybuf, strlen(bodybuf));
2031    TEST_CHK(status == FDB_RESULT_SUCCESS);
2032
2033    // Commit
2034    status = fdb_commit(file, FDB_COMMIT_NORMAL);
2035    TEST_CHK(status == FDB_RESULT_SUCCESS);
2036
2037    // update to trigger compaction
2038    for(i=0;i<n;i++) {
2039        sprintf(keybuf, "key%06d", i);
2040        sprintf(bodybuf, "body%06d", i);
2041        status = fdb_set_kv(db1, keybuf, strlen(keybuf),
2042                                 bodybuf, strlen(bodybuf));
2043        TEST_CHK(status == FDB_RESULT_SUCCESS);
2044
2045        status = fdb_get_file_info(file, &file_info);
2046        TEST_CHK(status == FDB_RESULT_SUCCESS);
2047        if (file_info.file_size > max_filesize) {
2048            max_filesize = file_info.file_size;
2049        }
2050    }
2051    // Commit
2052    status = fdb_commit(file, FDB_COMMIT_NORMAL);
2053    TEST_CHK(status == FDB_RESULT_SUCCESS);
2054
2055    status = fdb_get_file_info(file, &file_info);
2056    TEST_CHK(status == FDB_RESULT_SUCCESS);
2057    if (file_info.file_size > max_filesize) {
2058        max_filesize = file_info.file_size;
2059    }
2060
2061    printf("wait for daemon compaction completion... (max file size: %" _F64 ")\n", max_filesize);
2062    while (true) {
2063        sleep(1);
2064
2065        status = fdb_get_file_info(file, &file_info);
2066        TEST_CHK(status == FDB_RESULT_SUCCESS);
2067        if (file_info.file_size < max_filesize) {
2068            break;
2069        }
2070    }
2071    // should be compacted
2072    TEST_CHK(file_info.file_size < max_filesize);
2073
2074    status = fdb_close(file);
2075    TEST_CHK(status == FDB_RESULT_SUCCESS);
2076    status = fdb_shutdown();
2077    TEST_CHK(status == FDB_RESULT_SUCCESS);
2078
2079    memleak_end();
2080
2081    TEST_RESULT("auto compaction with custom comparison function");
2082}
2083
2084struct cb_txn_args {
2085    fdb_file_handle *file;
2086    fdb_kvs_handle *handle;
2087    int ndocs;
2088    int nupdates;
2089    int done;
2090};
2091
2092static int cb_txn(fdb_file_handle *fhandle,
2093                  fdb_compaction_status status, const char *kv_name,
2094                  fdb_doc *doc, uint64_t old_offset, uint64_t new_offset,
2095                  void *ctx)
2096{
2097    struct cb_txn_args *args = (struct cb_txn_args *)ctx;
2098    (void) fhandle;
2099    (void) doc;
2100    (void) old_offset;
2101    (void) new_offset;
2102
2103    if (status == FDB_CS_END && !args->done) {
2104        int i;
2105        int n = 10;
2106        char keybuf[256], bodybuf[256];
2107        fdb_status s;
2108
2109        // begin transaction
2110        fdb_begin_transaction(args->file, FDB_ISOLATION_READ_COMMITTED);
2111        // insert new docs, but do not end transaction
2112        for (i=0;i<n/2;++i){
2113            sprintf(keybuf, "txn%04d", i);
2114            sprintf(bodybuf, "txn_body%04d", i);
2115            s = fdb_set_kv(args->handle, keybuf, strlen(keybuf)+1,
2116                                         bodybuf, strlen(bodybuf)+1);
2117            (void)s;
2118        }
2119        args->done = 1;
2120    }
2121
2122    return 0;
2123}
2124
2125void compaction_with_concurrent_transaction_test()
2126{
2127    TEST_INIT();
2128    memleak_start();
2129
2130    int i, r;
2131    int n = 10;
2132    size_t valuelen;
2133    char keybuf[256], bodybuf[256];
2134    fdb_file_handle *dbfile, *txn_file;
2135    fdb_kvs_handle *db, *txn;
2136    fdb_status s;
2137    fdb_config fconfig = fdb_get_default_config();
2138    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2139    void *value;
2140    struct cb_txn_args cb_args;
2141
2142    memset(&cb_args, 0x0, sizeof(struct cb_txn_args));
2143    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2144    fconfig.compaction_cb = cb_txn;
2145    fconfig.compaction_cb_ctx = &cb_args;
2146    fconfig.compaction_cb_mask = FDB_CS_BEGIN |
2147                                 FDB_CS_END;
2148
2149    // remove previous compact_test files
2150    r = system(SHELL_DEL" compact_test* > errorlog.txt");
2151    (void)r;
2152
2153    // open db
2154    fdb_open(&dbfile, "./compact_test1", &fconfig);
2155    fdb_kvs_open(dbfile, &db, "db", &kvs_config);
2156
2157    fdb_open(&txn_file, "./compact_test1", &fconfig);
2158    fdb_kvs_open(txn_file, &txn, "db", &kvs_config);
2159    cb_args.file = txn_file;
2160    cb_args.handle = txn;
2161
2162    // write docs & commit
2163    for (i=0;i<n;++i){
2164        sprintf(keybuf, "key%04d", i);
2165        sprintf(bodybuf, "body%04d", i);
2166        s = fdb_set_kv(db, keybuf, strlen(keybuf)+1,
2167                           bodybuf, strlen(bodybuf)+1);
2168        TEST_CHK(s == FDB_RESULT_SUCCESS);
2169    }
2170    s = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2171    TEST_CHK(s == FDB_RESULT_SUCCESS);
2172    cb_args.ndocs = n;
2173    cb_args.nupdates = 2;
2174
2175    s = fdb_compact(dbfile, "./compact_test2");
2176    TEST_CHK(s == FDB_RESULT_SUCCESS);
2177
2178    // insert new docs through transaction
2179    for (i=n/2;i<n;++i){
2180        sprintf(keybuf, "txn%04d", i);
2181        sprintf(bodybuf, "txn_body%04d", i);
2182        s = fdb_set_kv(txn, keybuf, strlen(keybuf)+1,
2183                            bodybuf, strlen(bodybuf)+1);
2184        TEST_CHK(s == FDB_RESULT_SUCCESS);
2185    }
2186    // all txn docs should be retrieved
2187    for (i=0;i<n;++i){
2188        sprintf(keybuf, "txn%04d", i);
2189        sprintf(bodybuf, "txn_body%04d", i);
2190        s = fdb_get_kv(txn, keybuf, strlen(keybuf)+1,
2191                            &value, &valuelen);
2192        TEST_CHK(s == FDB_RESULT_SUCCESS);
2193        TEST_CMP(value, bodybuf, valuelen);
2194        fdb_free_block(value);
2195    }
2196    s = fdb_end_transaction(txn_file, FDB_COMMIT_MANUAL_WAL_FLUSH);
2197    TEST_CHK(s == FDB_RESULT_SUCCESS);
2198    fdb_close(txn_file);
2199
2200    s = fdb_compact(dbfile, "./compact_test3");
2201    TEST_CHK(s == FDB_RESULT_SUCCESS);
2202
2203    fdb_close(dbfile);
2204    fdb_shutdown();
2205    memleak_end();
2206    TEST_RESULT("compaction with concurrent transaction test");
2207}
2208
2209struct cb_upt_args {
2210    fdb_file_handle *file;
2211    fdb_kvs_handle *handle;
2212    fdb_kvs_handle *handle2;
2213    int ndocs;
2214    int nupdates;
2215    int done;
2216    int nmoves;
2217};
2218
2219static fdb_compact_decision cb_upt(fdb_file_handle *fhandle,
2220                                   fdb_compaction_status status,
2221                                   const char *kv_name,
2222                                   fdb_doc *doc, uint64_t old_offset,
2223                                   uint64_t new_offset,
2224                                   void *ctx)
2225{
2226    TEST_INIT();
2227    struct cb_upt_args *args = (struct cb_upt_args *)ctx;
2228    (void) fhandle;
2229    (void) doc;
2230    (void) old_offset;
2231    (void) new_offset;
2232
2233    if (status == FDB_CS_MOVE_DOC) {
2234        int i, j;
2235        int n = 10;
2236        char keybuf[256], bodybuf[256];
2237        char keystr[] = "new%04d";
2238        char valuestr[] = "new_body%04d";
2239        fdb_status s;
2240        args->nmoves++;
2241        if (args->nmoves > n/2 && !args->done) {
2242            // verify if the key is stripped off its prefix
2243            fdb_doc tdoc;
2244            memset(&tdoc, 0, sizeof(fdb_doc));
2245            tdoc.key = doc->key;
2246            tdoc.keylen = doc->keylen;
2247            s = fdb_get(args->handle, &tdoc);
2248            TEST_CHK(s == FDB_RESULT_SUCCESS);
2249            free(tdoc.meta);
2250            free(tdoc.body);
2251            // insert new docs
2252            for (i=0;i<n/2;++i){
2253                sprintf(keybuf, keystr, i);
2254                sprintf(bodybuf, valuestr, i);
2255                s = fdb_set_kv(args->handle, keybuf, strlen(keybuf)+1,
2256                                             bodybuf, strlen(bodybuf)+1);
2257                TEST_CHK(s == FDB_RESULT_SUCCESS);
2258                s = fdb_commit(args->file, FDB_COMMIT_NORMAL);
2259                TEST_CHK(s == FDB_RESULT_SUCCESS);
2260
2261                for (j=0; j<=i; ++j){
2262                    void *v_out;
2263                    size_t vlen_out;
2264                    sprintf(keybuf, keystr, i);
2265                    sprintf(bodybuf, valuestr, i);
2266                    s = fdb_get_kv(args->handle2, keybuf, strlen(keybuf)+1,
2267                        &v_out, &vlen_out);
2268                    TEST_CHK(s == FDB_RESULT_SUCCESS);
2269                    fdb_free_block(v_out);
2270                }
2271            }
2272            args->done = 1;
2273        }
2274        if (args->nmoves > n && args->done == 1) {
2275            // the first phase is done. insert new docs.
2276            for (i=0;i<2;++i){
2277                sprintf(keybuf, "xxx%d", i);
2278                sprintf(bodybuf, "xxxvalue%d", i);
2279                s = fdb_set_kv(args->handle, keybuf, strlen(keybuf)+1,
2280                                             bodybuf, strlen(bodybuf)+1);
2281                TEST_CHK(s == FDB_RESULT_SUCCESS);
2282                s = fdb_commit(args->file, FDB_COMMIT_NORMAL);
2283                TEST_CHK(s == FDB_RESULT_SUCCESS);
2284            }
2285            args->done = 2;
2286        }
2287        if (args->nmoves == (n*3/2 + 2) && args->done == 2) {
2288            // during the second-second phase,
2289            // insert new docs, and do not commit.
2290            for (i=0;i<2;++i){
2291                sprintf(keybuf, "zzz%d", i);
2292                sprintf(bodybuf, "zzzvalue%d", i);
2293                s = fdb_set_kv(args->handle, keybuf, strlen(keybuf)+1,
2294                                             bodybuf, strlen(bodybuf)+1);
2295                TEST_CHK(s == FDB_RESULT_SUCCESS);
2296            }
2297            args->done = 3;
2298        }
2299        TEST_CMP(kv_name, "db", 2);
2300    }
2301
2302    return 0;
2303}
2304
2305void compaction_with_concurrent_update_test()
2306{
2307    TEST_INIT();
2308    memleak_start();
2309
2310    int i, r;
2311    int n = 10;
2312    char keybuf[256], bodybuf[256];
2313    fdb_file_handle *dbfile, *dbfile2, *dbfile3;
2314    fdb_kvs_handle *db, *db2, *db3;
2315    fdb_status s;
2316    fdb_config fconfig = fdb_get_default_config();
2317    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2318    fdb_kvs_info info;
2319    fdb_seqnum_t seqnum;
2320    struct cb_upt_args cb_args;
2321    void *value;
2322    size_t valuelen;
2323
2324    memset(&cb_args, 0x0, sizeof(struct cb_upt_args));
2325    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2326    fconfig.compaction_cb = cb_upt;
2327    fconfig.compaction_cb_ctx = &cb_args;
2328    fconfig.compaction_cb_mask = FDB_CS_BEGIN |
2329                                 FDB_CS_END | FDB_CS_MOVE_DOC;
2330
2331    // remove previous compact_test files
2332    r = system(SHELL_DEL" compact_test* > errorlog.txt");
2333    (void)r;
2334
2335    // open db
2336    fdb_open(&dbfile, "./compact_test1", &fconfig);
2337    fdb_kvs_open(dbfile, &db, "db", &kvs_config);
2338
2339    fdb_open(&dbfile2, "./compact_test1", &fconfig);
2340    fdb_kvs_open(dbfile2, &db2, "db", &kvs_config);
2341    fdb_open(&dbfile3, "./compact_test1", &fconfig);
2342    fdb_kvs_open(dbfile3, &db3, "db", &kvs_config);
2343
2344    cb_args.file = dbfile2;
2345    cb_args.handle = db2;
2346    cb_args.handle2 = db3;
2347
2348    // write docs & commit
2349    for (i=0;i<n;++i){
2350        sprintf(keybuf, "key%04d", i);
2351        sprintf(bodybuf, "body%04d", i);
2352        s = fdb_set_kv(db, keybuf, strlen(keybuf)+1,
2353                           bodybuf, strlen(bodybuf)+1);
2354        TEST_CHK(s == FDB_RESULT_SUCCESS);
2355    }
2356    s = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2357    TEST_CHK(s == FDB_RESULT_SUCCESS);
2358    cb_args.ndocs = n;
2359    cb_args.nupdates = 2;
2360    cb_args.nmoves = 0;
2361
2362    s = fdb_compact(dbfile, "./compact_test2");
2363    TEST_CHK(s == FDB_RESULT_SUCCESS);
2364
2365    // insert new docs after compaction
2366    for (i=n/2;i<n;++i){
2367        sprintf(keybuf, "new%04d", i);
2368        sprintf(bodybuf, "new_body%04d", i);
2369        s = fdb_set_kv(db2, keybuf, strlen(keybuf)+1,
2370                            bodybuf, strlen(bodybuf)+1);
2371        TEST_CHK(s == FDB_RESULT_SUCCESS);
2372    }
2373    // all interleaved docs should be retrieved
2374    // 1. inserted during the first phase of the compaction
2375    for (i=0;i<n;++i){
2376        sprintf(keybuf, "new%04d", i);
2377        sprintf(bodybuf, "new_body%04d", i);
2378        s = fdb_get_kv(db2, keybuf, strlen(keybuf)+1,
2379                            &value, &valuelen);
2380        TEST_CHK(s == FDB_RESULT_SUCCESS);
2381        TEST_CMP(value, bodybuf, valuelen);
2382        fdb_free_block(value);
2383    }
2384    // 2. inserted during the second phase of the compaction
2385    for (i=0; i<2; ++i) {
2386        sprintf(keybuf, "xxx%d", i);
2387        sprintf(bodybuf, "xxxvalue%d", i);
2388        s = fdb_get_kv(db2, keybuf, strlen(keybuf)+1,
2389                            &value, &valuelen);
2390        TEST_CHK(s == FDB_RESULT_SUCCESS);
2391        TEST_CMP(value, bodybuf, valuelen);
2392        fdb_free_block(value);
2393
2394        sprintf(keybuf, "zzz%d", i);
2395        sprintf(bodybuf, "zzzvalue%d", i);
2396        s = fdb_get_kv(db2, keybuf, strlen(keybuf)+1,
2397                            &value, &valuelen);
2398        TEST_CHK(s == FDB_RESULT_SUCCESS);
2399        TEST_CMP(value, bodybuf, valuelen);
2400        fdb_free_block(value);
2401    }
2402
2403    fdb_close(dbfile2);
2404    fdb_close(dbfile3);
2405
2406    s = fdb_get_kvs_info(db, &info);
2407    TEST_CHK(s == FDB_RESULT_SUCCESS);
2408    seqnum = info.last_seqnum;
2409
2410    cb_args.nmoves = 0;
2411    s = fdb_compact(dbfile, "./compact_test3");
2412    TEST_CHK(s == FDB_RESULT_SUCCESS);
2413
2414    s = fdb_get_kvs_info(db, &info);
2415    TEST_CHK(s == FDB_RESULT_SUCCESS);
2416    // No docs are inserted during the compaction.
2417    // Seqnum should be same.
2418    TEST_CHK(info.last_seqnum == seqnum);
2419
2420    // insert the last document
2421    sprintf(keybuf, "last");
2422    sprintf(bodybuf, "last_value");
2423    s = fdb_set_kv(db, keybuf, strlen(keybuf)+1, bodybuf, strlen(bodybuf)+1);
2424    TEST_CHK(s == FDB_RESULT_SUCCESS);
2425    s = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2426    TEST_CHK(s == FDB_RESULT_SUCCESS);
2427
2428    fdb_close(dbfile);
2429    fdb_shutdown();
2430    memleak_end();
2431    TEST_RESULT("compaction with concurrent update test");
2432}
2433
2434static int compaction_del_cb(fdb_file_handle *fhandle,
2435                             fdb_compaction_status status,
2436                             const char *kv_name,
2437                             fdb_doc *doc, uint64_t old_offset,
2438                             uint64_t new_offset,
2439                             void *ctx)
2440{
2441    TEST_INIT();
2442    fdb_status s;
2443    fdb_file_handle *dbfile;
2444    fdb_kvs_handle *db = *(fdb_kvs_handle **)ctx;
2445    int i;
2446    int n = 10; // if changing this change caller function too
2447    char keybuf[256];
2448    (void) doc;
2449    (void) new_offset;
2450    (void) old_offset;
2451
2452    if (status == FDB_CS_BEGIN) {
2453        TEST_CHK(false);
2454    } else if (status == FDB_CS_END) {
2455        TEST_CHK(!kv_name);
2456        if (db) { // At end of first phase, mutate more docs...
2457            s = fdb_open(&dbfile, "./compact_test1", &fhandle->root->config);
2458            TEST_CHK(s == FDB_RESULT_SUCCESS);
2459            fdb_kvs_open_default(dbfile, &db, &db->kvs_config);
2460            for (i = 0; i < n; ++i){
2461                sprintf(keybuf, "key%d", i);
2462                s = fdb_del_kv(db, keybuf, strlen(keybuf));
2463                TEST_CHK(s == FDB_RESULT_SUCCESS);
2464            }
2465            // Now insert and delete a bunch of keys (all in WAL)
2466            for (i = 0; i < n; ++i){
2467                sprintf(keybuf, "KEY%d", i);
2468                s = fdb_set_kv(db, keybuf, strlen(keybuf), NULL, 0);
2469                TEST_CHK(s == FDB_RESULT_SUCCESS);
2470                s = fdb_del_kv(db, keybuf, strlen(keybuf));
2471                TEST_CHK(s == FDB_RESULT_SUCCESS);
2472            }
2473            sprintf(keybuf, "key%d", i);
2474            s = fdb_set_kv(db, keybuf, strlen(keybuf), NULL, 0);
2475            TEST_CHK(s == FDB_RESULT_SUCCESS);
2476            s = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2477            TEST_CHK(s == FDB_RESULT_SUCCESS);
2478            fdb_close(dbfile);
2479            *(fdb_kvs_handle **)ctx = NULL; // Don't run in 2nd, 3rd phases
2480        }
2481    } else if (status == FDB_CS_FLUSH_WAL) {
2482        TEST_CHK(false);
2483    } else if (status == FDB_CS_MOVE_DOC) {
2484        TEST_CHK(false);
2485    } else { // FDB_CS_BATCH_MOVE
2486        TEST_CHK(false);
2487    }
2488    return 0;
2489}
2490
2491void compact_deleted_doc_test()
2492{
2493    TEST_INIT();
2494    memleak_start();
2495    int i, r;
2496    int n = 10; // if changing this change the callback too
2497    fdb_file_handle *dbfile;
2498    fdb_kvs_handle *db, *cb_db;
2499    fdb_doc *rdoc;
2500    fdb_status s;
2501    char keybuf[256];
2502    void *value;
2503    size_t valuelen;
2504
2505    // remove previous compact_test files
2506    r = system(SHELL_DEL " compact_test* > errorlog.txt");
2507    (void)r;
2508
2509    fdb_config fconfig = fdb_get_default_config();
2510    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2511    fconfig.multi_kv_instances = false;
2512    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2513
2514    fconfig.compaction_cb = compaction_del_cb;
2515    fconfig.compaction_cb_ctx = &cb_db;
2516    fconfig.compaction_cb_mask = FDB_CS_END;
2517
2518    // open db
2519    fdb_open(&dbfile, "./compact_test1", &fconfig);
2520    fdb_kvs_open_default(dbfile, &db, &kvs_config);
2521    cb_db = db;
2522
2523    for (i=0;i<n;++i){
2524        sprintf(keybuf, "key%d", i);
2525        s = fdb_set_kv(db, keybuf, strlen(keybuf),
2526                            (void*)"value", 5);
2527        TEST_CHK(s == FDB_RESULT_SUCCESS);
2528    }
2529    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2530
2531    // At end of phase 1, all documents get deleted
2532    s = fdb_compact(dbfile, "compact_test2");
2533    TEST_CHK(s == FDB_RESULT_SUCCESS);
2534    TEST_CHK(wal_get_num_flushable(dbfile->root->file) == 0);
2535
2536    for (i=0;i<n;++i){
2537        sprintf(keybuf, "key%d", i);
2538        fdb_doc_create(&rdoc, keybuf, strlen(keybuf), NULL, 0, NULL, 0);
2539        s = fdb_get_metaonly(db, rdoc);
2540        TEST_CHK(s == FDB_RESULT_KEY_NOT_FOUND);
2541        fdb_doc_free(rdoc);
2542    }
2543
2544    // documents inserted in the middle of phase 1 should be present
2545    sprintf(keybuf, "key%d", i);
2546    s = fdb_get_kv(db, keybuf, strlen(keybuf), &value, &valuelen);
2547    TEST_CHK(s == FDB_RESULT_SUCCESS);
2548    fdb_free_block(value);
2549
2550    fdb_kvs_close(db);
2551    fdb_close(dbfile);
2552    fdb_shutdown();
2553    memleak_end();
2554    TEST_RESULT("compact deleted doc test");
2555}
2556
2557void compact_upto_twice_test()
2558{
2559    TEST_INIT();
2560    memleak_start();
2561    int i, r;
2562    fdb_file_handle *dbfile;
2563    fdb_kvs_handle *db;
2564    fdb_snapshot_info_t *markers;
2565    fdb_status status;
2566    uint64_t num_markers;
2567    char keybuf[256];
2568
2569    // remove previous compact_test files
2570    r = system(SHELL_DEL " compact_test* > errorlog.txt");
2571    (void)r;
2572
2573    fdb_config fconfig = fdb_get_default_config();
2574    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2575    fconfig.wal_threshold = 1024;
2576    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2577
2578    // open db
2579    fdb_open(&dbfile, "./compact_test1", &fconfig);
2580    fdb_kvs_open_default(dbfile, &db, &kvs_config);
2581
2582    for (i=0;i<10;++i){
2583        sprintf(keybuf, "key%d", i);
2584        status = fdb_set_kv(db, keybuf, strlen(keybuf),
2585                            (void*)"value", 5);
2586        TEST_CHK(status == FDB_RESULT_SUCCESS);
2587        fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2588    }
2589
2590    // compact upto twice with incrementing seqnums
2591    status = fdb_get_all_snap_markers(dbfile, &markers,
2592                                      &num_markers);
2593    TEST_CHK(status == FDB_RESULT_SUCCESS);
2594    status = fdb_compact_upto(dbfile, NULL, markers[num_markers-1].marker);
2595    TEST_CHK(status == FDB_RESULT_SUCCESS);
2596    status = fdb_free_snap_markers(markers, num_markers);
2597    TEST_CHK(status == FDB_RESULT_SUCCESS);
2598
2599    // compact upto
2600    status = fdb_get_all_snap_markers(dbfile, &markers,
2601                                      &num_markers);
2602    TEST_CHK(status == FDB_RESULT_SUCCESS);
2603    status = fdb_compact_upto(dbfile, NULL, markers[num_markers-2].marker);
2604    TEST_CHK(status == FDB_RESULT_SUCCESS);
2605    status = fdb_free_snap_markers(markers, num_markers);
2606    TEST_CHK(status == FDB_RESULT_SUCCESS);
2607
2608    fdb_close(dbfile);
2609    fdb_shutdown();
2610    memleak_end();
2611    TEST_RESULT("compact upto twice");
2612}
2613
2614void wal_delete_compact_upto_test()
2615{
2616    TEST_INIT();
2617    memleak_start();
2618    int i, r;
2619    fdb_file_handle *dbfile;
2620    fdb_kvs_handle *db;
2621    fdb_snapshot_info_t *markers;
2622    fdb_status status;
2623    uint64_t num_markers;
2624    fdb_kvs_info kvs_info;
2625    char keybuf[256];
2626    void *value;
2627    size_t valuelen;
2628
2629    // remove previous compact_test files
2630    r = system(SHELL_DEL " compact_test* > errorlog.txt");
2631    (void)r;
2632
2633    fdb_config fconfig = fdb_get_default_config();
2634    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2635    fconfig.wal_threshold = 19;
2636    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2637
2638    // open db
2639    fdb_open(&dbfile, "./compact_test5", &fconfig);
2640    fdb_kvs_open_default(dbfile, &db, &kvs_config);
2641
2642    // insert a few keys...
2643    for (i=0;i<10;++i){
2644        sprintf(keybuf, "key%d", i);
2645        status = fdb_set_kv(db, keybuf, strlen(keybuf),
2646                            (void*)"value", 5);
2647        TEST_CHK(status == FDB_RESULT_SUCCESS);
2648    }
2649
2650    status = fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2651    TEST_CHK(status == FDB_RESULT_SUCCESS);
2652
2653    // insert keys such that last insert causes wal flush...
2654    for (; i<20;++i){
2655        sprintf(keybuf, "key%d", i);
2656        status = fdb_set_kv(db, keybuf, strlen(keybuf),
2657                            (void*)"value", 5);
2658        TEST_CHK(status == FDB_RESULT_SUCCESS);
2659    }
2660    // Now delete half of the newly inserted keys
2661    for (i = 15; i < 20; ++i){
2662        sprintf(keybuf, "key%d", i);
2663        status = fdb_del_kv(db, keybuf, strlen(keybuf));
2664        TEST_CHK(status == FDB_RESULT_SUCCESS);
2665    }
2666    // Now deleted items are in the unflushed wal..
2667    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2668    TEST_CHK(status == FDB_RESULT_SUCCESS);
2669
2670    // Create another header for compact_upto()
2671    sprintf(keybuf, "key%d", i);
2672    status = fdb_set_kv(db, keybuf, strlen(keybuf),
2673            (void*)"value", 5);
2674    TEST_CHK(status == FDB_RESULT_SUCCESS);
2675    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2676    TEST_CHK(status == FDB_RESULT_SUCCESS);
2677
2678    // Get all snap markers..
2679    status = fdb_get_all_snap_markers(dbfile, &markers,
2680                                      &num_markers);
2681    TEST_CHK(status == FDB_RESULT_SUCCESS);
2682
2683    // compact upto the delete wal's marker...
2684    status = fdb_compact_upto(dbfile, NULL, markers[1].marker);
2685    TEST_CHK(status == FDB_RESULT_SUCCESS);
2686    status = fdb_free_snap_markers(markers, num_markers);
2687    TEST_CHK(status == FDB_RESULT_SUCCESS);
2688
2689    status = fdb_get_kvs_info(db, &kvs_info);
2690    TEST_CHK(status == FDB_RESULT_SUCCESS);
2691
2692    // Deleted items should not be found..
2693    for (i = 15; i < 20; ++i){
2694        sprintf(keybuf, "key%d", i);
2695        status = fdb_get_kv(db, keybuf, strlen(keybuf), &value, &valuelen);
2696        TEST_CHK(status == FDB_RESULT_KEY_NOT_FOUND);
2697    }
2698
2699    fdb_kvs_close(db);
2700    fdb_close(dbfile);
2701    fdb_shutdown();
2702    memleak_end();
2703    TEST_RESULT("compact upto with wal deletes test");
2704}
2705
2706void compact_upto_post_snapshot_test()
2707{
2708    TEST_INIT();
2709    memleak_start();
2710    int i, r;
2711    fdb_file_handle *dbfile;
2712    fdb_kvs_handle *db, *snap_db;
2713    fdb_snapshot_info_t *markers;
2714    fdb_status status;
2715    fdb_iterator *iterator;
2716    fdb_doc *rdoc = NULL;
2717    fdb_kvs_info kvs_info;
2718    uint64_t num_markers;
2719    char keybuf[256];
2720
2721    // remove previous compact_test files
2722    r = system(SHELL_DEL " compact_test* > errorlog.txt");
2723    (void)r;
2724
2725    fdb_config fconfig = fdb_get_default_config();
2726    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2727    fconfig.wal_threshold = 1024;
2728    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2729    // prevent block reusing to keep snapshots
2730    fconfig.block_reusing_threshold = 0;
2731
2732    // open db
2733    fdb_open(&dbfile, "./compact_test1", &fconfig);
2734    fdb_kvs_open_default(dbfile, &db, &kvs_config);
2735
2736    for (i=0;i<10;++i){
2737        sprintf(keybuf, "key%d", i);
2738        status = fdb_set_kv(db, keybuf, strlen(keybuf),
2739                            (void*)"value", 5);
2740        TEST_CHK(status == FDB_RESULT_SUCCESS);
2741        fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2742    }
2743
2744    // check  db info
2745    status = fdb_get_kvs_info(db, &kvs_info);
2746    TEST_CHK(status == FDB_RESULT_SUCCESS);
2747    TEST_CHK(kvs_info.last_seqnum == 10);
2748
2749    // open snapshot at seqnum 5
2750    status = fdb_snapshot_open(db, &snap_db, 5);
2751    TEST_CHK(status == FDB_RESULT_SUCCESS);
2752
2753    // check snap info
2754    status = fdb_get_kvs_info(snap_db, &kvs_info);
2755    TEST_CHK(status == FDB_RESULT_SUCCESS);
2756    TEST_CHK(kvs_info.last_seqnum == 5);
2757
2758
2759    // compact upto
2760    status = fdb_get_all_snap_markers(dbfile, &markers,
2761                                      &num_markers);
2762    TEST_CHK(status == FDB_RESULT_SUCCESS);
2763    status = fdb_compact_upto(dbfile, NULL, markers[5].marker);
2764    TEST_CHK(status == FDB_RESULT_SUCCESS);
2765    status = fdb_free_snap_markers(markers, num_markers);
2766    TEST_CHK(status == FDB_RESULT_SUCCESS);
2767
2768    // check db and snap info
2769    status = fdb_get_kvs_info(db, &kvs_info);
2770    TEST_CHK(status == FDB_RESULT_SUCCESS);
2771    TEST_CHK(kvs_info.last_seqnum == 10);
2772    status = fdb_get_kvs_info(snap_db, &kvs_info);
2773    TEST_CHK(status == FDB_RESULT_SUCCESS);
2774    TEST_CHK(kvs_info.last_seqnum == 5);
2775
2776
2777    // iterate over snapshot
2778    status = fdb_iterator_init(snap_db,
2779                               &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
2780    TEST_CHK(status == FDB_RESULT_SUCCESS);
2781    i = 0;
2782    do {
2783        status = fdb_iterator_get(iterator, &rdoc);
2784        TEST_CHK(status == FDB_RESULT_SUCCESS);
2785        fdb_doc_free(rdoc);
2786        rdoc = NULL;
2787        i++;
2788
2789        // check db and snap info
2790        status = fdb_get_kvs_info(db, &kvs_info);
2791        TEST_CHK(status == FDB_RESULT_SUCCESS);
2792        TEST_CHK(kvs_info.last_seqnum == 10);
2793        status = fdb_get_kvs_info(snap_db, &kvs_info);
2794        TEST_CHK(status == FDB_RESULT_SUCCESS);
2795        TEST_CHK(kvs_info.last_seqnum == 5);
2796    } while (fdb_iterator_next(iterator) != FDB_RESULT_ITERATOR_FAIL);
2797
2798    TEST_CHK(i == 5);
2799
2800    fdb_iterator_close(iterator);
2801    fdb_kvs_close(snap_db);
2802    fdb_close(dbfile);
2803    fdb_shutdown();
2804    memleak_end();
2805    TEST_RESULT("compact upto post snapshot test");
2806}
2807
2808void compact_upto_overwrite_test(int opt)
2809{
2810    TEST_INIT();
2811
2812    int n = 10, value_len=32;
2813    int i, r, idx, c;
2814    char cmd[256];
2815    char key[256], *value;
2816    char keystr[] = "key%06d";
2817    char valuestr[] = "value%08d";
2818    char valuestr2[] = "updated_value%08d";
2819    fdb_file_handle *db_file;
2820    fdb_kvs_handle *db, *snap;
2821    fdb_config config;
2822    fdb_kvs_config kvs_config;
2823    fdb_status s;
2824    fdb_kvs_info kvs_info;
2825    fdb_iterator *fit;
2826    fdb_doc *doc;
2827    fdb_snapshot_info_t *markers;
2828    fdb_seqnum_t seqnum;
2829    fdb_commit_opt_t commit_opt;
2830    uint64_t n_markers;
2831
2832    sprintf(cmd, SHELL_DEL " compact_test* > errorlog.txt");
2833    r = system(cmd);
2834    (void)r;
2835
2836    memleak_start();
2837
2838    value = (char*)malloc(value_len);
2839
2840    config = fdb_get_default_config();
2841    config.durability_opt = FDB_DRB_ASYNC;
2842    config.seqtree_opt = FDB_SEQTREE_USE;
2843    config.wal_flush_before_commit = true;
2844    config.multi_kv_instances = true;
2845    config.buffercache_size = 0;
2846
2847    commit_opt = (opt)?FDB_COMMIT_NORMAL:FDB_COMMIT_MANUAL_WAL_FLUSH;
2848
2849    kvs_config = fdb_get_default_kvs_config();
2850
2851    s = fdb_open(&db_file, "./compact_test", &config);
2852    TEST_CHK(s == FDB_RESULT_SUCCESS);
2853    s = fdb_kvs_open(db_file, &db, "db", &kvs_config);
2854    TEST_CHK(s == FDB_RESULT_SUCCESS);
2855
2856    // === write ===
2857    for (i=0;i<n;++i){
2858        idx = i % (n/2);
2859        sprintf(key, keystr, idx);
2860        memset(value, 'x', value_len);
2861        memcpy(value + value_len - 6, "<end>", 6);
2862        if (i < (n/2)) {
2863            sprintf(value, valuestr, idx);
2864        } else {
2865            sprintf(value, valuestr2, idx);
2866        }
2867        s = fdb_set_kv(db, key, strlen(key)+1, value, value_len);
2868        TEST_CHK(s == FDB_RESULT_SUCCESS);
2869
2870        if (opt == 2) {
2871            // HB+trie, WAL, HB+trie, WAL...
2872            commit_opt = (i%2)?FDB_COMMIT_NORMAL:FDB_COMMIT_MANUAL_WAL_FLUSH;
2873        } else if (opt == 3) {
2874            // WAL, HB+trie, WAL, HB+trie...
2875            commit_opt = (i%2)?FDB_COMMIT_MANUAL_WAL_FLUSH:FDB_COMMIT_NORMAL;
2876        }
2877
2878        s = fdb_commit(db_file, commit_opt);
2879        TEST_CHK(s == FDB_RESULT_SUCCESS);
2880        s = fdb_get_kvs_info(db, &kvs_info);
2881        TEST_CHK(s == FDB_RESULT_SUCCESS);
2882    }
2883
2884    s = fdb_get_all_snap_markers(db_file, &markers, &n_markers);
2885    TEST_CHK(s == FDB_RESULT_SUCCESS);
2886
2887    int upto = n_markers/2;
2888    s = fdb_compact_upto(db_file, "./compact_test2", markers[upto].marker);
2889    TEST_CHK(s == FDB_RESULT_SUCCESS);
2890
2891    // iterating using snapshots with various seqnums
2892    for (i=n_markers-1; i>=0; --i) {
2893        seqnum = markers[i].kvs_markers->seqnum;
2894
2895        s = fdb_snapshot_open(db, &snap, seqnum);
2896        if (i<=upto) {
2897            TEST_CHK(s == FDB_RESULT_SUCCESS);
2898        } else {
2899            // seqnum < (n/2) must fail
2900            TEST_CHK(s != FDB_RESULT_SUCCESS);
2901            continue;
2902        }
2903
2904        s = fdb_iterator_init(snap, &fit, NULL, 0, NULL, 0, 0);
2905        TEST_CHK(s == FDB_RESULT_SUCCESS);
2906        c = 0;
2907        do {
2908            doc = NULL;
2909            s = fdb_iterator_get(fit, &doc);
2910            if (s != FDB_RESULT_SUCCESS) break;
2911            idx = c;
2912            sprintf(key, keystr, idx);
2913            memset(value, 'x', value_len);
2914            memcpy(value + value_len - 6, "<end>", 6);
2915            if ((fdb_seqnum_t)c >= seqnum-(n/2)) {
2916                sprintf(value, valuestr, idx);
2917            } else {
2918                sprintf(value, valuestr2, idx);
2919            }
2920            TEST_CMP(doc->key, key, doc->keylen);
2921            TEST_CMP(doc->body, value, doc->bodylen);
2922            c++;
2923            fdb_doc_free(doc);
2924        } while(fdb_iterator_next(fit) == FDB_RESULT_SUCCESS);
2925        s = fdb_iterator_close(fit);
2926        TEST_CHK(s == FDB_RESULT_SUCCESS);
2927        TEST_CHK(c == (n/2));
2928
2929        s = fdb_kvs_close(snap);
2930        TEST_CHK(s == FDB_RESULT_SUCCESS);
2931    }
2932
2933    // iterating using the original handle
2934    s = fdb_iterator_init(db, &fit, NULL, 0, NULL, 0, 0);
2935    TEST_CHK(s == FDB_RESULT_SUCCESS);
2936    c = 0;
2937    do {
2938        doc = NULL;
2939        s = fdb_iterator_get(fit, &doc);
2940        if (s != FDB_RESULT_SUCCESS) break;
2941        idx = c;
2942        sprintf(key, keystr, idx);
2943        memset(value, 'x', value_len);
2944        memcpy(value + value_len - 6, "<end>", 6);
2945        sprintf(value, valuestr2, idx);
2946        TEST_CMP(doc->key, key, doc->keylen);
2947        TEST_CMP(doc->body, value, doc->bodylen);
2948        c++;
2949        fdb_doc_free(doc);
2950    } while(fdb_iterator_next(fit) == FDB_RESULT_SUCCESS);
2951    s = fdb_iterator_close(fit);
2952    TEST_CHK(s == FDB_RESULT_SUCCESS);
2953    TEST_CHK(c == (n/2));
2954
2955    s = fdb_free_snap_markers(markers, n_markers);
2956    TEST_CHK(s == FDB_RESULT_SUCCESS);
2957    free(value);
2958    s = fdb_close(db_file);
2959    TEST_CHK(s == FDB_RESULT_SUCCESS);
2960    s = fdb_shutdown();
2961    TEST_CHK(s == FDB_RESULT_SUCCESS);
2962
2963    memleak_end();
2964
2965    sprintf(cmd, "compact upto overwrite test");
2966    if (opt == 0) {
2967        strcat(cmd, " (HB+trie)");
2968    } else if (opt == 1) {
2969        strcat(cmd, " (WAL)");
2970    } else if (opt == 2) {
2971        strcat(cmd, " (mixed, HB+trie/WAL)");
2972    } else if (opt == 3) {
2973        strcat(cmd, " (mixed, WAL/HB+trie)");
2974    }
2975    TEST_RESULT(cmd);
2976}
2977static int compaction_cb_get(fdb_file_handle *fhandle,
2978                         fdb_compaction_status status, const char *kv_name,
2979                         fdb_doc *doc, uint64_t old_offset,
2980                         uint64_t new_offset, void *ctx)
2981{
2982    TEST_INIT();
2983    fdb_status s;
2984    struct cb_args *args = (struct cb_args *)ctx;
2985    fdb_kvs_handle *snap_db;
2986    fdb_snapshot_info_t *markers;
2987    uint64_t num_markers;
2988    fdb_seqnum_t seqno;
2989
2990    (void) doc;
2991    (void) old_offset;
2992    (void) new_offset;
2993
2994    if (status == FDB_CS_MOVE_DOC) {
2995        TEST_CHK(kv_name);
2996    } else {
2997        TEST_CHK(!kv_name);
2998    }
2999
3000    // use snap markers
3001    s = fdb_get_all_snap_markers(fhandle, &markers, &num_markers);
3002    TEST_CHK(s == FDB_RESULT_SUCCESS);
3003    seqno = markers[0].kvs_markers[0].seqnum;
3004
3005    // snapshot open
3006    s = fdb_snapshot_open(args->handle, &snap_db, seqno);
3007    TEST_CHK(s==FDB_RESULT_SUCCESS);
3008    s = fdb_kvs_close(snap_db);
3009    TEST_CHK(s==FDB_RESULT_SUCCESS);
3010    return 0;
3011}
3012
3013void compact_with_snapshot_open_test()
3014{
3015  TEST_INIT();
3016  memleak_start();
3017
3018  int i, r;
3019  int n = 100000;
3020  char keybuf[256], bodybuf[256];
3021  fdb_file_handle *dbfile;
3022  fdb_kvs_handle *db, *db2, *snap_db;
3023  fdb_status s;
3024  fdb_config fconfig = fdb_get_default_config();
3025  fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
3026  struct cb_args cb_args;
3027
3028  memset(&cb_args, 0x0, sizeof(struct cb_args));
3029  fconfig.wal_threshold = 1024;
3030  fconfig.flags = FDB_OPEN_FLAG_CREATE;
3031  fconfig.compaction_cb = compaction_cb_get;
3032  fconfig.compaction_cb_ctx = &cb_args;
3033  fconfig.compaction_cb_mask = FDB_CS_BEGIN |
3034                               FDB_CS_MOVE_DOC |
3035                               FDB_CS_FLUSH_WAL |
3036                               FDB_CS_END;
3037  // remove previous compact_test files
3038  r = system(SHELL_DEL" compact_test* > errorlog.txt");
3039  (void)r;
3040
3041  // open two handles for kvs
3042  fdb_open(&dbfile, "./compact_test1", &fconfig);
3043  fdb_kvs_open(dbfile, &db, "db", &kvs_config);
3044  fdb_kvs_open(dbfile, &db2, "db", &kvs_config);
3045  for (i=0;i<n;++i){
3046      sprintf(keybuf, "key%04d", i);
3047      sprintf(bodybuf, "body%04d", i);
3048      s = fdb_set_kv(db, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3049      TEST_CHK(s == FDB_RESULT_SUCCESS);
3050  }
3051
3052
3053  // commit
3054  s = fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
3055  TEST_CHK(s == FDB_RESULT_SUCCESS);
3056
3057  // point compact callback handle to db2
3058  cb_args.handle = db2;
3059
3060  // compact
3061  s = fdb_compact(dbfile, NULL);
3062  TEST_CHK(s == FDB_RESULT_SUCCESS);
3063
3064  // open compaction end
3065  s = fdb_snapshot_open(db2, &snap_db, n);
3066  TEST_CHK(s == FDB_RESULT_SUCCESS);
3067
3068  fdb_kvs_close(snap_db);
3069  s = fdb_close(dbfile);
3070  TEST_CHK(s == FDB_RESULT_SUCCESS);
3071  s = fdb_shutdown();
3072  TEST_CHK(s == FDB_RESULT_SUCCESS);
3073  TEST_RESULT("compact with snapshot_open test");
3074}
3075
3076static int compaction_cb_markers(fdb_file_handle *fhandle,
3077                         fdb_compaction_status status, const char *kv_name,
3078                         fdb_doc *doc, uint64_t old_offset,
3079                         uint64_t new_offset, void *ctx)
3080{
3081    TEST_INIT();
3082    uint64_t i, j;
3083    fdb_status s;
3084    fdb_kvs_handle *db, *snap_db;
3085    fdb_snapshot_info_t *markers;
3086    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
3087    fdb_seqnum_t seqno;
3088    uint64_t num_markers;
3089    struct cb_args *args = (struct cb_args *)ctx;
3090    uint64_t n = (uint64_t) args->n_moved_docs;
3091
3092    (void) status;
3093    (void) doc;
3094    (void) old_offset;
3095    (void) new_offset;
3096
3097    // get snap markers
3098    s = fdb_get_all_snap_markers(fhandle, &markers, &num_markers);
3099    TEST_CHK(s == FDB_RESULT_SUCCESS);
3100    TEST_CHK(markers[0].num_kvs_markers == 5);
3101    seqno = markers[0].kvs_markers[0].seqnum;
3102    TEST_CHK(seqno == n);
3103
3104    // snapshot open for each kvs for each marker
3105    for(j=0;j<num_markers;++j){
3106        for (i=0;i<(uint64_t)markers[j].num_kvs_markers;++i){
3107            // open kv for this marker
3108            fdb_kvs_open(fhandle, &db,
3109                         markers[j].kvs_markers[i].kv_store_name, &kvs_config);
3110            // snapshot the kv
3111            s = fdb_snapshot_open(db, &snap_db, seqno);
3112            TEST_CHK(s==FDB_RESULT_SUCCESS);
3113
3114            s = fdb_kvs_close(snap_db);
3115            TEST_CHK(s==FDB_RESULT_SUCCESS);
3116            s = fdb_kvs_close(db);
3117            TEST_CHK(s==FDB_RESULT_SUCCESS);
3118        }
3119    }
3120    return 0;
3121}
3122
3123void compact_with_snapshot_open_multi_kvs_test()
3124{
3125    TEST_INIT();
3126    memleak_start();
3127
3128    int i, j, r;
3129    int n = 1000;
3130    char keybuf[256], bodybuf[256];
3131    fdb_file_handle *dbfile;
3132    fdb_kvs_handle *db1, *db2, *db3, *db4, *db5;
3133    fdb_status s;
3134    fdb_config fconfig = fdb_get_default_config();
3135    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
3136    struct cb_args cb_args;
3137
3138    memset(&cb_args, 0x0, sizeof(struct cb_args));
3139
3140    fconfig.wal_threshold = 1024;
3141    fconfig.flags = FDB_OPEN_FLAG_CREATE;
3142    fconfig.compaction_cb = compaction_cb_markers;
3143    fconfig.compaction_cb_ctx = &cb_args;
3144    fconfig.compaction_cb_mask = FDB_CS_BEGIN |
3145                                 FDB_CS_MOVE_DOC |
3146                                 FDB_CS_FLUSH_WAL |
3147                                 FDB_CS_END;
3148
3149    // remove previous compact_test files
3150    r = system(SHELL_DEL" compact_test* > errorlog.txt");
3151    (void)r;
3152
3153    // open two handles for kvs
3154    fdb_open(&dbfile, "./compact_test1", &fconfig);
3155    fdb_kvs_open(dbfile, &db1, "db1", &kvs_config);
3156    fdb_kvs_open(dbfile, &db2, "db2", &kvs_config);
3157    fdb_kvs_open(dbfile, &db3, "db3", &kvs_config);
3158    fdb_kvs_open(dbfile, &db4, "db4", &kvs_config);
3159    fdb_kvs_open(dbfile, &db5, "db5", &kvs_config);
3160    for (j=0;j<5;++j){
3161        for (i=0;i<n;++i){
3162            sprintf(keybuf, "key%04d", i);
3163            sprintf(bodybuf, "body%04d", i);
3164            fdb_set_kv(db1, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3165            fdb_set_kv(db2, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3166            fdb_set_kv(db3, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3167            fdb_set_kv(db4, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3168            fdb_set_kv(db5, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3169        }
3170
3171        // commit
3172        s = fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
3173        TEST_CHK(s == FDB_RESULT_SUCCESS);
3174
3175        // compact
3176        cb_args.n_moved_docs = n*(j+1);
3177        s = fdb_compact(dbfile, NULL);
3178        TEST_CHK(s == FDB_RESULT_SUCCESS);
3179    }
3180
3181    s = fdb_close(dbfile);
3182    TEST_CHK(s == FDB_RESULT_SUCCESS);
3183    s = fdb_shutdown();
3184    TEST_CHK(s == FDB_RESULT_SUCCESS);
3185    TEST_RESULT("compact with snapshot_open multi kvs test");
3186}
3187
3188
3189static atomic_uint8_t cancel_test_signal_begin(0);
3190static atomic_uint8_t cancel_test_signal_end(0);
3191
3192static int cb_cancel_test(fdb_file_handle *fhandle,
3193                          fdb_compaction_status status, const char *kv_name,
3194                          fdb_doc *doc, uint64_t old_offset, uint64_t new_offset,
3195                          void *ctx)
3196{
3197    (void) fhandle;
3198    (void) kv_name;
3199    (void) doc;
3200    (void) old_offset;
3201    (void) new_offset;
3202    (void) ctx;
3203
3204    if (status == FDB_CS_BEGIN) {
3205        cancel_test_signal_begin = 1;
3206    } else if (status == FDB_CS_END) {
3207        cancel_test_signal_end = 1;
3208    }
3209
3210    return 0;
3211}
3212
3213void *db_compact_during_compaction_cancellation(void *args)
3214{
3215
3216    TEST_INIT();
3217
3218    fdb_file_handle *dbfile;
3219    fdb_status status;
3220    fdb_config config;
3221
3222    // Open Database File
3223    config = fdb_get_default_config();
3224    config.compaction_cb = cb_cancel_test;
3225    config.compaction_cb_ctx = NULL;
3226    config.compaction_cb_mask = FDB_CS_BEGIN | FDB_CS_END;
3227
3228    status = fdb_open(&dbfile, "compact_test", &config);
3229    TEST_CHK(status == FDB_RESULT_SUCCESS);
3230    // compaction thread enters here
3231    status = fdb_compact(dbfile, NULL);
3232    TEST_CHK(status == FDB_RESULT_SUCCESS ||
3233             status == FDB_RESULT_COMPACTION_CANCELLATION ||
3234             status == FDB_RESULT_FAIL_BY_ROLLBACK);
3235    fdb_close(dbfile);
3236
3237    // shutdown
3238    thread_exit(0);
3239    return NULL;
3240}
3241
3242typedef enum {
3243    COMPACTION_CANCEL_MODE = 0,
3244    COMPACTION_ROLLBACK_MODE = 1
3245} compaction_test_mode;
3246
3247void compaction_cancellation_test(compaction_test_mode mode)
3248{
3249    TEST_INIT();
3250
3251    memleak_start();
3252
3253    int i, r, n=100000;
3254    fdb_file_handle *file;
3255    fdb_kvs_handle *kvs;
3256    fdb_status status;
3257    fdb_config config;
3258    fdb_kvs_config kvs_config;
3259
3260    r = system(SHELL_DEL" compact_test* > errorlog.txt");
3261    (void)r;
3262
3263    // Open Database File
3264    config = fdb_get_default_config();
3265    status = fdb_open(&file, "compact_test", &config);
3266    TEST_CHK(status == FDB_RESULT_SUCCESS);
3267
3268    // Open KV Store
3269    kvs_config = fdb_get_default_kvs_config();
3270    status = fdb_kvs_open_default(file, &kvs, &kvs_config);
3271    TEST_CHK(status == FDB_RESULT_SUCCESS);
3272
3273    // Load kv pairs
3274    for(i=0;i<n;i++) {
3275        char str[15];
3276        sprintf(str, "%d", i);
3277        status = fdb_set_kv(kvs, str, strlen(str), (void*)"value", 5);
3278        TEST_CHK(status == FDB_RESULT_SUCCESS);
3279        // Commit every 100 SETs
3280        if (i % 100 == 0) {
3281            status = fdb_commit(file, FDB_COMMIT_NORMAL);
3282            TEST_CHK(status == FDB_RESULT_SUCCESS);
3283        }
3284    }
3285
3286    thread_t tid;
3287    void *thread_ret;
3288    bool rollback_failed = false;
3289
3290    thread_create(&tid, db_compact_during_compaction_cancellation, NULL);
3291
3292    // wait until compaction begins
3293    while (!cancel_test_signal_begin) {
3294        usleep(1000); // Sleep for 1ms
3295    }
3296
3297    if (mode == COMPACTION_ROLLBACK_MODE) {
3298        // append more mutations
3299        for(i=0;i<n/10;i++) {
3300            char str[15];
3301            sprintf(str, "%d", i);
3302            status = fdb_set_kv(kvs, str, strlen(str), (void*)"value", 5);
3303            TEST_CHK(status == FDB_RESULT_SUCCESS);
3304            // Commit every 100 SETs
3305            if (i % 100 == 0) {
3306                status = fdb_commit(file, FDB_COMMIT_NORMAL);
3307                TEST_CHK(status == FDB_RESULT_SUCCESS);
3308            }
3309        }
3310    }
3311
3312    if (mode == COMPACTION_CANCEL_MODE) {
3313        // Cancel the compaction task
3314        status = fdb_cancel_compaction(file);
3315        TEST_CHK(status == FDB_RESULT_SUCCESS);
3316    } else if (mode == COMPACTION_ROLLBACK_MODE) {
3317        // Rollback
3318
3319        // wait until the 1st phase is done
3320        while (!cancel_test_signal_end) {
3321            usleep(1000); // Sleep for 1ms
3322        }
3323
3324        status = fdb_rollback(&kvs, n/2 + 1);
3325        if (status != FDB_RESULT_SUCCESS) {
3326            // if compactor thread is done before reaching this line,
3327            // rollback may fail.
3328            rollback_failed = true;
3329        }
3330    }
3331    // join compactor
3332    thread_join(tid, &thread_ret);
3333
3334    // if rollback failed, we don't need to compact the file again.
3335    if (!rollback_failed) {
3336        // Compact the database
3337        status = fdb_compact(file, NULL);
3338        TEST_CHK(status == FDB_RESULT_SUCCESS);
3339    }
3340
3341    char key[15];
3342    void *value;
3343    size_t val_size;
3344
3345    for(i=0;i<n;i++) {
3346        sprintf(key, "%d", i);
3347        status = fdb_get_kv(kvs, key, strlen(key), &value, &val_size);
3348        if (mode == COMPACTION_CANCEL_MODE) {
3349            // compaction cancel mode: all docs should be retrieved
3350            TEST_CHK(status == FDB_RESULT_SUCCESS);
3351            fdb_free_block(value);
3352        } else if (mode == COMPACTION_ROLLBACK_MODE) {
3353            // rollback mode: only the first half docs should be retrieved
3354            if (i<=n/2 || rollback_failed) {
3355                // if rollback failed, all doc should be retrieved
3356                TEST_CHK(status == FDB_RESULT_SUCCESS);
3357                fdb_free_block(value);
3358            } else {
3359                TEST_CHK(status != FDB_RESULT_SUCCESS);
3360            }
3361        }
3362    }
3363
3364    status = fdb_close(file);
3365    TEST_CHK(status == FDB_RESULT_SUCCESS);
3366    status = fdb_shutdown();
3367    TEST_CHK(status == FDB_RESULT_SUCCESS);
3368
3369    memleak_end();
3370
3371    if (mode == COMPACTION_CANCEL_MODE) {
3372        TEST_RESULT("compaction cancellation test");
3373    } else if (mode == COMPACTION_ROLLBACK_MODE) {
3374        TEST_RESULT("rollback during the 2nd phase of compaction test");
3375    }
3376}
3377
3378int main(){
3379    int i;
3380
3381    compact_deleted_doc_test();
3382    compact_upto_test(false); // single kv instance in file
3383    compact_upto_test(true); // multiple kv instance in file
3384    wal_delete_compact_upto_test();
3385    for (i=0;i<4;++i) {
3386        compact_upto_overwrite_test(i);
3387    }
3388    compact_with_snapshot_open_multi_kvs_test();
3389    compact_with_snapshot_open_test();
3390    compact_upto_post_snapshot_test();
3391    compact_upto_twice_test();
3392    compaction_callback_test(true); // multi kv instance mode
3393    compaction_callback_test(false); // single kv instance mode
3394    compact_wo_reopen_test();
3395    compact_with_reopen_test();
3396    compact_reopen_with_iterator();
3397    compact_reopen_named_kvs();
3398    estimate_space_upto_test(false); // single kv instance in file
3399    estimate_space_upto_test(true); // multiple kv instance in file
3400    // Since we call unlink() to the old file after compaction,
3401    // cloning old file after the compaction doesn't work anymore.
3402    // So we temporarily disable this test.
3403    //auto_recover_compact_ok_test();
3404    unlink_after_compaction_test();
3405    db_compact_overwrite();
3406    db_compact_during_doc_delete(NULL);
3407    compaction_with_concurrent_transaction_test();
3408    compaction_with_concurrent_update_test();
3409    auto_compaction_with_custom_cmp_function();
3410    compaction_daemon_test(20);
3411    auto_compaction_with_concurrent_insert_test(20);
3412    compaction_cancellation_test(COMPACTION_CANCEL_MODE);
3413    // Disable it temporarily until it is resolved.
3414    //compaction_cancellation_test(COMPACTION_ROLLBACK_MODE);
3415
3416    return 0;
3417}
3418
3419