text:
#include <turbodbc_numpy/datetime_column.h>
#include <turbodbc_numpy/ndarrayobject.h>
#include <turbodbc_numpy/make_numpy_array.h>
#include <turbodbc/time_helpers.h>

#include <Python.h>

#include <boost/date_time/gregorian/gregorian_types.hpp>
#include <boost/date_time/posix_time/posix_time_types.hpp>

#ifdef _WIN32
#include <windows.h>
#endif
#include <sql.h>
#include <cstring>

#ifdef __GNUC__
#define EXTENSION __extension__
#else
#define EXTENSION
#endif

namespace turbodbc_numpy {

namespace {

	std::string get_type_descriptor(turbodbc::type_code type)
	{
		if (type == turbodbc::type_code::timestamp) {
			return "datetime64[us]";
		} else {
			return "datetime64[D]";
		}
	}

	datetime_column::converter make_converter(turbodbc::type_code type)
	{
		if (type == turbodbc::type_code::timestamp) {
			return turbodbc::timestamp_to_microseconds;
		} else {
			return turbodbc::date_to_days;
		}
	}

	PyArrayObject * get_array_ptr(pybind11::object & object)
	{
		return reinterpret_cast<PyArrayObject *>(object.ptr());
	}

}

datetime_column::datetime_column(turbodbc::type_code type) :
	type_(type),
	data_(make_empty_numpy_array(get_type_descriptor(type_))),
	mask_(make_empty_numpy_array(numpy_bool_type)),
	size_(0),
	converter_(make_converter(type_))
{
}

datetime_column::~datetime_column() = default;

void datetime_column::do_append(cpp_odbc::multi_value_buffer const & buffer, std::size_t n_values)
{
	auto const old_size = size_;
	resize(old_size + n_values);

	auto const data_pointer = static_cast<int64_t *>(PyArray_DATA(get_array_ptr(data_))) + old_size;
	auto const mask_pointer = static_cast<std::int8_t *>(PyArray_DATA(get_array_ptr(mask_))) + old_size;
	std::memset(mask_pointer, 0, n_values);

	for (std::size_t i = 0; i != n_values; ++i) {
		auto element = buffer[i];
		if (element.indicator == SQL_NULL_DATA) {
			mask_pointer[i] = 1;
		} else {
			reinterpret_cast<intptr_t *>(data_pointer)[i] = converter_(element.data_pointer);
		}
	}
}

pybind11::object datetime_column::do_get_data()
{
	return data_;
}

pybind11::object datetime_column::do_get_mask()
{
	return mask_;
}

void datetime_column::resize(std::size_t new_size)
{
	npy_intp size = new_size;
	PyArray_Dims new_dimensions = {&size, 1};
	int const no_reference_check = 0;
	EXTENSION PyArray_Resize(get_array_ptr(data_), &new_dimensions, no_reference_check, NPY_ANYORDER);
	EXTENSION PyArray_Resize(get_array_ptr(mask_), &new_dimensions, no_reference_check, NPY_ANYORDER);
	size_ = new_size;
}

}

max_stars_repo_name: arikfr/turbodbc

max_issues_repo_name: arikfr/turbodbc

max_forks_repo_name: arikfr/turbodbc


text:
/*
 * Copyright (C) 2015 Love Park Robotics, LLC
 *
 * 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 distribted 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.
 */

#include <cstdint>
#include <exception>
#include <fstream>
#include <iostream>
#include <memory>
#include <sstream>
#include <string>
#include <vector>
#include <boost/program_options.hpp>
#include "o3d3xx_camera.h"

//
// Imports a camera configuration or just an application based upon the IFM
// zip-compressed JSON file(s)
//

namespace po = boost::program_options;

int main(int argc, const char* argv[])
{
  int idx = -1;

  std::string camera_ip;
  std::uint32_t xmlrpc_port;
  std::string password;
  std::string infile;

  try
    {
      o3d3xx::CmdLineOpts opts("o3d3xx IFM Import");

      po::options_description import_opts("Import Information");
      import_opts.add_options()
        ("file,f", po::value<std::string>(), "Input IFM file")
        ("app,a",
         "Flag indicating that the file to import is an application")
        ("config,c",
         "Flag indicating that the file to import is a camera configuration");

      opts.visible.add(import_opts);

      if (! opts.Parse(argc, argv, &camera_ip, &xmlrpc_port, &password))
        {
          return 0;
        }

      if (opts.vm.count("app") && opts.vm.count("config"))
        {
          std::cerr << "Only one of `app' or `config' may be supplied!"
                    << std::endl << std::flush;

         throw(o3d3xx::error_t(O3D3XX_INVALID_ARGUMENT));
        }
      else if (! (opts.vm.count("app") || opts.vm.count("config")))
        {
          std::cerr << "Exactly one of `app' or `config' must be supplied!"
                    << std::endl << std::flush;

          throw(o3d3xx::error_t(O3D3XX_INVALID_ARGUMENT));
        }

      infile.assign(opts.vm["file"].as<std::string>());
      std::ifstream ifs(infile, std::ios::in|std::ios::binary);
      if (! ifs)
        {
          if (opts.vm.count("file") == 0)
            {
              std::cerr << "The `file' option is required!"
                        << std::endl << std::flush;
            }

          throw o3d3xx::error_t(O3D3XX_IO_ERROR);
        }

      // read the file into a byte buffer
      ifs.unsetf(std::ios::skipws);
      std::streampos file_size;
      ifs.seekg(0, std::ios::end);
      file_size = ifs.tellg();
      ifs.seekg(0, std::ios::beg);

      std::vector<std::uint8_t> bytes;
      bytes.reserve(file_size);

      bytes.insert(bytes.begin(),
                   std::istream_iterator<std::uint8_t>(ifs),
                   std::istream_iterator<std::uint8_t>());

      // send the byte buffer to the camera for import
      o3d3xx::Camera::Ptr cam =
        std::make_shared<o3d3xx::Camera>(camera_ip, xmlrpc_port, password);

      if (opts.vm.count("app"))
        {
          // import an app
          idx = cam->ImportIFMApp(bytes);
          std::cout << "New application created at index: "
                    << idx << std::endl << std::flush;
        }
      else
        {
          // import a "configuration"
          std::cerr << "Config import: not yet implemented!"
                    << std::endl;
        }
    }
  catch (const std::exception& e)
    {
      std::cerr << "Failed to import file '"
                << infile << "':"
                << std::endl << e.what() << std::endl;
      return 1;
    }

  return 0;
}

ext: cpp


text:
/*
// Licensed to DynamoBI Corporation (DynamoBI) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  DynamoBI licenses this file
// to you 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.
*/

#include "fennel/common/CommonPreamble.h"
#include "fennel/common/FemEnums.h"
#include "fennel/test/ExecStreamUnitTestBase.h"
#include "fennel/lcs/LcsClusterAppendExecStream.h"
#include "fennel/sorter/ExternalSortExecStream.h"
#include "fennel/lbm/LbmGeneratorExecStream.h"
#include "fennel/lbm/LbmSplicerExecStream.h"
#include "fennel/lbm/LbmSearchExecStream.h"
#include "fennel/ldbtest/LbmExecStreamTestBase.h"
#include "fennel/btree/BTreeBuilder.h"
#include "fennel/ftrs/BTreeInsertExecStream.h"
#include "fennel/ftrs/BTreeSearchExecStream.h"
#include "fennel/ftrs/BTreeExecStream.h"
#include "fennel/tuple/StandardTypeDescriptor.h"
#include "fennel/tuple/TupleDescriptor.h"
#include "fennel/exec/MockProducerExecStream.h"
#include "fennel/exec/ValuesExecStream.h"
#include "fennel/exec/ExecStreamGraph.h"
#include "fennel/exec/ExecStreamEmbryo.h"
#include "fennel/exec/SplitterExecStream.h"
#include "fennel/exec/BarrierExecStream.h"
#include "fennel/cache/Cache.h"
#include "fennel/common/SearchEndpoint.h"
#include <stdarg.h>

#include <boost/test/test_tools.hpp>

using namespace fennel;

/**
 * Testcase for scanning a single bitmap index using equality search on
 * all index keys
 */
class LbmSearchTest : public LbmExecStreamTestBase
{
protected:
    TupleAttributeDescriptor attrDesc_char1;
    TupleAttributeDescriptor attrDesc_nullableInt64;

    /**
     * BTrees corresponding to the clusters
     */
    vector<boost::shared_ptr<BTreeDescriptor> > bTreeClusters;

    /**
     * Saved root pageids of btrees corresponding to clusters; used to
     * append to existing table and to read from them
     */
    vector<PageId> savedBTreeClusterRootIds;

    /**
     * BTrees corresponding to the bitmaps
     */
    vector<boost::shared_ptr<BTreeDescriptor> > bTreeBitmaps;

    /**
     * Saved root pageids of btrees corresponding to bitmaps; used to
     * append to existing table and to read from them
     */
    vector<PageId> savedBTreeBitmapRootIds;

    /**
     * Initializes a BTreeExecStreamParam structure
     */
    void initBTreeExecStreamParam(
        BTreeExecStreamParams &param, shared_ptr<BTreeDescriptor> pBTreeDesc);

    /**
     * Initializes BTreeParams structure
     */
    void initBTreeParam(
        BTreeParams &param, shared_ptr<BTreeDescriptor> pBTreeDesc);

    /**
     * Initializes a cluster scan def structure for a LcsRowScanBase exec
     * stream
     */
    void initClusterScanDef(
        LcsRowScanBaseExecStreamParams &generatorParams,
        struct LcsClusterScanDef &clusterScanDef, uint bTreeIndex);

    /**
     * Initializes BTreeExecStreamParam corresponding to a bitmap index
     *
     * @param tupleDesc tuple descriptor corresponding to bitmap index
     * @param keyProj projection corresponding to the bitmap index keys;
     * excludes start rid
     * @param nKeys number of keys in the bitmap index; excludes start
     * rid from key count
     */
    void  initBTreeBitmapDesc(
        TupleDescriptor &tupleDesc, TupleProjection &keyProj, uint nKeys);

    /**
     * Initializes a tuple descriptor corresponding to a bitmap index
     *
     * @param tupleDesc tuple descriptor corresponding to bitmap index
     * @param nKeys number of keys in the bitmap index; excludes start
     * rid from key count
     */
    void initBTreeTupleDesc(TupleDescriptor &tupleDesc, uint nKeys);

    /**
     * Loads a table with nClusters clusters, 1 column per cluster, and nRows
     * rows.
     *
     *<p>

     * Each column has a repeating sequence of values based on the value in the
     * repeatSeqValues vector.  E.g., a repeating sequence of n will have
     * values:
     *
     * (0, 1, 2, ..., n-1, 0, 1, 2, ..., n-1, 0, 1, 2, ...).
     *
     * Then a single bitmap index is created on all columns.
     *
     * @param nRows number of rows to load
     * @param nClusters number of clusters to create
     * @param repeatSeqValues repeating sequence values for each column
     * @param newRoot if true, append to existing table
     */
    void loadTableAndIndex(
        uint nRows, uint nClusters, std::vector<int> const &repeatSeqValues,
        bool newRoot);

    /**
     * Tests equality scan on the table created in loadTableAndIndex, using
     * the entire index key (minus the rid)
     *
     * @param nRows total number of rows in the table
     * @param nKeys number of keys in the index (excluding startrid)
     * @param repeatSeqValues initial repeating sequence values for each column
     * @param useDynamicKeys if true, pass search keys through dynamic
     * parameters
     * @param includeRid if true, include rid >= 0 in the search
     */
    void testScanFullKey(
        uint nRows, uint nKeys, std::vector<int> const &repeatSeqValues,
        bool useDynamicKeys, bool includeRid);

    /**
     * Tests equality scan on the table created in loadTableAndIndex, using
     * the all keys except the rid and the last key
     *
     * @param nRows total number of rows in the table
     * @param nKeys number of keys in the index (excluding startrid)
     * @param repeatSeqValues initial repeating sequence values for each column
     */
    void testScanPartialKey(
        uint nRows, uint nKeys, std::vector<int> const &repeatSeqValues);

