Speed up date_histogram by precomputing ranges (#61467)

A few of us were talking about ways to speed up the `date_histogram`
using the index for the timestamp rather than the doc values. To do that
we'd have to pre-compute all of the "round down" points in the index. It
turns out that *just* precomputing those values speeds up rounding
fairly significantly:
```
Benchmark  (count)      (interval)                   (range)            (zone)  Mode  Cnt          Score         Error  Units
before    10000000  calendar month  2000-10-28 to 2000-10-31               UTC  avgt   10   96461080.982 ±  616373.011  ns/op
before    10000000  calendar month  2000-10-28 to 2000-10-31  America/New_York  avgt   10  130598950.850 ± 1249189.867  ns/op
after     10000000  calendar month  2000-10-28 to 2000-10-31               UTC  avgt   10   52311775.080 ±  107171.092  ns/op
after     10000000  calendar month  2000-10-28 to 2000-10-31  America/New_York  avgt   10   54800134.968 ±  373844.796  ns/op
```

That's a 46% speed up when there isn't a time zone and a 58% speed up
when there is.

This doesn't work for every time zone, specifically those that have two
midnights in a single day due to daylight savings time will produce wonky
results. So they don't get the optimization.

Second, this requires a few expensive computation up front to make the
transition array. And if the transition array is too large then we give
up and use the original mechanism, throwing away all of the work we did
to build the array. This seems appropriate for most usages of `round`,
but this change uses it for *all* usages of `round`. That seems ok for
now, but it might be worth investigating in a follow up.

I ran a macrobenchmark as well which showed an 11% preformance
improvement. *BUT* the benchmark wasn't tuned for my desktop so it
overwhelmed it and might have produced "funny" results. I think it is
pretty clear that this is an improvement, but know the measurement is
weird:

```
Benchmark  (count)      (interval)                   (range)            (zone)  Mode  Cnt          Score         Error  Units
before    10000000  calendar month  2000-10-28 to 2000-10-31               UTC  avgt   10   96461080.982 ±  616373.011  ns/op
before    10000000  calendar month  2000-10-28 to 2000-10-31  America/New_York  avgt   10  g± 1249189.867  ns/op
after     10000000  calendar month  2000-10-28 to 2000-10-31               UTC  avgt   10   52311775.080 ±  107171.092  ns/op
after     10000000  calendar month  2000-10-28 to 2000-10-31  America/New_York  avgt   10   54800134.968 ±  373844.796  ns/op

Before:
|               Min Throughput | hourly_agg |        0.11 |  ops/s |
|            Median Throughput | hourly_agg |        0.11 |  ops/s |
|               Max Throughput | hourly_agg |        0.11 |  ops/s |
|      50th percentile latency | hourly_agg |      650623 |     ms |
|      90th percentile latency | hourly_agg |      821478 |     ms |
|      99th percentile latency | hourly_agg |      859780 |     ms |
|     100th percentile latency | hourly_agg |      864030 |     ms |
| 50th percentile service time | hourly_agg |     9268.71 |     ms |
| 90th percentile service time | hourly_agg |        9380 |     ms |
| 99th percentile service time | hourly_agg |     9626.88 |     ms |
|100th percentile service time | hourly_agg |     9884.27 |     ms |
|                   error rate | hourly_agg |           0 |      % |

After:
|               Min Throughput | hourly_agg |        0.12 |  ops/s |
|            Median Throughput | hourly_agg |        0.12 |  ops/s |
|               Max Throughput | hourly_agg |        0.12 |  ops/s |
|      50th percentile latency | hourly_agg |      519254 |     ms |
|      90th percentile latency | hourly_agg |      653099 |     ms |
|      99th percentile latency | hourly_agg |      683276 |     ms |
|     100th percentile latency | hourly_agg |      686611 |     ms |
| 50th percentile service time | hourly_agg |     8371.41 |     ms |
| 90th percentile service time | hourly_agg |     8407.02 |     ms |
| 99th percentile service time | hourly_agg |     8536.64 |     ms |
|100th percentile service time | hourly_agg |     8538.54 |     ms |
|                   error rate | hourly_agg |           0 |      % |
```

server/src/main/java/org/elasticsearch/common/LocalTimeOffset.java

@@ -22,6 +22,7 @@ package org.elasticsearch.common;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.ZoneId;
+import java.time.temporal.ChronoField;
 import java.time.zone.ZoneOffsetTransition;
 import java.time.zone.ZoneOffsetTransitionRule;
 import java.time.zone.ZoneRules;
@@ -174,6 +175,12 @@ public abstract class LocalTimeOffset {
      */
     protected abstract LocalTimeOffset offsetContaining(long utcMillis);
 
+    /**
+     * Does this transition or any previous transitions move back to the
+     * previous day? See {@link Lookup#anyMoveBackToPreviousDay()} for rules.
+     */
+    protected abstract boolean anyMoveBackToPreviousDay();
+
     @Override
     public String toString() {
         return toString(millis);
@@ -195,6 +202,15 @@ public abstract class LocalTimeOffset {
          */
         public abstract LocalTimeOffset fixedInRange(long minUtcMillis, long maxUtcMillis);
 
+        /**
+         * Do any of the transitions move back to the previous day?
+         * <p>
+         * Note: If an overlap occurs at, say, 1 am and jumps back to
+         * <strong>exactly</strong> midnight then it doesn't count because
+         * midnight is still counted as being in the "next" day.
+         */
+        public abstract boolean anyMoveBackToPreviousDay();
+
         /**
          * The number of offsets in the lookup. Package private for testing.
          */
@@ -225,6 +241,11 @@ public abstract class LocalTimeOffset {
             return this;
         }
 
+        @Override
+        protected boolean anyMoveBackToPreviousDay() {
+            return false;
+        }
+
         @Override
         protected String toString(long millis) {
             return Long.toString(millis);
@@ -298,6 +319,11 @@ public abstract class LocalTimeOffset {
             return firstMissingLocalTime;
         }
 
+        @Override
+        protected boolean anyMoveBackToPreviousDay() {
+            return previous().anyMoveBackToPreviousDay();
+        }
+
         @Override
         protected String toString(long millis) {
             return "Gap of " + millis + "@" + Instant.ofEpochMilli(startUtcMillis());
@@ -307,13 +333,21 @@ public abstract class LocalTimeOffset {
     public static class Overlap extends Transition {
         private final long firstOverlappingLocalTime;
         private final long firstNonOverlappingLocalTime;
-
-        private Overlap(long millis, LocalTimeOffset previous, long startUtcMillis,
-                long firstOverlappingLocalTime, long firstNonOverlappingLocalTime) {
+        private final boolean movesBackToPreviousDay;
+
+        private Overlap(
+            long millis,
+            LocalTimeOffset previous,
+            long startUtcMillis,
+            long firstOverlappingLocalTime,
+            long firstNonOverlappingLocalTime,
+            boolean movesBackToPreviousDay
+        ) {
             super(millis, previous, startUtcMillis);
             this.firstOverlappingLocalTime = firstOverlappingLocalTime;
             this.firstNonOverlappingLocalTime = firstNonOverlappingLocalTime;
             assert firstOverlappingLocalTime < firstNonOverlappingLocalTime;
+            this.movesBackToPreviousDay = movesBackToPreviousDay;
         }
 
         @Override
@@ -341,6 +375,11 @@ public abstract class LocalTimeOffset {
             return firstOverlappingLocalTime;
         }
 
+        @Override
+        protected boolean anyMoveBackToPreviousDay() {
+            return movesBackToPreviousDay || previous().anyMoveBackToPreviousDay();
+        }
+
         @Override
         protected String toString(long millis) {
             return "Overlap of " + millis + "@" + Instant.ofEpochMilli(startUtcMillis());
@@ -375,6 +414,11 @@ public abstract class LocalTimeOffset {
         public String toString() {
             return String.format(Locale.ROOT, "FixedLookup[for %s at %s]", zone, fixed);
         }
+
+        @Override
+        public boolean anyMoveBackToPreviousDay() {
+            return false;
+        }
     }
 
     /**
@@ -406,6 +450,11 @@ public abstract class LocalTimeOffset {
         int size() {
             return size;
         }
+
+        @Override
+        public boolean anyMoveBackToPreviousDay() {
+            return lastOffset.anyMoveBackToPreviousDay();
+        }
     }
 
     /**
@@ -453,6 +502,11 @@ public abstract class LocalTimeOffset {
             return offsets.length;
         }
 
+        @Override
+        public boolean anyMoveBackToPreviousDay() {
+            return offsets[offsets.length - 1].anyMoveBackToPreviousDay();
+        }
+
         @Override
         public String toString() {
             return String.format(Locale.ROOT, "TransitionArrayLookup[for %s between %s and %s]",
@@ -505,7 +559,25 @@ public abstract class LocalTimeOffset {
             }
             long firstOverlappingLocalTime = utcStart + offsetAfterMillis;
             long firstNonOverlappingLocalTime = utcStart + offsetBeforeMillis;
-            return new Overlap(offsetAfterMillis, previous, utcStart, firstOverlappingLocalTime, firstNonOverlappingLocalTime);
+            return new Overlap(
+                offsetAfterMillis,
+                previous,
+                utcStart,
+                firstOverlappingLocalTime,
+                firstNonOverlappingLocalTime,
+                movesBackToPreviousDay(transition)
+            );
+        }
+
+        private static boolean movesBackToPreviousDay(ZoneOffsetTransition transition) {
+            if (transition.getDateTimeBefore().getDayOfMonth() == transition.getDateTimeAfter().getDayOfMonth()) {
+                return false;
+            }
+            if (transition.getDateTimeBefore().getLong(ChronoField.NANO_OF_DAY) == 0L) {
+                // If we change *at* midnight this is ok.
+                return false;
+            }
+            return true;
         }
     }
 

server/src/main/java/org/elasticsearch/common/Rounding.java

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.common;
 
+import org.apache.lucene.util.ArrayUtil;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.LocalTimeOffset.Gap;
 import org.elasticsearch.common.LocalTimeOffset.Overlap;
@@ -43,6 +44,7 @@ import java.time.temporal.TemporalField;
 import java.time.temporal.TemporalQueries;
 import java.time.zone.ZoneOffsetTransition;
 import java.time.zone.ZoneRules;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Locale;
 import java.util.Objects;
@@ -404,6 +406,34 @@ public abstract class Rounding implements Writeable {
         }
     }
 
+    private abstract class PreparedRounding implements Prepared {
+        /**
+         * Attempt to build a {@link Prepared} implementation that relies on pre-calcuated
+         * "round down" points. If there would be more than {@code max} points then return
+         * the original implementation, otherwise return the new, faster implementation.
+         */
+        protected Prepared maybeUseArray(long minUtcMillis, long maxUtcMillis, int max) {
+            long[] values = new long[1];
+            long rounded = round(minUtcMillis);
+            int i = 0;
+            values[i++] = rounded;
+            while ((rounded = nextRoundingValue(rounded)) <= maxUtcMillis) {
+                if (i >= max) {
+                    return this;
+                }
+                /*
+                 * We expect a time in the last transition (rounded - 1) to round
+                 * to the last value we calculated. If it doesn't then we're
+                 * probably doing something wrong here....
+                 */
+                assert values[i - 1] == round(rounded - 1);
+                values = ArrayUtil.grow(values, i + 1);
+                values[i++]= rounded;
+            }
+            return new ArrayRounding(values, i, this);
+        }
+    }
+
     static class TimeUnitRounding extends Rounding {
         static final byte ID = 1;
 
@@ -468,6 +498,15 @@ public abstract class Rounding implements Writeable {
 
         @Override
         public Prepared prepare(long minUtcMillis, long maxUtcMillis) {
+            /*
+             * 128 is a power of two that isn't huge. We might be able to do
+             * better if the limit was based on the actual type of prepared
+             * rounding but this'll do for now.
+             */
+            return prepareOffsetOrJavaTimeRounding(minUtcMillis, maxUtcMillis).maybeUseArray(minUtcMillis, maxUtcMillis, 128);
+        }
+
+        private TimeUnitPreparedRounding prepareOffsetOrJavaTimeRounding(long minUtcMillis, long maxUtcMillis) {
             long minLookup = minUtcMillis - unit.extraLocalOffsetLookup();
             long maxLookup = maxUtcMillis;
 
@@ -486,7 +525,6 @@ public abstract class Rounding implements Writeable {
                 // Range too long, just use java.time
                 return prepareJavaTime();
             }
-
             LocalTimeOffset fixedOffset = lookup.fixedInRange(minLookup, maxLookup);
             if (fixedOffset != null) {
                 // The time zone is effectively fixed
@@ -515,7 +553,7 @@ public abstract class Rounding implements Writeable {
         }
 
         @Override
-        Prepared prepareJavaTime() {
+        TimeUnitPreparedRounding prepareJavaTime() {
             if (unitRoundsToMidnight) {
                 return new JavaTimeToMidnightRounding();
             }
@@ -554,7 +592,7 @@ public abstract class Rounding implements Writeable {
             return "Rounding[" + unit + " in " + timeZone + "]";
         }
 
-        private abstract class TimeUnitPreparedRounding implements Prepared {
+        private abstract class TimeUnitPreparedRounding extends PreparedRounding {
             @Override
             public double roundingSize(long utcMillis, DateTimeUnit timeUnit) {
                 if (timeUnit.isMillisBased == unit.isMillisBased) {
@@ -648,6 +686,14 @@ public abstract class Rounding implements Writeable {
             public long beforeOverlap(long localMillis, Overlap overlap) {
                 return overlap.previous().localToUtc(localMillis, this);
             }
+
+            @Override
+            protected Prepared maybeUseArray(long minUtcMillis, long maxUtcMillis, int max) {
+                if (lookup.anyMoveBackToPreviousDay()) {
+                    return this;
+                }
+                return super.maybeUseArray(minUtcMillis, maxUtcMillis, max);
+            }
         }
 
         private class NotToMidnightRounding extends AbstractNotToMidnightRounding implements LocalTimeOffset.Strategy {
@@ -707,6 +753,12 @@ public abstract class Rounding implements Writeable {
                 return firstTimeOnDay(localMidnight);
             }
 
+            @Override
+            protected Prepared maybeUseArray(long minUtcMillis, long maxUtcMillis, int max) {
+                // We don't have the right information needed to know if this is safe for this time zone so we always use java rounding
+                return this;
+            }
+
             private long firstTimeOnDay(LocalDateTime localMidnight) {
                 assert localMidnight.toLocalTime().equals(LocalTime.of(0, 0, 0)) : "firstTimeOnDay should only be called at midnight";
 
@@ -1107,7 +1159,7 @@ public abstract class Rounding implements Writeable {
 
         @Override
         public Prepared prepare(long minUtcMillis, long maxUtcMillis) {
-            return wrapPreparedRounding(delegate.prepare(minUtcMillis, maxUtcMillis));
+            return wrapPreparedRounding(delegate.prepare(minUtcMillis - offset, maxUtcMillis - offset));
         }
 
         @Override
@@ -1182,4 +1234,41 @@ public abstract class Rounding implements Writeable {
                 throw new ElasticsearchException("unknown rounding id [" + id + "]");
         }
     }
+
+    /**
+     * Implementation of {@link Prepared} using pre-calculated "round down" points.
+     */
+    private static class ArrayRounding implements Prepared {
+        private final long[] values;
+        private final int max;
+        private final Prepared delegate;
+
+        private ArrayRounding(long[] values, int max, Prepared delegate) {
+            this.values = values;
+            this.max = max;
+            this.delegate = delegate;
+        }
+
+        @Override
+        public long round(long utcMillis) {
+            assert values[0] <= utcMillis : "utcMillis must be after " + values[0];
+            int idx = Arrays.binarySearch(values, 0, max, utcMillis);
+            assert idx != -1 : "The insertion point is before the array! This should have tripped the assertion above.";
+            assert -1 - idx <= values.length : "This insertion point is after the end of the array.";
+            if (idx < 0) {
+                idx = -2 - idx;
+            }
+            return values[idx];
+        }
+
+        @Override
+        public long nextRoundingValue(long utcMillis) {
+            return delegate.nextRoundingValue(utcMillis);
+        }
+
+        @Override
+        public double roundingSize(long utcMillis, DateTimeUnit timeUnit) {
+            return delegate.roundingSize(utcMillis, timeUnit);
+        }
+    }
 }

server/src/test/java/org/elasticsearch/common/LocalTimeOffsetTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.common;
 
+import org.elasticsearch.bootstrap.JavaVersion;
 import org.elasticsearch.common.LocalTimeOffset.Gap;
 import org.elasticsearch.common.LocalTimeOffset.Overlap;
 import org.elasticsearch.common.time.DateFormatter;
@@ -51,20 +52,22 @@ public class LocalTimeOffsetTests extends ESTestCase {
     }
 
     public void testUtc() {
-        assertFixOffset(ZoneId.of("UTC"), 0);
+        assertFixedOffset(ZoneId.of("UTC"), 0);
     }
 
     public void testFixedOffset() {
         ZoneOffset zone = ZoneOffset.ofTotalSeconds(between((int) -TimeUnit.HOURS.toSeconds(18), (int) TimeUnit.HOURS.toSeconds(18)));
-        assertFixOffset(zone, zone.getTotalSeconds() * 1000);
+        assertFixedOffset(zone, zone.getTotalSeconds() * 1000);
     }
 
-    private void assertFixOffset(ZoneId zone, long offsetMillis) {
+    private void assertFixedOffset(ZoneId zone, long offsetMillis) {
         LocalTimeOffset fixed = LocalTimeOffset.fixedOffset(zone);
         assertThat(fixed, notNullValue());
 
         LocalTimeOffset.Lookup lookup = LocalTimeOffset.lookup(zone, Long.MIN_VALUE, Long.MAX_VALUE);
         assertThat(lookup.size(), equalTo(1));
+        assertThat(lookup.anyMoveBackToPreviousDay(), equalTo(false));
+
         long min = randomLong();
         long max = randomValueOtherThan(min, ESTestCase::randomLong);
         if (min > max) {
@@ -72,8 +75,10 @@ public class LocalTimeOffsetTests extends ESTestCase {
             min = max;
             max = s;
         }
+
         LocalTimeOffset fixedInRange = lookup.fixedInRange(min, max);
         assertThat(fixedInRange, notNullValue());
+        assertThat(fixedInRange.anyMoveBackToPreviousDay(), equalTo(false));
 
         assertRoundingAtOffset(randomBoolean() ? fixed : fixedInRange, randomLong(), offsetMillis);
     }
@@ -118,6 +123,7 @@ public class LocalTimeOffsetTests extends ESTestCase {
         assertRoundingAtOffset(lookup.lookup(max), max, minMaxOffset);
         assertThat(lookup.fixedInRange(min, max), nullValue());
         assertThat(lookup.fixedInRange(min, sameOffsetAsMin), sameInstance(lookup.lookup(min)));
+        assertThat(lookup.anyMoveBackToPreviousDay(), equalTo(false)); // None of the examples we use this method with move back
     }
 
     // Some sanity checks for when you pas a single time. We don't expect to do this much but it shouldn't be totally borked.
@@ -241,6 +247,28 @@ public class LocalTimeOffsetTests extends ESTestCase {
         assertThat(gapOffset.localToUtc(localSkippedTime, useValueForGap(gapValue)), equalTo(gapValue));
     }
 
+    public void testKnownMovesBackToPreviousDay() {
+        assertKnownMovesBacktoPreviousDay("America/Goose_Bay", "2010-11-07T03:01:00");
+        assertKnownMovesBacktoPreviousDay("America/Moncton", "2006-10-29T03:01:00");
+        assertKnownMovesBacktoPreviousDay("America/Moncton", "2005-10-29T03:01:00");
+        assertKnownMovesBacktoPreviousDay("America/St_Johns", "2010-11-07T02:31:00");
+        assertKnownMovesBacktoPreviousDay("Canada/Newfoundland", "2010-11-07T02:31:00");
+        assertKnownMovesBacktoPreviousDay("Europe/Paris", "1911-03-1T00:01:00");
+        if (JavaVersion.current().compareTo(JavaVersion.parse("11")) > 0) {
+            // Added in java 12
+            assertKnownMovesBacktoPreviousDay("Pacific/Guam", "1969-01-25T13:01:00");
+            assertKnownMovesBacktoPreviousDay("Pacific/Saipan", "1969-01-25T13:01:00");
+        }
+    }
+
+    private void assertKnownMovesBacktoPreviousDay(String zone, String time) {
+        long utc = utcTime(time);
+        assertTrue(
+            zone + " just after " + time + " should move back",
+            LocalTimeOffset.lookup(ZoneId.of(zone), utc, utc + 1).anyMoveBackToPreviousDay()
+        );
+    }
+
     private static long utcTime(String time) {
         return DateFormatter.forPattern("date_optional_time").parseMillis(time);
     }

server/src/test/java/org/elasticsearch/common/RoundingTests.java

@@ -36,6 +36,8 @@ import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.TemporalAccessor;
 import java.time.zone.ZoneOffsetTransition;
+import java.time.zone.ZoneOffsetTransitionRule;
+import java.time.zone.ZoneRules;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -214,6 +216,9 @@ public class RoundingTests extends ESTestCase {
         assertThat(rounding.nextRoundingValue(0), equalTo(oneDay - twoHours));
         assertThat(rounding.withoutOffset().round(0), equalTo(0L));
         assertThat(rounding.withoutOffset().nextRoundingValue(0), equalTo(oneDay));
+
+        rounding = Rounding.builder(Rounding.DateTimeUnit.DAY_OF_MONTH).timeZone(ZoneId.of("America/New_York")).offset(-twoHours).build();
+        assertThat(rounding.round(time("2020-11-01T09:00:00")), equalTo(time("2020-11-01T02:00:00")));
     }
 
     /**
@@ -231,51 +236,55 @@ public class RoundingTests extends ESTestCase {
         for (int i = 0; i < 1000; ++i) {
             Rounding.DateTimeUnit unit = randomFrom(Rounding.DateTimeUnit.values());
             ZoneId tz = randomZone();
-            Rounding rounding = new Rounding.TimeUnitRounding(unit, tz);
-            long[] bounds = randomDateBounds();
-            Rounding.Prepared prepared = rounding.prepare(bounds[0], bounds[1]);
-
-            // Check that rounding is internally consistent and consistent with nextRoundingValue
-            long date = dateBetween(bounds[0], bounds[1]);
-            long unitMillis = unit.getField().getBaseUnit().getDuration().toMillis();
-            // FIXME this was copy pasted from the other impl and not used. breaks the nasty date actually gets assigned
-            if (randomBoolean()) {
-                nastyDate(date, tz, unitMillis);
-            }
-            final long roundedDate = prepared.round(date);
-            final long nextRoundingValue = prepared.nextRoundingValue(roundedDate);
-
-            assertInterval(roundedDate, date, nextRoundingValue, rounding, tz);
-
-            // check correct unit interval width for units smaller than a day, they should be fixed size except for transitions
-            if (unitMillis <= 86400 * 1000) {
-                // if the interval defined didn't cross timezone offset transition, it should cover unitMillis width
-                int offsetRounded = tz.getRules().getOffset(Instant.ofEpochMilli(roundedDate - 1)).getTotalSeconds();
-                int offsetNextValue = tz.getRules().getOffset(Instant.ofEpochMilli(nextRoundingValue + 1)).getTotalSeconds();
-                if (offsetRounded == offsetNextValue) {
-                    assertThat("unit interval width not as expected for [" + unit + "], [" + tz + "] at "
-                        + Instant.ofEpochMilli(roundedDate), nextRoundingValue - roundedDate, equalTo(unitMillis));
-                }
+            long[] bounds = randomDateBounds(unit);
+            assertUnitRoundingSameAsJavaUtilTimeImplementation(unit, tz, bounds[0], bounds[1]);
+        }
+    }
+
+    private void assertUnitRoundingSameAsJavaUtilTimeImplementation(Rounding.DateTimeUnit unit, ZoneId tz, long start, long end) {
+        Rounding rounding = new Rounding.TimeUnitRounding(unit, tz);
+        Rounding.Prepared prepared = rounding.prepare(start, end);
+
+        // Check that rounding is internally consistent and consistent with nextRoundingValue
+        long date = dateBetween(start, end);
+        long unitMillis = unit.getField().getBaseUnit().getDuration().toMillis();
+        // FIXME this was copy pasted from the other impl and not used. breaks the nasty date actually gets assigned
+        if (randomBoolean()) {
+            nastyDate(date, tz, unitMillis);
+        }
+        final long roundedDate = prepared.round(date);
+        final long nextRoundingValue = prepared.nextRoundingValue(roundedDate);
+
+        assertInterval(roundedDate, date, nextRoundingValue, rounding, tz);
+
+        // check correct unit interval width for units smaller than a day, they should be fixed size except for transitions
+        if (unitMillis <= 86400 * 1000) {
+            // if the interval defined didn't cross timezone offset transition, it should cover unitMillis width
+            int offsetRounded = tz.getRules().getOffset(Instant.ofEpochMilli(roundedDate - 1)).getTotalSeconds();
+            int offsetNextValue = tz.getRules().getOffset(Instant.ofEpochMilli(nextRoundingValue + 1)).getTotalSeconds();
+            if (offsetRounded == offsetNextValue) {
+                assertThat("unit interval width not as expected for [" + unit + "], [" + tz + "] at "
+                    + Instant.ofEpochMilli(roundedDate), nextRoundingValue - roundedDate, equalTo(unitMillis));
             }
+        }
 
-            // Round a whole bunch of dates and make sure they line up with the known good java time implementation
-            Rounding.Prepared javaTimeRounding = rounding.prepareJavaTime();
-            for (int d = 0; d < 1000; d++) {
-                date = dateBetween(bounds[0], bounds[1]);
-                long javaRounded = javaTimeRounding.round(date);
-                long esRounded = prepared.round(date);
-                if (javaRounded != esRounded) {
-                    fail("Expected [" + rounding + "] to round [" + Instant.ofEpochMilli(date) + "] to ["
-                            + Instant.ofEpochMilli(javaRounded) + "] but instead rounded to [" + Instant.ofEpochMilli(esRounded) + "]");
-                }
-                long javaNextRoundingValue = javaTimeRounding.nextRoundingValue(date);
-                long esNextRoundingValue = prepared.nextRoundingValue(date);
-                if (javaNextRoundingValue != esNextRoundingValue) {
-                    fail("Expected [" + rounding + "] to round [" + Instant.ofEpochMilli(date) + "] to ["
-                            + Instant.ofEpochMilli(esRounded) + "] and nextRoundingValue to be ["
-                            + Instant.ofEpochMilli(javaNextRoundingValue) + "] but instead was to ["
-                            + Instant.ofEpochMilli(esNextRoundingValue) + "]");
-                }
+        // Round a whole bunch of dates and make sure they line up with the known good java time implementation
+        Rounding.Prepared javaTimeRounding = rounding.prepareJavaTime();
+        for (int d = 0; d < 1000; d++) {
+            date = dateBetween(start, end);
+            long javaRounded = javaTimeRounding.round(date);
+            long esRounded = prepared.round(date);
+            if (javaRounded != esRounded) {
+                fail("Expected [" + rounding + "] to round [" + Instant.ofEpochMilli(date) + "] to ["
+                        + Instant.ofEpochMilli(javaRounded) + "] but instead rounded to [" + Instant.ofEpochMilli(esRounded) + "]");
+            }
+            long javaNextRoundingValue = javaTimeRounding.nextRoundingValue(date);
+            long esNextRoundingValue = prepared.nextRoundingValue(date);
+            if (javaNextRoundingValue != esNextRoundingValue) {
+                fail("Expected [" + rounding + "] to round [" + Instant.ofEpochMilli(date) + "] to ["
+                        + Instant.ofEpochMilli(esRounded) + "] and nextRoundingValue to be ["
+                        + Instant.ofEpochMilli(javaNextRoundingValue) + "] but instead was to ["
+                        + Instant.ofEpochMilli(esNextRoundingValue) + "]");
             }
         }
     }
@@ -715,6 +724,70 @@ public class RoundingTests extends ESTestCase {
         }
     }
 
+    /**
+     * Tests for DST transitions that cause the rounding to jump "backwards" because they round
+     * from one back to the previous day. Usually these rounding start before 
+     */
+    public void testForwardsBackwardsTimeZones() {
+        for (String zoneId : JAVA_ZONE_IDS) {
+            ZoneId tz = ZoneId.of(zoneId);
+            ZoneRules rules = tz.getRules();
+            for (ZoneOffsetTransition transition : rules.getTransitions()) {
+                checkForForwardsBackwardsTransition(tz, transition);
+            }
+            int firstYear;
+            if (rules.getTransitions().isEmpty()) {
+                // Pick an arbitrary year to start the range
+                firstYear = 1999;
+            } else {
+                ZoneOffsetTransition lastTransition = rules.getTransitions().get(rules.getTransitions().size() - 1);
+                firstYear = lastTransition.getDateTimeAfter().getYear() + 1;
+            }
+            // Pick an arbitrary year to end the range too
+            int lastYear = 2050;
+            int year = randomFrom(firstYear, lastYear);
+            for (ZoneOffsetTransitionRule transitionRule : rules.getTransitionRules()) {
+                ZoneOffsetTransition transition = transitionRule.createTransition(year);
+                checkForForwardsBackwardsTransition(tz, transition);
+            }
+        }
+    }
+
+    private void checkForForwardsBackwardsTransition(ZoneId tz, ZoneOffsetTransition transition) {
+        if (transition.getDateTimeBefore().getYear() < 1950) {
+            // We don't support transitions far in the past at all
+            return;
+        }
+        if (false == transition.isOverlap()) {
+            // Only overlaps cause the array rounding to have trouble
+            return;
+        }
+        if (transition.getDateTimeBefore().getDayOfMonth() == transition.getDateTimeAfter().getDayOfMonth()) {
+            // Only when the rounding changes the day
+            return;
+        }
+        if (transition.getDateTimeBefore().getMinute() == 0) {
+            // But roundings that change *at* midnight are safe because they don't "jump" to the next day.
+            return;
+        }
+        logger.info(
+            "{} from {}{} to {}{}",
+            tz,
+            transition.getDateTimeBefore(),
+            transition.getOffsetBefore(),
+            transition.getDateTimeAfter(),
+            transition.getOffsetAfter()
+        );
+        long millisSinceEpoch = TimeUnit.SECONDS.toMillis(transition.toEpochSecond());
+        long twoHours = TimeUnit.HOURS.toMillis(2);
+        assertUnitRoundingSameAsJavaUtilTimeImplementation(
+            Rounding.DateTimeUnit.DAY_OF_MONTH,
+            tz,
+            millisSinceEpoch - twoHours,
+            millisSinceEpoch + twoHours
+        );
+    }
+
     /**
      * tests for dst transition with overlaps and day roundings.
      */
@@ -957,6 +1030,11 @@ public class RoundingTests extends ESTestCase {
             return t <= time("2010-03-03T00:00:00Z")
                 || t >= time("2010-03-07T00:00:00Z");
         }
+        if (tz.getId().equals("Pacific/Guam") || tz.getId().equals("Pacific/Saipan")) {
+            // Clocks went back at 00:01 in 1969, causing overlapping days.
+            return t <= time("1969-01-25T00:00:00Z")
+                || t >= time("1969-01-26T00:00:00Z");  
+        }
 
         return true;
     }
@@ -965,8 +1043,13 @@ public class RoundingTests extends ESTestCase {
         return Math.abs(randomLong() % (2 * (long) 10e11)); // 1970-01-01T00:00:00Z - 2033-05-18T05:33:20.000+02:00
     }
 
-    private static long[] randomDateBounds() {
+    private static long[] randomDateBounds(Rounding.DateTimeUnit unit) {
         long b1 = randomDate();
+        if (randomBoolean()) {
+            // Sometimes use a fairly close date
+            return new long[] {b1, b1 + unit.extraLocalOffsetLookup() * between(1, 40)};
+        }
+        // Otherwise use a totally random date
         long b2 = randomValueOtherThan(b1, RoundingTests::randomDate);
         if (b1 < b2) {
             return new long[] {b1, b2};