Parallel transform (#11)

Parallel transform. 
* Added timegm.h from FreeBSD sources, enabled it for the benchmark.

Author: anton-malakhov

benchmark.cpp

@@ -7,6 +7,7 @@
 #include <print.h>
 #include <transform.h>
 #include <sort.h>
+#include <timegm.h>
 
 #include <cmath>
 
@@ -17,7 +18,8 @@
 //++++++++++++++++++++++++++++++
 // BENCHMARK
 //++++++++++++++++++++++++++++++
-std::shared_ptr<arrow::Table> taxi1(std::shared_ptr<arrow::Table> table) {
+std::shared_ptr<arrow::Table>
+taxi1(std::shared_ptr<arrow::Table> table) {
 	//SELECT cab_type, count(cab_type)
 	//FROM trips
 	//GROUP BY cab_type;
@@ -27,7 +29,8 @@ std::shared_ptr<arrow::Table> taxi1(std::shared_ptr<arrow::Table> table) {
 	return aggregate(table, taxi1_group_by, {&taxi1_task});
 }
 
-std::shared_ptr<arrow::Table> taxi2(std::shared_ptr<arrow::Table> table) {
+std::shared_ptr<arrow::Table>
+taxi2(std::shared_ptr<arrow::Table> table) {
 	//SELECT passenger_count, avg(total_amount)
 	//FROM trips
 	//GROUP BY passenger_count;
@@ -37,23 +40,27 @@ std::shared_ptr<arrow::Table> taxi2(std::shared_ptr<arrow::Table> table) {
 	return aggregate(table, taxi2_group_by, {&taxi2_task});
 }
 
-std::shared_ptr<arrow::Table> taxi3(std::shared_ptr<arrow::Table> table) {
+std::shared_ptr<arrow::Table>
+taxi3(std::shared_ptr<arrow::Table> table) {
 	//SELECT passenger_count,
 	//	EXTRACT(year from pickup_datetime) as year,
 	//	count(*)
 	//FROM trips
 	//GROUP BY passenger_count,
 	//	year;
 	printf("NAME: Taxi number 3\n");
-	auto year = [](int64_t time) { time_t tt = static_cast<time_t>(time); return int64_t(gmtime(&tt)->tm_year + 1900); }; // gmtime (not localtime) because of python
+	auto year = [](int64_t time) {
+	  time_t tt = static_cast<time_t>(time); struct tm r;
+	  return int64_t(_der_gmtime(tt, &r)->tm_year + 1900); }; // gmtime (not localtime) because of python
 	auto taxi3_table = transform<int64_t, int64_t, arrow::TimestampArray, arrow::Int64Builder>(table, 2, year);
 	group *taxi3_group_by = group_by(taxi3_table, {2, 10});
 	aggregate_task taxi3_task = {count, 0};
 	auto a = aggregate(taxi3_table, taxi3_group_by, {&taxi3_task});
   return sort(a, {0, 1}, {asc, asc}, flat); // Only one chunk for sort here, not a good checking - see sortAll
 }
 
-std::shared_ptr<arrow::Table> taxi4(std::shared_ptr<arrow::Table> table) {
+std::shared_ptr<arrow::Table>
+taxi4(std::shared_ptr<arrow::Table> table) {
 	//SELECT passenger_count,
 	//	EXTRACT(year from pickup_datetime) as year,
 	//	round(trip_distance) distance,
@@ -65,7 +72,9 @@ std::shared_ptr<arrow::Table> taxi4(std::shared_ptr<arrow::Table> table) {
 	//ORDER BY year,
 	//	trips desc;
 	printf("NAME: Taxi number 4\n");
-	auto year = [](int64_t time) { time_t tt = static_cast<time_t>(time); return int64_t(gmtime(&tt)->tm_year + 1900); }; // gmtime (not localtime) because of python
+  auto year = [](int64_t time) {
+    time_t tt = static_cast<time_t>(time); struct tm r;
+    return int64_t(_der_gmtime(tt, &r)->tm_year + 1900); }; // gmtime (not localtime) because of python
 	auto taxi4_table = transform<int64_t, int64_t, arrow::TimestampArray, arrow::Int64Builder>(table, 2, year);
 	auto taxi4_table1 = transform<double, double, arrow::DoubleArray, arrow::DoubleBuilder>(taxi4_table, 11, round);
 	group *taxi4_group_by = group_by(taxi4_table1, {2, 10, 11});

group_by.h

@@ -68,7 +68,7 @@ struct position {
 #if 1 //USE_TBB
 struct mult_group_map_t : public tbb::concurrent_hash_map<position, int> {
   using tbb::concurrent_hash_map<position, int>::concurrent_hash_map;
-#if 0
+#if TBB_INTERFACE_VERSION < 11007
   const_pointer fast_find(const position& k) {
     return internal_fast_find(k);
   }

timegm.h

@@ -0,0 +1,136 @@
+/*
+ * Copyright (c) 1997 Kungliga Tekniska Högskolan
+ * (Royal Institute of Technology, Stockholm, Sweden).
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * 3. Neither the name of the Institute nor the names of its contributors
+ *    may be used to endorse or promote products derived from this software
+ *    without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE INSTITUTE AND CONTRIBUTORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED.  IN NO EVENT SHALL THE INSTITUTE OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+
+#include <time.h>
+
+#define ASN1_MAX_YEAR 2000
+
+static int
+is_leap(unsigned y)
+{
+    y += 1900;
+    return (y % 4) == 0 && ((y % 100) != 0 || (y % 400) == 0);
+}
+
+static const unsigned ndays[2][12] ={
+    {31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31},
+    {31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31}};
+
+/*
+ * This is a simplifed version of timegm(3) that doesn't accept out of
+ * bound values that timegm(3) normally accepts but those are not
+ * valid in asn1 encodings.
+ */
+
+time_t
+_der_timegm (struct tm *tm)
+{
+  time_t res = 0;
+  int i;
+
+  /*
+   * See comment in _der_gmtime
+   */
+  if (tm->tm_year > ASN1_MAX_YEAR)
+      return 0;
+
+  if (tm->tm_year < 0)
+      return -1;
+  if (tm->tm_mon < 0 || tm->tm_mon > 11)
+      return -1;
+  if (tm->tm_mday < 1 || tm->tm_mday > (int)ndays[is_leap(tm->tm_year)][tm->tm_mon])
+      return -1;
+  if (tm->tm_hour < 0 || tm->tm_hour > 23)
+      return -1;
+  if (tm->tm_min < 0 || tm->tm_min > 59)
+      return -1;
+  if (tm->tm_sec < 0 || tm->tm_sec > 59)
+      return -1;
+
+  for (i = 70; i < tm->tm_year; ++i)
+    res += is_leap(i) ? 366 : 365;
+
+  for (i = 0; i < tm->tm_mon; ++i)
+    res += ndays[is_leap(tm->tm_year)][i];
+  res += tm->tm_mday - 1;
+  res *= 24;
+  res += tm->tm_hour;
+  res *= 60;
+  res += tm->tm_min;
+  res *= 60;
+  res += tm->tm_sec;
+  return res;
+}
+
+struct tm *
+_der_gmtime(time_t t, struct tm *tm)
+{
+    time_t secday = t % (3600 * 24);
+    time_t days = t / (3600 * 24);
+
+    memset(tm, 0, sizeof(*tm));
+
+    tm->tm_sec = secday % 60;
+    tm->tm_min = (secday % 3600) / 60;
+    tm->tm_hour = secday / 3600;
+
+    /*
+     * Refuse to calculate time ~ 2000 years into the future, this is
+     * not possible for systems where time_t is a int32_t, however,
+     * when time_t is a int64_t, that can happen, and this becomes a
+     * denial of sevice.
+     */
+    if (days > (ASN1_MAX_YEAR * 365))
+  return NULL;
+
+    tm->tm_year = 70;
+    while(1) {
+  unsigned dayinyear = (is_leap(tm->tm_year) ? 366 : 365);
+  if (days < dayinyear)
+      break;
+  tm->tm_year += 1;
+  days -= dayinyear;
+    }
+    tm->tm_mon = 0;
+
+    while (1) {
+  unsigned daysinmonth = ndays[is_leap(tm->tm_year)][tm->tm_mon];
+  if (days < daysinmonth)
+      break;
+  days -= daysinmonth;
+  tm->tm_mon++;
+    }
+    tm->tm_mday = days + 1;
+
+    return tm;
+}

transform.h

@@ -2,7 +2,7 @@
 #define TRANSFORM_H
 
 #include <arrow/api.h>
-
+#include <tbb/tbb.h>
 #include <print.h>
 
 //++++++++++++++++++++++++++++++
@@ -12,35 +12,36 @@
 // One column tranformation per time. Should aggregation be the same? No vector of tasks?
 template <typename T_c, typename T_n, typename T2, typename T4>
 std::shared_ptr<arrow::Table> transform(std::shared_ptr<arrow::Table> table, int column_id, T_n (*transformation)(T_c)) {
-	printf("TASK: transforming of column\n");
-	auto begin = std::chrono::steady_clock::now();
-
-	auto column = table->column(column_id)->data();
-	arrow::ArrayVector new_chunks(column->num_chunks());
-	std::shared_ptr<arrow::Array> data;
-	for (int i = 0; i < column->num_chunks(); i++) {
-		T4 bld; // template
-		auto c = std::dynamic_pointer_cast<T2>(column->chunk(i)); // template
-		if (c == NULL) {
-			printf("Type of %d column is wrong!\n", column_id + 1);
-		}
-		// TBB in parallel for all available chunks or sequential for each incoming chunk
-		for (int j = 0; j < c->length(); j++) {
-			auto value = c->Value(j);
-			T_n new_value = transformation(value);
-			bld.Append(new_value); // resize?
-		}
-		bld.Finish(&data);
-		new_chunks[i] = data;
-	}
-
-	std::shared_ptr<arrow::Field> new_field = std::make_shared<arrow::Field>("name", data->type());
-	std::shared_ptr<arrow::Column> new_column = std::make_shared<arrow::Column>(new_field, new_chunks);
-	std::shared_ptr<arrow::Table> new_table;
-	table->SetColumn(column_id, new_column, &new_table);
-
-	print_time(begin);
-	return new_table;
+    printf("TASK: transforming of column\n");
+    auto begin = std::chrono::steady_clock::now();
+    const auto column = table->column(column_id)->data();
+    arrow::ArrayVector new_chunks(column->num_chunks());
+    printf("Parallel: %d, chunk[0]->length: %d\n", column->num_chunks(), column->chunk(0)->length());
+    tbb::parallel_for(0, column->num_chunks(), [&new_chunks, &column, column_id, transformation](int i){
+    //for(int i = column->num_chunks()-1; i >= 0; i--){
+        auto c = std::dynamic_pointer_cast<T2>(column->chunk(i)); // template
+        if (c == NULL) {
+            printf("Type of %d column is wrong!\n", column_id + 1); abort();
+        }
+        T4 bld; // template
+        bld.Resize(c->length());
+        for (int j = 0; j < c->length(); j++) {
+            auto value = c->Value(j);
+            auto new_value = transformation(value);
+            bld.Append(new_value); // resize?
+        }
+        std::shared_ptr<arrow::Array> data;
+        bld.Finish(&data);
+        new_chunks[i] = data;
+    });
+
+    std::shared_ptr<arrow::Field> new_field = std::make_shared<arrow::Field>("name", new_chunks[0]->type());
+    std::shared_ptr<arrow::Column> new_column = std::make_shared<arrow::Column>(new_field, new_chunks);
+    std::shared_ptr<arrow::Table> new_table;
+    table->SetColumn(column_id, new_column, &new_table);
+
+    print_time(begin);
+    return new_table;
 }
 
 #endif