refactoring write-recording/rollback and rogging/recovery of Update: …

…focusing TestUndo.

lib/recovery/log_manager.go

@@ -160,6 +160,14 @@ func (log_manager *LogManager) AppendLogRecord(log_record *LogRecord) types.LSN
 		binary.Write(buf, binary.LittleEndian, log_record.Prev_page_id)
 		pageIdInBytes := buf.Bytes()
 		copy(log_manager.log_buffer[pos:], pageIdInBytes)
+	} else if log_record.Log_record_type == RESERVE_SPACE {
+		buf := new(bytes.Buffer)
+		binary.Write(buf, binary.LittleEndian, log_record.Reserving_rid)
+		ridInBytes := buf.Bytes()
+		copy(log_manager.log_buffer[pos:], ridInBytes)
+		pos += uint32(unsafe.Sizeof(log_record.Reserving_rid))
+		// we have provided serialize function for tuple class
+		log_record.Reserving_tuple.SerializeTo(log_manager.log_buffer[pos:])
 	}
 	// TODO: (SDB) need to implement serialization of RESERVE_SPACE type log
 

lib/recovery/log_recovery/log_recovery.go

@@ -81,8 +81,11 @@ func (log_recovery *LogRecovery) DeserializeLogRecord(data []byte, log_record *r
 		log_record.New_tuple.DeserializeFrom(data[pos:])
 	} else if log_record.Log_record_type == recovery.NEWPAGE {
 		binary.Read(bytes.NewBuffer(data[pos:]), binary.LittleEndian, &log_record.Prev_page_id)
+	} else if log_record.Log_record_type == recovery.RESERVE_SPACE {
+		binary.Read(bytes.NewBuffer(data[pos:]), binary.LittleEndian, &log_record.Reserving_rid)
+		pos += uint32(unsafe.Sizeof(log_record.Reserving_rid))
+		log_record.Reserving_tuple.DeserializeFrom(data[pos:])
 	}
-	// TODO: (SDB) need to implement deserialization of RESERVE_SPACE type log
 
 	//fmt.Println(log_record)
 

lib/recovery/recovery_test/log_recovery_test.go

@@ -160,7 +160,7 @@ func TestUndo(t *testing.T) {
 		samehada_instance.GetLogManager(),
 		samehada_instance.GetLockManager(),
 		txn)
-	first_page_id := test_table.GetFirstPageId()
+	//first_page_id := test_table.GetFirstPageId()
 
 	col1 := column.NewColumn("a", types.Varchar, false, index_constants.INDEX_KIND_INVALID, types.PageID(-1), nil)
 	col2 := column.NewColumn("b", types.Integer, false, index_constants.INDEX_KIND_INVALID, types.PageID(-1), nil)
