OmniSciDB  c1a53651b2
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
TableFunctionsCommon.hpp File Reference
#include <filesystem>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <tuple>
#include <unordered_map>
#include <vector>
#include "QueryEngine/heavydbTypes.h"
+ Include dependency graph for TableFunctionsCommon.hpp:
+ This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Namespaces

 FileUtilities
 

Enumerations

enum  BoundsType { Min, Max }
 
enum  IntervalType { Inclusive, Exclusive }
 

Functions

template<typename T >
NEVER_INLINE HOST std::pair< T, T > get_column_min_max (const Column< T > &col)
 
NEVER_INLINE HOST std::pair
< int32_t, int32_t > 
get_column_min_max (const Column< TextEncodingDict > &col)
 
template<typename T >
NEVER_INLINE HOST double get_column_mean (const T *data, const int64_t num_rows)
 
template<typename T >
NEVER_INLINE HOST double get_column_mean (const Column< T > &col)
 
template<typename T >
NEVER_INLINE HOST double get_column_std_dev (const Column< T > &col, const double mean)
 
template<typename T >
NEVER_INLINE HOST double get_column_std_dev (const T *data, const int64_t num_rows, const double mean)
 
template<typename T >
void z_std_normalize_col (const T *input_data, T *output_data, const int64_t num_rows, const double mean, const double std_dev)
 
template<typename T >
std::vector< std::vector< T > > z_std_normalize_data (const std::vector< T * > &input_data, const int64_t num_rows)
 
template<typename T >
NEVER_INLINE HOST std::tuple
< T, T, bool > 
get_column_metadata (const Column< T > &col)
 
NEVER_INLINE HOST std::tuple
< int32_t, int32_t, bool > 
get_column_metadata (const Column< TextEncodingDict > &col)
 
template<typename T1 , typename T2 >
NEVER_INLINE HOST T1 distance_in_meters (const T1 fromlon, const T1 fromlat, const T2 tolon, const T2 tolat)
 
int64_t x_y_bin_to_bin_index (const int64_t x_bin, const int64_t y_bin, const int64_t num_x_bins)
 
std::pair< int64_t, int64_t > bin_to_x_y_bin_indexes (const int64_t bin, const int64_t num_x_bins)
 
std::vector
< std::filesystem::path > 
FileUtilities::get_fs_paths (const std::string &file_or_directory)
 
template<typename T >
NEVER_INLINE HOST bool is_valid_tf_input (const T input, const T bounds_val, const BoundsType bounds_type, const IntervalType interval_type)
 

Enumeration Type Documentation

enum BoundsType
Enumerator
Min 
Max 

Definition at line 89 of file TableFunctionsCommon.hpp.

Enumerator
Inclusive 
Exclusive 

Definition at line 91 of file TableFunctionsCommon.hpp.

Function Documentation

std::pair<int64_t, int64_t> bin_to_x_y_bin_indexes ( const int64_t  bin,
const int64_t  num_x_bins 
)
inline

Definition at line 80 of file TableFunctionsCommon.hpp.

Referenced by GeoRaster< T, Z >::get_xy_coords_for_bin_idx().

81  {
82  return std::make_pair(bin % num_x_bins, bin / num_x_bins);
83 }

+ Here is the caller graph for this function:

template<typename T1 , typename T2 >
NEVER_INLINE HOST T1 distance_in_meters ( const T1  fromlon,
const T1  fromlat,
const T2  tolon,
const T2  tolat 
)

Definition at line 400 of file TableFunctionsCommon.cpp.

400  {
401  T1 latitudeArc = (fromlat - tolat) * 0.017453292519943295769236907684886;
402  T1 longitudeArc = (fromlon - tolon) * 0.017453292519943295769236907684886;
403  T1 latitudeH = sin(latitudeArc * 0.5);
404  latitudeH *= latitudeH;
405  T1 lontitudeH = sin(longitudeArc * 0.5);
406  lontitudeH *= lontitudeH;
407  T1 tmp = cos(fromlat * 0.017453292519943295769236907684886) *
408  cos(tolat * 0.017453292519943295769236907684886);
409  return 6372797.560856 * (2.0 * asin(sqrt(latitudeH + tmp * lontitudeH)));
410 }
template<typename T >
NEVER_INLINE HOST double get_column_mean ( const T *  data,
const int64_t  num_rows 
)

