s

Author: Devjiu

omniscidb/Calcite/CMakeLists.txt

@@ -35,7 +35,7 @@ endif()
 add_custom_command(
   OUTPUT
     ${CMAKE_BINARY_DIR}/bin/calcite-1.0-SNAPSHOT-jar-with-dependencies.jar
-    COMMAND ${MVN_PATH_COMMAND} ${MVN_EXECUTABLE} -l ${CMAKE_BINARY_DIR}/mvn_build.log -e clean install -Dmaven.compiler.showDeprecation=true -Dmaven.compiler.showWarnings=true -Domnisci.release.version="${OMNISCI_JAR_RELEASE_VERSION}" -Djava.net.preferIPv4Stack=true -Dmaven.wagon.http.retryHandler.count=3 -DMAPD_LOG_DIR="${CMAKE_BINARY_DIR}"
+    COMMAND ${MVN_PATH_COMMAND} ${MVN_EXECUTABLE} -e clean install -Dmaven.compiler.showDeprecation=true -Dmaven.compiler.showWarnings=true -Domnisci.release.version="${OMNISCI_JAR_RELEASE_VERSION}" -Djava.net.preferIPv4Stack=true -Dmaven.wagon.http.retryHandler.count=3 -DMAPD_LOG_DIR="${CMAKE_BINARY_DIR}"
     COMMAND ${MKDIR_COMMAND}
     COMMAND ${CMAKE_COMMAND} -E copy ${CMAKE_CURRENT_SOURCE_DIR}/java/calcite/target/calcite-1.0-SNAPSHOT-jar-with-dependencies.jar ${CMAKE_BINARY_DIR}/bin
     WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/java

omniscidb/DataMgr/Chunk/Chunk.cpp

@@ -34,6 +34,10 @@ std::shared_ptr<Chunk> Chunk::getChunk(ColumnInfoPtr col_info,
                                        const int deviceId,
                                        const size_t numBytes,
                                        const size_t numElems) {
+  std::ostringstream ss;
+  ss << __func__ << " getting chunk: " << col_info->toString();
+  LOG(WARNING) << ss.str();
+  auto timer = DEBUG_TIMER("ChunkNS getChunk");
   std::shared_ptr<Chunk> chunkp = std::make_shared<Chunk>(Chunk(col_info));
   chunkp->getChunkBuffer(data_mgr, key, memoryLevel, deviceId, numBytes, numElems);
   return chunkp;

omniscidb/DataMgr/PersistentStorageMgr/PersistentStorageMgr.cpp

@@ -46,6 +46,11 @@ void PersistentStorageMgr::deleteBuffersWithPrefix(const ChunkKey& chunk_key_pre
 
 AbstractBuffer* PersistentStorageMgr::getBuffer(const ChunkKey& chunk_key,
                                                 const size_t num_bytes) {
+  std::ostringstream ss;
+  ss << __func__ << " PersistentStrg: " << chunk_key.size() << "-" << chunk_key[0]
+     << " b: " << num_bytes;
+  LOG(WARNING) << ss.str();
+  auto timer = DEBUG_TIMER("PersistentStrg getBuffer");
   return getStorageMgrForTableKey(chunk_key)->getBuffer(chunk_key, num_bytes);
 }
 

omniscidb/QueryEngine/ColumnFetcher.cpp

@@ -173,6 +173,8 @@ const int8_t* ColumnFetcher::getOneTableColumnFragment(
     const Data_Namespace::MemoryLevel memory_level,
     const int device_id,
     DeviceAllocator* allocator) const {
+  auto timer = DEBUG_TIMER(__func__);
+  INJECT_TIMER(getOneTableColumnFragment);
   int db_id = col_info->db_id;
   int table_id = col_info->table_id;
   int col_id = col_info->column_id;
@@ -238,6 +240,8 @@ const int8_t* ColumnFetcher::getAllTableColumnFragments(
     const int device_id,
     DeviceAllocator* device_allocator,
     const size_t thread_idx) const {
+  auto timer = DEBUG_TIMER(__func__);
+  INJECT_TIMER(getAllTableColumnFragments);
   int db_id = col_info->db_id;
   int table_id = col_info->table_id;
   int col_id = col_info->column_id;
@@ -250,6 +254,7 @@ const int8_t* ColumnFetcher::getAllTableColumnFragments(
   const InputDescriptor table_desc(db_id, table_id, int(0));
   {
     std::lock_guard<std::mutex> columnar_conversion_guard(columnar_fetch_mutex_);
+    auto timer = DEBUG_TIMER("lock taken, execution started");
     auto column_it = columnarized_scan_table_cache_.find({table_id, col_id});
     if (column_it == columnarized_scan_table_cache_.end()) {
       for (size_t frag_id = 0; frag_id < frag_count; ++frag_id) {

omniscidb/QueryEngine/Execute.cpp

@@ -2329,6 +2329,7 @@ void Executor::executeWorkUnitPerFragment(
     for (auto fragment_index : fragment_indexes) {
       // We may want to consider in the future allowing this to execute on devices other
       // than CPU
+      LOG(ERROR) << "kernel create and run with fragment_idx: " << fragment_index;
       FragmentsList fragments_list{{db_id, table_id, {fragment_index}}};
       ExecutionKernel kernel(ra_exe_unit,
                              co.device_type,
@@ -2990,6 +2991,8 @@ Executor::getRowCountAndOffsetForAllFrags(
     const CartesianProduct<std::vector<std::vector<size_t>>>& frag_ids_crossjoin,
     const std::vector<InputDescriptor>& input_descs,
     const std::map<TableRef, const TableFragments*>& all_tables_fragments) {
+  auto timer = DEBUG_TIMER(__func__);
+  INJECT_TIMER(getRowCountAndOffsetForAllFrags);
   std::vector<std::vector<int64_t>> all_num_rows;
   std::vector<std::vector<uint64_t>> all_frag_offsets;
   const auto tab_id_to_frag_offsets =
@@ -3128,6 +3131,7 @@ FetchResult Executor::fetchChunks(
         memory_level_for_column = Data_Namespace::CPU_LEVEL;
       }
       if (needFetchAllFragments(*col_id, ra_exe_unit, selected_fragments)) {
+        VLOG(1) << "Need to fetch all frags.";
         // determine if we need special treatment to linearlize multi-frag table
         // i.e., a column that is classified as varlen type, i.e., array
         // for now, we can support more types in this way
@@ -3352,6 +3356,8 @@ void Executor::buildSelectedFragsMapping(
     const std::list<std::shared_ptr<const InputColDescriptor>>& col_global_ids,
     const FragmentsList& selected_fragments,
     const RelAlgExecutionUnit& ra_exe_unit) {
+  auto timer = DEBUG_TIMER(__func__);
+  INJECT_TIMER(buildSelectedFragsMapping);
   local_col_to_frag_pos.resize(plan_state_->global_to_local_col_ids_.size());
   size_t frag_pos{0};
   const auto& input_descs = ra_exe_unit.input_descs;

omniscidb/QueryEngine/RelAlgExecutor.cpp

@@ -264,6 +264,33 @@ ExecutionResult RelAlgExecutor::executeRelAlgQueryNoRetry(const CompilationOptio
     return registerResultSetTable({rs}, {}, true);
   }
 
+  std::stringstream ss;
+  std::vector<const hdk::ir::Node*> nodes;
+  for (size_t i = 0; i < query_seq.size(); i++) {
+    nodes.emplace_back(query_seq.step(i));
+  }
+  size_t ctr = nodes.size();
+  size_t tab_ctr = 0;
+  for (auto& body : boost::adaptors::reverse(nodes)) {
+    const auto index = ctr--;
+    const auto tabs = std::string(tab_ctr++, '\t');
+    CHECK(body);
+    ss << tabs << std::to_string(index) << " : " << body->toString() << "\n";
+    if (auto sort = dynamic_cast<const hdk::ir::Sort*>(body)) {
+      ss << tabs << "  : " << sort->getInput(0)->toString() << "\n";
+    }
+  }
+  const auto& subqueries = getSubqueries();
+  if (!subqueries.empty()) {
+    ss << "Subqueries: "
+       << "\n";
+    for (const auto& subquery : subqueries) {
+      const auto ra = subquery->node();
+      ss << "\t" << ra->toString() << "\n";
+    }
+  }
+  LOG(ERROR) << " dag?: " << ss.str();
+
   if (eo.find_push_down_candidates) {
     // this extra logic is mainly due to current limitations on multi-step queries
     // and/or subqueries.

omniscidb/ResultSetRegistry/ColumnarResults.cpp

@@ -20,6 +20,7 @@
 #include "Shared/Intervals.h"
 #include "Shared/funcannotations.h"
 #include "Shared/likely.h"
+#include "Shared/measure.h"
 #include "Shared/thread_count.h"
 
 #include <tbb/parallel_for.h>
@@ -135,6 +136,9 @@ ColumnarResults::ColumnarResults(std::shared_ptr<RowSetMemoryOwner> row_set_mem_
 std::unique_ptr<ColumnarResults> ColumnarResults::mergeResults(
     std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
     const std::vector<std::unique_ptr<ColumnarResults>>& sub_results) {
+  auto timer = DEBUG_TIMER(__func__);
+  INJECT_TIMER(mergeResults);
+
   if (sub_results.empty()) {
     return nullptr;
   }