Stitch extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() == 0) return null; List<DataBag> bags = new ArrayList<DataBag>(input.size()); for (int i = 0; i < input.size(); i++) { Object o = input.get(i); try { bags.add((DataBag)o); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Stitch expected bags as input but argument " + i + " is a " + DataType.findTypeName(o); throw new ExecException(msg, errCode, PigException.INPUT); } } if (bags.size() == 1) return bags.get(0); DataBag output = BagFactory.getInstance().newDefaultBag(); List<Iterator<Tuple>> iters = new ArrayList<Iterator<Tuple>>(bags.size()); for (DataBag bag : bags) { iters.add(bag.iterator()); } while (iters.get(0).hasNext()) { Tuple outTuple = TupleFactory.getInstance().newTuple(); for (Iterator<Tuple> iter : iters) { if (iter.hasNext()) { Tuple t = iter.next(); List<Object> fields = t.getAll(); for (Object field : fields) { outTuple.append(field); } } } output.add(outTuple); } return output; }  @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch);  }
@Test public void testNoInput() throws Exception { Stitch func = new Stitch(); assertNull(func.exec(null)); Tuple t = TupleFactory.getInstance().newTuple(); assertNull(func.exec(t)); }
@Test public void testBadInput() throws Exception { Stitch func = new Stitch(); boolean caught = false; Tuple t = TupleFactory.getInstance().newTuple(); t.append("Mary had a little lamb"); try { func.exec(t); } catch (ExecException ee) { caught = true; } assertTrue(caught); DataBag b = BagFactory.getInstance().newDefaultBag(); b.add(t); t = TupleFactory.getInstance().newTuple(); t.append(b); t.append("its fleece was white as snow"); caught = false; try { func.exec(t); } catch (ExecException ee) { caught = true; } assertTrue(caught); }
@Test public void testSingleInput() throws Exception { Stitch func = new Stitch(); Tuple t = TupleFactory.getInstance().newTuple(); t.append(new Integer(1)); DataBag b = BagFactory.getInstance().newDefaultBag(); b.add(t); t = TupleFactory.getInstance().newTuple(); t.append(b); DataBag out = func.exec(t); assertEquals(1, out.size()); assertEquals(new Integer(1), out.iterator().next().get(0)); }
@Test public void testDoubleInput() throws Exception { Stitch func = new Stitch(); DataBag b1 = BagFactory.getInstance().newDefaultBag(); Tuple t = TupleFactory.getInstance().newTuple(); t.append("a"); t.append("b"); b1.add(t); t = TupleFactory.getInstance().newTuple(); t.append("c"); t.append("d"); b1.add(t); DataBag b2 = BagFactory.getInstance().newDefaultBag(); t = TupleFactory.getInstance().newTuple(); t.append("1"); t.append("2"); b2.add(t); t = TupleFactory.getInstance().newTuple(); t.append("3"); t.append("4"); b2.add(t); t = TupleFactory.getInstance().newTuple(); t.append(b1); t.append(b2); DataBag out = func.exec(t); assertEquals(2, out.size()); Iterator<Tuple> iter = out.iterator(); t = iter.next(); assertEquals(4, t.size()); assertEquals("a", t.get(0)); assertEquals("b", t.get(1)); assertEquals("1", t.get(2)); assertEquals("2", t.get(3)); t = iter.next(); assertEquals(4, t.size()); assertEquals("c", t.get(0)); assertEquals("d", t.get(1)); assertEquals("3", t.get(2)); assertEquals("4", t.get(3)); }
@Test public void testSecondShort() throws Exception { Stitch func = new Stitch(); DataBag b1 = BagFactory.getInstance().newDefaultBag(); Tuple t = TupleFactory.getInstance().newTuple(); t.append("a"); t.append("b"); b1.add(t); t = TupleFactory.getInstance().newTuple(); t.append("c"); t.append("d"); b1.add(t); DataBag b2 = BagFactory.getInstance().newDefaultBag(); t = TupleFactory.getInstance().newTuple(); t.append("1"); t.append("2"); b2.add(t); t = TupleFactory.getInstance().newTuple(); t.append(b1); t.append(b2); DataBag out = func.exec(t); assertEquals(2, out.size()); Iterator<Tuple> iter = out.iterator(); t = iter.next(); assertEquals(4, t.size()); assertEquals("a", t.get(0)); assertEquals("b", t.get(1)); assertEquals("1", t.get(2)); assertEquals("2", t.get(3)); t = iter.next(); assertEquals(2, t.size()); assertEquals("c", t.get(0)); assertEquals("d", t.get(1)); }
@Test public void testFirstShort() throws Exception { Stitch func = new Stitch(); DataBag b1 = BagFactory.getInstance().newDefaultBag(); Tuple t = TupleFactory.getInstance().newTuple(); t.append("a"); t.append("b"); b1.add(t); DataBag b2 = BagFactory.getInstance().newDefaultBag(); t = TupleFactory.getInstance().newTuple(); t.append("1"); t.append("2"); b2.add(t); t = TupleFactory.getInstance().newTuple(); t.append("3"); t.append("4"); b2.add(t); t = TupleFactory.getInstance().newTuple(); t.append(b1); t.append(b2); DataBag out = func.exec(t); assertEquals(1, out.size()); Iterator<Tuple> iter = out.iterator(); t = iter.next(); assertEquals(4, t.size()); assertEquals("a", t.get(0)); assertEquals("b", t.get(1)); assertEquals("1", t.get(2)); assertEquals("2", t.get(3)); }