Support for the reactive version of SQL queries.

Signed-off-by: Youssef Aouichaoui <[email protected]>

Author: youssef3wi

src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java

@@ -17,6 +17,7 @@
 
 import static co.elastic.clients.util.ApiTypeHelper.*;
 import static org.springframework.data.elasticsearch.client.elc.TypeUtils.*;
+import static org.springframework.data.elasticsearch.core.sql.types.ResponseFormat.json;
 
 import co.elastic.clients.elasticsearch._types.Result;
 import co.elastic.clients.elasticsearch.core.*;
@@ -25,6 +26,8 @@
 import co.elastic.clients.json.JsonpMapper;
 import co.elastic.clients.transport.Version;
 import co.elastic.clients.transport.endpoints.BooleanResponse;
+import org.springframework.data.elasticsearch.core.query.SqlQuery;
+import org.springframework.data.elasticsearch.core.sql.SqlResponse;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -88,6 +91,7 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch
 	private static final Log LOGGER = LogFactory.getLog(ReactiveElasticsearchTemplate.class);
 
 	private final ReactiveElasticsearchClient client;
+	private final ReactiveElasticsearchSqlClient sqlClient;
 	private final RequestConverter requestConverter;
 	private final ResponseConverter responseConverter;
 	private final JsonpMapper jsonpMapper;
