Merge branch '8.16' into patch-1

Author: hanjongho

docs/changelog/117297.yaml

@@ -0,0 +1,5 @@
+pr: 117297
+summary: Fix CCS exchange when multi cluster aliases point to same cluster
+area: ES|QL
+type: bug
+issues: []

docs/changelog/117316.yaml

@@ -0,0 +1,5 @@
+pr: 117316
+summary: Fix validation of SORT by aggregate functions
+area: ES|QL
+type: bug
+issues: []

docs/reference/esql/functions/aggregation-functions.asciidoc

@@ -20,7 +20,7 @@ The <<esql-stats-by>> command supports these aggregate functions:
 * <<esql-sum>>
 * <<esql-top>>
 * <<esql-values>>
-* experimental:[] <<esql-weighted_avg>>
+* <<esql-weighted_avg>>
 // end::agg_list[]
 
 include::layout/avg.asciidoc[]

docs/reference/esql/functions/mv-functions.asciidoc

@@ -19,6 +19,7 @@
 * <<esql-mv_median>>
 * <<esql-mv_median_absolute_deviation>>
 * <<esql-mv_min>>
+* <<esql-mv_percentile>>
 * <<esql-mv_pseries_weighted_sum>>
 * <<esql-mv_sort>>
 * <<esql-mv_slice>>
@@ -37,6 +38,7 @@ include::layout/mv_max.asciidoc[]
 include::layout/mv_median.asciidoc[]
 include::layout/mv_median_absolute_deviation.asciidoc[]
 include::layout/mv_min.asciidoc[]
+include::layout/mv_percentile.asciidoc[]
 include::layout/mv_pseries_weighted_sum.asciidoc[]
 include::layout/mv_slice.asciidoc[]
 include::layout/mv_sort.asciidoc[]

docs/reference/esql/processing-commands/stats.asciidoc

@@ -1,16 +1,18 @@
 [discrete]
 [[esql-stats-by]]
-=== `STATS ... BY`
+=== `STATS`
 
-The `STATS ... BY` processing command groups rows according to a common value
+The `STATS` processing command groups rows according to a common value
 and calculates one or more aggregated values over the grouped rows.
 
 **Syntax**
 
 [source,esql]
 ----
-STATS [column1 =] expression1[, ..., [columnN =] expressionN]
-[BY grouping_expression1[, ..., grouping_expressionN]]
+STATS [column1 =] expression1 [WHERE boolean_expression1][,
+      ...,
+      [columnN =] expressionN [WHERE boolean_expressionN]]
+      [BY grouping_expression1[, ..., grouping_expressionN]]
 ----
 
 *Parameters*
@@ -28,14 +30,18 @@ An expression that computes an aggregated value.
 An expression that outputs the values to group by.
 If its name coincides with one of the computed columns, that column will be ignored.
 
+`boolean_expressionX`::
+The condition that must be met for a row to be included in the evaluation of `expressionX`.
+
 NOTE: Individual `null` values are skipped when computing aggregations.
 
 *Description*
 
-The `STATS ... BY` processing command groups rows according to a common value
-and calculate one or more aggregated values over the grouped rows. If `BY` is
-omitted, the output table contains exactly one row with the aggregations applied
-over the entire dataset.
+The `STATS` processing command groups rows according to a common value
+and calculates one or more aggregated values over the grouped rows. For the
+calculation of each aggregated value, the rows in a group can be filtered with
+`WHERE`. If `BY` is omitted, the output table contains exactly one row with
+the aggregations applied over the entire dataset.
 
 The following <<esql-agg-functions,aggregation functions>> are supported:
 
@@ -90,6 +96,29 @@ include::{esql-specs}/stats.csv-spec[tag=statsCalcMultipleValues]
 include::{esql-specs}/stats.csv-spec[tag=statsCalcMultipleValues-result]
 |===
 
