NullTupleFieldsToNull extends EvalFunc<Tuple> { @Override public Tuple exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() != 1) { return null; } Tuple first = (Tuple) tuple.get(0); if (first == null) { return null; } for (int i = 0; i < first.size(); i++) { if (!first.isNull(i)) { return first; } } return null; }  @Override Tuple exec(Tuple tuple); @Override Schema outputSchema(Schema input);  }
@Test public void testUDF() throws IOException { NullTupleFieldsToNull udf = new NullTupleFieldsToNull(); TupleFactory tupleFactory = TupleFactory.getInstance(); BagFactory bagFactory = BagFactory.getInstance(); DataBag emptyBag = bagFactory.newDefaultBag(); DataBag bag = bagFactory.newDefaultBag(Arrays.asList(new Tuple[]{tupleFactory.newTuple("tup1")})); Tuple nullTuple = tupleFactory.newTuple(Lists.newArrayList(null, null, null)); Tuple tuple = tupleFactory.newTuple(Lists.newArrayList(null, null, "tup1")); assertNull(udf.exec(null)); assertNull(udf.exec(tupleFactory.newTuple())); assertNull(udf.exec(tupleFactory.newTuple((Tuple)null))); assertNull(udf.exec(tupleFactory.newTuple(tupleFactory.newTuple()))); assertNull(udf.exec(tupleFactory.newTuple(nullTuple))); assertEquals(tuple, udf.exec(tupleFactory.newTuple(tuple))); assertNull(udf.exec(tupleFactory.newTuple(Lists.newArrayList(bag, emptyBag)))); }