ARIES recovery update.

delete - primary key vs before image (SQL exception)
insert - before image (SQL exception)
bulkload works.
truncate works.

src/ee/common/NValue.hpp

@@ -44,8 +44,6 @@
 #include "serializeio.h"
 #include "SerializableEEException.h"
 #include "SQLException.h"
-//#include "common/StringRef.h"
-#include "common/ThreadLocalPool.h"
 #include "types.h"
 #include "value_defs.h"
 
@@ -1731,9 +1729,9 @@ inline const NValue NValue::deserializeFromTupleStorage(const void *storage,
                                                  const ValueType type,
                                                  const bool isInlined)
 {
-		VOLT_WARN("From :: Type : %d :: Storage : %p isInlined : %d ", type, storage, (int)isInlined);
+		//VOLT_WARN("ARIES :: From :: Type : %d :: Storage : %p isInlined : %d ", type, storage, (int)isInlined);
 
-        NValue retval(type);
+		NValue retval(type);
         switch (type) {
           case VALUE_TYPE_TIMESTAMP:
             retval.getTimestamp() = *reinterpret_cast<const int64_t*>(storage);
@@ -1863,7 +1861,7 @@ inline void NValue::serializeToTupleStorageAllocateForObjects(void *storage, con
 inline void NValue::serializeToTupleStorage(void *storage, const bool isInlined, const int32_t maxLength) const
 {
     const ValueType type = getValueType();
-	VOLT_WARN("To :: Type : %d :: Storage : %p isInlined : %d maxLength: %d", type, storage, (int)isInlined, maxLength);
+	//VOLT_WARN("ARIES :: To :: Type : %d :: Storage : %p isInlined : %d maxLength: %d", type, storage, (int)isInlined, maxLength);
 
     switch (type) {
       case VALUE_TYPE_TIMESTAMP:

src/ee/execution/VoltDBEngine.cpp

@@ -761,13 +761,15 @@ bool VoltDBEngine::loadTable(PersistentTable *table,
 
 		LogManager* m_logManager = getLogManager();
 		Logger m_ariesLogger = m_logManager->getAriesLogger();
+		/*
 		const Logger *logger = m_logManager->getThreadLogger(LOGGERID_MM_ARIES);
 
 		assert(logger != NULL);
 
 		// we could ALSO directly write via writeToAriesLogBuffer(buffer, size)
 		// but not doing that for consistency while logging to Aries.
 
+		// CHANGE :: skip
 		logger->log(LOGLEVEL_INFO, output.data(), output.position());
 
 		// CAREFUL -- the number of bytes might just be too many
@@ -781,13 +783,13 @@ bool VoltDBEngine::loadTable(PersistentTable *table,
 		int64_t value = htonll(numBytes);
 
 		// first log the size of the bulkload array
-		logger->log(LOGLEVEL_INFO, reinterpret_cast<char*>(&value),
-				sizeof(value));
+		// CHANGE :: skip
+		logger->log(LOGLEVEL_INFO, reinterpret_cast<char*>(&value), sizeof(value));
 
 		// next log the raw bytes of the bulkload array
-		logger->log(LOGLEVEL_INFO,
-				reinterpret_cast<const char *>(serializeIn.getRawPointer(0)),
-				numBytes);
+		// CHANGE :: skip
+		logger->log(LOGLEVEL_INFO, reinterpret_cast<const char *>(serializeIn.getRawPointer(0)), numBytes);
+		*/
 
 		delete[] logrecordBuffer;
 		logrecordBuffer = NULL;
@@ -1656,6 +1658,13 @@ void VoltDBEngine::doAriesRecovery(char *logData, size_t length, int64_t replay_
 			VOLT_WARN("Log record recovery : BULKLOAD");
 		} else if (logrecord.getType() == LogRecord::T_DELETE) {
 			beforeImage = logrecord.getTupleBeforeImage();
+			TableTuple* primaryKey = logrecord.getPrimaryKey();
+
+			if(primaryKey != NULL)
+				VOLT_WARN("DEBUG PKEY : %s", primaryKey->debugNoHeader().c_str());
+			if(beforeImage != NULL)
+				VOLT_WARN("DEBUG BEFORE IMAGE : %s", beforeImage->debugNoHeader().c_str());
+
 			table->deleteTuple(*beforeImage, true);
 
 			logrecord.dellocateBeforeImageData();

src/ee/executors/deleteexecutor.cpp

@@ -95,6 +95,7 @@ bool DeleteExecutor::p_init(AbstractPlanNode *abstract_node, const catalog::Data
 
 bool DeleteExecutor::p_execute(const NValueArray &params, ReadWriteTracker *tracker) {
     assert(m_targetTable);
+
     if (m_truncate) {
         VOLT_TRACE("truncating table %s...", m_targetTable->name().c_str());
         // count the truncated tuples as deleted
@@ -218,10 +219,21 @@ bool DeleteExecutor::p_execute(const NValueArray &params, ReadWriteTracker *trac
 		LogManager* m_logManager = this->m_engine->getLogManager();
 		Logger m_ariesLogger = m_logManager->getAriesLogger();
 		//VOLT_WARN("m_logManager : %p AriesLogger : %p",&m_logManager, &m_ariesLogger);
+
+		/*
 		const Logger *logger = m_logManager->getThreadLogger(LOGGERID_MM_ARIES);
 
+		// CHANGE ::
 		logger->log(LOGLEVEL_INFO, output.data(), output.position());
 
+		if(beforeImage != NULL){
+			VOLT_WARN("DEBUG : %s", beforeImage->debugNoHeader().c_str());
+		}
+		else if(keyTuple != NULL){
+			VOLT_WARN("DEBUG : beforeImage null :: keyTuple : %s", keyTuple->debugNoHeader().c_str());
+		}
+		*/
+
 		delete[] logrecordBuffer;
 		logrecordBuffer = NULL;
 

src/ee/logging/Logrecord.cpp

@@ -292,6 +292,10 @@ TableTuple* LogRecord::initRecordTuple() {
 		recordTuple->setNValue(11, ValueFactory::getNullStringValue());
 	}
 
+	// debug
+	std::string recordValue = recordTuple->debugNoHeader();
+	VOLT_WARN("Log Record Tuple : %s",recordValue.c_str());
+
 	return recordTuple;
 }
 

src/ee/logging/Logrecord.h

@@ -121,6 +121,14 @@ class LogRecord {
 		return NULL;
 	}
 
+	inline TableTuple* getPrimaryKey() {
+		if (isValid) {
+			return primaryKey;
+		}
+
+		return NULL;
+	}
+
     void dellocateBeforeImageData()
     {
     	if (beforeImageData != NULL) {

tests/frontend/org/voltdb/regressionsuites/TestSnapshotRecovery.java

@@ -57,6 +57,7 @@
 import edu.brown.benchmark.ycsb.YCSBConstants;
 import edu.brown.benchmark.ycsb.YCSBLoader;
 import edu.brown.benchmark.ycsb.YCSBProjectBuilder;
+import edu.brown.benchmark.ycsb.YCSBUtil;
 import edu.brown.benchmark.ycsb.procedures.DeleteRecord;
 import edu.brown.benchmark.ycsb.procedures.InsertRecord;
 import edu.brown.benchmark.ycsb.procedures.ReadRecord;
@@ -192,8 +193,6 @@ public CatalogContext getCatalogContext() {
         loader.load();
     }    
 
-    int numDeletedTuples = 2;
-
     public void testSaveAndRestoreYCSB() throws IOException, InterruptedException, ProcCallException {
 
         System.out.println("Starting testSaveAndRestoreYCSB");
@@ -230,7 +229,8 @@ public void testSaveAndRestoreYCSB() throws IOException, InterruptedException, P
 
         long key = NUM_TUPLES / 2;
         String procName = ReadRecord.class.getSimpleName();
-        Object params[] = { key };
+        Object params[] ;
+        params = new Object[]{ key };
 
         cresponse = client.callProcedure(procName, params);
         assertNotNull(cresponse);
@@ -242,13 +242,36 @@ public void testSaveAndRestoreYCSB() throws IOException, InterruptedException, P
         assert(adv);
         assertEquals(key, vt.getLong(0));
 
+        // Delete, then Insert these many tuples back
+        int numTestTuples = 2;
+
         System.out.println("Read Record Test Passed");
 
-        for (long k_itr = 0; k_itr < numDeletedTuples; k_itr++) {
+        for (long k_itr = 0; k_itr < numTestTuples; k_itr++) {
             procName = DeleteRecord.class.getSimpleName();
-            Object paramst[] = { k_itr };
+            key = k_itr;
+            params =  new Object[]{ key };
 
-            cresponse = client.callProcedure(procName, paramst);
+            cresponse = client.callProcedure(procName, params);
+            assertEquals(Status.OK, cresponse.getStatus());
+            results = cresponse.getResults();
+
+            assertEquals(1, results.length);
+            assertNotNull(cresponse);
+        }
+
+        System.out.println("Delete Record Test Passed");
+
+        for (long k_itr = 0; k_itr < numTestTuples; k_itr++) {
+            procName = InsertRecord.class.getSimpleName();
+            key = k_itr;
+            String fields[] = new String[YCSBConstants.NUM_COLUMNS];
+            for (int i = 0; i < fields.length; i++) {
+                fields[i] = YCSBUtil.astring(YCSBConstants.COLUMN_LENGTH, YCSBConstants.COLUMN_LENGTH);
+            } // FOR
+            params = new Object[]{ key, fields };
+
+            cresponse = client.callProcedure(procName, params);
             assertEquals(Status.OK, cresponse.getStatus());
             results = cresponse.getResults();
 
@@ -385,7 +408,7 @@ private void checkYCSBTable(Client client, int numTuples) {
         VoltTable vt = null;
         boolean adv = true;
 
-        for(key_itr = numDeletedTuples ; key_itr < numTuples ; key_itr++){
+        for(key_itr = 0 ; key_itr < numTuples ; key_itr++){
             procName = ReadRecord.class.getSimpleName();
             Object params[] = { key_itr };