Fix bulking behavior in gremlin-go for TP 4.0 (#3397)

Author: xiazcy

docs/src/reference/gremlin-variants.asciidoc

@@ -216,9 +216,13 @@ Some connection options can also be set on individual requests made through the
 results, err := g.With("evaluationTimeout", 500).V().Out("knows").ToList()
 ----
 
-The following options are allowed on a per-request basis in this fashion: `batchSize`, `userAgent` and
+The following options are allowed on a per-request basis in this fashion: `batchSize`, `bulkResults`, `userAgent` and
 `evaluationTimeout`.
 
+NOTE: When submitting traversals through `DriverRemoteConnection`, `bulkResults` defaults to `true` per-request
+to optimize result transfer. This does not apply to direct `Client.Submit()` calls, where `bulkResults` must be
+set explicitly if desired.
+
 anchor:go-imports[]
 [[gremlin-go-imports]]
 === Common Imports
@@ -351,7 +355,7 @@ options := new(RequestOptionsBuilder).
 resultSet, err := client.SubmitWithOptions("g.V(x).count()", options)
 ----
 
-The following options are allowed on a per-request basis in this fashion: `batchSize`, `userAgent`,
+The following options are allowed on a per-request basis in this fashion: `batchSize`, `bulkResults`, `userAgent`,
 `evaluationTimeout` and `materializeProperties`.
 `RequestOptions` may also contain a map of variable `bindings` to be applied to the supplied
 traversal string.
@@ -813,9 +817,13 @@ GraphTraversalSource g = traversal().with(conf);
 List<Vertex> vertices = g.with(Tokens.ARGS_EVAL_TIMEOUT, 500L).V().out("knows").toList()
 ----
 
-The following options are allowed on a per-request basis in this fashion: `batchSize`, `userAgent`,
+The following options are allowed on a per-request basis in this fashion: `batchSize`, `bulkResults`, `userAgent`,
 `materializeProperties` and `evaluationTimeout`. Use of `Tokens` to reference these options is preferred.
 
+NOTE: When submitting traversals through `DriverRemoteConnection`, `bulkResults` defaults to `true` per-request
+to optimize result transfer. This does not apply to direct `Client.submit()` calls, where `bulkResults` must be
+set explicitly if desired.
+
 anchor:java-imports[]
 [[gremlin-java-imports]]
 === Common Imports
@@ -1590,6 +1598,10 @@ const vertices = await g.with_('evaluationTimeout', 500).V().out('knows').toList
 The following options are allowed on a per-request basis in this fashion: `batchSize`, `requestId`, `userAgent`,
 `bulkResults`, `materializeProperties` and `evaluationTimeout`.
 
+NOTE: When submitting traversals through `DriverRemoteConnection`, `bulkResults` defaults to `true` per-request
+to optimize result transfer. This does not apply to direct `Client.submit()` calls, where `bulkResults` must be
+set explicitly if desired.
+
 [[gremlin-javascript-imports]]
 === Common Imports
 
@@ -2124,10 +2136,14 @@ For instance to set request timeout to 500 milliseconds:
 var l = g.With(Tokens.ArgsEvalTimeout, 500).V().Out("knows").Count().ToList();
 ----
 
-The following options are allowed on a per-request basis in this fashion: `batchSize`, `userAgent`,
+The following options are allowed on a per-request basis in this fashion: `batchSize`, `bulkResults`, `userAgent`,
 `materializeProperties`, and `evaluationTimeout`. These options are available as constants on the
 `Gremlin.Net.Driver.Tokens` class.
 
+NOTE: When submitting traversals through `DriverRemoteConnection`, `bulkResults` defaults to `true` per-request
+to optimize result transfer. This does not apply to direct `GremlinClient.SubmitAsync()` calls, where `bulkResults`
+must be set explicitly if desired.
+
 [[gremlin-dotnet-imports]]
 === Common Imports
 
@@ -2270,7 +2286,7 @@ feature is to set a per-request override to the `evaluationTimeout` so that it o
 include::../../../gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Docs/Reference/GremlinVariantsTests.cs[tags=submittingScriptsWithTimeout]
 ----
 
-The following options are allowed on a per-request basis in this fashion: `batchSize`, `userAgent`, `materializeProperties`
+The following options are allowed on a per-request basis in this fashion: `batchSize`, `bulkResults`, `userAgent`, `materializeProperties`
 and `evaluationTimeout`. These options are available as constants on the `Gremlin.Net.Driver.Tokens` class.
 
 ==== Request Interceptors
@@ -2582,6 +2598,10 @@ vertices = g.with_('evaluationTimeout', 500).V().out('knows').to_list()
 The following options are allowed on a per-request basis in this fashion: `batchSize`, `bulkResults`, `language`,
 `materializeProperties`, `userAgent`, and `evaluationTimeout`.
 
+NOTE: When submitting traversals through `DriverRemoteConnection`, `bulkResults` defaults to `True` per-request
+to optimize result transfer. This does not apply to direct `Client.submit()` calls, where `bulkResults` must be
+set explicitly if desired.
+
 anchor:python-imports[]
 [[gremlin-python-imports]]
 === Common Imports
@@ -2808,7 +2828,7 @@ request.
 result_set = client.submit('g.V().repeat(both()).times(100)', request_options={'evaluationTimeout': 5000})
 ----
 
-The following options are allowed on a per-request basis in this fashion: `batchSize`, `requestId`, `userAgent`,
+The following options are allowed on a per-request basis in this fashion: `batchSize`, `bulkResults`, `requestId`, `userAgent`,
 `materializeProperties` and `evaluationTimeout` (formerly `scriptEvaluationTimeout` which is also supported but now deprecated).
 
 IMPORTANT: The preferred method for setting a per-request timeout for scripts is demonstrated above, but those familiar

gremlin-go/driver/client.go

@@ -170,6 +170,12 @@ func (client *Client) submitGremlinLang(gremlinLang *GremlinLang) (ResultSet, er
 		requestOptionsBuilder = applyOptionsConfig(requestOptionsBuilder, gremlinLang.optionsStrategies[0].configuration)
 	}
 
+	// default bulkResults to true when using DRC through request options
+	// consistent with Java RequestOptions.getRequestOptions and Python extract_request_options
+	if requestOptionsBuilder.bulkResults == nil {
+		requestOptionsBuilder.SetBulkResults(true)
+	}
+
 	request := MakeStringRequest(gremlinLang.GetGremlin(), client.traversalSource, requestOptionsBuilder.Create())
 	return client.conn.submit(&request)
 }
