private Schema getSchema() {
    try {
      final String schemaString = getProperties().getProperty(SCHEMA);
      if (schemaString == null) {
        throw new ParquetEncodingException("Can not store relation in Parquet as the schema is unknown");
      }
      return Utils.getSchemaFromString(schemaString);
    } catch (ParserException e) {
      throw new ParquetEncodingException("can not get schema from context", e);
    }
  }
@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);
    }
  }