Definition at line 110 of file TableFunctionsCommon.cpp.

References max_inputs_per_thread, threading_serial::parallel_for(), and heavydb.dtypes::T.

Referenced by get_column_mean(), and z_std_normalize_data().

110  {
111  // const int64_t num_rows = col.size();
112  const size_t max_thread_count = std::thread::hardware_concurrency();
113  const size_t max_inputs_per_thread = 20000;
114  const size_t num_threads = std::min(
115  max_thread_count, ((num_rows + max_inputs_per_thread - 1) / max_inputs_per_thread));
116 
117  std::vector<double> local_col_sums(num_threads, 0.);
118  std::vector<int64_t> local_col_non_null_counts(num_threads, 0L);
119  tbb::task_arena limited_arena(num_threads);
120  limited_arena.execute([&] {
121  tbb::parallel_for(tbb::blocked_range<int64_t>(0, num_rows),
122  [&](const tbb::blocked_range<int64_t>& r) {
123  const int64_t start_idx = r.begin();
124  const int64_t end_idx = r.end();
125  double local_col_sum = 0.;
126  int64_t local_col_non_null_count = 0;
127  for (int64_t r = start_idx; r < end_idx; ++r) {
128  const T val = data[r];
129  if (val == inline_null_value<T>()) {
130  continue;
131  }
132  local_col_sum += data[r];
133  local_col_non_null_count++;
134  }
135  size_t thread_idx = tbb::this_task_arena::current_thread_index();
136  local_col_sums[thread_idx] += local_col_sum;
137  local_col_non_null_counts[thread_idx] += local_col_non_null_count;
138  });
139  });
140 
141  double col_sum = 0.0;
142  int64_t col_non_null_count = 0L;
143 
144  for (size_t thread_idx = 0; thread_idx < num_threads; ++thread_idx) {
145  col_sum += local_col_sums[thread_idx];
146  col_non_null_count += local_col_non_null_counts[thread_idx];
147  }
148 
149  return col_non_null_count == 0 ? 0 : col_sum / col_non_null_count;
150 }
const size_t max_inputs_per_thread
void parallel_for(const blocked_range< Int > &range, const Body &body, const Partitioner &p=Partitioner())

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

template<typename T >
NEVER_INLINE HOST double get_column_mean ( const Column< T > &  col)

Definition at line 171 of file TableFunctionsCommon.cpp.

References get_column_mean(), Column< T >::getPtr(), and Column< T >::size().

171  {
172  return get_column_mean(col.getPtr(), col.size());
173 }
DEVICE int64_t size() const
Definition: heavydbTypes.h:751
DEVICE T * getPtr() const
Definition: heavydbTypes.h:750
NEVER_INLINE HOST double get_column_mean(const T *data, const int64_t num_rows)

+ Here is the call graph for this function:

template<typename T >
NEVER_INLINE HOST std::tuple<T, T, bool> get_column_metadata ( const Column< T > &  col)

Definition at line 259 of file TableFunctionsCommon.cpp.

References Column< T >::isNull(), max_inputs_per_thread, threading_serial::parallel_for(), Column< T >::size(), and heavydb.dtypes::T.

Referenced by get_column_metadata().

