This is an automated email from the ASF dual-hosted git repository.
Cole-Greer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/master by this push:
new 5271e1a764 [gremlin-dotnet] Handle non-bulked results in remote
traversals (#3525)
5271e1a764 is described below
commit 5271e1a764660136d811296638b24e859f2ab1b7
Author: Yang Xia <[email protected]>
AuthorDate: Mon Jul 20 09:21:40 2026 -0700
[gremlin-dotnet] Handle non-bulked results in remote traversals (#3525)
The `gremlin-dotnet` remote traversal path (`DriverRemoteConnection` /
`TransactionRemoteConnection`) assumed every server result was a `Traverser`.
It submitted as `SubmitAsync<Traverser>`, so `ResultSet<Traverser>` performed
an unconditional `(Traverser)item` cast during iteration.
When a compatible HTTP endpoint returns non-bulked results, e.g. a raw
`Int64` for `g.V().count()`, or any traversal run with `bulkResults=false`, the
values arrive as raw scalars rather than `Traverser` instances, and the cast
throws `InvalidCastException`. In effect, `bulkResults=false` was unusable on
the remote traversal path.
Submit as `SubmitAsync<object>` and adapt the stream at the iteration
boundary: pass `Traverser` instances through unchanged, and wrap raw values in
`new Traverser(item, 1)`. This mirrors the other GLVs. The generic
`ResultSet<T>` and the plain `client.SubmitAsync<T>` path are unchanged.
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 | 39 ++++++++++++
.../DriverRemoteConnection/GraphTraversalTests.cs | 28 +++++++++
.../Driver/DriverRemoteConnectionTests.cs | 72 +++++++++++++++++++++-
6 files changed, 157 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..0f14c88b84 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,43 @@ namespace Gremlin.Net.IntegrationTest.Driver
}
}
+ [Fact]
+ public async Task ShouldReturnRawValuesWithBulkResultsFalse()
+ {
+ var gremlinServer = new GremlinServer(TestHost, TestPort);
+ using (var gremlinClient = new GremlinClient(gremlinServer))
+ {
+ // bulkResults=false: the ResultSet contains raw values (no
Traverser wrapping).
+ var requestMessage =
RequestMessage.Build("g.inject(1,2,3,2,1)")
+ .AddBulkResults(false).Create();
+
+ var response = await
gremlinClient.SubmitAsync<object>(requestMessage);
+ var results = await response.ToListAsync();
+
+ Assert.Equal(5, results.Count);
+ Assert.All(results, r => Assert.IsNotType<Traverser>(r));
+ }
+ }
+
+ [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);
});
