private void runTest(WindowingSpec.Direction startDirection, int startAmount,
                       WindowingSpec.Direction endDirection, int endAmount) throws Exception {

    for (LinkedList<List<IntWritable>> partition : TEST_PARTITIONS){
      BoundaryDef startBoundary = new BoundaryDef(startDirection, startAmount);
      BoundaryDef endBoundary = new BoundaryDef(endDirection, endAmount);
      AtomicInteger readCounter = new AtomicInteger(0);

      int[] expectedBoundaryStarts = new int[partition.size()];
      int[] expectedBoundaryEnds = new int[partition.size()];
      int expectedReadCountWithoutCache = -1;

      for (PTFInvocationSpec.Order order : ORDERS) {
        for (Integer cacheSize : CACHE_SIZES) {
          LOG.info(Thread.currentThread().getStackTrace()[2].getMethodName());
          LOG.info("Cache: " + cacheSize + " order: " + order);
          BoundaryCache cache = cacheSize == null ? null : new BoundaryCache(cacheSize);
          Pair<PTFPartition, ValueBoundaryScanner> mocks = setupMocks(partition,
                  ORDER_BY_COL, startBoundary, endBoundary, order, cache, readCounter);
          PTFPartition ptfPartition = mocks.getLeft();
          ValueBoundaryScanner scanner = mocks.getRight();
          for (int i = 0; i < partition.size(); ++i) {
            scanner.handleCache(i, ptfPartition);

            scanner.enableBinarySearch = false;
            int startWithoutBinaryPreSearch = scanner.computeStart(i, ptfPartition);
            int endWithoutBinaryPreSearch = scanner.computeEnd(i, ptfPartition) - 1;

            scanner.enableBinarySearch = true;
            int start = scanner.computeStart(i, ptfPartition);
            int end = scanner.computeEnd(i, ptfPartition) - 1;

            Integer col0 = ofNullable(partition.get(i).get(0)).map(v -> v.get()).orElse(null);
            Integer col1 = ofNullable(partition.get(i).get(1)).map(v -> v.get()).orElse(null);
            Integer col2 = ofNullable(partition.get(i).get(2)).map(v -> v.get()).orElse(null);
            LOG.info(String.format("%d|\t%d\t%d\t%d\t|%d-%d", i, col0, col1, col2, start, end));

            assertEquals("start ranges are not equal with/without binary pre-search",
                startWithoutBinaryPreSearch, start);
            assertEquals("end ranges are not equal with/without binary pre-search",
                endWithoutBinaryPreSearch, end);

            if (cache == null) {
              //Cache-less version should be baseline
              expectedBoundaryStarts[i] = start;
              expectedBoundaryEnds[i] = end;
            } else {
              assertEquals(
                  String.format("expected boundary start doesn't match, order: %s, cache size: %d, i: %d",
                      order, cacheSize, i),
                  expectedBoundaryStarts[i], start);
              assertEquals(
                  String.format("expected boundary end doesn't match, order: %s, cache size: %d, i: %d",
                      order, cacheSize, i),
                  expectedBoundaryEnds[i], end);
            }
          }
          if (cache == null) {
            expectedReadCountWithoutCache = readCounter.get();
          } else {
            //Read count should be smaller with cache being used, but larger than the minimum of
            // reading every row once.
            assertTrue(expectedReadCountWithoutCache >= readCounter.get());
            if (startAmount != UNBOUNDED_AMOUNT || endAmount != UNBOUNDED_AMOUNT) {
              assertTrue(partition.size() <= readCounter.get());
            }
          }
          readCounter.set(0);
        }
      }
    }
  }
Executes test on a given window definition. Such a test will be executed against the values set
in ORDERS and CACHE_SIZES, validating ORDERS X CACHE_SIZES test cases. Cache size of null will
be used to setup baseline.
@param startDirection
@param startAmount
@param endDirection
@param endAmount
@throws Exception
ql/src/test/org/apache/hadoop/hive/ql/udf/ptf/TestBoundaryCache.java