TSDB: Implement downsampling on time-series indices

docs/changelog/85708.yaml

@@ -0,0 +1,5 @@
+pr: 85708
+summary: "TSDB: Implement downsampling on time-series indices"
+area: "TSDB"
+type: feature
+issues: []

server/src/main/java/org/elasticsearch/index/mapper/TimeSeriesParams.java

@@ -23,10 +23,20 @@ public final class TimeSeriesParams {
     private TimeSeriesParams() {}
 
     public enum MetricType {
-        gauge,
-        counter,
-        histogram,
-        summary
+        gauge(new String[] { "value_count", "sum", "min", "max" }),
+        counter(new String[] { "max" }),
+        histogram(new String[] { "value_count" }), // TODO Add more aggs
+        summary(new String[] { "value_count", "sum", "min", "max" });
+
+        private final String[] supportedAggs;
+
+        MetricType(String[] supportedAggs) {
+            this.supportedAggs = supportedAggs;
+        }
+
+        public String[] supportedAggs() {
+            return supportedAggs;
+        }
     }
 
     public static FieldMapper.Parameter<MetricType> metricParam(Function<FieldMapper, MetricType> initializer, MetricType... values) {

server/src/main/java/org/elasticsearch/search/aggregations/timeseries/TimeSeriesIndexSearcher.java

@@ -78,7 +78,7 @@ public void search(Query query, BucketCollector bucketCollector) throws IOExcept
         PriorityQueue<LeafWalker> queue = new PriorityQueue<>(searcher.getIndexReader().leaves().size()) {
             @Override
             protected boolean lessThan(LeafWalker a, LeafWalker b) {
-                return a.timestamp < b.timestamp;
+                return a.timestamp > b.timestamp;
             }
         };
 

server/src/test/java/org/elasticsearch/search/aggregations/timeseries/TimeSeriesIndexSearcherTests.java

@@ -46,26 +46,28 @@ public class TimeSeriesIndexSearcherTests extends ESTestCase {
     // Open a searcher over a set of leaves
     // Collection should be in order
 
+    private static final int THREADS = 5;
+    private static final int DOCS_PER_THREAD = 500;
+
     public void testCollectInOrderAcrossSegments() throws IOException, InterruptedException {
 
         Directory dir = newDirectory();
         IndexWriterConfig iwc = newIndexWriterConfig();
         iwc.setIndexSort(
             new Sort(
                 new SortField(TimeSeriesIdFieldMapper.NAME, SortField.Type.STRING),
-                new SortField(DataStream.TimestampField.FIXED_TIMESTAMP_FIELD, SortField.Type.LONG)
+                new SortField(DataStream.TimestampField.FIXED_TIMESTAMP_FIELD, SortField.Type.LONG, true)
             )
         );
         RandomIndexWriter iw = new RandomIndexWriter(random(), dir, iwc);
 
         AtomicInteger clock = new AtomicInteger(0);
 
-        final int THREADS = 5;
         ExecutorService indexer = Executors.newFixedThreadPool(THREADS);
         for (int i = 0; i < THREADS; i++) {
             indexer.submit(() -> {
                 Document doc = new Document();
-                for (int j = 0; j < 500; j++) {
+                for (int j = 0; j < DOCS_PER_THREAD; j++) {
                     String tsid = "tsid" + randomIntBetween(0, 30);
                     long time = clock.addAndGet(randomIntBetween(0, 10));
                     doc.clear();
@@ -90,8 +92,8 @@ public void testCollectInOrderAcrossSegments() throws IOException, InterruptedEx
 
         BucketCollector collector = new BucketCollector() {
 
-            BytesRef currentTSID = null;
-            long currentTimestamp = 0;
+            BytesRef previousTSID = null;
+            long previousTimestamp = 0;
             long total = 0;
 
             @Override
@@ -107,16 +109,18 @@ public LeafBucketCollector getLeafCollector(AggregationExecutionContext aggCtx)
                     public void collect(int doc, long owningBucketOrd) throws IOException {
                         assertTrue(tsid.advanceExact(doc));
                         assertTrue(timestamp.advanceExact(doc));
-                        BytesRef latestTSID = tsid.lookupOrd(tsid.ordValue());
-                        long latestTimestamp = timestamp.longValue();
-                        if (currentTSID != null) {
-                            assertTrue(currentTSID + "->" + latestTSID.utf8ToString(), latestTSID.compareTo(currentTSID) >= 0);
-                            if (latestTSID.equals(currentTSID)) {
-                                assertTrue(currentTimestamp + "->" + latestTimestamp, latestTimestamp >= currentTimestamp);
+                        BytesRef currentTSID = tsid.lookupOrd(tsid.ordValue());
+                        assertEquals(aggCtx.getTsid(), currentTSID);
+                        long currentTimestamp = timestamp.longValue();
+                        logger.info("{} -> {} / {} -> {}", previousTSID, currentTSID, previousTimestamp, currentTimestamp);
+                        if (previousTSID != null) {
+                            assertTrue(previousTSID + "->" + currentTSID.utf8ToString(), currentTSID.compareTo(previousTSID) >= 0);
+                            if (currentTSID.equals(previousTSID)) {
+                                assertTrue(previousTimestamp + "->" + currentTimestamp, currentTimestamp <= previousTimestamp);
                             }
                         }
-                        currentTimestamp = latestTimestamp;
-                        currentTSID = BytesRef.deepCopyOf(latestTSID);
+                        previousTimestamp = currentTimestamp;
+                        previousTSID = BytesRef.deepCopyOf(currentTSID);
                         total++;
                     }
                 };
@@ -129,7 +133,7 @@ public void preCollection() throws IOException {
 
             @Override
             public void postCollection() throws IOException {
-                assertEquals(2500, total);
+                assertEquals(THREADS * DOCS_PER_THREAD, total);
             }
 
             @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java

@@ -27,7 +27,6 @@
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.NetworkPlugin;
 import org.elasticsearch.plugins.Plugin;
-import org.elasticsearch.rollup.RollupV2;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.ParseField;
@@ -375,6 +374,9 @@ public List<ActionType<? extends ActionResponse>> getClientActions() {
                 DeleteRollupJobAction.INSTANCE,
                 GetRollupJobsAction.INSTANCE,
                 GetRollupCapsAction.INSTANCE,
+                // TSDB Downsampling / Rollup
+                RollupIndexerAction.INSTANCE,
+                RollupAction.INSTANCE,
                 // ILM
                 DeleteLifecycleAction.INSTANCE,
                 GetLifecycleAction.INSTANCE,
@@ -411,12 +413,6 @@ public List<ActionType<? extends ActionResponse>> getClientActions() {
             )
         );
 
-        // rollupV2
-        if (RollupV2.isEnabled()) {
-            actions.add(RollupIndexerAction.INSTANCE);
-            actions.add(RollupAction.INSTANCE);
-        }
-
         return actions;
     }
 
@@ -520,6 +516,7 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, WaitForSnapshotAction.NAME, WaitForSnapshotAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, SearchableSnapshotAction.NAME, SearchableSnapshotAction::new),
                 new NamedWriteableRegistry.Entry(LifecycleAction.class, MigrateAction.NAME, MigrateAction::readFrom),
+                new NamedWriteableRegistry.Entry(LifecycleAction.class, RollupILMAction.NAME, RollupILMAction::new),
                 // Transforms
                 new NamedWriteableRegistry.Entry(Metadata.Custom.class, TransformMetadata.TYPE, TransformMetadata::new),
                 new NamedWriteableRegistry.Entry(NamedDiff.class, TransformMetadata.TYPE, TransformMetadata.TransformMetadataDiff::new),
@@ -565,10 +562,6 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
             )
         );
 
-        if (RollupV2.isEnabled()) {
-            namedWriteables.add(new NamedWriteableRegistry.Entry(LifecycleAction.class, RollupILMAction.NAME, RollupILMAction::new));
-        }
-
         return namedWriteables;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/TimeseriesLifecycleType.java

@@ -10,7 +10,6 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.rollup.RollupV2;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -56,7 +55,7 @@ public class TimeseriesLifecycleType implements LifecycleType {
         UnfollowAction.NAME,
         RolloverAction.NAME,
         ReadOnlyAction.NAME,
-        RollupV2.isEnabled() ? RollupILMAction.NAME : null,
+        RollupILMAction.NAME,
         ShrinkAction.NAME,
         ForceMergeAction.NAME,
         SearchableSnapshotAction.NAME
@@ -68,7 +67,8 @@ public class TimeseriesLifecycleType implements LifecycleType {
         AllocateAction.NAME,
         MigrateAction.NAME,
         ShrinkAction.NAME,
-        ForceMergeAction.NAME
+        ForceMergeAction.NAME,
+        RollupILMAction.NAME
     );
     public static final List<String> ORDERED_VALID_COLD_ACTIONS = Stream.of(
         SetPriorityAction.NAME,
@@ -78,7 +78,7 @@ public class TimeseriesLifecycleType implements LifecycleType {
         AllocateAction.NAME,
         MigrateAction.NAME,
         FreezeAction.NAME,
-        RollupV2.isEnabled() ? RollupILMAction.NAME : null
+        RollupILMAction.NAME
     ).filter(Objects::nonNull).toList();
     public static final List<String> ORDERED_VALID_FROZEN_ACTIONS = List.of(UnfollowAction.NAME, SearchableSnapshotAction.NAME);
     public static final List<String> ORDERED_VALID_DELETE_ACTIONS = List.of(WaitForSnapshotAction.NAME, DeleteAction.NAME);