diff --git a/docs/cypher-compatibility.md b/docs/cypher-compatibility.md new file mode 100644 index 0000000000..ac5f312a13 --- /dev/null +++ b/docs/cypher-compatibility.md @@ -0,0 +1,96 @@ +# Cypher compatibility + +This note records the Cypher behavior verified for this change. It does not claim full openCypher or Neo4j +compatibility. The runtime used Java 17.0.20.1, TinkerPop 3.8.1, `org.opencypher.gremlin:translation:1.0.4`, +and RocksDB, rebased onto the Apache master that includes the Java 17 / TinkerPop 3.8.1 upgrade. + +## Request forms + +The endpoint is `/graphspaces/{graphspace}/graphs/{graph}/cypher`. + +| Method and content type | Request | Verified behavior | +| --- | --- | --- | +| GET | `?cypher=` | Query-string form works: `testGet` | +| POST `application/json` | Raw Cypher text | Legacy raw-body form works: `testPost` | +| POST `text/plain` | Any body | Rejected with HTTP 415: `testRejectPlainTextPost` | +| POST `application/json` | JSON object below | Separate bindings work: `testParameters` | + +```json +{ + "cypher": "MATCH (n:person) WHERE n.name = $name RETURN n.name", + "parameters": {"name": "marko"} +} +``` + +POST bodies containing JSON arrays, strings, numbers, booleans, or null are rejected with HTTP 400. +Malformed JSON objects and trailing tokens are also request errors and do not fall back to raw Cypher. + +For the JSON-object form, `cypher` must be a nonblank string and `parameters`, when present, must be an +object. Omitting `parameters` means an empty map. Missing bindings produce an execution error; an explicit +null value is accepted. Tests also cover bindings named `id` and `label`, and the fixed limit of 16 +top-level parameters. This Cypher processor does not support a `maxParameters` configuration setting. +Values containing quotes and newlines remain bound data rather than changing query text. + +A Map counts as one top-level parameter. Group related values into one Map and reference its fields +explicitly; the vertex label and required properties must match the graph's existing schema: + +```json +{ + "cypher": "CREATE (n:person {name:$props.name, age:$props.age, city:$props.city}) RETURN n.name", + "parameters": {"props": {"name": "new-person", "age": 20, "city": "Beijing"}} +} +``` + +Do not use the shorthand `CREATE (n:person $props)`: the pinned translator does not assign the bound +Map's properties in that path. Live REST verification confirmed that one Map containing 17 entries +works with explicit property references, with all 17 persisted values verified through native REST. +A 17-entry returned Map also succeeds, while 17 top-level bindings produce an execution error. + +The translator's reserved null-marker string `" cypher.null"` (two leading spaces) is rejected in binding +values, including nested maps and lists, to avoid silently converting a user string to null. Actual JSON +null remains accepted. The same marker in query literals or stored properties is an existing translator +limitation and is outside this compatibility target. +Check the HTTP status and body `status.code`: execution failures retain HTTP 200 with `status.code` 400 +and null result data. + +Traversal iteration failures, including Java errors, attempt rollback on the execution thread and return +a terminal failure response. A rollback exception is attached to the original failure. + +## Verified behavior + +The API fixture uses a strong schema with a `SECONDARY` index on `city` and a `RANGE` index on `age`. +Acceptance tests verified: + +| Area | Verified cases and test methods | +| --- | --- | +| Reads | Label scans, equality and range predicates, computed regex full-string matching, boolean combinations, directed one- and two-hop patterns, empty results: `testGet`, `testExactReadsAndPredicates`, `testComputedRegexExecutesExtensionPredicate`, `testComputedRegexRequiresWholeStringMatch`, `testRelationQuery` | +| Results | Aliases, scalar and node values, nested maps/lists, relationship ids, path shape, and null/missing-property semantics: `testReturnNodeIdAsPrimitiveValue`, `testReturnNodeDoesNotLeakInternalIdTypes`, `testReturnNestedIdDoesNotLeakInternalIdTypes`, `testReturnRelationIdDoesNotLeakInternalIdTypes`, `testReturnPathShape`, `testNullAndMissingProperty` | +| Aggregation and pagination | `DISTINCT`, `count`/`sum`/`min`/`max`/`avg`, `ORDER BY`, `SKIP`, and `LIMIT`: `testDuplicatesDistinctAndPagination`, `testAggregates` | +| Parameters | String, number, boolean, null, empty and missing bindings; quoting/newline safety: `testParameters`, `testRequestPreservesQueryAndParameterValues`, `testRequestAcceptsEmptyParameters`, `testAcceptCypherParameterNamesAndNullValues`, `testEnforceParameterCountBoundary` | +| Writes | Vertex and edge `CREATE`, property `SET`, edge and vertex `DELETE`; state checked through native REST reads: `testCreate`, `testCreateSetAndDeleteWithNativeReadback` | +| Failures and routing | Invalid syntax, request shape, binding keys, schema values, authentication, and graph routing: `testInvalidRequests`, `testRejectInvalidQuery`, `testRejectInvalidBindingShapeAndKeys`, `testAuthenticationAndGraphRouting`, `testSpecifiedGraphRouting` | +| Failed write | A rejected multi-vertex create left no residue in native reads or in 32 later query/native-read checks: `testFailedWriteDoesNotLeakIntoLaterRequests` | + +Write verification describes how acceptance was checked; clients are not required to issue native REST reads +after each write. A successful Cypher response alone was not used as proof of persisted state. + +## Verification record + +The final verification covers the rebased tree, including non-object JSON request rejection, the +unsupported `text/plain` media type, computed regex fixture cases, and fatal-error rollback and response. +`CypherApiTest` passed all 23 cases, `CypherClientTest` all seven, and `CypherOpProcessorTest` all eight, +with no Cypher skips. The related predicate and Gremlin HTTP ownership/context tests also passed. +`AbstractRestClientTest` passed all nine cases, including charset serialization and an ASCII-default JVM. +Gremlin API tests passed ten cases with one existing non-shared-backend skip; Login API tests passed all +three. EditorConfig formatting, the root clean compile, and reactor installation passed on Java 17. +The live RocksDB server reported Gremlin 3.8.1; its packaged API JAR and compiled target JAR were +byte-identical before the API run. + +Test sources: [`CypherApiTest`](../hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/CypherApiTest.java), +[`CypherClientTest`](../hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/cypher/CypherClientTest.java), +and [`CypherOpProcessorTest`](../hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessorTest.java). + +## Outside this compatibility target + +Advanced constructs including `MERGE`, `OPTIONAL MATCH`, `WITH`, `UNWIND`, `UNION`, and variable-length paths, +as well as HStore, Bolt, cross-request transactions, and performance, were not verified by this change. diff --git a/docs/upgrade-tinkerpop-3.8.md b/docs/upgrade-tinkerpop-3.8.md index 0cb692f0af..d1eaf2ecbf 100644 --- a/docs/upgrade-tinkerpop-3.8.md +++ b/docs/upgrade-tinkerpop-3.8.md @@ -221,6 +221,19 @@ The response has rows under `result.data`; each row is a map and `nodeId` is a JSON scalar, not a HugeGraph internal ID class. The Cypher API test covers this shape, nested IDs and paths. +## HTTP request errors + +The supplied WebSocket/HTTP channelizer releases each HTTP request buffer once, +including when a traversal alias is unknown. Such a request returns one HTTP 400, +and a later valid request on the same keep-alive connection receives its own +response. This avoids TinkerPop 3.8.1's duplicate release and second HTTP 500, +which could be mistaken for the next request's response by a connection pool. +The correction applies when the configured channelizer is TinkerPop's standard +`WsAndHttpChannelizer`; HugeGraph selects its compatibility implementation at +startup and retains TinkerPop's WebSocket handling. Custom channelizer classes +keep their configured behavior. Unknown aliases still require the caller to +use a graph or traversal-source name exposed by the server. + ## Compatibility verification The repository's checked-in tests establish the following boundaries: diff --git a/hugegraph-commons/README.md b/hugegraph-commons/README.md index d8cbcbc24a..71f6dea421 100644 --- a/hugegraph-commons/README.md +++ b/hugegraph-commons/README.md @@ -21,6 +21,10 @@ numeric or collection util classes to simplify the development of HugeGraph and - Util: performance analyzer, version checker, numeric and Collection utils, log and exception utils, etc. - Rpc: rpc component for inner module communication, currently it's based on [Sofa-RPC](https://github.com/sofastack/sofa-rpc) +The REST client serializes object bodies as JSON for `application/json`, including media types +with parameters such as `application/json; charset=UTF-16`. Request bodies use the declared +charset, or UTF-8 when none is specified; gzip preserves these encoded bytes. + You could use import the dependencies in `maven` like this: ```xml diff --git a/hugegraph-commons/hugegraph-common/README.md b/hugegraph-commons/hugegraph-common/README.md index bfeef03cd2..c50c1e8e7d 100644 --- a/hugegraph-commons/hugegraph-common/README.md +++ b/hugegraph-commons/hugegraph-common/README.md @@ -10,6 +10,12 @@ hugegraph-common encapsulates locks, configurations, events, iterators, rest and numeric or collection util classes to simplify the development of HugeGraph and its components. +## REST request encoding + +The shared REST client encodes request bodies using the charset declared in +`Content-Type`, or UTF-8 when no charset is declared. JSON and text requests do +not depend on the JVM's default charset. Gzip compresses those same encoded bytes. + ## Components - Lock: atomic lock, key lock, lock group and lock manager diff --git a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rest/AbstractRestClient.java b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rest/AbstractRestClient.java index 91fa90bf2b..ecb7deca30 100644 --- a/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rest/AbstractRestClient.java +++ b/hugegraph-commons/hugegraph-common/src/main/java/org/apache/hugegraph/rest/AbstractRestClient.java @@ -125,9 +125,10 @@ public AbstractRestClient(String url, RestClientConfig config) { } private static RequestBody buildRequestBody(Object body, RestHeaders headers) { - String contentType = parseContentType(headers); + MediaType contentType = MediaType.parse(parseContentType(headers)); String bodyContent; - if (RestHeaders.APPLICATION_JSON.equals(contentType)) { + if (contentType != null && "application".equals(contentType.type()) && + "json".equals(contentType.subtype())) { if (body == null) { bodyContent = "{}"; } else if (body instanceof String) { @@ -138,8 +139,7 @@ private static RequestBody buildRequestBody(Object body, RestHeaders headers) { } else { bodyContent = String.valueOf(body); } - RequestBody requestBody = RequestBody.create(bodyContent, - MediaType.parse(contentType)); + RequestBody requestBody = RequestBody.create(bodyContent, contentType); if (headers != null && "gzip".equals(headers.get(RestHeaders.CONTENT_ENCODING))) { diff --git a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/rest/AbstractRestClientTest.java b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/rest/AbstractRestClientTest.java index a19e44592f..ee2158b917 100644 --- a/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/rest/AbstractRestClientTest.java +++ b/hugegraph-commons/hugegraph-common/src/test/java/org/apache/hugegraph/unit/rest/AbstractRestClientTest.java @@ -72,6 +72,7 @@ public static void main(String[] args) throws IOException { tests.testJsonRequestBodyUsesUtf8(); tests.testGzipRequestBodyUsesUtf8(); tests.testRequestBodyRespectsExplicitCharset(); + tests.testJsonObjectRequestBodyRespectsExplicitCharset(); } } @@ -102,21 +103,39 @@ public void testRequestBodyRespectsExplicitCharset() throws IOException { buffer.readByteArray()); } + @Test + public void testJsonObjectRequestBodyRespectsExplicitCharset() throws IOException { + String value = "\u4f60\u597d\ud83d\ude80"; + String json = "{\"id\":\"" + value + "\"}"; + for (Charset charset : new Charset[]{StandardCharsets.UTF_8, StandardCharsets.UTF_16}) { + RestHeaders headers = new RestHeaders().add(RestHeaders.CONTENT_TYPE, + "application/json; charset=" + charset.name()); + assertBody(Collections.singletonMap("id", value), headers, json, charset, false); + headers.add(RestHeaders.CONTENT_ENCODING, "gzip"); + assertBody(Collections.singletonMap("id", value), headers, json, charset, true); + } + } + private static void assertUtf8Body(Object value, RestHeaders headers, String expected, boolean gzip) throws IOException { + assertBody(value, headers, expected, StandardCharsets.UTF_8, gzip); + } + + private static void assertBody(Object value, RestHeaders headers, String expected, + Charset charset, boolean gzip) throws IOException { RequestBody body = requestBody(value, headers); - Assert.assertEquals(StandardCharsets.UTF_8, body.contentType().charset()); + Assert.assertEquals(charset, body.contentType().charset()); Buffer buffer = new Buffer(); body.writeTo(buffer); if (gzip) { try (GzipSource source = new GzipSource(buffer)) { Buffer decoded = new Buffer(); decoded.writeAll(source); - Assert.assertArrayEquals(expected.getBytes(StandardCharsets.UTF_8), + Assert.assertArrayEquals(expected.getBytes(charset), decoded.readByteArray()); } } else { - Assert.assertArrayEquals(expected.getBytes(StandardCharsets.UTF_8), + Assert.assertArrayEquals(expected.getBytes(charset), buffer.readByteArray()); } } diff --git a/hugegraph-server/README.md b/hugegraph-server/README.md index d8fe1d7746..c83e4a305d 100644 --- a/hugegraph-server/README.md +++ b/hugegraph-server/README.md @@ -6,6 +6,8 @@ HugeGraph Server consists of two layers of functionality: the graph engine layer - Graph Engine Layer: - REST Server: Provides a RESTful API for querying graph/schema information, supports the [Gremlin](https://tinkerpop.apache.org/gremlin.html) and [Cypher](https://en.wikipedia.org/wiki/Cypher) query languages, and offers APIs for service monitoring and operations. + - See [Cypher compatibility notes](../docs/cypher-compatibility.md) for request forms, verified behavior, and + unverified scope. - Graph Engine: Supports both OLTP and OLAP graph computation types, with OLTP implementing the [Apache TinkerPop3](https://tinkerpop.apache.org) framework. - Backend Interface: Implements the storage of graph data to the backend. diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherAPI.java index a30bff73b8..939ccfe22b 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherAPI.java @@ -20,6 +20,7 @@ import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.util.Base64; +import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -29,11 +30,19 @@ import org.apache.hugegraph.api.filter.CompressInterceptor; import org.apache.hugegraph.util.E; import org.apache.hugegraph.util.Log; +import org.apache.tinkerpop.shaded.jackson.core.JsonProcessingException; +import org.apache.tinkerpop.shaded.jackson.databind.DeserializationFeature; +import org.apache.tinkerpop.shaded.jackson.databind.ObjectMapper; +import org.apache.tinkerpop.shaded.jackson.databind.ObjectReader; import org.slf4j.Logger; import com.codahale.metrics.annotation.Timed; import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.media.Content; +import io.swagger.v3.oas.annotations.media.ExampleObject; +import io.swagger.v3.oas.annotations.media.Schema; +import io.swagger.v3.oas.annotations.parameters.RequestBody; import io.swagger.v3.oas.annotations.tags.Tag; import jakarta.inject.Singleton; import jakarta.ws.rs.Consumes; @@ -55,6 +64,9 @@ public class CypherAPI extends API { private static final Logger LOG = Log.logger(CypherAPI.class); private static final Charset UTF8 = StandardCharsets.UTF_8; private static final String CLIENT_CONF = "conf/remote-objects.yaml"; + private static final ObjectReader REQUEST_READER = new ObjectMapper() + .readerFor(Object.class) + .with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); private final Base64.Decoder decoder = Base64.getUrlDecoder(); private final String basic = "Basic "; private final String bearer = "Bearer "; @@ -80,7 +92,8 @@ public CypherModel query(@Context HttpHeaders headers, @Parameter(description = "The cypher query string") @QueryParam("cypher") String cypher) { - return this.queryByCypher(headers, graphspace, graph, cypher); + return this.queryByCypher(headers, graphspace, graph, cypher, + Collections.emptyMap()); } @@ -89,32 +102,81 @@ public CypherModel query(@Context HttpHeaders headers, @CompressInterceptor.Compress @Consumes(APPLICATION_JSON) @Produces(APPLICATION_JSON_WITH_CHARSET) + @RequestBody(required = true, + description = "A nonblank raw Cypher query, or a JSON object with a nonblank " + + "string 'cypher' and optional object 'parameters'. Omitted " + + "parameters default to an empty object; null, arrays and " + + "scalars are rejected. The JSON request itself must be an object. " + + "Legacy raw Cypher sent as application/json " + + "is also accepted.", + content = { + @Content(mediaType = APPLICATION_JSON, + schema = @Schema(implementation = Map.class), + examples = @ExampleObject(name = "Parameterized query", + value = "{\"cypher\":\"MATCH (n:person) WHERE " + + "n.name = $name RETURN n.name\"," + + "\"parameters\":{\"name\":\"marko\"}}")) + }) public CypherModel post(@Context HttpHeaders headers, @Parameter(description = "The graph space name") @PathParam("graphspace") String graphspace, @Parameter(description = "The graph name") @PathParam("graph") String graph, - @Parameter(description = "The cypher query string") String cypher) { - return this.queryByCypher(headers, graphspace, graph, cypher); + Map parameters = Collections.emptyMap(); + if (looksLikeJson(cypher)) { + Object value; + try { + value = REQUEST_READER.readValue(cypher); + } catch (JsonProcessingException e) { + throw new IllegalArgumentException("Invalid Cypher request JSON", e); + } + E.checkArgument(value instanceof Map, "The Cypher request must be a JSON object"); + Map request = (Map) value; + Object query = request.get("cypher"); + E.checkArgument(query instanceof String, + "The cypher parameter must be a nonblank string"); + cypher = (String) query; + if (request.containsKey("parameters")) { + Object bindings = request.get("parameters"); + E.checkArgument(bindings instanceof Map, + "The parameters parameter must be an object"); + @SuppressWarnings("unchecked") + Map supplied = (Map) bindings; + parameters = supplied; + } + } + return this.queryByCypher(headers, graphspace, graph, cypher, parameters); + } + + private static boolean looksLikeJson(String body) { + if (body == null || body.isBlank()) { + return false; + } + String value = body.strip(); + char first = value.charAt(0); + return first == '{' || first == '[' || first == '"' || first == '-' || + (first >= '0' && first <= '9') || value.equals("null") || + value.equals("true") || value.equals("false"); } private CypherModel queryByCypher(HttpHeaders headers, String graphspace, - String graph, String cypher) { + String graph, String cypher, + Map parameters) { E.checkArgument(graphspace != null && !graphspace.isEmpty(), "The graphspace parameter can't be null or empty"); E.checkArgument(graph != null && !graph.isEmpty(), "The graph parameter can't be null or empty"); - E.checkArgument(cypher != null && !cypher.isEmpty(), - "The cypher parameter can't be null or empty"); + E.checkArgument(cypher != null && !cypher.isBlank(), + "The cypher parameter must be a nonblank string"); String graphInfo = graphspace + "-" + graph; Map aliases = new HashMap<>(2, 1); aliases.put("graph", graphInfo); aliases.put("g", "__g_" + graphInfo); - return this.client(headers).submitQuery(cypher, aliases); + return this.client(headers).submitQuery(cypher, aliases, parameters); } private CypherClient client(HttpHeaders headers) { diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherClient.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherClient.java index ebe0c7bb3c..dc9cf16d00 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherClient.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherClient.java @@ -18,6 +18,7 @@ package org.apache.hugegraph.api.cypher; import java.lang.reflect.Array; +import java.util.Collections; import java.util.IdentityHashMap; import java.util.Iterator; import java.util.LinkedHashMap; @@ -67,20 +68,25 @@ public final class CypherClient { } public CypherModel submitQuery(String cypherQuery, @Nullable Map aliases) { - E.checkArgument(cypherQuery != null && !cypherQuery.isEmpty(), - "The cypher-query parameter can't be null or empty"); + return this.submitQuery(cypherQuery, aliases, Collections.emptyMap()); + } - Cluster cluster = Cluster.open(getConfig()); - Client client = cluster.connect(); + public CypherModel submitQuery(String cypherQuery, @Nullable Map aliases, + Map parameters) { + E.checkArgument(cypherQuery != null && !cypherQuery.isBlank(), + "The cypher-query parameter must be a nonblank string"); + E.checkArgument(parameters != null, "The parameters parameter must be an object"); - if (aliases != null && !aliases.isEmpty()) { - client = client.alias(aliases); - } - - RequestMessage request = createRequest(cypherQuery); + RequestMessage request = createRequest(cypherQuery, parameters); + Cluster cluster = Cluster.open(getConfig()); + Client client = null; CypherModel res; try { + client = cluster.connect(); + if (aliases != null && !aliases.isEmpty()) { + client = client.alias(aliases); + } List list = this.doQueryList(client, request); res = CypherModel.dataOf(request.getRequestId().toString(), list); } catch (Exception e) { @@ -88,17 +94,23 @@ public CypherModel submitQuery(String cypherQuery, @Nullable Map cypherQuery), e); res = CypherModel.failOf(request.getRequestId().toString(), e.getMessage()); } finally { - client.close(); - cluster.close(); + try { + if (client != null) { + client.close(); + } + } finally { + cluster.close(); + } } return res; } - private RequestMessage createRequest(String cypherQuery) { + static RequestMessage createRequest(String cypherQuery, Map parameters) { return RequestMessage.build(Tokens.OPS_EVAL) .processor("cypher") .add(Tokens.ARGS_GREMLIN, cypherQuery) + .add(Tokens.ARGS_BINDINGS, parameters) .create(); } diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java index 6911b20302..f183871814 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java @@ -27,6 +27,7 @@ import org.apache.hugegraph.auth.HugeGraphAuthProxy.ContextThreadPoolExecutor; import org.apache.hugegraph.config.CoreOptions; import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.server.HugeGraphWsAndHttpChannelizer; import org.apache.hugegraph.testutil.Whitebox; import org.apache.hugegraph.util.Events; import org.apache.hugegraph.util.Log; @@ -35,6 +36,7 @@ import org.apache.tinkerpop.gremlin.server.GraphManager; import org.apache.tinkerpop.gremlin.server.GremlinServer; import org.apache.tinkerpop.gremlin.server.Settings; +import org.apache.tinkerpop.gremlin.server.channel.WsAndHttpChannelizer; import org.apache.tinkerpop.gremlin.server.util.ThreadFactoryUtil; import org.apache.tinkerpop.gremlin.structure.Graph; import org.slf4j.Logger; @@ -58,11 +60,18 @@ public ContextGremlinServer(final Settings settings, EventHub eventHub) { /* * pass custom Executor https://github.com/apache/tinkerpop/pull/813 */ - super(settings, newGremlinExecutorService(settings)); + super(configureChannelizer(settings), newGremlinExecutorService(settings)); this.eventHub = eventHub; this.listenChanges(); } + static Settings configureChannelizer(Settings settings) { + if (WsAndHttpChannelizer.class.getName().equals(settings.channelizer)) { + settings.channelizer = HugeGraphWsAndHttpChannelizer.class.getName(); + } + return settings; + } + private void listenChanges() { this.eventHub.listen(Events.GRAPH_CREATE, event -> { LOG.debug("GremlinServer accepts event '{}'", event.name()); diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessor.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessor.java index 2bcfa7ec83..6097dbeed9 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessor.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessor.java @@ -23,13 +23,15 @@ import static org.opencypher.gremlin.translation.StatementOption.EXPLAIN; import static org.slf4j.LoggerFactory.getLogger; +import java.util.ArrayDeque; +import java.util.Collection; +import java.util.Deque; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.Map; import java.util.Optional; -import java.util.concurrent.Future; import java.util.concurrent.FutureTask; import java.util.concurrent.TimeUnit; @@ -60,9 +62,11 @@ import org.opencypher.gremlin.traversal.ParameterNormalizer; import org.opencypher.gremlin.traversal.ProcedureContext; import org.opencypher.gremlin.traversal.ReturnNormalizer; +import org.opencypher.v9_0.expressions.Parameter; import org.slf4j.Logger; import io.netty.channel.ChannelHandlerContext; +import scala.Product; import scala.collection.Seq; /** @@ -102,12 +106,74 @@ public class CypherOpProcessor extends AbstractEvalOpProcessor { private static final String DEFAULT_TRANSLATOR_DEFINITION = "gremlin+cfog_server_extensions+inline_parameters"; + // translation 1.0.4 uses this string as null and cannot preserve it as bound data. + private static final String RESERVED_NULL_VALUE = " cypher.null"; + private static final Logger logger = getLogger(CypherOpProcessor.class); public CypherOpProcessor() { super(true); } + @Override + protected Optional> validateEvalMessage(RequestMessage message) + throws OpProcessorException { + Object query = message.getArgs().get(Tokens.ARGS_GREMLIN); + String error = null; + if (!(query instanceof String) || ((String) query).isBlank()) { + error = "The Cypher query must be a nonblank string"; + } else if (message.getArgs().containsKey(Tokens.ARGS_BINDINGS)) { + Object supplied = message.getArgs().get(Tokens.ARGS_BINDINGS); + if (!(supplied instanceof Map)) { + error = "Cypher bindings must be a map"; + } else { + Map bindings = (Map) supplied; + if (bindings.size() > this.maxParameters) { + error = "Cypher bindings exceed the maximum parameter count " + this.maxParameters; + } else { + // Cypher parameters do not enter the Gremlin script namespace. + for (Object key : bindings.keySet()) { + if (!(key instanceof String)) { + error = "Cypher binding keys must be nonnull strings"; + break; + } + } + if (error == null && containsReservedNullValue(bindings)) { + error = "Cypher bindings contain a string reserved for null by the translator"; + } + } + } + } + if (error != null) { + throw new OpProcessorException(error, ResponseMessage.build(message) + .code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS) + .statusMessage(error).create()); + } + return empty(); + } + + private static boolean containsReservedNullValue(Map bindings) { + Deque> pending = new ArrayDeque<>(); + pending.push(bindings.values().iterator()); + while (!pending.isEmpty()) { + Iterator values = pending.peek(); + if (!values.hasNext()) { + pending.pop(); + continue; + } + Object value = values.next(); + if (RESERVED_NULL_VALUE.equals(value)) { + return true; + } + if (value instanceof Map) { + pending.push(((Map) value).values().iterator()); + } else if (value instanceof Collection) { + pending.push(((Collection) value).iterator()); + } + } + return false; + } + @Override public String getName() { return "cypher"; @@ -133,6 +199,7 @@ private void evalCypher(Context context) throws OpProcessorException { Map parameters = ParameterNormalizer.normalize(getParameters(args)); ProcedureContext procedureContext = ProcedureContext.global(); CypherAst ast = CypherAst.parse(cypher, parameters, procedureContext.getSignatures()); + validateParameters(ast, parameters); String translatorDefinition = getTranslatorDefinition(context); @@ -161,24 +228,36 @@ private void evalCypher(Context context) throws OpProcessorException { parameters); ReturnNormalizer returnNormalizer = ReturnNormalizer.create(ast.getReturnTypes()); Iterator normalizedTraversal = returnNormalizer.normalize(traversal); - inTransaction(gts, () -> handleIterator(context, normalizedTraversal)); + handleTraversal(context, normalizedTraversal, traversal); } - private void inTransaction(GraphTraversalSource gts, Runnable runnable) { - Graph graph = gts.getGraph(); - boolean supportsTransactions = graph.features().graph().supportsTransactions(); - if (!supportsTransactions) { - runnable.run(); - return; - } - - try { - graph.tx().open(); - runnable.run(); - graph.tx().commit(); - } catch (Exception e) { - if (graph.tx().isOpen()) { - graph.tx().rollback(); + private static void validateParameters(CypherAst ast, Map parameters) { + // Inspect parsed parameter nodes: dollar signs in literals/comments are not bindings. + // The upstream translator otherwise silently turns an absent binding into null. + Deque pending = new ArrayDeque<>(); + pending.push(ast.statement()); + while (!pending.isEmpty()) { + Object node = pending.pop(); + if (node instanceof Parameter) { + String name = ((Parameter) node).name(); + if (!parameters.containsKey(name)) { + throw new IllegalArgumentException("Missing Cypher parameter: " + name); + } + continue; + } + scala.collection.Iterator children; + if (node instanceof scala.collection.Iterable) { + children = ((scala.collection.Iterable) node).iterator(); + } else if (node instanceof Product) { + children = ((Product) node).productIterator(); + } else { + continue; + } + while (children.hasNext()) { + Object child = children.next(); + if (child != null) { + pending.push(child); + } } } } @@ -221,22 +300,45 @@ private OpProcessorException opProcessorException(RequestMessage msg, String err .create()); } - @Override - protected void handleIterator(Context context, Iterator traversal) { + private void handleTraversal(Context context, Iterator traversal, + GraphTraversal source) { RequestMessage msg = context.getRequestMessage(); final long timeout = msg.getArgs().containsKey(Tokens.ARGS_EVAL_TIMEOUT) ? ((Number) msg.getArgs().get(Tokens.ARGS_EVAL_TIMEOUT)).longValue() : context.getSettings().evaluationTimeout; FutureTask evalFuture = new FutureTask<>(() -> { - try { + try (GraphTraversal ignored = source) { + // The parent commits before sending the terminal success response. Both + // iteration and failure rollback must run on this transaction-owning thread. super.handleIterator(context, traversal); + } catch (Error error) { + try { + attemptRollback(msg, context.getGraphManager(), + context.getSettings().strictTransactionManagement); + } catch (Exception rollbackFailure) { + error.addSuppressed(rollbackFailure); + } + // FutureTask retains the error, so send a terminal response here as well. + // This also covers errors raised after a timeout has cancelled the task. + logger.error("Fatal error during traversal iteration", error); + context.writeAndFlush(ResponseMessage.build(msg) + .code(SERVER_ERROR) + .statusMessage(error.getMessage()) + .statusAttributeException(error) + .create()); + throw error; } catch (Exception ex) { + try { + attemptRollback(msg, context.getGraphManager(), + context.getSettings().strictTransactionManagement); + } catch (Exception rollbackFailure) { + ex.addSuppressed(rollbackFailure); + } String errorMessage = getErrorMessage(msg, ex); logger.error("Error during traversal iteration", ex); - ChannelHandlerContext ctx = context.getChannelHandlerContext(); - ctx.writeAndFlush(ResponseMessage.build(msg) + context.writeAndFlush(ResponseMessage.build(msg) .code(SERVER_ERROR) .statusMessage(errorMessage) .statusAttributeException(ex) @@ -246,11 +348,10 @@ protected void handleIterator(Context context, Iterator traversal) { } ); - final Future executionFuture = context.getGremlinExecutor() - .getExecutorService().submit(evalFuture); + context.getGremlinExecutor().getExecutorService().execute(evalFuture); if (timeout > 0) { context.getScheduledExecutorService().schedule( - () -> executionFuture.cancel(true) + () -> evalFuture.cancel(true) , timeout, TimeUnit.MILLISECONDS); } diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HttpGremlinRequestHandler.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HttpGremlinRequestHandler.java new file mode 100644 index 0000000000..6eae82e21d --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HttpGremlinRequestHandler.java @@ -0,0 +1,67 @@ +/* + * 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.hugegraph.server; + +import java.util.Map; + +import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor; +import org.apache.tinkerpop.gremlin.server.GraphManager; +import org.apache.tinkerpop.gremlin.server.Settings; +import org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler; +import org.apache.tinkerpop.gremlin.util.MessageSerializer; + +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http.FullHttpRequest; + +@ChannelHandler.Sharable +public class HttpGremlinRequestHandler extends HttpGremlinEndpointHandler { + + public HttpGremlinRequestHandler(Map> serializers, + GremlinExecutor executor, + GraphManager graphManager, + Settings settings) { + super(serializers, executor, graphManager, settings); + } + + @Override + public void channelRead(ChannelHandlerContext context, Object message) { + if (!(message instanceof FullHttpRequest)) { + super.channelRead(context, message); + return; + } + + FullHttpRequest request = (FullHttpRequest) message; + try { + /* + * TinkerPop 3.8.1 releases the request before resolving aliases, + * then releases it again on an unknown alias. That second release + * writes a spurious 500 after the intended 400 on the same stream. + * Own the original reference here and let the upstream handler + * parse a view whose release is a no-op. It parses synchronously; + * asynchronous evaluation holds only the parsed RequestMessage. + */ + FullHttpRequest view = request.replace( + Unpooled.unreleasableBuffer(request.content())); + super.channelRead(context, view); + } finally { + request.release(); + } + } +} diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HugeGraphWsAndHttpChannelizer.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HugeGraphWsAndHttpChannelizer.java new file mode 100644 index 0000000000..94f358425f --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HugeGraphWsAndHttpChannelizer.java @@ -0,0 +1,55 @@ +/* + * 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.hugegraph.server; + +import org.apache.tinkerpop.gremlin.server.AbstractChannelizer; +import org.apache.tinkerpop.gremlin.server.handler.WsAndHttpChannelizerHandler; +import org.apache.tinkerpop.gremlin.server.util.ServerGremlinExecutor; + +import io.netty.channel.ChannelPipeline; + +public class HugeGraphWsAndHttpChannelizer extends AbstractChannelizer { + + private WsAndHttpChannelizerHandler handler; + + @Override + public void init(ServerGremlinExecutor executor) { + super.init(executor); + this.handler = new WsAndHttpChannelizerHandler(); + this.handler.init(executor, new HttpGremlinRequestHandler( + this.serializers, this.gremlinExecutor, + this.graphManager, this.settings)); + } + + @Override + public void configure(ChannelPipeline pipeline) { + this.handler.configure(pipeline); + pipeline.addAfter(PIPELINE_HTTP_REQUEST_DECODER, + "WsAndHttpChannelizerHandler", this.handler); + } + + @Override + public boolean supportsIdleMonitor() { + return true; + } + + @Override + public Object createIdleDetectionMessage() { + return this.handler.getWsChannelizer().createIdleDetectionMessage(); + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/CypherApiTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/CypherApiTest.java index 23abf42601..ba5884739a 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/CypherApiTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/CypherApiTest.java @@ -19,6 +19,9 @@ import static org.apache.hugegraph.testutil.Assert.assertContains; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; @@ -29,22 +32,49 @@ import com.google.common.collect.ImmutableMap; +import jakarta.ws.rs.client.Entity; +import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; public class CypherApiTest extends BaseApiTest { private static final String PATH = URL_PREFIX + "/cypher"; - private static final String QUERY = "MATCH (n:person) where n.city ='Beijing' return n"; - private static final String QUERY_RESULT = "Beijing"; + private static final String QUERY = "MATCH (n:cypher_person) WHERE n.city = 'Beijing' " + + "RETURN n.name AS name ORDER BY name"; @Before public void prepareSchema() { - BaseApiTest.initPropertyKey(); - BaseApiTest.initVertexLabel(); - BaseApiTest.initEdgeLabel(); - BaseApiTest.initIndexLabel(); - BaseApiTest.initVertex(); - BaseApiTest.initEdge(); + String schema = URL_PREFIX + "/schema/"; + for (String[] property : new String[][]{{"name", "TEXT"}, {"age", "INT"}, + {"city", "TEXT"}, {"active", "BOOLEAN"}, + {"note", "TEXT"}}) { + createAndAssert(schema + "propertykeys", JsonUtil.toJson(ImmutableMap.of( + "name", property[0], "data_type", property[1], + "cardinality", "SINGLE")), 202); + } + createAndAssert(schema + "vertexlabels", "{\"name\":\"cypher_person\", " + + "\"id_strategy\":\"PRIMARY_KEY\",\"primary_keys\":[\"name\"]," + + "\"properties\":[\"name\",\"age\",\"city\",\"active\",\"note\"]," + + "\"nullable_keys\":[\"active\",\"note\"]}"); + createAndAssert(schema + "edgelabels", "{\"name\":\"cypher_knows\", " + + "\"source_label\":\"cypher_person\",\"target_label\":\"cypher_person\"," + + "\"frequency\":\"SINGLE\",\"properties\":[],\"nullable_keys\":[]}"); + // Install indexes before inserting data; no existing data needs rebuilding. + for (String[] index : new String[][]{{"city", "SECONDARY"}, {"age", "RANGE"}}) { + createAndAssert(schema + "indexlabels", JsonUtil.toJson(ImmutableMap.builder() + .put("name", "cypher_person_by_" + index[0]) + .put("base_type", "VERTEX_LABEL") + .put("base_value", "cypher_person") + .put("index_type", index[1]) + .put("fields", Collections.singletonList(index[0])) + .put("rebuild", false).build()), 202); + } + addVertex("marko", 20, "Beijing", true); + addVertex("peter", 30, "Beijing", false); + addVertex("alice", 40, "Shanghai", true); + addVertex("dave", 30, "Shanghai", false); + addEdge("marko", "peter"); + addEdge("peter", "alice"); } @Test @@ -52,17 +82,23 @@ public void testGet() { Map params = ImmutableMap.of("cypher", QUERY); Response r = client().get(PATH, params); - this.validStatusAndTextContains(QUERY_RESULT, r); + assertColumn(assertCypherSuccessData(assertResponseStatus(200, r)), "name", "marko", "peter"); } @Test public void testPost() { - this.testCypherQueryAndContains(QUERY, QUERY_RESULT); + assertColumn(query(QUERY), "name", "marko", "peter"); + } + + @Test + public void testRejectPlainTextPost() { + Response response = client().post(PATH, Entity.entity(QUERY, MediaType.TEXT_PLAIN_TYPE)); + assertResponseStatus(415, response); } @Test public void testComputedRegexExecutesExtensionPredicate() { - String query = "MATCH (n:person) WHERE (n.name + '') =~ 'mar.*' " + + String query = "MATCH (n:cypher_person) WHERE (n.name + '') =~ 'mar.*' " + "RETURN n.name AS name"; String content = this.testCypherQueryAndContains(query, "marko"); Assert.assertEquals(List.of(Map.of("name", "marko")), assertCypherSuccessData(content)); @@ -70,7 +106,7 @@ public void testComputedRegexExecutesExtensionPredicate() { @Test public void testComputedRegexRequiresWholeStringMatch() { - String query = "MATCH (n:person) WHERE (n.name + '') =~ 'ark' " + + String query = "MATCH (n:cypher_person) WHERE (n.name + '') =~ 'ark' " + "RETURN n.name AS name"; String content = this.testCypherQueryAndContains(query, "data"); Assert.assertTrue(assertCypherSuccessData(content).isEmpty()); @@ -78,22 +114,22 @@ public void testComputedRegexRequiresWholeStringMatch() { @Test public void testCreate() { - this.testCypherQueryAndContains("CREATE (n:person { name : 'test', " + - "age: 20, city: 'Hefei' }) return n", - "Hefei"); + assertColumn(query("CREATE (n:cypher_person {name:'test', age:20, city:'Hefei'}) " + + "RETURN n.name AS name"), "name", "test"); + assertNativeVertex("test", 20, "Hefei"); } @Test public void testRelationQuery() { - String cypher = "MATCH (n:person)-[r:knows]->(friend:person)\n" + + String cypher = "MATCH (n:cypher_person)-[r:cypher_knows]->(friend:cypher_person)\n" + "WHERE n.name = 'marko'\n" + - "RETURN n, friend.name AS friend"; - this.testCypherQueryAndContains(cypher, "friend"); + "RETURN friend.name AS friend"; + assertColumn(query(cypher), "friend", "peter"); } @Test public void testReturnNodeIdAsPrimitiveValue() { - String cypher = "MATCH (n:person) WHERE n.name = 'marko' " + + String cypher = "MATCH (n:cypher_person) WHERE n.name = 'marko' " + "RETURN id(n) AS nodeId"; String content = this.testCypherQueryAndContains(cypher, "nodeId"); @@ -101,22 +137,27 @@ public void testReturnNodeIdAsPrimitiveValue() { Map row = assertSingleMapRow(data); Object nodeId = row.get("nodeId"); - Assert.assertNotNull(nodeId); + Assert.assertEquals(vertexId("marko"), nodeId); assertPrimitiveValue(nodeId); assertNoHugeGraphIdLeak(content); } @Test public void testReturnNodeDoesNotLeakInternalIdTypes() { - String cypher = "MATCH (n:person) WHERE n.name = 'marko' RETURN n"; + String cypher = "MATCH (n:cypher_person) WHERE n.name = 'marko' RETURN n"; String content = this.testCypherQueryAndContains(cypher, "marko"); + Map node = assertMapValue(assertSingleMapRow(assertCypherSuccessData(content)), "n"); + Assert.assertEquals("cypher_person", node.get("_label")); + Assert.assertEquals("marko", node.get("name")); + Assert.assertEquals(20, ((Number) node.get("age")).intValue()); + Assert.assertEquals("Beijing", node.get("city")); assertNoHugeGraphIdLeak(content); } @Test public void testReturnNestedIdDoesNotLeakInternalIdTypes() { - String cypher = "MATCH (n:person) WHERE n.name = 'marko' " + + String cypher = "MATCH (n:cypher_person) WHERE n.name = 'marko' " + "RETURN {nodeId: id(n), values: [id(n), n.name]} " + "AS payload"; @@ -127,7 +168,7 @@ public void testReturnNestedIdDoesNotLeakInternalIdTypes() { List values = assertListValue(payload, "values"); Object nodeId = payload.get("nodeId"); - Assert.assertNotNull(nodeId); + Assert.assertEquals(vertexId("marko"), nodeId); assertPrimitiveValue(nodeId); Assert.assertEquals(2, values.size()); Assert.assertEquals(nodeId, values.get(0)); @@ -137,7 +178,7 @@ public void testReturnNestedIdDoesNotLeakInternalIdTypes() { @Test public void testReturnRelationIdDoesNotLeakInternalIdTypes() { - String cypher = "MATCH (n:person)-[r:knows]->(friend:person) " + + String cypher = "MATCH (n:cypher_person)-[r:cypher_knows]->(friend:cypher_person) " + "WHERE n.name = 'marko' RETURN id(r) AS relationId"; String content = this.testCypherQueryAndContains(cypher, "relationId"); @@ -152,7 +193,7 @@ public void testReturnRelationIdDoesNotLeakInternalIdTypes() { @Test public void testReturnPathShape() { - String cypher = "MATCH p=(n:person)-[r:knows]->(friend:person) " + + String cypher = "MATCH p=(n:cypher_person)-[r:cypher_knows]->(friend:cypher_person) " + "WHERE n.name = 'marko' RETURN p AS path"; String content = this.testCypherQueryAndContains(cypher, "path"); @@ -166,17 +207,297 @@ public void testReturnPathShape() { Map target = assertMapValue(path, 2); Assert.assertEquals("node", source.get("_type")); - Assert.assertEquals("person", source.get("_label")); + Assert.assertEquals("cypher_person", source.get("_label")); Assert.assertEquals("marko", source.get("name")); - Assert.assertEquals("knows", relation.get("_label")); + Assert.assertEquals("cypher_knows", relation.get("_label")); Assert.assertEquals("node", target.get("_type")); - Assert.assertEquals("person", target.get("_label")); + Assert.assertEquals("cypher_person", target.get("_label")); Assert.assertEquals("peter", target.get("name")); assertContains("marko", content); assertContains("peter", content); assertNoHugeGraphIdLeak(content); } + @Test + public void testExactReadsAndPredicates() { + assertColumn(query("MATCH (n:cypher_person) RETURN n.name AS name ORDER BY name"), + "name", "alice", "dave", "marko", "peter"); + assertColumn(query("MATCH (n:cypher_person) WHERE n.age >= 30 AND " + + "(n.active = true OR n.name = 'dave') " + + "RETURN n.name AS name ORDER BY name"), "name", "alice", "dave"); + assertColumn(query("MATCH (n:cypher_person)-[:cypher_knows]->(m:cypher_person) " + + "RETURN n.name AS name ORDER BY name"), "name", "marko", "peter"); + assertColumn(query("MATCH (n:cypher_person)-[:cypher_knows]->()" + + "-[:cypher_knows]->(m:cypher_person) RETURN m.name AS name"), + "name", "alice"); + Assert.assertTrue(query("MATCH (n:cypher_person) WHERE n.name = 'absent' RETURN n").isEmpty()); + } + + @Test + public void testDuplicatesDistinctAndPagination() { + assertColumn(query("MATCH (n:cypher_person) RETURN n.city AS city ORDER BY city"), + "city", "Beijing", "Beijing", "Shanghai", "Shanghai"); + assertColumn(query("MATCH (n:cypher_person) RETURN DISTINCT n.city AS city ORDER BY city"), + "city", "Beijing", "Shanghai"); + assertColumn(query("MATCH (n:cypher_person) RETURN n.name AS name " + + "ORDER BY n.age, name SKIP 1 LIMIT 2"), "name", "dave", "peter"); + } + + @Test + public void testAggregates() { + Map row = assertSingleMapRow(query("MATCH (n:cypher_person) " + + "RETURN count(n) AS count, sum(n.age) AS sum, min(n.age) AS min, " + + "max(n.age) AS max, avg(n.age) AS avg")); + Assert.assertEquals(5, row.size()); + for (String key : Arrays.asList("count", "sum", "min", "max", "avg")) { + Assert.assertInstanceOf(Number.class, row.get(key)); + } + Assert.assertEquals(4L, ((Number) row.get("count")).longValue()); + Assert.assertEquals(120L, ((Number) row.get("sum")).longValue()); + Assert.assertEquals(20L, ((Number) row.get("min")).longValue()); + Assert.assertEquals(40L, ((Number) row.get("max")).longValue()); + Assert.assertEquals(30.0D, ((Number) row.get("avg")).doubleValue(), 0.0D); + } + + @Test + public void testParameters() { + assertColumn(bound("MATCH (n:cypher_person) WHERE n.name = $name AND " + + "n.age >= $age AND n.active = $active RETURN n.name AS name", + ImmutableMap.of("name", "marko", "age", 20, "active", true)), + "name", "marko"); + String text = "quote ' and \" newline\n MATCH (n) DELETE n //"; + assertColumn(bound("RETURN $text AS text", ImmutableMap.of("text", text)), "text", text); + assertColumn(bound("MATCH (n:cypher_person) RETURN n.name AS name ORDER BY name", + Collections.emptyMap()), "name", "alice", "dave", "marko", "peter"); + assertColumn(assertCypherSuccessData(assertResponseStatus(200, client().post(PATH, + "{\"cypher\":\"RETURN 1 AS value\"}"))), "value", 1); + assertExecutionError(client().post(PATH, JsonUtil.toJson(ImmutableMap.of( + "cypher", "RETURN $missing AS value", "parameters", Collections.emptyMap())))); + assertColumn(bound("RETURN $value AS value", Collections.singletonMap("value", null)), + "value", (Object) null); + Map reservedNames = assertSingleMapRow(bound("RETURN $id AS id, $label AS label", + ImmutableMap.of("id", "bound-id", "label", "bound-label"))); + Assert.assertEquals(ImmutableMap.of("id", "bound-id", "label", "bound-label"), reservedNames); + assertColumn(bound("RETURN '$not_a_binding' AS value // $ignored", + Collections.emptyMap()), "value", "$not_a_binding"); + assertExecutionError(client().post(PATH, JsonUtil.toJson(ImmutableMap.of( + "cypher", "RETURN {nested: $missing} AS value", + "parameters", Collections.emptyMap())))); + Assert.assertEquals(4, nativeVertices().size()); + } + + @Test + public void testRejectTranslatorNullMarker() { + String marker = " cypher.null"; + for (Object value : Arrays.asList(marker, List.of(marker), Map.of("nested", marker))) { + assertExecutionError(client().post(PATH, JsonUtil.toJson(ImmutableMap.of( + "cypher", "RETURN $value AS value", "parameters", Map.of("value", value))))); + } + assertColumn(bound("RETURN $value AS value", Map.of("value", "cypher.null")), + "value", "cypher.null"); + } + + @Test + public void testNullAndMissingProperty() { + List rows = query("MATCH (n:cypher_person) RETURN n.note AS note"); + Assert.assertEquals(4, rows.size()); + for (Object value : rows) { + Map row = (Map) value; + Assert.assertTrue(row.containsKey("note")); + Assert.assertNull(row.get("note")); + } + Assert.assertEquals(4, query("MATCH (n:cypher_person) WHERE n.note IS NULL RETURN n.name").size()); + Assert.assertTrue(query("MATCH (n:cypher_person) WHERE n.note IS NOT NULL RETURN n.name").isEmpty()); + } + + @Test + public void testCreateSetAndDeleteWithNativeReadback() { + assertColumn(query("CREATE (n:cypher_person {name:'created', age:50, city:'Paris'}) " + + "RETURN n.name AS name"), "name", "created"); + assertNativeVertex("created", 50, "Paris"); + assertColumn(query("MATCH (n:cypher_person) WHERE n.name = 'created' " + + "SET n.age = 51, n.city = 'London' RETURN n.name AS name"), "name", "created"); + assertNativeVertex("created", 51, "London"); + query("MATCH (a:cypher_person), (b:cypher_person) " + + "WHERE a.name = 'created' AND b.name = 'marko' CREATE (a)-[:cypher_knows]->(b)"); + List edges = nativeEdges(); + Assert.assertEquals(3, edges.size()); + boolean found = false; + for (Object value : edges) { + Map edge = (Map) value; + if (vertexId("created").equals(edge.get("outV"))) { + Assert.assertEquals(vertexId("marko"), edge.get("inV")); + Assert.assertEquals("cypher_knows", edge.get("label")); + found = true; + } + } + Assert.assertTrue(found); + query("MATCH (a:cypher_person)-[r:cypher_knows]->() WHERE a.name = 'created' DELETE r"); + Assert.assertEquals(2, nativeEdges().size()); + query("MATCH (n:cypher_person) WHERE n.name = 'created' DELETE n"); + Assert.assertEquals(4, nativeVertices().size()); + Assert.assertNull(nativeVertex("created")); + } + + @Test + public void testFailedWriteDoesNotLeakIntoLaterRequests() { + assertExecutionError(client().post(PATH, + "CREATE (good:cypher_person {name:'rollback', age:33, city:'Beijing'}), " + + "(bad:cypher_person {name:'invalid', age:34, city:'Shanghai', rogue:'x'}) " + + "RETURN good.name")); + Assert.assertNull(nativeVertex("rollback")); + // Failed-write residue became visible after later reads in the regression reproducer. + // Immediate REST readback alone therefore does not establish rollback. + for (int i = 0; i < 32; i++) { + Map row = assertSingleMapRow(query("MATCH (n:cypher_person) RETURN count(n) AS count")); + Assert.assertEquals(4L, ((Number) row.get("count")).longValue()); + Assert.assertEquals(4, nativeVertices().size()); + Assert.assertNull(nativeVertex("rollback")); + Assert.assertNull(nativeVertex("invalid")); + } + } + + @Test + public void testInvalidRequests() { + assertExecutionError(client().post(PATH, "MATCH ( RETURN broken")); + assertExecutionError(client().post(PATH, + "CREATE (:cypher_person {name:'wrongtype', age:'not an integer', city:'Paris'})")); + Assert.assertNull(nativeVertex("wrongtype")); + assertResponseStatus(400, client().get(PATH, ImmutableMap.of("cypher", ""))); + for (String body : Arrays.asList(" \n", "{", "{}", "{\"cypher\":\" \"}", + "{\"cypher\":1}", "[]", "[1, 2]", "\"RETURN 1\"", + " null ", "1", "-1.5", "true", "false", + "{\"cypher\":\"RETURN 1\"} garbage", + "{\"cypher\":\"RETURN 1\"} {}", + "{\"cypher\":\"RETURN 1\",\"parameters\":false}", + "{\"cypher\":\"RETURN 1\",\"parameters\":null}", + "{\"cypher\":\"RETURN 1\",\"parameters\":[]}", + "{\"cypher\":\"RETURN 1\",\"parameters\":3}")) { + assertResponseStatus(400, client().post(PATH, body)); + } + } + + @Test + public void testAuthenticationAndGraphRouting() { + RestClient invalid = new RestClient(BASE_URL, "admin", "invalid-cypher-password"); + try { + assertResponseStatus(401, invalid.post(PATH, "RETURN 1")); + } finally { + invalid.close(); + } + Response response = client().post("graphspaces/DEFAULT/graphs/cypher_missing_graph/cypher", + "MATCH (n) RETURN n"); + assertExecutionError(response); + assertColumn(query("MATCH (n:cypher_person) RETURN n.name AS name ORDER BY name"), + "name", "alice", "dave", "marko", "peter"); + } + + @Test + public void testSpecifiedGraphRouting() { + assumeStandaloneMode(); + String graph = "cypher_route_test"; + String path = "graphspaces/DEFAULT/graphs/" + graph; + assertResponseStatus(201, createGraphInRocksDB("DEFAULT", graph)); + try { + List rows = assertCypherSuccessData(assertResponseStatus(200, + client().post(path + "/cypher", "MATCH (n) RETURN count(n) AS count"))); + Map emptyGraph = assertSingleMapRow(rows); + Assert.assertEquals(0L, ((Number) emptyGraph.get("count")).longValue()); + Map defaultGraph = assertSingleMapRow(query("MATCH (n) RETURN count(n) AS count")); + Assert.assertEquals(4L, ((Number) defaultGraph.get("count")).longValue()); + } finally { + assertResponseStatus(204, client().delete(path, ImmutableMap.of( + "confirm_message", "I'm sure to drop the graph"))); + } + } + + private List query(String cypher) { + return assertCypherSuccessData(assertResponseStatus(200, client().post(PATH, cypher))); + } + + private List bound(String cypher, Map parameters) { + return query(JsonUtil.toJson(ImmutableMap.of("cypher", cypher, "parameters", parameters))); + } + + private static void assertColumn(List rows, String key, Object... expected) { + List actual = new ArrayList<>(); + for (Object value : rows) { + Assert.assertInstanceOf(Map.class, value); + Map row = (Map) value; + Assert.assertEquals(1, row.size()); + Assert.assertTrue(row.containsKey(key)); + actual.add(row.get(key)); + } + Assert.assertEquals(Arrays.asList(expected), actual); + } + + private static void assertExecutionError(Response response) { + String content = assertResponseStatus(200, response); + Map body = JsonUtil.fromJson(content, Map.class); + Map status = assertMapValue(body, "status"); + Assert.assertTrue(content, ((Number) status.get("code")).intValue() >= 400); + Assert.assertFalse(String.valueOf(status.get("message")).isEmpty()); + Assert.assertNull(assertMapValue(body, "result").get("data")); + } + + private void addVertex(String name, int age, String city, boolean active) { + createAndAssert(URL_PREFIX + "/graph/vertices", JsonUtil.toJson(ImmutableMap.of( + "label", "cypher_person", "properties", ImmutableMap.of( + "name", name, "age", age, "city", city, "active", active)))); + } + + private void addEdge(String source, String target) { + createAndAssert(URL_PREFIX + "/graph/edges", JsonUtil.toJson( + ImmutableMap.builder() + .put("label", "cypher_knows") + .put("outV", vertexId(source)) + .put("inV", vertexId(target)) + .put("outVLabel", "cypher_person") + .put("inVLabel", "cypher_person") + .put("properties", Collections.emptyMap()) + .build())); + } + + private List nativeVertices() { + return nativeElements("vertices"); + } + + private List nativeEdges() { + return nativeElements("edges"); + } + + private List nativeElements(String type) { + String content = assertResponseStatus(200, client().get(URL_PREFIX + "/graph/" + type, + ImmutableMap.of("limit", -1))); + return assertListValue(JsonUtil.fromJson(content, Map.class), type); + } + + private Map nativeVertex(String name) { + for (Object value : nativeVertices()) { + Map vertex = (Map) value; + if (name.equals(assertMapValue(vertex, "properties").get("name"))) { + return vertex; + } + } + return null; + } + + private Object vertexId(String name) { + Map vertex = nativeVertex(name); + Assert.assertNotNull(vertex); + return vertex.get("id"); + } + + private void assertNativeVertex(String name, int age, String city) { + Map vertex = nativeVertex(name); + Assert.assertNotNull(vertex); + Assert.assertEquals("cypher_person", vertex.get("label")); + Map properties = assertMapValue(vertex, "properties"); + Assert.assertEquals(name, properties.get("name")); + Assert.assertEquals(age, ((Number) properties.get("age")).intValue()); + Assert.assertEquals(city, properties.get("city")); + } + private String testCypherQueryAndContains(String cypher, String containsText) { Response r = client().post(PATH, cypher); @@ -186,6 +507,7 @@ private String testCypherQueryAndContains(String cypher, private String validStatusAndTextContains(String value, Response r) { String content = assertResponseStatus(200, r); assertContains(value, content); + assertCypherSuccessData(content); return content; } @@ -203,7 +525,7 @@ private static List assertCypherSuccessData(String content) { Assert.assertTrue(response.containsKey("requestId")); Map status = assertMapValue(response, "status"); - Assert.assertEquals(200, ((Number) status.get("code")).intValue()); + Assert.assertEquals(content, 200, ((Number) status.get("code")).intValue()); Assert.assertEquals("", status.get("message")); Map result = assertMapValue(response, "result"); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/cypher/CypherClientTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/cypher/CypherClientTest.java index 04b94cd99b..8303d79031 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/cypher/CypherClientTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/api/cypher/CypherClientTest.java @@ -17,6 +17,7 @@ package org.apache.hugegraph.api.cypher; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -27,10 +28,35 @@ import org.apache.hugegraph.unit.BaseUnitTest; import org.apache.tinkerpop.gremlin.process.traversal.Path; import org.apache.tinkerpop.gremlin.process.traversal.step.util.MutablePath; +import org.apache.tinkerpop.gremlin.util.Tokens; +import org.apache.tinkerpop.gremlin.util.message.RequestMessage; import org.junit.Test; public class CypherClientTest extends BaseUnitTest { + @Test + public void testRequestPreservesQueryAndParameterValues() { + String query = "MATCH (n:person) WHERE n.name = $name RETURN n.name"; + Map parameters = new LinkedHashMap<>(); + parameters.put("name", "quote'\"\nRETURN 999 //"); + parameters.put("age", 29); + parameters.put("active", true); + + RequestMessage request = CypherClient.createRequest(query, parameters); + + Assert.assertEquals("cypher", request.getProcessor()); + Assert.assertEquals(query, request.getArgs().get(Tokens.ARGS_GREMLIN)); + Assert.assertEquals(parameters, request.getArgs().get(Tokens.ARGS_BINDINGS)); + } + + @Test + public void testRequestAcceptsEmptyParameters() { + RequestMessage request = CypherClient.createRequest("RETURN 1", Collections.emptyMap()); + + Assert.assertEquals(Collections.emptyMap(), request.getArgs().get(Tokens.ARGS_BINDINGS)); + } + + @Test public void testNormalizeHandlesNullMapAndArrayValues() { Map value = new LinkedHashMap<>(); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/ContextGremlinServerHttpTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/ContextGremlinServerHttpTest.java new file mode 100644 index 0000000000..03c2bea0d0 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/ContextGremlinServerHttpTest.java @@ -0,0 +1,183 @@ +/* + * 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.hugegraph.auth; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.server.HugeGraphWsAndHttpChannelizer; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.util.JsonUtil; +import org.apache.tinkerpop.gremlin.driver.Client; +import org.apache.tinkerpop.gremlin.driver.Cluster; +import org.apache.tinkerpop.gremlin.server.Settings; +import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer; +import org.apache.tinkerpop.gremlin.server.channel.WsAndHttpChannelizer; +import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph; +import org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV3; +import org.apache.tinkerpop.gremlin.util.ser.GraphSONUntypedMessageSerializerV1; +import org.junit.Test; + +import io.netty.channel.Channel; + +public class ContextGremlinServerHttpTest { + + @Test + public void testOnlyStandardChannelizerIsAdapted() { + try { + Settings settings = new Settings(); + settings.channelizer = WsAndHttpChannelizer.class.getName(); + Assert.assertSame(settings, + ContextGremlinServer.configureChannelizer(settings)); + Assert.assertEquals(HugeGraphWsAndHttpChannelizer.class.getName(), + settings.channelizer); + settings.channelizer = "example.CustomChannelizer"; + ContextGremlinServer.configureChannelizer(settings); + Assert.assertEquals("example.CustomChannelizer", settings.channelizer); + settings.channelizer = WebSocketChannelizer.class.getName(); + ContextGremlinServer.configureChannelizer(settings); + Assert.assertEquals(WebSocketChannelizer.class.getName(), + settings.channelizer); + } finally { + HugeGraphAuthProxy.resetContext(); + } + } + + @Test + public void testUnknownAliasThenValidQueriesOnSameTcpConnection() + throws Exception { + Settings settings = new Settings(); + settings.host = "127.0.0.1"; + settings.port = 0; + settings.gremlinPool = 1; + settings.threadPoolBoss = 1; + settings.threadPoolWorker = 1; + settings.channelizer = WsAndHttpChannelizer.class.getName(); + Settings.SerializerSettings serializer = new Settings.SerializerSettings(); + serializer.className = GraphSONUntypedMessageSerializerV1.class.getName(); + Settings.SerializerSettings webSocketSerializer = new Settings.SerializerSettings(); + webSocketSerializer.className = GraphSONMessageSerializerV3.class.getName(); + settings.serializers = Arrays.asList(serializer, webSocketSerializer); + ContextGremlinServer server = new ContextGremlinServer( + settings, new EventHub("http-lifecycle-test", 1)); + try { + server.getServerGremlinExecutor().getGraphManager() + .putGraph("known", EmptyGraph.instance()); + server.start().get(30, TimeUnit.SECONDS); + Assert.assertTrue(server.getChannelizer() instanceof + HugeGraphWsAndHttpChannelizer); + Channel listening = Whitebox.getInternalState(server, + "serverSocketChannel"); + int port = ((InetSocketAddress) listening.localAddress()).getPort(); + try (Socket socket = new Socket("127.0.0.1", port)) { + socket.setSoTimeout(10000); + InputStream input = new BufferedInputStream(socket.getInputStream()); + send(socket, "missing", "101"); + Map rejected = read(input, 400); + Assert.assertTrue(rejected.get("message").toString() + .contains("Could not rebind")); + send(socket, "known", "202"); + Map accepted = read(input, 200); + Assert.assertEquals(Collections.singletonList(202), + ((Map) accepted.get("result")).get("data")); + send(socket, "known", "303"); + Map next = read(input, 200); + Assert.assertEquals(Collections.singletonList(303), + ((Map) next.get("result")).get("data")); + } + Cluster cluster = Cluster.build().addContactPoint("127.0.0.1") + .port(port) + .serializer(new GraphSONMessageSerializerV3()) + .create(); + try { + Client client = cluster.connect(); + try { + Assert.assertEquals(404, client.submit("404").all() + .get(10, TimeUnit.SECONDS) + .get(0).getInt()); + } finally { + client.close(); + } + } finally { + cluster.close(); + } + } finally { + server.stop().get(30, TimeUnit.SECONDS); + HugeGraphAuthProxy.resetContext(); + } + } + + private static void send(Socket socket, String alias, String script) + throws IOException { + String body = "{\"gremlin\":\"" + script + "\",\"language\":" + + "\"gremlin-groovy\",\"aliases\":{\"graph\":\"" + alias + "\"}}"; + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + String headers = "POST /gremlin HTTP/1.1\r\n" + + "Host: 127.0.0.1\r\n" + + "Content-Type: application/json\r\n" + + "Accept: application/json\r\n" + + "Connection: keep-alive\r\n" + + "Content-Length: " + bytes.length + "\r\n\r\n"; + socket.getOutputStream().write(headers.getBytes(StandardCharsets.US_ASCII)); + socket.getOutputStream().write(bytes); + socket.getOutputStream().flush(); + } + + private static Map read(InputStream input, int expectedStatus) + throws IOException { + String status = line(input); + Assert.assertTrue(status, status.startsWith("HTTP/1.1 " + expectedStatus + " ")); + int length = -1; + String header; + while (!(header = line(input)).isEmpty()) { + if (header.toLowerCase(Locale.ROOT).startsWith("content-length:")) { + length = Integer.parseInt(header.substring(header.indexOf(':') + 1).trim()); + } + } + Assert.assertTrue("Expected a bounded response body", length >= 0 && length < 65536); + byte[] bytes = input.readNBytes(length); + Assert.assertEquals(length, bytes.length); + return JsonUtil.fromJson(new String(bytes, StandardCharsets.UTF_8), Map.class); + } + + private static String line(InputStream input) throws IOException { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + int value; + while ((value = input.read()) != '\n') { + if (value < 0 || bytes.size() >= 8192) { + throw new IOException("Incomplete or excessive HTTP header"); + } + if (value != '\r') { + bytes.write(value); + } + } + return bytes.toString(StandardCharsets.US_ASCII); + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/HttpGremlinRequestHandlerTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/HttpGremlinRequestHandlerTest.java new file mode 100644 index 0000000000..dd33fe3c85 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/HttpGremlinRequestHandlerTest.java @@ -0,0 +1,284 @@ +/* + * 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.hugegraph.auth; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hugegraph.server.HttpGremlinRequestHandler; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.util.JsonUtil; +import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor; +import org.apache.tinkerpop.gremlin.server.GraphManager; +import org.apache.tinkerpop.gremlin.server.Settings; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.util.ser.GraphSONUntypedMessageSerializerV1; +import org.junit.Test; +import org.mockito.Mockito; + +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.DefaultFullHttpRequest; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.FullHttpResponse; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpHeaderValues; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpUtil; +import io.netty.handler.codec.http.HttpVersion; + +public class HttpGremlinRequestHandlerTest { + + @Test + public void testInvalidAliasRespondsOnceThenValidRequestWorks() + throws Exception { + try (GremlinExecutor executor = GremlinExecutor.build().create()) { + EmbeddedChannel channel = channel(executor); + try { + FullHttpRequest invalid = request("missing", "101"); + channel.writeInbound(invalid); + FullHttpResponse rejected = response(channel); + Assert.assertEquals(HttpResponseStatus.BAD_REQUEST, + rejected.status()); + rejected.release(); + Assert.assertEquals(0, invalid.refCnt()); + Assert.assertNull(channel.readOutbound()); + Assert.assertTrue(channel.isActive()); + + FullHttpRequest valid = request("known", "202"); + channel.writeInbound(valid); + FullHttpResponse accepted = response(channel); + try { + Assert.assertEquals(HttpResponseStatus.OK, + accepted.status()); + Map body = JsonUtil.fromJson( + accepted.content().toString(StandardCharsets.UTF_8), + Map.class); + Map result = (Map) body.get("result"); + Assert.assertEquals(Collections.singletonList(202), + result.get("data")); + } finally { + accepted.release(); + } + Assert.assertEquals(0, valid.refCnt()); + Assert.assertNull(channel.readOutbound()); + } finally { + channel.finishAndReleaseAll(); + } + } + } + + @Test + public void testEarlyErrorsReleaseRequestExactlyOnce() throws Exception { + try (GremlinExecutor executor = GremlinExecutor.build().create()) { + EmbeddedChannel channel = channel(executor); + try { + assertRejected(channel, requestBody("{"), + HttpResponseStatus.BAD_REQUEST); + assertRejected(channel, requestBody("{}"), + HttpResponseStatus.BAD_REQUEST); + FullHttpRequest method = request("known", "1"); + method.setMethod(HttpMethod.PUT); + assertRejected(channel, method, + HttpResponseStatus.METHOD_NOT_ALLOWED); + FullHttpRequest accept = request("known", "1"); + accept.headers().set(HttpHeaderNames.ACCEPT, + "application/unsupported"); + assertRejected(channel, accept, + HttpResponseStatus.BAD_REQUEST); + FullHttpRequest favicon = request("known", "1"); + favicon.setUri("/favicon.ico"); + assertRejected(channel, favicon, HttpResponseStatus.NOT_FOUND); + } finally { + channel.finishAndReleaseAll(); + } + } + } + + @Test + public void testConnectionCloseReleasesRequestOnSuccessAndFailure() + throws Exception { + try (GremlinExecutor executor = GremlinExecutor.build().create()) { + for (String alias : new String[]{"missing", "known"}) { + EmbeddedChannel channel = channel(executor); + FullHttpRequest request = request(alias, "202"); + request.headers().set(HttpHeaderNames.CONNECTION, + HttpHeaderValues.CLOSE); + try { + channel.writeInbound(request); + FullHttpResponse response = response(channel); + try { + Assert.assertEquals("known".equals(alias) ? + HttpResponseStatus.OK : + HttpResponseStatus.BAD_REQUEST, + response.status()); + } finally { + response.release(); + } + channel.runPendingTasks(); + Assert.assertFalse(channel.isActive()); + Assert.assertEquals(0, request.refCnt()); + Assert.assertNull(channel.readOutbound()); + } finally { + channel.finishAndReleaseAll(); + } + } + } + } + + @Test + public void testSharedBufferRetainsCallersReference() throws Exception { + try (GremlinExecutor executor = GremlinExecutor.build().create()) { + EmbeddedChannel channel = channel(executor); + FullHttpRequest request = request("missing", "1"); + request.retain(); + try { + channel.writeInbound(request); + FullHttpResponse rejected = response(channel); + Assert.assertEquals(HttpResponseStatus.BAD_REQUEST, + rejected.status()); + rejected.release(); + Assert.assertEquals(1, request.refCnt()); + Assert.assertNull(channel.readOutbound()); + } finally { + request.release(); + channel.finishAndReleaseAll(); + } + } + } + + @Test + public void testAsyncWriteFailureDoesNotRetainRequest() throws Exception { + CountDownLatch evaluating = new CountDownLatch(1); + CountDownLatch proceed = new CountDownLatch(1); + AtomicReference attempted = new AtomicReference<>(); + try (GremlinExecutor executor = GremlinExecutor.build() + .beforeEval(bindings -> { + evaluating.countDown(); + try { + Assert.assertTrue(proceed.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + }).create()) { + EmbeddedChannel channel = channel(executor); + channel.pipeline().addFirst("failed-write", + new ChannelOutboundHandlerAdapter() { + @Override + public void write(ChannelHandlerContext context, + Object message, + ChannelPromise promise) { + FullHttpResponse response = + (FullHttpResponse) message; + attempted.set(response); + response.release(); + promise.setFailure(new IOException("test write failure")); + } + }); + FullHttpRequest request = request("known", "202"); + try { + channel.writeInbound(request); + Assert.assertEquals(0, request.refCnt()); + Assert.assertTrue(evaluating.await(10, TimeUnit.SECONDS)); + proceed.countDown(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (attempted.get() == null && System.nanoTime() < deadline) { + channel.runPendingTasks(); + Thread.sleep(5L); + } + Assert.assertNotNull(attempted.get()); + Assert.assertEquals(0, attempted.get().refCnt()); + Assert.assertEquals(0, request.refCnt()); + Assert.assertNull(channel.readOutbound()); + } finally { + proceed.countDown(); + channel.finishAndReleaseAll(); + } + } + } + + private static void assertRejected(EmbeddedChannel channel, + FullHttpRequest request, + HttpResponseStatus expected) + throws InterruptedException { + channel.writeInbound(request); + FullHttpResponse response = response(channel); + try { + Assert.assertEquals(expected, response.status()); + } finally { + response.release(); + } + Assert.assertEquals(0, request.refCnt()); + Assert.assertNull(channel.readOutbound()); + Assert.assertTrue(channel.isActive()); + } + + private static EmbeddedChannel channel(GremlinExecutor executor) { + GraphManager manager = Mockito.mock(GraphManager.class); + Mockito.when(manager.getGraph("known")) + .thenReturn(Mockito.mock(Graph.class)); + return new EmbeddedChannel(new HttpGremlinRequestHandler( + Collections.singletonMap("application/json", + new GraphSONUntypedMessageSerializerV1()), + executor, manager, new Settings())); + } + + private static FullHttpRequest request(String alias, String script) { + String body = "{\"gremlin\":\"" + script + "\",\"language\":" + + "\"gremlin-groovy\",\"aliases\":{\"graph\":\"" + alias + "\"}}"; + return requestBody(body); + } + + private static FullHttpRequest requestBody(String body) { + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, HttpMethod.POST, "/gremlin", + Unpooled.copiedBuffer(body, StandardCharsets.UTF_8)); + request.headers().set(HttpHeaderNames.CONTENT_TYPE, + HttpHeaderValues.APPLICATION_JSON); + request.headers().set(HttpHeaderNames.ACCEPT, + HttpHeaderValues.APPLICATION_JSON); + request.headers().set(HttpHeaderNames.CONNECTION, + HttpHeaderValues.KEEP_ALIVE); + HttpUtil.setContentLength(request, request.content().readableBytes()); + return request; + } + + private static FullHttpResponse response(EmbeddedChannel channel) + throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (System.nanoTime() < deadline) { + channel.runPendingTasks(); + FullHttpResponse response = channel.readOutbound(); + if (response != null) { + return response; + } + Thread.sleep(5L); + } + throw new AssertionError("Gremlin HTTP response did not arrive"); + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessorTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessorTest.java new file mode 100644 index 0000000000..e98d4fbaf1 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessorTest.java @@ -0,0 +1,219 @@ +/* + * 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.hugegraph.opencypher; + +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hugegraph.testutil.Assert; +import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor; +import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal; +import org.apache.tinkerpop.gremlin.server.Context; +import org.apache.tinkerpop.gremlin.server.GraphManager; +import org.apache.tinkerpop.gremlin.server.Settings; +import org.apache.tinkerpop.gremlin.server.handler.StateKey; +import org.apache.tinkerpop.gremlin.server.op.AbstractEvalOpProcessor; +import org.apache.tinkerpop.gremlin.server.op.OpProcessorException; +import org.apache.tinkerpop.gremlin.util.Tokens; +import org.apache.tinkerpop.gremlin.util.message.RequestMessage; +import org.apache.tinkerpop.gremlin.util.message.ResponseMessage; +import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode; +import org.junit.Test; +import org.opencypher.gremlin.traversal.ParameterNormalizer; +import org.mockito.Mockito; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.embedded.EmbeddedChannel; + +public class CypherOpProcessorTest { + + private final CypherOpProcessor processor = new CypherOpProcessor(); + + @Test + public void testAcceptCypherParameterNamesAndNullValues() throws Exception { + Map bindings = new HashMap<>(); + bindings.put("value", null); + bindings.put("id", 1); + bindings.put("label", "person"); + Assert.assertFalse(this.processor.validateEvalMessage(request("RETURN $value", bindings)).isPresent()); + Assert.assertFalse(this.processor.validateEvalMessage(RequestMessage.build(Tokens.OPS_EVAL) + .addArg(Tokens.ARGS_GREMLIN, "RETURN 1").create()).isPresent()); + } + + @Test + public void testRejectInvalidQuery() { + assertInvalid(request(1, Collections.emptyMap())); + assertInvalid(request(" \n", Collections.emptyMap())); + } + + @Test + public void testRejectInvalidBindingShapeAndKeys() { + assertInvalid(request("RETURN 1", Collections.emptyList())); + assertInvalid(request("RETURN 1", Collections.singletonMap(1, "value"))); + assertInvalid(request("RETURN 1", Collections.singletonMap(null, "value"))); + } + + @Test + public void testRejectTranslatorNullSentinelInBindings() { + Object reserved = ParameterNormalizer.normalize(Collections.singletonMap("value", null)).get("value"); + Assert.assertEquals(" cypher.null", reserved); + assertInvalid(request("RETURN $value", Collections.singletonMap("value", reserved))); + assertInvalid(request("RETURN $value", Collections.singletonMap("value", + Collections.singleton(reserved)))); + assertInvalid(request("RETURN $value", Collections.singletonMap("value", + Collections.singletonMap("nested", reserved)))); + assertInvalid(request("RETURN $value", Collections.singletonMap("value", + Arrays.asList(null, Collections.singletonMap("nested", Collections.singletonList(reserved)))))); + } + + @Test + public void testAcceptNullAndStringsNearTranslatorSentinel() throws Exception { + Map bindings = Collections.singletonMap("value", + Arrays.asList(null, "cypher.null", " cypher.null", " cypher.null ", + Collections.singletonMap("nested", null))); + Assert.assertFalse(this.processor.validateEvalMessage(request("RETURN $value", bindings)).isPresent()); + } + + @Test + public void testEnforceParameterCountBoundary() throws Exception { + Map bindings = new HashMap<>(); + for (int i = 0; i < AbstractEvalOpProcessor.DEFAULT_MAX_PARAMETERS; i++) { + bindings.put("parameter" + i, i); + } + Assert.assertFalse(this.processor.validateEvalMessage(request("RETURN 1", bindings)).isPresent()); + bindings.put("oneTooMany", 1); + assertInvalid(request("RETURN 1", bindings)); + } + + @Test + public void testRollbackOnTraversalError() throws Exception { + assertRollbackOnTraversalError(false); + } + + @Test + public void testRollbackFailureIsSuppressedOnTraversalError() throws Exception { + assertRollbackOnTraversalError(true); + } + + private void assertRollbackOnTraversalError(boolean failRollback) throws Exception { + Context context = Mockito.mock(Context.class); + GraphManager manager = Mockito.mock(GraphManager.class); + GremlinExecutor gremlinExecutor = Mockito.mock(GremlinExecutor.class); + AtomicReference> submitted = new AtomicReference<>(); + ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, + new LinkedBlockingQueue<>()) { + @Override + public void execute(Runnable command) { + submitted.set((FutureTask) command); + super.execute(command); + } + }; + ChannelHandlerContext channelContext = Mockito.mock(ChannelHandlerContext.class); + EmbeddedChannel channel = new EmbeddedChannel(); + channel.attr(StateKey.USE_BINARY).set(false); + Settings settings = new Settings(); + settings.evaluationTimeout = 0; + settings.strictTransactionManagement = false; + Mockito.when(context.getRequestMessage()).thenReturn(request("RETURN 1", Collections.emptyMap())); + Mockito.when(context.getSettings()).thenReturn(settings); + Mockito.when(context.getGraphManager()).thenReturn(manager); + Mockito.when(context.getGremlinExecutor()).thenReturn(gremlinExecutor); + Mockito.when(gremlinExecutor.getExecutorService()).thenReturn(executor); + Mockito.when(context.getChannelHandlerContext()).thenReturn(channelContext); + Mockito.when(channelContext.channel()).thenReturn(channel); + + AssertionError failure = new AssertionError("fatal traversal failure"); + IllegalStateException rollbackFailure = new IllegalStateException("rollback failure"); + CountDownLatch responded = new CountDownLatch(1); + AtomicReference response = new AtomicReference<>(); + Mockito.doAnswer(invocation -> { + response.set(invocation.getArgument(0)); + responded.countDown(); + return null; + }).when(context).writeAndFlush(Mockito.any(ResponseMessage.class)); + AtomicReference iterationThread = new AtomicReference<>(); + AtomicReference rollbackThread = new AtomicReference<>(); + Iterator iterator = Mockito.mock(Iterator.class); + Mockito.when(iterator.hasNext()).thenAnswer(invocation -> { + iterationThread.set(Thread.currentThread()); + throw failure; + }); + Mockito.doAnswer(invocation -> { + rollbackThread.set(Thread.currentThread()); + if (failRollback) { + throw rollbackFailure; + } + return null; + }).when(manager).rollbackAll(); + GraphTraversal source = Mockito.mock(GraphTraversal.class); + Method handleTraversal = CypherOpProcessor.class.getDeclaredMethod( + "handleTraversal", Context.class, Iterator.class, GraphTraversal.class); + handleTraversal.setAccessible(true); + try { + handleTraversal.invoke(this.processor, context, iterator, source); + Assert.assertTrue("Fatal errors must send a terminal response", + responded.await(5, TimeUnit.SECONDS)); + FutureTask task = submitted.get(); + ExecutionException result = Assert.assertThrows(ExecutionException.class, + () -> task.get(5, TimeUnit.SECONDS)); + Assert.assertSame(failure, result.getCause()); + Assert.assertNotNull(iterationThread.get()); + Assert.assertNotSame(Thread.currentThread(), iterationThread.get()); + Assert.assertSame(iterationThread.get(), rollbackThread.get()); + Mockito.verify(manager).rollbackAll(); + Mockito.verify(manager, Mockito.never()).commitAll(); + Mockito.verify(source).close(); + Mockito.verify(context).writeAndFlush(Mockito.any(ResponseMessage.class)); + Assert.assertEquals(ResponseStatusCode.SERVER_ERROR, response.get().getStatus().getCode()); + Assert.assertEquals(failure.getMessage(), response.get().getStatus().getMessage()); + Assert.assertNull(response.get().getResult().getData()); + Assert.assertEquals(failRollback ? 1 : 0, failure.getSuppressed().length); + if (failRollback) { + Assert.assertSame(rollbackFailure, failure.getSuppressed()[0]); + } + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + channel.finishAndReleaseAll(); + } + } + + private void assertInvalid(RequestMessage message) { + OpProcessorException error = Assert.assertThrows(OpProcessorException.class, + () -> this.processor.validateEvalMessage(message)); + Assert.assertEquals(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS, + error.getResponseMessage().getStatus().getCode()); + } + + private static RequestMessage request(Object query, Object bindings) { + return RequestMessage.build(Tokens.OPS_EVAL) + .addArg(Tokens.ARGS_GREMLIN, query) + .addArg(Tokens.ARGS_BINDINGS, bindings).create(); + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java index 617440b668..b834d19fae 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java @@ -20,6 +20,8 @@ import org.apache.hugegraph.api.auth.GraphSpaceAuthPayloadTest; import org.apache.hugegraph.api.auth.GraphSpaceGroupAPITest; import org.apache.hugegraph.api.cypher.CypherClientTest; +import org.apache.hugegraph.auth.ContextGremlinServerHttpTest; +import org.apache.hugegraph.auth.HttpGremlinRequestHandlerTest; import org.apache.hugegraph.auth.StandardAuthManagerV2Test; import org.apache.hugegraph.auth.WsAndHttpBasicAuthHandlerTest; import org.apache.hugegraph.backend.page.QueryListTest; @@ -28,6 +30,7 @@ import org.apache.hugegraph.meta.EtcdMetaDriverTest; import org.apache.hugegraph.meta.MetaManagerSchemaCacheClearEventTest; import org.apache.hugegraph.meta.managers.AuthMetaManagerTest; +import org.apache.hugegraph.opencypher.CypherOpProcessorTest; import org.apache.hugegraph.store.client.OrderedScanSecurityTest; import org.apache.hugegraph.tinkerpop.HugeGraphTestInfrastructureTest; import org.apache.hugegraph.traversal.optimize.TraversalUtilOptimizeTest; @@ -130,7 +133,10 @@ GremlinQueryAPITest.class, CypherClientTest.class, CypherGremlinPredicatesTest.class, + CypherOpProcessorTest.class, WsAndHttpBasicAuthHandlerTest.class, + HttpGremlinRequestHandlerTest.class, + ContextGremlinServerHttpTest.class, GraphSpaceGroupAPITest.class, GraphSpaceAuthPayloadTest.class, StandardAuthManagerV2Test.class,