    /**
     * Performs an index search using the key values/directives passed in
     *
     * @param totalKeys total number of keys in the index (excluding startrid)
     *
     * @param nKeys number of keys to use in index search; always excludes
     * startrid
     *
     * @param bufSize size of input buffer containing search keys/directive
     *
     * @param inputBuffer buffer containing search key/directive tuples to
     * be passed into the index scan
     *
     * @param expectedNBitmaps expected number of bitmaps in result
     *
     * @param expectedBitmaps buffer containing expected bitmap result
     *
     * @param dynamicRootPageId if true, pass in the btree rootPageId as a
     * dynamic parameter
     *
     * @param useDynamicKeys if true, pass in search key values using dynamic
     * parameters
     *
     * @param includeRid include rid in search
     *
     * @param vals search key values
     */
    void testScanIdx(
        uint totalKeys, uint nKeys, uint bufSize,
        boost::shared_array<FixedBuffer> inputBuffer,
        uint expectedNBitmaps, PBuffer expectedBitmaps,
        bool dynamicRootPageId,
        bool useDynamicKeys,
        bool includeRid,
        const boost::scoped_array<uint64_t> &vals);

    /**
     * Initializes input search key and directives for an equality search
     *
     * @param nKeys number of keys to search on
     * @param nInputTuples number of search ranges to create
     * @param vals values to search on
     * @param lowerDirective lower bound search directive
     * @param upperDirective upper bound search directive
     * @param inputTupleAccessor accessor to marshal/unmarshal search key
     * @param inputTupleData tupledata storing search key
     * @param inputBuffer buffer storing search key
     * @param useDynamicKeys if true, search key values are passed to the
     * search stream using dynamic parameters
     */
    void initEqualSearch(
        uint nKeys, uint nInputTuples, boost::scoped_array<uint64_t> &vals,
        char &lowerDirective, char &upperDirective,
        TupleAccessor &inputTupleAccessor, TupleData &inputTupleData,
        boost::shared_array<FixedBuffer> &inputBuffer,
        bool useDynamicKeys);

    void setSearchKey(
        char lowerDirective, char upperDirective, uint64_t lowerVal,
        uint64_t upperVal, PBuffer inputBuf, uint &offset,
        TupleAccessor &inputTupleAccessor, TupleData &inputTupleData);

public:
    explicit LbmSearchTest()
    {
        FENNEL_UNIT_TEST_CASE(LbmSearchTest, testScanOneLevel);
        FENNEL_UNIT_TEST_CASE(LbmSearchTest, testScanTwoLevel);
        FENNEL_UNIT_TEST_CASE(LbmSearchTest, testMultipleRanges);
    }

    void testCaseSetUp();
    void testCaseTearDown();
    void testScans(uint nRows);

    void testScanOneLevel();
    void testScanTwoLevel();
    void testMultipleRanges();
};

void LbmSearchTest::testScanOneLevel()
{
    // single level btree
    testScans(100);
}

void LbmSearchTest::testScanTwoLevel()
{
    // with a 3-key index, 2000 rows will generate a 2-level btree
    testScans(2000);
}

void LbmSearchTest::testScans(uint nRows)
{
    uint nClusters = 3;
    std::vector<int> repeatSeqValues;

    // load the data
    repeatSeqValues.push_back(1);
    repeatSeqValues.push_back(5);
    repeatSeqValues.push_back(9);
    loadTableAndIndex(nRows, nClusters, repeatSeqValues, true);

    // Scan on all keys, passing the search keys through the input stream.
    // Then do the same, passing the keys via dynamic parameters.  Then,
    // finally, do the search, including a rid search.
    testScanFullKey(nRows, nClusters, repeatSeqValues, false, false);
    testScanFullKey(nRows, nClusters, repeatSeqValues, true, false);
    testScanFullKey(nRows, nClusters, repeatSeqValues, false, true);

    // scan on (nClusters - 1) keys
    testScanPartialKey(nRows, nClusters, repeatSeqValues);
}

void LbmSearchTest::testMultipleRanges()
{
    uint nRows = 20000;
    uint nClusters = 1;
    std::vector<int> repeatSeqValues;

    // load a table with a single index on a single column
    repeatSeqValues.push_back(100);
    loadTableAndIndex(nRows, nClusters, repeatSeqValues, true);

    // scan on all keys, just to make sure all key values really are there
    testScanFullKey(nRows, nClusters, repeatSeqValues, false, false);

    resetExecStreamTest();

    // Setup the following search keys.  Note that these keys weren't randomly
    // selected.  Some of them correspond to boundary conditions.
    // 1. key < 8
    // 2. key > 10 && key <= 17
    // 3. key >= 44 and key < 60
    // 4. key > 71

    TupleDescriptor inputTupleDesc;
    for (uint i = 0; i < 2; i++) {
        inputTupleDesc.push_back(attrDesc_char1);
        inputTupleDesc.push_back(attrDesc_nullableInt64);
    }
    TupleData inputTupleData(inputTupleDesc);
    TupleAccessor inputTupleAccessor;
    inputTupleAccessor.compute(inputTupleDesc);

    uint nInputTuples = 4;
    boost::shared_array<FixedBuffer> inputBuffer;
    inputBuffer.reset(
        new FixedBuffer[nInputTuples * inputTupleAccessor.getMaxByteCount()]);
    PBuffer inputBuf = inputBuffer.get();
    uint offset = 0;

    setSearchKey(
        '-', ')', 0, 8, inputBuf, offset, inputTupleAccessor, inputTupleData);
    setSearchKey(
        '(', ']', 10, 17, inputBuf, offset, inputTupleAccessor,
        inputTupleData);
    setSearchKey(
        '[', ')', 44, 60, inputBuf, offset, inputTupleAccessor,
        inputTupleData);
    setSearchKey(
        '(', '+', 71, 0, inputBuf, offset, inputTupleAccessor,
        inputTupleData);

    // setup the expected bitmap result values
    boost::scoped_array<FixedBuffer> expectedBitmaps;
    uint bufferSize = ((nRows / repeatSeqValues[0] / 8 + 1) * 60) * 24;
    expectedBitmaps.reset(new FixedBuffer[bufferSize]);
    PBuffer bitmapBuf = expectedBitmaps.get();
    uint expectedNBitmaps = 0;
    uint expectedBufSize = 0;
    // for each range, generate the bitmap values for each key in the desired
    // range
    for (uint i = 0; i < 8; i++) {
        generateBitmaps(
            nRows, i, repeatSeqValues[0], bitmapBuf, expectedBufSize,
            bufferSize, expectedNBitmaps);
    }
    for (uint i = 11; i <= 17; i++) {
        generateBitmaps(
            nRows, i, repeatSeqValues[0], bitmapBuf, expectedBufSize,
            bufferSize, expectedNBitmaps);
    }
    for (uint i = 44; i < 60; i++) {
        generateBitmaps(
            nRows, i, repeatSeqValues[0], bitmapBuf, expectedBufSize,
            bufferSize, expectedNBitmaps);
    }
    for (uint i = 72; i < repeatSeqValues[0]; i++) {
        generateBitmaps(
            nRows, i, repeatSeqValues[0], bitmapBuf, expectedBufSize,
            bufferSize, expectedNBitmaps);
    }

    boost::scoped_array<uint64_t> vals;
    testScanIdx(
        nClusters, nClusters, offset, inputBuffer, expectedNBitmaps,
        bitmapBuf, true, false, false, vals);
}

void LbmSearchTest::setSearchKey(
    char lowerDirective, char upperDirective, uint64_t lowerVal,
    uint64_t upperVal, PBuffer inputBuf, uint &offset,
    TupleAccessor &inputTupleAccessor, TupleData &inputTupleData)
{
    inputTupleData[0].pData = (PConstBuffer) &lowerDirective;
    inputTupleData[2].pData = (PConstBuffer) &upperDirective;
    if (lowerDirective != '-') {
        inputTupleData[1].pData = (PConstBuffer) &lowerVal;
    }
    if (upperDirective != '+') {
        inputTupleData[3].pData = (PConstBuffer) &upperVal;
    }
    inputTupleAccessor.marshal(inputTupleData, inputBuf + offset);
    offset += inputTupleAccessor.getCurrentByteCount();
}

void LbmSearchTest::testScanFullKey(
    uint nRows, uint nKeys, std::vector<int> const &repeatSeqValues,
    bool useDynamicKeys, bool includeRid)
{
    // search for key0 = <val0>, key1 = <val1>, ..., key(n-1) = <val(n-1)>
    uint nInputTuples = 1;
    boost::scoped_array<uint64_t> vals;
    char lowerDirective;
    char upperDirective;
    TupleAccessor inputTupleAccessor;
    TupleData inputTupleData;
    boost::shared_array<FixedBuffer> inputBuffer;

    initEqualSearch(
        nKeys, nInputTuples, vals, lowerDirective, upperDirective,
        inputTupleAccessor, inputTupleData, inputBuffer, useDynamicKeys);

    // do a search on each possible key combo
    uint skipRows = 1;
    for (uint i = 0; i < nKeys; i++) {
        skipRows *= repeatSeqValues[i];
    }
    for (uint i = 0; i < skipRows; i++) {
        // generate input keys for search
        for (uint j = 0; j < nKeys; j++) {
            vals[j] = i % repeatSeqValues[j];
        }
        inputTupleAccessor.marshal(inputTupleData, inputBuffer.get());

        // generate expected bitmap result
        boost::scoped_array<FixedBuffer> expectedBitmaps;
        uint bufferSize = (nRows / skipRows + 1) * 16;
        expectedBitmaps.reset(new FixedBuffer[bufferSize]);
        uint expectedNBitmaps = 0;
        uint expectedBufSize = 0;
        generateBitmaps(
            nRows, i, skipRows, expectedBitmaps.get(), expectedBufSize,
            bufferSize, expectedNBitmaps);

        testScanIdx(
            nKeys, nKeys, inputTupleAccessor.getCurrentByteCount(),
            inputBuffer, expectedNBitmaps, expectedBitmaps.get(), false,
            useDynamicKeys, includeRid, vals);
    }
}

void LbmSearchTest::testScanPartialKey(
    uint nRows, uint nKeys, std::vector<int> const &repeatSeqValues)
{
    // search for key0 = 0, key1 = 0, ..., key(n-2) = 0
    uint nInputTuples = 1;
    boost::scoped_array<uint64_t> vals;
    char lowerDirective;
    char upperDirective;
    TupleAccessor inputTupleAccessor;
    TupleData inputTupleData;
    boost::shared_array<FixedBuffer> inputBuffer;

    initEqualSearch(
        nKeys - 1, nInputTuples, vals, lowerDirective, upperDirective,
        inputTupleAccessor, inputTupleData, inputBuffer, false);

    // generate input keys for search
    for (uint j = 0; j < nKeys - 1; j++) {
        vals[j] = 0;
    }
    inputTupleAccessor.marshal(inputTupleData, inputBuffer.get());

    // Generate one set of bitmaps for each key combo that can be combined
    // with the partial key. E.g., if there are 3 keys, generate the
    // bitmaps that would be obtained from searching for (0, 0, 0), (0, 0, 1),
    // ..., (0, 0, repeatSeqValues[nKeys-1] - 1)

    uint skipRows = 1;
    for (uint i = 0; i < nKeys - 1; i++) {
        skipRows *= repeatSeqValues[i];
    }
    boost::scoped_array<FixedBuffer> expectedBitmaps;
    uint bufferSize = (nRows / skipRows / 8 + 1)
        * 12 * repeatSeqValues[nKeys - 1];
    expectedBitmaps.reset(new FixedBuffer[bufferSize]);
    PBuffer bitmapBuf = expectedBitmaps.get();
    uint expectedNBitmaps = 0;
    uint curBufSize = 0;

    for (uint i = 0; i < repeatSeqValues[nKeys - 1]; i++) {
        uint start;
        if (i == 0) {
            start = 0;
        } else {
            // look for the first rid where the last key is equal to "i" and
            // the preceeding keys are all 0
            for (start = i; start < nRows;
                 start += repeatSeqValues[nKeys - 1])
            {
                uint j;
                for (j = 0; j < nKeys - 1; j++) {
                    if (start % repeatSeqValues[j] != 0) {
                        break;
                    }
                }
                if (j == nKeys - 1) {
                    break;
                }
            }
            if (start >= nRows) {
                continue;
            }
        }
        generateBitmaps(
            nRows, start, skipRows * repeatSeqValues[nKeys - 1],
            bitmapBuf, curBufSize, bufferSize, expectedNBitmaps);
    }
    testScanIdx(
        nKeys, nKeys - 1, inputTupleAccessor.getCurrentByteCount(),
        inputBuffer, expectedNBitmaps, bitmapBuf, false, false, false, vals);
}

