pigServer.registerQuery("A = load 'table_testCrossAfterGroupAll' as (a0:int, a1:chararray);");
pigServer.registerQuery("B = group A all;");
pigServer.registerQuery("C = foreach B generate COUNT(A);");
pigServer.registerQuery("D = cross A, C;");
Path output = FileLocalizer.getTemporaryPath(pigServer.getPigContext());
ExecJob job = pigServer.store("D", output.toString());
FileSystem fs = output.getFileSystem(cluster.getConfiguration());
FileStatus[] partFiles = fs.listStatus(output, new PathFilter() {
@Override
public boolean accept(Path path) {
if (path.getName().startsWith("part")) {
return true;
}
return false;
}
});
// auto-parallelism is 2 in MR, 20 in Tez, so check >=2
Assert.assertTrue(partFiles.length >= 2);
// Check the output
Iterator<Tuple> iter = job.getResults();
List<Tuple> results = new ArrayList<Tuple>();
while (iter.hasNext()) {
results.add(iter.next());
}
Collections.sort(results);