pma_set now finds the leftmost non-deleted kv pair that matches the set key. closes #192

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

newbrt/pma.c

@@ -758,13 +758,48 @@ int toku_pma_cursor_get_current(PMA_CURSOR c, DBT *key, DBT *val, int even_delet
     return 0;
 }
 
+/* find the next matching key in the pma starting from index here */
+static int pma_next_key(PMA pma, DBT *k, DBT *v, int here, int n, int *found) {
+    assert(0 <= here);
+    *found = 0;
+    while (here < n && !kv_pair_inuse(pma->pairs[here]))
+        here += 1;
+    if (here < n) {
+        int cmp = pma_compare_dbt_kv(pma, k, v, kv_pair_ptr(pma->pairs[here]));
+        if (cmp == 0)
+            *found = 1;
+    }
+    return here;
+}
+
+/* find the previous matching key in the pma starting from index here */
+static int pma_prev_key(PMA pma, DBT *k, DBT *v, int here, int n, int *found) {
+    assert(here < n);
+    *found = 0;
+    while (0 <= here && !kv_pair_inuse(pma->pairs[here]))
+        here -= 1;
+    if (0 <= here) {
+        int cmp = pma_compare_dbt_kv(pma, k, v, kv_pair_ptr(pma->pairs[here]));
+        if (cmp == 0)
+            *found = 1;
+    }
+    return here;
+}
+
 int toku_pma_cursor_set_both(PMA_CURSOR c, DBT *key, DBT *val) {
     PMA pma = c->pma;
     unsigned int here; int found;
     here = pma_left_search(pma, key, val, 0, pma->N, &found);
     assert(here<=toku_pma_index_limit(pma));
     int r = DB_NOTFOUND;
+#if 1
+    /* skip any deleted pairs that match */
+    while (found && !kv_pair_valid(pma->pairs[here]))
+        here = pma_next_key(pma, key, val, here+1, pma->N, &found);
+    if (found) {
+#else
     if (found && kv_pair_valid(pma->pairs[here])) {
+#endif
         __pma_delete_resume(c->pma, c->position);
         c->position = here;
         r = 0;
@@ -792,33 +827,6 @@ int toku_pma_cursor_set_range_both(PMA_CURSOR c, DBT *key, DBT *val) {
     return r;
 }
 
-/* find the next matching key in the pma starting from index here */
-static int pma_next_key(PMA pma, DBT *k, DBT *v, int here, int n, int *found) {
-    assert(0 <= here);
-    *found = 0;
-    while (here < n && !kv_pair_inuse(pma->pairs[here]))
-        here += 1;
-    if (here < n) {
-        int cmp = pma_compare_dbt_kv(pma, k, v, kv_pair_ptr(pma->pairs[here]));
-        if (cmp == 0)
-            *found = 1;
-    }
-    return here;
-}
-
-/* find the previous matching key in the pma starting from index here */
-static int pma_prev_key(PMA pma, DBT *k, DBT *v, int here, int n, int *found) {
-    assert(here < n);
-    *found = 0;
-    while (0 <= here && !kv_pair_inuse(pma->pairs[here]))
-        here -= 1;
-    if (0 <= here) {
-        int cmp = pma_compare_dbt_kv(pma, k, v, kv_pair_ptr(pma->pairs[here]));
-        if (cmp == 0)
-            *found = 1;
-    }
-    return here;
-}
 
 /* set lastkeymatch if the kv pair under the cursor is the last one in the pma
    compare with the next and previous valid pma entries */

src/tests/test_dupsort_get.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <unistd.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <arpa/inet.h>
+#include <db.h>
+
+#include "test.h"
+
+
+
+void db_put(DB *db, int k, int v) {
+    DB_TXN * const null_txn = 0;
+    DBT key, val;
+    int r = db->put(db, null_txn, dbt_init(&key, &k, sizeof k), dbt_init(&val, &v, sizeof v), 0);
+    assert(r == 0);
+}
+
+void db_get(DB *db, int k) {
+    DB_TXN * const null_txn = 0;
+    DBT key, val;
+    int r = db->get(db, null_txn, dbt_init(&key, &k, sizeof k), dbt_init_malloc(&val), 0);
+    assert(r == 0);
+    int vv;
+    assert(val.size == sizeof vv);
+    memcpy(&vv, val.data, val.size);
+    printf("do_search %d\n", htonl(vv));
+    free(val.data);
+}
+
+void db_del(DB *db, int k) {
+    DB_TXN * const null_txn = 0;
+    DBT key;
+    int r = db->del(db, null_txn, dbt_init(&key, &k, sizeof k), 0);
+    assert(r == 0);
+}
+
+void expect_db_get(DB *db, int k, int v) {
+    DB_TXN * const null_txn = 0;
+    DBT key, val;
+    int r = db->get(db, null_txn, dbt_init(&key, &k, sizeof k), dbt_init_malloc(&val), 0);
+    assert(r == 0);
+    int vv;
+    assert(val.size == sizeof vv);
+    memcpy(&vv, val.data, val.size);
+    assert(vv == v);
+    free(val.data);
+}
+
+void expect_cursor_get(DBC *cursor, int k, int v) {
+    DBT key, val;
+    int r = cursor->c_get(cursor, dbt_init_malloc(&key), dbt_init_malloc(&val), DB_NEXT);
+    assert(r == 0);
+    assert(key.size == sizeof k);
+    int kk;
+    memcpy(&kk, key.data, key.size);
+    assert(val.size == sizeof v);
+    int vv;
+    memcpy(&vv, val.data, val.size);
+    if (kk != k || vv != v) printf("expect key %d got %d - %d %d\n", htonl(k), htonl(kk), htonl(v), htonl(vv));
+    assert(kk == k);
+    assert(vv == v);
+
+    free(key.data);
+    free(val.data);
+}
+
+void expect_cursor_set(DBC *cursor, int k) {
+    DBT key, val;
+    int r = cursor->c_get(cursor, dbt_init(&key, &k, sizeof k), dbt_init_malloc(&val), DB_SET);
+    assert(r == 0);
+    free(val.data);
+}
+
+void expect_cursor_get_both(DBC *cursor, int k, int v) {
+    DBT key, val;
+    int r = cursor->c_get(cursor, dbt_init(&key, &k, sizeof k), dbt_init(&val, &v, sizeof v), DB_GET_BOTH);
+    assert(r == 0);
+}
+
+void expect_cursor_get_current(DBC *cursor, int k, int v) {
+    DBT key, val;
+    int r = cursor->c_get(cursor, dbt_init_malloc(&key), dbt_init_malloc(&val), DB_CURRENT);
+    assert(r == 0);
+    int kk, vv;
+    assert(key.size == sizeof kk); memcpy(&kk, key.data, key.size); assert(kk == k);
+    assert(val.size == sizeof vv); memcpy(&vv, val.data, val.size); assert(vv == v);
+    free(key.data); free(val.data);
+}
+
+
+void test_dupsort_get(int n, int dup_mode) {
+    if (verbose) printf("test_dupsort_get:%d %d\n", n, dup_mode);
+
+    DB_ENV * const null_env = 0;
+    DB *db;
+    DB_TXN * const null_txn = 0;
+    const char * const fname = DIR "/" "test.dupsort.get.brt";
+    int r;
+
+    unlink(fname);
+
+    /* create the dup database file */
+    r = db_create(&db, null_env, 0); assert(r == 0);
+    r = db->set_flags(db, dup_mode); assert(r == 0);
+    r = db->set_pagesize(db, 4096); assert(r == 0);
+    r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666); assert(r == 0);
+
+    /* insert n duplicates */
+    int i;
+    for (i=0; i<n; i++) {
+        int k = htonl(n/2);
+        int v = htonl(i);
+        db_put(db, k, v);
+
+        expect_db_get(db, k, htonl(0));
+    } 
+
+    /* reopen the database to force nonleaf buffering */
+    r = db->close(db, 0); assert(r == 0);
+    r = db_create(&db, null_env, 0); assert(r == 0);
+    r = db->set_flags(db, dup_mode); assert(r == 0);
+    r = db->set_pagesize(db, 4096); assert(r == 0);
+    r = db->open(db, null_txn, fname, "main", DB_BTREE, 0, 0666); assert(r == 0);
+
+    db_del(db, htonl(n/2));
+
+    for (i=n-1; i>=0; i--)
+        db_put(db, htonl(n/2), htonl(i));
+
+    expect_db_get(db, htonl(n/2), htonl(0));
+
+    DBC *cursor;
+    r = db->cursor(db, null_txn, &cursor, 0); assert(r == 0);
+    expect_cursor_set(cursor, htonl(n/2));
+    expect_cursor_get_current(cursor, htonl(n/2), 0);
+    r = cursor->c_del(cursor, 0); assert(r == 0);
+    if (n > 1) expect_db_get(db, htonl(n/2), htonl(1));
+
+    r = cursor->c_close(cursor); assert(r == 0);
+
+    r = db->close(db, 0); assert(r == 0);
+}
+
+
+int main(int argc, const char *argv[]) {
+    int i;
+
+    parse_args(argc, argv);
+  
+    system("rm -rf " DIR);
+    mkdir(DIR, 0777);
+
+    for (i=1; i<65537; i *= 2) {
+        test_dupsort_get(i, DB_DUP + DB_DUPSORT);
+    }
+    return 0;
+}