/*
 * 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.impl.io;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.pig.Expression;
import org.apache.pig.FileInputLoadFunc;
import org.apache.pig.LoadFunc;
import org.apache.pig.LoadMetadata;
import org.apache.pig.ResourceSchema;
import org.apache.pig.ResourceStatistics;
import org.apache.pig.StoreFunc;
import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.BinSedesTuple;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.util.Utils;

/**
 * Store tuples (BinSedesTuples, specifically) using sequence files to leverage
 * sequence file's compression features. Replacement for previous use of {@code TFileStorage}, which
 * had some edge cases that were not properly handled there.
 */
@InterfaceAudience.Private
public class SequenceFileInterStorage extends FileInputLoadFunc implements
StoreFuncInterface, LoadMetadata {
    private static final Log mLog = LogFactory.getLog(SequenceFileInterStorage.class);
    public static final String useLog = "SequenceFile storage in use";

    final private NullWritable KEY0 = NullWritable.get();

    RecordReader<NullWritable, BinSedesTuple> recReader;
    RecordWriter<NullWritable, BinSedesTuple> recWriter;

    /**
     * Simple binary nested reader format
     */
    public SequenceFileInterStorage() throws IOException {
        mLog.debug(useLog);
    }

    @Override
    public Tuple getNext() throws IOException {
        try {
            if (recReader.nextKeyValue()) {
                return recReader.getCurrentValue();
            }
            else {
                return null;
            }
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
    }

    @Override
    public void putNext(Tuple t) throws IOException {
        try {
            recWriter.write(KEY0, (BinSedesTuple) t);
        }
        catch (InterruptedException e) {
            throw new IOException(e);
        }
    }

    @SuppressWarnings("rawtypes")
    @Override
    public InputFormat getInputFormat() {
        return new SequenceFileInputFormat<Object, Tuple>();
    }

    @Override
    public void setLocation(String location, Job job) throws IOException {
        FileInputFormat.setInputPaths(job, location);
    }

    @Override
    public ResourceSchema getSchema(String location, Job job)
                    throws IOException {
        return Utils.getSchema(this, location, true, job);
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public void prepareToRead(RecordReader reader, PigSplit split)
            throws IOException {
        recReader = reader;
    }

    @Override
    public ResourceStatistics getStatistics(String location, Job job)
            throws IOException {
        return null;
    }

    @Override
    public String[] getPartitionKeys(String location, Job job)
            throws IOException {
        return null;
    }

    @Override
    public void setPartitionFilter(Expression partitionFilter)
            throws IOException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String relToAbsPathForStoreLocation(String location, Path curDir)
                    throws IOException {
        return LoadFunc.getAbsolutePath(location, curDir);
    }

    @SuppressWarnings("rawtypes")
    @Override
    public OutputFormat getOutputFormat() throws IOException {
        return new SequenceFileTupleOutputFormat();
    }

    public static class SequenceFileTupleOutputFormat extends SequenceFileOutputFormat<NullWritable, BinSedesTuple> {
        protected SequenceFile.Writer getSequenceWriter(TaskAttemptContext context)
                throws IOException {
              Configuration conf = context.getConfiguration();

              CompressionCodec codec = null;
              CompressionType compressionType = CompressionType.NONE;
              if (getCompressOutput(context)) {
                // find the kind of compression to do
                compressionType = getOutputCompressionType(context);
                // find the right codec
                Class<?> codecClass = getOutputCompressorClass(context,
                                                               DefaultCodec.class);
                codec = (CompressionCodec)
                  ReflectionUtils.newInstance(codecClass, conf);
              }
              // get the path of the temporary output file
              Path file = getDefaultWorkFile(context, "");
              FileSystem fs = file.getFileSystem(conf);
              return SequenceFile.createWriter(fs, conf, file,
                       NullWritable.class,
                       BinSedesTuple.class,
                       compressionType,
                       codec,
                       context);
            }

            @Override
            public RecordWriter<NullWritable, BinSedesTuple>
                   getRecordWriter(TaskAttemptContext context
                                   ) throws IOException, InterruptedException {
              final SequenceFile.Writer out = getSequenceWriter(context);

              return new RecordWriter<NullWritable, BinSedesTuple>() {

                  @Override
                public void write(NullWritable key, BinSedesTuple value)
                    throws IOException {

                    out.append(key, value);
                  }

                  @Override
                public void close(TaskAttemptContext context) throws IOException {
                    out.close();
                  }
                };
            }
    }

    @Override
    public void setStoreLocation(String location, Job job) throws IOException {
        Configuration conf = job.getConfiguration();
        Utils.setMapredCompressionCodecProps(conf);
        FileOutputFormat.setOutputPath(job, new Path(location));
    }

    @Override
    public void checkSchema(ResourceSchema s) throws IOException {

    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public void prepareToWrite(RecordWriter writer) throws IOException {
        recWriter = writer;
    }

    @Override
    public void setStoreFuncUDFContextSignature(String signature) {
    }

    @Override
    public void cleanupOnFailure(String location, Job job) throws IOException {
        StoreFunc.cleanupOnFailureImpl(location, job);

    }

    @Override
    public void cleanupOnSuccess(String location, Job job) throws IOException {
        // DEFAULT : do nothing
    }


}