Fix transform incorrectly calculating date bucket on updating old data (#97992)

This pull request fixes #97101, where a continuous pivot transform was incorrectly calculating a metric aggregation on a date bucket when a document from that bucket was updated. This was due to the transform taking the updated document's timestamp field as an upper bound in the transforms optimization query and, consequently, ignoring all documents with a later timestamp in the updated metric aggregation calculation for the bucket.

contributed by @alyokaz 
closes #97101

docs/changelog/97992.yaml

@@ -0,0 +1,6 @@
+pr: 97401
+summary: Fix transform incorrectly calculating date bucket on updating old data
+area: Transform
+type: bug
+issues:
+ - 97101

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformPivotRestIT.java

@@ -1000,6 +1000,126 @@ public class TransformPivotRestIT extends TransformRestTestCase {
         Map<String, Object> indexStats = getAsMap(transformIndex + "/_stats");
         assertEquals(104, XContentMapValues.extractValue("_all.total.docs.count", indexStats));
         assertOnePivotValue(transformIndex + "/_search?q=by_hr:1484499600000", 4.0833333333);
+
+    }
+
+    // test that docs in same date bucket with a later date than the updated doc are not ignored by the transform.
+    @SuppressWarnings("unchecked")
+    public void testContinuousDateHistogramPivot() throws Exception {
+        String indexName = "continuous_reviews_date_histogram";
+
+        // ingest timestamp used to allow grouped on timestamp field to remain the same on update
+        Request createIngestPipeLine = new Request("PUT", "/_ingest/pipeline/es-timeadd");
+        createIngestPipeLine.setJsonEntity("""
+            {
+              "processors":[
+                    {
+                       "set":{
+                          "field":"_source.es_timestamp",
+                          "value":"{{_ingest.timestamp}}"
+                       }
+                    }
+                 ]
+            }""");
+        client().performRequest(createIngestPipeLine);
+
+        Request setDefaultPipeline = new Request("PUT", indexName);
+        setDefaultPipeline.setJsonEntity("""
+            {
+               "settings":{
+                  "index.default_pipeline":"es-timeadd"
+               }
+            }""");
+        client().performRequest(setDefaultPipeline);
+
+        Request createIndex = new Request("PUT", indexName + "/_doc/1");
+        createIndex.setJsonEntity("""
+            {
+                "user_id" : "user_1",
+                "timestamp": "2023-07-24T17:10:00.000Z",
+                "stars": 5
+            }""");
+        client().performRequest(createIndex);
+
+        var putRequest = new Request("PUT", indexName + "/_doc/2");
+        putRequest.setJsonEntity("""
+            {
+                "user_id" : "user_2",
+                "timestamp": "2023-07-24T17:55:00.000Z",
+                "stars": 5
+            }""");
+        client().performRequest(putRequest);
+
+        String transformId = "continuous_date_histogram_pivot";
+        String transformIndex = "pivot_reviews_via_date_continuous_histogram";
+        setupDataAccessRole(DATA_ACCESS_ROLE, indexName, transformIndex);
+
+        final Request createTransformRequest = createRequestWithAuth(
+            "PUT",
+            getTransformEndpoint() + transformId,
+            BASIC_AUTH_VALUE_TRANSFORM_ADMIN_WITH_SOME_DATA_ACCESS
+        );
+
+        String config = Strings.format("""
+            {
+              "source": {
+                "index": "%s"
+              },
+              "dest": {
+                "index": "%s"
+              },
+              "frequency": "1s",
+              "sync": {
+                "time": {
+                  "field": "es_timestamp",
+                  "delay": "1s"
+                }
+              },
+              "pivot": {
+                "group_by": {
+                  "by_hr": {
+                    "date_histogram": {
+                      "fixed_interval": "1h",
+                      "field": "timestamp"
+                    }
+                  }
+                },
+                "aggregations": {
+                  "total_rating": {
+                    "sum": {
+                      "field": "stars"
+                    }
+                  }
+                }
+              }
+            }""", indexName, transformIndex);
+
+        createTransformRequest.setJsonEntity(config);
+        var createTransformResponse = entityAsMap(client().performRequest(createTransformRequest));
+        assertThat(createTransformResponse.get("acknowledged"), equalTo(Boolean.TRUE));
+
+        startAndWaitForContinuousTransform(transformId, transformIndex, null);
+        assertTrue(indexExists(transformIndex));
+
+        // update stars field in first doc
+        Request updateDoc = new Request("PUT", indexName + "/_doc/1");
+        updateDoc.setJsonEntity("""
+            {
+                "user_id" : "user_1",
+                "timestamp": "2023-07-24T17:10:00.000Z",
+                "stars": 6
+            }""");
+        updateDoc.addParameter("refresh", "true");
+        client().performRequest(updateDoc);
+
+        waitForTransformCheckpoint(transformId, 2);
+        stopTransform(transformId, false);
+        refreshIndex(transformIndex);
+
+        var searchResponse = getAsMap(transformIndex + "/_search");
+        var hits = ((List<Map<String, Object>>) XContentMapValues.extractValue("hits.hits", searchResponse)).get(0);
+        var totalStars = (double) XContentMapValues.extractValue("_source.total_rating", hits);
+        assertEquals(11, totalStars, 0);
     }
 
     @SuppressWarnings("unchecked")

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/pivot/CompositeBucketsChangeCollector.java

@@ -408,7 +408,7 @@ public class CompositeBucketsChangeCollector implements ChangeCollector {
             if (lowerBoundResult != null && upperBoundResult != null) {
                 // we only need to round the lower bound, because the checkpoint will not contain new data for the upper bound
                 lowerBound = rounding.round((long) lowerBoundResult.value());
-                upperBound = (long) upperBoundResult.value();
+                upperBound = rounding.nextRoundingValue((long) upperBoundResult.value());
 
                 return false;
             }

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/pivot/CompositeBucketsChangeCollectorTests.java

@@ -11,17 +11,13 @@ import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchResponseSections;
 import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.index.query.QueryBuilder;
-import org.elasticsearch.index.query.RangeQueryBuilder;
 import org.elasticsearch.index.query.TermsQueryBuilder;
 import org.elasticsearch.search.aggregations.Aggregations;
 import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregation;
 import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregationBuilder;
-import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
-import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation.SingleValue;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
-import org.elasticsearch.xpack.core.transform.transforms.pivot.DateHistogramGroupSource;
 import org.elasticsearch.xpack.core.transform.transforms.pivot.GeoTileGroupSourceTests;
 import org.elasticsearch.xpack.core.transform.transforms.pivot.GroupConfig;
 import org.elasticsearch.xpack.core.transform.transforms.pivot.GroupConfigTests;
@@ -35,7 +31,6 @@ import org.mockito.stubbing.Answer;
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -43,7 +38,6 @@ import java.util.Map;
 
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -132,121 +126,6 @@ public class CompositeBucketsChangeCollectorTests extends ESTestCase {
         assertThat(((TermsQueryBuilder) queryBuilder).values(), containsInAnyOrder("id1", "id2", "id3"));
     }
 
-    public void testDateHistogramFieldCollector() throws IOException {
-        Map<String, SingleGroupSource> groups = new LinkedHashMap<>();
-
-        SingleGroupSource groupBy = new DateHistogramGroupSource(
-            "timestamp",
-            null,
-            false,
-            new DateHistogramGroupSource.FixedInterval(DateHistogramInterval.MINUTE),
-            null,
-            null
-        );
-        groups.put("output_timestamp", groupBy);
-
-        ChangeCollector collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, "timestamp");
-
-        QueryBuilder queryBuilder = collector.buildFilterQuery(
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 66_666L),
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 200_222L)
-        );
-
-        assertNotNull(queryBuilder);
-        assertThat(queryBuilder, instanceOf(RangeQueryBuilder.class));
-        // rounded down
-        assertThat(((RangeQueryBuilder) queryBuilder).from(), equalTo(Long.valueOf(60_000)));
-        assertTrue(((RangeQueryBuilder) queryBuilder).includeLower());
-        assertThat(((RangeQueryBuilder) queryBuilder).fieldName(), equalTo("timestamp"));
-
-        // timestamp field does not match
-        collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, "sync_timestamp");
-
-        SingleValue minTimestamp = mock(SingleValue.class);
-        when(minTimestamp.getName()).thenReturn("_transform_change_collector.output_timestamp.min");
-        when(minTimestamp.value()).thenReturn(122_633.0);
-
-        SingleValue maxTimestamp = mock(SingleValue.class);
-        when(maxTimestamp.getName()).thenReturn("_transform_change_collector.output_timestamp.max");
-        when(maxTimestamp.value()).thenReturn(302_523.0);
-
-        // simulate the agg response, that should inject
-        Aggregations aggs = new Aggregations(Arrays.asList(minTimestamp, maxTimestamp));
-        SearchResponseSections sections = new SearchResponseSections(null, aggs, null, false, null, null, 1);
-        SearchResponse response = new SearchResponse(sections, null, 1, 1, 0, 0, ShardSearchFailure.EMPTY_ARRAY, null);
-        collector.processSearchResponse(response);
-
-        // provide checkpoints, although they don't matter in this case
-        queryBuilder = collector.buildFilterQuery(
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 66_666L),
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 200_222L)
-        );
-
-        assertNotNull(queryBuilder);
-        assertThat(queryBuilder, instanceOf(RangeQueryBuilder.class));
-        // rounded down
-        assertThat(((RangeQueryBuilder) queryBuilder).from(), equalTo(Long.valueOf(120_000)));
-        assertTrue(((RangeQueryBuilder) queryBuilder).includeLower());
-        // the upper bound is not rounded
-        assertThat(((RangeQueryBuilder) queryBuilder).to(), equalTo(Long.valueOf(302_523)));
-        assertTrue(((RangeQueryBuilder) queryBuilder).includeUpper());
-        assertThat(((RangeQueryBuilder) queryBuilder).fieldName(), equalTo("timestamp"));
-
-        // field does not match, but output field equals sync field
-        collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, "output_timestamp");
-
-        when(minTimestamp.getName()).thenReturn("_transform_change_collector.output_timestamp.min");
-        when(minTimestamp.value()).thenReturn(242_633.0);
-
-        when(maxTimestamp.getName()).thenReturn("_transform_change_collector.output_timestamp.max");
-        when(maxTimestamp.value()).thenReturn(602_523.0);
-
-        // simulate the agg response, that should inject
-        collector.processSearchResponse(response);
-        queryBuilder = collector.buildFilterQuery(
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 66_666L),
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 200_222L)
-        );
-
-        assertNotNull(queryBuilder);
-
-        assertThat(queryBuilder, instanceOf(RangeQueryBuilder.class));
-        // rounded down
-        assertThat(((RangeQueryBuilder) queryBuilder).from(), equalTo(Long.valueOf(240_000)));
-        assertTrue(((RangeQueryBuilder) queryBuilder).includeLower());
-        // the upper bound is not rounded
-        assertThat(((RangeQueryBuilder) queryBuilder).to(), equalTo(Long.valueOf(602_523)));
-        assertTrue(((RangeQueryBuilder) queryBuilder).includeUpper());
-        assertThat(((RangeQueryBuilder) queryBuilder).fieldName(), equalTo("timestamp"));
-
-        // missing bucket disables optimization
-        groupBy = new DateHistogramGroupSource(
-            "timestamp",
-            null,
-            true,
-            new DateHistogramGroupSource.FixedInterval(DateHistogramInterval.MINUTE),
-            null,
-            null
-        );
-        groups.put("output_timestamp", groupBy);
-
-        collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, "timestamp");
-
-        queryBuilder = collector.buildFilterQuery(
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 66_666L),
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 200_222L)
-        );
-        assertNull(queryBuilder);
-
-        collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, "sync_timestamp");
-
-        queryBuilder = collector.buildFilterQuery(
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 66_666L),
-            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 200_222L)
-        );
-        assertNull(queryBuilder);
-    }
-
     public void testNoTermsFieldCollectorForScripts() throws IOException {
         Map<String, SingleGroupSource> groups = new LinkedHashMap<>();
 

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/pivot/DateHistogramFieldCollectorTests.java

@@ -0,0 +1,178 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.transform.transforms.pivot;
+
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchResponseSections;
+import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.RangeQueryBuilder;
+import org.elasticsearch.search.aggregations.Aggregations;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
+import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation.SingleValue;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
+import org.elasticsearch.xpack.core.transform.transforms.pivot.DateHistogramGroupSource;
+import org.elasticsearch.xpack.core.transform.transforms.pivot.SingleGroupSource;
+import org.elasticsearch.xpack.transform.transforms.Function.ChangeCollector;
+import org.junit.Before;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class DateHistogramFieldCollectorTests extends ESTestCase {
+    private Map<String, SingleGroupSource> groups;
+
+    private SingleValue minTimestamp;
+    private SingleValue maxTimestamp;
+
+    private static final String TIMESTAMP = "timestamp";
+    private static final String OUTPUT_TIMESTAMP = "output_timestamp";
+    private static final String SYNC_TIMESTAMP = "sync_timestamp";
+
+    private static final SingleGroupSource groupBy = new DateHistogramGroupSource(
+        TIMESTAMP,
+        null,
+        false,
+        new DateHistogramGroupSource.FixedInterval(DateHistogramInterval.MINUTE),
+        null,
+        null
+    );
+
+    private static final double MIN_TIMESTAMP_VALUE = 122_633;
+    private static final double MAX_TIMESTAMP_VALUE = 302_525;
+    private static final double EXPECTED_LOWER_BOUND = 120_000;
+
+    private static final double EXPECTED_UPPER_BOUND = 360_000;
+
+    @Before
+    public void setupDateHistogramFieldCollectorTest() {
+        minTimestamp = mock(NumericMetricsAggregation.SingleValue.class);
+        maxTimestamp = mock(NumericMetricsAggregation.SingleValue.class);
+
+        when(minTimestamp.getName()).thenReturn("_transform_change_collector.output_timestamp.min");
+        when(maxTimestamp.getName()).thenReturn("_transform_change_collector.output_timestamp.max");
+        when(minTimestamp.value()).thenReturn(MIN_TIMESTAMP_VALUE);
+        when(maxTimestamp.value()).thenReturn(MAX_TIMESTAMP_VALUE);
+
+        groups = new HashMap<>();
+    }
+
+    public void testWhenFieldAndSyncFieldSame() {
+        groups.put(OUTPUT_TIMESTAMP, groupBy);
+        ChangeCollector collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, TIMESTAMP);
+        QueryBuilder queryBuilder = buildFilterQuery(collector);
+
+        assertQuery(queryBuilder, 60_000.0, TIMESTAMP);
+    }
+
+    public void testWhenFieldAndSyncFieldDifferent() {
+        groups.put(OUTPUT_TIMESTAMP, groupBy);
+        ChangeCollector collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, SYNC_TIMESTAMP);
+
+        // simulate the agg response, that should inject
+        SearchResponse response = buildSearchResponse(minTimestamp, maxTimestamp);
+        collector.processSearchResponse(response);
+
+        // checkpoints are provided although are not used in this case
+        QueryBuilder queryBuilder = buildFilterQuery(collector);
+
+        assertQuery(queryBuilder, EXPECTED_LOWER_BOUND, EXPECTED_UPPER_BOUND, TIMESTAMP);
+    }
+
+    public void testWhenOutputAndSyncFieldSame() {
+        groups.put(OUTPUT_TIMESTAMP, groupBy);
+        ChangeCollector collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, SYNC_TIMESTAMP);
+
+        // simulate the agg response, that should inject
+        SearchResponse response = buildSearchResponse(minTimestamp, maxTimestamp);
+        collector.processSearchResponse(response);
+        QueryBuilder queryBuilder = buildFilterQuery(collector);
+
+        assertQuery(queryBuilder, EXPECTED_LOWER_BOUND, EXPECTED_UPPER_BOUND, TIMESTAMP);
+    }
+
+    public void testMissingBucketDisablesOptimization() {
+        // missing bucket disables optimization
+        DateHistogramGroupSource groupBy = new DateHistogramGroupSource(
+            TIMESTAMP,
+            null,
+            true,
+            new DateHistogramGroupSource.FixedInterval(DateHistogramInterval.MINUTE),
+            null,
+            null
+        );
+        groups.put(OUTPUT_TIMESTAMP, groupBy);
+
+        // field and sync_field are the same
+        ChangeCollector collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, TIMESTAMP);
+        QueryBuilder queryBuilder = buildFilterQuery(collector);
+
+        assertNull(queryBuilder);
+
+        // field and sync_field are different
+        collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, SYNC_TIMESTAMP);
+        queryBuilder = buildFilterQuery(collector);
+
+        assertNull(queryBuilder);
+    }
+
+    private static void assertQuery(
+        QueryBuilder queryBuilder,
+        Double expectedLowerBound,
+        Double expectedUpperBound,
+        String expectedFieldName
+    ) {
+        assertQuery(queryBuilder, expectedLowerBound, expectedFieldName);
+
+        // the upper bound is rounded up to the nearest time unit
+        assertThat(((RangeQueryBuilder) queryBuilder).to(), equalTo(expectedUpperBound.longValue()));
+        assertTrue(((RangeQueryBuilder) queryBuilder).includeUpper());
+    }
+
+    private static void assertQuery(QueryBuilder queryBuilder, Double expectedLowerBound, String expectedFieldName) {
+        assertNotNull(queryBuilder);
+        assertThat(queryBuilder, instanceOf(RangeQueryBuilder.class));
+
+        // lower bound is rounded down to the nearest time unit
+        assertThat(((RangeQueryBuilder) queryBuilder).from(), equalTo(expectedLowerBound.longValue()));
+        assertTrue(((RangeQueryBuilder) queryBuilder).includeLower());
+
+        assertThat(((RangeQueryBuilder) queryBuilder).fieldName(), equalTo(expectedFieldName));
+    }
+
+    // Util methods
+    private static QueryBuilder buildFilterQuery(ChangeCollector collector) {
+        return collector.buildFilterQuery(
+            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 66_666L),
+            new TransformCheckpoint("t_id", 42L, 42L, Collections.emptyMap(), 200_222L)
+        );
+    }
+
+    private static SearchResponse buildSearchResponse(SingleValue minTimestamp, SingleValue maxTimestamp) {
+        SearchResponseSections sections = new SearchResponseSections(
+            null,
+            new Aggregations(Arrays.asList(minTimestamp, maxTimestamp)),
+            null,
+            false,
+            null,
+            null,
+            1
+        );
+        return new SearchResponse(sections, null, 1, 1, 0, 0, ShardSearchFailure.EMPTY_ARRAY, null);
+    }
+
+}