/**
 * Utility function to throw an error if a vector is of unequal length.
 * \param[in] gl_sarray of type vector 
 */
void check_vector_equal_size(const gl_sarray& in) {
  // Initialize. 
  DASSERT_TRUE(in.dtype() == flex_type_enum::VECTOR); 
  size_t n_threads = thread::cpu_count();
  n_threads = std::max(n_threads, size_t(1));
  size_t m_size = in.size();
          
  // Throw the following error. 
  auto throw_error = [] (size_t row_number, size_t expected, size_t current) {
    std::stringstream ss;
    ss << "Vectors must be of the same size. Row " << row_number 
       << " contains a vector of size " << current << ". Expected a vector of"
       << " size " << expected << "." << std::endl;
    log_and_throw(ss.str());
  };
  
  // Within each block of the SArray, check that the vectors have the same size.
  std::vector<size_t> expected_sizes (n_threads, size_t(-1));
  in_parallel([&](size_t thread_idx, size_t n_threads) {
    size_t start_row = thread_idx * m_size / n_threads; 
    size_t end_row = (thread_idx + 1) * m_size / n_threads;
    size_t expected_size = size_t(-1);
    size_t row_number = start_row;
    for (const auto& v: in.range_iterator(start_row, end_row)) {
      if (v != FLEX_UNDEFINED) {
        if (expected_size == size_t(-1)) {
          expected_size = v.size();
          expected_sizes[thread_idx] = expected_size; 
        } else {
          DASSERT_TRUE(v.get_type() == flex_type_enum::VECTOR);
          if (expected_size != v.size()) {
            throw_error(row_number, expected_size, v.size());
          }
        }
      }
      row_number++;
    }
  });

  // Make sure sizes accross blocks are also the same. 
  size_t vector_size = size_t(-1);
  for (size_t thread_idx = 0; thread_idx < n_threads; thread_idx++) {
    // If this block contains all None values, skip it.
    if (expected_sizes[thread_idx] != size_t(-1)) {

      if (vector_size == size_t(-1)) {
          vector_size = expected_sizes[thread_idx]; 
      } else {
         if (expected_sizes[thread_idx] != vector_size) {
           throw_error(thread_idx * m_size / n_threads, 
                              vector_size, expected_sizes[thread_idx]);
         } 
      }
    }
  }
}

static void _to_serializable(flexible_type& data, schema_t& schema, const gl_sarray& input) {
  schema.insert(std::make_pair("type", JSON::types::SARRAY));
  flex_dict data_dict;
  data_dict.push_back(std::make_pair("dtype", flex_type_enum_to_name(input.dtype())));

  // will throw away schema from individual values
  flex_list values;
  for (const auto& value : input.range_iterator()) {
    flexible_type serialized_value;
    schema_t serialized_schema;
    _any_to_serializable(serialized_value, serialized_schema, value);
    values.push_back(serialized_value);
  }
  data_dict.push_back(std::make_pair("values", values));
  data = data_dict;
}

gl_sarray::gl_sarray(const gl_sarray& other) {
  m_sarray = other.get_proxy();
}

gl_sarray gl_sarray::append(const gl_sarray& other) const {
  return get_proxy()->append(other.get_proxy());
}

gl_sarray gl_sarray::operator[](const gl_sarray& slice) const {
  return get_proxy()->logical_filter(slice.get_proxy());
}