text:
// Copyright 2012 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

<<<<<<< HEAD
#include <boost/algorithm/string.hpp>
#include "text-converter.h"
=======
#include "exec/hdfs-rcfile-scanner.h"

#include <boost/algorithm/string.hpp>

#include "exec/hdfs-scan-node.h"
#include "exec/hdfs-sequence-scanner.h"
#include "exec/scanner-context.inline.h"
#include "exec/text-converter.inline.h"
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
#include "exprs/expr.h"
#include "runtime/descriptors.h"
#include "runtime/runtime-state.h"
#include "runtime/mem-pool.h"
#include "runtime/raw-value.h"
<<<<<<< HEAD
#include "runtime/row-batch.h"
#include "runtime/tuple-row.h"
#include "runtime/tuple.h"
#include "runtime/hdfs-fs-cache.h"
#include "runtime/string-value.h"
#include "util/runtime-profile.h"
#include "common/object-pool.h"
#include "gen-cpp/PlanNodes_types.h"
#include "exec/buffered-byte-stream.h"
#include "exec/hdfs-rcfile-scanner.h"
#include "exec/hdfs-sequence-scanner.h"
#include "exec/hdfs-scan-node.h"
#include "exec/serde-utils.inline.h"
#include "exec/text-converter.inline.h"
=======
#include "runtime/tuple-row.h"
#include "runtime/tuple.h"
#include "runtime/string-value.h"
#include "util/codec.h"
#include "util/string-parser.h"

#include "gen-cpp/PlanNodes_types.h"
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa

using namespace std;
using namespace boost;
using namespace impala;

const char* const HdfsRCFileScanner::RCFILE_KEY_CLASS_NAME =
  "org.apache.hadoop.hive.ql.io.RCFile$KeyBuffer";

const char* const HdfsRCFileScanner::RCFILE_VALUE_CLASS_NAME =
  "org.apache.hadoop.hive.ql.io.RCFile$ValueBuffer";

const char* const HdfsRCFileScanner::RCFILE_METADATA_KEY_NUM_COLS =
  "hive.io.rcfile.column.number";

const uint8_t HdfsRCFileScanner::RCFILE_VERSION_HEADER[4] = {'R', 'C', 'F', 1};

<<<<<<< HEAD
static const uint8_t RC_FILE_RECORD_DELIMITER = 0xff;

HdfsRCFileScanner::HdfsRCFileScanner(HdfsScanNode* scan_node, RuntimeState* state,
    MemPool* mem_pool)
    : HdfsScanner(scan_node, state, mem_pool),
      scan_range_fully_buffered_(false),
      key_buffer_pool_(new MemPool()),
      key_buffer_length_(0),
      compressed_data_buffer_length_(0),
      column_buffer_pool_(new MemPool()) {
  // Initialize the parser to find bytes that are 0xff.
  find_first_parser_.reset(new DelimitedTextParser(scan_node, RC_FILE_RECORD_DELIMITER));
}

HdfsRCFileScanner::~HdfsRCFileScanner() {
  // Collect the maximum amount of memory we used to process this file.
  COUNTER_UPDATE(scan_node_->memory_used_counter(),
      column_buffer_pool_->peak_allocated_bytes());
  COUNTER_UPDATE(scan_node_->memory_used_counter(),
      key_buffer_pool_->peak_allocated_bytes());
}

Status HdfsRCFileScanner::Prepare() {
  RETURN_IF_ERROR(HdfsScanner::Prepare());

  text_converter_.reset(new TextConverter(0));

  // Allocate the buffers for the key information that is used to read and decode
  // the column data from its run length encoding.
  int num_part_keys = scan_node_->num_partition_keys();
  num_cols_ = scan_node_->num_cols() - num_part_keys;

  col_buf_len_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  col_buf_uncompressed_len_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  col_key_bufs_ = reinterpret_cast<uint8_t**>
      (key_buffer_pool_->Allocate(sizeof(uint8_t*) * num_cols_));

  col_key_bufs_len_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  col_bufs_off_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  key_buf_pos_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  cur_field_length_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  cur_field_length_rep_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  col_buf_pos_ = reinterpret_cast<int32_t*>(
      key_buffer_pool_->Allocate(sizeof(int32_t) * num_cols_));

  return Status::OK;
}

Status HdfsRCFileScanner::Close() {
  return Status::OK;
}

