/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.pig.test;

import static org.junit.Assert.*;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Properties;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.pig.PigServer;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.test.utils.CloseAwareFSDataInputStream;
import org.apache.pig.test.utils.CloseAwareOutputStream;
import org.apache.tools.bzip2r.CBZip2InputStream;
import org.apache.tools.bzip2r.CBZip2OutputStream;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;

@RunWith(Parameterized.class)
public class TestBZip {
    private static Properties properties;
    private static MiniGenericCluster cluster;

    @Parameters(name = "pig.bzip.use.hadoop.inputformat = {0}.")
    public static Iterable<Object[]> data() {
        if ( HadoopShims.isHadoopYARN() ) {
            return Arrays.asList(new Object[][] {
                { false  },
                { true   }
            });
        } else {
            return Arrays.asList(new Object[][] {
                { false }
            });
        }
    }

    public TestBZip (Boolean useBzipFromHadoop) {
        properties = cluster.getProperties();
        properties.setProperty("pig.bzip.use.hadoop.inputformat", useBzipFromHadoop.toString());
    }

    @Rule
    public TemporaryFolder folder = new TemporaryFolder();

    @BeforeClass
    public static void oneTimeSetUp() throws Exception {
        cluster = MiniGenericCluster.buildCluster();
        properties = cluster.getProperties();
    }

    @AfterClass
    public static void oneTimeTearDown() throws Exception {
        cluster.shutDown();
    }

   /**
    * Tests the end-to-end writing and reading of a BZip file.
    */
    @Test
    public void testBzipInPig() throws Exception {
        PigServer pig = new PigServer(cluster.getExecType(), properties);

        File in = folder.newFile("junit-in.bz2");

        File out = folder.newFile("junit-out.bz2");
        out.delete();
        String clusterOutput = Util.removeColon(out.getAbsolutePath());

        CBZip2OutputStream cos =
            new CBZip2OutputStream(new FileOutputStream(in));
        for (int i = 1; i < 100; i++) {
            StringBuffer sb = new StringBuffer();
            sb.append(i).append("\n").append(-i).append("\n");
            byte bytes[] = sb.toString().getBytes();
            cos.write(bytes);
        }
        cos.close();

        pig.registerQuery("AA = load '"
                + Util.generateURI(in.getAbsolutePath(), pig.getPigContext())
                + "';");
        pig.registerQuery("A = foreach (group (filter AA by $0 > 0) all) generate flatten($1);");
        pig.registerQuery("store A into '" + Util.encodeEscape(clusterOutput) + "';");
        FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(
                pig.getPigContext().getProperties()));
        FileStatus[] outputFiles = fs.listStatus(new Path(clusterOutput),
                Util.getSuccessMarkerPathFilter());
        FSDataInputStream is = fs.open(outputFiles[0].getPath());
        CBZip2InputStream cis = new CBZip2InputStream(is, -1, out.length());

        // Just a sanity check, to make sure it was a bzip file; we
        // will do the value verification later
        assertEquals(100, cis.read(new byte[100]));
        cis.close();

        pig.registerQuery("B = load '" + Util.encodeEscape(clusterOutput) + "';");

        Iterator<Tuple> i = pig.openIterator("B");
        HashMap<Integer, Integer> map = new HashMap<Integer, Integer>();
        while (i.hasNext()) {
            Integer val = DataType.toInteger(i.next().get(0));
            map.put(val, val);
        }

        assertEquals(new Integer(99), new Integer(map.keySet().size()));

