@Test public void testOutputFormat() throws Exception { DataSet<Embedding> input = getExecutionEnvironment().fromElements( createEmbedding(m, e0, n) ); DataSet<Embedding> candidateEdges = getExecutionEnvironment().fromElements( createEmbedding(n, e1, a), createEmbedding(a, e2, b), createEmbedding(b, e3, c) ); DataSet<Embedding> result = getOperator( input, candidateEdges, 2, 1, 3, ExpandDirection.OUT, new ArrayList<>(), new ArrayList<>(), -1 ).evaluate(); assertEquals(3, result.count()); assertEveryEmbedding(result, embedding -> { assertEquals(5, embedding.size()); embedding.getIdList(3); }); assertEmbeddingExists( result, embedding -> (embedding.getIdList(3)).size() == 1 ); assertEmbeddingExists( result, embedding -> (embedding.getIdList(3)).size() == 3 ); assertEmbeddingExists( result, embedding -> (embedding.getIdList(3)).size() == 5 ); }
@Override public DataSet<Embedding> evaluate() { DataSet<ExpandEmbedding> initialWorkingSet = preProcess(); DataSet<ExpandEmbedding> iterationResults = iterate(initialWorkingSet); return postProcess(iterationResults); }
ExpandEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { DataSet<ExpandEmbedding> initialWorkingSet = preProcess(); DataSet<ExpandEmbedding> iterationResults = iterate(initialWorkingSet); return postProcess(iterationResults); } }
ExpandEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { DataSet<ExpandEmbedding> initialWorkingSet = preProcess(); DataSet<ExpandEmbedding> iterationResults = iterate(initialWorkingSet); return postProcess(iterationResults); } ExpandEmbeddings(DataSet<Embedding> input, DataSet<Embedding> candidateEdges,
    int expandColumn, int lowerBound, int upperBound, ExpandDirection direction,
    List<Integer> distinctVertexColumns, List<Integer> distinctEdgeColumns, int closingColumn,
    JoinOperatorBase.JoinHint joinHint); }
ExpandEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { DataSet<ExpandEmbedding> initialWorkingSet = preProcess(); DataSet<ExpandEmbedding> iterationResults = iterate(initialWorkingSet); return postProcess(iterationResults); } ExpandEmbeddings(DataSet<Embedding> input, DataSet<Embedding> candidateEdges,
    int expandColumn, int lowerBound, int upperBound, ExpandDirection direction,
    List<Integer> distinctVertexColumns, List<Integer> distinctEdgeColumns, int closingColumn,
    JoinOperatorBase.JoinHint joinHint); @Override DataSet<Embedding> evaluate(); @Override void setName(String newName); @Override String getName(); }
ExpandEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { DataSet<ExpandEmbedding> initialWorkingSet = preProcess(); DataSet<ExpandEmbedding> iterationResults = iterate(initialWorkingSet); return postProcess(iterationResults); } ExpandEmbeddings(DataSet<Embedding> input, DataSet<Embedding> candidateEdges,
    int expandColumn, int lowerBound, int upperBound, ExpandDirection direction,
    List<Integer> distinctVertexColumns, List<Integer> distinctEdgeColumns, int closingColumn,
    JoinOperatorBase.JoinHint joinHint); @Override DataSet<Embedding> evaluate(); @Override void setName(String newName); @Override String getName();  }