@Test
public void testProductMomentCorrelationCoefficient() {

  double[] values1 = IntStream.range(0, 10000)
          .mapToDouble(i -> ThreadLocalRandom.current().nextDouble())
          .toArray();
  double[] values2 = IntStream.range(0, 10000)
          .mapToDouble(i -> -ThreadLocalRandom.current().nextDouble())
          .toArray();
  double[] statistics = new double[6];
  for (int i = 0; i < 10000; ++i) {
    double sp = values1[i];
    double sq = values2[i];
    double spp = sp * sp;
    double sqq = sq * sq;
    double spq = sp * sq;
    statistics[0] += sp;
    statistics[1] += sq;
    statistics[2] += spp;
    statistics[3] += sqq;
    statistics[4] += spq;
    statistics[5] += 1;
  }
  double sp = statistics[0];
  double sq = statistics[1];
  double spp = statistics[2];
  double sqq = statistics[3];
  double spq = statistics[4];
  double n = statistics[5];
  double pmccExpected = (n * spq - sq * sp) / (Math.sqrt((n * spp - sp * sp) * (n * sqq - sq * sq)));
  DoubleArrayPageWriter writer1 = new DoubleArrayPageWriter(Involutions::reverse);
  DoubleArrayPageWriter writer2 = new DoubleArrayPageWriter(Involutions::reverse);
  SplitMapPageWriter filterWriter = new SplitMapPageWriter(Involutions::reverse);
  int key = 0;
  int multiple = 0;
  for (int k = 0; k < 20; ++k) {
    for (int i = 0; i < 500; ++i) {
      filterWriter.add(key + i);
      writer1.add(key + i, values1[i + multiple * 500]);
      writer2.add(key + i, values2[i + multiple * 500]);
    }
    ++multiple;
    key += 1 << 16;
  }
  SplitMap filter = filterWriter.toSplitMap();
  PrefixIndex<ChunkedDoubleArray> pi1 = writer1.toIndex();
  PrefixIndex<ChunkedDoubleArray> pi2 = writer2.toIndex();
  double pmcc = filter.stream()
          .parallel()
          .map(partition -> partition.reduce(SimpleLinearRegression.<InputModel>reducer(pi1, pi2)))
          .collect(SimpleLinearRegression.pmcc());
  assertEquals(pmcc, pmccExpected, 1E-5);
}