NLJoinExec extends CommonJoinExec { public Tuple next() throws IOException { while (!context.isStopped()) { if (needNewOuter) { outerTuple = leftChild.next(); if (outerTuple == null) { return null; } needNewOuter = false; } innerTuple = rightChild.next(); if (innerTuple == null) { needNewOuter = true; rightChild.rescan(); continue; } frameTuple.set(outerTuple, innerTuple); if (hasJoinQual) { if (joinQual.eval(frameTuple).isTrue()) { projector.eval(frameTuple, outTuple); return outTuple; } } else { projector.eval(frameTuple, outTuple); return outTuple; } } return null; } NLJoinExec(TaskAttemptContext context, JoinNode plan, PhysicalExec outer,
      PhysicalExec inner); Tuple next(); @Override void rescan();  }
@Test public final void testNLCrossJoin() throws IOException, PlanningException { FileFragment[] empFrags = FileStorageManager.splitNG(conf, "default.e", employee.getMeta(), new Path(employee.getPath()), Integer.MAX_VALUE); FileFragment[] peopleFrags = FileStorageManager.splitNG(conf, "default.p", people.getMeta(), new Path(people.getPath()), Integer.MAX_VALUE); FileFragment[] merged = TUtil.concat(empFrags, peopleFrags); Path workDir = CommonTestingUtil.getTestDir(TajoTestingCluster.DEFAULT_TEST_DIRECTORY + "/testNLCrossJoin"); TaskAttemptContext ctx = new TaskAttemptContext(new QueryContext(conf), LocalTajoTestingUtility.newTaskAttemptId(), merged, workDir); ctx.setEnforcer(new Enforcer()); Expr expr = analyzer.parse(QUERIES[0]); LogicalNode plan = planner.createPlan(LocalTajoTestingUtility.createDummyContext(conf), expr).getRootBlock().getRoot(); PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf); PhysicalExec exec = phyPlanner.createPlan(ctx, plan); int i = 0; exec.init(); while (exec.next() != null) { i++; } exec.close(); assertEquals(50*50/2, i); }
@Test public final void testNLInnerJoin() throws IOException, PlanningException { FileFragment[] empFrags = FileStorageManager.splitNG(conf, "default.e", employee.getMeta(), new Path(employee.getPath()), Integer.MAX_VALUE); FileFragment[] peopleFrags = FileStorageManager.splitNG(conf, "default.p", people.getMeta(), new Path(people.getPath()), Integer.MAX_VALUE); FileFragment[] merged = TUtil.concat(empFrags, peopleFrags); Path workDir = CommonTestingUtil.getTestDir(TajoTestingCluster.DEFAULT_TEST_DIRECTORY + "/testNLInnerJoin"); TaskAttemptContext ctx = new TaskAttemptContext(new QueryContext(conf), LocalTajoTestingUtility.newTaskAttemptId(masterPlan), merged, workDir); ctx.setEnforcer(new Enforcer()); Expr expr = analyzer.parse(QUERIES[1]); LogicalNode plan = planner.createPlan(LocalTajoTestingUtility.createDummyContext(conf), expr).getRootBlock().getRoot(); PhysicalPlanner phyPlanner = new PhysicalPlannerImpl(conf); PhysicalExec exec = phyPlanner.createPlan(ctx, plan); Tuple tuple; int i = 1; int count = 0; exec.init(); while ((tuple = exec.next()) != null) { count++; assertTrue(i == tuple.get(0).asInt4()); assertTrue(i == tuple.get(1).asInt4()); assertTrue(("dept_" + i).equals(tuple.get(2).asChars())); assertTrue(10 + i == tuple.get(3).asInt4()); i += 2; } exec.close(); assertEquals(50 / 2, count); }