//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).

#ifndef ROCKSDB_LITE

#include "rocksdb/utilities/document_db.h"

#include "rocksdb/cache.h"
#include "rocksdb/table.h"
#include "rocksdb/filter_policy.h"
#include "rocksdb/comparator.h"
#include "rocksdb/db.h"
#include "rocksdb/slice.h"
#include "rocksdb/utilities/json_document.h"
#include "util/coding.h"
#include "util/mutexlock.h"
#include "port/port.h"

namespace rocksdb {

// IMPORTANT NOTE: Secondary index column families should be very small and
// generally fit in memory. Assume that accessing secondary index column
// families is much faster than accessing primary index (data heap) column
// family. Accessing a key (i.e. checking for existance) from a column family in
// RocksDB is not much faster than accessing both key and value since they are
// kept together and loaded from storage together.

namespace {
// < 0   <=>  lhs < rhs
// == 0  <=>  lhs == rhs
// > 0   <=>  lhs == rhs
// TODO(icanadi) move this to JSONDocument?
int DocumentCompare(const JSONDocument& lhs, const JSONDocument& rhs) {
  assert(lhs.IsObject() == false && rhs.IsObject() == false &&
         lhs.type() == rhs.type());

  switch (lhs.type()) {
    case JSONDocument::kNull:
      return 0;
    case JSONDocument::kBool:
      return static_cast<int>(lhs.GetBool()) - static_cast<int>(rhs.GetBool());
    case JSONDocument::kDouble: {
      double res = lhs.GetDouble() - rhs.GetDouble();
      return res == 0.0 ? 0 : (res < 0.0 ? -1 : 1);
    }
    case JSONDocument::kInt64: {
      int64_t res = lhs.GetInt64() - rhs.GetInt64();
      return res == 0 ? 0 : (res < 0 ? -1 : 1);
    }
    case JSONDocument::kString:
      return Slice(lhs.GetString()).compare(Slice(rhs.GetString()));
    default:
      assert(false);
  }
  return 0;
}
}  // namespace

class Filter {
 public:
  // returns nullptr on parse failure
  static Filter* ParseFilter(const JSONDocument& filter);

  struct Interval {
    JSONDocument upper_bound;
    JSONDocument lower_bound;
    bool upper_inclusive;
    bool lower_inclusive;
    Interval()
        : upper_bound(),
          lower_bound(),
          upper_inclusive(false),
          lower_inclusive(false) {}
    Interval(const JSONDocument& ub, const JSONDocument& lb, bool ui, bool li)
        : upper_bound(ub),
          lower_bound(lb),
          upper_inclusive(ui),
          lower_inclusive(li) {
    }

    void UpdateUpperBound(const JSONDocument& ub, bool inclusive);
    void UpdateLowerBound(const JSONDocument& lb, bool inclusive);
  };

  bool SatisfiesFilter(const JSONDocument& document) const;
  const Interval* GetInterval(const std::string& field) const;

 private:
  explicit Filter(const JSONDocument& filter) : filter_(filter.Copy()) {
    assert(filter_.IsOwner());
  }

  // copied from the parameter
  const JSONDocument filter_;
  // constant after construction
  std::unordered_map<std::string, Interval> intervals_;
};

void Filter::Interval::UpdateUpperBound(const JSONDocument& ub,
                                        bool inclusive) {
  bool update = upper_bound.IsNull();
  if (!update) {
    int cmp = DocumentCompare(upper_bound, ub);
    update = (cmp > 0) || (cmp == 0 && !inclusive);
  }
  if (update) {
    upper_bound = ub;
    upper_inclusive = inclusive;
  }
}

void Filter::Interval::UpdateLowerBound(const JSONDocument& lb,
                                        bool inclusive) {
  bool update = lower_bound.IsNull();
  if (!update) {
    int cmp = DocumentCompare(lower_bound, lb);
    update = (cmp < 0) || (cmp == 0 && !inclusive);
  }
  if (update) {
    lower_bound = lb;
    lower_inclusive = inclusive;
  }
}

Filter* Filter::ParseFilter(const JSONDocument& filter) {
  if (filter.IsObject() == false) {
    return nullptr;
  }

  std::unique_ptr<Filter> f(new Filter(filter));

  for (const auto& items : f->filter_.Items()) {
    if (items.first.size() && items.first[0] == '$') {
      // fields starting with '$' are commands
      continue;
    }
    assert(f->intervals_.find(items.first) == f->intervals_.end());
    if (items.second.IsObject()) {
      if (items.second.Count() == 0) {
        // uhm...?
        return nullptr;
      }
      Interval interval;
      for (const auto& condition : items.second.Items()) {
        if (condition.second.IsObject() || condition.second.IsArray()) {
          // comparison operators not defined on objects. invalid array
          return nullptr;
        }
        // comparison operators:
        if (condition.first == "$gt") {
          interval.UpdateLowerBound(condition.second, false);
        } else if (condition.first == "$gte") {
          interval.UpdateLowerBound(condition.second, true);
        } else if (condition.first == "$lt") {
          interval.UpdateUpperBound(condition.second, false);
        } else if (condition.first == "$lte") {
          interval.UpdateUpperBound(condition.second, true);
        } else {
          // TODO(icanadi) more logical operators
          return nullptr;
        }
      }
      f->intervals_.insert({items.first, interval});
    } else {
      // equality
      f->intervals_.insert(
          {items.first, Interval(items.second,
                                 items.second, true, true)});
    }
  }

  return f.release();
}

const Filter::Interval* Filter::GetInterval(const std::string& field) const {
  auto itr = intervals_.find(field);
  if (itr == intervals_.end()) {
    return nullptr;
  }
  // we can do that since intervals_ is constant after construction
  return &itr->second;
}

bool Filter::SatisfiesFilter(const JSONDocument& document) const {
  for (const auto& interval : intervals_) {
    if (!document.Contains(interval.first)) {
      // doesn't have the value, doesn't satisfy the filter
      // (we don't support null queries yet)
      return false;
    }
    auto value = document[interval.first];
    if (!interval.second.upper_bound.IsNull()) {
      if (value.type() != interval.second.upper_bound.type()) {
        // no cross-type queries yet
        // TODO(icanadi) do this at least for numbers!
        return false;
      }
      int cmp = DocumentCompare(interval.second.upper_bound, value);
      if (cmp < 0 || (cmp == 0 && interval.second.upper_inclusive == false)) {
        // bigger (or equal) than upper bound
        return false;
      }
    }
    if (!interval.second.lower_bound.IsNull()) {
      if (value.type() != interval.second.lower_bound.type()) {
        // no cross-type queries yet
        return false;
      }
      int cmp = DocumentCompare(interval.second.lower_bound, value);
      if (cmp > 0 || (cmp == 0 && interval.second.lower_inclusive == false)) {
        // smaller (or equal) than the lower bound
        return false;
      }
    }
  }
  return true;
}

class Index {
 public:
  Index() = default;
  virtual ~Index() {}

  virtual const char* Name() const = 0;

  // Functions that are executed during write time
  // ---------------------------------------------
  // GetIndexKey() generates a key that will be used to index document and
  // returns the key though the second std::string* parameter
  virtual void GetIndexKey(const JSONDocument& document,
                           std::string* key) const = 0;
  // Keys generated with GetIndexKey() will be compared using this comparator.
  // It should be assumed that there will be a suffix added to the index key
  // according to IndexKey implementation
  virtual const Comparator* GetComparator() const = 0;

  // Functions that are executed during query time
  // ---------------------------------------------
  enum Direction {
    kForwards,
    kBackwards,
  };
  // Returns true if this index can provide some optimization for satisfying
  // filter. False otherwise
  virtual bool UsefulIndex(const Filter& filter) const = 0;
  // For every filter (assuming UsefulIndex()) there is a continuous interval of
  // keys in the index that satisfy the index conditions. That interval can be
  // three things:
  // * [A, B]
  // * [A, infinity>
  // * <-infinity, B]
  //
  // Query engine that uses this Index for optimization will access the interval
  // by first calling Position() and then iterating in the Direction (returned
  // by Position()) while ShouldContinueLooking() is true.
  // * For [A, B] interval Position() will Seek() to A and return kForwards.
  // ShouldContinueLooking() will be true until the iterator value gets beyond B
  // -- then it will return false
  // * For [A, infinity> Position() will Seek() to A and return kForwards.
  // ShouldContinueLooking() will always return true
  // * For <-infinity, B] Position() will Seek() to B and return kBackwards.
  // ShouldContinueLooking() will always return true (given that iterator is
  // advanced by calling Prev())
  virtual Direction Position(const Filter& filter,
                             Iterator* iterator) const = 0;
  virtual bool ShouldContinueLooking(const Filter& filter,
                                     const Slice& secondary_key,
                                     Direction direction) const = 0;

  // Static function that is executed when Index is created
  // ---------------------------------------------
  // Create Index from user-supplied description. Return nullptr on parse
  // failure.
  static Index* CreateIndexFromDescription(const JSONDocument& description,
                                           const std::string& name);

