Merge branch 'main' into http-aggregator-removal

Author: mhl-b

docs/changelog/129170.yaml

@@ -0,0 +1,5 @@
+pr: 129170
+summary: Add Support for LIKE (LIST)
+area: ES|QL
+type: enhancement
+issues: []

docs/reference/query-languages/esql/_snippets/operators/detailedDescription/like.md

@@ -10,6 +10,12 @@ ROW message = "foo * bar"
 ```
 
 
+```esql
+ROW message = "foobar"
+| WHERE message like ("foo*", "bar?")
+```
+
+
 To reduce the overhead of escaping, we suggest using triple quotes strings `"""`
 
 ```esql

docs/reference/query-languages/esql/kibana/definition/operators/like.json

@@ -118,7 +118,8 @@ public void testLookAheadTimeSettingHigherThanTimeSeriesPollIntervalSetting() {
         updateIndexSettings(indexSettings);
     }
 
-    private void updateClusterSettings(Settings settings) {
+    @Override
+    protected void updateClusterSettings(Settings settings) {
         clusterAdmin().updateSettings(
             new ClusterUpdateSettingsRequest(TEST_REQUEST_TIMEOUT, TEST_REQUEST_TIMEOUT).persistentSettings(settings)
         ).actionGet();

docs/reference/query-languages/esql/kibana/definition/operators/not like.json

@@ -264,14 +264,33 @@ public void testExpectedShardSizeIsPresent() throws InterruptedException {
     public void testHeapUsageEstimateIsPresent() {
         InternalClusterInfoService clusterInfoService = (InternalClusterInfoService) getInstanceFromNode(ClusterInfoService.class);
         ClusterInfoServiceUtils.refresh(clusterInfoService);
-        ClusterState state = getInstanceFromNode(ClusterService.class).state();
         Map<String, ShardHeapUsage> shardHeapUsages = clusterInfoService.getClusterInfo().getShardHeapUsages();
         assertNotNull(shardHeapUsages);
-        assertEquals(state.nodes().size(), shardHeapUsages.size());
-        for (DiscoveryNode node : state.nodes()) {
-            assertTrue(shardHeapUsages.containsKey(node.getId()));
-            ShardHeapUsage shardHeapUsage = shardHeapUsages.get(node.getId());
-            assertThat(shardHeapUsage.estimatedFreeBytes(), lessThanOrEqualTo(shardHeapUsage.totalBytes()));
+        // Not collecting yet because it is disabled
+        assertTrue(shardHeapUsages.isEmpty());
+
+        // Enable collection for shard heap usages
+        updateClusterSettings(
+            Settings.builder()
+                .put(InternalClusterInfoService.CLUSTER_ROUTING_ALLOCATION_SHARD_HEAP_THRESHOLD_DECIDER_ENABLED.getKey(), true)
+                .build()
+        );
+        try {
+            ClusterInfoServiceUtils.refresh(clusterInfoService);
+            ClusterState state = getInstanceFromNode(ClusterService.class).state();
+            shardHeapUsages = clusterInfoService.getClusterInfo().getShardHeapUsages();
+            assertEquals(state.nodes().size(), shardHeapUsages.size());
+            for (DiscoveryNode node : state.nodes()) {
+                assertTrue(shardHeapUsages.containsKey(node.getId()));
+                ShardHeapUsage shardHeapUsage = shardHeapUsages.get(node.getId());
+                assertThat(shardHeapUsage.estimatedFreeBytes(), lessThanOrEqualTo(shardHeapUsage.totalBytes()));
+            }
+        } finally {
+            updateClusterSettings(
+                Settings.builder()
+                    .putNull(InternalClusterInfoService.CLUSTER_ROUTING_ALLOCATION_SHARD_HEAP_THRESHOLD_DECIDER_ENABLED.getKey())
+                    .build()
+            );
         }
     }
 

docs/reference/query-languages/esql/kibana/docs/operators/like.md

@@ -33,6 +33,7 @@ public class ClusterInfoSimulator {
     private final CopyOnFirstWriteMap<String, Long> shardSizes;
     private final Map<ShardId, Long> shardDataSetSizes;
     private final Map<NodeAndShard, String> dataPath;
+    private final Map<String, ShardHeapUsage> shardHeapUsages;
 
     public ClusterInfoSimulator(RoutingAllocation allocation) {
         this.allocation = allocation;
@@ -41,6 +42,7 @@ public ClusterInfoSimulator(RoutingAllocation allocation) {
         this.shardSizes = new CopyOnFirstWriteMap<>(allocation.clusterInfo().shardSizes);
         this.shardDataSetSizes = Map.copyOf(allocation.clusterInfo().shardDataSetSizes);
         this.dataPath = Map.copyOf(allocation.clusterInfo().dataPath);
+        this.shardHeapUsages = allocation.clusterInfo().getShardHeapUsages();
     }
 
     /**
@@ -154,7 +156,7 @@ public ClusterInfo getClusterInfo() {
             shardDataSetSizes,
             dataPath,
             Map.of(),
-            Map.of()
+            shardHeapUsages
         );
     }
 }

docs/reference/query-languages/esql/kibana/docs/operators/not like.md

@@ -83,7 +83,15 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
         Property.NodeScope
     );
 
+    public static final Setting<Boolean> CLUSTER_ROUTING_ALLOCATION_SHARD_HEAP_THRESHOLD_DECIDER_ENABLED = Setting.boolSetting(
+        "cluster.routing.allocation.shard_heap.threshold_enabled",
+        false,
+        Property.Dynamic,
+        Property.NodeScope
+    );
+
     private volatile boolean diskThresholdEnabled;
+    private volatile boolean shardHeapThresholdEnabled;
     private volatile TimeValue updateFrequency;
     private volatile TimeValue fetchTimeout;
 
@@ -130,12 +138,20 @@ public InternalClusterInfoService(
             DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_THRESHOLD_ENABLED_SETTING,
             this::setDiskThresholdEnabled
         );
+        clusterSettings.initializeAndWatch(
+            CLUSTER_ROUTING_ALLOCATION_SHARD_HEAP_THRESHOLD_DECIDER_ENABLED,
+            this::setShardHeapThresholdEnabled
+        );
     }
 
     private void setDiskThresholdEnabled(boolean diskThresholdEnabled) {
         this.diskThresholdEnabled = diskThresholdEnabled;
     }
 
+    private void setShardHeapThresholdEnabled(boolean shardHeapThresholdEnabled) {
+        this.shardHeapThresholdEnabled = shardHeapThresholdEnabled;
+    }
+
     private void setFetchTimeout(TimeValue fetchTimeout) {
         this.fetchTimeout = fetchTimeout;
     }
@@ -185,20 +201,44 @@ void execute() {
             logger.trace("starting async refresh");
 
             try (var ignoredRefs = fetchRefs) {
-                if (diskThresholdEnabled) {
-                    try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
-                        fetchIndicesStats();
-                    }
-                } else {
-                    logger.trace("skipping collecting disk usage info from cluster, notifying listeners with empty cluster info");
-                    indicesStatsSummary = IndicesStatsSummary.EMPTY;
+                maybeFetchIndicesStats(diskThresholdEnabled);
+                maybeFetchNodeStats(diskThresholdEnabled || shardHeapThresholdEnabled);
+                maybeFetchNodesHeapUsage(shardHeapThresholdEnabled);
+            }
+        }
+
+        private void maybeFetchIndicesStats(boolean shouldFetch) {
+            if (shouldFetch) {
+                try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
+                    fetchIndicesStats();
                 }
+            } else {
+                logger.trace("skipping collecting disk usage info from cluster, notifying listeners with empty indices stats");
+                indicesStatsSummary = IndicesStatsSummary.EMPTY;
+            }
+        }
+
+        private void maybeFetchNodeStats(boolean shouldFetch) {
+            if (shouldFetch) {
                 try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
                     fetchNodeStats();
                 }
+            } else {
+                logger.trace("skipping collecting node stats from cluster, notifying listeners with empty node stats");
+                leastAvailableSpaceUsages = Map.of();
+                mostAvailableSpaceUsages = Map.of();
+                maxHeapPerNode = Map.of();
+            }
+        }
+
+        private void maybeFetchNodesHeapUsage(boolean shouldFetch) {
+            if (shouldFetch) {
                 try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
                     fetchNodesHeapUsage();
                 }
+            } else {
+                logger.trace("skipping collecting shard heap usage from cluster, notifying listeners with empty shard heap usage");
+                shardHeapUsagePerNode = Map.of();
             }
         }