From 129342b42cce0e4da0868cbc32e8bd64c2dc761e Mon Sep 17 00:00:00 2001
From: michim <michim@8dad8b1f-cf64-0410-95b6-bcf113ffbcfe>
Date: Thu, 12 May 2011 20:56:04 +0000
Subject: [PATCH] added remove().

git-svn-id: svn+ssh://svn.corp.yahoo.com/yahoo/yrl/labs/pnuts/code/logstore@2547 8dad8b1f-cf64-0410-95b6-bcf113ffbcfe
---
 logstore.cpp               | 13 ++++++++++++-
 sherpa/LSMServerHandler.cc | 26 ++++++++++----------------
 sherpa/main/lsm_client.cc  |  6 ++++++
 3 files changed, 28 insertions(+), 17 deletions(-)

diff --git a/logstore.cpp b/logstore.cpp
index 61b0257..ffe5469 100644
--- a/logstore.cpp
+++ b/logstore.cpp
@@ -437,6 +437,11 @@ datatuple * logtable<TUPLE>::findTuple(int xid, const datatuple::key_t key, size
 
     rwlc_unlock(header_mut);
     datatuple::freetuple(search_tuple);
+    if (ret_tuple != NULL && ret_tuple->isDelete()) {
+        // this is a tombstone. don't return it
+        datatuple::freetuple(ret_tuple);
+        return NULL;
+    }
     return ret_tuple;
 
 }
@@ -529,7 +534,13 @@ datatuple * logtable<TUPLE>::findTuple_first(int xid, datatuple::key_t key, size
     }
 
     datatuple::freetuple(search_tuple);
-    
+
+    if (ret_tuple != NULL && ret_tuple->isDelete()) {
+        // this is a tombstone. don't return it
+        datatuple::freetuple(ret_tuple);
+        return NULL;
+    }
+   
     return ret_tuple;
 
 }
diff --git a/sherpa/LSMServerHandler.cc b/sherpa/LSMServerHandler.cc
index 7d9835e..0e77f9b 100644
--- a/sherpa/LSMServerHandler.cc
+++ b/sherpa/LSMServerHandler.cc
@@ -246,7 +246,8 @@ get(datatuple* tuple)
 {
     // -1 is invalid txn id 
     //return ltable_->findTuple_first(-1, tuple->strippedkey(), tuple->strippedkeylen());
-    return ltable_->findTuple_first(-1, tuple->rawkey(), tuple->rawkeylen());
+    datatuple* tup = ltable_->findTuple_first(-1, tuple->rawkey(), tuple->rawkeylen());
+    return tup;
 }
 
 void LSMServerHandler::
@@ -349,7 +350,6 @@ update(const std::string& databaseName,
        const std::string& recordBody) 
 {
     uint32_t id = getDatabaseId(databaseName);
-    std::cout << "michim: enter update" << std::endl;
     if (id == 0) {
         return sherpa::ResponseCode::DatabaseNotFound;
     }
@@ -357,7 +357,6 @@ update(const std::string& databaseName,
     if (oldRecordBody == NULL) {
         return sherpa::ResponseCode::RecordNotFound;
     }
-    std::cout << "michim: updating record" << std::endl;
     datatuple::freetuple(oldRecordBody);
     datatuple* tup = buildTuple(id, recordName, recordBody);
     return insert(tup);
@@ -366,22 +365,17 @@ update(const std::string& databaseName,
 ResponseCode::type LSMServerHandler::
 remove(const std::string& databaseName, const std::string& recordName) 
 {
-/*
-    uint32_t id;
-    ResponseCode::type rc = getDatabaseId(databaseName, id);
-    if (rc != sherpa::ResponseCode::Ok) {
-        return rc;
+    uint32_t id = getDatabaseId(databaseName);
+    if (id == 0) {
+        return sherpa::ResponseCode::DatabaseNotFound;
     }
-    insertDatabaseId(const_cast<std::string&>(recordName), id);
-    Bdb::ResponseCode dbrc = db_[id % numPartitions_]->remove(recordName);
-    if (dbrc == Bdb::Ok) {
-        return sherpa::ResponseCode::Ok;
-    } else if (dbrc == Bdb::KeyNotFound) {
+    datatuple* oldRecordBody = get(id, recordName);
+    if (oldRecordBody == NULL) {
         return sherpa::ResponseCode::RecordNotFound;
-    } else {
     }
-*/
-    return sherpa::ResponseCode::Error;
+    datatuple::freetuple(oldRecordBody);
+    datatuple* tup = buildTuple(id, recordName);
+    return insert(tup);
 }
 
 /*
diff --git a/sherpa/main/lsm_client.cc b/sherpa/main/lsm_client.cc
index 3f1ae38..477dcc9 100644
--- a/sherpa/main/lsm_client.cc
+++ b/sherpa/main/lsm_client.cc
@@ -103,6 +103,12 @@ int main(int argc, char **argv) {
     client.get(getResponse, db, "new key");
     cout << getResponse.responseCode << endl;
     cout << getResponse.value << endl;
+
+    cout << client.remove(db1, "new key") << endl;
+    client.get(getResponse, db1, "new key");
+    cout << getResponse.responseCode << endl;
+    cout << client.remove(db1, "new key") << endl;
+
     return 0;
 
     /*