You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by xi...@apache.org on 2024/10/10 21:21:15 UTC
(tinkerpop) branch graphbinary-bulked-response updated (25dd5c68ff -> 7490784ae7)
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a change to branch graphbinary-bulked-response
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
from 25dd5c68ff add bulked response option to java
new 99248a5042 add in python
new 7490784ae7 add bulked as ctx attr to preserve state & added remote feature test for bulked config
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../driver/handler/GremlinResponseHandler.java | 3 ++-
.../handler/HttpGremlinResponseStreamDecoder.java | 5 ++++
.../main/python/gremlin_python/driver/client.py | 6 ++++-
.../python/gremlin_python/driver/connection.py | 11 ++++++++-
.../driver/driver_remote_connection.py | 4 +++-
.../main/python/gremlin_python/driver/request.py | 2 +-
.../python/gremlin_python/driver/serializer.py | 28 +++++++++++++++++-----
.../python/gremlin_python/process/traversal.py | 22 ++++++++---------
.../server/handler/HttpGremlinEndpointHandler.java | 7 +++---
...=> GraphBinaryLangBulkedRemoteFeatureTest.java} | 6 ++---
.../gremlin/driver/remote/RemoteWorld.java | 10 ++++++++
.../io.cucumber.core.backend.ObjectFactory | 1 +
12 files changed, 77 insertions(+), 28 deletions(-)
copy gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/{GraphBinaryGroovyRemoteFeatureTest.java => GraphBinaryLangBulkedRemoteFeatureTest.java} (88%)
(tinkerpop) 02/02: add bulked as ctx attr to preserve state & added remote feature test for bulked config
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch graphbinary-bulked-response
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 7490784ae77b7565713080f49aeb6aa9fa9bd6db
Author: Yang Xia <55...@users.noreply.github.com>
AuthorDate: Thu Oct 10 14:21:08 2024 -0700
add bulked as ctx attr to preserve state & added remote feature test for bulked config
---
.../driver/handler/GremlinResponseHandler.java | 3 +-
.../handler/HttpGremlinResponseStreamDecoder.java | 5 +++
.../server/handler/HttpGremlinEndpointHandler.java | 7 +--
.../GraphBinaryLangBulkedRemoteFeatureTest.java | 51 ++++++++++++++++++++++
.../gremlin/driver/remote/RemoteWorld.java | 10 +++++
.../io.cucumber.core.backend.ObjectFactory | 1 +
6 files changed, 73 insertions(+), 4 deletions(-)
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
index b44e26bcc2..3dfd983481 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
@@ -74,8 +74,9 @@ public class GremlinResponseHandler extends SimpleChannelInboundHandler<Response
if ((null == statusCode) || (statusCode == HttpResponseStatus.OK)) {
final List<Object> data = response.getResult().getData();
+ final boolean bulked = (boolean) channelHandlerContext.channel().attr(AttributeKey.valueOf("isBulked")).get();
// unrolls the collection into individual results to be handled by the queue.
- if (response.getResult().isBulked()) {
+ if (bulked) {
for (Iterator<Object> iter = data.iterator(); iter.hasNext(); ) {
Object obj = iter.next();
long bulk = (long) iter.next();
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
index 5c8c38edf9..16400ad9e6 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
@@ -49,6 +49,7 @@ public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<De
private static final AttributeKey<HttpResponseStatus> RESPONSE_STATUS = AttributeKey.valueOf("responseStatus");
private static final AttributeKey<String> RESPONSE_ENCODING = AttributeKey.valueOf("responseSerializer");
private static final AttributeKey<Long> BYTES_READ = AttributeKey.valueOf("bytesRead");
+ private static final AttributeKey<Boolean> IS_BULKED = AttributeKey.valueOf("isBulked");
private final MessageSerializer<?> serializer;
private final long maxResponseContentLength;
@@ -62,6 +63,7 @@ public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<De
@Override
protected void decode(ChannelHandlerContext ctx, DefaultHttpObject msg, List<Object> out) throws Exception {
final Attribute<Boolean> isFirstChunk = ((AttributeMap) ctx).attr(IS_FIRST_CHUNK);
+ final Attribute<Boolean> isBulked = ((AttributeMap) ctx).attr(IS_BULKED);
final Attribute<HttpResponseStatus> responseStatus = ((AttributeMap) ctx).attr(RESPONSE_STATUS);
final Attribute<String> responseEncoding = ((AttributeMap) ctx).attr(RESPONSE_ENCODING);
@@ -95,6 +97,9 @@ public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<De
out.add(response);
} else {
final ResponseMessage chunk = serializer.readChunk(content, isFirstChunk.get());
+ if (isFirstChunk.get()){
+ isBulked.set(chunk.getResult().isBulked());
+ }
isFirstChunk.set(false);
out.add(chunk);
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
index e92ba8d5a0..c5b4f1f6e7 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
@@ -313,8 +313,10 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
final Bindings mergedBindings = mergeBindingsFromRequest(context, new SimpleBindings(graphManager.getAsBindings()));
final Object result = scriptEngine.eval(message.getGremlin(), mergedBindings);
- final boolean bulking = Objects.equals(context.getChannelHandlerContext().channel().attr(StateKey.REQUEST_HEADERS).get().get(Tokens.BULKED), "true")
- || (args.containsKey(Tokens.BULKED) && (boolean) args.get(Tokens.BULKED));
+ final boolean bulking = ((Objects.equals(context.getChannelHandlerContext().channel().attr(StateKey.REQUEST_HEADERS).get().get(Tokens.BULKED), "true")
+ && result instanceof Traversal)
+ || (args.containsKey(Tokens.BULKED) && (boolean) args.get(Tokens.BULKED))
+ && Objects.equals(language, "gremlin-lang"));
if (bulking) {
// optimization for driver requests
@@ -421,7 +423,6 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
// while waiting for the client to catch up
if (aggregate.size() < resultIterationBatchSize && itty.hasNext()) {
if (bulking) {
- // bulking will return
Traverser traverser = (Traverser) itty.next();
aggregate.add(traverser.get());
aggregate.add(traverser.bulk());
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/GraphBinaryLangBulkedRemoteFeatureTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/GraphBinaryLangBulkedRemoteFeatureTest.java
new file mode 100644
index 0000000000..1e5b0a7317
--- /dev/null
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/GraphBinaryLangBulkedRemoteFeatureTest.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * http://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.apache.tinkerpop.gremlin.driver.remote;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Stage;
+import io.cucumber.guice.CucumberModules;
+import io.cucumber.junit.Cucumber;
+import io.cucumber.junit.CucumberOptions;
+import org.apache.tinkerpop.gremlin.features.AbstractGuiceFactory;
+import org.apache.tinkerpop.gremlin.features.World;
+import org.junit.runner.RunWith;
+
+@RunWith(Cucumber.class)
+@CucumberOptions(
+ tags = "not @RemoteOnly and not @GraphComputerOnly and not @AllowNullPropertyValues",
+ glue = { "org.apache.tinkerpop.gremlin.features" },
+ objectFactory = GraphBinaryLangBulkedRemoteFeatureTest.RemoteGuiceFactory.class,
+ features = { "classpath:/org/apache/tinkerpop/gremlin/test/features" },
+ plugin = {"progress", "junit:target/cucumber.xml"})
+public class GraphBinaryLangBulkedRemoteFeatureTest extends AbstractFeatureTest {
+ public static class RemoteGuiceFactory extends AbstractGuiceFactory {
+ public RemoteGuiceFactory() {
+ super(Guice.createInjector(Stage.PRODUCTION, CucumberModules.createScenarioModule(), new ServiceModule()));
+ }
+ }
+
+ public static final class ServiceModule extends AbstractModule {
+ @Override
+ protected void configure() {
+ bind(World.class).to(RemoteWorld.GraphBinaryLangBulkedRemoteWorld.class);
+ }
+ }
+}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java
index b1c66c1d39..84109b080e 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java
@@ -169,6 +169,16 @@ public abstract class RemoteWorld implements World {
}
}
+ public static class GraphBinaryLangBulkedRemoteWorld extends RemoteWorld {
+ public GraphBinaryLangBulkedRemoteWorld() { super(createTestCluster(Serializers.GRAPHBINARY_V4)); }
+
+ @Override
+ public GraphTraversalSource getGraphTraversalSource(final LoadGraphWith.GraphData graphData) {
+ final GraphTraversalSource g = super.getGraphTraversalSource(graphData);
+ return g.with("language", "gremlin-lang").with("bulked", true);
+ }
+ }
+
public static class GraphBinaryGroovyRemoteWorld extends RemoteWorld {
public GraphBinaryGroovyRemoteWorld() { super(createTestCluster(Serializers.GRAPHBINARY_V4)); }
diff --git a/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory b/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory
index 5b6b888d13..7b70b113b0 100644
--- a/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory
+++ b/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory
@@ -1,3 +1,4 @@
org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryLangRemoteFeatureTest$RemoteGuiceFactory
+org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryLangBulkedRemoteFeatureTest$RemoteGuiceFactory
org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryGroovyRemoteFeatureTest$RemoteGuiceFactory
org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryRemoteComputerFeatureTest$RemoteGuiceFactory
(tinkerpop) 01/02: add in python
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch graphbinary-bulked-response
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 99248a50421e99bab9d3bce9090feb16ec0bbd21
Author: Yang Xia <55...@users.noreply.github.com>
AuthorDate: Wed Oct 9 20:14:52 2024 -0700
add in python
---
.../main/python/gremlin_python/driver/client.py | 6 ++++-
.../python/gremlin_python/driver/connection.py | 11 ++++++++-
.../driver/driver_remote_connection.py | 4 +++-
.../main/python/gremlin_python/driver/request.py | 2 +-
.../python/gremlin_python/driver/serializer.py | 28 +++++++++++++++++-----
.../python/gremlin_python/process/traversal.py | 22 ++++++++---------
6 files changed, 52 insertions(+), 21 deletions(-)
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/client.py b/gremlin-python/src/main/python/gremlin_python/driver/client.py
index 16bd6f8270..a3add1f225 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/client.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/client.py
@@ -42,13 +42,14 @@ class Client:
def __init__(self, url, traversal_source, protocol_factory=None,
transport_factory=None, pool_size=None, max_workers=None,
message_serializer=None, auth=None, headers=None,
- enable_user_agent_on_connect=True, **transport_kwargs):
+ enable_user_agent_on_connect=True, enable_bulked_result=False, **transport_kwargs):
log.info("Creating Client with url '%s'", url)
self._closed = False
self._url = url
self._headers = headers
self._enable_user_agent_on_connect = enable_user_agent_on_connect
+ self._enable_bulked_result = enable_bulked_result
self._traversal_source = traversal_source
if "max_content_length" not in transport_kwargs:
transport_kwargs["max_content_length"] = 10 * 1024 * 1024
@@ -172,4 +173,7 @@ class Client:
else:
message.fields['bindings'] = request_options['params']
+ print("---client")
+ print(message)
+
return conn.write(message)
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/connection.py b/gremlin-python/src/main/python/gremlin_python/driver/connection.py
index de9a81034d..a6c907bb9f 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/connection.py
@@ -25,7 +25,8 @@ __author__ = 'David M. Brown (davebshow@gmail.com)'
class Connection:
def __init__(self, url, traversal_source, protocol, transport_factory,
- executor, pool, headers=None, enable_user_agent_on_connect=True):
+ executor, pool, headers=None, enable_user_agent_on_connect=True,
+ enable_bulked_result=False):
self._url = url
self._headers = headers
self._traversal_source = traversal_source
@@ -41,6 +42,11 @@ class Connection:
if self._headers is None:
self._headers = dict()
self._headers[useragent.userAgentHeader] = useragent.userAgent
+ self._enable_bulked_result = enable_bulked_result
+ if self._enable_bulked_result:
+ if self._headers is None:
+ self._headers = dict()
+ self._headers["bulked"] = "true"
def connect(self):
if self._transport:
@@ -63,6 +69,9 @@ class Connection:
future_write = self._executor.submit(
self._protocol.write, request_message)
+ print("---conn")
+ print(request_message)
+
def cb(f):
try:
f.result()
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py b/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
index a48dc0087b..dc8b95eb7a 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
@@ -35,7 +35,7 @@ class DriverRemoteConnection(RemoteConnection):
transport_factory=None, pool_size=None, max_workers=None,
auth=None,
message_serializer=None, headers=None,
- enable_user_agent_on_connect=True, **transport_kwargs):
+ enable_user_agent_on_connect=True, enable_bulked_result=False, **transport_kwargs):
log.info("Creating DriverRemoteConnection with url '%s'", str(url))
self.__url = url
self.__traversal_source = traversal_source
@@ -47,6 +47,7 @@ class DriverRemoteConnection(RemoteConnection):
self.__message_serializer = message_serializer
self.__headers = headers
self.__enable_user_agent_on_connect = enable_user_agent_on_connect
+ self.__enable_bulked_result = enable_bulked_result
self.__transport_kwargs = transport_kwargs
if message_serializer is None:
@@ -60,6 +61,7 @@ class DriverRemoteConnection(RemoteConnection):
auth=auth,
headers=headers,
enable_user_agent_on_connect=enable_user_agent_on_connect,
+ enable_bulked_result=enable_bulked_result,
**transport_kwargs)
self._url = self._client._url
self._traversal_source = self._client._traversal_source
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/request.py b/gremlin-python/src/main/python/gremlin_python/driver/request.py
index 5e04f0bb85..db524d6642 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/request.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/request.py
@@ -24,4 +24,4 @@ RequestMessage = collections.namedtuple(
'RequestMessage', ['fields', 'gremlin'])
Tokens = ['batchSize', 'bindings', 'g', 'gremlin', 'language',
- 'evaluationTimeout', 'materializeProperties', 'timeoutMs', 'userAgent']
+ 'evaluationTimeout', 'materializeProperties', 'timeoutMs', 'userAgent', 'bulked']
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/serializer.py b/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
index 3fa80a055b..50c92190ce 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
@@ -21,6 +21,10 @@ import base64
import logging
import struct
import io
+
+from gremlin_python.process.traversal import Traverser
+from numpy.lib.utils import source
+
try:
import ujson as json
if int(json.__version__[0]) < 2:
@@ -95,6 +99,7 @@ class GraphBinarySerializersV4(object):
return bytes(ba)
def deserialize_message(self, message, is_first_chunk=False):
+ bulking = False
if len(message) == 0:
return {'status': {'code': 204},
'result': {'meta': {},
@@ -105,8 +110,9 @@ class GraphBinarySerializersV4(object):
if is_first_chunk:
b.read(1) # version
+ bulking = b.read(1)[0] == 0x01
- result, readable = self.read_payload(b)
+ result, readable = self.read_payload(b, bulking)
if not readable:
return {
'result': {'meta': {},
@@ -125,18 +131,28 @@ class GraphBinarySerializersV4(object):
'result': {'meta': {},
'data': result}}
+ print('raw result')
+ print(result)
+
return msg
- def read_payload(self, buffer):
+ def read_payload(self, buffer, bulking):
results = []
readable = True
while buffer.readable(): # find method or way to access readable bytes without using buffer.getvalue()
if buffer.tell() == len(buffer.getvalue()):
readable = False
break
- data = self._graphbinary_reader.to_object(buffer)
- if data == Marker.end_of_stream():
- break
- results.append(data)
+ if bulking:
+ item = self._graphbinary_reader.to_object(buffer)
+ if item == Marker.end_of_stream():
+ break
+ bulk = self._graphbinary_reader.to_object(buffer)
+ results.append(Traverser(item, bulk))
+ else:
+ data = self._graphbinary_reader.to_object(buffer)
+ if data == Marker.end_of_stream():
+ break
+ results.append(data)
return results, readable
diff --git a/gremlin-python/src/main/python/gremlin_python/process/traversal.py b/gremlin-python/src/main/python/gremlin_python/process/traversal.py
index dce8524b98..53a792b2f5 100644
--- a/gremlin-python/src/main/python/gremlin_python/process/traversal.py
+++ b/gremlin-python/src/main/python/gremlin_python/process/traversal.py
@@ -55,15 +55,17 @@ class Traversal(object):
self.traversal_strategies.apply_strategies(self)
if self.last_traverser is None:
self.last_traverser = next(self.traversers)
- # TODO affected by bulking, revisit after bulking is added back
- # object = self.last_traverser.object
- # self.last_traverser.bulk = self.last_traverser.bulk - 1
- # if self.last_traverser.bulk <= 0:
- # self.last_traverser = None
- # return object
+ print("---calling next???")
res = self.last_traverser
- self.last_traverser = None
- return res
+ if isinstance(res, Traverser):
+ obj = res.object
+ self.last_traverser.bulk = self.last_traverser.bulk - 1
+ if self.last_traverser.bulk <= 0:
+ self.last_traverser = None
+ return obj
+ else:
+ self.last_traverser = None
+ return res
def toList(self):
warnings.warn(
@@ -100,7 +102,6 @@ class Traversal(object):
DeprecationWarning)
return self.next_traverser()
- # TODO: affected by bulking, revisit after bulking is added back
def next_traverser(self):
if self.traversers is None:
self.traversal_strategies.apply_strategies(self)
@@ -118,7 +119,6 @@ class Traversal(object):
DeprecationWarning)
return self.has_next()
- # TODO: revisit after bulking is added back
def has_next(self):
if self.traversers is None:
self.traversal_strategies.apply_strategies(self)
@@ -127,7 +127,7 @@ class Traversal(object):
self.last_traverser = next(self.traversers)
except StopIteration:
return False
- return not (self.last_traverser is None) # and self.last_traverser.bulk > 0
+ return not (self.last_traverser is None) and self.last_traverser.bulk > 0 if isinstance(self.last_traverser, Traverser) else True
def next(self, amount=None):
if amount is None: