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.

#include "exec/hdfs-parquet-scanner.h"

#include <boost/algorithm/string.hpp>
#include <gutil/strings/substitute.h>
#include <limits> // for std::numeric_limits

#include "common/object-pool.h"
#include "exec/hdfs-scan-node.h"
#include "exec/scanner-context.inline.h"
#include "exec/read-write-util.h"
#include "exprs/expr.h"
#include "runtime/descriptors.h"
#include "runtime/runtime-state.h"
#include "runtime/mem-pool.h"
#include "runtime/raw-value.h"
#include "runtime/row-batch.h"
#include "runtime/tuple-row.h"
#include "runtime/tuple.h"
#include "runtime/string-value.h"
#include "util/bitmap.h"
#include "util/bit-util.h"
#include "util/decompress.h"
#include "util/debug-util.h"
#include "util/dict-encoding.h"
#include "util/rle-encoding.h"
#include "util/runtime-profile.h"
#include "rpc/thrift-util.h"

using namespace std;
using namespace boost;
using namespace boost::algorithm;
using namespace impala;
using namespace strings;

Status HdfsParquetScanner::IssueInitialRanges(HdfsScanNode* scan_node,
    const std::vector<HdfsFileDesc*>& files) {
  vector<DiskIoMgr::ScanRange*> footer_ranges;
  for (int i = 0; i < files.size(); ++i) {
    for (int j = 0; j < files[i]->splits.size(); ++j) {
      DiskIoMgr::ScanRange* split = files[i]->splits[j];

      // Since Parquet scanners always read entire files, only read a file if we're
      // assigned the first split to avoid reading multi-block files with multiple
      // scanners.
      // We only process the split that starts at offset 0.
      if (split->offset() != 0) {
        // We are expecting each file to be one hdfs block (so all the scan range offsets
        // should be 0).  This is not incorrect but we will issue a warning.
        stringstream ss;
        ss << "Parquet file should not be split into multiple hdfs-blocks."
           << " file=" << files[i]->filename;
        scan_node->runtime_state()->LogError(ss.str());
        // We assign the entire file to one scan range, so mark all but one split
        // (i.e. the first split) as complete
        scan_node->RangeComplete(THdfsFileFormat::PARQUET, THdfsCompression::NONE);
        continue;
      }

      // Compute the offset of the file footer
      DCHECK_GT(files[i]->file_length, 0);
      int64_t footer_size = min(static_cast<int64_t>(FOOTER_SIZE), files[i]->file_length);
      int64_t footer_start = files[i]->file_length - footer_size;

      ScanRangeMetadata* metadata =
          reinterpret_cast<ScanRangeMetadata*>(split->meta_data());
      DiskIoMgr::ScanRange* footer_range = scan_node->AllocateScanRange(
          files[i]->filename.c_str(), footer_size,
          footer_start, metadata->partition_id, split->disk_id(), split->try_cache());
      footer_ranges.push_back(footer_range);
    }
  }
  RETURN_IF_ERROR(scan_node->AddDiskIoRanges(footer_ranges));
  return Status::OK;
}

namespace impala {

HdfsParquetScanner::HdfsParquetScanner(HdfsScanNode* scan_node, RuntimeState* state)
    : HdfsScanner(scan_node, state),
      metadata_range_(NULL),
      dictionary_pool_(new MemPool(scan_node->mem_tracker())),
      assemble_rows_timer_(scan_node_->materialize_tuple_timer()) {
  assemble_rows_timer_.Stop();
}

HdfsParquetScanner::~HdfsParquetScanner() {
}

// Reader for a single column from the parquet file.  It's associated with a
// ScannerContext::Stream and is responsible for decoding the data.
// Super class for per-type column readers. This contains most of the logic,
// the type specific functions must be implemented in the subclass.
class HdfsParquetScanner::BaseColumnReader {
 public:
  virtual ~BaseColumnReader() {}

  // This is called once for each row group in the file.
  Status Reset(const parquet::SchemaElement* schema_element,
      const parquet::ColumnMetaData* metadata, ScannerContext::Stream* stream) {
    DCHECK(stream != NULL);
    DCHECK(metadata != NULL);

    field_repetition_type_ = schema_element->repetition_type;
    num_buffered_values_ = 0;
    data_ = NULL;
    stream_ = stream;
    metadata_ = metadata;
    dict_decoder_base_ = NULL;
    num_values_read_ = 0;
    if (metadata_->codec != parquet::CompressionCodec::UNCOMPRESSED) {
      RETURN_IF_ERROR(Codec::CreateDecompressor(
          NULL, false, PARQUET_TO_IMPALA_CODEC[metadata_->codec], &decompressor_));
    }
    return Status::OK;
  }

  // Called once when the scanner is complete for final cleanup.
  void Close() {
    if (decompressor_.get() != NULL) decompressor_->Close();
  }

  int64_t total_len() const { return metadata_->total_compressed_size; }
  const SlotDescriptor* slot_desc() const { return desc_; }
  int file_idx() const { return file_idx_; }
  THdfsCompression::type codec() const {
    if (metadata_ == NULL) return THdfsCompression::NONE;
    return PARQUET_TO_IMPALA_CODEC[metadata_->codec];
  }

  // Read the next value into tuple for this column.  Returns false if there are no
  // more values in the file.
  // *conjuncts_failed is an in/out parameter. If false, it means this row has already
  // been filtered out (i.e. ReadValue is really a SkipValue()) and should be set to
  // true if ReadValue() can filter out this row.
  // TODO: this is the function that needs to be codegen'd (e.g. CodegenReadValue())
  // The codegened functions from all the materialized cols will then be combined
  // into one function.
  // TODO: another option is to materialize col by col for the entire row batch in
  // one call.  e.g. MaterializeCol would write out 1024 values.  Our row batches
  // are currently dense so we'll need to figure out something there.
  bool ReadValue(MemPool* pool, Tuple* tuple, bool* conjuncts_failed);

  // TODO: Some encodings might benefit a lot from a SkipValues(int num_rows) if
  // we know this row can be skipped. This could be very useful with stats and big
  // sections can be skipped. Implement that when we can benefit from it.

 protected:
  friend class HdfsParquetScanner;

  HdfsParquetScanner* parent_;
  const SlotDescriptor* desc_;

  // Index of this column in the this parquet file.
  int file_idx_;

  // This is either required, optional or repeated.
  // If it is required, the column cannot have NULLs.
  parquet::FieldRepetitionType::type field_repetition_type_;

  const parquet::ColumnMetaData* metadata_;
  scoped_ptr<Codec> decompressor_;
  ScannerContext::Stream* stream_;

  // Pool to allocate decompression buffers from.
  boost::scoped_ptr<MemPool> decompressed_data_pool_;

  // Header for current data page.
  parquet::PageHeader current_page_header_;

  // Num values remaining in the current data page
  int num_buffered_values_;

  // Pointer to start of next value in data page
  uint8_t* data_;

  // Decoder for definition.  Only one of these is valid at a time, depending on
  // the data page metadata.
  RleDecoder rle_def_levels_;
  BitReader bit_packed_def_levels_;

  // Decoder for dictionary-encoded columns. Set by the subclass.
  DictDecoderBase* dict_decoder_base_;

  // The number of values seen so far. Updated per data page.
  int64_t num_values_read_;

  // Cache of the bitmap_filter_ (if any) for this slot.
  const Bitmap* bitmap_filter_;
  // Cache of hash_seed_ to use with bitmap_filter_.
  uint32_t hash_seed_;

  // Bitmap filters are optional (i.e. they can be ignored and the results will be
  // correct). Keep track of stats to determine if the filter is not effective. If
  // the number of rows filtered out is too low, this is not worth the cost.
  // TODO: this should be cost based taking into account how much we save when we
  // filter a row.
  int64_t rows_returned_;
  int64_t bitmap_filter_rows_rejected_;

  BaseColumnReader(HdfsParquetScanner* parent, const SlotDescriptor* desc, int file_idx)
    : parent_(parent),
      desc_(desc),
      file_idx_(file_idx),
      field_repetition_type_(parquet::FieldRepetitionType::OPTIONAL),
      metadata_(NULL),
      stream_(NULL),
      decompressed_data_pool_(new MemPool(parent->scan_node_->mem_tracker())),
      num_buffered_values_(0),
      num_values_read_(0) {
    RuntimeState* state = parent_->scan_node_->runtime_state();
    bitmap_filter_ = state->GetBitmapFilter(desc_->id());
    hash_seed_ = state->fragment_hash_seed();
    rows_returned_ = 0;
    bitmap_filter_rows_rejected_ = 0;
  }

  // Read the next data page.  If a dictionary page is encountered, that will
  // be read and this function will continue reading for the next data page.
  Status ReadDataPage();

  // Returns the definition level for the next value
  // Returns -1 if there was a error parsing it.
  int ReadDefinitionLevel();

  // Creates a dictionary decoder from values/size. Subclass must implement this
  // and set dict_decoder_base_.
  virtual void CreateDictionaryDecoder(uint8_t* values, int size) = 0;

  // Initializes the reader with the data contents. This is the content for
  // the entire decompressed data page. Decoders can initialize state from
  // here.
  virtual Status InitDataPage(uint8_t* data, int size) = 0;

  // Writes the next value into *slot using pool if necessary.
  // Returns false if there was an error.
  // Subclass must implement this.
  // TODO: we need to remove this with codegen.
  virtual bool ReadSlot(void* slot, MemPool* pool, bool* conjuncts_failed) = 0;
};

// Per column type reader.
template<typename T>
class HdfsParquetScanner::ColumnReader : public HdfsParquetScanner::BaseColumnReader {
 public:
  ColumnReader(HdfsParquetScanner* parent, const SlotDescriptor* desc, int file_idx)
    : BaseColumnReader(parent, desc, file_idx) {
    DCHECK_NE(desc->type().type, TYPE_BOOLEAN);
    if (desc->type().type == TYPE_DECIMAL) {
      fixed_len_size_ = ParquetPlainEncoder::DecimalSize(desc->type());
    } else if (desc->type().type == TYPE_VARCHAR) {
      fixed_len_size_ = desc->type().len;
    } else {
      fixed_len_size_ = -1;
    }
  }

 protected:
  virtual void CreateDictionaryDecoder(uint8_t* values, int size) {
    dict_decoder_.reset(new DictDecoder<T>(values, size, fixed_len_size_));
    dict_decoder_base_ = dict_decoder_.get();
  }

  virtual Status InitDataPage(uint8_t* data, int size) {
    if (current_page_header_.data_page_header.encoding ==
          parquet::Encoding::PLAIN_DICTIONARY) {
      if (dict_decoder_.get() == NULL) {
        return Status("File corrupt. Missing dictionary page.");
      }
      dict_decoder_->SetData(data, size);
    }

    // Check if we should disable the bitmap filter. We'll do this if the filter
    // is not removing a lot of rows.
    // TODO: how to pick the selectivity?
    if (bitmap_filter_ != NULL && rows_returned_ > 10000 &&
        bitmap_filter_rows_rejected_ < rows_returned_ * .1) {
      bitmap_filter_ = NULL;
    }
    return Status::OK;
  }

  virtual bool ReadSlot(void* slot, MemPool* pool, bool* conjuncts_failed)  {
    parquet::Encoding::type page_encoding =
        current_page_header_.data_page_header.encoding;
    bool result = true;
    if (page_encoding == parquet::Encoding::PLAIN_DICTIONARY) {
      result = dict_decoder_->GetValue(reinterpret_cast<T*>(slot));
    } else {
      DCHECK(page_encoding == parquet::Encoding::PLAIN);
      data_ += ParquetPlainEncoder::Decode<T>(data_, fixed_len_size_,
          reinterpret_cast<T*>(slot));
      if (parent_->scan_node_->requires_compaction()) {
        CopySlot(reinterpret_cast<T*>(slot), pool);
      }
    }
    ++rows_returned_;
    if (!*conjuncts_failed && bitmap_filter_ != NULL) {
      uint32_t h = RawValue::GetHashValue(slot, desc_->type(), hash_seed_);
      *conjuncts_failed = !bitmap_filter_->Get<true>(h);
      ++bitmap_filter_rows_rejected_;
    }
    return result;
  }

 private:
  void CopySlot(T* slot, MemPool* pool) {
    // no-op for non-string columns.
  }

  scoped_ptr<DictDecoder<T> > dict_decoder_;