@@ -207,26 +207,12 @@ func TestUndo(t *testing.T) {
 
 	fmt.Println("Log page content is written to disk")
 	samehada_instance.GetLogManager().Flush()
-	fmt.Println("Table page content is written to disk")
-	samehada_instance.GetBufferPoolManager().FlushPage(first_page_id)
-
-	fmt.Println("System crash before commit")
-	// delete samehada_instance
-	samehada_instance.Shutdown(samehada.ShutdownPatternCloseFiles)
-
-	fmt.Println("System restarted..")
-	samehada_instance = samehada.NewSamehadaInstance(t.Name(), common.BufferPoolMaxFrameNumForTest)
-	txn = samehada_instance.GetTransactionManager().Begin(nil)
-
-	test_table = access.NewTableHeap(
-		samehada_instance.GetBufferPoolManager(),
-		samehada_instance.GetLogManager(),
-		samehada_instance.GetLockManager(),
-		txn)
+	//fmt.Println("Table page content is written to disk")
+	//samehada_instance.GetBufferPoolManager().FlushPage(first_page_id)
 
 	fmt.Println("Check if deleted tuple does not exist before recovery")
 	old_tuple1, _ := test_table.GetTuple(rid1, txn)
-	testingpkg.Assert(t, old_tuple1 == nil, "")
+	testingpkg.Assert(t, old_tuple1.Size() == 0, "handled as self deleted case")
 
 	fmt.Println("Check if updated tuple values are effected before recovery")
 	var old_tuple2 *tuple.Tuple
@@ -242,6 +228,20 @@ func TestUndo(t *testing.T) {
 	samehada_instance.GetLogManager().DeactivateLogging()
 	testingpkg.AssertFalse(t, samehada_instance.GetLogManager().IsEnabledLogging(), "common.EnableLogging is not false!")
 
+	fmt.Println("System crash before commit")
+	// delete samehada_instance
+	samehada_instance.Shutdown(samehada.ShutdownPatternCloseFiles)
+
+	fmt.Println("System restarted..")
+	samehada_instance = samehada.NewSamehadaInstance(t.Name(), common.BufferPoolMaxFrameNumForTest)
+	txn = samehada_instance.GetTransactionManager().Begin(nil)
+
+	test_table = access.NewTableHeap(
+		samehada_instance.GetBufferPoolManager(),
+		samehada_instance.GetLogManager(),
+		samehada_instance.GetLockManager(),
+		txn)
+
 	fmt.Println("Recovery started..")
 	log_recovery := log_recovery.NewLogRecovery(
 		samehada_instance.GetDiskManager(),

lib/samehada/samehada_instance.go

@@ -88,7 +88,7 @@ func (si *SamehadaInstance) Shutdown(shutdownPat ShutdownPattern) {
 		si.disk_manager.RemoveLogFile()
 	case ShutdownPatternCloseFiles:
 		si.log_manager.Flush()
-		si.bpm.FlushAllDirtyPages()
+		//si.bpm.FlushAllDirtyPages()
 		// close only
 		si.disk_manager.ShutDown()
 	case ShutdownPatternRemoveLogOnly:

lib/storage/access/table_page.go

@@ -242,9 +242,9 @@ func (tp *TablePage) UpdateTuple(new_tuple *tuple.Tuple, update_col_idxs []int,
 		txn.SetPrevLSN(lsn)
 	}
 
-	if new_tuple.Size() <= tuple_size {
+	if update_tuple.Size() < tuple_size {
 		// add dummy tuple which reserves space for update is aborted
-		tp.ReserveSpaceForRollbackUpdate(nil, tuple_size-new_tuple.Size(), txn, log_manager)
+		tp.ReserveSpaceForRollbackUpdate(nil, tuple_size-update_tuple.Size(), txn, log_manager)
 	}
 
 	// Perform the update.
@@ -285,7 +285,7 @@ func (tp *TablePage) ReserveSpaceForRollbackUpdate(rid *page.RID, size uint32, t
 	tp.setTuple(dummy_rid.GetSlotNum(), dummy_tuple)
 
 	if size > 0 {
-		tp.SetTupleSize(dummy_rid.GetSlotNum(), SetDeletedFlag(size))
+		tp.SetTupleSize(dummy_rid.GetSlotNum(), SetReservedFlag(size))
 	}
 
 	if log_manager.IsEnabledLogging() {

lib/storage/access/transaction_manager.go

@@ -192,7 +192,7 @@ func (transaction_manager *TransactionManager) Abort(catalog_ catalog_interface.
 				indexes := catalog_.GetRollbackNeededIndexes(indexMap, item.oid)
 				for _, index_ := range indexes {
 					if index_ != nil {
-						index_.DeleteEntry(item.tuple1, *item.rid, txn)
+						index_.DeleteEntry(item.tuple2, *item.rid, txn)
 					}
 				}
 			}
@@ -237,6 +237,7 @@ func (transaction_manager *TransactionManager) Abort(catalog_ catalog_interface.
 				}
 			}
 		} else if item.wtype == RESERVE_SPACE {
+			// TODO: (SDB) critical section for this operation should be concatenated with UPDATE case...
 			if common.EnableDebug && common.ActiveLogKindSetting&common.COMMIT_ABORT_HANDLE_INFO > 0 {
 				fmt.Printf("TransactionManager::Commit handle UPDATE write log. txn.txn_id:%v dbgInfo:%s rid:%v\n", txn.txn_id, txn.dbgInfo, item.rid)
 			}