 private:
  // No copying allowed
  Index(const Index&);
  void operator=(const Index&);
};

// Encoding helper function
namespace {
std::string InternalSecondaryIndexName(const std::string& user_name) {
  return "index_" + user_name;
}

// Don't change these, they are persisted in secondary indexes
enum JSONPrimitivesEncoding : char {
  kNull = 0x1,
  kBool = 0x2,
  kDouble = 0x3,
  kInt64 = 0x4,
  kString = 0x5,
};

// encodes simple JSON members (meaning string, integer, etc)
// the end result of this will be lexicographically compared to each other
bool EncodeJSONPrimitive(const JSONDocument& json, std::string* dst) {
  // TODO(icanadi) revise this at some point, have a custom comparator
  switch (json.type()) {
    case JSONDocument::kNull:
      dst->push_back(kNull);
      break;
    case JSONDocument::kBool:
      dst->push_back(kBool);
      dst->push_back(static_cast<char>(json.GetBool()));
      break;
    case JSONDocument::kDouble:
      dst->push_back(kDouble);
      PutFixed64(dst, static_cast<uint64_t>(json.GetDouble()));
      break;
    case JSONDocument::kInt64:
      dst->push_back(kInt64);
      {
        auto val = json.GetInt64();
        dst->push_back((val < 0) ? '0' : '1');
        PutFixed64(dst, static_cast<uint64_t>(val));
      }
      break;
    case JSONDocument::kString:
      dst->push_back(kString);
      dst->append(json.GetString());
      break;
    default:
      return false;
  }
  return true;
}

}  // namespace

// format of the secondary key is:
// <secondary_key><primary_key><offset_of_primary_key uint32_t>
class IndexKey {
 public:
  IndexKey() : ok_(false) {}
  explicit IndexKey(const Slice& slice) {
    if (slice.size() < sizeof(uint32_t)) {
      ok_ = false;
      return;
    }
    uint32_t primary_key_offset =
        DecodeFixed32(slice.data() + slice.size() - sizeof(uint32_t));
    if (primary_key_offset >= slice.size() - sizeof(uint32_t)) {
      ok_ = false;
      return;
    }
    parts_[0] = Slice(slice.data(), primary_key_offset);
    parts_[1] = Slice(slice.data() + primary_key_offset,
                      slice.size() - primary_key_offset - sizeof(uint32_t));
    ok_ = true;
  }
  IndexKey(const Slice& secondary_key, const Slice& primary_key) : ok_(true) {
    parts_[0] = secondary_key;
    parts_[1] = primary_key;
  }

  SliceParts GetSliceParts() {
    uint32_t primary_key_offset = static_cast<uint32_t>(parts_[0].size());
    EncodeFixed32(primary_key_offset_buf_, primary_key_offset);
    parts_[2] = Slice(primary_key_offset_buf_, sizeof(uint32_t));
    return SliceParts(parts_, 3);
  }

  const Slice& GetPrimaryKey() const { return parts_[1]; }
  const Slice& GetSecondaryKey() const { return parts_[0]; }

  bool ok() const { return ok_; }

 private:
  bool ok_;
  // 0 -- secondary key
  // 1 -- primary key
  // 2 -- primary key offset
  Slice parts_[3];
  char primary_key_offset_buf_[sizeof(uint32_t)];
};

class SimpleSortedIndex : public Index {
 public:
  SimpleSortedIndex(const std::string& field, const std::string& name)
      : field_(field), name_(name) {}

  virtual const char* Name() const override { return name_.c_str(); }

  virtual void GetIndexKey(const JSONDocument& document, std::string* key) const
      override {
    if (!document.Contains(field_)) {
      if (!EncodeJSONPrimitive(JSONDocument(JSONDocument::kNull), key)) {
        assert(false);
      }
    } else {
      if (!EncodeJSONPrimitive(document[field_], key)) {
        assert(false);
      }
    }
  }
  virtual const Comparator* GetComparator() const override {
    return BytewiseComparator();
  }

  virtual bool UsefulIndex(const Filter& filter) const override {
    return filter.GetInterval(field_) != nullptr;
  }
  // REQUIRES: UsefulIndex(filter) == true
  virtual Direction Position(const Filter& filter,
                             Iterator* iterator) const override {
    auto interval = filter.GetInterval(field_);
    assert(interval != nullptr);  // because index is useful
    Direction direction;

    const JSONDocument* limit;
    if (!interval->lower_bound.IsNull()) {
      limit = &(interval->lower_bound);
      direction = kForwards;
    } else {
      limit = &(interval->upper_bound);
      direction = kBackwards;
    }

    std::string encoded_limit;
    if (!EncodeJSONPrimitive(*limit, &encoded_limit)) {
      assert(false);
    }
    iterator->Seek(Slice(encoded_limit));

    return direction;
  }
  // REQUIRES: UsefulIndex(filter) == true
  virtual bool ShouldContinueLooking(
      const Filter& filter, const Slice& secondary_key,
      Index::Direction direction) const override {
    auto interval = filter.GetInterval(field_);
    assert(interval != nullptr);  // because index is useful
    if (direction == kForwards) {
      if (interval->upper_bound.IsNull()) {
        // continue looking, no upper bound
        return true;
      }
      std::string encoded_upper_bound;
      if (!EncodeJSONPrimitive(interval->upper_bound, &encoded_upper_bound)) {
        // uhm...?
        // TODO(icanadi) store encoded upper and lower bounds in Filter*?
        assert(false);
      }
      // TODO(icanadi) we need to somehow decode this and use DocumentCompare()
      int compare = secondary_key.compare(Slice(encoded_upper_bound));
      // if (current key is bigger than upper bound) OR (current key is equal to
      // upper bound, but inclusive is false) THEN stop looking. otherwise,
      // continue
      return (compare > 0 ||
              (compare == 0 && interval->upper_inclusive == false))
                 ? false
                 : true;
    } else {
      assert(direction == kBackwards);
      if (interval->lower_bound.IsNull()) {
        // continue looking, no lower bound
        return true;
      }
      std::string encoded_lower_bound;
      if (!EncodeJSONPrimitive(interval->lower_bound, &encoded_lower_bound)) {
        // uhm...?
        // TODO(icanadi) store encoded upper and lower bounds in Filter*?
        assert(false);
      }
      // TODO(icanadi) we need to somehow decode this and use DocumentCompare()
      int compare = secondary_key.compare(Slice(encoded_lower_bound));
      // if (current key is smaller than lower bound) OR (current key is equal
      // to lower bound, but inclusive is false) THEN stop looking. otherwise,
      // continue
      return (compare < 0 ||
              (compare == 0 && interval->lower_inclusive == false))
                 ? false
                 : true;
    }

    assert(false);
    // this is here just so compiler doesn't complain
    return false;
  }

