text: #include "experiments/data-provider.h"
  #include <stdlib.h>
  #include <math.h>
  #include <iostream>
  
  using namespace std;
  using namespace boost;
  using namespace impala;
  
  DataProvider::DataProvider(MemPool* pool, RuntimeProfile* profile) :
    pool_(pool),
    profile_(profile),
    num_rows_(0),
    batch_size_(0),
    rows_returned_(0),
    data_(NULL),
    row_size_(0) {
    SetSeed(0);
  
    bytes_generated_ = ADD_COUNTER(profile, "BytesGenerated", TCounterType::BYTES);
  }
  
  void DataProvider::Reset(int num_rows, int batch_size, const vector<DataProvider::ColDesc>& cols) {
    num_rows_ = num_rows;
    batch_size_ = batch_size;
    rows_returned_ = 0;
    row_size_ = 0;
    cols_ = cols;
    for (int i = 0; i < cols_.size(); ++i) {
      row_size_ += cols[i].bytes;
    }
    data_.reset(new char[row_size_ * batch_size_]);
    COUNTER_SET(bytes_generated_, 0L);
  }
  
  void DataProvider::SetSeed(int seed) {
    rand_generator_.seed(seed);
  }
  
  void RandString(MemPool* pool, StringValue* result,
      const StringValue& min, const StringValue& max, double r,
      variate_generator<minstd_rand&, uniform_real<> >& rand) {
    int min_len = min.len;
    int max_len = max.len;
    int len = r * (max_len - min_len) + min_len;
    char* ptr = reinterpret_cast<char*>(pool->Allocate(len));
    result->len = len;
    result->ptr = ptr;
  
    for (int i = 0; i < len; ++i) {
      int min_char = i < min_len ? min.ptr[i] : 'a';
      int max_char = (i < max_len ? max.ptr[i] : 'z') + 1;
      ptr[i] = rand() * (max_char - min_char) + min_char;
    }
  }
  
  void* DataProvider::NextBatch(int* rows_returned) {
    int num_rows = min(batch_size_, num_rows_ - rows_returned_);
    *rows_returned = num_rows;
    if (num_rows == 0) return NULL;
    COUNTER_ADD(bytes_generated_, num_rows * row_size_);
  
    uniform_real<> dist(0,1);
    variate_generator<minstd_rand&, uniform_real<> > rand_double(rand_generator_, dist);
  
    char* data = data_.get();
    for (int i = 0, row_idx = rows_returned_; i < num_rows; ++i, ++row_idx) {
      for (int j = 0; j < cols_.size(); ++j) {
        double r = rand_double();
        const ColDesc& col = cols_[j];
        switch (col.type) {
          case TYPE_BOOLEAN:
            *reinterpret_cast<bool*>(data) = col.Generate<bool>(r, row_idx);
            break;
          case TYPE_TINYINT:
            *reinterpret_cast<int8_t*>(data) = col.Generate<int8_t>(r, row_idx);
            break;
          case TYPE_SMALLINT:
            *reinterpret_cast<int16_t*>(data) = col.Generate<int16_t>(r, row_idx);
            break;
          case TYPE_INT:
            *reinterpret_cast<int32_t*>(data) = col.Generate<int32_t>(r, row_idx);
            break;
          case TYPE_BIGINT:
            *reinterpret_cast<int64_t*>(data) = col.Generate<int64_t>(r, row_idx);
            break;
          case TYPE_FLOAT:
            *reinterpret_cast<float*>(data) = col.Generate<float>(r, row_idx);
            break;
          case TYPE_DOUBLE:
            *reinterpret_cast<double*>(data) = col.Generate<double>(r, row_idx);
            break;
          case TYPE_VARCHAR:
          case TYPE_STRING: {
            // TODO: generate sequential strings
            StringValue* str = reinterpret_cast<StringValue*>(data);
            RandString(pool_, str, col.min.s, col.max.s, r, rand_double);
            break;
          }
          default:
            break;
        }
        data += col.bytes;
      }
    }
    rows_returned_ += num_rows;
    return reinterpret_cast<void*>(data_.get());
  }
  
  void DataProvider::Print(ostream* stream, char* data, int rows) const {
    char* next_col = reinterpret_cast<char*>(data);
    for (int i = 0; i < rows; ++i) {
      for (int j = 0; j < cols_.size(); ++j) {
        switch (cols_[j].type) {
          case TYPE_BOOLEAN:
            *stream << (*reinterpret_cast<int8_t*>(next_col) ? "true" : "false");
            break;
          case TYPE_TINYINT:
            *stream << (int)*reinterpret_cast<int8_t*>(next_col);
            break;
          case TYPE_SMALLINT:
            *stream << *reinterpret_cast<int16_t*>(next_col);
            break;
          case TYPE_INT:
            *stream << *reinterpret_cast<int32_t*>(next_col);
            break;
          case TYPE_BIGINT:
            *stream << *reinterpret_cast<int64_t*>(next_col);
            break;
          case TYPE_FLOAT:
            *stream << *reinterpret_cast<float*>(next_col);
            break;
          case TYPE_DOUBLE:
            *stream << *reinterpret_cast<double*>(next_col);
            break;
          case TYPE_STRING:
          case TYPE_VARCHAR:
            *stream << *reinterpret_cast<StringValue*>(next_col);
            break;
          default:
            *stream << "BAD" << endl;
            return;
        }
        if (j != cols_.size() - 1) *stream << ", ";
        next_col += cols_[j].bytes;
      }
      *stream << endl;
    }
  }
  
  