Query 22

#include <algorithm>
#include <ctime>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <iterator>
#include <map>
#include <numeric>
#include <sstream>
#include <stdint.h>
#include <thrust/device_vector.h>
#include <thrust/extrema.h>
#include <thrust/reduce.h>
#include <thrust/sequence.h>
#include <thrust/sort.h>
#include <thrust/unique.h>
#include <time.h>
#include <random>
#include <thrust/system/cuda/experimental/pinned_allocator.h>

using namespace std;
using namespace thrust::placeholders;


struct increase_pointers {
  char **char_pos;
  char *first_pos;

  increase_pointers(char **_char_pos, char *_first_pos)
      : char_pos(_char_pos), first_pos(_first_pos) {}
  template <typename IndexType>
  __host__ __device__ void operator()(const IndexType &i) {
    char_pos[i] = first_pos + 9 * i;
  }
};

// applying WHERE conditions
struct check_records {
  const unsigned int *time6;
  const char *bool5;
  bool *res;
  const unsigned int *time6_from;
  const unsigned int *time6_to;
  const char *bool5_val;

  check_records(const unsigned int *_time6, const char *_bool5, bool *_res,
                const unsigned int *_time6_from, const unsigned int *_time6_to,
                const char *_bool5_val)
      : time6(_time6), bool5(_bool5), res(_res), time6_from(_time6_from),
        time6_to(_time6_to), bool5_val(_bool5_val) {}
  template <typename IndexType>
  __host__ __device__ void operator()(const IndexType &i) {

    if (time6[i] >= time6_from[0] && time6[i] <= time6_to[0] &&
        bool5[i] == bool5_val[0]) {
      res[i] = 1;
    } else
      res[i] = 0;
  }
};

// sort fixed-length strings
struct sort_str {
  __host__ __device__ bool operator()(const char *t1, const char *t2) {
    for (unsigned int i = 0; i < 8; i++) {
      if (t1[i] < t2[i])
        return 1;
      if (t1[i] > t2[i])
        return 0;
    };
    return 0;
  }
};

