### Elasticsearch Version

8.6.2

### Installed Plugins

_No response_

### Java Version

_bundled_

### OS Version

Linux 20.04.2-Ubuntu

### Problem Description

Continuous tranform giving invalid result when date_histogram is used in pivot 

### Steps to Reproduce

// Step 1 Add ingest pipeline
PUT /_ingest/pipeline/es-timeadd_1?pretty
{
  "processors": [
    {
      "set": {
        "field": "_source.es_timestamp",
        "value": "{{_ingest.timestamp}}"
      }
    }
  ]
}

// Step2 Create index using above pipeline
PUT /test
{
  "settings": {
    "index.default_pipeline" : "es-timeadd"
  }
}

// Step3 Insert two documents
PUT /test/_doc/1
{
  "id" : "ritesh",
  "gen_timestamp_min" : "2023-06-21T15:54:00.000Z",
  "flow_count" : 5
}

PUT /test/_doc/2
{
  "id" : "ritesh",
  "gen_timestamp_min" : "2023-06-21T15:55:00.000Z",
  "flow_count" : 5
}

// Step4 - Create transform
PUT /_transform/test-transform
{
  "source": {
    "index": "test"
  },
  "dest" : {
    "index" : "test-transformed-index"
  },
  "sync": {
    "time": {
      "field": "es_timestamp",
      "delay": "0s"
    }
  },
  "pivot": {
    "group_by": {
      "gen_timestamp_hour": {
        "date_histogram": {
          "field": "gen_timestamp_min",
          "fixed_interval": "1h"
        }
      },
      "asset": {
        "terms": {
          "field": "id.keyword"
        }
      }
    },
    "aggregations": {
      "flow_count": {
        "sum": {
          "field": "flow_count"
        }
      }
    }
  }
}

// Step 5 - Start Transform
POST /_transform/test-transform/_start

// Wait for few seconds
// Step6 - Check data in destination index
GET test-transformed-index/_search
{
}

We get below output, In which sum of flow counts is 10 which is valid
{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "test-transformed-index",
        "_id": "cgAnWNlrvx_jB6WvSZa9f0n5AAAAAAAA",
        "_score": 1,
        "_source": {
          "gen_timestamp_hour": "2023-06-21T15:00:00.000Z",
          "flow_countNAME": 10,
          "asset": "ritesh"
        }
      }
    ]
  }
}

// Step 7 Update document in source index with id 1, as below
PUT /test/_doc/1
{
  "id" : "ritesh",
  "gen_timestamp_min" : "2023-06-21T15:54:00.000Z",
  "flow_count" : 6
}

// Step 8 Check data in output index, Total flow count should be 11, instead we see only 6 as below
GET test-transformed-index/_search
{
  
}

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "test-transformed-index",
        "_type" : "_doc",
        "_id" : "cgAnWNlrvx_jB6WvSZa9f0n5AAAAAAAA",
        "_score" : 1.0,
        "_source" : {
          "gen_timestamp_hour" : "2023-06-21T15:00:00.000Z",
          "flow_countNAME" : 6.0,
          "asset" : "ritesh"
        }
      }
    ]
  }
}

Root Cause - Which reprocessing in continuous tranform, It picks data upto gen_timestamp_min of modified document, It should process all the data in given group by.




### Logs (if relevant)

_No response_
x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/pivot/CompositeBucketsChangeCollectorTests.java
/**
     * Simple unit tests to check that a field collector has been implemented for any single source type
     */
    public void testCreateFieldCollector() {
        GroupConfig groupConfig = GroupConfigTests.randomGroupConfig();
        Map<String, FieldCollector> changeCollector = CompositeBucketsChangeCollector.createFieldCollectors(
            groupConfig.getGroups(),
            randomBoolean() ? randomAlphaOfLengthBetween(3, 10) : null
        );
        assertNotNull(changeCollector);
    }
    public void testPageSize() throws IOException {
        Map<String, SingleGroupSource> groups = new LinkedHashMap<>();
        
        SingleGroupSource termsGroupBy = TermsGroupSourceTests.randomTermsGroupSourceNoScript();
        groups.put("terms", termsGroupBy);
        ChangeCollector collector = CompositeBucketsChangeCollector.buildChangeCollector(groups, null);
        assertEquals(1_000, getCompositeAggregationBuilder(collector.buildChangesQuery(new SearchSourceBuilder(), null, 1_000)).size());
        assertEquals(10_000, getCompositeAggregationBuilder(collector.buildChangesQuery(new SearchSourceBuilder(), null, 10_000)).size());
        assertEquals(10, getCompositeAggregationBuilder(collector.buildChangesQuery(new SearchSourceBuilder(), null, 10)).size());
        assertEquals(65536, getCompositeAggregationBuilder(collector.buildChangesQuery(new SearchSourceBuilder(), null, 100_000)).size());
        assertEquals(
            65536,
            getCompositeAggregationBuilder(collector.buildChangesQuery(new SearchSourceBuilder(), null, Integer.MAX_VALUE)).size()
        );