+To filter the rows that go into an aggregation, use the `WHERE` clause:
+
+[source.merge.styled,esql]
+----
+include::{esql-specs}/stats.csv-spec[tag=aggFiltering]
+----
+[%header.monospaced.styled,format=dsv,separator=|]
+|===
+include::{esql-specs}/stats.csv-spec[tag=aggFiltering-result]
+|===
+
+The aggregations can be mixed, with and without a filter and grouping is
+optional as well:
+
+[source.merge.styled,esql]
+----
+include::{esql-specs}/stats.csv-spec[tag=aggFilteringNoGroup]
+----
+[%header.monospaced.styled,format=dsv,separator=|]
+|===
+include::{esql-specs}/stats.csv-spec[tag=aggFilteringNoGroup-result]
+|===
+
 [[esql-stats-mv-group]]
 If the grouping key is multivalued then the input row is in all groups:
 
@@ -109,7 +138,7 @@ It's also possible to group by multiple values:
 include::{esql-specs}/stats.csv-spec[tag=statsGroupByMultipleValues]
 ----
 
-If the all grouping keys are multivalued then the input row is in all groups:
+If all the grouping keys are multivalued then the input row is in all groups:
 
 [source.merge.styled,esql]
 ----
@@ -121,7 +150,7 @@ include::{esql-specs}/stats.csv-spec[tag=multi-mv-group-result]
 |===
 
 Both the aggregating functions and the grouping expressions accept other
-functions. This is useful for using `STATS...BY` on multivalue columns.
+functions. This is useful for using `STATS` on multivalue columns.
 For example, to calculate the average salary change, you can use `MV_AVG` to
 first average the multiple values per employee, and use the result with the
 `AVG` function:

docs/reference/release-notes/8.16.1.asciidoc

@@ -1,8 +1,6 @@
 [[release-notes-8.16.1]]
 == {es} version 8.16.1
 
-coming[8.16.1]
-
 Also see <<breaking-changes-8.16,Breaking changes in 8.16>>.
 
 [[bug-8.16.1]]

muted-tests.yml

@@ -286,6 +286,9 @@ tests:
 - class: org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderIT
   method: testEnterpriseDownloaderTask
   issue: https://github.com/elastic/elasticsearch/issues/115163
+- class: org.elasticsearch.xpack.inference.InferenceRestIT
+  method: test {p0=inference/40_semantic_text_query/Query a field that uses the default ELSER 2 endpoint}
+  issue: https://github.com/elastic/elasticsearch/issues/117027
 
 # Examples:
 #

test/framework/src/main/java/org/elasticsearch/test/AbstractMultiClustersTestCase.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.plugins.Plugin;
@@ -44,6 +45,7 @@
 
 import static org.elasticsearch.discovery.DiscoveryModule.DISCOVERY_SEED_PROVIDERS_SETTING;
 import static org.elasticsearch.discovery.SettingsBasedSeedHostsProvider.DISCOVERY_SEED_HOSTS_SETTING;
+import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.not;
@@ -149,19 +151,23 @@ public static void stopClusters() throws IOException {
     }
 
     protected void disconnectFromRemoteClusters() throws Exception {
-        Settings.Builder settings = Settings.builder();
         final Set<String> clusterAliases = clusterGroup.clusterAliases();
         for (String clusterAlias : clusterAliases) {
             if (clusterAlias.equals(LOCAL_CLUSTER) == false) {
-                settings.putNull("cluster.remote." + clusterAlias + ".seeds");
-                settings.putNull("cluster.remote." + clusterAlias + ".mode");
-                settings.putNull("cluster.remote." + clusterAlias + ".proxy_address");
+                removeRemoteCluster(clusterAlias);
             }
         }
+    }
+
+    protected void removeRemoteCluster(String clusterAlias) throws Exception {
+        Settings.Builder settings = Settings.builder();
+        settings.putNull("cluster.remote." + clusterAlias + ".seeds");
+        settings.putNull("cluster.remote." + clusterAlias + ".mode");
+        settings.putNull("cluster.remote." + clusterAlias + ".proxy_address");
         client().admin().cluster().prepareUpdateSettings(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT).setPersistentSettings(settings).get();
         assertBusy(() -> {
             for (TransportService transportService : cluster(LOCAL_CLUSTER).getInstances(TransportService.class)) {
-                assertThat(transportService.getRemoteClusterService().getRegisteredRemoteClusterNames(), empty());
+                assertThat(transportService.getRemoteClusterService().getRegisteredRemoteClusterNames(), not(contains(clusterAlias)));
             }
         });
     }
