[ML] Make delete intervening results more selective (#82437)

When reverting an anomaly detection job model snapshot using
the delete_intervening_results=true option, we were previously
being overzealous in the types of results that would be deleted.
Forecasts should _not_ be deleted in this case, as restarting
the datafeed after reverting the model snapshot will have no way
to automatically regenerate them.

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/RevertModelSnapshotIT.java

@@ -136,7 +136,11 @@ public class RevertModelSnapshotIT extends MlNativeAutodetectIntegTestCase {
                 .collect(Collectors.joining())
         );
         flushJob(job.getId(), true);
+        String forecastId = forecast(job.getId(), TimeValue.timeValueHours(10), TimeValue.timeValueDays(100));
+        waitForecastToFinish(job.getId(), forecastId);
         closeJob(job.getId());
+        long numForecastDocs = countForecastDocs(job.getId(), forecastId);
+        assertThat(numForecastDocs, greaterThan(0L));
 
         ModelSizeStats modelSizeStats1 = getJobStats(job.getId()).get(0).getModelSizeStats();
         Quantiles quantiles1 = getQuantiles(job.getId());
@@ -220,6 +224,9 @@ public class RevertModelSnapshotIT extends MlNativeAutodetectIntegTestCase {
         // Check annotations with event type from {delayed_data, model_change} have been removed if deleteInterveningResults flag is set
         assertThatNumberOfAnnotationsIsEqualTo(deleteInterveningResults ? 3 : 5);
 
+        // Reverting should not have deleted any forecast docs
+        assertThat(countForecastDocs(job.getId(), forecastId), is(numForecastDocs));
+
         // Re-run 2nd half of data
         openJob(job.getId());
         postData(
@@ -239,6 +246,9 @@ public class RevertModelSnapshotIT extends MlNativeAutodetectIntegTestCase {
         assertThat(finalPostRevertPointBucket.getTimestamp(), equalTo(finalPreRevertPointBucket.getTimestamp()));
         assertThat(finalPostRevertPointBucket.getAnomalyScore(), equalTo(finalPreRevertPointBucket.getAnomalyScore()));
         assertThat(finalPostRevertPointBucket.getEventCount(), equalTo(finalPreRevertPointBucket.getEventCount()));
+
+        // Re-running should not have deleted any forecast docs
+        assertThat(countForecastDocs(job.getId(), forecastId), is(numForecastDocs));
     }
 
     private Job.Builder buildAndRegisterJob(String jobId, TimeValue bucketSpan) throws Exception {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobDataDeleter.java

@@ -59,6 +59,11 @@ import org.elasticsearch.xpack.core.ml.job.persistence.ElasticsearchMappings;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.CategorizerState;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSnapshot;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.Quantiles;
+import org.elasticsearch.xpack.core.ml.job.results.AnomalyRecord;
+import org.elasticsearch.xpack.core.ml.job.results.Bucket;
+import org.elasticsearch.xpack.core.ml.job.results.BucketInfluencer;
+import org.elasticsearch.xpack.core.ml.job.results.Influencer;
+import org.elasticsearch.xpack.core.ml.job.results.ModelPlot;
 import org.elasticsearch.xpack.core.ml.job.results.Result;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.core.security.user.XPackUser;
@@ -185,14 +190,25 @@ public class JobDataDeleter {
     }
 
     /**
-     * Asynchronously delete all result types (Buckets, Records, Influencers) from {@code cutOffTime}
+     * Asynchronously delete all result types (Buckets, Records, Influencers) from {@code cutOffTime}.
+     * Forecasts are <em>not</em> deleted, as they will not be automatically regenerated after
+     * restarting a datafeed following a model snapshot reversion.
      *
      * @param cutoffEpochMs Results at and after this time will be deleted
      * @param listener Response listener
      */
     public void deleteResultsFromTime(long cutoffEpochMs, ActionListener<Boolean> listener) {
         QueryBuilder query = QueryBuilders.boolQuery()
-            .filter(QueryBuilders.existsQuery(Result.RESULT_TYPE.getPreferredName()))
+            .filter(
+                QueryBuilders.termsQuery(
+                    Result.RESULT_TYPE.getPreferredName(),
+                    AnomalyRecord.RESULT_TYPE_VALUE,
+                    Bucket.RESULT_TYPE_VALUE,
+                    BucketInfluencer.RESULT_TYPE_VALUE,
+                    Influencer.RESULT_TYPE_VALUE,
+                    ModelPlot.RESULT_TYPE_VALUE
+                )
+            )
             .filter(QueryBuilders.rangeQuery(Result.TIMESTAMP.getPreferredName()).gte(cutoffEpochMs));
         DeleteByQueryRequest dbqRequest = new DeleteByQueryRequest(AnomalyDetectorsIndex.jobResultsAliasedName(jobId)).setQuery(query)
             .setIndicesOptions(IndicesOptions.lenientExpandOpen())