void LbmSearchTest::initEqualSearch(
    uint nKeys, uint nInputTuples, boost::scoped_array<uint64_t> &vals,
    char &lowerDirective, char &upperDirective,
    TupleAccessor &inputTupleAccessor, TupleData &inputTupleData,
    boost::shared_array<FixedBuffer> &inputBuffer,
    bool useDynamicKeys)
{
    TupleDescriptor inputTupleDesc;
    for (uint i = 0; i < 2; i++) {
        inputTupleDesc.push_back(attrDesc_char1);
        for (uint j = 0; j < nKeys; j++) {
            inputTupleDesc.push_back(attrDesc_nullableInt64);
        }
    }

    inputTupleData.compute(inputTupleDesc);

    vals.reset(new uint64_t[nKeys]);
    lowerDirective = '[';
    inputTupleData[0].pData = (PConstBuffer) &lowerDirective;
    upperDirective = ']';
    inputTupleData[nKeys + 1].pData = (PConstBuffer) &upperDirective;
    for (uint i = 0; i < nKeys; i++) {
        // If keys are being passed through dynamic parameters, set them to
        // NULL in the input row
        if (useDynamicKeys) {
            inputTupleData[i + 1].pData = NULL;
            inputTupleData[i + 1].cbData = 0;
            inputTupleData[nKeys + 1 + i + 1].pData = NULL;
            inputTupleData[nKeys + 1 + i + 1].cbData = 0;
        } else {
            inputTupleData[i + 1].pData = (PConstBuffer) &vals[i];
            inputTupleData[nKeys + 1 + i + 1].pData = (PConstBuffer) &vals[i];
        }
    }

    inputTupleAccessor.compute(inputTupleDesc);

    inputBuffer.reset(
        new FixedBuffer[nInputTuples * inputTupleAccessor.getMaxByteCount()]);
}

void LbmSearchTest::loadTableAndIndex(
    uint nRows, uint nClusters, std::vector<int> const &repeatSeqValues,
    bool newRoot)
{
    // 0. reset member fields.
    for (uint i = 0; i < bTreeClusters.size(); i++) {
        bTreeClusters[i]->segmentAccessor.reset();
    }
    for (uint i = 0; i < bTreeBitmaps.size(); i++) {
        bTreeBitmaps[i]->segmentAccessor.reset();
    }
    bTreeClusters.clear();
    bTreeBitmaps.clear();

    // 1. setup mock input stream

    MockProducerExecStreamParams mockParams;
    for (uint i = 0; i < nClusters; i++) {
        mockParams.outputTupleDesc.push_back(attrDesc_int64);
    }
    mockParams.nRows = nRows;

    vector<boost::shared_ptr<ColumnGenerator<int64_t> > > columnGenerators;
    SharedInt64ColumnGenerator col;
    assert(repeatSeqValues.size() == nClusters);
    for (uint i = 0; i < repeatSeqValues.size(); i++) {
        col =
            SharedInt64ColumnGenerator(
                new RepeatingSeqColumnGenerator(repeatSeqValues[i]));
        columnGenerators.push_back(col);
    }
    mockParams.pGenerator.reset(
        new CompositeExecStreamGenerator(columnGenerators));

    ExecStreamEmbryo mockStreamEmbryo;
    mockStreamEmbryo.init(new MockProducerExecStream(), mockParams);
    mockStreamEmbryo.getStream()->setName("MockProducerExecStream");

    // 2. setup splitter stream for cluster loads

    SplitterExecStreamParams splitterParams;
    ExecStreamEmbryo splitterStreamEmbryo;
    splitterStreamEmbryo.init(new SplitterExecStream(), splitterParams);
    splitterStreamEmbryo.getStream()->setName("ClusterSplitterExecStream");

    // 3. setup loader streams

    vector<ExecStreamEmbryo> lcsAppendEmbryos;
    for (uint i = 0; i < nClusters; i++) {
        LcsClusterAppendExecStreamParams lcsAppendParams;
        boost::shared_ptr<BTreeDescriptor> pBTreeDesc =
            boost::shared_ptr<BTreeDescriptor> (new BTreeDescriptor());
        bTreeClusters.push_back(pBTreeDesc);

        // initialize the btree parameter portion of lcsAppendParams
        // BTree tuple desc has two columns (rid, clusterPageid)
        (lcsAppendParams.tupleDesc).push_back(attrDesc_int64);
        (lcsAppendParams.tupleDesc).push_back(attrDesc_int64);

        // BTree key only has one column which is the first column.
        (lcsAppendParams.keyProj).push_back(0);

        initBTreeExecStreamParam(lcsAppendParams, pBTreeDesc);

        // output two values (rows inserted, starting rid value)
        lcsAppendParams.outputTupleDesc.push_back(attrDesc_int64);
        lcsAppendParams.outputTupleDesc.push_back(attrDesc_int64);

        lcsAppendParams.inputProj.push_back(i);

        // create an empty page to start the btree

        if (newRoot) {
            BTreeBuilder builder(*pBTreeDesc, pRandomSegment);
            builder.createEmptyRoot();
            savedBTreeClusterRootIds.push_back(builder.getRootPageId());
        }
        lcsAppendParams.rootPageId = pBTreeDesc->rootPageId =
            savedBTreeClusterRootIds[i];

        // Now use the above initialized parameter

        ExecStreamEmbryo lcsAppendStreamEmbryo;
        lcsAppendStreamEmbryo.init(
            new LcsClusterAppendExecStream(), lcsAppendParams);
        std::ostringstream oss;
        oss << "LcsClusterAppendExecStream" << "#" << i;
        lcsAppendStreamEmbryo.getStream()->setName(oss.str());
        lcsAppendEmbryos.push_back(lcsAppendStreamEmbryo);
    }

    // 4. setup barrier stream for cluster loads

    BarrierExecStreamParams barrierParams;
    barrierParams.outputTupleDesc.push_back(attrDesc_int64);
    barrierParams.outputTupleDesc.push_back(attrDesc_int64);
    barrierParams.returnMode = BARRIER_RET_ANY_INPUT;

    ExecStreamEmbryo clusterBarrierStreamEmbryo;
    clusterBarrierStreamEmbryo.init(new BarrierExecStream(), barrierParams);
    clusterBarrierStreamEmbryo.getStream()->setName("ClusterBarrierExecStream");

    // create a DAG with the above, but without the final output sink
    prepareDAG(
        mockStreamEmbryo, splitterStreamEmbryo, lcsAppendEmbryos,
        clusterBarrierStreamEmbryo, false);

    // 5. setup splitter stream for create bitmaps

    splitterStreamEmbryo.init(
        new SplitterExecStream(), splitterParams);
    splitterStreamEmbryo.getStream()->setName("BitmapSplitterExecStream");

    // create streams for bitmap generator, sort, and bitmap splicer to
    // build an index on all columns

    std::vector<std::vector<ExecStreamEmbryo> > createBitmapStreamList;
        std::vector<ExecStreamEmbryo> createBitmapStream;

    // 6. setup generator

    LbmGeneratorExecStreamParams generatorParams;
    struct LcsClusterScanDef clusterScanDef;
    clusterScanDef.clusterTupleDesc.push_back(attrDesc_int64);

    for (uint j = 0; j < nClusters; j++) {
        initClusterScanDef(generatorParams, clusterScanDef, j);
    }

    TupleProjection proj;
    for (uint j = 0; j < nClusters; j++) {
        proj.push_back(j);
    }
    generatorParams.outputProj = proj;
    generatorParams.insertRowCountParamId = DynamicParamId(1);
    generatorParams.createIndex = false;

    boost::shared_ptr<BTreeDescriptor> pBTreeDesc =
        boost::shared_ptr<BTreeDescriptor> (new BTreeDescriptor());
    bTreeBitmaps.push_back(pBTreeDesc);

    // BTree tuple desc has the key columns + starting Rid + varbinary
    // field for bit segments/bit descriptors
    uint nKeys = nClusters;
    initBTreeTupleDesc(generatorParams.outputTupleDesc, nKeys);

    initBTreeBitmapDesc(
        generatorParams.tupleDesc, generatorParams.keyProj, nKeys);
    initBTreeExecStreamParam(generatorParams, pBTreeDesc);

    // create an empty page to start the btree

    if (newRoot) {
        BTreeBuilder builder(*pBTreeDesc, pRandomSegment);
        builder.createEmptyRoot();
        savedBTreeBitmapRootIds.push_back(builder.getRootPageId());
    }
    generatorParams.rootPageId = pBTreeDesc->rootPageId =
        savedBTreeBitmapRootIds[0];

    ExecStreamEmbryo generatorStreamEmbryo;
    generatorStreamEmbryo.init(
        new LbmGeneratorExecStream(), generatorParams);
    std::ostringstream oss;
    oss << "LbmGeneratorExecStream" << "#" << 0;
    generatorStreamEmbryo.getStream()->setName(oss.str());
    createBitmapStream.push_back(generatorStreamEmbryo);

    // 7. setup sorter

    ExternalSortExecStreamParams sortParams;
    initBTreeBitmapDesc(
        sortParams.outputTupleDesc, sortParams.keyProj, nKeys);
    sortParams.distinctness = DUP_ALLOW;
    sortParams.pTempSegment = pRandomSegment;
    sortParams.pCacheAccessor = pCache;
    sortParams.scratchAccessor =
        pSegmentFactory->newScratchSegment(pCache, 10);
    sortParams.storeFinalRun = false;
    sortParams.partitionKeyCount = 0;
    sortParams.estimatedNumRows = MAXU;
    sortParams.earlyClose = false;

    ExecStreamEmbryo sortStreamEmbryo;
    sortStreamEmbryo.init(
        ExternalSortExecStream::newExternalSortExecStream(), sortParams);
    sortStreamEmbryo.getStream()->setName("ExternalSortExecStream");
    std::ostringstream oss2;
    oss2 << "ExternalSortExecStream" << "#" << 0;
    sortStreamEmbryo.getStream()->setName(oss2.str());
    createBitmapStream.push_back(sortStreamEmbryo);

    // 8. setup splicer

    LbmSplicerExecStreamParams splicerParams;
    splicerParams.createNewIndex = false;
    splicerParams.scratchAccessor =
        pSegmentFactory->newScratchSegment(pCache, 15);
    splicerParams.pCacheAccessor = pCache;
    BTreeParams bTreeParams;
    initBTreeBitmapDesc(
        bTreeParams.tupleDesc, bTreeParams.keyProj, nKeys);
    initBTreeParam(bTreeParams, pBTreeDesc);
    bTreeParams.rootPageId = pBTreeDesc->rootPageId;
    splicerParams.bTreeParams.push_back(bTreeParams);
    splicerParams.insertRowCountParamId = DynamicParamId(1);
    splicerParams.writeRowCountParamId = DynamicParamId(0);
    splicerParams.outputTupleDesc.push_back(attrDesc_int64);

    ExecStreamEmbryo splicerStreamEmbryo;
    splicerStreamEmbryo.init(new LbmSplicerExecStream(), splicerParams);
    std::ostringstream oss3;
    oss3 << "LbmSplicerExecStream" << "#" << 0;
    splicerStreamEmbryo.getStream()->setName(oss3.str());
    createBitmapStream.push_back(splicerStreamEmbryo);

    // connect the sorter and splicer to generator and then add this
    // newly connected stream to the list of create bitmap stream embryos
    createBitmapStreamList.push_back(createBitmapStream);

    // 9. setup barrier stream for create bitmaps

    barrierParams.outputTupleDesc.clear();
    barrierParams.outputTupleDesc.push_back(attrDesc_int64);

    ExecStreamEmbryo barrierStreamEmbryo;
    barrierStreamEmbryo.init(
        new BarrierExecStream(), barrierParams);
    barrierStreamEmbryo.getStream()->setName("BitmapBarrierExecStream");

    // create the bitmap stream graph, with the load stream graph from
    // above as the source
    SharedExecStream pOutputStream = prepareDAG(
        clusterBarrierStreamEmbryo, splitterStreamEmbryo,
        createBitmapStreamList, barrierStreamEmbryo, true, false);

    // set up a generator which can produce the expected output
    RampExecStreamGenerator expectedResultGenerator(mockParams.nRows);

    verifyOutput(*pOutputStream, 1, expectedResultGenerator);
}

void LbmSearchTest::initBTreeExecStreamParam(
    BTreeExecStreamParams &param, shared_ptr<BTreeDescriptor> pBTreeDesc)
{
    param.scratchAccessor = pSegmentFactory->newScratchSegment(pCache, 15);
    param.pCacheAccessor = pCache;
    initBTreeParam(param, pBTreeDesc);
}

