[Transform] Use deduced mappings for determining proper fields' format even if `deduce_mappings==false` (#103682)

docs/changelog/103682.yaml

@@ -0,0 +1,6 @@
+pr: 103682
+summary: Use deduced mappings for determining proper fields' format even if `deduce_mappings==false`
+area: Transform
+type: bug
+issues:
+ - 103115

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/transform/preview_transforms.yml

@@ -240,6 +240,21 @@ setup:
               "deduce_mappings": false
             }
           }
+  - match: { preview.0.airline: foo }
+  - match: { preview.0.by-hour: "2017-02-18T00:00:00.000Z" }
+  - match: { preview.0.avg_response: 1.0 }
+  - match: { preview.0.time.max: "2017-02-18T00:30:00.000Z" }
+  - match: { preview.0.time.min: "2017-02-18T00:00:00.000Z" }
+  - match: { preview.1.airline: bar }
+  - match: { preview.1.by-hour: "2017-02-18T01:00:00.000Z" }
+  - match: { preview.1.avg_response: 42.0 }
+  - match: { preview.1.time.max: "2017-02-18T01:00:00.000Z" }
+  - match: { preview.1.time.min: "2017-02-18T01:00:00.000Z" }
+  - match: { preview.2.airline: foo }
+  - match: { preview.2.by-hour: "2017-02-18T01:00:00.000Z" }
+  - match: { preview.2.avg_response: 42.0 }
+  - match: { preview.2.time.max: "2017-02-18T01:01:00.000Z" }
+  - match: { preview.2.time.min: "2017-02-18T01:01:00.000Z" }
   - match: { generated_dest_index.mappings.properties: {} }
 
 ---

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformDestIndexIT.java

