package com.splicemachine.derby.impl.job.operation;

import com.google.common.collect.Lists;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.hadoop.hbase.filter.FilterBase;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.List;

/**
 * @author Scott Fines
 * Created on: 5/24/13
 */
public class SuccessFilter extends FilterBase {
    private static final Logger LOG = Logger.getLogger(SuccessFilter.class);
    @SuppressWarnings("unused")
	private static final long serialVersionUID = 1l;
    private List<byte[]> failedTasks;

    private boolean filterRow;

    public SuccessFilter() {
        super();
    }

    public SuccessFilter(List<byte[]> failedTasks) {
        this.failedTasks = failedTasks;
    }

    @Override
    public void reset() {
        filterRow = false;
    }

    @Override
    public boolean filterRowKey(byte[] buffer, int offset, int length) {
        int postfixOffset = offset+length;
        for(byte[] failedTask:failedTasks){
            int postOffset = postfixOffset-failedTask.length;
            if(Bytes.compareTo(buffer,postOffset,failedTask.length,failedTask,0,failedTask.length)==0){
                SpliceLogUtils.trace(LOG,"Found a row from a failed task, skipping");
                //we have a task which failed
                filterRow=true;
                break;
            }
        }

        return filterRow;
    }

    @Override
    public boolean filterRow() {
        return this.filterRow;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(failedTasks.size());
        for(byte[] failedTask: failedTasks){
            out.writeInt(failedTask.length);
            out.write(failedTask);
        }
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        int size = in.readInt();
        failedTasks = Lists.newArrayListWithExpectedSize(size);
        for(int i=0;i<size;i++){
            byte[] next = new byte[in.readInt()];
            in.readFully(next);
            failedTasks.add(next);
        }
    }
}

@@ -132,16 +132,115 @@
 ogUtils;
+%0A%0Aimport org.apache.hadoop.hbase.KeyValue;%0Aimport org.apache.hadoop.hbase.filter.Filter.ReturnCode;
 %0Aimport 
@@ -357,16 +357,17 @@
 Logger;%0A
+%0A
 import j
@@ -761,16 +761,47 @@
 edTasks;
+%0A    private int postfixOffset;
 %0A%0A    pr
@@ -1044,37 +1044,774 @@
  = false;%0A    %7D%0A
-%0A
     
+%0A    %0A    /**%0A     * %0A     * Used to filter row key.  Focuses on not forcing a reseek.%0A     * %0A     */%0A    @Override%0A%09public ReturnCode filterKeyValue(KeyValue keyValue) %7B%0A    %09postfixOffset = keyValue.getRowOffset()+keyValue.getRowLength();%0A        for(byte%5B%5D failedTask:failedTasks)%7B%0A            int postOffset = postfixOffset-failedTask.length;%0A            if(Bytes.compareTo(keyValue.getBuffer(),postOffset,failedTask.length,failedTask,0,failedTask.length)==0)%7B%0A                SpliceLogUtils.trace(LOG,%22Found a row from a failed task, skipping%22);%0A                //we have a task which failed%0A                return ReturnCode.SKIP;%0A            %7D%0A        %7D%0A        return ReturnCode.INCLUDE;%0A%09%7D%0A%0A    /* DO NOT USE: Forces a reseek.%0A%09
 @Override%0A    pu
@@ -2366,32 +2366,35 @@
 ilterRow;%0A    %7D%0A
+%09*/
 %0A    @Override%0A 

Remove reseeks when scanning from temp