[15721] TileGroup Compaction and GC Fixes

src/catalog/manager.cpp

@@ -47,4 +47,4 @@ void Manager::DropIndirectionArray(const oid_t oid) {
 void Manager::ClearIndirectionArray() { indirection_array_locator_.clear(); }
 
 }  // namespace catalog
-}  // namespace peloton
+}  // namespace peloton

src/codegen/deleter.cpp

@@ -38,6 +38,7 @@ void Deleter::Delete(uint32_t tile_group_id, uint32_t tuple_offset) {
 
   auto *txn = executor_context_->GetTransaction();
   auto tile_group = table_->GetTileGroupById(tile_group_id);
+  PELOTON_ASSERT(tile_group != nullptr);
   auto *tile_group_header = tile_group->GetHeader();
 
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();

src/codegen/inserter.cpp

@@ -34,6 +34,8 @@ void Inserter::Init(storage::DataTable *table,
 char *Inserter::AllocateTupleStorage() {
   location_ = table_->GetEmptyTupleSlot(nullptr);
 
+  PELOTON_ASSERT(location_.IsNull() == false);
+
   // Get the tile offset assuming that it is a row store
   auto tile_group = table_->GetTileGroupById(location_.block);
   auto layout = tile_group->GetLayout();
@@ -54,8 +56,11 @@ void Inserter::Insert() {
   auto *txn = executor_context_->GetTransaction();
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
 
-  ContainerTuple<storage::TileGroup> tuple(
-      table_->GetTileGroupById(location_.block).get(), location_.offset);
+  auto tile_group = table_->GetTileGroupById(location_.block).get();
+  PELOTON_ASSERT(tile_group != nullptr);
+
+  ContainerTuple<storage::TileGroup> tuple(tile_group, location_.offset);
+
   ItemPointer *index_entry_ptr = nullptr;
   bool result = table_->InsertTuple(&tuple, location_, txn, &index_entry_ptr);
   if (result == false) {

src/codegen/updater.cpp

@@ -61,6 +61,7 @@ char *Updater::Prepare(uint32_t tile_group_id, uint32_t tuple_offset) {
 
   auto *txn = executor_context_->GetTransaction();
   auto tile_group = table_->GetTileGroupById(tile_group_id).get();
+  PELOTON_ASSERT(tile_group != nullptr);
   auto *tile_group_header = tile_group->GetHeader();
   old_location_.block = tile_group_id;
   old_location_.offset = tuple_offset;
@@ -91,6 +92,7 @@ char *Updater::PreparePK(uint32_t tile_group_id, uint32_t tuple_offset) {
 
   auto *txn = executor_context_->GetTransaction();
   auto tile_group = table_->GetTileGroupById(tile_group_id).get();
+  PELOTON_ASSERT(tile_group != nullptr);
   auto *tile_group_header = tile_group->GetHeader();
 
   // Check ownership
@@ -135,6 +137,7 @@ void Updater::Update() {
             table_->GetOid());
   auto *txn = executor_context_->GetTransaction();
   auto tile_group = table_->GetTileGroupById(old_location_.block).get();
+  PELOTON_ASSERT(tile_group != nullptr);
   auto *tile_group_header = tile_group->GetHeader();
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
   // Either update in-place
@@ -147,8 +150,10 @@ void Updater::Update() {
   }
 
   // Or, update with a new version
-  ContainerTuple<storage::TileGroup> new_tuple(
-    table_->GetTileGroupById(new_location_.block).get(), new_location_.offset);
+  auto new_tile_group = table_->GetTileGroupById(new_location_.block);
+  PELOTON_ASSERT(new_tile_group != nullptr);
+  ContainerTuple<storage::TileGroup> new_tuple(new_tile_group.get(),
+                                               new_location_.offset);
   ItemPointer *indirection =
       tile_group_header->GetIndirection(old_location_.offset);
   auto result = table_->InstallVersion(&new_tuple, target_list_, txn,
@@ -171,6 +176,7 @@ void Updater::UpdatePK() {
             table_->GetOid());
   auto *txn = executor_context_->GetTransaction();
   auto tile_group = table_->GetTileGroupById(new_location_.block).get();
+  PELOTON_ASSERT(tile_group != nullptr);
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
 
   // Insert a new tuple

src/common/container/cuckoo_map.cpp

@@ -18,6 +18,9 @@
 #include "common/item_pointer.h"
 #include "common/logger.h"
 #include "common/macros.h"
+#include "common/container/lock_free_queue.h"
+#include "gc/recycle_stack.h"
+#include "storage/data_table.h"
 
 namespace peloton {
 
@@ -125,4 +128,7 @@ template class CuckooMap<std::shared_ptr<oid_t>, std::shared_ptr<oid_t>>;
 // Used in StatementCacheManager
 template class CuckooMap<StatementCache *, StatementCache *>;
 
+// Used in TransactionLevelGCManager
+template class CuckooMap<oid_t, std::shared_ptr<gc::RecycleStack>>;
+
 }  // namespace peloton

src/common/container/lock_free_array.cpp

@@ -53,6 +53,7 @@ template <typename ValueType>
 void LOCK_FREE_ARRAY_TYPE::Erase(const std::size_t &offset,
                                  const ValueType &invalid_value) {
   LOG_TRACE("Erase at %lu", offset);
+  PELOTON_ASSERT(lock_free_array.size() > offset);
   lock_free_array.at(offset) = invalid_value;
 }
 
@@ -107,6 +108,20 @@ bool LOCK_FREE_ARRAY_TYPE::Contains(const ValueType &value) const {
   return exists;
 }
 
+template <typename ValueType>
+ssize_t LOCK_FREE_ARRAY_TYPE::Lookup(const ValueType &value) {
+  for (std::size_t array_itr = 0; array_itr < lock_free_array.size();
+       array_itr++) {
+    auto array_value = lock_free_array.at(array_itr);
+    // Check array value
+    if (array_value == value) {
+      return array_itr;
+    }
+  }
+
+  return -1;
+}
+
 // Explicit template instantiation
 template class LockFreeArray<std::shared_ptr<oid_t>>;
 

src/common/init.cpp

@@ -52,7 +52,7 @@ void PelotonInit::Initialize() {
   threadpool::MonoQueuePool::GetExecutionInstance().Startup();
 
   int parallelism = (CONNECTION_THREAD_COUNT + 3) / 4;
-  storage::DataTable::SetActiveTileGroupCount(parallelism);
+  storage::DataTable::SetDefaultActiveTileGroupCount(parallelism);
   storage::DataTable::SetActiveIndirectionArrayCount(parallelism);
 
   // start epoch.

src/common/internal_types.cpp

@@ -2967,8 +2967,8 @@ std::string GCVersionTypeToString(GCVersionType type) {
     case GCVersionType::ABORT_UPDATE: {
       return "ABORT_UPDATE";
     }
-    case GCVersionType::ABORT_DELETE: {
-      return "ABORT_DELETE";
+    case GCVersionType::TOMBSTONE: {
+      return "TOMBSTONE";
     }
     case GCVersionType::ABORT_INSERT: {
       return "ABORT_INSERT";
@@ -2997,8 +2997,8 @@ GCVersionType StringToGCVersionType(const std::string &str) {
     return GCVersionType::COMMIT_INS_DEL;
   } else if (upper_str == "ABORT_UPDATE") {
     return GCVersionType::ABORT_UPDATE;
-  } else if (upper_str == "ABORT_DELETE") {
-    return GCVersionType::ABORT_DELETE;
+  } else if (upper_str == "TOMBSTONE") {
+    return GCVersionType::TOMBSTONE;
   } else if (upper_str == "ABORT_INSERT") {
     return GCVersionType::ABORT_INSERT;
   } else if (upper_str == "ABORT_INS_DEL") {

src/concurrency/timestamp_ordering_transaction_manager.cpp

@@ -290,9 +290,9 @@ bool TimestampOrderingTransactionManager::PerformRead(TransactionContext *const
   //////////////////////////////////////////////////////////
   else {
     PELOTON_ASSERT(current_txn->GetIsolationLevel() ==
-                       IsolationLevelType::SERIALIZABLE ||
-                   current_txn->GetIsolationLevel() ==
-                       IsolationLevelType::REPEATABLE_READS);
+                  IsolationLevelType::SERIALIZABLE ||
+              current_txn->GetIsolationLevel() ==
+                  IsolationLevelType::REPEATABLE_READS);
 
     oid_t tuple_id = location.offset;
 
@@ -580,6 +580,9 @@ void TimestampOrderingTransactionManager::PerformDelete(
   current_txn->RecordDelete(old_location);
 }
 
+// Performs Delete on a tuple that was created by the current transaction, and
+// never
+// installed into the database
 void TimestampOrderingTransactionManager::PerformDelete(
     TransactionContext *const current_txn, const ItemPointer &location) {
   PELOTON_ASSERT(!current_txn->IsReadOnly());
@@ -737,6 +740,9 @@ ResultType TimestampOrderingTransactionManager::CommitTransaction(
       gc_set->operator[](tile_group_id)[tuple_slot] =
           GCVersionType::COMMIT_DELETE;
 
+      gc_set->operator[](new_version.block)[new_version.offset] =
+          GCVersionType::TOMBSTONE;
+
       log_manager.LogDelete(ItemPointer(tile_group_id, tuple_slot));
 
     } else if (tuple_entry.second == RWType::INSERT) {
@@ -855,9 +861,11 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
       // before we unlink the aborted version from version list
       ItemPointer *index_entry_ptr =
           tile_group_header->GetIndirection(tuple_slot);
-      UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
-          index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
-      PELOTON_ASSERT(res == true);
+      if (index_entry_ptr) {
+        UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
+            index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
+        PELOTON_ASSERT(res == true);
+      }
       //////////////////////////////////////////////////
 
       // we should set the version before releasing the lock.
@@ -903,9 +911,11 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
       // before we unlink the aborted version from version list
       ItemPointer *index_entry_ptr =
           tile_group_header->GetIndirection(tuple_slot);
-      UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
-          index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
-      PELOTON_ASSERT(res == true);
+      if (index_entry_ptr) {
+        UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
+            index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
+        PELOTON_ASSERT(res == true);
+      }
       //////////////////////////////////////////////////
 
       // we should set the version before releasing the lock.
@@ -923,7 +933,7 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
 
       // add the version to gc set.
       gc_set->operator[](new_version.block)[new_version.offset] =
-          GCVersionType::ABORT_DELETE;
+          GCVersionType::TOMBSTONE;
 
     } else if (tuple_entry.second == RWType::INSERT) {
       tile_group_header->SetBeginCommitId(tuple_slot, MAX_CID);

src/executor/hybrid_scan_executor.cpp

@@ -122,6 +122,9 @@ bool HybridScanExecutor::DInit() {
         tile_group = table_->GetTileGroup(table_tile_group_count_ - 1);
       }
 
+      // TODO: Handle possibility of freed tile_group
+      PELOTON_ASSERT(tile_group != nullptr);
+
       oid_t tuple_id = 0;
       ItemPointer location(tile_group->GetTileGroupId(), tuple_id);
       block_threshold = location.block;
@@ -189,6 +192,10 @@ bool HybridScanExecutor::SeqScanUtil() {
   while (current_tile_group_offset_ < table_tile_group_count_) {
     LOG_TRACE("Current tile group offset : %u", current_tile_group_offset_);
     auto tile_group = table_->GetTileGroup(current_tile_group_offset_++);
+    if (tile_group == nullptr) {
+      continue;
+    }
+
     auto tile_group_header = tile_group->GetHeader();
 
     oid_t active_tuple_count = tile_group->GetNextTupleSlot();
@@ -380,6 +387,9 @@ bool HybridScanExecutor::ExecPrimaryIndexLookup() {
 
     auto storage_manager = storage::StorageManager::GetInstance();
     auto tile_group = storage_manager->GetTileGroup(tuple_location.block);
+
+    // TODO: Handle possibility of freed tile_group
+    PELOTON_ASSERT(tile_group != nullptr);
     auto tile_group_header = tile_group.get()->GetHeader();
 
     // perform transaction read
@@ -427,6 +437,9 @@ bool HybridScanExecutor::ExecPrimaryIndexLookup() {
         }
 
         tile_group = storage_manager->GetTileGroup(tuple_location.block);
+
+        // TODO: Handle possibility of freed tile_group
+        PELOTON_ASSERT(tile_group != nullptr);
         tile_group_header = tile_group.get()->GetHeader();
       }
     }
@@ -437,6 +450,9 @@ bool HybridScanExecutor::ExecPrimaryIndexLookup() {
     auto storage_manager = storage::StorageManager::GetInstance();
     auto tile_group = storage_manager->GetTileGroup(tuples.first);
 
+    // TODO: Handle possibility of freed tile_group
+    PELOTON_ASSERT(tile_group != nullptr);
+
     std::unique_ptr<LogicalTile> logical_tile(LogicalTileFactory::GetTile());
 
     // Add relevant columns to logical tile

src/executor/index_scan_executor.cpp

@@ -217,6 +217,7 @@ bool IndexScanExecutor::ExecPrimaryIndexLookup() {
   for (auto tuple_location_ptr : tuple_location_ptrs) {
     ItemPointer tuple_location = *tuple_location_ptr;
     auto tile_group = storage_manager->GetTileGroup(tuple_location.block);
+    PELOTON_ASSERT(tile_group != nullptr);
     auto tile_group_header = tile_group.get()->GetHeader();
     size_t chain_length = 0;
 
@@ -294,6 +295,7 @@ bool IndexScanExecutor::ExecPrimaryIndexLookup() {
               *(tile_group_header->GetIndirection(tuple_location.offset));
           auto storage_manager = storage::StorageManager::GetInstance();
           tile_group = storage_manager->GetTileGroup(tuple_location.block);
+          PELOTON_ASSERT(tile_group != nullptr);
           tile_group_header = tile_group.get()->GetHeader();
           chain_length = 0;
           continue;
@@ -320,6 +322,7 @@ bool IndexScanExecutor::ExecPrimaryIndexLookup() {
         // search for next version.
         auto storage_manager = storage::StorageManager::GetInstance();
         tile_group = storage_manager->GetTileGroup(tuple_location.block);
+        PELOTON_ASSERT(tile_group != nullptr);
         tile_group_header = tile_group.get()->GetHeader();
         continue;
       }
@@ -355,6 +358,7 @@ bool IndexScanExecutor::ExecPrimaryIndexLookup() {
       // into the result vector
       auto storage_manager = storage::StorageManager::GetInstance();
       auto tile_group = storage_manager->GetTileGroup(current_tile_group_oid);
+      PELOTON_ASSERT(tile_group != nullptr);
       std::unique_ptr<LogicalTile> logical_tile(LogicalTileFactory::GetTile());
       // Add relevant columns to logical tile
       logical_tile->AddColumns(tile_group, full_column_ids_);
@@ -374,6 +378,7 @@ bool IndexScanExecutor::ExecPrimaryIndexLookup() {
   // Add the remaining tuples to the result vector
   if ((current_tile_group_oid != INVALID_OID) && (!tuples.empty())) {
     auto tile_group = storage_manager->GetTileGroup(current_tile_group_oid);
+    PELOTON_ASSERT(tile_group != nullptr);
     std::unique_ptr<LogicalTile> logical_tile(LogicalTileFactory::GetTile());
     // Add relevant columns to logical tile
     logical_tile->AddColumns(tile_group, full_column_ids_);
@@ -464,6 +469,7 @@ bool IndexScanExecutor::ExecSecondaryIndexLookup() {
     ItemPointer tuple_location = *tuple_location_ptr;
     if (tuple_location.block != last_block) {
       tile_group = storage_manager->GetTileGroup(tuple_location.block);
+      PELOTON_ASSERT(tile_group != nullptr);
       tile_group_header = tile_group.get()->GetHeader();
     }
 #ifdef LOG_TRACE_ENABLED
@@ -565,6 +571,7 @@ bool IndexScanExecutor::ExecSecondaryIndexLookup() {
           tuple_location =
               *(tile_group_header->GetIndirection(tuple_location.offset));
           tile_group = storage_manager->GetTileGroup(tuple_location.block);
+          PELOTON_ASSERT(tile_group != nullptr);
           tile_group_header = tile_group.get()->GetHeader();
           chain_length = 0;
           continue;
@@ -594,6 +601,7 @@ bool IndexScanExecutor::ExecSecondaryIndexLookup() {
 
         // search for next version.
         tile_group = storage_manager->GetTileGroup(tuple_location.block);
+        PELOTON_ASSERT(tile_group != nullptr);
         tile_group_header = tile_group.get()->GetHeader();
       }
     }
@@ -621,6 +629,7 @@ bool IndexScanExecutor::ExecSecondaryIndexLookup() {
       // Since the tile_group_oids differ, fill in the current tile group
       // into the result vector
       auto tile_group = storage_manager->GetTileGroup(current_tile_group_oid);
+      PELOTON_ASSERT(tile_group != nullptr);
       std::unique_ptr<LogicalTile> logical_tile(LogicalTileFactory::GetTile());
       // Add relevant columns to logical tile
       logical_tile->AddColumns(tile_group, full_column_ids_);
@@ -640,6 +649,7 @@ bool IndexScanExecutor::ExecSecondaryIndexLookup() {
   // Add the remaining tuples (if any) to the result vector
   if ((current_tile_group_oid != INVALID_OID) && (!tuples.empty())) {
     auto tile_group = storage_manager->GetTileGroup(current_tile_group_oid);
+    PELOTON_ASSERT(tile_group != nullptr);
     std::unique_ptr<LogicalTile> logical_tile(LogicalTileFactory::GetTile());
     // Add relevant columns to logical tile
     logical_tile->AddColumns(tile_group, full_column_ids_);
@@ -691,6 +701,7 @@ bool IndexScanExecutor::CheckKeyConditions(const ItemPointer &tuple_location) {
 
   auto storage_manager = storage::StorageManager::GetInstance();
   auto tile_group = storage_manager->GetTileGroup(tuple_location.block);
+  PELOTON_ASSERT(tile_group != nullptr);
   ContainerTuple<storage::TileGroup> tuple(tile_group.get(),
                                            tuple_location.offset);
 

src/executor/seq_scan_executor.cpp

@@ -154,6 +154,12 @@ bool SeqScanExecutor::DExecute() {
     while (current_tile_group_offset_ < table_tile_group_count_) {
       auto tile_group =
           target_table_->GetTileGroup(current_tile_group_offset_++);
+
+      if (tile_group == nullptr) {
+        // tile group was freed so, continue to next tile group
+        continue;
+      }
+
       auto tile_group_header = tile_group->GetHeader();
 
       oid_t active_tuple_count = tile_group->GetNextTupleSlot();