[9.2] [Transform] Fix transform producing empty dest index when source query references runtime fields (elastic#142450) (elastic#142830)

* [Transform] Fix transform producing empty dest index when source query references runtime fields (elastic#142450)

When a transform's source query references a runtime field (e.g., a range filter on a runtime-mapped field), the destination index is produced empty even though _transform/_preview returns correct results. The root cause is that ClientTransformIndexer.injectPointInTimeIfNeeded() passes the source query as the PIT indexFilter, but OpenPointInTimeRequest does not support runtime_mappings. During the can_match phase, the query on the unknown runtime field is rewritten to match_none, all shards are filtered out, and the PIT opens with zero search contexts — causing every subsequent search to return empty results. Preview is unaffected because it uses a normal SearchSourceBuilder with both the query and runtime_mappings and does not use PIT.

This change skips the PIT index filter optimization when the source config has non-empty runtime_mappings, since the filter cannot resolve runtime fields. The PIT is still opened for snapshot consistency, just without shard pre-filtering. Additionally, two other code paths that use the source query without propagating runtime_mappings are fixed: the initial progress search in TransformIndexer and the sourceHasChanged check in TimeBasedCheckpointProvider. All three fixes follow the same principle — wherever the source query is used, runtime_mappings from the source config must be propagated alongside it.

Fixes elastic#113156

(cherry picked from commit cf28432)

# Conflicts:
#	x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/ClientTransformIndexerTests.java

* fix compilation issues

Author: valeriy42

docs/changelog/142450.yaml

@@ -0,0 +1,7 @@
+area: Transform
+issues:
+ - 113156
+pr: 142450
+summary: Fix transform producing empty dest index when source query references runtime
+  fields
+type: bug

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProvider.java

@@ -71,6 +71,7 @@ public void sourceHasChanged(TransformCheckpoint lastCheckpoint, ActionListener<
         SearchSourceBuilder sourceBuilder = new SearchSourceBuilder().size(0)
             // we only want to know if there is at least 1 new document
             .trackTotalHitsUpTo(1)
+            .runtimeMappings(transformConfig.getSource().getRuntimeMappings())
             .query(queryBuilder);
         SearchRequest searchRequest = new SearchRequest(transformConfig.getSource().getIndex()).allowPartialSearchResults(false)
             .indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN)

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

@@ -550,8 +550,11 @@ private void injectPointInTimeIfNeeded(
 
         // no pit, create a new one
         OpenPointInTimeRequest pitRequest = new OpenPointInTimeRequest(searchRequest.indices()).keepAlive(PIT_KEEP_ALIVE);
-        // use index filter for better performance
-        pitRequest.indexFilter(transformConfig.getSource().getQueryConfig().getQuery());
+        // Only use index filter when there are no runtime mappings, because OpenPointInTimeRequest
+        // does not support runtime_mappings and the query may reference runtime fields
+        if (transformConfig.getSource().getRuntimeMappings().isEmpty()) {
+            pitRequest.indexFilter(transformConfig.getSource().getQueryConfig().getQuery());
+        }
 
         ClientHelper.executeWithHeadersAsync(
             transformConfig.getHeaders(),

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

@@ -323,7 +323,9 @@ protected void onStart(long now, ActionListener<Boolean> listener) {
 
                     // get progress information
                     SearchRequest request = new SearchRequest(transformConfig.getSource().getIndex());
-                    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+                    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().runtimeMappings(
+                        transformConfig.getSource().getRuntimeMappings()
+                    );
 
                     function.buildSearchQueryForInitialProgress(searchSourceBuilder);
                     searchSourceBuilder.query(QueryBuilders.boolQuery().filter(buildFilterQuery()).filter(searchSourceBuilder.query()));

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/checkpoint/TimeBasedCheckpointProviderTests.java

@@ -31,7 +31,9 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.transform.TransformConfigVersion;
 import org.elasticsearch.xpack.core.transform.action.GetCheckpointAction;
+import org.elasticsearch.xpack.core.transform.transforms.QueryConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SettingsConfig;
+import org.elasticsearch.xpack.core.transform.transforms.SourceConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TimeSyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
@@ -50,6 +52,7 @@
 
 import java.time.Clock;
 import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -283,6 +286,41 @@ private void testCreateNextCheckpoint(
         assertThat(exceptionHolder.get(), is(nullValue()));
     }
 
+    public void testSourceHasChangedIncludesRuntimeMappings() throws InterruptedException {
+        // Arrange: create a config with explicit runtime_mappings
+        Map<String, Object> runtimeMappings = Map.of(
+            "total_price_with_tax",
+            Map.of("type", "double", "script", Map.of("source", "emit(1.0)"))
+        );
+        SourceConfig sourceWithRuntimeMappings = new SourceConfig(new String[] { "source_index" }, QueryConfig.matchAll(), runtimeMappings);
+        TransformConfig transformConfig = new TransformConfig.Builder(TransformConfigTests.randomTransformConfig()).setSource(
+            sourceWithRuntimeMappings
+        ).setSyncConfig(new TimeSyncConfig(TIMESTAMP_FIELD, TimeValue.ZERO)).build();
+
+        final SearchResponse searchResponse = newSearchResponse(0);
+        try {
+            doAnswer(withResponse(searchResponse)).when(client).execute(eq(TransportSearchAction.TYPE), any(), any());
+            TimeBasedCheckpointProvider provider = newCheckpointProvider(transformConfig);
+
+            // Act: call sourceHasChanged
+            CountDownLatch latch = new CountDownLatch(1);
+            provider.sourceHasChanged(TransformCheckpoint.EMPTY, new LatchedActionListener<>(ActionListener.wrap(r -> {}, e -> {}), latch));
+            assertThat(latch.await(100, TimeUnit.MILLISECONDS), is(true));
+
+            // Assert: the search request should include runtime_mappings
+            ArgumentCaptor<SearchRequest> searchRequestCaptor = ArgumentCaptor.forClass(SearchRequest.class);
+            verify(client).execute(eq(TransportSearchAction.TYPE), searchRequestCaptor.capture(), any());
+            SearchRequest capturedRequest = searchRequestCaptor.getValue();
+            assertThat(
+                "sourceHasChanged search should include runtime_mappings from the source config",
+                capturedRequest.source().runtimeMappings(),
+                is(equalTo(runtimeMappings))
+            );
+        } finally {
+            searchResponse.decRef();
+        }
+    }
+
     private TimeBasedCheckpointProvider newCheckpointProvider(TransformConfig transformConfig) {
         return new TimeBasedCheckpointProvider(
             clock,

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

@@ -51,6 +51,7 @@
 import org.elasticsearch.transport.ActionNotFoundTransportException;
 import org.elasticsearch.xpack.core.indexing.IndexerState;
 import org.elasticsearch.xpack.core.transform.TransformMetadata;
+import org.elasticsearch.xpack.core.transform.transforms.QueryConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SettingsConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SourceConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
@@ -77,6 +78,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -150,43 +152,7 @@ public void testPitInjection() throws InterruptedException {
 
         try (var threadPool = createThreadPool()) {
             final var client = new PitMockClient(threadPool, true);
-            MockClientTransformIndexer indexer = new MockClientTransformIndexer(
-                mock(ThreadPool.class),
-                mock(ClusterService.class),
-                mock(IndexNameExpressionResolver.class),
-                mock(TransformExtension.class),
-                new TransformServices(
-                    mock(IndexBasedTransformConfigManager.class),
-                    mock(TransformCheckpointService.class),
-                    mock(TransformAuditor.class),
-                    new TransformScheduler(Clock.systemUTC(), mock(ThreadPool.class), Settings.EMPTY, TimeValue.ZERO),
-                    mock(TransformNode.class)
-                ),
-                mock(CheckpointProvider.class),
-                new AtomicReference<>(IndexerState.STOPPED),
-                null,
-                new ParentTaskAssigningClient(client, new TaskId("dummy-node:123456")),
-                mock(TransformIndexerStats.class),
-                config,
-                null,
-                new TransformCheckpoint(
-                    "transform",
-                    Instant.now().toEpochMilli(),
-                    0L,
-                    Collections.emptyMap(),
-                    Instant.now().toEpochMilli()
-                ),
-                new TransformCheckpoint(
-                    "transform",
-                    Instant.now().toEpochMilli(),
-                    2L,
-                    Collections.emptyMap(),
-                    Instant.now().toEpochMilli()
-                ),
-                new SeqNoPrimaryTermAndIndex(1, 1, TransformInternalIndexConstants.LATEST_INDEX_NAME),
-                mock(TransformContext.class),
-                false
-            );
+            MockClientTransformIndexer indexer = createMockIndexerForPitTest(client, config);
 
             this.<SearchResponse>assertAsync(listener -> indexer.doNextSearch(0, listener), response -> {
                 assertEquals(new BytesArray("the_pit_id+"), response.pointInTimeId());
@@ -258,43 +224,7 @@ public void testPitInjectionIfPitNotSupported() throws InterruptedException {
 
         try (var threadPool = createThreadPool()) {
             final var client = new PitMockClient(threadPool, false);
-            MockClientTransformIndexer indexer = new MockClientTransformIndexer(
-                mock(ThreadPool.class),
-                mock(ClusterService.class),
-                mock(IndexNameExpressionResolver.class),
-                mock(TransformExtension.class),
-                new TransformServices(
-                    mock(IndexBasedTransformConfigManager.class),
-                    mock(TransformCheckpointService.class),
-                    mock(TransformAuditor.class),
-                    new TransformScheduler(Clock.systemUTC(), mock(ThreadPool.class), Settings.EMPTY, TimeValue.ZERO),
-                    mock(TransformNode.class)
-                ),
-                mock(CheckpointProvider.class),
-                new AtomicReference<>(IndexerState.STOPPED),
-                null,
-                new ParentTaskAssigningClient(client, new TaskId("dummy-node:123456")),
-                mock(TransformIndexerStats.class),
-                config,
-                null,
-                new TransformCheckpoint(
-                    "transform",
-                    Instant.now().toEpochMilli(),
-                    0L,
-                    Collections.emptyMap(),
-                    Instant.now().toEpochMilli()
-                ),
-                new TransformCheckpoint(
-                    "transform",
-                    Instant.now().toEpochMilli(),
-                    2L,
-                    Collections.emptyMap(),
-                    Instant.now().toEpochMilli()
-                ),
-                new SeqNoPrimaryTermAndIndex(1, 1, TransformInternalIndexConstants.LATEST_INDEX_NAME),
-                mock(TransformContext.class),
-                false
-            );
+            MockClientTransformIndexer indexer = createMockIndexerForPitTest(client, config);
 
             this.<SearchResponse>assertAsync(
                 listener -> indexer.doNextSearch(0, listener),
@@ -463,6 +393,64 @@ public void testDisablePitWhenThereIsRemoteIndexInSource() throws InterruptedExc
         }
     }
 
+    public void testPitIndexFilterOmittedWhenRuntimeMappingsPresent() throws InterruptedException {
+        // Arrange: create a config with runtime_mappings and PIT enabled
+        Map<String, Object> runtimeMappings = Map.of(
+            "total_price_with_tax",
+            Map.of("type", "double", "script", Map.of("source", "emit(1.0)"))
+        );
+        TransformConfig config = newPitEnabledConfigWithSource(
+            new SourceConfig(new String[] { "source_index" }, QueryConfig.matchAll(), runtimeMappings)
+        );
+
+        try (var threadPool = createThreadPool()) {
+            final var client = new PitMockClient(threadPool, true);
+            MockClientTransformIndexer indexer = createMockIndexerForPitTest(client, config);
+
+            // Act: trigger a search which opens a PIT
+            this.<SearchResponse>assertAsync(
+                listener -> indexer.doNextSearch(0, listener),
+                response -> { assertNotNull(response.pointInTimeId()); }
+            );
+
+            // Assert: the PIT request should NOT have an indexFilter when runtime_mappings are present,
+            // because OpenPointInTimeRequest does not support runtime_mappings and the query may reference runtime fields
+            OpenPointInTimeRequest capturedPitRequest = client.getLastOpenPitRequest();
+            assertNotNull("Expected a PIT open request to have been captured", capturedPitRequest);
+            assertNull(
+                "PIT indexFilter should be null when runtime_mappings are present, "
+                    + "because OpenPointInTimeRequest does not support runtime_mappings",
+                capturedPitRequest.indexFilter()
+            );
+        }
+    }
+
+    public void testPitIndexFilterSetWhenNoRuntimeMappings() throws InterruptedException {
+        // Arrange: create a config WITHOUT runtime_mappings and PIT enabled
+        TransformConfig config = newPitEnabledConfigWithSource(
+            new SourceConfig(new String[] { "source_index" }, QueryConfig.matchAll(), Collections.emptyMap())
+        );
+
+        try (var threadPool = createThreadPool()) {
+            final var client = new PitMockClient(threadPool, true);
+            MockClientTransformIndexer indexer = createMockIndexerForPitTest(client, config);
+
+            // Act: trigger a search which opens a PIT
+            this.<SearchResponse>assertAsync(
+                listener -> indexer.doNextSearch(0, listener),
+                response -> { assertNotNull(response.pointInTimeId()); }
+            );
+
+            // Assert: the PIT request SHOULD have an indexFilter when no runtime_mappings are present
+            OpenPointInTimeRequest capturedPitRequest = client.getLastOpenPitRequest();
+            assertNotNull("Expected a PIT open request to have been captured", capturedPitRequest);
+            assertNotNull(
+                "PIT indexFilter should be set when no runtime_mappings are present for performance optimization",
+                capturedPitRequest.indexFilter()
+            );
+        }
+    }
+
     public void testHandlePitIndexNotFound() throws InterruptedException {
         // simulate a deleted index due to ILM
         try (var threadPool = createThreadPool()) {
@@ -548,6 +536,40 @@ private IndexNameExpressionResolver resolver() {
         return resolver;
     }
 
+    private static TransformConfig newPitEnabledConfigWithSource(SourceConfig sourceConfig) {
+        return new TransformConfig.Builder(TransformConfigTests.randomTransformConfig()).setSource(sourceConfig)
+            .setSettings(new SettingsConfig.Builder().setUsePit(true).build())
+            .build();
+    }
+
+    private MockClientTransformIndexer createMockIndexerForPitTest(PitMockClient client, TransformConfig config) {
+        return new MockClientTransformIndexer(
+            mock(ThreadPool.class),
+            mock(ClusterService.class),
+            mock(IndexNameExpressionResolver.class),
+            mock(TransformExtension.class),
+            new TransformServices(
+                mock(IndexBasedTransformConfigManager.class),
+                mock(TransformCheckpointService.class),
+                mock(TransformAuditor.class),
+                new TransformScheduler(Clock.systemUTC(), mock(ThreadPool.class), Settings.EMPTY, TimeValue.ZERO),
+                mock(TransformNode.class)
+            ),
+            mock(CheckpointProvider.class),
+            new AtomicReference<>(IndexerState.STOPPED),
+            null,
+            new ParentTaskAssigningClient(client, new TaskId("dummy-node:123456")),
+            mock(TransformIndexerStats.class),
+            config,
+            null,
+            new TransformCheckpoint("transform", Instant.now().toEpochMilli(), 0L, Collections.emptyMap(), Instant.now().toEpochMilli()),
+            new TransformCheckpoint("transform", Instant.now().toEpochMilli(), 2L, Collections.emptyMap(), Instant.now().toEpochMilli()),
+            new SeqNoPrimaryTermAndIndex(1, 1, TransformInternalIndexConstants.LATEST_INDEX_NAME),
+            mock(TransformContext.class),
+            false
+        );
+    }
+
     private static class MockClientTransformIndexer extends ClientTransformIndexer {
 
         MockClientTransformIndexer(
@@ -600,12 +622,17 @@ private static class PitMockClient extends NoOpClient {
         private final boolean pitSupported;
         private AtomicLong pitContextCounter = new AtomicLong();
         private List<Runnable> beforeCloseListeners = new ArrayList<>();
+        private final AtomicReference<OpenPointInTimeRequest> lastOpenPitRequest = new AtomicReference<>();
 
         PitMockClient(ThreadPool threadPool, boolean pitSupported) {
             super(threadPool);
             this.pitSupported = pitSupported;
         }
 
+        public OpenPointInTimeRequest getLastOpenPitRequest() {
+            return lastOpenPitRequest.get();
+        }
+
         public void addBeforeCloseListener(Runnable listener) {
             this.beforeCloseListeners.add(listener);
         }
@@ -621,7 +648,8 @@ protected <Request extends ActionRequest, Response extends ActionResponse> void
             Request request,
             ActionListener<Response> listener
         ) {
-            if (request instanceof OpenPointInTimeRequest) {
+            if (request instanceof OpenPointInTimeRequest openPitRequest) {
+                lastOpenPitRequest.set(openPitRequest);
                 if (pitSupported) {
                     pitContextCounter.incrementAndGet();
                     OpenPointInTimeResponse response = new OpenPointInTimeResponse(new BytesArray("the_pit_id"), 1, 1, 0, 0);