Merge branch 'main' of https://github.com/elastic/elasticsearch into esql-inference-runner-refactoring

Author: afoucret

docs/reference/query-languages/esql/_snippets/functions/layout/categorize.md

@@ -89,7 +89,8 @@ EngineConfig engineConfigWithLargerIndexingMemory(EngineConfig config) {
                 config.isPromotableToPrimary(),
                 config.getMapperService(),
                 config.getEngineResetLock(),
-                config.getMergeMetrics()
+                config.getMergeMetrics(),
+                config.getIndexDeletionPolicyWrapper()
             );
         }
 

docs/reference/query-languages/esql/_snippets/functions/layout/sample.md

@@ -165,6 +165,10 @@ public boolean hasGlobalBlock(ClusterBlock block) {
         return global.contains(block);
     }
 
+    public boolean hasGlobalBlock(ProjectId projectId, ClusterBlock block) {
+        return global(projectId).contains(block);
+    }
+
     public boolean hasGlobalBlockWithId(final int blockId) {
         for (ClusterBlock clusterBlock : global) {
             if (clusterBlock.id() == blockId) {

docs/reference/query-languages/esql/_snippets/functions/layout/scalb.md

@@ -802,6 +802,28 @@ public <K, V> Map<K, V> readMapValues(final Writeable.Reader<V> valueReader, fin
         return map;
     }
 
+    /**
+     * Reads a multiple {@code V}-values and then converts them to a {@code Map} using keyMapper.
+     *
+     * @param valueReader The value reader
+     * @param keyMapper function to create a key from a value
+     * @param constructor map constructor
+     * @return Never {@code null}.
+     */
+    public <K, V, M extends Map<K, V>> M readMapValues(
+        final Writeable.Reader<V> valueReader,
+        final Function<V, K> keyMapper,
+        final IntFunction<M> constructor
+    ) throws IOException {
+        final int size = readArraySize();
+        final M map = constructor.apply(size);
+        for (int i = 0; i < size; i++) {
+            V value = valueReader.read(this);
+            map.put(keyMapper.apply(value), value);
+        }
+        return map;
+    }
+
     /**
      * If the returned map contains any entries it will be mutable. If it is empty it might be immutable.
      */

docs/reference/query-languages/esql/_snippets/functions/parameters/categorize.md

@@ -71,4 +71,11 @@ public Map<String, String> getUserData() throws IOException {
     public String toString() {
         return "FilterIndexCommit{" + "in=" + in + '}';
     }
+
+    public static IndexCommit unwrap(IndexCommit in) {
+        while (in instanceof FilterIndexCommit) {
+            in = ((FilterIndexCommit) in).getIndexCommit();
+        }
+        return in;
+    }
 }

server/src/internalClusterTest/java/org/elasticsearch/indices/IndexingMemoryControllerIT.java

@@ -38,7 +38,7 @@
  * In particular, this policy will delete index commits whose max sequence number is at most
  * the current global checkpoint except the index commit which has the highest max sequence number among those.
  */
-public class CombinedDeletionPolicy extends IndexDeletionPolicy {
+public class CombinedDeletionPolicy extends ElasticsearchIndexDeletionPolicy {
     private final Logger logger;
     private final TranslogDeletionPolicy translogDeletionPolicy;
     private final SoftDeletesPolicy softDeletesPolicy;
@@ -48,13 +48,6 @@ public class CombinedDeletionPolicy extends IndexDeletionPolicy {
     // when checking for externally acquired index commits that haven't been released
     private final Set<IndexCommit> internallyAcquiredIndexCommits;
 
-    interface CommitsListener {
-
-        void onNewAcquiredCommit(IndexCommit commit, Set<String> additionalFiles);
-
-        void onDeletedCommit(IndexCommit commit);
-    }
-
     @Nullable
     private final CommitsListener commitsListener;
 
@@ -187,7 +180,6 @@ private void deleteCommit(IndexCommit commit) throws IOException {
         assert commit.isDeleted() == false : "Index commit [" + commitDescription(commit) + "] is deleted twice";
         logger.debug("Delete index commit [{}]", commitDescription(commit));
         commit.delete();
-        assert commit.isDeleted() : "Deletion commit [" + commitDescription(commit) + "] was suppressed";
     }
 
     private void updateRetentionPolicy() throws IOException {
@@ -204,7 +196,8 @@ protected int getDocCountOfCommit(IndexCommit indexCommit) throws IOException {
         return SegmentInfos.readCommit(indexCommit.getDirectory(), indexCommit.getSegmentsFileName()).totalMaxDoc();
     }
 
-    SafeCommitInfo getSafeCommitInfo() {
+    @Override
+    public SafeCommitInfo getSafeCommitInfo() {
         return safeCommitInfo;
     }
 
@@ -214,7 +207,8 @@ SafeCommitInfo getSafeCommitInfo() {
      *
      * @param acquiringSafeCommit captures the most recent safe commit point if true; otherwise captures the most recent commit point.
      */
-    synchronized IndexCommit acquireIndexCommit(boolean acquiringSafeCommit) {
+    @Override
+    public synchronized IndexCommit acquireIndexCommit(boolean acquiringSafeCommit) {
         return acquireIndexCommit(acquiringSafeCommit, false);
     }
 
@@ -241,7 +235,8 @@ protected IndexCommit wrapCommit(IndexCommit indexCommit, boolean acquiredIntern
      *
      * @return true if the acquired commit can be clean up.
      */
-    synchronized boolean releaseCommit(final IndexCommit acquiredCommit) {
+    @Override
+    public synchronized boolean releaseIndexCommit(final IndexCommit acquiredCommit) {
         final SnapshotIndexCommit snapshotIndexCommit = (SnapshotIndexCommit) acquiredCommit;
         final IndexCommit releasingCommit = snapshotIndexCommit.getIndexCommit();
         assert acquiredIndexCommits.containsKey(releasingCommit)
@@ -316,7 +311,8 @@ private static Set<String> listOfNewFileNames(IndexCommit previous, IndexCommit
     /**
      * Checks whether the deletion policy is holding on to externally acquired index commits
      */
-    synchronized boolean hasAcquiredIndexCommitsForTesting() {
+    @Override
+    public synchronized boolean hasAcquiredIndexCommitsForTesting() {
         // We explicitly check only external commits and disregard internal commits acquired by the commits listener
         for (var e : acquiredIndexCommits.entrySet()) {
             if (internallyAcquiredIndexCommits.contains(e.getKey()) == false || e.getValue() > 1) {
@@ -329,7 +325,8 @@ synchronized boolean hasAcquiredIndexCommitsForTesting() {
     /**
      * Checks if the deletion policy can delete some index commits with the latest global checkpoint.
      */
-    boolean hasUnreferencedCommits() {
+    @Override
+    public boolean hasUnreferencedCommits() {
         return maxSeqNoOfNextSafeCommit <= globalCheckpointSupplier.getAsLong();
     }
 

server/src/main/java/org/elasticsearch/cluster/block/ClusterBlocks.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.engine;
+
+import org.apache.lucene.index.IndexCommit;
+import org.apache.lucene.index.IndexDeletionPolicy;
+
+import java.util.Set;
+
+public abstract class ElasticsearchIndexDeletionPolicy extends IndexDeletionPolicy {
+
+    /**
+     * Captures the most recent commit point or the most recent safe commit point.
+     * Index files of the capturing commit point won't be released until the commit reference is closed.
+     *
+     * @param acquiringSafeCommit captures the most recent safe commit point if true; otherwise captures the most recent commit point.
+     */
+    public abstract IndexCommit acquireIndexCommit(boolean acquiringSafeCommit);
+
+    /**
+     * Releases an index commit that was acquired by {@link #acquireIndexCommit(boolean)}.
+     *
+     * @return true if the acquired commit can be clean up.
+     */
+    public abstract boolean releaseIndexCommit(IndexCommit acquiredIndexCommit);
+
+    /**
+     * @return information about the safe commit
+     */
+    public abstract SafeCommitInfo getSafeCommitInfo();
+
+    public abstract boolean hasAcquiredIndexCommitsForTesting();
+
+    public abstract boolean hasUnreferencedCommits();
+
+    public interface CommitsListener {
+
+        void onNewAcquiredCommit(IndexCommit commit, Set<String> additionalFiles);
+
+        void onDeletedCommit(IndexCommit commit);
+    }
+}