[ 8.12] Downsampling supports date_histogram with tz (#104390)

* Port pr-103511 to branch 8.12

* mark changed transport version as patch

docs/changelog/103511.yaml

@@ -0,0 +1,6 @@
+pr: 103511
+summary: Downsampling supports `date_histogram` with tz
+area: Downsampling
+type: bug
+issues:
+ - 101309

docs/reference/data-streams/downsampling.asciidoc

@@ -135,7 +135,29 @@ downsampled.
 * For
 <<search-aggregations-bucket-datehistogram-aggregation,date histogram aggregations>>,
 only `fixed_intervals` (and not calendar-aware intervals) are supported.
-* Only Coordinated Universal Time (UTC) date-times are supported.
+* Timezone support comes with caveats:
+
+** Date histograms at intervals that are multiples of an hour are based on
+values generated at UTC. This works well for timezones that are on the hour, e.g.
++5:00 or -3:00, but requires offsetting the reported time buckets, e.g.
+`2020-01-01T10:30:00.000` instead of `2020-03-07T10:00:00.000` for
+timezone +5:30 (India), if downsampling aggregates values per hour. In this case,
+the results include the field `downsampled_results_offset: true`, to indicate that
+the time buckets are shifted. This can be avoided if a downsampling interval of 15
+minutes is used, as it allows properly calculating hourly values for the shifted
+buckets.
+
+** Date histograms at intervals that are multiples of a day are similarly
+affected, in case downsampling aggregates values per day. In this case, the
+beginning of each day is always calculated at UTC when generated the downsampled
+values, so the time buckets need to be shifted, e.g. reported as
+`2020-03-07T19:00:00.000` instead of `2020-03-07T00:00:00.000` for timezone `America/New_York`.
+The field `downsampled_results_offset: true` is added in this case too.
+
+** Daylight savings and similar peculiarities around timezones affect
+reported results, as <<datehistogram-aggregation-time-zone,documented>>
+for date histogram aggregation. Besides, downsampling at daily interval
+hinders tracking any information related to daylight savings changes.
 
 [discrete]
 [[downsampling-restrictions]]

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -191,6 +191,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion MISSED_INDICES_UPDATE_EXCEPTION_ADDED = def(8_558_00_0);
     public static final TransportVersion INFERENCE_SERVICE_EMBEDDING_SIZE_ADDED = def(8_559_00_0);
     public static final TransportVersion ENRICH_ELASTICSEARCH_VERSION_REMOVED = def(8_560_00_0);
+    public static final TransportVersion DATE_HISTOGRAM_SUPPORT_DOWNSAMPLED_TZ = def(8_560_00_1);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

@@ -26,7 +26,6 @@
 import org.elasticsearch.cluster.routing.allocation.IndexMetadataUpdater;
 import org.elasticsearch.cluster.routing.allocation.decider.DiskThresholdDecider;
 import org.elasticsearch.cluster.routing.allocation.decider.ShardsLimitAllocationDecider;
-import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -138,14 +137,9 @@ public class IndexMetadata implements Diffable<IndexMetadata>, ToXContentFragmen
         EnumSet.of(ClusterBlockLevel.WRITE)
     );
 
