import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
private void setUpProjFil(boolean withNulls) throws Exception {
    if (withNulls)
        inp = GenRandomData.genRandSmallTupDataBagWithNulls(r, 10, 100);
    else
        inp = GenRandomData.genRandSmallTupDataBag(r, 10, 100);
    t = GenRandomData.genRandSmallBagTuple(r, 10, 100);
    projFil = GenPhyOp.topFilterOpWithProj(1, 50);
    POProject inpPrj = GenPhyOp.exprProject();
    Tuple tmpTpl = new DefaultTuple();
    tmpTpl.append(inp);
    inpPrj.setColumn(0);
    inpPrj.setResultType(DataType.TUPLE);
    inpPrj.setOverloaded(true);
    inpPrj.attachInput(tmpTpl);
    List<PhysicalOperator> inputs = new ArrayList<PhysicalOperator>();
    inputs.add(inpPrj);
    projFil.setInputs(inputs);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
private void setUp2() throws PlanException, ExecException{
    lr = GenPhyOp.topLocalRearrangeOPWithPlanPlain(0,0,db.iterator().next());
    List<PhysicalPlan> plans = lr.getPlans();
    POLocalRearrange lrT = GenPhyOp.topLocalRearrangeOPWithPlanPlain(0, 1, db.iterator().next());
    List<PhysicalPlan> plansT = lrT.getPlans();
    plans.add(plansT.get(0));
    lr.setPlans(plans);
    
    POProject proj = GenPhyOp.exprProject();
    proj.setColumn(0);
    proj.setResultType(DataType.TUPLE);
    proj.setOverloaded(true);
    Tuple t = new DefaultTuple();
    t.append(db);
    proj.attachInput(t);
    List<PhysicalOperator> inputs = new ArrayList<PhysicalOperator>();
    inputs.add(proj);
    lr.setInputs(inputs);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testStore() throws Exception {
    inpDB = GenRandomData.genRandSmallTupDataBag(new Random(), 10, 100);
    storeAndCopyLocally(inpDB);

    int size = 0;
    BufferedReader br = new BufferedReader(new FileReader(outputFileName));
    for(String line=br.readLine();line!=null;line=br.readLine()){
        String[] flds = line.split("\t",-1);
        Tuple t = new DefaultTuple();
        t.append(flds[0].compareTo("")!=0 ? flds[0] : null);
        t.append(flds[1].compareTo("")!=0 ? Integer.parseInt(flds[1]) : null);

        System.err.println("Simple data: ");
        System.err.println(line);
        System.err.println("t: ");
        System.err.println(t);
        assertTrue(TestHelper.bagContains(inpDB, t));
        ++size;
    }
    assertEquals(size, inpDB.size());
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    Random r = new Random();
    db = GenRandomData.genRandSmallTupDataBagWithNulls(r, 10, 100);
    projDB = TestHelper.projectBag(db, 0);
    fe = GenPhyOp.topForEachOPWithPlan(0, db.iterator().next());
    POProject proj = GenPhyOp.exprProject();
    proj.setColumn(0);
    proj.setResultType(DataType.TUPLE);
    proj.setOverloaded(true);
    Tuple t = new DefaultTuple();
    t.append(db);
    proj.attachInput(t);
    List<PhysicalOperator> inputs = new ArrayList<PhysicalOperator>();
    inputs.add(proj);
    fe.setInputs(inputs);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
private void setUp2() throws PlanException, ExecException{
    lr = GenPhyOp.topLocalRearrangeOPWithPlanPlain(0,0,db.iterator().next());
    List<PhysicalPlan> plans = lr.getPlans();
    POLocalRearrange lrT = GenPhyOp.topLocalRearrangeOPWithPlanPlain(0, 1, db.iterator().next());
    List<PhysicalPlan> plansT = lrT.getPlans();
    plans.add(plansT.get(0));
    lr.setPlans(plans);

    POProject proj = GenPhyOp.exprProject();
    proj.setColumn(0);
    proj.setResultType(DataType.TUPLE);
    proj.setOverloaded(true);
    Tuple t = new DefaultTuple();
    t.append(db);
    proj.attachInput(t);
    List<PhysicalOperator> inputs = new ArrayList<PhysicalOperator>();
    inputs.add(proj);
    lr.setInputs(inputs);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testStore() throws Exception {
    inpDB = GenRandomData.genRandSmallTupDataBag(new Random(), 10, 100);
    storeAndCopyLocally(inpDB);

    int size = 0;
    BufferedReader br = new BufferedReader(new FileReader(outputFileName));
    for(String line=br.readLine();line!=null;line=br.readLine()){
        String[] flds = line.split("\t",-1);
        Tuple t = new DefaultTuple();
        t.append(flds[0].compareTo("")!=0 ? flds[0] : null);
        t.append(flds[1].compareTo("")!=0 ? Integer.parseInt(flds[1]) : null);

        System.err.println("Simple data: ");
        System.err.println(line);
        System.err.println("t: ");
        System.err.println(t);
        assertTrue(TestHelper.bagContains(inpDB, t));
        ++size;
    }
    assertEquals(size, inpDB.size());
    br.close();
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
private void setUpProjFil(boolean withNulls) throws Exception{
    Random r = new Random();
    if(withNulls)
        inp = GenRandomData.genRandSmallTupDataBagWithNulls(r, 10, 100);
    else
        inp = GenRandomData.genRandSmallTupDataBag(r, 10, 100);
    t = GenRandomData.genRandSmallBagTuple(r, 10, 100);
    projFil = GenPhyOp.topFilterOpWithProj(1, 50);
    POProject inpPrj = GenPhyOp.exprProject();
    Tuple tmpTpl = new DefaultTuple();
    tmpTpl.append(inp);
    inpPrj.setColumn(0);
    inpPrj.setResultType(DataType.TUPLE);
    inpPrj.setOverloaded(true);
    inpPrj.attachInput(tmpTpl);
    List<PhysicalOperator> inputs = new ArrayList<PhysicalOperator>();
    inputs.add(inpPrj);
    projFil.setInputs(inputs);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testStore() throws Exception {
    inpDB = GenRandomData.genRandSmallTupDataBag(new Random(), 10, 100);
    storeAndCopyLocally(inpDB);
    
    int size = 0;
    BufferedReader br = new BufferedReader(new FileReader(outputFileName));
    for(String line=br.readLine();line!=null;line=br.readLine()){
        String[] flds = line.split(":",-1);
        Tuple t = new DefaultTuple();
        t.append(flds[0].compareTo("")!=0 ? flds[0] : null);
        t.append(flds[1].compareTo("")!=0 ? Integer.parseInt(flds[1]) : null);
        
        System.err.println("Simple data: ");
        System.err.println(line);
        System.err.println("t: ");
        System.err.println(t);
        assertEquals(true, TestHelper.bagContains(inpDB, t));
        ++size;
    }
    assertEquals(true, size==inpDB.size());
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    Random r = new Random();
    db = GenRandomData.genRandSmallTupDataBagWithNulls(r, 10, 100);
    projDB = TestHelper.projectBag(db,0);
    fe = GenPhyOp.topForEachOPWithPlan(0,db.iterator().next());
    POProject proj = GenPhyOp.exprProject();
    proj.setColumn(0);
    proj.setResultType(DataType.TUPLE);
    proj.setOverloaded(true);
    Tuple t = new DefaultTuple();
    t.append(db);
    proj.attachInput(t);
    List<PhysicalOperator> inputs = new ArrayList<PhysicalOperator>();
    inputs.add(proj);
    fe.setInputs(inputs);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
protected Tuple createTuple(Object... entry) {
  Tuple result = new DefaultTuple();
  for (Object o : entry) {
    result.append(o);
  }
  return result;
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
/**
 * Called by the default implementation of {@link #readArray} to add a
 * value. Overridden to append to pig bag.
 */
@Override
protected void addToArray(Object array, long pos, Object e) {
    if (e instanceof Tuple) {
        ((DataBag) array).add((Tuple) e);
    } else {
        Tuple t = new DefaultTuple();
        t.append(e);
        ((DataBag) array).add(t);
    }
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testFilterCount1() throws IOException, ParserException {
    Data data = resetData(pigServer);
    data.set("foo", tuple(1,1,3), tuple(1,2,3), tuple(2,1,3), tuple(2,1,3));
    String script = "test   = load 'foo' using mock.Storage() as (col1: int, col2: int, col3: int);" +
            "test2 = group test by col1;" +
            "test3 = foreach test2 {" +
            "        filter_one    = filter test by (col2==1);" +
            "        filter_notone = filter test by (col2!=1);" +
            "        generate group as col1, COUNT(filter_one) as cnt_one, COUNT(filter_notone) as cnt_notone;};";
    pigServer.registerQuery(script);
    Iterator<Tuple> it = pigServer.openIterator("test3");
    Tuple[] expected = new DefaultTuple[2];
    expected[0] = (Tuple) Util.getPigConstant("(1,1L,1L)");
    expected[1] = (Tuple) Util.getPigConstant("(2,2L,0L)");
    Object[] results = new Object[2];
    int i = 0;
    while(it.hasNext()) {
        if(i == 2) {
            fail("Got more tuples than expected!");
        }
        Tuple t = it.next();
        if(t.get(0).equals(1)) {
            // this is the first tuple
            results[0] = t;
        } else {
            results[1] = t;
        }
        i++;
    }
    for (int j = 0; j < expected.length; j++) {
        assertTrue(expected[j].equals(results[j]));
    }
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
private Tuple castToDBA(Tuple in) throws ExecException{
    Tuple res = new DefaultTuple();
    for (int i=0;i<in.size();i++) {
        DataByteArray dba = new DataByteArray(in.get(i).toString());
        res.append(dba);
    }
    return res;
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testGetNextNullInput() throws Exception {
    File f1 = Util.createInputFile("tmp", "a.txt", new String[] {"1\t2\t3", "4\t5\t6"});
    File f2 = Util.createInputFile("tmp", "b.txt", new String[] {"7\t8\t9", "1\t200\t300"});
    File f3 = Util.createInputFile("tmp", "c.txt", new String[] {"1\t20\t30"});
    //FileLocalizer.deleteTempFiles();
    pigServer.registerQuery("a = load '" + f1.getAbsolutePath() + "' ;");
    pigServer.registerQuery("b = load '" + f2.getAbsolutePath() + "';");
    pigServer.registerQuery("c = union a, b;");
    pigServer.registerQuery("d = load '" + f3.getAbsolutePath() + "' ;");
    pigServer.registerQuery("e = cogroup c by $0 inner, d by $0 inner;");
    pigServer.explain("e", System.err);
    // output should be
    // (1,{(1,2,3),(1,200,300)},{(1,20,30)})
    Tuple expectedResult = new DefaultTuple();
    expectedResult.append(new DataByteArray("1"));
    Tuple[] secondFieldContents = new DefaultTuple[2];
    secondFieldContents[0] = Util.createTuple(Util.toDataByteArrays(new String[] {"1", "2", "3"}));
    secondFieldContents[1] = Util.createTuple(Util.toDataByteArrays(new String[] {"1", "200", "300"}));
    DataBag secondField = Util.createBag(secondFieldContents);
    expectedResult.append(secondField);
    DataBag thirdField = Util.createBag(new Tuple[]{Util.createTuple(Util.toDataByteArrays(new String[]{"1", "20", "30"}))});
    expectedResult.append(thirdField);
    Iterator<Tuple> it = pigServer.openIterator("e");
    assertEquals(expectedResult, it.next());
    assertFalse(it.hasNext());
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testSingleTupleBagAcess() throws Exception {
    Tuple inputTuple = new DefaultTuple();
    inputTuple.append("a");
    inputTuple.append("b");

    SingleTupleBag bg = new SingleTupleBag(inputTuple);
    Iterator<Tuple> it = bg.iterator();
    assertEquals(inputTuple, it.next());
    assertFalse(it.hasNext());
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
public static DataBag projectBag(DataBag db2, int i) throws ExecException {
    DataBag ret = DefaultBagFactory.getInstance().newDefaultBag();
    for (Tuple tuple : db2) {
        Object o = tuple.get(i);
        Tuple t1 = new DefaultTuple();
        t1.append(o);
        ret.add(t1);
    }
    return ret;
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Override
public Tuple getNext() throws IOException {
    String line = is.readLine();
    if(line == null)
        return null;
    String[] members = line.split("\t");
    DefaultTuple tuple = new DefaultTuple();
    for(String member : members)
        tuple.append(member);
    return tuple;
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
private void setUp1() throws PlanException, ExecException{
    lr = GenPhyOp.topLocalRearrangeOPWithPlanPlain(0,0,db.iterator().next());
    POProject proj = GenPhyOp.exprProject();
    proj.setColumn(0);
    proj.setResultType(DataType.TUPLE);
    proj.setOverloaded(true);
    Tuple t = new DefaultTuple();
    t.append(db);
    proj.attachInput(t);
    List<PhysicalOperator> inputs = new ArrayList<PhysicalOperator>();
    inputs.add(proj);
    lr.setInputs(inputs);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testGetNextTuple1() throws ExecException, PlanException {
    setUp1();
    int size=0;
    for(Result res=lr.getNextTuple();res.returnStatus!=POStatus.STATUS_EOP;res=lr.getNextTuple()){
        Tuple t = (Tuple)res.result;
        String key = (String)t.get(1);
        Tuple val = (Tuple)t.get(2);
        // The input data has 2 columns of which the first
        // is the key
        // With the optimized LocalRearrange, the part
        // of the "value" present in the "key" is 
        // excluded from the "value". So to reconstruct
        // the true "value", create a tuple with "key" in
        // first position and the "value" (val) we currently
        // have in the second position
        assertEquals(1, val.size());
        
        Tuple actualVal = new DefaultTuple();
        actualVal.append(key);
        actualVal.append(val.get(0));
        //Check if the index is same as input index
        assertEquals((byte)0, (byte)(Byte)t.get(0));
        
        //Check if the input bag contains the value tuple
        assertTrue(TestHelper.bagContains(db, actualVal));
        
        //Check if the input key and the output key are same
        String inpKey = (String)actualVal.get(0);
        assertEquals(0, inpKey.compareTo((String)t.get(1)));
        ++size;
    }
    
    //check if all the tuples in the input are generated
    assertEquals(db.size(), size);
}
 

import org.apache.pig.data.DefaultTuple; //导入依赖的package包/类
@Test
public void testStoreComplexData() throws Exception {
    inpDB = GenRandomData.genRandFullTupTextDataBag(new Random(), 10, 100);
    storeAndCopyLocally(inpDB);
    PigStorage ps = new PigStorage("\t");
    int size = 0;
    BufferedReader br = new BufferedReader(new FileReader(outputFileName));
    for(String line=br.readLine();line!=null;line=br.readLine()){
        String[] flds = line.split("\t",-1);
        Tuple t = new DefaultTuple();

        ResourceFieldSchema bagfs = GenRandomData.getSmallTupDataBagFieldSchema();
        ResourceFieldSchema tuplefs = GenRandomData.getSmallTupleFieldSchema();

        t.append(flds[0].compareTo("")!=0 ? ps.getLoadCaster().bytesToBag(flds[0].getBytes(), bagfs) : null);
        t.append(flds[1].compareTo("")!=0 ? new DataByteArray(flds[1].getBytes()) : null);
        t.append(flds[2].compareTo("")!=0 ? ps.getLoadCaster().bytesToCharArray(flds[2].getBytes()) : null);
        t.append(flds[3].compareTo("")!=0 ? ps.getLoadCaster().bytesToDouble(flds[3].getBytes()) : null);
        t.append(flds[4].compareTo("")!=0 ? ps.getLoadCaster().bytesToFloat(flds[4].getBytes()) : null);
        t.append(flds[5].compareTo("")!=0 ? ps.getLoadCaster().bytesToInteger(flds[5].getBytes()) : null);
        t.append(flds[6].compareTo("")!=0 ? ps.getLoadCaster().bytesToLong(flds[6].getBytes()) : null);
        t.append(flds[7].compareTo("")!=0 ? ps.getLoadCaster().bytesToMap(flds[7].getBytes()) : null);
        t.append(flds[8].compareTo("")!=0 ? ps.getLoadCaster().bytesToTuple(flds[8].getBytes(), tuplefs) : null);
        t.append(flds[9].compareTo("")!=0 ? ps.getLoadCaster().bytesToBoolean(flds[9].getBytes()) : null);
        t.append(flds[10].compareTo("")!=0 ? ps.getLoadCaster().bytesToDateTime(flds[10].getBytes()) : null);
        assertEquals(true, TestHelper.bagContains(inpDB, t));
        ++size;
    }
    assertEquals(true, size==inpDB.size());
}