Over extends EvalFunc<DataBag> { @Override public Schema outputSchema(Schema inputSch) { try { FieldSchema field; if (searchInnerType) { field = new FieldSchema(inputSch.getField(0)); while (searchInnerType) { if (field.schema != null && field.schema.getFields().size() > 1) { searchInnerType = false; } else { if (field.type == DataType.TUPLE || field.type == DataType.BAG) { field = new FieldSchema(field.schema.getField(0)); } else { field.alias = field.alias + "_over"; searchInnerType = false; } } } searchInnerType = true; } else if (returnType == DataType.UNKNOWN) { return Schema.generateNestedSchema(DataType.BAG, DataType.NULL); } else { field = new Schema.FieldSchema(returnName, returnType); } Schema outputTupleSchema = new Schema(field); return new Schema(new Schema.FieldSchema(getSchemaName(this .getClass().getName().toLowerCase(), inputSch), outputTupleSchema, DataType.BAG)); } catch (FrontendException fe) { throw new RuntimeException("Unable to create nested schema", fe); } } Over();  Over(String typespec); @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch);  }
@Test public void testSchema() throws Exception { Over func = new Over(); Schema in = Schema.generateNestedSchema(DataType.BAG, DataType.INTEGER); Schema out = func.outputSchema(in); assertEquals("{{NULL}}", out.toString()); func = new Over("chararray"); in = Schema.generateNestedSchema(DataType.BAG, DataType.INTEGER); out = func.outputSchema(in); assertEquals("{org.apache.pig.piggybank.evaluation.over_1: {result: chararray}}", out.toString()); func = new Over("Int"); in = Schema.generateNestedSchema(DataType.BAG, DataType.INTEGER); out = func.outputSchema(in); assertEquals("{org.apache.pig.piggybank.evaluation.over_2: {result: int}}", out.toString()); func = new Over("DOUBLE"); in = Schema.generateNestedSchema(DataType.BAG, DataType.INTEGER); out = func.outputSchema(in); assertEquals("{org.apache.pig.piggybank.evaluation.over_3: {result: double}}", out.toString()); func = new Over("BIGDECIMAL"); in = Schema.generateNestedSchema(DataType.BAG, DataType.INTEGER); out = func.outputSchema(in); assertEquals("{org.apache.pig.piggybank.evaluation.over_4: {result: bigdecimal}}", out.toString()); func = new Over("bob:chararray"); in = Schema.generateNestedSchema(DataType.BAG, DataType.INTEGER); out = func.outputSchema(in); assertEquals("{org.apache.pig.piggybank.evaluation.over_5: {bob: chararray}}", out.toString()); func = new Over("true"); in = Schema.generateNestedSchema(DataType.BAG, DataType.BIGDECIMAL); in.getField(0).schema.getField(0).alias="test"; out = func.outputSchema(in); assertEquals("{org.apache.pig.piggybank.evaluation.over_6: {test_over: bigdecimal}}", out.toString()); }