Merge branch 'main' into 2025/03/20/onResponseSent-redundant-response

Author: DaveCTurner

docs/changelog/122250.yaml

@@ -0,0 +1,6 @@
+pr: 122250
+summary: "ESQL: Align `RENAME` behavior with `EVAL` for sequential processing"
+area: ES|QL
+type: enhancement
+issues:
+  - 121739

muted-tests.yml

@@ -399,6 +399,9 @@ tests:
 - class: org.elasticsearch.reservedstate.service.ReservedClusterStateServiceTests
   method: testProcessMultipleChunks
   issue: https://github.com/elastic/elasticsearch/issues/125305
+- class: org.elasticsearch.index.mapper.NativeArrayIntegrationTestCase
+  method: testSynthesizeArrayRandomIgnoresMalformed
+  issue: https://github.com/elastic/elasticsearch/issues/125319
 
 # Examples:
 #

x-pack/plugin/esql/qa/testFixtures/src/main/resources/rename.csv-spec

@@ -214,3 +214,85 @@ x:keyword
 Facello
 Simmel
 ;
+
+swappingNames
+required_capability: rename_sequential_processing
+FROM employees
+| SORT emp_no ASC
+| KEEP first_name, last_name
+| RENAME first_name AS last_name, last_name AS first_name, first_name as name
+| LIMIT 2
+;
+
+name:keyword
+Georgi
+Bezalel
+;
+
+complexSwappingNames
+required_capability: rename_sequential_processing
+FROM employees
+| SORT emp_no ASC
+| KEEP first_name, last_name, emp_no
+| RENAME first_name AS last_name, last_name AS first_name, first_name as emp_no, emp_no AS first_name
+| LIMIT 2
+;
+
+first_name:keyword
+Georgi
+Bezalel
+;
+
+
+reuseRenamedAlias
+required_capability: rename_sequential_processing
+FROM employees
+| SORT emp_no ASC
+| KEEP first_name, last_name
+| LIMIT 2
+| RENAME first_name AS x, x AS y, last_name as x
+;
+
+y:keyword   | x:keyword
+Georgi      | Facello
+Bezalel     | Simmel
+;
+
+
+multipleRenamesToSameAliasLastOnePrevails
+required_capability: rename_sequential_processing
+FROM employees
+| SORT emp_no ASC
+| KEEP first_name, last_name
+| LIMIT 2
+| RENAME first_name AS x, last_name as x
+;
+
+x:keyword
+Facello
+Simmel
+;
+
+
+swapNames
+required_capability: rename_sequential_processing
+ROW a="keyword", b=5
+| RENAME a AS temp, b AS a, temp AS b
+;
+
+b:keyword | a:integer
+keyword   | 5
+;
+
+
+multipleRenames
+required_capability: rename_sequential_processing
+ROW a="keyword", b=5, c=null
+| RENAME a AS c, b AS a
+| RENAME c AS b
+| RENAME a AS b, b AS a
+;
+
+a:integer
+5
+;

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

