@Test public void testMultipleJoinPartners() throws Exception { Embedding l1 = new Embedding(); l1.add(v0, PropertyValue.create("Foobar"), PropertyValue.create(21)); Embedding l2 = new Embedding(); l2.add(v1, PropertyValue.create("Foobar"), PropertyValue.create(42)); DataSet<Embedding> left = getExecutionEnvironment().fromElements(l1, l2); Embedding r1 = new Embedding(); r1.add(v2, PropertyValue.create("Foobar"), PropertyValue.create(21)); Embedding r2 = new Embedding(); r2.add(v3, PropertyValue.create("Foobar"), PropertyValue.create(42)); DataSet<Embedding> right = getExecutionEnvironment().fromElements(r1, r2); PhysicalOperator join = new ValueJoin(left, right, Lists.newArrayList(0, 1), Lists.newArrayList(0, 1), 1); DataSet<Embedding> result = join.evaluate(); assertEquals(2, result.count()); assertEmbeddingExists(result, v0, v2); assertEmbeddingExists(result, v1, v3); }
@Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractPropertyJoinColumns(leftJoinProperties)) .equalTo(new ExtractPropertyJoinColumns(rightJoinProperties)) .with(new MergeEmbeddings( rightColumns, Lists.newArrayListWithCapacity(0), distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight )) .name(getName()); }
ValueJoin implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractPropertyJoinColumns(leftJoinProperties)) .equalTo(new ExtractPropertyJoinColumns(rightJoinProperties)) .with(new MergeEmbeddings( rightColumns, Lists.newArrayListWithCapacity(0), distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight )) .name(getName()); } }
ValueJoin implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractPropertyJoinColumns(leftJoinProperties)) .equalTo(new ExtractPropertyJoinColumns(rightJoinProperties)) .with(new MergeEmbeddings( rightColumns, Lists.newArrayListWithCapacity(0), distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight )) .name(getName()); } ValueJoin(DataSet<Embedding> left, DataSet<Embedding> right,
    List<Integer> leftJoinProperties, List<Integer> rightJoinProperties,
    int rightColumns);  ValueJoin(DataSet<Embedding> left, DataSet<Embedding> right,
    List<Integer> leftJoinProperties, List<Integer> rightJoinProperties, int rightColumns,
    List<Integer> distinctVertexColumnsLeft, List<Integer> distinctVertexColumnsRight,
    List<Integer> distinctEdgeColumnsLeft, List<Integer> distinctEdgeColumnsRight,
    JoinOperatorBase.JoinHint joinHint); }
ValueJoin implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractPropertyJoinColumns(leftJoinProperties)) .equalTo(new ExtractPropertyJoinColumns(rightJoinProperties)) .with(new MergeEmbeddings( rightColumns, Lists.newArrayListWithCapacity(0), distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight )) .name(getName()); } ValueJoin(DataSet<Embedding> left, DataSet<Embedding> right,
    List<Integer> leftJoinProperties, List<Integer> rightJoinProperties,
    int rightColumns);  ValueJoin(DataSet<Embedding> left, DataSet<Embedding> right,
    List<Integer> leftJoinProperties, List<Integer> rightJoinProperties, int rightColumns,
    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(); }
ValueJoin implements PhysicalOperator { @Override public DataSet<Embedding> evaluate() { return left.join(right, joinHint) .where(new ExtractPropertyJoinColumns(leftJoinProperties)) .equalTo(new ExtractPropertyJoinColumns(rightJoinProperties)) .with(new MergeEmbeddings( rightColumns, Lists.newArrayListWithCapacity(0), distinctVertexColumnsLeft, distinctVertexColumnsRight, distinctEdgeColumnsLeft, distinctEdgeColumnsRight )) .name(getName()); } ValueJoin(DataSet<Embedding> left, DataSet<Embedding> right,
    List<Integer> leftJoinProperties, List<Integer> rightJoinProperties,
    int rightColumns);  ValueJoin(DataSet<Embedding> left, DataSet<Embedding> right,
    List<Integer> leftJoinProperties, List<Integer> rightJoinProperties, int rightColumns,
    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();  }