Address review

Author: pmenon

src/codegen/operator/order_by_translator.cpp

@@ -303,13 +303,6 @@ void OrderByTranslator::Produce() const {
     CodeGen &codegen = GetCodeGen();
     auto *sorter_ptr = LoadStatePtr(sorter_id_);
 
-#if 0
-    // The tuples have been materialized into the buffer space, NOW SORT!!!
-    if (!child_pipeline_.IsParallel()) {
-      sorter_.Sort(codegen, sorter_ptr);
-    }
-#endif
-
     // Now iterate over the sorted list
     auto *i32_type = codegen.Int32Type();
     auto vec_size = Vector::kDefaultVectorSize.load();

src/codegen/operator/table_scan_translator.cpp

@@ -147,8 +147,10 @@ TableScanTranslator::TableScanTranslator(const planner::SeqScanPlan &scan,
   }
 }
 
-// TODO merge serial and parallel ...
-// TODO cleanup zonemap stuff
+// TODO merge serial and parallel since there is a lot of duplication
+// TODO cleanup zonemap stuff - zonemaps hardcode memory pointers which may not
+// live as long as the compiled query. Hence, it may point to invalid memory
+// at some invocation of the query.
 
 llvm::Value *TableScanTranslator::LoadTablePtr(CodeGen &codegen) const {
   const storage::DataTable &table = *GetScanPlan().GetTable();
@@ -202,7 +204,7 @@ void TableScanTranslator::ProduceParallel() const {
   // The input arguments
   const storage::DataTable &table = *GetScanPlan().GetTable();
 
-  // The use RuntimeFunctions::ExecuteTableScan() to launch a parallel scan.
+  // We use RuntimeFunctions::ExecuteTableScan() to launch a parallel scan.
   // We pass in the database and table IDs to scan the correct table.
   auto *dispatcher =
       RuntimeFunctionsProxy::ExecuteTableScan.GetFunction(codegen);

src/codegen/runtime_functions.cpp

@@ -169,11 +169,11 @@ void RuntimeFunctions::ExecuteTableScan(
   thread_states.Allocate(num_tasks);
 
   // Create count down latch
-  common::synchronization::CountDownLatch latch(num_tasks);
+  common::synchronization::CountDownLatch latch{num_tasks};
 
   // Now, submit the tasks
   for (uint32_t task_id = 0; task_id < num_tasks; task_id++) {
-    bool last_task = task_id == num_tasks - 1;
+    bool last_task = (task_id == num_tasks - 1);
     auto tilegroup_start = task_id * num_tilegroups_per_task;
     auto tilegroup_stop =
         last_task ? num_tilegroups : tilegroup_start + num_tilegroups_per_task;
@@ -216,7 +216,7 @@ void RuntimeFunctions::ExecutePerState(
 
   // Create count down latch
   uint32_t num_tasks = thread_states.NumThreads();
-  common::synchronization::CountDownLatch latch(num_tasks);
+  common::synchronization::CountDownLatch latch{num_tasks};
 
   // Loop over states
   for (uint32_t tid = 0; tid < num_tasks; tid++) {

src/codegen/util/sorter.cpp

@@ -115,25 +115,31 @@ void Sorter::SortParallel(
     sorters.push_back(sorter);
   }
 
-  // Sort each run in parallel
+  // This is a 2D matrix that stores, for each sorter, a list of separators. A
+  // separator separates a sorted run into equal, evenly spaces peices.
+  // separators[i][j] lists the j-th separator key in the i-th separator.
+  // We use these separator from each sorter to evenly split the inputs across
+  // all worker threads by taking a median-of-medians.
   std::vector<std::vector<char *>> separators(sorters.size());
   for (uint32_t i = 0; i < separators.size(); i++) {
     separators[i].resize(sorters.size());
   }
 
-  // Time
+  // Time it all
   Timer<std::milli> timer;
   timer.Start();
 
-  common::synchronization::CountDownLatch sort_latch(sorters.size());
+  common::synchronization::CountDownLatch sort_latch{sorters.size()};
+
+  // Sort each run in parallel
   auto &work_pool = threadpool::MonoQueuePool::GetExecutionInstance();
   for (uint32_t sort_idx = 0; sort_idx < sorters.size(); sort_idx++) {
     work_pool.SubmitTask([&sorters, &separators, &sort_latch, sort_idx]() {
       // First sort
       auto *sorter = sorters[sort_idx];
       sorter->Sort();
 
-      // Now compute local separators
+      // Now compute local separators that "evenly" divide the input
       auto part_size = sorter->NumTuples() / sorters.size();
       for (uint32_t idx = 0; idx < separators.size() - 1; idx++) {
         separators[idx][sort_idx] = sorter->tuples_[(idx + 1) * part_size];
@@ -201,7 +207,9 @@ void Sorter::SortParallel(
   timer.Reset();
   timer.Start();
 
-  common::synchronization::CountDownLatch merge_latch(merge_work.size());
+  common::synchronization::CountDownLatch merge_latch{merge_work.size()};
+
+  // Divide up the merging work across all threads
   auto heap_cmp = [this](const std::pair<char **, char **> &l,
                          const std::pair<char **, char **> &r) {
     return !(cmp_func_(*l.first, *r.first) < 0);

src/include/common/synchronization/count_down_latch.h

@@ -28,7 +28,7 @@ namespace synchronization {
 class CountDownLatch {
  public:
   /// Constructor
-  explicit CountDownLatch(uint32_t count);
+  explicit CountDownLatch(uint64_t count);
 
   /// This class cannot be copy or move-constructed
   DISALLOW_COPY_AND_MOVE(CountDownLatch);
@@ -43,7 +43,7 @@ class CountDownLatch {
   uint32_t GetCount();
 
  private:
-  uint32_t count_;
+  uint64_t count_;
   std::mutex mutex_;
   std::condition_variable cv_;
 };
@@ -54,7 +54,7 @@ class CountDownLatch {
 ///
 ////////////////////////////////////////////////////////////////////////////////
 
-inline CountDownLatch::CountDownLatch(uint32_t count) : count_(count) {}
+inline CountDownLatch::CountDownLatch(uint64_t count) : count_(count) {}
 
 inline bool CountDownLatch::Await(uint64_t nanos) {
   std::unique_lock<std::mutex> lock(mutex_);