[TSDB] Cache rollup bucket timestamp to reduce rounding cost (#88420)

Author: weizijun

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/v2/RollupShardIndexer.java

@@ -209,6 +209,7 @@ private class TimeSeriesBucketCollector extends BucketCollector {
         private long bucketsCreated;
         private final RollupBucketBuilder rollupBucketBuilder = new RollupBucketBuilder();
         long lastTimestamp = Long.MAX_VALUE;
+        long lastHistoTimestamp = Long.MAX_VALUE;
         BytesRef lastTsid = null;
 
         TimeSeriesBucketCollector(BulkProcessor bulkProcessor) {
@@ -232,14 +233,18 @@ public void collect(int docId, long owningBucketOrd) throws IOException {
                     final BytesRef tsid = aggCtx.getTsid();
                     assert tsid != null : "Document without [" + TimeSeriesIdFieldMapper.NAME + "] field was found.";
                     final long timestamp = aggCtx.getTimestamp();
-                    final long histoTimestamp = rounding.round(timestamp);
+
+                    boolean tsidChanged = tsid.equals(rollupBucketBuilder.tsid()) == false;
+                    if (tsidChanged || timestamp < lastHistoTimestamp) {
+                        lastHistoTimestamp = rounding.round(timestamp);
+                    }
 
                     logger.trace(
                         "Doc: [{}] - _tsid: [{}], @timestamp: [{}}] -> rollup bucket ts: [{}]",
                         docId,
                         DocValueFormat.TIME_SERIES_ID.format(tsid),
                         timestampFormat.format(timestamp),
-                        timestampFormat.format(histoTimestamp)
+                        timestampFormat.format(lastHistoTimestamp)
                     );
 
                     /*
@@ -262,15 +267,15 @@ public void collect(int docId, long owningBucketOrd) throws IOException {
                     lastTsid = BytesRef.deepCopyOf(tsid);
                     lastTimestamp = timestamp;
 
-                    if (tsid.equals(rollupBucketBuilder.tsid()) == false || rollupBucketBuilder.timestamp() != histoTimestamp) {
+                    if (tsidChanged || rollupBucketBuilder.timestamp() != lastHistoTimestamp) {
                         // Flush rollup doc if not empty
                         if (rollupBucketBuilder.isEmpty() == false) {
                             Map<String, Object> doc = rollupBucketBuilder.buildRollupDocument();
                             indexBucket(doc);
                         }
 
                         // Create new rollup bucket
-                        rollupBucketBuilder.init(tsid, histoTimestamp);
+                        rollupBucketBuilder.init(tsid, lastHistoTimestamp);
                         bucketsCreated++;
                     }