tokudb_load now supports dupsort databases. we defer judgement on db flags...

tokudb_load now supports dupsort databases.  we defer judgement on db flags until the db->open method is called.  this allows incremental update of the flags.
closes #314


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

Makefile

 TAGS: */*.c */*.h
 	etags */*.c */*.h
 
-SRCDIRS = newbrt src src/range_tree src/range_tree/tests src/lock_tree src/lock_tree/tests cxx cxx/tests \
+SRCDIRS = newbrt src src/tests src/range_tree src/range_tree/tests src/lock_tree src/lock_tree/tests cxx cxx/tests \
 		utils db-benchmark-test db-benchmark-test-cxx
 
 build:

src/tests/test_dup_dup.c

@@ -69,12 +69,15 @@ void test_dup_key(int dup_mode, u_int32_t put_flags, int rexpect, int rexpectdup
     db->set_errfile(db, stderr);
     r = db->set_flags(db, dup_mode);
     if (maybe_do_db_dup_warning(r, dup_mode)) {
-        r = db->close(db, 0); assert(r == 0);
-	return;
+        r = db->close(db, 0); assert(r == 0); return;
     }
     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);
+    r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666); 
+    if (maybe_do_db_dup_warning(r, dup_mode)) {
+        r = db->close(db, 0); assert(r == 0); return;
+    }
+    assert(r == 0);
 
     db_put(db, 0, 0, put_flags, rexpect);
     db_put(db, 0, 1, put_flags, rexpectdupdup);
@@ -117,12 +120,15 @@ void test_dup_dup(int dup_mode, u_int32_t put_flags, int rexpect, int rexpectdup
     db->set_errfile(db, stderr);
     r = db->set_flags(db, dup_mode);
     if (maybe_do_db_dup_warning(r, dup_mode)) {
-        r = db->close(db, 0); assert(r == 0);
-	return;
+        r = db->close(db, 0); assert(r == 0); return;
     }
     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);
+    r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666); 
+    if (maybe_do_db_dup_warning(r, dup_mode)) {
+        r = db->close(db, 0); assert(r == 0); return;
+    }
+    assert(r == 0);
 
     db_put(db, 0, 0, put_flags, rexpect);
     db_put(db, 0, 0, put_flags, rexpectdupdup);

src/tests/test_dup_flags.c

@@ -36,7 +36,15 @@ void test_dup_flags(int dup_flags) {
 #endif
     assert(r == 0);
     u_int32_t flags; r = db->get_flags(db, &flags); assert(r == 0); assert(flags == dup_flags);
-    r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666); assert(r == 0);
+    r = db->open(db, null_txn, fname, "main", DB_BTREE, DB_CREATE, 0666); 
+#if USE_TDB
+    if (r != 0 && dup_flags == DB_DUP) {
+        if (verbose) 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->close(db, 0); assert(r == 0);
 
     /* verify dup flags match */

src/ydb.c

@@ -1631,6 +1631,14 @@ static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *db
     if (is_db_excl && !is_db_create) return EINVAL;
     if (dbtype==DB_UNKNOWN && is_db_excl) return EINVAL;
 
+    /* tokudb supports no duplicates and sorted duplicates only */
+    unsigned int tflags;
+    r = toku_brt_get_flags(db->i->brt, &tflags);
+    if (r != 0) 
+        return r;
+    if ((tflags & TOKU_DB_DUP) && !(tflags & TOKU_DB_DUPSORT))
+        return EINVAL;
+
     if (db_opened(db))
         return EINVAL;              /* It was already open. */
     
@@ -1843,12 +1851,10 @@ 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;
     
-    /* 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 & DB_DUP)
+        tflags += TOKU_DB_DUP;
+    if (flags & DB_DUPSORT)
+        tflags += TOKU_DB_DUPSORT;
     r = toku_brt_set_flags(db->i->brt, tflags);
     return r;
 }

utils/tokudb_load.c

@@ -785,12 +785,16 @@ int get_dbt(DBT* pdbt)
    return EXIT_FAILURE;
 }
 
+#ifndef DB_YESOVERWRITE
+#define DB_YESOVERWRITE 0
+#endif
+
 int insert_pair(DBT* key, DBT* data)
 {
    DB_ENV* dbenv = g.dbenv;
    DB* db = g.db;
 
-   int retval = db->put(db, NULL, key, data, g.overwritekeys ? 0 : DB_NOOVERWRITE);
+   int retval = db->put(db, NULL, key, data, g.overwritekeys ? DB_YESOVERWRITE : DB_NOOVERWRITE);
    if (retval != 0) {
       //TODO: Check for transaction failures/etc.. retry if necessary.
       ERROR(retval, "DB->put");