This is an automated email from the ASF dual-hosted git repository.
yaozhq pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/geaflow.git
The following commit(s) were added to refs/heads/master by this push:
new 562037c0 [ISSUE-563] refactor: Update filter instantiation method and
optimize edge query
562037c0 is described below
commit 562037c03e03847431b3769bc9be6d6780abfa8e
Author: yaozhongq <[email protected]>
AuthorDate: Fri Aug 1 16:48:11 2025 +0800
[ISSUE-563] refactor: Update filter instantiation method and optimize edge
query
* refactor: Update filter instantiation method and optimize edge query
interface
* fix: code conflict
---
.../function/vc/base/VertexCentricFunction.java | 17 +---
.../vc/context/dynamic/DynamicEdgeQueryImpl.java | 24 +----
.../vc/context/statical/StaticEdgeQueryImpl.java | 21 +---
.../context/dynamic/DynamicEdgeQueryImplTest.java | 14 ++-
.../context/statical/StaticEdgeQueryImplTest.java | 14 ++-
.../dsl/common/algo/AlgorithmRuntimeContext.java | 106 ++++++++++++++++++++-
.../engine/AbstractTraversalRuntimeContext.java | 2 +-
.../GeaFlowAlgorithmDynamicRuntimeContext.java | 91 ++++++++++++------
.../engine/GeaFlowAlgorithmRuntimeContext.java | 53 +++++++++--
.../engine/GeaFlowDynamicVCTraversalFunction.java | 8 +-
.../function/graph/MatchEdgeFunctionImpl.java | 6 +-
.../function/graph/MatchVertexFunctionImpl.java | 2 +-
.../dsl/runtime/traversal/data/EdgeGroup.java | 5 +
.../traversal/operator/StepLoopUntilOperator.java | 2 +-
.../geaflow/dsl/runtime/plan/StepPlanTest.java | 21 ++--
.../geaflow/dsl/runtime/query/QueryTester.java | 4 +-
.../dsl/runtime/query/udf/TestEdgeIteratorUdf.java | 21 ++++
.../geaflow/store/memory/GraphMemoryStoreTest.java | 2 +-
.../state/pushdown/filter/EdgeLabelFilter.java | 2 +-
.../state/pushdown/filter/EdgeTsFilter.java | 2 +-
.../state/pushdown/filter/EdgeValueDropFilter.java | 2 +-
.../geaflow/state/pushdown/filter/EmptyFilter.java | 2 +-
.../state/pushdown/filter/InEdgeFilter.java | 2 +-
.../state/pushdown/filter/OutEdgeFilter.java | 2 +-
.../state/pushdown/filter/VertexLabelFilter.java | 2 +-
.../pushdown/filter/VertexMustContainFilter.java | 2 +-
.../state/pushdown/filter/VertexTsFilter.java | 2 +-
.../pushdown/filter/VertexValueDropFilter.java | 2 +-
.../pushdown/inner/CodeGenFilterConverterTest.java | 4 +-
.../state/pushdown/inner/PushDownCodeGenJMH.java | 4 +-
.../apache/geaflow/state/query/QueryCondition.java | 2 +-
.../apache/geaflow/state/MemoryGraphStateTest.java | 6 +-
.../state/RocksDBDtPartitionGraphStateTest.java | 52 +++++-----
.../geaflow/state/RocksDBGraphStateTest.java | 12 +--
.../state/RocksDBLabelPartitionGraphStateTest.java | 44 ++++-----
.../apache/geaflow/state/StaticGraphStateTest.java | 28 +++---
36 files changed, 377 insertions(+), 208 deletions(-)
diff --git
a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/function/vc/base/VertexCentricFunction.java
b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/function/vc/base/VertexCentricFunction.java
index 918e7d62..afdf52b5 100644
---
a/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/function/vc/base/VertexCentricFunction.java
+++
b/geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/function/vc/base/VertexCentricFunction.java
@@ -94,34 +94,19 @@ public interface VertexCentricFunction<K, VV, EV, M>
extends Function {
*/
List<IEdge<K, EV>> getEdges();
- /**
- * Returns the both edges iterator.
- */
- CloseableIterator<IEdge<K, EV>> getEdgesIterator();
-
/**
* Returns the out edges.
*/
List<IEdge<K, EV>> getOutEdges();
- /**
- * Returns the out edges iterator.
- */
- CloseableIterator<IEdge<K, EV>> getOutEdgesIterator();
-
/**
* Returns the in edges.
*/
List<IEdge<K, EV>> getInEdges();
- /**
- * Returns the in edges iterator.
- */
- CloseableIterator<IEdge<K, EV>> getInEdgesIterator();
-
/**
* Get the edges which satisfies filter condition.
*/
- List<IEdge<K, EV>> getEdges(IFilter edgeFilter);
+ CloseableIterator<IEdge<K, EV>> getEdges(IFilter<?> edgeFilter);
}
}
diff --git
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImpl.java
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImpl.java
index f8b28d3f..31c26394 100644
---
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImpl.java
+++
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImpl.java
@@ -58,39 +58,23 @@ public class DynamicEdgeQueryImpl<K, VV, EV> implements
EdgeQuery<K, EV> {
return edgeState.query(versionId, vId).asList();
}
- @Override
- public CloseableIterator<IEdge<K, EV>> getEdgesIterator() {
- DynamicEdgeState<K, VV, EV> edgeState = graphState.dynamicGraph().E();
- return edgeState.query(versionId, vId).iterator();
- }
@Override
public List<IEdge<K, EV>> getOutEdges() {
DynamicEdgeState<K, VV, EV> edgeState = graphState.dynamicGraph().E();
- return edgeState.query(versionId,
vId).by(OutEdgeFilter.instance()).asList();
+ return edgeState.query(versionId,
vId).by(OutEdgeFilter.getInstance()).asList();
}
- @Override
- public CloseableIterator<IEdge<K, EV>> getOutEdgesIterator() {
- DynamicEdgeState<K, VV, EV> edgeState = graphState.dynamicGraph().E();
- return edgeState.query(versionId,
vId).by(OutEdgeFilter.instance()).iterator();
- }
@Override
public List<IEdge<K, EV>> getInEdges() {
DynamicEdgeState<K, VV, EV> edgeState = graphState.dynamicGraph().E();
- return edgeState.query(versionId,
vId).by(InEdgeFilter.instance()).asList();
- }
-
- @Override
- public CloseableIterator<IEdge<K, EV>> getInEdgesIterator() {
- DynamicEdgeState<K, VV, EV> edgeState = graphState.dynamicGraph().E();
- return edgeState.query(versionId,
vId).by(InEdgeFilter.instance()).iterator();
+ return edgeState.query(versionId,
vId).by(InEdgeFilter.getInstance()).asList();
}
@Override
- public List<IEdge<K, EV>> getEdges(IFilter edgeFilter) {
+ public CloseableIterator<IEdge<K, EV>> getEdges(IFilter edgeFilter) {
DynamicEdgeState<K, VV, EV> edgeState = graphState.dynamicGraph().E();
- return edgeState.query(versionId, vId).by(edgeFilter).asList();
+ return edgeState.query(versionId, vId).by(edgeFilter).iterator();
}
}
diff --git
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImpl.java
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImpl.java
index d675bca0..78265253 100644
---
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImpl.java
+++
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImpl.java
@@ -51,33 +51,20 @@ public class StaticEdgeQueryImpl<K, VV, EV> implements
EdgeQuery<K, EV> {
return graphState.staticGraph().E().query(vId).asList();
}
- @Override
- public CloseableIterator<IEdge<K, EV>> getEdgesIterator() {
- return graphState.staticGraph().E().query(vId).iterator();
- }
-
@Override
public List<IEdge<K, EV>> getOutEdges() {
- return
graphState.staticGraph().E().query(vId).by(OutEdgeFilter.instance()).asList();
+ return
graphState.staticGraph().E().query(vId).by(OutEdgeFilter.getInstance()).asList();
}
- @Override
- public CloseableIterator<IEdge<K, EV>> getOutEdgesIterator() {
- return
graphState.staticGraph().E().query(vId).by(OutEdgeFilter.instance()).iterator();
- }
@Override
public List<IEdge<K, EV>> getInEdges() {
- return
graphState.staticGraph().E().query(vId).by(InEdgeFilter.instance()).asList();
+ return
graphState.staticGraph().E().query(vId).by(InEdgeFilter.getInstance()).asList();
}
- @Override
- public CloseableIterator<IEdge<K, EV>> getInEdgesIterator() {
- return
graphState.staticGraph().E().query(vId).by(InEdgeFilter.instance()).iterator();
- }
@Override
- public List<IEdge<K, EV>> getEdges(IFilter edgeFilter) {
- return graphState.staticGraph().E().query(vId).by(edgeFilter).asList();
+ public CloseableIterator<IEdge<K, EV>> getEdges(IFilter edgeFilter) {
+ return (CloseableIterator)
graphState.staticGraph().E().query(vId).by(edgeFilter).iterator();
}
}
diff --git
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImplTest.java
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImplTest.java
index 82e56fb1..daa9134e 100644
---
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImplTest.java
+++
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/dynamic/DynamicEdgeQueryImplTest.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.stream.Collectors;
import
org.apache.geaflow.api.graph.function.vc.base.VertexCentricFunction.EdgeQuery;
import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.iterator.CloseableIterator;
import org.apache.geaflow.common.type.primitive.IntegerType;
import org.apache.geaflow.model.graph.edge.EdgeDirection;
import org.apache.geaflow.model.graph.edge.IEdge;
@@ -113,14 +114,17 @@ public class DynamicEdgeQueryImplTest {
@Test
public void testTestGetEdges() {
- List<IEdge<Integer, Integer>> outEdges =
edgeQuery.getEdges(OutEdgeFilter.instance());
- Assert.assertEquals(outEdges,
+ CloseableIterator<IEdge<Integer, Integer>> outEdges =
edgeQuery.getEdges(OutEdgeFilter.getInstance());
+ List<IEdge<Integer, Integer>> outEdgesList = new ArrayList<>();
+ outEdges.forEachRemaining(outEdgesList::add);
+ Assert.assertEquals(outEdgesList,
edges.stream().filter(x -> x.getDirect() ==
EdgeDirection.OUT).collect(
Collectors.toList()));
-
- List<IEdge<Integer, Integer>> inEdges =
edgeQuery.getEdges(InEdgeFilter.instance());
- Assert.assertEquals(inEdges,
+ CloseableIterator<IEdge<Integer, Integer>> inEdges =
edgeQuery.getEdges(InEdgeFilter.getInstance());
+ List<IEdge<Integer, Integer>> inEdgesList = new ArrayList<>();
+ inEdges.forEachRemaining(inEdgesList::add);
+ Assert.assertEquals(inEdgesList,
edges.stream().filter(x -> x.getDirect() ==
EdgeDirection.IN).collect(
Collectors.toList()));
}
diff --git
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImplTest.java
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImplTest.java
index f7b21cc4..2271c32d 100644
---
a/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImplTest.java
+++
b/geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/algo/vc/context/statical/StaticEdgeQueryImplTest.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.stream.Collectors;
import
org.apache.geaflow.api.graph.function.vc.base.VertexCentricFunction.EdgeQuery;
import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.iterator.CloseableIterator;
import org.apache.geaflow.common.type.primitive.IntegerType;
import org.apache.geaflow.model.graph.edge.EdgeDirection;
import org.apache.geaflow.model.graph.edge.IEdge;
@@ -112,14 +113,17 @@ public class StaticEdgeQueryImplTest {
@Test
public void testTestGetEdges() {
- List<IEdge<Integer, Integer>> outEdges =
edgeQuery.getEdges(OutEdgeFilter.instance());
- Assert.assertEquals(outEdges,
+ CloseableIterator<IEdge<Integer, Integer>> outEdges =
edgeQuery.getEdges(OutEdgeFilter.getInstance());
+ List<IEdge<Integer, Integer>> outEdgesList = new ArrayList<>();
+ outEdges.forEachRemaining(outEdgesList::add);
+ Assert.assertEquals(outEdgesList,
edges.stream().filter(x -> x.getDirect() ==
EdgeDirection.OUT).collect(
Collectors.toList()));
-
- List<IEdge<Integer, Integer>> inEdges =
edgeQuery.getEdges(InEdgeFilter.instance());
- Assert.assertEquals(inEdges,
+ CloseableIterator<IEdge<Integer, Integer>> inEdges =
edgeQuery.getEdges(InEdgeFilter.getInstance());
+ List<IEdge<Integer, Integer>> inEdgesList = new ArrayList<>();
+ inEdges.forEachRemaining(inEdgesList::add);
+ Assert.assertEquals(inEdgesList,
edges.stream().filter(x -> x.getDirect() ==
EdgeDirection.IN).collect(
Collectors.toList()));
}
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmRuntimeContext.java
b/geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmRuntimeContext.java
index 31fd99a3..9e444400 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmRuntimeContext.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmRuntimeContext.java
@@ -26,26 +26,128 @@ import org.apache.geaflow.dsl.common.data.Row;
import org.apache.geaflow.dsl.common.data.RowEdge;
import org.apache.geaflow.dsl.common.types.GraphSchema;
import org.apache.geaflow.model.graph.edge.EdgeDirection;
+import org.apache.geaflow.state.pushdown.filter.IFilter;
+/**
+ * Interface defining methods for managing and interacting with the runtime
context of a graph algorithm.
+ *
+ * @param <K> The type of vertex IDs.
+ * @param <M> The type of messages that can be sent between vertices.
+ */
public interface AlgorithmRuntimeContext<K, M> {
+ /**
+ * Loads all edges in the specified direction.
+ *
+ * @param direction The direction of the edges to be loaded.
+ * @return A list of RowEdge objects representing the edges.
+ */
List<RowEdge> loadEdges(EdgeDirection direction);
+ /**
+ * Returns an iterator over all edges in the specified direction.
+ *
+ * @param direction The direction of the edges to iterate over.
+ * @return An iterator over RowEdge objects representing the edges.
+ */
+ CloseableIterator<RowEdge> loadEdgesIterator(EdgeDirection direction);
+
+ /**
+ * Returns an iterator over edges filtered by the provided IFilter.
+ *
+ * @param filter The filter to apply when loading edges.
+ * @return An iterator over RowEdge objects representing the filtered
edges.
+ */
+ CloseableIterator<RowEdge> loadEdgesIterator(IFilter filter);
+
+ /**
+ * Loads static edges in the specified direction.
+ *
+ * @param direction The direction of the edges to be loaded.
+ * @return A list of RowEdge objects representing the static edges.
+ */
List<RowEdge> loadStaticEdges(EdgeDirection direction);
+ /**
+ * Returns an iterator over static edges in the specified direction.
+ *
+ * @param direction The direction of the edges to iterate over.
+ * @return An iterator over RowEdge objects representing the static edges.
+ */
+ CloseableIterator<RowEdge> loadStaticEdgesIterator(EdgeDirection
direction);
+
+ /**
+ * Returns an iterator over static edges filtered by the provided IFilter.
+ *
+ * @param filter The filter to apply when loading edges.
+ * @return An iterator over RowEdge objects representing the filtered
static edges.
+ */
+ CloseableIterator<RowEdge> loadStaticEdgesIterator(IFilter filter);
+
+ /**
+ * Loads dynamic edges (changed during execution) in the specified
direction.
+ *
+ * @param direction The direction of the edges to be loaded.
+ * @return A list of RowEdge objects representing the dynamic edges.
+ */
List<RowEdge> loadDynamicEdges(EdgeDirection direction);
- CloseableIterator<RowEdge> loadStaticEdgesIterator(EdgeDirection
direction);
+ /**
+ * Returns an iterator over dynamic edges in the specified direction.
+ *
+ * @param direction The direction of the edges to iterate over.
+ * @return An iterator over RowEdge objects representing the dynamic edges.
+ */
+ CloseableIterator<RowEdge> loadDynamicEdgesIterator(EdgeDirection
direction);
+
+ /**
+ * Returns an iterator over dynamic edges filtered by the provided IFilter.
+ *
+ * @param filter The filter to apply when loading edges.
+ * @return An iterator over RowEdge objects representing the filtered
dynamic edges.
+ */
+ CloseableIterator<RowEdge> loadDynamicEdgesIterator(IFilter filter);
+ /**
+ * Sends a message to the specified vertex.
+ *
+ * @param vertexId The ID of the vertex to which the message should be
sent.
+ * @param message The message to send.
+ */
void sendMessage(K vertexId, M message);
+ /**
+ * Updates the current vertex's value with the new row data.
+ *
+ * @param value The new row data to set as the vertex value.
+ */
void updateVertexValue(Row value);
+ /**
+ * Takes a row of data, typically received from another source.
+ *
+ * @param value The row data to take.
+ */
void take(Row value);
+ /**
+ * Gets the unique identifier for the current iteration.
+ *
+ * @return The current iteration ID.
+ */
long getCurrentIterationId();
+ /**
+ * Retrieves the schema information for the graph.
+ *
+ * @return The GraphSchema object containing the graph structure details.
+ */
GraphSchema getGraphSchema();
+ /**
+ * Retrieves the configuration settings for the algorithm runtime context.
+ *
+ * @return The Configuration object containing the settings.
+ */
Configuration getConfig();
-}
+}
\ No newline at end of file
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/AbstractTraversalRuntimeContext.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/AbstractTraversalRuntimeContext.java
index edf2916e..d067875b 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/AbstractTraversalRuntimeContext.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/AbstractTraversalRuntimeContext.java
@@ -195,7 +195,7 @@ public abstract class AbstractTraversalRuntimeContext
implements TraversalRuntim
@Override
public EdgeGroup loadEdges(IFilter loadEdgesFilter) {
- return EdgeGroup.of((List) edgeQuery.getEdges(loadEdgesFilter));
+ return EdgeGroup.of((CloseableIterator)
edgeQuery.getEdges(loadEdgesFilter));
}
@Override
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 c2b0a06e..07105aa4 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
@@ -40,6 +40,11 @@ import org.apache.geaflow.model.graph.edge.IEdge;
import org.apache.geaflow.model.graph.vertex.IVertex;
import org.apache.geaflow.model.traversal.ITraversalResponse;
import org.apache.geaflow.model.traversal.TraversalType.ResponseType;
+import org.apache.geaflow.state.iterator.IteratorWithClose;
+import org.apache.geaflow.state.pushdown.filter.EmptyFilter;
+import org.apache.geaflow.state.pushdown.filter.IFilter;
+import org.apache.geaflow.state.pushdown.filter.InEdgeFilter;
+import org.apache.geaflow.state.pushdown.filter.OutEdgeFilter;
public class GeaFlowAlgorithmDynamicRuntimeContext implements
AlgorithmRuntimeContext<Object, Object> {
@@ -59,15 +64,12 @@ public class GeaFlowAlgorithmDynamicRuntimeContext
implements AlgorithmRuntimeCo
private long iterationId = -1L;
- public GeaFlowAlgorithmDynamicRuntimeContext(
- GeaFlowAlgorithmDynamicAggTraversalFunction traversalFunction,
- IncVertexCentricTraversalFuncContext<Object, Row, Row, Object, Row>
traversalContext,
- GraphSchema graphSchema) {
+ public
GeaFlowAlgorithmDynamicRuntimeContext(GeaFlowAlgorithmDynamicAggTraversalFunction
traversalFunction,
+
IncVertexCentricTraversalFuncContext<Object, Row, Row, Object, Row>
traversalContext, GraphSchema graphSchema) {
this.traversalFunction = traversalFunction;
this.incVCTraversalCtx = traversalContext;
this.graphSchema = graphSchema;
- TraversalGraphSnapShot<Object, Row, Row> graphSnapShot =
incVCTraversalCtx.getHistoricalGraph()
- .getSnapShot(0L);
+ TraversalGraphSnapShot<Object, Row, Row> graphSnapShot =
incVCTraversalCtx.getHistoricalGraph().getSnapShot(0L);
this.vertexQuery = graphSnapShot.vertex();
this.edgeQuery = graphSnapShot.edges();
}
@@ -109,6 +111,25 @@ public class GeaFlowAlgorithmDynamicRuntimeContext
implements AlgorithmRuntimeCo
}
}
+ @Override
+ public CloseableIterator<RowEdge> loadEdgesIterator(EdgeDirection
direction) {
+ switch (direction) {
+ case OUT:
+ return loadEdgesIterator(OutEdgeFilter.getInstance());
+ case IN:
+ return loadEdgesIterator(InEdgeFilter.getInstance());
+ case BOTH:
+ return loadEdgesIterator(EmptyFilter.getInstance());
+ default:
+ throw new GeaFlowDSLException("Illegal edge direction: " +
direction);
+ }
+ }
+
+ @Override
+ public CloseableIterator<RowEdge> loadEdgesIterator(IFilter filter) {
+ return (CloseableIterator) edgeQuery.getEdges(filter);
+ }
+
public List<RowEdge> loadDynamicEdges(EdgeDirection direction) {
List<IEdge<Object, Row>> edges =
incVCTraversalCtx.getTemporaryGraph().getEdges();
List<RowEdge> rowEdges = new ArrayList<>();
@@ -142,47 +163,63 @@ public class GeaFlowAlgorithmDynamicRuntimeContext
implements AlgorithmRuntimeCo
}
public List<RowEdge> loadStaticEdges(EdgeDirection direction) {
- List<IEdge<Object, Row>> edges;
switch (direction) {
case OUT:
- edges = this.incVCTraversalCtx.getHistoricalGraph()
- .getSnapShot(0).edges().getOutEdges();
- break;
+ return (List) edgeQuery.getOutEdges();
case IN:
- edges = this.incVCTraversalCtx.getHistoricalGraph()
- .getSnapShot(0).edges().getInEdges();
- break;
+ return (List) edgeQuery.getInEdges();
case BOTH:
- edges = this.incVCTraversalCtx.getHistoricalGraph()
- .getSnapShot(0).edges().getEdges();
- break;
+ return (List) edgeQuery.getEdges();
default:
throw new GeaFlowDSLException("Illegal edge direction: " +
direction);
}
- List<RowEdge> rowEdges = new ArrayList<>();
- if (edges == null) {
- return rowEdges;
- }
- for (IEdge<Object, Row> edge : edges) {
- rowEdges.add((RowEdge) edge);
- }
- return rowEdges;
}
@Override
public CloseableIterator<RowEdge> loadStaticEdgesIterator(EdgeDirection
direction) {
switch (direction) {
case OUT:
- return (CloseableIterator) edgeQuery.getOutEdgesIterator();
+ return loadStaticEdgesIterator(OutEdgeFilter.getInstance());
+ case IN:
+ return loadStaticEdgesIterator(InEdgeFilter.getInstance());
+ case BOTH:
+ return loadStaticEdgesIterator(EmptyFilter.getInstance());
+ default:
+ throw new GeaFlowDSLException("Illegal edge direction: " +
direction);
+ }
+ }
+
+ @Override
+ public CloseableIterator<RowEdge> loadStaticEdgesIterator(IFilter filter) {
+ return (CloseableIterator) edgeQuery.getEdges(filter);
+ }
+
+ @Override
+ public CloseableIterator<RowEdge> loadDynamicEdgesIterator(EdgeDirection
direction) {
+ switch (direction) {
+ case OUT:
+ return loadDynamicEdgesIterator(OutEdgeFilter.getInstance());
case IN:
- return (CloseableIterator) edgeQuery.getInEdgesIterator();
+ return loadDynamicEdgesIterator(InEdgeFilter.getInstance());
case BOTH:
- return (CloseableIterator) edgeQuery.getEdgesIterator();
+ return loadDynamicEdgesIterator(EmptyFilter.getInstance());
default:
throw new GeaFlowDSLException("Illegal edge direction: " +
direction);
}
}
+ @Override
+ public CloseableIterator<RowEdge> loadDynamicEdgesIterator(IFilter filter)
{
+ List<IEdge<Object, Row>> edges =
incVCTraversalCtx.getTemporaryGraph().getEdges();
+ List<RowEdge> res = new ArrayList<>();
+ for (IEdge<Object, Row> edge : edges) {
+ if (filter.filter(edge)) {
+ res.add((RowEdge) edge);
+ }
+ }
+ return IteratorWithClose.wrap(res.iterator());
+ }
+
@Override
public void sendMessage(Object vertexId, Object message) {
incVCTraversalCtx.sendMessage(vertexId, message);
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 fd8892ef..01794a74 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
@@ -26,6 +26,7 @@ import
org.apache.geaflow.api.graph.function.aggregate.VertexCentricAggContextFu
import
org.apache.geaflow.api.graph.function.vc.VertexCentricTraversalFunction.TraversalEdgeQuery;
import
org.apache.geaflow.api.graph.function.vc.VertexCentricTraversalFunction.VertexCentricTraversalFuncContext;
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.data.Row;
@@ -36,6 +37,10 @@ import
org.apache.geaflow.dsl.runtime.traversal.message.ITraversalAgg;
import org.apache.geaflow.model.graph.edge.EdgeDirection;
import org.apache.geaflow.model.traversal.ITraversalResponse;
import org.apache.geaflow.model.traversal.TraversalType.ResponseType;
+import org.apache.geaflow.state.pushdown.filter.EmptyFilter;
+import org.apache.geaflow.state.pushdown.filter.IFilter;
+import org.apache.geaflow.state.pushdown.filter.InEdgeFilter;
+import org.apache.geaflow.state.pushdown.filter.OutEdgeFilter;
public class GeaFlowAlgorithmRuntimeContext implements
AlgorithmRuntimeContext<Object, Object> {
@@ -85,29 +90,63 @@ public class GeaFlowAlgorithmRuntimeContext implements
AlgorithmRuntimeContext<O
}
@Override
- public List<RowEdge> loadStaticEdges(EdgeDirection direction) {
- throw new RuntimeException("GeaFlowAlgorithmRuntimeContext not support
loadStaticEdges");
+ public CloseableIterator<RowEdge> loadEdgesIterator(EdgeDirection
direction) {
+ switch (direction) {
+ case OUT:
+ return loadEdgesIterator(OutEdgeFilter.getInstance());
+ case IN:
+ return loadEdgesIterator(InEdgeFilter.getInstance());
+ case BOTH:
+ return loadEdgesIterator(EmptyFilter.getInstance());
+ default:
+ throw new GeaFlowDSLException("Illegal edge direction: " +
direction);
+ }
}
@Override
- public List<RowEdge> loadDynamicEdges(EdgeDirection direction) {
- throw new RuntimeException("GeaFlowAlgorithmRuntimeContext not support
loadDynamicEdges");
+ public CloseableIterator<RowEdge> loadEdgesIterator(IFilter filter) {
+ return (CloseableIterator) edgeQuery.getEdges(filter);
+ }
+
+ @Override
+ public List<RowEdge> loadStaticEdges(EdgeDirection direction) {
+ return loadEdges(direction);
}
@Override
public CloseableIterator<RowEdge> loadStaticEdgesIterator(EdgeDirection
direction) {
switch (direction) {
case OUT:
- return (CloseableIterator) edgeQuery.getOutEdgesIterator();
+ return loadStaticEdgesIterator(OutEdgeFilter.getInstance());
case IN:
- return (CloseableIterator) edgeQuery.getInEdgesIterator();
+ return loadStaticEdgesIterator(InEdgeFilter.getInstance());
case BOTH:
- return (CloseableIterator) edgeQuery.getEdgesIterator();
+ return loadStaticEdgesIterator(EmptyFilter.getInstance());
default:
throw new GeaFlowDSLException("Illegal edge direction: " +
direction);
}
}
+ @Override
+ public CloseableIterator<RowEdge> loadStaticEdgesIterator(IFilter filter) {
+ return (CloseableIterator) edgeQuery.getEdges(filter);
+ }
+
+ @Override
+ public List<RowEdge> loadDynamicEdges(EdgeDirection direction) {
+ throw new GeaflowRuntimeException("GeaFlowAlgorithmRuntimeContext not
support loadDynamicEdges");
+ }
+
+ @Override
+ public CloseableIterator<RowEdge> loadDynamicEdgesIterator(EdgeDirection
direction) {
+ throw new GeaflowRuntimeException("GeaFlowAlgorithmRuntimeContext not
support loadDynamicEdgesIterator");
+ }
+
+ @Override
+ public CloseableIterator<RowEdge> loadDynamicEdgesIterator(IFilter filter)
{
+ throw new GeaflowRuntimeException("GeaFlowAlgorithmRuntimeContext not
support loadDynamicEdgesIterator");
+ }
+
@Override
public void sendMessage(Object vertexId, Object message) {
traversalContext.sendMessage(vertexId, message);
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowDynamicVCTraversalFunction.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowDynamicVCTraversalFunction.java
index d4cc4749..9049033e 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowDynamicVCTraversalFunction.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/engine/GeaFlowDynamicVCTraversalFunction.java
@@ -113,13 +113,13 @@ public class GeaFlowDynamicVCTraversalFunction implements
EdgeDirection direction = ((MatchEdgeOperator)
step).getFunction().getDirection();
switch (direction) {
case OUT:
- incrMessageEdgeFilter.add(InEdgeFilter.instance());
+ incrMessageEdgeFilter.add(InEdgeFilter.getInstance());
break;
case IN:
- incrMessageEdgeFilter.add(OutEdgeFilter.instance());
+ incrMessageEdgeFilter.add(OutEdgeFilter.getInstance());
break;
default:
- incrMessageEdgeFilter.add(EmptyFilter.of());
+ incrMessageEdgeFilter.add(EmptyFilter.getInstance());
}
}
@@ -166,7 +166,7 @@ public class GeaFlowDynamicVCTraversalFunction implements
private void sendEvolveMessage(Object vertexId, TraversalRuntimeContext
context) {
context.setVertex(IdOnlyVertex.of(vertexId));
- IFilter edgeFilter = incrMessageEdgeFilter.isEmpty() ?
EmptyFilter.of() : incrMessageEdgeFilter.get(
+ IFilter edgeFilter = incrMessageEdgeFilter.isEmpty() ?
EmptyFilter.getInstance() : incrMessageEdgeFilter.get(
(int) (context.getIterationId() - 1));
EdgeGroup rowEdges = context.loadEdges(edgeFilter);
StepOperator<StepRecord, StepRecord> operator =
commonFunction.getExecuteDagGroup()
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchEdgeFunctionImpl.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchEdgeFunctionImpl.java
index a7cc5c01..7bffa5d7 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchEdgeFunctionImpl.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchEdgeFunctionImpl.java
@@ -76,13 +76,13 @@ public class MatchEdgeFunctionImpl implements
MatchEdgeFunction {
IFilter directionFilter;
switch (direction) {
case OUT:
- directionFilter = OutEdgeFilter.instance();
+ directionFilter = OutEdgeFilter.getInstance();
break;
case IN:
- directionFilter = InEdgeFilter.instance();
+ directionFilter = InEdgeFilter.getInstance();
break;
case BOTH:
- directionFilter = EmptyFilter.of();
+ directionFilter = EmptyFilter.getInstance();
break;
default:
throw new IllegalArgumentException("Illegal edge direction: "
+ direction);
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchVertexFunctionImpl.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchVertexFunctionImpl.java
index c991245d..7e1f6919 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchVertexFunctionImpl.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/graph/MatchVertexFunctionImpl.java
@@ -73,7 +73,7 @@ public class MatchVertexFunctionImpl implements
MatchVertexFunction {
vertexTypes.stream().map(BinaryString::toString)
.collect(Collectors.toSet()));
} else {
- this.vertexFilter = EmptyFilter.of();
+ this.vertexFilter = EmptyFilter.getInstance();
}
for (IFilter filter : pushDownFilters) {
this.vertexFilter = this.vertexFilter == null ? filter :
this.vertexFilter.and(filter);
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/data/EdgeGroup.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/data/EdgeGroup.java
index 870b7a4e..b5905f2b 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/data/EdgeGroup.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/data/EdgeGroup.java
@@ -22,6 +22,7 @@ package org.apache.geaflow.dsl.runtime.traversal.data;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
import java.util.Iterator;
import java.util.List;
import java.util.function.Function;
@@ -44,6 +45,10 @@ public class EdgeGroup implements Iterable<RowEdge> {
return new EdgeGroup(edges);
}
+ public static EdgeGroup of(Iterator<RowEdge> edges) {
+ return new EdgeGroup(Lists.newArrayList(edges));
+ }
+
@Override
public Iterator<RowEdge> iterator() {
return edges.iterator();
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/operator/StepLoopUntilOperator.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/operator/StepLoopUntilOperator.java
index 491da5b1..d96b77a2 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/operator/StepLoopUntilOperator.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/traversal/operator/StepLoopUntilOperator.java
@@ -101,7 +101,7 @@ public class StepLoopUntilOperator extends
AbstractStepOperator<StepBoolFunction
final RowVertex vertexRecord;
if (vertexId instanceof IdOnlyVertex && !(vertexId.getId()
instanceof VirtualId)) {
vertexRecord = context.loadVertex(vertexId.getId(),
- EmptyFilter.of(),
+ EmptyFilter.getInstance(),
graphSchema,
addingVertexFieldTypes);
} else {
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/plan/StepPlanTest.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/plan/StepPlanTest.java
index a5dd3285..eb08dd52 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/plan/StepPlanTest.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/plan/StepPlanTest.java
@@ -65,22 +65,23 @@ public class StepPlanTest {
.withOutputType(VoidType.INSTANCE)
.withGraphSchema(createGraph())
.vertexMatch(new
MatchVertexFunctionImpl(Sets.newHashSet(BinaryString.fromString(
- "person")), "a", EmptyFilter.of()))
- .withInputPathSchema(PathType.EMPTY)
- .withOutputPathSchema(PathType.EMPTY)
- .withOutputType(VoidType.INSTANCE)
- .withGraphSchema(createGraph())
+ "person")), "a", EmptyFilter.getInstance()))
+ .withInputPathSchema(PathType.EMPTY)
+ .withOutputPathSchema(PathType.EMPTY)
+ .withOutputType(VoidType.INSTANCE)
+ .withGraphSchema(createGraph())
.edgeMatch(new MatchEdgeFunctionImpl(EdgeDirection.OUT,
Sets.newHashSet(), "e"))
.withInputPathSchema(PathType.EMPTY)
.withOutputPathSchema(PathType.EMPTY)
.withOutputType(VoidType.INSTANCE)
.withGraphSchema(createGraph())
.vertexMatch(new
MatchVertexFunctionImpl(Sets.newHashSet(BinaryString.fromString(
- "person")), "b", EmptyFilter.of()))
- .withInputPathSchema(PathType.EMPTY)
- .withOutputPathSchema(PathType.EMPTY)
- .withOutputType(VoidType.INSTANCE)
- .withGraphSchema(createGraph());
+ "person")), "b", EmptyFilter.getInstance()))
+ .withInputPathSchema(PathType.EMPTY)
+ .withOutputPathSchema(PathType.EMPTY)
+ .withOutputType(VoidType.INSTANCE)
+ .withGraphSchema(createGraph())
+ ;
String planDesc = logicalPlan.getPlanDesc();
LOGGER.info("Logical plan:\n{}", planDesc);
Assert.assertEquals(planDesc,
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/QueryTester.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/QueryTester.java
index 6ddcd691..ccf8754b 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/QueryTester.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/QueryTester.java
@@ -193,7 +193,7 @@ public class QueryTester implements Serializable {
private void compareResult(String actualResult, String expectResult) {
if (compareWithOrder) {
- Assert.assertEquals(actualResult, expectResult);
+ Assert.assertEquals(expectResult, actualResult);
} else {
String[] actualLines = actualResult.split("\n");
String[] expectLines = expectResult.split("\n");
@@ -209,7 +209,7 @@ public class QueryTester implements Serializable {
String actualSort = StringUtils.join(actualLines, "\n");
String expectSort = StringUtils.join(expectLines, "\n");
if (!Objects.equals(actualSort, expectSort)) {
- Assert.assertEquals(actualResult, expectResult);
+ Assert.assertEquals(expectResult, actualResult);
}
}
}
diff --git
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/udf/TestEdgeIteratorUdf.java
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/udf/TestEdgeIteratorUdf.java
index f3dae075..815d0d55 100644
---
a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/udf/TestEdgeIteratorUdf.java
+++
b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/udf/TestEdgeIteratorUdf.java
@@ -19,7 +19,10 @@
package org.apache.geaflow.dsl.runtime.query.udf;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
import java.util.Iterator;
+import java.util.List;
import java.util.Optional;
import org.apache.geaflow.common.iterator.CloseableIterator;
import org.apache.geaflow.common.type.primitive.LongType;
@@ -71,6 +74,24 @@ public class TestEdgeIteratorUdf implements
AlgorithmUserFunction<Object, Long>
count++;
}
}
+ List<RowEdge> bothEdge =
Lists.newArrayList(context.loadStaticEdgesIterator(EdgeDirection.BOTH));
+ List<RowEdge> inEdge =
Lists.newArrayList(context.loadStaticEdgesIterator(EdgeDirection.IN));
+ List<RowEdge> outEdge =
Lists.newArrayList(context.loadStaticEdgesIterator(EdgeDirection.OUT));
+ Preconditions.checkState(bothEdge.size() == inEdge.size() +
outEdge.size(), "Static edge not equal");
+ List<RowEdge> bothEdgeList =
context.loadStaticEdges(EdgeDirection.BOTH);
+ Preconditions.checkState(bothEdgeList.size() == inEdge.size() +
outEdge.size(), "Static edge not equal");
+
+ bothEdge =
Lists.newArrayList(context.loadDynamicEdgesIterator(EdgeDirection.BOTH));
+ inEdge =
Lists.newArrayList(context.loadDynamicEdgesIterator(EdgeDirection.IN));
+ outEdge =
Lists.newArrayList(context.loadDynamicEdgesIterator(EdgeDirection.OUT));
+ Preconditions.checkState(bothEdge.size() == inEdge.size() +
outEdge.size(), "Dynamic edge not equal");
+
+ bothEdge =
Lists.newArrayList(context.loadEdgesIterator(EdgeDirection.BOTH));
+ inEdge =
Lists.newArrayList(context.loadEdgesIterator(EdgeDirection.IN));
+ outEdge =
Lists.newArrayList(context.loadEdgesIterator(EdgeDirection.OUT));
+ Preconditions.checkState(bothEdge.size() == inEdge.size() +
outEdge.size(), "History edge not equal");
+
+
}
@Override
diff --git
a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-memory/src/test/java/org/apache/geaflow/store/memory/GraphMemoryStoreTest.java
b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-memory/src/test/java/org/apache/geaflow/store/memory/GraphMemoryStoreTest.java
index 7b07d465..d33d7e9f 100644
---
a/geaflow/geaflow-plugins/geaflow-store/geaflow-store-memory/src/test/java/org/apache/geaflow/store/memory/GraphMemoryStoreTest.java
+++
b/geaflow/geaflow-plugins/geaflow-store/geaflow-store-memory/src/test/java/org/apache/geaflow/store/memory/GraphMemoryStoreTest.java
@@ -143,7 +143,7 @@ public class GraphMemoryStoreTest {
list =
Lists.newArrayList(store.getOneDegreeGraphIterator(StatePushDown.of().withFilter(
- GraphFilter.of(VertexMustContainFilter.instance()))));
+ GraphFilter.of(VertexMustContainFilter.getInstance()))));
Assert.assertEquals(list.size(), 1);
list =
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeLabelFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeLabelFilter.java
index facd804f..43c02245 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeLabelFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeLabelFilter.java
@@ -38,7 +38,7 @@ public class EdgeLabelFilter<K, EV> implements IEdgeFilter<K,
EV> {
this.labels = Sets.newHashSet(labels);
}
- public static <K, EV> EdgeLabelFilter<K, EV> instance(String... labels) {
+ public static <K, EV> EdgeLabelFilter<K, EV> getInstance(String... labels)
{
return new EdgeLabelFilter<>(labels);
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeTsFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeTsFilter.java
index b860ab2b..2b5e5446 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeTsFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeTsFilter.java
@@ -31,7 +31,7 @@ public class EdgeTsFilter<K, EV> implements IEdgeFilter<K,
EV> {
this.timeRange = timeRange;
}
- public static <K, EV> EdgeTsFilter<K, EV> instance(long start, long end) {
+ public static <K, EV> EdgeTsFilter<K, EV> getInstance(long start, long
end) {
return new EdgeTsFilter<>(TimeRange.of(start, end));
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeValueDropFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeValueDropFilter.java
index 46c9c4c1..4c93042f 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeValueDropFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EdgeValueDropFilter.java
@@ -25,7 +25,7 @@ public class EdgeValueDropFilter<K, VV> implements
IEdgeFilter<K, VV> {
private static final EdgeValueDropFilter filter = new
EdgeValueDropFilter();
- public static <K, EV> EdgeValueDropFilter<K, EV> instance() {
+ public static <K, EV> EdgeValueDropFilter<K, EV> getInstance() {
return filter;
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EmptyFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EmptyFilter.java
index 5a248bee..8a81d4c0 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EmptyFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/EmptyFilter.java
@@ -25,7 +25,7 @@ public class EmptyFilter implements IFilter {
private static final EmptyFilter FILTER = new EmptyFilter();
- public static EmptyFilter of() {
+ public static EmptyFilter getInstance() {
return FILTER;
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/InEdgeFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/InEdgeFilter.java
index 2b67257a..c9dcc2bc 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/InEdgeFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/InEdgeFilter.java
@@ -36,7 +36,7 @@ public class InEdgeFilter<K, EV> implements IEdgeFilter<K,
EV> {
return FilterType.IN_EDGE;
}
- public static <K, EV> InEdgeFilter<K, EV> instance() {
+ public static <K, EV> InEdgeFilter<K, EV> getInstance() {
return inEdgeFilter;
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/OutEdgeFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/OutEdgeFilter.java
index 418e9801..e8240d29 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/OutEdgeFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/OutEdgeFilter.java
@@ -26,7 +26,7 @@ public class OutEdgeFilter<K, EV> implements IEdgeFilter<K,
EV> {
private static final OutEdgeFilter outEdgeFilter = new OutEdgeFilter();
- public static <K, EV> OutEdgeFilter<K, EV> instance() {
+ public static <K, EV> OutEdgeFilter<K, EV> getInstance() {
return outEdgeFilter;
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexLabelFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexLabelFilter.java
index 2440d642..c84c4752 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexLabelFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexLabelFilter.java
@@ -38,7 +38,7 @@ public class VertexLabelFilter<K, VV> implements
IVertexFilter<K, VV> {
this.labels = Sets.newHashSet(labels);
}
- public static <K, VV> VertexLabelFilter<K, VV> instance(String... labels) {
+ public static <K, VV> VertexLabelFilter<K, VV> getInstance(String...
labels) {
return new VertexLabelFilter<>(labels);
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexMustContainFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexMustContainFilter.java
index c29a4f4f..745d84cd 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexMustContainFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexMustContainFilter.java
@@ -35,7 +35,7 @@ public class VertexMustContainFilter<K, VV, EV> implements
IOneDegreeGraphFilter
return FilterType.VERTEX_MUST_CONTAIN;
}
- public static <K, VV, EV> VertexMustContainFilter<K, VV, EV> instance() {
+ public static <K, VV, EV> VertexMustContainFilter<K, VV, EV> getInstance()
{
return filter;
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexTsFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexTsFilter.java
index de4555d4..fb7b5723 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexTsFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexTsFilter.java
@@ -31,7 +31,7 @@ public class VertexTsFilter<K, VV> implements
IVertexFilter<K, VV> {
this.timeRange = timeRange;
}
- public static <K, VV> VertexTsFilter<K, VV> instance(long start, long end)
{
+ public static <K, VV> VertexTsFilter<K, VV> getInstance(long start, long
end) {
return new VertexTsFilter<>(TimeRange.of(start, end));
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexValueDropFilter.java
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexValueDropFilter.java
index 2e5d026a..197b0a8f 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexValueDropFilter.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/main/java/org/apache/geaflow/state/pushdown/filter/VertexValueDropFilter.java
@@ -25,7 +25,7 @@ public class VertexValueDropFilter<K, VV> implements
IVertexFilter<K, VV> {
private static final VertexValueDropFilter filter = new
VertexValueDropFilter();
- public static <K, VV> VertexValueDropFilter<K, VV> instance() {
+ public static <K, VV> VertexValueDropFilter<K, VV> getInstance() {
return filter;
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/CodeGenFilterConverterTest.java
b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/CodeGenFilterConverterTest.java
index 5c48f8c7..ff2a3940 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/CodeGenFilterConverterTest.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/CodeGenFilterConverterTest.java
@@ -66,7 +66,7 @@ public class CodeGenFilterConverterTest {
.or(new VertexLabelFilter(Arrays.asList("label3")));
List<IFilter> list = new ArrayList<>();
for (int i = 0; i < 100; i++) {
- list.add(new AndFilter(i / 2 == 0 ? InEdgeFilter.instance() :
OutEdgeFilter.instance(),
+ list.add(new AndFilter(i / 2 == 0 ? InEdgeFilter.getInstance() :
OutEdgeFilter.getInstance(),
new EdgeLabelFilter("label" + i)).and(new
EdgeTsFilter(TimeRange.of(10, 100))));
}
complexFilter = new OrFilter(list);
@@ -81,7 +81,7 @@ public class CodeGenFilterConverterTest {
List<IGraphFilter> graphFilters = new ArrayList<>();
for (int i = 0; i < 100; i++) {
- graphFilters.add(new AndGraphFilter(Arrays.asList(GraphFilter.of(i
/ 2 == 0 ? InEdgeFilter.instance() : OutEdgeFilter.instance()),
+ graphFilters.add(new AndGraphFilter(Arrays.asList(GraphFilter.of(i
/ 2 == 0 ? InEdgeFilter.getInstance() : OutEdgeFilter.getInstance()),
GraphFilter.of(new EdgeLabelFilter("label" + i)),
GraphFilter.of(new EdgeTsFilter(TimeRange.of(10, 100))))));
}
diff --git
a/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/PushDownCodeGenJMH.java
b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/PushDownCodeGenJMH.java
index 6d8353ef..a446e155 100644
---
a/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/PushDownCodeGenJMH.java
+++
b/geaflow/geaflow-state/geaflow-state-common/src/test/java/org/apache/geaflow/state/pushdown/inner/PushDownCodeGenJMH.java
@@ -94,7 +94,7 @@ public class PushDownCodeGenJMH {
.or(new VertexLabelFilter(Arrays.asList("label3")));
List<IFilter> list = new ArrayList<>();
for (int i = 0; i < 100; i++) {
- list.add(new AndFilter(i / 2 == 0 ? InEdgeFilter.instance() :
OutEdgeFilter.instance(),
+ list.add(new AndFilter(i / 2 == 0 ? InEdgeFilter.getInstance() :
OutEdgeFilter.getInstance(),
new EdgeLabelFilter("label" + i % 5)).and(new
EdgeTsFilter(TimeRange.of(10, 100))));
}
complexFilter = new OrFilter(list);
@@ -109,7 +109,7 @@ public class PushDownCodeGenJMH {
List<IGraphFilter> graphFilters = new ArrayList<>();
for (int i = 0; i < 100; i++) {
- graphFilters.add(new AndGraphFilter(Arrays.asList(GraphFilter.of(i
/ 2 == 0 ? InEdgeFilter.instance() : OutEdgeFilter.instance()),
+ graphFilters.add(new AndGraphFilter(Arrays.asList(GraphFilter.of(i
/ 2 == 0 ? InEdgeFilter.getInstance() : OutEdgeFilter.getInstance()),
GraphFilter.of(new EdgeLabelFilter("label" + i)),
GraphFilter.of(new EdgeTsFilter(TimeRange.of(10, 100))))));
}
diff --git
a/geaflow/geaflow-state/geaflow-state-impl/src/main/java/org/apache/geaflow/state/query/QueryCondition.java
b/geaflow/geaflow-state/geaflow-state-impl/src/main/java/org/apache/geaflow/state/query/QueryCondition.java
index 4a2778d1..1a93afbd 100644
---
a/geaflow/geaflow-state/geaflow-state-impl/src/main/java/org/apache/geaflow/state/query/QueryCondition.java
+++
b/geaflow/geaflow-state/geaflow-state-impl/src/main/java/org/apache/geaflow/state/query/QueryCondition.java
@@ -34,7 +34,7 @@ public class QueryCondition<K> {
public K queryId;
public List<K> queryIds;
public boolean isFullScan;
- public IFilter[] stateFilters = new IFilter[]{EmptyFilter.of()};
+ public IFilter[] stateFilters = new IFilter[] {EmptyFilter.getInstance()};
public IProjector projector;
public IEdgeLimit limit;
public EdgeAtom order;
diff --git
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/MemoryGraphStateTest.java
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/MemoryGraphStateTest.java
index cad988be..f987ec52 100644
---
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/MemoryGraphStateTest.java
+++
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/MemoryGraphStateTest.java
@@ -193,16 +193,16 @@ public class MemoryGraphStateTest {
Assert.assertEquals(list.size(), 6);
list =
- graphState.staticGraph().E().query().by(InEdgeFilter.instance())
+ graphState.staticGraph().E().query().by(InEdgeFilter.getInstance())
.limit(1L, 1L).asList();
Assert.assertEquals(list.size(), 10);
- list = graphState.staticGraph().E().query().by(InEdgeFilter.instance())
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance())
.limit(1L, 2L).asList();
Assert.assertEquals(list.size(), 20);
List<String> targetIds =
- graphState.staticGraph().E().query().by(InEdgeFilter.instance())
+ graphState.staticGraph().E().query().by(InEdgeFilter.getInstance())
.select(new DstIdProjector<>()).limit(1L, 2L).asList();
Assert.assertEquals(targetIds.size(), 20);
diff --git
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBDtPartitionGraphStateTest.java
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBDtPartitionGraphStateTest.java
index 3d49819e..552ee656 100644
---
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBDtPartitionGraphStateTest.java
+++
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBDtPartitionGraphStateTest.java
@@ -135,7 +135,7 @@ public class RocksDBDtPartitionGraphStateTest {
graphState.manage().operate().finish();
List<IEdge<String, String>> edgeList1 =
graphState.staticGraph().E().query("0")
- .by(OutEdgeFilter.instance().and(EdgeTsFilter.instance(1735660800,
1735662000)))
+
.by(OutEdgeFilter.getInstance().and(EdgeTsFilter.getInstance(1735660800,
1735662000)))
.asList();
Assert.assertEquals(edgeList1.size(), 390);
graphState.manage().operate().close();
@@ -146,16 +146,16 @@ public class RocksDBDtPartitionGraphStateTest {
List<IEdge<String, String>> edgeList =
graphState.staticGraph().E().query("0").asList();
Assert.assertEquals(edgeList.size(), 750);
- edgeList =
graphState.staticGraph().E().query("0").by(OutEdgeFilter.instance()).asList();
+ edgeList =
graphState.staticGraph().E().query("0").by(OutEdgeFilter.getInstance()).asList();
Assert.assertEquals(edgeList.size(), 750);
- edgeList =
graphState.staticGraph().E().query("0").by(InEdgeFilter.instance()).asList();
+ edgeList =
graphState.staticGraph().E().query("0").by(InEdgeFilter.getInstance()).asList();
Assert.assertEquals(edgeList.size(), 0);
edgeList = graphState.staticGraph().E().query()
- .by(OutEdgeFilter.instance().and(EdgeTsFilter.instance(1735660800,
1735662000)))
+
.by(OutEdgeFilter.getInstance().and(EdgeTsFilter.getInstance(1735660800,
1735662000)))
.asList();
Assert.assertEquals(edgeList.size(), 780);
edgeList = graphState.staticGraph().E().query()
- .by(OutEdgeFilter.instance().and(EdgeTsFilter.instance(1736265600,
1736265699)))
+
.by(OutEdgeFilter.getInstance().and(EdgeTsFilter.getInstance(1736265600,
1736265699)))
.asList();
Assert.assertEquals(edgeList.size(), 99);
@@ -185,17 +185,17 @@ public class RocksDBDtPartitionGraphStateTest {
}
List<IEdge<String, String>> edges1 =
graphState.staticGraph().E().query("2")
- .by(EdgeTsFilter.instance(1735660800, 1735662800)).asList();
+ .by(EdgeTsFilter.getInstance(1735660800, 1735662800)).asList();
List<IEdge<String, String>> edges2 =
graphState.staticGraph().E().query("2")
- .by(EdgeTsFilter.instance(1736265600, 1736365600)).asList();
+ .by(EdgeTsFilter.getInstance(1736265600, 1736365600)).asList();
List<IEdge<String, String>> edges3 =
graphState.staticGraph().E().query("2")
- .by(EdgeTsFilter.instance(1835660800, 1835760800)).asList();
+ .by(EdgeTsFilter.getInstance(1835660800, 1835760800)).asList();
IVertex<String, String> vertex1 =
graphState.staticGraph().V().query("9999")
- .by(VertexTsFilter.instance(1736870400, 1736970400)).get();
+ .by(VertexTsFilter.getInstance(1736870400, 1736970400)).get();
IVertex<String, String> vertex2 =
graphState.staticGraph().V().query("9999")
- .by(VertexTsFilter.instance(1735660800, 1735760800)).get();
+ .by(VertexTsFilter.getInstance(1735660800, 1735760800)).get();
IVertex<String, String> vertex3 =
graphState.staticGraph().V().query("9999")
- .by(VertexTsFilter.instance(1835660800, 1835760800)).get();
+ .by(VertexTsFilter.getInstance(1835660800, 1835760800)).get();
Assert.assertEquals(edges1.size(), 2000);
Assert.assertEquals(edges2.size(), 10000);
@@ -246,12 +246,12 @@ public class RocksDBDtPartitionGraphStateTest {
graphState.manage().operate().finish();
Iterator<IVertex<String, String>> it =
graphState.staticGraph().V().query()
- .by(VertexTsFilter.instance(1736870400, 1736970400)).iterator();
+ .by(VertexTsFilter.getInstance(1736870400, 1736970400)).iterator();
List<IVertex<String, String>> vertices = Lists.newArrayList(it);
Assert.assertEquals(vertices.size(), 100);
it = graphState.staticGraph().V().query()
- .by(VertexTsFilter.instance(1836870400, 1836970400)).iterator();
+ .by(VertexTsFilter.getInstance(1836870400, 1836970400)).iterator();
vertices = Lists.newArrayList(it);
Assert.assertEquals(vertices.size(), 0);
@@ -265,14 +265,14 @@ public class RocksDBDtPartitionGraphStateTest {
Assert.assertEquals(Iterators.size(idIt), 150);
Iterator<OneDegreeGraph<String, String, String>> it2 =
graphState.staticGraph().VE().query()
- .by(VertexTsFilter.instance(1735660800, 1736265600)
- .and(EdgeTsFilter.instance(1735660800,
1736265600))).iterator();
+ .by(VertexTsFilter.getInstance(1735660800, 1736265600)
+ .and(EdgeTsFilter.getInstance(1735660800,
1736265600))).iterator();
List<OneDegreeGraph> res = Lists.newArrayList(it2);
Assert.assertEquals(res.size(), 100);
it2 = graphState.staticGraph().VE().query()
- .by(VertexTsFilter.instance(1836870400, 1836970400)
- .and(EdgeTsFilter.instance(1836870400,
1836970400))).iterator();
+ .by(VertexTsFilter.getInstance(1836870400, 1836970400)
+ .and(EdgeTsFilter.getInstance(1836870400,
1836970400))).iterator();
res = Lists.newArrayList(it2);
Assert.assertEquals(res.size(), 0);
@@ -325,7 +325,7 @@ public class RocksDBDtPartitionGraphStateTest {
long num = edges.stream().filter(e -> e.getDirect() ==
EdgeDirection.OUT).count();
Assert.assertEquals(num, 5000);
- edges =
graphState.staticGraph().E().query("2").by(OutEdgeFilter.instance()
+ edges =
graphState.staticGraph().E().query("2").by(OutEdgeFilter.getInstance()
.and(new EdgeTsFilter(TimeRange.of(1735660800,
1735661800)))).asList();
Assert.assertEquals(edges.size(), 1000);
@@ -340,7 +340,7 @@ public class RocksDBDtPartitionGraphStateTest {
edges = graphState.staticGraph().E().query("2")
.by(new EdgeTsFilter(TimeRange.of(1735667800, 1735670800)).and(
- EdgeLabelFilter.instance("world"))).asList();
+ EdgeLabelFilter.getInstance("world"))).asList();
Assert.assertEquals(edges.size(), 3000);
long minTime = edges.stream().mapToLong(e -> ((IDLabelTimeEdge)
e).getTime()).min()
.getAsLong();
@@ -415,15 +415,15 @@ public class RocksDBDtPartitionGraphStateTest {
.limit(1L, 1L).asList();
Assert.assertEquals(list.size(), 6);
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 1L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
1L)
.asList();
Assert.assertEquals(list.size(), 10);
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 2L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
2L)
.asList();
Assert.assertEquals(list.size(), 20);
- List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.instance())
+ List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance())
.select(new DstIdProjector<>()).limit(1L, 2L).asList();
Assert.assertEquals(targetIds.size(), 20);
@@ -487,10 +487,10 @@ public class RocksDBDtPartitionGraphStateTest {
edges = graphState.staticGraph().E().asList();
Assert.assertEquals(edges.size(), 180);
edges = graphState.staticGraph().E().query()
- .by(EdgeTsFilter.instance(1736265600, 1737265600)).asList();
+ .by(EdgeTsFilter.getInstance(1736265600, 1737265600)).asList();
Assert.assertEquals(edges.size(), 80);
edges = graphState.staticGraph().E().query()
- .by(EdgeTsFilter.instance(1735660800, 1735661800)).asList();
+ .by(EdgeTsFilter.getInstance(1735660800, 1735661800)).asList();
Assert.assertEquals(edges.size(), 100);
vertices = graphState.staticGraph().V().asList();
Assert.assertEquals(vertices.size(), 2);
@@ -502,10 +502,10 @@ public class RocksDBDtPartitionGraphStateTest {
graphState.manage().operate().recover();
graphState.staticGraph().V().add(new ValueTimeVertex<>("2", "world",
1737265600));
Assert.assertEquals(graphState.staticGraph().V().query("2")
- .by(VertexTsFilter.instance(1737265600,
1737265601)).get().getValue(), "world");
+ .by(VertexTsFilter.getInstance(1737265600,
1737265601)).get().getValue(), "world");
graphState.manage().operate().finish();
Assert.assertEquals(graphState.staticGraph().V().query("2")
- .by(VertexTsFilter.instance(1737265600,
1737265601)).get().getValue(), "world");
+ .by(VertexTsFilter.getInstance(1737265600,
1737265601)).get().getValue(), "world");
graphState.manage().operate().close();
graphState.manage().operate().drop();
diff --git
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBGraphStateTest.java
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBGraphStateTest.java
index 1290e10e..5c332793 100644
---
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBGraphStateTest.java
+++
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBGraphStateTest.java
@@ -129,9 +129,9 @@ public class RocksDBGraphStateTest {
List<IEdge<String, String>> edgeList =
graphState.staticGraph().E().query("0").asList();
Assert.assertEquals(edgeList.size(), 390);
- edgeList =
graphState.staticGraph().E().query("0").by(OutEdgeFilter.instance()).asList();
+ edgeList =
graphState.staticGraph().E().query("0").by(OutEdgeFilter.getInstance()).asList();
Assert.assertEquals(edgeList.size(), 390);
- edgeList =
graphState.staticGraph().E().query("0").by(InEdgeFilter.instance()).asList();
+ edgeList =
graphState.staticGraph().E().query("0").by(InEdgeFilter.getInstance()).asList();
Assert.assertEquals(edgeList.size(), 0);
graphState.manage().operate().close();
@@ -347,7 +347,7 @@ public class RocksDBGraphStateTest {
Assert.assertEquals(num, 5000);
edges = graphState.staticGraph().E().query("2")
- .by(OutEdgeFilter.instance().and(new EdgeTsFilter(TimeRange.of(0,
1000)))).asList();
+ .by(OutEdgeFilter.getInstance().and(new
EdgeTsFilter(TimeRange.of(0, 1000)))).asList();
Assert.assertEquals(edges.size(), 1000);
maxTime = edges.stream().mapToLong(e -> ((IDLabelTimeEdge)
e).getTime()).max().getAsLong();
@@ -427,15 +427,15 @@ public class RocksDBGraphStateTest {
.limit(1L, 1L).asList();
Assert.assertEquals(list.size(), 6);
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 1L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
1L)
.asList();
Assert.assertEquals(list.size(), 10);
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 2L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
2L)
.asList();
Assert.assertEquals(list.size(), 20);
- List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.instance())
+ List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance())
.select(new DstIdProjector<>()).limit(1L, 2L).asList();
Assert.assertEquals(targetIds.size(), 20);
diff --git
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBLabelPartitionGraphStateTest.java
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBLabelPartitionGraphStateTest.java
index a255ad7a..37e43730 100644
---
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBLabelPartitionGraphStateTest.java
+++
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/RocksDBLabelPartitionGraphStateTest.java
@@ -131,7 +131,7 @@ public class RocksDBLabelPartitionGraphStateTest {
graphState.manage().operate().finish();
List<IEdge<String, String>> edgeList1 =
graphState.staticGraph().E().query("0")
-
.by(OutEdgeFilter.instance().and(EdgeLabelFilter.instance("person"))).asList();
+
.by(OutEdgeFilter.getInstance().and(EdgeLabelFilter.getInstance("person"))).asList();
Assert.assertEquals(edgeList1.size(), 390);
graphState.manage().operate().close();
@@ -141,12 +141,12 @@ public class RocksDBLabelPartitionGraphStateTest {
List<IEdge<String, String>> edgeList =
graphState.staticGraph().E().query("0").asList();
Assert.assertEquals(edgeList.size(), 750);
- edgeList =
graphState.staticGraph().E().query("0").by(OutEdgeFilter.instance()).asList();
+ edgeList =
graphState.staticGraph().E().query("0").by(OutEdgeFilter.getInstance()).asList();
Assert.assertEquals(edgeList.size(), 750);
- edgeList =
graphState.staticGraph().E().query("0").by(InEdgeFilter.instance()).asList();
+ edgeList =
graphState.staticGraph().E().query("0").by(InEdgeFilter.getInstance()).asList();
Assert.assertEquals(edgeList.size(), 0);
edgeList = graphState.staticGraph().E().query("0")
-
.by(OutEdgeFilter.instance().and(EdgeLabelFilter.instance("person"))).asList();
+
.by(OutEdgeFilter.getInstance().and(EdgeLabelFilter.getInstance("person"))).asList();
Assert.assertEquals(edgeList.size(), 390);
graphState.manage().operate().close();
@@ -173,17 +173,17 @@ public class RocksDBLabelPartitionGraphStateTest {
}
List<IEdge<String, String>> edges1 =
graphState.staticGraph().E().query("2")
- .by(EdgeLabelFilter.instance("person")).asList();
+ .by(EdgeLabelFilter.getInstance("person")).asList();
List<IEdge<String, String>> edges2 =
graphState.staticGraph().E().query("2")
- .by(EdgeLabelFilter.instance("trade")).asList();
+ .by(EdgeLabelFilter.getInstance("trade")).asList();
List<IEdge<String, String>> edges3 =
graphState.staticGraph().E().query("2")
- .by(EdgeLabelFilter.instance("illegal")).asList();
+ .by(EdgeLabelFilter.getInstance("illegal")).asList();
IVertex<String, String> vertex1 =
graphState.staticGraph().V().query("9999")
- .by(VertexLabelFilter.instance("relation")).get();
+ .by(VertexLabelFilter.getInstance("relation")).get();
IVertex<String, String> vertex2 =
graphState.staticGraph().V().query("9999")
- .by(VertexLabelFilter.instance("person")).get();
+ .by(VertexLabelFilter.getInstance("person")).get();
IVertex<String, String> vertex3 =
graphState.staticGraph().V().query("9999")
- .by(VertexLabelFilter.instance("illegal")).get();
+ .by(VertexLabelFilter.getInstance("illegal")).get();
Assert.assertEquals(edges1.size(), 10000);
Assert.assertEquals(edges2.size(), 10000);
@@ -230,11 +230,11 @@ public class RocksDBLabelPartitionGraphStateTest {
graphState.manage().operate().finish();
Iterator<IVertex<String, String>> it =
graphState.staticGraph().V().query()
- .by(VertexLabelFilter.instance("entity")).iterator();
+ .by(VertexLabelFilter.getInstance("entity")).iterator();
List<IVertex<String, String>> vertices = Lists.newArrayList(it);
Assert.assertEquals(vertices.size(), 100);
- it =
graphState.staticGraph().V().query().by(VertexLabelFilter.instance("illegal"))
+ it =
graphState.staticGraph().V().query().by(VertexLabelFilter.getInstance("illegal"))
.iterator();
vertices = Lists.newArrayList(it);
Assert.assertEquals(vertices.size(), 0);
@@ -249,11 +249,11 @@ public class RocksDBLabelPartitionGraphStateTest {
Assert.assertEquals(Iterators.size(idIt), 150);
Iterator<OneDegreeGraph<String, String, String>> it2 =
graphState.staticGraph().VE().query()
- .by(VertexLabelFilter.instance("person")).iterator();
+ .by(VertexLabelFilter.getInstance("person")).iterator();
List<OneDegreeGraph> res = Lists.newArrayList(it2);
Assert.assertEquals(res.size(), 100);
- it2 =
graphState.staticGraph().VE().query().by(VertexLabelFilter.instance("entity"))
+ it2 =
graphState.staticGraph().VE().query().by(VertexLabelFilter.getInstance("entity"))
.iterator();
res = Lists.newArrayList(it2);
Assert.assertEquals(res.size(), 200);
@@ -297,7 +297,7 @@ public class RocksDBLabelPartitionGraphStateTest {
graphState.manage().operate().finish();
List<OneDegreeGraph<Integer, Object, Object>> list =
graphState.staticGraph().VE().query()
- .by(EdgeLabelFilter.instance("foo")).asList();
+ .by(EdgeLabelFilter.getInstance("foo")).asList();
Assert.assertEquals(list.size(), 1000);
int key = list.get(0).getKey();
Assert.assertEquals(list.get(0).getVertex(),
@@ -378,7 +378,7 @@ public class RocksDBLabelPartitionGraphStateTest {
Assert.assertEquals(num, 5000);
edges = graphState.staticGraph().E().query("2")
- .by(OutEdgeFilter.instance().and(new EdgeTsFilter(TimeRange.of(0,
1000)))).asList();
+ .by(OutEdgeFilter.getInstance().and(new
EdgeTsFilter(TimeRange.of(0, 1000)))).asList();
Assert.assertEquals(edges.size(), 1000);
maxTime = edges.stream().mapToLong(e -> ((IDLabelTimeEdge)
e).getTime()).max().getAsLong();
@@ -391,7 +391,7 @@ public class RocksDBLabelPartitionGraphStateTest {
Assert.assertEquals(num, 0);
edges = graphState.staticGraph().E().query("2")
- .by(new EdgeTsFilter(TimeRange.of(7000,
10000)).and(EdgeLabelFilter.instance("world")))
+ .by(new EdgeTsFilter(TimeRange.of(7000,
10000)).and(EdgeLabelFilter.getInstance("world")))
.asList();
Assert.assertEquals(edges.size(), 3000);
long minTime = edges.stream().mapToLong(e -> ((IDLabelTimeEdge)
e).getTime()).min()
@@ -466,15 +466,15 @@ public class RocksDBLabelPartitionGraphStateTest {
.limit(1L, 1L).asList();
Assert.assertEquals(list.size(), 6);
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 1L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
1L)
.asList();
Assert.assertEquals(list.size(), 10);
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 2L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
2L)
.asList();
Assert.assertEquals(list.size(), 20);
- List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.instance())
+ List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance())
.select(new DstIdProjector<>()).limit(1L, 2L).asList();
Assert.assertEquals(targetIds.size(), 20);
@@ -535,10 +535,10 @@ public class RocksDBLabelPartitionGraphStateTest {
graphState.manage().operate().archive();
edges = graphState.staticGraph().E().asList();
Assert.assertEquals(edges.size(), 180);
- edges =
graphState.staticGraph().E().query().by(EdgeLabelFilter.instance("person"))
+ edges =
graphState.staticGraph().E().query().by(EdgeLabelFilter.getInstance("person"))
.asList();
Assert.assertEquals(edges.size(), 80);
- edges =
graphState.staticGraph().E().query().by(EdgeLabelFilter.instance("foo")).asList();
+ edges =
graphState.staticGraph().E().query().by(EdgeLabelFilter.getInstance("foo")).asList();
Assert.assertEquals(edges.size(), 100);
vertices = graphState.staticGraph().V().asList();
Assert.assertEquals(vertices.size(), 2);
diff --git
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/StaticGraphStateTest.java
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/StaticGraphStateTest.java
index 29a5a08a..316381dc 100644
---
a/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/StaticGraphStateTest.java
+++
b/geaflow/geaflow-state/geaflow-state-impl/src/test/java/org/apache/geaflow/state/StaticGraphStateTest.java
@@ -210,7 +210,7 @@ public class StaticGraphStateTest {
.limit(1L, 1L).asList();
Assert.assertEquals(list.size(), 6);
- list = graphState.staticGraph().E().query("1", "2",
"3").by(InEdgeFilter.instance())
+ list = graphState.staticGraph().E().query("1", "2",
"3").by(InEdgeFilter.getInstance())
.limit(1L, 1L).asList();
Assert.assertEquals(list.size(), 3);
@@ -218,31 +218,31 @@ public class StaticGraphStateTest {
Assert.assertEquals(list.size(), 2);
list = graphState.staticGraph().E().query("11", "12", "13")
- .by(EdgeTsFilter.instance(10, 20).or(EdgeTsFilter.instance(50,
60)).singleLimit())
+ .by(EdgeTsFilter.getInstance(10,
20).or(EdgeTsFilter.getInstance(50, 60)).singleLimit())
.limit(2L, 1L).asList();
Assert.assertEquals(list.size(), 18);
list = graphState.staticGraph().E().query("11", "12", "13")
- .by(EdgeTsFilter.instance(10, 20).or(EdgeTsFilter.instance(50,
60))).limit(2L, 1L)
+ .by(EdgeTsFilter.getInstance(10,
20).or(EdgeTsFilter.getInstance(50, 60))).limit(2L, 1L)
.asList();
Assert.assertEquals(list.size(), 9);
// full limit
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 1L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
1L)
.asList();
Assert.assertEquals(list.size(), 100);
- list =
graphState.staticGraph().E().query().by(InEdgeFilter.instance()).limit(1L, 2L)
+ list =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance()).limit(1L,
2L)
.asList();
Assert.assertEquals(list.size(), 200);
list = graphState.staticGraph().E().query()
- .by(EdgeTsFilter.instance(10, 20).or(EdgeTsFilter.instance(50,
60)).singleLimit())
+ .by(EdgeTsFilter.getInstance(10,
20).or(EdgeTsFilter.getInstance(50, 60)).singleLimit())
.limit(2L, 1L).asList();
Assert.assertEquals(list.size(), 540);
list = graphState.staticGraph().E().query()
- .by(EdgeTsFilter.instance(10, 20).or(EdgeTsFilter.instance(50,
60))).limit(2L, 1L)
+ .by(EdgeTsFilter.getInstance(10,
20).or(EdgeTsFilter.getInstance(50, 60))).limit(2L, 1L)
.asList();
Assert.assertEquals(list.size(), 270);
@@ -293,37 +293,37 @@ public class StaticGraphStateTest {
graphState.manage().operate().finish();
// project test
- List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.instance())
+ List<String> targetIds =
graphState.staticGraph().E().query().by(InEdgeFilter.getInstance())
.select(new DstIdProjector<>()).limit(1L, 2L).asList();
Assert.assertEquals(targetIds.size(), 20);
targetIds = graphState.staticGraph().E().query()
-
.by(InEdgeFilter.instance().or(OutEdgeFilter.instance()).singleLimit())
+
.by(InEdgeFilter.getInstance().or(OutEdgeFilter.getInstance()).singleLimit())
.select(new DstIdProjector<>()).limit(1L, 2L).asList();
Assert.assertEquals(targetIds.size(), 30);
// full agg test
Map<String, Long> res = graphState.staticGraph().E().query()
- .by(InEdgeFilter.instance().and(new
EdgeLabelFilter("teacher"))).aggregate();
+ .by(InEdgeFilter.getInstance().and(new
EdgeLabelFilter("teacher"))).aggregate();
Assert.assertEquals(res.size(), 10);
Assert.assertTrue(res.get("2") == 1L);
res = graphState.staticGraph().E().query()
- .by(OutEdgeFilter.instance().and(new
EdgeLabelFilter("student"))).aggregate();
+ .by(OutEdgeFilter.getInstance().and(new
EdgeLabelFilter("student"))).aggregate();
Assert.assertEquals(res.size(), 10);
Assert.assertTrue(res.get("2") == 2L);
// key agg test
res = graphState.staticGraph().E().query("2", "5")
- .by(InEdgeFilter.instance().and(new
EdgeLabelFilter("teacher"))).aggregate();
+ .by(InEdgeFilter.getInstance().and(new
EdgeLabelFilter("teacher"))).aggregate();
Assert.assertEquals(res.size(), 2);
Assert.assertTrue(res.get("2") == 1L);
res = graphState.staticGraph().E().query("2", "5")
- .by(InEdgeFilter.instance().and(new EdgeLabelFilter("teacher")),
- OutEdgeFilter.instance().and(new
EdgeLabelFilter("student"))).aggregate();
+ .by(InEdgeFilter.getInstance().and(new EdgeLabelFilter("teacher")),
+ OutEdgeFilter.getInstance().and(new
EdgeLabelFilter("student"))).aggregate();
Assert.assertEquals(res.size(), 2);
Assert.assertTrue(res.get("2") == 1L);
Assert.assertTrue(res.get("5") == 2L);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
