Method testInputOutputSpecs

test/org/apache/pig/test/TestStreaming.java:619–682

()

    }

    @Test
    public void testInputOutputSpecs() throws Exception {
        File input = Util.createInputFile("tmp", "",
                                          new String[] {"A,1", "B,2", "C,3",
                                                        "D,2", "A,5", "B,5",
                                                        "C,8", "A,8", "D,8",
                                                        "A,9"});

        // Perl script
        String[] script =
            new String[] {
                          "#!/usr/bin/perl",
                          "open(INFILE,  $ARGV[0]) or die \"Can't open \".$ARGV[0].\"!: $!\";",
                          "open(OUTFILE, \">\", $ARGV[1]) or die \"Can't open \".$ARGV[1].\"!: $!\";",
                          "open(OUTFILE2, \">\", $ARGV[2]) or die \"Can't open \".$ARGV[2].\"!: $!\";",
                          "while (<INFILE>) {",
                          "  chomp $_;",
                          "  print OUTFILE \"$_\n\";",
                          "  print STDERR \"STDERR: $_\n\";",
                          "  print OUTFILE2 \"$_\n\";",
                          "}",
                         };
        File command = Util.createInputFile("script", "pl", script);

        // Expected results
        String[] expectedFirstFields =
            new String[] {"A", "B", "C", "A", "D", "A"};
        Integer[] expectedSecondFields = new Integer[] {5, 5, 8, 8, 8, 9};
        Tuple[] expectedResults =
                setupExpectedResults(Util.toDataByteArrays(expectedFirstFields),
                                     Util.toDataByteArrays(expectedSecondFields));
        // Pig query to run
        pigServer.registerQuery(
                "define CMD `perl " + command.getName() + " foo bar foobar` " +
                "ship ('" + Util.encodeEscape(command.toString()) + "') " +
                "input('foo' using " + PigStreaming.class.getName() + "(',')) " +
                "output('bar', " +
                "'foobar' using " + PigStreaming.class.getName() + "(',')) " +
                "stderr();");
        pigServer.registerQuery("IP = load '"
                + Util.generateURI(input.toString(),
                        pigServer.getPigContext()) + "' using "
                + PigStorage.class.getName() + "(',');");
        pigServer.registerQuery("FILTERED_DATA = filter IP by $1 > 3;");
        pigServer.registerQuery("OP = stream FILTERED_DATA through CMD;");

        String output = "/pig/out";
        pigServer.deleteFile(output);
        pigServer.store("OP", output, PigStorage.class.getName() + "(',')");

        pigServer.registerQuery("A = load '" + output + "/foobar" + "' using PigStorage(',');");
        Iterator<Tuple> iter = pigServer.openIterator("A");

        List<Tuple> outputs = new ArrayList<Tuple>();
        while (iter.hasNext()) {
            outputs.add(iter.next());
        }