Dist sort cpu

.github/workflows/conda-actions.yml

@@ -35,7 +35,7 @@ jobs:
     # Specify the correct host compilers
     - name: Install/Select gcc and g++
       run: |
-        sudo apt-get install -y gcc-${{ matrix.gcc }} g++-${{ matrix.gcc }}
+        sudo apt-get install -y gcc-${{ matrix.gcc }} g++-${{ matrix.gcc }} libopenmpi-dev
         echo "CC=/usr/bin/gcc-${{ matrix.gcc }}" >> $GITHUB_ENV
         echo "CXX=/usr/bin/g++-${{ matrix.gcc }}" >> $GITHUB_ENV
         echo "CUDAHOSTCXX=/usr/bin/g++-${{ matrix.gcc }}" >> $GITHUB_ENV

conda/environments/cylon.yml

@@ -7,7 +7,7 @@ dependencies:
   - cmake>=3.17
   - pyarrow=5.0.0
   - glog=0.5.0
-  - openmpi>=4.1
+  - openmpi>=4.1.3
   - ucx>=1.12.1
   - cython>=0.29,<0.30
   - numpy>=1.16

conda/environments/cylon_MacOS.yml

@@ -7,7 +7,7 @@ dependencies:
   - cmake>=3.17
   - pyarrow=5.0.0
   - glog=0.5.0
-  - openmpi>=4.1
+  - openmpi>=4.1.2
   - cython>=0.29,<0.30
   - numpy>=1.16
   - pandas>=1.0

conda/environments/gcylon.yml

@@ -11,8 +11,8 @@ dependencies:
   - cudf=21.10.01
   - cudatoolkit=11.2
   - glog=0.5.0
-  - openmpi>=4.1
-  - ucx>=1.12.0
+  - openmpi>=4.1.2
+  - ucx>=1.12.1
   - numpy>=1.16
   - pandas>=1.0
   - setuptools>=40.0,<60.0

cpp/src/cylon/arrow/arrow_comparator.cpp

@@ -759,10 +759,13 @@ Status CreateDualArrayIndexComparator(const std::shared_ptr<arrow::Array> &a1,
 
 Status TableRowIndexEqualTo::Make(const std::shared_ptr<arrow::Table> &table,
                                   const std::vector<int> &col_ids,
-                                  std::unique_ptr<TableRowIndexEqualTo> *out_equal_to) {
+                                  std::unique_ptr<TableRowIndexEqualTo> *out_equal_to,
+                                  const std::vector<bool> &sort_order) {
   auto comps = std::make_shared<std::vector<std::shared_ptr<ArrayIndexComparator>>>();
   comps->reserve(col_ids.size());
-  for (int col_id: col_ids) {
+  bool order_not_given = sort_order.size() == 0;
+  for (std::size_t i = 0; i < col_ids.size(); i++) {
+    int col_id = col_ids[i];
     if (table->column(col_id)->num_chunks() > 1) {
       return {Code::Invalid, "TableRowIndexEqualTo does not support multiple chunks"};
     }
@@ -772,7 +775,7 @@ Status TableRowIndexEqualTo::Make(const std::shared_ptr<arrow::Table> &table,
     } else {
       const auto &array = table->column(col_id)->chunk(0);
       std::unique_ptr<ArrayIndexComparator> comp;
-      RETURN_CYLON_STATUS_IF_FAILED(CreateArrayIndexComparator(array, &comp));
+      RETURN_CYLON_STATUS_IF_FAILED(CreateArrayIndexComparator(array, &comp, order_not_given || sort_order[i]));
       comps->emplace_back(std::move(comp));
     }
   }
@@ -920,7 +923,8 @@ Status DualTableRowIndexEqualTo::Make(const std::shared_ptr<arrow::Table> &t1,
                                       const std::shared_ptr<arrow::Table> &t2,
                                       const std::vector<int> &t1_indices,
                                       const std::vector<int> &t2_indices,
-                                      std::unique_ptr<DualTableRowIndexEqualTo> *out_equal_to) {
+                                      std::unique_ptr<DualTableRowIndexEqualTo> *out_equal_to,
+                                      const std::vector<bool> &sort_order) {
   int num_cols = (int) t1_indices.size();
   if (num_cols != (int) t2_indices.size()) {
     return {Code::Invalid, "sizes of indices of t1 and t2 are not equal!"};
@@ -929,6 +933,8 @@ Status DualTableRowIndexEqualTo::Make(const std::shared_ptr<arrow::Table> &t1,
   auto comps = std::make_shared<std::vector<std::shared_ptr<DualArrayIndexComparator>>>();
   comps->reserve(num_cols);
 
+  bool order_not_given = sort_order.size() == 0;
+
   for (int i = 0; i < num_cols; i++) {
     if (t1->column(t1_indices[i])->num_chunks() > 1
         || t2->column(t2_indices[i])->num_chunks() > 1) {
@@ -939,7 +945,9 @@ Status DualTableRowIndexEqualTo::Make(const std::shared_ptr<arrow::Table> &t1,
     const auto &a2 = util::GetChunkOrEmptyArray(t2->column(t2_indices[i]), 0);
 
     std::unique_ptr<DualArrayIndexComparator> comp;
-    RETURN_CYLON_STATUS_IF_FAILED(CreateDualArrayIndexComparator(a1, a2, &comp));
+
+    // asc=true if sort_order is not provided
+    RETURN_CYLON_STATUS_IF_FAILED(CreateDualArrayIndexComparator(a1, a2, &comp, order_not_given || sort_order[i]));
     comps->emplace_back(std::move(comp));
   }
 

cpp/src/cylon/arrow/arrow_comparator.hpp

@@ -173,8 +173,11 @@ class TableRowIndexEqualTo {
   // equality, less than, greater than
   int compare(const int64_t &record1, const int64_t &record2) const;
 
-  static Status Make(const std::shared_ptr<arrow::Table> &table, const std::vector<int> &col_ids,
-                     std::unique_ptr<TableRowIndexEqualTo> *out_equal_to);
+  static Status Make(const std::shared_ptr<arrow::Table> &table,
+                     const std::vector<int> &col_ids,
+                     std::unique_ptr<TableRowIndexEqualTo> *out_equal_to,
+                     const std::vector<bool> &sort_order = {});
+
   static Status Make(const std::vector<std::shared_ptr<arrow::Array>> &arrays,
                      std::unique_ptr<TableRowIndexEqualTo> *out_equal_to);
 
@@ -278,7 +281,8 @@ class DualTableRowIndexEqualTo {
                      const std::shared_ptr<arrow::Table> &t2,
                      const std::vector<int> &t1_indices,
                      const std::vector<int> &t2_indices,
-                     std::unique_ptr<DualTableRowIndexEqualTo> *out_equal_to);
+                     std::unique_ptr<DualTableRowIndexEqualTo> *out_equal_to,
+                     const std::vector<bool> &sort_order = {});
 
   bool operator()(const int64_t &record1, const int64_t &record2) const;