Merge branch 'main' into Add-Hugging-Face-Rerank-support

Author: Evgenii-Kazannik

plugins/examples/custom-suggester/src/main/java/org/elasticsearch/example/customsuggester/CustomSuggester.java

@@ -11,7 +11,7 @@
 
 import org.apache.lucene.search.IndexSearcher;
 import org.apache.lucene.util.CharsRefBuilder;
-import org.elasticsearch.common.text.Text;
+import org.elasticsearch.xcontent.Text;
 import org.elasticsearch.search.suggest.Suggest;
 import org.elasticsearch.search.suggest.Suggester;
 

plugins/examples/custom-suggester/src/main/java/org/elasticsearch/example/customsuggester/CustomSuggestion.java

@@ -11,7 +11,7 @@
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.text.Text;
+import org.elasticsearch.xcontent.Text;
 import org.elasticsearch.search.suggest.Suggest;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.XContentBuilder;

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexStateService.java

@@ -65,6 +65,7 @@
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.IndexReshardService;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.shard.IndexLongFieldRange;
 import org.elasticsearch.index.shard.ShardId;
@@ -352,6 +353,12 @@ static ClusterState addIndexClosedBlocks(
             );
         }
 
+        // Check if index closing conflicts with ongoing resharding
+        Set<Index> reshardingIndices = IndexReshardService.reshardingIndices(currentProjectState, indicesToClose);
+        if (reshardingIndices.isEmpty() == false) {
+            throw new IllegalArgumentException("Cannot close indices that are being resharded: " + reshardingIndices);
+        }
+
         final ClusterBlocks.Builder blocks = ClusterBlocks.builder(currentState.blocks());
 
         for (Index index : indicesToClose) {
@@ -940,6 +947,22 @@ static Tuple<ClusterState, List<IndexResult>> closeRoutingTable(
                     continue;
                 }
 
+                // Check if index closing conflicts with ongoing resharding
+                Set<Index> reshardingIndices = IndexReshardService.reshardingIndices(currentProjectState, Set.of(index));
+                if (reshardingIndices.isEmpty() == false) {
+                    closingResults.put(
+                        result.getKey(),
+                        new IndexResult(
+                            result.getKey(),
+                            new IllegalStateException(
+                                "verification of shards before closing " + index + " succeeded but index is being resharded in the meantime"
+                            )
+                        )
+                    );
+                    logger.debug("verification of shards before closing {} succeeded but index is being resharded in the meantime", index);
+                    continue;
+                }
+
                 blocks.removeIndexBlockWithId(projectId, index.getName(), INDEX_CLOSED_BLOCK_ID);
                 blocks.addIndexBlock(projectId, index.getName(), INDEX_CLOSED_BLOCK);
                 final IndexMetadata.Builder updatedMetadata = IndexMetadata.builder(indexMetadata).state(IndexMetadata.State.CLOSE);

server/src/main/java/org/elasticsearch/index/IndexReshardService.java

@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+import org.elasticsearch.cluster.ProjectState;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Resharding is currently only implemented in Serverless. This service only exposes minimal metadata about resharding
+ * needed by other services.
+ */
+public class IndexReshardService {
+    /**
+     * Returns the indices from the provided set that are currently being resharded.
+     */
+    public static Set<Index> reshardingIndices(final ProjectState projectState, final Set<Index> indicesToCheck) {
+        final Set<Index> indices = new HashSet<>();
+        for (Index index : indicesToCheck) {
+            IndexMetadata indexMetadata = projectState.metadata().index(index);
+
+            if (indexMetadata != null && indexMetadata.getReshardingMetadata() != null) {
+                indices.add(index);
+            }
+        }
+        return indices;
+    }
+}

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexStateServiceTests.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexVersion;
@@ -45,6 +46,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -164,6 +166,38 @@ public void testCloseRoutingTableWithSnapshottedIndex() {
         assertThat(updatedState.blocks().hasIndexBlockWithId(projectId, index.getName(), INDEX_CLOSED_BLOCK_ID), is(true));
     }
 
+    public void testCloseRoutingTableWithReshardingIndex() {
+        ClusterState state = stateWithProject("testCloseRoutingTableWithReshardingIndex", projectId);
+
+        String indexName = "resharding-index";
+        ClusterBlock block = MetadataIndexStateService.createIndexClosingBlock();
+        state = addOpenedIndex(projectId, indexName, randomIntBetween(1, 5), randomIntBetween(0, 5), state);
+
+        var updatedMetadata = IndexMetadata.builder(state.metadata().getProject(projectId).index(indexName))
+            .reshardingMetadata(IndexReshardingMetadata.newSplitByMultiple(randomIntBetween(1, 5), 2))
+            .build();
+        state = ClusterState.builder(state)
+            .putProjectMetadata(ProjectMetadata.builder(state.metadata().getProject(projectId)).put(updatedMetadata, true))
+            .build();
+
+        state = ClusterState.builder(state)
+            .blocks(ClusterBlocks.builder().blocks(state.blocks()).addIndexBlock(projectId, indexName, block))
+            .build();
+
+        final Index index = state.metadata().getProject(projectId).index(indexName).getIndex();
+        final Tuple<ClusterState, List<IndexResult>> result = MetadataIndexStateService.closeRoutingTable(
+            state,
+            projectId,
+            Map.of(index, block),
+            Map.of(index, new IndexResult(index)),
+            TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY
+        );
+        final ClusterState updatedState = result.v1();
+        assertIsOpened(index.getName(), updatedState, projectId);
+        assertThat(updatedState.blocks().hasIndexBlockWithId(projectId, index.getName(), INDEX_CLOSED_BLOCK_ID), is(true));
+        assertThat(result.v2().get(0).getException(), notNullValue());
+    }
+
     public void testAddIndexClosedBlocks() {
         final ClusterState initialState = stateWithProject("testAddIndexClosedBlocks", projectId);
         {

server/src/test/java/org/elasticsearch/index/IndexReshardServiceTests.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;
+
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.IndexReshardingMetadata;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class IndexReshardServiceTests extends ESTestCase {
+    public void testReshardingIndices() {
+        var indexMetadata1 = IndexMetadata.builder(randomAlphaOfLength(5)).settings(indexSettings(IndexVersion.current(), 1, 0)).build();
+        var indexMetadata2 = IndexMetadata.builder(indexMetadata1.getIndex().getName() + "2")
+            .settings(indexSettings(IndexVersion.current(), 1, 0))
+            .reshardingMetadata(IndexReshardingMetadata.newSplitByMultiple(1, 2))
+            .build();
+
+        var projectId = randomProjectIdOrDefault();
+
+        var projectMetadataBuilder = ProjectMetadata.builder(projectId);
+        projectMetadataBuilder.put(indexMetadata1, false);
+        projectMetadataBuilder.put(indexMetadata2, false);
+
+        final ClusterState clusterState = ClusterState.builder(new ClusterName("testReshardingIndices"))
+            .putProjectMetadata(projectMetadataBuilder)
+            .build();
+
+        Set<Index> indicesToCheck = new HashSet<>();
+        indicesToCheck.add(indexMetadata1.getIndex());
+        indicesToCheck.add(indexMetadata2.getIndex());
+
+        Set<Index> reshardingIndices = IndexReshardService.reshardingIndices(clusterState.projectState(projectId), indicesToCheck);
+
+        assertEquals(1, reshardingIndices.size());
+        assertTrue(reshardingIndices.contains(indexMetadata2.getIndex()));
+    }
+}

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java

@@ -46,7 +46,7 @@
 import static org.elasticsearch.xpack.esql.CsvTestUtils.isEnabled;
 import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.ENRICH_SOURCE_INDICES;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.classpathResources;
-import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.FORK_V3;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.FORK_V4;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.INLINESTATS;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.INLINESTATS_V2;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.INLINESTATS_V7;
@@ -132,7 +132,7 @@ protected void shouldSkipTest(String testName) throws IOException {
         assumeFalse("LOOKUP JOIN not yet supported in CCS", testCase.requiredCapabilities.contains(JOIN_LOOKUP_V12.capabilityName()));
         // Unmapped fields require a coorect capability response from every cluster, which isn't currently implemented.
         assumeFalse("UNMAPPED FIELDS not yet supported in CCS", testCase.requiredCapabilities.contains(UNMAPPED_FIELDS.capabilityName()));
-        assumeFalse("FORK not yet supported in CCS", testCase.requiredCapabilities.contains(FORK_V3.capabilityName()));
+        assumeFalse("FORK not yet supported in CCS", testCase.requiredCapabilities.contains(FORK_V4.capabilityName()));
     }
 
     @Override

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

@@ -3,7 +3,7 @@
 //
 
 simpleFork
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | FORK ( WHERE emp_no == 10001 )
@@ -18,7 +18,7 @@ emp_no:integer | _fork:keyword
 ;
 
 forkWithWhereSortAndLimit
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | FORK ( WHERE hire_date < "1985-03-01T00:00:00Z" | SORT first_name | LIMIT 5 )
@@ -38,7 +38,7 @@ emp_no:integer | first_name:keyword | _fork:keyword
 ;
 
 fiveFork
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | FORK ( WHERE emp_no == 10005 )
@@ -59,7 +59,7 @@ fork5         | 10001
 ;
 
 forkWithWhereSortDescAndLimit
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | FORK ( WHERE hire_date < "1985-03-01T00:00:00Z" | SORT first_name DESC | LIMIT 2 )
@@ -76,7 +76,7 @@ fork2         | 10087          | Xinglin
 ;
 
 forkWithCommonPrefilter
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | WHERE emp_no > 10050
@@ -94,7 +94,7 @@ fork2         | 10100
 ;
 
 forkWithSemanticSearchAndScore
-required_capability: fork_v3
+required_capability: fork_v4
 required_capability: semantic_text_field_caps
 required_capability: metadata_score
 
@@ -114,7 +114,7 @@ fork2         | 6.093784261960139E18  | 2           | all we have to decide is w
 ;
 
 forkWithEvals
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | FORK (WHERE emp_no == 10048 OR emp_no == 10081 | EVAL x = "abc" | EVAL y = 1)
@@ -131,7 +131,7 @@ fork2         | 10087          | def       | null      | 2
 ;
 
 forkWithStats
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | FORK (WHERE emp_no == 10048 OR emp_no == 10081)
@@ -152,7 +152,7 @@ fork4         | null           | 100    | 10001     | null
 ;
 
 forkWithDissect
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | WHERE emp_no == 10048 OR emp_no == 10081
@@ -172,7 +172,7 @@ fork2         | 10081          | Rosen     | 10081     | null      | Zhongwei
 ;
 
 forkWithMixOfCommands
-required_capability: fork_v3
+required_capability: fork_v4
 
 FROM employees
 | WHERE emp_no == 10048 OR emp_no == 10081
@@ -195,3 +195,24 @@ fork3         | 10081          | Rosen     | null      | null      | null
 fork4         | 10048          | abc       | aaa       | null      | null
 fork4         | 10081          | abc       | aaa       | null      | null
 ;
+
+forkWithFiltersOnConstantValues
+required_capability: fork_v4
+
+FROM employees
+| EVAL z = 1
+| WHERE z == 1
+| FORK (WHERE emp_no == 10048 OR emp_no == 10081 | WHERE z - 1 == 0)
+       (WHERE emp_no == 10081 OR emp_no == 10087 | EVAL a = "x" )
+       (STATS x = COUNT(*), y = MAX(emp_no), z = MIN(emp_no) | EVAL a = "y" )
+       (STATS x = COUNT(*), y = MIN(emp_no))
+| WHERE _fork == "fork2" OR a == "y"
+| KEEP _fork, emp_no, x, y, z
+| SORT _fork, emp_no
+;
+
+_fork:keyword | emp_no:integer | x:long | y:integer | z:integer
+fork2         | 10081          | null   | null      | 1
+fork2         | 10087          | null   | null      | 1
+fork3         | null           | 100    | 10100     | 10001
+;

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/ForkIT.java

@@ -585,6 +585,51 @@ public void testWithStatsSimple() {
         }
     }
 
+    public void testWithConditionOnForkField() {
+        var query = """
+                FROM test
+                | FORK ( WHERE content:"fox" | EVAL a = 1)
+                       ( WHERE content:"cat" | EVAL b = 2 )
+                       ( WHERE content:"dog" | EVAL c = 3 )
+                | WHERE _fork == "fork2"
+                | KEEP _fork, id, content, a, b, c
+                | SORT _fork
+            """;
+
+        try (var resp = run(query)) {
+            assertColumnNames(resp.columns(), List.of("_fork", "id", "content", "a", "b", "c"));
+
+            Iterable<Iterable<Object>> expectedValues = List.of(
+                Arrays.stream(new Object[] { "fork2", 5, "There is also a white cat", null, 2, null }).toList()
+            );
+            assertValues(resp.values(), expectedValues);
+        }
+    }
+
+    public void testWithFilteringOnConstantColumn() {
+        var query = """
+                FROM test
+                | FORK ( WHERE content:"fox" | EVAL a = 1)
+                       ( WHERE content:"cat" | EVAL a = 2 )
+                       ( WHERE content:"dog" | EVAL a = 3 )
+                | WHERE a == 3
+                | KEEP _fork, id, content, a
+                | SORT id
+                | LIMIT 3
+            """;
+
+        try (var resp = run(query)) {
+            assertColumnNames(resp.columns(), List.of("_fork", "id", "content", "a"));
+
+            Iterable<Iterable<Object>> expectedValues = List.of(
+                List.of("fork3", 2, "This is a brown dog", 3),
+                List.of("fork3", 3, "This dog is really brown", 3),
+                List.of("fork3", 4, "The dog is brown but this document is very very long", 3)
+            );
+            assertValues(resp.values(), expectedValues);
+        }
+    }
+
     public void testWithEvalWithConflictingTypes() {
         var query = """
                 FROM test

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

@@ -1013,7 +1013,7 @@ public enum Cap {
         /**
          * Support streaming of sub plan results
          */
-        FORK_V3(Build.current().isSnapshot()),
+        FORK_V4(Build.current().isSnapshot()),
 
         /**
          * Support for the {@code leading_zeros} named parameter.