package com.splicemachine.hbase.backup;

import com.splicemachine.derby.impl.job.ZkTask;
import com.splicemachine.derby.impl.job.coprocessor.RegionTask;
import com.splicemachine.derby.impl.job.operation.OperationJob;
import com.splicemachine.derby.impl.job.scheduler.SchedulerPriorities;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;

import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.List;
import java.util.concurrent.ExecutionException;

/**
 *
 *
 */
public class RestoreBackupTask extends ZkTask {
    private static final long serialVersionUID = 5l;
    private BackupItem backupItem;

    public RestoreBackupTask() {
    }

    public RestoreBackupTask(BackupItem backupItem, String jobId) {
        super(jobId, OperationJob.operationTaskPriority);
        this.backupItem = backupItem;
    }

    @Override
    protected String getTaskType() {
        return "restoreBackupTask";
    }

    @Override
    public void writeExternal(ObjectOutput out) throws IOException {
        super.writeExternal(out);
        out.writeObject(backupItem); // TODO Needs to be replaced with protobuf
    }

    @Override
    public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
        super.readExternal(in);
        backupItem = (BackupItem) in.readObject(); // TODO Needs to be replaced with protobuf
    }

    @Override
    public boolean invalidateOnClose() {
        return true;
    }

    @Override
    public RegionTask getClone() {
        return new RestoreBackupTask(backupItem, jobId);
    }

    @Override
    public void doExecute() throws ExecutionException, InterruptedException {
        if (LOG.isTraceEnabled())
            SpliceLogUtils.trace(LOG, String.format("executing %S backup on region %s", backupItem.getBackup().isIncrementalBackup() ? "incremental" : "full", region.toString()));
        try {
            BackupItem.RegionInfo regionInfo = getRegionInfo();
            if (regionInfo == null) {
                return;
            }
            List<Pair<byte[], String>> famPaths = regionInfo.getFamPaths();
            region.bulkLoadHFiles(famPaths);
        } catch (IOException e) {
            SpliceLogUtils.error(LOG, "Couldn't recover region " + region, e);
            throw new ExecutionException("Failed recovery of region " + region, e);
        }
    }

    private BackupItem.RegionInfo getRegionInfo() {
        Bytes.ByteArrayComparator comparator = new Bytes.ByteArrayComparator();
        for (BackupItem.RegionInfo ri : backupItem.getRegionInfoList()) {
            if (comparator.compare(ri.getHRegionInfo().getEndKey(), region.getEndKey()) == 0) {
                return ri;
            }
        }
        SpliceLogUtils.warn(LOG, "Couldn't find matching backup data for region " + region);
        return null;
    }

    @Override
    public int getPriority() {
        return SchedulerPriorities.INSTANCE.getBasePriority(RestoreBackupTask.class);
    }
}


@@ -34,16 +34,68 @@
 ackup;%0A%0A
+import com.splicemachine.constants.SpliceConstants;%0A
 import c
@@ -380,16 +380,128 @@
 gUtils;%0A
+import org.apache.hadoop.fs.FileSystem;%0Aimport org.apache.hadoop.fs.FileUtil;%0Aimport org.apache.hadoop.fs.Path;%0A
 import o
@@ -680,20 +680,71 @@
 va.util.
-List
+ArrayList;%0Aimport java.util.List;%0Aimport java.util.UUID
 ;%0Aimport
@@ -2411,24 +2411,101 @@
 FamPaths();%0A
+            List%3CPair%3Cbyte%5B%5D, String%3E%3E copyPaths = copyStoreFiles(famPaths);%0A
             
@@ -2522,27 +2522,28 @@
 kLoadHFiles(
-fam
+copy
 Paths);%0A    
@@ -2748,24 +2748,787 @@
    %7D%0A    %7D%0A%0A
+    private List%3CPair%3Cbyte%5B%5D, String%3E%3E copyStoreFiles(List%3CPair%3Cbyte%5B%5D, String%3E%3E famPaths) throws IOException %7B%0A        List%3CPair%3Cbyte%5B%5D, String%3E%3E copy = new ArrayList%3CPair%3Cbyte%5B%5D, String%3E%3E(famPaths.size());%0A        FileSystem fs = region.getFilesystem();%0A        for (Pair%3Cbyte%5B%5D, String%3E pair : famPaths) %7B%0A            Path srcPath = new Path(pair.getSecond());%0A            FileSystem srcFs = srcPath.getFileSystem(SpliceConstants.config);%0A            Path localDir = region.getRegionDir();%0A            Path tmpPath = getRandomFilename(fs, localDir);%0A            FileUtil.copy(srcFs, srcPath, fs, tmpPath, false, SpliceConstants.config);%0A            copy.add(new Pair%3Cbyte%5B%5D, String%3E(pair.getFirst(), tmpPath.toString()));%0A        %7D%0A        return copy;%0A    %7D%0A%0A
     private 
@@ -4125,12 +4125,237 @@
 );%0A    %7D
+%0A%0A    static Path getRandomFilename(final FileSystem fs,%0A                                  final Path dir)%0A            throws IOException %7B%0A        return new Path(dir, UUID.randomUUID().toString().replaceAll(%22-%22, %22%22));%0A    %7D
 %0A%7D%0A%0A

Copy store files before restore