@@ -178,12 +184,17 @@ protected void configureAndConnectsToRemoteClusters() throws Exception {
     }
 
     protected void configureRemoteCluster(String clusterAlias, Collection<String> seedNodes) throws Exception {
-        final String remoteClusterSettingPrefix = "cluster.remote." + clusterAlias + ".";
-        Settings.Builder settings = Settings.builder();
-        final List<String> seedAddresses = seedNodes.stream().map(node -> {
+        final var seedAddresses = seedNodes.stream().map(node -> {
             final TransportService transportService = cluster(clusterAlias).getInstance(TransportService.class, node);
-            return transportService.boundAddress().publishAddress().toString();
+            return transportService.boundAddress().publishAddress();
         }).toList();
+        configureRemoteClusterWithSeedAddresses(clusterAlias, seedAddresses);
+    }
+
+    protected void configureRemoteClusterWithSeedAddresses(String clusterAlias, Collection<TransportAddress> seedNodes) throws Exception {
+        final String remoteClusterSettingPrefix = "cluster.remote." + clusterAlias + ".";
+        Settings.Builder settings = Settings.builder();
+        final List<String> seedAddresses = seedNodes.stream().map(TransportAddress::toString).toList();
         boolean skipUnavailable = skipUnavailableForRemoteClusters().containsKey(clusterAlias)
             ? skipUnavailableForRemoteClusters().get(clusterAlias)
             : DEFAULT_SKIP_UNAVAILABLE;

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java

@@ -40,6 +40,7 @@
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -339,6 +340,10 @@ public boolean isEmpty() {
         return sinks.isEmpty();
     }
 
+    public Set<String> sinkKeys() {
+        return sinks.keySet();
+    }
+
     @Override
     protected void doStart() {
 

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

@@ -2318,6 +2318,42 @@ v:integer | job_positions:keyword
     10094 | Accountant
 ;
 
+docsStatsWithSimpleFiltering
+required_capability: per_agg_filtering
+// tag::aggFiltering[]
+FROM employees
+| STATS avg50s = AVG(salary)::LONG WHERE birth_date < "1960-01-01",
+        avg60s = AVG(salary)::LONG WHERE birth_date >= "1960-01-01"
+        BY gender
+| SORT gender
+// end::aggFiltering[]
+| WHERE gender IS NOT NULL
+;
+
+// tag::aggFiltering-result[]
+avg50s:long    |avg60s:long    |gender:keyword
+55462          |46637          |F
+48279          |44879          |M
+// end::aggFiltering-result[]
+;
+
+docsStatsWithFilteringNoGroups
+required_capability: per_agg_filtering
+// tag::aggFilteringNoGroup[]
+FROM employees
+| EVAL Ks = salary / 1000 // thousands
+| STATS under_40K = COUNT(*) WHERE Ks < 40,
+        inbetween = COUNT(*) WHERE 40 <= Ks AND Ks < 60,
+        over_60K  = COUNT(*) WHERE 60 <= Ks,
+        total     = COUNT(*)
+// end::aggFilteringNoGroup[]
+;
+
+// tag::aggFilteringNoGroup-result[]
+under_40K:long |inbetween:long |over_60K:long  |total:long
+36             |39             |25             |100
+// end::aggFilteringNoGroup-result[]
+;
 
 statsWithFiltering
 required_capability: per_agg_filtering