  // The size of this column with plain encoding for FIXED_LEN_BYTE_ARRAY, or
  // the max length for VARCHAR columns. Unused otherwise.
  int fixed_len_size_;
};

template<>
void HdfsParquetScanner::ColumnReader<StringValue>::CopySlot(
    StringValue* slot, MemPool* pool) {
  if (slot->len == 0) return;
  uint8_t* buffer = pool->Allocate(slot->len);
  memcpy(buffer, slot->ptr, slot->len);
  slot->ptr = reinterpret_cast<char*>(buffer);
}

class HdfsParquetScanner::BoolColumnReader : public HdfsParquetScanner::BaseColumnReader {
 public:
  BoolColumnReader(HdfsParquetScanner* parent, const SlotDescriptor* desc, int file_idx)
    : BaseColumnReader(parent, desc, file_idx) {
    DCHECK_EQ(desc->type().type, TYPE_BOOLEAN);
  }

 protected:
  virtual void CreateDictionaryDecoder(uint8_t* values, int size) {
    DCHECK(false) << "Dictionary encoding is not supported for bools. Should never "
                  << "have gotten this far.";
  }

  virtual Status InitDataPage(uint8_t* data, int size) {
    // Initialize bool decoder
    bool_values_ = BitReader(data, size);
    return Status::OK;
  }

  virtual bool ReadSlot(void* slot, MemPool* pool, bool* conjuncts_failed)  {
    bool valid = bool_values_.GetValue(1, reinterpret_cast<bool*>(slot));
    if (!valid) parent_->parse_status_ = Status("Invalid bool column.");
    return valid;
  }

 private:
  BitReader bool_values_;
};

}

Status HdfsParquetScanner::Prepare(ScannerContext* context) {
  RETURN_IF_ERROR(HdfsScanner::Prepare(context));
  num_cols_counter_ =
      ADD_COUNTER(scan_node_->runtime_profile(), "NumColumns", TCounterType::UNIT);

  scan_node_->IncNumScannersCodegenDisabled();
  return Status::OK;
}

void HdfsParquetScanner::Close() {
  vector<THdfsCompression::type> compression_types;
  for (int i = 0; i < column_readers_.size(); ++i) {
    if (column_readers_[i]->decompressed_data_pool_.get() != NULL) {
      // No need to commit the row batches with the AttachPool() calls
      // since AddFinalRowBatch() already does below.
      AttachPool(column_readers_[i]->decompressed_data_pool_.get(), false);
    }
    column_readers_[i]->Close();
    compression_types.push_back(column_readers_[i]->codec());
  }
  AttachPool(dictionary_pool_.get(), false);
  AddFinalRowBatch();

  // If this was a metadata only read (i.e. count(*)), there are no columns.
  if (compression_types.empty()) compression_types.push_back(THdfsCompression::NONE);
  scan_node_->RangeComplete(THdfsFileFormat::PARQUET, compression_types);
  assemble_rows_timer_.Stop();
  assemble_rows_timer_.ReleaseCounter();

  HdfsScanner::Close();
}

HdfsParquetScanner::BaseColumnReader* HdfsParquetScanner::CreateReader(
    SlotDescriptor* desc, int file_idx) {
  BaseColumnReader* reader = NULL;
  switch (desc->type().type) {
    case TYPE_BOOLEAN:
      reader = new BoolColumnReader(this, desc, file_idx);
      break;
    case TYPE_TINYINT:
      reader = new ColumnReader<int8_t>(this, desc, file_idx);
      break;
    case TYPE_SMALLINT:
      reader = new ColumnReader<int16_t>(this, desc, file_idx);
      break;
    case TYPE_INT:
      reader = new ColumnReader<int32_t>(this, desc, file_idx);
      break;
    case TYPE_BIGINT:
      reader = new ColumnReader<int64_t>(this, desc, file_idx);
      break;
    case TYPE_FLOAT:
      reader = new ColumnReader<float>(this, desc, file_idx);
      break;
    case TYPE_DOUBLE:
      reader = new ColumnReader<double>(this, desc, file_idx);
      break;
    case TYPE_TIMESTAMP:
      reader = new ColumnReader<TimestampValue>(this, desc, file_idx);
      break;
    case TYPE_STRING:
    case TYPE_VARCHAR:
      reader = new ColumnReader<StringValue>(this, desc, file_idx);
      break;
    case TYPE_DECIMAL:
      switch (desc->type().GetByteSize()) {
        case 4:
          reader = new ColumnReader<Decimal4Value>(this, desc, file_idx);
          break;
        case 8:
          reader = new ColumnReader<Decimal8Value>(this, desc, file_idx);
          break;
        case 16:
          reader = new ColumnReader<Decimal16Value>(this, desc, file_idx);
          break;
      }
      break;
    default:
      DCHECK(false);
  }
  return scan_node_->runtime_state()->obj_pool()->Add(reader);
}

// In 1.1, we had a bug where the dictionary page metadata was not set. Returns true
// if this matches those versions and compatibility workarounds need to be used.
static bool RequiresSkippedDictionaryHeaderCheck(
    const HdfsParquetScanner::FileVersion& v) {
  if (v.application != "impala") return false;
  return v.VersionEq(1,1,0) || (v.VersionEq(1,2,0) && v.is_impala_internal);
}

Status HdfsParquetScanner::BaseColumnReader::ReadDataPage() {
  Status status;
  uint8_t* buffer;
  int num_bytes;

  // We're about to move to the next data page.  The previous data page is
  // now complete, pass along the memory allocated for it.
  parent_->AttachPool(decompressed_data_pool_.get(), false);

  // Read the next data page, skipping page types we don't care about.
  // We break out of this loop on the non-error case (a data page was found or we read all
  // the pages).
  while (true) {
    DCHECK_EQ(num_buffered_values_, 0);
    if (num_values_read_ >= metadata_->num_values) {
      // No more pages to read
      DCHECK_EQ(num_values_read_, metadata_->num_values);
      break;
    }

    int64_t buffer_size;
    RETURN_IF_ERROR(stream_->GetBuffer(true, &buffer, &buffer_size));
    num_bytes = min(buffer_size, static_cast<int64_t>(MAX_PAGE_HEADER_SIZE));
    if (num_bytes == 0) {
      DCHECK(stream_->eosr());
      stringstream ss;
      ss << "Column metadata states there are " << metadata_->num_values
         << " values, but only read " << num_values_read_ << " values from column "
         << (desc_->col_pos() - parent_->scan_node_->num_partition_keys());
      if (parent_->scan_node_->runtime_state()->abort_on_error()) {
        return Status(ss.str());
      } else {
        parent_->scan_node_->runtime_state()->LogError(ss.str());
      }
    }

    // We don't know the actual header size until the thrift object is deserialized.
    uint32_t header_size = num_bytes;
    status = DeserializeThriftMsg(
        buffer, &header_size, true, &current_page_header_, true);
    if (!status.ok()) {
      if (header_size >= MAX_PAGE_HEADER_SIZE) {
        status.AddErrorMsg("ParquetScanner: Could not deserialize page header.");
        return status;
      }
      // Stitch the header bytes that are split across buffers.
      uint8_t header_buffer[MAX_PAGE_HEADER_SIZE];
      int32_t header_first_part = header_size;
      memcpy(header_buffer, buffer, header_first_part);

      if (!stream_->SkipBytes(header_first_part, &status)) return status;
      RETURN_IF_ERROR(stream_->GetBuffer(true, &buffer, &buffer_size));
      num_bytes = min(buffer_size, static_cast<int64_t>(MAX_PAGE_HEADER_SIZE));
      if (num_bytes == 0) return status;
      uint32_t header_second_part = ::min(num_bytes,
                                          MAX_PAGE_HEADER_SIZE - header_first_part);
      memcpy(header_buffer + header_first_part, buffer, header_second_part);
      header_size = MAX_PAGE_HEADER_SIZE;
      status =
          DeserializeThriftMsg(header_buffer, &header_size, true, &current_page_header_);

      if (!status.ok()) {
        status.AddErrorMsg("ParquetScanner: Could not deserialize page header");
        return status;
      }
      DCHECK_GT(header_size, header_first_part);
      header_size = header_size - header_first_part;
    }
    if (!stream_->SkipBytes(header_size, &status)) return status;

    int data_size = current_page_header_.compressed_page_size;
    int uncompressed_size = current_page_header_.uncompressed_page_size;

    if (current_page_header_.type == parquet::PageType::DICTIONARY_PAGE) {
      if (dict_decoder_base_ != NULL) {
        return Status("Column chunk should not contain two dictionary pages.");
      }
      if (desc_->type().type == TYPE_BOOLEAN) {
        return Status("Unexpected dictionary page. Dictionary page is not"
            " supported for booleans.");
      }
      const parquet::DictionaryPageHeader* dict_header = NULL;
      if (current_page_header_.__isset.dictionary_page_header) {
        dict_header = &current_page_header_.dictionary_page_header;
      } else {
        if (!RequiresSkippedDictionaryHeaderCheck(parent_->file_version_)) {
          return Status("Dictionary page does not have dictionary header set.");
        }
      }
      if (dict_header != NULL &&
          dict_header->encoding != parquet::Encoding::PLAIN &&
          dict_header->encoding != parquet::Encoding::PLAIN_DICTIONARY) {
        return Status("Only PLAIN and PLAIN_DICTIONARY encodings are supported "
            "for dictionary pages.");
      }

      if (!stream_->ReadBytes(data_size, &data_, &status)) return status;

      uint8_t* dict_values = NULL;
      if (decompressor_.get() != NULL) {
        dict_values = parent_->dictionary_pool_->Allocate(uncompressed_size);
        RETURN_IF_ERROR(decompressor_->ProcessBlock32(true, data_size, data_,
            &uncompressed_size, &dict_values));
        VLOG_FILE << "Decompressed " << data_size << " to " << uncompressed_size;
        data_size = uncompressed_size;
      } else {
        DCHECK_EQ(data_size, current_page_header_.uncompressed_page_size);
        // Copy dictionary from io buffer (which will be recycled as we read
        // more data) to a new buffer
        dict_values = parent_->dictionary_pool_->Allocate(data_size);
        memcpy(dict_values, data_, data_size);
      }

      CreateDictionaryDecoder(dict_values, data_size);
      if (dict_header != NULL &&
          dict_header->num_values != dict_decoder_base_->num_entries()) {
        return Status(Substitute(
            "Invalid dictionary. Expected $0 entries but data contained $1 entries",
            dict_header->num_values, dict_decoder_base_->num_entries()));
      }
      // Done with dictionary page, read next page
      continue;
    }

    if (current_page_header_.type != parquet::PageType::DATA_PAGE) {
      // We can safely skip non-data pages
      if (!stream_->SkipBytes(data_size, &status)) return status;
      continue;
    }

    // Read Data Page
    if (!stream_->ReadBytes(data_size, &data_, &status)) return status;
    num_buffered_values_ = current_page_header_.data_page_header.num_values;
    num_values_read_ += num_buffered_values_;

    if (decompressor_.get() != NULL) {
      SCOPED_TIMER(parent_->decompress_timer_);
      uint8_t* decompressed_buffer = decompressed_data_pool_->Allocate(uncompressed_size);
      RETURN_IF_ERROR(decompressor_->ProcessBlock32(true,
          current_page_header_.compressed_page_size, data_, &uncompressed_size,
          &decompressed_buffer));
      VLOG_FILE << "Decompressed " << current_page_header_.compressed_page_size
                << " to " << uncompressed_size;
      DCHECK_EQ(current_page_header_.uncompressed_page_size, uncompressed_size);
      data_ = decompressed_buffer;
      data_size = current_page_header_.uncompressed_page_size;
    } else {
      DCHECK_EQ(metadata_->codec, parquet::CompressionCodec::UNCOMPRESSED);
      DCHECK_EQ(current_page_header_.compressed_page_size, uncompressed_size);
    }

    if (field_repetition_type_ == parquet::FieldRepetitionType::OPTIONAL) {
      // Initialize the definition level data
      int32_t num_definition_bytes = 0;
      switch (current_page_header_.data_page_header.definition_level_encoding) {
        case parquet::Encoding::RLE:
          if (!ReadWriteUtil::Read(&data_, &data_size, &num_definition_bytes, &status)) {
            return status;
          }
          rle_def_levels_ = RleDecoder(data_, num_definition_bytes, 1);
          break;
        case parquet::Encoding::BIT_PACKED:
          num_definition_bytes = BitUtil::Ceil(num_buffered_values_, 8);
          bit_packed_def_levels_ = BitReader(data_, num_definition_bytes);
          break;
        default: {
          stringstream ss;
          ss << "Unsupported definition level encoding: "
            << current_page_header_.data_page_header.definition_level_encoding;
          return Status(ss.str());
        }
      }
      DCHECK_GT(num_definition_bytes, 0);
      data_ += num_definition_bytes;
      data_size -= num_definition_bytes;
    }

    // Data can be empty if the column contains all NULLs
    if (data_size != 0) RETURN_IF_ERROR(InitDataPage(data_, data_size));
    break;
  }

  return Status::OK;
}

