loc

Author: Devjiu

omniscidb/QueryEngine/ColumnFetcher.cpp

@@ -266,23 +266,98 @@ const int8_t* ColumnFetcher::getAllTableColumnFragments(
         }
         auto chunk_meta_it = fragment.getChunkMetadataMap().find(col_id);
         CHECK(chunk_meta_it != fragment.getChunkMetadataMap().end());
-        auto col_buffer = getOneTableColumnFragment(col_info,
-                                                    static_cast<int>(frag_id),
-                                                    all_tables_fragments,
-                                                    chunk_holder,
-                                                    chunk_iter_holder,
-                                                    Data_Namespace::CPU_LEVEL,
-                                                    int(0),
-                                                    device_allocator);
+        // auto col_buffer = getOneTableColumnFragment(col_info,
+        //                                             static_cast<int>(frag_id),
+        //                                             all_tables_fragments,
+        //                                             chunk_holder,
+        //                                             chunk_iter_holder,
+        //                                             Data_Namespace::CPU_LEVEL,
+        //                                             int(0),
+        //                                             device_allocator);
+        // getOneTableColumnFragment(
+        //   const Data_Namespace::MemoryLevel memory_level,
+        //   const int device_id,
+        //   DeviceAllocator* allocator)
+        std::shared_ptr<Chunk_NS::Chunk> chunk;
+        // Fixed length arrays are also included here.
+        const bool is_varlen = col_info->type->isString() || col_info->type->isArray();
+        {
+          ChunkKey chunk_key{
+              db_id, fragment.physicalTableId, col_id, fragment.fragmentId};
+          std::unique_ptr<std::lock_guard<std::mutex>> varlen_chunk_lock;
+          if (is_varlen) {
+            varlen_chunk_lock.reset(
+                new std::lock_guard<std::mutex>(varlen_chunk_fetch_mutex_));
+          }
+          chunk = data_provider_->getChunk(
+              col_info,
+              chunk_key,
+              Data_Namespace::CPU_LEVEL,
+              0,
+              chunk_meta_it->second->numBytes(),
+              chunk_meta_it->second->numElements());
+          std::lock_guard<std::mutex> chunk_list_lock(chunk_list_mutex_);
+          chunk_holder.push_back(chunk);
+        }
+        if (is_varlen) {
+          CHECK_GT(table_id, 0);
+          CHECK(chunk_meta_it != fragment.getChunkMetadataMap().end());
+          chunk_iter_holder.push_back(chunk->begin_iterator(chunk_meta_it->second));
+          auto& chunk_iter = chunk_iter_holder.back();
+          return reinterpret_cast<int8_t*>(&chunk_iter);
+        } else {
+          auto ab = chunk->getBuffer();
+          CHECK(ab->getMemoryPtr());
+          return ab->getMemoryPtr();  // @TODO(alex) change to use ChunkIter
+        }
         column_frags.push_back(
             std::make_unique<ColumnarResults>(executor_->row_set_mem_owner_,
                                               col_buffer,
                                               fragment.getNumTuples(),
                                               chunk_meta_it->second->type(),
                                               thread_idx));
       }
-      auto merged_results =
-          ColumnarResults::mergeResults(executor_->row_set_mem_owner_, column_frags);
+      // auto merged_results =
+      //     ColumnarResults::mergeResults(executor_->row_set_mem_owner_, column_frags);
+      // std::unique_ptr<ColumnarResults> ColumnarResults::mergeResults(
+      //   std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
+      //   const std::vector<std::unique_ptr<ColumnarResults>>& sub_results) {
+      if (column_frags.empty()) {
+        return nullptr;
+      }
+      const auto total_row_count = std::accumulate(
+          column_frags.begin(),
+          column_frags.end(),
+          size_t(0),
+          [](const size_t init, const std::unique_ptr<ColumnarResults>& result) {
+            return init + result->size();
+          });
+      std::unique_ptr<ColumnarResults> merged_results(
+          new ColumnarResults(total_row_count, column_frags[0]->target_types_));
+      const auto col_count = column_frags[0]->column_buffers_.size();
+      const auto nonempty_it = std::find_if(
+          column_frags.begin(),
+          column_frags.end(),
+          [](const std::unique_ptr<ColumnarResults>& needle) { return needle->size(); });
+      if (nonempty_it == column_frags.end()) {
+        return nullptr;
+      }
+      for (size_t col_idx = 0; col_idx < col_count; ++col_idx) {
+        const auto byte_width = (*nonempty_it)->columnType(col_idx)->size();
+        auto write_ptr = row_set_mem_owner->allocate(byte_width * total_row_count);
+        merged_results->column_buffers_.push_back(write_ptr);
+        for (auto& rs : column_frags) {
+          CHECK_EQ(col_count, rs->column_buffers_.size());
+          if (!rs->size()) {
+            continue;
+          }
+          CHECK_EQ(byte_width, rs->columnType(col_idx)->size());
+          memcpy(write_ptr, rs->column_buffers_[col_idx], rs->size() * byte_width);
+          write_ptr += rs->size() * byte_width;
+        }
+      }
+      return merged_results;
+
       table_column = merged_results.get();
       columnarized_scan_table_cache_.emplace(std::make_pair(table_id, col_id),
                                              std::move(merged_results));