 private:
  std::string field_;
  std::string name_;
};

Index* Index::CreateIndexFromDescription(const JSONDocument& description,
                                         const std::string& name) {
  if (!description.IsObject() || description.Count() != 1) {
    // not supported yet
    return nullptr;
  }
  const auto& field = *description.Items().begin();
  if (field.second.IsInt64() == false || field.second.GetInt64() != 1) {
    // not supported yet
    return nullptr;
  }
  return new SimpleSortedIndex(field.first, name);
}

class CursorWithFilterIndexed : public Cursor {
 public:
  CursorWithFilterIndexed(Iterator* primary_index_iter,
                          Iterator* secondary_index_iter, const Index* index,
                          const Filter* filter)
      : primary_index_iter_(primary_index_iter),
        secondary_index_iter_(secondary_index_iter),
        index_(index),
        filter_(filter),
        valid_(true),
        current_json_document_(nullptr) {
    assert(filter_.get() != nullptr);
    direction_ = index->Position(*filter_.get(), secondary_index_iter_.get());
    UpdateIndexKey();
    AdvanceUntilSatisfies();
  }

  virtual bool Valid() const override {
    return valid_ && secondary_index_iter_->Valid();
  }
  virtual void Next() override {
    assert(Valid());
    Advance();
    AdvanceUntilSatisfies();
  }
  // temporary object. copy it if you want to use it
  virtual const JSONDocument& document() const override {
    assert(Valid());
    return *current_json_document_;
  }
  virtual Status status() const override {
    if (!status_.ok()) {
      return status_;
    }
    if (!primary_index_iter_->status().ok()) {
      return primary_index_iter_->status();
    }
    return secondary_index_iter_->status();
  }

 private:
  void Advance() {
    if (direction_ == Index::kForwards) {
      secondary_index_iter_->Next();
    } else {
      secondary_index_iter_->Prev();
    }
    UpdateIndexKey();
  }
  void AdvanceUntilSatisfies() {
    bool found = false;
    while (secondary_index_iter_->Valid() &&
           index_->ShouldContinueLooking(
               *filter_.get(), index_key_.GetSecondaryKey(), direction_)) {
      if (!UpdateJSONDocument()) {
        // corruption happened
        return;
      }
      if (filter_->SatisfiesFilter(*current_json_document_)) {
        // we found satisfied!
        found = true;
        break;
      } else {
        // doesn't satisfy :(
        Advance();
      }
    }
    if (!found) {
      valid_ = false;
    }
  }

  bool UpdateJSONDocument() {
    assert(secondary_index_iter_->Valid());
    primary_index_iter_->Seek(index_key_.GetPrimaryKey());
    if (!primary_index_iter_->Valid()) {
      status_ = Status::Corruption(
          "Inconsistency between primary and secondary index");
      valid_ = false;
      return false;
    }
    current_json_document_.reset(
        JSONDocument::Deserialize(primary_index_iter_->value()));
    assert(current_json_document_->IsOwner());
    if (current_json_document_.get() == nullptr) {
      status_ = Status::Corruption("JSON deserialization failed");
      valid_ = false;
      return false;
    }
    return true;
  }
  void UpdateIndexKey() {
    if (secondary_index_iter_->Valid()) {
      index_key_ = IndexKey(secondary_index_iter_->key());
      if (!index_key_.ok()) {
        status_ = Status::Corruption("Invalid index key");
        valid_ = false;
      }
    }
  }
  std::unique_ptr<Iterator> primary_index_iter_;
  std::unique_ptr<Iterator> secondary_index_iter_;
  // we don't own index_
  const Index* index_;
  Index::Direction direction_;
  std::unique_ptr<const Filter> filter_;
  bool valid_;
  IndexKey index_key_;
  std::unique_ptr<JSONDocument> current_json_document_;
  Status status_;
};

class CursorFromIterator : public Cursor {
 public:
  explicit CursorFromIterator(Iterator* iter)
      : iter_(iter), current_json_document_(nullptr) {
    iter_->SeekToFirst();
    UpdateCurrentJSON();
  }

