@Test public void testLagWithRowsBehindNoDefault() throws Exception { Over func = new Over(); DataBag inbag = BagFactory.getInstance().newDefaultBag(); for (int i = 0; i < 10; i++) { Tuple t = TupleFactory.getInstance().newTuple(1); t.set(0, i); inbag.add(t); } Tuple t = TupleFactory.getInstance().newTuple(5); t.set(0, inbag); t.set(1, "lag"); t.set(2, -1); t.set(3, -1); t.set(4, 3); DataBag outbag = func.exec(t); assertEquals(10, outbag.size()); int count = -3; for (Tuple to : outbag) { assertEquals(1, to.size()); try { if (count >= 0) assertEquals(new Integer(count), to.get(0)); else assertNull(to.get(0)); } finally { count++; } } }
@Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() < 2) { int errCode = 2107; String msg = "Over expected 2 or more inputs but received " + input.size(); throw new ExecException(msg, errCode, PigException.INPUT); } DataBag inbag = null; try { inbag = (DataBag)input.get(0); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Over expected a bag for arg 1 but received " + DataType.findTypeName(input.get(0)); throw new ExecException(msg, errCode, PigException.INPUT); } if (!initialized) { init(input); } else { if (func instanceof ResetableEvalFunc) { ((ResetableEvalFunc)func).reset(); } } OverBag tmpbag = new OverBag(inbag, rowsBefore, rowsAfter); Tuple tmptuple = mTupleFactory.newTuple(1); tmptuple.set(0, tmpbag); DataBag outbag = BagFactory.getInstance().newDefaultBag(); for (int i = 0; i < inbag.size(); i++) { tmpbag.setCurrentRow(i); Tuple t = mTupleFactory.newTuple(1); t.set(0, func.exec(tmptuple)); outbag.add(t); } return outbag; }
Over extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() < 2) { int errCode = 2107; String msg = "Over expected 2 or more inputs but received " + input.size(); throw new ExecException(msg, errCode, PigException.INPUT); } DataBag inbag = null; try { inbag = (DataBag)input.get(0); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Over expected a bag for arg 1 but received " + DataType.findTypeName(input.get(0)); throw new ExecException(msg, errCode, PigException.INPUT); } if (!initialized) { init(input); } else { if (func instanceof ResetableEvalFunc) { ((ResetableEvalFunc)func).reset(); } } OverBag tmpbag = new OverBag(inbag, rowsBefore, rowsAfter); Tuple tmptuple = mTupleFactory.newTuple(1); tmptuple.set(0, tmpbag); DataBag outbag = BagFactory.getInstance().newDefaultBag(); for (int i = 0; i < inbag.size(); i++) { tmpbag.setCurrentRow(i); Tuple t = mTupleFactory.newTuple(1); t.set(0, func.exec(tmptuple)); outbag.add(t); } return outbag; } }
Over extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() < 2) { int errCode = 2107; String msg = "Over expected 2 or more inputs but received " + input.size(); throw new ExecException(msg, errCode, PigException.INPUT); } DataBag inbag = null; try { inbag = (DataBag)input.get(0); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Over expected a bag for arg 1 but received " + DataType.findTypeName(input.get(0)); throw new ExecException(msg, errCode, PigException.INPUT); } if (!initialized) { init(input); } else { if (func instanceof ResetableEvalFunc) { ((ResetableEvalFunc)func).reset(); } } OverBag tmpbag = new OverBag(inbag, rowsBefore, rowsAfter); Tuple tmptuple = mTupleFactory.newTuple(1); tmptuple.set(0, tmpbag); DataBag outbag = BagFactory.getInstance().newDefaultBag(); for (int i = 0; i < inbag.size(); i++) { tmpbag.setCurrentRow(i); Tuple t = mTupleFactory.newTuple(1); t.set(0, func.exec(tmptuple)); outbag.add(t); } return outbag; } Over();  Over(String typespec); }
Over extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() < 2) { int errCode = 2107; String msg = "Over expected 2 or more inputs but received " + input.size(); throw new ExecException(msg, errCode, PigException.INPUT); } DataBag inbag = null; try { inbag = (DataBag)input.get(0); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Over expected a bag for arg 1 but received " + DataType.findTypeName(input.get(0)); throw new ExecException(msg, errCode, PigException.INPUT); } if (!initialized) { init(input); } else { if (func instanceof ResetableEvalFunc) { ((ResetableEvalFunc)func).reset(); } } OverBag tmpbag = new OverBag(inbag, rowsBefore, rowsAfter); Tuple tmptuple = mTupleFactory.newTuple(1); tmptuple.set(0, tmpbag); DataBag outbag = BagFactory.getInstance().newDefaultBag(); for (int i = 0; i < inbag.size(); i++) { tmpbag.setCurrentRow(i); Tuple t = mTupleFactory.newTuple(1); t.set(0, func.exec(tmptuple)); outbag.add(t); } return outbag; } Over();  Over(String typespec); @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch); }
Over extends EvalFunc<DataBag> { @Override public DataBag exec(Tuple input) throws IOException { if (input == null || input.size() < 2) { int errCode = 2107; String msg = "Over expected 2 or more inputs but received " + input.size(); throw new ExecException(msg, errCode, PigException.INPUT); } DataBag inbag = null; try { inbag = (DataBag)input.get(0); } catch (ClassCastException cce) { int errCode = 2107; String msg = "Over expected a bag for arg 1 but received " + DataType.findTypeName(input.get(0)); throw new ExecException(msg, errCode, PigException.INPUT); } if (!initialized) { init(input); } else { if (func instanceof ResetableEvalFunc) { ((ResetableEvalFunc)func).reset(); } } OverBag tmpbag = new OverBag(inbag, rowsBefore, rowsAfter); Tuple tmptuple = mTupleFactory.newTuple(1); tmptuple.set(0, tmpbag); DataBag outbag = BagFactory.getInstance().newDefaultBag(); for (int i = 0; i < inbag.size(); i++) { tmpbag.setCurrentRow(i); Tuple t = mTupleFactory.newTuple(1); t.set(0, func.exec(tmptuple)); outbag.add(t); } return outbag; } Over();  Over(String typespec); @Override DataBag exec(Tuple input); @Override Schema outputSchema(Schema inputSch);  }