  @Test
  public void testComplexSchema() throws ExecException, Exception {

    String out = "target/out";
    PigServer pigServer = new PigServer(ExecType.LOCAL);
    Data data = Storage.resetData(pigServer);
    Collection<Tuple> list = new ArrayList<Tuple>();
    for (int i = 0; i < 1000; i++) {
      list.add(tuple("a"+i, bag(tuple("o", "b"))));
    }
    for (int i = 10; i < 2000; i++) {
      list.add(tuple("a"+i, bag(tuple("o", "b"), tuple("o", "b"), tuple("o", "b"), tuple("o", "b"))));
    }
    for (int i = 20; i < 3000; i++) {
      list.add(tuple("a"+i, bag(tuple("o", "b"), tuple("o", null), tuple(null, "b"), tuple(null, null))));
    }
    for (int i = 30; i < 4000; i++) {
      list.add(tuple("a"+i, null));
    }
    Collections.shuffle((List<?>)list);
    data.set("in", "a:chararray, b:{t:(c:chararray, d:chararray)}", list );
    pigServer.setBatchOn();
    pigServer.registerQuery("A = LOAD 'in' USING mock.Storage();");
    pigServer.deleteFile(out);
    pigServer.registerQuery("Store A into '"+out+"' using "+ParquetStorer.class.getName()+"();");
    if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) {
      throw new RuntimeException("Job failed", pigServer.executeBatch().get(0).getException());
    }

    {
      pigServer.registerQuery("B = LOAD '"+out+"' USING "+ParquetLoader.class.getName()+"();");
      pigServer.registerQuery("Store B into 'out' using mock.Storage();");
      if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) {
        throw new RuntimeException("Job failed", pigServer.executeBatch().get(0).getException());
      }

      List<Tuple> result = data.get("out");
      assertEquals(list, result);
      final Schema schema = data.getSchema("out");
      assertEquals("{a:chararray, b:{t:(c:chararray, d:chararray)}}".replaceAll(" ", ""), schema.toString().replaceAll(" ", ""));
    }

    {
      pigServer.registerQuery("C = LOAD '"+out+"' USING "+ParquetLoader.class.getName()+"('a:chararray');");
      pigServer.registerQuery("Store C into 'out2' using mock.Storage();");
      if (pigServer.executeBatch().get(0).getStatus() != JOB_STATUS.COMPLETED) {
        throw new RuntimeException("Job failed", pigServer.executeBatch().get(0).getException());
      }

      final Function<Tuple,Object> grabFirstColumn = new Function<Tuple,Object>() {
        @Override
        public Object apply(Tuple input) {
          try {
              return input.get(0);
          } catch (ExecException e) {
            throw new RuntimeException(e);
          }
        }
      };

      List<Tuple> result2 = data.get("out2");
      // Functional programming!!
      Object[] result2int = Collections2.transform(result2, grabFirstColumn).toArray();
      Object[] input2int = Collections2.transform(list, grabFirstColumn).toArray();

      assertArrayEquals(input2int, result2int);
    }
  }