void LbmSearchTest::initBTreeParam(
    BTreeParams &param, shared_ptr<BTreeDescriptor> pBTreeDesc)
{
    param.pSegment = pRandomSegment;
    param.pRootMap = 0;
    param.rootPageIdParamId = DynamicParamId(0);

    pBTreeDesc->segmentAccessor.pSegment = param.pSegment;
    pBTreeDesc->segmentAccessor.pCacheAccessor = pCache;
    pBTreeDesc->tupleDescriptor = param.tupleDesc;
    pBTreeDesc->keyProjection = param.keyProj;
    param.pageOwnerId = pBTreeDesc->pageOwnerId;
    param.segmentId = pBTreeDesc->segmentId;
}

void LbmSearchTest::initClusterScanDef(
    LcsRowScanBaseExecStreamParams &rowScanParams,
    struct LcsClusterScanDef &clusterScanDef,
    uint bTreeIndex)
{
    clusterScanDef.pSegment =
        bTreeClusters[bTreeIndex]->segmentAccessor.pSegment;
    clusterScanDef.pCacheAccessor =
        bTreeClusters[bTreeIndex]->segmentAccessor.pCacheAccessor;
    clusterScanDef.tupleDesc = bTreeClusters[bTreeIndex]->tupleDescriptor;
    clusterScanDef.keyProj = bTreeClusters[bTreeIndex]->keyProjection;
    clusterScanDef.rootPageId = bTreeClusters[bTreeIndex]->rootPageId;
    clusterScanDef.pageOwnerId = bTreeClusters[bTreeIndex]->pageOwnerId;
    clusterScanDef.segmentId = bTreeClusters[bTreeIndex]->segmentId;
    clusterScanDef.pRootMap = 0;
    clusterScanDef.rootPageIdParamId = DynamicParamId(0);
    rowScanParams.lcsClusterScanDefs.push_back(clusterScanDef);
}

void LbmSearchTest::initBTreeBitmapDesc(
    TupleDescriptor &tupleDesc, TupleProjection &keyProj, uint nKeys)
{
    initBTreeTupleDesc(tupleDesc, nKeys);

    // btree key consists of the key columns and the start rid
    for (uint j = 0; j < nKeys + 1; j++) {
        keyProj.push_back(j);
    }
}

void LbmSearchTest::initBTreeTupleDesc(
    TupleDescriptor &tupleDesc, uint nKeys)
{
    for (uint i = 0; i < nKeys; i++) {
        tupleDesc.push_back(attrDesc_int64);
    }
    // add on the rid and bitmaps
    tupleDesc.push_back(bitmapTupleDesc[0]);
    tupleDesc.push_back(bitmapTupleDesc[1]);
    tupleDesc.push_back(bitmapTupleDesc[2]);
}

void LbmSearchTest::testCaseSetUp()
{
    LbmExecStreamTestBase::testCaseSetUp();

    attrDesc_char1 = TupleAttributeDescriptor(
        stdTypeFactory.newDataType(STANDARD_TYPE_CHAR), false, 1);
    attrDesc_nullableInt64 = TupleAttributeDescriptor(
        stdTypeFactory.newDataType(STANDARD_TYPE_INT_64),
        true, sizeof(uint64_t));
}

void LbmSearchTest::testCaseTearDown()
{
    for (uint i = 0; i < bTreeClusters.size(); i++) {
        bTreeClusters[i]->segmentAccessor.reset();
    }
    for (uint i = 0; i < bTreeBitmaps.size(); i++) {
        bTreeBitmaps[i]->segmentAccessor.reset();
    }
    bTreeClusters.clear();
    bTreeBitmaps.clear();
    savedBTreeClusterRootIds.clear();
    savedBTreeBitmapRootIds.clear();

    LbmExecStreamTestBase::testCaseTearDown();
}

void LbmSearchTest::testScanIdx(
    uint totalKeys, uint nKeys, uint bufSize,
    boost::shared_array<FixedBuffer> inputBuffer,
    uint expectedNBitmaps, PBuffer expectedBitmaps,
    bool dynamicRootPageId,
    bool useDynamicKeys,
    bool includeRid,
    const boost::scoped_array<uint64_t> &vals)
{
    resetExecStreamTest();

    // setup input into index scan; values stream will read tuples from
    // inputBuffer

    ValuesExecStreamParams valuesParams;
    for (uint i = 0; i < 2; i++) {
        valuesParams.outputTupleDesc.push_back(attrDesc_char1);
        for (uint j = 0; j < nKeys; j++) {
            valuesParams.outputTupleDesc.push_back(attrDesc_nullableInt64);
        }
    }
    valuesParams.pTupleBuffer = inputBuffer;
    valuesParams.bufSize = bufSize;

    ExecStreamEmbryo valuesStreamEmbryo;
    valuesStreamEmbryo.init(new ValuesExecStream(), valuesParams);
    valuesStreamEmbryo.getStream()->setName("ValuesExecStream");

    // setup index scan stream

    LbmSearchExecStreamParams indexScanParams;

    // initialize parameters specific to indexScan
    indexScanParams.rowLimitParamId = DynamicParamId(0);
    if (includeRid) {
        indexScanParams.startRidParamId = DynamicParamId(1);
        SharedDynamicParamManager pDynamicParamManager =
            pGraph->getDynamicParamManager();
        pDynamicParamManager->createParam(DynamicParamId(1), attrDesc_int64);
        TupleDatum ridDatum;
        LcsRid rid = LcsRid(0);
        ridDatum.pData = (PConstBuffer) &rid;
        ridDatum.cbData = sizeof(LcsRid);
        pDynamicParamManager->writeParam(DynamicParamId(1), ridDatum);
    } else {
        indexScanParams.startRidParamId = DynamicParamId(0);
    }

    // initialize parameters for btree read
    initBTreeBitmapDesc(
        indexScanParams.tupleDesc, indexScanParams.keyProj, totalKeys);
    initBTreeExecStreamParam(indexScanParams, bTreeBitmaps[0]);
    bTreeBitmaps[0]->rootPageId = savedBTreeBitmapRootIds[0];

    if (!dynamicRootPageId) {
        indexScanParams.rootPageId = savedBTreeBitmapRootIds[0];
    } else {
        indexScanParams.rootPageId = NULL_PAGE_ID;
        indexScanParams.rootPageIdParamId = DynamicParamId(2);
        SharedDynamicParamManager pDynamicParamManager =
            pGraph->getDynamicParamManager();
        pDynamicParamManager->createParam(DynamicParamId(2), attrDesc_int64);
        TupleDatum rootPageIdDatum;
        rootPageIdDatum.pData = (PConstBuffer) &(savedBTreeBitmapRootIds[0]);
        rootPageIdDatum.cbData = sizeof(PageId);
        pDynamicParamManager->writeParam(DynamicParamId(2), rootPageIdDatum);
    }

    TupleProjection outputProj;
    for (uint i = totalKeys; i < totalKeys + 3; i++) {
        outputProj.push_back(i);
    }
    indexScanParams.outputProj = outputProj;

    // initialize parameters for btree search
    indexScanParams.outerJoin = false;
    TupleProjection inputKeyProj;
    for (uint i = 0; i < 2; i++) {
        for (uint j = 0; j < nKeys; j++) {
            inputKeyProj.push_back(i * (nKeys + 1) + j + 1);
        }
    }
    indexScanParams.inputKeyProj = inputKeyProj;
    indexScanParams.inputDirectiveProj.push_back(0);
    indexScanParams.inputDirectiveProj.push_back(nKeys + 1);

    // output is bitmap btree tuple without the key values, but with the rid
    indexScanParams.outputTupleDesc = bitmapTupleDesc;

    if (useDynamicKeys) {
        SharedDynamicParamManager pDynamicParamManager =
            pGraph->getDynamicParamManager();
        for (uint i = 3; i < nKeys * 2 + 3; i++) {
            indexScanParams.searchKeyParams.push_back(
                BTreeSearchKeyParameter(
                    DynamicParamId(i),
                    i - 3));
            pDynamicParamManager->createParam(
                DynamicParamId(i), attrDesc_int64);
            TupleDatum keyValDatum;
            keyValDatum.pData = (PConstBuffer) &(vals[(i - 3) % nKeys]);
            keyValDatum.cbData = sizeof(uint64_t);
            pDynamicParamManager->writeParam(DynamicParamId(i), keyValDatum);
        }
    }

    ExecStreamEmbryo indexScanStreamEmbryo;
    indexScanStreamEmbryo.init(new LbmSearchExecStream(), indexScanParams);
    indexScanStreamEmbryo.getStream()->setName("IndexScanStream");

    SharedExecStream pOutputStream = prepareTransformGraph(
        valuesStreamEmbryo, indexScanStreamEmbryo);

    bitmapTupleAccessor.setCurrentTupleBuf(expectedBitmaps);
    verifyBufferedOutput(
        *pOutputStream, bitmapTupleDesc, expectedNBitmaps, expectedBitmaps);
}

FENNEL_UNIT_TEST_SUITE(LbmSearchTest);

// End LbmSearchTest.cpp


text: #include <boost/range/mfc.hpp>

ext: hpp

max_stars_repo_name: miathedev/BoostForArduino

max_issues_repo_name: miathedev/BoostForArduino

max_forks_repo_name: miathedev/BoostForArduino


text:
//
// Created by Paul Botros on 10/28/19.
//

#include <thread>
#include <regex>
#include <chrono>
#include <boost/filesystem.hpp>
#include <arrow/io/file.h>
#include <arrow/util/logging.h>
#include <arrow/api.h>
#include <parquet/arrow/writer.h>
#include <parquet/properties.h>
#include <parquet/exception.h>
#include <fmt/format.h>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/json_parser.hpp>
#include <memory>
#include <boost/bind.hpp>
#include <glog/logging.h>

#include <utility>

#include "ingester.h"

// Max number of samples per stream to query Redis. Keep this low in order to prevent interference of the ingester with
// other writers/readers of Redis (Redis is single-threaded and so a single big read can block for awhile).
static const int64_t SAMPLES_PER_READ = 32;