@@ -179,12 +185,12 @@ func applyOptionsConfig(builder *RequestOptionsBuilder, config map[string]interf
 
 	// Map configuration keys to setter method names
 	setterMap := map[string]string{
-		"requestId":             "SetRequestId",
 		"evaluationTimeout":     "SetEvaluationTimeout",
 		"batchSize":             "SetBatchSize",
 		"userAgent":             "SetUserAgent",
 		"bindings":              "SetBindings",
 		"materializeProperties": "SetMaterializeProperties",
+		"bulkResults":           "SetBulkResults",
 	}
 
 	for key, value := range config {

gremlin-go/driver/client_test.go

@@ -158,6 +158,60 @@ func TestClient(t *testing.T) {
 		AssertMarkoVertexWithoutProperties(t, result)
 	})
 
+	t.Run("Test client.SubmitWithOptions() with bulkResults true", func(t *testing.T) {
+		skipTestsIfNotEnabled(t, integrationTestSuiteName, testNoAuthEnable)
+		client, err := NewClient(testNoAuthUrl,
+			func(settings *ClientSettings) {
+				settings.TlsConfig = testNoAuthTlsConfig
+			})
+		assert.NoError(t, err)
+		assert.NotNil(t, client)
+		defer client.Close()
+
+		resultSet, err := client.SubmitWithOptions("g.inject(1,2,3,2,1)",
+			new(RequestOptionsBuilder).SetBulkResults(true).Create())
+		assert.NoError(t, err)
+		assert.NotNil(t, resultSet)
+		results, err := resultSet.All()
+		assert.NoError(t, err)
+		// With bulkResults=true, the ResultSet contains Traverser objects (one per unique value).
+		// This is consistent with Java, Python, .NET, and JS which all expose Traverser objects
+		// at the ResultSet level when bulking is enabled.
+		// g.inject(1,2,3,2,1) has 3 unique values: 1 (bulk=2), 2 (bulk=2), 3 (bulk=1)
+		assert.Equal(t, 3, len(results))
+		totalBulk := int64(0)
+		for _, r := range results {
+			tr, err := r.GetTraverser()
+			assert.NoError(t, err)
+			totalBulk += tr.Bulk
+		}
+		assert.Equal(t, int64(5), totalBulk)
+	})
+
+	t.Run("Test client.SubmitWithOptions() with bulkResults false", func(t *testing.T) {
+		skipTestsIfNotEnabled(t, integrationTestSuiteName, testNoAuthEnable)
+		client, err := NewClient(testNoAuthUrl,
+			func(settings *ClientSettings) {
+				settings.TlsConfig = testNoAuthTlsConfig
+			})
+		assert.NoError(t, err)
+		assert.NotNil(t, client)
+		defer client.Close()
+
+		resultSet, err := client.SubmitWithOptions("g.inject(1,2,3,2,1)",
+			new(RequestOptionsBuilder).SetBulkResults(false).Create())
+		assert.NoError(t, err)
+		assert.NotNil(t, resultSet)
+		results, err := resultSet.All()
+		assert.NoError(t, err)
+		// With bulkResults=false, the ResultSet contains raw values (no Traverser wrapping).
+		assert.Equal(t, 5, len(results))
+		for _, r := range results {
+			_, err := r.GetTraverser()
+			assert.Error(t, err, "expected raw value, not Traverser")
+		}
+	})
+
 	t.Run("Test deserialization of VertexProperty with properties", func(t *testing.T) {
 		skipTestsIfNotEnabled(t, integrationTestSuiteName, testNoAuthEnable)
 		client, err := NewClient(testNoAuthUrl,

gremlin-go/driver/connection.go

@@ -317,7 +317,23 @@ func (c *connection) streamToResultSet(reader io.Reader, rs ResultSet) {
 			return
 		}
 
-		rs.Channel() <- &Result{obj}
+		if d.IsBulked() {
+			bulkObj, err := d.ReadFullyQualified()
+			if err != nil {
+				c.logHandler.logf(Error, failedToReceiveResponse, err.Error())
+				rs.setError(err)
+				return
+			}
+			bulk, ok := bulkObj.(int64)
+			if !ok {
+				c.logHandler.logf(Error, failedToReceiveResponse, "expected int64 bulk count")
+				rs.setError(fmt.Errorf("expected int64 bulk count, got %T", bulkObj))
+				return
+			}
+			rs.Channel() <- &Result{&Traverser{Bulk: bulk, Value: obj}}
+		} else {
+			rs.Channel() <- &Result{obj}
+		}
 	}
 }
 

gremlin-go/driver/connection_test.go

@@ -376,8 +376,11 @@ func TestConnection(t *testing.T) {
 		var names []string
 		for _, res := range allResults {
 			assert.NotNil(t, res)
-			vp, err := res.GetVertexProperty()
-			assert.Nil(t, err)
+			// DRC defaults bulkResults=true, so results should be Traverser-wrapped.
+			tr, ok := res.Data.(*Traverser)
+			assert.True(t, ok, "expected *Traverser from DRC path with bulkResults=true, got %T", res.Data)
+			vp, ok := tr.Value.(*VertexProperty)
+			assert.True(t, ok)
 			names = append(names, vp.Value.(string))
 		}
 		assert.True(t, sortAndCompareTwoStringSlices(names, testNames))
@@ -775,6 +778,35 @@ func TestConnection(t *testing.T) {
 			assert.Greater(t, len(props), 0)
 		}
 	})
+
+	t.Run("Test bulkResults with DRC request option", func(t *testing.T) {
+		skipTestsIfNotEnabled(t, integrationTestSuiteName, testNoAuthEnable)
+
+		g := getModernGraph(t, testNoAuthUrl, &tls.Config{})
+		defer g.remoteConnection.Close()
+
+		// bulkResults is defaulted to true in submitGremlinLang, results should still be correct
+		results, err := g.Inject(1, 2, 3, 2, 1).ToList()
+		assert.Nil(t, err)
+		assert.Equal(t, 5, len(results))
+	})
+
+	t.Run("Test bulkResults with explicit With option", func(t *testing.T) {
+		skipTestsIfNotEnabled(t, integrationTestSuiteName, testNoAuthEnable)
+
+		g := getModernGraph(t, testNoAuthUrl, &tls.Config{})
+		defer g.remoteConnection.Close()
+
+		// explicitly set bulkResults to true via With
+		results, err := g.With("bulkResults", true).Inject(1, 2, 3, 2, 1).ToList()
+		assert.Nil(t, err)
+		assert.Equal(t, 5, len(results))
+
+		// explicitly set bulkResults to false via With
+		results, err = g.With("bulkResults", false).Inject(1, 2, 3, 2, 1).ToList()
+		assert.Nil(t, err)
+		assert.Equal(t, 5, len(results))
+	})
 }
 
 func submitCount(i int, client *Client, t *testing.T) {

gremlin-go/driver/graphBinaryDeserializer.go

@@ -57,9 +57,10 @@ import (
 // The bufio.Reader wrapper provides efficient buffering without affecting the
 // streaming semantics - it simply reduces the number of underlying read syscalls.
 type GraphBinaryDeserializer struct {
-	r   *bufio.Reader
-	buf [8]byte
-	err error // sticky error
+	r      *bufio.Reader
+	buf    [8]byte
+	err    error // sticky error
+	bulked bool  // whether the response stream uses bulked encoding
 }
 
 // GraphBinary flag for bulked list/set
@@ -133,13 +134,24 @@ func (d *GraphBinaryDeserializer) readInt64() (int64, error) {
 }
 
 // ReadHeader reads and validates the GraphBinary response header.
+// The header consists of a version byte and a bulking flag byte (0x00 = not bulked, 0x01 = bulked).
 // This must be called before reading any objects from the stream.
 func (d *GraphBinaryDeserializer) ReadHeader() error {
 	if _, err := d.readByte(); err != nil {
 		return err
 	}
-	_, err := d.readByte()
-	return err
+	flag, err := d.readByte()
+	if err != nil {
+		return err
+	}
+	d.bulked = flag == 0x01
+	return nil
+}
+
+// IsBulked returns whether the response stream uses bulked encoding.
+// This is determined by the header flag read during ReadHeader().
+func (d *GraphBinaryDeserializer) IsBulked() bool {
+	return d.bulked
 }
 
 // ReadFullyQualified reads the next fully-qualified GraphBinary value from the stream.

gremlin-go/driver/request.go

@@ -19,6 +19,8 @@ under the License.
 
 package gremlingo
 
+import "strconv"
+
 // RequestMessage represents a request to the server.
 type RequestMessage struct {
 	Gremlin string
@@ -71,6 +73,10 @@ func MakeStringRequest(stringGremlin string, traversalSource string, requestOpti
 		newFields["materializeProperties"] = requestOptions.materializeProperties
 	}
 
+	if requestOptions.bulkResults != nil {
+		newFields["bulkResults"] = strconv.FormatBool(*requestOptions.bulkResults)
+	}
+
 	return RequestMessage{
 		Gremlin: stringGremlin,
 		Fields:  newFields,
@@ -82,7 +88,7 @@ func MakeStringRequest(stringGremlin string, traversalSource string, requestOpti
 var allowedReqArgs = map[string]bool{
 	"evaluationTimeout":     true,
 	"batchSize":             true,
-	"requestId":             true,
 	"userAgent":             true,
 	"materializeProperties": true,
+	"bulkResults":           true,
 }

gremlin-go/driver/requestOptions.go

@@ -19,31 +19,22 @@ under the License.
 
 package gremlingo
 
-import (
-	"github.com/google/uuid"
-)
-
 type RequestOptions struct {
-	requestID             uuid.UUID
 	evaluationTimeout     int
 	batchSize             int
 	userAgent             string
 	bindings              map[string]interface{}
 	materializeProperties string
+	bulkResults           *bool
 }
 
 type RequestOptionsBuilder struct {
-	requestID             uuid.UUID
 	evaluationTimeout     int
 	batchSize             int
 	userAgent             string
 	bindings              map[string]interface{}
 	materializeProperties string
-}
-
-func (builder *RequestOptionsBuilder) SetRequestId(requestId uuid.UUID) *RequestOptionsBuilder {
-	builder.requestID = requestId
-	return builder
+	bulkResults           *bool
 }
 
 func (builder *RequestOptionsBuilder) SetEvaluationTimeout(evaluationTimeout int) *RequestOptionsBuilder {
@@ -71,6 +62,11 @@ func (builder *RequestOptionsBuilder) SetMaterializeProperties(materializeProper
 	return builder
 }
 
+func (builder *RequestOptionsBuilder) SetBulkResults(bulkResults bool) *RequestOptionsBuilder {
+	builder.bulkResults = &bulkResults
+	return builder
+}
+
 func (builder *RequestOptionsBuilder) AddBinding(key string, binding interface{}) *RequestOptionsBuilder {
 	if builder.bindings == nil {
 		builder.bindings = make(map[string]interface{})
@@ -82,12 +78,12 @@ func (builder *RequestOptionsBuilder) AddBinding(key string, binding interface{}
 func (builder *RequestOptionsBuilder) Create() RequestOptions {
 	requestOptions := new(RequestOptions)
 
-	requestOptions.requestID = builder.requestID
 	requestOptions.evaluationTimeout = builder.evaluationTimeout
 	requestOptions.batchSize = builder.batchSize
 	requestOptions.userAgent = builder.userAgent
 	requestOptions.bindings = builder.bindings
 	requestOptions.materializeProperties = builder.materializeProperties
+	requestOptions.bulkResults = builder.bulkResults
 
 	return *requestOptions
 }