Wire project routing to FC (#138714)

Author: idegtiarenko

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java

@@ -566,16 +566,17 @@ private void resolveIndicesAndAnalyze(
         PreAnalysisResult result,
         ActionListener<Versioned<LogicalPlan>> logicalPlanListener
     ) {
-        SubscribableListener.<PreAnalysisResult>newForked(l -> preAnalyzeMainIndices(preAnalysis, executionInfo, result, requestFilter, l))
-            .andThenApply(r -> {
-                if (r.indexResolution.isEmpty() == false // Rule out ROW case with no FROM clauses
-                    && executionInfo.isCrossClusterSearch()
-                    && executionInfo.getRunningClusterAliases().findAny().isEmpty()) {
-                    LOGGER.debug("No more clusters to search, ending analysis stage");
-                    throw new NoClustersToSearchException();
-                }
-                return r;
-            })
+        SubscribableListener.<PreAnalysisResult>newForked(
+            l -> preAnalyzeMainIndices(preAnalysis, configuration, executionInfo, result, requestFilter, l)
+        ).andThenApply(r -> {
+            if (r.indexResolution.isEmpty() == false // Rule out ROW case with no FROM clauses
+                && executionInfo.isCrossClusterSearch()
+                && executionInfo.getRunningClusterAliases().findAny().isEmpty()) {
+                LOGGER.debug("No more clusters to search, ending analysis stage");
+                throw new NoClustersToSearchException();
+            }
+            return r;
+        })
             .<PreAnalysisResult>andThen((l, r) -> preAnalyzeLookupIndices(preAnalysis.lookupIndices().iterator(), r, executionInfo, l))
             .<PreAnalysisResult>andThen((l, r) -> {
                 enrichPolicyResolver.resolvePolicies(preAnalysis.enriches(), executionInfo, l.map(r::withEnrichResolution));
@@ -813,6 +814,7 @@ private void validateRemoteVersions(EsqlExecutionInfo executionInfo) {
      */
     private void preAnalyzeMainIndices(
         PreAnalyzer.PreAnalysis preAnalysis,
+        Configuration configuration,
         EsqlExecutionInfo executionInfo,
         PreAnalysisResult result,
         QueryBuilder requestFilter,
@@ -842,7 +844,16 @@ private void preAnalyzeMainIndices(
             forAll(
                 preAnalysis.indexes().entrySet().iterator(),
                 result,
-                (e, r, l) -> preAnalyzeFlatMainIndices(e.getKey(), e.getValue(), preAnalysis, executionInfo, r, requestFilter, l),
+                (e, r, l) -> preAnalyzeFlatMainIndices(
+                    e.getKey(),
+                    e.getValue(),
+                    configuration.projectRouting(),
+                    preAnalysis,
+                    executionInfo,
+                    r,
+                    requestFilter,
+                    l
+                ),
                 listener
             );
         }
@@ -883,6 +894,7 @@ private void preAnalyzeMainIndices(
     private void preAnalyzeFlatMainIndices(
         IndexPattern indexPattern,
         IndexMode indexMode,
+        String projectRouting,
         PreAnalyzer.PreAnalysis preAnalysis,
         EsqlExecutionInfo executionInfo,
         PreAnalysisResult result,
@@ -891,6 +903,7 @@ private void preAnalyzeFlatMainIndices(
     ) {
         indexResolver.resolveFlatWorldIndicesVersioned(
             indexPattern.indexPattern(),
+            projectRouting,
             result.fieldNames,
             createQueryFilter(indexMode, requestFilter),
             indexMode == IndexMode.TIME_SERIES,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/IndexResolver.java

@@ -102,7 +102,7 @@ public IndexResolver(Client client) {
      */
     public void resolveIndices(String indexPattern, Set<String> fieldNames, ActionListener<IndexResolution> listener) {
         doResolveIndices(
-            createFieldCapsRequest(DEFAULT_OPTIONS, indexPattern, fieldNames, null, false, false),
+            createFieldCapsRequest(DEFAULT_OPTIONS, indexPattern, null, fieldNames, null, false, false),
             indexPattern,
             false,
             false,
@@ -126,7 +126,7 @@ public void resolveIndicesVersioned(
         ActionListener<Versioned<IndexResolution>> listener
     ) {
         doResolveIndices(
-            createFieldCapsRequest(DEFAULT_OPTIONS, indexPattern, fieldNames, requestFilter, includeAllDimensions, false),
+            createFieldCapsRequest(DEFAULT_OPTIONS, indexPattern, null, fieldNames, requestFilter, includeAllDimensions, false),
             indexPattern,
             useAggregateMetricDoubleWhenNotSupported,
             useDenseVectorWhenNotSupported,
@@ -140,6 +140,7 @@ public void resolveIndicesVersioned(
 
     public void resolveFlatWorldIndicesVersioned(
         String indexPattern,
+        String projectRouting,
         Set<String> fieldNames,
         QueryBuilder requestFilter,
         boolean includeAllDimensions,
@@ -148,7 +149,7 @@ public void resolveFlatWorldIndicesVersioned(
         ActionListener<Versioned<IndexResolution>> listener
     ) {
         doResolveIndices(
-            createFieldCapsRequest(FLAT_WORLD_OPTIONS, indexPattern, fieldNames, requestFilter, includeAllDimensions, true),
+            createFieldCapsRequest(FLAT_WORLD_OPTIONS, indexPattern, projectRouting, fieldNames, requestFilter, includeAllDimensions, true),
             indexPattern,
             useAggregateMetricDoubleWhenNotSupported,
             useDenseVectorWhenNotSupported,
@@ -478,6 +479,7 @@ private static EsField conflictingMetricTypes(String name, String fullName, Fiel
     private static FieldCapabilitiesRequest createFieldCapsRequest(
         IndicesOptions options,
         String index,
+        @Nullable String projectRouting,
         Set<String> fieldNames,
         QueryBuilder requestFilter,
         boolean includeAllDimensions,
@@ -500,6 +502,7 @@ private static FieldCapabilitiesRequest createFieldCapsRequest(
         }
         request.setMergeResults(false);
         request.includeResolvedTo(includeResolvedTo);
+        request.projectRouting(projectRouting);
         return request;
     }