[8.19] Align LINQ-to-ES|QL integration with Elastic.Esql 0.11.0 (#8886) (#8887)

Co-authored-by: Florian Bernd <git@flobernd.de>

Author: github-actions[bot]

src/Elastic.Clients.Elasticsearch/Elastic.Clients.Elasticsearch.csproj

@@ -31,7 +31,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Elastic.Esql" Version="0.10.0" />
+    <PackageReference Include="Elastic.Esql" Version="0.11.0" />
     <PackageReference Include="Elastic.Transport" Version="0.16.0" />
     <PackageReference Include="PolySharp" Version="1.15.0">
       <PrivateAssets>all</PrivateAssets>

src/Elastic.Clients.Elasticsearch/_Shared/Api/Esql/EsqlQueryExecutor.cs

@@ -12,6 +12,7 @@
 using Elastic.Esql.Execution;
 using Elastic.Esql.QueryModel;
 using Elastic.Transport;
+using Elastic.Transport.Products.Elasticsearch;
 
 #if NET10_0_OR_GREATER
 using System.IO.Pipelines;
@@ -37,7 +38,7 @@ public IEsqlResponse ExecuteQuery(string esql, EsqlParameters? parameters, objec
 		var queryOptions = ResolveOptions(options);
 		var request = BuildQueryRequest(esql, parameters, queryOptions);
 		request.BeforeRequest();
-		var response = _client.DoRequest<EsqlQueryRequest, StreamResponse, EsqlQueryRequestParameters>(request);
+		var response = _client.DoRequest<EsqlQueryRequest, ElasticsearchStreamResponse, EsqlQueryRequestParameters>(request);
 		return new EsqlTransportResponse(response);
 	}
 
@@ -46,8 +47,13 @@ public async Task<IEsqlAsyncResponse> ExecuteQueryAsync(string esql, EsqlParamet
 		var queryOptions = ResolveOptions(options);
 		var request = BuildQueryRequest(esql, parameters, queryOptions);
 		request.BeforeRequest();
-		var response = await _client.DoRequestAsync<EsqlQueryRequest, StreamResponse, EsqlQueryRequestParameters>(request, cancellationToken)
+#if NET10_0_OR_GREATER
+		var response = await _client.DoRequestAsync<EsqlQueryRequest, ElasticsearchPipeResponse, EsqlQueryRequestParameters>(request, cancellationToken)
+			.ConfigureAwait(false);
+#else
+		var response = await _client.DoRequestAsync<EsqlQueryRequest, ElasticsearchStreamResponse, EsqlQueryRequestParameters>(request, cancellationToken)
 			.ConfigureAwait(false);
+#endif
 		return new EsqlTransportAsyncResponse(response);
 	}
 
@@ -56,7 +62,7 @@ public IEsqlResponse SubmitAsyncQuery(string esql, EsqlParameters? parameters, o
 		var queryOptions = ResolveOptions(options);
 		var request = BuildAsyncQueryRequest(esql, parameters, queryOptions, asyncOptions);
 		request.BeforeRequest();
-		var response = _client.DoRequest<AsyncQueryRequest, StreamResponse, AsyncQueryRequestParameters>(request);
+		var response = _client.DoRequest<AsyncQueryRequest, ElasticsearchStreamResponse, AsyncQueryRequestParameters>(request);
 		return new EsqlTransportResponse(response);
 	}
 
@@ -65,8 +71,13 @@ public async Task<IEsqlAsyncResponse> SubmitAsyncQueryAsync(string esql, EsqlPar
 		var queryOptions = ResolveOptions(options);
 		var request = BuildAsyncQueryRequest(esql, parameters, queryOptions, asyncOptions);
 		request.BeforeRequest();
-		var response = await _client.DoRequestAsync<AsyncQueryRequest, StreamResponse, AsyncQueryRequestParameters>(request, cancellationToken)
+#if NET10_0_OR_GREATER
+		var response = await _client.DoRequestAsync<AsyncQueryRequest, ElasticsearchPipeResponse, AsyncQueryRequestParameters>(request, cancellationToken)
 			.ConfigureAwait(false);
+#else
+		var response = await _client.DoRequestAsync<AsyncQueryRequest, ElasticsearchStreamResponse, AsyncQueryRequestParameters>(request, cancellationToken)
+			.ConfigureAwait(false);
+#endif
 		return new EsqlTransportAsyncResponse(response);
 	}
 
@@ -77,7 +88,7 @@ public IEsqlResponse PollAsyncQuery(string queryId, object? options)
 		if (queryOptions?.RequestConfiguration is not null)
 			request.RequestConfiguration = queryOptions.RequestConfiguration;
 		request.BeforeRequest();
-		var response = _client.DoRequest<AsyncQueryGetRequest, StreamResponse, AsyncQueryGetRequestParameters>(request);
+		var response = _client.DoRequest<AsyncQueryGetRequest, ElasticsearchStreamResponse, AsyncQueryGetRequestParameters>(request);
 		return new EsqlTransportResponse(response);
 	}
 
