Method simpleMultiQueryTest2

test/org/apache/pig/test/TestPigRunner.java:377–433

()

    }

    @Test
    public void simpleMultiQueryTest2() throws Exception {
        final String OUTPUT_FILE_2 = "output2";

        PrintWriter w = new PrintWriter(new FileWriter(PIG_FILE));
        w.println("A = load '" + INPUT_FILE + "' as (a0:int, a1:int, a2:int);");
        w.println("B = filter A by a0 >= 4;");
        w.println("C = filter A by a0 < 4;");
        w.println("D = group C by a0;");
        w.println("E = foreach D generate group, COUNT(C);");
        w.println("store B into '" + OUTPUT_FILE_2 + "';");
        w.println("store E into '" + OUTPUT_FILE + "';");
        w.close();

        try {
            String[] args = { "-x", execType, PIG_FILE };
            PigStats stats = PigRunner.run(args, new TestNotificationListener(execType));
            assertTrue(stats.isSuccessful());
            if (execType.equals("spark")) {
                // In spark mode,the number of spark job is calculated by the number of POStore.
                // 2 POStore generates 2 spark jobs.
                assertEquals(stats.getJobGraph().size(), 2);
            } else {
                assertEquals(stats.getJobGraph().size(), 1);
            }

            // Each output file should include the following:
            // output:
            //   5\t3\t4\n
            //   5\t6\t7\n
            // output2:
            //   1\t1\n
            //   3\t2\n
            final int numOfRecords1 = 2;
            final int numOfRecords2 = 2;
            final int numOfCharsPerRecord1 = 6;
            final int numOfCharsPerRecord2 = 4;
            assertEquals(numOfRecords1 + numOfRecords2, stats.getRecordWritten());
            assertEquals((numOfRecords1 * numOfCharsPerRecord1) + (numOfRecords2 * numOfCharsPerRecord2),
                    stats.getBytesWritten());
            assertTrue(stats.getOutputNames().size() == 2);
            for (String fname : stats.getOutputNames()) {
                assertTrue(fname.equals(OUTPUT_FILE) || fname.equals(OUTPUT_FILE_2));
                if (fname.equals(OUTPUT_FILE)) {
                    assertEquals(2, stats.getNumberRecords(fname));
                } else {
                    assertEquals(2, stats.getNumberRecords(fname));
                }
            }
            assertEquals("A,B,C,D,E",
                    ((JobStats)stats.getJobGraph().getSinks().get(0)).getAlias());
        } finally {
            new File(PIG_FILE).delete();
            Util.deleteFile(cluster, OUTPUT_FILE);
            Util.deleteFile(cluster, OUTPUT_FILE_2);
        }
    }