        for (int j = 1; j < 100; j++) {
            assertEquals(new Integer(j), map.get(j));
        }
    }

   /**
    * Tests the end-to-end writing and reading of a BZip file using absolute path with a trailing /.
    */
    @Test
    public void testBzipInPig2() throws Exception {
        PigServer pig = new PigServer(cluster.getExecType(), properties);

        File in = folder.newFile("junit-in.bz2");

        File out = folder.newFile("junit-out.bz2");
        out.delete();
        String clusterOutput = Util.removeColon(out.getAbsolutePath());

        CBZip2OutputStream cos =
            new CBZip2OutputStream(new FileOutputStream(in));
        for (int i = 1; i < 100; i++) {
            StringBuffer sb = new StringBuffer();
            sb.append(i).append("\n").append(-i).append("\n");
            byte bytes[] = sb.toString().getBytes();
            cos.write(bytes);
        }
        cos.close();

        pig.registerQuery("AA = load '"
                + Util.generateURI(in.getAbsolutePath(), pig.getPigContext())
                + "';");
        pig.registerQuery("A = foreach (group (filter AA by $0 > 0) all) generate flatten($1);");
        pig.registerQuery("store A into '" + Util.encodeEscape(clusterOutput) + "/';");
        FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(
                pig.getPigContext().getProperties()));
        FileStatus[] outputFiles = fs.listStatus(new Path(clusterOutput),
                Util.getSuccessMarkerPathFilter());
        FSDataInputStream is = fs.open(outputFiles[0].getPath());
        CBZip2InputStream cis = new CBZip2InputStream(is, -1, out.length());

        // Just a sanity check, to make sure it was a bzip file; we
        // will do the value verification later
        assertEquals(100, cis.read(new byte[100]));
        cis.close();

        pig.registerQuery("B = load '" + Util.encodeEscape(clusterOutput) + "';");

        Iterator<Tuple> i = pig.openIterator("B");
        HashMap<Integer, Integer> map = new HashMap<Integer, Integer>();
        while (i.hasNext()) {
            Integer val = DataType.toInteger(i.next().get(0));
            map.put(val, val);
        }

        assertEquals(new Integer(99), new Integer(map.keySet().size()));

        for (int j = 1; j < 100; j++) {
            assertEquals(new Integer(j), map.get(j));
        }
    }

    //see PIG-2391
    @Test
    public void testBz2() throws Exception {
        String[] inputData = new String[] {
                "1\t2\r3\t4", // '\r' case - this will be split into two tuples
                "5\t6\r", // '\r\n' case
                "7\t8", // '\n' case
                "9\t10\r" // '\r\n' at the end of file
        };

        // bzip compressed input
        File in = folder.newFile("junit-in.bz2");
        String compressedInputFileName = in.getAbsolutePath();
        String clusterCompressedFilePath = Util.removeColon(compressedInputFileName);

        try {
            CBZip2OutputStream cos =
                new CBZip2OutputStream(new FileOutputStream(in));
            for (int i = 0; i < inputData.length; i++) {
                StringBuffer sb = new StringBuffer();
                sb.append(inputData[i]).append("\n");
                byte bytes[] = sb.toString().getBytes();
                cos.write(bytes);
            }
            cos.close();

            Util.copyFromLocalToCluster(cluster, compressedInputFileName,
                    clusterCompressedFilePath);

            // pig script to read compressed input
            PigServer pig = new PigServer(cluster.getExecType(), properties);

            // pig script to read compressed input
            String script ="a = load '" + Util.encodeEscape(clusterCompressedFilePath) +"';";
            pig.registerQuery(script);

            pig.registerQuery("store a into 'intermediate.bz';");
            pig.registerQuery("b = load 'intermediate.bz';");
            Iterator<Tuple> it2 = pig.openIterator("b");
            while (it2.hasNext()) {
                it2.next();
            }
        } finally {
            Util.deleteFile(cluster, "intermediate.bz");
            Util.deleteFile(cluster, "final.bz");
        }
    }
    /**
     * Tests that '\n', '\r' and '\r\n' are treated as record delims when using
     * bzip just like they are when using uncompressed text
     */
    @Test
    public void testRecordDelims() throws Exception {
        String[] inputData = new String[] {
                "1\t2\r3\t4", // '\r' case - this will be split into two tuples
                "5\t6\r", // '\r\n' case
                "7\t8", // '\n' case
                "9\t10\r" // '\r\n' at the end of file
        };

        // bzip compressed input
        File in = folder.newFile("junit-in.bz2");
        String compressedInputFileName = in.getAbsolutePath();
        String clusterCompressedFilePath = Util.removeColon(compressedInputFileName);

        String unCompressedInputFileName = "testRecordDelims-uncomp.txt";
        Util.createInputFile(cluster, unCompressedInputFileName, inputData);

        try {
            CBZip2OutputStream cos =
                new CBZip2OutputStream(new FileOutputStream(in));
            for (int i = 0; i < inputData.length; i++) {
                StringBuffer sb = new StringBuffer();
                sb.append(inputData[i]).append("\n");
                byte bytes[] = sb.toString().getBytes();
                cos.write(bytes);
            }
            cos.close();

            Util.copyFromLocalToCluster(cluster, compressedInputFileName,
                    clusterCompressedFilePath);

            // pig script to read uncompressed input
            String script = "a = load '" + unCompressedInputFileName +"';";
            PigServer pig = new PigServer(cluster.getExecType(), properties);
            pig.registerQuery(script);
            Iterator<Tuple> it1 = pig.openIterator("a");

            // pig script to read compressed input
            script = "a = load '" + Util.encodeEscape(clusterCompressedFilePath) +"';";
            pig.registerQuery(script);
            Iterator<Tuple> it2 = pig.openIterator("a");

            while(it1.hasNext()) {
                Tuple t1 = it1.next();
                Tuple t2 = it2.next();
                assertEquals(t1, t2);
            }

            assertFalse(it2.hasNext());

        } finally {
            Util.deleteFile(cluster, unCompressedInputFileName);
            Util.deleteFile(cluster, clusterCompressedFilePath);
        }

    }

    /**
     * Tests the end-to-end writing and reading of an empty BZip file.
     */
     @Test
     public void testEmptyBzipInPig() throws Exception {
        PigServer pig = new PigServer(cluster.getExecType(), properties);

        File in = folder.newFile("junit-in.tmp");

        File out = folder.newFile("junit-out.bz2");
        out.delete();
        String clusterOutputFilePath = Util.removeColon(out.getAbsolutePath());

        FileOutputStream fos = new FileOutputStream(in);
        fos.write("55\n".getBytes());
        fos.close();
        System.out.println(in.getAbsolutePath());

        pig.registerQuery("AA = load '"
                + Util.generateURI(in.getAbsolutePath(), pig.getPigContext())
                + "';");
        pig.registerQuery("A=foreach (group (filter AA by $0 < '0') all) generate flatten($1);");
        pig.registerQuery("store A into '" + Util.encodeEscape(clusterOutputFilePath) + "';");
        FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(
                pig.getPigContext().getProperties()));
        FileStatus[] outputFiles = fs.listStatus(new Path(clusterOutputFilePath),
                Util.getSuccessMarkerPathFilter());
        FSDataInputStream is = fs.open(outputFiles[0].getPath());
        CBZip2InputStream cis = new CBZip2InputStream(is, -1, out.length());

        // Just a sanity check, to make sure it was a bzip file; we
        // will do the value verification later
        assertEquals(-1, cis.read(new byte[100]));
        cis.close();

        pig.registerQuery("B = load '" + Util.encodeEscape(clusterOutputFilePath) + "';");
        pig.openIterator("B");
    }

    /**
     * Tests the writing and reading of an empty BZip file.
     */
    @Test
    public void testEmptyBzip() throws Exception {
        File tmp = folder.newFile("junit.tmp");
        CBZip2OutputStream cos = new CBZip2OutputStream(new FileOutputStream(
                tmp));
        cos.close();
        assertNotSame(0, tmp.length());
        FileSystem fs = FileSystem.getLocal(new Configuration(false));
        CBZip2InputStream cis = new CBZip2InputStream(
                fs.open(new Path(tmp.getAbsolutePath())), -1, tmp.length());
        assertEquals(-1, cis.read(new byte[100]));
        cis.close();
    }

    @Test
    public void testInnerStreamGetsClosed() throws Exception {
        File tmp = folder.newFile("junit.tmp");

        CloseAwareOutputStream out = new CloseAwareOutputStream(new FileOutputStream(tmp));
        CBZip2OutputStream cos = new CBZip2OutputStream(out);
        assertFalse(out.isClosed());
        cos.close();
        assertTrue(out.isClosed());

        FileSystem fs = FileSystem.getLocal(new Configuration(false));
        Path path = new Path(tmp.getAbsolutePath());
        CloseAwareFSDataInputStream in = new CloseAwareFSDataInputStream(fs.open(path));
        CBZip2InputStream cis = new CBZip2InputStream(in, -1, tmp.length());
        assertFalse(in.isClosed());
        cis.close();
        assertTrue(in.isClosed());
    }

    /**
     * Tests the case where a bzip block ends exactly at the end of the {@link InputSplit}
     * with the block header ending a few bits into the last byte of current
     * InputSplit. This case results in dropped records in Pig 0.6 release
     * This test also tests that bzip files couple of dirs deep can be read by
     * specifying the top level dir.
     */
    @Test
    public void testBlockHeaderEndingAtSplitNotByteAligned() throws IOException {
        // the actual input file is at
        // test/org/apache/pig/test/data/bzipdir1.bz2/bzipdir2.bz2/recordLossblockHeaderEndsAt136500.txt.bz2
        // In this test we will load test/org/apache/pig/test/data/bzipdir1.bz2 to also
        // test that the BZip2TextInputFormat can read subdirs recursively
        String inputFileName =
            "test/org/apache/pig/test/data/bzipdir1.bz2";
        Long expectedCount = 74999L; // number of lines in above file
        // the first block in the above file exactly ends a few bits into the
        // byte at position 136500
        int splitSize = 136500;
        try {
            Util.copyFromLocalToCluster(cluster, inputFileName, inputFileName);
            testCount(inputFileName, expectedCount, splitSize, "PigStorage()");
            testCount(inputFileName, expectedCount, splitSize, "TextLoader()");
        } finally {
            Util.deleteFile(cluster, inputFileName);
        }
    }

    /**
     *  Tests the case where a bzip block ends exactly at the end of the input
     *  split (byte aligned with the last byte) and the last byte is a carriage
     *  return.
     */
    @Test
    public void testBlockHeaderEndingWithCR() throws IOException {
        String inputFileName =
            "test/org/apache/pig/test/data/blockEndingInCR.txt.bz2";
        // number of lines in above file (the value is 1 more than bzcat | wc -l
        // since there is a '\r' which is also treated as a record delim
        Long expectedCount = 82094L;
        // the first block in the above file exactly ends at the byte at
        // position 136498 and the last byte is a carriage return ('\r')
        try {
            int splitSize = 136498;
            Util.copyFromLocalToCluster(cluster, inputFileName, inputFileName);
            testCount(inputFileName, expectedCount, splitSize, "PigStorage()");
        } finally {
            Util.deleteFile(cluster, inputFileName);
        }
    }

    /**
     * Tests the case where a bzip block ends exactly at the end of the input
     * split and has more data which results in overcounting (record duplication)
     * in Pig 0.6
     *
     */
    @Test
    public void testBlockHeaderEndingAtSplitOverCounting() throws IOException {

        String inputFileName =
            "test/org/apache/pig/test/data/blockHeaderEndsAt136500.txt.bz2";
        Long expectedCount = 1041046L; // number of lines in above file
        // the first block in the above file exactly ends a few bits into the
        // byte at position 136500
        int splitSize = 136500;
        try {
            Util.copyFromLocalToCluster(cluster, inputFileName, inputFileName);
            testCount(inputFileName, expectedCount, splitSize, "PigStorage()");
        } finally {
            Util.deleteFile(cluster, inputFileName);
        }
    }

    private void testCount(String inputFileName, Long expectedCount,
            int splitSize, String loadFuncSpec) throws IOException {
        String outputFile = "/tmp/bz-output";
        // simple load-store script to verify that the bzip input is getting
        // split
        String scriptToTestSplitting = "a = load '" +inputFileName + "' using " +
        loadFuncSpec + "; store a into '" + outputFile + "';";

        String script = "a = load '" + inputFileName + "';" +
                "b = group a all;" +
                "c = foreach b generate COUNT_STAR(a);";
        Properties props = new Properties();
        for (Entry<Object, Object> entry : properties.entrySet()) {
            props.put(entry.getKey(), entry.getValue());
        }
        props.setProperty(MRConfiguration.MAX_SPLIT_SIZE, Integer.toString(splitSize));
        props.setProperty("pig.noSplitCombination", "true");
        PigServer pig = new PigServer(cluster.getExecType(), props);
        FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(props));
        fs.delete(new Path(outputFile), true);
        Util.registerMultiLineQuery(pig, scriptToTestSplitting);

        // verify that > 1 maps were launched due to splitting of the bzip input
        FileStatus[] files = fs.listStatus(new Path(outputFile));
        int numPartFiles = 0;
        for (FileStatus fileStatus : files) {
            if(fileStatus.getPath().getName().startsWith("part")) {
                numPartFiles++;
            }
        }
        assertEquals(true, numPartFiles > 1);

        // verify record count to verify we read bzip data correctly
        Util.registerMultiLineQuery(pig, script);
        Iterator<Tuple> it = pig.openIterator("c");
        Long result = (Long) it.next().get(0);
        assertEquals(expectedCount, result);

    }

    @Test
    public void testBzipStoreInMultiQuery() throws Exception {
        String[] inputData = new String[] {
                "1\t2\r3\t4"
        };

        try {
            String inputFileName = "input.txt";
            Util.createInputFile(cluster, inputFileName, inputData);

            PigServer pig = new PigServer(cluster.getExecType(), properties);

            pig.setBatchOn();
            pig.registerQuery("a = load '" +  inputFileName + "';");
            pig.registerQuery("store a into 'output.bz2';");
            pig.registerQuery("store a into 'output';");
            pig.executeBatch();

            FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(
                    pig.getPigContext().getProperties()));
            FileStatus[] outputFiles = fs.listStatus(new Path("output"),
                    Util.getSuccessMarkerPathFilter());
            assertTrue(outputFiles[0].getLen() > 0);

            outputFiles = fs.listStatus(new Path("output.bz2"),
                    Util.getSuccessMarkerPathFilter());
            assertTrue(outputFiles[0].getLen() > 0);
        } finally {
            Util.deleteFile(cluster, "input.txt");
            Util.deleteFile(cluster, "output.bz2");
            Util.deleteFile(cluster, "output");
        }
    }

    @Test
    public void testBzipStoreInMultiQuery2() throws Exception {
        String[] inputData = new String[] {
                "1\t2\r3\t4"
        };

        String inputFileName = "input2.txt";
        Util.createInputFile(cluster, inputFileName, inputData);

        try {
            PigServer pig = new PigServer(cluster.getExecType(), properties);
            PigContext pigContext = pig.getPigContext();
            pigContext.getProperties().setProperty( "output.compression.enabled", "true" );
            pigContext.getProperties().setProperty( "output.compression.codec", "org.apache.hadoop.io.compress.BZip2Codec" );

            pig.setBatchOn();
            pig.registerQuery("a = load '" +  inputFileName + "';");
            pig.registerQuery("store a into 'output2.bz2';");
            pig.registerQuery("store a into 'output2';");
            pig.executeBatch();

            FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(
                    pig.getPigContext().getProperties()));
            FileStatus[] outputFiles = fs.listStatus(new Path("output2"),
                    Util.getSuccessMarkerPathFilter());
            assertTrue(outputFiles[0].getLen() > 0);

            outputFiles = fs.listStatus(new Path("output2.bz2"),
                    Util.getSuccessMarkerPathFilter());
            assertTrue(outputFiles[0].getLen() > 0);
        } finally {
            Util.deleteFile(cluster,"input2.txt");
            Util.deleteFile(cluster,"output2.bz2");
            Util.deleteFile(cluster,"output2");
        }
    }

    /**
     * Tests that Pig's Bzip2TextInputFormat throws an IOException when the input files to be loaded are actually
     * a result of concatenating 2 or more bz2 files. It should not silently ignore part
     * of the input data.  When, hadoop's TextInpuFormat is used(PIG-3251), it should
     * successfully read this concatenated bzip file to the end.
     */
    @Test
    public void testBZ2Concatenation() throws Exception {
        String[] inputData1 = new String[] {
                "1\ta",
                "2\taa"
        };
        String[] inputData2 = new String[] {
                "1\tb",
                "2\tbb"
        };
        String[] inputDataMerged = new String[] {
                "1\ta",
                "2\taa",
                "1\tb",
                "2\tbb"
        };

        // bzip compressed input file1
        File in1 = folder.newFile("junit-in1.bz2");
        String compressedInputFileName1 = in1.getAbsolutePath();

        // file2
        File in2 = folder.newFile("junit-in2.bz2");
        String compressedInputFileName2 = in2.getAbsolutePath();

        String unCompressedInputFileName = "testRecordDelims-uncomp.txt";
        Util.createInputFile(cluster, unCompressedInputFileName, inputDataMerged);

        try {
            CBZip2OutputStream cos =
                new CBZip2OutputStream(new FileOutputStream(in1));
            for (int i = 0; i < inputData1.length; i++) {
                StringBuffer sb = new StringBuffer();
                sb.append(inputData1[i]).append("\n");
                byte bytes[] = sb.toString().getBytes();
                cos.write(bytes);
            }
            cos.close();

            CBZip2OutputStream cos2 =
                new CBZip2OutputStream(new FileOutputStream(in2));
            for (int i = 0; i < inputData2.length; i++) {
                StringBuffer sb = new StringBuffer();
                sb.append(inputData2[i]).append("\n");
                byte bytes[] = sb.toString().getBytes();
                cos2.write(bytes);
            }
            cos2.close();

            // cat
            catInto(compressedInputFileName2, compressedInputFileName1);
            Util.copyFromLocalToCluster(cluster, compressedInputFileName1,
                    compressedInputFileName1);

            // pig script to read uncompressed input
            String script = "a = load '" + Util.encodeEscape(unCompressedInputFileName) +"';";
            PigServer pig = new PigServer(cluster.getExecType(), properties);
            pig.registerQuery(script);
            Iterator<Tuple> it1 = pig.openIterator("a");

            // pig script to read compressed concatenated input
            script = "a = load '" + Util.encodeEscape(compressedInputFileName1) +"';";
            pig.registerQuery(script);

            try {
              Iterator<Tuple> it2 = pig.openIterator("a");
              while(it1.hasNext()) {
                  Tuple t1 = it1.next();
                  Tuple t2 = it2.next();
                  assertEquals(t1, t2);
              }

              assertFalse(it2.hasNext());

              // When pig.bzip.use.hadoop.inputformat=true, it should successfully read the concatenated bzip file
              assertEquals("IOException should be thrown when pig's own Bzip2TextInputFormat is used",
                           properties.getProperty("pig.bzip.use.hadoop.inputformat"),
                           "true");

            } catch (IOException e) {
                assertEquals("IOException should only be thrown when pig's own Bzip2TextInputFormat is used",
                             properties.getProperty("pig.bzip.use.hadoop.inputformat"),
                             "false");
            }

        } finally {
            Util.deleteFile(cluster, unCompressedInputFileName);
        }

    }

    /*
     * Concatenate the contents of src file to the contents of dest file
     */
    private void catInto(String src, String dest) throws IOException {
        FileOutputStream out = new FileOutputStream(new File(dest) , true);
        FileInputStream in = new FileInputStream(new File(src));
        byte[] buffer = new byte[4096];
        int bytesread;
        while ((bytesread = in.read(buffer)) != -1) {
            out.write(buffer,0, bytesread);
        }
        in.close();
        out.close();
    }

    // See PIG-1714
    @Test
    public void testBzipStoreInMultiQuery3() throws Exception {
        String[] inputData = new String[] {
                "1\t2\r3\t4"
        };

        String inputFileName = "input3.txt";
        Util.createInputFile(cluster, inputFileName, inputData);

        String inputScript = "set mapred.output.compress true\n" +
                "set mapreduce.output.fileoutputformat.compress true\n" +
                "set mapred.output.compression.codec org.apache.hadoop.io.compress.BZip2Codec\n" +
                "set mapreduce.output.fileoutputformat.compress.codec org.apache.hadoop.io.compress.BZip2Codec\n" +
                "a = load '" + inputFileName + "';\n" +
                "store a into 'output3.bz2';\n" +
                "store a into 'output3';";

        String inputScriptName = "script3.txt";
        PrintWriter pw = new PrintWriter(new FileWriter(inputScriptName));
        pw.println(inputScript);
        pw.close();

        try {
            PigServer pig = new PigServer(cluster.getExecType(), properties);

            FileInputStream fis = new FileInputStream(inputScriptName);
            pig.registerScript(fis);

            FileSystem fs = FileSystem.get(ConfigurationUtil.toConfiguration(
                    pig.getPigContext().getProperties()));
            FileStatus[] outputFiles = fs.listStatus(new Path("output3"),
                    Util.getSuccessMarkerPathFilter());
            assertTrue(outputFiles[0].getLen() > 0);

            outputFiles = fs.listStatus(new Path("output3.bz2"),
                    Util.getSuccessMarkerPathFilter());
            assertTrue(outputFiles[0].getLen() > 0);
        } finally {
            Util.deleteFile(cluster, "input3.txt");
            Util.deleteFile(cluster, "output3.bz2");
            Util.deleteFile(cluster, "output3");
        }
    }

}