@VisibleForTesting
  SegmentSizeRecommendations estimate(long generatedSegmentSize, int desiredSegmentSize,
      int numRecordsOfGeneratedSegment, long numRecordsPerPush) {

    // calc num rows in desired segment
    double sizeRatio = (double) desiredSegmentSize / generatedSegmentSize;
    long numRowsInDesiredSegment = Math.round(numRecordsOfGeneratedSegment * sizeRatio);

    // calc optimal num segment
    long optimalNumSegments = Math.round(numRecordsPerPush / (double) numRowsInDesiredSegment);
    optimalNumSegments = Math.max(optimalNumSegments, 1);

    // revise optimal num rows in segment
    long optimalNumRowsInSegment = Math.round(numRecordsPerPush / (double) optimalNumSegments);

    // calc optimal segment size
    double rowRatio = (double) optimalNumRowsInSegment / numRecordsOfGeneratedSegment;
    long optimalSegmentSize = Math.round(generatedSegmentSize * rowRatio);

    return new SegmentSizeRecommendations(optimalNumRowsInSegment, optimalNumSegments, optimalSegmentSize);
  }
@Test
  public void testEstimate() {

    /*
     * numRecordsPerPush -> num records per push
     * numRecordsOfGeneratedSegment -> num records of generated segment
     * generatedSegmentSize  -> generated segment size
     * desiredSegmentSize  -> desired segment size
     */

    long numRecordsPerPush = 20 * MILLION;
    int numRecordsOfGeneratedSegment = 5 * MILLION;
    long generatedSegmentSize = 50 * MEGA_BYTE;
    int desiredSegmentSize = 120 * MEGA_BYTE;
    SegmentSizeRecommendations params =
        RULE.estimate(generatedSegmentSize, desiredSegmentSize, numRecordsOfGeneratedSegment, numRecordsPerPush);
    assertEquals(params.getNumSegments(), 2);
    assertEquals(params.getSegmentSize(), 100 * MEGA_BYTE);
    assertEquals(params.getNumRowsPerSegment(), 10 * MILLION);

    numRecordsPerPush = 22 * MILLION;
    numRecordsOfGeneratedSegment = 5 * MILLION;
    generatedSegmentSize = 50 * MEGA_BYTE;
    desiredSegmentSize = 120 * MEGA_BYTE;
    params = RULE.estimate(generatedSegmentSize, desiredSegmentSize, numRecordsOfGeneratedSegment, numRecordsPerPush);
    assertEquals(params.getNumSegments(), 2);
    assertEquals(params.getSegmentSize(), 110 * MEGA_BYTE);
    assertEquals(params.getNumRowsPerSegment(), 11 * MILLION);

    numRecordsPerPush = 18 * MILLION;
    numRecordsOfGeneratedSegment = 5 * MILLION;
    generatedSegmentSize = 50 * MEGA_BYTE;
    desiredSegmentSize = 120 * MEGA_BYTE;
    params = RULE.estimate(generatedSegmentSize, desiredSegmentSize, numRecordsOfGeneratedSegment, numRecordsPerPush);
    assertEquals(params.getNumSegments(), 2);
    assertEquals(params.getSegmentSize(), 90 * MEGA_BYTE);
    assertEquals(params.getNumRowsPerSegment(), 9 * MILLION);

    numRecordsPerPush = 16 * MILLION;
    numRecordsOfGeneratedSegment = 5 * MILLION;
    generatedSegmentSize = 50 * MEGA_BYTE;
    desiredSegmentSize = 120 * MEGA_BYTE;
    params = RULE.estimate(generatedSegmentSize, desiredSegmentSize, numRecordsOfGeneratedSegment, numRecordsPerPush);
    assertEquals(params.getNumSegments(), 1);
    assertEquals(params.getSegmentSize(), 160 * MEGA_BYTE);
    assertEquals(params.getNumRowsPerSegment(), 16 * MILLION);

    numRecordsPerPush = 2 * MILLION;
    numRecordsOfGeneratedSegment = 5 * MILLION;
    generatedSegmentSize = 50 * MEGA_BYTE;
    desiredSegmentSize = 120 * MEGA_BYTE;
    params = RULE.estimate(generatedSegmentSize, desiredSegmentSize, numRecordsOfGeneratedSegment, numRecordsPerPush);
    assertEquals(params.getNumSegments(), 1);
    assertEquals(params.getSegmentSize(), 20 * MEGA_BYTE);
    assertEquals(params.getNumRowsPerSegment(), 2 * MILLION);
  }