diff --git a/docs/changelog.md b/docs/changelog.md index 7b2f4d6616..33974a984e 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -314,6 +314,20 @@ mandatory and cannot be disabled. The default ExecutorService core pool size is the default value is considered to be optimal unless users want to artificially limit parallelism of CQL results deserialization jobs. +##### A new multi-query method added into `KeyColumnValueStore` (affects storage adapters implementations only) + +A new method `Map> getMultiSlices(MultiKeysQueryGroups multiKeysQueryGroups, StoreTransaction txh)` +is added into `KeyColumnValueStore` which is now preferred for multi-queries (batch queries) with multiple slice queries. +In case a multi-query executes more than one Slice query per multi-query execution then those Slice queries will be +grouped per same key sets and the new `getMultiSlices` query will be executed with groups of different `SliceQuery` for the same sets of keys. + +Notice, if the storage doesn't have multiQuery feature enabled - the method won't be used. Hence, it's not necessary to implement it. + +In case storage backend has multiQuery feature enabled, then it is highly recommended to overwrite the default (non-optimized) implementation +and optimize this method execution to execute all the slice queries for all the requested keys in the shortest time possible +(for example, by using asynchronous programming, slice queries grouping, multi-threaded execution, or any other technique which +is efficient for the respective storage adapter). + ##### Removal of deprecated classes/methods/functionalities ###### Methods diff --git a/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphBaseTest.java b/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphBaseTest.java index b751da7130..efff716143 100644 --- a/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphBaseTest.java +++ b/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphBaseTest.java @@ -616,7 +616,7 @@ public static void evaluateQuery(JanusGraphVertexQuery query, RelationCategory r int subQueryCounter = 0; for (SimpleQueryProfiler subProfiler : profiler) { assertNotNull(subProfiler); - if (subProfiler.getGroupName().equals(QueryProfiler.OPTIMIZATION)) continue; + if (subProfiler.getGroupName().equals(QueryProfiler.OPTIMIZATION) || Boolean.TRUE.equals(subProfiler.getAnnotation(QueryProfiler.MULTI_SLICES_ANNOTATION))) continue; if (subQuerySpecs.length == 2) { //0=>fitted, 1=>ordered assertEquals(subQuerySpecs[0], subProfiler.getAnnotation(QueryProfiler.FITTED_ANNOTATION)); assertEquals(subQuerySpecs[1], subProfiler.getAnnotation(QueryProfiler.ORDERED_ANNOTATION)); diff --git a/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphTest.java b/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphTest.java index e0905bc074..e727e8fc4c 100644 --- a/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphTest.java +++ b/janusgraph-backend-testutils/src/main/java/org/janusgraph/graphdb/JanusGraphTest.java @@ -197,7 +197,6 @@ import static org.apache.tinkerpop.gremlin.process.traversal.Order.asc; import static org.apache.tinkerpop.gremlin.process.traversal.Order.desc; -import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.has; import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.hasNot; import static org.apache.tinkerpop.gremlin.structure.Direction.BOTH; import static org.apache.tinkerpop.gremlin.structure.Direction.IN; @@ -5817,8 +5816,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName()); // `foo` and `fooBar` properties are prefetched separately - assertEquals(6, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(2, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 2, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 2, profile.getMetrics())); // test batching for 3 `has()` steps of required properties only profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar").has("barFoo", "Foo"), @@ -5826,8 +5825,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName()); // `foo`, `fooBar`, and `barFoo` properties are prefetched separately - assertEquals(9, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(3, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 3, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 3, profile.getMetrics())); // test `has()` step `REQUIRED_AND_NEXT_PROPERTIES` mode with not following any properties steps profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar"), @@ -5835,8 +5834,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_AND_NEXT_PROPERTIES.getConfigName()); // `foo` and `fooBar` properties are prefetched separately - assertEquals(6, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(2, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 2, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 2, profile.getMetrics())); // test `has()` step `REQUIRED_AND_NEXT_PROPERTIES` mode with following valueMap step profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar").valueMap("barFoo"), @@ -5844,8 +5843,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_AND_NEXT_PROPERTIES.getConfigName()); // `foo`, `fooBar`, and `barFoo` properties are prefetched separately - assertEquals(9, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(3, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 3, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 3, profile.getMetrics())); // test `has()` step `REQUIRED_AND_NEXT_PROPERTIES` mode with following properties step profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar").properties("barFoo"), @@ -5853,8 +5852,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_AND_NEXT_PROPERTIES.getConfigName()); // `foo`, `fooBar`, and `barFoo` properties are prefetched separately - assertEquals(9, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(3, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 3, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 3, profile.getMetrics())); // test `has()` step `REQUIRED_AND_NEXT_PROPERTIES` mode with following values step profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar").values("barFoo"), @@ -5862,8 +5861,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_AND_NEXT_PROPERTIES.getConfigName()); // `foo`, `fooBar`, and `barFoo` properties are prefetched separately - assertEquals(9, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(3, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 3, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 3, profile.getMetrics())); // test `has()` step `REQUIRED_AND_NEXT_PROPERTIES` mode with following elementMap step profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar").elementMap("barFoo"), @@ -5871,8 +5870,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_AND_NEXT_PROPERTIES.getConfigName()); // `foo`, `fooBar`, and `barFoo` properties are prefetched separately - assertEquals(9, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(3, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 3, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 3, profile.getMetrics())); // test `has()` step `REQUIRED_AND_NEXT_PROPERTIES` mode with following valueMap step (all properties) profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar").valueMap(), @@ -5974,8 +5973,8 @@ private void testLimitBatchSizeForHasStep(int numV, int barrierSize, int limit, option(PROPERTY_PREFETCHING), false, option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_AND_NEXT_PROPERTIES_OR_ALL.getConfigName()); // `foo`, `fooBar`, and `barFoo` properties are prefetched separately - assertEquals(9, countBackendQueriesOfSize(barrierSize, profile.getMetrics())); - assertEquals(3, countBackendQueriesOfSize(numV - 3 * barrierSize, profile.getMetrics())); + assertEquals(3, countBackendQueriesOfSize(barrierSize * 3, profile.getMetrics())); + assertEquals(1, countBackendQueriesOfSize((numV - 3 * barrierSize) * 3, profile.getMetrics())); // test `has()` step `REQUIRED_AND_NEXT_PROPERTIES_OR_ALL` mode with not following values step profile = testLimitedBatch(() -> graph.traversal().V(cs).barrier(barrierSize).has("foo", "bar").has("fooBar", "Bar") @@ -6093,6 +6092,70 @@ private void testLimitBatchSizeForMultiQueryOfConnectiveSteps(JanusGraphVertex[] assertEquals((int) Math.ceil((double) Math.min(bs.length, limit) / barrierSize), countOptimizationQueries(profile.getMetrics())); } + @Test + public void testMultiSliceDBCachedRequests(){ + clopen(option(DB_CACHE), false); + mgmt.makeVertexLabel("testVertex").make(); + finishSchema(); + int numV = 100; + JanusGraphVertex a = graph.addVertex(); + JanusGraphVertex[] bs = new JanusGraphVertex[numV]; + JanusGraphVertex[] cs = new JanusGraphVertex[numV]; + for (int i = 0; i < numV; ++i) { + bs[i] = graph.addVertex(); + cs[i] = graph.addVertex("testVertex"); + cs[i].property("foo", "bar"); + cs[i].property("fooBar", "Bar"); + cs[i].property("barFoo", "Foo"); + a.addEdge("knows", bs[i]); + bs[i].addEdge("knows", cs[i]); + } + + clopen(option(USE_MULTIQUERY), true, option(LIMITED_BATCH), true, + option(DB_CACHE), true, option(DB_CACHE_TIME), 1000000L, + option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName(), + option(PROPERTIES_BATCH_MODE), MultiQueryPropertiesStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName()); + assertEquals(numV, graph.traversal().V(cs).values("foo").toList().size()); + graph.tx().rollback(); + assertEquals(numV, graph.traversal().V(cs).values("foo").toList().size()); + graph.tx().rollback(); + assertEquals(numV * 2, graph.traversal().V(cs).values("foo", "fooBar").toList().size()); + graph.tx().rollback(); + assertEquals(numV * 2, graph.traversal().V(cs).values("foo", "fooBar").toList().size()); + graph.tx().rollback(); + assertEquals(numV * 3, graph.traversal().V(cs).values("foo", "barFoo", "fooBar").toList().size()); + graph.tx().rollback(); + assertEquals(numV * 3, graph.traversal().V(cs).values("foo", "barFoo", "fooBar").toList().size()); + + clopen(option(USE_MULTIQUERY), true, option(LIMITED_BATCH), true, + option(DB_CACHE), true, option(DB_CACHE_TIME), 1000000L, + option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName(), + option(PROPERTIES_BATCH_MODE), MultiQueryPropertiesStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName()); + assertEquals(numV, graph.traversal().V(cs).values("foo").toList().size()); + graph.tx().rollback(); + assertEquals(numV * 3, graph.traversal().V(cs).values("foo", "barFoo", "fooBar").toList().size()); + graph.tx().rollback(); + + clopen(option(USE_MULTIQUERY), true, option(LIMITED_BATCH), true, + option(DB_CACHE), true, option(DB_CACHE_TIME), 1000000L, + option(HAS_STEP_BATCH_MODE), MultiQueryHasStepStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName(), + option(PROPERTIES_BATCH_MODE), MultiQueryPropertiesStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName()); + assertEquals(3, graph.traversal().V(cs[1], cs[5], cs[10]).values("foo").toList().size()); + graph.tx().rollback(); + assertEquals(3, graph.traversal().V(cs[1], cs[7], cs[10]).values("foo").toList().size()); + graph.tx().rollback(); + assertEquals(6, graph.traversal().V(cs[1], cs[6], cs[8]).values("foo", "fooBar").toList().size()); + graph.tx().rollback(); + assertEquals(6, graph.traversal().V(cs[1], cs[7], cs[10]).values("foo", "fooBar").toList().size()); + graph.tx().rollback(); + assertEquals(9, graph.traversal().V(cs[1], cs[5], cs[10]).values("foo", "barFoo", "fooBar").toList().size()); + graph.tx().rollback(); + assertEquals(numV * 3, graph.traversal().V(cs).values("foo", "barFoo", "fooBar").toList().size()); + graph.tx().rollback(); + assertEquals(numV * 2, graph.traversal().V(cs).values("foo", "fooBar").toList().size()); + assertEquals(numV * 3, graph.traversal().V().values("foo", "barFoo", "fooBar").toList().size()); + } + private TraversalMetrics testLimitedBatch(Supplier> traversal, Object... settings){ assertEqualResultWithAndWithoutLimitBatchSize(traversal); if(settings.length == 0){ @@ -7791,7 +7854,7 @@ public void testVertexCentricQueryProfiling() { }}; mMultiLabels.getAnnotations().remove("percentDur"); assertEquals(annotations, mMultiLabels.getAnnotations()); - assertEquals(3, mMultiLabels.getNested().size()); + assertEquals(4, mMultiLabels.getNested().size()); Metrics friendMetrics = (Metrics) mMultiLabels.getNested().toArray()[1]; assertEquals("OR-query", friendMetrics.getName()); // FIXME: assertTrue(friendMetrics.getDuration(TimeUnit.MICROSECONDS) > 0); @@ -7810,6 +7873,14 @@ public void testVertexCentricQueryProfiling() { put("query", "friend-no-index:SliceQuery[0x7180,0x7181)"); // no vertex-centric index found }}; assertEquals(annotations, friendNoIndexMetrics.getAnnotations()); + Metrics backendQueryMetrics = (Metrics) mMultiLabels.getNested().toArray()[3]; + assertEquals("backend-query", backendQueryMetrics.getName()); + Map annotatnions = backendQueryMetrics.getAnnotations(); + assertEquals(3, annotatnions.size()); + assertEquals("true", annotatnions.get("multiSlices")); + assertEquals(2, annotatnions.get("queriesAmount")); + assertTrue(annotatnions.get("queries").equals("[2069:byTime:SliceQuery[0xB0E0FF7FFFFF9B,0xB0E0FF7FFFFF9C), friend-no-index:SliceQuery[0x7180,0x7181)]") || + annotatnions.get("queries").equals("[friend-no-index:SliceQuery[0x7180,0x7181), 2069:byTime:SliceQuery[0xB0E0FF7FFFFF9B,0xB0E0FF7FFFFF9C)]")); } @Test diff --git a/janusgraph-benchmark/src/main/java/org/janusgraph/CQLMultiQueryMultiSlicesBenchmark.java b/janusgraph-benchmark/src/main/java/org/janusgraph/CQLMultiQueryMultiSlicesBenchmark.java new file mode 100644 index 0000000000..1d52d77af4 --- /dev/null +++ b/janusgraph-benchmark/src/main/java/org/janusgraph/CQLMultiQueryMultiSlicesBenchmark.java @@ -0,0 +1,188 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph; + +import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal; +import org.apache.tinkerpop.gremlin.structure.Vertex; +import org.janusgraph.core.JanusGraph; +import org.janusgraph.core.JanusGraphFactory; +import org.janusgraph.core.JanusGraphTransaction; +import org.janusgraph.core.PropertyKey; +import org.janusgraph.core.schema.JanusGraphManagement; +import org.janusgraph.diskstorage.BackendException; +import org.janusgraph.diskstorage.configuration.ModifiableConfiguration; +import org.janusgraph.diskstorage.configuration.WriteConfiguration; +import org.janusgraph.diskstorage.cql.CQLConfigOptions; +import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration; +import org.janusgraph.graphdb.tinkerpop.optimize.strategy.MultiQueryPropertiesStrategyMode; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; + +@BenchmarkMode(Mode.AverageTime) +@Fork(1) +@State(Scope.Benchmark) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +public class CQLMultiQueryMultiSlicesBenchmark { + @Param({"5000", "50000"}) + int verticesAmount; + + JanusGraph graph; + + public WriteConfiguration getConfiguration() { + ModifiableConfiguration config = GraphDatabaseConfiguration.buildGraphConfiguration(); + config.set(GraphDatabaseConfiguration.STORAGE_BACKEND,"cql"); + config.set(CQLConfigOptions.LOCAL_DATACENTER, "dc1"); + config.set(GraphDatabaseConfiguration.USE_MULTIQUERY, true); + config.set(GraphDatabaseConfiguration.PROPERTIES_BATCH_MODE, MultiQueryPropertiesStrategyMode.REQUIRED_PROPERTIES_ONLY.getConfigName()); + config.set(GraphDatabaseConfiguration.PROPERTY_PREFETCHING, false); + return config.getConfiguration(); + } + + @Setup + public void setUp() throws Exception { + graph = JanusGraphFactory.open(getConfiguration()); + int propertiesAmount = 10; + + JanusGraphManagement mgmt = graph.openManagement(); + PropertyKey name = mgmt.makePropertyKey("name").dataType(String.class).make(); + + for(int i=0;i getValuesMultiplePropertiesWithAllMultiQuerySlicesUnderMaxRequestsPerConnection() { + JanusGraphTransaction tx = graph.buildTransaction() + .start(); + List result = toVerticesTraversal(tx) + .barrier(CQLConfigOptions.MAX_REQUESTS_PER_CONNECTION.getDefaultValue() / 10 - 2) + .values("prop0", "prop1", "prop2", "prop3", "prop4", "prop5", "prop6", "prop7", "prop8", "prop9") + .toList(); + tx.rollback(); + return result; + } + + @Benchmark + public List getValuesMultiplePropertiesWithUnlimitedBatch() { + JanusGraphTransaction tx = graph.buildTransaction() + .start(); + List result = toVerticesTraversal(tx) + .barrier(Integer.MAX_VALUE) + .values("prop0", "prop1", "prop2", "prop3", "prop4", "prop5", "prop6", "prop7", "prop8", "prop9") + .toList(); + tx.rollback(); + return result; + } + + @Benchmark + public List getValuesMultiplePropertiesWithSmallBatch() { + JanusGraphTransaction tx = graph.buildTransaction() + .start(); + List result = toVerticesTraversal(tx) + .barrier(10) + .values("prop0", "prop1", "prop2", "prop3", "prop4", "prop5", "prop6", "prop7", "prop8", "prop9") + .toList(); + tx.rollback(); + return result; + } + + @Benchmark + public List getValuesThreePropertiesWithAllMultiQuerySlicesUnderMaxRequestsPerConnection() { + JanusGraphTransaction tx = graph.buildTransaction() + .start(); + List result = toVerticesTraversal(tx) + .barrier(100) + .values("prop1", "prop3", "prop8") + .toList(); + tx.rollback(); + return result; + } + + @Benchmark + public List getValuesAllPropertiesWithAllMultiQuerySlicesUnderMaxRequestsPerConnection() { + JanusGraphTransaction tx = graph.buildTransaction() + .start(); + List result = toVerticesTraversal(tx) + .barrier(CQLConfigOptions.MAX_REQUESTS_PER_CONNECTION.getDefaultValue() / 10 - 2) + .values() + .toList(); + tx.rollback(); + return result; + } + + @Benchmark + public List getValuesAllPropertiesWithUnlimitedBatch() { + JanusGraphTransaction tx = graph.buildTransaction() + .start(); + List result = toVerticesTraversal(tx) + .barrier(Integer.MAX_VALUE) + .values() + .toList(); + tx.rollback(); + return result; + } + + @Benchmark + public List vertexCentricPropertiesFetching() { + JanusGraphTransaction tx = graph.buildTransaction() + .start(); + List vertices = toVerticesTraversal(tx) + .toList(); + List result = new ArrayList<>(vertices.size() * 10); + for(Vertex vertex : vertices){ + vertex.properties("prop0", "prop1", "prop2", "prop3", "prop4", "prop5", "prop6", "prop7", "prop8", "prop9") + .forEachRemaining(property -> result.add(property.value())); + } + tx.rollback(); + return result; + } + + private GraphTraversal toVerticesTraversal(JanusGraphTransaction tx){ + return tx.traversal().V() + .has("name", "testVertex"); + } +} diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/BackendTransaction.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/BackendTransaction.java index 9041388b4d..2e9bfa7ee4 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/BackendTransaction.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/BackendTransaction.java @@ -24,6 +24,8 @@ import org.janusgraph.diskstorage.keycolumnvalue.KeyIterator; import org.janusgraph.diskstorage.keycolumnvalue.KeyRangeQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; +import org.janusgraph.diskstorage.keycolumnvalue.KeysQueriesGroup; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.StoreFeatures; import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; @@ -308,45 +310,83 @@ public String toString() { } }); } else { - final Map results = new HashMap<>(keys.size()); - if (threadPool == null || keys.size() < MIN_TASKS_TO_PARALLELIZE) { - for (StaticBuffer key : keys) { - results.put(key,edgeStoreQuery(new KeySliceQuery(key, query))); - } - } else { - final CountDownLatch doneSignal = new CountDownLatch(keys.size()); - final AtomicInteger failureCount = new AtomicInteger(0); - EntryList[] resultArray = new EntryList[keys.size()]; - for (int i = 0; i < keys.size(); i++) { - final int pos = i; - threadPool.execute(() -> { - try { - resultArray[pos] = edgeStoreQuery(new KeySliceQuery(keys.get(pos), query)); - } catch (Exception e) { - failureCount.incrementAndGet(); - log.warn("Individual query in multi-transaction failed: ", e); - } finally { - doneSignal.countDown(); - } - }); - } - try { - doneSignal.await(); - } catch (InterruptedException e) { - throw new JanusGraphException("Interrupted while waiting for multi-query to complete", e); + return multiThreadedEdgeStoreMultiQuery(keys, query); + } + } + + public Map> edgeStoreMultiQuery(final MultiKeysQueryGroups multiSliceQueriesWithKeys) { + if (storeFeatures.hasMultiQuery()) { + return executeRead(new Callable>>() { + @Override + public Map> call() throws Exception { + return cacheEnabled?edgeStore.getMultiSlices(multiSliceQueriesWithKeys, storeTx): + edgeStore.getMultiSlicesNoCache(multiSliceQueriesWithKeys, storeTx); } - if (failureCount.get() > 0) { - throw new JanusGraphException("Could not successfully complete multi-query. " + failureCount.get() + " individual queries failed."); + + @Override + public String toString() { + return "MultiEdgeStoreQuery"; } - for (int i=0;i> results = new HashMap<>(); + for(KeysQueriesGroup queriesForKeysPair : multiSliceQueriesWithKeys.getQueryGroups()){ + List keys = queriesForKeysPair.getKeysGroup(); + for(SliceQuery query : queriesForKeysPair.getQueries()){ + Map sliceResult = multiThreadedEdgeStoreMultiQuery(keys, query); + results.put(query, sliceResult); } } return results; } } + /** + * This method uses `threadPool` to execute a slice query for each key in parallel. + * Notice, however, that it is recommended to send all these keys in balk for execution directly to the storage + * implementation if the storage has multiQuery (`storeFeatures.hasMultiQuery()`) support because many + * storage backends can leverage optimized execution for multiple keys (asynchronous or grouped execution) and + * thus, don't require a separate thread for each SliceQuery + key pair execution. + */ + private Map multiThreadedEdgeStoreMultiQuery(final List keys, final SliceQuery query){ + final Map results = new HashMap<>(keys.size()); + if (threadPool == null || keys.size() < MIN_TASKS_TO_PARALLELIZE) { + for (StaticBuffer key : keys) { + results.put(key,edgeStoreQuery(new KeySliceQuery(key, query))); + } + } else { + final CountDownLatch doneSignal = new CountDownLatch(keys.size()); + final AtomicInteger failureCount = new AtomicInteger(0); + EntryList[] resultArray = new EntryList[keys.size()]; + for (int i = 0; i < keys.size(); i++) { + final int pos = i; + threadPool.execute(() -> { + try { + resultArray[pos] = edgeStoreQuery(new KeySliceQuery(keys.get(pos), query)); + } catch (Exception e) { + failureCount.incrementAndGet(); + log.warn("Individual query in multi-transaction failed: ", e); + } finally { + doneSignal.countDown(); + } + }); + } + try { + doneSignal.await(); + } catch (InterruptedException e) { + throw new JanusGraphException("Interrupted while waiting for multi-query to complete", e); + } + if (failureCount.get() > 0) { + throw new JanusGraphException("Could not successfully complete multi-query. " + failureCount.get() + " individual queries failed."); + } + for (int i=0;i getSlice(List keys, SliceQuery query, StoreTransaction txh) throws BackendException { return store.getSlice(keys, query, unwrapTx(txh)); } + + @Override + public Map> getMultiSlices(MultiKeysQueryGroups multiKeysQueryGroups, StoreTransaction txh) throws BackendException { + return store.getMultiSlices(multiKeysQueryGroups, unwrapTx(txh)); + } } diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KCVSUtil.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KCVSUtil.java index 356ab6bb75..dd6c6a4e67 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KCVSUtil.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KCVSUtil.java @@ -22,6 +22,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -106,7 +107,7 @@ public static boolean matches(KeyRangeQuery query, StaticBuffer key, StaticBuffe } - public static Map emptyResults(List keys) { + public static Map emptyResults(Collection keys) { final Map result = new HashMap<>(keys.size()); for (StaticBuffer key : keys) { result.put(key,EntryList.EMPTY_LIST); diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyColumnValueStore.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyColumnValueStore.java index 362124e842..2ab88e02fc 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyColumnValueStore.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyColumnValueStore.java @@ -20,6 +20,7 @@ import org.janusgraph.diskstorage.StaticBuffer; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -56,7 +57,7 @@ public interface KeyColumnValueStore { * Retrieves the list of entries (i.e. column-value pairs) as specified by the given {@link SliceQuery} for all * of the given keys together. * - * @param keys List of keys + * @param keys Collection of keys * @param query Slicequery specifying matching entries * @param txh Transaction * @return The result of the query for each of the given keys as a map from the key to the list of result entries. @@ -64,6 +65,41 @@ public interface KeyColumnValueStore { */ Map getSlice(List keys, SliceQuery query, StoreTransaction txh) throws BackendException; + /** + * Retrieves the list of entries (i.e. column-value pairs) for each provided {@link SliceQuery} for all respective given keys together. + *
+ * The default implementation of this method is not optimized and instead uses blocking calls for each separate + * {@link SliceQuery} via {@link #getSlice(List, SliceQuery, StoreTransaction)}. + *
+ * It is highly advisable to overwrite this implementation and use optimized implementation which requests all Slice queries in parallel + * by using asynchronous slice queries evaluation / grouped slice queries evaluation / parallel slice queries evaluation using a thread pool or any other + * optimized queries evaluation which can retrieve results for all requested queries ({@link SliceQuery}) in the + * shortest time possible. + *
+ * Backend implementations which are not using blocking IO calls to the underlying storage backend (i.e. `in-memory` storage implementation) + * don't need to overwrite or optimized this method because both parallelized implementation and a sequential implementation + * will be performed the same for such storage implementations. + * + * @param multiKeysQueryGroups List of Tuples where key is a List of {@link SliceQuery} queries which has + * to be performed and a value is a list of keys for which all the queries have + * to be performed. + * @param txh Transaction + * @return The result of the query for each of the given {@link SliceQuery} and each of the given key. + * {@link EntryList} result should be provided for each of the {@link StaticBuffer} of each of the {@link SliceQuery}. + * @throws org.janusgraph.diskstorage.BackendException + */ + default Map> getMultiSlices(MultiKeysQueryGroups multiKeysQueryGroups, StoreTransaction txh) throws BackendException{ + Map> result = new HashMap<>(); + for(KeysQueriesGroup queriesForKeysPair : multiKeysQueryGroups.getQueryGroups()){ + List keys = queriesForKeysPair.getKeysGroup(); + for(SliceQuery query : queriesForKeysPair.getQueries()){ + Map sliceResult = getSlice(keys, query, txh); + result.put(query, sliceResult); + } + } + return result; + } + /** * Verifies acquisition of locks {@code txh} from previous calls to * {@link #acquireLock(StaticBuffer, StaticBuffer, StaticBuffer, StoreTransaction)} diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyMultiQuery.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyMultiQuery.java new file mode 100644 index 0000000000..8fb3917de2 --- /dev/null +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyMultiQuery.java @@ -0,0 +1,56 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.diskstorage.keycolumnvalue; + +import com.google.common.collect.Iterables; +import org.janusgraph.graphdb.query.Query; + +import java.util.List; + +/** + * Represents a list of queries which should be performed for a specified key. + * + * @param key + */ +public class KeyMultiQuery { + + private final K key; + + private final List columnQueries; + + private final List columnSliceQueries; + + public KeyMultiQuery(K key, List columnQueries, List columnSliceQueries) { + this.key = key; + this.columnQueries = columnQueries; + this.columnSliceQueries = columnSliceQueries; + } + + public K getKey() { + return key; + } + + public List getColumnQueries() { + return columnQueries; + } + + public List getColumnSliceQueries() { + return columnSliceQueries; + } + + public Iterable getAllQueries() { + return Iterables.concat(columnQueries, columnSliceQueries); + } +} diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyRangeQuery.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyRangeQuery.java index 00191fd2cc..dfca0f04b1 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyRangeQuery.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeyRangeQuery.java @@ -80,8 +80,7 @@ public int hashCode() { @Override public boolean equals(Object other) { if (this==other) return true; - else if (other==null) return false; - else if (!getClass().isInstance(other)) return false; + if (!(other instanceof KeyRangeQuery)) return false; KeyRangeQuery oth = (KeyRangeQuery)other; return keyStart.equals(oth.keyStart) && keyEnd.equals(oth.keyEnd) && super.equals(oth); } diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeySliceQuery.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeySliceQuery.java index 7fcca8209c..692edb8e1b 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeySliceQuery.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeySliceQuery.java @@ -66,8 +66,7 @@ public int hashCode() { @Override public boolean equals(Object other) { if (this==other) return true; - else if (other==null) return false; - else if (!getClass().isInstance(other)) return false; + if (!(other instanceof KeySliceQuery)) return false; KeySliceQuery oth = (KeySliceQuery)other; return key.equals(oth.key) && super.equals(oth); } diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeysQueriesGroup.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeysQueriesGroup.java new file mode 100644 index 0000000000..ade1c5205d --- /dev/null +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/KeysQueriesGroup.java @@ -0,0 +1,43 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.diskstorage.keycolumnvalue; + +import org.janusgraph.graphdb.query.Query; + +import java.util.List; + +public class KeysQueriesGroup { + + private final List keysGroup; + + private List queries; + + public KeysQueriesGroup(List keysGroup, List queries) { + this.keysGroup = keysGroup; + this.queries = queries; + } + + public List getKeysGroup() { + return keysGroup; + } + + public List getQueries() { + return queries; + } + + public void setQueries(List queries) { + this.queries = queries; + } +} diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiKeyMultiQuery.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiKeyMultiQuery.java new file mode 100644 index 0000000000..63e08434a6 --- /dev/null +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiKeyMultiQuery.java @@ -0,0 +1,35 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.diskstorage.keycolumnvalue; + +import java.util.List; + +/** + * Represents a list of keys and a list of queries for each of the key which should be performed for a specified key. + * + * @param key + */ +public class MultiKeyMultiQuery { + + private final List> keyMultiQueries; + + public MultiKeyMultiQuery(List> keyMultiQueries) { + this.keyMultiQueries = keyMultiQueries; + } + + public List> getKeyMultiQueries() { + return keyMultiQueries; + } +} diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiKeysQueryGroups.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiKeysQueryGroups.java new file mode 100644 index 0000000000..9c4bde8f13 --- /dev/null +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiKeysQueryGroups.java @@ -0,0 +1,43 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.diskstorage.keycolumnvalue; + +import org.janusgraph.graphdb.query.Query; + +import java.util.List; + +public class MultiKeysQueryGroups { + + private List> queryGroups; + + private final MultiQueriesByKeysGroupsContext multiQueryContext; + + public MultiKeysQueryGroups(List> queryGroups, MultiQueriesByKeysGroupsContext multiQueryContext) { + this.queryGroups = queryGroups; + this.multiQueryContext = multiQueryContext; + } + + public List> getQueryGroups() { + return queryGroups; + } + + public MultiQueriesByKeysGroupsContext getMultiQueryContext() { + return multiQueryContext; + } + + public void setQueryGroups(List> queryGroups) { + this.queryGroups = queryGroups; + } +} diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiQueriesByKeysGroupsContext.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiQueriesByKeysGroupsContext.java new file mode 100644 index 0000000000..27bb074333 --- /dev/null +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/MultiQueriesByKeysGroupsContext.java @@ -0,0 +1,77 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.diskstorage.keycolumnvalue; + +import org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil; + +import java.util.List; + +public class MultiQueriesByKeysGroupsContext { + + /** + * This is a list of all unique keys which could be used in this multi-query. + * It doesn't mean that all these keys should be performed for each query. + * This array should be considered to be a mata-data which may be useful for queries grouping and should not + * be considered that all these keys should be requested. + */ + private final K[] allKeysArr; + + /** + * This is the tree where each left node represents a missing key from `allKeysArr` and each right node represents a selected + * key from `allKeysArr`. The first child node represents the key with index 0 from `allKeysArr`, + * while the leaf nodes are always represent a query with the index `allKeysArr.length-1`. + * Moreover, each leaf node is of type {@code MultiSliceQueriesGroupingUtil.DataTreeNode>} , + * where `K` is of key type. + * Each leaf node represents a separate group where all queries are unique. + */ + private final MultiSliceQueriesGroupingUtil.TreeNode groupingRootTreeNode; + + /** + * All parent nodes which has at least one child node which is a leaf node with data. + */ + private final List allLeafParents; + + private final int totalAmountOfQueries; + + public MultiQueriesByKeysGroupsContext(K[] allKeysArr, MultiQueriesByKeysGroupsContext cloneContext) { + this.allKeysArr = allKeysArr; + this.groupingRootTreeNode = cloneContext.getGroupingRootTreeNode(); + this.totalAmountOfQueries = cloneContext.totalAmountOfQueries; + this.allLeafParents = cloneContext.getAllLeafParents(); + } + + public MultiQueriesByKeysGroupsContext(K[] allKeysArr, MultiSliceQueriesGroupingUtil.TreeNode groupingRootTreeNode, int totalAmountOfQueries, List allLeafParents) { + this.allKeysArr = allKeysArr; + this.groupingRootTreeNode = groupingRootTreeNode; + this.totalAmountOfQueries = totalAmountOfQueries; + this.allLeafParents = allLeafParents; + } + + public K[] getAllKeysArr() { + return allKeysArr; + } + + public MultiSliceQueriesGroupingUtil.TreeNode getGroupingRootTreeNode() { + return groupingRootTreeNode; + } + + public int getTotalAmountOfQueries() { + return totalAmountOfQueries; + } + + public List getAllLeafParents() { + return allLeafParents; + } +} diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/SliceQuery.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/SliceQuery.java index fcb3fc8451..fdd284c6fe 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/SliceQuery.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/SliceQuery.java @@ -88,8 +88,9 @@ public boolean equals(Object other) { if (this == other) return true; - if (other == null && !getClass().isInstance(other)) + if(!(other instanceof SliceQuery)){ return false; + } SliceQuery oth = (SliceQuery) other; return sliceStart.equals(oth.sliceStart) diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/StoreFeatures.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/StoreFeatures.java index 66beea265c..3e7b282ef2 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/StoreFeatures.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/StoreFeatures.java @@ -18,6 +18,8 @@ import org.janusgraph.diskstorage.keycolumnvalue.scan.ScanJob; import org.janusgraph.diskstorage.util.time.TimestampProviders; +import java.util.List; + /** * Describes features supported by a storage backend. * @@ -56,7 +58,7 @@ public interface StoreFeatures { /** * Whether this storage backend supports query operations on multiple keys * via - * {@link KeyColumnValueStore#getSlice(java.util.List, SliceQuery, StoreTransaction)} + * {@link KeyColumnValueStore#getSlice(List, SliceQuery, StoreTransaction)} and {@link KeyColumnValueStore#getMultiSlices(MultiKeysQueryGroups, StoreTransaction)} */ boolean hasMultiQuery(); diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/ExpirationKCVSCache.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/ExpirationKCVSCache.java index 1516d945ce..bf24ad678c 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/ExpirationKCVSCache.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/ExpirationKCVSCache.java @@ -17,17 +17,22 @@ import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.base.Preconditions; +import org.apache.commons.lang3.tuple.Pair; import org.janusgraph.core.JanusGraphException; import org.janusgraph.diskstorage.BackendException; import org.janusgraph.diskstorage.EntryList; import org.janusgraph.diskstorage.StaticBuffer; import org.janusgraph.diskstorage.keycolumnvalue.KeyColumnValueStore; import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; +import org.janusgraph.diskstorage.keycolumnvalue.KeysQueriesGroup; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; import org.janusgraph.diskstorage.util.CacheMetricsAction; +import org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil; import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -106,35 +111,143 @@ public EntryList getSlice(final KeySliceQuery query, final StoreTransaction txh) @Override public Map getSlice(final List keys, final SliceQuery query, final StoreTransaction txh) throws BackendException { final Map results = new HashMap<>(keys.size()); - final List remainingKeys = new ArrayList<>(keys.size()); - KeySliceQuery[] ksqs = new KeySliceQuery[keys.size()]; incActionBy(keys.size(), CacheMetricsAction.RETRIEVAL,txh); - //Find all cached queries - for (int i=0;i, Map> misses = fillResultAndReturnMisses(results, query, keys); + final List remainingKeys = misses.getKey(); + final Map keySliceQueries = misses.getValue(); + //Request remaining ones from backend if (!remainingKeys.isEmpty()) { incActionBy(remainingKeys.size(), CacheMetricsAction.MISS,txh); Map subresults = store.getSlice(remainingKeys, query, unwrapTx(txh)); - for (int i=0;i { + KeySliceQuery ksqs = keySliceQueries.get(key); + if(ksqs != null){ + cache.put(ksqs,subresult); } - } + results.put(key,subresult); + }); } return results; } + @Override + public Map> getMultiSlices(final MultiKeysQueryGroups multiKeysQueryGroups, + final StoreTransaction txh) throws BackendException { + Map> result = new HashMap<>(multiKeysQueryGroups.getMultiQueryContext().getTotalAmountOfQueries()); + Map> remainingKeysPerQuery = new HashMap<>(multiKeysQueryGroups.getMultiQueryContext().getTotalAmountOfQueries()); + + List> allQueryGroups = multiKeysQueryGroups.getQueryGroups(); + List>> updatedQueryGroups = new ArrayList<>(); + + for(KeysQueriesGroup keyQueriesGroup : allQueryGroups){ + List currentKeys = keyQueriesGroup.getKeysGroup(); + List currentQueries = keyQueriesGroup.getQueries(); + if(currentKeys.isEmpty() || currentQueries.isEmpty()){ + continue; + } + + incActionBy(currentKeys.size()*currentQueries.size(), CacheMetricsAction.RETRIEVAL,txh); + List remainingCurrentGroupQueries = new ArrayList<>(currentQueries.size()); + + for(SliceQuery query : currentQueries){ + Map currentQueryResult = result.computeIfAbsent(query, q -> new HashMap<>(currentKeys.size())); + Pair, Map> misses = fillResultAndReturnMisses(currentQueryResult, query, currentKeys); + final List remainingCurrentGroupKeys = misses.getKey(); + remainingKeysPerQuery.put(query, misses.getValue()); + if(remainingCurrentGroupKeys.size() == currentKeys.size()){ + remainingCurrentGroupQueries.add(query); + } else if(!remainingCurrentGroupKeys.isEmpty()){ + updatedQueryGroups.add(Pair.of(query, remainingCurrentGroupKeys)); + } + } + + if(remainingCurrentGroupQueries.size() != currentQueries.size()){ + keyQueriesGroup.setQueries(remainingCurrentGroupQueries); + } + } + + // move queries with updated groups to new existing or new groups. + MultiSliceQueriesGroupingUtil.moveQueriesToNewLeafNode(updatedQueryGroups, + multiKeysQueryGroups.getMultiQueryContext().getAllKeysArr(), + multiKeysQueryGroups.getMultiQueryContext().getGroupingRootTreeNode(), + allQueryGroups); + + allQueryGroups = filterEmptyGroups(allQueryGroups); + multiKeysQueryGroups.setQueryGroups(allQueryGroups); + + //Request remaining ones from backend + if(!allQueryGroups.isEmpty()){ + Map> subresults = store.getMultiSlices(multiKeysQueryGroups, unwrapTx(txh)); + subresults.forEach((sliceQuery, sliceQueryResultsPerKey) -> { + + if(!sliceQueryResultsPerKey.isEmpty()){ + incActionBy(sliceQueryResultsPerKey.size(), CacheMetricsAction.MISS,txh); + } + + // populate cache with new results for any key with non-expired keySliceQuery + Map queryKeySliceQueriesPerVertexKey = remainingKeysPerQuery.get(sliceQuery); + if(queryKeySliceQueriesPerVertexKey != null){ + sliceQueryResultsPerKey.forEach((key, keyResult) -> { + KeySliceQuery ksqs = queryKeySliceQueriesPerVertexKey.get(key); + if(ksqs != null){ + cache.put(ksqs,keyResult); + } + }); + } + + // add requested results into a final resulting map + Map currentSliceQueryResults = result.get(sliceQuery); + if(currentSliceQueryResults == null){ + currentSliceQueryResults = sliceQueryResultsPerKey; + result.put(sliceQuery, currentSliceQueryResults); + } else { + currentSliceQueryResults.putAll(sliceQueryResultsPerKey); + } + }); + } + + return result; + } + + private List> filterEmptyGroups(List> originalGroups){ + List> filteredGroups = new ArrayList<>(originalGroups.size()); + for(KeysQueriesGroup group : originalGroups){ + if(!group.getKeysGroup().isEmpty() && !group.getQueries().isEmpty()){ + filteredGroups.add(group); + } + } + return filteredGroups; + } + + + /** + * Fills the result with non-expired cached data. Returns any keys which are missed as well as their optional `KeySliceQuery` in case it's not expired. + * If `KeySliceQuery` is currently considered to be expired then `null` will be returned for respective `StaticBuffer`. + * This method doesn't execute any slice queries to the storage backend. + */ + private Pair, Map> fillResultAndReturnMisses(final Map results, final SliceQuery query, final Collection keys){ + final Map keySliceQueries = new HashMap<>(keys.size()); + final List remainingKeys = new ArrayList<>(keys.size()); + //Find all cached queries + for (StaticBuffer key : keys) { + KeySliceQuery ksqs = new KeySliceQuery(key,query); + if (isExpired(ksqs)){ + remainingKeys.add(key); + keySliceQueries.put(key, null); + } else { + EntryList result = cache.getIfPresent(ksqs); + if (result == null){ + remainingKeys.add(key); + keySliceQueries.put(key, ksqs); + } else { + results.put(key, result); + } + } + } + return Pair.of(remainingKeys, keySliceQueries); + } + @Override public void clearCache() { // We should not call `expiredKeys.clear();` directly because there could be a race condition @@ -243,7 +356,4 @@ void stopThread() { } } - - - } diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/KCVSCache.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/KCVSCache.java index f5956a670c..69fc93265b 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/KCVSCache.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/keycolumnvalue/cache/KCVSCache.java @@ -21,6 +21,7 @@ import org.janusgraph.diskstorage.keycolumnvalue.KCVSProxy; import org.janusgraph.diskstorage.keycolumnvalue.KeyColumnValueStore; import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; import org.janusgraph.diskstorage.util.CacheMetricsAction; @@ -85,4 +86,7 @@ public Map getSliceNoCache(List keys, Sli return store.getSlice(keys,query,unwrapTx(txh)); } + public Map> getMultiSlicesNoCache(MultiKeysQueryGroups multiSliceQueriesWithKeys, StoreTransaction txh) throws BackendException { + return store.getMultiSlices(multiSliceQueriesWithKeys, unwrapTx(txh)); + } } diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/CompletableFutureUtil.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/CompletableFutureUtil.java index f7bbd2a311..f137a0d70f 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/CompletableFutureUtil.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/CompletableFutureUtil.java @@ -60,6 +60,7 @@ public static T get(CompletableFuture future){ } } + // public static Map unwrap(Map> futureMap) throws Throwable{ Map resultMap = new HashMap<>(futureMap.size()); Throwable firstException = null; @@ -81,6 +82,27 @@ public static Map unwrap(Map> futureMap) throw return resultMap; } + public static Map> unwrapMapOfMaps(Map>> futureMap) throws Throwable{ + Map> resultMap = new HashMap<>(futureMap.size()); + Throwable firstException = null; + for(Map.Entry>> entry : futureMap.entrySet()){ + try{ + resultMap.put(entry.getKey(), unwrap(entry.getValue())); + } catch (Throwable throwable){ + if(firstException == null){ + firstException = throwable; + } else { + firstException.addSuppressed(throwable); + } + } + } + + if(firstException != null){ + throw firstException; + } + return resultMap; + } + public static void awaitAll(Collection> futureCollection) throws Throwable{ Throwable firstException = null; for(CompletableFuture future : futureCollection){ diff --git a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/MetricInstrumentedStore.java b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/MetricInstrumentedStore.java index 17d43f20c6..0936159d06 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/MetricInstrumentedStore.java +++ b/janusgraph-core/src/main/java/org/janusgraph/diskstorage/util/MetricInstrumentedStore.java @@ -26,6 +26,7 @@ import org.janusgraph.diskstorage.keycolumnvalue.KeyRangeQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySlicesIterator; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.diskstorage.keycolumnvalue.MultiSlicesQuery; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; @@ -128,6 +129,20 @@ public Map getSlice(final List keys, }); } + @Override + public Map> getMultiSlices(MultiKeysQueryGroups multiKeysQueryGroups, StoreTransaction txh) throws BackendException { + return runWithMetrics(txh, metricsStoreName, M_GET_SLICE, () -> { + final Map> results = backend.getMultiSlices(multiKeysQueryGroups, txh); + + for (final Map queryResults : results.values()) { + for (final EntryList result : queryResults.values()) { + recordSliceMetrics(txh, result); + } + } + return results; + }); + } + @Override public void mutate(final StaticBuffer key, final List additions, diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/database/StandardJanusGraph.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/database/StandardJanusGraph.java index 0bd3e58b90..2f9560cefb 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/graphdb/database/StandardJanusGraph.java +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/database/StandardJanusGraph.java @@ -49,6 +49,9 @@ import org.janusgraph.diskstorage.keycolumnvalue.KeyIterator; import org.janusgraph.diskstorage.keycolumnvalue.KeyRangeQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; +import org.janusgraph.diskstorage.keycolumnvalue.KeysQueriesGroup; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; +import org.janusgraph.diskstorage.keycolumnvalue.MultiQueriesByKeysGroupsContext; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.StoreFeatures; import org.janusgraph.diskstorage.keycolumnvalue.cache.KCVSCache; @@ -68,6 +71,7 @@ import org.janusgraph.graphdb.tinkerpop.optimize.strategy.JanusGraphHasStepStrategy; import org.janusgraph.graphdb.tinkerpop.optimize.strategy.JanusGraphLocalQueryOptimizerStrategy; import org.janusgraph.graphdb.tinkerpop.optimize.strategy.JanusGraphUnusedMultiQueryRemovalStrategy; +import org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil; import org.janusgraph.util.IDUtils; import org.janusgraph.graphdb.database.index.IndexInfoRetriever; import org.janusgraph.graphdb.database.index.IndexUpdate; @@ -567,6 +571,51 @@ public List edgeMultiQuery(List vertexIdsAsObjects, SliceQuer return resultList; } + private StaticBuffer[] convertVertexIds(Object[] vertexIds){ + StaticBuffer[] convertedIds = new StaticBuffer[vertexIds.length]; + for(int i=0;i> edgeMultiQuery(MultiKeysQueryGroups groupedMultiSliceQueries, BackendTransaction tx) { + assert groupedMultiSliceQueries != null && !groupedMultiSliceQueries.getQueryGroups().isEmpty(); + + Object[] vertexIds = groupedMultiSliceQueries.getMultiQueryContext().getAllKeysArr(); + StaticBuffer[] vertexKeys = convertVertexIds(vertexIds); + Map vertexIdToKey = new HashMap<>(vertexKeys.length); + Map keyToVertexId = new HashMap<>(vertexKeys.length); + for(int i=0; i> groupedMultiSliceQueriesWithKeys = new ArrayList<>(groupedMultiSliceQueries.getQueryGroups().size()); + + for(KeysQueriesGroup queriesToVertexIdsPaid : groupedMultiSliceQueries.getQueryGroups()){ + List keys = new ArrayList<>(queriesToVertexIdsPaid.getKeysGroup().size()); + for (Object vertexIdsAsObject : queriesToVertexIdsPaid.getKeysGroup()) { + keys.add(vertexIdToKey.get(vertexIdsAsObject)); + } + groupedMultiSliceQueriesWithKeys.add(new KeysQueriesGroup<>(keys, queriesToVertexIdsPaid.getQueries())); + } + + MultiSliceQueriesGroupingUtil.replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(groupedMultiSliceQueries.getMultiQueryContext().getAllLeafParents(), vertexIdToKey); + final Map> resultWithKeys = tx.edgeStoreMultiQuery( + new MultiKeysQueryGroups<>(groupedMultiSliceQueriesWithKeys, new MultiQueriesByKeysGroupsContext<>(vertexKeys, groupedMultiSliceQueries.getMultiQueryContext()))); + + final Map> resultWithVertexIds = new HashMap<>(resultWithKeys.size()); + resultWithKeys.forEach((sliceQuery, keyToResultMap) -> { + Map vertexIdToResultMap = new HashMap<>(keyToResultMap.size()); + keyToResultMap.forEach((key, result) -> vertexIdToResultMap.put(keyToVertexId.get(key), result)); + resultWithVertexIds.put(sliceQuery, vertexIdToResultMap); + }); + + return resultWithVertexIds; + } + private ModifiableConfiguration getGlobalSystemConfig(Backend backend) { return new ModifiableConfiguration(GraphDatabaseConfiguration.ROOT_NS, diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/BackendQueryHolder.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/BackendQueryHolder.java index 02e5a66916..b923419674 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/BackendQueryHolder.java +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/BackendQueryHolder.java @@ -18,6 +18,8 @@ import org.janusgraph.graphdb.query.profile.ProfileObservable; import org.janusgraph.graphdb.query.profile.QueryProfiler; +import java.util.Objects; + /** * Holds a {@link BackendQuery} and captures additional information that pertains to its execution and to be used by a * {@link QueryExecutor}: @@ -81,4 +83,21 @@ public void observeWith(QueryProfiler parentProfiler, boolean hasSiblings) { profiler.setAnnotation(QueryProfiler.QUERY_ANNOTATION,backendQuery); if (backendQuery instanceof ProfileObservable) ((ProfileObservable)backendQuery).observeWith(profiler); } + + @Override + public int hashCode() { + return Objects.hash(backendQuery, isFitted, isFitted); + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if(!(other instanceof BackendQueryHolder)){ + return false; + } + BackendQueryHolder oth = (BackendQueryHolder) other; + return isFitted == oth.isFitted && isSorted == oth.isSorted && Objects.equals(backendQuery, oth.backendQuery); + } } diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/profile/QueryProfiler.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/profile/QueryProfiler.java index e3596ba755..6e1cda7f9f 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/profile/QueryProfiler.java +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/profile/QueryProfiler.java @@ -14,10 +14,12 @@ package org.janusgraph.graphdb.query.profile; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.graphdb.query.Query; import org.janusgraph.graphdb.query.graph.JointIndexQuery.Subquery; import java.util.Collection; +import java.util.Map; import java.util.function.Function; /** @@ -39,6 +41,10 @@ public interface QueryProfiler { String QUERY_ANNOTATION = "query"; String FULLSCAN_ANNOTATION = "fullscan"; String INDEX_ANNOTATION = "index"; + String QUERIES_ANNOTATION = "queries"; + String QUERIES_AMOUNT_ANNOTATION = "queriesAmount"; + String MULTI_SLICES_ANNOTATION = "multiSlices"; + String QUERY_LIMITS_ANNOTATION = "limitsPerQuery"; /* ================================================================================== GROUP NAMES @@ -120,6 +126,38 @@ static R profile(String groupName, QueryPr return result; } + static Map> profile(QueryProfiler profiler, MultiKeysQueryGroups multiSliceQueries, boolean multiQuery, Function, Map>> queryExecutor) { + return profile(BACKEND_QUERY,profiler,multiSliceQueries,multiQuery,queryExecutor); + } + + static Map> profile(String groupName, QueryProfiler profiler, MultiKeysQueryGroups multiSliceQueries, boolean multiQuery, Function, Map>> queryExecutor) { + final QueryProfiler sub = profiler.addNested(groupName); + QueryProfilerUtil.setMultiSliceQueryAnnotations(sub, multiSliceQueries); + sub.startTimer(); + final Map> result = queryExecutor.apply(multiSliceQueries); + sub.stopTimer(); + long resultSize = 0; + if (multiQuery && profiler!=QueryProfiler.NO_OP) { + //The result set is a collection of collections, but don't do this computation if profiling is disabled + for(Map resultsPerSlice : result.values()){ + for (R resultsPerVertex : resultsPerSlice.values()) { + for(Object r : resultsPerVertex){ + if (r instanceof Collection) resultSize+=((Collection)r).size(); + else resultSize++; + } + } + } + } else { + for(Map resultsPerSlice : result.values()){ + for (R resultsPerVertex : resultsPerSlice.values()) { + resultSize += resultsPerVertex.size(); + } + } + } + sub.setResultSize(resultSize); + return result; + } + static QueryProfiler startProfile(QueryProfiler profiler, Subquery query) { final QueryProfiler sub = profiler.addNested(BACKEND_QUERY); sub.setAnnotation(QUERY_ANNOTATION, query); diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/profile/QueryProfilerUtil.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/profile/QueryProfilerUtil.java new file mode 100644 index 0000000000..32c6b09ac5 --- /dev/null +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/profile/QueryProfilerUtil.java @@ -0,0 +1,61 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.graphdb.query.profile; + +import org.janusgraph.diskstorage.keycolumnvalue.KeysQueriesGroup; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; +import org.janusgraph.graphdb.query.Query; + +import java.util.ArrayList; +import java.util.List; + +public class QueryProfilerUtil { + + private QueryProfilerUtil(){} + + public static void setMultiSliceQueryAnnotations(QueryProfiler profiler, MultiKeysQueryGroups multiSliceQueries){ + if(profiler == QueryProfiler.NO_OP){ + return; + } + profiler.setAnnotation(QueryProfiler.MULTI_SLICES_ANNOTATION, true); + List allQueries = new ArrayList<>(); + ArrayList allLimits = null; + boolean hasLimit = false; + int queriesCount = 0; + for(KeysQueriesGroup groupedQueries : multiSliceQueries.getQueryGroups()){ + for(Q query : groupedQueries.getQueries()){ + allQueries.add(query); + if(hasLimit){ + allLimits.add(query.hasLimit() ? query.getLimit() : -1); + } else if(query.hasLimit()){ + hasLimit = true; + allLimits = new ArrayList<>(); + allLimits.ensureCapacity(queriesCount+1); + for(int i=0;i Map execute(RelationCategory returnType, Result profiler.setAnnotation(QueryProfiler.MULTIQUERY_ANNOTATION,true); profiler.setAnnotation(QueryProfiler.NUMVERTICES_ANNOTATION,vertices.size()); if (!bq.isEmpty()) { - for (BackendQueryHolder sq : bq.getQueries()) { - Collection adjVertices = getResolvedAdjVertices(); - //Overwrite with more accurate size accounting for partitioned vertices - profiler.setAnnotation(QueryProfiler.NUMVERTICES_ANNOTATION,adjVertices.size()); - tx.executeMultiQuery(adjVertices, sq.getBackendQuery(), sq.getProfiler()); + Collection adjVertices = getResolvedAdjVertices(); + //Overwrite with more accurate size accounting for partitioned vertices + profiler.setAnnotation(QueryProfiler.NUMVERTICES_ANNOTATION,adjVertices.size()); + if(bq.getQueries().size() == 1){ + BackendQueryHolder query = bq.getQueries().get(0); + // if it's just a single query - there is no need to group any queries together. + // Thus, we can execute `executeMultiQuery` instead of `executeMultiSliceMultiQuery` + tx.executeMultiQuery(adjVertices, query.getBackendQuery(), query.getProfiler()); + } else { + tx.executeMultiSliceMultiQuery(adjVertices, bq.getQueries(), profiler); } for (InternalVertex v : vertices) { result.put(v, resultConstructor.getResult(v, bq)); diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/vertex/VertexCentricQueryBuilder.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/vertex/VertexCentricQueryBuilder.java index 323bbcaed1..2015b49f0d 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/vertex/VertexCentricQueryBuilder.java +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/query/vertex/VertexCentricQueryBuilder.java @@ -28,6 +28,7 @@ import org.janusgraph.graphdb.query.QueryProcessor; import org.janusgraph.graphdb.query.profile.QueryProfiler; +import java.util.Collections; import java.util.List; /** @@ -72,8 +73,9 @@ protected VertexCentricQueryBuilder getThis() { protected Q execute(RelationCategory returnType, ResultConstructor resultConstructor) { BaseVertexCentricQuery bq = super.constructQuery(returnType); if (bq.isEmpty()) return resultConstructor.emptyResult(); - if (returnType==RelationCategory.PROPERTY && hasSingleType() && !hasQueryOnlyLoaded() - && tx.getConfiguration().hasPropertyPrefetching()) { + boolean prefetchAllVertexProperties = returnType==RelationCategory.PROPERTY && hasSingleType() && !hasQueryOnlyLoaded() + && tx.getConfiguration().hasPropertyPrefetching(); + if (prefetchAllVertexProperties) { //Preload properties vertex.query().properties().iterator().hasNext(); } @@ -83,11 +85,24 @@ protected Q execute(RelationCategory returnType, ResultConstructor resultC profiler.setAnnotation(QueryProfiler.PARTITIONED_VERTEX_ANNOTATION,true); profiler.setAnnotation(QueryProfiler.NUMVERTICES_ANNOTATION,vertices.size()); if (vertices.size()>1) { - for (BackendQueryHolder sq : bq.getQueries()) { - tx.executeMultiQuery(vertices, sq.getBackendQuery(),sq.getProfiler()); + if(bq.getQueries().size() == 1){ + BackendQueryHolder query = bq.getQueries().get(0); + // if it's just a single query - there is no need to group any queries together. + // Thus, we can execute `executeMultiQuery` instead of `executeMultiSliceMultiQuery` + tx.executeMultiQuery(vertices, query.getBackendQuery(),query.getProfiler()); + } else { + tx.executeMultiSliceMultiQuery(vertices, bq.getQueries(), profiler); } } - } else profiler.setAnnotation(QueryProfiler.NUMVERTICES_ANNOTATION,1); + } else { + profiler.setAnnotation(QueryProfiler.NUMVERTICES_ANNOTATION,1); + // in case there are more than 1 slice query expected for the provided vertex + // then it still makes sense to pre-fetch vertex data via multi-slice query (all queries in parallel) + // instead of executing each separate query sequentially + if(!prefetchAllVertexProperties && bq.getQueries().size() > 1){ + tx.executeMultiSliceMultiQuery(Collections.singletonList(vertex), bq.getQueries(), profiler); + } + } return resultConstructor.getResult(vertex,bq); } diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/transaction/StandardJanusGraphTx.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/transaction/StandardJanusGraphTx.java index b84afa8d74..939ce23962 100644 --- a/janusgraph-core/src/main/java/org/janusgraph/graphdb/transaction/StandardJanusGraphTx.java +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/transaction/StandardJanusGraphTx.java @@ -49,6 +49,7 @@ import org.janusgraph.diskstorage.BackendTransaction; import org.janusgraph.diskstorage.EntryList; import org.janusgraph.diskstorage.indexing.IndexTransaction; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.util.time.TimestampProvider; import org.janusgraph.graphdb.database.EdgeSerializer; @@ -67,6 +68,7 @@ import org.janusgraph.graphdb.internal.InternalVertexLabel; import org.janusgraph.graphdb.internal.JanusGraphSchemaCategory; import org.janusgraph.graphdb.internal.RelationCategory; +import org.janusgraph.graphdb.query.BackendQueryHolder; import org.janusgraph.graphdb.query.MetricsQueryExecutor; import org.janusgraph.graphdb.query.Query; import org.janusgraph.graphdb.query.QueryExecutor; @@ -132,6 +134,7 @@ import org.janusgraph.graphdb.types.vertices.JanusGraphSchemaVertex; import org.janusgraph.graphdb.types.vertices.PropertyKeyVertex; import org.janusgraph.graphdb.util.IndexHelper; +import org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil; import org.janusgraph.graphdb.util.ProfiledIterator; import org.janusgraph.graphdb.util.SubqueryIterator; import org.janusgraph.graphdb.util.VertexCentricEdgeIterable; @@ -1258,6 +1261,16 @@ public void executeMultiQuery(final Collection vertices, final S } } + public void executeMultiSliceMultiQuery(final Collection vertices, final List> queries, QueryProfiler profiler) { + MultiKeysQueryGroups groupedMultiSliceQueries = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups(vertices, queries); + if (!groupedMultiSliceQueries.getQueryGroups().isEmpty()) { + Map> allResults = QueryProfiler.profile(profiler, groupedMultiSliceQueries, true, q -> graph.edgeMultiQuery(q, txHandle)); + Map vertexIdToVertexMap = vertices.stream().collect(Collectors.toMap(JanusGraphElement::id, v -> v)); + allResults.forEach((sliceQuery, resultsPerQuery) -> resultsPerQuery.forEach((vertexId, vertexSliceResult) -> + ((CacheVertex) vertexIdToVertexMap.get(vertexId)).loadRelations(sliceQuery, query -> vertexSliceResult))); + } + } + public final QueryExecutor edgeProcessor; public final QueryExecutor edgeProcessorImpl = new QueryExecutor() { diff --git a/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/MultiSliceQueriesGroupingUtil.java b/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/MultiSliceQueriesGroupingUtil.java new file mode 100644 index 0000000000..fd34919df0 --- /dev/null +++ b/janusgraph-core/src/main/java/org/janusgraph/graphdb/util/MultiSliceQueriesGroupingUtil.java @@ -0,0 +1,327 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.graphdb.util; + +import org.apache.commons.lang3.mutable.MutableInt; +import org.apache.commons.lang3.tuple.Pair; +import org.janusgraph.diskstorage.keycolumnvalue.KeysQueriesGroup; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; +import org.janusgraph.diskstorage.keycolumnvalue.MultiQueriesByKeysGroupsContext; +import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; +import org.janusgraph.graphdb.internal.InternalVertex; +import org.janusgraph.graphdb.query.BackendQueryHolder; +import org.janusgraph.graphdb.vertices.CacheVertex; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +/** + * Utility class which helps to group necessary queries to matched vertices for the following multi-query execution. + */ +public class MultiSliceQueriesGroupingUtil { + + private static final MultiKeysQueryGroups EMPTY_QUERY = new MultiKeysQueryGroups<>(Collections.emptyList(), + new MultiQueriesByKeysGroupsContext<>(new Object[0], new TreeNode(), 0, Collections.emptyList())); + + /** + * Moves queries `updatedQueryGroup` to either existing leaf nodes or generates new leaf nodes for the necessary key sets. + * + * @param updatedQueryGroup Query groups to which a query should be moved. + * @param allVertices All vertices + * @param groupingRootTreeNode Root tree node which represents chains where each left node represents a missing key + * from `allVertices` array, and each right node represents existing key from `allVertices` + * array. + * @param remainingQueryGroups Remaining groups where queries from `updatedQueryGroup` should be added. Notice, if + * a new leaf node is added into `groupingRootTreeNode` then a new group is added to + * `remainingQueryGroups`. + */ + public static void moveQueriesToNewLeafNode(List>> updatedQueryGroup, + K[] allVertices, + TreeNode groupingRootTreeNode, + List> remainingQueryGroups){ + + for(Pair> keyNewQueriesGroup : updatedQueryGroup){ + boolean newChainGenerated = false; + TreeNode previousNode = groupingRootTreeNode; + TreeNode currentNode = groupingRootTreeNode; + boolean rightNode = false; + Iterator usedKeyIt = keyNewQueriesGroup.getValue().iterator(); + for(int keyIndex = 0; keyIndex < allVertices.length; keyIndex++){ + + K usedKey = usedKeyIt.hasNext() ? usedKeyIt.next() : null; + + while (keyIndex < allVertices.length && usedKey != allVertices[keyIndex]){ + rightNode = false; + if(currentNode.left == null){ + currentNode.left = new TreeNode(); + newChainGenerated = true; + } + previousNode = currentNode; + currentNode = currentNode.left; + ++keyIndex; + } + + if(keyIndex == allVertices.length){ + break; + } + + rightNode = true; + if(currentNode.right == null){ + currentNode.right = new TreeNode(); + newChainGenerated = true; + } + previousNode = currentNode; + currentNode = currentNode.right; + } + + DataTreeNode> chainLeafNode; + if(newChainGenerated){ + KeysQueriesGroup data = new KeysQueriesGroup<>(keyNewQueriesGroup.getValue(), new ArrayList<>()); + chainLeafNode = new DataTreeNode<>(data); + if(rightNode){ + previousNode.right = chainLeafNode; + } else { + previousNode.left = chainLeafNode; + } + remainingQueryGroups.add(data); + } else { + chainLeafNode = (DataTreeNode>) currentNode; + } + chainLeafNode.data.getQueries().add(keyNewQueriesGroup.getKey()); + } + } + + /** + * Replaces child leaf nodes which have data {@code KeysQueriesGroup} with new leaf nodes which have keys replaces + * by `oldToNewKeysMap`. The resulting data of child nodes will be {@code KeysQueriesGroup}. + * + * @param allLeafParents parent nodes + * @param oldToNewKeysMap map to replace old type keys with new type keys + * @param current key type + * @param new key type + */ + public static void replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(List allLeafParents, Map oldToNewKeysMap){ + for(TreeNode node : allLeafParents){ + if(node.left instanceof DataTreeNode){ + node.left = generateConvertedLeafNode((DataTreeNode>) node.left, oldToNewKeysMap); + } + if(node.right instanceof DataTreeNode){ + node.right = generateConvertedLeafNode((DataTreeNode>) node.right, oldToNewKeysMap); + } + } + } + + private static DataTreeNode> generateConvertedLeafNode(DataTreeNode> currentLeafNode, Map oldToNewKeysMap){ + KeysQueriesGroup data = currentLeafNode.data; + List keysGroup = new ArrayList<>(data.getKeysGroup().size()); + for(C currentKey : data.getKeysGroup()){ + keysGroup.add(oldToNewKeysMap.get(currentKey)); + } + KeysQueriesGroup newData = new KeysQueriesGroup<>(keysGroup, data.getQueries()); + return new DataTreeNode<>(newData); + } + + /** + * Queries grouping algorithm for the same sets of keys (vertices). + *
+ * This algorithm uses a binary prefix tree to find a group with the same vertices (same vertices Set). + * All and only the leaf nodes store the final computed data (queries used for this leaf node + vertices used for this leaf nodes). + * Each leaf node represents a group of queries which always have the same vertices. + * Time complexity is always O(N+M), where N is the vertices amount and M is the queries amount. + * Space complexity in most cases is O(N+M), or O(N*M) in the worst case. + * + * @param vertices all vertices. + * @param queries all queries which should be executed for all vertices. + * @return Generated grouped multi-query representation. + */ + public static MultiKeysQueryGroups toMultiKeysQueryGroups(final Collection vertices, final List> queries){ + if(queries.isEmpty()){ + return EMPTY_QUERY; + } + MutableInt verticesMutableSize = new MutableInt(); + InternalVertex[] cacheableVertices = filterNonCacheableVertices(vertices, verticesMutableSize); + final int verticesSize = verticesMutableSize.intValue(); + Object[] vertexIds = toIds(cacheableVertices, verticesSize); + if(verticesSize == 0){ + return EMPTY_QUERY; + } + List> result = new ArrayList<>(); + TreeNode root = new TreeNode(); + boolean[] useVertex = new boolean[verticesSize]; + List allLeafParents = new ArrayList<>(); + for(BackendQueryHolder queryHolder : queries){ + final SliceQuery query = queryHolder.getBackendQuery(); + TreeNode currentNode = root; + int usedVertices = 0; + for(int i=0; i>) currentNode).data.getQueries().add(query); + } + return new MultiKeysQueryGroups<>(result, new MultiQueriesByKeysGroupsContext<>(vertexIds, root, queries.size(), allLeafParents)); + } + + private static InternalVertex[] filterNonCacheableVertices(Collection vertices, MutableInt verticesMutableSize){ + InternalVertex[] cacheableVertices = new InternalVertex[vertices.size()]; + int i = 0; + for(InternalVertex v : vertices){ + if(!v.isNew() && v.hasId() && (v instanceof CacheVertex)){ + cacheableVertices[i++] = v; + } + } + verticesMutableSize.setValue(i); + return cacheableVertices; + } + + private static Object[] toPartiallyFilledVertexIds(InternalVertex[] cacheableVertices, boolean[] useVertex, int fillUpToIndex, int totalVerticesSize){ + Object[] vertexIds = new Object[totalVerticesSize]; + for(int i=0, j=0; i<=fillUpToIndex; i++){ + if(useVertex[i]){ + vertexIds[j++] = cacheableVertices[i].id(); + } + } + return vertexIds; + } + + private static Object[] toIds(InternalVertex[] cacheableVertices, int totalVerticesSize){ + Object[] vertexIds = new Object[totalVerticesSize]; + for(int i=0; i> result, + List allLeafParents){ + TreeNode previousNode = currentNode; + currentNode = new TreeNode(); + assignChild(previousNode, currentNode, childNodeHasLoadedRelations); + Object[] queryVertexIds = toPartiallyFilledVertexIds(cacheableVertices, useVertex, currentIndex, totalVerticesSize); + ChainCreationResult newChain = generateNewNodesChain( + cacheableVertices, queryVertexIds, previousNode, currentNode, currentIndex, usedVertices, totalVerticesSize, childNodeHasLoadedRelations, query); + previousNode = newChain.latestParent; + usedVertices = newChain.usedVerticesCount; + if(usedVertices != totalVerticesSize){ + if(usedVertices == 0){ + queryVertexIds = new Object[0]; + } else { + Object[] trimmedQueryVertexIds = new Object[usedVertices]; + System.arraycopy(queryVertexIds, 0, trimmedQueryVertexIds, 0, usedVertices); + queryVertexIds = trimmedQueryVertexIds; + } + } + + KeysQueriesGroup data = new KeysQueriesGroup<>(Arrays.asList(queryVertexIds), new ArrayList<>()); + currentNode = new DataTreeNode<>(data); + assignChild(previousNode, currentNode, newChain.lastVertexHasLoadedRelations); + if(previousNode.left == null || previousNode.right == null){ + allLeafParents.add(previousNode); + } + if(usedVertices > 0){ + result.add(data); + } + return currentNode; + } + + private static void assignChild(TreeNode parent, TreeNode child, boolean childNodeHasLoadedRelations){ + if(childNodeHasLoadedRelations){ + parent.left = child; + } else { + parent.right = child; + } + } + + private static ChainCreationResult generateNewNodesChain(InternalVertex[] cacheableVertices, + Object[] queryVertexIds, + TreeNode previousNode, + TreeNode currentNode, + int currentIndex, + int usedVertices, + int totalVerticesSize, + boolean lastChildHasLoadedRelations, + SliceQuery currentQuery){ + boolean hasLoadedRelations = lastChildHasLoadedRelations; + while (++currentIndex < totalVerticesSize){ + previousNode = currentNode; + hasLoadedRelations = cacheableVertices[currentIndex].hasLoadedRelations(currentQuery); + if(hasLoadedRelations){ + currentNode.left = new TreeNode(); + currentNode = currentNode.left; + } else { + currentNode.right = new TreeNode(); + currentNode = currentNode.right; + queryVertexIds[usedVertices++] = cacheableVertices[currentIndex].id(); + } + } + return new ChainCreationResult(previousNode, hasLoadedRelations, usedVertices); + } + + public static class TreeNode { + public TreeNode left; + public TreeNode right; + public TreeNode() { + } + public TreeNode(TreeNode left, TreeNode right) { + this.left = left; + this.right = right; + } + } + + public static class DataTreeNode extends TreeNode{ + public final Q data; + public DataTreeNode(Q data) { + this.data = data; + } + } + + private static class ChainCreationResult{ + // parent before the leaf node + public final TreeNode latestParent; + public final boolean lastVertexHasLoadedRelations; + public final int usedVerticesCount; + + private ChainCreationResult(TreeNode latestParent, boolean lastVertexHasLoadedRelations, int usedVerticesCount) { + this.latestParent = latestParent; + this.lastVertexHasLoadedRelations = lastVertexHasLoadedRelations; + this.usedVerticesCount = usedVerticesCount; + } + } +} diff --git a/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/MultiSliceQueriesGroupingUtilTest.java b/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/MultiSliceQueriesGroupingUtilTest.java new file mode 100644 index 0000000000..e8fe7344b5 --- /dev/null +++ b/janusgraph-core/src/test/java/org/janusgraph/graphdb/util/MultiSliceQueriesGroupingUtilTest.java @@ -0,0 +1,754 @@ +// Copyright 2023 JanusGraph Authors +// +// Licensed 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.janusgraph.graphdb.util; + +import org.apache.commons.lang3.mutable.MutableBoolean; +import org.apache.commons.lang3.tuple.Pair; +import org.janusgraph.diskstorage.keycolumnvalue.KeysQueriesGroup; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; +import org.janusgraph.diskstorage.keycolumnvalue.MultiQueriesByKeysGroupsContext; +import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; +import org.janusgraph.graphdb.internal.InternalVertex; +import org.janusgraph.graphdb.query.BackendQueryHolder; +import org.janusgraph.graphdb.vertices.CacheVertex; +import org.janusgraph.graphdb.vertices.StandardVertex; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; + +import static org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil.DataTreeNode; +import static org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil.TreeNode; +import static org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil.moveQueriesToNewLeafNode; +import static org.janusgraph.graphdb.util.MultiSliceQueriesGroupingUtil.replaceCurrentLeafNodeWithUpdatedTypeLeafNodes; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; + +public class MultiSliceQueriesGroupingUtilTest { + + private static final AtomicLong VERTEX_ID = new AtomicLong(1); + + ///// Grouping testing + + @Test + public void testGroupingNoVertices() { + + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups(Collections.emptyList(), + Collections.singletonList(new BackendQueryHolder<>(mock(SliceQuery.class), false, false))); + + assertEquals(0, result.getQueryGroups().size()); + assertEquals(0, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(0, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(0, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + } + + @Test + public void testGroupingNoQueries() { + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + Collections.singletonList(alwaysUseVertex()), + Collections.emptyList()); + + assertEquals(0, result.getQueryGroups().size()); + assertEquals(0, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(0, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(0, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + + result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + Collections.emptyList(), + Collections.emptyList()); + + assertEquals(0, result.getQueryGroups().size()); + assertEquals(0, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(0, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(0, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + } + + @Test + public void testGroupingNonCacheableVertices() { + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + Collections.singletonList(nonCacheVertex()), + Collections.singletonList(new BackendQueryHolder<>(mock(SliceQuery.class), false, false))); + assertEquals(0, result.getQueryGroups().size()); + assertEquals(0, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(0, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(0, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + + result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + Arrays.asList(nonCacheVertex(), alwaysUseVertex(), nonCacheVertex()), + Collections.singletonList(new BackendQueryHolder<>(mock(SliceQuery.class), false, false))); + assertEquals(1, result.getQueryGroups().size()); + assertEquals(1, result.getQueryGroups().get(0).getQueries().size()); + assertEquals(1, result.getQueryGroups().get(0).getKeysGroup().size()); + assertEquals(1, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(1, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(1, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + } + + @Test + public void testGroupingAlwaysSkipVertex() { + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + Collections.singletonList(alwaysSkipVertex()), + Collections.singletonList(new BackendQueryHolder<>(mock(SliceQuery.class), false, false))); + assertEquals(0, result.getQueryGroups().size()); + assertEquals(1, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(1, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(1, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + + result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + Arrays.asList(alwaysSkipVertex(), alwaysUseVertex(), nonCacheVertex()), + Collections.singletonList(new BackendQueryHolder<>(mock(SliceQuery.class), false, false))); + assertEquals(1, result.getQueryGroups().size()); + assertEquals(1, result.getQueryGroups().get(0).getQueries().size()); + assertEquals(1, result.getQueryGroups().get(0).getKeysGroup().size()); + assertEquals(2, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(1, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(1, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + } + + @Test + public void testGroupingByQueries() { + + ArrayList> queries = new ArrayList<>(); + for (int i=0; i<10; i++){ + queries.add(new BackendQueryHolder<>(mock(SliceQuery.class), false, false)); + } + + ArrayList alwaysUseVertices = new ArrayList<>(); + for(int i=0;i<10;i++){ + alwaysUseVertices.add(alwaysUseVertex()); + } + InternalVertex vertex1Query1 = vertex(queries.get(1).getBackendQuery()); + + InternalVertex vertex1Query2 = vertex(queries.get(2).getBackendQuery(), queries.get(3).getBackendQuery()); + InternalVertex vertex2Query2 = vertex(queries.get(2).getBackendQuery(), queries.get(3).getBackendQuery()); + InternalVertex vertex3Query2 = vertex(queries.get(2).getBackendQuery(), queries.get(3).getBackendQuery()); + + InternalVertex vertex1Query5_9 = vertex(queries.get(5).getBackendQuery(), queries.get(9).getBackendQuery()); + InternalVertex vertex1Query8 = vertex(queries.get(8).getBackendQuery()); + + InternalVertex vertex1Query7 = vertex(queries.get(7).getBackendQuery()); + InternalVertex vertex2Query7 = vertex(queries.get(7).getBackendQuery()); + + InternalVertex vertex1Query1_2_4_7 = vertex(queries.get(1).getBackendQuery(), queries.get(2).getBackendQuery(), + queries.get(3).getBackendQuery(), queries.get(7).getBackendQuery()); + InternalVertex vertex2Query1_2_4_7 = vertex(queries.get(1).getBackendQuery(), queries.get(2).getBackendQuery(), + queries.get(3).getBackendQuery(), queries.get(7).getBackendQuery()); + + ArrayList allVertices = new ArrayList<>(); + allVertices.addAll(alwaysUseVertices); + allVertices.add(alwaysSkipVertex()); + allVertices.add(alwaysSkipVertex()); + allVertices.add(alwaysSkipVertex()); + allVertices.add(vertex1Query1); + allVertices.add(vertex1Query2); + allVertices.add(vertex2Query2); + allVertices.add(vertex3Query2); + allVertices.add(vertex1Query5_9); + allVertices.add(vertex1Query8); + allVertices.add(vertex1Query7); + allVertices.add(vertex2Query7); + allVertices.add(vertex1Query1_2_4_7); + allVertices.add(vertex2Query1_2_4_7); + + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + allVertices, queries); + + assertEquals(6, result.getQueryGroups().size()); + assertEquals(alwaysUseVertices.size()+13, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(queries.size(), result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(6, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + assertGroupedByUniqueQueriesAndVertexSets(result); + + Map queryVerticesSizes = new HashMap<>(); + for(Integer queryNumber : Arrays.asList(0, 4, 6)){ + queryVerticesSizes.put(queries.get(queryNumber).getBackendQuery(), alwaysUseVertices.size()); + } + queryVerticesSizes.put(queries.get(1).getBackendQuery(), alwaysUseVertices.size() + 3); + queryVerticesSizes.put(queries.get(2).getBackendQuery(), alwaysUseVertices.size() + 5); + queryVerticesSizes.put(queries.get(3).getBackendQuery(), alwaysUseVertices.size() + 5); + queryVerticesSizes.put(queries.get(5).getBackendQuery(), alwaysUseVertices.size() + 1); + queryVerticesSizes.put(queries.get(7).getBackendQuery(), alwaysUseVertices.size() + 4); + queryVerticesSizes.put(queries.get(8).getBackendQuery(), alwaysUseVertices.size() + 1); + queryVerticesSizes.put(queries.get(9).getBackendQuery(), alwaysUseVertices.size() + 1); + + assertQueriesVertexSizes(result, queryVerticesSizes); + + result.getQueryGroups().forEach(group -> { + if(group.getQueries().contains(queries.get(0).getBackendQuery())){ + assertEquals(3, group.getQueries().size()); + assertTrue(group.getQueries().contains(queries.get(4).getBackendQuery())); + assertTrue(group.getQueries().contains(queries.get(6).getBackendQuery())); + } + if(group.getQueries().contains(queries.get(1).getBackendQuery())){ + assertEquals(1, group.getQueries().size()); + } + if(group.getQueries().contains(queries.get(2).getBackendQuery())){ + assertEquals(2, group.getQueries().size()); + assertTrue(group.getQueries().contains(queries.get(3).getBackendQuery())); + } + if(group.getQueries().contains(queries.get(5).getBackendQuery())){ + assertEquals(2, group.getQueries().size()); + assertTrue(group.getQueries().contains(queries.get(9).getBackendQuery())); + } + if(group.getQueries().contains(queries.get(7).getBackendQuery())){ + assertEquals(1, group.getQueries().size()); + } + if(group.getQueries().contains(queries.get(8).getBackendQuery())){ + assertEquals(1, group.getQueries().size()); + } + }); + } + + @Test + public void testGroupingByQueriesTwoElementsDifference() { + // this module doesn't have JUnit platform, thus, we are not using @ParametrizedTest here. + testGroupingByQueriesTwoElementsDifference(true); + testGroupingByQueriesTwoElementsDifference(false); + } + + private void testGroupingByQueriesTwoElementsDifference(boolean useFirst){ + ArrayList> queries = new ArrayList<>(); + queries.add(new BackendQueryHolder<>(mock(SliceQuery.class), false, false)); + queries.add(new BackendQueryHolder<>(mock(SliceQuery.class), false, false)); + + ArrayList allVertices = new ArrayList<>(); + InternalVertex vertex1 = null; + InternalVertex vertex2 = null; + + if(useFirst){ + vertex1 = vertex(queries.get(0).getBackendQuery()); + vertex2 = vertex(queries.get(1).getBackendQuery()); + allVertices.add(vertex1); + allVertices.add(vertex2); + } + for(int i=0;i<10;i++){ + allVertices.add(alwaysUseVertex()); + } + if(!useFirst){ + vertex1 = vertex(queries.get(0).getBackendQuery()); + vertex2 = vertex(queries.get(1).getBackendQuery()); + allVertices.add(vertex1); + allVertices.add(vertex2); + } + + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + allVertices, queries); + + assertEquals(2, result.getQueryGroups().size()); + assertEquals(allVertices.size(), result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(queries.size(), result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(2, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + assertGroupedByUniqueQueriesAndVertexSets(result); + + Map queryVerticesSizes = new HashMap<>(); + queryVerticesSizes.put(queries.get(0).getBackendQuery(), allVertices.size() - 1); + queryVerticesSizes.put(queries.get(1).getBackendQuery(), allVertices.size() - 1); + assertQueriesVertexSizes(result, queryVerticesSizes); + + for(KeysQueriesGroup group : result.getQueryGroups()){ + if(group.getQueries().contains(queries.get(0).getBackendQuery())){ + assertEquals(1, group.getQueries().size()); + assertTrue(group.getKeysGroup().contains(vertex1.id())); + assertFalse(group.getKeysGroup().contains(vertex2.id())); + } + if(group.getQueries().contains(queries.get(1).getBackendQuery())){ + assertEquals(1, group.getQueries().size()); + assertTrue(group.getKeysGroup().contains(vertex2.id())); + assertFalse(group.getKeysGroup().contains(vertex1.id())); + } + } + } + + @Test + public void testGroupingByQueriesLastOneElementDifference(){ + ArrayList> queries = new ArrayList<>(); + queries.add(new BackendQueryHolder<>(mock(SliceQuery.class), false, false)); + queries.add(new BackendQueryHolder<>(mock(SliceQuery.class), false, false)); + + ArrayList allVertices = new ArrayList<>(); + + for(int i=0;i<10;i++){ + allVertices.add(alwaysUseVertex()); + } + InternalVertex vertex1 = vertex(queries.get(0).getBackendQuery()); + allVertices.add(vertex1); + + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + allVertices, queries); + + assertEquals(2, result.getQueryGroups().size()); + assertEquals(allVertices.size(), result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(queries.size(), result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(1, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + assertGroupedByUniqueQueriesAndVertexSets(result); + + Map queryVerticesSizes = new HashMap<>(); + queryVerticesSizes.put(queries.get(0).getBackendQuery(), allVertices.size()); + queryVerticesSizes.put(queries.get(1).getBackendQuery(), allVertices.size() - 1); + assertQueriesVertexSizes(result, queryVerticesSizes); + + for(KeysQueriesGroup group : result.getQueryGroups()){ + if(group.getQueries().contains(queries.get(0).getBackendQuery())){ + assertEquals(1, group.getQueries().size()); + assertTrue(group.getKeysGroup().contains(vertex1.id())); + assertEquals(allVertices.size(), group.getKeysGroup().size()); + } + if(group.getQueries().contains(queries.get(1).getBackendQuery())){ + assertEquals(1, group.getQueries().size()); + assertFalse(group.getKeysGroup().contains(vertex1.id())); + assertEquals(allVertices.size()-1, group.getKeysGroup().size()); + } + } + } + + @Test + public void testGroupingVertexMultipleQueries() { + MultiKeysQueryGroups result = MultiSliceQueriesGroupingUtil.toMultiKeysQueryGroups( + Collections.singletonList(alwaysUseVertex()), + Arrays.asList(new BackendQueryHolder<>(mock(SliceQuery.class), false, false), + new BackendQueryHolder<>(mock(SliceQuery.class), false, false), + new BackendQueryHolder<>(mock(SliceQuery.class), false, false))); + assertEquals(1, result.getQueryGroups().size()); + assertEquals(3, result.getQueryGroups().get(0).getQueries().size()); + assertEquals(1, result.getQueryGroups().get(0).getKeysGroup().size()); + assertEquals(1, result.getMultiQueryContext().getAllKeysArr().length); + assertEquals(3, result.getMultiQueryContext().getTotalAmountOfQueries()); + assertEquals(1, result.getMultiQueryContext().getAllLeafParents().size()); + assertContextIsValid(result.getMultiQueryContext()); + } + + ///// Nodes replacement testing + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodesEmptyList() { + List allLeafParents = Collections.emptyList(); + Map oldToNewKeysMap = new HashMap<>(); + MultiSliceQueriesGroupingUtil.replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + assertTrue(allLeafParents.isEmpty()); + assertTrue(oldToNewKeysMap.isEmpty()); + } + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodeBothChildrenDataTreeNode() { + // Create leaf nodes with both children being instances of DataTreeNode + DataTreeNode> leftChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(1), Collections.emptyList())); + DataTreeNode> rightChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(2), Collections.emptyList())); + TreeNode parentNode = new TreeNode(leftChild, rightChild); + List allLeafParents = Collections.singletonList(parentNode); + + // Create a mapping from Integer to String + Map oldToNewKeysMap = new HashMap<>(); + oldToNewKeysMap.put(1, "one"); + oldToNewKeysMap.put(2, "two"); + + replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + + // Verify that both children are replaced with updated type leaf nodes + assertTrue(parentNode.left instanceof DataTreeNode); + assertTrue(parentNode.right instanceof DataTreeNode); + assertEquals("one", ((DataTreeNode) parentNode.left).data.getKeysGroup().get(0)); + assertEquals("two", ((DataTreeNode) parentNode.right).data.getKeysGroup().get(0)); + } + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodesOnlyLeftChildDataTreeNode() { + // Create a leaf node with only the left child being an instance of DataTreeNode + DataTreeNode> leftChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(1), Collections.emptyList())); + TreeNode parentNode = new TreeNode(leftChild, new TreeNode()); + List allLeafParents = Collections.singletonList(parentNode); + + // Create a mapping from Integer to String + Map oldToNewKeysMap = new HashMap<>(); + oldToNewKeysMap.put(1, "one"); + + replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + + // Verify that only the left child is replaced with an updated type leaf node + assertTrue(parentNode.left instanceof DataTreeNode); + assertFalse(parentNode.right instanceof DataTreeNode); + assertEquals("one", ((DataTreeNode) parentNode.left).data.getKeysGroup().get(0)); + } + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodesOnlyRightChildDataTreeNode() { + // Create a leaf node with only the right child being an instance of DataTreeNode + TreeNode parentNode = new TreeNode(new TreeNode(), new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(2), Collections.emptyList()))); + List allLeafParents = Collections.singletonList(parentNode); + + // Create a mapping from Integer to String + Map oldToNewKeysMap = new HashMap<>(); + oldToNewKeysMap.put(2, "two"); + + replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + + // Verify that only the right child is replaced with an updated type leaf node + assertFalse(parentNode.left instanceof DataTreeNode); + assertTrue(parentNode.right instanceof DataTreeNode); + assertEquals("two", ((DataTreeNode) parentNode.right).data.getKeysGroup().get(0)); + } + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodesNoDataTreeNodeChildren() { + // Create leaf nodes with both children not being instances of DataTreeNode + TreeNode parentNode = new TreeNode(new TreeNode(), new TreeNode()); + List allLeafParents = Collections.singletonList(parentNode); + + // Create a mapping from Integer to String + Map oldToNewKeysMap = new HashMap<>(); + oldToNewKeysMap.put(1, "one"); + oldToNewKeysMap.put(2, "two"); + + replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + + // Verify that no modifications occur + assertFalse(parentNode.left instanceof DataTreeNode); + assertFalse(parentNode.right instanceof DataTreeNode); + } + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodesEmptyMap() { + // Create leaf nodes with both children being instances of DataTreeNode + DataTreeNode> leftChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(1), Collections.emptyList())); + DataTreeNode> rightChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(2), Collections.emptyList())); + TreeNode parentNode = new TreeNode(leftChild, rightChild); + List allLeafParents = Collections.singletonList(parentNode); + + // Empty mapping + Map oldToNewKeysMap = Collections.emptyMap(); + + replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + + // Verify that no modifications occur + assertTrue(parentNode.left instanceof DataTreeNode); + assertTrue(parentNode.right instanceof DataTreeNode); + assertNull(((DataTreeNode) parentNode.left).data.getKeysGroup().get(0)); + assertNull(((DataTreeNode) parentNode.right).data.getKeysGroup().get(0)); + } + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodesKeysNotPresentInData() { + // Create leaf nodes with both children being instances of DataTreeNode + DataTreeNode> leftChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(1), Collections.emptyList())); + DataTreeNode> rightChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(2), Collections.emptyList())); + TreeNode parentNode = new TreeNode(leftChild, rightChild); + List allLeafParents = Collections.singletonList(parentNode); + + // Create a mapping from Integer to String + Map oldToNewKeysMap = new HashMap<>(); + oldToNewKeysMap.put(3, "three"); + + replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + + // Verify that no modifications occur + assertTrue(parentNode.left instanceof DataTreeNode); + assertTrue(parentNode.right instanceof DataTreeNode); + assertNull(((DataTreeNode) parentNode.left).data.getKeysGroup().get(0)); + assertNull(((DataTreeNode) parentNode.right).data.getKeysGroup().get(0)); + } + + @Test + public void testReplaceCurrentLeafNodeWithUpdatedTypeLeafNodesValidMapping() { + // Create leaf nodes with both children being instances of DataTreeNode + DataTreeNode> leftChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(1), Collections.emptyList())); + DataTreeNode> rightChild = new DataTreeNode<>(new KeysQueriesGroup<>(Collections.singletonList(2), Collections.emptyList())); + TreeNode parentNode = new TreeNode(leftChild, rightChild); + List allLeafParents = Collections.singletonList(parentNode); + + // Create a mapping from Integer to String + Map oldToNewKeysMap = new HashMap<>(); + oldToNewKeysMap.put(1, "one"); + oldToNewKeysMap.put(2, "two"); + + replaceCurrentLeafNodeWithUpdatedTypeLeafNodes(allLeafParents, oldToNewKeysMap); + + // Verify that both children are replaced with updated type leaf nodes + assertTrue(parentNode.left instanceof DataTreeNode); + assertTrue(parentNode.right instanceof DataTreeNode); + assertEquals("one", ((DataTreeNode) parentNode.left).data.getKeysGroup().get(0)); + assertEquals("two", ((DataTreeNode) parentNode.right).data.getKeysGroup().get(0)); + } + + // Queries movement testing + + @Test + public void testMoveQueriesToNewLeafNodeEmptyUpdatedQueryGroup() { + List>> updatedQueryGroup = Collections.emptyList(); + Integer[] allVertices = {1, 2, 3}; + TreeNode groupingRootTreeNode = new TreeNode(null, new TreeNode()); + List> remainingQueryGroups = new ArrayList<>(); + + moveQueriesToNewLeafNode(updatedQueryGroup, allVertices, groupingRootTreeNode, remainingQueryGroups); + + assertNull(groupingRootTreeNode.left); + assertNull(groupingRootTreeNode.right.right); + assertNull(groupingRootTreeNode.right.left); + assertTrue(remainingQueryGroups.isEmpty()); + } + + @Test + public void testMoveQueriesToNewLeafNodeMatchingVertices() { + // Create leaf nodes with matching vertices + TreeNode leftChild = new TreeNode(); + TreeNode rightChild = new TreeNode(); + TreeNode groupingRootTreeNode = new TreeNode(leftChild, rightChild); + List>> updatedQueryGroup = new ArrayList<>(); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Arrays.asList(1, 2))); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Collections.singletonList(3))); + Integer[] allVertices = {1, 2, 3}; + List> remainingQueryGroups = new ArrayList<>(); + + moveQueriesToNewLeafNode(updatedQueryGroup, allVertices, groupingRootTreeNode, remainingQueryGroups); + + // Verify that the queries are added to the corresponding leaf nodes + assertEquals(1, ((DataTreeNode) leftChild.left.right).data.getQueries().size()); + assertEquals(1, ((DataTreeNode) rightChild.right.left).data.getQueries().size()); + assertEquals(2, remainingQueryGroups.size()); + } + + @Test + public void testMoveQueriesToNewLeafNodeNewChainNodes() { + // Create a leaf node without matching vertices + List>> updatedQueryGroup = new ArrayList<>(); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Collections.singletonList(1))); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Collections.singletonList(2))); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Collections.singletonList(3))); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Arrays.asList(1,2,3))); + + Integer[] allVertices = {1, 2, 3}; + List> remainingQueryGroups = new ArrayList<>(); + KeysQueriesGroup existingQueryData = new KeysQueriesGroup<>(Arrays.asList(allVertices), new ArrayList<>()); + remainingQueryGroups.add(existingQueryData); + TreeNode root = new TreeNode(null, new TreeNode(null, new TreeNode(null, new DataTreeNode<>(existingQueryData)))); + + moveQueriesToNewLeafNode(updatedQueryGroup, allVertices, root, remainingQueryGroups); + + // Verify that new chain nodes are generated and queries are added to the corresponding leaf nodes + assertEquals(3, ((DataTreeNode>) root.right.right.right).data.getKeysGroup().size()); + assertTrue(((DataTreeNode>) root.right.right.right).data.getKeysGroup().containsAll(Arrays.asList(allVertices))); + assertEquals(1, ((DataTreeNode>) root.right.right.right).data.getQueries().size()); + assertEquals(updatedQueryGroup.get(3).getKey(), ((DataTreeNode>) root.right.right.right).data.getQueries().iterator().next()); + assertEquals(existingQueryData, ((DataTreeNode>) root.right.right.right).data); + + assertEquals(1, ((DataTreeNode>) root.right.left.left).data.getQueries().size()); + assertEquals(1, ((DataTreeNode>) root.left.right.left).data.getQueries().size()); + assertEquals(1, ((DataTreeNode>) root.left.left.right).data.getQueries().size()); + assertEquals(updatedQueryGroup.get(0).getKey(), ((DataTreeNode>) root.right.left.left).data.getQueries().iterator().next()); + assertEquals(updatedQueryGroup.get(1).getKey(), ((DataTreeNode>) root.left.right.left).data.getQueries().iterator().next()); + assertEquals(updatedQueryGroup.get(2).getKey(), ((DataTreeNode>) root.left.left.right).data.getQueries().iterator().next()); + assertNull(root.right.right.left); + assertNull(root.right.left.right); + assertNull(root.left.right.right); + assertNull(root.left.left.left); + + assertEquals(4, remainingQueryGroups.size()); + assertTrue(remainingQueryGroups.containsAll(Arrays.asList( + ((DataTreeNode>) root.right.right.right).data, + ((DataTreeNode>) root.right.left.left).data, + ((DataTreeNode>) root.left.right.left).data, + ((DataTreeNode>) root.left.left.right).data + ))); + } + + @Test + public void testMoveQueriesToNewLeafNodeEmptyRemainingQueryGroups() { + TreeNode leaf1 = new TreeNode(); + TreeNode leaf2 = new TreeNode(); + TreeNode groupingRootTreeNode = new TreeNode(leaf1, leaf2); + List>> updatedQueryGroup = new ArrayList<>(); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Arrays.asList(1, 2))); + Integer[] allVertices = {1, 2, 3}; + List> remainingQueryGroups = new ArrayList<>(); + + moveQueriesToNewLeafNode(updatedQueryGroup, allVertices, groupingRootTreeNode, remainingQueryGroups); + + assertEquals(1, remainingQueryGroups.size()); + } + + @Test + public void testMoveQueriesToNewLeafNodeNonEmptyRemainingQueryGroups() { + List>> updatedQueryGroup = new ArrayList<>(); + updatedQueryGroup.add(Pair.of(mock(SliceQuery.class), Collections.singletonList(1))); + Integer[] allVertices = {1}; + List> remainingQueryGroups = new ArrayList<>(); + remainingQueryGroups.add(new KeysQueriesGroup<>(Collections.emptyList(), Collections.emptyList())); + TreeNode groupingRootTreeNode = new TreeNode(new DataTreeNode<>(remainingQueryGroups.get(0)), null); + + moveQueriesToNewLeafNode(updatedQueryGroup, allVertices, groupingRootTreeNode, remainingQueryGroups); + + assertEquals(2, remainingQueryGroups.size()); + assertEquals(1, remainingQueryGroups.get(1).getKeysGroup().size()); + assertTrue(remainingQueryGroups.get(1).getKeysGroup().contains(1)); + assertEquals(1, remainingQueryGroups.get(1).getQueries().size()); + assertTrue(remainingQueryGroups.get(1).getQueries().contains(updatedQueryGroup.get(0).getKey())); + } + + private void assertContextIsValid(MultiQueriesByKeysGroupsContext context){ + assertTrue(context.getAllLeafParents().size() <= context.getTotalAmountOfQueries()); + + Object[] keys = context.getAllKeysArr(); + Set keysSet = new HashSet<>(Arrays.asList(keys)); + assertEquals(keysSet.size(), keys.length); + + Set allLeafDepths = new HashSet<>(1); + addAllLeafDepths(context.getGroupingRootTreeNode(), 0, allLeafDepths); + if(keysSet.isEmpty()){ + assertTrue(allLeafDepths.isEmpty()); + } else { + assertEquals(1, allLeafDepths.size()); + Integer calculatedTreeDepth = allLeafDepths.iterator().next(); + assertEquals(keys.length, calculatedTreeDepth); + } + + List calculatedLeafNodes = new ArrayList<>(); + addAllLeafNodes(context.getGroupingRootTreeNode(), calculatedLeafNodes); + assertTrue(calculatedLeafNodes.size() >= context.getAllLeafParents().size()); + + for(DataTreeNode leafNode : calculatedLeafNodes){ + boolean childOfLeafParents = false; + for(TreeNode parentNode : context.getAllLeafParents()){ + if(parentNode.left == leafNode || parentNode.right == leafNode){ + assertFalse(childOfLeafParents, "This child node was already a child of another parent. " + + "Each parent should have unique children nodes."); + childOfLeafParents = true; + } + } + assertTrue(childOfLeafParents, "Calculated leaf node is not a child of any of the context's leaf parent nodes."); + } + } + + private void addAllLeafDepths(TreeNode node, int currentDepth, Set depths){ + if(node == null){ + return; + } + if(node instanceof DataTreeNode){ + assertNull(node.left); + assertNull(node.right); + depths.add(currentDepth); + return; + } + addAllLeafDepths(node.left, currentDepth+1, depths); + addAllLeafDepths(node.right, currentDepth+1, depths); + } + + private void addAllLeafNodes(TreeNode node, List leafNodes){ + if(node == null){ + return; + } + if(node instanceof DataTreeNode){ + assertNull(node.left); + assertNull(node.right); + leafNodes.add((DataTreeNode) node); + return; + } + addAllLeafNodes(node.left, leafNodes); + addAllLeafNodes(node.right, leafNodes); + } + + private StandardVertex nonCacheVertex(){ + return mock(StandardVertex.class); + } + + private CacheVertex vertex(SliceQuery ... queriesWithHasLoadedRelations){ + return mockVertex(CacheVertex.class, queriesWithHasLoadedRelations); + } + + private CacheVertex alwaysSkipVertex(){ + return mockVertex(CacheVertex.class, true); + } + + private CacheVertex alwaysUseVertex(){ + return mockVertex(CacheVertex.class, false); + } + + private T mockVertex(Class type, boolean hasLoadedRelations){ + T vertex = mock(type); + doReturn(hasLoadedRelations).when(vertex).hasLoadedRelations(any()); + doReturn(false).when(vertex).isNew(); + doReturn(true).when(vertex).hasId(); + doReturn(VERTEX_ID.incrementAndGet()).when(vertex).id(); + doReturn(hasLoadedRelations).when(vertex).hasLoadedRelations(any()); + return vertex; + } + + private T mockVertex(Class type, SliceQuery ... queriesWithHasLoadedRelations){ + T vertex = mock(type); + doReturn(false).when(vertex).isNew(); + doReturn(true).when(vertex).hasId(); + doReturn(VERTEX_ID.incrementAndGet()).when(vertex).id(); + Set falseQueries = new HashSet<>(Arrays.asList(queriesWithHasLoadedRelations)); + doAnswer(invocationOnMock -> { + return !falseQueries.contains(invocationOnMock.getArgument(0)); + }).when(vertex).hasLoadedRelations(any()); + return vertex; + } + + + private void assertQueriesVertexSizes(MultiKeysQueryGroups result, Map queryVerticesSizes){ + result.getQueryGroups().forEach(group -> group.getQueries().forEach(query -> assertEquals(queryVerticesSizes.get(query), group.getKeysGroup().size()))); + } + + private void assertGroupedByUniqueQueriesAndVertexSets(MultiKeysQueryGroups result){ + + // Check all slice queries are unique across all collections + result.getQueryGroups().forEach(group -> { + group.getQueries().forEach(query -> { + MutableBoolean duplicate = new MutableBoolean(false); + result.getQueryGroups().forEach(group2 -> { + group2.getQueries().forEach(query2 -> { + if(query == query2){ + assertFalse(duplicate.booleanValue()); + duplicate.setTrue(); + } + }); + }); + }); + }); + + // Check all vertex sets are unique across all pairs + result.getQueryGroups().forEach(group -> { + MutableBoolean duplicate = new MutableBoolean(false); + result.getQueryGroups().forEach(group2 -> { + if(group.getKeysGroup().size() == group2.getKeysGroup().size() && group.getKeysGroup().containsAll(group2.getKeysGroup())){ + assertFalse(duplicate.booleanValue()); + duplicate.setTrue(); + } + }); + }); + } + +} diff --git a/janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLKeyColumnValueStore.java b/janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLKeyColumnValueStore.java index ceb1dec7d6..4a10023655 100644 --- a/janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLKeyColumnValueStore.java +++ b/janusgraph-cql/src/main/java/org/janusgraph/diskstorage/cql/CQLKeyColumnValueStore.java @@ -53,12 +53,15 @@ import org.janusgraph.diskstorage.keycolumnvalue.KeyRangeQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySlicesIterator; +import org.janusgraph.diskstorage.keycolumnvalue.KeysQueriesGroup; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.diskstorage.keycolumnvalue.MultiSlicesQuery; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; import org.janusgraph.diskstorage.util.CompletableFutureUtil; import org.janusgraph.diskstorage.util.backpressure.QueryBackPressure; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -400,6 +403,32 @@ public Map getSlice(final List keys, fina } } + // This implementation is better optimized than `KeyColumnValueStoreUtil.getMultiRangeSliceNonOptimized(this, multiSliceQueriesForKeys, txh);` + // because it sends all slice queries in parallel instead of using blocking calls to + // `getSlice(final Collection keys, final SliceQuery query, final StoreTransaction txh)` + @Override + public Map> getMultiSlices(MultiKeysQueryGroups multiKeysQueryGroups, StoreTransaction txh) throws BackendException { + try { + Map>> futureResult = new HashMap<>(multiKeysQueryGroups.getMultiQueryContext().getTotalAmountOfQueries()); + for(KeysQueriesGroup queriesForKeysPair : multiKeysQueryGroups.getQueryGroups()){ + Collection keys = queriesForKeysPair.getKeysGroup(); + //TODO: instead of using a separate query for the same keys it would be great to chain all slice queries for the same + // key into a single CQL query. This could result in a better performance, but it's not yet possible + // because CQL doesn't have `OR` operator and batching multiple CQL `select` queries isn't possible right now. + // See issue: https://github.com/JanusGraph/janusgraph/issues/3816 + for(SliceQuery query : queriesForKeysPair.getQueries()){ + Map> futureQueryResult = futureResult.computeIfAbsent(query, sliceQuery -> new HashMap<>(keys.size())); + for(StaticBuffer key : keys){ + futureQueryResult.put(key, cqlSliceFunction.getSlice(new KeySliceQuery(key, query), txh)); + } + } + } + return CompletableFutureUtil.unwrapMapOfMaps(futureResult); + } catch (Throwable e) { + throw EXCEPTION_MAPPER.apply(e); + } + } + public BatchableStatement deleteColumn(final StaticBuffer key, final StaticBuffer column) { return deleteColumn(key, column, null); } diff --git a/janusgraph-inmemory/src/test/java/org/janusgraph/graphdb/inmemory/InMemoryGraphTest.java b/janusgraph-inmemory/src/test/java/org/janusgraph/graphdb/inmemory/InMemoryGraphTest.java index 20cf6efd07..6d7b6d77ed 100644 --- a/janusgraph-inmemory/src/test/java/org/janusgraph/graphdb/inmemory/InMemoryGraphTest.java +++ b/janusgraph-inmemory/src/test/java/org/janusgraph/graphdb/inmemory/InMemoryGraphTest.java @@ -63,6 +63,9 @@ public void testLimitBatchSizeForMultiQueryRepeatStep() {} @Override @Test @Disabled public void testLimitBatchSizeForMultiQuery() {} + @Override @Test @Disabled + public void testMultiSliceDBCachedRequests(){} + @Override @Test @Disabled public void testMaskableGraphConfig() {} diff --git a/janusgraph-test/src/main/java/org/janusgraph/diskstorage/cache/KCVSCacheTest.java b/janusgraph-test/src/main/java/org/janusgraph/diskstorage/cache/KCVSCacheTest.java index 032d1924d7..f5a4ee2487 100644 --- a/janusgraph-test/src/main/java/org/janusgraph/diskstorage/cache/KCVSCacheTest.java +++ b/janusgraph-test/src/main/java/org/janusgraph/diskstorage/cache/KCVSCacheTest.java @@ -25,6 +25,7 @@ import org.janusgraph.diskstorage.keycolumnvalue.KeyRangeQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.KeySlicesIterator; +import org.janusgraph.diskstorage.keycolumnvalue.MultiKeysQueryGroups; import org.janusgraph.diskstorage.keycolumnvalue.MultiSlicesQuery; import org.janusgraph.diskstorage.keycolumnvalue.SliceQuery; import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; @@ -210,6 +211,12 @@ public Map getSlice(List keys, SliceQuery return store.getSlice(keys,query,txh); } + @Override + public Map> getMultiSlices(MultiKeysQueryGroups multiKeysQueryGroups, StoreTransaction txh) throws BackendException { + multiKeysQueryGroups.getQueryGroups().forEach(group -> getSliceCounter.addAndGet(group.getQueries().size())); + return store.getMultiSlices(multiKeysQueryGroups,txh); + } + @Override public void mutate(StaticBuffer key, List additions, List deletions, StoreTransaction txh) throws BackendException { store.mutate(key,additions,deletions,txh);