@@ -99,6 +103,7 @@ public ReactiveElasticsearchTemplate(ReactiveElasticsearchClient client, Elastic
 		Assert.notNull(client, "client must not be null");
 
 		this.client = client;
+		this.sqlClient = client.sql();
 		this.jsonpMapper = client._transport().jsonpMapper();
 		requestConverter = new RequestConverter(converter, jsonpMapper);
 		responseConverter = new ResponseConverter(jsonpMapper);
@@ -646,6 +651,16 @@ public BaseQueryBuilder queryBuilderWithIds(List<String> ids) {
 		return NativeQuery.builder().withIds(ids);
 	}
 
+	@Override
+	public Mono<SqlResponse> search(SqlQuery query) {
+		Assert.notNull(query, "Query must not be null.");
+		Assert.isTrue(query.getFormat() == null || json.equals(query.getFormat()),
+				"The Elasticsearch Java Client only supports JSON format.");
+
+		co.elastic.clients.elasticsearch.sql.QueryRequest request = requestConverter.sqlQueryRequest(query);
+		return sqlClient.query(request).onErrorMap(this::translateException).map(responseConverter::sqlResponse);
+	}
+
 	/**
 	 * Callback interface to be used with {@link #execute(ReactiveElasticsearchTemplate.ClientCallback<>)} for operating
 	 * directly on {@link ReactiveElasticsearchClient}.

src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchOperations.java

@@ -21,6 +21,7 @@
 import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
 import org.springframework.data.elasticsearch.core.routing.RoutingResolver;
 import org.springframework.data.elasticsearch.core.script.ReactiveScriptOperations;
+import org.springframework.data.elasticsearch.core.sql.ReactiveSqlOperations;
 import org.springframework.lang.Nullable;
 
 /**
@@ -31,7 +32,7 @@
  * @since 3.2
  */
 public interface ReactiveElasticsearchOperations
-		extends ReactiveDocumentOperations, ReactiveSearchOperations, ReactiveScriptOperations {
+		extends ReactiveDocumentOperations, ReactiveSearchOperations, ReactiveScriptOperations, ReactiveSqlOperations {
 
 	/**
 	 * Get the {@link ElasticsearchConverter} used.

src/main/java/org/springframework/data/elasticsearch/core/sql/ReactiveSqlOperations.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.sql;
+
+import org.springframework.data.elasticsearch.core.query.SqlQuery;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * The reactive version of operations for the
+ * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/sql-search-api.html">SQL search API</a>.
+ *
+ * @author Aouichaoui Youssef
+ * @since 5.4
+ */
+public interface ReactiveSqlOperations {
+	/**
+	 * Execute the criteria {@code query} against elasticsearch and return result as {@link SqlResponse}
+	 *
+	 * @param query the query to execute
+	 * @return {@link SqlResponse} containing the list of found objects
+	 */
+	Mono<SqlResponse> search(SqlQuery query);
+}

src/main/java/org/springframework/data/elasticsearch/core/sql/SqlOperations.java

@@ -17,6 +17,13 @@
 
 import org.springframework.data.elasticsearch.core.query.SqlQuery;
 
+/**
+ * The operations for the
+ * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/sql-search-api.html">SQL search API</a>.
+ *
+ * @author Aouichaoui Youssef
+ * @since 5.4
+ */
 public interface SqlOperations {
 	/**
 	 * Execute the criteria {@code query} against elasticsearch and return result as {@link SqlResponse}

src/test/java/org/springframework/data/elasticsearch/core/sql/ReactiveSqlOperationsIntegrationTests.java

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.core.sql;
+
+import static org.springframework.data.elasticsearch.core.IndexOperationsAdapter.blocking;
+
+import java.util.List;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations;
+import org.springframework.data.elasticsearch.core.query.SqlQuery;
+import org.springframework.data.elasticsearch.junit.jupiter.ReactiveElasticsearchTemplateConfiguration;
+import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest;
+import org.springframework.lang.Nullable;
+import org.springframework.test.context.ContextConfiguration;
+
+import reactor.test.StepVerifier;
+
+/**
+ * Testing the reactive querying using SQL syntax.
+ *
+ * @author Youssef Aouichaoui
+ */
+@SpringIntegrationTest
+@ContextConfiguration(classes = { ReactiveSqlOperationsIntegrationTests.Config.class })
+@DisplayName("Using Elasticsearch SQL Reactive Client")
+public class ReactiveSqlOperationsIntegrationTests {
+	@Autowired ReactiveElasticsearchOperations operations;
+
+	@BeforeEach
+	void setUp() {
+		// create index
+		blocking(operations.indexOps(EntityForSQL.class)).createWithMapping();
+
+		// add data
+		operations
+				.saveAll(List.of(EntityForSQL.builder().withViews(3).build(), EntityForSQL.builder().withViews(0).build()),
+						EntityForSQL.class)
+				.blockLast();
+	}
+
+	@AfterEach
+	void tearDown() {
+		// delete index
+		blocking(operations.indexOps(EntityForSQL.class)).delete();
+	}
+
+	// begin configuration region
+	@Configuration
+	@Import({ ReactiveElasticsearchTemplateConfiguration.class })
+	static class Config {}
+	// end region
+
+	@Test
+	void when_search_with_an_sql_query() {
+		// Given
+		SqlQuery query = SqlQuery.builder("SELECT * FROM entity_for_sql WHERE views = 0").build();
+
+		// When
+
+		// Then
+		operations.search(query).as(StepVerifier::create).expectNextCount(1).verifyComplete();
+	}
+
+	// begin region
+	@Document(indexName = "entity_for_sql")
+	static class EntityForSQL {
+		@Id private String id;
+		private final Integer views;
+
+		public EntityForSQL(EntityForSQL.Builder builder) {
+			this.views = builder.views;
+		}
+
+		@Nullable
+		public String getId() {
+			return id;
+		}
+
+		public Integer getViews() {
+			return views;
+		}
+
+		public static EntityForSQL.Builder builder() {
+			return new EntityForSQL.Builder();
+		}
+
+		static class Builder {
+			private Integer views = 0;
+
+			public EntityForSQL.Builder withViews(Integer views) {
+				this.views = views;
+
+				return this;
+			}
+
+			public EntityForSQL build() {
+				return new EntityForSQL(this);
+			}
+		}
+	}
+	// end region
+}

src/test/java/org/springframework/data/elasticsearch/core/sql/SqlOperationsIntegrationTests.java

@@ -103,7 +103,6 @@ void when_search_with_an_sql_query_and_txt_format() {
 	// begin region
 	@Document(indexName = "entity_for_sql")
 	static class EntityForSQL {
-		@Nullable
 		@Id private String id;
 		private final Integer views;
 
@@ -116,10 +115,6 @@ public String getId() {
 			return id;
 		}
 
-		public void setId(@Nullable String id) {
-			this.id = id;
-		}
-
 		public Integer getViews() {
 			return views;
 		}
@@ -129,7 +124,7 @@ public static Builder builder() {
 		}
 
 		static class Builder {
-			private Integer views;
+			private Integer views = 0;
 
 			public Builder withViews(Integer views) {
 				this.views = views;