feat(tianmu):delta store,Batch merge from Rocksdb to base layer

stoneatom · Mar 7, 2023 · 8beb08f · 8beb08f
1 parent ca806c6
commit 8beb08f
Show file tree

Hide file tree

Showing 17 changed files with 877 additions and 417 deletions.
diff --git a/storage/tianmu/core/blocked_mem_table.cpp b/storage/tianmu/core/blocked_mem_table.cpp
@@ -73,14 +73,15 @@ void MemBlockManager::FreeBlock(void *b) {
         current_size -= block_size;
       } else
         free_blocks.push_back(b);
-    } else
-        // often freed block are immediately required by other - keep some of them
-        // above the limit in the pool
-        if (current_size > size_limit + block_size * (no_threads + 1)) {
-      dealloc(b);
-      current_size -= block_size;
-    } else
-      free_blocks.push_back(b);
+    } else {
+      // often freed block are immediately required by other - keep some of them
+      // above the limit in the pool
+      if (current_size > size_limit + block_size * (no_threads + 1)) {
+        dealloc(b);
+        current_size -= block_size;
+      } else
+        free_blocks.push_back(b);
+    }
   }  // else not found (already erased)
 }
 

diff --git a/storage/tianmu/core/column_share.cpp b/storage/tianmu/core/column_share.cpp
@@ -141,6 +141,9 @@ void ColumnShare::scan_dpn(common::TX_ID xid) {
 
   ASSERT(hdr.numOfPacks <= capacity, "bad dpn index");
 
+  // get column saved auto inc
+  auto_inc_.store(hdr.auto_inc);
+
   if (hdr.numOfPacks == 0) {
     for (uint32_t i = 0; i < capacity; i++) {
       start[i].reset();

diff --git a/storage/tianmu/core/column_share.h b/storage/tianmu/core/column_share.h
@@ -48,7 +48,7 @@ struct alignas(128) COL_VER_HDR_V3 {
   uint64_t numOfPacks;    // number of packs
   uint64_t numOfDeleted;  // number of deleted
 
-  uint64_t auto_inc_next;
+  uint64_t auto_inc;  // only use in commit write
   int64_t min;
   int64_t max;
   uint32_t dict_ver;  // dict file version name. 0 means n/a
@@ -100,6 +100,8 @@ class ColumnShare final {
   }
   std::string GetFieldName() const { return field_name_; }
 
+  void Truncate() { auto_inc_.store(0); }
+
  private:
   void Init(common::TX_ID xid);
   void map_dpn();
@@ -115,6 +117,7 @@ class ColumnShare final {
   common::PackType pt;
   uint32_t col_id;
   std::string field_name_;
+  std::atomic<uint64_t> auto_inc_{0};
   struct seg {
     uint64_t offset;
     uint64_t len;

diff --git a/storage/tianmu/core/column_type.h b/storage/tianmu/core/column_type.h
@@ -96,6 +96,7 @@ struct ColumnType {
   void OverrideInternalSize(uint size) { internal_size = size; };
   int GetDisplaySize() const { return display_size; }
   bool Lookup() const { return fmt == common::PackFmt::LOOKUP; }
+  bool IsLookup() const { return fmt == common::PackFmt::LOOKUP; }
   ColumnType RemovedLookup() const;
 
   bool IsNumeric() const {

diff --git a/storage/tianmu/core/descriptor.cpp b/storage/tianmu/core/descriptor.cpp
@@ -1717,10 +1717,11 @@ void Descriptor::CoerceColumnType(vcolumn::VirtualColumn *&for_typecast) {
 
   if (tcc) {
     if (tianmu_control_.isOn()) {
-      tianmu_control_.lock(current_txn_->GetThreadID())
-          << "Type conversion for VC:"
-          << (for_typecast == val1.vc ? val1.vc_id : for_typecast == val2.vc ? val2.vc_id : val1.vc_id)
-          << system::unlock;
+      tianmu_control_.lock(current_txn_->GetThreadID()) << "Type conversion for VC:"
+                                                        << (for_typecast == val1.vc   ? val1.vc_id
+                                                            : for_typecast == val2.vc ? val2.vc_id
+                                                                                      : val1.vc_id)
+                                                        << system::unlock;
     }
 
     table->AddVirtColumn(tcc);