@Test public void testGetIntervalsLowWatermarkExceedsHighWatermark() { Map<Long, Long> expected = new HashMap<Long, Long>(); Map<Long, Long> actual = this.simpleWatermark.getIntervals(110, 100, 10, 10); Assert.assertEquals(actual, expected); }
@Override public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionInterval, int maxIntervals) { Preconditions.checkArgument(partitionInterval >= 1, "Invalid value for partitionInterval, value should be at least 1."); Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); HashMap<Long, Long> intervalMap = new HashMap<>(); long nextNum; long interval = getInterval(lowWatermarkValue, highWatermarkValue, partitionInterval, maxIntervals); LOG.info("Recalculated partition interval:" + interval); if (interval == 0) { return intervalMap; } long startNum = lowWatermarkValue; long endNum = highWatermarkValue; boolean longOverflow = false; if (startNum == endNum) { intervalMap.put(startNum, endNum); return intervalMap; } while (startNum < endNum && !longOverflow) { longOverflow = (Long.MAX_VALUE - interval < startNum); nextNum = longOverflow ? endNum : Math.min(startNum + interval, endNum); intervalMap.put(startNum, nextNum); startNum = nextNum; } return intervalMap; }
SimpleWatermark implements Watermark { @Override public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionInterval, int maxIntervals) { Preconditions.checkArgument(partitionInterval >= 1, "Invalid value for partitionInterval, value should be at least 1."); Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); HashMap<Long, Long> intervalMap = new HashMap<>(); long nextNum; long interval = getInterval(lowWatermarkValue, highWatermarkValue, partitionInterval, maxIntervals); LOG.info("Recalculated partition interval:" + interval); if (interval == 0) { return intervalMap; } long startNum = lowWatermarkValue; long endNum = highWatermarkValue; boolean longOverflow = false; if (startNum == endNum) { intervalMap.put(startNum, endNum); return intervalMap; } while (startNum < endNum && !longOverflow) { longOverflow = (Long.MAX_VALUE - interval < startNum); nextNum = longOverflow ? endNum : Math.min(startNum + interval, endNum); intervalMap.put(startNum, nextNum); startNum = nextNum; } return intervalMap; } }
SimpleWatermark implements Watermark { @Override public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionInterval, int maxIntervals) { Preconditions.checkArgument(partitionInterval >= 1, "Invalid value for partitionInterval, value should be at least 1."); Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); HashMap<Long, Long> intervalMap = new HashMap<>(); long nextNum; long interval = getInterval(lowWatermarkValue, highWatermarkValue, partitionInterval, maxIntervals); LOG.info("Recalculated partition interval:" + interval); if (interval == 0) { return intervalMap; } long startNum = lowWatermarkValue; long endNum = highWatermarkValue; boolean longOverflow = false; if (startNum == endNum) { intervalMap.put(startNum, endNum); return intervalMap; } while (startNum < endNum && !longOverflow) { longOverflow = (Long.MAX_VALUE - interval < startNum); nextNum = longOverflow ? endNum : Math.min(startNum + interval, endNum); intervalMap.put(startNum, nextNum); startNum = nextNum; } return intervalMap; } SimpleWatermark(String watermarkColumn, String watermarkFormat); }
SimpleWatermark implements Watermark { @Override public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionInterval, int maxIntervals) { Preconditions.checkArgument(partitionInterval >= 1, "Invalid value for partitionInterval, value should be at least 1."); Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); HashMap<Long, Long> intervalMap = new HashMap<>(); long nextNum; long interval = getInterval(lowWatermarkValue, highWatermarkValue, partitionInterval, maxIntervals); LOG.info("Recalculated partition interval:" + interval); if (interval == 0) { return intervalMap; } long startNum = lowWatermarkValue; long endNum = highWatermarkValue; boolean longOverflow = false; if (startNum == endNum) { intervalMap.put(startNum, endNum); return intervalMap; } while (startNum < endNum && !longOverflow) { longOverflow = (Long.MAX_VALUE - interval < startNum); nextNum = longOverflow ? endNum : Math.min(startNum + interval, endNum); intervalMap.put(startNum, nextNum); startNum = nextNum; } return intervalMap; } SimpleWatermark(String watermarkColumn, String watermarkFormat); @Override String getWatermarkCondition(QueryBasedExtractor<?, ?> extractor, long watermarkValue, String operator); @Override int getDeltaNumForNextWatermark(); @Override HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionInterval,
      int maxIntervals); static long adjustWatermark(String baseWatermark, int diff); }
SimpleWatermark implements Watermark { @Override public HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionInterval, int maxIntervals) { Preconditions.checkArgument(partitionInterval >= 1, "Invalid value for partitionInterval, value should be at least 1."); Preconditions.checkArgument(maxIntervals > 0, "Invalid value for maxIntervals, positive value expected."); HashMap<Long, Long> intervalMap = new HashMap<>(); long nextNum; long interval = getInterval(lowWatermarkValue, highWatermarkValue, partitionInterval, maxIntervals); LOG.info("Recalculated partition interval:" + interval); if (interval == 0) { return intervalMap; } long startNum = lowWatermarkValue; long endNum = highWatermarkValue; boolean longOverflow = false; if (startNum == endNum) { intervalMap.put(startNum, endNum); return intervalMap; } while (startNum < endNum && !longOverflow) { longOverflow = (Long.MAX_VALUE - interval < startNum); nextNum = longOverflow ? endNum : Math.min(startNum + interval, endNum); intervalMap.put(startNum, nextNum); startNum = nextNum; } return intervalMap; } SimpleWatermark(String watermarkColumn, String watermarkFormat); @Override String getWatermarkCondition(QueryBasedExtractor<?, ?> extractor, long watermarkValue, String operator); @Override int getDeltaNumForNextWatermark(); @Override HashMap<Long, Long> getIntervals(long lowWatermarkValue, long highWatermarkValue, long partitionInterval,
      int maxIntervals); static long adjustWatermark(String baseWatermark, int diff);  }