[TSDB] Improve downsampling performance by using tsid ordinals (#90088)

This PR modifies downsampling operation so that it uses global ordinal to track tsid changes

PR depends on the work done in #90035

Relates to #74660

Author: csoulios

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

@@ -254,9 +254,10 @@ public LeafBucketCollector getLeafCollector(final AggregationExecutionContext ag
                 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 int tsidOrd = aggCtx.getTsidOrd();
                     final long timestamp = aggCtx.getTimestamp();
 
-                    boolean tsidChanged = tsid.equals(rollupBucketBuilder.tsid()) == false;
+                    boolean tsidChanged = tsidOrd != rollupBucketBuilder.tsidOrd();
                     if (tsidChanged || timestamp < lastHistoTimestamp) {
                         lastHistoTimestamp = Math.max(
                             rounding.round(timestamp),
@@ -303,7 +304,7 @@ public void collect(int docId, long owningBucketOrd) throws IOException {
 
                         // Create new rollup bucket
                         if (tsidChanged) {
-                            rollupBucketBuilder.resetTsid(tsid, lastHistoTimestamp);
+                            rollupBucketBuilder.resetTsid(tsid, tsidOrd, lastHistoTimestamp);
                         } else {
                             rollupBucketBuilder.resetTimestamp(lastHistoTimestamp);
                         }
@@ -370,6 +371,7 @@ public ScoreMode scoreMode() {
 
     private class RollupBucketBuilder {
         private BytesRef tsid;
+        private int tsidOrd = -1;
         private long timestamp;
         private int docCount;
         private final Map<String, MetricFieldProducer> metricFieldProducers;
@@ -383,8 +385,9 @@ private class RollupBucketBuilder {
         /**
          * tsid changed, reset tsid and timestamp
          */
-        public RollupBucketBuilder resetTsid(BytesRef tsid, long timestamp) {
+        public RollupBucketBuilder resetTsid(BytesRef tsid, int tsidOrd, long timestamp) {
             this.tsid = BytesRef.deepCopyOf(tsid);
+            this.tsidOrd = tsidOrd;
             return resetTimestamp(timestamp);
         }
 
@@ -493,6 +496,10 @@ public BytesRef tsid() {
             return tsid;
         }
 
+        public int tsidOrd() {
+            return tsidOrd;
+        }
+
         public int docCount() {
             return docCount;
         }

x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java

@@ -477,7 +477,6 @@ public void onFailure(Exception e) {
         assertBusy(() -> assertTrue("In progress rollup did not complete", rollupListener.success), 60, TimeUnit.SECONDS);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/88800")
     public void testRollupDatastream() throws Exception {
         DownsampleConfig config = new DownsampleConfig(randomInterval());
         String dataStreamName = createDataStream();