// TODO More codegen here as well.
inline int HdfsParquetScanner::BaseColumnReader::ReadDefinitionLevel() {
  if (field_repetition_type_ == parquet::FieldRepetitionType::REQUIRED) {
    // This column is required so there is nothing encoded for the definition
    // levels.
    return 1;
  }

  uint8_t definition_level;
  bool valid = false;
  switch (current_page_header_.data_page_header.definition_level_encoding) {
    case parquet::Encoding::RLE:
      valid = rle_def_levels_.Get(&definition_level);
      break;
    case parquet::Encoding::BIT_PACKED: {
      valid = bit_packed_def_levels_.GetValue(1, &definition_level);
      break;
    }
    default:
      DCHECK(false);
  }
  if (!valid) return -1;
  return definition_level;
}

inline bool HdfsParquetScanner::BaseColumnReader::ReadValue(
    MemPool* pool, Tuple* tuple, bool* conjuncts_failed) {
  if (num_buffered_values_ == 0) {
    parent_->assemble_rows_timer_.Stop();
    parent_->parse_status_ = ReadDataPage();
    // We don't return Status objects as parameters because they are too
    // expensive for per row/per col calls.  If ReadDataPage failed,
    // return false to indicate this column reader is done.
    if (num_buffered_values_ == 0 || !parent_->parse_status_.ok()) return false;
    parent_->assemble_rows_timer_.Start();
  }

  --num_buffered_values_;
  int definition_level = ReadDefinitionLevel();
  if (definition_level < 0) return false;

  if (definition_level == 0) {
    // Null value
    tuple->SetNull(desc_->null_indicator_offset());
    return true;
  }

  DCHECK_EQ(definition_level, 1);
  return ReadSlot(tuple->GetSlot(desc_->tuple_offset()), pool, conjuncts_failed);
}

Status HdfsParquetScanner::ProcessSplit() {
  HdfsFileDesc* file_desc = scan_node_->GetFileDesc(stream_->filename());
  DCHECK(file_desc != NULL);

  // First process the file metadata in the footer
  bool eosr;
  RETURN_IF_ERROR(ProcessFooter(&eosr));

  if (eosr) return Status::OK;

  // We've processed the metadata and there are columns that need to be
  // materialized.
  COUNTER_SET(num_cols_counter_, static_cast<int64_t>(column_readers_.size()));
  RETURN_IF_ERROR(CreateColumnReaders());

  // Iterate through each row group in the file and read all the materialized columns
  // per row group.  Row groups are independent, so this this could be parallelized.
  // However, having multiple row groups per file should be seen as an edge case and
  // we can do better parallelizing across files instead.
  // TODO: not really an edge case since MR writes multiple row groups
  for (int i = 0; i < file_metadata_.row_groups.size(); ++i) {
    // Attach any resources and clear the streams before starting a new row group. These
    // streams could either be just the footer stream or streams for the previous row
    // group.
    context_->ReleaseCompletedResources(batch_, /* done */ true);
    // Commit the rows to flush the row batch from the previous row group
    CommitRows(0);

    RETURN_IF_ERROR(InitColumns(i));
    RETURN_IF_ERROR(AssembleRows(i));
  }

  return Status::OK;
}

// TODO: this needs to be codegen'd.  The ReadValue function needs to be codegen'd,
// specific to type and encoding and then inlined into AssembleRows().
Status HdfsParquetScanner::AssembleRows(int row_group_idx) {
  assemble_rows_timer_.Start();
  // Read at most as many rows as stated in the metadata
  int64_t expected_rows_in_group = file_metadata_.row_groups[row_group_idx].num_rows;
  int64_t rows_read = 0;
  bool reached_limit = scan_node_->ReachedLimit();
  bool cancelled = context_->cancelled();
  int num_column_readers = column_readers_.size();
  MemPool* pool;

  while (!reached_limit && !cancelled && rows_read < expected_rows_in_group) {
    Tuple* tuple;
    TupleRow* row;
    int64_t row_mem_limit = static_cast<int64_t>(GetMemory(&pool, &tuple, &row));
    int64_t expected_rows_to_read = expected_rows_in_group - rows_read;
    int64_t num_rows = std::min(expected_rows_to_read, row_mem_limit);

    int num_to_commit = 0;
    if (num_column_readers > 0) {
      for (int i = 0; i < num_rows; ++i) {
        bool conjuncts_failed = false;
        InitTuple(template_tuple_, tuple);
        for (int c = 0; c < num_column_readers; ++c) {
          if (!column_readers_[c]->ReadValue(pool, tuple, &conjuncts_failed)) {
            assemble_rows_timer_.Stop();
            // This column is complete and has no more data.  This indicates
            // we are done with this row group.
            // For correctly formed files, this should be the first column we
            // are reading.
            DCHECK(c == 0 || !parse_status_.ok())
              << "c=" << c << " " << parse_status_.GetErrorMsg();;
            COUNTER_ADD(scan_node_->rows_read_counter(), i);
            RETURN_IF_ERROR(CommitRows(num_to_commit));

            // If we reach this point, it means that we reached the end of file for
            // this column. Test if the expected number of rows from metadata matches
            // the actual number of rows in the file.
            rows_read += i;
            if (rows_read != expected_rows_in_group) {
              HdfsParquetScanner::BaseColumnReader* reader = column_readers_[c];
              DCHECK(reader->stream_ != NULL);
              stringstream ss;
              ss << "Metadata states that in group " << reader->stream_->filename()
                 << "[" << row_group_idx << "] there are " << expected_rows_in_group
                 << " rows, but only " << rows_read << " rows were read.";
              if (scan_node_->runtime_state()->abort_on_error()) {
                return Status(ss.str());
              } else {
                scan_node_->runtime_state()->LogError(ss.str());
              }
            }
            return parse_status_;
          }
        }
        if (conjuncts_failed) continue;
        row->SetTuple(scan_node_->tuple_idx(), tuple);
        if (EvalConjuncts(row)) {
          row = next_row(row);
          tuple = next_tuple(tuple);
          ++num_to_commit;
        }
      }
    } else {
      // Special case when there is no data for the accessed column(s) in the file.
      // This can happen, for example, due to schema evolution (alter table add column).
      // Since all the tuples are same, evaluating conjuncts only for the first tuple.
      DCHECK_GT(num_rows, 0);
      InitTuple(template_tuple_, tuple);
      row->SetTuple(scan_node_->tuple_idx(), tuple);
      if (EvalConjuncts(row)) {
        row = next_row(row);
        tuple = next_tuple(tuple);

        for (int i = 1; i < num_rows; ++i) {
          InitTuple(template_tuple_, tuple);
          row->SetTuple(scan_node_->tuple_idx(), tuple);
          row = next_row(row);
          tuple = next_tuple(tuple);
        }
        num_to_commit += num_rows;
      }
    }
    rows_read += num_rows;
    COUNTER_ADD(scan_node_->rows_read_counter(), num_rows);
    RETURN_IF_ERROR(CommitRows(num_to_commit));

    reached_limit = scan_node_->ReachedLimit();
    cancelled = context_->cancelled();
  }

  if (!reached_limit && !cancelled && (num_column_readers > 0)) {
    // If we get to this point, it means that we have read as many rows as the metadata
    // told us we should read. Attempt to read one more row and if that succeeds report
    // the error.
    DCHECK_EQ(rows_read, expected_rows_in_group);
    uint8_t dummy_tuple_mem[tuple_byte_size_];
    Tuple* dummy_tuple = reinterpret_cast<Tuple*>(&dummy_tuple_mem);
    InitTuple(template_tuple_, dummy_tuple);
    bool conjuncts_failed = false;
    if (column_readers_[0]->ReadValue(pool, dummy_tuple, &conjuncts_failed)) {
      // If another tuple is successfully read, it means that there are still values
      // in the file.
      HdfsParquetScanner::BaseColumnReader* reader = column_readers_[0];
      DCHECK(reader->stream_ != NULL);
      stringstream ss;
      ss << "Metadata states that in group " << reader->stream_->filename() << "["
         << row_group_idx << "] there are " << expected_rows_in_group << " rows, but"
         << " there is at least one more row in the file.";
      if (scan_node_->runtime_state()->abort_on_error()) {
        return Status(ss.str());
      } else {
        scan_node_->runtime_state()->LogError(ss.str());
      }
    }
  }

  assemble_rows_timer_.Stop();
  return parse_status_;
}

Status HdfsParquetScanner::ProcessFooter(bool* eosr) {
  *eosr = false;

  uint8_t* buffer;
  int64_t len;

  RETURN_IF_ERROR(stream_->GetBuffer(false, &buffer, &len));
  DCHECK(stream_->eosr());

  // Number of bytes in buffer after the fixed size footer is accounted for.
  int remaining_bytes_buffered = len - sizeof(int32_t) - sizeof(PARQUET_VERSION_NUMBER);

  // Make sure footer has enough bytes to contain the required information.
  if (remaining_bytes_buffered < 0) {
    stringstream ss;
    ss << "File " << stream_->filename() << " is invalid.  Missing metadata.";
    return Status(ss.str());
  }

  // Validate magic file bytes are correct
  uint8_t* magic_number_ptr = buffer + len - sizeof(PARQUET_VERSION_NUMBER);
  if (memcmp(magic_number_ptr,
      PARQUET_VERSION_NUMBER, sizeof(PARQUET_VERSION_NUMBER) != 0)) {
    stringstream ss;
    ss << "File " << stream_->filename() << " is invalid.  Invalid file footer: "
        << string((char*)magic_number_ptr, sizeof(PARQUET_VERSION_NUMBER));
    return Status(ss.str());
  }

  // The size of the metadata is encoded as a 4 byte little endian value before
  // the magic number
  uint8_t* metadata_size_ptr = magic_number_ptr - sizeof(int32_t);
  uint32_t metadata_size = *reinterpret_cast<uint32_t*>(metadata_size_ptr);
  uint8_t* metadata_ptr = metadata_size_ptr - metadata_size;
  // If the metadata was too big, we need to stitch it before deserializing it.
  // In that case, we stitch the data in this buffer.
  vector<uint8_t> metadata_buffer;
  metadata_range_ = stream_->scan_range();

  if (UNLIKELY(metadata_size > remaining_bytes_buffered)) {
    // In this case, the metadata is bigger than our guess meaning there are
    // not enough bytes in the footer range from IssueInitialRanges().
    // We'll just issue more ranges to the IoMgr that is the actual footer.
    const HdfsFileDesc* file_desc = scan_node_->GetFileDesc(metadata_range_->file());
    // The start of the metadata is:
    // file_length - 4-byte metadata size - footer-size - metadata size
    int64_t metadata_start = file_desc->file_length -
      sizeof(int32_t) - sizeof(PARQUET_VERSION_NUMBER) - metadata_size;
    int metadata_bytes_to_read = metadata_size;

    // IoMgr can only do a fixed size Read(). The metadata could be larger
    // so we stitch it here.
    // TODO: consider moving this stitching into the scanner context. The scanner
    // context usually handles the stitching but no other scanner need this logic
    // now.
    metadata_buffer.resize(metadata_size);
    metadata_ptr = &metadata_buffer[0];
    int copy_offset = 0;
    DiskIoMgr* io_mgr = scan_node_->runtime_state()->io_mgr();

    while (metadata_bytes_to_read > 0) {
      int to_read = ::min(io_mgr->max_read_buffer_size(), metadata_bytes_to_read);
      DiskIoMgr::ScanRange* range = scan_node_->AllocateScanRange(
          metadata_range_->file(), to_read, metadata_start + copy_offset, -1,
          metadata_range_->disk_id(), metadata_range_->try_cache());

      DiskIoMgr::BufferDescriptor* io_buffer = NULL;
      RETURN_IF_ERROR(io_mgr->Read(scan_node_->reader_context(), range, &io_buffer));
      memcpy(metadata_ptr + copy_offset, io_buffer->buffer(), io_buffer->len());
      io_buffer->Return();

      metadata_bytes_to_read -= to_read;
      copy_offset += to_read;
    }
    DCHECK_EQ(metadata_bytes_to_read, 0);
  }
  // Deserialize file header
  Status status =
      DeserializeThriftMsg(metadata_ptr, &metadata_size, true, &file_metadata_);
  if (!status.ok()) {
    stringstream ss;
    ss << "File " << stream_->filename() << " has invalid file metadata at file offset "
        << (metadata_size + sizeof(PARQUET_VERSION_NUMBER) + sizeof(uint32_t)) << ". "
        << "Error = " << status.GetErrorMsg();
    return Status(ss.str());
  }

  RETURN_IF_ERROR(ValidateFileMetadata());

  // Tell the scan node this file has been taken care of.
  HdfsFileDesc* desc = scan_node_->GetFileDesc(stream_->filename());
  scan_node_->MarkFileDescIssued(desc);

  if (scan_node_->materialized_slots().empty()) {
    // No materialized columns.  We can serve this query from just the metadata.  We
    // don't need to read the column data.
    int64_t num_tuples = file_metadata_.num_rows;
    COUNTER_ADD(scan_node_->rows_read_counter(), num_tuples);

    while (num_tuples > 0) {
      MemPool* pool;
      Tuple* tuple;
      TupleRow* current_row;
      int max_tuples = GetMemory(&pool, &tuple, &current_row);
      max_tuples = min(static_cast<int64_t>(max_tuples), num_tuples);
      num_tuples -= max_tuples;

      int num_to_commit = WriteEmptyTuples(context_, current_row, max_tuples);
      RETURN_IF_ERROR(CommitRows(num_to_commit));
    }

    *eosr = true;
    return Status::OK;
  } else if (file_metadata_.num_rows == 0) {
    // Empty file
    *eosr = true;
    return Status::OK;
  }

  if (file_metadata_.row_groups.empty()) {
    stringstream ss;
    ss << "Invalid file. This file: " << stream_->filename() << " has no row groups";
    return Status(ss.str());
  }

  return Status::OK;
}

