/*
 * Copyright 2012 - 2016 Splice Machine, Inc.
 *
 * Licensed 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 com.splicemachine.derby.stream.output.direct;

import com.splicemachine.db.iapi.error.StandardException;
import com.splicemachine.derby.iapi.sql.execute.SpliceOperation;
import com.splicemachine.derby.impl.sql.execute.operations.TriggerHandler;
import com.splicemachine.derby.stream.iapi.OperationContext;
import com.splicemachine.derby.stream.iapi.TableWriter;
import com.splicemachine.kvpair.KVPair;
import com.splicemachine.metrics.MetricFactory;
import com.splicemachine.metrics.Metrics;
import com.splicemachine.pipeline.Exceptions;
import com.splicemachine.pipeline.PipelineDriver;
import com.splicemachine.pipeline.api.WriteStats;
import com.splicemachine.pipeline.callbuffer.RecordingCallBuffer;
import com.splicemachine.pipeline.client.WriteCoordinator;
import com.splicemachine.pipeline.config.ForwardingWriteConfiguration;
import com.splicemachine.pipeline.config.WriteConfiguration;
import com.splicemachine.pipeline.utils.PipelineUtils;
import com.splicemachine.primitives.Bytes;
import com.splicemachine.si.api.txn.TxnView;

import java.io.IOException;
import java.util.Iterator;

/**
 * @author Scott Fines
 *         Date: 1/13/16
 */
public class DirectPipelineWriter implements TableWriter<KVPair>,AutoCloseable{
    private final long destConglomerate;
    private TxnView txn;
    private final OperationContext opCtx;
    private final boolean skipIndex;

    private RecordingCallBuffer<KVPair> writeBuffer;

    public DirectPipelineWriter(long destConglomerate,TxnView txn,OperationContext opCtx,boolean skipIndex){
        this.destConglomerate=destConglomerate;
        this.txn=txn;
        this.opCtx=opCtx;
        this.skipIndex=skipIndex;
    }

    @Override
    public void open() throws StandardException{
        WriteCoordinator wc = PipelineDriver.driver().writeCoordinator();
        WriteConfiguration writeConfiguration = new DirectWriteConfiguration(wc.defaultWriteConfiguration());
        try{
            this.writeBuffer = wc.writeBuffer(
                    Bytes.toBytes(Long.toString(destConglomerate)),
                    txn,
                    PipelineUtils.noOpFlushHook,
                    writeConfiguration,
                    Metrics.basicMetricFactory());
        }catch(IOException e){
            throw Exceptions.parseException(e);
        }
    }

    @Override
    public void open(TriggerHandler triggerHandler,SpliceOperation dmlWriteOperation) throws StandardException{
        open();
    }

    @Override
    public void write(KVPair row) throws StandardException{
        try{
            writeBuffer.add(row);
        }catch(Exception e){
            throw Exceptions.parseException(e);
        }
    }

    @Override
    public void write(Iterator<KVPair> rows) throws StandardException{
        while(rows.hasNext()){
            write(rows.next());
        }
    }

    @Override
    public void close() throws StandardException{
        try{
            writeBuffer.flushBufferAndWait();
            writeBuffer.close();
            WriteStats ws = writeBuffer.getWriteStats();
            opCtx.recordPipelineWrites(ws.getWrittenCounter());
            opCtx.recordRetry(ws.getRetryCounter());
            opCtx.recordThrownErrorRows(ws.getThrownErrorsRows());
            opCtx.recordRetriedRows(ws.getRetriedRows());
            opCtx.recordPartialRows(ws.getPartialRows());
            opCtx.recordPartialThrownErrorRows(ws.getPartialThrownErrorRows());
            opCtx.recordPartialRetriedRows(ws.getPartialRetriedRows());
            opCtx.recordPartialIgnoredRows(ws.getPartialIgnoredRows());
            opCtx.recordPartialWrite(ws.getPartialWrite());
            opCtx.recordIgnoredRows(ws.getIgnoredRows());
            opCtx.recordCatchThrownRows(ws.getCatchThrownRows());
            opCtx.recordCatchRetriedRows(ws.getCatchRetriedRows());
            opCtx.recordRegionTooBusy(ws.getRegionTooBusy());
        }catch(Exception e){
            throw Exceptions.parseException(e);
        }
    }

    @Override
    public void setTxn(TxnView txn){
        this.txn = txn;
    }

    @Override
    public TxnView getTxn(){
        return txn;
    }

    @Override
    public byte[] getDestinationTable(){
        return Bytes.toBytes(Long.toString(destConglomerate));
    }

    @Override
    public OperationContext getOperationContext() {
        return opCtx;
    }

    public static class DirectWriteConfiguration extends ForwardingWriteConfiguration {
        protected DirectWriteConfiguration(WriteConfiguration delegate) {
            super(delegate);
        }

        @Override
        public MetricFactory getMetricFactory() {
            return Metrics.basicMetricFactory();
        }
    }
}

@@ -3657,24 +3657,61 @@
 iteStats();%0A
+            if (opCtx != null) %7B%0A    
             
@@ -3766,32 +3766,36 @@
 ));%0A            
+ 
+   
 opCtx.recordRetr
@@ -3823,32 +3823,36 @@
 ));%0A            
+   
+ 
 opCtx.recordThro
@@ -3894,32 +3894,36 @@
 ));%0A            
+    
 opCtx.recordRetr
@@ -3956,32 +3956,36 @@
 ));%0A            
+ 
+   
 opCtx.recordPart
@@ -4006,32 +4006,36 @@
 PartialRows());%0A
+    
             opCt
@@ -4102,32 +4102,36 @@
 ));%0A            
+    
 opCtx.recordPart
@@ -4178,32 +4178,36 @@
 ));%0A            
+  
+  
 opCtx.recordPart
@@ -4242,32 +4242,36 @@
 IgnoredRows());%0A
+    
             opCt
@@ -4318,32 +4318,36 @@
 ));%0A            
+    
 opCtx.recordIgno
@@ -4380,32 +4380,36 @@
 ));%0A            
+  
+  
 opCtx.recordCatc
@@ -4438,32 +4438,36 @@
 hThrownRows());%0A
+    
             opCt
@@ -4522,32 +4522,36 @@
 ));%0A            
+    
 opCtx.recordRegi
@@ -4580,24 +4580,38 @@
 TooBusy());%0A
+            %7D%0A
         %7Dcat

Fix mem profile build