@@ -88,8 +99,13 @@ public async Task<IEsqlAsyncResponse> PollAsyncQueryAsync(string queryId, object
 		if (queryOptions?.RequestConfiguration is not null)
 			request.RequestConfiguration = queryOptions.RequestConfiguration;
 		request.BeforeRequest();
-		var response = await _client.DoRequestAsync<AsyncQueryGetRequest, StreamResponse, AsyncQueryGetRequestParameters>(request, cancellationToken)
+#if NET10_0_OR_GREATER
+		var response = await _client.DoRequestAsync<AsyncQueryGetRequest, ElasticsearchPipeResponse, AsyncQueryGetRequestParameters>(request, cancellationToken)
 			.ConfigureAwait(false);
+#else
+		var response = await _client.DoRequestAsync<AsyncQueryGetRequest, ElasticsearchStreamResponse, AsyncQueryGetRequestParameters>(request, cancellationToken)
+			.ConfigureAwait(false);
+#endif
 		return new EsqlTransportAsyncResponse(response);
 	}
 
@@ -231,9 +247,9 @@ JsonValueKind.Number when element.TryGetInt64(out var l) => FieldValue.Long(l),
 
 internal sealed class EsqlTransportResponse : IEsqlResponse
 {
-	private readonly StreamResponse _response;
+	private readonly ElasticsearchStreamResponse _response;
 
-	public EsqlTransportResponse(StreamResponse response) => _response = response;
+	public EsqlTransportResponse(ElasticsearchStreamResponse response) => _response = response;
 
 	public Stream Body => _response.Body;
 
@@ -243,28 +259,21 @@ internal sealed class EsqlTransportResponse : IEsqlResponse
 #if NET10_0_OR_GREATER
 internal sealed class EsqlTransportAsyncResponse : IEsqlAsyncResponse
 {
-	private readonly StreamResponse _response;
+	private readonly ElasticsearchPipeResponse _response;
 
-	public EsqlTransportAsyncResponse(StreamResponse response)
-	{
-		_response = response;
-		Body = PipeReader.Create(response.Body);
-	}
+	public EsqlTransportAsyncResponse(ElasticsearchPipeResponse response) => _response = response;
 
-	public PipeReader Body { get; }
+	public PipeReader Body => _response.Body;
 
-	public async ValueTask DisposeAsync()
-	{
-		await Body.CompleteAsync().ConfigureAwait(false);
-		_response.Dispose();
-	}
+	public async ValueTask DisposeAsync() =>
+		await _response.DisposeAsync().ConfigureAwait(false);
 }
 #else
 internal sealed class EsqlTransportAsyncResponse : IEsqlAsyncResponse
 {
-	private readonly StreamResponse _response;
+	private readonly ElasticsearchStreamResponse _response;
 
-	public EsqlTransportAsyncResponse(StreamResponse response) => _response = response;
+	public EsqlTransportAsyncResponse(ElasticsearchStreamResponse response) => _response = response;
 
 	public Stream Body => _response.Body;
 

src/Elastic.Clients.Elasticsearch/_Shared/Api/Esql/EsqlQueryRequest.cs

@@ -41,31 +41,31 @@ public partial class EsqlNamespacedClient
 	/// </summary>
 	/// <returns>The ES|QL query result as a generic stream response.</returns>
 	/// <remarks>The response must be disposed after use.</remarks>
-	public virtual Task<StreamResponse> QueryAsStreamAsync<TDocument>(
+	public virtual Task<ElasticsearchStreamResponse> QueryAsStreamAsync<TDocument>(
 		Action<EsqlQueryRequestDescriptor<TDocument>> configureRequest,
 		CancellationToken cancellationToken = default)
 	{
 		var descriptor = new EsqlQueryRequestDescriptor<TDocument>();
 		configureRequest?.Invoke(descriptor);
 		var request = descriptor.Instance;
 		request.BeforeRequest();
-		return DoRequestAsync<EsqlQueryRequest, StreamResponse, EsqlQueryRequestParameters>(request, cancellationToken);
+		return DoRequestAsync<EsqlQueryRequest, ElasticsearchStreamResponse, EsqlQueryRequestParameters>(request, cancellationToken);
 	}
 
 	/// <summary>
 	/// Executes an ES|QL request and returns the response as a stream.
 	/// </summary>
 	/// <returns>The ES|QL query result as a generic stream response.</returns>
 	/// <remarks>The response must be disposed after use.</remarks>
-	public virtual Task<StreamResponse> QueryAsStreamAsync(
+	public virtual Task<ElasticsearchStreamResponse> QueryAsStreamAsync(
 		Action<EsqlQueryRequestDescriptor> configureRequest,
 		CancellationToken cancellationToken = default)
 	{
 		var descriptor = new EsqlQueryRequestDescriptor();
 		configureRequest?.Invoke(descriptor);
 		var request = descriptor.Instance;
 		request.BeforeRequest();
-		return DoRequestAsync<EsqlQueryRequest, StreamResponse, EsqlQueryRequestParameters>(request, cancellationToken);
+		return DoRequestAsync<EsqlQueryRequest, ElasticsearchStreamResponse, EsqlQueryRequestParameters>(request, cancellationToken);
 	}
 
 	#endregion

src/Elastic.Clients.Elasticsearch/_Shared/Api/Esql/EsqlQueryResponse.cs

@@ -436,7 +436,6 @@ protected ConnectionConfigurationBase(
 		: base(nodePool, requestInvoker, serializer, registration ?? new ElasticsearchProductRegistration(typeof(ElasticsearchClient)))
 	{
 		UserAgent(ConnectionConfiguration.DefaultUserAgent);
-		ResponseBuilder(new EsqlResponseBuilder());
 	}
 
 	bool TransportClientConfigurationValues.IncludeServerStackTraceOnError => _includeServerStackTraceOnError;