More tests pass for keyrange.

The same 4 cases mentioned in [3932], except now three out of four are tested:
   1. flattened nodup (tested in [3932])
   2. unflattened nodup (tested in this change set)
   3. flattened dupsort (tested in this change set
   4. unflattened dupsort (untested)

Addresses #764.


git-svn-id: file:///svn/tokudb@3944 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt.c

@@ -3163,40 +3163,56 @@ static void toku_brt_keyrange_internal (BRT brt, CACHEKEY nodename, DBT *key, u_
 	node = node_v;
     }
     if (node->height>0) {
-	int prevcomp=-1;
+	int n_keys = node->u.n.n_children-1;
+	int compares[n_keys];
 	int i;
-	int comp;
-	for (i=0; i+1<node->u.n.n_children; i++) {
+	for (i=0; i<n_keys; i++) {
 	    struct kv_pair *pivot = node->u.n.childkeys[i];
 	    DBT dbt;
-	    comp = brt->compare_fun(brt->db, toku_fill_dbt(&dbt, kv_pair_key(pivot), kv_pair_keylen(pivot)), key);
-	    if (comp<0) {
-		*less += BNC_SUBTREE_LEAFENTRY_ESTIMATE(node, i);
-	    } else if (comp==0 && prevcomp==0) {
-		*equal += BNC_SUBTREE_LEAFENTRY_ESTIMATE(node, i);
-	    } else if (prevcomp>0) {
-		*greater += BNC_SUBTREE_LEAFENTRY_ESTIMATE(node, i);
-	    } else {
-		// prevcomp<=0 && comp>=0, but not both zero, so we have a subtree
-		toku_brt_keyrange_internal(brt, BNC_DISKOFF(node, i), key, less, equal, greater);
+	    compares[i] = brt->compare_fun(brt->db, toku_fill_dbt(&dbt, kv_pair_key(pivot), kv_pair_keylen(pivot)), key);
 	}
-	    prevcomp=comp;
-	}
-	if (prevcomp>0) {
-	    *greater += BNC_SUBTREE_LEAFENTRY_ESTIMATE(node, i);
+	for (i=0; i<node->u.n.n_children; i++) {
+	    int prevcomp = (i==0) ? -1 : compares[i-1];
+	    int nextcomp = (i+1 >= n_keys) ? 1 : compares[i];
+	    int subest = BNC_SUBTREE_LEAFENTRY_ESTIMATE(node, i);
+	    if (nextcomp < 0) {
+		// We're definitely looking too far to the left
+		*less += subest;
+	    } else if (prevcomp > 0) {
+		// We're definitely looking too far to the right
+		*greater += subest;
+	    } else if (prevcomp == 0 && nextcomp == 0) {
+		// We're looking at a subtree that contains all zeros
+		*equal   += subest;
 	    } else {
+		// nextcomp>=0 and prevcomp<=0, so something in the subtree could match
+		// but they are not both zero, so it's not the whole subtree, so we need to recurse
 		toku_brt_keyrange_internal(brt, BNC_DISKOFF(node, i), key, less, equal, greater);
 	    }
+	}
     } else {
 	BRT_CMD_S cmd = { BRT_INSERT, 0, .u.id={key,0}};
 	struct cmd_leafval_bessel_extra be = {brt, &cmd, 0};
 	u_int32_t idx;
 	int r = toku_omt_find_zero(node->u.l.buffer, toku_cmd_leafval_bessel, &be, 0, &idx);
 	*less += idx;
+	if (r==0 && (brt->flags & TOKU_DB_DUP)) {
+	    // There is something, and so we now want to find the rightmost extent.
+	    u_int32_t idx2;
+	    r = toku_omt_find(node->u.l.buffer, toku_cmd_leafval_bessel, &be, +1, 0, &idx2);
+	    if (r==0) {
+		*greater += toku_omt_size(node->u.l.buffer)-idx2;
+		*equal   += idx2-idx;
+	    } else {
+		*equal   += toku_omt_size(node->u.l.buffer)-idx;
+	    }
+	    //printf("%s:%d (%llu, %llu, %llu)\n", __FILE__, __LINE__, (unsigned long long)*less, (unsigned long long)*equal, (unsigned long long)*greater);
+	} else {
 	    *greater += toku_omt_size(node->u.l.buffer)-idx;
 	    if (r==0) {
-	    (*equal)++;
 		(*greater)--;
+		(*equal)++;
+	    }
 	}
     }
     {
@@ -3211,6 +3227,7 @@ int toku_brt_keyrange (BRT brt, DBT *key, u_int64_t *less,  u_int64_t *equal,  u
 	assert(rr == 0);
     }
     CACHEKEY *rootp = toku_calculate_root_offset_pointer(brt);
+
     *less = *equal = *greater = 0;
     toku_brt_keyrange_internal (brt, *rootp, key, less, equal, greater);
     {

newbrt/tests/Makefile

@@ -58,6 +58,7 @@ REGRESSION_TESTS = \
 	fifo-test \
 	list-test \
 	keyrange \
+	keyrange-dupsort \
 	keyrange-unflat \
 	log-test \
 	log-test2 \

newbrt/tests/keyrange-dupsort.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2008 Tokutek Inc.  All rights reserved."
+
+#include "brt.h"
+#include "test.h"
+#include "toku_assert.h"
+#include "key.h"
+
+#include <unistd.h>
+
+static TOKUTXN const null_txn = 0;
+static DB * const null_db = 0;
+
+static void test_flat (void) {
+    char fname[]= __FILE__ ".brt";
+    u_int64_t limit=100;
+    u_int64_t ilimit=100;
+    unlink(fname);
+    CACHETABLE ct;
+    int r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);                                assert(r==0);
+    BRT t;
+    r = toku_brt_create(&t); assert(r==0);
+    r = toku_brt_set_flags(t, TOKU_DB_DUP + TOKU_DB_DUPSORT); assert(r == 0);
+    r = toku_brt_set_nodesize(t, 4096); assert(r == 0);
+    r = toku_brt_open(t, fname, fname, 0, 1, 1, 0, ct, null_txn, (DB*)0);
+    u_int64_t i;
+    for (i=0; i<limit; i++) {
+	u_int64_t j;
+	for (j=0; j<ilimit; j++) {
+	    char key[100],val[100];
+	    snprintf(key, 100, "%08llu", (unsigned long long)2*i+1);
+	    snprintf(val, 100, "%08llu", (unsigned long long)2*j+1);
+	    DBT k,v;
+	    r = toku_brt_insert(t, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v,val, 1+strlen(val)), null_txn);
+	    assert(r == 0);
+	}
+    }
+    // flatten it.
+    for (i=0; i<limit; i++) {
+	u_int64_t j;
+	for (j=0; j<ilimit; j++) {
+	    char key[100],val[100];
+	    snprintf(key, 100, "%08llu", (unsigned long long)2*i+1);
+	    snprintf(val, 100, "%08llu", (unsigned long long)2*j+1);
+	    DBT k,v;
+	    r = toku_brt_lookup(t, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v, val, 1+strlen(val)));
+	    assert(r==0);
+	}
+    }
+    for (i=0; i<limit; i++) {
+	char key[100];
+	snprintf(key, 100, "%08llu", (unsigned long long)2*i+1);
+	DBT k;
+	u_int64_t less,equal,greater;
+	r = toku_brt_keyrange(t, toku_fill_dbt(&k, key, 1+strlen(key)), &less, &equal, &greater);
+	assert(r == 0);
+	//printf("key %llu/%llu %llu %llu %llu\n", (unsigned long long)2*i+1, (unsigned long long)2*limit+1, (unsigned long long)less, (unsigned long long)equal, (unsigned long long)greater);
+	assert(less==ilimit*i);
+	assert(equal==ilimit);
+	assert(less+equal+greater == limit*ilimit);
+    }
+    for (i=0; i<1+limit; i++) {
+	char key[100];
+	snprintf(key, 100, "%08llu", (unsigned long long)2*i);
+	DBT k;
+	u_int64_t less,equal,greater;
+	r = toku_brt_keyrange(t, toku_fill_dbt(&k, key, 1+strlen(key)), &less, &equal, &greater);
+	assert(r == 0);
+	//printf("key %llu/%llu %llu %llu %llu\n", (unsigned long long)2*i, (unsigned long long)2*limit, (unsigned long long)less, (unsigned long long)equal, (unsigned long long)greater);
+	assert(less==ilimit*i);
+	assert(equal==0);
+	assert(less+equal+greater == limit*ilimit);
+    }
+    r = toku_close_brt(t, 0);          assert(r==0);
+    r = toku_cachetable_close(&ct);    assert(r==0);
+}
+
+int main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+
+    test_flat();
+    toku_malloc_cleanup();
+    if (verbose) printf("test ok\n");
+    return 0;
+}
+