Status HdfsParquetScanner::CreateColumnReaders() {
  DCHECK(column_readers_.empty());
  int num_partition_keys = scan_node_->num_partition_keys();

  for (int i = 0; i < scan_node_->materialized_slots().size(); ++i) {
    SlotDescriptor* slot_desc = scan_node_->materialized_slots()[i];
    int col_idx = slot_desc->col_pos() - num_partition_keys;
    if (col_idx >= file_metadata_.schema[0].num_children) {
      // In this case, we are selecting a column that is not in the file.
      // Update the template tuple to put a NULL in this slot.
      if (template_tuple_ == NULL) {
        template_tuple_ = scan_node_->InitEmptyTemplateTuple();
      }
      template_tuple_->SetNull(slot_desc->null_indicator_offset());
      continue;
    }

    column_readers_.push_back(CreateReader(slot_desc, col_idx));
  }
  return Status::OK;
}

Status HdfsParquetScanner::InitColumns(int row_group_idx) {
  parquet::RowGroup& row_group = file_metadata_.row_groups[row_group_idx];

  // All the scan ranges (one for each col).
  vector<DiskIoMgr::ScanRange*> col_ranges;

  for (int i = 0; i < column_readers_.size(); ++i) {
    const SlotDescriptor* slot_desc = column_readers_[i]->desc_;
    int file_col_idx = column_readers_[i]->file_idx();

    const parquet::SchemaElement& schema_element =
        file_metadata_.schema[file_col_idx + 1];
    const parquet::ColumnChunk& col_chunk = row_group.columns[file_col_idx];
    int64_t col_start = col_chunk.meta_data.data_page_offset;

    RETURN_IF_ERROR(ValidateColumn(slot_desc, file_col_idx));

    // If there is a dictionary page, the file format requires it to come before
    // any data pages.  We need to start reading the column from the data page.
    if (col_chunk.meta_data.__isset.dictionary_page_offset) {
      if (col_chunk.meta_data.dictionary_page_offset >= col_start) {
        stringstream ss;
        ss << "File " << stream_->filename() << ": metadata is corrupt. "
           << "Dictionary page (offset=" << col_chunk.meta_data.dictionary_page_offset
           << ") must come before any data pages (offset=" << col_start << ").";
        return Status(ss.str());
      }
      col_start = col_chunk.meta_data.dictionary_page_offset;
    }
    int64_t col_len = col_chunk.meta_data.total_compressed_size;
    if (file_version_.application == "parquet-mr" && file_version_.VersionLt(1, 2, 9)) {
      // The Parquet MR writer had a bug in 1.2.8 and below where it didn't include the
      // dictionary page header size in total_compressed_size and total_uncompressed_size
      // (see IMPALA-694). We pad col_len to compensate.
      col_len += MAX_PAGE_HEADER_SIZE;
    }

    // TODO: this will need to change when we have co-located files and the columns
    // are different files.
    if (!col_chunk.file_path.empty()) {
      DCHECK_EQ(col_chunk.file_path, string(metadata_range_->file()));
    }

    DiskIoMgr::ScanRange* col_range = scan_node_->AllocateScanRange(
        metadata_range_->file(), col_len, col_start, file_col_idx,
        metadata_range_->disk_id(), metadata_range_->try_cache());
    col_ranges.push_back(col_range);

    // Get the stream that will be used for this column
    ScannerContext::Stream* stream = context_->AddStream(col_range);
    DCHECK(stream != NULL);

    RETURN_IF_ERROR(column_readers_[i]->Reset(&schema_element,
        &col_chunk.meta_data, stream));

    if (!scan_node_->materialized_slots()[i]->type().IsStringType() ||
        col_chunk.meta_data.codec != parquet::CompressionCodec::UNCOMPRESSED) {
      // Non-string types are always compact.  Compressed columns don't reference data
      // in the io buffers after tuple materialization.  In both cases, we can set compact
      // to true and recycle buffers more promptly.
      stream->set_contains_tuple_data(false);
    }
  }
  DCHECK_EQ(col_ranges.size(), column_readers_.size());
  DCHECK_GE(scan_node_->materialized_slots().size(), column_readers_.size());

  // The super class stream is not longer valid/used.  It was used
  // just to parse the file header.
  stream_ = NULL;

  // Issue all the column chunks to the io mgr and have them scheduled immediately.
  // This means these ranges aren't returned via DiskIoMgr::GetNextRange and
  // instead are scheduled to be read immediately.
  RETURN_IF_ERROR(scan_node_->runtime_state()->io_mgr()->AddScanRanges(
      scan_node_->reader_context(), col_ranges, true));

  return Status::OK;
}

HdfsParquetScanner::FileVersion::FileVersion(const string& created_by) {
  string created_by_lower = created_by;
  std::transform(created_by_lower.begin(), created_by_lower.end(),
      created_by_lower.begin(), ::tolower);
  is_impala_internal = false;

  vector<string> tokens;
  split(tokens, created_by_lower, is_any_of(" "), token_compress_on);
  // Boost always creates at least one token
  DCHECK_GT(tokens.size(), 0);
  application = tokens[0];

  if (tokens.size() >= 3 && tokens[1] == "version") {
    string version_string = tokens[2];
    // Ignore any trailing extra characters
    int n = version_string.find_first_not_of("0123456789.");
    string version_string_trimmed = version_string.substr(0, n);

    vector<string> version_tokens;
    split(version_tokens, version_string_trimmed, is_any_of("."));
    version.major = version_tokens.size() >= 1 ? atoi(version_tokens[0].c_str()) : 0;
    version.minor = version_tokens.size() >= 2 ? atoi(version_tokens[1].c_str()) : 0;
    version.patch = version_tokens.size() >= 3 ? atoi(version_tokens[2].c_str()) : 0;

    if (application == "impala") {
      if (version_string.find("-internal") != string::npos) is_impala_internal = true;
    }
  } else {
    version.major = 0;
    version.minor = 0;
    version.patch = 0;
  }
}

bool HdfsParquetScanner::FileVersion::VersionLt(int major, int minor, int patch) const {
  if (version.major < major) return true;
  if (version.major > major) return false;
  DCHECK_EQ(version.major, major);
  if (version.minor < minor) return true;
  if (version.minor > minor) return false;
  DCHECK_EQ(version.minor, minor);
  return version.patch < patch;
}

bool HdfsParquetScanner::FileVersion::VersionEq(int major, int minor, int patch) const {
  return version.major == major && version.minor == minor && version.patch == patch;
}

Status HdfsParquetScanner::ValidateFileMetadata() {
  if (file_metadata_.version > PARQUET_CURRENT_VERSION) {
    stringstream ss;
    ss << "File: " << stream_->filename() << " is of an unsupported version. "
       << "file version: " << file_metadata_.version;
    return Status(ss.str());
  }

  // Parse out the created by application version string
  if (file_metadata_.__isset.created_by) {
    file_version_ = FileVersion(file_metadata_.created_by);
  }
  return Status::OK;
}

bool IsEncodingSupported(parquet::Encoding::type e) {
  switch (e) {
    case parquet::Encoding::PLAIN:
    case parquet::Encoding::PLAIN_DICTIONARY:
    case parquet::Encoding::BIT_PACKED:
    case parquet::Encoding::RLE:
      return true;
    default:
      return false;
  }
}

Status HdfsParquetScanner::ValidateColumn(const SlotDescriptor* slot_desc, int col_idx) {
  parquet::ColumnChunk& file_data = file_metadata_.row_groups[0].columns[col_idx];
  const parquet::SchemaElement& schema_element = file_metadata_.schema[col_idx + 1];

  // Check the encodings are supported
  vector<parquet::Encoding::type>& encodings = file_data.meta_data.encodings;
  for (int i = 0; i < encodings.size(); ++i) {
    if (!IsEncodingSupported(encodings[i])) {
      stringstream ss;
      ss << "File '" << stream_->filename() << "' uses an unsupported encoding: "
         << PrintEncoding(encodings[i]) << " for column '" << schema_element.name
         << "'.";
      return Status(ss.str());
    }
  }

  // Check the compression is supported
  if (file_data.meta_data.codec != parquet::CompressionCodec::UNCOMPRESSED &&
      file_data.meta_data.codec != parquet::CompressionCodec::SNAPPY &&
      file_data.meta_data.codec != parquet::CompressionCodec::GZIP) {
    stringstream ss;
    ss << "File '" << stream_->filename() << "' uses an unsupported compression: "
        << file_data.meta_data.codec << " for column '" << schema_element.name
        << "'.";
    return Status(ss.str());
  }

  // Check the type in the file is compatible with the catalog metadata.
  parquet::Type::type type = IMPALA_TO_PARQUET_TYPES[slot_desc->type().type];
  if (type != file_data.meta_data.type) {
    stringstream ss;
    ss << "File '" << stream_->filename() << "' has an incompatible type with the"
       << " table schema for column '" << schema_element.name << "'.  Expected type: "
       << type << ".  Actual type: " << file_data.meta_data.type;
    return Status(ss.str());
  }

  // Check that the column is not nested
  const vector<string> schema_path = file_data.meta_data.path_in_schema;
  if (schema_path.size() != 1) {
    stringstream ss;
    ss << "File '" << stream_->filename() << "' contains a nested schema for column '"
       << schema_element.name << "'.  This is currently not supported.";
    return Status(ss.str());
  }

  // Check that this column is optional or required
  if (schema_element.repetition_type != parquet::FieldRepetitionType::OPTIONAL &&
      schema_element.repetition_type != parquet::FieldRepetitionType::REQUIRED) {
    stringstream ss;
    ss << "File '" << stream_->filename() << "' column '" << schema_element.name
       << "' contains an unsupported column repetition type: "
       << schema_element.repetition_type;
    return Status(ss.str());
  }

  // Check the decimal scale in the file matches the metastore scale and precision.
  // We fail the query if the metadata makes it impossible for us to safely read
  // the file. If we don't require the metadata, we will fail the query if
  // abort_on_error is true, otherwise we will just log a warning.
  bool is_converted_type_decimal = schema_element.__isset.converted_type &&
      schema_element.converted_type == parquet::ConvertedType::DECIMAL;
  if (slot_desc->type().type == TYPE_DECIMAL) {
    // We require that the scale and byte length be set.
    if (schema_element.type != parquet::Type::FIXED_LEN_BYTE_ARRAY) {
      stringstream ss;
      ss << "File '" << stream_->filename() << "' column '" << schema_element.name
         << "' should be a decimal column encoded using FIXED_LEN_BYTE_ARRAY.";
      return Status(ss.str());
    }

    if (!schema_element.__isset.type_length) {
      stringstream ss;
      ss << "File '" << stream_->filename() << "' column '" << schema_element.name
         << "' does not have type_length set.";
      return Status(ss.str());
    }

    int expected_len = ParquetPlainEncoder::DecimalSize(slot_desc->type());
    if (schema_element.type_length != expected_len) {
      stringstream ss;
      ss << "File '" << stream_->filename() << "' column '" << schema_element.name
         << "' has an invalid type length. Expecting: " << expected_len
         << " len in file: " << schema_element.type_length;
      return Status(ss.str());
    }

    if (!schema_element.__isset.scale) {
      stringstream ss;
      ss << "File '" << stream_->filename() << "' column '" << schema_element.name
         << "' does not have the scale set.";
      return Status(ss.str());
    }

    if (schema_element.scale != slot_desc->type().scale) {
      // TODO: we could allow a mismatch and do a conversion at this step.
      stringstream ss;
      ss << "File '" << stream_->filename() << "' column '" << schema_element.name
         << "' has a scale that does not match the table metadata scale."
         << " File metadata scale: " << schema_element.scale
         << " Table metadata scale: " << slot_desc->type().scale;
      return Status(ss.str());
    }

    // The other decimal metadata should be there but we don't need it.
    if (!schema_element.__isset.precision) {
      stringstream ss;
      ss << "File '" << stream_->filename() << "' column '" << schema_element.name
         << "' does not have the precision set.";
      if (state_->abort_on_error()) return Status(ss.str());
      state_->LogError(ss.str());
    } else {
      if (schema_element.precision != slot_desc->type().precision) {
        // TODO: we could allow a mismatch and do a conversion at this step.
        stringstream ss;
        ss << "File '" << stream_->filename() << "' column '" << schema_element.name
          << "' has a precision that does not match the table metadata precision."
          << " File metadata precision: " << schema_element.precision
          << " Table metadata precision: " << slot_desc->type().precision;
        if (state_->abort_on_error()) return Status(ss.str());
        state_->LogError(ss.str());
      }
    }

    if (!is_converted_type_decimal) {
      // TODO: is this validation useful? It is not required at all to read the data and
      // might only serve to reject otherwise perfectly readable files.
      stringstream ss;
      ss << "File '" << stream_->filename() << "' column '" << schema_element.name
         << "' does not have converted type set to DECIMAL.";
      if (state_->abort_on_error()) return Status(ss.str());
      state_->LogError(ss.str());
    }
  } else if (schema_element.__isset.scale || schema_element.__isset.precision ||
      is_converted_type_decimal) {
    stringstream ss;
    ss << "File '" << stream_->filename() << "' column '" << schema_element.name
       << "' contains decimal data but the table metadata has type " << slot_desc->type();
    if (state_->abort_on_error()) return Status(ss.str());
    state_->LogError(ss.str());
  }

  return Status::OK;
}