-    // TODO: refactor this method after adding more downsampling metadata
-    public boolean isDownsampledIndex() {
-        final String sourceIndex = settings.get(IndexMetadata.INDEX_DOWNSAMPLE_SOURCE_NAME_KEY);
-        final String indexDownsamplingStatus = settings.get(IndexMetadata.INDEX_DOWNSAMPLE_STATUS_KEY);
-        final boolean downsamplingSuccess = DownsampleTaskStatus.SUCCESS.name()
-            .toLowerCase(Locale.ROOT)
-            .equals(indexDownsamplingStatus != null ? indexDownsamplingStatus.toLowerCase(Locale.ROOT) : DownsampleTaskStatus.UNKNOWN);
-        return Strings.isNullOrEmpty(sourceIndex) == false && downsamplingSuccess;
+    @Nullable
+    public String getDownsamplingInterval() {
+        return settings.get(IndexMetadata.INDEX_DOWNSAMPLE_INTERVAL_KEY);
     }
 
     public enum State implements Writeable {
@@ -1235,6 +1229,7 @@ public Index getResizeSourceIndex() {
     public static final String INDEX_DOWNSAMPLE_ORIGIN_UUID_KEY = "index.downsample.origin.uuid";
 
     public static final String INDEX_DOWNSAMPLE_STATUS_KEY = "index.downsample.status";
+    public static final String INDEX_DOWNSAMPLE_INTERVAL_KEY = "index.downsample.interval";
     public static final Setting<String> INDEX_DOWNSAMPLE_SOURCE_UUID = Setting.simpleString(
         INDEX_DOWNSAMPLE_SOURCE_UUID_KEY,
         Property.IndexScope,
@@ -1277,6 +1272,14 @@ public String toString() {
         Property.InternalIndex
     );
 
+    public static final Setting<String> INDEX_DOWNSAMPLE_INTERVAL = Setting.simpleString(
+        INDEX_DOWNSAMPLE_INTERVAL_KEY,
+        "",
+        Property.IndexScope,
+        Property.InternalIndex,
+        Property.PrivateIndex
+    );
+
     // LIFECYCLE_NAME is here an as optimization, see LifecycleSettings.LIFECYCLE_NAME and
     // LifecycleSettings.LIFECYCLE_NAME_SETTING for the 'real' version
     public static final String LIFECYCLE_NAME = "index.lifecycle.name";

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -76,6 +76,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         IndexMetadata.INDEX_DOWNSAMPLE_ORIGIN_NAME,
         IndexMetadata.INDEX_DOWNSAMPLE_ORIGIN_UUID,
         IndexMetadata.INDEX_DOWNSAMPLE_STATUS,
+        IndexMetadata.INDEX_DOWNSAMPLE_INTERVAL,
         SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_DEBUG_SETTING,
         SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_WARN_SETTING,
         SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_INFO_SETTING,

server/src/main/java/org/elasticsearch/index/query/QueryRewriteContext.java

@@ -285,6 +285,13 @@ public IndexSettings getIndexSettings() {
         return indexSettings;
     }
 
+    /**
+     * Returns the MappingLookup for the queried index.
+     */
+    public MappingLookup getMappingLookup() {
+        return mappingLookup;
+    }
+
     /**
      *  Given an index pattern, checks whether it matches against the current shard. The pattern
      *  may represent a fully qualified index name if the search targets remote shards.

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateHistogramAggregationBuilder.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.mapper.DataStreamTimestampFieldMapper;
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
@@ -36,6 +37,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.SimpleTimeZone;
 import java.util.function.Consumer;
 
 import static java.util.Map.entry;
@@ -406,23 +408,46 @@ protected ValuesSourceAggregatorFactory innerBuild(
     ) throws IOException {
         final DateIntervalWrapper.IntervalTypeEnum dateHistogramIntervalType = dateHistogramInterval.getIntervalType();
 
-        if (context.getIndexSettings().getIndexMetadata().isDownsampledIndex()
-            && DateIntervalWrapper.IntervalTypeEnum.CALENDAR.equals(dateHistogramIntervalType)) {
-            throw new IllegalArgumentException(
-                config.getDescription()
-                    + " is not supported for aggregation ["
-                    + getName()
-                    + "] with interval type ["
-                    + dateHistogramIntervalType.getPreferredName()
-                    + "]"
-            );
-        }
-
+        boolean downsampledResultsOffset = false;
         final ZoneId tz = timeZone();
-        if (context.getIndexSettings().getIndexMetadata().isDownsampledIndex() && tz != null && ZoneId.of("UTC").equals(tz) == false) {
-            throw new IllegalArgumentException(
-                config.getDescription() + " is not supported for aggregation [" + getName() + "] with timezone [" + tz + "]"
-            );
+
+        String downsamplingInterval = context.getIndexSettings().getIndexMetadata().getDownsamplingInterval();
+        if (downsamplingInterval != null) {
+            if (DateIntervalWrapper.IntervalTypeEnum.CALENDAR.equals(dateHistogramIntervalType)) {
+                throw new IllegalArgumentException(
+                    config.getDescription()
+                        + " is not supported for aggregation ["
+                        + getName()
+                        + "] with interval type ["
+                        + dateHistogramIntervalType.getPreferredName()
+                        + "]"
+                );
+            }
+
+            // Downsampled data in time-series indexes contain aggregated values that get calculated over UTC-based intervals.
+            // When they get aggregated using a different timezone, the resulting buckets may need to be offset to account for
+            // the difference between UTC (where stored data refers to) and the requested timezone. For instance:
+            // a. A TZ shifted by -01:15 over hourly downsampled data will lead to buckets with times XX:45, instead of XX:00
+            // b. A TZ shifted by +07:00 over daily downsampled data will lead to buckets with times 07:00, instead of 00:00
+            // c. Intervals over DST are approximate, not including gaps in time buckets. This applies to date histogram aggregation in
+            // general.
+            if (tz != null && ZoneId.of("UTC").equals(tz) == false && field().equals(DataStreamTimestampFieldMapper.DEFAULT_PATH)) {
+
+                // Get the downsampling interval.
+                DateHistogramInterval interval = new DateHistogramInterval(downsamplingInterval);
+                long downsamplingResolution = interval.estimateMillis();
+                long aggregationResolution = dateHistogramInterval.getAsFixedInterval().estimateMillis();
+
+                // If the aggregation resolution is not a multiple of the downsampling resolution, the reported time for each
+                // bucket needs to be shifted by the mod - in addition to rounding that's applied as usual.
+                // Note that the aggregation resolution gets shifted to match the specified timezone. Timezone.getOffset() normally expects
+                // a date but it can also process an offset (interval) in milliseconds as it uses the Unix epoch for reference.
+                long aggregationOffset = SimpleTimeZone.getTimeZone(tz).getOffset(aggregationResolution) % downsamplingResolution;
+                if (aggregationOffset != 0) {
+                    downsampledResultsOffset = true;
+                    offset += aggregationOffset;
+                }
+            }
         }
 
         DateHistogramAggregationSupplier aggregatorSupplier = context.getValuesSourceRegistry().getAggregator(REGISTRY_KEY, config);
@@ -473,6 +498,7 @@ protected ValuesSourceAggregatorFactory innerBuild(
             order,
             keyed,
             minDocCount,
+            downsampledResultsOffset,
             rounding,
             roundedBounds,
             roundedHardBounds,

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateHistogramAggregationSupplier.java

@@ -29,6 +29,7 @@ Aggregator build(
         BucketOrder order,
         boolean keyed,
         long minDocCount,
+        boolean downsampledResultsOffset,
         @Nullable LongBounds extendedBounds,
         @Nullable LongBounds hardBounds,
         ValuesSourceConfig valuesSourceConfig,

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateHistogramAggregator.java

@@ -79,6 +79,7 @@ public static Aggregator build(
         BucketOrder order,
         boolean keyed,
         long minDocCount,
+        boolean downsampledResultsOffset,
         @Nullable LongBounds extendedBounds,
         @Nullable LongBounds hardBounds,
         ValuesSourceConfig valuesSourceConfig,
@@ -96,6 +97,7 @@ public static Aggregator build(
             order,
             keyed,
             minDocCount,
+            downsampledResultsOffset,
             extendedBounds,
             hardBounds,
             valuesSourceConfig,
@@ -115,6 +117,7 @@ public static Aggregator build(
             order,
             keyed,
             minDocCount,
+            downsampledResultsOffset,
             extendedBounds,
             hardBounds,
             valuesSourceConfig,
@@ -133,6 +136,7 @@ private static FromDateRange adaptIntoRangeOrNull(
         BucketOrder order,
         boolean keyed,
         long minDocCount,
+        boolean downsampledResultsOffset,
         @Nullable LongBounds extendedBounds,
         @Nullable LongBounds hardBounds,
         ValuesSourceConfig valuesSourceConfig,
@@ -191,6 +195,7 @@ private static FromDateRange adaptIntoRangeOrNull(
             minDocCount,
             extendedBounds,
             keyed,
+            downsampledResultsOffset,
             fixedRoundingPoints
         );
     }
@@ -227,6 +232,7 @@ private static RangeAggregator.Range[] ranges(LongBounds hardBounds, long[] fixe
     private final boolean keyed;
 
     private final long minDocCount;
+    private final boolean downsampledResultsOffset;
     private final LongBounds extendedBounds;
     private final LongBounds hardBounds;
 
@@ -240,6 +246,7 @@ private static RangeAggregator.Range[] ranges(LongBounds hardBounds, long[] fixe
         BucketOrder order,
         boolean keyed,
         long minDocCount,
+        boolean downsampledResultsOffset,
         @Nullable LongBounds extendedBounds,
         @Nullable LongBounds hardBounds,
         ValuesSourceConfig valuesSourceConfig,
@@ -255,6 +262,7 @@ private static RangeAggregator.Range[] ranges(LongBounds hardBounds, long[] fixe
         order.validate(this);
         this.keyed = keyed;
         this.minDocCount = minDocCount;
+        this.downsampledResultsOffset = downsampledResultsOffset;
         this.extendedBounds = extendedBounds;
         this.hardBounds = hardBounds;
         // TODO: Stop using null here
@@ -328,6 +336,7 @@ public InternalAggregation[] buildAggregations(long[] owningBucketOrds) throws I
                 emptyBucketInfo,
                 formatter,
                 keyed,
+                downsampledResultsOffset,
                 metadata()
             );
         });
@@ -347,6 +356,7 @@ public InternalAggregation buildEmptyAggregation() {
             emptyBucketInfo,
             formatter,
             keyed,
+            downsampledResultsOffset,
             metadata()
         );
     }
@@ -392,6 +402,7 @@ static class FromDateRange extends AdaptingAggregator implements SizedBucketAggr
         private final long minDocCount;
         private final LongBounds extendedBounds;
         private final boolean keyed;
+        private final boolean downsampledResultsOffset;
         private final long[] fixedRoundingPoints;
 
         FromDateRange(
@@ -405,6 +416,7 @@ static class FromDateRange extends AdaptingAggregator implements SizedBucketAggr
             long minDocCount,
             LongBounds extendedBounds,
             boolean keyed,
+            boolean downsampledResultsOffset,
             long[] fixedRoundingPoints
         ) throws IOException {
             super(parent, subAggregators, delegate);
@@ -416,6 +428,7 @@ static class FromDateRange extends AdaptingAggregator implements SizedBucketAggr
             this.minDocCount = minDocCount;
             this.extendedBounds = extendedBounds;
             this.keyed = keyed;
+            this.downsampledResultsOffset = downsampledResultsOffset;
             this.fixedRoundingPoints = fixedRoundingPoints;
         }
 
@@ -454,6 +467,7 @@ protected InternalAggregation adapt(InternalAggregation delegateResult) {
                 emptyBucketInfo,
                 format,
                 keyed,
+                downsampledResultsOffset,
                 range.getMetadata()
             );
         }

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/DateHistogramAggregatorFactory.java

@@ -52,6 +52,7 @@ public static void registerAggregators(ValuesSourceRegistry.Builder builder) {
                 order,
                 keyed,
                 minDocCount,
+                downsampledResultsOffset,
                 extendedBounds,
                 hardBounds,
                 valuesSourceConfig,
@@ -71,6 +72,7 @@ public static void registerAggregators(ValuesSourceRegistry.Builder builder) {
                     order,
                     keyed,
                     minDocCount,
+                    downsampledResultsOffset,
                     extendedBounds,
                     hardBounds,
                     valuesSourceConfig,
@@ -88,6 +90,7 @@ public static void registerAggregators(ValuesSourceRegistry.Builder builder) {
     private final BucketOrder order;
     private final boolean keyed;
     private final long minDocCount;
+    private final boolean downsampledResultsOffset;
     private final LongBounds extendedBounds;
     private final LongBounds hardBounds;
     private final Rounding rounding;
@@ -98,6 +101,7 @@ public DateHistogramAggregatorFactory(
         BucketOrder order,
         boolean keyed,
         long minDocCount,
+        boolean downsampledResultsOffset,
         Rounding rounding,
         LongBounds extendedBounds,
         LongBounds hardBounds,
@@ -111,6 +115,7 @@ public DateHistogramAggregatorFactory(
         this.aggregatorSupplier = aggregationSupplier;
         this.order = order;
         this.keyed = keyed;
+        this.downsampledResultsOffset = downsampledResultsOffset;
         this.minDocCount = minDocCount;
         this.extendedBounds = extendedBounds;
         this.hardBounds = hardBounds;
@@ -139,6 +144,7 @@ protected Aggregator doCreateInternal(Aggregator parent, CardinalityUpperBound c
             order,
             keyed,
             minDocCount,
+            downsampledResultsOffset,
             extendedBounds,
             hardBounds,
             config,
@@ -159,6 +165,7 @@ protected Aggregator createUnmapped(Aggregator parent, Map<String, Object> metad
             order,
             keyed,
             minDocCount,
+            downsampledResultsOffset,
             extendedBounds,
             hardBounds,
             config,