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 testMultipleSchema() throws ExecException, Exception {
    String out = "target/out";
    int rows = 1000;
    Properties props = new Properties();
    props.setProperty("parquet.compression", "uncompressed");
    props.setProperty("parquet.page.size", "1000");
    PigServer pigServer = new PigServer(ExecType.LOCAL, props);
    Data data = Storage.resetData(pigServer);
    Collection<Tuple> list1 = new ArrayList<Tuple>();
    for (int i = 0; i < rows; i++) {
      list1.add(tuple("a" + i));
    }
    Collection<Tuple> list2 = new ArrayList<Tuple>();
    for (int i = 0; i < rows; i++) {
      list2.add(tuple("b" + i));
    }
    data.set("a", "a:chararray", list1);
    data.set("b", "b:chararray", list2);
    pigServer.setBatchOn();
    pigServer.registerQuery("A = LOAD 'a' USING mock.Storage();");
    pigServer.registerQuery("B = LOAD 'b' USING mock.Storage();");
    pigServer.deleteFile(out);
    pigServer.registerQuery("Store A into '" + out + "/a' using " + ParquetStorer.class.getName() + "();");
    pigServer.registerQuery("Store B into '" + out + "/b' 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");

    final Schema schema = data.getSchema("out");
    assertEquals(2, schema.size());
    // union could be in either order
    int ai;
    int bi;
    if ("a".equals(schema.getField(0).alias)) {
      ai = 0;
      bi = 1;
      assertEquals("a", schema.getField(0).alias);
      assertEquals("b", schema.getField(1).alias);
    } else {
      ai = 1;
      bi = 0;
      assertEquals("b", schema.getField(0).alias);
      assertEquals("a", schema.getField(1).alias);
    }

    assertEquals(rows * 2, result.size());

    int a = 0;
    int b = 0;
    for (Tuple tuple : result) {
      String fa = (String) tuple.get(ai);
      String fb = (String) tuple.get(bi);
      if (fa != null) {
        assertEquals("a" + a, fa);
        ++a;
      }
      if (fb != null) {
        assertEquals("b" + b, fb);
        ++b;
      }
    }
  }