IdReplacerUDF extends EvalFunc<Tuple> { @Override public Tuple exec(Tuple input) throws IOException { checkArguments(input); String oldIdPath = (String) input.get(0); String newId = (String) input.get(1); Tuple record = getRecord(input); Schema recordSchema = getRecordSchema(getInputSchema()); replaceIdFieldInTuple(newId, oldIdPath, record, recordSchema); return record; }  @Override /** * UDF replaces original id with new id string. The input tuple should contain following elements: * a dot-separated list of field ids representing a path in the input tree structure (first element), * new id string (second element), the rest of the tuple should contain the consecutive fields of the input record. * * Replacing of multiple id fields is not implemented yet. Only one field replacing is implemented. * * @param tuple a tuple (id path - dot-separated list of fields indicating position in the record schema tree, new id string, input tuple fields...) * @return * @throws IOException */ Tuple exec(Tuple input); @Override Schema outputSchema(Schema input);  }
@Test(expected=IllegalArgumentException.class) public void testExecOnNull() throws Exception { idReplacer.exec(null); }
@Test(expected=IllegalArgumentException.class) public void testExecOnEmpty() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); idReplacer.exec(tupleFactory.newTuple()); }
@Test(expected=IllegalArgumentException.class) public void testExecNotEnoughArgs() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); idReplacer.exec(tupleFactory.newTuple(Lists.newArrayList( "str1", "str2"))); }
@Test public void testExecSingleIdField() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); Schema schema = new Schema(); schema.add(new FieldSchema(null, DataType.CHARARRAY)); schema.add(new FieldSchema(null, DataType.CHARARRAY)); schema.add(new FieldSchema("id", DataType.CHARARRAY)); idReplacer.setInputSchema(schema); String idFieldName = "id"; String newId = "updatedId"; String oldId = "oldId"; Tuple result = idReplacer.exec(tupleFactory.newTuple(Lists.newArrayList( idFieldName, newId, oldId))); assertNotNull(result); assertEquals(1, result.getAll().size()); assertEquals(newId, result.get(0)); }
@SuppressWarnings("unchecked") @Test public void testExecMultipleIdFields() throws Exception { TupleFactory tupleFactory = TupleFactory.getInstance(); Schema schema = new Schema(); schema.add(new FieldSchema(null, DataType.CHARARRAY)); schema.add(new FieldSchema(null, DataType.CHARARRAY)); schema.add(new FieldSchema("id", DataType.CHARARRAY)); schema.add(new FieldSchema("id", DataType.CHARARRAY)); schema.add(new FieldSchema("irrelevant", DataType.BOOLEAN)); idReplacer.setInputSchema(schema); String idFieldName = "id"; String newId = "updatedId"; String oldId1 = "oldId1"; String oldId2 = "oldId2"; Tuple result = idReplacer.exec(tupleFactory.newTuple(Lists.newArrayList( idFieldName, newId, oldId1, oldId2, true))); assertNotNull(result); assertEquals(3, result.getAll().size()); assertEquals(oldId1, result.get(0)); assertEquals(newId, result.get(1)); assertTrue((boolean)result.get(2)); }