text:
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF 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 <boost/scoped_ptr.hpp>
#include <boost/bind.hpp>
#include <boost/filesystem.hpp>

#include <set>
#include <string>
#include <limits> // for std::numeric_limits<int>::max()

#include "testutil/gtest-util.h"
#include "codegen/llvm-codegen.h"
#include "gutil/gscoped_ptr.h"
#include "runtime/buffered-tuple-stream.inline.h"
#include "runtime/collection-value.h"
#include "runtime/collection-value-builder.h"
#include "runtime/raw-value.h"
#include "runtime/row-batch.h"
#include "runtime/string-value.inline.h"
#include "runtime/test-env.h"
#include "runtime/tmp-file-mgr.h"
#include "service/fe-support.h"
#include "testutil/desc-tbl-builder.h"
#include "util/test-info.h"

#include "gen-cpp/Types_types.h"
#include "gen-cpp/ImpalaInternalService_types.h"

#include "common/names.h"

using kudu::FreeDeleter;

static const int BATCH_SIZE = 250;
static const int IO_BLOCK_SIZE = 8 * 1024 * 1024;
static const uint32_t PRIME = 479001599;

namespace impala {

static const StringValue STRINGS[] = {
  StringValue("ABC"),
  StringValue("HELLO"),
  StringValue("123456789"),
  StringValue("FOOBAR"),
  StringValue("ONE"),
  StringValue("THREE"),
  StringValue("abcdefghijklmno"),
  StringValue("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
  StringValue("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
};

static const int NUM_STRINGS = sizeof(STRINGS) / sizeof(StringValue);

class SimpleTupleStreamTest : public testing::Test {
 protected:
  virtual void SetUp() {
    test_env_.reset(new TestEnv());
    ASSERT_OK(test_env_->Init());

    CreateDescriptors();

    mem_pool_.reset(new MemPool(&tracker_));
  }

  virtual void CreateDescriptors() {
    vector<bool> nullable_tuples(1, false);
    vector<TTupleId> tuple_ids(1, static_cast<TTupleId>(0));

    DescriptorTblBuilder int_builder(test_env_->exec_env()->frontend(), &pool_);
    int_builder.DeclareTuple() << TYPE_INT;
    int_desc_ = pool_.Add(new RowDescriptor(
        *int_builder.Build(), tuple_ids, nullable_tuples));

    DescriptorTblBuilder string_builder(test_env_->exec_env()->frontend(), &pool_);
    string_builder.DeclareTuple() << TYPE_STRING;
    string_desc_ = pool_.Add(new RowDescriptor(
        *string_builder.Build(), tuple_ids, nullable_tuples));
  }

  virtual void TearDown() {
    runtime_state_ = NULL;
    client_ = NULL;
    pool_.Clear();
    mem_pool_->FreeAll();
    test_env_.reset();
  }

  /// Setup a block manager with the provided settings and client with no reservation,
  /// tracked by tracker_.
  void InitBlockMgr(int64_t limit, int block_size) {
    ASSERT_OK(test_env_->CreateQueryStateWithBlockMgr(
        0, limit, block_size, nullptr, &runtime_state_));
    MemTracker* client_tracker =
        pool_.Add(new MemTracker(-1, "client", runtime_state_->instance_mem_tracker()));
    ASSERT_OK(runtime_state_->block_mgr()->RegisterClient(
        "", 0, false, client_tracker, runtime_state_, &client_));
  }

  /// Generate the ith element of a sequence of int values.
  int GenIntValue(int i) {
    // Multiply by large prime to get varied bit patterns.
    return i * PRIME;
  }

  /// Generate the ith element of a sequence of bool values.
  bool GenBoolValue(int i) {
    // Use a middle bit of the int value.
    return ((GenIntValue(i) >> 8) & 0x1) != 0;
  }

  /// Count the total number of slots per row based on the given row descriptor.
  int CountSlotsPerRow(const RowDescriptor& row_desc) {
    int slots_per_row = 0;
    for (int i = 0; i < row_desc.tuple_descriptors().size(); ++i) {
      TupleDescriptor* tuple_desc = row_desc.tuple_descriptors()[i];
      slots_per_row += tuple_desc->slots().size();
    }
    return slots_per_row;
  }

  /// Allocate a row batch with 'num_rows' of rows with layout described by 'row_desc'.
  /// 'offset' is used to account for rows occupied by any previous row batches. This is
  /// needed to match the values generated in VerifyResults(). If 'gen_null' is true,
  /// some tuples will be set to NULL.
  virtual RowBatch* CreateBatch(
      const RowDescriptor* row_desc, int offset, int num_rows, bool gen_null) {
    RowBatch* batch = pool_.Add(new RowBatch(row_desc, num_rows, &tracker_));
    int num_tuples = row_desc->tuple_descriptors().size();

    int idx = offset * CountSlotsPerRow(*row_desc);
    for (int row_idx = 0; row_idx < num_rows; ++row_idx) {
      TupleRow* row = batch->GetRow(row_idx);
      for (int tuple_idx = 0; tuple_idx < num_tuples; ++tuple_idx) {
        TupleDescriptor* tuple_desc = row_desc->tuple_descriptors()[tuple_idx];
        Tuple* tuple = Tuple::Create(tuple_desc->byte_size(), batch->tuple_data_pool());
        bool is_null = gen_null && !GenBoolValue(idx);
        for (int slot_idx = 0; slot_idx < tuple_desc->slots().size(); ++slot_idx, ++idx) {
          SlotDescriptor* slot_desc = tuple_desc->slots()[slot_idx];
          void* slot = tuple->GetSlot(slot_desc->tuple_offset());
          switch (slot_desc->type().type) {
            case TYPE_INT:
              *reinterpret_cast<int*>(slot) = GenIntValue(idx);
              break;
            case TYPE_STRING:
              *reinterpret_cast<StringValue*>(slot) = STRINGS[idx % NUM_STRINGS];
              break;
            default:
              // The memory has been zero'ed out already by Tuple::Create().
              break;
          }
        }
        if (is_null) {
          row->SetTuple(tuple_idx, NULL);
        } else {
          row->SetTuple(tuple_idx, tuple);
        }
      }
      batch->CommitLastRow();
    }
    return batch;
  }

  virtual RowBatch* CreateIntBatch(int offset, int num_rows, bool gen_null) {
    return CreateBatch(int_desc_, offset, num_rows, gen_null);
  }

  virtual RowBatch* CreateStringBatch(int offset, int num_rows, bool gen_null) {
    return CreateBatch(string_desc_, offset, num_rows, gen_null);
  }

  void AppendValue(uint8_t* ptr, vector<int>* results) {
    if (ptr == NULL) {
      // For the tests indicate null-ability using the max int value
      results->push_back(std::numeric_limits<int>::max());
    } else {
      results->push_back(*reinterpret_cast<int*>(ptr));
    }
  }

  void AppendValue(uint8_t* ptr, vector<StringValue>* results) {
    if (ptr == NULL) {
      results->push_back(StringValue());
    } else {
      StringValue sv = *reinterpret_cast<StringValue*>(ptr);
      uint8_t* copy = mem_pool_->Allocate(sv.len);
      memcpy(copy, sv.ptr, sv.len);
      sv.ptr = reinterpret_cast<char*>(copy);
      results->push_back(sv);
    }
  }

  template <typename T>
  void AppendRowTuples(TupleRow* row, RowDescriptor* row_desc, vector<T>* results) {
    DCHECK(row != NULL);
    const int num_tuples = row_desc->tuple_descriptors().size();

    for (int tuple_idx = 0; tuple_idx < num_tuples; ++tuple_idx) {
      TupleDescriptor* tuple_desc = row_desc->tuple_descriptors()[tuple_idx];
      Tuple* tuple = row->GetTuple(tuple_idx);
      const int num_slots = tuple_desc->slots().size();
      for (int slot_idx = 0; slot_idx < num_slots; ++slot_idx) {
        SlotDescriptor* slot_desc = tuple_desc->slots()[slot_idx];
        if (tuple == NULL) {
          AppendValue(NULL, results);
        } else {
          void* slot = tuple->GetSlot(slot_desc->tuple_offset());
          AppendValue(reinterpret_cast<uint8_t*>(slot), results);
        }
      }
    }
  }

  template <typename T>
  void ReadValues(BufferedTupleStream* stream, RowDescriptor* desc, vector<T>* results,
      int num_batches = -1) {
    bool eos = false;
    RowBatch batch(desc, BATCH_SIZE, &tracker_);
    int batches_read = 0;
    do {
      batch.Reset();
      EXPECT_OK(stream->GetNext(&batch, &eos));
      ++batches_read;
      for (int i = 0; i < batch.num_rows(); ++i) {
        AppendRowTuples(batch.GetRow(i), desc, results);
      }
    } while (!eos && (num_batches < 0 || batches_read <= num_batches));
  }

  void GetExpectedValue(int idx, bool is_null, int* val) {
    if (is_null) {
      *val = std::numeric_limits<int>::max();
    } else {
      *val = GenIntValue(idx);
    }
  }

  void GetExpectedValue(int idx, bool is_null, StringValue* val) {
    if (is_null) {
      *val = StringValue();
    } else {
      *val = STRINGS[idx % NUM_STRINGS];
    }
  }

  template <typename T>
  void VerifyResults(const RowDescriptor& row_desc, const vector<T>& results,
      int num_rows, bool gen_null) {
    int idx = 0;
    for (int row_idx = 0; row_idx < num_rows; ++row_idx) {
      const int num_tuples = row_desc.tuple_descriptors().size();
      for (int tuple_idx = 0; tuple_idx < num_tuples; ++tuple_idx) {
        const TupleDescriptor* tuple_desc = row_desc.tuple_descriptors()[tuple_idx];
        const int num_slots = tuple_desc->slots().size();
        bool is_null = gen_null && !GenBoolValue(idx);
        for (int slot_idx = 0; slot_idx < num_slots; ++slot_idx, ++idx) {
          T expected_val;
          GetExpectedValue(idx, is_null, &expected_val);
          ASSERT_EQ(results[idx], expected_val)
              << "results[" << idx << "] " << results[idx] << " != "
              << expected_val << " row_idx=" << row_idx
              << " tuple_idx=" << tuple_idx << " slot_idx=" << slot_idx
              << " gen_null=" << gen_null;
        }
      }
    }
    DCHECK_EQ(results.size(), idx);
  }

  // Test adding num_batches of ints to the stream and reading them back.
  // If unpin_stream is true, operate the stream in unpinned mode.
  // Assumes that enough buffers are available to read and write the stream.
  template <typename T>
  void TestValues(int num_batches, RowDescriptor* desc, bool gen_null,
      bool unpin_stream, int num_rows = BATCH_SIZE, bool use_small_buffers = true) {
    BufferedTupleStream stream(runtime_state_, desc, runtime_state_->block_mgr(), client_,
        use_small_buffers, false);
    ASSERT_OK(stream.Init(-1, NULL, true));
    bool got_write_buffer;
    ASSERT_OK(stream.PrepareForWrite(&got_write_buffer));
    ASSERT_TRUE(got_write_buffer);

    if (unpin_stream) {
      ASSERT_OK(stream.UnpinStream(BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT));
    }
    // Add rows to the stream
    int offset = 0;
    for (int i = 0; i < num_batches; ++i) {
      RowBatch* batch = NULL;

      Status status;
      ASSERT_TRUE(sizeof(T) == sizeof(int) || sizeof(T) == sizeof(StringValue));
      batch = CreateBatch(desc, offset, num_rows, gen_null);
      for (int j = 0; j < batch->num_rows(); ++j) {
        bool b = stream.AddRow(batch->GetRow(j), &status);
        ASSERT_OK(status);
        if (!b) {
          ASSERT_TRUE(stream.using_small_buffers());
          bool got_buffer;
          ASSERT_OK(stream.SwitchToIoBuffers(&got_buffer));
          ASSERT_TRUE(got_buffer);
          b = stream.AddRow(batch->GetRow(j), &status);
          ASSERT_OK(status);
        }
        ASSERT_TRUE(b);
      }
      offset += batch->num_rows();
      // Reset the batch to make sure the stream handles the memory correctly.
      batch->Reset();
    }

    bool got_read_buffer;
    ASSERT_OK(stream.PrepareForRead(false, &got_read_buffer));
    ASSERT_TRUE(got_read_buffer);

    // Read all the rows back
    vector<T> results;
    ReadValues(&stream, desc, &results);

    // Verify result
    VerifyResults<T>(*desc, results, num_rows * num_batches, gen_null);

    stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
  }

  void TestIntValuesInterleaved(int num_batches, int num_batches_before_read,
      bool unpin_stream) {
    for (int small_buffers = 0; small_buffers < 2; ++small_buffers) {
      BufferedTupleStream stream(runtime_state_, int_desc_, runtime_state_->block_mgr(),
          client_, small_buffers == 0, // initial small buffers
          true); // read_write
      ASSERT_OK(stream.Init(-1, NULL, true));
      bool got_write_buffer;
      ASSERT_OK(stream.PrepareForWrite(&got_write_buffer));
      ASSERT_TRUE(got_write_buffer);
      bool got_read_buffer;
      ASSERT_OK(stream.PrepareForRead(true, &got_read_buffer));
      ASSERT_TRUE(got_read_buffer);
      if (unpin_stream) {
        ASSERT_OK(stream.UnpinStream(BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT));
      }

      vector<int> results;

      for (int i = 0; i < num_batches; ++i) {
        RowBatch* batch = CreateIntBatch(i * BATCH_SIZE, BATCH_SIZE, false);
        for (int j = 0; j < batch->num_rows(); ++j) {
          Status status;
          bool b = stream.AddRow(batch->GetRow(j), &status);
          ASSERT_TRUE(b);
          ASSERT_OK(status);
        }
        // Reset the batch to make sure the stream handles the memory correctly.
        batch->Reset();
        if (i % num_batches_before_read == 0) {
          ReadValues(&stream, int_desc_, &results,
              (rand() % num_batches_before_read) + 1);
        }
      }
      ReadValues(&stream, int_desc_, &results);

      VerifyResults<int>(*int_desc_, results, BATCH_SIZE * num_batches, false);

      stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
    }
  }

  void TestUnpinPin(bool varlen_data);

  void TestTransferMemory(bool pinned_stream, bool read_write);

  scoped_ptr<TestEnv> test_env_;
  RuntimeState* runtime_state_;
  BufferedBlockMgr::Client* client_;

  MemTracker tracker_;
  ObjectPool pool_;
  RowDescriptor* int_desc_;
  RowDescriptor* string_desc_;
  scoped_ptr<MemPool> mem_pool_;
};

// Tests with a non-NULLable tuple per row.
class SimpleNullStreamTest : public SimpleTupleStreamTest {
 protected:
  virtual void CreateDescriptors() {
    vector<bool> nullable_tuples(1, true);
    vector<TTupleId> tuple_ids(1, static_cast<TTupleId>(0));

    DescriptorTblBuilder int_builder(test_env_->exec_env()->frontend(), &pool_);
    int_builder.DeclareTuple() << TYPE_INT;
    int_desc_ = pool_.Add(new RowDescriptor(
        *int_builder.Build(), tuple_ids, nullable_tuples));

    DescriptorTblBuilder string_builder(test_env_->exec_env()->frontend(), &pool_);
    string_builder.DeclareTuple() << TYPE_STRING;
    string_desc_ = pool_.Add(new RowDescriptor(
        *string_builder.Build(), tuple_ids, nullable_tuples));
  }
}; // SimpleNullStreamTest

// Tests with multiple non-NULLable tuples per row.
class MultiTupleStreamTest : public SimpleTupleStreamTest {
 protected:
  virtual void CreateDescriptors() {
    vector<bool> nullable_tuples;
    nullable_tuples.push_back(false);
    nullable_tuples.push_back(false);
    nullable_tuples.push_back(false);

    vector<TTupleId> tuple_ids;
    tuple_ids.push_back(static_cast<TTupleId>(0));
    tuple_ids.push_back(static_cast<TTupleId>(1));
    tuple_ids.push_back(static_cast<TTupleId>(2));

    DescriptorTblBuilder int_builder(test_env_->exec_env()->frontend(), &pool_);
    int_builder.DeclareTuple() << TYPE_INT;
    int_builder.DeclareTuple() << TYPE_INT;
    int_builder.DeclareTuple() << TYPE_INT;
    int_desc_ = pool_.Add(new RowDescriptor(
        *int_builder.Build(), tuple_ids, nullable_tuples));

    DescriptorTblBuilder string_builder(test_env_->exec_env()->frontend(), &pool_);
    string_builder.DeclareTuple() << TYPE_STRING;
    string_builder.DeclareTuple() << TYPE_STRING;
    string_builder.DeclareTuple() << TYPE_STRING;
    string_desc_ = pool_.Add(new RowDescriptor(
        *string_builder.Build(), tuple_ids, nullable_tuples));
  }
};

// Tests with multiple NULLable tuples per row.
class MultiNullableTupleStreamTest : public SimpleTupleStreamTest {
 protected:
  virtual void CreateDescriptors() {
    vector<bool> nullable_tuples;
    nullable_tuples.push_back(false);
    nullable_tuples.push_back(true);
    nullable_tuples.push_back(true);

    vector<TTupleId> tuple_ids;
    tuple_ids.push_back(static_cast<TTupleId>(0));
    tuple_ids.push_back(static_cast<TTupleId>(1));
    tuple_ids.push_back(static_cast<TTupleId>(2));

    DescriptorTblBuilder int_builder(test_env_->exec_env()->frontend(), &pool_);
    int_builder.DeclareTuple() << TYPE_INT;
    int_builder.DeclareTuple() << TYPE_INT;
    int_builder.DeclareTuple() << TYPE_INT;
    int_desc_ = pool_.Add(new RowDescriptor(
        *int_builder.Build(), tuple_ids, nullable_tuples));

    DescriptorTblBuilder string_builder(test_env_->exec_env()->frontend(), &pool_);
    string_builder.DeclareTuple() << TYPE_STRING;
    string_builder.DeclareTuple() << TYPE_STRING;
    string_builder.DeclareTuple() << TYPE_STRING;
    string_desc_ = pool_.Add(new RowDescriptor(
        *string_builder.Build(), tuple_ids, nullable_tuples));
  }
};

/// Tests with collection types.
class ArrayTupleStreamTest : public SimpleTupleStreamTest {
 protected:
  RowDescriptor* array_desc_;

  virtual void CreateDescriptors() {
    // tuples: (array<string>, array<array<int>>) (array<int>)
    vector<bool> nullable_tuples(2, true);
    vector<TTupleId> tuple_ids;
    tuple_ids.push_back(static_cast<TTupleId>(0));
    tuple_ids.push_back(static_cast<TTupleId>(1));
    ColumnType string_array_type;
    string_array_type.type = TYPE_ARRAY;
    string_array_type.children.push_back(TYPE_STRING);

    ColumnType int_array_type;
    int_array_type.type = TYPE_ARRAY;
    int_array_type.children.push_back(TYPE_STRING);

    ColumnType nested_array_type;
    nested_array_type.type = TYPE_ARRAY;
    nested_array_type.children.push_back(int_array_type);

    DescriptorTblBuilder builder(test_env_->exec_env()->frontend(), &pool_);
    builder.DeclareTuple() << string_array_type << nested_array_type;
    builder.DeclareTuple() << int_array_type;
    array_desc_ = pool_.Add(new RowDescriptor(
        *builder.Build(), tuple_ids, nullable_tuples));
  }
};

// Basic API test. No data should be going to disk.
TEST_F(SimpleTupleStreamTest, Basic) {
  InitBlockMgr(-1, IO_BLOCK_SIZE);
  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);
  TestValues<int>(100, int_desc_, false, true);
  TestValues<int>(1, int_desc_, false, false);
  TestValues<int>(10, int_desc_, false, false);
  TestValues<int>(100, int_desc_, false, false);

  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
  TestValues<StringValue>(100, string_desc_, false, true);
  TestValues<StringValue>(1, string_desc_, false, false);
  TestValues<StringValue>(10, string_desc_, false, false);
  TestValues<StringValue>(100, string_desc_, false, false);

  TestIntValuesInterleaved(1, 1, true);
  TestIntValuesInterleaved(10, 5, true);
  TestIntValuesInterleaved(100, 15, true);
  TestIntValuesInterleaved(1, 1, false);
  TestIntValuesInterleaved(10, 5, false);
  TestIntValuesInterleaved(100, 15, false);
}

// Test with only 1 buffer.
TEST_F(SimpleTupleStreamTest, OneBufferSpill) {
  // Each buffer can only hold 100 ints, so this spills quite often.
  int buffer_size = 100 * sizeof(int);
  InitBlockMgr(buffer_size, buffer_size);
  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);

  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
}

// Test with a few buffers.
TEST_F(SimpleTupleStreamTest, ManyBufferSpill) {
  int buffer_size = 100 * sizeof(int);
  InitBlockMgr(10 * buffer_size, buffer_size);

  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);
  TestValues<int>(100, int_desc_, false, true);
  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
  TestValues<StringValue>(100, string_desc_, false, true);