  virtual bool Valid() const override { return status_.ok() && iter_->Valid(); }
  virtual void Next() override {
    iter_->Next();
    UpdateCurrentJSON();
  }
  virtual const JSONDocument& document() const override {
    assert(Valid());
    return *current_json_document_;
  };
  virtual Status status() const override {
    if (!status_.ok()) {
      return status_;
    }
    return iter_->status();
  }

  // not part of public Cursor interface
  Slice key() const { return iter_->key(); }

 private:
  void UpdateCurrentJSON() {
    if (Valid()) {
      current_json_document_.reset(JSONDocument::Deserialize(iter_->value()));
      if (current_json_document_.get() == nullptr) {
        status_ = Status::Corruption("JSON deserialization failed");
      }
    }
  }

  Status status_;
  std::unique_ptr<Iterator> iter_;
  std::unique_ptr<JSONDocument> current_json_document_;
};

class CursorWithFilter : public Cursor {
 public:
  CursorWithFilter(Cursor* base_cursor, const Filter* filter)
      : base_cursor_(base_cursor), filter_(filter) {
    assert(filter_.get() != nullptr);
    SeekToNextSatisfies();
  }
  virtual bool Valid() const override { return base_cursor_->Valid(); }
  virtual void Next() override {
    assert(Valid());
    base_cursor_->Next();
    SeekToNextSatisfies();
  }
  virtual const JSONDocument& document() const override {
    assert(Valid());
    return base_cursor_->document();
  }
  virtual Status status() const override { return base_cursor_->status(); }

 private:
  void SeekToNextSatisfies() {
    for (; base_cursor_->Valid(); base_cursor_->Next()) {
      if (filter_->SatisfiesFilter(base_cursor_->document())) {
        break;
      }
    }
  }
  std::unique_ptr<Cursor> base_cursor_;
  std::unique_ptr<const Filter> filter_;
};

class CursorError : public Cursor {
 public:
  explicit CursorError(Status s) : s_(s) { assert(!s.ok()); }
  virtual Status status() const override { return s_; }
  virtual bool Valid() const override { return false; }
  virtual void Next() override {}
  virtual const JSONDocument& document() const override {
    assert(false);
    // compiler complains otherwise
    return trash_;
  }

 private:
  Status s_;
  JSONDocument trash_;
};

class DocumentDBImpl : public DocumentDB {
 public:
  DocumentDBImpl(
      DB* db, ColumnFamilyHandle* primary_key_column_family,
      const std::vector<std::pair<Index*, ColumnFamilyHandle*>>& indexes,
      const Options& rocksdb_options)
      : DocumentDB(db),
        primary_key_column_family_(primary_key_column_family),
        rocksdb_options_(rocksdb_options) {
    for (const auto& index : indexes) {
      name_to_index_.insert(
          {index.first->Name(), IndexColumnFamily(index.first, index.second)});
    }
  }

  ~DocumentDBImpl() {
    for (auto& iter : name_to_index_) {
      delete iter.second.index;
      delete iter.second.column_family;
    }
    delete primary_key_column_family_;
  }

  virtual Status CreateIndex(const WriteOptions& write_options,
                             const IndexDescriptor& index) override {
    auto index_obj =
        Index::CreateIndexFromDescription(*index.description, index.name);
    if (index_obj == nullptr) {
      return Status::InvalidArgument("Failed parsing index description");
    }

    ColumnFamilyHandle* cf_handle;
    Status s =
        CreateColumnFamily(ColumnFamilyOptions(rocksdb_options_),
                           InternalSecondaryIndexName(index.name), &cf_handle);
    if (!s.ok()) {
      delete index_obj;
      return s;
    }

    MutexLock l(&write_mutex_);

    std::unique_ptr<CursorFromIterator> cursor(new CursorFromIterator(
        DocumentDB::NewIterator(ReadOptions(), primary_key_column_family_)));

    WriteBatch batch;
    for (; cursor->Valid(); cursor->Next()) {
      std::string secondary_index_key;
      index_obj->GetIndexKey(cursor->document(), &secondary_index_key);
      IndexKey index_key(Slice(secondary_index_key), cursor->key());
      batch.Put(cf_handle, index_key.GetSliceParts(), SliceParts());
    }

    if (!cursor->status().ok()) {
      delete index_obj;
      return cursor->status();
    }

    {
      MutexLock l_nti(&name_to_index_mutex_);
      name_to_index_.insert(
          {index.name, IndexColumnFamily(index_obj, cf_handle)});
    }

    return DocumentDB::Write(write_options, &batch);
  }

