package com.backtype.hadoop;

import com.backtype.hadoop.FileCopyInputFormat.FileCopyArgs;
import com.backtype.support.Utils;
import org.apache.hadoop.fs.FileChecksum;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.UUID;

public abstract class AbstractFileCopyMapper extends MapReduceBase implements Mapper<Text, Text, NullWritable, NullWritable> {
    public static Logger LOG = LoggerFactory.getLogger(AbstractFileCopyMapper.class);

    private FileSystem fsSource;
    private FileSystem fsDest;
    private String tmpRoot;

    private void setStatus(Reporter rprtr, String msg) {
        LOG.info(msg);
        rprtr.setStatus(msg);
    }

    public void map(Text source, Text target, OutputCollector<NullWritable, NullWritable> oc, Reporter rprtr) throws IOException {
        Path sourceFile = new Path(source.toString());
        Path finalFile = new Path(target.toString());
        // Path tmpFile = new Path(tmpRoot, UUID.randomUUID().toString());

        setStatus(rprtr, "Copying " + sourceFile.toString() + " to " + finalFile.toString());

        if(fsDest.exists(finalFile)) {
            FileChecksum fc1 = fsSource.getFileChecksum(sourceFile);
            FileChecksum fc2 = fsDest.getFileChecksum(finalFile);
            if(fc1 != null && fc2 != null && !fc1.equals(fc2) ||
               fsSource.getContentSummary(sourceFile).getLength()!=fsDest.getContentSummary(finalFile).getLength() ||
               ((fc1==null || fc2==null) && !Utils.firstNBytesSame(fsSource, sourceFile, fsDest, finalFile, 1024*1024))) {
                throw new IOException("Target file already exists and is different! " + finalFile.toString());
            } else {
                return;
            }
        }

        fsDest.mkdirs(finalFile.getParent());

        copyFile(fsSource, sourceFile, fsDest, finalFile, rprtr);

        // setStatus(rprtr, "Renaming " + tmpFile.toString() + " to " + finalFile.toString());

        // fsDest.mkdirs(finalFile.getParent());
        // if(!fsDest.rename(tmpFile, finalFile))
        //    throw new IOException("could not rename " + tmpFile.toString() + " to " + finalFile.toString());

        // this is a bit of a hack; if we don't do this explicit rename, the owner of the file will
        // be hadoop each time.
        //fsDest.setOwner(finalFile, this.owner, fs.getGroup());
    }

    protected abstract void copyFile(FileSystem fsSource, Path source, FileSystem fsDest, Path target, Reporter reporter) throws IOException;

    @Override
    public void configure(JobConf job) {
        FileCopyArgs args = (FileCopyArgs) Utils.getObject(job, FileCopyInputFormat.ARGS);
        try {
            tmpRoot = job.get("hadoop.tmp.dir") != null ? job.get("hadoop.tmp.dir") + Path.SEPARATOR + "filecopy" : args.tmpRoot;
            fsSource = new Path(args.source).getFileSystem(job);
            fsDest = new Path(args.dest).getFileSystem(job);

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

@@ -1127,19 +1127,16 @@
 %0A       
- //
  Path tm
@@ -1197,103 +1197,8 @@
 );%0A%0A
-        setStatus(rprtr, %22Copying %22 + sourceFile.toString() + %22 to %22 + finalFile.toString());%0A%0A
     
@@ -1862,35 +1862,202 @@
         
+if (
 fsDest.
-mkdirs(final
+getUri().getScheme().equalsIgnoreCase(%22hdfs%22)) %7B%0A%0A            setStatus(rprtr, %22Copying %22 + sourceFile.toString() + %22 to %22 + tmpFile.toString());%0A            fsDest.mkdirs(tmp
 File.get
@@ -2063,25 +2063,28 @@
 tParent());%0A
-%0A
+    
         copy
@@ -2114,29 +2114,27 @@
 le, fsDest, 
-final
+tmp
 File, rprtr)
@@ -2144,18 +2144,19 @@
         
-//
+   
  setStat
@@ -2241,18 +2241,19 @@
         
-//
+   
  fsDest.
@@ -2291,18 +2291,19 @@
         
+   
-//
  if(!fsD
@@ -2342,18 +2342,20 @@
         
-//
+    
     thro
@@ -2440,32 +2440,280 @@
 e.toString());%0A%0A
+        %7D else %7B%0A%0A            setStatus(rprtr, %22Copying %22 + sourceFile.toString() + %22 to %22 + finalFile.toString());%0A%0A            fsDest.mkdirs(finalFile.getParent());%0A            copyFile(fsSource, sourceFile, fsDest, finalFile, rprtr);%0A        %7D%0A%0A
         // this 

use tmp file stage if dest fs is hdfs, otherwise skip tmp file stage