<commit_before>/* 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/>.
 */

#include "TupleOutputStream.h"
#include "TupleOutputStreamProcessor.h"
#include "StreamPredicate.h"
#include "TupleSerializer.h"
#include "tabletuple.h"
#include <limits>

namespace voltdb {

/** Default constructor. */
TupleOutputStreamProcessor::TupleOutputStreamProcessor()
    : boost::ptr_vector<TupleOutputStream>()
{
    clearState();
}

/** Constructor with initial size. */
TupleOutputStreamProcessor::TupleOutputStreamProcessor(std::size_t nBuffers)
    : boost::ptr_vector<TupleOutputStream>(nBuffers)
{
    clearState();
}

/** Constructor for a single stream. Convenient for backward compatibility in tests. */
TupleOutputStreamProcessor::TupleOutputStreamProcessor(void *data, std::size_t length)
    : boost::ptr_vector<TupleOutputStream>(1)
{
    clearState();
    add(data, length);
}

/** Private method used by constructors, etc. to clear state. */
void TupleOutputStreamProcessor::clearState()
{
    m_maxTupleLength = 0;
    m_predicates = NULL;
    m_table = NULL;
    m_totalBytesSerialized = 0;
}

/** Convenience method to create and add a new TupleOutputStream. */
TupleOutputStream &TupleOutputStreamProcessor::add(void *data, std::size_t length)
{
    std::auto_ptr<TupleOutputStream> out(new TupleOutputStream(data, length));
    push_back(out);
    return *out;
}

/** Start serializing. */
void TupleOutputStreamProcessor::open(PersistentTable &table,
                                      std::size_t maxTupleLength,
                                      int32_t partitionId,
                                      StreamPredicateList &predicates,
                                      int32_t totalPartitions)
{
    m_table = &table;
    m_maxTupleLength = maxTupleLength;
    // It must be either one predicate per output stream or none at all.
    bool havePredicates = !predicates.empty();
    if (havePredicates && predicates.size() != size()) {
        throwFatalException("serializeMore() expects either no predicates or one per output stream.");
    }
    m_predicates = &predicates;
    m_totalPartitions = totalPartitions;
    for (TupleOutputStreamProcessor::iterator iter = begin(); iter != end(); ++iter) {
        iter->startRows(partitionId);
    }
    m_totalBytesSerialized = 0;
    // Recalculate the serialization throttle threshold based on the number of partitions.
    if (m_totalPartitions > 0) {
        m_bytesSerializedThreshold = m_bytesSerializedThresholdPerPartition * m_totalPartitions;
    }
    else {
        m_bytesSerializedThreshold = m_bytesSerializedThresholdPerPartition;
    }
}

/** Stop serializing. */
void TupleOutputStreamProcessor::close()
{
    for (TupleOutputStreamProcessor::iterator iter = begin(); iter != end(); ++iter) {
        iter->endRows();
    }
    clearState();
}

/**
 * Write a tuple to the output streams.
 * Expects buffer space was already checked.
 * Returns true when the caller should yield to allow other work to proceed.
 */
bool TupleOutputStreamProcessor::writeRow(TupleSerializer &serializer, TableTuple &tuple)
{
    if (m_table == NULL) {
        throwFatalException("TupleOutputStreamProcessor::writeRow() was called before initialize().");
    }
<commit_msg>ENG-4290 Fixed minor error in exception message.<commit_after>/* 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/>.
 */

#include "TupleOutputStream.h"
#include "TupleOutputStreamProcessor.h"
#include "StreamPredicate.h"
#include "TupleSerializer.h"
#include "tabletuple.h"
#include <limits>

namespace voltdb {

/** Default constructor. */
TupleOutputStreamProcessor::TupleOutputStreamProcessor()
    : boost::ptr_vector<TupleOutputStream>()
{
    clearState();
}

/** Constructor with initial size. */
TupleOutputStreamProcessor::TupleOutputStreamProcessor(std::size_t nBuffers)
    : boost::ptr_vector<TupleOutputStream>(nBuffers)
{
    clearState();
}

/** Constructor for a single stream. Convenient for backward compatibility in tests. */
TupleOutputStreamProcessor::TupleOutputStreamProcessor(void *data, std::size_t length)
    : boost::ptr_vector<TupleOutputStream>(1)
{
    clearState();
    add(data, length);
}

/** Private method used by constructors, etc. to clear state. */
void TupleOutputStreamProcessor::clearState()
{
    m_maxTupleLength = 0;
    m_predicates = NULL;
    m_table = NULL;
    m_totalBytesSerialized = 0;
}

/** Convenience method to create and add a new TupleOutputStream. */
TupleOutputStream &TupleOutputStreamProcessor::add(void *data, std::size_t length)
{
    std::auto_ptr<TupleOutputStream> out(new TupleOutputStream(data, length));
    push_back(out);
    return *out;
}

/** Start serializing. */
void TupleOutputStreamProcessor::open(PersistentTable &table,
                                      std::size_t maxTupleLength,
                                      int32_t partitionId,
                                      StreamPredicateList &predicates,
                                      int32_t totalPartitions)
{
    m_table = &table;
    m_maxTupleLength = maxTupleLength;
    // It must be either one predicate per output stream or none at all.
    bool havePredicates = !predicates.empty();
    if (havePredicates && predicates.size() != size()) {
        throwFatalException("serializeMore() expects either no predicates or one per output stream.");
    }
    m_predicates = &predicates;
    m_totalPartitions = totalPartitions;
    for (TupleOutputStreamProcessor::iterator iter = begin(); iter != end(); ++iter) {
        iter->startRows(partitionId);
    }
    m_totalBytesSerialized = 0;
    // Recalculate the serialization throttle threshold based on the number of partitions.
    if (m_totalPartitions > 0) {
        m_bytesSerializedThreshold = m_bytesSerializedThresholdPerPartition * m_totalPartitions;
    }
    else {
        m_bytesSerializedThreshold = m_bytesSerializedThresholdPerPartition;
    }
}

/** Stop serializing. */
void TupleOutputStreamProcessor::close()
{
    for (TupleOutputStreamProcessor::iterator iter = begin(); iter != end(); ++iter) {
        iter->endRows();
    }
    clearState();
}

/**
 * Write a tuple to the output streams.
 * Expects buffer space was already checked.
 * Returns true when the caller should yield to allow other work to proceed.
 */
bool TupleOutputStreamProcessor::writeRow(TupleSerializer &serializer, TableTuple &tuple)
{
    if (m_table == NULL) {
        throwFatalException("TupleOutputStreamProcessor::writeRow() was called before open().");
    }

{'fraction_non_alphanumeric': 0.096, 'fraction_numerical': 0.005079365079365079, 'mean_word_length': 3.5924198250728865, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6192428', 'n_tokens_mistral': 2124, 'n_tokens_neox': 2069, 'n_words': 670}