  virtual Status DropIndex(const std::string& name) override {
    MutexLock l(&write_mutex_);

    auto index_iter = name_to_index_.find(name);
    if (index_iter == name_to_index_.end()) {
      return Status::InvalidArgument("No such index");
    }

    Status s = DropColumnFamily(index_iter->second.column_family);
    if (!s.ok()) {
      return s;
    }

    delete index_iter->second.index;
    delete index_iter->second.column_family;

    // remove from name_to_index_
    {
      MutexLock l_nti(&name_to_index_mutex_);
      name_to_index_.erase(index_iter);
    }

    return Status::OK();
  }

  virtual Status Insert(const WriteOptions& options,
                        const JSONDocument& document) override {
    WriteBatch batch;

    if (!document.IsObject()) {
      return Status::InvalidArgument("Document not an object");
    }
    if (!document.Contains(kPrimaryKey)) {
      return Status::InvalidArgument("No primary key");
    }
    auto primary_key = document[kPrimaryKey];
    if (primary_key.IsNull() ||
        (!primary_key.IsString() && !primary_key.IsInt64())) {
      return Status::InvalidArgument(
          "Primary key format error");
    }
    std::string encoded_document;
    document.Serialize(&encoded_document);
    std::string primary_key_encoded;
    if (!EncodeJSONPrimitive(primary_key, &primary_key_encoded)) {
      // previous call should be guaranteed to pass because of all primary_key
      // conditions checked before
      assert(false);
    }
    Slice primary_key_slice(primary_key_encoded);

    // Lock now, since we're starting DB operations
    MutexLock l(&write_mutex_);
    // check if there is already a document with the same primary key
    PinnableSlice value;
    Status s = DocumentDB::Get(ReadOptions(), primary_key_column_family_,
                               primary_key_slice, &value);
    if (!s.IsNotFound()) {
      return s.ok() ? Status::InvalidArgument("Duplicate primary key!") : s;
    }

    batch.Put(primary_key_column_family_, primary_key_slice, encoded_document);

    for (const auto& iter : name_to_index_) {
      std::string secondary_index_key;
      iter.second.index->GetIndexKey(document, &secondary_index_key);
      IndexKey index_key(Slice(secondary_index_key), primary_key_slice);
      batch.Put(iter.second.column_family, index_key.GetSliceParts(),
                SliceParts());
    }

    return DocumentDB::Write(options, &batch);
  }

  virtual Status Remove(const ReadOptions& read_options,
                        const WriteOptions& write_options,
                        const JSONDocument& query) override {
    MutexLock l(&write_mutex_);
    std::unique_ptr<Cursor> cursor(
        ConstructFilterCursor(read_options, nullptr, query));

    WriteBatch batch;
    for (; cursor->status().ok() && cursor->Valid(); cursor->Next()) {
      const auto& document = cursor->document();
      if (!document.IsObject()) {
        return Status::Corruption("Document corruption");
      }
      if (!document.Contains(kPrimaryKey)) {
        return Status::Corruption("Document corruption");
      }
      auto primary_key = document[kPrimaryKey];
      if (primary_key.IsNull() ||
          (!primary_key.IsString() && !primary_key.IsInt64())) {
        return Status::Corruption("Document corruption");
      }

      // TODO(icanadi) Instead of doing this, just get primary key encoding from
      // cursor, as it already has this information
      std::string primary_key_encoded;
      if (!EncodeJSONPrimitive(primary_key, &primary_key_encoded)) {
        // previous call should be guaranteed to pass because of all primary_key
        // conditions checked before
        assert(false);
      }
      Slice primary_key_slice(primary_key_encoded);
      batch.Delete(primary_key_column_family_, primary_key_slice);

      for (const auto& iter : name_to_index_) {
        std::string secondary_index_key;
        iter.second.index->GetIndexKey(document, &secondary_index_key);
        IndexKey index_key(Slice(secondary_index_key), primary_key_slice);
        batch.Delete(iter.second.column_family, index_key.GetSliceParts());
      }
    }

    if (!cursor->status().ok()) {
      return cursor->status();
    }

    return DocumentDB::Write(write_options, &batch);
  }

