@Test public void join2Test_1parent_1child() throws Exception { CompositeMetadata amd = getCmd("A", projection("[{'field':'obj1.c'},{'field':'obj1.c.*.b'}]")); CompositeMetadata cmd = amd.getChildMetadata(new Path("obj1.c")); CompositeMetadata bmd = cmd.getChildMetadata(new Path("obj1.c.*.b")); Assert.assertNotNull(bmd); Assert.assertNotNull(cmd); QueryPlan qp = new QueryPlan(amd, new SimpleScorer()); ExecutionBlock ablock = new ExecutionBlock(amd, qp.getNode(amd)); ExecutionBlock bblock = new ExecutionBlock(amd, qp.getNode(bmd)); ExecutionBlock cblock = new ExecutionBlock(amd, qp.getNode(cmd)); cblock.setAssociationQuery(ablock, new AssociationQuery(amd, cmd, (ResolvedReferenceField) amd.resolve(new Path("obj1.c")), new ArrayList<Conjunct>())); cblock.setAssociationQuery(bblock, new AssociationQuery(amd, cmd, (ResolvedReferenceField) cmd.resolve(new Path("b")), new ArrayList<Conjunct>())); qp.flip(qp.getNode(bmd), qp.getNode(cmd)); cblock.addSourceBlock(ablock); cblock.addSourceBlock(bblock); ablock.linkBlocks(); bblock.linkBlocks(); cblock.linkBlocks(); TestStep a = new TestStep(ablock, resultDoc(ablock, "{'_id':1}"), resultDoc(ablock, "{'_id':2}"), resultDoc(ablock, "{'_id':3}"), resultDoc(ablock, "{'_id':4}"), resultDoc(ablock, "{'_id':5}"), resultDoc(ablock, "{'_id':6}")); TestStep b = new TestStep(bblock, resultDoc(bblock, "{'_id':'a'}"), resultDoc(bblock, "{'_id':'b'}"), resultDoc(bblock, "{'_id':'c'}"), resultDoc(bblock, "{'_id':'d'}")); Join join = new Join(cblock, new Source[]{new Source<>(a), new Source<>(b)}); FindRequest freq = new FindRequest(); freq.setEntityVersion(new EntityVersion("test", "1")); OperationContext opctx = new OperationContext(freq, null, new Factory(), CRUDOperation.FIND, null, null, new HashSet<String>(), null); ExecutionContext ctx = new ExecutionContext(opctx, Executors.newSingleThreadExecutor()); StepResult<JoinTuple> result = join.getResults(ctx); Stream<JoinTuple> stream = result.stream(); stream.forEach(tuple -> System.out.println(tuple)); Iterator<JoinTuple> itr = result.stream().iterator(); for (int i : new int[]{1, 2, 3, 4, 5, 6}) { for (String j : new String[]{"a", "b", "c", "d"}) { JoinTuple tp = itr.next(); Assert.assertEquals(i, tp.getParentDocument().getDoc().getRoot().get("_id").asInt()); Assert.assertEquals("obj1", tp.getParentDocumentSlot().getLocalContainerName().toString()); Assert.assertEquals(j, tp.getChildTuple().get(0).getDoc().getRoot().get("_id").asText()); } } Assert.assertFalse(itr.hasNext()); }
@Override public StepResult<JoinTuple> getResults(ExecutionContext ctx) { int parentIndex = -1; Future<StepResult<ResultDocument>>[] futureResults = new Future[sources.length]; int i = 0; for (Source<ResultDocument> source : sources) { if (source.getStep().getBlock().getMetadata() == block.getMetadata().getParent()) { parentIndex = i; } futureResults[i++] = ctx.getExecutor().submit(() -> { return source.getStep().getResults(ctx); }); } Tuples<ResultDocument> tuples = new Tuples(); i = 0; for (Future<StepResult<ResultDocument>> futureResult : futureResults) { tuples.add(() -> { try { return futureResult.get().stream().iterator(); } catch (InterruptedException x) { throw new RuntimeException(x); } catch (ExecutionException ex) { throw new RuntimeException(ex); } } ); } if (ctx.hasErrors()) { return StepResult.EMPTY; } return new JoinStream(tuples, parentIndex, parentIndex == -1 ? null : block.getAssociationQueryForEdge(sources[parentIndex].getBlock()). getReference()); }
Join extends Step<JoinTuple> { @Override public StepResult<JoinTuple> getResults(ExecutionContext ctx) { int parentIndex = -1; Future<StepResult<ResultDocument>>[] futureResults = new Future[sources.length]; int i = 0; for (Source<ResultDocument> source : sources) { if (source.getStep().getBlock().getMetadata() == block.getMetadata().getParent()) { parentIndex = i; } futureResults[i++] = ctx.getExecutor().submit(() -> { return source.getStep().getResults(ctx); }); } Tuples<ResultDocument> tuples = new Tuples(); i = 0; for (Future<StepResult<ResultDocument>> futureResult : futureResults) { tuples.add(() -> { try { return futureResult.get().stream().iterator(); } catch (InterruptedException x) { throw new RuntimeException(x); } catch (ExecutionException ex) { throw new RuntimeException(ex); } } ); } if (ctx.hasErrors()) { return StepResult.EMPTY; } return new JoinStream(tuples, parentIndex, parentIndex == -1 ? null : block.getAssociationQueryForEdge(sources[parentIndex].getBlock()). getReference()); } }
Join extends Step<JoinTuple> { @Override public StepResult<JoinTuple> getResults(ExecutionContext ctx) { int parentIndex = -1; Future<StepResult<ResultDocument>>[] futureResults = new Future[sources.length]; int i = 0; for (Source<ResultDocument> source : sources) { if (source.getStep().getBlock().getMetadata() == block.getMetadata().getParent()) { parentIndex = i; } futureResults[i++] = ctx.getExecutor().submit(() -> { return source.getStep().getResults(ctx); }); } Tuples<ResultDocument> tuples = new Tuples(); i = 0; for (Future<StepResult<ResultDocument>> futureResult : futureResults) { tuples.add(() -> { try { return futureResult.get().stream().iterator(); } catch (InterruptedException x) { throw new RuntimeException(x); } catch (ExecutionException ex) { throw new RuntimeException(ex); } } ); } if (ctx.hasErrors()) { return StepResult.EMPTY; } return new JoinStream(tuples, parentIndex, parentIndex == -1 ? null : block.getAssociationQueryForEdge(sources[parentIndex].getBlock()). getReference()); } Join(ExecutionBlock block, Source<ResultDocument>[] sources); }
Join extends Step<JoinTuple> { @Override public StepResult<JoinTuple> getResults(ExecutionContext ctx) { int parentIndex = -1; Future<StepResult<ResultDocument>>[] futureResults = new Future[sources.length]; int i = 0; for (Source<ResultDocument> source : sources) { if (source.getStep().getBlock().getMetadata() == block.getMetadata().getParent()) { parentIndex = i; } futureResults[i++] = ctx.getExecutor().submit(() -> { return source.getStep().getResults(ctx); }); } Tuples<ResultDocument> tuples = new Tuples(); i = 0; for (Future<StepResult<ResultDocument>> futureResult : futureResults) { tuples.add(() -> { try { return futureResult.get().stream().iterator(); } catch (InterruptedException x) { throw new RuntimeException(x); } catch (ExecutionException ex) { throw new RuntimeException(ex); } } ); } if (ctx.hasErrors()) { return StepResult.EMPTY; } return new JoinStream(tuples, parentIndex, parentIndex == -1 ? null : block.getAssociationQueryForEdge(sources[parentIndex].getBlock()). getReference()); } Join(ExecutionBlock block, Source<ResultDocument>[] sources); Source<ResultDocument>[] getSources(); @Override StepResult<JoinTuple> getResults(ExecutionContext ctx); @Override JsonNode toJson(); @Override JsonNode explain(ExecutionContext ctx); }
Join extends Step<JoinTuple> { @Override public StepResult<JoinTuple> getResults(ExecutionContext ctx) { int parentIndex = -1; Future<StepResult<ResultDocument>>[] futureResults = new Future[sources.length]; int i = 0; for (Source<ResultDocument> source : sources) { if (source.getStep().getBlock().getMetadata() == block.getMetadata().getParent()) { parentIndex = i; } futureResults[i++] = ctx.getExecutor().submit(() -> { return source.getStep().getResults(ctx); }); } Tuples<ResultDocument> tuples = new Tuples(); i = 0; for (Future<StepResult<ResultDocument>> futureResult : futureResults) { tuples.add(() -> { try { return futureResult.get().stream().iterator(); } catch (InterruptedException x) { throw new RuntimeException(x); } catch (ExecutionException ex) { throw new RuntimeException(ex); } } ); } if (ctx.hasErrors()) { return StepResult.EMPTY; } return new JoinStream(tuples, parentIndex, parentIndex == -1 ? null : block.getAssociationQueryForEdge(sources[parentIndex].getBlock()). getReference()); } Join(ExecutionBlock block, Source<ResultDocument>[] sources); Source<ResultDocument>[] getSources(); @Override StepResult<JoinTuple> getResults(ExecutionContext ctx); @Override JsonNode toJson(); @Override JsonNode explain(ExecutionContext ctx);  }