namespace river {

StreamIngester::StreamIngester(const RedisConnection &connection,
                               const string &output_directory,
                               bool *terminated,
                               const string& stream_filter,
                               int64_t samples_per_row_group,
                               int minimum_age_seconds_before_deletion,
                               int stalled_timeout_ms,
                               int stale_period_ms)
        : _connection(connection),
          _output_directory(output_directory),
          _terminated(terminated),
          _samples_per_row_group(samples_per_row_group),
          _minimum_age_seconds_before_deletion(minimum_age_seconds_before_deletion),
          _stalled_timeout_ms(stalled_timeout_ms),
          _stale_period_ms(stale_period_ms),
          _stream_filter(stream_filter) {
    // Create the output directory if necessary
    if (boost::filesystem::exists(output_directory)) {
        if (!boost::filesystem::is_directory(output_directory)) {
            throw StreamIngesterException(fmt::format("Non-directory filepath given: {}", output_directory));
        }
    } else {
        if (!boost::filesystem::create_directory(output_directory)) {
            throw StreamIngesterException(fmt::format("Failed to create directory {}", output_directory));
        }
    }

    this->_redis = internal::Redis::Create(connection);

    auto func = boost::bind(&StreamIngester::ingest_single, this, _1);
    _pool = make_unique<IngesterThreadPool<string, StreamIngestionResult>>(4, func);
}

void StreamIngester::Ingest() {
    auto stream_names = _redis->ListStreamNames(_stream_filter);

    if (stream_names.empty()) {
        LOG(INFO) << "No streams found to persist." << endl;
        return;
    }

    for (const auto &stream_name : stream_names) {
        bool did_enqueue = false;
        {
            lock_guard<mutex> lock(_streams_in_progress_mtx);
            if (_streams_in_progress.find(stream_name) == _streams_in_progress.end()) {
                LOG(INFO) << "Stream " << stream_name << " enqueued." << endl;
                _streams_in_progress.insert(stream_name);
                did_enqueue = true;
            }
        }

        if (did_enqueue) {
            _pool->enqueue_stream(stream_name);
        }
    }
}

void StreamIngester::Stop() {
    _pool->stop();
}

boost::optional<boost::variant<exception, StreamIngestionResult>>
StreamIngester::GetResult(const string &stream_name) {
    boost::variant<exception, StreamIngestionResult> ret;
    bool present = _pool->visit_result(stream_name,
                                       [&](const exception &e) {
                                           ret = e;
                                           LOG(ERROR) << "Exception thrown! " << e.what() << endl;
                                       },
                                       [&](const StreamIngestionResult &r) {
                                           ret = r;
                                       });
    if (!present) {
        return boost::none;
    } else {
        return ret;
    }
}

StreamIngestionResult StreamIngester::ingest_single(string stream_name) {
    if (*_terminated) {
        return StreamIngestionResult::IN_PROGRESS;
    }

    LOG(INFO) << "Starting ingestion of stream " << stream_name << " [output directory " << _output_directory << "]."
         << endl;

    try {
        auto ingester = internal::SingleStreamIngester(_connection,
                                                       stream_name,
                                                       _output_directory,
                                                       _terminated,
                                                       _samples_per_row_group,
                                                       _minimum_age_seconds_before_deletion,
                                                       _stalled_timeout_ms,
                                                       _stale_period_ms);
        auto ret = ingester.Ingest();
        {
            lock_guard<mutex> lock(_streams_in_progress_mtx);
            _streams_in_progress.erase(stream_name);
        }
        return ret;
    } catch (const exception& e) {
        {
            lock_guard<mutex> lock(_streams_in_progress_mtx);
            _streams_in_progress.erase(stream_name);
        }
        throw;
    }
}

namespace internal {
static shared_ptr<arrow::Schema> to_arrow(const StreamSchema &stream_schema);
template<class ArrayT>
static shared_ptr<ArrayT> get_last(const shared_ptr<arrow::Table> &table, const string &column_name, int *chunk_idx);

SingleStreamIngester::SingleStreamIngester(const RedisConnection &connection,
                                           const string &stream_name,
                                           const string &output_directory,
                                           bool *terminated,
                                           int64_t samples_per_row_group,
                                           int minimum_age_seconds_before_deletion,
                                           int stalled_timeout_ms,
                                           int stale_period_ms)
        : _connection(connection),
          _samples_per_row_group(samples_per_row_group),
          _minimum_age_seconds_before_deletion(minimum_age_seconds_before_deletion),
          _stalled_timeout_ms(stalled_timeout_ms),
          _stale_period_ms(stale_period_ms),
          stream_name_(stream_name),
          should_ingest(true),
          _terminated(terminated) {
    this->reader = std::make_unique<StreamReader>(connection);
    this->reader->Initialize(stream_name);
    auto local_schema = reader->schema();
    this->schema = make_unique<StreamSchema>(local_schema);

    parent_directory = boost::filesystem::path(output_directory) / boost::filesystem::path(stream_name);
    // Create the directory if necessary
    if (boost::filesystem::exists(parent_directory)) {
        if (!boost::filesystem::is_directory(parent_directory)) {
            throw StreamIngesterException(
                    fmt::format("Non-directory filepath given: {}", parent_directory.make_preferred().string()));
        }
    } else {
        if (!boost::filesystem::create_directory(parent_directory)) {
            throw StreamIngesterException(
                    fmt::format("Failed to create directory {}", parent_directory.make_preferred().string()));
        }
    }
}

template<class DataT, class BuilderT>
void write_to_numeric_array(const int64_t row_group_size,
                            const int within_sample_offset,
                            const int *sizes,
                            const char *read_buffer,
                            shared_ptr<arrow::Array> &array) {
    BuilderT builder;
    int64_t column_data_index = 0;
    for (int64_t j = 0; j < row_group_size; j++) {
        PARQUET_THROW_NOT_OK(builder.AppendValues(
                reinterpret_cast<const DataT *>(read_buffer + within_sample_offset + column_data_index),
                1));
        column_data_index += sizes[j];
    }
    PARQUET_THROW_NOT_OK(builder.Finish(&array));
}

StreamIngestionResult SingleStreamIngester::Ingest() {
    append_metadata(StreamIngestionResult::IN_PROGRESS);

    int sample_size = schema->sample_size();

    vector<int64_t> data_indices(_samples_per_row_group);
    vector<char> read_buffer(sample_size * _samples_per_row_group);
    vector<int> sizes(_samples_per_row_group);
    vector<string> keys(_samples_per_row_group);

    // Determine the next index for the file by looking at the current directory.
    int file_data_index;
    string last_key;
    int64_t global_offset;
    read_existing_files(&file_data_index, &last_key, &global_offset);

    // Seek the reader to the last key that we persisted, and continue from there.
    int64_t seek_ret = reader->Seek(last_key);
    if (seek_ret < 0) {
        throw StreamIngesterException(fmt::format("#seek() returned -1? For key {}", last_key));
    }

    StreamIngestionResult ingestion_status = StreamIngestionResult::IN_PROGRESS;
    while (should_ingest && !(*_terminated)) {
        LOG_EVERY_N(INFO, 10) << "New loop for stream " << stream_name_;
        int64_t row_group_size = 0;
        string eof_key;
        while (should_ingest && !(*_terminated) && row_group_size < _samples_per_row_group) {
            LOG_EVERY_N(INFO, 500) << "Fetching new samples. Size " << row_group_size
                                  << " for stream " << stream_name_;
            int64_t remaining_samples_in_row_group = _samples_per_row_group - row_group_size;
            auto samples_to_read = remaining_samples_in_row_group > SAMPLES_PER_READ ? SAMPLES_PER_READ
                                                                                : remaining_samples_in_row_group;

            int *sizes_ptr = &sizes[row_group_size];
            string *keys_ptr = &keys[row_group_size];
            int64_t num_read = reader->ReadBytes(&read_buffer[row_group_size * sample_size],
                                                 samples_to_read,
                                                 &sizes_ptr,
                                                 &keys_ptr,
                                                 static_cast<int>(_stalled_timeout_ms));
            if (num_read == 0) {
                LOG(INFO) << fmt::format("Stream {} has stalled; no responses after {} ms [file index {}].",
                                         stream_name_, _stalled_timeout_ms, file_data_index);
                should_ingest = false;
                break;
            } else if (num_read < 0) {
                should_ingest = false;
                ingestion_status = StreamIngestionResult::COMPLETED;
                eof_key = reader->eof_key();
                LOG(INFO) << fmt::format("EOF encountered in stream {}, num_read=0, global_offset={}",
                                         stream_name_,
                                         global_offset);
                break;
            }

            for (int64_t i = 0; i < num_read; i++) {
                data_indices[row_group_size + i] = global_offset + i;
            }

            row_group_size += num_read;
            global_offset += num_read;
        }

        if (row_group_size > 0) {
            const string &this_data_filepath = data_filepath(file_data_index);
            if (boost::filesystem::exists(this_data_filepath)) {
                LOG(INFO) << "Filepath " << this_data_filepath << " already exists. Refusing to overwrite any files.";
                throw StreamIngesterException(
                        fmt::format("Data file already exists; we will not overwrite. File={}", this_data_filepath));
            }

            LOG(INFO) << "Creating batch of length " << row_group_size << ". Total offset is now " << global_offset;

            std::vector<std::shared_ptr<arrow::Array>> arrays;

            // 1. Write sample_index
            arrow::Int64Builder sample_idx_builder;
            PARQUET_THROW_NOT_OK(sample_idx_builder.AppendValues(&data_indices.front(), row_group_size));
            std::shared_ptr<arrow::Array> sample_idx_array;
            PARQUET_THROW_NOT_OK(sample_idx_builder.Finish(&sample_idx_array));
            arrays.push_back(sample_idx_array);

            LOG(INFO) << "Successfully created sample_indices.";

            // 2. Write keys
            arrow::StringBuilder keys_builder;
            for (auto it = keys.begin(); it < keys.begin() + row_group_size; it++) {
                PARQUET_THROW_NOT_OK(keys_builder.Append(*it));
            }
            std::shared_ptr<arrow::Array> keys_array;
            PARQUET_THROW_NOT_OK(keys_builder.Finish(&keys_array));
            arrays.push_back(keys_array);
            LOG(INFO) << "Successfully created keys.";

            // 3. Write timestamp_ms (derived from keys)
            arrow::Int64Builder timestamps_builder;
            for (auto it = keys.begin(); it < keys.begin() + row_group_size; it++) {
                auto timestamp_ms = std::chrono::time_point_cast<std::chrono::milliseconds>(KeyTimestamp(it->c_str()));
                PARQUET_THROW_NOT_OK(timestamps_builder.Append(timestamp_ms.time_since_epoch().count()));
            }
            std::shared_ptr<arrow::Array> timestamps_array;
            PARQUET_THROW_NOT_OK(timestamps_builder.Finish(&timestamps_array));
            arrays.push_back(timestamps_array);
            LOG(INFO) << "Successfully created timestamps.";

            // 4. All the data fields given in the schema.
            int within_sample_offset = 0;
            for (const auto& field : schema->field_definitions) {
                std::shared_ptr<arrow::Array> column_array;
                switch (field.type) {
                    case FieldDefinition::DOUBLE: {
                        write_to_numeric_array<double, arrow::DoubleBuilder>(
                                row_group_size, within_sample_offset, &sizes.front(), &read_buffer.front(), column_array);
                        break;
                    }
                    case FieldDefinition::FLOAT: {
                        write_to_numeric_array<float, arrow::FloatBuilder>(
                                row_group_size, within_sample_offset, &sizes.front(), &read_buffer.front(), column_array);
                        break;
                    }
                    case FieldDefinition::INT32: {
                        write_to_numeric_array<int32_t, arrow::Int32Builder>(
                                row_group_size, within_sample_offset, &sizes.front(), &read_buffer.front(), column_array);
                        break;
                    }
                    case FieldDefinition::INT64: {
                        write_to_numeric_array<int64_t, arrow::Int64Builder>(
                                row_group_size, within_sample_offset, &sizes.front(), &read_buffer.front(), column_array);
                        break;
                    }
                    case FieldDefinition::FIXED_WIDTH_BYTES: {
                        arrow::FixedSizeBinaryBuilder builder(arrow::fixed_size_binary(field.size));
                        int64_t column_data_index = 0;
                        for (int64_t j = 0; j < row_group_size; j++) {
                            PARQUET_THROW_NOT_OK(builder.AppendValues(
                                    reinterpret_cast<uint8_t *>(&read_buffer[within_sample_offset + column_data_index]),
                                    1));
                            column_data_index += sizes[j];
                        }
                        PARQUET_THROW_NOT_OK(builder.Finish(&column_array));
                        break;
                    }
                    case FieldDefinition::VARIABLE_WIDTH_BYTES: {
                        arrow::BinaryBuilder builder;
                        int64_t column_data_index = 0;
                        for (int64_t j = 0; j < row_group_size; j++) {
                            PARQUET_THROW_NOT_OK(builder.Append(
                                    reinterpret_cast<const uint8_t *>(&read_buffer[within_sample_offset + column_data_index]),
                                    sizes[j]));
                            column_data_index += sizes[j];
                        }
                        PARQUET_THROW_NOT_OK(builder.Finish(&column_array));
                        break;
                    }
                    default:
                        throw StreamIngesterException("Unhandled field type");
                }

                within_sample_offset += field.size;
                arrays.push_back(column_array);
                LOG(INFO) << "Successfully created column array for field " << field.name;
            }

            auto arrow_schema = to_arrow(*this->schema);
            shared_ptr<arrow::Table> table = arrow::Table::Make(arrow_schema, arrays);

            boost::filesystem::path temp = parent_directory / boost::filesystem::unique_path();
            LOG(INFO) << "Writing file to temp filepath " << temp << "...";
            write_parquet_file(temp.make_preferred().string(), *table);
            LOG(INFO) << fmt::format("Successfully wrote to file path. Renaming temporary file {} to final path: {}",
                                     temp.make_preferred().string(), this_data_filepath);
            boost::filesystem::rename(temp, this_data_filepath);
            LOG(INFO) << fmt::format("Successfully moved temporary file {} to final path: {}",
                                     temp.make_preferred().string(), this_data_filepath);
            file_data_index++;
        }

        if (!eof_key.empty()) {
            combine_all_files();
            append_metadata(ingestion_status);
            delete_up_to(eof_key);
        } else if (row_group_size > 0) {
            string last_key_persisted = string(keys[row_group_size - 1]);
            delete_up_to(last_key_persisted);
        }
    }

    reader->Stop();

    if (ingestion_status == StreamIngestionResult::IN_PROGRESS && !(*_terminated)) {
        add_eof_if_necessary();
    }
    return ingestion_status;
}

void SingleStreamIngester::delete_up_to(const string& last_key_persisted) {
    class SeekListener : public internal::StreamReaderListener {
    public:
        vector <pair<string, string>> stream_keys_to_delete_for_live;
        string last_stream_key;

        void OnStreamKeyChange(const string &old_stream_key, const string &new_stream_key) override {
            if (!old_stream_key.empty()) {
                // For a live stream (i.e. no EOF ingested yet), we only want to delete streams that are wholly behind
                // the given key.
                stream_keys_to_delete_for_live.emplace_back(old_stream_key, new_stream_key);
            }
            last_stream_key = new_stream_key;
        };

        ~SeekListener() = default;
    };

    unique_ptr<StreamReader> seek_reader = make_unique<StreamReader>(_connection);
    auto listener = make_unique<SeekListener>();
    seek_reader->AddListener(listener.get());
    seek_reader->Initialize(stream_name_);
    int64_t seek_ret = seek_reader->Seek(last_key_persisted);
    bool is_eof = seek_ret < 0;

    vector<pair<string, string>> stream_keys_to_delete = listener->stream_keys_to_delete_for_live;
    if (is_eof) {
        stream_keys_to_delete.emplace_back(listener->last_stream_key, "");
    }

    if (stream_keys_to_delete.empty()) {
        LOG(INFO) << fmt::format("Nothing to delete for stream {} up to key {}", stream_name_, last_key_persisted);
        return;
    }

    auto elapsed_seconds = std::chrono::duration_cast<std::chrono::seconds>(
            std::chrono::system_clock::now() - KeyTimestamp(last_key_persisted.c_str()));
    if (elapsed_seconds.count() > _minimum_age_seconds_before_deletion) {
        long long to_sleep = _minimum_age_seconds_before_deletion - elapsed_seconds.count() + 1;
        if (to_sleep > 0) {
            LOG(INFO) << fmt::format(
                    "Sleeping for {} seconds until we can delete up to this key.", to_sleep);
            std::this_thread::sleep_for(chrono::seconds(to_sleep));
        }
    }

    unique_ptr<internal::Redis> redis = internal::Redis::Create(_connection);
    for (const pair<string, string> &to_del : stream_keys_to_delete) {
        const string &stream_key_to_del = to_del.first;
        const string &stream_key_following = to_del.second;

        if (!stream_key_following.empty()) {
            // Change the first_stream_key in the metadata to the one after this, so we always have an intact readable
            // stream.
            redis->SetMetadata(stream_name_, {{"first_stream_key", stream_key_following}});
            LOG(INFO) << "First_stream_key changed to " << stream_key_following << "." << endl;
        }

        redis->Unlink(stream_key_to_del);
        LOG(INFO) << "Stream key " << stream_key_to_del << " deleted." << endl;
    }

    if (is_eof) {
        redis->DeleteMetadata(stream_name_);
        LOG(INFO) << "Stream metadata for " << stream_name_ << " deleted." << endl;
    }
}

void SingleStreamIngester::add_eof_if_necessary() {
    class TailListener : public internal::StreamReaderListener {
    public:
        string last_stream_key;

        void OnStreamKeyChange(const string &, const string &new_stream_key) override {
            last_stream_key = new_stream_key;
        };

        ~TailListener() = default;
    };

    unique_ptr<StreamReader> tail_reader = make_unique<StreamReader>(_connection);
    auto listener = make_unique<TailListener>();
    tail_reader->AddListener(listener.get());
    tail_reader->Initialize(stream_name_);
    const int bufsize = tail_reader->schema().sample_size();
    vector<char> buf(bufsize);
    char key[512];
    int64_t sample_index;
    int64_t num_read = tail_reader->TailBytes(&buf.front(), 1000, key, &sample_index);
    if (num_read < 0) {
        // There's already an EOF, nothing to do here.
        return;
    }

    if (num_read == 0) {
        // We timed out OR we've consumed everything; can't tell.
        LOG(INFO) << "No elements read; cannot differentiate between an empty stream and repeated timeouts, so not doing "
                "anything with stream " << stream_name_ << endl;
        return;
    }

    uint64_t last_sample_written_at_ms, b;
    internal::DecodeCursor(key, &last_sample_written_at_ms, &b);

    unique_ptr<internal::Redis> redis = internal::Redis::Create(_connection);
    int64_t time_us = redis->TimeUs();

    int64_t elapsed_us = time_us - last_sample_written_at_ms * 1000;
    if (elapsed_us > _stale_period_ms * 1000) {
        auto reply = redis->Xadd(listener->last_stream_key, {
            {"eof", "1",},
            {"sample_index", fmt::format_int(sample_index).str()}});
        LOG(INFO) << fmt::format("Forcibly added an EOF to stream {} at key {} [stream was {} seconds old]",
                                 stream_name_, reply->str, elapsed_us / 1000000.0) << endl;
    }
}

vector<boost::filesystem::path> SingleStreamIngester::list_existing_files() {
    vector<boost::filesystem::path> paths;
    for (auto it = boost::filesystem::directory_iterator(parent_directory);
         it != boost::filesystem::directory_iterator();
         it++) {
        boost::filesystem::path path = it->path();
        string stem = path.stem().make_preferred().string();
        if (boost::starts_with(stem, "data_") && path.extension().compare(".parquet") == 0) {
            paths.push_back(path);
        }
    }
    return paths;
}

void SingleStreamIngester::combine_all_files() {
    vector<boost::filesystem::path> p = this->list_existing_files();

    if (p.empty()) {
        LOG(INFO) << fmt::format("No previous files found in directory {}. Nothing to do.",
                                 parent_directory.make_preferred().string()) << endl;
        return;
    }

    const string &this_data_filepath = combined_data_filepath();
    if (boost::filesystem::exists(this_data_filepath)) {
        LOG(INFO) << "Combined filepath " << this_data_filepath << " already exists. Refusing to overwrite any files.";
        throw StreamIngesterException(
                fmt::format("Combined file already exists; we will not overwrite. File={}", this_data_filepath));
    }

    string temp_filepath = (parent_directory / boost::filesystem::unique_path()).make_preferred().string();

    std::vector<std::shared_ptr<arrow::Array>> arrays;

    auto arrow_schema = to_arrow(*this->schema);
#ifdef PARQUET_ASSIGN_OR_THROW
    PARQUET_ASSIGN_OR_THROW(
            shared_ptr<arrow::io::OutputStream> sink, arrow::io::FileOutputStream::Open(temp_filepath));
#else
    shared_ptr<arrow::io::OutputStream> sink;
    PARQUET_THROW_NOT_OK(arrow::io::FileOutputStream::Open(temp_filepath, &sink));
#endif
    unique_ptr<parquet::arrow::FileWriter> writer;

    parquet::WriterProperties::Builder builder;
    builder.compression(parquet::Compression::SNAPPY);
    shared_ptr<parquet::WriterProperties> props = builder.build();

    // Open a writer
    PARQUET_THROW_NOT_OK(parquet::arrow::FileWriter::Open(
            *arrow_schema.get(),
            arrow::default_memory_pool(),
            sink,
            props,
            &writer));

    LOG(INFO) << "Beginning combining of files to temp file " << temp_filepath << endl;

    // Sort ascending order
    sort(p.begin(), p.end(), less<>());
    for (auto &path : p) {
        // Read the file on disk into a table:
#ifdef PARQUET_ASSIGN_OR_THROW
        PARQUET_ASSIGN_OR_THROW(
                std::shared_ptr<arrow::io::ReadableFile> infile,
                arrow::io::ReadableFile::Open(path.make_preferred().string(),
                                              arrow::default_memory_pool()));
#else
        std::shared_ptr<arrow::io::ReadableFile> infile;
            PARQUET_THROW_NOT_OK(arrow::io::ReadableFile::Open(
            path.make_preferred().string(),
            arrow::default_memory_pool(),
            &infile));
#endif

        std::unique_ptr<parquet::arrow::FileReader> file_reader;
        PARQUET_THROW_NOT_OK(
                parquet::arrow::OpenFile(infile, arrow::default_memory_pool(), &file_reader));
        std::shared_ptr<arrow::Table> table;
        PARQUET_THROW_NOT_OK(file_reader->ReadTable(&table));

        // And then write this table to the existing file
        int64_t num_rows = table->num_rows();
        LOG(INFO) << "Writing contents of " << path
                  << " to combined data filepath (" << num_rows << " rows)" << endl;
        PARQUET_THROW_NOT_OK(writer->WriteTable(*table.get(), num_rows));
        LOG(INFO) << "Done writing " << num_rows << " rows." << endl;
    }

    PARQUET_THROW_NOT_OK(writer->Close());

    LOG(INFO) << fmt::format("Renaming temporary file {} to final path: {}",
                             temp_filepath, this_data_filepath);
    boost::filesystem::rename(temp_filepath, this_data_filepath);
    LOG(INFO) << fmt::format("Successfully moved temporary file {} to final path: {}",
                             temp_filepath, this_data_filepath);

    // And then we can remove all old files
    for (auto &path : p) {
        boost::filesystem::remove(path);
        LOG(INFO) << fmt::format("Removed file {}", path);
    }
}

void SingleStreamIngester::read_existing_files(int *next_data_filepath_index, string *last_key, int64_t *global_index) {
    vector<boost::filesystem::path> p = this->list_existing_files();

    if (p.empty()) {
        *next_data_filepath_index = 0;
        *last_key = "0-0";
        *global_index = 0;
        LOG(INFO) << fmt::format("No previous files found in directory {}. Starting from the start.",
                            parent_directory.make_preferred().string()) << endl;
        return;
    }

    // Sort descending order
    sort(p.begin(), p.end(), greater<>());
    boost::filesystem::path last_path = p.at(0);
    string stem = last_path.stem().make_preferred().string();
    stem = regex_replace(stem, regex("^data_"), "");
    int64_t last_index = strtoll(stem.c_str(), nullptr, 10);
    *next_data_filepath_index = last_index + 1;

    // Read the last file on disk, and find the greatest "sample_index" it is at.
#ifdef PARQUET_ASSIGN_OR_THROW
    PARQUET_ASSIGN_OR_THROW(
            std::shared_ptr<arrow::io::ReadableFile> infile,
            arrow::io::ReadableFile::Open(last_path.make_preferred().string(),
                                          arrow::default_memory_pool()));
#else
    std::shared_ptr<arrow::io::ReadableFile> infile;
    PARQUET_THROW_NOT_OK(arrow::io::ReadableFile::Open(
            last_path.make_preferred().string(),
            arrow::default_memory_pool(),
            &infile));
#endif

    std::unique_ptr<parquet::arrow::FileReader> file_reader;
    PARQUET_THROW_NOT_OK(
            parquet::arrow::OpenFile(infile, arrow::default_memory_pool(), &file_reader));
    std::shared_ptr<arrow::Table> table;
    PARQUET_THROW_NOT_OK(file_reader->ReadTable(&table));

    int key_idx;
    auto key_array = get_last<arrow::StringArray>(table, "key", &key_idx);

    int sample_index_idx;
    auto sample_index_array = get_last<arrow::Int64Array>(table, "sample_index", &sample_index_idx);

    if (!key_array || !sample_index_array) {
        LOG(INFO) << fmt::format("No data found in the loaded table from file {}.", last_path.make_preferred().string());
        *last_key = "0-0";
        *global_index = 0;
        return;
    }

    *last_key = key_array->GetString(key_idx);
    *global_index = sample_index_array->Value(sample_index_idx) + 1;

    LOG(INFO) << fmt::format("Starting from existing files. last_key={}, global_index={}, found in filename {}. "
                        "new data file index {}",
                        *last_key,
                        *global_index,
                        last_path.make_preferred().string(),
                        *next_data_filepath_index) << endl;
}

template<class ArrayT>
inline shared_ptr<ArrayT> get_last(const shared_ptr<arrow::Table> &table, const string &column_name, int *idx) {
    auto key_column = table->GetColumnByName(column_name);
    if (key_column->length() == 0 || key_column->num_chunks() == 0) {
        *idx = 0;
        return nullptr;
    }

    auto last_chunk = key_column->chunk(key_column->num_chunks() - 1);
    auto array = std::static_pointer_cast<ArrayT>(last_chunk);
    *idx = array->length() - 1;
    return array;
}

shared_ptr<arrow::Schema> to_arrow(const StreamSchema &stream_schema) {
    vector<std::shared_ptr<arrow::Field>> fields;
    fields.push_back(arrow::field("sample_index", arrow::int64(), false));
    fields.push_back(arrow::field("key", arrow::utf8(), false));
    fields.push_back(arrow::field("timestamp_ms", arrow::int64(), false));

    for (auto &field : stream_schema.field_definitions) {
        shared_ptr<arrow::DataType> type;
        switch (field.type) {
            case FieldDefinition::DOUBLE:
                type = arrow::float64();
                break;
            case FieldDefinition::FLOAT:
                type = arrow::float32();
                break;
            case FieldDefinition::INT32:
                type = arrow::int32();
                break;
            case FieldDefinition::INT64:
                type = arrow::int64();
                break;
            case FieldDefinition::FIXED_WIDTH_BYTES:
                type = arrow::fixed_size_binary(field.size);
                break;
            case FieldDefinition::VARIABLE_WIDTH_BYTES:
                type = arrow::binary();
                break;
            default:
                throw StreamIngesterException("Unhandled data type! ");
        }
        fields.push_back(arrow::field(field.name, type, false));
    }

    return arrow::schema(fields);
}

void SingleStreamIngester::append_metadata(StreamIngestionResult result) {
    boost::property_tree::ptree root;
    const string &filename = metadata_filepath();
    if (boost::filesystem::exists(filename)) {
        boost::property_tree::read_json(filename, root);
    }

    for (const auto &it : reader->Metadata()) {
        root.put(it.first, it.second);
    }
    root.put("stream_name", stream_name_);
    root.put("local_minus_server_clock_us", std::to_string(reader->local_minus_server_clock_us()));
    root.put("initialized_at_us", std::to_string(reader->initialized_at_us()));

    string result_str;
    switch (result) {
        case IN_PROGRESS:
            result_str = "IN_PROGRESS";
            break;
        case COMPLETED:
            result_str = "COMPLETED";
            break;
    }
    root.put("ingestion_status", result_str);

    std::ofstream ofs(filename, std::ofstream::out);
    boost::property_tree::write_json(ofs, root);
}

string SingleStreamIngester::metadata_filepath() {
    return (parent_directory / boost::filesystem::path("metadata.json")).make_preferred().string();
}

string SingleStreamIngester::combined_data_filepath() {
    return (parent_directory / boost::filesystem::path("data.parquet")).make_preferred().string();
}

string SingleStreamIngester::data_filepath(int index) {
    return (parent_directory / boost::filesystem::path(fmt::format("data_{:0>4}.parquet", index)))
            .make_preferred().string();
}

void SingleStreamIngester::write_parquet_file(const string &filepath, const arrow::Table& table) {
    if (boost::filesystem::exists(filepath)) {
        throw StreamIngesterException(
                fmt::format("Data file already exists; we will not overwrite. File={}", filepath));
    }

#ifdef PARQUET_ASSIGN_OR_THROW
    PARQUET_ASSIGN_OR_THROW(
            shared_ptr<arrow::io::OutputStream> sink, arrow::io::FileOutputStream::Open(filepath));
#else
    shared_ptr<arrow::io::OutputStream> sink;
    PARQUET_THROW_NOT_OK(arrow::io::FileOutputStream::Open(filepath, &sink));
#endif

    parquet::WriterProperties::Builder builder;
    builder.compression(parquet::Compression::SNAPPY);
    shared_ptr<parquet::WriterProperties> props = builder.build();

    PARQUET_THROW_NOT_OK(
            parquet::arrow::WriteTable(table, arrow::default_memory_pool(), sink, 1024 * 1024 * 4, props));
    LOG(INFO) << "Successfully wrote table to file. filepath = " << filepath << endl;
}
}
}

