remove unsort duplicates. closes #179

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

src/tests/test_cursor_delete.c

@@ -30,6 +30,12 @@ void cursor_expect(DBC *cursor, int k, int v, int op) {
     free(val.data);
 }
 
+void cursor_expect_fail(DBC *cursor, int op, int expectr) {
+    DBT key, val;
+    int r = cursor->c_get(cursor, dbt_init_malloc(&key), dbt_init_malloc(&val), op);
+    assert(r == expectr);
+}
+ 
 /* generate a multi-level tree and delete all entries with a cursor
    verify that the pivot flags are toggled (currently by inspection) */
 
@@ -79,9 +85,8 @@ void test_cursor_delete(int dup_mode) {
 }
 
 /* insert duplicate duplicates into a sorted duplicate tree */
-
-void test_cursor_delete_dupsort(int dup_mode) {
-    if (verbose) printf("test_cursor_delete_dupsort:%d\n", dup_mode);
+void test_cursor_delete_dupsort() {
+    if (verbose) printf("test_cursor_delete_dupsort\n");
 
     int pagesize = 4096;
     int elementsize = 32;
@@ -98,7 +103,7 @@ void test_cursor_delete_dupsort(int dup_mode) {
 
     /* 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_flags(db, DB_DUP + DB_DUPSORT); assert(r == 0);
     r = db->set_pagesize(db, pagesize); assert(r == 0);
     r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666); assert(r == 0);
 
@@ -107,18 +112,22 @@ void test_cursor_delete_dupsort(int dup_mode) {
         int k = htonl(1);
         int v = htonl(1);
         DBT key, val;
-        r = db->put(db, null_txn, dbt_init(&key, &k, sizeof k), dbt_init(&val, &v, sizeof v), 0); assert(r == 0);
+        r = db->put(db, null_txn, dbt_init(&key, &k, sizeof k), dbt_init(&val, &v, sizeof v), 0);
+        if (i == 0) 
+            assert(r == 0); 
+        else 
+            assert(r == DB_KEYEXIST);
     }
 
     /* verify the sort order with a cursor */
     DBC *cursor;
     r = db->cursor(db, null_txn, &cursor, 0); assert(r == 0);
 
-    for (i=0; i<n; i++) {
     cursor_expect(cursor, htonl(1), htonl(1), DB_NEXT); 
         
     r = cursor->c_del(cursor, 0); assert(r == 0);
-     }
+
+    cursor_expect_fail(cursor, DB_NEXT, DB_NOTFOUND);
 
     r = cursor->c_close(cursor); assert(r == 0);
 
@@ -133,9 +142,11 @@ int main(int argc, const char *argv[]) {
     mkdir(DIR, 0777);
     
     test_cursor_delete(0);
+#if USE_BDB
     test_cursor_delete(DB_DUP);
+#endif
     test_cursor_delete(DB_DUP + DB_DUPSORT);
-    test_cursor_delete_dupsort(DB_DUP + DB_DUPSORT);
+    test_cursor_delete_dupsort();
 
     return 0;
 }

src/tests/test_db_set_flags.c

@@ -40,8 +40,8 @@ int main(int argc, const char *argv[]) {
     
     test_db_set_flags(0, 0, 0, 0);
     test_db_set_flags(0, 0, DB_DUP, EINVAL);
-    test_db_set_flags(DB_DUP, 0, DB_DUP, EINVAL);
-    test_db_set_flags(DB_DUP, 0, 0, 0);
+    test_db_set_flags(DB_DUP+DB_DUPSORT, 0, DB_DUP+DB_DUPSORT, EINVAL);
+    test_db_set_flags(DB_DUP+DB_DUPSORT, 0, 0, 0);
 
     return 0;
 }

src/tests/test_dup_delete.c

@@ -728,6 +728,7 @@ int main(int argc, const char *argv[]) {
     system("rm -rf " DIR);
     mkdir(DIR, 0777);
 
+#if USE_BDB
     /* dup tests */
     for (i = 1; i <= (1<<16); i *= 2) {
         test_dup_delete(i, DB_DUP);
@@ -735,15 +736,14 @@ int main(int argc, const char *argv[]) {
         test_all_dup_delete_insert(i);
         test_walk_empty(i, DB_DUP);
     }
+#endif
 
-#if USE_TDB
     /* dupsort tests */
     for (i = 1; i <= (1<<16); i *= 2) {
         test_dup_delete(i, DB_DUP + DB_DUPSORT);
         test_dup_delete_insert(i, DB_DUP + DB_DUPSORT);
         test_walk_empty(i, DB_DUP + DB_DUPSORT);
     }
-#endif
 
     return 0;
 }

src/tests/test_dup_dup.c

@@ -41,6 +41,13 @@ void test_dup_dup(int dup_mode, u_int32_t put_flags, int rexpect, int rexpectdup
     r = db_create(&db, null_env, 0);
     assert(r == 0);
     r = db->set_flags(db, dup_mode);
+#if USE_TDB
+    if (r != 0 && dup_mode == DB_DUP) {
+        printf("%s:%d:WARNING: tokudb does not support DB_DUP\n", __FILE__, __LINE__);
+        r = db->close(db, 0); assert(r == 0);
+        return;
+    }
+#endif
     assert(r == 0);
     r = db->set_pagesize(db, 4096);
     assert(r == 0);

src/tests/test_dup_flags.c

@@ -28,6 +28,13 @@ void test_dup_flags(int dup_flags) {
     r = db_create(&db, null_env, 0);
     assert(r == 0);
     r = db->set_flags(db, dup_flags);
+#if USE_TDB
+    if (r != 0 && dup_flags == DB_DUP) {
+        printf("%s:%d: WARNING: tokudb does not support DB_DUP\n", __FILE__, __LINE__);
+        r = db->close(db, 0); assert(r == 0);
+        return;
+    }
+#endif
     assert(r == 0);
     r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666);
     assert(r == 0);

src/tests/test_dup_insert.c

@@ -57,7 +57,7 @@ void test_insert(int n, int dup_mode) {
 
     int values[n];
     for (i=0; i<n; i++)
-        values[i] = htonl(random());
+        values[i] = htonl((random() << 16) + i);
     int sortvalues[n];
     for (i=0; i<n; i++)
         sortvalues[i] = values[i];
@@ -65,10 +65,7 @@ void test_insert(int n, int dup_mode) {
         return memcmp(a, b, sizeof (int));
     }
     qsort(sortvalues, n, sizeof sortvalues[0], mycmp);
-#if USE_BDB
-    for (i=1; i<n; i++) 
-        if (sortvalues[i-1] == sortvalues[i]) printf("dup %d\n", i);
-#endif
+
     /* insert n-1 unique keys {0, 1,  n-1} - {n/2} */
     for (i=0; i<n; i++) {
         if (i == n/2)
@@ -86,18 +83,6 @@ void test_insert(int n, int dup_mode) {
         int v = values[i];
         DBT key, val;
         r = db->put(db, null_txn, dbt_init(&key, &k, sizeof k), dbt_init(&val, &v, sizeof v), 0);
-#if USE_BDB
-        if (r != 0) {
-            void find_dup_val(int v, int m) {
-                int i;
-                printf("dup values[%d]=%d: ", m, v);
-                for (i=0; i<m; i++)
-                    if (values[i] == v) printf("%d ", i);
-                printf("\n");
-            }
-            find_dup_val(values[i], i);
-        }
-#endif
         assert(r == 0);
     } 
 
@@ -175,7 +160,7 @@ void test_nonleaf_insert(int n, int dup_mode) {
 
     int values[n];
     for (i=0; i<n; i++)
-        values[i] = htonl(random());
+        values[i] = htonl((random() << 16) + i);
     int sortvalues[n];
     for (i=0; i<n; i++)
         sortvalues[i] = values[i];
@@ -280,18 +265,20 @@ int main(int argc, const char *argv[]) {
     }
 
     /* dup tests */
+#if USE_TDB
+    printf("%s:%d:WARNING:tokudb does not support DB_DUP\n", __FILE__, __LINE__);
+#else
     for (i = 1; i <= (1<<16); i *= 2) {
         test_insert(i, DB_DUP);
         test_nonleaf_insert(i, DB_DUP);
     }
+#endif
 
-#if USE_TDB
     /* dupsort tests */
     for (i = 1; i <= (1<<16); i *= 2) {
         test_insert(i, DB_DUP + DB_DUPSORT);
         test_nonleaf_insert(i, DB_DUP + DB_DUPSORT);
     }
-#endif
 
     return 0;
 }

src/tests/test_dup_search.c

@@ -333,16 +333,29 @@ int main(int argc, const char *argv[]) {
     mkdir(DIR, 0777);
 
     /* dup search */
+#if USE_TDB
+    printf("%s:%d:WARNING:tokudb does not support DB_DUP\n", __FILE__, __LINE__);
+#else
     for (i = 1; i <= (1<<16); i *= 2) {
          test_ici_search(i, DB_DUP);
          test_icdi_search(i, DB_DUP);
          test_i0i1ci0_search(i, DB_DUP);
     }
+#endif
+
+    /* dupsort search */
+    for (i = 1; i <= (1<<16); i *= 2) {
+         test_ici_search(i, DB_DUP + DB_DUPSORT);
+         test_icdi_search(i, DB_DUP + DB_DUPSORT);
+         test_i0i1ci0_search(i, DB_DUP + DB_DUPSORT);
+    }
 
     /* insert data in descending order */
     for (i = 1; i <= (1<<16); i *= 2) {
         test_reverse_search(i, 0);
+#if USE_BDB
         test_reverse_search(i, DB_DUP);
+#endif
         test_reverse_search(i, DB_DUP + DB_DUPSORT);
     }
 

src/ydb.c

@@ -1384,15 +1384,12 @@ static int toku_db_set_flags(DB * db, u_int32_t flags) {
     int r = toku_brt_get_flags(db->i->brt, &tflags);
     if (r!=0) return r;
     
-    if (flags & DB_DUP) {
-        tflags |= TOKU_DB_DUP;
-        flags &= ~DB_DUP;
-    }
-    if (flags & DB_DUPSORT) {
-        tflags |= TOKU_DB_DUPSORT;
-        flags &= ~DB_DUPSORT;
+    /* we support no duplicates and sorted duplicates */
+    if (flags) {
+        if (flags != (DB_DUP + DB_DUPSORT))
+            return EINVAL;
+        tflags += TOKU_DB_DUP + TOKU_DB_DUPSORT;
     }
-    if (flags != 0) return EINVAL;
     r = toku_brt_set_flags(db->i->brt, tflags);
     return r;
 }