This is an automated email from the ASF dual-hosted git repository. xiazcy pushed a commit to branch dotnet-bulk-fix in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit f49266185b88aa53a18b0ab83ef89adc7b673412 Author: Yang Xia <[email protected]> AuthorDate: Fri Jul 17 14:58:29 2026 -0700 fix(gremlin-dotnet): Handle non-bulked remote traversal results Remote traversals assumed every result was a Traverser and threw InvalidCastException on non-bulked scalar responses (bulkResults=false). Wrap raw values in a Traverser at the iteration boundary, matching the Java, Python, Go, and JavaScript drivers. Assisted-by: Kiro:claude-opus-4-8 --- .../Driver/Remote/DriverRemoteConnection.cs | 2 +- .../Driver/Remote/DriverRemoteTraversal.cs | 21 ++++++- .../Driver/Remote/TransactionRemoteConnection.cs | 2 +- .../Driver/GremlinClientTests.cs | 37 +++++++++++ .../DriverRemoteConnection/GraphTraversalTests.cs | 28 +++++++++ .../Driver/DriverRemoteConnectionTests.cs | 72 +++++++++++++++++++++- 6 files changed, 155 insertions(+), 7 deletions(-) diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs index 3f3ef7462b..57cefda32a 100644 --- a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs +++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs @@ -142,7 +142,7 @@ namespace Gremlin.Net.Driver.Remote requestMsg.AddField(Tokens.ArgsBulkResults, true); } - var resultSet = await _client.SubmitAsync<Traverser>(requestMsg.Create(), cancellationToken) + var resultSet = await _client.SubmitAsync<object>(requestMsg.Create(), cancellationToken) .ConfigureAwait(false); return new DriverRemoteTraversal<TStart, TEnd>(resultSet); } diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs index e4ff99563a..fa19257584 100644 --- a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs +++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs @@ -22,22 +22,37 @@ #endregion using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; using Gremlin.Net.Process.Traversal; namespace Gremlin.Net.Driver.Remote { /// <summary> /// A traversal returned from a remote Gremlin Server submission, wrapping a - /// <see cref="ResultSet{T}"/> of <see cref="Traverser"/> instances. + /// <see cref="ResultSet{T}"/> of result objects. /// </summary> internal class DriverRemoteTraversal<TStart, TEnd> : DefaultTraversal<TStart, TEnd> { - public DriverRemoteTraversal(ResultSet<Traverser> resultSet) + public DriverRemoteTraversal(ResultSet<object> resultSet) { - Traversers = resultSet; + Traversers = AdaptResults(resultSet); } /// <inheritdoc /> public override GremlinLang GremlinLang => throw new NotSupportedException("Remote traversals do not have GremlinLang"); + + // Bulked responses arrive as Traversers; non-bulked responses arrive as raw values + // (e.g. a long for g.V().count()) and are wrapped with a bulk of 1, as the other GLVs do. + private static async IAsyncEnumerable<Traverser> AdaptResults(ResultSet<object> resultSet, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + await foreach (var item in resultSet.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + yield return item as Traverser ?? new Traverser(item, 1); + } + } } } \ No newline at end of file diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/TransactionRemoteConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/TransactionRemoteConnection.cs index ac60914b5a..c80d5ab80a 100644 --- a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/TransactionRemoteConnection.cs +++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/TransactionRemoteConnection.cs @@ -87,7 +87,7 @@ namespace Gremlin.Net.Driver.Remote requestMsg.AddField(Tokens.ArgsTransactionId, _transactionId); // Route through Transaction's serialized submission to guarantee ordering - var resultSet = await _transaction.SubmitAsync<Traverser>(requestMsg.Create(), cancellationToken) + var resultSet = await _transaction.SubmitAsync<object>(requestMsg.Create(), cancellationToken) .ConfigureAwait(false); return new DriverRemoteTraversal<TStart, TEnd>(resultSet); } diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs index a35f026522..f060897343 100644 --- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs +++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs @@ -23,11 +23,13 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Threading.Tasks; using Gremlin.Net.Driver; using Gremlin.Net.Driver.Exceptions; using Gremlin.Net.Driver.Messages; using Gremlin.Net.IntegrationTest.Util; +using Gremlin.Net.Process.Traversal; using Gremlin.Net.Structure; using Microsoft.Extensions.Logging; using NSubstitute; @@ -135,6 +137,41 @@ namespace Gremlin.Net.IntegrationTest.Driver } } + [Fact] + public async Task ShouldReturnRawValuesWithBulkResultsFalse() + { + var gremlinServer = new GremlinServer(TestHost, TestPort); + using (var gremlinClient = new GremlinClient(gremlinServer)) + { + // bulkResults=false: each value is returned raw and read as its value type. + var requestMessage = RequestMessage.Build("g.inject(1,2,3,2,1)") + .AddBulkResults(false).Create(); + + var response = await gremlinClient.SubmitAsync<int>(requestMessage); + + Assert.Equal(new List<int> {1, 2, 3, 2, 1}, await response.ToListAsync()); + } + } + + [Fact] + public async Task ShouldReturnTraversersWithBulkResultsTrue() + { + var gremlinServer = new GremlinServer(TestHost, TestPort); + using (var gremlinClient = new GremlinClient(gremlinServer)) + { + // bulkResults=true: values are bulked into Traversers, read as Traverser. + var requestMessage = RequestMessage.Build("g.inject(1,2,3,2,1)") + .AddBulkResults(true).Create(); + + var response = await gremlinClient.SubmitAsync<Traverser>(requestMessage); + var results = await response.ToListAsync(); + + // 3 unique values with bulk counts summing to the 5 injected. + Assert.Equal(3, results.Count); + Assert.Equal(5L, results.Sum(t => t.Bulk)); + } + } + [Fact] public async Task ShouldThrowOnExecutionOfSimpleInvalidScript() { diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs index 489994b4c0..0c9104ba8c 100644 --- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs +++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs @@ -61,6 +61,34 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection Assert.Equal(2, count); } + [Fact] + public void g_V_Count_WithBulkResultsFalse() + { + // bulkResults=false makes the server return a raw scalar instead of a bulked Traverser. + var connection = _connectionFactory.CreateRemoteConnection(); + var g = AnonymousTraversalSource.Traversal().With(connection) + .With(Tokens.ArgsBulkResults, false); + + var count = g.V().Count().Next(); + + Assert.Equal(6, count); + } + + [Fact] + public void g_V_Values_WithBulkResultsFalse() + { + // Non-bulked multi-result response: each raw value must be yielded. + var connection = _connectionFactory.CreateRemoteConnection(); + var g = AnonymousTraversalSource.Traversal().With(connection) + .With(Tokens.ArgsBulkResults, false); + + var names = g.V().Values<string>("name").ToList(); + + Assert.Equal(6, names.Count); + Assert.Contains("marko", names); + Assert.Contains("lop", names); + } + [Fact] public void g_V_Count_Clone() { diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs index e4c1b1fb05..4d21579498 100644 --- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs +++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs @@ -218,20 +218,88 @@ namespace Gremlin.Net.UnitTest.Driver Assert.Equal(false, capturedRequest!.Fields[Tokens.ArgsBulkResults]); } + [Fact] + public async Task ShouldYieldRawScalarFromNonBulkedResponse() + { + // A raw (non-bulked) scalar must be yielded directly, not cast-failed as a Traverser. + var client = CreateResultClient(10L); + var connection = new DriverRemoteConnection(client, "g"); + var gl = new GremlinLang(); + gl.AddStep("V", Array.Empty<object>()); + + var traversal = await connection.SubmitAsync<object, long>(gl); + var results = traversal.ToList(); + + Assert.Equal(new long[] { 10L }, results); + } + + [Fact] + public async Task ShouldExpandBulkedTraverserFromResponse() + { + // A bulked Traverser must be expanded by its bulk count. + var client = CreateResultClient(new Traverser("a", 3)); + var connection = new DriverRemoteConnection(client, "g"); + var gl = new GremlinLang(); + gl.AddStep("V", Array.Empty<object>()); + + var traversal = await connection.SubmitAsync<object, string>(gl); + var results = traversal.ToList(); + + Assert.Equal(new[] { "a", "a", "a" }, results); + } + + [Fact] + public async Task ShouldHandleMixOfRawValuesAndTraversers() + { + // Raw values and bulked Traversers may interleave in one response. + var client = CreateResultClient(1L, new Traverser(2L, 2), 3L); + var connection = new DriverRemoteConnection(client, "g"); + var gl = new GremlinLang(); + gl.AddStep("V", Array.Empty<object>()); + + var traversal = await connection.SubmitAsync<object, long>(gl); + var results = traversal.ToList(); + + Assert.Equal(new long[] { 1L, 2L, 2L, 3L }, results); + } + + /// <summary> + /// Creates a mock IGremlinClient whose submission returns a ResultSet streaming the + /// provided items (raw values and/or Traverser instances). + /// </summary> + private static IGremlinClient CreateResultClient(params object[] items) + { + var client = Substitute.For<IGremlinClient>(); + client.SubmitAsync<object>(Arg.Any<RequestMessage>(), Arg.Any<CancellationToken>()) + .Returns(_ => + { + var channel = System.Threading.Channels.Channel.CreateUnbounded<object>(); + foreach (var item in items) + { + channel.Writer.TryWrite(item); + } + channel.Writer.Complete(); + var cts = new CancellationTokenSource(); + var resultSet = new ResultSet<object>(channel.Reader, cts, Task.CompletedTask); + return Task.FromResult(resultSet); + }); + return client; + } + /// <summary> /// Creates a mock IGremlinClient that captures the submitted RequestMessage. /// </summary> private static IGremlinClient CreateCapturingClient(Action<RequestMessage> capture) { var client = Substitute.For<IGremlinClient>(); - client.SubmitAsync<Traverser>(Arg.Any<RequestMessage>(), Arg.Any<CancellationToken>()) + client.SubmitAsync<object>(Arg.Any<RequestMessage>(), Arg.Any<CancellationToken>()) .Returns(callInfo => { capture(callInfo.Arg<RequestMessage>()); var channel = System.Threading.Channels.Channel.CreateUnbounded<object>(); channel.Writer.Complete(); var cts = new CancellationTokenSource(); - var emptyResult = new ResultSet<Traverser>( + var emptyResult = new ResultSet<object>( channel.Reader, cts, Task.CompletedTask); return Task.FromResult(emptyResult); });