max_stars_repo_name: mjsML/river

max_issues_repo_name: mjsML/river

max_forks_repo_name: mjsML/river


text:
#include <iostream>
#include <ios>
#include <fstream>
#include <map>
#include <algorithm>

#include <boost/filesystem.hpp>
#include <zeep/xml/document.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/format.hpp>
//#include <boost/timer/timer.hpp>
#include <boost/regex.hpp>

#include "M6Lib.h"
#include "M6File.h"
#include "M6Index.h"
#include "M6Tokenizer.h"
#include "M6Error.h"
#include "M6Lexicon.h"
#include "M6DocStore.h"
#include "M6Document.h"
#include "M6Databank.h"

#include <boost/test/unit_test.hpp>

using namespace std;
namespace fs = boost::filesystem;
namespace ba = boost::algorithm;
namespace io = boost::iostreams;

vector<string> testdocs;

BOOST_AUTO_TEST_CASE(test_store_0)
{
    cout << "testing document store (initialising)" << endl;

    ifstream text("test/pdbfind2-head.txt");
    BOOST_REQUIRE(text.is_open());

    stringstream doc;

    for (;;)
    {
        string line;
        getline(text, line);

        if (line.empty())
        {
            if (text.eof())
                break;
            continue;
        }

        doc << line << endl;

        if (line == "//")
        {
            testdocs.push_back(doc.str());
            doc.str("");
            doc.clear();
        }
    }
}

