Allow IndexCoordinates in repositories

It is common to have multiple indexes with the same document structure.
E.g. rolling indices over time and having a wildcard alias.
Sometimes one wants to select different IndexCoordinates than specified in
`@Document(indexName = "...", ...)`. Also you might want to use the same
Repository to do cross cluster searches and for this need to specify the IndexCoordinates.

It would therefore be helpful to support IndexCoordinates as an argument to repository methods
the same way ScrollPosition, Sort and Pageable are handled.

This PR is an attempt to introduce this change. I keep it as draft to get some comments
I can add tests to it if we can agree on the change being valid.

src/main/java/org/springframework/data/elasticsearch/repository/query/AbstractElasticsearchRepositoryQuery.java

@@ -79,11 +79,13 @@ public Object execute(Object[] parameters) {
 
 		Query query = createQuery(parameters);
 
-		IndexCoordinates index = elasticsearchOperations.getIndexCoordinatesFor(clazz);
+		IndexCoordinates index = parameterAccessor
+				.getIndexCoordinatesOrDefaults(elasticsearchOperations.getIndexCoordinatesFor(clazz));
 
 		Object result = null;
 
 		if (isDeleteQuery()) {
+			index = elasticsearchOperations.getIndexCoordinatesFor(clazz);
 			result = countOrGetDocumentsForDelete(query, parameterAccessor);
 			elasticsearchOperations.delete(query, clazz, index);
 			elasticsearchOperations.indexOps(index).refresh();

src/main/java/org/springframework/data/elasticsearch/repository/query/AbstractReactiveElasticsearchRepositoryQuery.java

@@ -98,7 +98,7 @@ private Object execute(ElasticsearchParametersParameterAccessor parameterAccesso
 		queryMethod.addMethodParameter(query, parameterAccessor, elasticsearchOperations.getElasticsearchConverter());
 
 		String indexName = queryMethod.getEntityInformation().getIndexName();
-		IndexCoordinates index = IndexCoordinates.of(indexName);
+		IndexCoordinates index = parameterAccessor.getIndexCoordinatesOrDefaults(IndexCoordinates.of(indexName));
 
 		ReactiveElasticsearchQueryExecution execution = getExecution(parameterAccessor,
 				new ResultProcessingConverter(processor));

src/main/java/org/springframework/data/elasticsearch/repository/query/ElasticsearchParameter.java

@@ -16,6 +16,7 @@
 package org.springframework.data.elasticsearch.repository.query;
 
 import org.springframework.core.MethodParameter;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.query.RuntimeField;
 import org.springframework.data.elasticsearch.core.query.ScriptedField;
 import org.springframework.data.repository.query.Parameter;
@@ -42,7 +43,8 @@ class ElasticsearchParameter extends Parameter {
 
 	@Override
 	public boolean isSpecialParameter() {
-		return super.isSpecialParameter() || isScriptedFieldParameter() || isRuntimeFieldParameter();
+		return super.isSpecialParameter() || isScriptedFieldParameter() || isRuntimeFieldParameter()
+				|| isIndexCoordinatesParameter();
 	}
 
 	public Boolean isScriptedFieldParameter() {
@@ -52,4 +54,7 @@ public Boolean isScriptedFieldParameter() {
 	public Boolean isRuntimeFieldParameter() {
 		return RuntimeField.class.isAssignableFrom(getType());
 	}
+	public Boolean isIndexCoordinatesParameter() {
+			return IndexCoordinates.class.isAssignableFrom(getType());
+	}
 }

src/main/java/org/springframework/data/elasticsearch/repository/query/ElasticsearchParameterAccessor.java

@@ -15,7 +15,9 @@
  */
 package org.springframework.data.elasticsearch.repository.query;
 
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.lang.NonNull;
 
 /**
  * @author Christoph Strobl
@@ -29,4 +31,6 @@ public interface ElasticsearchParameterAccessor extends ParameterAccessor {
 	 * @return
 	 */
 	Object[] getValues();
+
+	IndexCoordinates getIndexCoordinatesOrDefaults(@NonNull IndexCoordinates defaults);
 }

src/main/java/org/springframework/data/elasticsearch/repository/query/ElasticsearchParameters.java

@@ -29,10 +29,11 @@
  * @since 3.2
  */
 public class ElasticsearchParameters extends Parameters<ElasticsearchParameters, ElasticsearchParameter> {
-
 	private final List<ElasticsearchParameter> scriptedFields = new ArrayList<>();
 	private final List<ElasticsearchParameter> runtimeFields = new ArrayList<>();
 
+	private final int indexCoordinatesIndex;
+
 	public ElasticsearchParameters(Method method, TypeInformation<?> domainType) {
 
 		super(method, parameter -> new ElasticsearchParameter(parameter, domainType));
@@ -50,17 +51,35 @@ public ElasticsearchParameters(Method method, TypeInformation<?> domainType) {
 				runtimeFields.add(parameter);
 			}
 		}
+		this.indexCoordinatesIndex = initIndexCoordinatesIndex();
+	}
 
+	private int initIndexCoordinatesIndex() {
+		int index = 0;
+		List<Integer> foundIndices = new ArrayList<>();
+		for (ElasticsearchParameter parameter : this) {
+			if (parameter.isIndexCoordinatesParameter()) {
+				foundIndices.add(index);
+			}
+			index++;
+		}
+		if (foundIndices.size() > 1) {
+			throw new IllegalArgumentException(this + " can only contain at most one IndexCoordinates parameter.");
+		}
+		return foundIndices.isEmpty() ? -1 : foundIndices.get(0);
 	}
 
 	private ElasticsearchParameter parameterFactory(MethodParameter methodParameter, TypeInformation<?> domainType) {
 		return new ElasticsearchParameter(methodParameter, domainType);
 	}
 
+
 	private ElasticsearchParameters(List<ElasticsearchParameter> parameters) {
 		super(parameters);
+		this.indexCoordinatesIndex = initIndexCoordinatesIndex();
 	}
 
+
 	@Override
 	protected ElasticsearchParameters createFrom(List<ElasticsearchParameter> parameters) {
 		return new ElasticsearchParameters(parameters);
@@ -73,4 +92,12 @@ List<ElasticsearchParameter> getScriptedFields() {
 	List<ElasticsearchParameter> getRuntimeFields() {
 		return runtimeFields;
 	}
+
+	public boolean hasIndexCoordinatesParameter() {
+		return this.indexCoordinatesIndex != -1;
+	}
+
+	public int getIndexCoordinatesIndex() {
+		return indexCoordinatesIndex;
+	}
 }

src/main/java/org/springframework/data/elasticsearch/repository/query/ElasticsearchParametersParameterAccessor.java

@@ -15,7 +15,13 @@
  */
 package org.springframework.data.elasticsearch.repository.query;
 
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
+import org.springframework.data.repository.query.Parameters;
 import org.springframework.data.repository.query.ParametersParameterAccessor;
+import org.springframework.lang.NonNull;
 
 /**
  * @author Christoph Strobl
@@ -25,6 +31,7 @@ class ElasticsearchParametersParameterAccessor extends ParametersParameterAccess
 		implements ElasticsearchParameterAccessor {
 
 	private final Object[] values;
+	private final ElasticsearchParameters eleasticSearchParameters;
 
 	/**
 	 * Creates a new {@link ElasticsearchParametersParameterAccessor}.
@@ -36,10 +43,20 @@ class ElasticsearchParametersParameterAccessor extends ParametersParameterAccess
 
 		super(method.getParameters(), values);
 		this.values = values;
+		this.eleasticSearchParameters = method.getParameters();
 	}
 
 	@Override
 	public Object[] getValues() {
 		return values;
 	}
+
+
+	@Override
+	public IndexCoordinates getIndexCoordinatesOrDefaults(@NonNull IndexCoordinates defaults) {
+		if (!eleasticSearchParameters.hasIndexCoordinatesParameter()) {
+			return defaults;
+		}
+		return (IndexCoordinates) getValues()[eleasticSearchParameters.getIndexCoordinatesIndex()];
+	}
 }

src/main/java/org/springframework/data/elasticsearch/repository/query/ElasticsearchQueryMethod.java

@@ -340,6 +340,11 @@ private String[] mapParameters(String[] source, ParameterAccessor parameterAcces
 		return fieldNames.toArray(new String[0]);
 	}
 
+	@Override
+	public ElasticsearchParameters getParameters() {
+		return (ElasticsearchParameters) super.getParameters();
+	}
+
 	// region Copied from QueryMethod base class
 	/*
 	 * Copied from the QueryMethod class adding support for collections of SearchHit instances. No static method here.

src/main/java/org/springframework/data/elasticsearch/repository/query/ReactiveElasticsearchQueryMethod.java

@@ -142,11 +142,6 @@ public boolean isStreamQuery() {
 		return true;
 	}
 
-	@Override
-	public ElasticsearchParameters getParameters() {
-		return (ElasticsearchParameters) super.getParameters();
-	}
-
 	@Override
 	protected boolean isAllowedGenericType(ParameterizedType methodGenericReturnType) {
 		return super.isAllowedGenericType(methodGenericReturnType)

src/test/java/org/springframework/data/elasticsearch/repository/query/StubParameterAccessor.java

@@ -17,12 +17,13 @@
 
 import java.util.Arrays;
 import java.util.Iterator;
-import java.util.Optional;
 
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.ScrollPosition;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.repository.query.ParameterAccessor;
+import org.springframework.lang.NonNull;
 
 /**
  * Simple {@link ParameterAccessor} that returns the given parameters unfiltered.
@@ -97,6 +98,11 @@ public Object[] getValues() {
 		return this.values;
 	}
 
+	@Override
+	public IndexCoordinates getIndexCoordinatesOrDefaults(@NonNull IndexCoordinates defaults) {
+		return defaults;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.query.ParameterAccessor#findDynamicProjection()