make secondary work a little longer. Addresses #48.

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

src/tests/test_db_secondary.c

@@ -5,13 +5,7 @@
 #include <db.h>
 #include <assert.h>
 #include <errno.h>
-/*#include <limits.h>
-#include <stdio.h>
-#include <stdlib.h>
 #include <sys/stat.h>
-#include <sys/types.h>
-#include <fcntl.h>
-#include <unistd.h>*/
 
 #include "test.h"
 
@@ -60,7 +54,7 @@ void second_setup() {
 
     /* Open/create primary */
     r = db_create(&dbp, dbenv, 0);                                              CKERR(r);
-    r = dbp->open(dbp, NULL, "students.db", NULL, DB_BTREE, DB_CREATE, 0600);   CKERR(r);
+    r = dbp->open(dbp, NULL, DIR "/students.db", NULL, DB_BTREE, DB_CREATE, 0600);   CKERR(r);
 
     /*
      * Open/create secondary. Note that it supports duplicate data
@@ -68,7 +62,7 @@ void second_setup() {
      */
     r = db_create(&sdbp, dbenv, 0);                                             CKERR(r);
     r = sdbp->set_flags(sdbp, DB_DUP | DB_DUPSORT);                             CKERR(r);
-    r = sdbp->open(sdbp, NULL, "lastname.db", NULL, DB_BTREE, DB_CREATE, 0600); CKERR(r);
+    r = sdbp->open(sdbp, NULL, DIR "/lastname.db", NULL, DB_BTREE, DB_CREATE, 0600); CKERR(r);
     
 
     /* Associate the secondary with the primary. */
@@ -181,6 +175,9 @@ void verify_gone() {
 int main() {
     int r;
 
+    system("rm -rf " DIR);
+    mkdir(DIR, 0777);
+
     second_setup();
     insert_test();
     delete_from_primary();

src/ydb.c

@@ -616,39 +616,50 @@ static int toku_db_del(DB * db, DB_TXN * txn, DBT * key, u_int32_t flags) {
     return r;
 }
 
-static int toku_db_get(DB * db, DB_TXN * txn, DBT * key, DBT * data, u_int32_t flags) {
-    assert(flags == 0);
+
+// Internal flag for when doing a get, don't do the work for associate.
+#define DB_NO_ASSOCIATE 42
+
+static int toku_db_get (DB * db, DB_TXN * txn, DBT * key, DBT * data, u_int32_t flags) {
     int r;
     unsigned int brtflags;
-    DBT primary_key;
-    DBT *sdata;
-    DB  *primary = db->i->primary;
-    int is_secondary = primary!=0;
-    if (is_secondary) {
-	memset(&primary_key, 0, sizeof(primary_key));
-	sdata=&primary_key;
-    } else {
-	sdata=data;
-    }
+    assert(flags == 0 || flags==DB_NO_ASSOCIATE); // We aren't ready to handle flags such as DB_GET_BOTH  or DB_READ_COMMITTED or DB_READ_UNCOMMITTED or DB_RMW
+    if (db->i->primary==0 || flags==DB_NO_ASSOCIATE) {
+	// It's a get on a primary
 	toku_brt_get_flags(db->i->brt, &brtflags);
 	if (brtflags & TOKU_DB_DUPSORT) {
 	    DBC *dbc;
 	    r = db->cursor(db, txn, &dbc, 0);
-        if (r == 0) {
-            r = dbc->c_get(dbc, key, sdata, DB_SET);
-            dbc->c_close(dbc);
-        }
-    } else
-        r = toku_brt_lookup(db->i->brt, key, sdata);
 	    if (r!=0) return r;
-    if (is_secondary) {
-	// Time to do a get on the primary
-	return primary->get(primary, txn, &primary_key, data, 0);
+	    r = dbc->c_get(dbc, key, data, DB_SET);
+	    int r2 = dbc->c_close(dbc);
+	    if (r!=0) return r;
+	    return r2;
+	} else
+	    return toku_brt_lookup(db->i->brt, key, data);
     } else {
-	return r;
+	// It's a get on a secondary.
+	DBT primary_key;
+	memset(&primary_key, 0, sizeof(primary_key));
+	r = toku_db_get (db, txn, key, &primary_key, DB_NO_ASSOCIATE); // Lookup the primary
+	if (r!=0) return r;
+	// If data and primary_key are both zeroed, the temporary storage used to fill in data is different in the two cases because they come from different trees.
+	assert(db->i->brt != db->i->primary->i->brt); // Make sure they realy are different trees.
+	return toku_db_get (db->i->primary, txn, &primary_key, data, 0); 
     }
 }
 
+static int toku_db_pget (DB *db, DB_TXN *txn, DBT *key, DBT *pkey, DBT *data, u_int32_t flags) {
+    int r;
+    if (!db->i->primary) return EINVAL; // pget doesn't work on a primary.
+    assert(flags==0); // not ready to handle all those other options
+    r = toku_db_get (db, txn, key, pkey, DB_NO_ASSOCIATE);
+    if (r!=0) return r;
+    assert(db!=db->i->primary);
+    r = toku_db_get (db->i->primary, txn, pkey, data, 0);
+    return r;
+}
+
 static int toku_db_key_range(DB * db, DB_TXN * txn, DBT * dbt, DB_KEY_RANGE * kr, u_int32_t flags) {
     db=db; txn=txn; dbt=dbt; kr=kr; flags=flags;
     barf();
@@ -931,6 +942,7 @@ int db_create(DB ** db, DB_ENV * env, u_int32_t flags) {
     result->get = toku_db_get;
     result->key_range = toku_db_key_range;
     result->open = toku_db_open;
+    result->pget = toku_db_pget;
     result->put = toku_db_put;
     result->remove = toku_db_remove;
     result->rename = toku_db_rename;