int main(int ac, char **av) {
  unsigned int time6_from = 19000101, time6_to = 20300101;
  uint64_t line_count, segment_size = 200000000;
  char bool5_val = 0;
  string usage = "Usage : query1 [-line_count LINE_COUNT] [-segment_size SEGMENT_SIZE]";

  if (ac != 5) {
    cout << usage << endl;
    exit(1);
  };

  for (unsigned int i = 1; i < ac; i++) {
    if (strcmp(av[i], "-line_count") == 0) {
      if (i + 1 < ac) {
        line_count = atoi(av[i + 1]);
        i++;
      } 
	  else {
        cout << usage << endl;
        exit(1);
	  };		
    }
    else if (strcmp(av[i], "-segment_size") == 0) {
      if (i + 1 < ac) {
        segment_size = atoi(av[i + 1]);
        i++;
      } 
	  else {
        cout << usage << endl;
        exit(1);
      };	
    };
  };

  std::clock_t start1 = 0, start2;
  string file_name = "col2.bin";
  std::fstream f(file_name, std::ios_base::in | ios::binary);
  thrust::device_vector<uint64_t> id2;
  thrust::host_vector<uint64_t, thrust::cuda::experimental::pinned_allocator<uint64_t> > keys(segment_size);  
  file_name = "col6.bin";
  std::fstream f1(file_name, std::ios_base::in | ios::binary);
  thrust::device_vector<unsigned int> time6;
  thrust::host_vector<unsigned int, thrust::cuda::experimental::pinned_allocator<unsigned int> > date_keys(segment_size);  
  file_name = "col5.bin";
  std::fstream f2(file_name, std::ios_base::in | ios::binary);
  thrust::device_vector<char> bool5;
  thrust::host_vector<char, thrust::cuda::experimental::pinned_allocator<char> > bool_keys(segment_size);  
  file_name = "col3.txt";
  std::fstream f3(file_name, std::ios_base::in | ios::binary);  
  thrust::host_vector<char> file_buffer(9*segment_size);
  thrust::device_vector<char> char3(9*segment_size); 
  thrust::device_vector<bool> res;
  unsigned int curr_cnt;
  unsigned long long int total_count = 0, total_line_cnt = 0;
  thrust::host_vector<char> max_seg_char(8);
  uint64_t tt_count = 0;

  uint64_t * m_A; 
  cudaMallocManaged(&m_A, sizeof(uint64_t)*line_count); 
  thrust::device_ptr<uint64_t> h_values(m_A); 

  uint64_t * m_A_tmp; 
  cudaMallocManaged(&m_A_tmp, sizeof(uint64_t)*line_count); 
  thrust::device_ptr<uint64_t> h_values_tmp(m_A_tmp);     
  uint64_t sum = 0;
  bool data_entered = 0;
    
  
  while(1) {
	  while(total_line_cnt < line_count) {
		  curr_cnt = 0;	  
		  if (f) {
			f.read((char*)keys.data(), segment_size*8);
			curr_cnt = f.gcount()/8;
			total_line_cnt = total_line_cnt + curr_cnt;
		  } 
		  else {
			cout << "Could not open file " << file_name << endl;
			exit(0);
		  };

		  curr_cnt = 0;
		  if (f1) {
			f1.read((char*)date_keys.data(), segment_size*4);
			curr_cnt = f1.gcount()/4;
		  }
		  else {
			cout << "Could not open file " << file_name << endl;
			exit(0);
		  };

		  curr_cnt = 0;
		  if (f2) {
			f2.read(bool_keys.data(), segment_size);
			curr_cnt = f2.gcount();
		  }
		  else {
			cout << "Could not open file " << file_name << endl;
			exit(0);
		  };

		  if (f3) {
			f3.read(file_buffer.data(), curr_cnt*9);
		  }
		  else {
			cout << "Could not open file " << file_name << endl;
		  };
		  
		  start2 = std::clock();
		  id2.resize(curr_cnt);
		  thrust::copy(keys.begin(), keys.begin() + curr_cnt, id2.begin());
		  time6.resize(curr_cnt);
		  thrust::copy(date_keys.begin(), date_keys.begin() + curr_cnt, time6.begin());
		  bool5.resize(curr_cnt);
		  thrust::copy(bool_keys.begin(), bool_keys.begin() + curr_cnt, bool5.begin());
		  char3.resize(curr_cnt*9);		
		  thrust::copy(file_buffer.begin(), file_buffer.begin() + 9*curr_cnt, char3.begin());
		  
	 
		  // Now we have all 4 arrays in a device memory
		  thrust::device_vector<bool> res(curr_cnt);
		  
		  if(!data_entered) {
			  string s;
			  cout << "Enter date_from (default:19000101)" <<  endl;  
			  getline(cin,s);	  
			  if(s.length())
				  time6_from = stoi(s);
			  else
				  time6_from = 19000101;
			  cout << "Enter date_to (default:20300101)" <<  endl;  
			  getline(cin,s);	  
			  if(s.length())
				  time6_to = stoi(s);
			  else
				  time6_to = 20300101;
			  cout << "Enter bool (default:0)" <<  endl;  
			  getline(cin,s);	  
			  if(s.length())
				  bool5_val = s[0];  
			  else
				  bool5_val = '0';  
			  data_entered = 1;
		  };	  
		  
		  start1 = 0;  	  
		  
		  // SQL WHERE condition check
		  thrust::device_vector<unsigned int> dev_time6_from(1);
		  thrust::device_vector<unsigned int> dev_time6_to(1);
		  thrust::device_vector<char> dev_bool5(1);
		  dev_time6_from[0] = time6_from;
		  dev_time6_to[0] = time6_to;
		  dev_bool5[0] = bool5_val;

		  start2 = std::clock();
		  thrust::counting_iterator<unsigned int> begin(0);
		  check_records ff(
			  (const unsigned int *)thrust::raw_pointer_cast(time6.data()),
			  (const char *)thrust::raw_pointer_cast(bool5.data()),
			  thrust::raw_pointer_cast(res.data()),
			  (const unsigned int *)thrust::raw_pointer_cast(dev_time6_from.data()),
			  (const unsigned int *)thrust::raw_pointer_cast(dev_time6_to.data()),
			  (const char *)thrust::raw_pointer_cast(dev_bool5.data()));
		  thrust::for_each(begin, begin + res.size(), ff);

	  
		  if(thrust::count(res.begin(), res.end(),1) == 0)
			continue;
		  // copy_if the results
		  thrust::device_vector<uint64_t> id2_cpy(res.size());
		  auto w_count = thrust::copy_if(id2.begin(), id2.end(), res.begin(),
										 id2_cpy.begin(), thrust::identity<bool>()) -
						 id2_cpy.begin();
		  tt_count = tt_count + w_count;
		  
		  // SQL DISTINCT
		  thrust::sort(id2_cpy.begin(), id2_cpy.begin() + w_count);
		  auto distinct_cnt =
			  thrust::unique(id2_cpy.begin(), id2_cpy.begin() + w_count) -
			  id2_cpy.begin();
		  cout << "segment distinct count " << distinct_cnt << endl;

		  if(total_count == 0) {
			thrust::copy(id2_cpy.begin(), id2_cpy.begin() + distinct_cnt, h_values);    		
		  }	
		  else {		
			thrust::merge(h_values, h_values + total_count, id2_cpy.begin(), id2_cpy.begin() + distinct_cnt, h_values_tmp);  		
			thrust::copy(h_values_tmp, h_values_tmp + distinct_cnt + total_count, h_values);		
		  };	
		  total_count = total_count + distinct_cnt;
		  
		  // SQL AVG  - just sum the all segments values and then divide by segment count
		  sum = sum + thrust::reduce(id2_cpy.begin(), id2_cpy.end());  	  
		  cout << "segment sum " << sum << endl;	
		  
		  // MAX(char3)	  
		  thrust::host_vector<char> max_char(8);
		  thrust::device_vector<char *> char_pos(curr_cnt);
		  increase_pointers func(thrust::raw_pointer_cast(char_pos.data()),
								 thrust::raw_pointer_cast(char3.data()));
		  thrust::for_each(begin, begin + curr_cnt, func);
		  
		  thrust::device_vector<char *> char_pos_res(w_count);
		  
		  // apply WHERE conditions
		  thrust::copy_if(char_pos.begin(), char_pos.end(), res.begin(),
						  char_pos_res.begin(), thrust::identity<bool>());
		  auto end =
			  thrust::max_element(char_pos_res.begin(), char_pos_res.end(), sort_str());
		  auto device_str = *(end);
		  cudaMemcpy(max_char.data(), (void *)device_str, 8, cudaMemcpyDeviceToHost);
		  for (int z = 0; z < 8; z++) {
			if(max_char[z] > max_seg_char[z]) 
			  max_seg_char = max_char;
			else 
			  if(max_char[z] < max_seg_char[z])
				break;
		  };		 
		  cout << "segment MAX(char3) = ";
		  for (int z = 0; z < 8; z++)
			cout << max_char[z];
		  cout << endl;  	  
		  start1 = start1 + (std::clock() - start2);
		  
	  };		

	  start2 = std::clock();
	  res.resize(0);
	  res.shrink_to_fit();				     
	  id2.resize(0);
	  id2.shrink_to_fit();				 
	  time6.resize(0);
	  time6.shrink_to_fit();				 
	  bool5.resize(0);
	  bool5.shrink_to_fit();				 	
	  char3.resize(0);
	  char3.shrink_to_fit();				 
		
	  auto final_distinct_cnt =
		  thrust::unique(h_values, h_values + total_count) - h_values;
	  		  
	  cout << "merge distinct count " << final_distinct_cnt << endl;   
	  cout << "AVG(id2) = " << sum / tt_count << endl;   
	  
	  cout << "MAX(char3) = ";
	  for (int z = 0; z < 8; z++)
		cout << max_seg_char[z];
	  cout << endl;  
	  
	  start1 = start1 + (std::clock() - start2);
	  std::cout << "gpu running time " <<  start1/(double)CLOCKS_PER_SEC   << '\n';	
	  
	  data_entered = 0;
	  total_line_cnt = 0;
	  total_count = 0;
	  f.seekg(ios_base::beg);
	  f1.seekg(ios_base::beg);
	  f2.seekg(ios_base::beg);
	  f3.seekg(ios_base::beg);
	  
  }
  cudaFree(m_A);	  
  cudaFree(m_A_tmp);  
  return 0;
}