  virtual Status Update(const ReadOptions& read_options,
                        const WriteOptions& write_options,
                        const JSONDocument& filter,
                        const JSONDocument& updates) override {
    MutexLock l(&write_mutex_);
    std::unique_ptr<Cursor> cursor(
        ConstructFilterCursor(read_options, nullptr, filter));

    if (!updates.IsObject()) {
        return Status::Corruption("Bad update document format");
    }
    WriteBatch batch;
    for (; cursor->status().ok() && cursor->Valid(); cursor->Next()) {
      const auto& old_document = cursor->document();
      JSONDocument new_document(old_document);
      if (!new_document.IsObject()) {
        return Status::Corruption("Document corruption");
      }
      // TODO(icanadi) Make this nicer, something like class Filter
      for (const auto& update : updates.Items()) {
        if (update.first == "$set") {
          JSONDocumentBuilder builder;
          bool res __attribute__((unused)) = builder.WriteStartObject();
          assert(res);
          for (const auto& itr : update.second.Items()) {
            if (itr.first == kPrimaryKey) {
              return Status::NotSupported("Please don't change primary key");
            }
            res = builder.WriteKeyValue(itr.first, itr.second);
            assert(res);
          }
          res = builder.WriteEndObject();
          assert(res);
          JSONDocument update_document = builder.GetJSONDocument();
          builder.Reset();
          res = builder.WriteStartObject();
          assert(res);
          for (const auto& itr : new_document.Items()) {
            if (update_document.Contains(itr.first)) {
              res = builder.WriteKeyValue(itr.first,
                                          update_document[itr.first]);
            } else {
              res = builder.WriteKeyValue(itr.first, new_document[itr.first]);
            }
            assert(res);
          }
          res = builder.WriteEndObject();
          assert(res);
          new_document = builder.GetJSONDocument();
          assert(new_document.IsOwner());
        } else {
          // TODO(icanadi) more commands
          return Status::InvalidArgument("Can't understand update command");
        }
      }

      // TODO(icanadi) reuse some of this code
      if (!new_document.Contains(kPrimaryKey)) {
        return Status::Corruption("Corrupted document -- primary key missing");
      }
      auto primary_key = new_document[kPrimaryKey];
      if (primary_key.IsNull() ||
          (!primary_key.IsString() && !primary_key.IsInt64())) {
        // This will happen when document on storage doesn't have primary key,
        // since we don't support any update operations on primary key. That's
        // why this is corruption error
        return Status::Corruption("Corrupted document -- primary key missing");
      }
      std::string encoded_document;
      new_document.Serialize(&encoded_document);
      std::string primary_key_encoded;
      if (!EncodeJSONPrimitive(primary_key, &primary_key_encoded)) {
        // previous call should be guaranteed to pass because of all primary_key
        // conditions checked before
        assert(false);
      }
      Slice primary_key_slice(primary_key_encoded);
      batch.Put(primary_key_column_family_, primary_key_slice,
                encoded_document);

      for (const auto& iter : name_to_index_) {
        std::string old_key, new_key;
        iter.second.index->GetIndexKey(old_document, &old_key);
        iter.second.index->GetIndexKey(new_document, &new_key);
        if (old_key == new_key) {
          // don't need to update this secondary index
          continue;
        }

        IndexKey old_index_key(Slice(old_key), primary_key_slice);
        IndexKey new_index_key(Slice(new_key), primary_key_slice);

        batch.Delete(iter.second.column_family, old_index_key.GetSliceParts());
        batch.Put(iter.second.column_family, new_index_key.GetSliceParts(),
                  SliceParts());
      }
    }

    if (!cursor->status().ok()) {
      return cursor->status();
    }

    return DocumentDB::Write(write_options, &batch);
  }

  virtual Cursor* Query(const ReadOptions& read_options,
                        const JSONDocument& query) override {
    Cursor* cursor = nullptr;

    if (!query.IsArray()) {
      return new CursorError(
          Status::InvalidArgument("Query has to be an array"));
    }

    // TODO(icanadi) support index "_id"
    for (size_t i = 0; i < query.Count(); ++i) {
      const auto& command_doc = query[i];
      if (command_doc.Count() != 1) {
        // there can be only one key-value pair in each of array elements.
        // key is the command and value are the params
        delete cursor;
        return new CursorError(Status::InvalidArgument("Invalid query"));
      }
      const auto& command = *command_doc.Items().begin();

      if (command.first == "$filter") {
        cursor = ConstructFilterCursor(read_options, cursor, command.second);
      } else {
        // only filter is supported for now
        delete cursor;
        return new CursorError(Status::InvalidArgument("Invalid query"));
      }
    }

    if (cursor == nullptr) {
      cursor = new CursorFromIterator(
          DocumentDB::NewIterator(read_options, primary_key_column_family_));
    }

    return cursor;
  }

  // RocksDB functions
  using DB::Get;
  virtual Status Get(const ReadOptions& options,
                     ColumnFamilyHandle* column_family, const Slice& key,
                     PinnableSlice* value) override {
    return Status::NotSupported("");
  }
  virtual Status Get(const ReadOptions& options, const Slice& key,
                     std::string* value) override {
    return Status::NotSupported("");
  }
  virtual Status Write(const WriteOptions& options,
                       WriteBatch* updates) override {
    return Status::NotSupported("");
  }
  virtual Iterator* NewIterator(const ReadOptions& options,
                                ColumnFamilyHandle* column_family) override {
    return nullptr;
  }
  virtual Iterator* NewIterator(const ReadOptions& options) override {
    return nullptr;
  }