259  {
260  T col_min = std::numeric_limits<T>::max();
261  T col_max = std::numeric_limits<T>::lowest();
262  bool has_nulls = false;
263  const int64_t num_rows = col.size();
264  const size_t max_thread_count = std::thread::hardware_concurrency();
265  const size_t max_inputs_per_thread = 200000;
266  const size_t num_threads = std::min(
267  max_thread_count, ((num_rows + max_inputs_per_thread - 1) / max_inputs_per_thread));
268 
269  std::vector<T> local_col_mins(num_threads, std::numeric_limits<T>::max());
270  std::vector<T> local_col_maxes(num_threads, std::numeric_limits<T>::lowest());
271  std::vector<bool> local_col_has_nulls(num_threads, false);
272  tbb::task_arena limited_arena(num_threads);
273 
274  limited_arena.execute([&] {
275  tbb::parallel_for(tbb::blocked_range<int64_t>(0, num_rows),
276  [&](const tbb::blocked_range<int64_t>& r) {
277  const int64_t start_idx = r.begin();
278  const int64_t end_idx = r.end();
279  T local_col_min = std::numeric_limits<T>::max();
280  T local_col_max = std::numeric_limits<T>::lowest();
281  bool local_has_nulls = false;
282  for (int64_t r = start_idx; r < end_idx; ++r) {
283  if (col.isNull(r)) {
284  local_has_nulls = true;
285  continue;
286  }
287  if (col[r] < local_col_min) {
288  local_col_min = col[r];
289  }
290  if (col[r] > local_col_max) {
291  local_col_max = col[r];
292  }
293  }
294  const size_t thread_idx =
295  tbb::this_task_arena::current_thread_index();
296  if (local_has_nulls) {
297  local_col_has_nulls[thread_idx] = true;
298  }
299  if (local_col_min < local_col_mins[thread_idx]) {
300  local_col_mins[thread_idx] = local_col_min;
301  }
302  if (local_col_max > local_col_maxes[thread_idx]) {
303  local_col_maxes[thread_idx] = local_col_max;
304  }
305  });
306  });
307 
308  for (size_t thread_idx = 0; thread_idx < num_threads; ++thread_idx) {
309  if (local_col_has_nulls[thread_idx]) {
310  has_nulls = true;
311  }
312  if (local_col_mins[thread_idx] < col_min) {
313  col_min = local_col_mins[thread_idx];
314  }
315  if (local_col_maxes[thread_idx] > col_max) {
316  col_max = local_col_maxes[thread_idx];
317  }
318  }
319  return {col_min, col_max, has_nulls};
320 }
DEVICE int64_t size() const
Definition: heavydbTypes.h:751
const size_t max_inputs_per_thread
DEVICE bool isNull(int64_t index) const
Definition: heavydbTypes.h:754
void parallel_for(const blocked_range< Int > &range, const Body &body, const Partitioner &p=Partitioner())

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

NEVER_INLINE HOST std::tuple<int32_t, int32_t, bool> get_column_metadata ( const Column< TextEncodingDict > &  col)

Definition at line 335 of file TableFunctionsCommon.cpp.

References get_column_metadata(), Column< TextEncodingDict >::getPtr(), and Column< TextEncodingDict >::size().

336  {
337  Column<int32_t> int_alias_col(reinterpret_cast<int32_t*>(col.getPtr()), col.size());
338  return get_column_metadata(int_alias_col);
339 }
DEVICE TextEncodingDict * getPtr() const
Definition: heavydbTypes.h:949
NEVER_INLINE HOST std::tuple< T, T, bool > get_column_metadata(const Column< T > &col)
DEVICE int64_t size() const
Definition: heavydbTypes.h:950

+ Here is the call graph for this function:

template<typename T >
NEVER_INLINE HOST std::pair<T, T> get_column_min_max ( const Column< T > &  col)

Definition at line 32 of file TableFunctionsCommon.cpp.

References max_inputs_per_thread, threading_serial::parallel_for(), Column< T >::size(), and heavydb.dtypes::T.

Referenced by ct_union_pushdown_stats__cpu_template(), GeoRaster< T, Z >::GeoRaster(), get_column_min_max(), get_min_or_max(), and get_min_or_max_union().

