ftloader-test-writer.cc 7.59 KB
Newer Older
1 2
/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
3
#ident "$Id$"
4 5 6 7 8 9 10
#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."

// test the loader write dbfile function

#include "includes.h"
#include "test.h"
11
#include "ftloader-internal.h"
12
#include <inttypes.h>
13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28


static void traceit(const char *s) {
    time_t t = time(NULL);
    printf("%.24s %s\n", ctime(&t), s);
    fflush(stdout);
}

static int qsort_compare_ints (const void *a, const void *b) {
    int avalue = *(int*)a;
    int bvalue = *(int*)b;
    if (avalue<bvalue) return -1;
    if (avalue>bvalue) return +1;
    return 0;
}

Zardosht Kasheff's avatar
Zardosht Kasheff committed
29
static int compare_ints (DB *UU(desc), const DBT *akey, const DBT *bkey) {
30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45
    assert(akey->size==sizeof(int));
    assert(bkey->size==sizeof(int));
    return qsort_compare_ints(akey->data, bkey->data);
}

static void err_cb(DB *db UU(), int dbn UU(), int err UU(), DBT *key UU(), DBT *val UU(), void *extra UU()) {
    fprintf(stderr, "error in test");
    abort();
}

static void verify_dbfile(int n, const char *name) {
    if (verbose) traceit("verify");

    int r;

    CACHETABLE ct;
46
    r = toku_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER); assert(r==0);
47 48

    TOKUTXN const null_txn = NULL;
49 50 51
    FT_HANDLE t = NULL;
    r = toku_ft_handle_create(&t); assert(r == 0);
    r = toku_ft_set_bt_compare(t, compare_ints);
52
    assert(r==0);
53
    r = toku_ft_handle_open(t, name, 0, 0, ct, null_txn); assert(r==0);
54

55
    if (verbose) traceit("Verifying brt internals");
56
    r = toku_verify_ft(t);
57 58
    if (verbose) traceit("Verified brt internals");

59
    FT_CURSOR cursor = NULL;
Yoni Fogel's avatar
Yoni Fogel committed
60
    r = toku_ft_cursor(t, &cursor, NULL, false, false); assert(r == 0);
61

62
    size_t userdata = 0;
63 64 65 66 67
    int i;
    for (i=0; ; i++) {
	int kk = i;
	int vv = i;
	struct check_pair pair = {sizeof kk, &kk, sizeof vv, &vv, 0};
68
        r = toku_ft_cursor_get(cursor, NULL, lookup_checkf, &pair, DB_NEXT);
69 70 71 72 73
        if (r != 0) {
	    assert(pair.call_count ==0);
	    break;
	}
	assert(pair.call_count==1);
74
        userdata += pair.keylen + pair.vallen;
75 76 77 78
    }

    assert(i == n);

79
    r = toku_ft_cursor_close(cursor); assert(r == 0);
80

81 82
    struct ftstat64_s s;
    r = toku_ft_handle_stat64(t, NULL, &s); assert(r == 0);
Yoni Fogel's avatar
Yoni Fogel committed
83
    assert(s.nkeys == (uint64_t)n && s.ndata == (uint64_t)n && s.dsize == userdata);
84

85
    r = toku_close_ft_handle_nolsn(t, 0); assert(r==0);
86 87 88 89
    r = toku_cachetable_close(&ct);assert(r==0);
    if (verbose) traceit("verify done");
}

90
static void test_write_dbfile (char *tf_template, int n, char *output_name, TXNID xid) {
91 92 93
    if (verbose) traceit("test start");

    DB *dest_db = NULL;
94 95 96 97 98
    struct ft_loader_s bl;
    ZERO_STRUCT(bl);
    bl.temp_file_template = tf_template;
    bl.reserved_memory = 512*1024*1024;
    bl.load_root_xid = xid;
99
    if (xid) {
100
        XCALLOC_N(1, bl.root_xids_that_created);
101 102
        bl.root_xids_that_created[0] = 0;
    }
103 104 105
    int r = ft_loader_init_file_infos(&bl.file_infos); CKERR(r);
    ft_loader_lock_init(&bl);
    ft_loader_set_fractal_workers_count_from_c(&bl);
106

107 108 109 110 111
    struct merge_fileset fs;
    init_merge_fileset(&fs);

    // put rows in the row set
    struct rowset aset;
112
    uint64_t size_est = 0;
113
    init_rowset(&aset, toku_ft_loader_get_rowset_budget_for_testing());
114
    for (int i=0; i<n; i++) {
115 116 117 118
	DBT key;
        toku_fill_dbt(&key, &i, sizeof i);
	DBT val;
        toku_fill_dbt(&val, &i, sizeof i);
119
	add_row(&aset, &key, &val);
120
	size_est += ft_loader_leafentry_size(key.size, val.size, xid);
121
     }
122

123
    toku_ft_loader_set_n_rows(&bl, n);
124

125 126 127
    ft_loader_init_error_callback(&bl.error_callback);
    ft_loader_set_error_function(&bl.error_callback, err_cb, NULL);
    r = ft_loader_sort_and_write_rows(&aset, &fs, &bl, 0, dest_db, compare_ints);  CKERR(r);
128
    // destroy_rowset(&aset);
129
    
130
    ft_loader_fi_close_all(&bl.file_infos);
131 132 133 134 135 136 137 138 139 140 141 142

    QUEUE q;
    r = queue_create(&q, 0xFFFFFFFF); // infinite queue.
    assert(r==0);
    r = merge_files(&fs, &bl, 0, dest_db, compare_ints, 0, q); CKERR(r);
    assert(fs.n_temp_files==0);

    QUEUE q2;
    r = queue_create(&q2, 0xFFFFFFFF); // infinite queue.
    assert(r==0);

    size_t num_found = 0;
143
    size_t found_size_est = 0;
144 145 146 147 148
    while (1) {
	void *v;
	r = queue_deq(q, &v, NULL, NULL);
	if (r==EOF) break;
	struct rowset *rs = (struct rowset *)v;
149
	if (verbose) printf("v=%p\n", v);
150

151
	for (size_t i=0; i<rs->n_rows; i++) {
152 153 154
	    struct row *row = &rs->rows[i];
	    assert(row->klen==sizeof(int));
	    assert(row->vlen==sizeof(int));
155
	    assert((int)(num_found+i)==*(int*)(rs->data+row->off));
156
	    found_size_est += ft_loader_leafentry_size(row->klen, row->vlen, xid);
157
 	}
158 159 160 161 162 163 164

	num_found += rs->n_rows;

	r = queue_enq(q2, v, 0, NULL);
	assert(r==0);
    }
    assert((int)num_found == n);
165 166
    assert(found_size_est == size_est);
 
167 168 169 170 171 172
    r = queue_eof(q2);
    assert(r==0);

    r = queue_destroy(q);
    assert(r==0);

173 174
    DESCRIPTOR_S desc;
    toku_fill_dbt(&desc.dbt, "abcd", 4);
175 176 177 178 179

    int fd = open(output_name, O_RDWR | O_CREAT | O_BINARY, S_IRWXU|S_IRWXG|S_IRWXO);
    assert(fd>=0);

    if (verbose) traceit("write to file");
180
    r = toku_loader_write_brt_from_q_in_C(&bl, &desc, fd, 1000, q2, size_est, 0, 0, 0, TOKU_DEFAULT_COMPRESSION_METHOD);
181 182 183
    assert(r==0);

    r = queue_destroy(q2);
184
    assert_zero(r);
185 186
   
    destroy_merge_fileset(&fs);
Yoni Fogel's avatar
Yoni Fogel committed
187
    ft_loader_fi_destroy(&bl.file_infos, false);
188 189 190 191

    // walk a cursor through the dbfile and verify the rows
    verify_dbfile(n, output_name);

192 193
    ft_loader_destroy_error_callback(&bl.error_callback);
    ft_loader_lock_destroy(&bl);
194 195

    toku_free(bl.root_xids_that_created);
196 197
}

198
static int nrows = 1;
199
static TXNID xid = 0;
200 201

static int usage(const char *progname) {
202
    fprintf(stderr, "Usage:\n %s [-h] [-v] [-q] [-r %d] [-x %" PRIu64 "] [-s] directory\n", progname, nrows, xid);
203
    return 1;
204 205
}

206 207 208 209
int test_main (int argc, const char *argv[]) {
    const char *progname=argv[0];
    argc--; argv++;
    while (argc>0) {
210
	if (strcmp(argv[0], "-h") == 0 || strcmp(argv[0], "--help") == 0) {
211
	    return usage(progname);
212
        } else if (strcmp(argv[0], "-v") == 0 || strcmp(argv[0], "--verbose") == 0) {
213
	    verbose=1;
214
	} else if (strcmp(argv[0], "-q") == 0) {
215
	    verbose=0;
216
        } else if (strcmp(argv[0], "-r") == 0) {
217
            argc--; argv++;
218
            nrows = atoi(argv[0]);
219 220 221
        } else if (strcmp(argv[0], "-x") == 0) {
            argc--; argv++;
            xid = atol(argv[0]);
222
        } else if (strcmp(argv[0], "-s") == 0) {
223
            toku_ft_loader_set_size_factor(1);
224
	} else if (argv[0][0] == '-' || argc != 1) {
225 226
            return usage(progname);
	} else {
227 228 229 230
            break;
        }
	argc--; argv++;
    }
231
    assert(argc == 1); // argv[1] is the directory in which to do the test.
232 233 234
    const char* directory = argv[0];
    char unlink_all[strlen(directory)+20];
    snprintf(unlink_all, strlen(directory)+20, "rm -rf %s", directory);
235 236 237 238 239
    int r;
    r = system(unlink_all);
    CKERR(r);
    r = toku_os_mkdir(directory, 0755);
    CKERR(r);
240 241

    int  templen = strlen(directory)+15;
242 243
    char tf_template[templen];
    int tlen = snprintf(tf_template, templen, "%s/tempXXXXXX", directory);
244 245 246 247 248 249
    assert (tlen>0 && tlen<templen);

    char output_name[templen];
    int  olen = snprintf(output_name, templen, "%s/test.tokudb", directory);
    assert (olen>0 && olen<templen);

250
    test_write_dbfile(tf_template, nrows, output_name, xid);
251

252
#if 0
253 254
    r = system(unlink_all);
    CKERR(r);
255
#endif
256 257 258
    return 0;
}