 private:
  Cursor* ConstructFilterCursor(ReadOptions read_options, Cursor* cursor,
                                const JSONDocument& query) {
    std::unique_ptr<const Filter> filter(Filter::ParseFilter(query));
    if (filter.get() == nullptr) {
      return new CursorError(Status::InvalidArgument("Invalid query"));
    }

    IndexColumnFamily tmp_storage(nullptr, nullptr);

    if (cursor == nullptr) {
      IndexColumnFamily* index_column_family = nullptr;
      if (query.Contains("$index") && query["$index"].IsString()) {
        {
          auto index_name = query["$index"];
          MutexLock l(&name_to_index_mutex_);
          auto index_iter = name_to_index_.find(index_name.GetString());
          if (index_iter != name_to_index_.end()) {
            tmp_storage = index_iter->second;
            index_column_family = &tmp_storage;
          } else {
            return new CursorError(
                Status::InvalidArgument("Index does not exist"));
          }
        }
      }

      if (index_column_family != nullptr &&
          index_column_family->index->UsefulIndex(*filter.get())) {
        std::vector<Iterator*> iterators;
        Status s = DocumentDB::NewIterators(
            read_options,
            {primary_key_column_family_, index_column_family->column_family},
            &iterators);
        if (!s.ok()) {
          delete cursor;
          return new CursorError(s);
        }
        assert(iterators.size() == 2);
        return new CursorWithFilterIndexed(iterators[0], iterators[1],
                                           index_column_family->index,
                                           filter.release());
      } else {
        return new CursorWithFilter(
            new CursorFromIterator(DocumentDB::NewIterator(
                read_options, primary_key_column_family_)),
            filter.release());
      }
    } else {
      return new CursorWithFilter(cursor, filter.release());
    }
    assert(false);
    return nullptr;
  }

  // currently, we lock and serialize all writes to rocksdb. reads are not
  // locked and always get consistent view of the database. we should optimize
  // locking in the future
  port::Mutex write_mutex_;
  port::Mutex name_to_index_mutex_;
  const char* kPrimaryKey = "_id";
  struct IndexColumnFamily {
    IndexColumnFamily(Index* _index, ColumnFamilyHandle* _column_family)
        : index(_index), column_family(_column_family) {}
    Index* index;
    ColumnFamilyHandle* column_family;
  };


  // name_to_index_ protected:
  // 1) when writing -- 1. lock write_mutex_, 2. lock name_to_index_mutex_
  // 2) when reading -- lock name_to_index_mutex_ OR write_mutex_
  std::unordered_map<std::string, IndexColumnFamily> name_to_index_;
  ColumnFamilyHandle* primary_key_column_family_;
  Options rocksdb_options_;
};

namespace {
Options GetRocksDBOptionsFromOptions(const DocumentDBOptions& options) {
  Options rocksdb_options;
  rocksdb_options.max_background_compactions = options.background_threads - 1;
  rocksdb_options.max_background_flushes = 1;
  rocksdb_options.write_buffer_size = options.memtable_size;
  rocksdb_options.max_write_buffer_number = 6;
  BlockBasedTableOptions table_options;
  table_options.block_cache = NewLRUCache(options.cache_size);
  rocksdb_options.table_factory.reset(NewBlockBasedTableFactory(table_options));
  return rocksdb_options;
}
}  // namespace

Status DocumentDB::Open(const DocumentDBOptions& options,
                        const std::string& name,
                        const std::vector<DocumentDB::IndexDescriptor>& indexes,
                        DocumentDB** db, bool read_only) {
  Options rocksdb_options = GetRocksDBOptionsFromOptions(options);
  rocksdb_options.create_if_missing = true;

  std::vector<ColumnFamilyDescriptor> column_families;
  column_families.push_back(ColumnFamilyDescriptor(
      kDefaultColumnFamilyName, ColumnFamilyOptions(rocksdb_options)));
  for (const auto& index : indexes) {
    column_families.emplace_back(InternalSecondaryIndexName(index.name),
                                 ColumnFamilyOptions(rocksdb_options));
  }
  std::vector<ColumnFamilyHandle*> handles;
  DB* base_db;
  Status s;
  if (read_only) {
    s = DB::OpenForReadOnly(DBOptions(rocksdb_options), name, column_families,
                            &handles, &base_db);
  } else {
    s = DB::Open(DBOptions(rocksdb_options), name, column_families, &handles,
                 &base_db);
  }
  if (!s.ok()) {
    return s;
  }

  std::vector<std::pair<Index*, ColumnFamilyHandle*>> index_cf(indexes.size());
  assert(handles.size() == indexes.size() + 1);
  for (size_t i = 0; i < indexes.size(); ++i) {
    auto index = Index::CreateIndexFromDescription(*indexes[i].description,
                                                   indexes[i].name);
    index_cf[i] = {index, handles[i + 1]};
  }
  *db = new DocumentDBImpl(base_db, handles[0], index_cf, rocksdb_options);
  return Status::OK();
}

}  // namespace rocksdb
#endif  // ROCKSDB_LITE