[Transform] Add support for search-time runtime fields. (#67643)

client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/SourceConfig.java

@@ -24,10 +24,12 @@
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
@@ -49,15 +51,19 @@ public class SourceConfig implements ToXContentObject {
         String[] index = ((List<String>)args[0]).toArray(new String[0]);
         // default handling: if the user does not specify a query, we default to match_all
         QueryConfig queryConfig = (QueryConfig) args[1];
-        return new SourceConfig(index, queryConfig);
+        @SuppressWarnings("unchecked")
+        Map<String, Object> runtimeMappings = (Map<String, Object>) args[2];
+        return new SourceConfig(index, queryConfig, runtimeMappings);
     });
     static {
         PARSER.declareStringArray(constructorArg(), INDEX);
         PARSER.declareObject(optionalConstructorArg(), (p, c) -> QueryConfig.fromXContent(p), QUERY);
+        PARSER.declareObject(optionalConstructorArg(), (p, c) -> p.map(), SearchSourceBuilder.RUNTIME_MAPPINGS_FIELD);
     }
 
     private final String[] index;
     private final QueryConfig queryConfig;
+    private final Map<String, Object> runtimeMappings;
 
     /**
      * Create a new SourceConfig for the provided indices.
@@ -67,19 +73,20 @@ public class SourceConfig implements ToXContentObject {
      * @param index Any number of indices. At least one non-null, non-empty, index should be provided
      */
     public SourceConfig(String... index) {
-        this.index = index;
-        this.queryConfig = null;
+        this(index, null, null);
     }
 
     /**
      * Create a new SourceConfig for the provided indices, from which data is gathered with the provided {@link QueryConfig}
      *
      * @param index Any number of indices. At least one non-null, non-empty, index should be provided
      * @param queryConfig A QueryConfig object that contains the desired query. Defaults to MatchAll query.
+     * @param runtimeMappings Search-time runtime fields that can be used by the transform
      */