max_stars_repo_name: AtScaleInc/Impala

max_issues_repo_name: AtScaleInc/Impala

max_forks_repo_name: AtScaleInc/Impala


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

This software is licensed under the Apache 2 license, quoted below.

Copyright (C) 2013 Andrey Budnik <budnik27@gmail.com>

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.

===========================================================================
*/

#define BOOST_SPIRIT_THREADSAFE

#include <sstream>
#include <boost/property_tree/json_parser.hpp>
#include <boost/filesystem.hpp>
#include <boost/bind.hpp>
#include "admin.h"
#include "user_command.h"
#include "job_manager.h"
#include "worker_manager.h"
#include "common/crutches.h"
#include "common/service_locator.h"

namespace master {

int AdminCommand_Run::Execute( const boost::property_tree::ptree &params,
                               std::string &result )
{
    std::string filePath, jobAlias;
    try
    {
        filePath = params.get<std::string>( "file" );
        if ( filePath[0] != '/' )
        {
            IJobManager *jobManager = common::GetService< IJobManager >();
            filePath = jobManager->GetJobsDir() + '/' + filePath;
        }

        if ( params.count( "alias" ) > 0 )
        {
            jobAlias = params.get<std::string>( "alias" );
        }
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_Run::Execute: " << e.what() );
        return JSON_RPC_INVALID_PARAMS;
    }

    if ( !UserCommand().Run( filePath, jobAlias, result ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_Stop::Execute( const boost::property_tree::ptree &params,
                                std::string &result )
{
    std::string jobName;
    int64_t jobId = -1;
    try
    {
        std::string value = params.get<std::string>( "job_id" );
        if ( !value.empty() && isalpha( value[0] ) )
        {
            jobName = value;
        }
        else
        {
            jobId = params.get<int64_t>( "job_id" );
        }
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_Stop::Execute: " << e.what() );
        return JSON_RPC_INVALID_PARAMS;
    }

    if ( jobName.empty() )
    {
        if ( !UserCommand().Stop( jobId ) )
            return JSON_RPC_INTERNAL_ERROR;
    }
    else
    {
        if ( !UserCommand().StopNamed( jobName ) )
            return JSON_RPC_INTERNAL_ERROR;
    }

    return 0;
}

int AdminCommand_StopGroup::Execute( const boost::property_tree::ptree &params,
                                     std::string &result )
{
    int64_t groupId;
    try
    {
        groupId = params.get<int64_t>( "group_id" );
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_StopGroup::Execute: " << e.what() );
        return JSON_RPC_INVALID_PARAMS;
    }

    if ( !UserCommand().StopGroup( groupId ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_StopAll::Execute( const boost::property_tree::ptree &params,
                                   std::string &result )
{
    if ( !UserCommand().StopAll() )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_AddHosts::Execute( const boost::property_tree::ptree &params,
                                    std::string &result )
{
    try
    {
        int i = 0;
        std::string groupName, host;
        for( const auto &v : params.get_child( "hosts" ) )
        {
            if ( i++ % 2 > 0 )
            {
                groupName = v.second.get_value< std::string >();
                UserCommand().AddWorkerHost( groupName, host );
            }
            else
                host = v.second.get_value< std::string >();
        }
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_AddHosts::Execute: " << e.what() );
        return JSON_RPC_INTERNAL_ERROR;
    }
    return 0;
}

int AdminCommand_DeleteHosts::Execute( const boost::property_tree::ptree &params,
                                       std::string &result )
{
    try
    {;
        std::string host;
        for( const auto &v : params.get_child( "hosts" ) )
        {
            host = v.second.get_value< std::string >();
            UserCommand().DeleteHost( host );
        }
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_DeleteHosts::Execute: " << e.what() );
        return JSON_RPC_INTERNAL_ERROR;
    }
    return 0;
}

int AdminCommand_AddGroup::Execute( const boost::property_tree::ptree &params,
                                    std::string &result )
{
    std::string filePath, fileName;
    try
    {
        filePath = params.get<std::string>( "file" );
        if ( filePath[0] != '/' )
        {
            IWorkerManager *workerManager = common::GetService< IWorkerManager >();
            filePath = workerManager->GetConfigDir() + '/' + filePath;
        }

        boost::filesystem::path p( filePath );
        fileName = common::PathToString( p.filename() );
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_AddGroup::Execute: " << e.what() );
        return JSON_RPC_INVALID_PARAMS;
    }

    if ( !UserCommand().AddGroup( filePath, fileName ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_DeleteGroup::Execute( const boost::property_tree::ptree &params,
                                       std::string &result )
{
    std::string group;
    try
    {
        group = params.get<std::string>( "group" );
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_DeleteGroup::Execute: " << e.what() );
        return JSON_RPC_INVALID_PARAMS;
    }

    if ( !UserCommand().DeleteGroup( group ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_Info::Execute( const boost::property_tree::ptree &params,
                                std::string &result )
{
    int64_t jobId;
    try
    {
        jobId = params.get<int64_t>( "job_id" );
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminCommand_Info::Execute: " << e.what() );
        return JSON_RPC_INVALID_PARAMS;
    }

    if ( !UserCommand().Info( jobId, result ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_Stat::Execute( const boost::property_tree::ptree &params,
                                std::string &result )
{
    if ( !UserCommand().GetStatistics( result ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_Jobs::Execute( const boost::property_tree::ptree &params,
                                std::string &result )
{
    if ( !UserCommand().GetAllJobInfo( result ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_Ls::Execute( const boost::property_tree::ptree &params,
                              std::string &result )
{
    if ( !UserCommand().GetWorkersStatistics( result ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

int AdminCommand_Cron::Execute( const boost::property_tree::ptree &params,
                                std::string &result )
{
    if ( !UserCommand().GetCronInfo( result ) )
        return JSON_RPC_INTERNAL_ERROR;

    return 0;
}

void AdminSession::Start()
{
    remoteIP_ = socket_.remote_endpoint().address().to_string();

    PLOG( "AdminSession::Start: ip=" << remoteIP_ );

    socket_.async_read_some( boost::asio::buffer( buffer_ ),
                             boost::bind( &AdminSession::HandleRead, shared_from_this(),
                                          boost::asio::placeholders::error,
                                          boost::asio::placeholders::bytes_transferred ) );
}

void AdminSession::HandleRead( const boost::system::error_code &error, size_t bytes_transferred )
{
    if ( !error )
    {
        request_.append( buffer_.begin(), buffer_.begin() + bytes_transferred );

        if ( !common::JsonRpc::ValidateJsonBraces( request_ ) )
        {
            socket_.async_read_some( boost::asio::buffer( buffer_ ),
                                     boost::bind( &AdminSession::HandleRead, shared_from_this(),
                                                  boost::asio::placeholders::error,
                                                  boost::asio::placeholders::bytes_transferred ) );
        }
        else
        {
            HandleRequest();

            // read next command
            request_.clear();
            socket_.async_read_some( boost::asio::buffer( buffer_ ),
                                     boost::bind( &AdminSession::HandleRead, shared_from_this(),
                                                  boost::asio::placeholders::error,
                                                  boost::asio::placeholders::bytes_transferred ) );
        }
    }
    else
    {
        PLOG_WRN( "AdminSession::HandleRead error=" << error.message() );
    }
}

void AdminSession::HandleWrite( const boost::system::error_code& error, size_t bytes_transferred )
{
    if ( error )
    {
        PLOG_ERR( "AdminSession::HandleWrite error=" << error.message() );
    }
}

void AdminSession::HandleRequest()
{
    PLOG( request_ );
    std::string requestId, result;
    int errCode = requestHandler_.HandleRequest( request_, requestId, result );

    try
    {
        boost::property_tree::ptree ptree;
        ptree.put( "jsonrpc", common::JsonRpc::GetProtocolVersion() );
        ptree.put( "id", requestId );

        if ( errCode )
        {
            std::string description;
            requestHandler_.GetErrorDescription( errCode, description );
            boost::property_tree::ptree perror;
            perror.put( "code", errCode );
            perror.put( "message", description );
            ptree.add_child( "error", perror );
        }
        else
        {
            ptree.put( "result", result );
        }

        std::ostringstream ss;
        boost::property_tree::write_json( ss, ptree, false );
        response_ = ss.str();

        boost::asio::async_write( socket_,
                                  boost::asio::buffer( response_ ),
                                  boost::bind( &AdminSession::HandleWrite, shared_from_this(),
                                               boost::asio::placeholders::error,
                                               boost::asio::placeholders::bytes_transferred ) );
    }
    catch( std::exception &e )
    {
        PLOG_ERR( "AdminSession::HandleRequest: " << e.what() );
    }
}

void AdminConnection::InitializeRpcHandlers()
{
    common::JsonRpc &rpc = requestHandler_;
    rpc.RegisterHandler( "run",          new AdminCommand_Run );
    rpc.RegisterHandler( "stop",         new AdminCommand_Stop );
    rpc.RegisterHandler( "stop_group",   new AdminCommand_StopGroup );
    rpc.RegisterHandler( "stop_all",     new AdminCommand_StopAll );
    rpc.RegisterHandler( "add_hosts",    new AdminCommand_AddHosts );
    rpc.RegisterHandler( "delete_hosts", new AdminCommand_DeleteHosts );
    rpc.RegisterHandler( "add_group",    new AdminCommand_AddGroup );
    rpc.RegisterHandler( "delete_group", new AdminCommand_DeleteGroup );
    rpc.RegisterHandler( "info",         new AdminCommand_Info );
    rpc.RegisterHandler( "stat",         new AdminCommand_Stat );
    rpc.RegisterHandler( "jobs",         new AdminCommand_Jobs );
    rpc.RegisterHandler( "ls",           new AdminCommand_Ls );
    rpc.RegisterHandler( "cron",         new AdminCommand_Cron );
}

void AdminConnection::StartAccept()
{
    session_ptr session( new AdminSession( io_service_, requestHandler_ ) );
    acceptor_.async_accept( session->GetSocket(),
                            boost::bind( &AdminConnection::HandleAccept, this,
                                         session, boost::asio::placeholders::error ) );
}

void AdminConnection::HandleAccept( session_ptr session, const boost::system::error_code &error )
{
    if ( !error )
    {
        io_service_.post( boost::bind( &AdminSession::Start, session ) );
        StartAccept();
    }
    else
    {
        PLOG_ERR( "AdminConnection::HandleAccept: " << error.message() );
    }
}

} // namespace master

ext: cpp

max_stars_repo_name: abudnik/prun

max_issues_repo_name: abudnik/prun

max_forks_repo_name: abudnik/prun


text:
/*  _______________________________________________________________________

    DAKOTA: Design Analysis Kit for Optimization and Terascale Applications
    Copyright 2014 Sandia Corporation.
    This software is distributed under the GNU Lesser General Public License.
    For more information, see the README file in the top Dakota directory.
    _______________________________________________________________________ */

//- Class:       SharedResponseData
//- Description: Encapsulation of data that can be shared among Response
//-              instances.
//- Owner:       Mike Eldred
//- Version: $Id: SharedResponseData.hpp 6784 2010-05-18 21:10:35Z mseldre $

#ifndef SHARED_RESPONSE_DATA_H
#define SHARED_RESPONSE_DATA_H

#include "dakota_data_types.hpp"
#include "dakota_global_defs.hpp"
#include "DataResponses.hpp"
#include <boost/shared_ptr.hpp>
#include <boost/serialization/access.hpp>
#include <boost/serialization/tracking.hpp>

namespace Dakota {

// forward declarations
class ProblemDescDB;

/// The representation of a SharedResponseData instance.  This representation,
/// or body, may be shared by multiple SharedResponseData handle instances.

/** The SharedResponseData/SharedResponseDataRep pairs utilize a
    handle-body idiom (Coplien, Advanced C++). */

class SharedResponseDataRep
{
  //
  //- Heading: Friends
  //

  friend class SharedResponseData;

  /// allow boost access to serialize this class
  friend class boost::serialization::access;

public:

  /// destructor must be public for shared_ptr
  ~SharedResponseDataRep();

private:

  //
  //- Heading: Constructor and destructor
  //

  /// default constructor
  SharedResponseDataRep();
  /// standard constructor
  SharedResponseDataRep(const ProblemDescDB& problem_db);
  /// alternate on-the-fly constructor
  SharedResponseDataRep(const ActiveSet& set);

  //
  //- Heading: Member functions
  //

  /// copy the data from srd_rep to the current representation
  void copy_rep(SharedResponseDataRep* srd_rep);

  /// serialize the core shared response data: write and read are
  /// symmetric for this class
  template<class Archive>
  void serialize(Archive& ar, const unsigned int version);

  /// experimental operator== for use in unit testing
  bool operator==(const SharedResponseDataRep& other);

  /// build/update the unrolled field labels based on fieldLabels and
  /// group lengths
  void build_field_labels();

  //
  //- Heading: Data
  //

  /// enumeration of BASE_RESPONSE, SIMULATION_RESPONSE, or EXPERIMENT_RESPONSE
  short responseType;

  /// data set type for primary response: generic, objective, calibration
  short primaryFnType;

  /// response identifier string from the input file
  String responsesId;

  /// fine-grained (unrolled) set of response function identifiers
  /// used to improve output readability; length Response::functionValues
  StringArray functionLabels;

  /// labels for each field group
  StringArray fieldLabels;

  /// number of scalar responses
  size_t numScalarResponses;
  /// index of field lengths for field data
  IntVector fieldRespGroupLengths;

  /// dimensions of each function
  IntVector numCoordsPerField;

};

inline SharedResponseDataRep::SharedResponseDataRep():
  // copy and reshape immediately apply copy_rep, so don't initialize anything:
  //:responseType(BASE_RESPONSE)//, // overridden in derived class ctors
  primaryFnType(GENERIC_FNS)
  //responsesId("NO_SPECIFICATION"), numScalarResponses(0),
{
#ifdef REFCOUNT_DEBUG
  Cout << "SharedResponseDataRep::SharedResponseDataRep() called to build "
       << "empty body object." << std::endl;
#endif
}

inline SharedResponseDataRep::~SharedResponseDataRep()
{ }

/// Container class encapsulating variables data that can be shared
/// among a set of Response instances.

/** An array of Response objects (e.g., Analyzer::allResponse) contains
    repeated configuration data (id's, labels, counts).  SharedResponseData
    employs a handle-body idiom to allow this shared data to be managed
    in a single object with many references to it, one per Response
    object in the array.  This allows scaling to larger sample sets. */

class SharedResponseData
{
public:

  //
  //- Heading: Friends
  //

  /// allow boost access to serialize this class
  friend class boost::serialization::access;

  //
  //- Heading: Constructors, destructor, and operators
  //

  /// default constructor
  SharedResponseData();
  /// standard constructor
  explicit SharedResponseData(const ProblemDescDB& problem_db);
  /// alternate on-the-fly constructor (explicit disallows its use for implicit
  /// type conversion so that instantiations of Response(set) are invalid)
  explicit SharedResponseData(const ActiveSet& set);
  /// copy constructor
  SharedResponseData(const SharedResponseData& srd);
  /// destructor
  ~SharedResponseData();

  /// assignment operator
  SharedResponseData& operator=(const SharedResponseData& srd);

  /// experimental operator== for use in unit testing
  bool operator==(const SharedResponseData& other);

  //
  //- Heading: member functions
  //

  /// number of scalar responses
  size_t num_scalar_responses() const;
  /// number of field response groups
  size_t num_field_response_groups() const;
  /// total number of response groups (scalars + field groups)
  size_t num_response_groups() const;
  /// total number of field functions (1-norm of fieldRespGroupLengths)
  size_t num_field_functions() const;
  /// total number of response functions (scalars + 1-norm of
  /// fieldRespGroupLengths)
  size_t num_functions() const;

  /// index of field lengths for field data
  const IntVector& field_lengths() const;
  /// set field lengths (if experiment different from simulation)
  void field_lengths(const IntVector& field_lengths);
  /// dimensions of each function
  const IntVector& num_coords_per_field() const;

  /// return a response function identifier string
  const String& function_label(size_t i) const;
  /// return the response function identifier strings
  const StringArray& function_labels() const;
  /// return the response function identifier strings
  StringArray& function_labels();
  /// set a response function identifier string
  void function_label(const String& label, size_t i);
  /// set the response function identifier strings
  void function_labels(const StringArray& labels);

  /// return the coarse (per-group) field response labels
  const StringArray& field_group_labels();
  /// set the coarse field group labels (must agree with number fields)
  void field_group_labels(const StringArray& field_labels);

  /// return the response identifier
  const String& responses_id() const;

  /// return the response type: {BASE,SIMULATION,EXPERIMENT}_RESPONSE
  short response_type() const;
  /// set the response type: {BASE,SIMULATION,EXPERIMENT}_RESPONSE
  void response_type(short type);

  /// get the primary function type (generic, objective, calibration)
  short primary_fn_type() const;
  /// set the primary function type (generic, objective, calibration)
  void primary_fn_type(short type);

  /// create a deep copy of the current object and return by value
  SharedResponseData copy() const;
  /// reshape the data, disconnecting a shared rep if necessary
  void reshape(size_t num_fns);

  /// return true if empty handle with null representation
  bool is_null() const;

  /// how many handles (including this) are sharing this
  /// representation (body); for debugging/testing only
  long reference_count() const;

private:

  /// serialize through the pointer, which requires object tracking:
  /// write and read are symmetric for this class
  template<class Archive>
  void serialize(Archive& ar, const unsigned int version);

  //
  //- Heading: Private data members
  //

  /// pointer to the body (handle-body idiom)
  boost::shared_ptr<SharedResponseDataRep> srdRep;
};

inline SharedResponseData::SharedResponseData()
{
#ifdef REFCOUNT_DEBUG
  Cout << "SharedResponseData::SRD(); default constructed handle.\n";
  Cout << "  srdRep use_count = " << srdRep.use_count() << std::endl;
#endif
}

inline SharedResponseData::SharedResponseData(const ProblemDescDB& problem_db):
  srdRep(new SharedResponseDataRep(problem_db))
{
#ifdef REFCOUNT_DEBUG
  Cout << "SharedResponseData::SRD(PDB) problem DB constructed handle.\n";
  Cout << "  srdRep use_count = " << srdRep.use_count() << std::endl;
#endif
}

inline SharedResponseData::SharedResponseData(const ActiveSet& set):
  srdRep(new SharedResponseDataRep(set))
{
#ifdef REFCOUNT_DEBUG
  Cout << "SharedResponseData::SRD(AS) ActiveSet constructed handle.\n";
  Cout << "  srdRep use_count = " << srdRep.use_count() << std::endl;
#endif
}

inline SharedResponseData::SharedResponseData(const SharedResponseData& srd)
{
#ifdef REFCOUNT_DEBUG
  Cout << "SharedResponseData::SRD(SRD) copy constructor.\n";
  Cout << "  srdRep use_count before = " << srdRep.use_count() << std::endl;
#endif
  // share the representation (body)
  srdRep = srd.srdRep;
#ifdef REFCOUNT_DEBUG
  Cout << "  srdRep use_count after  = " << srdRep.use_count() << std::endl;
#endif
}

inline SharedResponseData& SharedResponseData::
operator=(const SharedResponseData& srd)
{
#ifdef REFCOUNT_DEBUG
  Cout << "SharedResponseData::operator=.\n";
  Cout << "  srdRep use_count before = " << srdRep.use_count() << std::endl;
#endif
  // share the inbound representation (body) by copying the pointer
  srdRep = srd.srdRep;
#ifdef REFCOUNT_DEBUG
  Cout << "  srdRep use_count after  = " << srdRep.use_count() << std::endl;
#endif
  return *this;
}

inline SharedResponseData::~SharedResponseData()
{
  /* empty dtor in case we add virtual functions */
#ifdef REFCOUNT_DEBUG
  Cout << "SharedResponseData::~SRD called.\n";
  Cout << "  srdRep use_count = " << srdRep.use_count() << std::endl;
#endif
}

inline size_t SharedResponseData::num_scalar_responses() const
{ return srdRep->numScalarResponses; }

inline size_t SharedResponseData::num_field_response_groups() const
{ return srdRep->fieldRespGroupLengths.length(); }

inline size_t SharedResponseData::num_response_groups() const
{ return srdRep->numScalarResponses + srdRep->fieldRespGroupLengths.length(); }

inline size_t SharedResponseData::num_field_functions() const
{ return srdRep->fieldRespGroupLengths.normOne(); }

inline size_t SharedResponseData::num_functions() const
{ return srdRep->numScalarResponses + srdRep->fieldRespGroupLengths.normOne(); }

inline const IntVector& SharedResponseData::field_lengths() const
{ return srdRep->fieldRespGroupLengths; }

inline const IntVector& SharedResponseData::num_coords_per_field() const
{ return srdRep->numCoordsPerField; }

inline const String& SharedResponseData::responses_id() const
{ return srdRep->responsesId; }

inline short SharedResponseData::response_type() const
{ return srdRep->responseType; }

// TODO: Seems this could require disconnecting/new rep?
inline void SharedResponseData::response_type(short type)
{ srdRep->responseType = type; }

inline short SharedResponseData::primary_fn_type() const
{ return srdRep->primaryFnType; }

inline const String& SharedResponseData::function_label(size_t i) const
{ return srdRep->functionLabels[i]; }

inline const StringArray& SharedResponseData::function_labels() const
{ return srdRep->functionLabels; }

inline StringArray& SharedResponseData::function_labels()
{ return srdRep->functionLabels; }

inline void SharedResponseData::function_label(const String& label, size_t i)
{ srdRep->functionLabels[i] = label; }

inline void SharedResponseData::function_labels(const StringArray& labels)
{ srdRep->functionLabels = labels; }

inline const StringArray& SharedResponseData::field_group_labels()
{ return srdRep->fieldLabels; }

inline bool SharedResponseData::is_null() const
{ return (srdRep == NULL); }

} // namespace Dakota

// Since we may serialize this class through a temporary, force
// serialization mode and no tracking.  We allow tracking on
// SharedResponseDataRep as we want to serialize each unique pointer
// exactly once (may need to revisit this).
BOOST_CLASS_IMPLEMENTATION(Dakota::SharedResponseData,
 			   boost::serialization::object_serializable)
BOOST_CLASS_TRACKING(Dakota::SharedResponseData,
  		     boost::serialization::track_never)

#endif

ext: hpp

max_stars_repo_name: seakers/ExtUtils

max_issues_repo_name: seakers/ExtUtils

max_forks_repo_name: seakers/ExtUtils


text:
/********************************************************************
Unit tests for \ref record_remote
*********************************************************************/

#include "StdAfx.h"
#include "record_remote.h"

#include <iterator>

#include <boost/test/unit_test.hpp>
BOOST_AUTO_TEST_SUITE( test_record_remote )

	using namespace mem_engine ;

	BOOST_AUTO_TEST_CASE( set_trans )
	{
		try
		{
			record_remote rec(L"Felix.RemoteRecordFake") ;

			//	void set_trans( const wstring rich_trans ) ;
			rec.set_trans( L"here is a wstring text" ) ;
			string actual = string2string(rec.get_trans_rich()).c_str() ;
			string expected = "here is a wstring text" ;
			BOOST_CHECK_EQUAL(expected, actual) ;
			BOOST_CHECK ( rec.get_trans_plain() == L"here is a wstring text" ) ;
		}
		catch (_com_error& e)
		{
			TRACE(e.Description()) ;
			TRACE(e.ErrorMessage()) ;
			TRACE(e.Error()) ;
			CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
			BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
		}
	}
	//BOOST_AUTO_TEST_CASE( set_trans_modified_changes )
	//{
	//	try
	//	{
	//		record_remote rec(L"Felix.RemoteRecordFake") ;

	//		wstring billy = L"billy" ;
	//		rec.set_modified_by(billy) ;
	//		BOOST_CHECK_EQUAL(rec.get_modified_by(), L"billy") ;

	//		wstring me = L"Ryan Francis Ginstrom the First but Not Last" ;
	//		mem_engine::set_record_username(me) ;

	//		rec.set_source(L"crippy croppy") ;
	//		BOOST_CHECK_EQUAL(rec.get_modified_by(), L"Ryan Francis Ginstrom the First but Not Last") ;
	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}

	////	bool is_validated() const  ;
	//BOOST_AUTO_TEST_CASE( is_validated )
	//{
	//	try
	//	{
	//		record_remote rec(L"Felix.RemoteRecordFake") ;

	//		BOOST_CHECK ( rec.is_validated() == false ) ;

	//		rec.set_validated_on( ) ;
	//		BOOST_CHECK ( rec.is_validated() == true ) ;

	//		rec.set_validated_off() ;
	//		BOOST_CHECK ( rec.is_validated() == false ) ;
	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}

	////	size_t get_refcount() const ;
	//BOOST_AUTO_TEST_CASE( get_refcount )
	//{
	//	try
	//	{
	//		record_remote rec(L"Felix.RemoteRecordFake") ;

	//		BOOST_CHECK ( rec.get_refcount() == 0u ) ;

	//		rec.increment_refcount() ;

	//		BOOST_CHECK ( rec.get_refcount() == 1u ) ;

	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}

	////	set_reliability( const _bstr_t &text ) ;
	////	set_reliability( const wstring reliability ) ;
	////	set_reliability( const size_t reliability ) ;
	//BOOST_AUTO_TEST_CASE( set_reliability )
	//{
	//	try
	//	{

	//		record_remote rec(L"Felix.RemoteRecordFake") ;

	//		BOOST_CHECK ( rec.get_reliability() == 0u ) ;

	//		rec.set_reliability( 10u ) ;

	//		BOOST_CHECK_EQUAL( rec.get_reliability(), 9u) ;

	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}

	//}

	////	void translation_record_tester::set_validated( const bool validated );
	////	void translation_record_tester::set_validated( const wstring validated );
	//BOOST_AUTO_TEST_CASE( set_validated )
	//{
	//	try
	//	{

	//		record_remote rec(L"Felix.RemoteRecordFake") ;

	//		BOOST_CHECK ( rec.is_validated() == false ) ;

	//		rec.set_validated_on() ;

	//		BOOST_CHECK ( rec.is_validated() == true ) ;

	//		rec.set_validated_off() ;

	//		BOOST_CHECK ( rec.is_validated() == false ) ;
	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}

	////	void translation_record_tester::increment_reliability();
	//BOOST_AUTO_TEST_CASE( increment_reliability )
	//{
	//	try
	//	{
	//		record_remote rec(L"Felix.RemoteRecordFake") ;

	//		BOOST_CHECK ( rec.get_reliability() == 0u ) ;

	//		rec.increment_reliability() ;
	//		BOOST_CHECK ( rec.get_reliability() == 1u ) ;

	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}

	////	void translation_record_tester::reset_refcount();
	//BOOST_AUTO_TEST_CASE( reset_refcount )
	//{
	//	try
	//	{
	//		record_remote rec(L"Felix.RemoteRecordFake") ;

	//		rec.set_refcount( 10u ) ;
	//		rec.reset_refcount() ;
	//		BOOST_CHECK ( rec.get_refcount() == 0u ) ;
	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}

	//BOOST_AUTO_TEST_CASE( test_clone)
	//{
	//	try
	//	{
	//		record_pointer rec(new record_remote(L"Felix.RemoteRecordFake")) ;
	//		rec->set_trans(L"spam") ;
	//		rec->set_source(L"egg") ;

	//		record_pointer cloned = record_pointer(rec->clone()) ;

	//		BOOST_CHECK_EQUAL(rec->get_source_rich(), cloned->get_source_rich()) ;
	//		BOOST_CHECK_EQUAL(rec->get_trans_rich(), cloned->get_trans_rich()) ;
	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}

	//BOOST_AUTO_TEST_CASE( set_id)
	//{
	//	try
	//	{
	//		record_pointer rec(new record_remote(L"Felix.RemoteRecordFake")) ;
	//		BOOST_CHECK_EQUAL(rec->get_id(), 0u) ;
	//		rec->set_id(3u) ;
	//		BOOST_CHECK_EQUAL(rec->get_id(), 3u) ;
	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}
	//BOOST_AUTO_TEST_CASE( copy_id)
	//{
	//	try
	//	{
	//		record_pointer rec(new record_remote(L"Felix.RemoteRecordFake")) ;
	//		rec->set_id(3u) ;

	//		record_remote rec2(rec);
	//		BOOST_CHECK_EQUAL(rec2.get_id(), 3u) ;
	//	}
	//	catch (_com_error& e)
	//	{
	//		TRACE(e.Description()) ;
	//		TRACE(e.ErrorMessage()) ;
	//		TRACE(e.Error()) ;
	//		CStringA msg(static_cast< LPCWSTR >( e.ErrorMessage() )) ;
	//		BOOST_FAIL(static_cast< LPCSTR >( msg )) ;
	//	}
	//}
	//// copy_from_self
	//BOOST_AUTO_TEST_CASE( copy_from_self_source)
	//{
	//	record_pointer rec(new record_remote(L"Felix.RemoteRecordFake")) ;
	//	rec->set_source(L"spam") ;
	//	record_pointer other(new record_remote(L"Felix.RemoteRecordFake")) ;
	//	rec->copy_from_self(other) ;
	//	string actual = (LPCSTR)CStringA(other->get_source_rich().c_str()) ;
	//	string expected = "spam" ;
	//	BOOST_CHECK_EQUAL(expected, actual) ;
	//}

	//BOOST_AUTO_TEST_CASE( copy_from_self_modified_by)
	//{
	//	record_pointer rec(new record_remote(L"Felix.RemoteRecordFake")) ;
	//	rec->set_modified_by(L"spam") ;
	//	record_pointer other(new record_remote(L"Felix.RemoteRecordFake")) ;
	//	rec->copy_from_self(other) ;
	//	string actual = (LPCSTR)CStringA(other->get_modified_by().c_str()) ;
	//	string expected = "spam" ;
	//	BOOST_CHECK_EQUAL(expected, actual) ;
	//	BOOST_CHECK(other->get_source_rich().empty()) ;
	//}

BOOST_AUTO_TEST_SUITE_END()

max_stars_repo_name: ultimatezen/felix

max_issues_repo_name: ultimatezen/felix

max_forks_repo_name: ultimatezen/felix


text:
#ifndef _RABIN_KARP_ALGORITHM_H_
#define _RABIN_KARP_ALGORITHM_H
#include<iostream>
#include<fstream>
#include<string>
#include<vector>
#include <boost/algorithm/string.hpp>
#include<cstdlib>
std::string reading_file(std::string path)
{
    std::ifstream myfile = std::ifstream(path, std::ios::in);
    std::string line;
    std::string output;
    if(myfile.is_open())
    {
        while(std::getline(myfile, line))
        {
            output += line;
        }
    }
    else{
        std::cout << "I wasn't able to find the file in the following path: " + path << std::endl;
    }
    output += '\n';
    //std::cout << output;
    return output;
}

//I will return a counter of occurance and the index of where the word was found in the text.
std::pair<int, std::vector<std::pair<int, std::string>>> bruteForce(std::string txt, std::string word)
{
    int counter = 0;
    std::vector<std::pair<int, std::string>> indexs;
    word = boost::algorithm::to_lower_copy(word);
    //std::cout << word << '\n';
    int lenWord = word.length();
    int lenText = txt.length();
    for(int ind=0; ind < lenText; ind++)
    {
        std::string putativeMatch = boost::algorithm::to_lower_copy(txt.substr(ind, lenWord));
        //std::cout << putativeMatch << '\n';

        if(putativeMatch.compare(word) == 0)
        {
            counter +=1;
            int begin = ind - 10 > 0?ind-10: 0;
            int end = ind + lenWord + 10 < lenText?lenWord + 10  + (ind - begin): lenWord + 10;
            indexs.push_back(std::pair<int, std::string>(ind, txt.substr(begin, end)));
        }
    }
    return std::pair<int, std::vector<std::pair<int, std::string>>>(counter, indexs);
}

std::pair<int, std::vector<std::pair<int, std::string>>> rabinKarp(std::string txt, std::string word, int primeNumber)
{
    int counter = 0;
    std::vector<std::pair<int, std::string>> indexs;
    word = boost::algorithm::to_lower_copy(word);
    //I will do hashing using division method, in which I will sum throgh the characters of the string of txt and word
    int charSize = 256;//a
    int shiftBase  = 1;//a^|u|-1 mod p
    int hashWord = 0;
    int hashSubText = 0;
    //std::cout << word << '\n';
    int lenWord = word.length();
    int lenText = txt.length();
    for(int i=0; i < lenWord - 1; i++)
    {
        shiftBase = (charSize * shiftBase)%primeNumber;
    }

    std::string putativeMatch = boost::algorithm::to_lower_copy(txt.substr(0, lenWord));

    //Calculat h(|s| from txt) and h(|s| from word)
    for(int i=0; i< lenWord; i++)
    {
        hashWord = (hashWord * charSize + word[i])%primeNumber;
        hashSubText = (hashSubText * charSize + putativeMatch[i])%primeNumber;
    }
    //IF the match was at the beginning.
    if( hashWord == hashSubText)
    {
        if(putativeMatch.compare(word) == 0)
        {
            counter += 1;
            indexs.push_back(std::pair<int, std::string>(0, txt.substr(0, 10)));
        }
    }
    char lastChar;
    for(int ind=lenWord; ind < lenText; ind++)
    {
        lastChar = putativeMatch[0];
        putativeMatch = boost::algorithm::to_lower_copy(txt.substr(ind - lenWord + 1, lenWord));
        int begin = ind - lenWord - 10 > 0?ind - lenWord - 10: 0;
        int end = ind + 10 < lenText? 10 + (ind - begin): lenWord + 10 + 1;
        //std::cout << putativeMatch << '\n';
        if(ind + 1 < lenText)
        {
            //skip step
            hashSubText = (hashSubText - lastChar * shiftBase)%primeNumber;
            //append step
            hashSubText = (hashSubText * charSize + putativeMatch[lenWord -1]) %primeNumber;
            if(hashSubText < 0)
                hashSubText = hashSubText + primeNumber; // 0<x<primeNumber
        }
        //Is  h(k1)==h(k2)?
        if(hashWord == hashSubText)
        {
            //Is k1==k2?
            if(putativeMatch.compare(word) == 0)
            {
                counter += 1;
                indexs.push_back(std::pair<int, std::string>(ind - lenWord, txt.substr(begin, end)));
            }
        }
    }
    return std::pair<int, std::vector<std::pair<int, std::string>>>(counter, indexs);
}

#endif /*_RABIN_KARP_ALGORITHM_H*/

max_stars_repo_name: AlazzR/CLRS-Introduction-to-Algorithm

max_issues_repo_name: AlazzR/CLRS-Introduction-to-Algorithm

max_forks_repo_name: AlazzR/CLRS-Introduction-to-Algorithm


text:
/**
 * @author Mike Bogochow
 * @version 2.4.0, Dec 8, 2015
 *
 * @file Auctioneer.cpp
 *
 * Auctioneer class implementation
 */

#include "Auctioneer.h"
#include "../lib_auction/AuctionDefs.h"
#include "../lib_auction/DebugPrinter.h"

#include "MBUtils.h"

#include <boost/algorithm/string/predicate.hpp> // starts_with
#include <chrono>
#include <thread>

Auctioneer::Auctioneer(void)
{
  numBidders = 0;
  numTargets = 0;

  roundNumber = 0;
  numRecvdBids = 0;
  bids = nullptr;
}

Auctioneer::~Auctioneer(void)
{
  if (bids != nullptr)
    delete bids;
}

bool
Auctioneer::OnNewMail(MOOSMSG_LIST &NewMail)
{
  bool ret = AuctionMOOSApp::OnNewMail(NewMail);

  MOOSMSG_LIST::reverse_iterator p;
  for(p = NewMail.rbegin(); p != NewMail.rend(); p++)
  {
    CMOOSMsg &msg = *p;
    std::string key   = msg.GetKey();

    if (boost::starts_with(key, MVAR_BID_HEADER))
    {
      int bidder = getBidder(key);
      bids[bidder] = bidFromString(msg.GetString());
      numRecvdBids += 1;
      dp.dprintf(LVL_MIN_VERB, "Got %s mail: %s\n", key.c_str(),
          msg.GetString().c_str());
    }
  }

  return ret;
}

bool
Auctioneer::Iterate(void)
{
  bool ret = AuctionMOOSApp::Iterate();

  dp.dprintf(LVL_MAX_VERB, "roundNum <= numNodes (%lu <= %lu)?\n", roundNumber,
      numTargets);
  if (roundNumber <= numTargets)
  {
    dp.dprintf(LVL_MID_VERB, "numReceivedBids=%lu / %lu\n", numRecvdBids,
        numBidders);

    if (roundNumber == 0)
      doNotify(MVAR_BID_START, ++roundNumber);
    else if (numRecvdBids == numBidders)
    { // All bids received for round
      WinningBid winner = { 0, MAX_VERTEX, MAX_WEIGHT };
      // Calculate winner
      for (size_t i = 0; i < numBidders; i++)
      {
        if (bids[i].second < winner.bid)
        {
          winner.winner = i;
          winner.target = bids[i].first;
          winner.bid = bids[i].second;
        }
      }

      // Send winner
      doNotify(MVAR_BID_WINNER, winningBidToString(winner));
      doNotify(MVAR_BID_START, ++roundNumber);

      numRecvdBids = 0;
    }
    else
    { // Keep sending the round number in case they didn't get the message.
      doNotify(MVAR_BID_START, roundNumber);
      if (roundNumber == 1)
        doNotify(MVAR_BID_TARGETS, targets);
    }
  }
  else
  {
    // Exit pAuctioneer
    doNotify("EXITED_NORMALLY", "pAuctioneer");
    RequestQuit();
  }

  return ret;
}

bool
Auctioneer::OnStartUp(void)
{
  bool ret;

  // DebugOutput config
  int debugLevel;
  if (!m_MissionReader.GetConfigurationParam("DebugOutput", debugLevel))
    debugLevel = LVL_OFF;
  dp.setLevel((DebugLevel)debugLevel);

  ret = AuctionMOOSApp::OnStartUp();

  if (ret)
  {
    // NumBidders config
    std::string tmp;
    if (!m_MissionReader.GetConfigurationParam("NumBidders", tmp))
      ret = MissingRequiredParam("NumBidders");
    else
    {
      numBidders = boost::lexical_cast<size_t>(tmp);
      this->bids = new Bid[numBidders];
    }

    if (ret)
    {
      // Targets config
      if (!m_MissionReader.GetConfigurationParam("Targets", targets))
        ret = MissingRequiredParam("Targets");
      else
      {
        numTargets = getStringPathSize(targets);
        doNotify(MVAR_BID_TARGETS, targets);
      }

      if (ret)
        RegisterVariables();
    }
  }

  return ret;
}

bool
Auctioneer::OnConnectToServer(void)
{
  bool ret = AuctionMOOSApp::OnConnectToServer();
  RegisterVariables();
  return ret;
}

void
Auctioneer::RegisterVariables(void)
{
  for (size_t i = 0; i < numBidders; i++)
  {
    std::string key = getBidVar(i);
    dp.dprintf(LVL_MIN_VERB, "Registering %s\n", key.c_str());
    m_Comms.Register(key, 0);
  }
}

max_stars_repo_name: mbogochow/proj

max_issues_repo_name: mbogochow/proj

max_forks_repo_name: mbogochow/proj


text:
/* redis_async_test.cc
   Wolfgang Sourdeau, 11 December 2012
   Copyright (c) 2012 Datacratic.  All rights reserved.

   Test commands supported by our Redis class
*/

#define BOOST_TEST_MAIN
#define BOOST_TEST_DYN_LINK

#include <unordered_map>

#include <boost/test/unit_test.hpp>

#include "soa/service/redis.h"
#include "soa/service/testing/redis_temporary_server.h"

using namespace std;
using namespace Redis;

BOOST_AUTO_TEST_CASE( test_redis_commands )
{
    RedisTemporaryServer redis;
    AsyncConnection connection(redis);

    /* MSET is supported */
    Command mset(MSET);
    mset.addArg("testkey1"); mset.addArg(1234);
    mset.addArg("testkey2"); mset.addArg(9876);
    mset.addArg("testkey3"); mset.addArg("textvalue");

    Result result = connection.exec(mset);
    BOOST_CHECK_EQUAL(result.ok(), true);

    /* MGET is supported and that all slots requested must be returned in the
       form of an array, including empty ones, without errors */
    Command mget(MGET);
    mget.addArg("testkey1");
    mget.addArg("testkey3");
    mget.addArg("poil");

    result = connection.exec(mget);
    BOOST_CHECK_EQUAL(result.ok(), true);
    Reply reply = result.reply();
    BOOST_CHECK_EQUAL(reply.type(), ARRAY);

    Reply subreply = reply[0]; /* testkey1 */
    BOOST_CHECK_EQUAL(subreply.type(), STRING);
    BOOST_CHECK_EQUAL(subreply.asString(), "1234");
    /* FIXME: expected behaviour:
    BOOST_CHECK_EQUAL(subreply.type(), INTEGER);
    BOOST_CHECK_EQUAL(subreply.asInt(), 1234);
    */

    subreply = reply[1]; /* testkey3 */
    BOOST_CHECK_EQUAL(subreply.type(), STRING);
    BOOST_CHECK_EQUAL(string(subreply), "textvalue");

    subreply = reply[2]; /* poil */
    BOOST_CHECK_EQUAL(subreply.type(), NIL);

    /* SADD is supported and each slot is used once per key, even when
       specified multiple times */
    Command sadd(SADD);
    sadd.addArg("my-new-set");
    sadd.addArg("testkey1");
    sadd.addArg("testkey2");
    sadd.addArg("testkey2");

    result = connection.exec(sadd);
    BOOST_CHECK_EQUAL(result.ok(), true);

    /* SMEMBERS is supported and does return the values registered above */
    Command smembers(SMEMBERS);
    smembers.addArg("my-new-set");

    result = connection.exec(smembers);
    BOOST_CHECK_EQUAL(result.ok(), true);
    reply = result.reply();
    BOOST_CHECK_EQUAL(reply.type(), ARRAY);
    BOOST_CHECK_EQUAL(reply.length(), 2); /* testkey1, testkey2 */

    /* members are not specifically returned in order */
    unordered_map<string, bool> keyMap;
    subreply = reply[0];
    BOOST_CHECK_EQUAL(subreply.type(), STRING);
    keyMap[subreply.asString()] = true;
    subreply = reply[1];
    BOOST_CHECK_EQUAL(subreply.type(), STRING);
    keyMap[subreply.asString()] = true;

    BOOST_CHECK_EQUAL(keyMap["testkey1"], true);
    BOOST_CHECK_EQUAL(keyMap["testkey2"], true);

    /* SMOVE is supported, it moves a key from one set to another*/
    Command smove(SMOVE);
    smove.addArg("my-new-set");
    smove.addArg("my-destination-set");
    smove.addArg("testkey2");

    result = connection.exec(smove);
    BOOST_CHECK_EQUAL(result.ok(), true);

    /* verify that the key was properly moved */
    Command smembersNewSet(SMEMBERS);
    smembersNewSet.addArg("my-new-set");
    Command smembersDestSet(SMEMBERS);
    smembersDestSet.addArg("my-destination-set");

    result = connection.exec(smembersNewSet);
    BOOST_CHECK_EQUAL(result.ok(), true);
    reply = result.reply();
    BOOST_CHECK_EQUAL(reply.type(), ARRAY);
    BOOST_CHECK_EQUAL(reply.length(), 1);
    BOOST_CHECK_EQUAL(reply[0].type(), STRING);
    BOOST_CHECK_EQUAL(reply[0].asString(), "testkey1");

    result = connection.exec(smembersDestSet);
    BOOST_CHECK_EQUAL(result.ok(), true);
    reply = result.reply();
    BOOST_CHECK_EQUAL(reply.type(), ARRAY);
    BOOST_CHECK_EQUAL(reply.length(), 1);
    BOOST_CHECK_EQUAL(reply[0].type(), STRING);
    BOOST_CHECK_EQUAL(reply[0].asString(), "testkey2");

    /* check a SMOVE failure */
    result = connection.exec(smove);
    BOOST_CHECK_EQUAL(result.ok(), true);
    BOOST_CHECK_EQUAL(result.reply().type(), INTEGER);
    BOOST_CHECK_EQUAL(result.reply().asInt(), 0);

    /* SISMEMBER is supported, it checks for the membership of a
     * key to a set */
    Command sIsMemberTrue(SISMEMBER);
    sIsMemberTrue.addArg("my-new-set");
    sIsMemberTrue.addArg("testkey1");

    result = connection.exec(sIsMemberTrue);
    BOOST_CHECK_EQUAL(result.ok(), true);
    BOOST_CHECK_EQUAL(result.reply().type(), INTEGER);
    BOOST_CHECK_EQUAL(result.reply().asInt(), 1);

    Command sIsMemberFalse(SISMEMBER);
    sIsMemberFalse.addArg("my-new-set");
    sIsMemberFalse.addArg("testkey2");

    result = connection.exec(sIsMemberFalse);
    BOOST_CHECK_EQUAL(result.ok(), true);
    BOOST_CHECK_EQUAL(result.reply().type(), INTEGER);
    BOOST_CHECK_EQUAL(result.reply().asInt(), 0);

    /* EXISTS is supported it checks for the existance of a key */
    Command existsTrue(EXISTS);
    existsTrue.addArg("testkey2");

    result = connection.exec(existsTrue);
    BOOST_CHECK_EQUAL(result.ok(), true);
    BOOST_CHECK_EQUAL(result.reply().type(), INTEGER);
    BOOST_CHECK_EQUAL(result.reply().asInt(), 1);

    Command existsFalse(EXISTS);
    existsFalse.addArg("testkey4");

    result = connection.exec(existsFalse);
    BOOST_CHECK_EQUAL(result.ok(), true);
    BOOST_CHECK_EQUAL(result.reply().type(), INTEGER);
    BOOST_CHECK_EQUAL(result.reply().asInt(), 0);
}

ext: cc

max_stars_repo_name: etnrlz/rtbkit

max_issues_repo_name: TuanTranEngineer/rtbkit

max_forks_repo_name: TuanTranEngineer/rtbkit