|
| 1 | +/* |
| 2 | + * Copyright © 2021-present Arcade Data Ltd (info@arcadedata.com) |
| 3 | + * |
| 4 | + * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | + * you may not use this file except in compliance with the License. |
| 6 | + * You may obtain a copy of the License at |
| 7 | + * |
| 8 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | + * |
| 10 | + * Unless required by applicable law or agreed to in writing, software |
| 11 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | + * See the License for the specific language governing permissions and |
| 14 | + * limitations under the License. |
| 15 | + * |
| 16 | + * SPDX-FileCopyrightText: 2021-present Arcade Data Ltd (info@arcadedata.com) |
| 17 | + * SPDX-License-Identifier: Apache-2.0 |
| 18 | + */ |
| 19 | +package com.arcadedb.query.sql.executor; |
| 20 | + |
| 21 | +import com.arcadedb.database.Document; |
| 22 | +import com.arcadedb.database.Identifiable; |
| 23 | +import com.arcadedb.database.RID; |
| 24 | +import com.arcadedb.exception.RecordNotFoundException; |
| 25 | +import com.arcadedb.exception.TimeoutException; |
| 26 | +import com.arcadedb.graph.GraphTraversalProvider; |
| 27 | +import com.arcadedb.graph.Vertex; |
| 28 | + |
| 29 | +import java.util.ArrayList; |
| 30 | +import java.util.List; |
| 31 | +import java.util.NoSuchElementException; |
| 32 | + |
| 33 | +/** |
| 34 | + * Fused multi-hop MATCH step that uses CSR arrays to traverse a chain of consecutive |
| 35 | + * out/in hops in a single DFS pass, without materializing intermediate Result objects. |
| 36 | + * |
| 37 | + * @author Luca Garulli (l.garulli@arcadedata.com) |
| 38 | + * <p> |
| 39 | + * For example, in a MATCH pattern like: |
| 40 | + * <pre> |
| 41 | + * {as: a} -KNOWS-> {as: b} -FOLLOWS-> {as: c} -LIKES-> {as: d} |
| 42 | + * </pre> |
| 43 | + * Instead of creating 3 separate MatchStep/MatchEdgeTraverser instances (each materializing |
| 44 | + * a full ResultInternal per hop), this step traverses the entire chain using int[] nodeId |
| 45 | + * arrays and only materializes vertex results at each alias binding point. |
| 46 | + */ |
| 47 | +class MatchGAVFusedStep extends AbstractExecutionStep { |
| 48 | + private final List<EdgeTraversal> edges; |
| 49 | + private final GraphTraversalProvider provider; |
| 50 | + private ResultSet upstream; |
| 51 | + private Result lastUpstreamRecord; |
| 52 | + private List<ResultInternal> bufferedResults; |
| 53 | + private int bufferIndex; |
| 54 | + |
| 55 | + MatchGAVFusedStep(final CommandContext context, final List<EdgeTraversal> edges, final GraphTraversalProvider provider) { |
| 56 | + super(context); |
| 57 | + this.edges = edges; |
| 58 | + this.provider = provider; |
| 59 | + } |
| 60 | + |
| 61 | + @Override |
| 62 | + public void reset() { |
| 63 | + this.upstream = null; |
| 64 | + this.lastUpstreamRecord = null; |
| 65 | + this.bufferedResults = null; |
| 66 | + this.bufferIndex = 0; |
| 67 | + } |
| 68 | + |
| 69 | + @Override |
| 70 | + public ResultSet syncPull(final CommandContext context, final int nRecords) throws TimeoutException { |
| 71 | + return new ResultSet() { |
| 72 | + int localCount = 0; |
| 73 | + |
| 74 | + @Override |
| 75 | + public boolean hasNext() { |
| 76 | + if (localCount >= nRecords) |
| 77 | + return false; |
| 78 | + return fetchNextIfNeeded(); |
| 79 | + } |
| 80 | + |
| 81 | + @Override |
| 82 | + public Result next() { |
| 83 | + if (localCount >= nRecords) |
| 84 | + throw new NoSuchElementException(); |
| 85 | + if (!fetchNextIfNeeded()) |
| 86 | + throw new NoSuchElementException(); |
| 87 | + localCount++; |
| 88 | + return bufferedResults.get(bufferIndex++); |
| 89 | + } |
| 90 | + }; |
| 91 | + } |
| 92 | + |
| 93 | + private boolean fetchNextIfNeeded() { |
| 94 | + // If we have buffered results remaining, return them |
| 95 | + if (bufferedResults != null && bufferIndex < bufferedResults.size()) |
| 96 | + return true; |
| 97 | + |
| 98 | + // Fetch next upstream record and compute all fused results |
| 99 | + while (true) { |
| 100 | + if (upstream == null || !upstream.hasNext()) |
| 101 | + upstream = getPrev().syncPull(context, 100); |
| 102 | + if (!upstream.hasNext()) |
| 103 | + return false; |
| 104 | + |
| 105 | + lastUpstreamRecord = upstream.next(); |
| 106 | + bufferedResults = computeFusedResults(lastUpstreamRecord); |
| 107 | + bufferIndex = 0; |
| 108 | + |
| 109 | + if (!bufferedResults.isEmpty()) |
| 110 | + return true; |
| 111 | + } |
| 112 | + } |
| 113 | + |
| 114 | + /** |
| 115 | + * Executes the fused chain traversal for a single upstream record using CSR DFS. |
| 116 | + */ |
| 117 | + private List<ResultInternal> computeFusedResults(final Result upstream) { |
| 118 | + final List<ResultInternal> results = new ArrayList<>(); |
| 119 | + |
| 120 | + // Get the starting point from the first edge's source alias |
| 121 | + final String startAlias = edges.get(0).out ? edges.get(0).edge.out.alias : edges.get(0).edge.in.alias; |
| 122 | + Identifiable startingElem = upstream.getElementProperty(startAlias); |
| 123 | + if (startingElem instanceof Result r) |
| 124 | + startingElem = r.getElement().orElse(null); |
| 125 | + if (startingElem == null) |
| 126 | + return results; |
| 127 | + |
| 128 | + final int startNodeId = provider.getNodeId(startingElem.getIdentity()); |
| 129 | + if (startNodeId < 0) |
| 130 | + return results; // Not in CSR, caller should fall back to standard path |
| 131 | + |
| 132 | + // Build direction and edge label arrays for the chain |
| 133 | + final int chainLen = edges.size(); |
| 134 | + final Vertex.DIRECTION[] directions = new Vertex.DIRECTION[chainLen]; |
| 135 | + final String[][] edgeLabelsPerHop = new String[chainLen][]; |
| 136 | + final String[] aliases = new String[chainLen]; // alias bound at each hop's endpoint |
| 137 | + |
| 138 | + for (int i = 0; i < chainLen; i++) { |
| 139 | + final EdgeTraversal et = edges.get(i); |
| 140 | + directions[i] = et.out ? Vertex.DIRECTION.OUT : Vertex.DIRECTION.IN; |
| 141 | + // Extract edge labels from the method call parameters |
| 142 | + final var methodParams = et.edge.item.getMethod().params; |
| 143 | + if (methodParams != null && !methodParams.isEmpty()) { |
| 144 | + final String[] labels = new String[methodParams.size()]; |
| 145 | + for (int p = 0; p < methodParams.size(); p++) |
| 146 | + labels[p] = methodParams.get(p).toString().replace("'", "").replace("\"", "").trim(); |
| 147 | + edgeLabelsPerHop[i] = labels; |
| 148 | + } else { |
| 149 | + edgeLabelsPerHop[i] = new String[0]; |
| 150 | + } |
| 151 | + // The alias for this hop's endpoint |
| 152 | + aliases[i] = et.out ? et.edge.in.alias : et.edge.out.alias; |
| 153 | + } |
| 154 | + |
| 155 | + // DFS through CSR, collecting path nodeIds |
| 156 | + final int[] pathNodeIds = new int[chainLen + 1]; |
| 157 | + pathNodeIds[0] = startNodeId; |
| 158 | + dfs(pathNodeIds, 0, chainLen, directions, edgeLabelsPerHop, aliases, upstream, results); |
| 159 | + |
| 160 | + return results; |
| 161 | + } |
| 162 | + |
| 163 | + /** |
| 164 | + * Recursive DFS through CSR arrays. Materializes a ResultInternal only at the final hop, |
| 165 | + * carrying all intermediate alias bindings. |
| 166 | + */ |
| 167 | + private void dfs(final int[] pathNodeIds, final int hop, final int chainLen, |
| 168 | + final Vertex.DIRECTION[] directions, final String[][] edgeLabelsPerHop, |
| 169 | + final String[] aliases, final Result upstream, final List<ResultInternal> results) { |
| 170 | + final int currentNodeId = pathNodeIds[hop]; |
| 171 | + final int[] neighbors = edgeLabelsPerHop[hop].length == 0 |
| 172 | + ? provider.getNeighborIds(currentNodeId, directions[hop]) |
| 173 | + : provider.getNeighborIds(currentNodeId, directions[hop], edgeLabelsPerHop[hop]); |
| 174 | + |
| 175 | + for (final int neighborId : neighbors) { |
| 176 | + pathNodeIds[hop + 1] = neighborId; |
| 177 | + |
| 178 | + if (hop == chainLen - 1) { |
| 179 | + // Final hop: materialize the full result with all alias bindings |
| 180 | + final ResultInternal result = buildResult(pathNodeIds, aliases, upstream); |
| 181 | + if (result != null) |
| 182 | + results.add(result); |
| 183 | + } else { |
| 184 | + // Intermediate hop: continue DFS with raw int IDs |
| 185 | + dfs(pathNodeIds, hop + 1, chainLen, directions, edgeLabelsPerHop, aliases, upstream, results); |
| 186 | + } |
| 187 | + } |
| 188 | + } |
| 189 | + |
| 190 | + /** |
| 191 | + * Builds a ResultInternal from the CSR path, binding each alias to its vertex. |
| 192 | + */ |
| 193 | + private ResultInternal buildResult(final int[] pathNodeIds, final String[] aliases, final Result upstream) { |
| 194 | + final ResultInternal result = new ResultInternal(context.getDatabase()); |
| 195 | + |
| 196 | + // Copy all properties from upstream |
| 197 | + for (final String prop : upstream.getPropertyNames()) |
| 198 | + result.setProperty(prop, upstream.getProperty(prop)); |
| 199 | + |
| 200 | + // Bind each hop's endpoint alias |
| 201 | + for (int i = 0; i < aliases.length; i++) { |
| 202 | + final RID rid = provider.getRID(pathNodeIds[i + 1]); |
| 203 | + if (rid == null) |
| 204 | + return null; |
| 205 | + try { |
| 206 | + final Document doc = (Document) rid.getRecord(); |
| 207 | + // Check if this alias was already bound — if so, verify it matches |
| 208 | + final Object prevValue = upstream.getProperty(aliases[i]); |
| 209 | + if (prevValue != null) { |
| 210 | + Identifiable prevElem = null; |
| 211 | + if (prevValue instanceof Identifiable id) |
| 212 | + prevElem = id; |
| 213 | + else if (prevValue instanceof Result r) |
| 214 | + prevElem = r.getElement().orElse(null); |
| 215 | + if (prevElem != null && !prevElem.getIdentity().equals(rid)) |
| 216 | + return null; // Doesn't match bound value |
| 217 | + } |
| 218 | + result.setProperty(aliases[i], new ResultInternal(doc)); |
| 219 | + } catch (final RecordNotFoundException e) { |
| 220 | + return null; // Deleted since CSR build |
| 221 | + } |
| 222 | + } |
| 223 | + |
| 224 | + return result; |
| 225 | + } |
| 226 | + |
| 227 | + @Override |
| 228 | + public String prettyPrint(final int depth, final int indent) { |
| 229 | + final String spaces = ExecutionStepInternal.getIndent(depth, indent); |
| 230 | + final StringBuilder sb = new StringBuilder(); |
| 231 | + sb.append(spaces).append("+ MATCH GAV FUSED (").append(edges.size()).append(" hops)\n"); |
| 232 | + for (final EdgeTraversal edge : edges) { |
| 233 | + sb.append(spaces).append(" "); |
| 234 | + sb.append("{").append(edge.edge.out.alias).append("}"); |
| 235 | + sb.append(edge.edge.item.getMethod()); |
| 236 | + sb.append("{").append(edge.edge.in.alias).append("}\n"); |
| 237 | + } |
| 238 | + return sb.toString(); |
| 239 | + } |
| 240 | +} |
0 commit comments