-    SourceConfig(String[] index, QueryConfig queryConfig) {
+    SourceConfig(String[] index, QueryConfig queryConfig, Map<String, Object> runtimeMappings) {
         this.index = index;
         this.queryConfig = queryConfig;
+        this.runtimeMappings = runtimeMappings;
     }
 
     public String[] getIndex() {
@@ -90,6 +97,10 @@ public QueryConfig getQueryConfig() {
         return queryConfig;
     }
 
+    public Map<String, Object> getRuntimeMappings() {
+        return runtimeMappings;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -99,6 +110,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (queryConfig != null) {
             builder.field(QUERY.getPreferredName(), queryConfig);
         }
+        if (runtimeMappings != null) {
+            builder.field(SearchSourceBuilder.RUNTIME_MAPPINGS_FIELD.getPreferredName(), runtimeMappings);
+        }
         builder.endObject();
         return builder;
     }
@@ -113,14 +127,16 @@ public boolean equals(Object other) {
         }
 
         SourceConfig that = (SourceConfig) other;
-        return Arrays.equals(index, that.index) && Objects.equals(queryConfig, that.queryConfig);
+        return Arrays.equals(index, that.index)
+            && Objects.equals(queryConfig, that.queryConfig)
+            && Objects.equals(runtimeMappings, that.runtimeMappings);
     }
 
     @Override
     public int hashCode(){
         // Using Arrays.hashCode as Objects.hash does not deeply hash nested arrays. Since we are doing Array.equals, this is necessary
-        int hash = Arrays.hashCode(index);
-        return 31 * hash + (queryConfig == null ? 0 : queryConfig.hashCode());
+        int indexArrayHash = Arrays.hashCode(index);
+        return Objects.hash(indexArrayHash, queryConfig, runtimeMappings);
     }
 
     public static Builder builder() {
@@ -130,6 +146,7 @@ public static Builder builder() {
     public static class Builder {
         private String[] index;
         private QueryConfig queryConfig;
+        private Map<String, Object> runtimeMappings;
 
         /**
          * Sets what indices from which to fetch data
@@ -160,8 +177,13 @@ public Builder setQuery(QueryBuilder query) {
             return this.setQueryConfig(new QueryConfig(query));
         }
 
+        public Builder setRuntimeMappings(Map<String, Object> runtimeMappings) {
+            this.runtimeMappings = runtimeMappings;
+            return this;
+        }
+
         public SourceConfig build() {
-            return new SourceConfig(index, queryConfig);
+            return new SourceConfig(index, queryConfig, runtimeMappings);
         }
     }
 }

client/rest-high-level/src/test/java/org/elasticsearch/client/transform/transforms/SourceConfigTests.java

@@ -26,18 +26,28 @@
 import org.elasticsearch.test.AbstractXContentTestCase;
 
 import java.io.IOException;
+import java.util.Map;
 import java.util.function.Predicate;
 
 import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonMap;
+import static java.util.stream.Collectors.toMap;
 
 
 public class SourceConfigTests extends AbstractXContentTestCase<SourceConfig> {
 
     public static SourceConfig randomSourceConfig() {
-        return new SourceConfig(generateRandomStringArray(10, 10, false, false),
-            QueryConfigTests.randomQueryConfig());
+        return new SourceConfig(
+            generateRandomStringArray(10, 10, false, false),
+            QueryConfigTests.randomQueryConfig(),
+            randomRuntimeMappings());
     }
 
+    private static Map<String, Object> randomRuntimeMappings() {
+        return randomList(0, 10, () -> randomAlphaOfLengthBetween(1, 10)).stream()
+            .distinct()
+            .collect(toMap(f -> f, f -> singletonMap("type", randomFrom("boolean", "date", "double", "keyword", "long"))));
+    }
 
     @Override
     protected SourceConfig doParseInstance(XContentParser parser) throws IOException {

docs/reference/rest-api/common-parms.asciidoc

@@ -678,7 +678,7 @@ objects define the `group by` fields and the aggregation to reduce the data.
 end::pivot[]
 
 tag::pivot-aggs[]
-Defines how to aggregate the grouped data. The following aggregations are 
+Defines how to aggregate the grouped data. The following aggregations are
 currently supported:
 +
 --
@@ -705,7 +705,7 @@ currently supported:
 end::pivot-aggs[]
 
 tag::pivot-group-by[]
-Defines how to group the data. More than one grouping can be defined per pivot. 
+Defines how to group the data. More than one grouping can be defined per pivot.
 The following groupings are currently supported:
 +
 --
@@ -917,6 +917,10 @@ A query clause that retrieves a subset of data from the source index. See
 <<query-dsl>>.
 end::source-query-transforms[]
 
+tag::source-runtime-mappings-transforms[]
+Definitions of search-time runtime fields that can be used by the transform.
+end::source-runtime-mappings-transforms[]
+
 tag::state-transform[]
 The status of the {transform}, which can be one of the following values:
 +
@@ -968,7 +972,7 @@ default value is `60s`.
 end::sync-time-delay[]
 
 tag::transform-latest[]
-The `latest` method transforms the data by finding the latest document for each 
+The `latest` method transforms the data by finding the latest document for each
 unique key.
 end::transform-latest[]
 

docs/reference/transform/apis/preview-transform.asciidoc

@@ -18,10 +18,10 @@ Previews a {transform}.
 [[preview-transform-prereq]]
 == {api-prereq-title}
 
-If the {es} {security-features} are enabled, you must have the following 
+If the {es} {security-features} are enabled, you must have the following
 privileges:
 
-* `manage_transform` 
+* `manage_transform`
 * source index: `read`, `view_index_metadata`
 
 The built-in `transform_admin` role has the `manage_transform` privilege.
@@ -136,6 +136,10 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=source-index-transform
 `query`:::
 (Optional, object)
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=source-query-transforms]
+
+`runtime_mappings`:::
+(Optional, object)
+include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=source-runtime-mappings-transforms]
 ====
 //End source
 
@@ -160,7 +164,7 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=sync-time]
 `delay`::::
 (Optional, <<time-units, time units>>)
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=sync-time-delay]
-    
+
 `field`::::
 (Optional, string)
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=sync-time-field]
@@ -191,11 +195,11 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=transform-settings-max
 == {api-response-body-title}
 
 `preview`::
-(array) An array of documents. In particular, they are the JSON representation 
-of the documents that would be created in the destination index by the 
+(array) An array of documents. In particular, they are the JSON representation
+of the documents that would be created in the destination index by the
 {transform}.
 
-//Begin generated_dest_index  
+//Begin generated_dest_index
 `generated_dest_index`::
 (object) Contains details about the destination index.
 +
@@ -207,7 +211,7 @@ of the documents that would be created in the destination index by the
 (object) The aliases for the destination index.
 
 `mappings`:::
-(object) The <<mapping,mappings>> for each document in the destination index. 
+(object) The <<mapping,mappings>> for each document in the destination index.
 
 `settings`:::
 (object) The <<index-modules-settings,index settings>> for the destination index.

docs/reference/transform/apis/put-transform.asciidoc

@@ -33,10 +33,10 @@ For more information, see <<built-in-roles>>, <<security-privileges>>, and
 == {api-description-title}
 
 This API defines a {transform}, which copies data from source indices,
-transforms it, and persists it into an entity-centric destination index. If you 
-choose to use the pivot method for your {transform}, the entities are defined by 
-the set of `group_by` fields in the `pivot` object.  If you choose to use the 
-latest method, the entities are defined by the `unique_key` field values in the 
+transforms it, and persists it into an entity-centric destination index. If you
+choose to use the pivot method for your {transform}, the entities are defined by
+the set of `group_by` fields in the `pivot` object.  If you choose to use the
+latest method, the entities are defined by the `unique_key` field values in the
 `latest` object.
 
 You can also think of the destination index as a two-dimensional tabular data structure (known as a {dataframe}). The ID for each document in the
@@ -56,12 +56,12 @@ it had at the time of creation and uses those same roles. If those roles do not
 have the required privileges on the source and destination indices, the
 {transform} fails when it attempts unauthorized operations.
 
-IMPORTANT: You must use {kib} or this API to create a {transform}. Do not put a 
+IMPORTANT: You must use {kib} or this API to create a {transform}. Do not put a
 {transform} directly into any `.transform-internal*` indices using the {es}
 index API. If {es} {security-features} are enabled, do not give users any
 privileges on `.transform-internal*` indices. If you used {transforms} prior to
 7.5, also do not give users any privileges on `.data-frame-internal*` indices.
-            
+
 You must choose either the latest or pivot method for your {transform}; you
 cannot use both in a single {transform}.
 
@@ -187,6 +187,10 @@ include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=source-index-transform
 `query`:::
 (Optional, object)
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=source-query-transforms]
+
+`runtime_mappings`:::
+(Optional, object)
+include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=source-runtime-mappings-transforms]
 ====
 //End source
 
@@ -289,7 +293,7 @@ When the {transform} is created, you receive the following results:
 }
 ----
 
-The following {transform} uses the `latest` method: 
+The following {transform} uses the `latest` method:
 
 [source,console]
 --------------------------------------------------
@@ -315,4 +319,4 @@ PUT _transform/ecommerce_transform2
   }
 }
 --------------------------------------------------
-// TEST[setup:kibana_sample_data_ecommerce]
+// TEST[setup:kibana_sample_data_ecommerce]