@Test public void testJoin() throws Exception { Embedding l = new Embedding(); l.add(v0, PropertyValue.create("Foobar")); l.add(e0, PropertyValue.create(42)); l.add(v1); DataSet<Embedding> left = getExecutionEnvironment().fromElements(l); Embedding r = new Embedding(); r.add(v1, PropertyValue.create("Baz")); DataSet<Embedding> right = getExecutionEnvironment().fromElements(r); PhysicalOperator join = new JoinEmbeddings(left, right, 1, 2, 0); DataSet<Embedding> result = join.evaluate(); assertEquals(1, result.count()); assertEveryEmbedding(result, embedding -> embedding.getProperties().equals(Lists.newArrayList( PropertyValue.create("Foobar"), PropertyValue.create(42), PropertyValue.create("Baz") ) )); assertEmbeddingExists(result, v0, e0, v1); }
@Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractJoinColumns(leftJoinColumns)) .equalTo(new ExtractJoinColumns(rightJoinColumns)) .with(new MergeEmbeddings(rightColumns, rightJoinColumns, distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight)) .name(getName()); }
JoinEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractJoinColumns(leftJoinColumns)) .equalTo(new ExtractJoinColumns(rightJoinColumns)) .with(new MergeEmbeddings(rightColumns, rightJoinColumns, distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight)) .name(getName()); } }
JoinEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractJoinColumns(leftJoinColumns)) .equalTo(new ExtractJoinColumns(rightJoinColumns)) .with(new MergeEmbeddings(rightColumns, rightJoinColumns, distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight)) .name(getName()); } JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    int leftJoinColumn, int rightJoinColumn);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns,
    List<Integer> distinctVertexColumnsLeft, List<Integer> distinctVertexColumnsRight,
    List<Integer> distinctEdgeColumnsLeft, List<Integer> distinctEdgeColumnsRight);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns,
    List<Integer> distinctVertexColumnsLeft, List<Integer> distinctVertexColumnsRight,
    List<Integer> distinctEdgeColumnsLeft, List<Integer> distinctEdgeColumnsRight,
    JoinOperatorBase.JoinHint joinHint); }
JoinEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractJoinColumns(leftJoinColumns)) .equalTo(new ExtractJoinColumns(rightJoinColumns)) .with(new MergeEmbeddings(rightColumns, rightJoinColumns, distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight)) .name(getName()); } JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    int leftJoinColumn, int rightJoinColumn);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns,
    List<Integer> distinctVertexColumnsLeft, List<Integer> distinctVertexColumnsRight,
    List<Integer> distinctEdgeColumnsLeft, List<Integer> distinctEdgeColumnsRight);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns,
    List<Integer> distinctVertexColumnsLeft, List<Integer> distinctVertexColumnsRight,
    List<Integer> distinctEdgeColumnsLeft, List<Integer> distinctEdgeColumnsRight,
    JoinOperatorBase.JoinHint joinHint); @Override DataSet<Embedding> evaluate(); @Override void setName(String newName); @Override String getName(); }
JoinEmbeddings implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractJoinColumns(leftJoinColumns)) .equalTo(new ExtractJoinColumns(rightJoinColumns)) .with(new MergeEmbeddings(rightColumns, rightJoinColumns, distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight)) .name(getName()); } JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    int leftJoinColumn, int rightJoinColumn);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns,
    List<Integer> distinctVertexColumnsLeft, List<Integer> distinctVertexColumnsRight,
    List<Integer> distinctEdgeColumnsLeft, List<Integer> distinctEdgeColumnsRight);  JoinEmbeddings(DataSet<Embedding> left, DataSet<Embedding> right,
    int rightColumns,
    List<Integer> leftJoinColumns, List<Integer> rightJoinColumns,
    List<Integer> distinctVertexColumnsLeft, List<Integer> distinctVertexColumnsRight,
    List<Integer> distinctEdgeColumnsLeft, List<Integer> distinctEdgeColumnsRight,
    JoinOperatorBase.JoinHint joinHint); @Override DataSet<Embedding> evaluate(); @Override void setName(String newName); @Override String getName();  }