Method testBlockBoundary

test/org/apache/pig/test/TestPigStorage.java:113–177

()

    }

    @Test
    public void testBlockBoundary() throws ExecException {

        MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
        Properties properties = cluster.getProperties();

        // This tests PigStorage loader with records exactly
        // on the boundary of the file blocks.
        Properties props = new Properties();
        for (Entry<Object, Object> entry : properties.entrySet()) {
            props.put(entry.getKey(), entry.getValue());
        }
        props.setProperty(MRConfiguration.MAX_SPLIT_SIZE, "20");
        Util.resetStateForExecModeSwitch();
        PigServer pigServer = new PigServer(cluster.getExecType(), props);
        String[] inputs = {
                "abcdefgh1", "abcdefgh2", "abcdefgh3",
                "abcdefgh4", "abcdefgh5", "abcdefgh6",
                "abcdefgh7", "abcdefgh8", "abcdefgh9"
        };

        String[] expected = {
                "(abcdefgh1)", "(abcdefgh2)", "(abcdefgh3)",
                "(abcdefgh4)", "(abcdefgh5)", "(abcdefgh6)",
                "(abcdefgh7)", "(abcdefgh8)", "(abcdefgh9)"
        };

        System.setProperty("pig.overrideBlockSize", "20");

        String INPUT_FILE = "tmp.txt";

        try {

            PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE));
            for (String s : inputs) {
                w.println(s);
            }
            w.close();

            Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);

            pigServer.registerQuery("a = load '" + INPUT_FILE + "';");

            Iterator<Tuple> iter = pigServer.openIterator("a");
            int counter = 0;
            while (iter.hasNext()){
                assertEquals(expected[counter++].toString(), iter.next().toString());
            }

            assertEquals(expected.length, counter);

        } catch (Exception e) {
            e.printStackTrace();
            Assert.fail();
        } finally {
            new File(INPUT_FILE).delete();
            try {
                Util.deleteFile(cluster, INPUT_FILE);