Merge branch 'main' into fix/bulk-pipeline

Author: samxbr

modules/reindex/src/main/java/org/elasticsearch/reindex/ReindexPlugin.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.index.reindex.BulkByScrollTask;
 import org.elasticsearch.index.reindex.DeleteByQueryAction;
@@ -44,6 +45,11 @@ public class ReindexPlugin extends Plugin implements ActionPlugin {
 
     public static final ActionType<ListTasksResponse> RETHROTTLE_ACTION = new ActionType<>("cluster:admin/reindex/rethrottle");
 
+    /**
+     * Whether the feature flag to guard the work to make reindex more resilient while it is under development.
+     */
+    static boolean REINDEX_RESILIENCE_ENABLED = new FeatureFlag("reindex_resilience").isEnabled();
+
     @Override
     public List<ActionHandler> getActions() {
         return Arrays.asList(

muted-tests.yml

@@ -447,6 +447,9 @@ tests:
 - class: org.elasticsearch.test.rest.yaml.CcsCommonYamlTestSuiteIT
   method: test {p0=scroll/20_keep_alive/Max keep alive}
   issue: https://github.com/elastic/elasticsearch/issues/138680
+- class: org.elasticsearch.xpack.unsignedlong.UnsignedLongSyntheticSourceNativeArrayIntegrationTests
+  method: testSynthesizeArrayRandom
+  issue: https://github.com/elastic/elasticsearch/issues/138684
 
 # Examples:
 #

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java

@@ -148,11 +148,7 @@ protected LookupEnrichQueryGenerator queryList(
      * In those cases, it is safe to ignore the plan sent and return null
      */
     private static PhysicalPlan localLookupNodePlanning(PhysicalPlan physicalPlan) {
-        if (physicalPlan instanceof FragmentExec fragmentExec) {
-            LocalMapper localMapper = new LocalMapper();
-            return localMapper.map(fragmentExec.fragment());
-        }
-        return null;
+        return physicalPlan instanceof FragmentExec fragmentExec ? LocalMapper.INSTANCE.map(fragmentExec.fragment()) : null;
     }
 
     @Override

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/PlannerUtils.java

@@ -144,9 +144,8 @@ public static PlanReduction reductionPlan(PhysicalPlan plan) {
             return SimplePlanReduction.NO_REDUCTION;
         }
         final LogicalPlan pipelineBreaker = pipelineBreakers.getFirst();
-        final LocalMapper mapper = new LocalMapper();
         int estimatedRowSize = fragment.estimatedRowSize();
-        return switch (mapper.map(pipelineBreaker)) {
+        return switch (LocalMapper.INSTANCE.map(pipelineBreaker)) {
             case TopNExec topN -> new TopNReduction(EstimatesRowSize.estimateRowSize(estimatedRowSize, topN));
             case AggregateExec aggExec -> getPhysicalPlanReduction(estimatedRowSize, aggExec.withMode(AggregatorMode.INTERMEDIATE));
             case PhysicalPlan p -> getPhysicalPlanReduction(estimatedRowSize, p);
@@ -218,7 +217,6 @@ public static PhysicalPlan localPlan(
         LocalLogicalPlanOptimizer logicalOptimizer,
         LocalPhysicalPlanOptimizer physicalOptimizer
     ) {
-        final LocalMapper localMapper = new LocalMapper();
         var isCoordPlan = new Holder<>(Boolean.TRUE);
         Set<PhysicalPlan> lookupJoinExecRightChildren = plan.collect(LookupJoinExec.class::isInstance)
             .stream()
@@ -234,7 +232,7 @@ public static PhysicalPlan localPlan(
             }
             isCoordPlan.set(Boolean.FALSE);
             LogicalPlan optimizedFragment = logicalOptimizer.localOptimize(f.fragment());
-            PhysicalPlan physicalFragment = localMapper.map(optimizedFragment);
+            PhysicalPlan physicalFragment = LocalMapper.INSTANCE.map(optimizedFragment);
             QueryBuilder filter = f.esFilter();
             if (filter != null) {
                 physicalFragment = physicalFragment.transformUp(

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/mapper/LocalMapper.java

@@ -41,6 +41,10 @@
  */
 public class LocalMapper {
 
+    public static LocalMapper INSTANCE = new LocalMapper();
+
+    private LocalMapper() {}
+
     public PhysicalPlan map(LogicalPlan p) {
 
         if (p instanceof LeafPlan leaf) {

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

@@ -133,7 +133,7 @@ public static Optional<ReductionPlan> planReduceDriverTopN(
     }
 
     private static PhysicalPlan toPhysical(LogicalPlan plan, LocalPhysicalOptimizerContext context) {
-        return new InsertFieldExtraction().apply(new ReplaceSourceAttributes().apply(new LocalMapper().map(plan)), context);
+        return new InsertFieldExtraction().apply(new ReplaceSourceAttributes().apply(LocalMapper.INSTANCE.map(plan)), context);
     }
 
     private LateMaterializationPlanner() { /* static class */ }