StringListToListWithIndexes extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() != 1) { throw new IOException("UDF StringListToListWithIndexes takes one argument: a list of strings"); } DataBag stringBag = (DataBag) tuple.get(0); BagFactory bagFactory = BagFactory.getInstance(); DataBag indexedBag = bagFactory.newDefaultBag(); if (stringBag == null) { return indexedBag; } TupleFactory tupleFactory = TupleFactory.getInstance(); int i = 0; for (Tuple stringTuple : stringBag) { Tuple indexedTuple = tupleFactory.newTuple(Lists.newArrayList(i++, stringTuple.get(0))); indexedBag.add(indexedTuple); } return indexedBag; }  @Override DataBag exec(Tuple tuple); @Override Schema outputSchema(Schema input);  }
@Test(expected=IOException.class) public void testExecOnNull() throws Exception { udf.exec(null); }
@Test(expected=IOException.class) public void testExecOnEmpty() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); udf.exec(tupleFactory.newTuple()); }
@Test(expected=IOException.class) public void testExecTooManyArgs() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); udf.exec(tupleFactory.newTuple(Lists.newArrayList("str1", "str2"))); }
@Test(expected=ClassCastException.class) public void testExecOnString() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); udf.exec(tupleFactory.newTuple(Lists.newArrayList("str1"))); }
@Test public void testExecWithEmptyDataBag() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); BagFactory bagFactory = BagFactory.getInstance(); DataBag resultBag = udf.exec(tupleFactory.newTuple(bagFactory.newDefaultBag())); assertNotNull(resultBag); assertEquals(0, resultBag.size()); }
@Test public void testExec() throws Exception { String tuple1Name = "tuple1"; String tuple2Name = "tuple2"; TupleFactory tupleFactory = TupleFactory.getInstance(); BagFactory bagFactory = BagFactory.getInstance(); DataBag sourceBag = bagFactory.newDefaultBag(Lists.newArrayList( tupleFactory.newTuple(tuple1Name), tupleFactory.newTuple(tuple2Name))); DataBag resultBag = udf.exec(tupleFactory.newTuple(sourceBag)); assertNotNull(resultBag); Iterator<Tuple> tupleIt = resultBag.iterator(); assertTrue(tupleIt.hasNext()); Tuple firstTuple = tupleIt.next(); assertEquals(0, firstTuple.get(0)); assertEquals(tuple1Name, firstTuple.get(1)); assertTrue(tupleIt.hasNext()); Tuple secondTuple = tupleIt.next(); assertEquals(1, secondTuple.get(0)); assertEquals(tuple2Name, secondTuple.get(1)); assertFalse(tupleIt.hasNext()); }