s

Author: Devjiu

omniscidb/DataMgr/BufferMgr/Buffer.cpp

@@ -33,6 +33,7 @@ Buffer::Buffer(BufferMgr* bm,
     , mem_(nullptr)
     , bm_(bm)
     , seg_it_(seg_it)
+    , is_seg_invalid(false)
     , page_size_(page_size)
     , num_pages_(0)
     , pin_count_(0)
@@ -53,6 +54,7 @@ Buffer::Buffer(BufferMgr* bm,
     , mem_(const_cast<int8_t*>(token->getMemoryPtr()))
     , bm_(bm)
     , seg_it_()
+    , is_seg_invalid(true)
     , page_size_(page_size)
     , num_pages_(0)
     , pin_count_(0)

omniscidb/DataMgr/BufferMgr/Buffer.h

@@ -138,7 +138,8 @@ class Buffer : public AbstractBuffer {
     std::unique_lock<std::mutex> pin_lock(pin_mutex_);
     int res = (--pin_count_);
     if (!res && delete_on_unpin_) {
-      pin_lock.unlock();
+      LOG(ERROR) << "unpin just delete pin = 0";
+      // pin_lock.unlock();
       delete this;
     }
     return res;
@@ -151,9 +152,11 @@ class Buffer : public AbstractBuffer {
   inline void deleteWhenUnpinned() override {
     std::unique_lock<std::mutex> pin_lock(pin_mutex_);
     if (pin_count_) {
+      LOG(ERROR) << "on unpin";
       delete_on_unpin_ = true;
     } else {
-      pin_lock.unlock();
+      LOG(ERROR) << "just delete pin = 0";
+      // pin_lock.unlock();
       delete this;
     }
   }
@@ -180,6 +183,7 @@ class Buffer : public AbstractBuffer {
 
   BufferMgr* bm_;
   BufferList::iterator seg_it_;
+  bool is_seg_invalid;
   size_t page_size_;  /// the size of each page in the buffer
   size_t num_pages_;
   int epoch_;  /// indicates when the buffer was last flushed

omniscidb/DataMgr/BufferMgr/BufferMgr.cpp

@@ -102,12 +102,10 @@ void BufferMgr::clear() {
   // pinned. Here we delete unpinned buffers and mark pinned buffers
   // for removal to have them deleted when unpinned.
   for (auto& buf : chunk_index_) {
-    // Iterator can be empty, check Buffer c-tors
-    if (buf.second == BufferList::iterator()) {
-      if (buf.second->buffer) {
-        buf.second->buffer->deleteWhenUnpinned();
-        buf.second->buffer = nullptr;
-      }
+    if (buf.second->buffer) {
+      buf.second->buffer->deleteWhenUnpinned();
+      LOG(ERROR) << "buffer ptr after delete: " << static_cast<void*>(buf.second->buffer);
+      buf.second->buffer = nullptr;
     }
   }
 
@@ -499,6 +497,8 @@ void BufferMgr::clearSlabs() {
     for (auto& segment : segment_list) {
       if (segment.mem_status == FREE) {
         // no need to free
+        //} else if (segment.buffer) {
+
       } else if (segment.buffer->getPinCount() < 1) {
         deleteBuffer(segment.chunk_key, true);
       } else {
@@ -835,8 +835,10 @@ void BufferMgr::free(AbstractBuffer* buffer) {
   if (casted_buffer == 0) {
     LOG(FATAL) << "Wrong buffer type - expects base class pointer to Buffer type.";
   }
-  if (casted_buffer->seg_it_ == BufferList::iterator()) {
+  if (casted_buffer->is_seg_invalid) {
+    LOG(ERROR) << "worked it";
     delete buffer;
+    return;
   }
 
   deleteBuffer(casted_buffer->seg_it_->chunk_key);

omniscidb/QueryEngine/ColumnFetcher.cpp

@@ -284,6 +284,17 @@ const int8_t* ColumnFetcher::getAllTableColumnFragments(
       total_row_count += rows_in_frag;
     }
 
+    if (total_row_count == 0) {
+      std::unique_ptr<ColumnarResults> merged_results(nullptr);
+
+      table_column = merged_results.get();
+      columnarized_scan_table_cache_.emplace(std::make_pair(table_id, col_id),
+                                             std::move(merged_results));
+
+      return ColumnFetcher::transferColumnIfNeeded(
+          table_column, 0, memory_level, device_id, device_allocator);
+    }
+
     const auto type_width = col_info->type->size();
     auto write_ptr =
         executor_->row_set_mem_owner_->allocate(type_width * total_row_count);
@@ -300,17 +311,7 @@ const int8_t* ColumnFetcher::getAllTableColumnFragments(
       valid_fragments.push_back(frag_id);
     }
 
-    if (write_ptrs.empty()) {
-      std::unique_ptr<ColumnarResults> merged_results(nullptr);
-
-      table_column = merged_results.get();
-      columnarized_scan_table_cache_.emplace(std::make_pair(table_id, col_id),
-                                             std::move(merged_results));
-
-      return ColumnFetcher::transferColumnIfNeeded(
-          table_column, 0, memory_level, device_id, device_allocator);
-    }
-
+    CHECK(!write_ptrs.empty());
     size_t valid_frag_count = valid_fragments.size();
     tbb::parallel_for(
         tbb::blocked_range<size_t>(0, valid_frag_count),

omniscidb/QueryEngine/JoinHashTable/Runtime/JoinColumnIterator.h

@@ -73,7 +73,6 @@ struct JoinColumnIterator {
   DEVICE FORCE_INLINE JoinColumnIterator& operator++() {
     index += step;
     index_inside_chunk += step;
-    // this loop is made to find index_of_chunk by total index of element
     while (chunk_data &&
            index_inside_chunk >= join_chunk_array[index_of_chunk].num_elems) {
       index_inside_chunk -= join_chunk_array[index_of_chunk].num_elems;

omniscidb/Tests/ArrowBasedExecuteTest.cpp

@@ -9268,15 +9268,6 @@ TEST_F(Select, Joins_Fixed_Size_Array_Multi_Frag) {
         v<int64_t>(run_simple_agg(
             "SELECT COUNT(1) FROM mf_t_arr r1, mf_t_arr r2 WHERE r1.t2[1] = r2.t2[1]",
             dt)));
-    // {
-    //   std::ostringstream oss_l;
-    //   oss_l << "SELECT * FROM  mf_d_arr t1, mf_d_arr t2 WHERE ";
-    //   auto part = oss_l.str();
-    //   auto q_{part + "t1.c2[1] = t2.c2[1];"};
-    //   const auto rows = run_multiple_agg(q_, dt);
-    //   LOG(ERROR) << rows->contentToString(true);
-    //   LOG(ERROR) << "should be done.";
-    // }
   }
 
   dropTable("mf_f_arr");

python/pyhdk/hdk.py

@@ -2899,9 +2899,6 @@ def if_then_else(self, cond, true_val, false_val):
         """
         return self._builder.if_then_else(cond, true_val, false_val)
 
-    def clear_cache(self):
-        self._executor.clearMemory(self._data_mgr, 1)
-
 
 def init(**kwargs):
     if init._instance is None: