diff --git a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/LayeredSubgraphAssembler.java b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/LayeredSubgraphAssembler.java new file mode 100644 index 000000000..7536737f6 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/LayeredSubgraphAssembler.java @@ -0,0 +1,167 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.geaflow.state.sampling.LocalNeighborhood; + +/** + * Window-local assembler. It is intentionally not serializable or checkpointed; reusable data is + * held by each vertex as {@link LocalNeighborhood} instead. + */ +public class LayeredSubgraphAssembler { + + private final Map> assemblies = new LinkedHashMap<>(); + private final long maxSampledNodes; + private final long maxSampledEdges; + private final Comparator idComparator; + + public LayeredSubgraphAssembler() { + this(SubgraphSamplingSpec.DEFAULT_MAX_SAMPLED_NODES, + SubgraphSamplingSpec.DEFAULT_MAX_SAMPLED_EDGES, null); + } + + public LayeredSubgraphAssembler(long maxSampledNodes, long maxSampledEdges, + Comparator idComparator) { + if (maxSampledNodes < 1 || maxSampledEdges < 1) { + throw new IllegalArgumentException("sampling limits must be greater than zero"); + } + this.maxSampledNodes = maxSampledNodes; + this.maxSampledEdges = maxSampledEdges; + this.idComparator = idComparator; + } + + public void start(K rootId, int maxDepth, LocalNeighborhood rootNeighborhood) { + Objects.requireNonNull(rootId, "rootId"); + Objects.requireNonNull(rootNeighborhood, "rootNeighborhood"); + if (maxDepth < 1) { + throw new IllegalArgumentException("maxDepth must be greater than zero"); + } + if (!Objects.equals(rootId, rootNeighborhood.getVertex().getId())) { + throw new IllegalArgumentException("root neighborhood vertex id does not match rootId"); + } + if (assemblies.containsKey(rootId)) { + throw new IllegalStateException("sampling assembly already exists for rootId=" + rootId); + } + SampledSubgraph subgraph = new SampledSubgraph<>(rootId, + rootNeighborhood.getSnapshotVersion(), maxSampledNodes, maxSampledEdges, idComparator); + subgraph.addNeighborhood(0, rootNeighborhood, maxDepth > 0); + Assembly assembly = new Assembly<>(maxDepth, subgraph); + assembly.minDepthByVertex.put(rootId, 0); + assembly.completedVertices.add(rootId); + assemblies.put(rootId, assembly); + } + + public boolean registerRequest(K rootId, K vertexId, int depth) { + Objects.requireNonNull(vertexId, "vertexId"); + Assembly assembly = requireAssembly(rootId); + validateDepth(depth, assembly.maxDepth); + Integer knownDepth = assembly.minDepthByVertex.get(vertexId); + if (knownDepth != null && knownDepth <= depth) { + return false; + } + if (knownDepth == null && assembly.minDepthByVertex.size() + 1L > maxSampledNodes) { + throw new SubgraphSamplingLimitException(rootId, "nodes", + assembly.minDepthByVertex.size() + 1L, maxSampledNodes); + } + assembly.minDepthByVertex.put(vertexId, depth); + assembly.completedVertices.remove(vertexId); + return true; + } + + public boolean add(SubgraphSamplingResponse response) { + Objects.requireNonNull(response, "response"); + Assembly assembly = assemblies.get(response.getRootId()); + if (assembly == null) { + return false; + } + validateDepth(response.getDepth(), assembly.maxDepth); + K vertexId = response.getNeighborhood().getVertex().getId(); + Integer knownDepth = assembly.minDepthByVertex.get(vertexId); + if (knownDepth == null) { + throw new IllegalStateException(String.format( + "sampling response was not requested, rootId=%s, vertexId=%s", + response.getRootId(), vertexId)); + } + if (response.getDepth() > knownDepth || assembly.completedVertices.contains(vertexId)) { + return false; + } + if (response.getDepth() < knownDepth) { + throw new IllegalStateException(String.format( + "sampling response depth precedes request, rootId=%s, vertexId=%s, depth=%s", + response.getRootId(), vertexId, response.getDepth())); + } + assembly.subgraph.addNeighborhood(response.getDepth(), response.getNeighborhood(), + response.getDepth() < assembly.maxDepth); + assembly.completedVertices.add(vertexId); + return true; + } + + public SampledSubgraph take(K rootId) { + Assembly assembly = assemblies.remove(rootId); + if (assembly == null) { + return null; + } + Set pending = new LinkedHashSet<>(assembly.minDepthByVertex.keySet()); + pending.removeAll(assembly.completedVertices); + if (!pending.isEmpty()) { + throw new IllegalStateException("sampling responses missing for rootId=" + rootId + + ", vertexIds=" + pending); + } + assembly.subgraph.validateComplete(); + return assembly.subgraph; + } + + public void clear() { + assemblies.clear(); + } + + private Assembly requireAssembly(K rootId) { + Assembly assembly = assemblies.get(Objects.requireNonNull(rootId, "rootId")); + if (assembly == null) { + throw new IllegalStateException("sampling assembly does not exist for rootId=" + rootId); + } + return assembly; + } + + private void validateDepth(int depth, int maxDepth) { + if (depth < 1 || depth > maxDepth) { + throw new IllegalArgumentException("sampling depth is outside configured range: " + depth); + } + } + + private static class Assembly { + + private final int maxDepth; + private final SampledSubgraph subgraph; + private final Map minDepthByVertex = new LinkedHashMap<>(); + private final Set completedVertices = new LinkedHashSet<>(); + + private Assembly(int maxDepth, SampledSubgraph subgraph) { + this.maxDepth = maxDepth; + this.subgraph = subgraph; + } + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/LogicalEdgeId.java b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/LogicalEdgeId.java new file mode 100644 index 000000000..448c0b524 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/LogicalEdgeId.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.io.Serializable; +import java.util.Objects; +import org.apache.geaflow.model.graph.IGraphElementWithLabelField; +import org.apache.geaflow.model.graph.IGraphElementWithTimeField; +import org.apache.geaflow.model.graph.edge.IEdge; + +/** Stable identity of a normalized logical edge, independent of storage replicas and values. */ +public final class LogicalEdgeId implements Serializable { + + private final K sourceId; + private final K targetId; + private final String label; + private final Long time; + + public LogicalEdgeId(K sourceId, K targetId, String label, Long time) { + this.sourceId = Objects.requireNonNull(sourceId, "sourceId"); + this.targetId = Objects.requireNonNull(targetId, "targetId"); + this.label = label; + this.time = time; + } + + public static LogicalEdgeId fromNormalized(IEdge edge) { + Objects.requireNonNull(edge, "edge"); + String edgeLabel = edge instanceof IGraphElementWithLabelField + ? ((IGraphElementWithLabelField) edge).getLabel() : null; + Long edgeTime = edge instanceof IGraphElementWithTimeField + ? ((IGraphElementWithTimeField) edge).getTime() : null; + return new LogicalEdgeId<>(edge.getSrcId(), edge.getTargetId(), edgeLabel, edgeTime); + } + + public K getSourceId() { + return sourceId; + } + + public K getTargetId() { + return targetId; + } + + public String getLabel() { + return label; + } + + public Long getTime() { + return time; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (!(other instanceof LogicalEdgeId)) { + return false; + } + LogicalEdgeId that = (LogicalEdgeId) other; + return Objects.equals(sourceId, that.sourceId) + && Objects.equals(targetId, that.targetId) + && Objects.equals(label, that.label) + && Objects.equals(time, that.time); + } + + @Override + public int hashCode() { + return Objects.hash(sourceId, targetId, label, time); + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SampledSubgraph.java b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SampledSubgraph.java new file mode 100644 index 000000000..318282148 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SampledSubgraph.java @@ -0,0 +1,194 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.geaflow.model.graph.IGraphElementWithLabelField; +import org.apache.geaflow.model.graph.IGraphElementWithTimeField; +import org.apache.geaflow.model.graph.edge.IEdge; +import org.apache.geaflow.model.graph.vertex.IVertex; +import org.apache.geaflow.state.sampling.LocalNeighborhood; + +/** + * Short-lived layered view assembled from reusable one-hop vertex state. + */ +public class SampledSubgraph implements Serializable { + + private final K rootId; + private final long snapshotVersion; + private final Map> vertices = new LinkedHashMap<>(); + private final List>> edgeLayers = new ArrayList<>(); + private final Set> edgeIdentities = new LinkedHashSet<>(); + private final long maxSampledNodes; + private final long maxSampledEdges; + private final transient Comparator idComparator; + + public SampledSubgraph(K rootId, long snapshotVersion) { + this(rootId, snapshotVersion, SubgraphSamplingSpec.DEFAULT_MAX_SAMPLED_NODES, + SubgraphSamplingSpec.DEFAULT_MAX_SAMPLED_EDGES, null); + } + + public SampledSubgraph(K rootId, long snapshotVersion, long maxSampledNodes, + long maxSampledEdges, Comparator idComparator) { + this.rootId = Objects.requireNonNull(rootId, "rootId"); + this.snapshotVersion = snapshotVersion; + this.maxSampledNodes = maxSampledNodes; + this.maxSampledEdges = maxSampledEdges; + this.idComparator = idComparator; + } + + public K getRootId() { + return rootId; + } + + public long getSnapshotVersion() { + return snapshotVersion; + } + + public void addVertex(IVertex vertex) { + Objects.requireNonNull(vertex, "vertex"); + K vertexId = Objects.requireNonNull(vertex.getId(), "vertexId"); + if (!vertices.containsKey(vertexId) && vertices.size() + 1L > maxSampledNodes) { + throw new SubgraphSamplingLimitException(rootId, "nodes", + vertices.size() + 1L, maxSampledNodes); + } + vertices.put(vertexId, vertex); + } + + public void addNeighborhood(int depth, LocalNeighborhood neighborhood, + boolean includeEdges) { + if (depth < 0) { + throw new IllegalArgumentException("sampling neighborhood depth must not be negative"); + } + Objects.requireNonNull(neighborhood, "neighborhood"); + if (neighborhood.getSnapshotVersion() != snapshotVersion) { + throw new IllegalArgumentException("neighborhood snapshot does not match assembly snapshot"); + } + addVertex(neighborhood.getVertex()); + if (!includeEdges) { + return; + } + while (edgeLayers.size() <= depth) { + edgeLayers.add(new ArrayList<>()); + } + List> layer = edgeLayers.get(depth); + for (IEdge edge : neighborhood.getEdges()) { + addEdge(layer, edge); + } + } + + public Map> getVertices() { + if (idComparator == null || vertices.size() < 2) { + return Collections.unmodifiableMap(vertices); + } + List ids = new ArrayList<>(vertices.keySet()); + ids.sort((left, right) -> { + if (Objects.equals(left, rootId)) { + return Objects.equals(right, rootId) ? 0 : -1; + } + if (Objects.equals(right, rootId)) { + return 1; + } + return idComparator.compare(left, right); + }); + Map> ordered = new LinkedHashMap<>(); + for (K id : ids) { + ordered.put(id, vertices.get(id)); + } + return Collections.unmodifiableMap(ordered); + } + + public List>> getEdgeLayers() { + List>> layers = new ArrayList<>(edgeLayers.size()); + for (List> layer : edgeLayers) { + List> ordered = new ArrayList<>(layer); + if (idComparator != null) { + ordered.sort(this::compareEdges); + } + layers.add(Collections.unmodifiableList(ordered)); + } + return Collections.unmodifiableList(layers); + } + + public void validateComplete() { + for (List> layer : edgeLayers) { + for (IEdge edge : layer) { + if (!vertices.containsKey(edge.getSrcId()) || !vertices.containsKey(edge.getTargetId())) { + throw new IllegalStateException(String.format( + "sampled subgraph contains dangling edge, rootId=%s, srcId=%s, targetId=%s", + rootId, edge.getSrcId(), edge.getTargetId())); + } + } + } + } + + private void addEdge(List> layer, IEdge edge) { + Objects.requireNonNull(edge, "edge"); + Objects.requireNonNull(edge.getSrcId(), "edge.srcId"); + Objects.requireNonNull(edge.getTargetId(), "edge.targetId"); + // Incoming storage copies and outgoing copies represent the same logical edge after normalization. + LogicalEdgeId identity = LogicalEdgeId.fromNormalized(edge); + if (edgeIdentities.add(identity)) { + if (edgeIdentities.size() > maxSampledEdges) { + edgeIdentities.remove(identity); + throw new SubgraphSamplingLimitException(rootId, "edges", + edgeIdentities.size() + 1L, maxSampledEdges); + } + layer.add(edge); + } + } + + private int compareEdges(IEdge left, IEdge right) { + int result = idComparator.compare(left.getSrcId(), right.getSrcId()); + if (result == 0) { + result = idComparator.compare(left.getTargetId(), right.getTargetId()); + } + if (result == 0) { + result = String.valueOf(labelOf(left)).compareTo(String.valueOf(labelOf(right))); + } + if (result == 0) { + result = String.valueOf(timeOf(left)).compareTo(String.valueOf(timeOf(right))); + } + if (result == 0) { + result = String.valueOf(left.getValue()).compareTo(String.valueOf(right.getValue())); + } + return result; + } + + private static String labelOf(IEdge edge) { + return edge instanceof IGraphElementWithLabelField + ? ((IGraphElementWithLabelField) edge).getLabel() : null; + } + + private static Long timeOf(IEdge edge) { + return edge instanceof IGraphElementWithTimeField + ? ((IGraphElementWithTimeField) edge).getTime() : null; + } + +} diff --git a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingLimitException.java b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingLimitException.java new file mode 100644 index 000000000..9be028436 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingLimitException.java @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +/** Raised when a sampled subgraph would exceed its per-root resource budget. */ +public class SubgraphSamplingLimitException extends IllegalStateException { + + public SubgraphSamplingLimitException(Object rootId, String resource, long actual, long limit) { + super(String.format("subgraph sampling limit exceeded, rootId=%s, resource=%s, " + + "actual=%s, limit=%s", rootId, resource, actual, limit)); + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingRequest.java b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingRequest.java new file mode 100644 index 000000000..ce6b0e074 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingRequest.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.io.Serializable; +import java.util.Objects; + +/** A request to expand one vertex for one sampling layer. */ +public class SubgraphSamplingRequest implements Serializable { + + private final K rootId; + private final int depth; + + public SubgraphSamplingRequest(K rootId, int depth) { + if (depth < 1) { + throw new IllegalArgumentException("sampling request depth must be greater than zero"); + } + this.rootId = Objects.requireNonNull(rootId, "rootId"); + this.depth = depth; + } + + public K getRootId() { + return rootId; + } + + public int getDepth() { + return depth; + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingResponse.java b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingResponse.java new file mode 100644 index 000000000..b845dc753 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingResponse.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.io.Serializable; +import java.util.Objects; +import org.apache.geaflow.state.sampling.LocalNeighborhood; + +/** One vertex's local sampling result returned to the requesting root. */ +public class SubgraphSamplingResponse implements Serializable { + + private final K rootId; + private final int depth; + private final LocalNeighborhood neighborhood; + + public SubgraphSamplingResponse(K rootId, int depth, + LocalNeighborhood neighborhood) { + if (depth < 1) { + throw new IllegalArgumentException("sampling response depth must be greater than zero"); + } + this.rootId = Objects.requireNonNull(rootId, "rootId"); + this.depth = depth; + this.neighborhood = Objects.requireNonNull(neighborhood, "neighborhood"); + } + + public K getRootId() { + return rootId; + } + + public int getDepth() { + return depth; + } + + public LocalNeighborhood getNeighborhood() { + return neighborhood; + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingSpec.java b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingSpec.java new file mode 100644 index 000000000..8f12dd783 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingSpec.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.io.Serializable; +import java.util.Objects; +import org.apache.geaflow.model.graph.edge.EdgeDirection; + +/** + * Configuration for bounded iterative multi-hop sampling. + */ +public class SubgraphSamplingSpec implements Serializable { + + public static final long DEFAULT_MAX_SAMPLED_NODES = 10000L; + public static final long DEFAULT_MAX_SAMPLED_EDGES = 100000L; + public static final long DEFAULT_MAX_RETURNED_EDGES = 100000L; + + private final int hops; + private final int fanout; + private final EdgeDirection direction; + private final long maxReturnedEdges; + private final long seed; + + public SubgraphSamplingSpec(int hops, int fanout, EdgeDirection direction) { + this(hops, fanout, direction, DEFAULT_MAX_RETURNED_EDGES); + } + + public SubgraphSamplingSpec(int hops, int fanout, EdgeDirection direction, + long maxReturnedEdges) { + this(hops, fanout, direction, maxReturnedEdges, 0L); + } + + public SubgraphSamplingSpec(int hops, int fanout, EdgeDirection direction, + long maxReturnedEdges, long seed) { + if (hops < 1) { + throw new IllegalArgumentException("sampling hops must be greater than zero"); + } + if (fanout == 0 || fanout < -1) { + throw new IllegalArgumentException("fanout must be -1 or greater than zero"); + } + if (maxReturnedEdges < 1) { + throw new IllegalArgumentException("maxReturnedEdges must be greater than zero"); + } + this.hops = hops; + this.fanout = fanout; + this.direction = Objects.requireNonNull(direction, "direction"); + this.maxReturnedEdges = maxReturnedEdges; + this.seed = seed; + } + + public int getHops() { + return hops; + } + + public int getFanout() { + return fanout; + } + + public EdgeDirection getDirection() { + return direction; + } + + public long getMaxReturnedEdges() { + return maxReturnedEdges; + } + + public long getSeed() { + return seed; + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/LayeredSubgraphAssemblerTest.java b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/LayeredSubgraphAssemblerTest.java new file mode 100644 index 000000000..de506288e --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/LayeredSubgraphAssemblerTest.java @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.util.Collections; +import java.util.List; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.impl.ValueEdge; +import org.apache.geaflow.model.graph.edge.impl.ValueLabelEdge; +import org.apache.geaflow.model.graph.edge.impl.ValueLabelTimeEdge; +import org.apache.geaflow.model.graph.vertex.impl.ValueVertex; +import org.apache.geaflow.state.sampling.LocalNeighborhood; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class LayeredSubgraphAssemblerTest { + + @Test + public void testAssemblesOneHopPerLayerAndReleasesResult() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + LocalNeighborhood root = neighborhood(1L, 2L, 7L); + LocalNeighborhood firstHop = neighborhood(2L, 3L, 7L); + LocalNeighborhood frontier = neighborhood(3L, 4L, 7L); + + assembler.start(1L, 2, root); + Assert.assertTrue(assembler.registerRequest(1L, 2L, 1)); + assembler.add(new SubgraphSamplingResponse<>(1L, 1, firstHop)); + Assert.assertTrue(assembler.registerRequest(1L, 3L, 2)); + assembler.add(new SubgraphSamplingResponse<>(1L, 2, frontier)); + + SampledSubgraph subgraph = assembler.take(1L); + Assert.assertEquals(subgraph.getVertices().size(), 3); + Assert.assertEquals(subgraph.getVertices().get(3L).getValue(), Integer.valueOf(3)); + Assert.assertFalse(subgraph.getVertices().containsKey(4L)); + Assert.assertEquals(subgraph.getEdgeLayers().size(), 2); + Assert.assertEquals(subgraph.getEdgeLayers().get(0).get(0).getTargetId(), Long.valueOf(2L)); + Assert.assertEquals(subgraph.getEdgeLayers().get(1).get(0).getTargetId(), Long.valueOf(3L)); + Assert.assertTrue(subgraph.getEdgeLayers().stream() + .flatMap(List::stream) + .noneMatch(edge -> Long.valueOf(4L).equals(edge.getTargetId()))); + Assert.assertNull(assembler.take(1L)); + } + + @Test + public void testAcceptsOutOfOrderResponsesAndIgnoresDuplicates() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 1, new LocalNeighborhood<>(new ValueVertex<>(1L, 1), + java.util.Arrays.asList(new ValueEdge<>(1L, 2L, 1, EdgeDirection.OUT), + new ValueEdge<>(1L, 3L, 1, EdgeDirection.OUT)), 7L)); + Assert.assertTrue(assembler.registerRequest(1L, 2L, 1)); + Assert.assertTrue(assembler.registerRequest(1L, 3L, 1)); + + SubgraphSamplingResponse responseForThree = + new SubgraphSamplingResponse<>(1L, 1, neighborhood(3L, 4L, 7L)); + SubgraphSamplingResponse responseForTwo = + new SubgraphSamplingResponse<>(1L, 1, neighborhood(2L, 5L, 7L)); + Assert.assertTrue(assembler.add(responseForThree)); + Assert.assertTrue(assembler.add(responseForTwo)); + Assert.assertFalse(assembler.add(responseForThree)); + + Assert.assertEquals(assembler.take(1L).getVertices().keySet(), + new java.util.LinkedHashSet<>(java.util.Arrays.asList(1L, 2L, 3L))); + } + + @Test + public void testKeepsAssembliesIsolatedAndClearRemovesThem() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 1, neighborhood(1L, 2L, 7L)); + assembler.start(10L, 1, neighborhood(10L, 11L, 7L)); + + Assert.assertTrue(assembler.registerRequest(1L, 2L, 1)); + Assert.assertTrue(assembler.registerRequest(10L, 11L, 1)); + Assert.assertTrue(assembler.add(new SubgraphSamplingResponse<>(1L, 1, + neighborhood(2L, 3L, 7L)))); + Assert.assertTrue(assembler.add(new SubgraphSamplingResponse<>(10L, 1, + neighborhood(11L, 12L, 7L)))); + Assert.assertEquals(assembler.take(1L).getRootId(), Long.valueOf(1L)); + Assert.assertEquals(assembler.take(10L).getRootId(), Long.valueOf(10L)); + + assembler.start(20L, 1, neighborhood(20L, 21L, 7L)); + assembler.clear(); + Assert.assertNull(assembler.take(20L)); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsRequestDepthOutsideAssembly() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 1, neighborhood(1L, 2L, 7L)); + assembler.registerRequest(1L, 2L, 2); + } + + @Test(expectedExceptions = IllegalStateException.class) + public void testRejectsResponseForUnrequestedVertex() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 1, neighborhood(1L, 2L, 7L)); + assembler.add(new SubgraphSamplingResponse<>(1L, 1, neighborhood(3L, 4L, 7L))); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsNeighborhoodFromPreviousWindow() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 1, neighborhood(1L, 2L, 7L)); + assembler.registerRequest(1L, 2L, 1); + assembler.add(new SubgraphSamplingResponse<>(1L, 1, neighborhood(2L, 3L, 6L))); + } + + @Test + public void testDoesNotDuplicateLogicalEdgeAcrossLayers() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 2, neighborhood(1L, 2L, 7L)); + assembler.registerRequest(1L, 2L, 1); + assembler.add(new SubgraphSamplingResponse<>(1L, 1, + neighborhoodWithEdge(2L, 1L, 2L, 7L))); + + SampledSubgraph subgraph = assembler.take(1L); + long edgeCount = subgraph.getEdgeLayers().stream().mapToLong(List::size).sum(); + Assert.assertEquals(edgeCount, 1L); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsNeighborhoodFromFutureWindow() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 1, neighborhood(1L, 2L, 7L)); + assembler.registerRequest(1L, 2L, 1); + assembler.add(new SubgraphSamplingResponse<>(1L, 1, neighborhood(2L, 3L, 8L))); + } + + @Test + public void testCycleRegistersEachVertexOnlyOnce() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 3, neighborhood(1L, 2L, 7L)); + + Assert.assertTrue(assembler.registerRequest(1L, 2L, 1)); + Assert.assertFalse(assembler.registerRequest(1L, 1L, 2)); + Assert.assertFalse(assembler.registerRequest(1L, 2L, 2)); + assembler.add(new SubgraphSamplingResponse<>(1L, 1, + neighborhoodWithEdge(2L, 2L, 1L, 7L))); + + Assert.assertEquals(assembler.take(1L).getVertices().size(), 2); + } + + @Test(expectedExceptions = SubgraphSamplingLimitException.class) + public void testRejectsNodeBudgetOverflow() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(1, 10, Long::compare); + assembler.start(1L, 1, neighborhood(1L, 2L, 7L)); + assembler.registerRequest(1L, 2L, 1); + } + + @Test(expectedExceptions = SubgraphSamplingLimitException.class) + public void testRejectsEdgeBudgetOverflow() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(10, 1, Long::compare); + LocalNeighborhood root = new LocalNeighborhood<>( + new ValueVertex<>(1L, 1), java.util.Arrays.asList( + new ValueEdge<>(1L, 2L, 1, EdgeDirection.OUT), + new ValueEdge<>(1L, 3L, 1, EdgeDirection.OUT)), 7L); + assembler.start(1L, 1, root); + } + + @Test(expectedExceptions = IllegalStateException.class) + public void testRejectsMissingResponse() { + LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + assembler.start(1L, 1, neighborhood(1L, 2L, 7L)); + assembler.registerRequest(1L, 2L, 1); + assembler.take(1L); + } + + @Test + public void testStructuralEdgeIdentityPreservesCollisionsAndLabels() { + SampledSubgraph subgraph = new SampledSubgraph<>("root", 1L); + LocalNeighborhood neighborhood = new LocalNeighborhood<>( + new ValueVertex<>("root", 0), java.util.Arrays.asList( + new ValueLabelEdge<>("a->b", "c", 1, "first"), + new ValueLabelEdge<>("a", "b->c", 1, "first"), + new ValueLabelEdge<>("a", "b->c", 1, "second")), 1L); + subgraph.addNeighborhood(0, neighborhood, true); + + Assert.assertEquals(subgraph.getEdgeLayers().get(0).size(), 3); + } + + @Test + public void testLogicalEdgeIdentityIgnoresReplicaDirectionAndValue() { + ValueLabelEdge out = new ValueLabelEdge<>(1L, 2L, "first", "knows"); + out.setDirect(EdgeDirection.OUT); + ValueLabelEdge inReplica = new ValueLabelEdge<>(1L, 2L, "second", "knows"); + inReplica.setDirect(EdgeDirection.IN); + ValueLabelEdge reciprocal = new ValueLabelEdge<>(2L, 1L, "third", "knows"); + + SampledSubgraph subgraph = new SampledSubgraph<>(1L, 1L); + subgraph.addNeighborhood(0, new LocalNeighborhood<>(new ValueVertex<>(1L, 1), + java.util.Arrays.asList(out, inReplica, reciprocal), 1L), true); + + Assert.assertEquals(subgraph.getEdgeLayers().get(0).size(), 2); + } + + @Test + public void testLogicalEdgeIdentityPreservesTemporalParallelEdges() { + SampledSubgraph subgraph = new SampledSubgraph<>(1L, 1L); + subgraph.addNeighborhood(0, new LocalNeighborhood<>(new ValueVertex<>(1L, 1), + java.util.Arrays.asList( + new ValueLabelTimeEdge<>(1L, 2L, "same", "knows", 10L), + new ValueLabelTimeEdge<>(1L, 2L, "same", "knows", 11L)), 1L), true); + + Assert.assertEquals(subgraph.getEdgeLayers().get(0).size(), 2); + } + + private LocalNeighborhood neighborhood(long source, long target, + long version) { + return neighborhoodWithEdge(source, source, target, version); + } + + private LocalNeighborhood neighborhoodWithEdge(long vertexId, + long source, + long target, + long version) { + ValueVertex vertex = new ValueVertex<>(vertexId, (int) vertexId); + ValueEdge edge = new ValueEdge<>(source, target, 1, EdgeDirection.OUT); + return new LocalNeighborhood<>(vertex, Collections.singletonList(edge), version); + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingEndToEndTest.java b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingEndToEndTest.java new file mode 100644 index 000000000..fde7db764 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingEndToEndTest.java @@ -0,0 +1,311 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.IEdge; +import org.apache.geaflow.model.graph.edge.impl.ValueEdge; +import org.apache.geaflow.model.graph.vertex.impl.ValueVertex; +import org.apache.geaflow.state.sampling.DeterministicNeighborSampler; +import org.apache.geaflow.state.sampling.LocalNeighborhood; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class SubgraphSamplingEndToEndTest { + + private static final long SNAPSHOT_VERSION = 7L; + private static final long SAMPLING_VERSION = 42L; + private static final long SEED = 17L; + + @Test + public void testRunsThreeHopSamplingWithOutOfOrderResponses() { + SubgraphSamplingSpec spec = new SubgraphSamplingSpec( + 3, -1, EdgeDirection.OUT, 100L, SEED); + RecordingGraph graph = diamondAndCycleGraph(); + SamplingDriver driver = new SamplingDriver(graph, spec); + + SampledSubgraph result = driver.run(1L); + + Assert.assertEquals(new LinkedHashSet<>(driver.getRequestTrace().subList(0, 2)), + new LinkedHashSet<>(Arrays.asList("1->2@1", "1->3@1"))); + Assert.assertEquals(new LinkedHashSet<>(driver.getRequestTrace().subList(2, 4)), + new LinkedHashSet<>(Arrays.asList("1->4@2", "1->5@2"))); + Assert.assertEquals(driver.getRequestTrace().get(4), "1->6@3"); + Assert.assertNotEquals(driver.getRequestTrace().subList(0, 2), + driver.getEnqueuedRequestTrace().subList(0, 2)); + Assert.assertEquals(new LinkedHashSet<>(driver.getResponseTrace().subList(0, 2)), + new LinkedHashSet<>(Arrays.asList("2->1@1", "3->1@1"))); + Assert.assertEquals(new LinkedHashSet<>(driver.getResponseTrace().subList(2, 4)), + new LinkedHashSet<>(Arrays.asList("4->1@2", "5->1@2"))); + Assert.assertEquals(driver.getResponseTrace().get(4), "6->1@3"); + Assert.assertEquals(driver.getTerminalResponseEdgeCounts(), Collections.singletonList(0)); + Assert.assertEquals(driver.getRequestDepths(), Arrays.asList(1, 1, 2, 2, 3)); + Assert.assertEquals(driver.getResponseCount(), driver.getRequestCount()); + Assert.assertEquals(driver.getRequestCount(), 5); + Assert.assertEquals(result.getVertices().keySet(), new LinkedHashSet<>(Arrays.asList( + 1L, 2L, 3L, 4L, 5L, 6L))); + Assert.assertEquals(result.getEdgeLayers().size(), 3); + Assert.assertEquals(result.getEdgeLayers().get(0).size(), 2); + Assert.assertEquals(result.getEdgeLayers().get(1).size(), 4); + Assert.assertEquals(result.getEdgeLayers().get(2).size(), 3); + Assert.assertEquals(graph.getSampleReads().size(), 5); + Assert.assertEquals(graph.getVertexOnlyReads(), Collections.singletonList("6@3")); + } + + @Test + public void testDoesNotReadEdgesForTerminalDepth() { + SubgraphSamplingSpec spec = new SubgraphSamplingSpec( + 2, -1, EdgeDirection.OUT, 100L, SEED); + RecordingGraph graph = lineGraph(); + SamplingDriver driver = new SamplingDriver(graph, spec); + + SampledSubgraph result = driver.run(1L); + + Assert.assertEquals(graph.getSampleReads(), Arrays.asList("1@0", "2@1")); + Assert.assertEquals(graph.getVertexOnlyReads(), Collections.singletonList("3@2")); + Assert.assertEquals(driver.getTerminalResponseEdgeCounts(), Collections.singletonList(0)); + Assert.assertEquals(result.getVertices().keySet(), new LinkedHashSet<>(Arrays.asList( + 1L, 2L, 3L))); + Assert.assertEquals(result.getEdgeLayers().size(), 2); + Assert.assertEquals(result.getEdgeLayers().get(0).size(), 1); + Assert.assertEquals(result.getEdgeLayers().get(1).size(), 1); + Assert.assertEquals(driver.getRequestDepths(), Arrays.asList(1, 2)); + } + + @Test + public void testEndToEndSamplingHonorsPositiveFanout() { + SubgraphSamplingSpec spec = new SubgraphSamplingSpec( + 1, 1, EdgeDirection.OUT, 100L, SEED); + RecordingGraph graph = fanoutGraph(); + SamplingDriver driver = new SamplingDriver(graph, spec); + + SampledSubgraph result = driver.run(1L); + + Assert.assertEquals(result.getVertices().size(), 2); + Assert.assertEquals(result.getEdgeLayers().get(0).size(), 1); + Assert.assertEquals(driver.getRequestCount(), 1); + Assert.assertEquals(graph.getSampleReads(), Collections.singletonList("1@0")); + Assert.assertEquals(graph.getVertexOnlyReads().size(), 1); + } + + private static RecordingGraph diamondAndCycleGraph() { + Map>> adjacency = new LinkedHashMap<>(); + adjacency.put(1L, Arrays.asList(edge(1L, 2L), edge(1L, 3L))); + adjacency.put(2L, Arrays.asList(edge(2L, 4L), edge(2L, 5L))); + adjacency.put(3L, Arrays.asList(edge(3L, 4L), edge(3L, 5L))); + adjacency.put(4L, Arrays.asList(edge(4L, 1L), edge(4L, 6L))); + adjacency.put(5L, Collections.singletonList(edge(5L, 6L))); + adjacency.put(6L, Collections.emptyList()); + return new RecordingGraph(adjacency); + } + + private static RecordingGraph lineGraph() { + Map>> adjacency = new LinkedHashMap<>(); + adjacency.put(1L, Collections.singletonList(edge(1L, 2L))); + adjacency.put(2L, Collections.singletonList(edge(2L, 3L))); + adjacency.put(3L, Collections.emptyList()); + return new RecordingGraph(adjacency); + } + + private static RecordingGraph fanoutGraph() { + Map>> adjacency = new LinkedHashMap<>(); + adjacency.put(1L, Arrays.asList(edge(1L, 2L), edge(1L, 3L), edge(1L, 4L))); + adjacency.put(2L, Collections.emptyList()); + adjacency.put(3L, Collections.emptyList()); + adjacency.put(4L, Collections.emptyList()); + return new RecordingGraph(adjacency); + } + + private static IEdge edge(long source, long target) { + return new ValueEdge<>(source, target, 1, EdgeDirection.OUT); + } + + private static byte[] longBytes(long value) { + return new byte[]{ + (byte) (value >>> 56), (byte) (value >>> 48), (byte) (value >>> 40), + (byte) (value >>> 32), (byte) (value >>> 24), (byte) (value >>> 16), + (byte) (value >>> 8), (byte) value}; + } + + private static final class SamplingDriver { + + private final RecordingGraph graph; + private final SubgraphSamplingSpec spec; + private final LayeredSubgraphAssembler assembler = + new LayeredSubgraphAssembler<>(); + private final List enqueuedRequestTrace = new ArrayList<>(); + private final List requestTrace = new ArrayList<>(); + private final List responseTrace = new ArrayList<>(); + private final List requestDepths = new ArrayList<>(); + private final List terminalResponseEdgeCounts = new ArrayList<>(); + + private SamplingDriver(RecordingGraph graph, SubgraphSamplingSpec spec) { + this.graph = graph; + this.spec = spec; + } + + private SampledSubgraph run(long rootId) { + LocalNeighborhood rootNeighborhood = + graph.sample(rootId, 0, spec); + assembler.start(rootId, spec.getHops(), rootNeighborhood); + Set frontier = neighbors(rootId, rootNeighborhood.getEdges()); + + for (int depth = 1; depth <= spec.getHops(); depth++) { + List requests = new ArrayList<>(); + for (Long vertexId : frontier) { + if (assembler.registerRequest(rootId, vertexId, depth)) { + requests.add(new RoutedRequest(vertexId, + new SubgraphSamplingRequest<>(rootId, depth))); + enqueuedRequestTrace.add(rootId + "->" + vertexId + "@" + depth); + } + } + + // Reverse delivery order to ensure assembly does not depend on transport order. + Collections.reverse(requests); + Set nextFrontier = new LinkedHashSet<>(); + for (RoutedRequest request : requests) { + Long requestRootId = request.request.getRootId(); + int requestDepth = request.request.getDepth(); + Assert.assertEquals(requestRootId, Long.valueOf(rootId)); + Assert.assertEquals(requestDepth, depth); + requestTrace.add(requestRootId + "->" + request.vertexId + "@" + requestDepth); + requestDepths.add(requestDepth); + LocalNeighborhood neighborhood = requestDepth == spec.getHops() + ? graph.vertexOnly(request.vertexId, requestDepth) + : graph.sample(request.vertexId, requestDepth, spec); + responseTrace.add(request.vertexId + "->" + requestRootId + "@" + requestDepth); + if (requestDepth == spec.getHops()) { + terminalResponseEdgeCounts.add(neighborhood.getEdges().size()); + } + SubgraphSamplingResponse response = + new SubgraphSamplingResponse<>(requestRootId, requestDepth, neighborhood); + Assert.assertTrue(assembler.add(response)); + if (requestDepth < spec.getHops()) { + nextFrontier.addAll(neighbors(request.vertexId, neighborhood.getEdges())); + } + } + frontier = nextFrontier; + } + return assembler.take(rootId); + } + + private Set neighbors(long vertexId, List> edges) { + Set neighbors = new LinkedHashSet<>(); + for (IEdge edge : edges) { + if (Long.valueOf(vertexId).equals(edge.getSrcId())) { + neighbors.add(edge.getTargetId()); + } else { + neighbors.add(edge.getSrcId()); + } + } + return neighbors; + } + + private List getRequestTrace() { + return requestTrace; + } + + private List getEnqueuedRequestTrace() { + return enqueuedRequestTrace; + } + + private List getResponseTrace() { + return responseTrace; + } + + private List getRequestDepths() { + return requestDepths; + } + + private int getRequestCount() { + return requestTrace.size(); + } + + private int getResponseCount() { + return responseTrace.size(); + } + + private List getTerminalResponseEdgeCounts() { + return terminalResponseEdgeCounts; + } + } + + private static final class RoutedRequest { + + private final Long vertexId; + private final SubgraphSamplingRequest request; + + private RoutedRequest(Long vertexId, SubgraphSamplingRequest request) { + this.vertexId = vertexId; + this.request = request; + } + } + + private static final class RecordingGraph { + + private final Map>> adjacency; + private final List sampleReads = new ArrayList<>(); + private final List vertexOnlyReads = new ArrayList<>(); + + private RecordingGraph(Map>> adjacency) { + this.adjacency = adjacency; + } + + private LocalNeighborhood sample( + long vertexId, int depth, SubgraphSamplingSpec spec) { + if (depth >= spec.getHops()) { + Assert.fail("terminal depth must not read or sample adjacent edges"); + } + sampleReads.add(vertexId + "@" + depth); + List> sampled = DeterministicNeighborSampler.sample( + vertexId, adjacency.get(vertexId), spec.getDirection(), spec.getFanout(), + Long::compare, spec.getMaxReturnedEdges(), spec.getSeed(), SAMPLING_VERSION, + SubgraphSamplingEndToEndTest::longBytes); + return neighborhood(vertexId, sampled); + } + + private LocalNeighborhood vertexOnly(long vertexId, int depth) { + vertexOnlyReads.add(vertexId + "@" + depth); + return neighborhood(vertexId, Collections.emptyList()); + } + + private LocalNeighborhood neighborhood( + long vertexId, List> edges) { + return new LocalNeighborhood<>(new ValueVertex<>(vertexId, (int) vertexId), + edges, SNAPSHOT_VERSION, SAMPLING_VERSION); + } + + private List getSampleReads() { + return sampleReads; + } + + private List getVertexOnlyReads() { + return vertexOnlyReads; + } + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingMessageTest.java b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingMessageTest.java new file mode 100644 index 000000000..7ae6af8cd --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingMessageTest.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import java.util.Collections; +import org.apache.geaflow.model.graph.vertex.impl.ValueVertex; +import org.apache.geaflow.state.sampling.LocalNeighborhood; +import org.testng.annotations.Test; + +public class SubgraphSamplingMessageTest { + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsNonPositiveRequestDepth() { + new SubgraphSamplingRequest<>(1L, 0); + } + + @Test(expectedExceptions = NullPointerException.class) + public void testRejectsNullRequestRoot() { + new SubgraphSamplingRequest<>(null, 1); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsNonPositiveResponseDepth() { + new SubgraphSamplingResponse<>(1L, 0, neighborhood(1L)); + } + + @Test(expectedExceptions = NullPointerException.class) + public void testRejectsNullResponseRoot() { + new SubgraphSamplingResponse<>(null, 1, neighborhood(1L)); + } + + @Test(expectedExceptions = NullPointerException.class) + public void testRejectsNullResponseNeighborhood() { + new SubgraphSamplingResponse<>(1L, 1, null); + } + + private LocalNeighborhood neighborhood(long vertexId) { + return new LocalNeighborhood<>(new ValueVertex<>(vertexId, (int) vertexId), + Collections.emptyList(), 7L); + } +} diff --git a/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingSpecTest.java b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingSpecTest.java new file mode 100644 index 000000000..603549009 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-api/src/test/java/org/apache/geaflow/api/graph/sampling/SubgraphSamplingSpecTest.java @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.api.graph.sampling; + +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class SubgraphSamplingSpecTest { + + @Test + public void testHopsAndFanoutAreScalar() { + SubgraphSamplingSpec spec = new SubgraphSamplingSpec(2, 10, EdgeDirection.OUT); + + Assert.assertEquals(spec.getHops(), 2); + Assert.assertEquals(spec.getFanout(), 10); + Assert.assertEquals(spec.getMaxReturnedEdges(), 100000L); + Assert.assertEquals(spec.getSeed(), 0L); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsNonPositiveHops() { + new SubgraphSamplingSpec(0, 10, EdgeDirection.OUT); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsZeroFanout() { + new SubgraphSamplingSpec(2, 0, EdgeDirection.OUT); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsFanoutBelowUnlimitedMarker() { + new SubgraphSamplingSpec(2, -2, EdgeDirection.OUT); + } + + @Test(expectedExceptions = NullPointerException.class) + public void testRejectsNullDirection() { + new SubgraphSamplingSpec(2, 1, null); + } + + @Test + public void testUnlimitedFanoutKeepsPerVertexEdgeBudget() { + SubgraphSamplingSpec spec = new SubgraphSamplingSpec(2, -1, EdgeDirection.BOTH); + + Assert.assertEquals(spec.getFanout(), -1); + Assert.assertEquals(spec.getMaxReturnedEdges(), 100000L); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsInvalidReturnedEdgeBudget() { + new SubgraphSamplingSpec(2, -1, EdgeDirection.BOTH, 0); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsNegativeReturnedEdgeBudget() { + new SubgraphSamplingSpec(2, -1, EdgeDirection.BOTH, -1); + } +} diff --git a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/cache/TemporaryGraphCache.java b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/cache/TemporaryGraphCache.java index c6030489c..f19cfade1 100644 --- a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/cache/TemporaryGraphCache.java +++ b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/cache/TemporaryGraphCache.java @@ -55,6 +55,7 @@ public IVertex getVertex(K vId) { public void addEdge(IEdge edge) { this.vertexIds.add(edge.getSrcId()); + this.vertexIds.add(edge.getTargetId()); List> edges = this.vertexEdges.getOrDefault(edge.getSrcId(), new ArrayList<>()); edges.add(edge); diff --git a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/cache/TemporaryGraphCacheTest.java b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/cache/TemporaryGraphCacheTest.java new file mode 100644 index 000000000..1f3cd1568 --- /dev/null +++ b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/cache/TemporaryGraphCacheTest.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.operator.impl.graph.compute.dynamic.cache; + +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.impl.ValueEdge; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TemporaryGraphCacheTest { + + @Test + public void testEdgeTriggersSourceAndTargetVertices() { + TemporaryGraphCache cache = new TemporaryGraphCache<>(); + cache.addEdge(new ValueEdge<>(1L, 2L, 1, EdgeDirection.OUT)); + + Assert.assertTrue(cache.getAllEvolveVId().contains(1L)); + Assert.assertTrue(cache.getAllEvolveVId().contains(2L)); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmSamplingRuntimeContext.java b/geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmSamplingRuntimeContext.java new file mode 100644 index 000000000..1466b52f2 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmSamplingRuntimeContext.java @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.dsl.common.algo; + +import java.util.Comparator; +import java.util.List; +import java.util.function.Function; +import org.apache.geaflow.api.graph.sampling.SubgraphSamplingSpec; +import org.apache.geaflow.common.iterator.CloseableIterator; +import org.apache.geaflow.common.type.IType; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.RowEdge; +import org.apache.geaflow.dsl.common.data.RowVertex; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.IEdge; +import org.apache.geaflow.state.sampling.DeterministicNeighborSampler; +import org.apache.geaflow.state.sampling.LocalNeighborhood; + +/** Runtime-facing contract for reusable one-hop sampling. */ +public interface AlgorithmSamplingRuntimeContext extends AlgorithmRuntimeContext { + + default LocalNeighborhood sampleOneHop(RowVertex vertex, + EdgeDirection direction, + int fanout) { + return sampleOneHop(vertex, direction, fanout, + DeterministicNeighborSampler.DEFAULT_MAX_RETURNED_EDGES, 0L, + getSamplingSnapshotVersion()); + } + + default LocalNeighborhood sampleOneHop(RowVertex vertex, + EdgeDirection direction, + int fanout, + long maxReturnedEdges, + long seed, + long samplingVersion) { + try (CloseableIterator iterator = loadStaticEdgesIterator(direction)) { + Iterable edges = () -> iterator; + @SuppressWarnings({"unchecked", "rawtypes"}) + IType idType = (IType) getGraphSchema().getIdType(); + Comparator comparator = idType::compare; + Function idEncoder = idType::serialize; + @SuppressWarnings({"unchecked", "rawtypes"}) + List> sampled = (List) DeterministicNeighborSampler.sample( + vertex.getId(), edges, direction, fanout, comparator, maxReturnedEdges, + seed, samplingVersion, idEncoder); + return new LocalNeighborhood<>(vertex, sampled, getSamplingSnapshotVersion(), + samplingVersion); + } + } + + default LocalNeighborhood sampleOneHop(RowVertex vertex, + SubgraphSamplingSpec spec, + long samplingVersion) { + return sampleOneHop(vertex, spec.getDirection(), spec.getFanout(), + spec.getMaxReturnedEdges(), spec.getSeed(), samplingVersion); + } + + long getSamplingSnapshotVersion(); + + default long getNeighborhoodChangeVersion(Object vertexId) { + return Long.MIN_VALUE; + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicAggTraversalFunction.java b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicAggTraversalFunction.java index 98c475b15..658793939 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicAggTraversalFunction.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicAggTraversalFunction.java @@ -57,6 +57,8 @@ public class GeaFlowAlgorithmDynamicAggTraversalFunction private static final Logger LOGGER = LoggerFactory.getLogger(GeaFlowAlgorithmDynamicAggTraversalFunction.class); private static final String STATE_SUFFIX = "UpdatedValueState"; + private static final String NEIGHBORHOOD_CHANGE_VERSION_STATE_SUFFIX = + "NeighborhoodChangeVersionState"; private final AlgorithmUserFunction userFunction; @@ -64,15 +66,16 @@ public class GeaFlowAlgorithmDynamicAggTraversalFunction private GraphSchema graphSchema; - private IncVertexCentricTraversalFuncContext traversalContext; + private transient IncVertexCentricTraversalFuncContext traversalContext; - private GeaFlowAlgorithmDynamicRuntimeContext algorithmCtx; + private transient GeaFlowAlgorithmDynamicRuntimeContext algorithmCtx; - private MutableGraph mutableGraph; + private transient MutableGraph mutableGraph; private transient Set initVertices; private transient KeyValueState vertexUpdateValues; + private transient KeyValueState neighborhoodChangeVersions; private boolean materializeInFinish; @@ -109,12 +112,21 @@ public void open( IKeyGroupAssigner keyGroupAssigner = KeyGroupAssignerFactory.createKeyGroupAssigner( keyGroup, taskIndex, maxParallelism); descriptor.withKeyGroupAssigner(keyGroupAssigner); - long recoverWindowId = traversalContext.getRuntimeContext().getWindowId(); + final long recoverWindowId = traversalContext.getRuntimeContext().getWindowId(); this.vertexUpdateValues = StateFactory.buildKeyValueState(descriptor, traversalContext.getRuntimeContext().getConfiguration()); + KeyValueStateDescriptor changeVersionDescriptor = KeyValueStateDescriptor.build( + traversalContext.getTraversalOpName() + "_" + NEIGHBORHOOD_CHANGE_VERSION_STATE_SUFFIX, + traversalContext.getRuntimeContext().getConfiguration().getString(SYSTEM_STATE_BACKEND_TYPE)); + changeVersionDescriptor.withKeyGroup(keyGroup); + changeVersionDescriptor.withKeyGroupAssigner(keyGroupAssigner); + this.neighborhoodChangeVersions = StateFactory.buildKeyValueState(changeVersionDescriptor, + traversalContext.getRuntimeContext().getConfiguration()); if (recoverWindowId > 1) { this.vertexUpdateValues.manage().operate().setCheckpointId(recoverWindowId - 1); this.vertexUpdateValues.manage().operate().recover(); + this.neighborhoodChangeVersions.manage().operate().setCheckpointId(recoverWindowId - 1); + this.neighborhoodChangeVersions.manage().operate().recover(); } } @@ -129,6 +141,9 @@ public void init(ITraversalRequest traversalRequest) { // false when called after the first time to avoid redundant invocation. if (vertexId != null && needInit(vertexId)) { RowVertex vertex = (RowVertex) algorithmCtx.loadVertex(); + if (vertex == null) { + vertex = (RowVertex) algorithmCtx.getIncVCTraversalCtx().getTemporaryGraph().getVertex(); + } if (vertex != null) { algorithmCtx.setVertexId(vertex.getId()); Row newValue = getVertexNewValue(vertex.getId()); @@ -146,8 +161,15 @@ public Row getVertexNewValue(Object vertexId) { return vertexUpdateValues.get(vertexId); } + public long getNeighborhoodChangeVersion(Object vertexId) { + Long version = neighborhoodChangeVersions.get(vertexId); + return version == null ? Long.MIN_VALUE : version; + } + @Override public void evolve(Object vertexId, TemporaryGraph temporaryGraph) { + neighborhoodChangeVersions.put(vertexId, + traversalContext.getRuntimeContext().getWindowId()); if (!materializeInFinish) { IVertex vertex = temporaryGraph.getVertex(); List> edges = temporaryGraph.getEdges(); @@ -182,6 +204,9 @@ public void compute(Object vertexId, Iterator messages) { } } else { vertex = (RowVertex) algorithmCtx.loadVertex(); + if (vertex == null) { + vertex = (RowVertex) algorithmCtx.getIncVCTraversalCtx().getTemporaryGraph().getVertex(); + } } if (vertex != null) { Row newValue = getVertexNewValue(vertex.getId()); @@ -193,6 +218,9 @@ public void compute(Object vertexId, Iterator messages) { public void finish(Object vertexId, MutableGraph mutableGraph) { algorithmCtx.setVertexId(vertexId); RowVertex graphVertex = (RowVertex) algorithmCtx.loadVertex(); + if (graphVertex == null) { + graphVertex = (RowVertex) algorithmCtx.getIncVCTraversalCtx().getTemporaryGraph().getVertex(); + } if (graphVertex != null) { Row newValue = getVertexNewValue(graphVertex.getId()); userFunction.finish(graphVertex, Optional.ofNullable(newValue)); @@ -221,6 +249,9 @@ public void finish() { this.vertexUpdateValues.manage().operate().setCheckpointId(windowId); this.vertexUpdateValues.manage().operate().finish(); this.vertexUpdateValues.manage().operate().archive(); + this.neighborhoodChangeVersions.manage().operate().setCheckpointId(windowId); + this.neighborhoodChangeVersions.manage().operate().finish(); + this.neighborhoodChangeVersions.manage().operate().archive(); } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicRuntimeContext.java b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicRuntimeContext.java index d929ae441..8203a75fd 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicRuntimeContext.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicRuntimeContext.java @@ -29,7 +29,7 @@ import org.apache.geaflow.api.graph.function.vc.VertexCentricTraversalFunction.TraversalVertexQuery; import org.apache.geaflow.common.config.Configuration; import org.apache.geaflow.common.iterator.CloseableIterator; -import org.apache.geaflow.dsl.common.algo.AlgorithmRuntimeContext; +import org.apache.geaflow.dsl.common.algo.AlgorithmSamplingRuntimeContext; import org.apache.geaflow.dsl.common.data.Row; import org.apache.geaflow.dsl.common.data.RowEdge; import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; @@ -46,7 +46,7 @@ import org.apache.geaflow.state.pushdown.filter.InEdgeFilter; import org.apache.geaflow.state.pushdown.filter.OutEdgeFilter; -public class GeaFlowAlgorithmDynamicRuntimeContext implements AlgorithmRuntimeContext { +public class GeaFlowAlgorithmDynamicRuntimeContext implements AlgorithmSamplingRuntimeContext { private final IncVertexCentricTraversalFuncContext incVCTraversalCtx; @@ -175,6 +175,16 @@ public List loadStaticEdges(EdgeDirection direction) { } } + @Override + public long getSamplingSnapshotVersion() { + return incVCTraversalCtx.getRuntimeContext().getWindowId(); + } + + @Override + public long getNeighborhoodChangeVersion(Object vertexId) { + return traversalFunction.getNeighborhoodChangeVersion(vertexId); + } + @Override public CloseableIterator loadStaticEdgesIterator(EdgeDirection direction) { switch (direction) { diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmRuntimeContext.java b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmRuntimeContext.java index 7696b4f10..d3d88b22a 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmRuntimeContext.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmRuntimeContext.java @@ -28,7 +28,7 @@ import org.apache.geaflow.common.config.Configuration; import org.apache.geaflow.common.exception.GeaflowRuntimeException; import org.apache.geaflow.common.iterator.CloseableIterator; -import org.apache.geaflow.dsl.common.algo.AlgorithmRuntimeContext; +import org.apache.geaflow.dsl.common.algo.AlgorithmSamplingRuntimeContext; import org.apache.geaflow.dsl.common.data.Row; import org.apache.geaflow.dsl.common.data.RowEdge; import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException; @@ -42,7 +42,7 @@ import org.apache.geaflow.state.pushdown.filter.InEdgeFilter; import org.apache.geaflow.state.pushdown.filter.OutEdgeFilter; -public class GeaFlowAlgorithmRuntimeContext implements AlgorithmRuntimeContext { +public class GeaFlowAlgorithmRuntimeContext implements AlgorithmSamplingRuntimeContext { private final VertexCentricTraversalFuncContext traversalContext; @@ -113,6 +113,11 @@ public List loadStaticEdges(EdgeDirection direction) { return loadEdges(direction); } + @Override + public long getSamplingSnapshotVersion() { + return traversalContext.getRuntimeContext().getWindowId(); + } + @Override public CloseableIterator loadStaticEdgesIterator(EdgeDirection direction) { switch (direction) { diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicAggTraversalFunctionTest.java b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicAggTraversalFunctionTest.java new file mode 100644 index 000000000..2c0cd92d8 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicAggTraversalFunctionTest.java @@ -0,0 +1,103 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.dsl.runtime.engine; + +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.HashSet; +import org.apache.geaflow.api.graph.function.vc.IncVertexCentricTraversalFunction.IncVertexCentricTraversalFuncContext; +import org.apache.geaflow.api.graph.function.vc.base.IncVertexCentricFunction.TemporaryGraph; +import org.apache.geaflow.dsl.common.algo.AlgorithmUserFunction; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.types.GraphSchema; +import org.apache.geaflow.state.KeyValueState; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class GeaFlowAlgorithmDynamicAggTraversalFunctionTest { + + @Test + public void testEvolvePersistsNeighborhoodChangeVersion() throws Exception { + GeaFlowAlgorithmDynamicAggTraversalFunction function = + new GeaFlowAlgorithmDynamicAggTraversalFunction(mock(GraphSchema.class), + mock(AlgorithmUserFunction.class), new Object[0]); + IncVertexCentricTraversalFuncContext traversalContext = + mock(IncVertexCentricTraversalFuncContext.class, RETURNS_DEEP_STUBS); + TemporaryGraph temporaryGraph = mock(TemporaryGraph.class); + KeyValueState changeVersions = mock(KeyValueState.class); + when(traversalContext.getRuntimeContext().getWindowId()).thenReturn(7L); + when(temporaryGraph.getEdges()).thenReturn(Collections.emptyList()); + when(changeVersions.get(2L)).thenReturn(5L); + setField(function, "traversalContext", traversalContext); + setField(function, "neighborhoodChangeVersions", changeVersions); + + function.evolve(2L, temporaryGraph); + + verify(changeVersions).put(2L, 7L); + Assert.assertEquals(function.getNeighborhoodChangeVersion(2L), 5L); + } + + @Test + public void testMissingChangeVersionIsStatic() throws Exception { + GeaFlowAlgorithmDynamicAggTraversalFunction function = + new GeaFlowAlgorithmDynamicAggTraversalFunction(mock(GraphSchema.class), + mock(AlgorithmUserFunction.class), new Object[0]); + setField(function, "neighborhoodChangeVersions", mock(KeyValueState.class)); + + Assert.assertEquals(function.getNeighborhoodChangeVersion(1L), Long.MIN_VALUE); + } + + @Test + public void testFinishCheckpointsNeighborhoodChangeVersions() throws Exception { + AlgorithmUserFunction userFunction = mock(AlgorithmUserFunction.class); + GeaFlowAlgorithmDynamicAggTraversalFunction function = + new GeaFlowAlgorithmDynamicAggTraversalFunction(mock(GraphSchema.class), + userFunction, new Object[0]); + IncVertexCentricTraversalFuncContext traversalContext = + mock(IncVertexCentricTraversalFuncContext.class, RETURNS_DEEP_STUBS); + KeyValueState vertexValues = mock(KeyValueState.class, RETURNS_DEEP_STUBS); + KeyValueState changeVersions = mock(KeyValueState.class, RETURNS_DEEP_STUBS); + GeaFlowAlgorithmDynamicRuntimeContext algorithmContext = + mock(GeaFlowAlgorithmDynamicRuntimeContext.class); + when(traversalContext.getRuntimeContext().getWindowId()).thenReturn(7L); + setField(function, "traversalContext", traversalContext); + setField(function, "algorithmCtx", algorithmContext); + setField(function, "initVertices", new HashSet<>()); + setField(function, "vertexUpdateValues", vertexValues); + setField(function, "neighborhoodChangeVersions", changeVersions); + + function.finish(); + + verify(changeVersions.manage().operate()).setCheckpointId(7L); + verify(changeVersions.manage().operate()).finish(); + verify(changeVersions.manage().operate()).archive(); + } + + private void setField(Object target, String name, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicRuntimeContextTest.java b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicRuntimeContextTest.java new file mode 100644 index 000000000..5cb2b699b --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowAlgorithmDynamicRuntimeContextTest.java @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.dsl.runtime.engine; + +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.Arrays; +import java.util.Collections; +import org.apache.geaflow.api.context.RuntimeContext; +import org.apache.geaflow.api.graph.sampling.SubgraphSamplingSpec; +import org.apache.geaflow.api.graph.function.vc.IncVertexCentricTraversalFunction.IncVertexCentricTraversalFuncContext; +import org.apache.geaflow.api.graph.function.vc.IncVertexCentricTraversalFunction.TraversalGraphSnapShot; +import org.apache.geaflow.api.graph.function.vc.IncVertexCentricTraversalFunction.TraversalHistoricalGraph; +import org.apache.geaflow.api.graph.function.vc.VertexCentricTraversalFunction.TraversalEdgeQuery; +import org.apache.geaflow.api.graph.function.vc.VertexCentricTraversalFunction.TraversalVertexQuery; +import org.apache.geaflow.api.graph.function.vc.base.IncVertexCentricFunction.TemporaryGraph; +import org.apache.geaflow.common.iterator.CloseableIterator; +import org.apache.geaflow.common.type.primitive.LongType; +import org.apache.geaflow.dsl.common.algo.AlgorithmUserFunction; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.RowEdge; +import org.apache.geaflow.dsl.common.data.RowVertex; +import org.apache.geaflow.dsl.common.types.GraphSchema; +import org.apache.geaflow.dsl.common.data.impl.types.ObjectEdge; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.IEdge; +import org.apache.geaflow.state.pushdown.filter.OutEdgeFilter; +import org.apache.geaflow.state.sampling.LocalNeighborhood; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class GeaFlowAlgorithmDynamicRuntimeContextTest { + + @Test + public void testSamplingUsesMaterializedSnapshotOnly() { + IncVertexCentricTraversalFuncContext traversalContext = mock( + IncVertexCentricTraversalFuncContext.class); + TraversalHistoricalGraph historicalGraph = mock(TraversalHistoricalGraph.class); + TraversalGraphSnapShot snapshot = mock(TraversalGraphSnapShot.class); + TraversalVertexQuery vertexQuery = mock(TraversalVertexQuery.class); + TraversalEdgeQuery edgeQuery = mock(TraversalEdgeQuery.class); + RuntimeContext runtimeContext = mock(RuntimeContext.class); + TemporaryGraph temporaryGraph = mock(TemporaryGraph.class); + CloseableIterator> edgeIterator = mock(CloseableIterator.class); + GraphSchema graphSchema = mock(GraphSchema.class); + + RowEdge edge = new ObjectEdge(1L, 2L, ObjectRow.create(1.0D)); + edge.setDirect(EdgeDirection.OUT); + when(edgeIterator.hasNext()).thenReturn(true, false); + when(edgeIterator.next()).thenReturn(edge); + when(traversalContext.getHistoricalGraph()).thenReturn(historicalGraph); + when(historicalGraph.getSnapShot(0L)).thenReturn(snapshot); + when(snapshot.vertex()).thenReturn(vertexQuery); + when(snapshot.edges()).thenReturn(edgeQuery); + when(edgeQuery.getOutEdges()).thenReturn(Collections.singletonList(edge)); + when(edgeQuery.getEdges(OutEdgeFilter.getInstance())).thenReturn(edgeIterator); + when(traversalContext.getRuntimeContext()).thenReturn(runtimeContext); + when(runtimeContext.getWindowId()).thenReturn(7L); + when(traversalContext.getTemporaryGraph()).thenReturn(temporaryGraph); + when(temporaryGraph.getEdges()).thenReturn(Arrays.asList(edge)); + doReturn(LongType.INSTANCE).when(graphSchema).getIdType(); + + GeaFlowAlgorithmDynamicRuntimeContext context = new GeaFlowAlgorithmDynamicRuntimeContext( + new GeaFlowAlgorithmDynamicAggTraversalFunction(graphSchema, + mock(AlgorithmUserFunction.class), new Object[0]), traversalContext, graphSchema); + RowVertex vertex = mock(RowVertex.class); + when(vertex.getId()).thenReturn(1L); + + LocalNeighborhood neighborhood = context.sampleOneHop(vertex, EdgeDirection.OUT, -1); + + Assert.assertEquals(neighborhood.getEdges().size(), 1); + Assert.assertEquals(neighborhood.getSnapshotVersion(), 7L); + verify(temporaryGraph, never()).getEdges(); + } + + @Test + public void testSamplingSpecPropagatesVersionAndClosesStaticIterator() { + IncVertexCentricTraversalFuncContext traversalContext = mock( + IncVertexCentricTraversalFuncContext.class); + TraversalHistoricalGraph historicalGraph = mock(TraversalHistoricalGraph.class); + TraversalGraphSnapShot snapshot = mock(TraversalGraphSnapShot.class); + TraversalVertexQuery vertexQuery = mock(TraversalVertexQuery.class); + TraversalEdgeQuery edgeQuery = mock(TraversalEdgeQuery.class); + RuntimeContext runtimeContext = mock(RuntimeContext.class); + CloseableIterator> edgeIterator = mock(CloseableIterator.class); + GraphSchema graphSchema = mock(GraphSchema.class); + + RowEdge first = edge(1L, 2L); + RowEdge second = edge(1L, 3L); + RowEdge third = edge(1L, 4L); + when(edgeIterator.hasNext()).thenReturn(true, true, true, false); + when(edgeIterator.next()).thenReturn(first, second, third); + when(traversalContext.getHistoricalGraph()).thenReturn(historicalGraph); + when(historicalGraph.getSnapShot(0L)).thenReturn(snapshot); + when(snapshot.vertex()).thenReturn(vertexQuery); + when(snapshot.edges()).thenReturn(edgeQuery); + when(edgeQuery.getEdges(OutEdgeFilter.getInstance())).thenReturn(edgeIterator); + when(traversalContext.getRuntimeContext()).thenReturn(runtimeContext); + when(runtimeContext.getWindowId()).thenReturn(7L); + doReturn(LongType.INSTANCE).when(graphSchema).getIdType(); + + GeaFlowAlgorithmDynamicRuntimeContext context = new GeaFlowAlgorithmDynamicRuntimeContext( + new GeaFlowAlgorithmDynamicAggTraversalFunction(graphSchema, + mock(AlgorithmUserFunction.class), new Object[0]), traversalContext, graphSchema); + RowVertex vertex = mock(RowVertex.class); + when(vertex.getId()).thenReturn(1L); + + LocalNeighborhood neighborhood = context.sampleOneHop(vertex, + new SubgraphSamplingSpec(1, 1, EdgeDirection.OUT, 100L, 17L), 9L); + + Assert.assertEquals(neighborhood.getEdges().size(), 1); + Assert.assertEquals(neighborhood.getSnapshotVersion(), 7L); + Assert.assertEquals(neighborhood.getSamplingVersion(), 9L); + verify(edgeIterator).close(); + } + + private RowEdge edge(long source, long target) { + RowEdge edge = new ObjectEdge(source, target, ObjectRow.create(1.0D)); + edge.setDirect(EdgeDirection.OUT); + return edge; + } +} diff --git a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/sampling/DeterministicNeighborSampler.java b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/sampling/DeterministicNeighborSampler.java new file mode 100644 index 000000000..58e2ec04a --- /dev/null +++ b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/sampling/DeterministicNeighborSampler.java @@ -0,0 +1,292 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.state.sampling; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.PriorityQueue; +import java.util.function.Function; +import org.apache.geaflow.model.graph.IGraphElementWithLabelField; +import org.apache.geaflow.model.graph.IGraphElementWithTimeField; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.IEdge; + +/** + * Storage-independent seeded one-hop neighbor sampling. + * The supplied ID encoder must return a stable, canonical byte representation across workers. + */ +public final class DeterministicNeighborSampler { + + public static final long DEFAULT_MAX_RETURNED_EDGES = 100000L; + + private DeterministicNeighborSampler() { + } + + public static List> sample(K vertexId, + Iterable> edges, + EdgeDirection direction, + int fanout, + Comparator idComparator, + Function idEncoder) { + return sample(vertexId, edges, direction, fanout, idComparator, + DEFAULT_MAX_RETURNED_EDGES, 0L, 0L, idEncoder); + } + + public static List> sample(K vertexId, + Iterable> edges, + EdgeDirection direction, + int fanout, + Comparator idComparator, + long maxReturnedEdges, + long seed, + long samplingVersion, + Function idEncoder) { + return select(vertexId, edges, direction, fanout, idComparator, maxReturnedEdges, + seed, samplingVersion, idEncoder, true); + } + + /** Project an already direction-filtered local neighborhood to a smaller fanout. */ + public static List> project(K vertexId, + Iterable> edges, + EdgeDirection direction, + int fanout, + Comparator idComparator, + Function idEncoder) { + return project(vertexId, edges, direction, fanout, idComparator, + DEFAULT_MAX_RETURNED_EDGES, 0L, 0L, idEncoder); + } + + public static List> project(K vertexId, + Iterable> edges, + EdgeDirection direction, + int fanout, + Comparator idComparator, + long maxReturnedEdges, + long seed, + long samplingVersion, + Function idEncoder) { + return select(vertexId, edges, direction, fanout, idComparator, maxReturnedEdges, + seed, samplingVersion, idEncoder, false); + } + + private static List> select(K vertexId, + Iterable> edges, + EdgeDirection direction, + int fanout, + Comparator idComparator, + long maxReturnedEdges, + long seed, + long samplingVersion, + Function idEncoder, + boolean filterAndNormalize) { + validate(vertexId, edges, direction, fanout, idComparator, maxReturnedEdges, idEncoder); + long vertexHash = hashId(vertexId, idEncoder); + Comparator> rankComparator = (left, right) -> { + int result = Long.compareUnsigned(left.score, right.score); + return result != 0 ? result + : compareIds(left.neighborId, right.neighborId, idComparator, idEncoder); + }; + Map> selected = new HashMap<>(); + PriorityQueue> worstFirst = + fanout < 0 ? null : new PriorityQueue<>(fanout, rankComparator.reversed()); + + for (IEdge sourceEdge : edges) { + Objects.requireNonNull(sourceEdge, "edge"); + if (filterAndNormalize && !matchesDirection(sourceEdge, direction)) { + continue; + } + IEdge edge = filterAndNormalize ? normalize(sourceEdge) : sourceEdge; + K neighborId = neighborId(vertexId, edge); + if (neighborId == null) { + continue; + } + NeighborGroup group = selected.get(neighborId); + if (group != null) { + group.edges.add(edge); + continue; + } + + group = new NeighborGroup<>(neighborId, + sampleScore(seed, samplingVersion, vertexHash, direction, + hashId(neighborId, idEncoder)), edge); + if (fanout < 0 || selected.size() < fanout) { + selected.put(neighborId, group); + if (worstFirst != null) { + worstFirst.add(group); + } + } else if (rankComparator.compare(group, worstFirst.peek()) < 0) { + NeighborGroup removed = worstFirst.remove(); + selected.remove(removed.neighborId); + selected.put(neighborId, group); + worstFirst.add(group); + } + } + + List> groups = new ArrayList<>(selected.values()); + groups.sort(rankComparator); + List> result = new ArrayList<>(); + for (NeighborGroup group : groups) { + group.edges.sort((left, right) -> compareEdges(left, right, idComparator, idEncoder)); + result.addAll(group.edges); + if (result.size() > maxReturnedEdges) { + throw new IllegalStateException(String.format( + "one-hop sampling edge limit exceeded, vertexId=%s, actual=%s, limit=%s", + vertexId, result.size(), maxReturnedEdges)); + } + } + return result; + } + + private static void validate(Object vertexId, Iterable edges, EdgeDirection direction, + int fanout, Comparator idComparator, long maxReturnedEdges, + Function idEncoder) { + Objects.requireNonNull(vertexId, "vertexId"); + Objects.requireNonNull(edges, "edges"); + Objects.requireNonNull(direction, "direction"); + Objects.requireNonNull(idComparator, "idComparator"); + Objects.requireNonNull(idEncoder, "idEncoder"); + if (fanout == 0 || fanout < -1) { + throw new IllegalArgumentException("fanout must be -1 or greater than zero"); + } + if (maxReturnedEdges < 1) { + throw new IllegalArgumentException("maxReturnedEdges must be greater than zero"); + } + } + + private static boolean matchesDirection(IEdge edge, EdgeDirection direction) { + return direction == EdgeDirection.BOTH || edge.getDirect() == direction; + } + + private static IEdge normalize(IEdge edge) { + if (edge.getDirect() != EdgeDirection.IN) { + return edge; + } + IEdge reversed = edge.reverse(); + // The reversed endpoints now represent the logical outgoing direction. + reversed.setDirect(EdgeDirection.OUT); + return reversed; + } + + private static K neighborId(K vertexId, IEdge edge) { + if (Objects.equals(vertexId, edge.getSrcId())) { + return edge.getTargetId(); + } + if (Objects.equals(vertexId, edge.getTargetId())) { + return edge.getSrcId(); + } + return null; + } + + private static long sampleScore(long seed, long samplingVersion, long vertexHash, + EdgeDirection direction, long neighborHash) { + long value = mix64(seed) ^ Long.rotateLeft(mix64(samplingVersion), 11); + value ^= Long.rotateLeft(vertexHash, 23); + value ^= Long.rotateLeft(mix64(direction.ordinal()), 37); + value ^= Long.rotateLeft(neighborHash, 47); + return mix64(value); + } + + private static long hashId(K value, Function idEncoder) { + return stableHash(Objects.requireNonNull(idEncoder.apply(value), "idEncoder result")); + } + + private static long stableHash(byte[] value) { + long hash = 0xcbf29ce484222325L; + for (byte current : value) { + hash ^= current & 0xffL; + hash *= 0x100000001b3L; + } + return mix64(hash); + } + + private static long mix64(long value) { + value = (value ^ (value >>> 30)) * 0xbf58476d1ce4e5b9L; + value = (value ^ (value >>> 27)) * 0x94d049bb133111ebL; + return value ^ (value >>> 31); + } + + private static int compareIds(K left, K right, Comparator idComparator, + Function idEncoder) { + int result = idComparator.compare(left, right); + return result != 0 ? result : compareBytes( + Objects.requireNonNull(idEncoder.apply(left), "idEncoder result"), + Objects.requireNonNull(idEncoder.apply(right), "idEncoder result")); + } + + private static int compareEdges(IEdge left, IEdge right, + Comparator idComparator, + Function idEncoder) { + int result = compareIds(left.getSrcId(), right.getSrcId(), idComparator, idEncoder); + if (result == 0) { + result = compareIds(left.getTargetId(), right.getTargetId(), idComparator, idEncoder); + } + if (result == 0) { + result = left.getDirect().compareTo(right.getDirect()); + } + if (result == 0) { + result = String.valueOf(labelOf(left)).compareTo(String.valueOf(labelOf(right))); + } + if (result == 0) { + result = String.valueOf(timeOf(left)).compareTo(String.valueOf(timeOf(right))); + } + if (result == 0) { + result = String.valueOf(left.getValue()).compareTo(String.valueOf(right.getValue())); + } + return result; + } + + private static int compareBytes(byte[] left, byte[] right) { + int length = Math.min(left.length, right.length); + for (int i = 0; i < length; i++) { + int result = Integer.compare(left[i] & 0xff, right[i] & 0xff); + if (result != 0) { + return result; + } + } + return Integer.compare(left.length, right.length); + } + + private static String labelOf(IEdge edge) { + return edge instanceof IGraphElementWithLabelField + ? ((IGraphElementWithLabelField) edge).getLabel() : null; + } + + private static Long timeOf(IEdge edge) { + return edge instanceof IGraphElementWithTimeField + ? ((IGraphElementWithTimeField) edge).getTime() : null; + } + + private static final class NeighborGroup { + + private final K neighborId; + private final long score; + private final List> edges = new ArrayList<>(); + + private NeighborGroup(K neighborId, long score, IEdge firstEdge) { + this.neighborId = neighborId; + this.score = score; + this.edges.add(firstEdge); + } + } +} diff --git a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/sampling/LocalNeighborhood.java b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/sampling/LocalNeighborhood.java new file mode 100644 index 000000000..405f6aedf --- /dev/null +++ b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/sampling/LocalNeighborhood.java @@ -0,0 +1,110 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.state.sampling; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Objects; +import java.util.function.Function; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.IEdge; +import org.apache.geaflow.model.graph.vertex.IVertex; + +/** + * Versioned one-hop state owned by a single vertex. + */ +public class LocalNeighborhood implements Serializable { + + private final IVertex vertex; + private final List> edges; + private final long snapshotVersion; + private final long samplingVersion; + + public LocalNeighborhood(IVertex vertex, List> edges, + long snapshotVersion) { + this(vertex, edges, snapshotVersion, snapshotVersion); + } + + public LocalNeighborhood(IVertex vertex, List> edges, + long snapshotVersion, long samplingVersion) { + this.vertex = Objects.requireNonNull(vertex, "vertex"); + this.edges = new ArrayList<>(Objects.requireNonNull(edges, "edges")); + for (IEdge edge : this.edges) { + Objects.requireNonNull(edge, "edge"); + } + this.snapshotVersion = snapshotVersion; + this.samplingVersion = samplingVersion; + } + + public IVertex getVertex() { + return vertex; + } + + public List> getEdges() { + return Collections.unmodifiableList(edges); + } + + public long getSnapshotVersion() { + return snapshotVersion; + } + + public long getSamplingVersion() { + return samplingVersion; + } + + public boolean matches(long expectedSnapshotVersion, long expectedSamplingVersion) { + return snapshotVersion == expectedSnapshotVersion + && samplingVersion == expectedSamplingVersion; + } + + public LocalNeighborhood revalidate(IVertex currentVertex, + long currentSnapshotVersion) { + if (currentSnapshotVersion < snapshotVersion) { + throw new IllegalArgumentException("cannot revalidate a neighborhood to an older snapshot"); + } + return new LocalNeighborhood<>(currentVertex, edges, currentSnapshotVersion, samplingVersion); + } + + /** + * Create a bounded view of this already direction-filtered neighborhood. + */ + public LocalNeighborhood project(EdgeDirection direction, int fanout, + Comparator idComparator, + Function idEncoder) { + return new LocalNeighborhood<>(vertex, + DeterministicNeighborSampler.project(vertex.getId(), edges, direction, fanout, + idComparator, idEncoder), + snapshotVersion, samplingVersion); + } + + public LocalNeighborhood project(EdgeDirection direction, int fanout, + Comparator idComparator, + long maxReturnedEdges, + long seed, + Function idEncoder) { + return new LocalNeighborhood<>(vertex, DeterministicNeighborSampler.project(vertex.getId(), + edges, direction, fanout, idComparator, maxReturnedEdges, seed, samplingVersion, + idEncoder), + snapshotVersion, samplingVersion); + } +} diff --git a/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/sampling/DeterministicNeighborSamplerTest.java b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/sampling/DeterministicNeighborSamplerTest.java new file mode 100644 index 000000000..7e22133db --- /dev/null +++ b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/sampling/DeterministicNeighborSamplerTest.java @@ -0,0 +1,291 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.state.sampling; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.IEdge; +import org.apache.geaflow.model.graph.edge.impl.ValueEdge; +import org.apache.geaflow.model.graph.vertex.impl.ValueVertex; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class DeterministicNeighborSamplerTest { + + @Test + public void testSamplingIsBoundedAndIndependentOfInputOrder() { + List> first = Arrays.asList( + edge(1L, 2L), edge(1L, 3L), edge(1L, 4L)); + List> second = Arrays.asList( + edge(1L, 4L), edge(1L, 2L), edge(1L, 3L)); + + List> sampledFirst = + DeterministicNeighborSampler.sample(1L, first, EdgeDirection.OUT, 2, + Long::compare, DeterministicNeighborSamplerTest::longBytes); + List> sampledSecond = + DeterministicNeighborSampler.sample(1L, second, EdgeDirection.OUT, 2, + Long::compare, DeterministicNeighborSamplerTest::longBytes); + + Assert.assertEquals(sampledFirst.size(), 2); + Assert.assertEquals(targetIds(sampledFirst), targetIds(sampledSecond)); + } + + @Test + public void testPositiveFanoutDoesNotMaterializeAllCandidates() { + List> edges = new ArrayList<>(); + for (long target = 2L; target < 102L; target++) { + edges.add(edge(1L, target)); + } + + List> sampled = DeterministicNeighborSampler.sample( + 1L, edges, EdgeDirection.OUT, 3, Long::compare, 3L, 17L, 9L, + DeterministicNeighborSamplerTest::longBytes); + + Assert.assertEquals(sampled.stream().map(IEdge::getTargetId).distinct().count(), 3L); + Assert.assertEquals(sampled.size(), 3); + } + + @Test + public void testSeedAndVersionAreStableAndInputOrderIndependent() { + List> first = new ArrayList<>(); + for (long target = 2L; target < 42L; target++) { + first.add(edge(1L, target)); + } + List> reversed = new ArrayList<>(first); + java.util.Collections.reverse(reversed); + + List firstSample = targetIds(DeterministicNeighborSampler.sample( + 1L, first, EdgeDirection.OUT, 5, Long::compare, 5L, 123L, 7L, + DeterministicNeighborSamplerTest::longBytes)); + List reorderedSample = targetIds(DeterministicNeighborSampler.sample( + 1L, reversed, EdgeDirection.OUT, 5, Long::compare, 5L, 123L, 7L, + DeterministicNeighborSamplerTest::longBytes)); + List nextVersionSample = targetIds(DeterministicNeighborSampler.sample( + 1L, first, EdgeDirection.OUT, 5, Long::compare, 5L, 123L, 8L, + DeterministicNeighborSamplerTest::longBytes)); + + Assert.assertEquals(firstSample, reorderedSample); + Assert.assertNotEquals(firstSample, nextVersionSample); + } + + @Test + public void testDirectionAndUnlimitedFanout() { + IEdge out = edge(1L, 2L); + IEdge in = edge(1L, 3L); + in.setDirect(EdgeDirection.IN); + + List> sampledIn = DeterministicNeighborSampler.sample( + 1L, Arrays.asList(out, in), EdgeDirection.IN, -1, Long::compare, + DeterministicNeighborSamplerTest::longBytes); + Assert.assertEquals(sampledIn.size(), 1); + Assert.assertEquals(sampledIn.get(0).getSrcId(), Long.valueOf(3L)); + Assert.assertEquals(sampledIn.get(0).getTargetId(), Long.valueOf(1L)); + Assert.assertEquals(sampledIn.get(0).getDirect(), EdgeDirection.OUT); + Assert.assertEquals( + DeterministicNeighborSampler.sample(1L, Arrays.asList(out, in), EdgeDirection.BOTH, + -1, Long::compare, DeterministicNeighborSamplerTest::longBytes).size(), + 2); + } + + @Test + public void testIgnoresUnrelatedEdges() { + IEdge related = edge(1L, 2L); + IEdge unrelated = edge(3L, 4L); + List> edges = Arrays.asList(related, unrelated); + + List> sampled = DeterministicNeighborSampler.sample( + 1L, edges, EdgeDirection.OUT, -1, Long::compare, 1L, 0L, 0L, + DeterministicNeighborSamplerTest::longBytes); + List> projected = DeterministicNeighborSampler.project( + 1L, edges, EdgeDirection.OUT, -1, Long::compare, + DeterministicNeighborSamplerTest::longBytes); + + Assert.assertEquals(sampled, Collections.singletonList(related)); + Assert.assertEquals(projected, Collections.singletonList(related)); + } + + @Test + public void testIncomingNormalizationDoesNotMutateInputEdge() { + IEdge incoming = edge(2L, 1L); + incoming.setDirect(EdgeDirection.IN); + + List> sampled = DeterministicNeighborSampler.sample( + 1L, Collections.singletonList(incoming), EdgeDirection.IN, -1, Long::compare, + DeterministicNeighborSamplerTest::longBytes); + + Assert.assertEquals(incoming.getSrcId(), Long.valueOf(2L)); + Assert.assertEquals(incoming.getTargetId(), Long.valueOf(1L)); + Assert.assertEquals(incoming.getDirect(), EdgeDirection.IN); + Assert.assertEquals(sampled.get(0).getSrcId(), Long.valueOf(1L)); + Assert.assertEquals(sampled.get(0).getTargetId(), Long.valueOf(2L)); + Assert.assertEquals(sampled.get(0).getDirect(), EdgeDirection.OUT); + } + + @Test + public void testComparatorTieUsesStableIdFallback() { + List> edges = Arrays.asList(edge(1L, 3L), edge(1L, 2L)); + + List first = targetIds(DeterministicNeighborSampler.sample( + 1L, edges, EdgeDirection.OUT, 1, (left, right) -> 0, + 100L, 17L, 7L, DeterministicNeighborSamplerTest::longBytes)); + List second = targetIds(DeterministicNeighborSampler.sample( + 1L, Arrays.asList(edges.get(1), edges.get(0)), EdgeDirection.OUT, 1, + (left, right) -> 0, 100L, 17L, 7L, + DeterministicNeighborSamplerTest::longBytes)); + + Assert.assertEquals(first, second); + Assert.assertEquals(first.size(), 1); + } + + @Test + public void testFanoutCountsNeighborsAndKeepsSelectedParallelEdges() { + List> edges = Arrays.asList( + edge(1L, 2L), edgeWithValue(1L, 2L, "parallel"), edge(1L, 3L), edge(1L, 4L)); + + List> sampled = DeterministicNeighborSampler.sample( + 1L, edges, EdgeDirection.OUT, 2, Long::compare, + DeterministicNeighborSamplerTest::longBytes); + Map allCounts = edges.stream().collect(Collectors.groupingBy( + IEdge::getTargetId, Collectors.counting())); + Map sampledCounts = sampled.stream().collect(Collectors.groupingBy( + IEdge::getTargetId, Collectors.counting())); + + Assert.assertEquals(sampledCounts.size(), 2); + for (Map.Entry entry : sampledCounts.entrySet()) { + Assert.assertEquals(entry.getValue(), allCounts.get(entry.getKey())); + } + } + + @Test + public void testNeighborhoodMatchesSnapshotAndSamplingVersion() { + LocalNeighborhood neighborhood = new LocalNeighborhood<>( + new ValueVertex<>(1L, "vertex"), Arrays.asList(edge(1L, 2L)), 7L, 3L); + + Assert.assertTrue(neighborhood.matches(7L, 3L)); + Assert.assertFalse(neighborhood.matches(8L, 3L)); + Assert.assertFalse(neighborhood.matches(7L, 4L)); + } + + @Test(expectedExceptions = IllegalStateException.class) + public void testRejectsReturnedEdgeOverflow() { + DeterministicNeighborSampler.sample(1L, + Arrays.asList(edge(1L, 2L), edge(1L, 3L)), EdgeDirection.OUT, -1, + Long::compare, 1, 0L, 0L, DeterministicNeighborSamplerTest::longBytes); + } + + @Test + public void testSamplesIdsWithoutCallingToString() { + List> edges = Arrays.asList( + stableEdge(1L, 2L, "first"), stableEdge(1L, 2L, "second"), + stableEdge(1L, 3L, "third")); + Comparator comparator = (left, right) -> 0; + + List> first = DeterministicNeighborSampler.sample( + new StableId(1L), edges, EdgeDirection.OUT, -1, comparator, 100L, 17L, 7L, + id -> longBytes(id.value)); + List> second = DeterministicNeighborSampler.sample( + new StableId(1L), Arrays.asList(edges.get(2), edges.get(1), edges.get(0)), + EdgeDirection.OUT, -1, comparator, 100L, 17L, 7L, id -> longBytes(id.value)); + + if (first.size() != second.size()) { + Assert.fail("sampling result sizes differ"); + } + for (int i = 0; i < first.size(); i++) { + if (first.get(i) != second.get(i)) { + Assert.fail("sampling result order differs"); + } + } + } + + @Test(expectedExceptions = NullPointerException.class, + expectedExceptionsMessageRegExp = "idEncoder result") + public void testRejectsNullIdEncoding() { + DeterministicNeighborSampler.sample(1L, Collections.singletonList(edge(1L, 2L)), + EdgeDirection.OUT, -1, Long::compare, 100L, 0L, 0L, id -> null); + } + + @Test(expectedExceptions = NullPointerException.class, + expectedExceptionsMessageRegExp = "idEncoder") + public void testRejectsNullIdEncoder() { + DeterministicNeighborSampler.sample(1L, Collections.singletonList(edge(1L, 2L)), + EdgeDirection.OUT, -1, Long::compare, 100L, 0L, 0L, + (Function) null); + } + + private static IEdge edge(long source, long target) { + return edgeWithValue(source, target, "value"); + } + + private static IEdge edgeWithValue(long source, long target, String value) { + ValueEdge edge = new ValueEdge<>(source, target, value); + edge.setDirect(EdgeDirection.OUT); + return edge; + } + + private static List targetIds(List> edges) { + return edges.stream().map(IEdge::getTargetId).collect(Collectors.toList()); + } + + private static IEdge stableEdge(long source, long target, String value) { + ValueEdge edge = new ValueEdge<>(new StableId(source), + new StableId(target), value); + edge.setDirect(EdgeDirection.OUT); + return edge; + } + + private static byte[] longBytes(long value) { + return new byte[]{ + (byte) (value >>> 56), (byte) (value >>> 48), (byte) (value >>> 40), + (byte) (value >>> 32), (byte) (value >>> 24), (byte) (value >>> 16), + (byte) (value >>> 8), (byte) value}; + } + + private static final class StableId { + + private final long value; + + private StableId(long value) { + this.value = value; + } + + @Override + public boolean equals(Object other) { + return other instanceof StableId && value == ((StableId) other).value; + } + + @Override + public int hashCode() { + return Long.hashCode(value); + } + + @Override + public String toString() { + throw new AssertionError("sampling must not call StableId.toString()"); + } + } + +} diff --git a/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/sampling/LocalNeighborhoodTest.java b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/sampling/LocalNeighborhoodTest.java new file mode 100644 index 000000000..68f34bff1 --- /dev/null +++ b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/sampling/LocalNeighborhoodTest.java @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.geaflow.state.sampling; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.apache.geaflow.model.graph.edge.EdgeDirection; +import org.apache.geaflow.model.graph.edge.IEdge; +import org.apache.geaflow.model.graph.edge.impl.ValueEdge; +import org.apache.geaflow.model.graph.vertex.impl.ValueVertex; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class LocalNeighborhoodTest { + + @Test + public void testProjectPreservesVersionsAndBoundsNeighbors() { + LocalNeighborhood neighborhood = neighborhood( + Arrays.asList(edge(1L, 2L), edge(1L, 3L)), 7L, 11L); + + LocalNeighborhood projected = neighborhood.project( + EdgeDirection.OUT, 1, Long::compare, 100L, 17L, LocalNeighborhoodTest::longBytes); + + Assert.assertEquals(projected.getEdges().size(), 1); + Assert.assertEquals(projected.getSnapshotVersion(), 7L); + Assert.assertEquals(projected.getSamplingVersion(), 11L); + Assert.assertTrue(projected.matches(7L, 11L)); + } + + @Test + public void testRevalidatePreservesSamplingVersionOnNewSnapshot() { + LocalNeighborhood neighborhood = neighborhood( + Collections.singletonList(edge(1L, 2L)), 7L, 11L); + + LocalNeighborhood revalidated = neighborhood.revalidate( + new ValueVertex<>(1L, 2), 8L); + + Assert.assertEquals(revalidated.getSnapshotVersion(), 8L); + Assert.assertEquals(revalidated.getSamplingVersion(), 11L); + Assert.assertEquals(revalidated.getEdges().size(), 1); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testRejectsRevalidationToOlderSnapshot() { + LocalNeighborhood neighborhood = neighborhood( + Collections.emptyList(), 7L, 11L); + + neighborhood.revalidate(new ValueVertex<>(1L, 2), 6L); + } + + @Test + public void testCopiesInputEdgesAndProtectsReturnedEdges() { + List> input = new ArrayList<>(); + input.add(edge(1L, 2L)); + LocalNeighborhood neighborhood = neighborhood(input, 7L, 11L); + + input.clear(); + Assert.assertEquals(neighborhood.getEdges().size(), 1); + try { + neighborhood.getEdges().clear(); + Assert.fail("neighborhood edges must be immutable"); + } catch (UnsupportedOperationException expected) { + // Expected defensive view. + } + } + + private LocalNeighborhood neighborhood( + List> edges, long snapshotVersion, long samplingVersion) { + return new LocalNeighborhood<>(new ValueVertex<>(1L, 1), edges, + snapshotVersion, samplingVersion); + } + + private IEdge edge(long source, long target) { + return new ValueEdge<>(source, target, 1, EdgeDirection.OUT); + } + + private static byte[] longBytes(long value) { + return new byte[]{ + (byte) (value >>> 56), (byte) (value >>> 48), (byte) (value >>> 40), + (byte) (value >>> 32), (byte) (value >>> 24), (byte) (value >>> 16), + (byte) (value >>> 8), (byte) value}; + } +}