32  {
33  T col_min = std::numeric_limits<T>::max();
34  T col_max = std::numeric_limits<T>::lowest();
35  const int64_t num_rows = col.size();
36  const size_t max_thread_count = std::thread::hardware_concurrency();
37  const size_t max_inputs_per_thread = 20000;
38  const size_t num_threads = std::min(
39  max_thread_count, ((num_rows + max_inputs_per_thread - 1) / max_inputs_per_thread));
40 
41  std::vector<T> local_col_mins(num_threads, std::numeric_limits<T>::max());
42  std::vector<T> local_col_maxes(num_threads, std::numeric_limits<T>::lowest());
43  tbb::task_arena limited_arena(num_threads);
44 
45  limited_arena.execute([&] {
46  tbb::parallel_for(tbb::blocked_range<int64_t>(0, num_rows),
47  [&](const tbb::blocked_range<int64_t>& r) {
48  const int64_t start_idx = r.begin();
49  const int64_t end_idx = r.end();
50  T local_col_min = std::numeric_limits<T>::max();
51  T local_col_max = std::numeric_limits<T>::lowest();
52  for (int64_t r = start_idx; r < end_idx; ++r) {
53  const T val = col[r];
54  if (val == inline_null_value<T>()) {
55  continue;
56  }
57  if (val < local_col_min) {
58  local_col_min = val;
59  }
60  if (val > local_col_max) {
61  local_col_max = val;
62  }
63  }
64  size_t thread_idx = tbb::this_task_arena::current_thread_index();
65  if (local_col_min < local_col_mins[thread_idx]) {
66  local_col_mins[thread_idx] = local_col_min;
67  }
68  if (local_col_max > local_col_maxes[thread_idx]) {
69  local_col_maxes[thread_idx] = local_col_max;
70  }
71  });
72  });
73 
74  for (size_t thread_idx = 0; thread_idx < num_threads; ++thread_idx) {
75  if (local_col_mins[thread_idx] < col_min) {
76  col_min = local_col_mins[thread_idx];
77  }
78  if (local_col_maxes[thread_idx] > col_max) {
79  col_max = local_col_maxes[thread_idx];
80  }
81  }
82  return std::make_pair(col_min, col_max);
83 }
DEVICE int64_t size() const
Definition: heavydbTypes.h:751
const size_t max_inputs_per_thread
void parallel_for(const blocked_range< Int > &range, const Body &body, const Partitioner &p=Partitioner())

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

NEVER_INLINE HOST std::pair<int32_t, int32_t> get_column_min_max ( const Column< TextEncodingDict > &  col)

Definition at line 98 of file TableFunctionsCommon.cpp.

References get_column_min_max(), Column< TextEncodingDict >::getPtr(), and Column< TextEncodingDict >::size().

98  {
99  Column<int32_t> int_alias_col(reinterpret_cast<int32_t*>(col.getPtr()), col.size());
100  return get_column_min_max(int_alias_col);
101 }
NEVER_INLINE HOST std::pair< T, T > get_column_min_max(const Column< T > &col)
DEVICE TextEncodingDict * getPtr() const
Definition: heavydbTypes.h:949
DEVICE int64_t size() const
Definition: heavydbTypes.h:950

+ Here is the call graph for this function:

template<typename T >
NEVER_INLINE HOST double get_column_std_dev ( const Column< T > &  col,
const double  mean 
)

Definition at line 183 of file TableFunctionsCommon.cpp.

References get_column_std_dev(), Column< T >::getPtr(), and Column< T >::size().

Referenced by get_column_std_dev(), and z_std_normalize_data().

183  {
184  return get_column_std_dev(col.getPtr(), col.size(), mean);
185 }
DEVICE int64_t size() const
Definition: heavydbTypes.h:751
DEVICE T * getPtr() const
Definition: heavydbTypes.h:750
NEVER_INLINE HOST double get_column_std_dev(const Column< T > &col, const double mean)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

template<typename T >
NEVER_INLINE HOST double get_column_std_dev ( const T *  data,
const int64_t  num_rows,
const double  mean 
)

Definition at line 197 of file TableFunctionsCommon.cpp.

References max_inputs_per_thread, threading_serial::parallel_for(), and heavydb.dtypes::T.

199  {
200  // const int64_t num_rows = col.size();
201  const size_t max_thread_count = std::thread::hardware_concurrency();
202  const size_t max_inputs_per_thread = 200000;
203  const size_t num_threads = std::min(
204  max_thread_count, ((num_rows + max_inputs_per_thread - 1) / max_inputs_per_thread));
205 
206  std::vector<double> local_col_squared_residuals(num_threads, 0.);
207  std::vector<int64_t> local_col_non_null_counts(num_threads, 0L);
208  tbb::task_arena limited_arena(num_threads);
209 
210  limited_arena.execute([&] {
211  tbb::parallel_for(tbb::blocked_range<int64_t>(0, num_rows),
212  [&](const tbb::blocked_range<int64_t>& r) {
213  const int64_t start_idx = r.begin();
214  const int64_t end_idx = r.end();
215  double local_col_squared_residual = 0.;
216  int64_t local_col_non_null_count = 0;
217  for (int64_t r = start_idx; r < end_idx; ++r) {
218  const T val = data[r];
219  if (val == inline_null_value<T>()) {
220  continue;
221  }
222  const double residual = val - mean;
223  local_col_squared_residual += (residual * residual);
224  local_col_non_null_count++;
225  }
226  size_t thread_idx = tbb::this_task_arena::current_thread_index();
227  local_col_squared_residuals[thread_idx] +=
228  local_col_squared_residual;
229  local_col_non_null_counts[thread_idx] += local_col_non_null_count;
230  });
231  });
232 
233  double col_sum_squared_residual = 0.0;
234  int64_t col_non_null_count = 0;
235 
236  for (size_t thread_idx = 0; thread_idx < num_threads; ++thread_idx) {
237  col_sum_squared_residual += local_col_squared_residuals[thread_idx];
238  col_non_null_count += local_col_non_null_counts[thread_idx];
239  }
240 
241  return col_non_null_count == 0 ? 0
242  : sqrt(col_sum_squared_residual / col_non_null_count);
243 }
const size_t max_inputs_per_thread
void parallel_for(const blocked_range< Int > &range, const Body &body, const Partitioner &p=Partitioner())