  TestIntValuesInterleaved(1, 1, true);
  TestIntValuesInterleaved(10, 5, true);
  TestIntValuesInterleaved(100, 15, true);
}

void SimpleTupleStreamTest::TestUnpinPin(bool varlen_data) {
  int buffer_size = 100 * sizeof(int);
  InitBlockMgr(3 * buffer_size, buffer_size);
  RowDescriptor* row_desc = varlen_data ? string_desc_ : int_desc_;

  BufferedTupleStream stream(
      runtime_state_, row_desc, runtime_state_->block_mgr(), client_, true, false);
  ASSERT_OK(stream.Init(-1, NULL, true));
  bool got_write_buffer;
  ASSERT_OK(stream.PrepareForWrite(&got_write_buffer));
  ASSERT_TRUE(got_write_buffer);

  int offset = 0;
  bool full = false;
  while (!full) {
    RowBatch* batch = varlen_data ? CreateStringBatch(offset, BATCH_SIZE, false)
                                  : CreateIntBatch(offset, BATCH_SIZE, false);
    int j = 0;
    for (; j < batch->num_rows(); ++j) {
      Status status;
      full = !stream.AddRow(batch->GetRow(j), &status);
      ASSERT_OK(status);
      if (full) break;
    }
    offset += j;
  }

  ASSERT_OK(stream.UnpinStream(BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT));

  bool pinned = false;
  ASSERT_OK(stream.PinStream(false, &pinned));
  ASSERT_TRUE(pinned);

  // Read and verify result a few times. We should be able to reread the stream if
  // we don't use delete on read mode.
  int read_iters = 3;
  for (int i = 0; i < read_iters; ++i) {
    bool delete_on_read = i == read_iters - 1;
    bool got_read_buffer;
    ASSERT_OK(stream.PrepareForRead(delete_on_read, &got_read_buffer));
    ASSERT_TRUE(got_read_buffer);

    if (varlen_data) {
      vector<StringValue> results;
      ReadValues(&stream, row_desc, &results);
      VerifyResults<StringValue>(*string_desc_, results, offset, false);
    } else {
      vector<int> results;
      ReadValues(&stream, row_desc, &results);
      VerifyResults<int>(*int_desc_, results, offset, false);
    }
  }

  // After delete_on_read, all blocks aside from the last should be deleted.
  // Note: this should really be 0, but the BufferedTupleStream returns eos before
  // deleting the last block, rather than after, so the last block isn't deleted
  // until the stream is closed.
  ASSERT_EQ(stream.bytes_in_mem(false), buffer_size);

  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);

  ASSERT_EQ(stream.bytes_in_mem(false), 0);
}