@@ -407,6 +407,12 @@ public enum Cap {
          */
         UNION_TYPES_FIX_RENAME_RESOLUTION,
 
+        /**
+         * Execute `RENAME` operations sequentially from left to right,
+         * see <a href="https://github.com/elastic/elasticsearch/issues/122250"> ESQL: Align RENAME behavior with EVAL for sequential processing #122250 </a>
+         */
+        RENAME_SEQUENTIAL_PROCESSING,
+
         /**
          * Fix for union-types when some indexes are missing the required field. Done in #111932.
          */

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

@@ -986,6 +986,7 @@ public static List<NamedExpression> projectionsForRename(Rename rename, List<Att
                 if (alias.child() instanceof UnresolvedAttribute ua && alias.name().equals(ua.name()) == false) {
                     // remove attributes overwritten by a renaming: `| keep a, b, c | rename a as b`
                     projections.removeIf(x -> x.name().equals(alias.name()));
+                    childrenOutput.removeIf(x -> x.name().equals(alias.name()));
 
                     var resolved = maybeResolveAttribute(ua, childrenOutput, logger);
                     if (resolved instanceof UnsupportedAttribute || resolved.resolved()) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/adaptiveallocations/AdaptiveAllocationsScalerService.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.xpack.core.ml.action.GetDeploymentStatsAction;
 import org.elasticsearch.xpack.core.ml.action.UpdateTrainedModelDeploymentAction;
 import org.elasticsearch.xpack.core.ml.inference.assignment.AssignmentStats;
+import org.elasticsearch.xpack.core.ml.inference.assignment.RoutingState;
 import org.elasticsearch.xpack.core.ml.inference.assignment.TrainedModelAssignment;
 import org.elasticsearch.xpack.core.ml.inference.assignment.TrainedModelAssignmentMetadata;
 import org.elasticsearch.xpack.ml.MachineLearning;
@@ -213,6 +214,7 @@ Collection<DoubleWithAttributes> observeDouble(Function<AdaptiveAllocationsScale
     private volatile Scheduler.Cancellable cancellable;
     private final AtomicBoolean busy;
     private final long scaleToZeroAfterNoRequestsSeconds;
+    private final long scaleUpCooldownTimeMillis;
     private final Set<String> deploymentIdsWithInFlightScaleFromZeroRequests = new ConcurrentSkipListSet<>();
     private final Map<String, String> lastWarningMessages = new ConcurrentHashMap<>();
 
@@ -224,7 +226,17 @@ public AdaptiveAllocationsScalerService(
         MeterRegistry meterRegistry,
         boolean isNlpEnabled
     ) {
-        this(threadPool, clusterService, client, inferenceAuditor, meterRegistry, isNlpEnabled, DEFAULT_TIME_INTERVAL_SECONDS);
+        this(
+            threadPool,
+            clusterService,
+            client,
+            inferenceAuditor,
+            meterRegistry,
+            isNlpEnabled,
+            DEFAULT_TIME_INTERVAL_SECONDS,
+            SCALE_TO_ZERO_AFTER_NO_REQUESTS_TIME_SECONDS,
+            SCALE_UP_COOLDOWN_TIME_MILLIS
+        );
     }
 
     // visible for testing
@@ -235,7 +247,9 @@ public AdaptiveAllocationsScalerService(
         InferenceAuditor inferenceAuditor,
         MeterRegistry meterRegistry,
         boolean isNlpEnabled,
-        int timeIntervalSeconds
+        int timeIntervalSeconds,
+        long scaleToZeroAfterNoRequestsSeconds,
+        long scaleUpCooldownTimeMillis
     ) {
         this.threadPool = threadPool;
         this.clusterService = clusterService;
@@ -244,14 +258,15 @@ public AdaptiveAllocationsScalerService(
         this.meterRegistry = meterRegistry;
         this.isNlpEnabled = isNlpEnabled;
         this.timeIntervalSeconds = timeIntervalSeconds;
+        this.scaleToZeroAfterNoRequestsSeconds = scaleToZeroAfterNoRequestsSeconds;
+        this.scaleUpCooldownTimeMillis = scaleUpCooldownTimeMillis;
 
         lastInferenceStatsByDeploymentAndNode = new HashMap<>();
         lastInferenceStatsTimestampMillis = null;
         lastScaleUpTimesMillis = new HashMap<>();
         scalers = new HashMap<>();
         metrics = new Metrics();
         busy = new AtomicBoolean(false);
-        scaleToZeroAfterNoRequestsSeconds = SCALE_TO_ZERO_AFTER_NO_REQUESTS_TIME_SECONDS;
     }
 
     public synchronized void start() {
@@ -375,6 +390,9 @@ private void processDeploymentStats(GetDeploymentStatsAction.Response statsRespo
 
         Map<String, Stats> recentStatsByDeployment = new HashMap<>();
         Map<String, Integer> numberOfAllocations = new HashMap<>();
+        // Check for recent scale ups in the deployment stats, because a different node may have
+        // caused a scale up when an inference request arrives and there were zero allocations.
+        Set<String> hasRecentObservedScaleUp = new HashSet<>();
 
         for (AssignmentStats assignmentStats : statsResponse.getStats().results()) {
             String deploymentId = assignmentStats.getDeploymentId();
@@ -401,6 +419,12 @@ private void processDeploymentStats(GetDeploymentStatsAction.Response statsRespo
                         (key, value) -> value == null ? recentStats : value.add(recentStats)
                     );
                 }
+                if (nodeStats.getRoutingState() != null && nodeStats.getRoutingState().getState() == RoutingState.STARTING) {
+                    hasRecentObservedScaleUp.add(deploymentId);
+                }
+                if (nodeStats.getStartTime() != null && now < nodeStats.getStartTime().toEpochMilli() + scaleUpCooldownTimeMillis) {
+                    hasRecentObservedScaleUp.add(deploymentId);
+                }
             }
         }
 
@@ -416,9 +440,12 @@ private void processDeploymentStats(GetDeploymentStatsAction.Response statsRespo
             Integer newNumberOfAllocations = adaptiveAllocationsScaler.scale();
             if (newNumberOfAllocations != null) {
                 Long lastScaleUpTimeMillis = lastScaleUpTimesMillis.get(deploymentId);
+                // hasRecentScaleUp indicates whether this service has recently scaled up the deployment.
+                // hasRecentObservedScaleUp indicates whether a deployment recently has started,
+                // potentially triggered by another node.
+                boolean hasRecentScaleUp = lastScaleUpTimeMillis != null && now < lastScaleUpTimeMillis + scaleUpCooldownTimeMillis;
                 if (newNumberOfAllocations < numberOfAllocations.get(deploymentId)
-                    && lastScaleUpTimeMillis != null
-                    && now < lastScaleUpTimeMillis + SCALE_UP_COOLDOWN_TIME_MILLIS) {
+                    && (hasRecentScaleUp || hasRecentObservedScaleUp.contains(deploymentId))) {
                     logger.debug("adaptive allocations scaler: skipping scaling down [{}] because of recent scaleup.", deploymentId);
                     continue;
                 }