Move DataExtractor interface and some utility functions from core to the ml plugin (#104797)

* Move DataExtractor interface to ML plugin

* Split core/ExtractorUtils into core/DatafeedConfigUtils and ml/DataExtractorUtils.

x-pack/plugin/core/src/main/java/module-info.java

@@ -81,7 +81,6 @@ module org.elasticsearch.xcore {
     exports org.elasticsearch.xpack.core.ml.annotations;
     exports org.elasticsearch.xpack.core.ml.autoscaling;
     exports org.elasticsearch.xpack.core.ml.calendars;
-    exports org.elasticsearch.xpack.core.ml.datafeed.extractor;
     exports org.elasticsearch.xpack.core.ml.datafeed;
     exports org.elasticsearch.xpack.core.ml.dataframe.analyses;
     exports org.elasticsearch.xpack.core.ml.dataframe.evaluation.classification;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfig.java

@@ -38,7 +38,6 @@ import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xpack.core.common.time.TimeUtils;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.core.ml.utils.MlStrings;
 import org.elasticsearch.xpack.core.ml.utils.QueryProvider;
@@ -144,7 +143,7 @@ public class DatafeedConfig implements SimpleDiffable<DatafeedConfig>, ToXConten
         }
 
         Builder.checkForOnlySingleTopLevelCompositeAggAndValidate(aggregations.getAggregatorFactories());
-        AggregationBuilder histogramAggregation = ExtractorUtils.getHistogramAggregation(aggregatorFactories);
+        AggregationBuilder histogramAggregation = DatafeedConfigUtils.getHistogramAggregation(aggregatorFactories);
         if (histogramAggregation instanceof CompositeAggregationBuilder
             && aggregations.getPipelineAggregatorFactories().isEmpty() == false) {
             throw ExceptionsHelper.badRequestException(
@@ -449,7 +448,7 @@ public class DatafeedConfig implements SimpleDiffable<DatafeedConfig>, ToXConten
      * @param namedXContentRegistry XContent registry to transform the lazily parsed aggregations
      */
     public long getHistogramIntervalMillis(NamedXContentRegistry namedXContentRegistry) {
-        return ExtractorUtils.getHistogramIntervalMillis(getParsedAggregations(namedXContentRegistry));
+        return DatafeedConfigUtils.getHistogramIntervalMillis(getParsedAggregations(namedXContentRegistry));
     }
 
     /**
@@ -461,7 +460,7 @@ public class DatafeedConfig implements SimpleDiffable<DatafeedConfig>, ToXConten
         if (hasAggregations() == false) {
             return false;
         }
-        AggregationBuilder maybeComposite = ExtractorUtils.getHistogramAggregation(
+        AggregationBuilder maybeComposite = DatafeedConfigUtils.getHistogramAggregation(
             getParsedAggregations(namedXContentRegistry).getAggregatorFactories()
         );
         return maybeComposite instanceof CompositeAggregationBuilder;
@@ -622,12 +621,14 @@ public class DatafeedConfig implements SimpleDiffable<DatafeedConfig>, ToXConten
         if (aggProvider == null || aggProvider.getParsedAggs() == null) {
             return ChunkingConfig.newAuto();
         } else {
-            AggregationBuilder histogram = ExtractorUtils.getHistogramAggregation(aggProvider.getParsedAggs().getAggregatorFactories());
+            AggregationBuilder histogram = DatafeedConfigUtils.getHistogramAggregation(
+                aggProvider.getParsedAggs().getAggregatorFactories()
+            );
             if (histogram instanceof CompositeAggregationBuilder) {
                 // Allow composite aggs to handle the underlying chunking and searching
                 return ChunkingConfig.newOff();
             }
-            long histogramIntervalMillis = ExtractorUtils.getHistogramIntervalMillis(histogram);
+            long histogramIntervalMillis = DatafeedConfigUtils.getHistogramIntervalMillis(histogram);
             if (histogramIntervalMillis <= 0) {
                 throw ExceptionsHelper.badRequestException(DATAFEED_AGGREGATIONS_INTERVAL_MUST_BE_GREATER_THAN_ZERO);
             }
@@ -1084,7 +1085,7 @@ public class DatafeedConfig implements SimpleDiffable<DatafeedConfig>, ToXConten
 
         private static void checkNoMoreHistogramAggregations(Collection<AggregationBuilder> aggregations) {
             for (AggregationBuilder agg : aggregations) {
-                if (ExtractorUtils.isHistogram(agg)) {
+                if (DatafeedConfigUtils.isHistogram(agg)) {
                     throw ExceptionsHelper.badRequestException(DATAFEED_AGGREGATIONS_MAX_ONE_DATE_HISTOGRAM);
                 }
                 checkNoMoreHistogramAggregations(agg.getSubAggregations());
@@ -1097,7 +1098,7 @@ public class DatafeedConfig implements SimpleDiffable<DatafeedConfig>, ToXConten
                 timeField = ((ValuesSourceAggregationBuilder<?>) histogramAggregation).field();
             }
             if (histogramAggregation instanceof CompositeAggregationBuilder) {
-                DateHistogramValuesSourceBuilder valueSource = ExtractorUtils.getDateHistogramValuesSource(
+                DateHistogramValuesSourceBuilder valueSource = DatafeedConfigUtils.getDateHistogramValuesSource(
                     (CompositeAggregationBuilder) histogramAggregation
                 );
                 timeField = valueSource.field();
@@ -1115,7 +1116,7 @@ public class DatafeedConfig implements SimpleDiffable<DatafeedConfig>, ToXConten
         }
 
         private static void checkHistogramIntervalIsPositive(AggregationBuilder histogramAggregation) {
-            long interval = ExtractorUtils.getHistogramIntervalMillis(histogramAggregation);
+            long interval = DatafeedConfigUtils.getHistogramIntervalMillis(histogramAggregation);
             if (interval <= 0) {
                 throw ExceptionsHelper.badRequestException(DATAFEED_AGGREGATIONS_INTERVAL_MUST_BE_GREATER_THAN_ZERO);
             }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/extractor/ExtractorUtils.java → x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfigUtils.java

@@ -4,14 +4,11 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-package org.elasticsearch.xpack.core.ml.datafeed.extractor;
+package org.elasticsearch.xpack.core.ml.datafeed;
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.Rounding;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.index.query.BoolQueryBuilder;
-import org.elasticsearch.index.query.QueryBuilder;
-import org.elasticsearch.index.query.RangeQueryBuilder;
 import org.elasticsearch.search.aggregations.AggregationBuilder;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregationBuilder;
@@ -29,21 +26,11 @@ import java.util.Collection;
 import java.util.concurrent.TimeUnit;
 
 /**
- * Collects common utility methods needed by various {@link DataExtractor} implementations
+ * Utility methods used for datafeed configuration.
  */
-public final class ExtractorUtils {
+public final class DatafeedConfigUtils {
 
-    private static final String EPOCH_MILLIS = "epoch_millis";
-
-    private ExtractorUtils() {}
-
-    /**
-     * Combines a user query with a time range query.
-     */
-    public static QueryBuilder wrapInTimeRangeQuery(QueryBuilder userQuery, String timeField, long start, long end) {
-        QueryBuilder timeQuery = new RangeQueryBuilder(timeField).gte(start).lt(end).format(EPOCH_MILLIS);
-        return new BoolQueryBuilder().filter(userQuery).filter(timeQuery);
-    }
+    private DatafeedConfigUtils() {}
 
     /**
      * Find the (date) histogram in {@code aggFactory} and extract its interval.

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/datafeed/extractor/ExtractorUtilsTests.java → x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfigUtilsTests.java

@@ -4,7 +4,7 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-package org.elasticsearch.xpack.core.ml.datafeed.extractor;
+package org.elasticsearch.xpack.core.ml.datafeed;
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.search.aggregations.AggregationBuilder;
@@ -23,7 +23,7 @@ import java.time.ZoneOffset;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 
-public class ExtractorUtilsTests extends ESTestCase {
+public class DatafeedConfigUtilsTests extends ESTestCase {
 
     public void testGetHistogramAggregation_DateHistogramHasSibling() {
         AvgAggregationBuilder avg = AggregationBuilders.avg("avg");
@@ -31,7 +31,7 @@ public class ExtractorUtilsTests extends ESTestCase {
 
         ElasticsearchException e = expectThrows(
             ElasticsearchException.class,
-            () -> ExtractorUtils.getHistogramAggregation(
+            () -> DatafeedConfigUtils.getHistogramAggregation(
                 new AggregatorFactories.Builder().addAggregator(avg).addAggregator(dateHistogram).getAggregatorFactories()
             )
         );
@@ -42,7 +42,9 @@ public class ExtractorUtilsTests extends ESTestCase {
         terms.subAggregation(avg);
         e = expectThrows(
             ElasticsearchException.class,
-            () -> ExtractorUtils.getHistogramAggregation(new AggregatorFactories.Builder().addAggregator(terms).getAggregatorFactories())
+            () -> DatafeedConfigUtils.getHistogramAggregation(
+                new AggregatorFactories.Builder().addAggregator(terms).getAggregatorFactories()
+            )
         );
         assertEquals("The date_histogram (or histogram) aggregation cannot have sibling aggregations", e.getMessage());
     }
@@ -52,20 +54,20 @@ public class ExtractorUtilsTests extends ESTestCase {
         TermsAggregationBuilder nestedTerms = AggregationBuilders.terms("nested_terms");
 
         DateHistogramAggregationBuilder dateHistogram = AggregationBuilders.dateHistogram("time");
-        AggregationBuilder histogramAggregationBuilder = ExtractorUtils.getHistogramAggregation(
+        AggregationBuilder histogramAggregationBuilder = DatafeedConfigUtils.getHistogramAggregation(
             new AggregatorFactories.Builder().addAggregator(dateHistogram).getAggregatorFactories()
         );
         assertEquals(dateHistogram, histogramAggregationBuilder);
 
         dateHistogram.subAggregation(avg).subAggregation(nestedTerms);
-        histogramAggregationBuilder = ExtractorUtils.getHistogramAggregation(
+        histogramAggregationBuilder = DatafeedConfigUtils.getHistogramAggregation(
             new AggregatorFactories.Builder().addAggregator(dateHistogram).getAggregatorFactories()
         );
         assertEquals(dateHistogram, histogramAggregationBuilder);
 
         TermsAggregationBuilder toplevelTerms = AggregationBuilders.terms("top_level");
         toplevelTerms.subAggregation(dateHistogram);
-        histogramAggregationBuilder = ExtractorUtils.getHistogramAggregation(
+        histogramAggregationBuilder = DatafeedConfigUtils.getHistogramAggregation(
             new AggregatorFactories.Builder().addAggregator(toplevelTerms).getAggregatorFactories()
         );
 
@@ -76,7 +78,9 @@ public class ExtractorUtilsTests extends ESTestCase {
         TermsAggregationBuilder terms = AggregationBuilders.terms("top_level");
         ElasticsearchException e = expectThrows(
             ElasticsearchException.class,
-            () -> ExtractorUtils.getHistogramAggregation(new AggregatorFactories.Builder().addAggregator(terms).getAggregatorFactories())
+            () -> DatafeedConfigUtils.getHistogramAggregation(
+                new AggregatorFactories.Builder().addAggregator(terms).getAggregatorFactories()
+            )
         );
         assertEquals("A date_histogram (or histogram) aggregation is required", e.getMessage());
     }
@@ -90,7 +94,7 @@ public class ExtractorUtilsTests extends ESTestCase {
             .subAggregation(maxTime);
         ElasticsearchException e = expectThrows(
             ElasticsearchException.class,
-            () -> ExtractorUtils.getHistogramIntervalMillis(dateHistogram)
+            () -> DatafeedConfigUtils.getHistogramIntervalMillis(dateHistogram)
         );
 
         assertThat(e.getMessage(), equalTo("ML requires date_histogram.time_zone to be UTC"));
@@ -104,7 +108,7 @@ public class ExtractorUtilsTests extends ESTestCase {
             .fixedInterval(new DateHistogramInterval(300000L + "ms"))
             .timeZone(zone)
             .subAggregation(maxTime);
-        assertThat(ExtractorUtils.getHistogramIntervalMillis(dateHistogram), is(300_000L));
+        assertThat(DatafeedConfigUtils.getHistogramIntervalMillis(dateHistogram), is(300_000L));
     }
 
     public void testGetHistogramIntervalMillis_GivenUtcTimeZones() {
@@ -115,22 +119,22 @@ public class ExtractorUtilsTests extends ESTestCase {
             .fixedInterval(new DateHistogramInterval("300000ms"))
             .timeZone(zone)
             .subAggregation(maxTime);
-        assertThat(ExtractorUtils.getHistogramIntervalMillis(dateHistogram), is(300_000L));
+        assertThat(DatafeedConfigUtils.getHistogramIntervalMillis(dateHistogram), is(300_000L));
     }
 
     public void testIsHistogram() {
-        assertTrue(ExtractorUtils.isHistogram(AggregationBuilders.dateHistogram("time")));
-        assertTrue(ExtractorUtils.isHistogram(AggregationBuilders.histogram("time")));
-        assertFalse(ExtractorUtils.isHistogram(AggregationBuilders.max("time")));
+        assertTrue(DatafeedConfigUtils.isHistogram(AggregationBuilders.dateHistogram("time")));
+        assertTrue(DatafeedConfigUtils.isHistogram(AggregationBuilders.histogram("time")));
+        assertFalse(DatafeedConfigUtils.isHistogram(AggregationBuilders.max("time")));
     }
 
     public void testValidateAndGetCalendarInterval() {
-        assertEquals(300 * 1000L, ExtractorUtils.validateAndGetCalendarInterval("5m"));
-        assertEquals(7200 * 1000L, ExtractorUtils.validateAndGetCalendarInterval("2h"));
-        assertEquals(86400L * 1000L, ExtractorUtils.validateAndGetCalendarInterval("1d"));
+        assertEquals(300 * 1000L, DatafeedConfigUtils.validateAndGetCalendarInterval("5m"));
+        assertEquals(7200 * 1000L, DatafeedConfigUtils.validateAndGetCalendarInterval("2h"));
+        assertEquals(86400L * 1000L, DatafeedConfigUtils.validateAndGetCalendarInterval("1d"));
     }
 
     public void testValidateAndGetCalendarInterval_intervalIsLongerThanAWeek() {
-        expectThrows(ElasticsearchException.class, () -> ExtractorUtils.validateAndGetCalendarInterval("8d"));
+        expectThrows(ElasticsearchException.class, () -> DatafeedConfigUtils.validateAndGetCalendarInterval("8d"));
     }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportPreviewDatafeedAction.java

@@ -35,10 +35,10 @@ import org.elasticsearch.xpack.core.ml.action.PreviewDatafeedAction;
 import org.elasticsearch.xpack.core.ml.datafeed.ChunkingConfig;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedTimingStats;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.security.SecurityContext;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 import org.elasticsearch.xpack.ml.datafeed.persistence.DatafeedConfigProvider;
 import org.elasticsearch.xpack.ml.job.persistence.JobConfigProvider;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedJob.java

@@ -26,7 +26,6 @@ import org.elasticsearch.xpack.core.ml.action.PersistJobAction;
 import org.elasticsearch.xpack.core.ml.action.PostDataAction;
 import org.elasticsearch.xpack.core.ml.annotations.Annotation;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.DataDescription;
 import org.elasticsearch.xpack.core.ml.job.messages.Messages;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.DataCounts;
@@ -35,6 +34,7 @@ import org.elasticsearch.xpack.core.security.user.InternalUsers;
 import org.elasticsearch.xpack.ml.annotations.AnnotationPersister;
 import org.elasticsearch.xpack.ml.datafeed.delayeddatacheck.DelayedDataDetector;
 import org.elasticsearch.xpack.ml.datafeed.delayeddatacheck.DelayedDataDetectorFactory.BucketWithMissingData;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 import org.elasticsearch.xpack.ml.notifications.AnomalyDetectionAuditor;
 

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/delayeddatacheck/DatafeedDelayedDataDetector.java

@@ -20,10 +20,10 @@ import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.xpack.core.action.util.PageParams;
 import org.elasticsearch.xpack.core.ml.action.GetBucketsAction;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils;
 import org.elasticsearch.xpack.core.ml.job.results.Bucket;
 import org.elasticsearch.xpack.core.ml.utils.Intervals;
 import org.elasticsearch.xpack.ml.datafeed.delayeddatacheck.DelayedDataDetectorFactory.BucketWithMissingData;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorUtils;
 
 import java.time.ZonedDateTime;
 import java.util.Collections;
@@ -129,7 +129,7 @@ public class DatafeedDelayedDataDetector implements DelayedDataDetector {
                 new DateHistogramAggregationBuilder(DATE_BUCKETS).fixedInterval(new DateHistogramInterval(bucketSpan + "ms"))
                     .field(timeField)
             )
-            .query(ExtractorUtils.wrapInTimeRangeQuery(datafeedQuery, timeField, start, end))
+            .query(DataExtractorUtils.wrapInTimeRangeQuery(datafeedQuery, timeField, start, end))
             .runtimeMappings(runtimeMappings);
 
         SearchRequest searchRequest = new SearchRequest(datafeedIndices).source(searchSourceBuilder).indicesOptions(indicesOptions);

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/extractor/DataExtractor.java → x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/DataExtractor.java

@@ -4,7 +4,7 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-package org.elasticsearch.xpack.core.ml.datafeed.extractor;
+package org.elasticsearch.xpack.ml.datafeed.extractor;
 
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.search.SearchResponse;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/DataExtractorFactory.java

@@ -16,7 +16,6 @@ import org.elasticsearch.license.RemoteClusterLicenseChecker;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.core.rollup.action.GetRollupIndexCapsAction;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/DataExtractorUtils.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.ml.datafeed.extractor;
+
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.RangeQueryBuilder;
+
+/**
+ * Utility methods for various DataExtractor implementations.
+ */
+public final class DataExtractorUtils {
+
+    private static final String EPOCH_MILLIS = "epoch_millis";
+
+    /**
+     * Combines a user query with a time range query.
+     */
+    public static QueryBuilder wrapInTimeRangeQuery(QueryBuilder userQuery, String timeField, long start, long end) {
+        QueryBuilder timeQuery = new RangeQueryBuilder(timeField).gte(start).lt(end).format(EPOCH_MILLIS);
+        return new BoolQueryBuilder().filter(userQuery).filter(timeQuery);
+    }
+}

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/aggregation/AbstractAggregationDataExtractor.java

@@ -17,10 +17,11 @@ import org.elasticsearch.search.aggregations.Aggregation;
 import org.elasticsearch.search.aggregations.Aggregations;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.xpack.core.ClientHelper;
+import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfigUtils;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorUtils;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -151,10 +152,10 @@ abstract class AbstractAggregationDataExtractor<T extends ActionRequestBuilder<S
         // For derivative aggregations the first bucket will always be null
         // so query one extra histogram bucket back and hope there is data
         // in that bucket
-        long histogramSearchStartTime = Math.max(0, context.start - ExtractorUtils.getHistogramIntervalMillis(context.aggs));
+        long histogramSearchStartTime = Math.max(0, context.start - DatafeedConfigUtils.getHistogramIntervalMillis(context.aggs));
 
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().size(0)
-            .query(ExtractorUtils.wrapInTimeRangeQuery(context.query, context.timeField, histogramSearchStartTime, context.end));
+            .query(DataExtractorUtils.wrapInTimeRangeQuery(context.query, context.timeField, histogramSearchStartTime, context.end));
 
         if (context.runtimeMappings.isEmpty() == false) {
             searchSourceBuilder.runtimeMappings(context.runtimeMappings);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/aggregation/AggregationDataExtractorFactory.java

@@ -13,10 +13,10 @@ import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.utils.Intervals;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 
 public record AggregationDataExtractorFactory(

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/aggregation/CompositeAggregationDataExtractor.java

@@ -17,11 +17,12 @@ import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregati
 import org.elasticsearch.search.aggregations.bucket.composite.CompositeAggregationBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.xpack.core.ClientHelper;
+import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfigUtils;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils;
 import org.elasticsearch.xpack.core.ml.utils.Intervals;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorUtils;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -70,7 +71,7 @@ class CompositeAggregationDataExtractor implements DataExtractor {
         this.context = Objects.requireNonNull(dataExtractorContext);
         this.timingStatsReporter = Objects.requireNonNull(timingStatsReporter);
         this.requestBuilder = Objects.requireNonNull(requestBuilder);
-        this.interval = ExtractorUtils.getHistogramIntervalMillis(compositeAggregationBuilder);
+        this.interval = DatafeedConfigUtils.getHistogramIntervalMillis(compositeAggregationBuilder);
         this.hasNext = true;
     }
 
@@ -127,7 +128,7 @@ class CompositeAggregationDataExtractor implements DataExtractor {
             () -> format("[%s] Executing composite aggregated search from [%s] to [%s]", context.jobId, context.start, context.end)
         );
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().size(0)
-            .query(ExtractorUtils.wrapInTimeRangeQuery(context.query, context.timeField, context.start, context.end));
+            .query(DataExtractorUtils.wrapInTimeRangeQuery(context.query, context.timeField, context.start, context.end));
 
         if (context.runtimeMappings.isEmpty() == false) {
             searchSourceBuilder.runtimeMappings(context.runtimeMappings);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/aggregation/CompositeAggregationDataExtractorFactory.java

@@ -16,11 +16,11 @@ import org.elasticsearch.search.aggregations.bucket.composite.CompositeValuesSou
 import org.elasticsearch.search.aggregations.bucket.composite.DateHistogramValuesSourceBuilder;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils;
+import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfigUtils;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.utils.Intervals;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 
 import java.util.ArrayList;
@@ -59,7 +59,7 @@ public class CompositeAggregationDataExtractorFactory implements DataExtractorFa
         this.job = Objects.requireNonNull(job);
         this.timingStatsReporter = Objects.requireNonNull(timingStatsReporter);
         this.parsedQuery = datafeedConfig.getParsedQuery(xContentRegistry);
-        AggregationBuilder aggregationBuilder = ExtractorUtils.getHistogramAggregation(
+        AggregationBuilder aggregationBuilder = DatafeedConfigUtils.getHistogramAggregation(
             datafeedConfig.getParsedAggregations(xContentRegistry).getAggregatorFactories()
         );
         if (aggregationBuilder instanceof CompositeAggregationBuilder == false) {
@@ -106,7 +106,7 @@ public class CompositeAggregationDataExtractorFactory implements DataExtractorFa
         compositeAggregationBuilder.size(numBuckets);
         subAggs.forEach(compositeAggregationBuilder::subAggregation);
         subPipelineAggs.forEach(compositeAggregationBuilder::subAggregation);
-        long histogramInterval = ExtractorUtils.getHistogramIntervalMillis(compositeAggregationBuilder);
+        long histogramInterval = DatafeedConfigUtils.getHistogramIntervalMillis(compositeAggregationBuilder);
         CompositeAggregationDataExtractorContext dataExtractorContext = new CompositeAggregationDataExtractorContext(
             job.getId(),
             job.getDataDescription().getTimeField(),

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/aggregation/RollupDataExtractorFactory.java

@@ -20,7 +20,6 @@ import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilde
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregationBuilder;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.utils.Intervals;
 import org.elasticsearch.xpack.core.rollup.action.RollableIndexCaps;
@@ -28,6 +27,7 @@ import org.elasticsearch.xpack.core.rollup.action.RollupJobCaps.RollupFieldCaps;
 import org.elasticsearch.xpack.core.rollup.action.RollupSearchAction;
 import org.elasticsearch.xpack.core.rollup.job.DateHistogramGroupConfig;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 
 import java.time.ZoneId;
@@ -42,9 +42,9 @@ import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-import static org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils.getHistogramAggregation;
-import static org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils.getHistogramIntervalMillis;
-import static org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils.validateAndGetCalendarInterval;
+import static org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfigUtils.getHistogramAggregation;
+import static org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfigUtils.getHistogramIntervalMillis;
+import static org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfigUtils.validateAndGetCalendarInterval;
 
 public class RollupDataExtractorFactory implements DataExtractorFactory {
 

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/chunked/ChunkedDataExtractor.java

@@ -21,11 +21,11 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils;
 import org.elasticsearch.xpack.core.rollup.action.RollupSearchAction;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorUtils;
 import org.elasticsearch.xpack.ml.datafeed.extractor.aggregation.RollupDataExtractorFactory;
 
 import java.io.IOException;
@@ -303,7 +303,7 @@ public class ChunkedDataExtractor implements DataExtractor {
 
         private SearchSourceBuilder rangeSearchBuilder() {
             return new SearchSourceBuilder().size(0)
-                .query(ExtractorUtils.wrapInTimeRangeQuery(context.query, context.timeField, currentStart, context.end))
+                .query(DataExtractorUtils.wrapInTimeRangeQuery(context.query, context.timeField, currentStart, context.end))
                 .runtimeMappings(context.runtimeMappings)
                 .aggregation(AggregationBuilders.min(EARLIEST_TIME).field(context.timeField))
                 .aggregation(AggregationBuilders.max(LATEST_TIME).field(context.timeField));

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/chunked/ChunkedDataExtractorFactory.java

@@ -11,10 +11,10 @@ import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.utils.Intervals;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 
 import java.util.Objects;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/scroll/ScrollDataExtractor.java

@@ -25,9 +25,9 @@ import org.elasticsearch.search.fetch.StoredFieldsContext;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.ExtractorUtils;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorUtils;
 import org.elasticsearch.xpack.ml.extractor.ExtractedField;
 
 import java.io.IOException;
@@ -158,7 +158,7 @@ class ScrollDataExtractor implements DataExtractor {
     private SearchRequestBuilder buildSearchRequest(long start) {
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().size(context.scrollSize)
             .sort(context.extractedFields.timeField(), SortOrder.ASC)
-            .query(ExtractorUtils.wrapInTimeRangeQuery(context.query, context.extractedFields.timeField(), start, context.end))
+            .query(DataExtractorUtils.wrapInTimeRangeQuery(context.query, context.extractedFields.timeField(), start, context.end))
             .runtimeMappings(context.runtimeMappings);
 
         SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client).setScroll(SCROLL_TIMEOUT)

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/extractor/scroll/ScrollDataExtractorFactory.java

@@ -19,11 +19,11 @@ import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.core.ml.utils.MlStrings;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 
 import java.util.Map;

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/action/TransportPreviewDatafeedActionTests.java

@@ -16,7 +16,7 @@ import org.elasticsearch.xpack.core.ml.action.PreviewDatafeedAction;
 import org.elasticsearch.xpack.core.ml.datafeed.ChunkingConfig;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchIntervalTests;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.junit.Before;
 import org.mockito.stubbing.Answer;
 

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/datafeed/DatafeedJobTests.java

@@ -36,7 +36,6 @@ import org.elasticsearch.xpack.core.ml.action.PostDataAction;
 import org.elasticsearch.xpack.core.ml.annotations.Annotation;
 import org.elasticsearch.xpack.core.ml.annotations.AnnotationIndex;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.core.ml.job.config.DataDescription;
 import org.elasticsearch.xpack.core.ml.job.messages.Messages;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.DataCounts;
@@ -46,6 +45,7 @@ import org.elasticsearch.xpack.core.security.user.InternalUsers;
 import org.elasticsearch.xpack.ml.annotations.AnnotationPersister;
 import org.elasticsearch.xpack.ml.datafeed.delayeddatacheck.DelayedDataDetector;
 import org.elasticsearch.xpack.ml.datafeed.delayeddatacheck.DelayedDataDetectorFactory.BucketWithMissingData;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 import org.elasticsearch.xpack.ml.notifications.AnomalyDetectionAuditor;
 import org.elasticsearch.xpack.ml.utils.persistence.ResultsPersisterService;

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/datafeed/extractor/aggregation/AggregationDataExtractorTests.java

@@ -23,9 +23,9 @@ import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedTimingStats;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter.DatafeedTimingStatsPersister;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.aggregation.AggregationTestUtils.Term;
 import org.junit.Before;
 

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/datafeed/extractor/aggregation/CompositeAggregationDataExtractorTests.java

@@ -28,9 +28,9 @@ import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInter
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedTimingStats;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter.DatafeedTimingStatsPersister;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.junit.Before;
 
 import java.io.BufferedReader;

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/datafeed/extractor/chunked/ChunkedDataExtractorTests.java

@@ -26,9 +26,9 @@ import org.elasticsearch.search.aggregations.metrics.Min;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedTimingStats;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter.DatafeedTimingStatsPersister;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractorFactory;
 import org.junit.Before;
 import org.mockito.Mockito;

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/datafeed/extractor/scroll/ScrollDataExtractorTests.java

@@ -36,9 +36,9 @@ import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedTimingStats;
 import org.elasticsearch.xpack.core.ml.datafeed.SearchInterval;
-import org.elasticsearch.xpack.core.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedTimingStatsReporter.DatafeedTimingStatsPersister;
+import org.elasticsearch.xpack.ml.datafeed.extractor.DataExtractor;
 import org.elasticsearch.xpack.ml.extractor.DocValueField;
 import org.elasticsearch.xpack.ml.extractor.ExtractedField;
 import org.elasticsearch.xpack.ml.extractor.TimeField;