@Test public void testUDF() throws IOException { StringBagsDifference udf = new StringBagsDifference(); TupleFactory tupleFactory = TupleFactory.getInstance(); BagFactory bagFactory = BagFactory.getInstance(); DataBag nullBag = null; DataBag emptyBag = bagFactory.newDefaultBag(); DataBag bag1 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup1"), tupleFactory.newTuple("tup2"))); DataBag bag2 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup3"), tupleFactory.newTuple("tup4"))); DataBag bag3 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup1"), tupleFactory.newTuple("tup4"), tupleFactory.newTuple("tup5"))); DataBag bag4 = bagFactory.newDefaultBag(Lists.newArrayList(tupleFactory.newTuple("tup1"), tupleFactory.newTuple("tup5"))); assertNull(udf.exec(null)); assertNull(udf.exec(tupleFactory.newTuple())); assertNull(udf.exec(tupleFactory.newTuple(nullBag))); assertNull(udf.exec(tupleFactory.newTuple(emptyBag))); assertNull(udf.exec(tupleFactory.newTuple(Lists.newArrayList(bag1, emptyBag, bag2, bag3)))); assertNull(udf.exec(tupleFactory.newTuple(Lists.newArrayList(nullBag, bag3, bag2)))); assertEquals(bag1, udf.exec(tupleFactory.newTuple(Lists.newArrayList(bag1, bag2)))); assertEquals(bag4, udf.exec(tupleFactory.newTuple(Lists.newArrayList(bag3, bag2)))); assertNull(udf.exec(tupleFactory.newTuple(Lists.newArrayList(bag4, bag3)))); }
@Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() != 2) { return null; } DataBag dbMain = (DataBag) tuple.get(0); DataBag dbSub = (DataBag) tuple.get(1); if (dbMain == null || dbSub == null) { return dbMain; } List<Tuple> tuples = new ArrayList<Tuple>(); Iterator<Tuple> itMain = dbMain.iterator(); while (itMain.hasNext()) { tuples.add(itMain.next()); } Iterator<Tuple> itSub = dbSub.iterator(); while (itSub.hasNext()) { tuples.remove(itSub.next()); } if (tuples.isEmpty()) { return null; } BagFactory bagFactory = BagFactory.getInstance(); return bagFactory.newDefaultBag(tuples); }
StringBagsDifference extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() != 2) { return null; } DataBag dbMain = (DataBag) tuple.get(0); DataBag dbSub = (DataBag) tuple.get(1); if (dbMain == null || dbSub == null) { return dbMain; } List<Tuple> tuples = new ArrayList<Tuple>(); Iterator<Tuple> itMain = dbMain.iterator(); while (itMain.hasNext()) { tuples.add(itMain.next()); } Iterator<Tuple> itSub = dbSub.iterator(); while (itSub.hasNext()) { tuples.remove(itSub.next()); } if (tuples.isEmpty()) { return null; } BagFactory bagFactory = BagFactory.getInstance(); return bagFactory.newDefaultBag(tuples); } }
StringBagsDifference extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() != 2) { return null; } DataBag dbMain = (DataBag) tuple.get(0); DataBag dbSub = (DataBag) tuple.get(1); if (dbMain == null || dbSub == null) { return dbMain; } List<Tuple> tuples = new ArrayList<Tuple>(); Iterator<Tuple> itMain = dbMain.iterator(); while (itMain.hasNext()) { tuples.add(itMain.next()); } Iterator<Tuple> itSub = dbSub.iterator(); while (itSub.hasNext()) { tuples.remove(itSub.next()); } if (tuples.isEmpty()) { return null; } BagFactory bagFactory = BagFactory.getInstance(); return bagFactory.newDefaultBag(tuples); }  }
StringBagsDifference extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() != 2) { return null; } DataBag dbMain = (DataBag) tuple.get(0); DataBag dbSub = (DataBag) tuple.get(1); if (dbMain == null || dbSub == null) { return dbMain; } List<Tuple> tuples = new ArrayList<Tuple>(); Iterator<Tuple> itMain = dbMain.iterator(); while (itMain.hasNext()) { tuples.add(itMain.next()); } Iterator<Tuple> itSub = dbSub.iterator(); while (itSub.hasNext()) { tuples.remove(itSub.next()); } if (tuples.isEmpty()) { return null; } BagFactory bagFactory = BagFactory.getInstance(); return bagFactory.newDefaultBag(tuples); }  @Override DataBag exec(Tuple tuple); @Override Schema outputSchema(Schema input); }
StringBagsDifference extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() != 2) { return null; } DataBag dbMain = (DataBag) tuple.get(0); DataBag dbSub = (DataBag) tuple.get(1); if (dbMain == null || dbSub == null) { return dbMain; } List<Tuple> tuples = new ArrayList<Tuple>(); Iterator<Tuple> itMain = dbMain.iterator(); while (itMain.hasNext()) { tuples.add(itMain.next()); } Iterator<Tuple> itSub = dbSub.iterator(); while (itSub.hasNext()) { tuples.remove(itSub.next()); } if (tuples.isEmpty()) { return null; } BagFactory bagFactory = BagFactory.getInstance(); return bagFactory.newDefaultBag(tuples); }  @Override DataBag exec(Tuple tuple); @Override Schema outputSchema(Schema input);  }