/* 
 * 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.
 */

using System.Text;
using ParquetSharp.Example.Data;
using ParquetSharp.Example.Data.Simple;
using ParquetSharp.External;
using ParquetSharp.Hadoop;
using ParquetSharp.Hadoop.Example;
using ParquetSharp.Hadoop.Metadata;
using ParquetSharp.IO.Api;
using ParquetSharp.Schema;
using Xunit;

namespace ParquetHadoopTests.Hadoop
{
    public class TestInputOutputFormatWithPadding
    {
        public const string FILE_CONTENT = "" +
            "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ," +
            "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ," +
            "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";

        public static MessageType PARQUET_TYPE = Types.buildMessage()
            .required(BINARY).@as(UTF8).named("uuid")
          .required(BINARY).@as(UTF8).named("char")
          .named("FormatTestObject");

        private static readonly Charset UTF_8 = Charset.forName("UTF-8");

        /**
         * ParquetInputFormat that will not split the input file (easier validation)
         */
        private class NoSplits : ParquetInputFormat
        {
            protected override bool isSplitable(JobContext context, Path filename)
            {
                return false;
            }
        }

        public class Writer : Mapper<LongWritable, Text, Void, Group>
        {
            public static readonly SimpleGroupFactory GROUP_FACTORY = new SimpleGroupFactory(PARQUET_TYPE);

            protected override void map(LongWritable key, Text value, Context context)
            {
                // writes each character of the line with a UUID
                string line = value.ToString();
                for (int i = 0; i < line.length(); i += 1)
                {
                    Group group = GROUP_FACTORY.newGroup();
                    group.add(0, Binary.fromString(UUID.randomUUID().ToString()));
                    group.add(1, Binary.fromString(line.substring(i, i + 1)));
                    context.write(null, group);
                }
            }
        }

        public class Reader : Mapper<Void, Group, LongWritable, Text>
        {
            protected override void map(Void key, Group value, Context context)
            {
                context.write(null, new Text(value.getString("char", 0)));
            }
        }

        // @Rule
        public TemporaryFolder temp = new TemporaryFolder();

        [Fact]
        public void testBasicBehaviorWithPadding()
        {
            ParquetFileWriter.BLOCK_FS_SCHEMES.add("file");

            File inputFile = temp.newFile();
            FileOutputStream @out = new FileOutputStream(inputFile);
            @out.write(FILE_CONTENT.getBytes("UTF-8"));
            @out.close();

            File tempFolder = temp.newFolder();
            tempFolder.delete();
            Path tempPath = new Path(tempFolder.toURI());

            File outputFolder = temp.newFile();
            outputFolder.delete();

            Configuration conf = new Configuration();
            // May test against multiple hadoop versions
            conf.set("dfs.block.size", "1024");
            conf.set("dfs.blocksize", "1024");
            conf.set("dfs.blockSize", "1024");
            conf.set("fs.local.block.size", "1024");

            // don't use a cached FS with a different block size
            conf.set("fs.file.impl.disable.cache", "true");

            // disable summary metadata, it isn't needed
            conf.set("parquet.enable.summary-metadata", "false");
            conf.set("parquet.example.schema", PARQUET_TYPE.ToString());

            {
                Job writeJob = new Job(conf, "write");
                writeJob.setInputFormatClass(typeof(TextInputFormat));
                TextInputFormat.addInputPath(writeJob, new Path(inputFile.ToString()));

                writeJob.setOutputFormatClass(typeof(ParquetOutputFormat));
                writeJob.setMapperClass(typeof(Writer));
                writeJob.setNumReduceTasks(0); // write directly to Parquet without reduce
                ParquetOutputFormat.setWriteSupportClass(writeJob, typeof(GroupWriteSupport));
                ParquetOutputFormat.setBlockSize(writeJob, 1024);
                ParquetOutputFormat.setPageSize(writeJob, 512);
                ParquetOutputFormat.setDictionaryPageSize(writeJob, 512);
                ParquetOutputFormat.setEnableDictionary(writeJob, true);
                ParquetOutputFormat.setMaxPaddingSize(writeJob, 1023); // always pad
                ParquetOutputFormat.setOutputPath(writeJob, tempPath);

                waitForJob(writeJob);
            }

            // make sure padding was added
            File parquetFile = getDataFile(tempFolder);
            ParquetMetadata footer = ParquetFileReader.readFooter(conf,
                new Path(parquetFile.ToString()), ParquetMetadataConverter.NO_FILTER);
            foreach (BlockMetaData block in footer.getBlocks())
            {
                Assert.True("Block should start at a multiple of the block size",
                    block.getStartingPos() % 1024 == 0);
            }

            {
                Job readJob = new Job(conf, "read");
                readJob.setInputFormatClass(typeof(NoSplits));
                ParquetInputFormat.setReadSupportClass(readJob, typeof(GroupReadSupport));
                TextInputFormat.addInputPath(readJob, tempPath);

                readJob.setOutputFormatClass(typeof(TextOutputFormat));
                readJob.setMapperClass(typeof(Reader));
                readJob.setNumReduceTasks(0); // write directly to text without reduce
                TextOutputFormat.setOutputPath(readJob, new Path(outputFolder.ToString()));

                waitForJob(readJob);
            }

            File dataFile = getDataFile(outputFolder);
            Assert.NotNull("Should find a data file", dataFile);

            StringBuilder contentBuilder = new StringBuilder();
            foreach (string line in Files.readAllLines(dataFile, UTF_8))
            {
                contentBuilder.Append(line);
            }
            string reconstructed = contentBuilder.ToString();
            Assert.Equal("Should match written file content",
                FILE_CONTENT, reconstructed);

            ParquetFileWriter.BLOCK_FS_SCHEMES.remove("file");
        }

        private void waitForJob(Job job)
        {
            job.submit();
            while (!job.isComplete())
            {
                sleep(100);
            }
            if (!job.isSuccessful())
            {
                throw new RuntimeException("job failed " + job.getJobName());
            }
        }

        private File getDataFile(File location)
        {
            File[] files = location.listFiles();
            File dataFile = null;
            if (files != null)
            {
                foreach (File file in files)
                {
                    if (file.getName().startsWith("part-"))
                    {
                        dataFile = file;
                        break;
                    }
                }
            }
            return dataFile;
        }
    }