Status HdfsRCFileScanner::InitCurrentScanRange(HdfsPartitionDescriptor* hdfs_partition,
    DiskIoMgr::ScanRange* scan_range, Tuple* template_tuple,
    ByteStream* current_byte_stream_) {
  RETURN_IF_ERROR(HdfsScanner::InitCurrentScanRange(hdfs_partition, scan_range,
      template_tuple, current_byte_stream_));
  end_of_scan_range_ = scan_range->len() + scan_range->offset();

  // Check the Location (file name) to see if we have changed files.
  // If this a new file then we need to read and process the header.
  if (previous_location_ != current_byte_stream_->GetLocation()) {
    RETURN_IF_ERROR(current_byte_stream_->Seek(0));
    RETURN_IF_ERROR(ReadFileHeader());
    // Save the current file name if we get the same file we can avoid
    // rereading the header information.
    previous_location_ = current_byte_stream_->GetLocation();
    // Save the seek offset of the end of the header information.  If we
    // get a scan range that starts at the beginning of the file we can avoid
    // reading the header but we must seek past it to get to the beginning of the data.
    current_byte_stream_->GetPosition(&header_end_);
  } else if (scan_range->offset() == 0) {
    // If are at the beginning of the file and we previously read the file header
    // we do not have to read it again but we need to seek past the file header.
    // We saved the offset above when we previously read and processed the header.
    RETURN_IF_ERROR(current_byte_stream_->Seek(header_end_));
  }

  // Offset may not point to row group boundary so we need to search for the next
  // sync block.
  if (scan_range->offset() != 0) {
    RETURN_IF_ERROR(current_byte_stream_->Seek(scan_range->offset()));
    RETURN_IF_ERROR(find_first_parser_->FindSyncBlock(end_of_scan_range_,
          SYNC_HASH_SIZE, &(sync_hash_[0]), current_byte_stream_));
  }
  int64_t position;
  RETURN_IF_ERROR(current_byte_stream_->GetPosition(&position));

  ResetRowGroup();

  scan_range_fully_buffered_ = false;
  previous_total_length_ = 0;
  tuple_ = NULL;

=======
// Macro to convert between SerdeUtil errors to Status returns.
#define RETURN_IF_FALSE(x) if (UNLIKELY(!(x))) return parse_status_

HdfsRCFileScanner::HdfsRCFileScanner(HdfsScanNode* scan_node, RuntimeState* state)
    : BaseSequenceScanner(scan_node, state) {
}

HdfsRCFileScanner::~HdfsRCFileScanner() {
}

Status HdfsRCFileScanner::Prepare(ScannerContext* context) {
  RETURN_IF_ERROR(BaseSequenceScanner::Prepare(context));
  text_converter_.reset(
      new TextConverter(0, scan_node_->hdfs_table()->null_column_value()));
  scan_node_->IncNumScannersCodegenDisabled();
  return Status::OK;
}

Status HdfsRCFileScanner::InitNewRange() {
  DCHECK(header_ != NULL);

  only_parsing_header_ = false;
  row_group_buffer_size_ = 0;

  // Can reuse buffer if there are no string columns (since the tuple won't contain
  // ptrs into the decompressed data).
  reuse_row_group_buffer_ = scan_node_->tuple_desc()->string_slots().empty();

  // The scanner currently copies all the column data out of the io buffer so the
  // stream never contains any tuple data.
  stream_->set_contains_tuple_data(false);

  if (header_->is_compressed) {
    RETURN_IF_ERROR(Codec::CreateDecompressor(NULL,
        reuse_row_group_buffer_, header_->codec, &decompressor_));
  }

  // Allocate the buffers for the key information that is used to read and decode
  // the column data.
  columns_.resize(reinterpret_cast<RcFileHeader*>(header_)->num_cols);
  int num_table_cols = scan_node_->num_cols() - scan_node_->num_partition_keys();
  for (int i = 0; i < columns_.size(); ++i) {
    if (i < num_table_cols) {
      int col_idx = i + scan_node_->num_partition_keys();
      columns_[i].materialize_column =
          scan_node_->GetMaterializedSlotIdx(col_idx) != HdfsScanNode::SKIP_COLUMN;
    } else {
      // Treat columns not found in table metadata as extra unmaterialized columns
      columns_[i].materialize_column = false;
    }
  }

  // TODO: Initialize codegen fn here
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
  return Status::OK;
}

Status HdfsRCFileScanner::ReadFileHeader() {
<<<<<<< HEAD
  vector<uint8_t> head;
  RETURN_IF_ERROR(SerDeUtils::ReadBytes(current_byte_stream_,
      sizeof(RCFILE_VERSION_HEADER), &head));
  if (!memcmp(&head[0], HdfsSequenceScanner::SEQFILE_VERSION_HEADER,
      sizeof(HdfsSequenceScanner::SEQFILE_VERSION_HEADER))) {
    version_ = SEQ6;
  } else if (!memcmp(&head[0], RCFILE_VERSION_HEADER, sizeof(RCFILE_VERSION_HEADER))) {
    version_ = RCF1;
  } else {
    stringstream ss;
    ss << "Invalid RCFILE_VERSION_HEADER: '"
       << SerDeUtils::HexDump(&head[0], sizeof(RCFILE_VERSION_HEADER)) << "'";
    return Status(ss.str());
  }

  if (version_ == SEQ6) {
    vector<char> buf;
    RETURN_IF_ERROR(SerDeUtils::ReadText(current_byte_stream_, &buf));
    if (strncmp(&buf[0], HdfsRCFileScanner::RCFILE_KEY_CLASS_NAME,
        strlen(HdfsRCFileScanner::RCFILE_KEY_CLASS_NAME))) {
      stringstream ss;
      ss << "Invalid RCFILE_KEY_CLASS_NAME: '"
         << string(&buf[0], strlen(HdfsRCFileScanner::RCFILE_KEY_CLASS_NAME))
         << "'";
      return Status(ss.str());
    }

    RETURN_IF_ERROR(SerDeUtils::ReadText(current_byte_stream_, &buf));
    if (strncmp(&buf[0], HdfsRCFileScanner::RCFILE_VALUE_CLASS_NAME,
        strlen(HdfsRCFileScanner::RCFILE_VALUE_CLASS_NAME))) {
      stringstream ss;
      ss << "Invalid RCFILE_VALUE_CLASS_NAME: '"
         << string(&buf[0], strlen(HdfsRCFileScanner::RCFILE_VALUE_CLASS_NAME))
         << "'";
=======
  uint8_t* header;

  RcFileHeader* rc_header = reinterpret_cast<RcFileHeader*>(header_);
  // Validate file version
  RETURN_IF_FALSE(stream_->ReadBytes(
      sizeof(RCFILE_VERSION_HEADER), &header, &parse_status_));
  if (!memcmp(header, HdfsSequenceScanner::SEQFILE_VERSION_HEADER,
      sizeof(HdfsSequenceScanner::SEQFILE_VERSION_HEADER))) {
    rc_header->version = SEQ6;
  } else if (!memcmp(header, RCFILE_VERSION_HEADER, sizeof(RCFILE_VERSION_HEADER))) {
    rc_header->version = RCF1;
  } else {
    stringstream ss;
    ss << "Invalid RCFILE_VERSION_HEADER: '"
       << ReadWriteUtil::HexDump(header, sizeof(RCFILE_VERSION_HEADER)) << "'";
    return Status(ss.str());
  }

  if (rc_header->version == SEQ6) {
    // Validate class name key/value
    uint8_t* class_name_key;
    int64_t len;
    RETURN_IF_FALSE(
        stream_->ReadText(&class_name_key, &len, &parse_status_));
    if (len != strlen(HdfsRCFileScanner::RCFILE_KEY_CLASS_NAME) ||
        memcmp(class_name_key, HdfsRCFileScanner::RCFILE_KEY_CLASS_NAME, len)) {
      stringstream ss;
      ss << "Invalid RCFILE_KEY_CLASS_NAME: '"
         << string(reinterpret_cast<char*>(class_name_key), len)
         << "' len=" << len;
      return Status(ss.str());
    }

    uint8_t* class_name_val;
    RETURN_IF_FALSE(
        stream_->ReadText(&class_name_val, &len, &parse_status_));
    if (len != strlen(HdfsRCFileScanner::RCFILE_VALUE_CLASS_NAME) ||
        memcmp(class_name_val, HdfsRCFileScanner::RCFILE_VALUE_CLASS_NAME, len)) {
      stringstream ss;
      ss << "Invalid RCFILE_VALUE_CLASS_NAME: '"
         << string(reinterpret_cast<char*>(class_name_val), len)
         << "' len=" << len;
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
      return Status(ss.str());
    }
  }

<<<<<<< HEAD
  RETURN_IF_ERROR(SerDeUtils::ReadBoolean(current_byte_stream_, &is_compressed_));

  if (version_ == SEQ6) {
    // Read the is_blk_compressed header field. This field should *always*
    // be FALSE, and is the result of a defect in the original RCFile
    // implementation contained in Hive.
    bool is_blk_compressed;
    RETURN_IF_ERROR(SerDeUtils::ReadBoolean(current_byte_stream_, &is_blk_compressed));
    if (is_blk_compressed) {
      stringstream ss;
      ss << "RC files do no support block compression, set in: '"
         << current_byte_stream_->GetLocation() << "'";
      return Status(ss.str());
    }
  }

  vector<char> codec;
  if (is_compressed_) {
    // Read the codec and get the right decompressor class.
    RETURN_IF_ERROR(SerDeUtils::ReadText(current_byte_stream_, &codec));
    string codec_str(&codec[0], codec.size());
    RETURN_IF_ERROR(Codec::CreateDecompressor(state_,
        column_buffer_pool_.get(), !has_noncompact_strings_, codec_str, &decompressor_));
  }

  VLOG_FILE << current_byte_stream_->GetLocation() << ": "
            << (is_compressed_ ?  "block compressed" : "not compressed");
  if (is_compressed_) VLOG_FILE << string(&codec[0], codec.size());

  RETURN_IF_ERROR(ReadFileHeaderMetadata());

  RETURN_IF_ERROR(SerDeUtils::ReadBytes(current_byte_stream_,
      HdfsRCFileScanner::SYNC_HASH_SIZE, &sync_hash_[0]));
  return Status::OK;
}

Status HdfsRCFileScanner::ReadFileHeaderMetadata() {
  int map_size = 0;
  vector<char> key;
  vector<char> value;

  RETURN_IF_ERROR(SerDeUtils::ReadInt(current_byte_stream_, &map_size));

  for (int i = 0; i < map_size; ++i) {
    RETURN_IF_ERROR(SerDeUtils::ReadText(current_byte_stream_, &key));
    RETURN_IF_ERROR(SerDeUtils::ReadText(current_byte_stream_, &value));

    if (!strncmp(&key[0], HdfsRCFileScanner::RCFILE_METADATA_KEY_NUM_COLS,
        strlen(HdfsRCFileScanner::RCFILE_METADATA_KEY_NUM_COLS))) {
      string tmp(&value[0], value.size());
      int file_num_cols = atoi(tmp.c_str());
      if (file_num_cols != num_cols_) {
        return Status("Unexpected hive.io.rcfile.column.number value!");
      }
=======
  // Check for compression
  RETURN_IF_FALSE(
      stream_->ReadBoolean(&header_->is_compressed, &parse_status_));
  if (rc_header->version == SEQ6) {
    // Read the is_blk_compressed header field. This field should *always*
    // be FALSE, and is the result of using the sequence file header format in the
    // original RCFile format.
    bool is_blk_compressed;
    RETURN_IF_FALSE(
        stream_->ReadBoolean(&is_blk_compressed, &parse_status_));
    if (is_blk_compressed) {
      stringstream ss;
      ss << "RC files do no support block compression.";
      return Status(ss.str());
    }
  }
  if (header_->is_compressed) {
    uint8_t* codec_ptr;
    int64_t len;
    // Read the codec and get the right decompressor class.
    RETURN_IF_FALSE(stream_->ReadText(&codec_ptr, &len, &parse_status_));
    header_->codec = string(reinterpret_cast<char*>(codec_ptr), len);
    Codec::CodecMap::const_iterator it = Codec::CODEC_MAP.find(header_->codec);
    DCHECK(it != Codec::CODEC_MAP.end());
    header_->compression_type = it->second;
  } else {
    header_->compression_type = THdfsCompression::NONE;
  }
  VLOG_FILE << stream_->filename() << ": "
            << (header_->is_compressed ?  "compressed" : "not compressed");
  if (header_->is_compressed) VLOG_FILE << header_->codec;

  RETURN_IF_ERROR(ReadNumColumnsMetadata());

  // Read file sync marker
  uint8_t* sync;
  RETURN_IF_FALSE(stream_->ReadBytes(SYNC_HASH_SIZE, &sync, &parse_status_));
  memcpy(header_->sync, sync, SYNC_HASH_SIZE);

  header_->header_size = stream_->total_bytes_returned() - SYNC_HASH_SIZE;
  return Status::OK;
}

Status HdfsRCFileScanner::ReadNumColumnsMetadata() {
  int map_size = 0;
  RETURN_IF_FALSE(stream_->ReadInt(&map_size, &parse_status_));

  for (int i = 0; i < map_size; ++i) {
    uint8_t* key, *value;
    int64_t key_len, value_len;
    RETURN_IF_FALSE(stream_->ReadText(&key, &key_len, &parse_status_));
    RETURN_IF_FALSE(stream_->ReadText(&value, &value_len, &parse_status_));

    if (key_len == strlen(RCFILE_METADATA_KEY_NUM_COLS) &&
        !memcmp(key, HdfsRCFileScanner::RCFILE_METADATA_KEY_NUM_COLS, key_len)) {
      string value_str(reinterpret_cast<char*>(value), value_len);
      StringParser::ParseResult result;
      int num_cols =
          StringParser::StringToInt<int>(value_str.c_str(), value_str.size(), &result);
      if (result != StringParser::PARSE_SUCCESS) {
        stringstream ss;
        ss << "Could not parse number of columns in file " << stream_->filename()
           << ": " << value_str;
        if (result == StringParser::PARSE_OVERFLOW) ss << " (result overflowed)";
        return Status(ss.str());
      }
      RcFileHeader* rc_header = reinterpret_cast<RcFileHeader*>(header_);
      rc_header->num_cols = num_cols;
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
    }
  }
  return Status::OK;
}

<<<<<<< HEAD
Status HdfsRCFileScanner::ReadSync() {
  uint8_t hash[HdfsRCFileScanner::SYNC_HASH_SIZE];
  RETURN_IF_ERROR(SerDeUtils::ReadBytes(current_byte_stream_,
      HdfsRCFileScanner::SYNC_HASH_SIZE, &hash[0]));
  if (memcmp(&hash[0], &sync_hash_[0], HdfsRCFileScanner::SYNC_HASH_SIZE)) {
    if (state_->LogHasSpace()) {
      stringstream ss;
      ss  << "Bad sync hash in current HdfsRCFileScanner: "
          << current_byte_stream_->GetLocation() << "." << endl
          << "Expected: '"
          << SerDeUtils::HexDump(sync_hash_, HdfsRCFileScanner::SYNC_HASH_SIZE)
          << "'" << endl
          << "Actual:   '"
          << SerDeUtils::HexDump(hash, HdfsRCFileScanner::SYNC_HASH_SIZE)
          << "'" << endl;
      state_->LogError(ss.str());
    }
    return Status("bad sync hash block");
  }
  return Status::OK;
=======
BaseSequenceScanner::FileHeader* HdfsRCFileScanner::AllocateFileHeader() {
  return new RcFileHeader;
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
}

void HdfsRCFileScanner::ResetRowGroup() {
  num_rows_ = 0;
  row_pos_ = 0;
  key_length_ = 0;
  compressed_key_length_ = 0;

<<<<<<< HEAD
  memset(col_buf_len_, 0, num_cols_ * sizeof(int32_t));
  memset(col_buf_uncompressed_len_, 0, num_cols_ * sizeof(int32_t));
  memset(col_key_bufs_len_, 0, num_cols_ * sizeof(int32_t));

  memset(key_buf_pos_, 0, num_cols_ * sizeof(int32_t));
  memset(cur_field_length_, 0, num_cols_ * sizeof(int32_t));
  memset(cur_field_length_rep_, 0, num_cols_ * sizeof(int32_t));

  memset(col_buf_pos_, 0, num_cols_ * sizeof(int32_t));
=======
  for (int i = 0; i < columns_.size(); ++i) {
    columns_[i].buffer_len = 0;
    columns_[i].buffer_pos = 0;
    columns_[i].uncompressed_buffer_len = 0;
    columns_[i].key_buffer_len = 0;
    columns_[i].key_buffer_pos = 0;
    columns_[i].current_field_len = 0;
    columns_[i].current_field_len_rep = 0;
  }

  // We are done with this row group, pass along external buffers if necessary.
  if (!reuse_row_group_buffer_) {
    AttachPool(data_buffer_pool_.get(), true);
    row_group_buffer_size_ = 0;
  }
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
}

Status HdfsRCFileScanner::ReadRowGroup() {
  ResetRowGroup();
<<<<<<< HEAD
  int64_t position;
  bool eof;

  while (num_rows_ == 0) {
    RETURN_IF_ERROR(ReadHeader());
    RETURN_IF_ERROR(ReadKeyBuffers());
    if (has_noncompact_strings_ || previous_total_length_ < total_col_length_) {
      column_buffer_ = column_buffer_pool_->Allocate(total_col_length_);
      previous_total_length_ = total_col_length_;
    }
    RETURN_IF_ERROR(ReadColumnBuffers());
    RETURN_IF_ERROR(current_byte_stream_->GetPosition(&position));
    if (position >= end_of_scan_range_) {
      RETURN_IF_ERROR(current_byte_stream_->Eof(&eof));
      if (eof) {
        scan_range_fully_buffered_ = true;
        break;
      }

      // We must read up to the next sync marker.
      int32_t record_length;
      RETURN_IF_ERROR(SerDeUtils::ReadInt(current_byte_stream_, &record_length));
      if (record_length == HdfsRCFileScanner::SYNC_MARKER) {
        // If the marker is there, it's an error not to have a Sync block following the
        // Marker.
        RETURN_IF_ERROR(ReadSync());
        scan_range_fully_buffered_ = true;
        break;
      } else {
        RETURN_IF_ERROR(current_byte_stream_->Seek(position));
      }
    }
=======

  while (num_rows_ == 0) {
    RETURN_IF_ERROR(ReadRowGroupHeader());
    RETURN_IF_ERROR(ReadKeyBuffers());
    if (!reuse_row_group_buffer_ || row_group_buffer_size_ < row_group_length_) {
      // Allocate a new buffer for reading the row group.  Row groups have a
      // fixed number of rows so take a guess at how big it will be based on
      // the previous row group size.
      // The row group length depends on the user data and can be very big. This
      // can cause us to go way over the mem limit so use TryAllocate instead.
      row_group_buffer_ = data_buffer_pool_->TryAllocate(row_group_length_);
      if (row_group_length_ > 0 && row_group_buffer_ == NULL) {
        return state_->SetMemLimitExceeded(
            scan_node_->mem_tracker(), row_group_length_);
      }
      row_group_buffer_size_ = row_group_length_;
    }
    RETURN_IF_ERROR(ReadColumnBuffers());
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
  }
  return Status::OK;
}

<<<<<<< HEAD
Status HdfsRCFileScanner::ReadHeader() {
  int32_t record_length;
  RETURN_IF_ERROR(SerDeUtils::ReadInt(current_byte_stream_, &record_length));
  // The sync block is marked with a record_length of -1.
  if (record_length == HdfsRCFileScanner::SYNC_MARKER) {
    RETURN_IF_ERROR(ReadSync());
    RETURN_IF_ERROR(SerDeUtils::ReadInt(current_byte_stream_, &record_length));
  }
  if (record_length < 0) {
    stringstream ss;
    int64_t position;
    current_byte_stream_->GetPosition(&position);
    position -= sizeof(int32_t);
    ss << "Bad record length: " << record_length << " in file: "
        << current_byte_stream_->GetLocation() << " at offset: " << position;
    return Status(ss.str());
  }
  RETURN_IF_ERROR(SerDeUtils::ReadInt(current_byte_stream_, &key_length_));
  if (key_length_ < 0) {
    stringstream ss;
    int64_t position;
    current_byte_stream_->GetPosition(&position);
    position -= sizeof(int32_t);
    ss << "Bad key length: " << key_length_ << " in file: "
        << current_byte_stream_->GetLocation() << " at offset: " << position;
    return Status(ss.str());
  }
  RETURN_IF_ERROR(SerDeUtils::ReadInt(current_byte_stream_, &compressed_key_length_));
  if (compressed_key_length_ < 0) {
    stringstream ss;
    int64_t position;
    current_byte_stream_->GetPosition(&position);
    position -= sizeof(int32_t);
    ss << "Bad compressed key length: " << compressed_key_length_ << " in file: "
        << current_byte_stream_->GetLocation() << " at offset: " << position;
=======
Status HdfsRCFileScanner::ReadRowGroupHeader() {
  int32_t record_length;
  RETURN_IF_FALSE(stream_->ReadInt(&record_length, &parse_status_));
  if (record_length < 0) {
    stringstream ss;
    int64_t position = stream_->file_offset();
    position -= sizeof(int32_t);
    ss << "Bad record length: " << record_length << " at offset: " << position;
    return Status(ss.str());
  }
  RETURN_IF_FALSE(stream_->ReadInt(&key_length_, &parse_status_));
  if (key_length_ < 0) {
    stringstream ss;
    int64_t position = stream_->file_offset();
    position -= sizeof(int32_t);
    ss << "Bad key length: " << key_length_ << " at offset: " << position;
    return Status(ss.str());
  }
  RETURN_IF_FALSE(stream_->ReadInt(&compressed_key_length_, &parse_status_));
  if (compressed_key_length_ < 0) {
    stringstream ss;
    int64_t position = stream_->file_offset();
    position -= sizeof(int32_t);
    ss << "Bad compressed key length: " << compressed_key_length_
       << " at offset: " << position;
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
    return Status(ss.str());
  }
  return Status::OK;
}

Status HdfsRCFileScanner::ReadKeyBuffers() {
<<<<<<< HEAD
  if (key_buffer_length_ < key_length_) {
    key_buffer_ = key_buffer_pool_->Allocate(key_length_);
    key_buffer_length_ = key_length_;
  }
  if (is_compressed_) {
    if (compressed_data_buffer_length_ < compressed_key_length_) {
      compressed_data_buffer_ = key_buffer_pool_->Allocate(compressed_key_length_);
      compressed_data_buffer_length_ = compressed_key_length_;
    }
    RETURN_IF_ERROR(SerDeUtils::ReadBytes(current_byte_stream_,
        compressed_key_length_, compressed_data_buffer_));
    RETURN_IF_ERROR(decompressor_->ProcessBlock(compressed_key_length_,
        compressed_data_buffer_, &key_length_, &key_buffer_));
  } else {
    RETURN_IF_ERROR(
        SerDeUtils::ReadBytes(current_byte_stream_, key_length_, key_buffer_));
  }

  total_col_length_ = 0;
  uint8_t* key_buf_ptr = key_buffer_;
  int bytes_read = SerDeUtils::GetVInt(key_buf_ptr, &num_rows_);
  key_buf_ptr += bytes_read;

  for (int col_idx = 0; col_idx < num_cols_; ++col_idx) {
    GetCurrentKeyBuffer(col_idx, !ReadColumn(col_idx), &key_buf_ptr);
    DCHECK_LE(key_buf_ptr, key_buffer_ + key_length_);
  }
  DCHECK_EQ(key_buf_ptr, key_buffer_ + key_length_);
=======
  if (key_buffer_.size() < key_length_) key_buffer_.resize(key_length_);
  uint8_t* key_buffer = &key_buffer_[0];

  if (header_->is_compressed) {
    uint8_t* compressed_buffer;
    RETURN_IF_FALSE(stream_->ReadBytes(
        compressed_key_length_, &compressed_buffer, &parse_status_));
    {
      SCOPED_TIMER(decompress_timer_);
      RETURN_IF_ERROR(decompressor_->ProcessBlock32(true, compressed_key_length_,
          compressed_buffer, &key_length_, &key_buffer));
      VLOG_FILE << "Decompressed " << compressed_key_length_ << " to " << key_length_;
    }
  } else {
    uint8_t* buffer;
    RETURN_IF_FALSE(
        stream_->ReadBytes(key_length_, &buffer, &parse_status_));
    // Make a copy of this buffer.  The underlying IO buffer will get recycled
    memcpy(key_buffer, buffer, key_length_);
  }

  row_group_length_ = 0;
  uint8_t* key_buf_ptr = key_buffer;
  int bytes_read = ReadWriteUtil::GetVInt(key_buf_ptr, &num_rows_);
  key_buf_ptr += bytes_read;

  for (int col_idx = 0; col_idx < columns_.size(); ++col_idx) {
    GetCurrentKeyBuffer(col_idx, !columns_[col_idx].materialize_column, &key_buf_ptr);
    DCHECK_LE(key_buf_ptr, key_buffer + key_length_);
  }
  DCHECK_EQ(key_buf_ptr, key_buffer + key_length_);
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa

  return Status::OK;
}

void HdfsRCFileScanner::GetCurrentKeyBuffer(int col_idx, bool skip_col_data,
                                            uint8_t** key_buf_ptr) {
<<<<<<< HEAD
  int bytes_read = SerDeUtils::GetVInt(*key_buf_ptr, &(col_buf_len_[col_idx]));
  *key_buf_ptr += bytes_read;

  bytes_read = SerDeUtils::GetVInt(*key_buf_ptr, &(col_buf_uncompressed_len_[col_idx]));
  *key_buf_ptr += bytes_read;

  int col_key_buf_len;
  bytes_read = SerDeUtils::GetVInt(*key_buf_ptr , &col_key_buf_len);
  *key_buf_ptr += bytes_read;

  if (!skip_col_data) {
    col_key_bufs_[col_idx] = *key_buf_ptr;

    // Set the offset for the start of the data for this column in the allocated buffer.
    col_bufs_off_[col_idx] = total_col_length_;
    total_col_length_ += col_buf_uncompressed_len_[col_idx];
=======
  ColumnInfo& col_info = columns_[col_idx];

  int bytes_read = ReadWriteUtil::GetVInt(*key_buf_ptr, &col_info.buffer_len);
  *key_buf_ptr += bytes_read;

  bytes_read = ReadWriteUtil::GetVInt(*key_buf_ptr, &col_info.uncompressed_buffer_len);
  *key_buf_ptr += bytes_read;

  int col_key_buf_len;
  bytes_read = ReadWriteUtil::GetVInt(*key_buf_ptr , &col_key_buf_len);
  *key_buf_ptr += bytes_read;

  if (!skip_col_data) {
    col_info.key_buffer = *key_buf_ptr;

    // Set the offset for the start of the data for this column in the allocated buffer.
    col_info.start_offset = row_group_length_;
    row_group_length_ += col_info.uncompressed_buffer_len;
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
  }
  *key_buf_ptr += col_key_buf_len;
}

inline Status HdfsRCFileScanner::NextField(int col_idx) {
<<<<<<< HEAD
  col_buf_pos_[col_idx] += cur_field_length_[col_idx];

  if (cur_field_length_rep_[col_idx] > 0) {
    // repeat the previous length
    --cur_field_length_rep_[col_idx];
  } else {
    DCHECK_GE(cur_field_length_rep_[col_idx], 0);
    // Get the next column length or repeat count
    int64_t length = 0;
    uint8_t* col_key_buf = &col_key_bufs_[col_idx][0];
    int bytes_read = SerDeUtils::GetVLong(col_key_buf, key_buf_pos_[col_idx], &length);
    if (bytes_read == -1) {
        int64_t position;
        current_byte_stream_->GetPosition(&position);
        stringstream ss;
        ss << "Invalid column length in file: "
           << current_byte_stream_->GetLocation() << " at offset: " << position;
        if (state_->LogHasSpace()) state_->LogError(ss.str());
        return Status(ss.str());
    }
    key_buf_pos_[col_idx] += bytes_read;
=======
  ColumnInfo& col_info = columns_[col_idx];
  col_info.buffer_pos += col_info.current_field_len;

  if (col_info.current_field_len_rep > 0) {
    // repeat the previous length
    --col_info.current_field_len_rep;
  } else {
    // Get the next column length or repeat count
    int64_t length = 0;
    uint8_t* col_key_buf = col_info.key_buffer;
    int bytes_read = ReadWriteUtil::GetVLong(
        col_key_buf, col_info.key_buffer_pos, &length);
    if (bytes_read == -1) {
        int64_t position = stream_->file_offset();
        stringstream ss;
        ss << "Invalid column length at offset: " << position;
        return Status(ss.str());
    }
    col_info.key_buffer_pos += bytes_read;
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa

    if (length < 0) {
      // The repeat count is stored as the logical negation of the number of repetitions.
      // See the column-key-buffer comment in hdfs-rcfile-scanner.h.
<<<<<<< HEAD
      cur_field_length_rep_[col_idx] = ~length - 1;
    } else {
      cur_field_length_[col_idx] = length;
=======
      col_info.current_field_len_rep = ~length - 1;
    } else {
      col_info.current_field_len = length;
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
    }
  }
  return Status::OK;
}

<<<<<<< HEAD
inline Status HdfsRCFileScanner::NextRow(bool* eorg) {
  // TODO: Wrap this in an iterator and prevent people from alternating
  // calls to NextField()/NextRow()
  if (row_pos_ >= num_rows_) {
    *eorg = true;
    return Status::OK;
  }
  for (int col_idx = 0; col_idx < num_cols_; ++col_idx) {
    if (ReadColumn(col_idx)) {
=======
inline Status HdfsRCFileScanner::NextRow() {
  // TODO: Wrap this in an iterator and prevent people from alternating
  // calls to NextField()/NextRow()
  DCHECK_LT(row_pos_, num_rows_);
  for (int col_idx = 0; col_idx < columns_.size(); ++col_idx) {
    if (columns_[col_idx].materialize_column) {
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
      RETURN_IF_ERROR(NextField(col_idx));
    }
  }
  ++row_pos_;
<<<<<<< HEAD
  *eorg = false;
=======
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
  return Status::OK;
}

Status HdfsRCFileScanner::ReadColumnBuffers() {
<<<<<<< HEAD
  for (int col_idx = 0; col_idx < num_cols_; ++col_idx) {
    if (!ReadColumn(col_idx)) {
      RETURN_IF_ERROR(SerDeUtils::SkipBytes(current_byte_stream_, col_buf_len_[col_idx]));
    } else {
      // TODO: Stream through these column buffers instead of reading everything
      // in at once.
      DCHECK_LE(
          col_buf_uncompressed_len_[col_idx] + col_bufs_off_[col_idx], total_col_length_);
      if (is_compressed_) {
        if (compressed_data_buffer_length_ < col_buf_len_[col_idx]) {
          compressed_data_buffer_ = key_buffer_pool_->Allocate(col_buf_len_[col_idx]);
          compressed_data_buffer_length_ = col_buf_len_[col_idx];
        }
        RETURN_IF_ERROR(SerDeUtils::ReadBytes(current_byte_stream_,
            col_buf_len_[col_idx], compressed_data_buffer_));
        uint8_t* compressed_output = column_buffer_ + col_bufs_off_[col_idx];
        RETURN_IF_ERROR(decompressor_->ProcessBlock(col_buf_len_[col_idx],
            compressed_data_buffer_, &col_buf_uncompressed_len_[col_idx],
            &compressed_output));
      } else {
        RETURN_IF_ERROR(SerDeUtils::ReadBytes(current_byte_stream_,
            col_buf_len_[col_idx], column_buffer_ + col_bufs_off_[col_idx]));
      }
=======
  for (int col_idx = 0; col_idx < columns_.size(); ++col_idx) {
    ColumnInfo& column = columns_[col_idx];
    if (!columns_[col_idx].materialize_column) {
      // Not materializing this column, just skip it.
      RETURN_IF_FALSE(
          stream_->SkipBytes(column.buffer_len, &parse_status_));
      continue;
    }

    // TODO: Stream through these column buffers instead of reading everything
    // in at once.
    DCHECK_LE(column.uncompressed_buffer_len + column.start_offset, row_group_length_);
    if (header_->is_compressed) {
      uint8_t* compressed_input;
      RETURN_IF_FALSE(stream_->ReadBytes(
          column.buffer_len, &compressed_input, &parse_status_));
      uint8_t* compressed_output = row_group_buffer_ + column.start_offset;
      {
        SCOPED_TIMER(decompress_timer_);
        RETURN_IF_ERROR(decompressor_->ProcessBlock32(true, column.buffer_len,
            compressed_input, &column.uncompressed_buffer_len,
            &compressed_output));
        VLOG_FILE << "Decompressed " << column.buffer_len << " to "
                  << column.uncompressed_buffer_len;
      }
    } else {
      uint8_t* uncompressed_data;
      RETURN_IF_FALSE(stream_->ReadBytes(
          column.buffer_len, &uncompressed_data, &parse_status_));
      // TODO: this is bad.  Remove this copy.
      memcpy(row_group_buffer_ + column.start_offset,
          uncompressed_data, column.buffer_len);
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
    }
  }
  return Status::OK;
}

<<<<<<< HEAD
// TODO: We should be able to skip over badly fomrated data and move to the
//       next sync block to restart the scan.
Status HdfsRCFileScanner::GetNext(RowBatch* row_batch, bool* eosr) {
  AllocateTupleBuffer(row_batch);
  // Indicates whether the current row has errors.
  bool error_in_row = false;

  if (scan_node_->ReachedLimit()) {
    tuple_ = NULL;
    *eosr = true;
    return Status::OK;
  }

  // HdfsRCFileScanner effectively does buffered IO, in that it reads the
  // rcfile into a column buffer, and sets an eof marker at that
  // point, before the buffer is drained.
  // In the following loop, eosr tracks the end of stream flag that
  // tells the scan node to move onto the next scan range.
  // scan_range_fully_buffered_ is set once the RC file has been fully
  // read into RCRowGroups.
  while (!scan_node_->ReachedLimit() && !row_batch->IsFull() && !(*eosr)) {
    if (num_rows_ == row_pos_) {
      // scan_range_fully_buffered_ is set iff the scan range has been
      // exhausted due to a previous read.
      if (scan_range_fully_buffered_) {
        *eosr = true;
        break;
      }
      RETURN_IF_ERROR(ReadRowGroup());

      if (num_rows_ == 0) break;
    }

    SCOPED_TIMER(scan_node_->materialize_tuple_timer());
    // Copy rows out of the current row group into the row_batch
    while (!scan_node_->ReachedLimit() && !row_batch->IsFull()) {
      bool eorg = false;
      RETURN_IF_ERROR(NextRow(&eorg));
      if (eorg) break;

      // Index into current row in row_batch.
      int row_idx = row_batch->AddRow();
      DCHECK(row_idx != RowBatch::INVALID_ROW_INDEX);
      TupleRow* current_row = row_batch->GetRow(row_idx);
      current_row->SetTuple(scan_node_->tuple_idx(), tuple_);
      // Initialize tuple_ from the partition key template tuple before writing the
      // slots
      InitTuple(template_tuple_, tuple_);

      const vector<SlotDescriptor*>& materialized_slots =
          scan_node_->materialized_slots();
      vector<SlotDescriptor*>::const_iterator it;

      for (it = materialized_slots.begin(); it != materialized_slots.end(); ++it) {
        const SlotDescriptor* slot_desc = *it;
        int rc_column_idx = slot_desc->col_pos() - scan_node_->num_partition_keys();

        const char* col_start = reinterpret_cast<const char*>(&((column_buffer_ +
            col_bufs_off_[rc_column_idx])[col_buf_pos_[rc_column_idx]]));
        int field_len = cur_field_length_[rc_column_idx];
        DCHECK_LE(col_start + field_len,
            reinterpret_cast<const char*>(column_buffer_ + total_col_length_));

        if (!text_converter_->WriteSlot(slot_desc, tuple_,
              col_start, field_len, !has_noncompact_strings_, false, tuple_pool_)) {
          ReportColumnParseError(slot_desc, col_start, field_len);
          error_in_row = true;
        }
      }

      if (error_in_row) {
        error_in_row = false;
        if (state_->LogHasSpace()) {
          stringstream ss;
          ss << "file: " << current_byte_stream_->GetLocation();
          state_->LogError(ss.str());
        }
        if (state_->abort_on_error()) {
          state_->ReportFileErrors(current_byte_stream_->GetLocation(), 1);
          return Status(state_->ErrorLog());
        }
      }

      // Evaluate the conjuncts and add the row to the batch
      if (ExecNode::EvalConjuncts(conjuncts_, num_conjuncts_, current_row)) {
        row_batch->CommitLastRow();
        if (scan_node_->ReachedLimit() || row_batch->IsFull()) {
          tuple_ = NULL;
          return Status::OK;
        }
        char* new_tuple = reinterpret_cast<char*>(tuple_);
        new_tuple += tuple_byte_size_;
        tuple_ = reinterpret_cast<Tuple*>(new_tuple);
      }
    }
  }

  // Note that number of rows remaining == 0 -> scan_range_fully_buffered_
  // at this point, hence we don't need to explicitly check that
  // scan_range_fully_buffered_ == true.
  if (scan_node_->ReachedLimit() || num_rows_ == 0 || num_rows_ == row_pos_) {
    // We reached the limit, drained the row group or hit the end of the table.
    // No more work to be done. Clean up all pools with the last row batch.
    *eosr = true;
  } else {
    DCHECK(row_batch->IsFull());
    // The current row_batch is full, but we haven't yet reached our limit.
    // Hang on to the last chunks. We'll continue from there in the next
    // call to GetNext().
    *eosr = false;
  }
  // Maintian ownership of last memory chunk if not at the end of the scan range.
  if (has_noncompact_strings_) {
    row_batch->tuple_data_pool()->AcquireData(column_buffer_pool_.get(), !*eosr);
  }
  row_batch->tuple_data_pool()->AcquireData(tuple_pool_, !*eosr);

=======
Status HdfsRCFileScanner::ProcessRange() {
  ResetRowGroup();

  // HdfsRCFileScanner effectively does buffered IO, in that it reads all the
  // materialized columns into a row group buffer.
  // It will then materialize tuples from the row group buffer.  When the row
  // group is complete, it will move onto the next row group.
  while (!finished()) {
    DCHECK_EQ(num_rows_, row_pos_);
    // Finished materializing this row group, read the next one.
    RETURN_IF_ERROR(ReadRowGroup());
    if (num_rows_ == 0) break;

    while (num_rows_ != row_pos_) {
      SCOPED_TIMER(scan_node_->materialize_tuple_timer());

      // Indicates whether the current row has errors.
      bool error_in_row = false;
      const vector<SlotDescriptor*>& materialized_slots =
          scan_node_->materialized_slots();
      vector<SlotDescriptor*>::const_iterator it;

      // Materialize rows from this row group in row batch sizes
      MemPool* pool;
      Tuple* tuple;
      TupleRow* current_row;
      int max_tuples = GetMemory(&pool, &tuple, &current_row);
      max_tuples = min(max_tuples, num_rows_ - row_pos_);

      if (materialized_slots.empty()) {
        // If there are no materialized slots (e.g. count(*) or just partition cols)
        // we can shortcircuit the parse loop
        row_pos_ += max_tuples;
        int num_to_commit = WriteEmptyTuples(context_, current_row, max_tuples);
        COUNTER_ADD(scan_node_->rows_read_counter(), max_tuples);
        RETURN_IF_ERROR(CommitRows(num_to_commit));
        continue;
      }

      int num_to_commit = 0;
      for (int i = 0; i < max_tuples; ++i) {
        RETURN_IF_ERROR(NextRow());

        // Initialize tuple from the partition key template tuple before writing the
        // slots
        InitTuple(template_tuple_, tuple);

        for (it = materialized_slots.begin(); it != materialized_slots.end(); ++it) {
          const SlotDescriptor* slot_desc = *it;
          int file_column_idx = slot_desc->col_pos() - scan_node_->num_partition_keys();

          // Set columns missing in this file to NULL
          if (file_column_idx >= columns_.size()) {
            tuple->SetNull(slot_desc->null_indicator_offset());
            continue;
          }

          ColumnInfo& column = columns_[file_column_idx];
          DCHECK(column.materialize_column);

          const char* col_start = reinterpret_cast<const char*>(
              row_group_buffer_ + column.start_offset + column.buffer_pos);
          int field_len = column.current_field_len;
          DCHECK_LE(col_start + field_len,
              reinterpret_cast<const char*>(row_group_buffer_ + row_group_length_));

          if (!text_converter_->WriteSlot(slot_desc, tuple, col_start, field_len,
              false, false, pool)) {
            ReportColumnParseError(slot_desc, col_start, field_len);
            error_in_row = true;
          }
        }

        if (error_in_row) {
          error_in_row = false;
          if (state_->LogHasSpace()) {
            stringstream ss;
            ss << "file: " << stream_->filename();
            state_->LogError(ss.str());
          }
          if (state_->abort_on_error()) {
            state_->ReportFileErrors(stream_->filename(), 1);
            return Status(state_->ErrorLog());
          }
        }

        current_row->SetTuple(scan_node_->tuple_idx(), tuple);
        // Evaluate the conjuncts and add the row to the batch
        if (EvalConjuncts(current_row)) {
          ++num_to_commit;
          current_row = next_row(current_row);
          tuple = next_tuple(tuple);
        }
      }
      COUNTER_ADD(scan_node_->rows_read_counter(), max_tuples);
      RETURN_IF_ERROR(CommitRows(num_to_commit));
      if (scan_node_->ReachedLimit()) return Status::OK;
    }

    // RCFiles don't end with syncs
    if (stream_->eof()) return Status::OK;

    // Check for sync by looking for the marker that precedes syncs.
    int marker;
    RETURN_IF_FALSE(stream_->ReadInt(&marker, &parse_status_, /* peek */ true));
    if (marker == HdfsRCFileScanner::SYNC_MARKER) {
      RETURN_IF_FALSE(stream_->ReadInt(&marker, &parse_status_, /* peek */ false));
      RETURN_IF_ERROR(ReadSync());
    }
  }
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
  return Status::OK;
}

void HdfsRCFileScanner::DebugString(int indentation_level, stringstream* out) const {
  // TODO: Add more details of internal state.
<<<<<<< HEAD
  *out << string(indentation_level * 2, ' ');
  *out << "HdfsRCFileScanner(tupleid=" << scan_node_->tuple_idx() <<
    " file=" << current_byte_stream_->GetLocation();
=======
  *out << string(indentation_level * 2, ' ')
       << "HdfsRCFileScanner(tupleid=" << scan_node_->tuple_idx()
       << " file=" << stream_->filename();
>>>>>>> d520a9cdea2fc97e8d5da9fbb0244e60ee416bfa
  // TODO: Scanner::DebugString
  //  ExecNode::DebugString(indentation_level, out);
  *out << "])" << endl;
}

ext: cc

max_stars_repo_name: zhennan-sun/impala

max_issues_repo_name: zhennan-sun/impala

max_forks_repo_name: zhennan-sun/impala


text:
// Boost.Geometry (aka GGL, Generic Geometry Library)
// Unit Test

// Copyright (c) 2010-2012 Barend Gehrels, Amsterdam, the Netherlands.

// Use, modification and distribution is subject to 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)

#include <algorithm>
#include <iterator>
#include <sstream>
#include <string>

#include <geometry_test_common.hpp>

#include <boost/geometry/views/reversible_view.hpp>

#include <boost/geometry/io/wkt/wkt.hpp>
#include <boost/geometry/io/dsv/write.hpp>
#include <boost/geometry/geometries/geometries.hpp>
#include <boost/geometry/geometries/point_xy.hpp>
#include <boost/geometry/geometries/adapted/boost_tuple.hpp>

BOOST_GEOMETRY_REGISTER_BOOST_TUPLE_CS(cs::cartesian)

template <bg::iterate_direction Direction, typename Range>
void test_forward_or_reverse(Range const& range, std::string const& expected)
{
    typedef typename bg::reversible_view
        <
            Range const,
            Direction
        >::type view_type;

    view_type view(range);

    bool first = true;
    std::ostringstream out;
    for (typename boost::range_iterator<view_type const>::type
        it = boost::begin(view);
        it != boost::end(view);
        ++it, first = false)
    {
        out << (first ? "" : " ") << bg::dsv(*it);
    }
    BOOST_CHECK_EQUAL(out.str(), expected);
}

template <typename Geometry>
void test_geometry(std::string const& wkt,
            std::string const& expected_forward, std::string const& expected_reverse)
{
    Geometry geo;
    bg::read_wkt(wkt, geo);

    test_forward_or_reverse<bg::iterate_forward>(geo, expected_forward);
    test_forward_or_reverse<bg::iterate_reverse>(geo, expected_reverse);
}

template <typename P>
void test_all()
{
    test_geometry<bg::model::linestring<P> >(
            "linestring(1 1,2 2,3 3)",
            "(1, 1) (2, 2) (3, 3)",
            "(3, 3) (2, 2) (1, 1)");
}

int test_main(int, char* [])
{
    test_all<bg::model::d2::point_xy<double> >();
    test_all<bg::model::point<int, 2, bg::cs::cartesian> >();
    test_all<boost::tuple<double, double> >();

    return 0;
}

ext: cpp


text:
/*=============================================================================
    Spirit v1.6.0
    Copyright (c) 1998-2003 Joel de Guzman
    http://spirit.sourceforge.net/

    Permission to copy, use, modify, sell and distribute this software is
    granted provided this copyright notice appears in all copies. This
    software is provided "as is" without express or implied warranty, and
    with no claim as to its suitability for any purpose.
=============================================================================*/
#include <boost/spirit/core.hpp>
#include <boost/spirit/error_handling/exceptions.hpp>
#include <iostream>
#include <cassert>

///////////////////////////////////////////////////////////////////////////////
using namespace std;
using namespace boost::spirit;

///////////////////////////////////////////////////////////////////////////////
struct handler
{
    template <typename ScannerT, typename ErrorT>
    error_status<>
    operator()(ScannerT const& /*scan*/, ErrorT const& /*error*/) const
    {
        cout << "exception caught...Test concluded successfully" << endl;
        return error_status<>(error_status<>::fail);
    }
};

///////////////////////////////////////////////////////////////////////////////
int
main()
{
    cout << "/////////////////////////////////////////////////////////\n\n";
    cout << "\t\tExceptions Test...\n\n";
    cout << "/////////////////////////////////////////////////////////\n\n";

    assertion<int>  expect(0);
    guard<int>      my_guard;

    bool r =
        parse("abcx",
            my_guard(ch_p('a') >> 'b' >> 'c' >> expect( ch_p('d') ))
            [
                handler()
            ]
        ).full;

    assert(!r);
    return 0;
}


text:
// Boost.Geometry (aka GGL, Generic Geometry Library)
// QuickBook Example

// Copyright (c) 2011 Barend Gehrels, Amsterdam, the Netherlands.

// Use, modification and distribution is subject to 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)

//[interior_type
//`Shows how to use the interior_type metafunction

#include <iostream>
#include <typeinfo>

#include <boost/geometry.hpp>
#include <boost/geometry/geometries/polygon.hpp>
#include <boost/geometry/geometries/ring.hpp>
#include <boost/geometry/geometries/adapted/boost_array.hpp>

BOOST_GEOMETRY_REGISTER_BOOST_ARRAY_CS(cs::cartesian)

int main()
{
    // Define a polygon storing points in a deque and storing interior rings
    // in a list (note that std::list is not supported by most algorithms
    // because not supporting a random access iterator)
    typedef boost::geometry::model::polygon
        <
            boost::array<short, 3>,
            true, true,
            std::deque, std::list
        > polygon;

    std::cout << typeid(boost::geometry::interior_type<polygon>::type).name() << std::endl;

    return 0;
}

//]

//[interior_type_output
/*`
Output (using MSVC) is a long story (part manually replaced with ellipsis):
[pre
class std::list<class boost::geometry::model::ring<class boost::array<short,3>,1,1,class std::deque,class std::allocator>,class std::allocator<...> > >
]
*/
//]


text:
/**********************************************************************
*  Copyright (c) 2008-2014, Alliance for Sustainable Energy.
*  All rights reserved.
*
*  This library is free software; you can redistribute it and/or
*  modify it under the terms of the GNU Lesser General Public
*  License as published by the Free Software Foundation; either
*  version 2.1 of the License, or (at your option) any later version.
*
*  This library 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
*  Lesser General Public License for more details.
*
*  You should have received a copy of the GNU Lesser General Public
*  License along with this library; if not, write to the Free Software
*  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
**********************************************************************/

#include <utilities/core/Checksum.hpp>

#include <ios>
#include <sstream>

#include <boost/crc.hpp>
#include <boost/filesystem/fstream.hpp>

namespace openstudio {

  /// return 8 character hex checksum of string
  std::string checksum(const std::string& s)
  {
    std::stringstream ss(s);
    return checksum(ss);
  }

  /// return 8 character hex checksum of istream
  std::string checksum(std::istream& is)
  {
    boost::crc_32_type  crc;
    do{
      const std::streamsize n = 1024;
      char  buffer[n];
      is.read(buffer, n);
      crc.process_bytes(buffer, static_cast<size_t>(is.gcount()));
    } while ( is );

    std::stringstream ss;
    ss << std::hex << std::uppercase << crc.checksum();
    std::string result = "00000000";
    std::string checksum = ss.str();
    result.replace(8-checksum.size(), checksum.size(), checksum);

    return result;
  }

  /// return 8 character hex checksum of file contents
  std::string checksum(const path& p)
  {
    std::string result = "00000000";
    try{
      boost::filesystem::ifstream  ifs(p, std::ios_base::binary );
      if ( ifs ){
        result = checksum(ifs);
      }
    }catch(...){
    }
    return result;
  }

} // openstudio

max_stars_repo_name: ORNL-BTRIC/OpenStudio

max_issues_repo_name: ORNL-BTRIC/OpenStudio

max_forks_repo_name: ORNL-BTRIC/OpenStudio


text:
/*
 * Copyright 2012 Open Source Robotics Foundation
 *
 * 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.
 *
*/
/* Desc: The ODE base joint class
 * Author: Nate Koenig, Andrew Howard
 * Date: 12 Oct 2009
 */

#ifndef _ODEJOINT_HH_
#define _ODEJOINT_HH_

#include <boost/any.hpp>
#include <string>

#include "gazebo/physics/ode/ODEPhysics.hh"
#include "gazebo/physics/Joint.hh"

namespace gazebo
{
  namespace physics
  {
    /// \brief ODE joint interface
    class ODEJoint : public Joint
    {
      /// \brief internal variables used for cfm damping
      public:  enum CFMMode
      {
        /// \brief cfm damping not active
        NONE           = 0x00000000,
        /// \brief cfm damping active, joints within limits
        DAMPING_ACTIVE = 0x00000001,
        /// \brief cfm damping not active, enforcing joints limits
        JOINT_LIMIT    = 0x00000002
      };

      /// \brief Constructor.
      /// \param[in] _parent Parent of the Joint.
      public: ODEJoint(BasePtr _parent);

      /// \brief Destructor.
      public: virtual ~ODEJoint();

      // Documentation inherited.
      public: virtual void Load(sdf::ElementPtr _sdf);

      // Documentation inherited.
      public: virtual void Reset();

      // Documentation inherited.
      public: virtual LinkPtr GetJointLink(int _index) const;

      // Documentation inherited.
      public: virtual bool AreConnected(LinkPtr _one, LinkPtr _two) const;

      /// \brief Get an ODE joint parameter.
      ///
      /// The default function does nothing. This should be
      /// overriden in the child classes where appropriate.
      /// \param[in] _parameter ID of the parameter to get.
      /// \return Value of the parameter.
      public: virtual double GetParam(int _parameter) const;

      /// \brief Set an ODE joint paramter.
      ///
      /// By default this does nothing. It should be overridden in child
      /// classes where appropriate
      /// \param[in] _parameter ID of the parameter to set.
      /// \param[in] _value Value to set.
      public: virtual void SetParam(int _parameter, double _value);

      // Documentation inherited
      public: virtual void SetDamping(int _index, double _damping);

      // Documentation inherited.
      public: virtual void Attach(LinkPtr _parent, LinkPtr _child);

      // Documentation inherited.
      public: virtual void Detach();

      /// \brief Set the ERP of this joint.
      /// \param[in] _erp Error Reduction Parameter value.
      public: void SetERP(double _erp);

      /// \brief Get the ERP of this joint.
      /// \return The Error Reduction Parameter of this joint
      public: double GetERP();

      /// \brief Set the CFM of this joint.
      /// \param[in] _cfm The Constraint Force Mixing value
      public: void SetCFM(double _cfm);

      /// \brief Get the CFM of this joint
      /// \return The Constraint Force Mixing value
      public: double GetCFM();

      /// \brief Get the feedback data structure for this joint, if set
      /// \return Pointer to the joint feedback.
      public: dJointFeedback *GetFeedback();

      /// \brief simulating damping with CFM and meddling with Joint limits
      public: void CFMDamping();

      /// \brief Get access to stopCFM
      /// \return Returns joint's cfm for end stops
      public: double GetStopCFM()
      {
        return this->stopCFM;
      }

      /// \brief Get access to stopERP
      /// \return Returns joint's erp for end stops
      public: double GetStopERP()
      {
        return this->stopERP;
      }

      /// \brief internal variable to keep track of cfm damping internals
      private: int cfmDampingState[3];

      /// \brief internal variable to keep track if ConnectJointUpdate
      /// has been called on a damping method
      private: bool dampingInitialized;

      // Documentation inherited.
      public: virtual void SetHighStop(int _index, const math::Angle &_angle);

      // Documentation inherited.
      public: virtual void SetLowStop(int _index, const math::Angle &_angle);

      // Documentation inherited.
      public: virtual math::Angle GetHighStop(int _index);

      // Documentation inherited.
      public: virtual math::Angle GetLowStop(int _index);

      // Documentation inherited.
      public: virtual math::Vector3 GetLinkForce(unsigned int _index) const;

      // Documentation inherited.
      public: virtual math::Vector3 GetLinkTorque(unsigned int _index) const;

      // Documentation inherited.
      public: virtual void SetAttribute(Attribute _attr, int _index,
                                        double _value);

      // Documentation inherited.
      public: virtual void SetAttribute(const std::string &_key, int _index,
                                        const boost::any &_value);

      // Documentation inherited.
      public: virtual double GetAttribute(const std::string &_key,
                                                unsigned int _index);

      /// \brief This is our ODE ID
      protected: dJointID jointId;

      /// \brief Feedback data for this joint
      private: dJointFeedback *feedback;

      /// \brief Provide Feedback data for contact forces
      private: bool provideFeedback;

      // Documentation inherited.
      public: virtual JointWrench GetForceTorque(int _index);

      // Documentation inherited.
      public: virtual JointWrench GetForceTorque(unsigned int _index);

      /// \brief CFM for joint's limit constraint
      private: double stopCFM;

      /// \brief ERP for joint's limit constraint
      private: double stopERP;
    };
  }
}
#endif

ext: hh

max_stars_repo_name: thomas-moulard/gazebo-deb

max_issues_repo_name: thomas-moulard/gazebo-deb

max_forks_repo_name: thomas-moulard/gazebo-deb


text:
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
/*                                                                           */
/*  This file is part of the library KASKADE 7                               */
/*    see http://www.zib.de/projects/kaskade7-finite-element-toolbox         */
/*                                                                           */
/*  Copyright (C) 2016-2016 Zuse Institute Berlin                            */
/*                                                                           */
/*  KASKADE 7 is distributed under the terms of the ZIB Academic License.    */
/*    see $KASKADE/academic.txt                                              */
/*                                                                           */
/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#include <cassert>
#include <iomanip>

#include <boost/timer/timer.hpp>

#include "utilities/timing.hh"

namespace Kaskade
{

  struct Timings::Times
  {
    std::string             name;
    size_t                  calls;
    std::vector<Times>      children;
    boost::timer::cpu_timer timer;

    int maxNameLength(int indent) const
    {
      int len = name.size();
      for (auto const& c: children)
        len = std::max(len,indent+c.maxNameLength(2));
      return len;
    }

    void report(std::ostream& out, int indent, int pos, boost::timer::nanosecond_type full) const
    {
      out << std::setw(indent) << ""
          << std::setw(3) << std::right << 100*timer.elapsed().wall/full << "% "
          << std::setw(pos-indent) << std::left << name
          << std::setw(6) << std::right << calls << "  ";
      out << timer.format();

      full = timer.elapsed().wall;
      for (auto const& c: children)
        c.report(out,indent+2,pos,full);
    }
  };

  Timings& Timings::instance()
  {
    static Timings globalTimings;
    return globalTimings;
  }

  Timings::Timings()
  : all(new Times{"all",0,{},{}})
  {
    stack.push(all.get());
    all->timer.start();
  }

  std::ostream& Timings::report(std::ostream& out) const
  {
    int pos = all->maxNameLength(2)+4;
    out << std::setw(pos+4) << " " << "  calls   " << "\n";
    //~ out << std::setw(pos+30) << " " << "\n";
    all->report(out,0,pos,all->timer.elapsed().wall);
    return out;
  }

  void Timings::start(std::string const& name)
  {
    Times* section = nullptr;
    for (auto& c: stack.top()->children)
      if (name == c.name)
      {
        section = &c;
        section->timer.resume();
        break;
      }

    if (! section)
    {
      // section not found - create a new one. Note that this can invalidate pointers into
      // the children container. Fortunately, there are no pointers/iterators around: The
      // stack contains only pointers into higher level containers.
      stack.top()->children.push_back(Times{name,0,{},{}});
      section = &(stack.top()->children.back());
      section->timer.start();
    }

    ++section->calls;
    stack.push(section);
  }

  void Timings::stop(std::string const& name)
  {
    assert(stack.top()->name == name);
    stack.top()->timer.stop();
    stack.pop();
  }

  void Timings::clear()
  {
    while (stack.size()>1)  // remove everything (except the root)
      stack.pop();
    all->timer.start();      // reset timer to zero
    all->calls = 0;
    all->children.clear();
  }

  std::ostream& operator<<(std::ostream& out, Timings const& timings)
  {
    return timings.report(out);
  }

}

max_stars_repo_name: chenzongxiong/streambox

max_issues_repo_name: chenzongxiong/streambox

max_forks_repo_name: chenzongxiong/streambox


text:
#include <string> //std::string

#include <boost/log/trivial.hpp>

#include <graphtyper/graph/absolute_position.hpp>
#include <graphtyper/graph/genomic_region.hpp>
#include <graphtyper/graph/graph.hpp>
#include <graphtyper/graph/label.hpp>
#include <graphtyper/utilities/type_conversions.hpp>

#include <catch.hpp>

/*
TEST_CASE("Getting absolute and contig positions from genomic region")
{
  BOOST_LOG_TRIVIAL(debug) << "[" << __HERE__
                           << "] TEST CASE: Getting absolute and contig positions from genomic region.";

  gyper::GenomicRegion genomic_region;

  uint32_t const CHR01_LENGTH = 66ul;
  uint32_t const CHR02_LENGTH = 66ul;
  uint32_t const CHR03_LENGTH = 66ul;

  // Get absolute position
  {
    REQUIRE(genomic_region.get_absolute_position("chr1", 1) == 1);
    REQUIRE(genomic_region.get_absolute_position("chr1", 100) == 100);
    REQUIRE(genomic_region.get_absolute_position("chr2", 100) == 100 + CHR01_LENGTH);
    REQUIRE(genomic_region.get_absolute_position("chr4", 1) == 1 + CHR01_LENGTH + CHR02_LENGTH + CHR03_LENGTH);
  }

  // Get contig position
  {
    REQUIRE(genomic_region.get_contig_position(1, gyper::graph).first == "chr1");
    REQUIRE(genomic_region.get_contig_position(1, gyper::graph).second == 1ul);

    REQUIRE(genomic_region.get_contig_position(3, gyper::graph).first == "chr1");
    REQUIRE(genomic_region.get_contig_position(3, gyper::graph).second == 3ul);

    REQUIRE(genomic_region.get_contig_position(1 + CHR01_LENGTH, gyper::graph).first == "chr2");
    REQUIRE(genomic_region.get_contig_position(1 + CHR01_LENGTH, gyper::graph).second == 1ul);

    REQUIRE(genomic_region.get_contig_position(CHR01_LENGTH, gyper::graph).first == "chr1");
    REQUIRE(genomic_region.get_contig_position(CHR01_LENGTH, gyper::graph).second == CHR01_LENGTH);
  }
}
*/


text:
//==============================================================================
//         Copyright 2009 - 2013 LRI    UMR 8623 CNRS/Univ Paris Sud XI
//         Copyright 2012 - 2013 MetaScale SAS
//
//          Distributed under the Boost Software License, Version 1.0.
//                 See accompanying file LICENSE.txt or copy at
//                     http://www.boost.org/LICENSE_1_0.txt
//==============================================================================
#ifndef NT2_SDK_BENCH_DETAILS_DISPLAY_METRIC_HPP_INCLUDED
#define NT2_SDK_BENCH_DETAILS_DISPLAY_METRIC_HPP_INCLUDED

#include <nt2/sdk/bench/details/measure.hpp>
#include <nt2/sdk/bench/config.hpp>
#include <nt2/sdk/meta/type_id.hpp>
#include <boost/accumulators/statistics/count.hpp>
#include <iostream>
#include <sstream>
#include <string>

namespace nt2 { namespace details
{
  extern NT2_TEST_BENCHMARK_DECL std::string current_benchmark;

  /*
    display_metric is a callable object usable with fusion::for_each
    to compute metrics and to display their results
  */
  template<typename Experiment>
  struct display_metric
  {
    display_metric( Experiment const& xp
                  , details::times_set  const& t
                  , details::cycles_set const& c
                  )
          : experiment_(xp), times_(t), cycles_(c)
    {}

    template<typename T> inline void operator()(T& t) const
    {
      std::cout << current_benchmark<< '\t' << experiment_ << '\t'
                << t(experiment_,nt2::type_id(experiment_),times_,cycles_)
                << '\t' << t.unit()
                << '\t' << boost::accumulators::count(times_)
                << std::endl;
    }

    Experiment          const&  experiment_;
    details::times_set  const&  times_;
    details::cycles_set const&  cycles_;

    private:
    display_metric& operator=(display_metric const&);
  };
} }

#endif

ext: hpp


text:
/*
 * wave_equation.cpp
 *
 *  Created on: 11.07.2017
 *      Author: thies
 */

#include <base/AdaptiveMesh.h>
#include <base/ConstantMesh.h>
#include <base/DiscretizedFunction.h>
#include <base/SpaceTimeMesh.h>
#include <base/Util.h>
#include <deal.II/base/exceptions.h>
#include <deal.II/base/function.h>
#include <deal.II/base/logstream.h>
#include <deal.II/base/numbers.h>
#include <deal.II/base/point.h>
#include <deal.II/base/quadrature.h>
#include <deal.II/base/quadrature_lib.h>
#include <deal.II/base/timer.h>
#include <deal.II/fe/fe_q.h>
#include <deal.II/grid/grid_generator.h>
#include <deal.II/grid/tria.h>
#include <forward/L2RightHandSide.h>
#include <forward/WaveEquation.h>
#include <gtest/gtest.h>
#include <norms/L2L2.h>
#include <stddef.h>
#include <iostream>
#include <memory>
#include <vector>

namespace {

using namespace dealii;
using namespace wavepi::forward;
using namespace wavepi::base;
using namespace wavepi;

template<int dim>
class TestF: public LightFunction<dim> {
public:
   virtual ~TestF() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      if ((t <= 1) && (p.norm() < 0.5))
         return std::sin(t * 2 * numbers::PI);
      else
         return 0.0;
   }
};

template<int dim>
class TestG: public LightFunction<dim> {
public:
   virtual ~TestG() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      Point<dim> pc = Point<dim>::unit_vector(0);
      pc *= 0.5;

      if (std::abs(this->get_time() - 1.0) < 0.5 && (p.distance(pc) < 0.5))
         return std::sin(this->get_time() * 1 * numbers::PI);
      else
         return 0.0;
   }
};

template<int dim>
class TestH: public LightFunction<dim> {
public:
   virtual ~TestH() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      return p.norm() * this->get_time();
   }
};

template<int dim>
double rho(const Point<dim> &p, double t) {
   return p.norm() + t + 1.0;
}

template<int dim>
double c_squared(const Point<dim> &p, double t) {
   double tmp = p.norm() * t + 1.0;

   return tmp * tmp;
}

template<int dim>
class TestC: public LightFunction<dim> {
public:
   virtual ~TestC() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      return 1.0 / (rho(p, t) * c_squared(p, t));
   }
};

template<int dim>
class TestRho: public LightFunction<dim> {
public:
   virtual ~TestRho() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      return rho(p, t);
   }
};

template<int dim>
class TestNu: public LightFunction<dim> {
public:
   virtual ~TestNu() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      // do not just change this, reference_test_nu needs this
      return p.norm() * t;
   }
};

template<int dim>
class RhsTestNu: public LightFunction<dim> {
public:
   virtual ~RhsTestNu() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      // do not just change this, reference_test_nu needs this
      return nu.evaluate(p, t) * v->evaluate(p, t);
   }

   RhsTestNu(const std::shared_ptr<LightFunction<dim>> &v)
         : v(v) {
      AssertThrow(v, ExcNotInitialized());
   }

private:
   std::shared_ptr<LightFunction<dim>> v;  // pointer to u'
   TestNu<dim> nu;
};

template<int dim>
class TestQ: public LightFunction<dim> {
public:
   virtual ~TestQ() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const override {
      return p.norm() < 0.5 ? std::sin(t / 2 * 2 * numbers::PI) : 0.0;
   }

   static const Point<dim> q_position;
};

template<>
const Point<1> TestQ<1>::q_position = Point<1>(-1.0);
template<>
const Point<2> TestQ<2>::q_position = Point<2>(-1.0, 0.5);
template<>
const Point<3> TestQ<3>::q_position = Point<3>(-1.0, 0.5, 0.0);

template<int dim>
class DiscretizedFunctionDisguise: public LightFunction<dim> {
public:
   DiscretizedFunctionDisguise(std::shared_ptr<DiscretizedFunction<dim>> base)
         : base(base) {
   }
   virtual ~DiscretizedFunctionDisguise() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      return base->value(p, t);
   }
private:
   std::shared_ptr<DiscretizedFunction<dim>> base;
};

// checks, whether the matrix assembly of discretized parameters works correct
// (by supplying DiscretizedFunctions and DiscretizedFunctionDisguises)
template<int dim>
void run_discretized_test(int fe_order, int quad_order, int refines) {
   Timer timer;

   auto triangulation = std::make_shared<Triangulation<dim>>();
   GridGenerator::hyper_cube(*triangulation, -1, 1);
   Util::set_all_boundary_ids(*triangulation, 0);
   triangulation->refine_global(refines);

   double t_start = 0.0, t_end = 2.0, dt = t_end / 64.0;
   std::vector<double> times;

   for (size_t i = 0; t_start + i * dt <= t_end; i++)
      times.push_back(t_start + i * dt);

   FE_Q<dim> fe(fe_order);
   Quadrature<dim> quad = QGauss<dim>(quad_order);  // exact in poly degree 2n-1 (needed: fe_dim^3)

   std::shared_ptr<SpaceTimeMesh<dim>> mesh = std::make_shared<ConstantMesh<dim>>(times, fe, quad, triangulation);

   deallog << std::endl << "----------  n_dofs / timestep: " << mesh->get_dof_handler(0)->n_dofs();
   deallog << ", n_steps: " << times.size() << "  ----------" << std::endl;

   WaveEquation<dim> wave_eq(mesh);

   /* continuous */

   wave_eq.set_param_rho(std::make_shared<TestRho<dim>>());
   wave_eq.set_param_c(std::make_shared<TestC<dim>>());
   wave_eq.set_param_q(std::make_shared<TestQ<dim>>());
   wave_eq.set_param_nu(std::make_shared<TestNu<dim>>());

   timer.restart();
   DiscretizedFunction<dim> sol_cont = wave_eq.run(
         std::make_shared<L2RightHandSide<dim>>(std::make_shared<TestF<dim>>()), WaveEquation<dim>::Forward);
   sol_cont.set_norm(std::make_shared<norms::L2L2<dim>>());

   timer.stop();
   deallog << "continuous params: " << std::fixed << timer.wall_time() << " s of wall time" << std::endl;
   EXPECT_GT(sol_cont.norm(), 0.0);

   /* discretized */

   TestC<dim> c;
   auto c_disc = std::make_shared<DiscretizedFunction<dim>>(mesh, c);
   wave_eq.set_param_c(c_disc);

   TestRho<dim> rho;
   auto rho_disc = std::make_shared<DiscretizedFunction<dim>>(mesh, rho);
   wave_eq.set_param_rho(rho_disc);

   TestQ<dim> q;
   auto q_disc = std::make_shared<DiscretizedFunction<dim>>(mesh, q);
   wave_eq.set_param_q(q_disc);

   TestNu<dim> nu;
   auto nu_disc = std::make_shared<DiscretizedFunction<dim>>(mesh, nu);
   wave_eq.set_param_nu(nu_disc);

   TestF<dim> f;
   auto f_disc = std::make_shared<DiscretizedFunction<dim>>(mesh, f);

   timer.restart();
   DiscretizedFunction<dim> sol_disc = wave_eq.run(std::make_shared<L2RightHandSide<dim>>(f_disc),
         WaveEquation<dim>::Forward);
   sol_disc.set_norm(std::make_shared<norms::L2L2<dim>>());
   timer.stop();
   deallog << "all discretized: " << std::fixed << timer.wall_time() << " s of wall time" << std::endl;
   EXPECT_GT(sol_disc.norm(), 0.0);

   /* discretized, q disguised */

   auto c_disguised = std::make_shared<DiscretizedFunctionDisguise<dim>>(c_disc);
   auto rho_disguised = std::make_shared<DiscretizedFunctionDisguise<dim>>(rho_disc);
   auto q_disguised = std::make_shared<DiscretizedFunctionDisguise<dim>>(q_disc);
   auto nu_disguised = std::make_shared<DiscretizedFunctionDisguise<dim>>(nu_disc);
   auto f_disguised = std::make_shared<DiscretizedFunctionDisguise<dim>>(f_disc);

   wave_eq.set_param_q(q_disguised);

   timer.restart();
   DiscretizedFunction<dim> sol_disc_except_q = wave_eq.run(std::make_shared<L2RightHandSide<dim>>(f_disc),
         WaveEquation<dim>::Forward);
   sol_disc_except_q.set_norm(std::make_shared<norms::L2L2<dim>>());
   timer.stop();
   deallog << "all discretized, q disguised: " << std::fixed << timer.wall_time() << " s of wall time" << std::endl;
   EXPECT_GT(sol_disc_except_q.norm(), 0.0);

   /* discretized, a disguised */

   wave_eq.set_param_rho(rho_disguised);
   wave_eq.set_param_q(q_disc);

   timer.restart();
   DiscretizedFunction<dim> sol_disc_except_a = wave_eq.run(std::make_shared<L2RightHandSide<dim>>(f_disc),
         WaveEquation<dim>::Forward);
   sol_disc_except_a.set_norm(std::make_shared<norms::L2L2<dim>>());
   timer.stop();
   deallog << "all discretized, a disguised: " << std::fixed << timer.wall_time() << " s of wall time" << std::endl;
   EXPECT_GT(sol_disc_except_a.norm(), 0.0);

   /* disguised */

   wave_eq.set_param_nu(nu_disguised);
   wave_eq.set_param_q(q_disguised);
   wave_eq.set_param_rho(rho_disguised);
   wave_eq.set_param_c(c_disguised);

   timer.restart();
   DiscretizedFunction<dim> sol_disguised = wave_eq.run(std::make_shared<L2RightHandSide<dim>>(f_disguised),
         WaveEquation<dim>::Forward);
   sol_disguised.set_norm(std::make_shared<norms::L2L2<dim>>());
   timer.stop();
   deallog << "all discretized and disguised as continuous: " << std::fixed << timer.wall_time() << " s of wall time"
         << std::endl << std::endl;
   EXPECT_GT(sol_disguised.norm(), 0.0);

   /* results */
   DiscretizedFunction<dim> tmp(sol_disc_except_q);
   tmp -= sol_disguised;
   double err_disguised_vs_disc_except_q = tmp.norm() / sol_disguised.norm();

   deallog << "rel. error between disguised discrete and discrete (q disguised): " << std::scientific
         << err_disguised_vs_disc_except_q << std::endl;
   EXPECT_LT(err_disguised_vs_disc_except_q, 1e-6);

   tmp = sol_disc_except_a;
   tmp -= sol_disguised;
   double err_disguised_vs_disc_except_a = tmp.norm() / sol_disguised.norm();

   deallog << "rel. error between disguised discrete and discrete (a disguised): " << std::scientific
         << err_disguised_vs_disc_except_a << std::endl;
   EXPECT_LT(err_disguised_vs_disc_except_a, 1e-6);

   tmp = sol_disc;
   tmp -= sol_disguised;
   double err_disguised_vs_disc = tmp.norm() / sol_disguised.norm();

   deallog << "rel. error between disguised discrete and full discrete: " << std::scientific << err_disguised_vs_disc
         << std::endl << std::endl;
   EXPECT_LT(err_disguised_vs_disc, 1e-6);
}

// product of sines in space to have dirichlet b.c. in [0,pi], times a sum of sine and cosine in time.
// its time derivative is the same function with C[1] = C[0]*norm(k), C[0] = -C[1]*norm(k)
template<int dim>
class SeparationAnsatz: public LightFunction<dim> {
public:
   virtual ~SeparationAnsatz() = default;

   virtual double evaluate(const Point<dim> &p, const double t) const {
      double res = 1;

      for (size_t i = 0; i < dim; i++)
         res *= std::sin(k[i] * p[i]);

      res *= constants[0] * std::sin(std::sqrt(k.square()) * t) + constants[1] * std::cos(std::sqrt(k.square()) * t);
      return res;
   }

   SeparationAnsatz(Point<dim, int> k, Point<2> constants)
         : k(k), constants(constants) {
   }

private:
   Point<dim, int> k;
   Point<2> constants;
};

template<int dim>
void run_reference_test(std::shared_ptr<SpaceTimeMesh<dim>> mesh, Point<dim, int> k, Point<2> constants, bool expect =
      true, bool save = false) {
   deallog << std::endl << "----------  n_dofs(0): " << mesh->get_dof_handler(0)->n_dofs();
   deallog << ", n_steps: " << mesh->get_times().size() << "  ----------" << std::endl;

   WaveEquation<dim> wave_eq(mesh);

   Point<2> derivative_constants;
   derivative_constants[0] = -constants[1] * std::sqrt(k.square());
   derivative_constants[1] = constants[0] * std::sqrt(k.square());

   auto u = std::make_shared<SeparationAnsatz<dim>>(k, constants);
   auto v = std::make_shared<SeparationAnsatz<dim>>(k, derivative_constants);

   wave_eq.set_initial_values_u(u);
   wave_eq.set_initial_values_v(v);

   DiscretizedFunction<dim> solu = wave_eq.run(
         std::make_shared<L2RightHandSide<dim>>(std::make_shared<Functions::ZeroFunction<dim>>(1)),
         WaveEquation<dim>::Forward);
   DiscretizedFunction<dim> solv = solu.derivative();
   solu.throw_away_derivative();

   solu.set_norm(std::make_shared<norms::L2L2<dim>>());
   solv.set_norm(std::make_shared<norms::L2L2<dim>>());

   DiscretizedFunction<dim> refu(mesh, *u);
   DiscretizedFunction<dim> refv(mesh, *v);

   refu.set_norm(std::make_shared<norms::L2L2<dim>>());
   refv.set_norm(std::make_shared<norms::L2L2<dim>>());

   DiscretizedFunction<dim> tmp(solu);
   tmp -= refu;
   double err_u = tmp.norm() / refu.norm();

   tmp = solv;
   tmp -= refv;
   double err_v = tmp.norm() / refv.norm();

   if (expect) {
      EXPECT_LT(err_u, 1e-1);
      EXPECT_LT(err_v, 1e-1);
   }

   if (save) {
      solu.write_pvd("./", "solu", "u");
      refu.write_pvd("./", "refu", "uref");

      DiscretizedFunction<dim> tmp(solu);
      tmp -= refu;
      tmp.write_pvd("./", "diff", "udiff");
   }

   deallog << std::scientific << "forward : rerr(u) = " << err_u << ", rerr(v) = " << err_v << std::endl;

   solu = wave_eq.run(std::make_shared<L2RightHandSide<dim>>(std::make_shared<Functions::ZeroFunction<dim>>(1)),
         WaveEquation<dim>::Backward);
   solv = solu.derivative();
   solu.throw_away_derivative();

   solu.set_norm(std::make_shared<norms::L2L2<dim>>());
   solv.set_norm(std::make_shared<norms::L2L2<dim>>());

   tmp = solu;
   tmp -= refu;
   err_u = tmp.norm() / refu.norm();

   tmp = solv;
   tmp -= refv;
   err_v = tmp.norm() / refv.norm();

   if (expect) {
      EXPECT_LT(err_u, 1e-1);
      EXPECT_LT(err_v, 1e-1);
   }

   deallog << std::scientific << "backward: rerr(u) = " << err_u << ", rerr(v) = " << err_v << std::endl << std::endl;
}
}  // namespace

template<int dim>
void run_reference_test_constant(int fe_order, int quad_order, int refines, Point<dim, int> k, Point<2> constants,
      double t_end, int steps, bool expect = true, bool save = false) {
   auto triangulation = std::make_shared<Triangulation<dim>>();
   GridGenerator::hyper_cube(*triangulation, 0.0, numbers::PI);
   Util::set_all_boundary_ids(*triangulation, 0);
   triangulation->refine_global(refines);

   double t_start = 0.0, dt = t_end / steps;
   std::vector<double> times;

   for (size_t i = 0; t_start + i * dt <= t_end; i++)
      times.push_back(t_start + i * dt);

   FE_Q<dim> fe(fe_order);
   Quadrature<dim> quad = QGauss<dim>(quad_order);  // exact in poly degree 2n-1 (needed: fe_dim^3)

   std::shared_ptr<SpaceTimeMesh<dim>> mesh = std::make_shared<ConstantMesh<dim>>(times, fe, quad, triangulation);

   run_reference_test<dim>(mesh, k, constants, expect, save);
}

template<int dim>
void run_reference_test_nu(int fe_order, int quad_order, int refines, Point<dim, int> k, Point<2> constants,
      double t_end, int steps, bool expect = true, bool save = false) {
   auto triangulation = std::make_shared<Triangulation<dim>>();
   GridGenerator::hyper_cube(*triangulation, 0.0, numbers::PI);
   Util::set_all_boundary_ids(*triangulation, 0);
   triangulation->refine_global(refines);

   double t_start = 0.0, dt = t_end / steps;
   std::vector<double> times;

   for (size_t i = 0; t_start + i * dt <= t_end; i++)
      times.push_back(t_start + i * dt);

   FE_Q<dim> fe(fe_order);
   Quadrature<dim> quad = QGauss<dim>(quad_order);  // exact in poly degree 2n-1 (needed: fe_dim^3)

   std::shared_ptr<SpaceTimeMesh<dim>> mesh = std::make_shared<ConstantMesh<dim>>(times, fe, quad, triangulation);
   deallog << std::endl << "----------  n_dofs(0): " << mesh->get_dof_handler(0)->n_dofs();
   deallog << ", n_steps: " << mesh->get_times().size() << "  ----------" << std::endl;

   Point<2> derivative_constants;
   derivative_constants[0] = -constants[1] * std::sqrt(k.square());
   derivative_constants[1] = constants[0] * std::sqrt(k.square());

   auto u = std::make_shared<SeparationAnsatz<dim>>(k, constants);
   auto v = std::make_shared<SeparationAnsatz<dim>>(k, derivative_constants);

   WaveEquation<dim> wave_eq(mesh);
   wave_eq.set_param_nu(std::make_shared<TestNu<dim>>());

   wave_eq.set_initial_values_u(u);
   wave_eq.set_initial_values_v(v);

   auto x = std::make_shared<RhsTestNu<dim>>(v);

   DiscretizedFunction<dim> solu = wave_eq.run(std::make_shared<L2RightHandSide<dim>>(x), WaveEquation<dim>::Forward);
   DiscretizedFunction<dim> solv = solu.derivative();
   solu.throw_away_derivative();

   solu.set_norm(std::make_shared<norms::L2L2<dim>>());
   solv.set_norm(std::make_shared<norms::L2L2<dim>>());

   DiscretizedFunction<dim> refu(mesh, *u);
   DiscretizedFunction<dim> refv(mesh, *v);

   refu.set_norm(std::make_shared<norms::L2L2<dim>>());
   refv.set_norm(std::make_shared<norms::L2L2<dim>>());

   DiscretizedFunction<dim> tmp(solu);
   tmp -= refu;
   double err_u = tmp.norm() / refu.norm();

   tmp = solv;
   tmp -= refv;
   double err_v = tmp.norm() / refv.norm();

   if (expect) {
      EXPECT_LT(err_u, 1e-1);
      EXPECT_LT(err_v, 1e-1);
   }

   if (save) {
      solu.write_pvd("./", "solu", "u");
      refu.write_pvd("./", "refu", "uref");

      DiscretizedFunction<dim> tmp(solu);
      tmp -= refu;
      tmp.write_pvd("./", "diff", "udiff");
   }

   deallog << std::scientific << "forward : rerr(u) = " << err_u << ", rerr(v) = " << err_v << std::endl;

   // backward does not work for nu > 0
   /*
    wave_eq.set_run_direction(WaveEquation<dim>::Backward);
    solu = wave_eq.run();
    solv = solu.derivative();
    solu.throw_away_derivative();

    solu.set_norm(Norm::L2L2);
    solv.set_norm(Norm::L2L2);

    tmp = solu;
    tmp -= refu;
    err_u = tmp.norm() / refu.norm();

    tmp = solv;
    tmp -= refv;
    err_v = tmp.norm() / refv.norm();

    if (expect) {
    EXPECT_LT(err_u, 1e-1);
    EXPECT_LT(err_v, 1e-1);
    }

    deallog << std::scientific << "backward: rerr(u) = " << err_u << ", rerr(v) = " << err_v << std::endl << std::endl;
    */
}

template<int dim>
void run_reference_test_adaptive(int fe_order, int quad_order, int refines, Point<dim, int> k, Point<2> constants,
      double t_end, int steps, bool expect = true, bool save = false) {
   auto triangulation = std::make_shared<Triangulation<dim>>();
   GridGenerator::hyper_cube(*triangulation, 0.0, numbers::PI);
   Util::set_all_boundary_ids(*triangulation, 0);
   triangulation->refine_global(refines);

   double t_start = 0.0, dt = t_end / steps;
   std::vector<double> times;

   for (size_t i = 0; t_start + i * dt <= t_end; i++)
      times.push_back(t_start + i * dt);

   FE_Q<dim> fe(fe_order);
   Quadrature<dim> quad = QGauss<dim>(quad_order);  // exact in poly degree 2n-1 (needed: fe_dim^3)

   auto mesh = std::make_shared<AdaptiveMesh<dim>>(times, fe, quad, triangulation);

   // flag some cells for refinement, and refine them in some step
   for (auto cell : triangulation->active_cell_iterators())
      if (cell->center()[0] > numbers::PI / 2) cell->set_refine_flag();

   std::vector<bool> ref;
   std::vector<bool> coa;

   triangulation->save_refine_flags(ref);
   triangulation->save_coarsen_flags(coa);

   for (auto cell : triangulation->active_cell_iterators())
      cell->clear_refine_flag();

   std::vector<Patch> patches = mesh->get_forward_patches();
   patches[steps / 4].emplace_back(ref, coa);
   mesh->set_forward_patches(patches);

   mesh->get_dof_handler(0);

   run_reference_test<dim>(mesh, k, constants, expect, save);
}

template<int dim>
void run_reference_test_refined(int fe_order, int quad_order, int refines, Point<dim, int> k, Point<2> constants,
      double t_end, int steps, bool expect = true, bool save = false) {
   auto triangulation = std::make_shared<Triangulation<dim>>();
   GridGenerator::hyper_cube(*triangulation, 0.0, numbers::PI);
   Util::set_all_boundary_ids(*triangulation, 0);
   triangulation->refine_global(refines);

   // flag some cells for refinement and refine them
   for (auto cell : triangulation->active_cell_iterators())
      if (cell->center()[1] > numbers::PI / 2) cell->set_refine_flag();

   triangulation->execute_coarsening_and_refinement();

   double t_start = 0.0, dt = t_end / steps;
   std::vector<double> times;

   for (size_t i = 0; t_start + i * dt <= t_end; i++)
      times.push_back(t_start + i * dt);

   FE_Q<dim> fe(fe_order);
   Quadrature<dim> quad = QGauss<dim>(quad_order);  // exact in poly degree 2n-1 (needed: fe_dim^3)

   auto mesh = std::make_shared<ConstantMesh<dim>>(times, fe, quad, triangulation);

   run_reference_test<dim>(mesh, k, constants, expect, save);
}

TEST(WaveEquation, DiscretizedParameters1DFE1) {
   run_discretized_test<1>(1, 3, 8);
}

TEST(WaveEquation, DiscretizedParameters1DFE2) {
   run_discretized_test<1>(2, 4, 8);
}

TEST(WaveEquation, DiscretizedParameters2DFE1) {
   run_discretized_test<2>(1, 3, 3);
}

TEST(WaveEquation, DiscretizedParameters2DFE2) {
   run_discretized_test<2>(2, 4, 3);
}

TEST(WaveEquation, DiscretizedParameters3DFE1) {
   run_discretized_test<3>(1, 3, 1);
}

TEST(WaveEquation, ReferenceTest1DFE1) {
   for (int steps = 128; steps <= 1024; steps *= 2)
      run_reference_test_constant<1>(1, 3, 10, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps,
            steps >= 64);

   for (int refine = 7; refine >= 1; refine--)
      run_reference_test_constant<1>(1, 3, refine, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, 1024, false);
}

TEST(WaveEquation, ReferenceTest1DFE2) {
   for (int steps = 16; steps <= 128; steps *= 2)
      run_reference_test_constant<1>(2, 4, 7, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps,
            steps >= 64);

   for (int refine = 6; refine >= 1; refine--)
      run_reference_test_constant<1>(2, 4, refine, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, 128, false);
}

TEST(WaveEquation, ReferenceTest2DFE1) {
   for (int steps = 16; steps <= 256; steps *= 2)
      run_reference_test_constant<2>(1, 3, 5, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps,
            steps >= 64);

   for (int refine = 5; refine >= 1; refine--)
      run_reference_test_constant<2>(1, 3, refine, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, 256,
            false);
}

TEST(WaveEquation, ReferenceTestAdaptive2DFE1) {
   for (int steps = 16; steps <= 256; steps *= 2)
      run_reference_test_adaptive<2>(1, 3, 5, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps,
            steps >= 64, false);

   for (int refine = 5; refine >= 1; refine--)
      run_reference_test_adaptive<2>(1, 3, refine, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, 256,
            false);
}

TEST(WaveEquation, ReferenceTestRefined2DFE1) {
   for (int steps = 16; steps <= 256; steps *= 2)
      run_reference_test_refined<2>(1, 3, 5, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps,
            steps >= 64, false);

   for (int refine = 5; refine >= 1; refine--)
      run_reference_test_refined<2>(1, 3, refine, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, 256, false);
}

TEST(WaveEquation, ReferenceTest3DFE1) {
   for (int steps = 8; steps <= 32; steps *= 2)
      run_reference_test_constant<3>(1, 3, 3, Point<3, int>(1, 2, 3), Point<2>(0.7, 1.2), 2 * numbers::PI, steps,
            steps >= 32);

   for (int refine = 2; refine >= 0; refine--)
      run_reference_test_constant<3>(1, 3, refine, Point<3, int>(1, 2, 3), Point<2>(0.7, 1.2), 2 * numbers::PI, 32,
            false);
}

TEST(WaveEquation, ReferenceTestNu1DFE1) {
   for (int steps = 128; steps <= 1024; steps *= 2)
      run_reference_test_nu<1>(1, 3, 10, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps, steps >= 64);

   for (int refine = 7; refine >= 1; refine--)
      run_reference_test_nu<1>(1, 3, refine, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, 1024, false);
}

TEST(WaveEquation, ReferenceTestNu1DFE2) {
   for (int steps = 16; steps <= 128; steps *= 2)
      run_reference_test_nu<1>(2, 4, 7, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps, steps >= 64);

   for (int refine = 6; refine >= 1; refine--)
      run_reference_test_nu<1>(2, 4, refine, Point<1, int>(2), Point<2>(1.0, 1.5), 2 * numbers::PI, 128, false);
}

TEST(WaveEquation, ReferenceTestNu2DFE1) {
   for (int steps = 16; steps <= 512; steps *= 2)
      run_reference_test_nu<2>(1, 3, 6, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, steps, steps >= 64);

   for (int refine = 5; refine >= 1; refine--)
      run_reference_test_nu<2>(1, 3, refine, Point<2, int>(1, 2), Point<2>(1.0, 1.5), 2 * numbers::PI, 256, false);
}

TEST(WaveEquation, ReferenceTestNu3DFE1) {
   for (int steps = 8; steps <= 32; steps *= 2)
      run_reference_test_nu<3>(1, 3, 3, Point<3, int>(1, 2, 3), Point<2>(0.7, 1.2), 2 * numbers::PI, steps,
            steps >= 32);

   for (int refine = 2; refine >= 0; refine--)
      run_reference_test_nu<3>(1, 3, refine, Point<3, int>(1, 2, 3), Point<2>(0.7, 1.2), 2 * numbers::PI, 32, false);
}

max_stars_repo_name: thiesgerken/wavepi

max_issues_repo_name: thiesgerken/wavepi

max_forks_repo_name: thiesgerken/wavepi


text:
///
/// Copyright (c) 2009-2014 Nous Xiong (348944179 at qq dot com)
///
/// 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)
///
/// See https://github.com/nousxiong/gce for latest version.
///

#include <gce/actor/impl/tcp/socket.hpp>
#include <gce/actor/detail/cache_pool.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/placeholders.hpp>
#include <boost/bind.hpp>

namespace gce
{
namespace tcp
{
///----------------------------------------------------------------------------
socket::socket(io_service_t& ios)
  : user_(0)
  , reso_(ios)
  , sock_(ios)
  , closed_(false)
  , reconn_(false)
  , sending_(false)
  , sending_buffer_(0)
  , standby_buffer_(1)
  , sync_(ios)
  , waiting_end_(false)
{
}
///----------------------------------------------------------------------------
socket::socket(io_service_t& ios, std::string const& host, std::string const& port)
  : user_(0)
  , reso_(ios)
  , sock_(ios)
  , host_(host)
  , port_(port)
  , closed_(false)
  , reconn_(false)
  , sending_(false)
  , sending_buffer_(0)
  , standby_buffer_(1)
  , sync_(ios)
  , waiting_end_(false)
{
}
///----------------------------------------------------------------------------
socket::~socket()
{
}
///----------------------------------------------------------------------------
void socket::init(gce::detail::cache_pool* user)
{
  user_ = user;
}
///----------------------------------------------------------------------------
void socket::send(
  byte_t const* header, std::size_t header_size,
  byte_t const* body, std::size_t body_size
  )
{
  if (!waiting_end_)
  {
    gce::detail::bytes_t& send_buf = send_buffer_[standby_buffer_];
    send_buf.append(header, header_size);
    send_buf.append(body, body_size);

    if (!sending_)
    {
      begin_send();
    }
  }
}
///----------------------------------------------------------------------------
std::size_t socket::recv(byte_t* buf, std::size_t size, yield_t yield)
{
  return sock_.async_read_some(boost::asio::buffer(buf, size), yield);
}
///----------------------------------------------------------------------------
void socket::connect(yield_t yield)
{
  if (sending_)
  {
    reconn_ = true;
  }
  close_socket();
  boost::asio::ip::tcp::resolver::query query(host_, port_);
  boost::asio::ip::tcp::resolver::iterator itr = reso_.async_resolve(query, yield);
  if (yield.ec_ && *yield.ec_)
  {
    return;
  }
  boost::asio::async_connect(sock_, itr, yield);
}
///----------------------------------------------------------------------------
void socket::close()
{
  closed_ = true;
  if (!sending_)
  {
    close_socket();
  }
}
///----------------------------------------------------------------------------
void socket::wait_end(yield_t yield)
{
  waiting_end_ = true;
  if (sending_)
  {
    errcode_t ec;
    sync_.expires_from_now(infin);
    sync_.async_wait(yield[ec]);
  }
}
///----------------------------------------------------------------------------
void socket::reset()
{
  close_socket();
}
///----------------------------------------------------------------------------
void socket::close_socket()
{
  errcode_t ignore_ec;
  sock_.close(ignore_ec);
}
///----------------------------------------------------------------------------
void socket::begin_send()
{
  sending_ = true;
  strand_t& snd = user_->get_strand();
  std::swap(sending_buffer_, standby_buffer_);
  gce::detail::bytes_t const& bytes = send_buffer_[sending_buffer_];

  boost::asio::async_write(
    sock_,
    boost::asio::buffer(bytes.data(), bytes.size()),
    snd.wrap(
      boost::bind(
        &socket::end_send, this,
        boost::asio::placeholders::error
        )
      )
    );
}
///----------------------------------------------------------------------------
void socket::end_send(errcode_t const& errc)
{
  sending_ = false;
  send_buffer_[sending_buffer_].clear();

  if (!errc && !send_buffer_[standby_buffer_].empty())
  {
    begin_send();
  }
  else if (closed_)
  {
    close();
  }

  if (closed_ && !sending_)
  {
    errcode_t ignore_ec;
    sync_.cancel(ignore_ec);
  }

  reconn_ = false;
}
///----------------------------------------------------------------------------
}
}

max_stars_repo_name: nousxiong/gce

max_issues_repo_name: txwdyzcb/gce

max_forks_repo_name: txwdyzcb/gce


text:
// @COPYRIGHT@
// Licensed under MIT license (LICENSE.txt)

// clxx/app/options/options_description.hpp

/** // doc: clxx/app/options/options_description.hpp {{{
 * \file clxx/app/options/options_description.hpp
 * \brief Defines the \ref clxx::app::options::options_description class
 */ // }}}
#ifndef CLXX_APP_OPTIONS_OPTIONS_DESCRIPTION_HPP_INCLUDED
#define CLXX_APP_OPTIONS_OPTIONS_DESCRIPTION_HPP_INCLUDED

#include <boost/program_options/options_description.hpp>

namespace clxx { namespace app { namespace options {
/** // doc: option_description {{{
 * \todo Write documentation
 */ // }}}
using option_description = boost::program_options::option_description;
/** // doc: options_description_easy_init {{{
 * \todo Write documentation
 */ // }}}
using options_description_easy_init = boost::program_options::options_description_easy_init;
/** // doc: options_description {{{
 * \brief A set of option descriptions
 *
 * This is just a direct descendant of
 * boost::program_options::options_description
 * class. It's here in case we had to introduce any fixes/workarounds etc.
 */ // }}}
class options_description
  : public boost::program_options::options_description
{
public:
  /** // doc: option_description {{{
   * \brief A type used to represent single option description
   */ // }}}
  typedef boost::program_options::option_description option_description;
public:
  /** // doc: ~options_description() {{{
   * \brief Destructor
   */ // }}}
  ~options_description();
  /** // doc: options_description(...) {{{
   * \brief Constructor
   *
   * \param line_length Passed directly to the constructor of the base class
   *    \c boost::program_options::options_description.
   * \param min_description_length Passed directly to the constructor of the
   *    base class \c boost::program_options::options_description.
   */ // }}}
  options_description( unsigned line_length = m_default_line_length
        , unsigned min_description_length = m_default_line_length/2);
  /** // doc: options_description(...) {{{
   * \brief Constructor
   *
   * \param caption Passed directly to the constructor of the base class
   *    \c boost::program_options::options_description.
   * \param line_length Passed directly to the constructor of the base class
   *    \c boost::program_options::options_description.
   * \param min_description_length Passed directly to the constructor of the
   *    base class \c boost::program_options::options_description.
   */ // }}}
  options_description( const std::string & caption
        , unsigned line_length = m_default_line_length
        , unsigned min_description_length = m_default_line_length/2);
};
} } } // end namespace clxx::app::options

#endif /* CLXX_APP_OPTIONS_OPTIONS_DESCRIPTION_HPP_INCLUDED */
// vim: set expandtab tabstop=2 shiftwidth=2:
// vim: set foldmethod=marker foldcolumn=4:

max_stars_repo_name: ptomulik/clxx

max_issues_repo_name: ptomulik/clxx

max_forks_repo_name: ptomulik/clxx