@@ -182,7 +182,7 @@ public class TransformDestIndexIT extends TransformRestTestCase {
                     }
                   }
                 }""", destIndex);
-            Request createIndexTemplateRequest = new Request("PUT", "_template/test_dest_index_no_deduce_template");
+            Request createIndexTemplateRequest = new Request("PUT", "_template/test_dest_index_mappings_template");
             createIndexTemplateRequest.setJsonEntity(destIndexTemplate);
             createIndexTemplateRequest.setOptions(expectWarnings(RestPutIndexTemplateAction.DEPRECATION_WARNING));
             Map<String, Object> createIndexTemplateResponse = entityAsMap(client().performRequest(createIndexTemplateRequest));
@@ -261,6 +261,9 @@ public class TransformDestIndexIT extends TransformRestTestCase {
                 )
             )
         );
+        Map<String, Object> searchResult = getAsMap(destIndex + "/_search?q=reviewer:user_0");
+        String timestamp = (String) ((List<?>) XContentMapValues.extractValue("hits.hits._source.timestamp", searchResult)).get(0);
+        assertThat(timestamp, is(equalTo("2017-01-10T10:10:10.000Z")));
     }
 
     private static void assertAliases(String index, String... aliases) throws IOException {

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformPivotRestIT.java

@@ -1383,8 +1383,11 @@ public class TransformPivotRestIT extends TransformRestTestCase {
                     }
                   }
                 }
+              },
+              "settings": {
+                "deduce_mappings": %s
               }
-            }""", REVIEWS_INDEX_NAME, offset);
+            }""", REVIEWS_INDEX_NAME, offset, randomBoolean());
         createPreviewRequest.setJsonEntity(config);
 
         Map<String, Object> previewTransformResponse = entityAsMap(client().performRequest(createPreviewRequest));

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransformUpdater.java

@@ -299,7 +299,7 @@ public class TransformUpdater {
         TransformAuditor auditor,
         IndexNameExpressionResolver indexNameExpressionResolver,
         TransformConfig config,
-        Map<String, String> mappings,
+        Map<String, String> destIndexMappings,
         SeqNoPrimaryTermAndIndex seqNoPrimaryTermAndIndex,
         ClusterState clusterState,
         Settings destIndexSettings,
@@ -355,7 +355,7 @@ public class TransformUpdater {
                 clusterState,
                 config,
                 destIndexSettings,
-                mappings,
+                destIndexMappings,
                 createDestinationListener
             );
         } else {

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPreviewTransformAction.java

@@ -47,6 +47,7 @@ import org.elasticsearch.xpack.core.transform.action.PreviewTransformAction;
 import org.elasticsearch.xpack.core.transform.action.PreviewTransformAction.Request;
 import org.elasticsearch.xpack.core.transform.action.PreviewTransformAction.Response;
 import org.elasticsearch.xpack.core.transform.transforms.DestAlias;
+import org.elasticsearch.xpack.core.transform.transforms.SettingsConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SourceConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
@@ -65,6 +66,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static java.util.Collections.emptyMap;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.xpack.core.transform.action.PreviewTransformAction.DUMMY_DEST_INDEX_FOR_PREVIEW;
 import static org.elasticsearch.xpack.transform.utils.SecondaryAuthorizationUtils.getSecurityHeadersPreferringSecondary;
@@ -153,6 +155,7 @@ public class TransportPreviewTransformAction extends HandledTransportAction<Requ
                 config.getDestination().getIndex(),
                 config.getDestination().getAliases(),
                 config.getSyncConfig(),
+                config.getSettings(),
                 listener
             ),
             listener::onFailure
@@ -208,6 +211,7 @@ public class TransportPreviewTransformAction extends HandledTransportAction<Requ
         String dest,
         List<DestAlias> aliases,
         SyncConfig syncConfig,
+        SettingsConfig settingsConfig,
         ActionListener<Response> listener
     ) {
         Client parentTaskClient = new ParentTaskAssigningClient(client, parentTaskId);
@@ -285,12 +289,17 @@ public class TransportPreviewTransformAction extends HandledTransportAction<Requ
         }, listener::onFailure);
 
         ActionListener<Map<String, String>> deduceMappingsListener = ActionListener.wrap(deducedMappings -> {
-            mappings.set(deducedMappings);
+            if (Boolean.FALSE.equals(settingsConfig.getDeduceMappings())) {
+                mappings.set(emptyMap());
+            } else {
+                mappings.set(deducedMappings);
+            }
             function.preview(
                 parentTaskClient,
                 timeout,
                 filteredHeaders,
                 source,
+                // Use deduced mappings for generating preview even if "settings.deduce_mappings" is set to false
                 deducedMappings,
                 NUMBER_OF_PREVIEW_BUCKETS,
                 previewListener

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportPutTransformAction.java

@@ -109,7 +109,7 @@ public class TransportPutTransformAction extends AcknowledgedTransportMasterNode
 
         // <3> Create the transform
         ActionListener<ValidateTransformAction.Response> validateTransformListener = ActionListener.wrap(
-            validationResponse -> putTransform(request, listener),
+            unusedValidationResponse -> putTransform(request, listener),
             listener::onFailure
         );
 

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/persistence/TransformIndex.java

@@ -40,6 +40,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
+import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static java.util.stream.Collectors.joining;
 import static java.util.stream.Collectors.toMap;
@@ -138,7 +139,7 @@ public final class TransformIndex {
         if (dest.length == 0) {
             TransformDestIndexSettings generatedDestIndexSettings = createTransformDestIndexSettings(
                 destIndexSettings,
-                destIndexMappings,
+                Boolean.FALSE.equals(config.getSettings().getDeduceMappings()) ? emptyMap() : destIndexMappings,
                 config.getId(),
                 Clock.systemUTC()
             );

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/ClientTransformIndexer.java

@@ -288,13 +288,13 @@ class ClientTransformIndexer extends TransformIndexer {
         SchemaUtil.getDestinationFieldMappings(client, getConfig().getDestination().getIndex(), fieldMappingsListener);
     }
 
-    void validate(ActionListener<Void> listener) {
+    void validate(ActionListener<ValidateTransformAction.Response> listener) {
         ClientHelper.executeAsyncWithOrigin(
             client,
             ClientHelper.TRANSFORM_ORIGIN,
             ValidateTransformAction.INSTANCE,
             new ValidateTransformAction.Request(transformConfig, false, AcknowledgedRequest.DEFAULT_ACK_TIMEOUT),
-            ActionListener.wrap(response -> listener.onResponse(null), listener::onFailure)
+            listener
         );
     }
 

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformIndexer.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.transform.transforms;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
@@ -31,6 +32,7 @@ import org.elasticsearch.xpack.core.indexing.AsyncTwoPhaseIndexer;
 import org.elasticsearch.xpack.core.indexing.IndexerState;
 import org.elasticsearch.xpack.core.indexing.IterationResult;
 import org.elasticsearch.xpack.core.transform.TransformMessages;
+import org.elasticsearch.xpack.core.transform.action.ValidateTransformAction;
 import org.elasticsearch.xpack.core.transform.transforms.SettingsConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
@@ -58,6 +60,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Stream;
 
+import static java.util.Collections.emptyMap;
 import static org.elasticsearch.core.Strings.format;
 
 public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformIndexerPosition, TransformIndexerStats> {
@@ -174,7 +177,7 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
 
     abstract void persistState(TransformState state, ActionListener<Void> listener);
 
-    abstract void validate(ActionListener<Void> listener);
+    abstract void validate(ActionListener<ValidateTransformAction.Response> listener);
 
     @Override
     protected String getJobId() {
@@ -265,6 +268,8 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
             return;
         }
 
+        SetOnce<Map<String, String>> deducedDestIndexMappings = new SetOnce<>();
+
         ActionListener<Void> finalListener = ActionListener.wrap(r -> {
             try {
                 // if we haven't set the page size yet, if it is set we might have reduced it after running into an out of memory
@@ -326,8 +331,14 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
             }
         }, listener::onFailure);
 
-        ActionListener<Map<String, String>> fieldMappingsListener = ActionListener.wrap(mappings -> {
-            this.fieldMappings = mappings;
+        ActionListener<Map<String, String>> fieldMappingsListener = ActionListener.wrap(destIndexMappings -> {
+            if (destIndexMappings.isEmpty() == false) {
+                // If we managed to fetch destination index mappings, we use them from now on ...
+                this.fieldMappings = destIndexMappings;
+            } else {
+                // ... otherwise we fall back to index mappings deduced based on source indices
+                this.fieldMappings = deducedDestIndexMappings.get();
+            }
             configurationReadyListener.onResponse(null);
         }, listener::onFailure);
 
@@ -338,7 +349,8 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
         }, listener::onFailure);
 
         // If we are continuous, we will want to verify we have the latest stored configuration
-        ActionListener<Void> changedSourceListener = ActionListener.wrap(r -> {
+        ActionListener<ValidateTransformAction.Response> changedSourceListener = ActionListener.wrap(validationResponse -> {
+            deducedDestIndexMappings.set(validationResponse.getDestIndexMappings());
             if (isContinuous()) {
                 transformsConfigManager.getTransformConfiguration(getJobId(), ActionListener.wrap(config -> {
                     if (transformConfig.equals(config) && fieldMappings != null) {
@@ -377,7 +389,7 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
                 if (hasChanged) {
                     context.setChangesLastDetectedAt(instantOfTrigger);
                     logger.debug("[{}] source has changed, triggering new indexer run.", getJobId());
-                    changedSourceListener.onResponse(null);
+                    changedSourceListener.onResponse(new ValidateTransformAction.Response(emptyMap()));
                 } else {
                     logger.trace("[{}] source has not changed, finish indexer early.", getJobId());
                     // No changes, stop executing
@@ -396,7 +408,7 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
             hasSourceChanged = true;
             context.setLastSearchTime(instantOfTrigger);
             context.setChangesLastDetectedAt(instantOfTrigger);
-            changedSourceListener.onResponse(null);
+            changedSourceListener.onResponse(new ValidateTransformAction.Response(emptyMap()));
         }
     }
 

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/pivot/Pivot.java

@@ -43,7 +43,6 @@ import java.util.Map.Entry;
 import java.util.Set;
 import java.util.stream.Stream;
 
-import static java.util.Collections.emptyMap;
 import static java.util.stream.Collectors.toList;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 
@@ -93,10 +92,6 @@ public class Pivot extends AbstractCompositeAggFunction {
         SourceConfig sourceConfig,
         final ActionListener<Map<String, String>> listener
     ) {
-        if (Boolean.FALSE.equals(settings.getDeduceMappings())) {
-            listener.onResponse(emptyMap());
-            return;
-        }
         SchemaUtil.deduceMappings(
             client,
             headers,

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/TransformIndexerStateTests.java

@@ -35,6 +35,7 @@ import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.indexing.IndexerState;
 import org.elasticsearch.xpack.core.indexing.IterationResult;
+import org.elasticsearch.xpack.core.transform.action.ValidateTransformAction;
 import org.elasticsearch.xpack.core.transform.transforms.SettingsConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TimeSyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
@@ -248,7 +249,7 @@ public class TransformIndexerStateTests extends ESTestCase {
         }
 
         @Override
-        void validate(ActionListener<Void> listener) {
+        void validate(ActionListener<ValidateTransformAction.Response> listener) {
             listener.onResponse(null);
         }
     }
@@ -335,7 +336,7 @@ public class TransformIndexerStateTests extends ESTestCase {
         }
 
         @Override
-        void validate(ActionListener<Void> listener) {
+        void validate(ActionListener<ValidateTransformAction.Response> listener) {
             listener.onResponse(null);
         }
 

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/TransformIndexerTests.java

@@ -34,6 +34,7 @@ import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.indexing.IndexerState;
 import org.elasticsearch.xpack.core.indexing.IterationResult;
+import org.elasticsearch.xpack.core.transform.action.ValidateTransformAction;
 import org.elasticsearch.xpack.core.transform.transforms.TimeRetentionPolicyConfigTests;
 import org.elasticsearch.xpack.core.transform.transforms.TimeSyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
@@ -268,7 +269,7 @@ public class TransformIndexerTests extends ESTestCase {
         }
 
         @Override
-        void validate(ActionListener<Void> listener) {
+        void validate(ActionListener<ValidateTransformAction.Response> listener) {
             listener.onResponse(null);
         }
     }