Adding distributed scalar aggregates #570

nirandaperera · 2022-02-19T14:09:17Z

No description provided.

� Conflicts: � cpp/src/cylon/table.cpp

vibhatha

There are minor changes that I saw, nothing serious. Thank you for the PR.

vibhatha · 2022-02-27T16:33:16Z

cpp/src/cylon/column.hpp


-  static std::shared_ptr<Column> Make(const std::string &id, const std::shared_ptr<DataType> &type,
-                                      const std::shared_ptr<arrow::Array> &data_);
+  static std::shared_ptr<Column> Make(std::shared_ptr<arrow::Array> data_);


why is data_ instead of data

vibhatha · 2022-02-27T16:36:37Z

cpp/src/cylon/compute/aggregates.hpp

@@ -52,7 +50,8 @@ class Result {
 * Function pointer for aggregate functions
 */
 typedef Status
-(*AggregateOperation)(const std::shared_ptr<cylon::Table> &table, int32_t col_idx, std::shared_ptr<Result> &output);
+(*AggregateOperation)


It will be better if we can create an alias and use it when the function pointer is used.
Another thought will it be possible to use std::function<...>?

I'm thinking of deprecating these all together. I think the better approach would be to use templates with a lambda function. I'll change to that.

vibhatha · 2022-02-27T16:38:05Z

cpp/src/cylon/compute/aggregates.hpp

@@ -72,7 +73,9 @@ cylon::Status Sum(const std::shared_ptr<cylon::Table> &table, int32_t col_idx, s
 * @param output
 * @return
 */
-cylon::Status Count(const std::shared_ptr<cylon::Table> &table, int32_t col_idx, std::shared_ptr<Result> &output);
+cylon::Status Count(const std::shared_ptr<cylon::Table> &table,


Why are we using &output is it possible to use *output instead?
ALso, instead of returning cylon::Status, we can return a result object like Arrow is doing.
It could simplify things. Just a thought.

Yeah I think we should convert all outputs to pointers. I created an issue here.
#486

vibhatha · 2022-02-27T16:43:17Z

cpp/src/cylon/compute/scalar_aggregate.cpp

+
+  net::ReduceOp reduce_op() const override { return net::SUM; }
+
+  int32_t num_intermediate_results() const override { return 3; }


any specific reason why a number is returned?
may be we can add a constexpr and comment how this number is decided?

this is a vitual method. I think virtual constexpr are only supported from c++20 on wards,
https://stackoverflow.com/questions/34828161/can-virtual-functions-be-constexpr

The idea of this method is to get the number of intermediate elements produced by the local aggregation. Ex: std produces, count, sum, and sum of squares.

vibhatha · 2022-03-03T00:52:38Z

cpp/src/cylon/table.hpp

@@ -69,6 +69,7 @@ class Table {
   */
  static Status FromColumns(const std::shared_ptr<CylonContext> &ctx,
                            const std::vector<std::shared_ptr<Column>> &columns,
+                            const std::vector<std::string> &column_names,
                            std::shared_ptr<Table> &tableOut);


What if we use std::shared_ptr<Table> *tableOut instead?

vibhatha · 2022-03-03T01:26:43Z

python/pycylon/pycylon/data/aggregates.pyx

+from pycylon.data.aggregates cimport *
+from pycylon.common.status cimport CStatus
+from pycylon.api.lib cimport pycylon_unwrap_context, pycylon_unwrap_table, pycylon_wrap_column
+
+from pycylon.data.table import Table
+from pycylon.ctx.context import CylonContext
+from pycylon.data.column import Column


nit: we should probably re-order the imports. But I am not sure if we checked this for other PRs.

vibhatha · 2022-03-03T01:28:23Z

python/pycylon/pycylon/data/aggregates.pyx

@@ -36,3 +42,86 @@ AggregationOpString = {
    'quantile': CGroupByAggregationOp.CQUANTILE,
    'std': CGroupByAggregationOp.CSTDDEV,


nit:
STD instead std
QUANTILE instead quantile

vibhatha · 2022-03-03T01:28:51Z

python/pycylon/pycylon/data/aggregates.pyx

@@ -36,3 +42,86 @@ AggregationOpString = {
    'quantile': CGroupByAggregationOp.CQUANTILE,
    'std': CGroupByAggregationOp.CSTDDEV,
 }
+
+def sum_table(ctx: CylonContext, table: Table, skipna=True) -> Column:


skipna: bool = True

vibhatha · 2022-03-03T01:32:48Z

python/pycylon/pycylon/data/aggregates.pyx

+    cdef shared_ptr[CCylonContext] cctx = pycylon_unwrap_context(ctx)
+    cdef shared_ptr[CTable] ctable = pycylon_unwrap_table(table)
+    cdef CBasicOptions options = CBasicOptions(skipna)
+    cdef shared_ptr[CColumn] cresult


nit: this seems like a separate cdef function which can be called within each of these aggregate functions.
WDYT?

unless we use some sort of an enum, I'm not sure how we can achieve this.

vibhatha · 2022-03-03T01:36:07Z

python/pycylon/pycylon/data/column.pyx


 cdef class Column:
+    def __cinit__(self, array: pa.Array = None):


we could probably take the input as a list, a numpy array or an Arrow array. All 3 should be compatible and we can handle it inside the constructor.

nirandaperera added 11 commits February 15, 2022 13:15

init

6c8221f

incomplete

0ab5066

changing column API

a7eea23

Merge branch 'main' of https://github.com/cylondata/cylon into new_agg

40dae9d

� Conflicts: � cpp/src/cylon/table.cpp

refactoring column

c94184e

adding allreduce column

e8b006a

adding min, max, count

a0a8b78

adding mean

d5c6718

removing ctx from communicator methods

2c1ef14

adding variance and stddev

819d8a6

minor refactor

3fe8d67

nirandaperera marked this pull request as ready for review February 20, 2022 17:03

nirandaperera requested a review from supunkamburugamuve February 20, 2022 17:04

nirandaperera added 4 commits February 21, 2022 19:51

minor fix

27d0e7b

adding table aggregations

7af1e97

more tests

8da9a07

adding cython bindings and tests

1eafcae

nirandaperera requested a review from vibhatha March 1, 2022 16:59

vibhatha approved these changes Mar 3, 2022

View reviewed changes

nirandaperera added 2 commits March 4, 2022 11:51

Update aggregate_test.cpp

af7be3c

incorporating suggestions from the review

cc669d5

nirandaperera merged commit b2c0820 into cylondata:main Mar 4, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding distributed scalar aggregates #570

Adding distributed scalar aggregates #570

nirandaperera commented Feb 19, 2022

vibhatha left a comment

vibhatha Feb 27, 2022

vibhatha Feb 27, 2022

nirandaperera Mar 4, 2022

vibhatha Feb 27, 2022

nirandaperera Mar 4, 2022

vibhatha Feb 27, 2022

nirandaperera Mar 3, 2022 •

edited

Loading

vibhatha Mar 3, 2022

vibhatha Mar 3, 2022

vibhatha Mar 3, 2022

vibhatha Mar 3, 2022

vibhatha Mar 3, 2022

nirandaperera Mar 4, 2022

vibhatha Mar 3, 2022


		net::ReduceOp reduce_op() const override { return net::SUM; }

		int32_t num_intermediate_results() const override { return 3; }

		@@ -36,3 +42,86 @@ AggregationOpString = {
		'quantile': CGroupByAggregationOp.CQUANTILE,
		'std': CGroupByAggregationOp.CSTDDEV,


		cdef class Column:
		def __cinit__(self, array: pa.Array = None):

Adding distributed scalar aggregates #570

Adding distributed scalar aggregates #570

Conversation

nirandaperera commented Feb 19, 2022

vibhatha left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nirandaperera Mar 3, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nirandaperera Mar 3, 2022 •

edited

Loading