TEST_F(SimpleTupleStreamTest, UnpinPin) {
  TestUnpinPin(false);
}

TEST_F(SimpleTupleStreamTest, UnpinPinVarlen) {
  TestUnpinPin(false);
}

TEST_F(SimpleTupleStreamTest, SmallBuffers) {
  int buffer_size = IO_BLOCK_SIZE;
  InitBlockMgr(2 * buffer_size, buffer_size);

  BufferedTupleStream stream(
      runtime_state_, int_desc_, runtime_state_->block_mgr(), client_, true, false);
  ASSERT_OK(stream.Init(-1, NULL, false));
  bool got_write_buffer;
  ASSERT_OK(stream.PrepareForWrite(&got_write_buffer));
  ASSERT_TRUE(got_write_buffer);

  // Initial buffer should be small.
  EXPECT_LT(stream.bytes_in_mem(false), buffer_size);

  RowBatch* batch = CreateIntBatch(0, 1024, false);

  Status status;
  for (int i = 0; i < batch->num_rows(); ++i) {
    bool ret = stream.AddRow(batch->GetRow(i), &status);
    EXPECT_TRUE(ret);
    ASSERT_OK(status);
  }
  EXPECT_LT(stream.bytes_in_mem(false), buffer_size);
  EXPECT_LT(stream.byte_size(), buffer_size);
  ASSERT_TRUE(stream.using_small_buffers());

  // 40 MB of ints
  batch = CreateIntBatch(0, 10 * 1024 * 1024, false);
  for (int i = 0; i < batch->num_rows(); ++i) {
    bool ret = stream.AddRow(batch->GetRow(i), &status);
    ASSERT_OK(status);
    if (!ret) {
      ASSERT_TRUE(stream.using_small_buffers());
      bool got_buffer;
      ASSERT_OK(stream.SwitchToIoBuffers(&got_buffer));
      ASSERT_TRUE(got_buffer);
      ret = stream.AddRow(batch->GetRow(i), &status);
      ASSERT_OK(status);
    }
    ASSERT_TRUE(ret);
  }
  EXPECT_EQ(stream.bytes_in_mem(false), buffer_size);

  // TODO: Test for IMPALA-2330. In case SwitchToIoBuffers() fails to get buffer then
  // using_small_buffers() should still return true.
  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

void SimpleTupleStreamTest::TestTransferMemory(bool pin_stream, bool read_write) {
  // Use smaller buffers so that the explicit FLUSH_RESOURCES flag is required to
  // make the batch at capacity.
  int buffer_size = 4 * 1024;
  InitBlockMgr(100 * buffer_size, buffer_size);

  BufferedTupleStream stream(
      runtime_state_, int_desc_, runtime_state_->block_mgr(), client_, false, read_write);
  ASSERT_OK(stream.Init(-1, NULL, pin_stream));
  bool got_write_buffer;
  ASSERT_OK(stream.PrepareForWrite(&got_write_buffer));
  ASSERT_TRUE(got_write_buffer);
  RowBatch* batch = CreateIntBatch(0, 1024, false);

  // Construct a stream with 4 blocks.
  const int total_num_blocks = 4;
  while (stream.byte_size() < total_num_blocks * buffer_size) {
    Status status;
    for (int i = 0; i < batch->num_rows(); ++i) {
      bool ret = stream.AddRow(batch->GetRow(i), &status);
      EXPECT_TRUE(ret);
      ASSERT_OK(status);
    }
  }

  bool got_read_buffer;
  ASSERT_OK(stream.PrepareForRead(true, &got_read_buffer));
  ASSERT_TRUE(got_read_buffer);

  batch->Reset();
  stream.Close(batch, RowBatch::FlushMode::FLUSH_RESOURCES);
  if (pin_stream) {
    DCHECK_EQ(total_num_blocks, batch->num_blocks());
  } else if (read_write) {
    // Read and write block should be attached.
    DCHECK_EQ(2, batch->num_blocks());
  } else {
    // Read block should be attached.
    DCHECK_EQ(1, batch->num_blocks());
  }
  DCHECK(batch->AtCapacity()); // Flush resources flag should have been set.
  batch->Reset();
  DCHECK_EQ(0, batch->num_blocks());
}

/// Test attaching memory to a row batch from a pinned stream.
TEST_F(SimpleTupleStreamTest, TransferMemoryFromPinnedStreamReadWrite) {
  TestTransferMemory(true, true);
}

TEST_F(SimpleTupleStreamTest, TransferMemoryFromPinnedStreamNoReadWrite) {
  TestTransferMemory(true, false);
}

/// Test attaching memory to a row batch from an unpinned stream.
TEST_F(SimpleTupleStreamTest, TransferMemoryFromUnpinnedStreamReadWrite) {
  TestTransferMemory(false, true);
}

TEST_F(SimpleTupleStreamTest, TransferMemoryFromUnpinnedStreamNoReadWrite) {
  TestTransferMemory(false, false);
}

// Test that tuple stream functions if it references strings outside stream. The
// aggregation node relies on this since it updates tuples in-place.
TEST_F(SimpleTupleStreamTest, StringsOutsideStream) {
  int buffer_size = 8 * 1024 * 1024;
  InitBlockMgr(2 * buffer_size, buffer_size);
  Status status = Status::OK();

  int num_batches = 100;
  int rows_added = 0;
  DCHECK_EQ(string_desc_->tuple_descriptors().size(), 1);
  TupleDescriptor& tuple_desc = *string_desc_->tuple_descriptors()[0];

  set<SlotId> external_slots;
  for (int i = 0; i < tuple_desc.string_slots().size(); ++i) {
    external_slots.insert(tuple_desc.string_slots()[i]->id());
  }

  BufferedTupleStream stream(runtime_state_, string_desc_, runtime_state_->block_mgr(),
      client_, true, false, external_slots);
  for (int i = 0; i < num_batches; ++i) {
    RowBatch* batch = CreateStringBatch(rows_added, BATCH_SIZE, false);
    for (int j = 0; j < batch->num_rows(); ++j) {
      uint8_t* varlen_data;
      int fixed_size = tuple_desc.byte_size();
      uint8_t* tuple = stream.AllocateRow(fixed_size, 0, &varlen_data, &status);
      ASSERT_TRUE(tuple != NULL);
      ASSERT_TRUE(status.ok());
      // Copy fixed portion in, but leave it pointing to row batch's varlen data.
      memcpy(tuple, batch->GetRow(j)->GetTuple(0), fixed_size);
    }
    rows_added += batch->num_rows();
  }

  DCHECK_EQ(rows_added, stream.num_rows());

  for (int delete_on_read = 0; delete_on_read <= 1; ++delete_on_read) {
    // Keep stream in memory and test we can read ok.
    vector<StringValue> results;
    bool got_read_buffer;
    ASSERT_OK(stream.PrepareForRead(delete_on_read, &got_read_buffer));
    ASSERT_TRUE(got_read_buffer);
    ReadValues(&stream, string_desc_, &results);
    VerifyResults<StringValue>(*string_desc_, results, rows_added, false);
  }

  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

// Construct a big row by stiching together many tuples so the total row size
// will be close to the IO block size. With null indicators, stream will fail to
// be initialized; Without null indicators, things should work fine.
TEST_F(SimpleTupleStreamTest, BigRow) {
  InitBlockMgr(2 * IO_BLOCK_SIZE, IO_BLOCK_SIZE);
  vector<TupleId> tuple_ids;
  vector<bool> nullable_tuples;
  vector<bool> non_nullable_tuples;

  DescriptorTblBuilder big_row_builder(test_env_->exec_env()->frontend(), &pool_);
  // Each tuple contains 8 slots of TYPE_INT and a single byte for null indicator.
  const int num_tuples = IO_BLOCK_SIZE / (8 * sizeof(int) + 1);
  for (int tuple_idx = 0; tuple_idx < num_tuples; ++tuple_idx) {
    big_row_builder.DeclareTuple() << TYPE_INT << TYPE_INT << TYPE_INT << TYPE_INT
        << TYPE_INT << TYPE_INT << TYPE_INT << TYPE_INT;
    tuple_ids.push_back(static_cast<TTupleId>(tuple_idx));
    nullable_tuples.push_back(true);
    non_nullable_tuples.push_back(false);
  }
  DescriptorTbl *desc = big_row_builder.Build();

  // Construct a big row with all non-nullable tuples.
  RowDescriptor* row_desc = pool_.Add(new RowDescriptor(
      *desc, tuple_ids, non_nullable_tuples));
  ASSERT_FALSE(row_desc->IsAnyTupleNullable());
  // Test writing this row into the stream and then reading it back.
  TestValues<int>(1, row_desc, false, false, 1, false);
  TestValues<int>(1, row_desc, false, true, 1, false);

  // Construct a big row with nullable tuples. This requires space for null indicators
  // in the stream which, as a result, will fail to initialize.
  RowDescriptor* nullable_row_desc = pool_.Add(new RowDescriptor(
      *desc, tuple_ids, nullable_tuples));
  ASSERT_TRUE(nullable_row_desc->IsAnyTupleNullable());
  BufferedTupleStream nullable_stream(runtime_state_, nullable_row_desc,
      runtime_state_->block_mgr(), client_, false, false);
  Status status = nullable_stream.Init(-1, NULL, true);
  ASSERT_FALSE(status.ok());
  nullable_stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

// Test for IMPALA-3923: overflow of 32-bit int in GetRows().
TEST_F(SimpleTupleStreamTest, TestGetRowsOverflow) {
  InitBlockMgr(-1, 8 * 1024 * 1024);
  BufferedTupleStream stream(
      runtime_state_, int_desc_, runtime_state_->block_mgr(), client_, false, false);
  ASSERT_OK(stream.Init(-1, NULL, true));

  Status status;
  // Add more rows than can be fit in a RowBatch (limited by its 32-bit row count).
  // Actually adding the rows would take a very long time, so just set num_rows_.
  // This puts the stream in an inconsistent state, but exercises the right code path.
  stream.num_rows_ = 1L << 33;
  bool got_rows;
  scoped_ptr<RowBatch> overflow_batch;
  ASSERT_FALSE(stream.GetRows(&overflow_batch, &got_rows).ok());
  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

// Basic API test. No data should be going to disk.
TEST_F(SimpleNullStreamTest, Basic) {
  InitBlockMgr(-1, IO_BLOCK_SIZE);
  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);
  TestValues<int>(100, int_desc_, false, true);
  TestValues<int>(1, int_desc_, true, true);
  TestValues<int>(10, int_desc_, true, true);
  TestValues<int>(100, int_desc_, true, true);
  TestValues<int>(1, int_desc_, false, false);
  TestValues<int>(10, int_desc_, false, false);
  TestValues<int>(100, int_desc_, false, false);
  TestValues<int>(1, int_desc_, true, false);
  TestValues<int>(10, int_desc_, true, false);
  TestValues<int>(100, int_desc_, true, false);

  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
  TestValues<StringValue>(100, string_desc_, false, true);
  TestValues<StringValue>(1, string_desc_, true, true);
  TestValues<StringValue>(10, string_desc_, true, true);
  TestValues<StringValue>(100, string_desc_, true, true);
  TestValues<StringValue>(1, string_desc_, false, false);
  TestValues<StringValue>(10, string_desc_, false, false);
  TestValues<StringValue>(100, string_desc_, false, false);
  TestValues<StringValue>(1, string_desc_, true, false);
  TestValues<StringValue>(10, string_desc_, true, false);
  TestValues<StringValue>(100, string_desc_, true, false);

  TestIntValuesInterleaved(1, 1, true);
  TestIntValuesInterleaved(10, 5, true);
  TestIntValuesInterleaved(100, 15, true);
  TestIntValuesInterleaved(1, 1, false);
  TestIntValuesInterleaved(10, 5, false);
  TestIntValuesInterleaved(100, 15, false);
}

// Test tuple stream with only 1 buffer and rows with multiple tuples.
TEST_F(MultiTupleStreamTest, MultiTupleOneBufferSpill) {
  // Each buffer can only hold 100 ints, so this spills quite often.
  int buffer_size = 100 * sizeof(int);
  InitBlockMgr(buffer_size, buffer_size);
  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);

  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
}

// Test with a few buffers and rows with multiple tuples.
TEST_F(MultiTupleStreamTest, MultiTupleManyBufferSpill) {
  int buffer_size = 100 * sizeof(int);
  InitBlockMgr(10 * buffer_size, buffer_size);

  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);
  TestValues<int>(100, int_desc_, false, true);

  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
  TestValues<StringValue>(100, string_desc_, false, true);

  TestIntValuesInterleaved(1, 1, true);
  TestIntValuesInterleaved(10, 5, true);
  TestIntValuesInterleaved(100, 15, true);
}

// Test that we can allocate a row in the stream and copy in multiple tuples then
// read it back from the stream.
TEST_F(MultiTupleStreamTest, MultiTupleAllocateRow) {
  // Use small buffers so it will be flushed to disk.
  int buffer_size = 4 * 1024;
  InitBlockMgr(2 * buffer_size, buffer_size);
  Status status = Status::OK();

  int num_batches = 1;
  int rows_added = 0;
  BufferedTupleStream stream(
      runtime_state_, string_desc_, runtime_state_->block_mgr(), client_, false, false);
  ASSERT_OK(stream.Init(-1, NULL, false));
  bool got_write_buffer;
  ASSERT_OK(stream.PrepareForWrite(&got_write_buffer));
  ASSERT_TRUE(got_write_buffer);

  for (int i = 0; i < num_batches; ++i) {
    RowBatch* batch = CreateStringBatch(rows_added, 1, false);
    for (int j = 0; j < batch->num_rows(); ++j) {
      TupleRow* row = batch->GetRow(j);
      int64_t fixed_size = 0;
      int64_t varlen_size = 0;
      for (int k = 0; k < string_desc_->tuple_descriptors().size(); k++) {
        TupleDescriptor* tuple_desc = string_desc_->tuple_descriptors()[k];
        fixed_size += tuple_desc->byte_size();
        varlen_size += row->GetTuple(k)->VarlenByteSize(*tuple_desc);
      }
      uint8_t* varlen_data;
      uint8_t* fixed_data = stream.AllocateRow(fixed_size, varlen_size, &varlen_data,
          &status);
      ASSERT_TRUE(fixed_data != NULL);
      ASSERT_TRUE(status.ok());
      uint8_t* varlen_write_ptr = varlen_data;
      for (int k = 0; k < string_desc_->tuple_descriptors().size(); k++) {
        TupleDescriptor* tuple_desc = string_desc_->tuple_descriptors()[k];
        Tuple* src = row->GetTuple(k);
        Tuple* dst = reinterpret_cast<Tuple*>(fixed_data);
        fixed_data += tuple_desc->byte_size();
        memcpy(dst, src, tuple_desc->byte_size());
        for (int l = 0; l < tuple_desc->slots().size(); l++) {
          SlotDescriptor* slot = tuple_desc->slots()[l];
          StringValue* src_string = src->GetStringSlot(slot->tuple_offset());
          StringValue* dst_string = dst->GetStringSlot(slot->tuple_offset());
          dst_string->ptr = reinterpret_cast<char*>(varlen_write_ptr);
          memcpy(dst_string->ptr, src_string->ptr, src_string->len);
          varlen_write_ptr += src_string->len;
        }
      }
      ASSERT_EQ(varlen_data + varlen_size, varlen_write_ptr);
    }
    rows_added += batch->num_rows();
  }

  for (int i = 0; i < 3; ++i) {
    bool delete_on_read = i == 2;
    vector<StringValue> results;
    bool got_read_buffer;
    stream.PrepareForRead(delete_on_read, &got_read_buffer);
    ASSERT_TRUE(got_read_buffer);
    ReadValues(&stream, string_desc_, &results);
    VerifyResults<StringValue>(*string_desc_, results, rows_added, false);
  }

  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

// Test with rows with multiple nullable tuples.
TEST_F(MultiNullableTupleStreamTest, MultiNullableTupleOneBufferSpill) {
  // Each buffer can only hold 100 ints, so this spills quite often.
  int buffer_size = 100 * sizeof(int);
  InitBlockMgr(buffer_size, buffer_size);
  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);
  TestValues<int>(1, int_desc_, true, true);
  TestValues<int>(10, int_desc_, true, true);

  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
  TestValues<StringValue>(1, string_desc_, true, true);
  TestValues<StringValue>(10, string_desc_, true, true);
}

// Test with a few buffers.
TEST_F(MultiNullableTupleStreamTest, MultiNullableTupleManyBufferSpill) {
  int buffer_size = 100 * sizeof(int);
  InitBlockMgr(10 * buffer_size, buffer_size);

  TestValues<int>(1, int_desc_, false, true);
  TestValues<int>(10, int_desc_, false, true);
  TestValues<int>(100, int_desc_, false, true);
  TestValues<int>(1, int_desc_, true, true);
  TestValues<int>(10, int_desc_, true, true);
  TestValues<int>(100, int_desc_, true, true);

  TestValues<StringValue>(1, string_desc_, false, true);
  TestValues<StringValue>(10, string_desc_, false, true);
  TestValues<StringValue>(100, string_desc_, false, true);
  TestValues<StringValue>(1, string_desc_, true, true);
  TestValues<StringValue>(10, string_desc_, true, true);
  TestValues<StringValue>(100, string_desc_, true, true);

  TestIntValuesInterleaved(1, 1, true);
  TestIntValuesInterleaved(10, 5, true);
  TestIntValuesInterleaved(100, 15, true);
}

/// Test that ComputeRowSize handles nulls
TEST_F(MultiNullableTupleStreamTest, TestComputeRowSize) {
  InitBlockMgr(-1, 8 * 1024 * 1024);
  const vector<TupleDescriptor*>& tuple_descs = string_desc_->tuple_descriptors();
  // String in second tuple is stored externally.
  set<SlotId> external_slots;
  const SlotDescriptor* external_string_slot = tuple_descs[1]->slots()[0];
  external_slots.insert(external_string_slot->id());

  BufferedTupleStream stream(runtime_state_, string_desc_, runtime_state_->block_mgr(),
      client_, false, false, external_slots);
  gscoped_ptr<TupleRow, FreeDeleter> row(reinterpret_cast<TupleRow*>(
        malloc(tuple_descs.size() * sizeof(Tuple*))));
  gscoped_ptr<Tuple, FreeDeleter> tuple0(reinterpret_cast<Tuple*>(
        malloc(tuple_descs[0]->byte_size())));
  gscoped_ptr<Tuple, FreeDeleter> tuple1(reinterpret_cast<Tuple*>(
        malloc(tuple_descs[1]->byte_size())));
  gscoped_ptr<Tuple, FreeDeleter> tuple2(reinterpret_cast<Tuple*>(
        malloc(tuple_descs[2]->byte_size())));
  memset(tuple0.get(), 0, tuple_descs[0]->byte_size());
  memset(tuple1.get(), 0, tuple_descs[1]->byte_size());
  memset(tuple2.get(), 0, tuple_descs[2]->byte_size());

  // All nullable tuples are NULL.
  row->SetTuple(0, tuple0.get());
  row->SetTuple(1, NULL);
  row->SetTuple(2, NULL);
  EXPECT_EQ(tuple_descs[0]->byte_size(), stream.ComputeRowSize(row.get()));

  // Tuples are initialized to empty and have no var-len data.
  row->SetTuple(1, tuple1.get());
  row->SetTuple(2, tuple2.get());
  EXPECT_EQ(string_desc_->GetRowSize(), stream.ComputeRowSize(row.get()));

  // Tuple 0 has some data.
  const SlotDescriptor* string_slot = tuple_descs[0]->slots()[0];
  StringValue* sv = tuple0->GetStringSlot(string_slot->tuple_offset());
  *sv = STRINGS[0];
  int64_t expected_len = string_desc_->GetRowSize() + sv->len;
  EXPECT_EQ(expected_len, stream.ComputeRowSize(row.get()));

  // Check that external slots aren't included in count.
  sv = tuple1->GetStringSlot(external_string_slot->tuple_offset());
  sv->ptr = reinterpret_cast<char*>(1234);
  sv->len = 1234;
  EXPECT_EQ(expected_len, stream.ComputeRowSize(row.get()));

  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

/// Test that deep copy works with arrays by copying into a BufferedTupleStream, freeing
/// the original rows, then reading back the rows and verifying the contents.
TEST_F(ArrayTupleStreamTest, TestArrayDeepCopy) {
  Status status;
  InitBlockMgr(-1, IO_BLOCK_SIZE);
  const int NUM_ROWS = 4000;
  BufferedTupleStream stream(
      runtime_state_, array_desc_, runtime_state_->block_mgr(), client_, false, false);
  const vector<TupleDescriptor*>& tuple_descs = array_desc_->tuple_descriptors();
  // Write out a predictable pattern of data by iterating over arrays of constants.
  int strings_index = 0; // we take the mod of this as index into STRINGS.
  int array_lens[] = { 0, 1, 5, 10, 1000, 2, 49, 20 };
  int num_array_lens = sizeof(array_lens) / sizeof(array_lens[0]);
  int array_len_index = 0;
  ASSERT_OK(stream.Init(-1, NULL, false));
  bool got_write_buffer;
  ASSERT_OK(stream.PrepareForWrite(&got_write_buffer));
  ASSERT_TRUE(got_write_buffer);

  for (int i = 0; i < NUM_ROWS; ++i) {
    int expected_row_size = tuple_descs[0]->byte_size() + tuple_descs[1]->byte_size();
    gscoped_ptr<TupleRow, FreeDeleter> row(reinterpret_cast<TupleRow*>(
          malloc(tuple_descs.size() * sizeof(Tuple*))));
    gscoped_ptr<Tuple, FreeDeleter> tuple0(reinterpret_cast<Tuple*>(
          malloc(tuple_descs[0]->byte_size())));
    gscoped_ptr<Tuple, FreeDeleter> tuple1(reinterpret_cast<Tuple*>(
          malloc(tuple_descs[1]->byte_size())));
    memset(tuple0.get(), 0, tuple_descs[0]->byte_size());
    memset(tuple1.get(), 0, tuple_descs[1]->byte_size());
    row->SetTuple(0, tuple0.get());
    row->SetTuple(1, tuple1.get());

    // Only array<string> is non-null.
    tuple0->SetNull(tuple_descs[0]->slots()[1]->null_indicator_offset());
    tuple1->SetNull(tuple_descs[1]->slots()[0]->null_indicator_offset());
    const SlotDescriptor* array_slot_desc = tuple_descs[0]->slots()[0];
    const TupleDescriptor* item_desc = array_slot_desc->collection_item_descriptor();

    int array_len = array_lens[array_len_index++ % num_array_lens];
    CollectionValue* cv = tuple0->GetCollectionSlot(array_slot_desc->tuple_offset());
    cv->ptr = NULL;
    cv->num_tuples = 0;
    CollectionValueBuilder builder(cv, *item_desc, mem_pool_.get(), runtime_state_,
        array_len);
    Tuple* array_data;
    int num_rows;
    builder.GetFreeMemory(&array_data, &num_rows);
    expected_row_size += item_desc->byte_size() * array_len;

    // Fill the array with pointers to our constant strings.
    for (int j = 0; j < array_len; ++j) {
      const StringValue* string = &STRINGS[strings_index++ % NUM_STRINGS];
      array_data->SetNotNull(item_desc->slots()[0]->null_indicator_offset());
      RawValue::Write(string, array_data, item_desc->slots()[0], mem_pool_.get());
      array_data += item_desc->byte_size();
      expected_row_size += string->len;
    }
    builder.CommitTuples(array_len);

    // Check that internal row size computation gives correct result.
    EXPECT_EQ(expected_row_size, stream.ComputeRowSize(row.get()));
    bool b = stream.AddRow(row.get(), &status);
    ASSERT_TRUE(b);
    ASSERT_OK(status);
    mem_pool_->FreeAll(); // Free data as soon as possible to smoke out issues.
  }

  // Read back and verify data.
  bool got_read_buffer;
  stream.PrepareForRead(false, &got_read_buffer);
  ASSERT_TRUE(got_read_buffer);
  strings_index = 0;
  array_len_index = 0;
  bool eos = false;
  int rows_read = 0;
  RowBatch batch(array_desc_, BATCH_SIZE, &tracker_);
  do {
    batch.Reset();
    ASSERT_OK(stream.GetNext(&batch, &eos));
    for (int i = 0; i < batch.num_rows(); ++i) {
      TupleRow* row = batch.GetRow(i);
      Tuple* tuple0 = row->GetTuple(0);
      Tuple* tuple1 = row->GetTuple(1);
      ASSERT_TRUE(tuple0 != NULL);
      ASSERT_TRUE(tuple1 != NULL);
      const SlotDescriptor* array_slot_desc = tuple_descs[0]->slots()[0];
      ASSERT_FALSE(tuple0->IsNull(array_slot_desc->null_indicator_offset()));
      ASSERT_TRUE(tuple0->IsNull(tuple_descs[0]->slots()[1]->null_indicator_offset()));
      ASSERT_TRUE(tuple1->IsNull(tuple_descs[1]->slots()[0]->null_indicator_offset()));

      const TupleDescriptor* item_desc = array_slot_desc->collection_item_descriptor();
      int expected_array_len = array_lens[array_len_index++ % num_array_lens];
      CollectionValue* cv = tuple0->GetCollectionSlot(array_slot_desc->tuple_offset());
      ASSERT_EQ(expected_array_len, cv->num_tuples);
      for (int j = 0; j < cv->num_tuples; ++j) {
        Tuple* item = reinterpret_cast<Tuple*>(cv->ptr + j * item_desc->byte_size());
        const SlotDescriptor* string_desc = item_desc->slots()[0];
        ASSERT_FALSE(item->IsNull(string_desc->null_indicator_offset()));
        const StringValue* expected = &STRINGS[strings_index++ % NUM_STRINGS];
        const StringValue* actual = item->GetStringSlot(string_desc->tuple_offset());
        ASSERT_EQ(*expected, *actual);
      }
    }
    rows_read += batch.num_rows();
  } while (!eos);
  ASSERT_EQ(NUM_ROWS, rows_read);
  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

/// Test that ComputeRowSize handles nulls
TEST_F(ArrayTupleStreamTest, TestComputeRowSize) {
  InitBlockMgr(-1, 8 * 1024 * 1024);
  const vector<TupleDescriptor*>& tuple_descs = array_desc_->tuple_descriptors();
  set<SlotId> external_slots;
  // Second array slot in first tuple is stored externally.
  const SlotDescriptor* external_array_slot = tuple_descs[0]->slots()[1];
  external_slots.insert(external_array_slot->id());

  BufferedTupleStream stream(runtime_state_, array_desc_, runtime_state_->block_mgr(),
      client_, false, false, external_slots);
  gscoped_ptr<TupleRow, FreeDeleter> row(reinterpret_cast<TupleRow*>(
        malloc(tuple_descs.size() * sizeof(Tuple*))));
  gscoped_ptr<Tuple, FreeDeleter> tuple0(reinterpret_cast<Tuple*>(
        malloc(tuple_descs[0]->byte_size())));
  gscoped_ptr<Tuple, FreeDeleter> tuple1(reinterpret_cast<Tuple*>(
        malloc(tuple_descs[1]->byte_size())));
  memset(tuple0.get(), 0, tuple_descs[0]->byte_size());
  memset(tuple1.get(), 0, tuple_descs[1]->byte_size());

  // All tuples are NULL.
  row->SetTuple(0, NULL);
  row->SetTuple(1, NULL);
  EXPECT_EQ(0, stream.ComputeRowSize(row.get()));

  // Tuples are initialized to empty and have no var-len data.
  row->SetTuple(0, tuple0.get());
  row->SetTuple(1, tuple1.get());
  EXPECT_EQ(array_desc_->GetRowSize(), stream.ComputeRowSize(row.get()));

  // Tuple 0 has an array.
  int expected_row_size = array_desc_->GetRowSize();
  const SlotDescriptor* array_slot = tuple_descs[0]->slots()[0];
  const TupleDescriptor* item_desc = array_slot->collection_item_descriptor();
  int array_len = 128;
  CollectionValue* cv = tuple0->GetCollectionSlot(array_slot->tuple_offset());
  CollectionValueBuilder builder(cv, *item_desc, mem_pool_.get(), runtime_state_,
      array_len);
  Tuple* array_data;
  int num_rows;
  builder.GetFreeMemory(&array_data, &num_rows);
  expected_row_size += item_desc->byte_size() * array_len;

  // Fill the array with pointers to our constant strings.
  for (int i = 0; i < array_len; ++i) {
    const StringValue* str = &STRINGS[i % NUM_STRINGS];
    array_data->SetNotNull(item_desc->slots()[0]->null_indicator_offset());
    RawValue::Write(str, array_data, item_desc->slots()[0], mem_pool_.get());
    array_data += item_desc->byte_size();
    expected_row_size += str->len;
  }
  builder.CommitTuples(array_len);
  EXPECT_EQ(expected_row_size, stream.ComputeRowSize(row.get()));

  // Check that the external slot isn't included in size.
  cv = tuple0->GetCollectionSlot(external_array_slot->tuple_offset());
  // ptr of external slot shouldn't be dereferenced when computing size.
  cv->ptr = reinterpret_cast<uint8_t*>(1234);
  cv->num_tuples = 1234;
  EXPECT_EQ(expected_row_size, stream.ComputeRowSize(row.get()));

  // Check that the array is excluded if tuple 0's array has its null indicator set.
  tuple0->SetNull(array_slot->null_indicator_offset());
  EXPECT_EQ(array_desc_->GetRowSize(), stream.ComputeRowSize(row.get()));

  stream.Close(NULL, RowBatch::FlushMode::NO_FLUSH_RESOURCES);
}

// TODO: more tests.
//  - The stream can operate in many modes

}

int main(int argc, char** argv) {
  ::testing::InitGoogleTest(&argc, argv);
  impala::InitCommonRuntime(argc, argv, true, impala::TestInfo::BE_TEST);
  impala::InitFeSupport();
  impala::LlvmCodeGen::InitializeLlvm();
  return RUN_ALL_TESTS();
}

max_stars_repo_name: anujjamwal/Impala

max_issues_repo_name: anujjamwal/Impala

max_forks_repo_name: anujjamwal/Impala


text:
#include <iostream>
#include <fstream>
#include <string>
#include <cstdlib>
#include <unordered_map>
#include <boost/filesystem.hpp>

using namespace std;
namespace fs = boost::filesystem;

int main () {
  string orl_1;
  string orl_2;
  string cmd;
  ifstream result_or_oracle_1;
  ifstream result_or_oracle_2;

  fs::path exec_directory = fs::current_path();
  fs::path test_directory("./tests");

  for(auto iter = fs::recursive_directory_iterator(test_directory); iter != fs::recursive_directory_iterator(); ++iter) {
    if (iter->path().extension() == ".c") {
      cmd = "cd " + iter->path().parent_path().string() + "; " + exec_directory.string() + "/selfie -c " + iter->path().filename().string() + " -o " + iter->path().filename().stem().string();
      system(cmd.c_str());
      cmd = "cd " + iter->path().parent_path().string() + "; " + exec_directory.string() + "/phantom -l " + iter->path().filename().stem().string() + " -i 0";
      system(cmd.c_str());
      cmd = "cd " + iter->path().parent_path().string() + "; " + "rm " + iter->path().filename().stem().string();
      system(cmd.c_str());
    }
  }

  std::ofstream output_results("test_output.txt", std::ofstream::trunc);
  unordered_map<string, string> scanned_files;
  unordered_map<string, string>::const_iterator record;
  for(auto iter = fs::recursive_directory_iterator(test_directory); iter != fs::recursive_directory_iterator(); ++iter) {
    record = scanned_files.find(iter->path().filename().stem().string());
    if (record == scanned_files.end()) {
      if (iter->path().filename().extension().string() == ".oracle" || iter->path().filename().extension().string() == ".result") {
        scanned_files.insert(make_pair<string, string>(iter->path().filename().stem().c_str(), iter->path().filename().c_str()));
      }
    } else {
      if (iter->path().filename().extension().string() == ".oracle" || iter->path().filename().extension().string() == ".result") {
        string dir_1 = iter->path().parent_path().string() + "/" + record->second;
        string dir_2 = iter->path().parent_path().string() + "/" + iter->path().filename().string();
        result_or_oracle_1.open(dir_1);
        result_or_oracle_2.open(dir_2);
        if (result_or_oracle_1.is_open() && result_or_oracle_2.is_open()) {
          getline(result_or_oracle_1, orl_1);
          getline(result_or_oracle_2, orl_2);
          while (result_or_oracle_1.eof() == false && result_or_oracle_2.eof() == false) {
            if (orl_2.compare(orl_1) != 0) {
              output_results << "not equal1: " << dir_1 << endl;
              break;
            }
            getline(result_or_oracle_1, orl_1);
            getline(result_or_oracle_2, orl_2);
          }

          if (result_or_oracle_1.eof() == false) {
            output_results << "not equal2: " << dir_1 << endl;
          }

          if (result_or_oracle_2.eof() == false) {
            output_results << "not equal3: " << dir_1 << endl;
          }

          result_or_oracle_1.close();
          result_or_oracle_2.close();
        }
      }
    }
  }

  output_results.close();

  return 0;
}

ext: cpp


text:
//==================================================================================================
/*!

  Copyright 2016 NumScale SAS

  Distributed under the Boost Software License, Version 1.0.
  (See accompanying file LICENSE.md or copy at http://boost.org/LICENSE_1_0.txt)
*/
//==================================================================================================
#include <boost/simd/function/bitwise_cast.hpp>
#include <boost/simd/meta/is_bitwise_logical.hpp>
#include <boost/simd/constant/allbits.hpp>
#include <boost/simd/logical.hpp>
#include <boost/simd/pack.hpp>
#include <simd_test.hpp>

STF_CASE( "Check bitwise_cast between integer types" )
{
  namespace bs = boost::simd;
  static const std::size_t N = bs::pack<uint16_t>::static_size;

  {
    using ps_t = bs::pack<std::uint16_t,N>;
    using pc_t = bs::pack<std::uint8_t,2*N>;

    STF_EQUAL ( bs::bitwise_cast<ps_t>( pc_t(0x01) )  , ps_t(0x0101)  );
    STF_EQUAL ( bs::bitwise_cast<pc_t>( ps_t(0x0101) ), pc_t(0x01)     );
  }

  {
    using ps_t = bs::pack<std::uint16_t,N/2>;
    using pc_t = bs::pack<std::uint8_t,N>;

    STF_EQUAL ( bs::bitwise_cast<ps_t>( pc_t(0x01) )  , ps_t(0x0101)  );
    STF_EQUAL ( bs::bitwise_cast<pc_t>( ps_t(0x0101) ), pc_t(0x01)     );
  }

  {
    using ps_t = bs::pack<std::uint16_t,2*N>;
    using pc_t = bs::pack<std::uint8_t,4*N>;

    STF_EQUAL ( bs::bitwise_cast<ps_t>( pc_t(0x01) )  , ps_t(0x0101)  );
    STF_EQUAL ( bs::bitwise_cast<pc_t>( ps_t(0x0101) ), pc_t(0x01)     );
  }
}

STF_CASE( "Check bitwise_cast between integer & real types" )
{
  namespace bs = boost::simd;
  static const std::size_t N = bs::pack<float>::static_size;

  {
    using pf_t = bs::pack<float,N>;
    using pi_t = bs::pack<std::uint32_t,N>;
    STF_EQUAL ( bs::bitwise_cast<pi_t>( pf_t(1.f) )         , pi_t(0x3F800000UL)  );
    STF_EQUAL ( bs::bitwise_cast<pf_t>( pi_t(0x3F800000UL) ), pf_t(1.f)           );
  }

  {
    using pf_t = bs::pack<float,N/2>;
    using pi_t = bs::pack<std::uint32_t,N/2>;
    STF_EQUAL ( bs::bitwise_cast<pi_t>( pf_t(1.f) )         , pi_t(0x3F800000UL)  );
    STF_EQUAL ( bs::bitwise_cast<pf_t>( pi_t(0x3F800000UL) ), pf_t(1.f)           );
  }

  {
    using pf_t = bs::pack<float,N*2>;
    using pi_t = bs::pack<std::uint32_t,N*2>;
    STF_EQUAL ( bs::bitwise_cast<pi_t>( pf_t(1.f) )         , pi_t(0x3F800000UL)  );
    STF_EQUAL ( bs::bitwise_cast<pf_t>( pi_t(0x3F800000UL) ), pf_t(1.f)           );
  }
}

template<typename T, typename Env> void logical_test(Env& $, std::true_type const&)
{
  namespace bs = boost::simd;
  static const std::size_t N = bs::pack<T>::static_size;

  {
    using pa_t = bs::pack<T,N>;
    using pl_t = bs::pack<bs::logical<T>,N>;
    STF_IEEE_EQUAL ( bs::bitwise_cast<pa_t>( pl_t(true) )         , bs::Allbits<pa_t>() );
    STF_IEEE_EQUAL ( bs::bitwise_cast<pl_t>( bs::Allbits<pa_t>() ), pl_t(true)          );
  }

  {
    using pa_t = bs::pack<T,N/2>;
    using pl_t = bs::pack<bs::logical<T>,N/2>;
    STF_IEEE_EQUAL ( bs::bitwise_cast<pa_t>( pl_t(true) )         , bs::Allbits<pa_t>() );
    STF_IEEE_EQUAL ( bs::bitwise_cast<pl_t>( bs::Allbits<pa_t>() ), pl_t(true)          );
  }

  {
    using pa_t = bs::pack<T,N*2>;
    using pl_t = bs::pack<bs::logical<T>,N*2>;
    STF_IEEE_EQUAL ( bs::bitwise_cast<pa_t>( pl_t(true) )         , bs::Allbits<pa_t>() );
    STF_IEEE_EQUAL ( bs::bitwise_cast<pl_t>( bs::Allbits<pa_t>() ), pl_t(true)          );
  }
}

template<typename T, typename Env> void logical_test(Env& $, std::false_type const&)
{
  STF_PASS("Logical type are not bitwise compatible on this architecture.");
}

STF_CASE_TPL( "Check bitwise_cast between arithmetic & logical types", STF_NUMERIC_TYPES )
{
  logical_test<T>($, typename boost::simd::is_bitwise_logical<T>::type{});
}

template<typename Env> void logical_diff_test(Env& $, std::true_type const&)
{
  namespace bs = boost::simd;
  static const std::size_t N = bs::pack<float>::static_size;

  {
    using pf_t  = bs::pack<float,N>;
    using plf_t = bs::pack<bs::logical<float>,N>;
    using pli_t = bs::pack<bs::logical<std::uint32_t>,N>;

    STF_EQUAL     ( bs::bitwise_cast<pli_t>( bs::Allbits<pf_t>() ), pli_t(true)         );
    STF_IEEE_EQUAL( bs::bitwise_cast<pf_t>( pli_t(true) )         , bs::Allbits<pf_t>() );
    STF_IEEE_EQUAL( bs::bitwise_cast<plf_t>( pli_t(true) )        , plf_t(true)         );
  }

  {
    using pf_t  = bs::pack<float,N/2>;
    using plf_t = bs::pack<bs::logical<float>,N/2>;
    using pli_t = bs::pack<bs::logical<std::uint32_t>,N/2>;

    STF_EQUAL     ( bs::bitwise_cast<pli_t>( bs::Allbits<pf_t>() ), pli_t(true)         );
    STF_IEEE_EQUAL( bs::bitwise_cast<pf_t>( pli_t(true) )         , bs::Allbits<pf_t>() );
    STF_IEEE_EQUAL( bs::bitwise_cast<plf_t>( pli_t(true) )        , plf_t(true)         );
  }

  {
    using pf_t  = bs::pack<float,N*2>;
    using plf_t = bs::pack<bs::logical<float>,N*2>;
    using pli_t = bs::pack<bs::logical<std::uint32_t>,N*2>;

    STF_EQUAL     ( bs::bitwise_cast<pli_t>( bs::Allbits<pf_t>() ), pli_t(true)         );
    STF_IEEE_EQUAL( bs::bitwise_cast<pf_t>( pli_t(true) )         , bs::Allbits<pf_t>() );
    STF_IEEE_EQUAL( bs::bitwise_cast<plf_t>( pli_t(true) )        , plf_t(true)         );
  }
}

template<typename Env> void logical_diff_test(Env& $, std::false_type const&)
{
  STF_PASS("Logical types are not bitwise compatible on this architecture.");
}

STF_CASE( "Check bitwise_cast between logical types of different base types" )
{
  logical_diff_test($, typename boost::simd::is_bitwise_logical<float>::type{});
}

max_stars_repo_name: nickporubsky/boost-simd-clone


text:
#ifndef _HASH_HPP
#define _HASH_HPP
#include <string>
#include <vector>
#include <stdint.h>
#include <boost/multiprecision/gmp.hpp>
#include <boost/multiprecision/cpp_int.hpp>

namespace usebitcoins {
namespace hash {

// Generate BTC addresses deterministically by hashing the customer's ID
// (e.g., an email address)

using email_address_t = std::string;

/// Perform straightforward hash of arbitrary data
std::vector<uint8_t> blake2_hash(std::vector<uint8_t> src);

// Hash an arbitrary string via blake2 to a byte array
std::vector<uint8_t> hash_string(std::string_view src);

/// Hash email address to a 64-bit integer
uint64_t hash_email(const email_address_t& email_address);

/// Fit a byte array into a bignum
/// Assumes that `src` is arranged in little-endian order
boost::multiprecision::uint256_t pack_byte_array(const std::vector<uint8_t>& src);

} // namespace hash
} // namespace usebitcoins

#endif // _HASH_HPP

ext: hpp


text:
//
//  Polygone.cpp
//
//
//  Created by Victoire Chapelle on 14/01/2016.
//
//

#include "Polygon.h"

#include <math.h>
#include <string>

#include <boost/serialization/vector.hpp>

#include "Utils.h"
#include "optional.h"
#include "Segment.h"

namespace TP4
{
	std::experimental::optional<Polygon> make_polygon(const std::vector<Point> vertices)
	{
		if (vertices.size() > 2)
		{
			Point p1, p2;

			p1.first = vertices[0].first - vertices[vertices.size() - 1].first;
			p1.second = vertices[0].second - vertices[vertices.size() - 1].second;
			p2.first = vertices[1].first - vertices[0].first;
			p2.second = vertices[1].second - vertices[0].second;

			double det_value;
			double current_det_value;

			det_value = p1.first * p2.second - p1.second * p2.first;

			for (size_t i = 0; i < vertices.size(); ++i)
			{
				p1.first = vertices[mod(i + 1, vertices.size())].first - vertices[i].first;
				p1.second = vertices[mod(i + 1, vertices.size())].second - vertices[i].second;
				p2.first = vertices[mod(i + 2, vertices.size())].first - vertices[mod(i + 1, vertices.size())].first;
				p2.second = vertices[mod(i + 2, vertices.size())].second - vertices[mod(i + 1, vertices.size())].second;

				current_det_value = p1.first * p2.second - p1.second * p2.first;

				if (det_value*current_det_value < 0 - 0.00001)
					return std::experimental::nullopt;
			}

			return std::experimental::optional<Polygon>(Polygon(std::move(vertices)));
		}

		return std::experimental::nullopt;
	}

	Polygon Move(const Polygon& shape, coord_t dx, coord_t dy)
	{
		std::vector<Point> new_vertices;
		new_vertices.reserve(new_vertices.size());

		for (const Point& p : shape.vertices)
			new_vertices.emplace_back(p.first + dx, p.second + dy);

		return Polygon(std::move(new_vertices));
	}

	inline double squared_distance(Point a, Point b)
	{
		return (a.first - b.first) * (a.first - b.first)
			+ (a.second - b.second) * (a.second - b.second);
	}

	bool Is_contained(const Polygon& shape, Point point)
	{
		double sum = 0.0;

		// Calcul de la somme des cosinus que font les angles entre les sommets consécutifs et le point
		for (size_t i = 0; i < shape.vertices.size(); ++i)
		{
			auto next_point = shape.vertices[mod(i + 1, shape.vertices.size())];

			double a2 = squared_distance(shape.vertices[i], point);
			double c2 = squared_distance(shape.vertices[i], next_point);
			double b2 = squared_distance(next_point, point);

			if (abs(a2) < 0.001 || abs(b2) < 0.001)
				return true; // Le point est confondu avec un sommet

			auto seg = make_segment(shape.vertices[i], next_point);
			if(seg)
				if (Is_contained(*seg, point))
					return true;

			// Formule d'al-kashi : cos(angle) = (a² + b² - c²) / (2*a*b)
			double cos = (a2 + b2 - c2) / (2.0*sqrt(a2)*sqrt(b2));
		//	double cos = c2 / (length_1 + length_3 - 2.0*sqrt(length_3)*sqrt(length_1));

			sum += acos(cos);
		}

		// Si la somme est 1 le point est à l'intérieur (somme des angles = 360°)
		// TODO: faire mieux?
		if (abs(mod(sum, 2.0*3.14159) < 0.001))
			return true;
		return false;
	}

	Polygon::Polygon(const std::vector<Point>&& vertices)
		: vertices(std::move(vertices))
	{ }

	std::ostream& operator<<(std::ostream& flux, const Polygon& polygone)
	{
		flux << "{ ";
		for (size_t i = 0; i < polygone.vertices.size(); ++i)
			flux << "(" << polygone.vertices[i].first << ", " << polygone.vertices[i].second << (i == polygone.vertices.size() - 1 ? ") " : "); ");
		flux << "}";

		return flux;
	}
}


text:
///////////////////////////////////////////////////////////////////////////////
//
// Copyright (c) Tavendo GmbH
//
// Boost Software License - Version 1.0 - August 17th, 2003
//
// Permission is hereby granted, free of charge, to any person or organization
// obtaining a copy of the software and accompanying documentation covered by
// this license (the "Software") to use, reproduce, display, distribute,
// execute, and transmit the Software, and to prepare derivative works of the
// Software, and to permit third-parties to whom the Software is furnished to
// do so, all subject to the following:
//
// The copyright notices in the Software and this entire statement, including
// the above license grant, this restriction and the following disclaimer,
// must be included in all copies of the Software, in whole or in part, and
// all derivative works of the Software, unless such copies or derivative
// works are solely in the form of machine-executable object code generated by
// a source language processor.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
// SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
// FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
// ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
//
///////////////////////////////////////////////////////////////////////////////

#include "parameters.hpp"

#include <autobahn/autobahn.hpp>
#include <autobahn/wamp_websocketpp_websocket_transport.hpp>
#include <websocketpp/config/asio_no_tls_client.hpp>
#include <websocketpp/client.hpp>
#include <boost/version.hpp>
#include <iostream>
#include <memory>
#include <string>
#include <tuple>

void add2(autobahn::wamp_invocation invocation)
{
    auto a = invocation->argument<uint64_t>(0);
    auto b = invocation->argument<uint64_t>(1);

    std::cerr << "Procedure com.examples.calculator.add2 invoked: " << a << ", " << b << std::endl;

    invocation->result(std::make_tuple(a + b));
}

typedef websocketpp::client<websocketpp::config::asio_client> client;

int main(int argc, char** argv)
{
    std::cerr << "Boost: " << BOOST_VERSION << std::endl;
    try {
        auto parameters = get_parameters(argc, argv);

        std::cerr << "Connecting to realm: " << parameters->realm() << std::endl;

        boost::asio::io_service io;
        bool debug = parameters->debug();

        client ws_clinet;
        ws_clinet.init_asio(&io);
        auto transport = std::make_shared < autobahn::wamp_websocketpp_websocket_transport<websocketpp::config::asio_client> >(
            ws_clinet, "ws://127.0.0.1:8080/ws", debug);

        auto session = std::make_shared<autobahn::wamp_session>(io, debug);

        // Create a thread to run the telemetry loop
        transport->attach(std::static_pointer_cast<autobahn::wamp_transport_handler>(session));

        // Make sure the continuation futures we use do not run out of scope prematurely.
        // Since we are only using one thread here this can cause the io service to block
        // as a future generated by a continuation will block waiting for its promise to be
        // fulfilled when it goes out of scope. This would prevent the session from receiving
        // responses from the router.
        boost::future<void> connect_future;
        boost::future<void> start_future;
        boost::future<void> join_future;
        boost::future<void> provide_future;

        connect_future = transport->connect().then([&](boost::future<void> connected) {
            try {
                connected.get();
            } catch (const std::exception& e) {
                std::cerr << e.what() << std::endl;
                io.stop();
                return;
            }

            std::cerr << "transport connected" << std::endl;

            start_future = session->start().then([&](boost::future<void> started) {
                try {
                    started.get();
                } catch (const std::exception& e) {
                    std::cerr << e.what() << std::endl;
                    io.stop();
                    return;
                }

                std::cerr << "session started" << std::endl;

                join_future = session->join(parameters->realm()).then([&](boost::future<uint64_t> joined) {
                    try {
                        std::cerr << "joined realm: " << joined.get() << std::endl;
                    } catch (const std::exception& e) {
                        std::cerr << e.what() << std::endl;
                        io.stop();
                        return;
                    }

                    provide_future = session->provide("com.examples.calculator.add2", &add2).then(
                        [&](boost::future<autobahn::wamp_registration> registration) {
                        try {
                            std::cerr << "registered procedure:" << registration.get().id() << std::endl;
                        } catch (const std::exception& e) {
                            std::cerr << e.what() << std::endl;
                            io.stop();
                            return;
                        }
                    });
                });
            });
        });

        std::cerr << "starting io service" << std::endl;

        io.run();

        std::cerr << "stopped io service" << std::endl;
    }
    catch (const std::exception& e) {
        std::cerr << "exception: " << e.what() << std::endl;
        return -1;
    }

    return 0;
}

max_stars_repo_name: Costallat/hunter

max_issues_repo_name: koinos/hunter

max_forks_repo_name: koinos/hunter


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

using namespace bc;
using namespace bc::config;

#define BASE58_ENCODED_A "vYxp6yFC7qiVtK1RcGQQt3L6EqTc8YhEDLnSMLqDvp8D"
#define BASE58_DECODED_A \
{{ \
    0x03, 0x1b, 0xab, 0x84, 0xe6, 0x87, 0xe3, 0x65, 0x14, 0xee, 0xaf, 0x5a, \
    0x01, 0x7c, 0x30, 0xd3, 0x2c, 0x1f, 0x59, 0xdd, 0x4e, 0xa6, 0x62, 0x9d, \
    0xa7, 0x97, 0x0c, 0xa3, 0x74, 0x51, 0x3d, 0xd0,  0x06 \
}}

BOOST_AUTO_TEST_SUITE(base58_tests)

BOOST_AUTO_TEST_CASE(base58__constructor__default__does_not_throw)
{
    BOOST_REQUIRE_NO_THROW(base58());
}

BOOST_AUTO_TEST_CASE(base58__constructor__valid_string_cast__decodes)
{
    const data_chunk original(BASE58_DECODED_A);
    const data_chunk instance(base58(BASE58_ENCODED_A));
    BOOST_REQUIRE(original == instance);
}

////BOOST_AUTO_TEST_CASE(base58__constructor__bogus_string__throws_invalid_option)
////{
////    //BX_REQUIRE_THROW_INVALID_OPTION_VALUE(base58("bo-gus"));
////}

////BOOST_AUTO_TEST_CASE(base58__constructor__copy_address_primitives__round_trips)
////{
////    //BX_SERIALIZE_COPY_ROUND_TRIP(base58, BASE58_ENCODED_A);
////}

BOOST_AUTO_TEST_SUITE_END()


text:
/*
 *  Copyright (c), 2017, Adrien Devresse <adrien.devresse@epfl.ch>
 *
 *  Distributed under the Boost Software License, Version 1.0.
 *    (See accompanying file LICENSE_1_0.txt or copy at
 *          http://www.boost.org/LICENSE_1_0.txt)
 *
 */
#ifndef H5EASY_HPP
#define H5EASY_HPP

#include <string>
#include <vector>

// optionally enable xtensor plug-in and load the library
#ifdef XTENSOR_VERSION_MAJOR
#ifndef H5_USE_XTENSOR
#define H5_USE_XTENSOR
#endif
#endif

#ifdef H5_USE_XTENSOR
#include <xtensor/xarray.hpp>
#include <xtensor/xtensor.hpp>
#endif

// optionally enable Eigen plug-in and load the library
#ifdef EIGEN_WORLD_VERSION
#ifndef H5_USE_EIGEN
#define H5_USE_EIGEN
#endif
#endif

#ifdef H5_USE_EIGEN
#include <Eigen/Eigen>
#endif

// optionally enable OpenCV plug-in and load the library
#ifdef CV_MAJOR_VERSION
#ifndef H5_USE_OPENCV
#define H5_USE_OPENCV
#endif
#endif

#ifdef H5_USE_OPENCV
#include <opencv2/opencv.hpp>
#endif

#include "H5File.hpp"

namespace H5Easy {

using h5::Attribute;
using h5::AtomicType;
using h5::Chunking;
using h5::DataSet;
using h5::DataSetCreateProps;
using h5::DataSpace;
using h5::Deflate;
using h5::Exception;
using h5::File;
using h5::ObjectType;
using h5::Shuffle;

///
/// \brief Write mode for DataSets
enum class DumpMode {
    Create = 0, /*!< Dump only if DataSet does not exist, otherwise throw. */
    Overwrite = 1 /*!< Create or overwrite if DataSet of correct shape exists, otherwise throw. */
};

///
/// \brief Enable/disable automatic flushing after write operations.
enum class Flush
{
    False = 0, /*!< No automatic flushing. */
    True = 1 /*!< Automatic flushing. */
};

///
/// \brief Set compression level for written DataSets.
class Compression
{
public:

    //
    // \brief Enable compression with the highest compression level (9).
    // or disable compression (set compression level to 0).
    explicit Compression(bool enable = true);

    //
    // \brief Set compression level.
    template <class T>
    Compression(T level);

    //
    // \brief Return compression level.
    inline unsigned get() const;

private:
    unsigned m_compression_level;
};

///
/// \brief Options for dumping data.
///
/// By default:
/// - DumpMode::Create
/// - Flush::True
/// - Compression(false)
/// - ChunkSize: automatic.
class DumpOptions
{
public:
    ///
    /// \brief Constructor: accept all defaults.
    DumpOptions() = default;

    ///
    /// \brief Constructor: overwrite (some of the) defaults.
    /// \param Any of DumpMode, Flush, Compression in arbitrary number and order.
    template <class... Args>
    DumpOptions(Args... args)
    {
        set(args...);
    }

    ///
    /// \brief Overwrite setting.
    /// \param mode: DumpMode.
    inline void set(DumpMode mode);

    ///
    /// \brief Overwrite setting.
    /// \param flush Flush.
    inline void set(Flush mode);

    ///
    /// \brief Overwrite setting.
    /// \param level Compression.
    inline void set(const Compression& level);

    ///
    /// \brief Overwrite settings.
    /// \param Any of DumpMode, Flush, Compression in arbitrary number and order.
    template <class T, class... Args>
    inline void set(T arg, Args... args);

    ///
    /// \brief Set chunk-size. If the input is rank (size) zero, automatic chunking is enabled.
    /// \param shape Chunk size along each dimension.
    template <class T>
    inline void setChunkSize(const std::vector<T>& shape);

    ///
    /// \brief Set chunk-size. If the input is rank (size) zero, automatic chunking is enabled.
    /// \param shape Chunk size along each dimension.
    inline void setChunkSize(std::initializer_list<size_t> shape);

    ///
    /// \brief Check to overwrite.
    inline bool overwrite() const;

    ///
    /// \brief Check to flush.
    inline bool flush() const;

    ///
    /// \brief Check to compress.
    inline bool compress() const;

    ///
    /// \brief Get compression level.
    inline unsigned getCompressionLevel() const;

    ///
    /// \brief Check if chunk-size is manually set (or should be computed automatically).
    inline bool isChunked() const;

    ///
    /// \brief Get chunk size.
    inline std::vector<hsize_t> getChunkSize() const;

private:
    bool m_overwrite = false;
    bool m_flush = true;
    unsigned m_compression_level = 0;
    std::vector<hsize_t> m_chunk_size = {};
};

///
/// \brief Get the size of an existing DataSet in an open HDF5 file.
///
/// \param file A readable opened file
/// \param path Path of the DataSet
///
/// \return Size of the DataSet
inline size_t getSize(const File& file, const std::string& path);

///
/// \brief Get the shape of an existing DataSet in an readable file.
///
/// \param file A readable opened file
/// \param path Path of the DataSet
///
/// \return the shape of the DataSet
inline std::vector<size_t> getShape(const File& file, const std::string& path);

///
/// \brief Write object (templated) to a (new) DataSet in an open HDF5 file.
///
/// \param file Writeable opened file
/// \param path Path of the DataSet
/// \param data Data to write
/// \param mode Write mode
///
/// \return The newly created DataSet
///
template <class T>
inline DataSet dump(File& file,
                    const std::string& path,
                    const T& data,
                    DumpMode mode = DumpMode::Create);

///
/// \brief Write object (templated) to a (new) DataSet in an open HDF5 file.
///
/// \param file Writeable opened file
/// \param path Path of the DataSet
/// \param data Data to write
/// \param options Dump options
///
/// \return The newly created DataSet
///
template <class T>
inline DataSet dump(File& file,
                    const std::string& path,
                    const T& data,
                    const DumpOptions& options);

///
/// \brief Write a scalar to a (new, extendible) DataSet in an open HDF5 file.
///
/// \param file opened File (has to be writeable)
/// \param path path of the DataSet
/// \param data the data to write
/// \param idx the indices to which to write
///
/// \return The newly created DataSet
///
template <class T>
inline DataSet dump(File& file,
                    const std::string& path,
                    const T& data,
                    const std::vector<size_t>& idx);

///
/// \brief Write a scalar to a (new, extendable) DataSet in an open HDF5 file.
///
/// \param file open File (has to be writeable)
/// \param path path of the DataSet
/// \param data the data to write
/// \param idx the indices to which to write
///
/// \return The newly created DataSet
///
template <class T>
inline DataSet dump(File& file,
                    const std::string& path,
                    const T& data,
                    const std::initializer_list<size_t>& idx);

///
/// \brief Write a scalar to a (new, extendible) DataSet in an open HDF5 file.
///
/// \param file opened File (has to be writeable)
/// \param path path of the DataSet
/// \param data the data to write
/// \param idx the indices to which to write
/// \param options Dump options
///
/// \return The newly created DataSet
///
template <class T>
inline DataSet dump(File& file,
                    const std::string& path,
                    const T& data,
                    const std::vector<size_t>& idx,
                    const DumpOptions& options);

///
/// \brief Write a scalar to a (new, extendible) DataSet in an open HDF5 file.
///
/// \param file opened File (has to be writeable)
/// \param path path of the DataSet
/// \param data the data to write
/// \param idx the indices to which to write
/// \param options Dump options
///
/// \return The newly created DataSet
///
template <class T>
inline DataSet dump(File& file,
                    const std::string& path,
                    const T& data,
                    const std::initializer_list<size_t>& idx,
                    const DumpOptions& options);

///
/// \brief Load entry "(i,j)" from a rank-two DataSet in an open HDF5 file to a scalar.
///
/// \param file opened File (has to be writeable)
/// \param idx the indices to load
/// \param path path of the DataSet
///
/// \return the read data
///
template <class T>
inline T load(const File& file, const std::string& path, const std::vector<size_t>& idx);

///
/// \brief Load a DataSet in an open HDF5 file to an object (templated).
///
/// \param file opened File (has to be writeable)
/// \param path path of the DataSet
///
/// \return the read data
///
template <class T>
inline T load(const File& file, const std::string& path);

///
/// \brief Write object (templated) to a (new) Attribute in an open HDF5 file.
///
/// \param file Writeable opened file
/// \param path Path of the DataSet
/// \param key Name of the attribute
/// \param data Data to write
/// \param mode Write mode
///
/// \return The newly created DataSet
///
template <class T>
inline Attribute dumpAttribute(File& file,
                               const std::string& path,
                               const std::string& key,
                               const T& data,
                               DumpMode mode = DumpMode::Create);

///
/// \brief Write object (templated) to a (new) Attribute in an open HDF5 file.
///
/// \param file Writeable opened file
/// \param path Path of the DataSet
/// \param key Name of the attribute
/// \param data Data to write
/// \param options Dump options
///
/// \return The newly created DataSet
///
template <class T>
inline Attribute dumpAttribute(File& file,
                               const std::string& path,
                               const std::string& key,
                               const T& data,
                               const DumpOptions& options);

///
/// \brief Load a Attribute in an open HDF5 file to an object (templated).
///
/// \param file opened File (has to be writeable)
/// \param path path of the DataSet
/// \param key Name of the attribute
///
/// \return the read data
///
template <class T>
inline T loadAttribute(const File& file, const std::string& path, const std::string& key);

}  // namespace H5Easy

#include "h5easy_bits/H5Easy_Eigen.hpp"
#include "h5easy_bits/H5Easy_misc.hpp"
#include "h5easy_bits/H5Easy_opencv.hpp"
#include "h5easy_bits/H5Easy_public.hpp"
#include "h5easy_bits/H5Easy_scalar.hpp"
#include "h5easy_bits/H5Easy_vector.hpp"
#include "h5easy_bits/H5Easy_xtensor.hpp"

#endif  // H5EASY_HPP

max_stars_repo_name: PolasekT/ICTree

max_issues_repo_name: PolasekT/ICTree

max_forks_repo_name: PolasekT/ICTree