BOOST_AUTO_TEST_CASE(test_store_1)
{
    cout << "testing document store (store)" << endl;

    if (fs::exists("test/pdbfind2.docs"))
        fs::remove("test/pdbfind2.docs");

//    boost::timer::auto_cpu_timer t;

    M6DocStore store("test/pdbfind2.docs", eReadWrite);

    for (const string& doc : testdocs)
        store.StoreDocument(doc.c_str(), doc.length());
    store.Commit();

//    store.Dump();
    store.Validate();

    BOOST_CHECK_EQUAL(store.size(), testdocs.size());
}

BOOST_AUTO_TEST_CASE(test_store_2)
{
    cout << "testing document store (retrieve-1)" << endl;

//    boost::timer::auto_cpu_timer t;

    M6DocStore store("test/pdbfind2.docs", eReadOnly);

    store.Validate();

    for (uint32 i = 1; i <= testdocs.size(); ++i)
    {
        uint32 docPage, docSize;
        BOOST_CHECK(store.FetchDocument(i, docPage, docSize));

        io::filtering_stream<io::input> is;
        store.OpenDataStream(i, docPage, docSize, is);

        string docA;
        for (;;)
        {
            string line;
            getline(is, line);
            if (line.empty() and is.eof())
                break;
            docA += line + "\n";
        }

        string docB = testdocs[i - 1];

        BOOST_CHECK_EQUAL(docA.length(), docB.length());
        BOOST_CHECK_EQUAL(docA, docB);
    }
}

BOOST_AUTO_TEST_CASE(test_store_3)
{
    cout << "testing document store (retrieve-2 one line only)" << endl;

//    boost::timer::auto_cpu_timer t;

    M6DocStore store("test/pdbfind2.docs", eReadOnly);

    store.Validate();

    for (uint32 i = 1; i <= testdocs.size(); ++i)
    {
        uint32 docPage, docSize;
        BOOST_CHECK(store.FetchDocument(i, docPage, docSize));

        io::filtering_stream<io::input> is;
        store.OpenDataStream(i, docPage, docSize, is);

        string line;
        getline(is, line);
    }
}

BOOST_AUTO_TEST_CASE(test_store_4)
{
    cout << "testing document store (store using M6Databank)" << endl;

    boost::regex re("^ID\\s*:\\s+(.{4})");

//    boost::timer::auto_cpu_timer t;

    if (fs::exists("test/pdbfind2.m6"))
        fs::remove_all("test/pdbfind2.m6");

    M6Databank db("test/pdbfind2.m6", eReadWrite);

    M6Lexicon lexicon;
    db.StartBatchImport(lexicon);

    for (const string& text : testdocs)
    {
        M6InputDocument* doc = new M6InputDocument(db, text);

        boost::smatch m;
        BOOST_REQUIRE(boost::regex_search(text, m, re));

        string attr(m[1]);
        doc->SetAttribute("id", attr.c_str(), attr.length());

        db.Store(doc);
    }

    db.CommitBatchImport();

    db.Validate();

    BOOST_CHECK_EQUAL(db.size(), testdocs.size());
}

BOOST_AUTO_TEST_CASE(test_store_5)
{
    cout << "testing document store (retrieve using M6Databank)" << endl;

//    boost::timer::auto_cpu_timer t;

    M6Databank db("test/pdbfind2.m6", eReadOnly);

    db.Validate();

    BOOST_CHECK_EQUAL(db.size(), testdocs.size());

    for (uint32 i = 1; i <= testdocs.size(); ++i)
    {
        M6Document* doc = db.Fetch(i);

        BOOST_CHECK(doc != nullptr);

        if (doc == nullptr)
            continue;

        BOOST_CHECK_EQUAL(doc->GetText(), testdocs[i - 1]);

        delete doc;
    }
}

BOOST_AUTO_TEST_CASE(test_store_6)
{
    cout << "testing document store (retrieve attribute)" << endl;

    boost::regex re("^ID\\s*:\\s+(.{4})");

//    boost::timer::auto_cpu_timer t;

    M6Databank db("test/pdbfind2.m6", eReadOnly);
    BOOST_CHECK_EQUAL(db.size(), testdocs.size());

    for (uint32 i = 1; i <= testdocs.size(); ++i)
    {
        M6Document* doc = db.Fetch(i);

        BOOST_CHECK(doc != nullptr);

        if (doc == nullptr)
            continue;

        boost::smatch m;
        BOOST_REQUIRE(boost::regex_search(testdocs[i - 1], m, re));
        BOOST_CHECK_EQUAL(doc->GetAttribute("id"), m[1]);

        delete doc;
    }
}

BOOST_AUTO_TEST_CASE(test_store_7)
{
    cout << "dumping pdbfinder" << endl;

//    boost::timer::auto_cpu_timer t;

    M6Databank db("test/pdbfinder.m6", eReadOnly);
    uint32 size = db.size();

    for (uint32 i = 1; i <= size; ++i)
    {
        M6Document* doc = db.Fetch(i);

        BOOST_CHECK(doc != nullptr);

        if (doc == nullptr)
            continue;

        (void)doc->GetAttribute("id");
        (void)doc->GetAttribute("title");

        delete doc;
    }
}

max_stars_repo_name: mhekkel/mrs

max_issues_repo_name: mhekkel/mrs

max_forks_repo_name: mhekkel/mrs


text:
#pragma once
#include <Eigen/Dense>
#include <unsupported/Eigen/KroneckerProduct>

#include "yavque/utils.hpp"
Eigen::VectorXcd product_state(uint32_t n, const Eigen::VectorXcd& s)
{
	Eigen::VectorXcd res(1);
	res(0) = 1.0;
	for(uint32_t i = 0; i < n; i++)
	{
		res = Eigen::kroneckerProduct(res, s).eval();
	}
	return res;
}

template<typename T>
Eigen::VectorXcd
apply_kronecker(uint32_t N, const Eigen::Matrix<T, Eigen::Dynamic, Eigen::Dynamic>& m,
                const Eigen::VectorXcd& vec)
{
	Eigen::VectorXcd res = vec;
	for(uint32_t k = 0; k < N; ++k)
	{
		res = yavque::apply_single_qubit(res, m, k);
	}
	return res;
}

template<typename RandomEngine>
std::pair<uint32_t, uint32_t> random_connection(const int N, RandomEngine& re)
{
	std::uniform_int_distribution<uint32_t> uid1(0, N - 1);
	std::uniform_int_distribution<uint32_t> uid2(0, N - 2);

	auto r1 = uid1(re);
	auto r2 = uid2(re);

	if(r2 < r1)
		return std::make_pair(r1, r2);
	else
		return std::make_pair(r1, r2 + 1);
}

template<typename RandomEngine>
Eigen::MatrixXcd random_unitary(uint32_t dim, RandomEngine&& re)
{
	constexpr yavque::cx_double I(0.0, 1.0);
	std::normal_distribution<double> ndist;

	Eigen::MatrixXcd m(dim, dim);
	for(uint32_t i = 0; i < dim; ++i)
	{
		for(uint32_t j = 0; j < dim; ++j)
		{
			m(i, j) = ndist(re) + I * ndist(re);
		}
	}
	Eigen::HouseholderQR<Eigen::MatrixXcd> qr(m);
	return qr.householderQ();
}

template<typename RandomEngine>
Eigen::VectorXcd random_vector(uint32_t dim, RandomEngine&& re)
{
	constexpr yavque::cx_double I(0.0, 1.0);
	std::normal_distribution<double> ndist;

	Eigen::VectorXcd res(dim);
	for(uint32_t k = 0; k < dim; ++k)
	{
		res(k) = ndist(re) + I * ndist(re);
	}
	res.normalize();
	return res;
}

max_stars_repo_name: chaeyeunpark/Yavque

max_issues_repo_name: chaeyeunpark/Yavque

max_forks_repo_name: chaeyeunpark/Yavque


text:
/** \file PerfTools/EdmEvent/interface/EdmEventSize.cc
 *
 *  \author Vincenzo Innocente
 */
#include "PerfTools/EdmEvent/interface/EdmEventSize.h"
#include <valarray>
#include <functional>
#include <algorithm>
#include <boost/bind.hpp>
#include <ostream>
#include <limits>
#include <cassert>

#include "Rtypes.h"
#include "TROOT.h"
#include "TFile.h"
#include "TTree.h"
#include "TStyle.h"
#include "TObjArray.h"
#include "TBranch.h"
#include "TH1.h"
#include "TCanvas.h"
#include "Riostream.h"

#include "TBufferFile.h"

namespace {

  enum Indices { kUncompressed, kCompressed };

  typedef std::valarray<Long64_t> size_type;

  size_type getBasketSize(TBranch*);

  size_type getBasketSize(TObjArray* branches) {
    size_type result(static_cast<Long64_t>(0), 2);
    size_t n = branches->GetEntries();
    for (size_t i = 0; i < n; ++i) {
      TBranch* b = dynamic_cast<TBranch*>(branches->At(i));
      assert(b != nullptr);
      result += getBasketSize(b);
    }
    return result;
  }

  size_type getBasketSize(TBranch* b) {
    size_type result(static_cast<Long64_t>(0), 2);
    if (b != nullptr) {
      if (b->GetZipBytes() > 0) {
        result[kUncompressed] = b->GetTotBytes();
        result[kCompressed] = b->GetZipBytes();
      } else {
        result[kUncompressed] = b->GetTotalSize();
        result[kCompressed] = b->GetTotalSize();
      }
      result += getBasketSize(b->GetListOfBranches());
    }
    return result;
  }

  size_type getTotalSize(TBranch* br) {
    TBufferFile buf(TBuffer::kWrite, 10000);
    TBranch::Class()->WriteBuffer(buf, br);
    size_type size = getBasketSize(br);
    if (br->GetZipBytes() > 0)
      size[kUncompressed] += buf.Length();
    return size;
  }
}  // namespace

namespace perftools {

  EdmEventSize::EdmEventSize() : m_nEvents(0) {}

  EdmEventSize::EdmEventSize(std::string const& fileName, std::string const& treeName) : m_nEvents(0) {
    parseFile(fileName);
  }

  void EdmEventSize::parseFile(std::string const& fileName, std::string const& treeName) {
    m_fileName = fileName;
    m_branches.clear();

    TFile* file = TFile::Open(fileName.c_str());
    if (file == nullptr || (!(*file).IsOpen()))
      throw Error("unable to open data file " + fileName, 7002);

    TObject* o = file->Get(treeName.c_str());
    if (o == nullptr)
      throw Error("no object \"" + treeName + "\" found in file: " + fileName, 7003);

    TTree* events = dynamic_cast<TTree*>(o);
    if (events == nullptr)
      throw Error("object \"" + treeName + "\" is not a TTree in file: " + fileName, 7004);

    m_nEvents = events->GetEntries();
    if (m_nEvents == 0)
      throw Error("tree \"" + treeName + "\" in file " + fileName + " contains no Events", 7005);

    TObjArray* branches = events->GetListOfBranches();
    if (branches == nullptr)
      throw Error("tree \"" + treeName + "\" in file " + fileName + " contains no branches", 7006);

    const size_t n = branches->GetEntries();
    m_branches.reserve(n);
    for (size_t i = 0; i < n; ++i) {
      TBranch* b = dynamic_cast<TBranch*>(branches->At(i));
      if (b == nullptr)
        continue;
      std::string const name(b->GetName());
      if (name == "EventAux")
        continue;
      size_type s = getTotalSize(b);
      m_branches.push_back(
          BranchRecord(name, double(s[kCompressed]) / double(m_nEvents), double(s[kUncompressed]) / double(m_nEvents)));
    }
    std::sort(m_branches.begin(),
              m_branches.end(),
              boost::bind(std::greater<double>(),
                          boost::bind(&BranchRecord::compr_size, _1),
                          boost::bind(&BranchRecord::compr_size, _2)));
  }

  void EdmEventSize::sortAlpha() {
    std::sort(
        m_branches.begin(),
        m_branches.end(),
        boost::bind(
            std::less<std::string>(), boost::bind(&BranchRecord::name, _1), boost::bind(&BranchRecord::name, _2)));
  }

  namespace detail {
    // format as product:label (type)
    void shorterName(EdmEventSize::BranchRecord& br) {
      size_t b = br.fullName.find('_');
      size_t e = br.fullName.rfind('_');
      if (b == e)
        br.name = br.fullName;
      else {
        // remove type and process
        br.name = br.fullName.substr(b + 1, e - b - 1);
        // change label separator in :
        e = br.name.rfind('_');
        if (e != std::string::npos)
          br.name.replace(e, 1, ":");
        // add the type name
        br.name.append(" (" + br.fullName.substr(0, b) + ")");
      }
    }

  }  // namespace detail

  void EdmEventSize::formatNames() { std::for_each(m_branches.begin(), m_branches.end(), &detail::shorterName); }

  namespace detail {

    void dump(std::ostream& co, EdmEventSize::BranchRecord const& br) {
      co << br.name << " " << br.uncompr_size << " " << br.compr_size << "\n";
    }
  }  // namespace detail

  void EdmEventSize::dump(std::ostream& co, bool header) const {
    if (header) {
      co << "File " << m_fileName << " Events " << m_nEvents << "\n";
      co << "Branch Name | Average Uncompressed Size (Bytes/Event) | Average Compressed Size (Bytes/Event) \n";
    }
    std::for_each(m_branches.begin(), m_branches.end(), boost::bind(detail::dump, boost::ref(co), _1));
  }

  namespace detail {

    struct Hist {
      explicit Hist(int itop)
          : top(itop),
            uncompressed("uncompressed", "branch sizes", top, -0.5, -0.5 + top),
            compressed("compressed", "branch sizes", top, -0.5, -0.5 + top),
            cxAxis(compressed.GetXaxis()),
            uxAxis(uncompressed.GetXaxis()),
            x(0) {}

      void fill(EdmEventSize::BranchRecord const& br) {
        if (x < top) {
          cxAxis->SetBinLabel(x + 1, br.name.c_str());
          uxAxis->SetBinLabel(x + 1, br.name.c_str());
          compressed.Fill(x, br.compr_size);
          uncompressed.Fill(x, br.uncompr_size);
          x++;
        }
      }

      void finalize() {
        double mn = std::numeric_limits<double>::max();
        for (int i = 1; i <= top; ++i) {
          double cm = compressed.GetMinimum(i), um = uncompressed.GetMinimum(i);
          if (cm > 0 && cm < mn)
            mn = cm;
          if (um > 0 && um < mn)
            mn = um;
        }
        mn *= 0.8;
        double mx = std::max(compressed.GetMaximum(), uncompressed.GetMaximum());
        mx *= 1.2;
        uncompressed.SetMinimum(mn);
        uncompressed.SetMaximum(mx);
        compressed.SetMinimum(mn);
        //  compressed.SetMaximum( mx );
        cxAxis->SetLabelOffset(-0.32);
        cxAxis->LabelsOption("v");
        cxAxis->SetLabelSize(0.03);
        uxAxis->SetLabelOffset(-0.32);
        uxAxis->LabelsOption("v");
        uxAxis->SetLabelSize(0.03);
        compressed.GetYaxis()->SetTitle("Bytes");
        compressed.SetFillColor(kBlue);
        compressed.SetLineWidth(2);
        uncompressed.GetYaxis()->SetTitle("Bytes");
        uncompressed.SetFillColor(kRed);
        uncompressed.SetLineWidth(2);
      }

      int top;
      TH1F uncompressed;
      TH1F compressed;
      TAxis* cxAxis;
      TAxis* uxAxis;

      int x;
    };

  }  // namespace detail

  void EdmEventSize::produceHistos(std::string const& plot, std::string const& file, int top) const {
    if (top == 0)
      top = m_branches.size();
    detail::Hist h(top);
    std::for_each(m_branches.begin(), m_branches.end(), boost::bind(&detail::Hist::fill, boost::ref(h), _1));
    h.finalize();
    if (!plot.empty()) {
      gROOT->SetStyle("Plain");
      gStyle->SetOptStat(kFALSE);
      gStyle->SetOptLogy();
      TCanvas c;
      h.uncompressed.Draw();
      h.compressed.Draw("same");
      c.SaveAs(plot.c_str());
    }
    if (!file.empty()) {
      TFile f(file.c_str(), "RECREATE");
      h.compressed.Write();
      h.uncompressed.Write();
      f.Close();
    }
  }

}  // namespace perftools

ext: cc

max_stars_repo_name: NTrevisani/cmssw

max_issues_repo_name: NTrevisani/cmssw

max_forks_repo_name: NTrevisani/cmssw


text:
#include <voltu/types/intrinsics.hpp>
#include <Eigen/Eigen>

Eigen::Matrix3d voltu::Intrinsics::matrix() {
	Eigen::Matrix3d K;
	K <<		focal_x,			0.0,	-principle_x,
					0.0,		focal_y,	-principle_y,
					0.0,			0.0,			 1.0;
	return K;
}

Eigen::Vector2i voltu::Intrinsics::size() {
	return { width, height };
}

max_stars_repo_name: knicos/voltu

max_issues_repo_name: knicos/voltu

max_forks_repo_name: knicos/voltu


text:
    /**
  *********************************************************************
*************************************************************************
***
*** \file  GizmoEventSoundVisualization.cpp
*** \brief GizmoEventSoundVisualization class body
***
*****************************************
  *****************************************
    **/

/*

  Copyright (c) 2007, Tim Burrell
  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.

*/

#include "GizmoEventSoundVisualization.hpp"
#include "../libH/Debug.hpp"
#include "../libH/Exception.hpp"
#include <boost/shared_ptr.hpp>

using namespace std;
using namespace boost;
using namespace H;
using namespace Gizmod;

////////////////////////////////////////////////////////////////////////////
// Type Defs
///////////////////////////////////////

////////////////////////////////////////////////////////////////////////////
// Construction
///////////////////////////////////////

/**
 * \brief GizmoEventSoundVisualization Default Constructor
**/
GizmoEventSoundVisualization::GizmoEventSoundVisualization() : GizmoEvent(GIZMO_EVENTCLASS_SOUNDVISUALIZATION, false) {
	mVULeft = mVURight = mVUCombined = 0;
	mType = SOUNDVISUALIZATION_RENDER;
}

/**
 * \brief GizmoEventSoundVisualization Init Constructor
**/
GizmoEventSoundVisualization::GizmoEventSoundVisualization(float VULeft, float VURight, float VUCombined, bool IsRemote) : GizmoEvent(GIZMO_EVENTCLASS_SOUNDVISUALIZATION, IsRemote) {
	mVULeft = VULeft;
	mVURight = VURight;
	mVUCombined = VUCombined;
	mType = SOUNDVISUALIZATION_RENDER;
}

/**
 * \brief GizmoEventSoundVisualization Init Constructor
**/
GizmoEventSoundVisualization::GizmoEventSoundVisualization(SoundVisualizationEventType Type, bool IsRemote) : GizmoEvent(GIZMO_EVENTCLASS_SOUNDVISUALIZATION, IsRemote) {
	mVULeft = mVURight = mVUCombined = 0;
	mType = Type;
}

/**
 * \brief GizmoEventSoundVisualization Destructor
**/
GizmoEventSoundVisualization::~GizmoEventSoundVisualization() {
}

////////////////////////////////////////////////////////////////////////////
// Class Body
///////////////////////////////////////

/**
 * \brief  Get VU data for the left channel
 * \return The VU data
**/
float GizmoEventSoundVisualization::getVULeft() {
	return mVULeft;
}

/**
 * \brief  Get VU data for the right channel
 * \return The VU data
**/
float GizmoEventSoundVisualization::getVURight() {
	return mVURight;
}

/**
 * \brief  Get VU data for both channels
 * \return The VU data
**/
float GizmoEventSoundVisualization::getVUCombined() {
	return mVUCombined;
}

/**
 * \brief  Get VU data for both channels
 * \return The VU data
**/
SoundVisualizationEventType GizmoEventSoundVisualization::getType() {
	return mType;
}

max_stars_repo_name: pauljeremyturner/gizmod

max_issues_repo_name: pauljeremyturner/gizmod

max_forks_repo_name: pauljeremyturner/gizmod