/* This file is part of VoltDB.
 * Copyright (C) 2008-2013 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.UnmodifiableIterator;
import com.google.common.util.concurrent.Callables;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.voltcore.utils.DBBPool;
import org.voltcore.utils.Pair;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Callable;

/**
 * A class that manages streaming data from a single table. It knows how to interpret the return codes and check for
 * errors related to table data streaming using the table stream API.
 */
public class TableStreamer {
    private final int m_tableId;
    private final TableStreamType m_type;
    private final ImmutableList<SnapshotTableTask> m_tableTasks;

    public TableStreamer(int tableId, TableStreamType type, List<SnapshotTableTask> tableTasks)
    {
        m_tableId = tableId;
        m_type = type;
        m_tableTasks = ImmutableList.copyOf(tableTasks);
    }

    /**
     * Activate the stream with the given predicates on the given table.
     * @param context       Context
     * @param predicates    Predicates associated with the stream
     */
    public void activate(SystemProcedureExecutionContext context, byte[] predicates)
    {
        activate(context, Long.MAX_VALUE, predicates);
    }

    /**
     * Activate the stream with the given predicates on the given table.
     * @param context       Context
     * @param undoToken     The undo token
     * @param predicates    Predicates associated with the stream
     */
    public void activate(SystemProcedureExecutionContext context, long undoToken, byte[] predicates)
    {
        if (!context.activateTableStream(m_tableId, m_type, undoToken, predicates)) {
            VoltDB.crashLocalVoltDB("Attempted to activate copy on write mode for table "
                                    + m_tableId + " and failed", false, null);
        }
    }

    /**
     * Streams more tuples from the table.
     * @param context          Context
     * @param outputBuffers    Allocated buffers to hold output tuples
     * @return A future for all writes to data targets, and a boolean indicating if there's more left in the table.
     * The future could be null if nothing is serialized.
     */
    public Pair<ListenableFuture, Boolean> streamMore(SystemProcedureExecutionContext context,
                                                      List<DBBPool.BBContainer> outputBuffers)
    {
        ListenableFuture writeFuture = null;

        prepareBuffers(outputBuffers);

        Pair<Long, int[]> serializeResult = context.tableStreamSerializeMore(m_tableId, m_type, outputBuffers);
        for (int serializedBytes : serializeResult.getSecond()) {
            if (serializedBytes < 0) {
                VoltDB.crashLocalVoltDB("Failure while serialize data from a table for COW snapshot", false, null);
            }
        }

        if (serializeResult.getSecond()[0] > 0) {
            writeFuture = writeBlocksToTargets(outputBuffers, serializeResult.getSecond());
        } else {
            // Return all allocated snapshot output buffers
            for (DBBPool.BBContainer container : outputBuffers) {
                container.discard();
            }
        }

        return Pair.of(writeFuture, serializeResult.getFirst() > 0);
    }

    /**
     * Set the positions of the buffers to the start of the content, leaving some room for the headers.
     */
    private void prepareBuffers(List<DBBPool.BBContainer> buffers)
    {
        Preconditions.checkArgument(buffers.size() == m_tableTasks.size());

        UnmodifiableIterator<SnapshotTableTask> iterator = m_tableTasks.iterator();
        for (DBBPool.BBContainer container : buffers) {
            int headerSize = iterator.next().m_target.getHeaderSize();
            container.b.clear();
            container.b.position(headerSize);
        }
    }

    /**
     * Finalize the output buffers and write them to the corresponding data targets
     *
     * @return A future that can used to wait for all targets to finish writing the buffers
     */
    private ListenableFuture<?> writeBlocksToTargets(Collection<DBBPool.BBContainer> outputBuffers,
                                                     int[] serialized)
    {
        Preconditions.checkArgument(m_tableTasks.size() == serialized.length);
        Preconditions.checkArgument(outputBuffers.size() == serialized.length);

        final List<ListenableFuture<?>> writeFutures =
            new ArrayList<ListenableFuture<?>>(outputBuffers.size());

        // The containers, the data targets, and the serialized byte counts should all line up
        Iterator<DBBPool.BBContainer> containerIter = outputBuffers.iterator();
        int serializedIndex = 0;

        for (SnapshotTableTask task : m_tableTasks) {
            final DBBPool.BBContainer container = containerIter.next();

            /*
             * Finalize the buffer by setting position to 0 and limit to the last used byte
             */
            container.b.limit(serialized[serializedIndex++] + task.m_target.getHeaderSize());
            container.b.position(0);

            Callable<DBBPool.BBContainer> valueForTarget = Callables.returning(container);
            if (task.m_filters != null) {
                for (SnapshotDataFilter filter : task.m_filters) {
                    valueForTarget = filter.filter(valueForTarget);
                }
            }

            ListenableFuture<?> writeFuture = task.m_target.write(valueForTarget, m_tableId);
            if (writeFuture != null) {
                writeFutures.add(writeFuture);
            }
        }

        // Wraps all write futures in one future
        return Futures.allAsList(writeFutures);
    }
}

@@ -1501,16 +1501,128 @@
 eamer %7B%0A
+    // Error code returned by EE.tableStreamSerializeMore().%0A    private static byte SERIALIZATION_ERROR = -1;%0A%0A
     priv
@@ -3678,109 +3678,65 @@
     
-for (int serializedBytes : serializeResult.getSecond()) %7B%0A            if (serializedBytes %3C 0) %7B%0A    
+if (serializeResult.getFirst() == SERIALIZATION_ERROR) %7B%0A
     
@@ -3790,17 +3790,19 @@
 serializ
-e
+ing
  data fr
@@ -3808,33 +3808,27 @@
 rom 
-a 
 table 
-for COW snapshot%22
+%22 + m_tableId
 , fa
@@ -3839,30 +3839,16 @@
  null);%0A
-            %7D%0A
         

Add static error code for tableStreamSerializeMore() in TableStreamer.