+ Here is the call graph for this function:

template<typename T >
NEVER_INLINE HOST bool is_valid_tf_input ( const T  input,
const T  bounds_val,
const BoundsType  bounds_type,
const IntervalType  interval_type 
)

Definition at line 504 of file TableFunctionsCommon.cpp.

References Exclusive, Inclusive, Max, Min, and UNREACHABLE.

507  {
508  switch (bounds_type) {
509  case BoundsType::Min:
510  switch (interval_type) {
512  return input >= bounds_val;
514  return input > bounds_val;
515  default:
516  UNREACHABLE();
517  }
518  case BoundsType::Max:
519  switch (interval_type) {
521  return input <= bounds_val;
523  return input < bounds_val;
524  default:
525  UNREACHABLE();
526  }
527  break;
528  default:
529  UNREACHABLE();
530  }
531  UNREACHABLE();
532  return false; // To address compiler warning
533 }
#define UNREACHABLE()
Definition: Logger.h:337
template<typename T >
void z_std_normalize_col ( const T *  input_data,
T *  output_data,
const int64_t  num_rows,
const double  mean,
const double  std_dev 
)

Definition at line 342 of file TableFunctionsCommon.cpp.

References threading_serial::parallel_for().

Referenced by z_std_normalize_data().

346  {
347  if (std_dev <= 0.0) {
348  throw std::runtime_error("Standard deviation cannot be <= 0");
349  }
350  const double inv_std_dev = 1.0 / std_dev;
351 
352  tbb::parallel_for(tbb::blocked_range<int64_t>(0, num_rows),
353  [&](const tbb::blocked_range<int64_t>& r) {
354  const int64_t start_idx = r.begin();
355  const int64_t end_idx = r.end();
356  for (int64_t row_idx = start_idx; row_idx < end_idx; ++row_idx) {
357  output_data[row_idx] = (input_data[row_idx] - mean) * inv_std_dev;
358  }
359  });
360 }
void parallel_for(const blocked_range< Int > &range, const Body &body, const Partitioner &p=Partitioner())

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

template<typename T >
std::vector<std::vector<T> > z_std_normalize_data ( const std::vector< T * > &  input_data,
const int64_t  num_rows 
)

Definition at line 374 of file TableFunctionsCommon.cpp.

References get_column_mean(), get_column_std_dev(), and z_std_normalize_col().

Referenced by dbscan__cpu_template(), and kmeans__cpu_template().

375  {
376  const int64_t num_features = input_data.size();
377  std::vector<std::vector<T>> normalized_data(num_features);
378  for (int64_t feature_idx = 0; feature_idx < num_features; ++feature_idx) {
379  const auto mean = get_column_mean(input_data[feature_idx], num_rows);
380  const auto std_dev = get_column_std_dev(input_data[feature_idx], num_rows, mean);
381  normalized_data[feature_idx].resize(num_rows);
382  z_std_normalize_col(input_data[feature_idx],
383  normalized_data[feature_idx].data(),
384  num_rows,
385  mean,
386  std_dev);
387  }
388  return normalized_data;
389 }
void z_std_normalize_col(const T *input_data, T *output_data, const int64_t num_rows, const double mean, const double std_dev)
NEVER_INLINE HOST double get_column_std_dev(const Column< T > &col, const double mean)
NEVER_INLINE HOST double get_column_mean(const T *data, const int64_t num_rows)

+ Here is the call graph for this function:

+ Here is the caller graph for this function: