/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 **/

#include "storage/StorageBlock.hpp"

#include <memory>
#include <random>
#include <type_traits>
#include <unordered_map>
#include <utility>
#include <vector>

#include "catalog/CatalogRelationSchema.hpp"
#include "catalog/CatalogTypedefs.hpp"
#include "expressions/predicate/Predicate.hpp"
#include "expressions/scalar/Scalar.hpp"
#include "storage/BasicColumnStoreTupleStorageSubBlock.hpp"
#include "storage/BloomFilterIndexSubBlock.hpp"
#include "storage/CSBTreeIndexSubBlock.hpp"
#include "storage/CompressedColumnStoreTupleStorageSubBlock.hpp"
#include "storage/CompressedPackedRowStoreTupleStorageSubBlock.hpp"
#include "storage/CountedReference.hpp"
#include "storage/IndexSubBlock.hpp"
#include "storage/InsertDestinationInterface.hpp"
#include "storage/SMAIndexSubBlock.hpp"
#include "storage/SplitRowStoreTupleStorageSubBlock.hpp"
#include "storage/StorageBlockBase.hpp"
#include "storage/StorageBlockInfo.hpp"
#include "storage/StorageBlockLayout.hpp"
#include "storage/StorageBlockLayout.pb.h"
#include "storage/StorageConfig.h"
#include "storage/StorageErrors.hpp"
#include "storage/SubBlocksReference.hpp"
#include "storage/TupleIdSequence.hpp"
#include "storage/TupleStorageSubBlock.hpp"
#include "storage/ValueAccessor.hpp"
#include "storage/ValueAccessorUtil.hpp"
#include "types/TypedValue.hpp"
#include "types/containers/ColumnVector.hpp"
#include "types/containers/ColumnVectorsValueAccessor.hpp"
#include "types/containers/Tuple.hpp"
#include "types/operations/comparisons/ComparisonUtil.hpp"
#include "utility/Macros.hpp"

#include "glog/logging.h"

#ifdef QUICKSTEP_HAVE_BITWEAVING
#include "storage/bitweaving/BitWeavingIndexSubBlock.hpp"
#include "storage/bitweaving/BitWeavingHIndexSubBlock.hpp"
#include "storage/bitweaving/BitWeavingVIndexSubBlock.hpp"
#endif

using std::make_pair;
using std::pair;
using std::size_t;
using std::unique_ptr;
using std::unordered_map;
using std::vector;

namespace quickstep {

class Type;

StorageBlock::StorageBlock(const CatalogRelationSchema &relation,
                           const block_id id,
                           const StorageBlockLayout &layout,
                           const bool new_block,
                           void *block_memory,
                           const std::size_t block_memory_size)
    : StorageBlockBase(id, block_memory, block_memory_size),
      all_indices_consistent_(true),
      all_indices_inconsistent_(false),
      relation_(relation) {
  if (new_block) {
    if (block_memory_size_ < layout.getBlockHeaderSize()) {
      throw BlockMemoryTooSmall("StorageBlock", block_memory_size_);
    }

    layout.copyHeaderTo(block_memory_);
    DEBUG_ASSERT(*static_cast<const int*>(block_memory_) > 0);

    if (!block_header_.ParseFromArray(static_cast<char*>(block_memory_) + sizeof(int),
                                      *static_cast<const int*>(block_memory_))) {
      FATAL_ERROR("A StorageBlockLayout created a malformed StorageBlockHeader.");
    }

    // We mark a newly-created block as dirty, so that in the rare case that a
    // block is evicted before anything is inserted into it, we still write it
    // (and the header plus any sub-block specific fixed data structures) back
    // to disk.
    dirty_ = true;

    DEBUG_ASSERT(block_header_.IsInitialized());
    DEBUG_ASSERT(StorageBlockLayout::DescriptionIsValid(relation_, block_header_.layout()));
    DEBUG_ASSERT(block_header_.index_size_size() == block_header_.layout().index_description_size());
    DEBUG_ASSERT(block_header_.index_size_size() == block_header_.index_consistent_size());
  } else {
    if (block_memory_size < sizeof(int)) {
      throw MalformedBlock();
    }
    if (*static_cast<const int*>(block_memory_) <= 0) {
      throw MalformedBlock();
    }
    if (*static_cast<const int*>(block_memory_) + sizeof(int) > block_memory_size_) {
      throw MalformedBlock();
    }

    if (!block_header_.ParseFromArray(static_cast<char*>(block_memory_) + sizeof(int),
                                      *static_cast<const int*>(block_memory_))) {
      throw MalformedBlock();
    }
    if (!block_header_.IsInitialized()) {
      throw MalformedBlock();
    }
    if (!StorageBlockLayout::DescriptionIsValid(relation_, block_header_.layout())) {
      throw MalformedBlock();
    }
    if (block_header_.index_size_size() != block_header_.layout().index_description_size()) {
      throw MalformedBlock();
    }
    if (block_header_.index_size_size() != block_header_.index_consistent_size()) {
      throw MalformedBlock();
    }
  }

  size_t block_size_from_metadata = *static_cast<const int*>(block_memory_) + sizeof(int);
  block_size_from_metadata += block_header_.tuple_store_size();
  for (int index_num = 0;
       index_num < block_header_.index_size_size();
       ++index_num) {
    block_size_from_metadata += block_header_.index_size(index_num);
  }

  if (block_size_from_metadata > block_memory_size_) {
    throw MalformedBlock();
  } else if (block_size_from_metadata < block_memory_size_) {
    // WARNING: this isn't strictly an error, but it does indicate that there
    // is unallocated space in the block.
  }

  char *sub_block_address = static_cast<char*>(block_memory_)
                            + sizeof(int)
                            + *static_cast<const int*>(block_memory_);
  tuple_store_.reset(CreateTupleStorageSubBlock(
      relation_,
      block_header_.layout().tuple_store_description(),
      new_block,
      sub_block_address,
      block_header_.tuple_store_size()));
  sub_block_address += block_header_.tuple_store_size();
  ad_hoc_insert_supported_ = tuple_store_->supportsAdHocInsert();
  ad_hoc_insert_efficient_ = tuple_store_->adHocInsertIsEfficient();

  if (block_header_.index_size_size() > 0) {
    all_indices_inconsistent_ = true;
  }
  for (int index_num = 0;
       index_num < block_header_.index_size_size();
       ++index_num) {
    indices_.push_back(CreateIndexSubBlock(*tuple_store_,
                                           block_header_.layout().index_description(index_num),
                                           new_block, sub_block_address,
                                           block_header_.index_size(index_num)));
    sub_block_address += block_header_.index_size(index_num);
    if (!indices_.back().supportsAdHocAdd()) {
      ad_hoc_insert_efficient_ = false;
    }

    if (block_header_.index_consistent(index_num)) {
      indices_consistent_.push_back(true);
      all_indices_inconsistent_ = false;
    } else {
      indices_consistent_.push_back(false);
      all_indices_consistent_ = false;
    }
  }
}

bool StorageBlock::insertTuple(const Tuple &tuple) {
  if (!ad_hoc_insert_supported_) {
    return false;
  }

  const bool empty_before = tuple_store_->isEmpty();

  TupleStorageSubBlock::InsertResult tuple_store_insert_result = tuple_store_->insertTuple(tuple);
  if (tuple_store_insert_result.inserted_id < 0) {
    DEBUG_ASSERT(tuple_store_insert_result.ids_mutated == false);
    if (empty_before) {
      throw TupleTooLargeForBlock(tuple.getByteSize());
    } else {
      return false;
    }
  }

  bool update_succeeded = true;

  if (tuple_store_insert_result.ids_mutated) {
    update_succeeded = rebuildIndexes(true);
    if (!update_succeeded) {
      tuple_store_->deleteTuple(tuple_store_insert_result.inserted_id);
      if (!rebuildIndexes(true)) {
        // It should always be possible to rebuild an index with the tuples
        // which it originally contained.
        FATAL_ERROR("Rebuilding an IndexSubBlock failed after removing tuples.");
      }
    }
  } else {
    update_succeeded = insertEntryInIndexes(tuple_store_insert_result.inserted_id);
  }

  if (update_succeeded) {
    dirty_ = true;
    return true;
  } else {
    if (empty_before) {
      throw TupleTooLargeForBlock(tuple.getByteSize());
    } else {
      return false;
    }
  }
}

bool StorageBlock::insertTupleInBatch(const Tuple &tuple) {
  if (tuple_store_->insertTupleInBatch(tuple)) {
    invalidateAllIndexes();
    dirty_ = true;
    return true;
  } else {
    if (tuple_store_->isEmpty()) {
      throw TupleTooLargeForBlock(tuple.getByteSize());
    } else {
      return false;
    }
  }
}

tuple_id StorageBlock::bulkInsertTuples(ValueAccessor *accessor) {
  const tuple_id num_inserted = tuple_store_->bulkInsertTuples(accessor);
  if (num_inserted != 0) {
    invalidateAllIndexes();
    dirty_ = true;
  } else if (tuple_store_->isEmpty()) {
    if (!accessor->iterationFinishedVirtual()) {
      throw TupleTooLargeForBlock(0);
    }
  }
  return num_inserted;
}

tuple_id StorageBlock::bulkInsertTuplesWithRemappedAttributes(
    const std::vector<attribute_id> &attribute_map,
    ValueAccessor *accessor) {
  const tuple_id num_inserted
      = tuple_store_->bulkInsertTuplesWithRemappedAttributes(attribute_map,
                                                             accessor);
  if (num_inserted != 0) {
    invalidateAllIndexes();
    dirty_ = true;
  } else if (tuple_store_->isEmpty()) {
    if (!accessor->iterationFinishedVirtual()) {
      throw TupleTooLargeForBlock(0);
    }
  }
  return num_inserted;
}

tuple_id StorageBlock::bulkInsertPartialTuples(
    const std::vector<attribute_id> &attribute_map,
    ValueAccessor *accessor,
    const tuple_id max_num_tuples_to_insert) {
  const tuple_id num_inserted
      = tuple_store_->bulkInsertPartialTuples(attribute_map,
                                              accessor,
                                              max_num_tuples_to_insert);
  if (num_inserted != 0) {
    invalidateAllIndexes();
    dirty_ = true;
  } else if (tuple_store_->isEmpty()) {
    if (!accessor->iterationFinishedVirtual()) {
      throw TupleTooLargeForBlock(0);
    }
  }
  return num_inserted;
}

void StorageBlock::bulkInsertPartialTuplesFinalize(
    const tuple_id num_tuples_inserted) {
  tuple_store_->bulkInsertPartialTuplesFinalize(num_tuples_inserted);
}

void StorageBlock::sample(const bool is_block_sample,
                          const int percentage,
                          InsertDestinationInterface *destination) const {
  std::unique_ptr<ValueAccessor> accessor;
  // Bulk insert if the sampling method is block sample or if the sample
  // percent is 100
  if (is_block_sample || percentage == 100) {
    accessor.reset(tuple_store_->createValueAccessor());
    destination->bulkInsertTuples(accessor.get());
  } else {
    int actual_percentage = percentage;
    std::random_device random_device;
    std::mt19937 generator(random_device());
    std::unique_ptr<TupleIdSequence> sequence;
    // Get the tuple id sequence
    sequence.reset(tuple_store_->getExistenceMap());
    // Initialize a new TupleIdSequence from the existing sequence
    // but with a default bit vector of all zeros
    std::unique_ptr<TupleIdSequence> sequence_mask(
        new TupleIdSequence(sequence->length()));
    std::unordered_map<std::size_t, std::size_t> tuple_index_mapping;
    bool invert_bits = false;

    // If we have to set more number of 1 in the TupleIdSequence
    // calculate the number of 0's to be set(corresponding to the
    // rows that will not be picked) and set those rows to 1.
    if (percentage > 50) {
      actual_percentage = 100 - percentage;
      invert_bits = true;
    }
    const std::size_t num_tuples = sequence->length() * actual_percentage / 100;

    // Total possible tuple id space (N) is given by the length of
    // TupleIdSequence
    // Generate a random number between 0 and N
    for (std::size_t n = 0; n < num_tuples; n++) {
        std::uniform_real_distribution<> distribution(0, sequence->length() - (n + 1));
        std::size_t random_number = distribution(generator);
        sequence_mask->set(tuple_index_mapping.find(random_number) ==
                                   tuple_index_mapping.end()
                               ? random_number
                               : tuple_index_mapping[random_number],
                           true);
        tuple_index_mapping[random_number] = sequence->length() - (n + 1);
    }

    // Since the bits set correspond to actual zeros invert the bitvector
    if (invert_bits) {
      sequence_mask->invert();
    }
    accessor.reset(tuple_store_->createValueAccessor(sequence_mask.get()));
    destination->bulkInsertTuples(accessor.get());
  }
}

void StorageBlock::select(const vector<unique_ptr<const Scalar>> &selection,
                          const TupleIdSequence *filter,
                          InsertDestinationInterface *destination) const {
  ColumnVectorsValueAccessor temp_result;
  {
    SubBlocksReference sub_blocks_ref(*tuple_store_,
                                      indices_,
                                      indices_consistent_);

    std::unique_ptr<ValueAccessor> accessor(
        tuple_store_->createValueAccessor(filter));

    for (vector<unique_ptr<const Scalar>>::const_iterator selection_cit = selection.begin();
         selection_cit != selection.end();
         ++selection_cit) {
      // TODO(chasseur): Can probably elide some copies for parts of the
      // selection that are ScalarAttribute or ScalarLiteral.
      temp_result.addColumn((*selection_cit)->getAllValues(accessor.get(), &sub_blocks_ref));
    }
  }

  destination->bulkInsertTuples(&temp_result);
}

void StorageBlock::selectSimple(const std::vector<attribute_id> &selection,
                                const TupleIdSequence *filter,
                                InsertDestinationInterface *destination) const {
  std::unique_ptr<ValueAccessor> accessor(
      tuple_store_->createValueAccessor(filter));

  destination->bulkInsertTuplesWithRemappedAttributes(selection,
                                                      accessor.get());
}

// TODO(chasseur): Vectorization for updates.
StorageBlock::UpdateResult StorageBlock::update(
    const unordered_map<attribute_id, unique_ptr<const Scalar>> &assignments,
    const Predicate *predicate,
    InsertDestinationInterface *relocation_destination) {
  if (relation_.getID() != relocation_destination->getRelation().getID()) {
    FATAL_ERROR("StorageBlock::update() called with a relocation_destination "
                "that does not belong to the same relation.");
  }

  UpdateResult retval;
  // TODO(chasseur): Be smarter and only update indexes that need to be updated.
  std::unique_ptr<TupleIdSequence> matches(getMatchesForPredicate(predicate));

  // If nothing matches the predicate, return immediately.
  if (matches->empty()) {
    retval.indices_consistent = all_indices_consistent_;
    retval.relocation_destination_used = false;
    return retval;
  }

  // Remove index entries for tuples which will be updated.
  bool rebuild_some_indices = false;
  for (PtrVector<IndexSubBlock>::iterator it = indices_.begin();
       it != indices_.end();
       ++it) {
    if (it->supportsAdHocRemove()) {
      it->bulkRemoveEntries(*matches);
    } else {
      rebuild_some_indices = true;
    }
  }

  // To be safe, relocate ALL tuples if the relation is partitioned and we are
  // updating the partitioning attribute.
  const bool relocate_all =
      assignments.find(relocation_destination->getPartitioningAttribute()) != assignments.end();

  // IDs of tuples which should be re-added to indices.
  TupleIdSequence in_place_ids(tuple_store_->getMaxTupleID() + 1);
  // IDs of tuples which must be reinserted.
  TupleIdSequence relocate_ids(tuple_store_->getMaxTupleID() + 1);
  std::vector<Tuple> relocation_buffer;

  std::unique_ptr<ValueAccessor> accessor(tuple_store_->createValueAccessor());
  for (TupleIdSequence::const_iterator match_it = matches->begin();
       match_it != matches->end();
       ++match_it) {
    // Generate a map of updated values according to 'assignments'.
    std::unique_ptr<std::unordered_map<attribute_id, TypedValue>>
        updated_values(generateUpdatedValues(*accessor, *match_it, assignments));

    // Determine if the tuple can be modified in-place without deleting and
    // reinserting.
    if (!relocate_all
        && tuple_store_->canSetAttributeValuesInPlaceTyped(*match_it, *updated_values)) {
      // Update attribute values in place.
      for (std::unordered_map<attribute_id, TypedValue>::const_iterator update_it
               = updated_values->begin();
           update_it != updated_values->end();
           ++update_it) {
        tuple_store_->setAttributeValueInPlaceTyped(*match_it, update_it->first, update_it->second);
      }

      in_place_ids.set(*match_it, true);
    } else {
      // Make a copy of the tuple with the updated values.
      std::vector<TypedValue> updated_tuple_values;
      for (CatalogRelationSchema::const_iterator attr_it = relation_.begin();
           attr_it != relation_.end();
           ++attr_it) {
        std::unordered_map<attribute_id, TypedValue>::iterator update_it
            = updated_values->find(attr_it->getID());
        if (update_it == updated_values->end()) {
          updated_tuple_values.emplace_back(tuple_store_->getAttributeValueTyped(*match_it, attr_it->getID()));
          updated_tuple_values.back().ensureNotReference();
        } else {
          updated_tuple_values.emplace_back(std::move(update_it->second));
        }
      }

      relocation_buffer.emplace_back(std::move(updated_tuple_values));
      relocate_ids.set(*match_it, true);
    }
  }

  bool rebuild_all = false;
  retval.indices_consistent = all_indices_consistent_;
  retval.relocation_destination_used = false;
  if (!relocate_ids.empty()) {
    // Delete relocated tuples.
    if (tuple_store_->bulkDeleteTuples(&relocate_ids)) {
      rebuild_all = true;
    }

    if (relocate_all) {
      // Immediately bulk-insert into InsertDestination if required by
      // partitioning.
      retval.relocation_destination_used = true;
      relocation_destination->insertTuplesFromVector(relocation_buffer.begin(),
                                                     relocation_buffer.end());
    } else {
      // Reinsert into this block until space is exhausted.
      for (std::vector<Tuple>::const_iterator copy_it = relocation_buffer.begin();
          copy_it != relocation_buffer.end();
          ++copy_it) {
        if (rebuild_all || (!tuple_store_->adHocInsertIsEfficient())) {
          // If we must rebuild anyway, we might as well use the fast insert path.
          if (tuple_store_->insertTupleInBatch(*copy_it)) {
            rebuild_all = true;
          } else {
            retval.relocation_destination_used = true;
            relocation_destination->insertTuplesFromVector(copy_it, relocation_buffer.end());
            break;
          }
        } else {
          const TupleStorageSubBlock::InsertResult reinsert_result
              = tuple_store_->insertTuple(*copy_it);
          if (reinsert_result.inserted_id < 0) {
            DCHECK(!reinsert_result.ids_mutated);
            retval.relocation_destination_used = true;
            relocation_destination->insertTuplesFromVector(copy_it, relocation_buffer.end());
            break;
          }

          if (reinsert_result.ids_mutated) {
            rebuild_all = true;
          } else {
            // Only bother adding 'reinsert_id' to 'in_place_ids' if not rebuilding.
            in_place_ids.set(reinsert_result.inserted_id, true);
          }
        }
      }
    }
  }

  // TODO(chasseur): Consider doing a partial rollback when an index rebuild
  // fails.
  if (rebuild_all) {
    tuple_store_->rebuild();

    // Rebuild indexes.
    retval.indices_consistent = rebuildIndexes(false);
  } else if (!indices_.empty()) {
    all_indices_inconsistent_ = true;
    int index_num = 0;
    for (PtrVector<IndexSubBlock>::iterator it = indices_.begin();
         it != indices_.end();
         ++it, ++index_num) {
      bool add_successful;
      if (it->supportsAdHocAdd() && ((!rebuild_some_indices) || it->supportsAdHocRemove())) {
        add_successful = it->bulkAddEntries(in_place_ids);
#ifdef QUICKSTEP_REBUILD_INDEX_ON_UPDATE_OVERFLOW
        if (!add_successful) {
          add_successful = it->rebuild();
        }
#endif
      } else {
        add_successful = it->rebuild();
      }
      if (add_successful) {
        block_header_.set_index_consistent(index_num, true);
        indices_consistent_[index_num] = true;
        all_indices_inconsistent_ = false;
      } else {
        block_header_.set_index_consistent(index_num, false);
        indices_consistent_[index_num] = false;
        all_indices_consistent_ = false;
      }
    }

    updateHeader();
    retval.indices_consistent = all_indices_consistent_;
  }

  dirty_ = true;
  return retval;
}

namespace {

// Helper class to sort tuple.
//
// TODO(shoban): Refine this to use values directly instead of pointing to
// memory in storage block for better cache locality.
class SortReference {
 public:
  SortReference(const void *value, const tuple_id tuple) : tuple_(tuple), value_(value) {}

  inline const void* getDataPtr() const {
    return value_;
  }

  inline tuple_id getTupleID() const {
    return tuple_;
  }

 private:
  tuple_id tuple_;
  const void *value_;
};

}  // namespace

void StorageBlock::sortColumn(bool use_input_sequence,
                              const Scalar &sort_attribute,
                              bool sort_is_ascending,
                              bool null_first,
                              OrderedTupleIdSequence *sorted_sequence) const {
  const attribute_id sort_attr_id = sort_attribute.getAttributeIdForValueAccessor();
  DCHECK_NE(sort_attr_id, -1)
      << "Attempted to sort a StorageBlock on a sort attribute that is not "
      << "actually a ScalarAttribute.";

  const Type &sort_attr_type = sort_attribute.getType();

  std::unique_ptr<ValueAccessor> accessor;
  std::vector<SortReference> refs;
  OrderedTupleIdSequence nulls;

  refs.reserve(use_input_sequence ? sorted_sequence->size() : getTupleStorageSubBlock().numTuples());

  // TODO(shoban): Refer the TODO in sortColumnHelperValueAccessor() for
  // optimizing for NULL checking.
  ValueAccessor *all_accessor = tuple_store_->createValueAccessor(nullptr);
  InvokeOnValueAccessorNotAdapter(
      all_accessor,
      [&sort_attr_id,
       &use_input_sequence,
       &nulls,
       &refs,
       &accessor,
       &sorted_sequence](auto *all_accessor) -> void {  // NOLINT(build/c++11)
    if (use_input_sequence) {
      auto *seq_value_accessor = new OrderedTupleIdSequenceAdapterValueAccessor<
          typename std::remove_reference<decltype(*all_accessor)>::type>(
              all_accessor, *sorted_sequence);
      accessor.reset(seq_value_accessor);
      seq_value_accessor->beginIteration();
      while (seq_value_accessor->next()) {
        const void *value = seq_value_accessor->getUntypedValue(sort_attr_id);
        if (value) {
          refs.emplace_back(value, seq_value_accessor->getCurrentPosition());
        } else {
          nulls.emplace_back(seq_value_accessor->getCurrentPosition());
        }
      }
    } else {
      accessor.reset(all_accessor);
      while (all_accessor->next()) {
        const void *value = all_accessor->getUntypedValue(sort_attr_id);
        if (value) {
          refs.emplace_back(value, all_accessor->getCurrentPosition());
        } else {
          nulls.emplace_back(all_accessor->getCurrentPosition());
        }
      }
    }
  });

  if (use_input_sequence) {
    if (sort_is_ascending) {
      StableSortWrappedValues<SortReference, vector<SortReference>::iterator>(
          sort_attr_type, refs.begin(), refs.end());
    } else {
      // Use reverse iterators to sort in descending order.
      StableSortWrappedValues<SortReference, vector<SortReference>::reverse_iterator>(
          sort_attr_type, refs.rbegin(), refs.rend());
    }
  } else {
    if (sort_is_ascending) {
      SortWrappedValues<SortReference, vector<SortReference>::iterator>(
          sort_attr_type, refs.begin(), refs.end());
    } else {
      // Use reverse iterators to sort in descending order.
      SortWrappedValues<SortReference, vector<SortReference>::reverse_iterator>(
          sort_attr_type, refs.rbegin(), refs.rend());
    }
  }

  sorted_sequence->clear();
  if (null_first) {
    sorted_sequence->insert(sorted_sequence->end(), nulls.begin(), nulls.end());
  }
  for (const SortReference &ref : refs) {
    sorted_sequence->emplace_back(ref.getTupleID());
  }
  if (!null_first) {
    sorted_sequence->insert(sorted_sequence->end(), nulls.begin(), nulls.end());
  }
}

void StorageBlock::sort(const PtrVector<Scalar> &order_by,  // NOLINT(build/include_what_you_use)
                        const std::vector<bool> &sort_is_ascending,
                        const std::vector<bool> &null_first,
                        OrderedTupleIdSequence *sorted_sequence,
                        InsertDestinationInterface *output_destination) const {
  // TODO(shoban): We, currently, use a scheme where we stable sort the
  // tuple-id-sequence from last to first columns in ORDER BY clause to produce
  // the final sorted sequence. We could on the other hand sort from first to
  // last columns in ORDER BY clause, where the subsequent column's sort can be
  // broken down into sort of smaller ranges of tuple-id-sequences for which the
  // previous columns have the same value. This can have several pros and cons
  // the method used. Average-case asymptotics is definitely better in the
  // later. Need to do an analysis of the two methods.

  DEBUG_ASSERT(order_by.size() == sort_is_ascending.size());
  DEBUG_ASSERT(order_by.size() == null_first.size());
  DEBUG_ASSERT(order_by.size() > 0);

  // TODO(shoban): We should use reverse_iterator in conjunction with rbegin()
  // and rend() for better readability, if PtrVector supports it.
  PtrVector<Scalar>::const_iterator order_it = order_by.end();
  std::vector<bool>::const_iterator sort_is_ascending_it = sort_is_ascending.end();
  std::vector<bool>::const_iterator null_first_it = null_first.end();
  --order_it;
  --sort_is_ascending_it;
  --null_first_it;

  // Reserve capacity once. std::vector::clear() does not change capacity.
  sorted_sequence->reserve(getTupleStorageSubBlock().numTuples());

  // Sort the last column based on TupleIdSequence.
  // We can use regular sort on the last column, since there no inherent order
  // prior to this. std::sort() is an in-place impementation, whereas
  // std::stable_sort() uses extra memory. Hence, std::sort() is slight better
  // than std::stable_sort().
  sortColumn(false, *order_it, *sort_is_ascending_it, *null_first_it, sorted_sequence);

  while (order_it != order_by.begin()) {
    --order_it;
    --sort_is_ascending_it;
    --null_first_it;

    // Sort the other columns in reverse order based on OrderedTupleIdSequence
    // to maintain the previous sorted order.
    sortColumn(true, *order_it, *sort_is_ascending_it, *null_first_it, sorted_sequence);
  }

  // Write to output destination.
  if (output_destination) {
    std::unique_ptr<ValueAccessor> accessor(tuple_store_->createValueAccessor(nullptr));
    std::unique_ptr<ValueAccessor> ordered_accessor;
    InvokeOnValueAccessorNotAdapter(
        accessor.get(),
        [&](auto *accessor) -> void {  // NOLINT(build/c++11)
      ordered_accessor.reset(
          accessor->createSharedOrderedTupleIdSequenceAdapter(*sorted_sequence));
    });

    // TODO(quickstep-team): This might depending on the block layouts write one
    // or more sorted blocks for each input blocks. It would be useful to
    // remember the list of sorted blocks as an initial run for the merge phase
    // of sorting.
    output_destination->bulkInsertTuples(ordered_accessor.get(), true);
  }
}

void StorageBlock::deleteTuples(const Predicate *predicate) {
  std::unique_ptr<TupleIdSequence> matches(getMatchesForPredicate(predicate));

  if (!matches->empty()) {
    // First, remove entries from indices that support ad-hoc removal.
    bool rebuild_some_indices = false;
    for (PtrVector<IndexSubBlock>::iterator it = indices_.begin(); it != indices_.end(); ++it) {
      if (it->supportsAdHocRemove()) {
        it->bulkRemoveEntries(*matches);
      } else {
        rebuild_some_indices = true;
      }
    }

    // Delete tuples from the TupleStorageSubBlock.
    if (tuple_store_->bulkDeleteTuples(matches.get())) {
      // If the tuple-ID sequence was mutated, rebuild all indices.
      if (!rebuildIndexes(true)) {
        FATAL_ERROR("Rebuilding an IndexSubBlock failed after removing tuples.");
      }
    } else if (rebuild_some_indices) {
      // Rebuild any remaining indices that don't support ad-hoc removal.
      for (PtrVector<IndexSubBlock>::iterator it = indices_.begin(); it != indices_.end(); ++it) {
        if (!it->supportsAdHocRemove()) {
          if (!it->rebuild()) {
            FATAL_ERROR("Rebuilding an IndexSubBlock failed after removing tuples.");
          }
        }
      }
    }

    dirty_ = true;
  }
}

TupleStorageSubBlock* StorageBlock::CreateTupleStorageSubBlock(
    const CatalogRelationSchema &relation,
    const TupleStorageSubBlockDescription &description,
    const bool new_block,
    void *sub_block_memory,
    const std::size_t sub_block_memory_size) {
  DEBUG_ASSERT(description.IsInitialized());
  switch (description.sub_block_type()) {
    case TupleStorageSubBlockDescription::BASIC_COLUMN_STORE:
      return new BasicColumnStoreTupleStorageSubBlock(relation,
                                                      description,
                                                      new_block,
                                                      sub_block_memory,
                                                      sub_block_memory_size);
    case TupleStorageSubBlockDescription::COMPRESSED_PACKED_ROW_STORE:
      return new CompressedPackedRowStoreTupleStorageSubBlock(relation,
                                                              description,
                                                              new_block,
                                                              sub_block_memory,
                                                              sub_block_memory_size);
    case TupleStorageSubBlockDescription::COMPRESSED_COLUMN_STORE:
      return new CompressedColumnStoreTupleStorageSubBlock(relation,
                                                           description,
                                                           new_block,
                                                           sub_block_memory,
                                                           sub_block_memory_size);
    case TupleStorageSubBlockDescription::SPLIT_ROW_STORE:
      return new SplitRowStoreTupleStorageSubBlock(relation,
                                                   description,
                                                   new_block,
                                                   sub_block_memory,
                                                   sub_block_memory_size);
    default:
      if (new_block) {
        FATAL_ERROR("A StorageBlockLayout provided an unknown TupleStorageSubBlockType.");
      } else {
        throw MalformedBlock();
      }
  }
}

IndexSubBlock* StorageBlock::CreateIndexSubBlock(
    const TupleStorageSubBlock &tuple_store,
    const IndexSubBlockDescription &description,
    const bool new_block,
    void *sub_block_memory,
    const std::size_t sub_block_memory_size) {
  DEBUG_ASSERT(description.IsInitialized());
  switch (description.sub_block_type()) {
    case IndexSubBlockDescription::BLOOM_FILTER:
      return new BloomFilterIndexSubBlock(tuple_store,
                                          description,
                                          new_block,
                                          sub_block_memory,
                                          sub_block_memory_size);
    case IndexSubBlockDescription::CSB_TREE:
      return new CSBTreeIndexSubBlock(tuple_store,
                                      description,
                                      new_block,
                                      sub_block_memory,
                                      sub_block_memory_size);
    case IndexSubBlockDescription::SMA:
      return new SMAIndexSubBlock(tuple_store,
                                  description,
                                  new_block,
                                  sub_block_memory,
                                  sub_block_memory_size);
#ifdef QUICKSTEP_HAVE_BITWEAVING
    case IndexSubBlockDescription::BITWEAVING_V:
      return new BitWeavingVIndexSubBlock(tuple_store,
                                          description,
                                          new_block,
                                          sub_block_memory,
                                          sub_block_memory_size);
    case IndexSubBlockDescription::BITWEAVING_H:
      return new BitWeavingHIndexSubBlock(tuple_store,
                                          description,
                                          new_block,
                                          sub_block_memory,
                                          sub_block_memory_size);
#else
    case IndexSubBlockDescription::BITWEAVING_V:  // Fall through.
    case IndexSubBlockDescription::BITWEAVING_H:
      LOG(FATAL) << "Attempted to create a block with a bitweaving index "
                 << "but Quickstep was not compiled with bitweaving.";
#endif
    default:
      if (new_block) {
        FATAL_ERROR("A StorageBlockLayout provided an unknown IndexBlockType.");
      } else {
        throw MalformedBlock();
      }
  }
}

bool StorageBlock::insertEntryInIndexes(const tuple_id new_tuple) {
  DEBUG_ASSERT(ad_hoc_insert_supported_);
  DEBUG_ASSERT(new_tuple >= 0);
  DEBUG_ASSERT(all_indices_consistent_);

  for (PtrVector<IndexSubBlock>::iterator it = indices_.begin();
       it != indices_.end();
       ++it) {
    bool entry_added;
    if (it->supportsAdHocAdd()) {
      entry_added = it->addEntry(new_tuple);
    } else {
      entry_added = it->rebuild();
    }
    if (!entry_added) {
      // Roll back if index is full.
      //
      // TODO(chasseur): What about fragmented indexes, where rebuilding could allow success?
      bool rebuild_some_indices = false;
      for (PtrVector<IndexSubBlock>::iterator fixer_it = indices_.begin();
           fixer_it != it;
           ++fixer_it) {
        // Do ad-hoc removal for those indices which support it. Those that
        // don't are rebuilt after the entry is removed from the tuple_store_
        // below.
        if (fixer_it->supportsAdHocRemove()) {
          fixer_it->removeEntry(new_tuple);
        } else {
          rebuild_some_indices = true;
        }
      }

      if (tuple_store_->deleteTuple(new_tuple)) {
        // The tuple-ID sequence was mutated, so rebuild all indices.
        if (!rebuildIndexes(true)) {
          FATAL_ERROR("Rebuilding an IndexSubBlock failed after removing tuples.");
        }
      } else if (rebuild_some_indices) {
        // Rebuild those indices that were modified that don't support ad-hoc
        // removal.
        for (PtrVector<IndexSubBlock>::iterator fixer_it = indices_.begin();
             fixer_it != it;
             ++fixer_it) {
          if (!fixer_it->supportsAdHocRemove()) {
            if (!fixer_it->rebuild()) {
              // It should always be possible to rebuild an index with the
              // tuples which it originally contained.
              FATAL_ERROR("Rebuilding an IndexSubBlock failed after removing tuples.");
            }
          }
        }
      }

      return false;
    }
  }

  return true;
}

bool StorageBlock::bulkInsertEntriesInIndexes(TupleIdSequence *new_tuples,
                                              const bool roll_back_on_failure) {
  DEBUG_ASSERT(ad_hoc_insert_supported_);
  DEBUG_ASSERT(all_indices_consistent_);

  // If 'roll_back_on_failure' is false, we will allow some indices to become
  // inconsistent.
  if ((!indices_.empty()) && (!roll_back_on_failure)) {
    all_indices_inconsistent_ = true;
  }

  int index_num = 0;
  for (PtrVector<IndexSubBlock>::iterator it = indices_.begin();
       it != indices_.end();
       ++it, ++index_num) {
    bool entries_added;
    if (it->supportsAdHocAdd()) {
      entries_added = it->bulkAddEntries(*new_tuples);
    } else {
      entries_added = it->rebuild();
    }
    if (!entries_added) {
      if (roll_back_on_failure) {
        // Roll back if index full.
        bool rebuild_some_indices = false;
        for (PtrVector<IndexSubBlock>::iterator fixer_it = indices_.begin();
             fixer_it != it;
             ++fixer_it) {
          // Do ad-hoc removal for those indices which support it. Those that
          // don't are rebuilt after the entries are removed from the
          // tuple_store_ below.
          if (fixer_it->supportsAdHocRemove()) {
            fixer_it->bulkRemoveEntries(*new_tuples);
          } else {
            rebuild_some_indices = true;
          }
        }

        if (tuple_store_->bulkDeleteTuples(new_tuples)) {
          // The tuple-ID sequence was mutated, so rebuild all indices.
          if (!rebuildIndexes(true)) {
            FATAL_ERROR("Rebuilding an IndexSubBlock failed after removing tuples.");
          }
        } else if (rebuild_some_indices) {
          // Rebuild those indices that were modified that don't support ad-hoc
          // removal.
          for (PtrVector<IndexSubBlock>::iterator fixer_it = indices_.begin();
               fixer_it != it;
               ++fixer_it) {
            if (!fixer_it->supportsAdHocRemove()) {
              if (!fixer_it->rebuild()) {
                // It should always be possible to rebuild an index with the
                // tuples which it originally contained.
                FATAL_ERROR("Rebuilding an IndexSubBlock failed after removing tuples.");
              }
            }
          }
        }

        return false;
      } else {
        block_header_.set_index_consistent(index_num, false);
        indices_consistent_[index_num] = false;
        all_indices_consistent_ = false;
      }
    } else if (!roll_back_on_failure) {
      all_indices_inconsistent_ = false;
    }
  }

  if (!all_indices_consistent_) {
    updateHeader();
  }
  return all_indices_consistent_;
}

bool StorageBlock::rebuildIndexes(bool short_circuit) {
  if (indices_.empty()) {
    return true;
  }

  all_indices_consistent_ = true;
  all_indices_inconsistent_ = true;

  int index_num = 0;
  for (PtrVector<IndexSubBlock>::iterator it = indices_.begin();
       it != indices_.end();
       ++it, ++index_num) {
    if (it->rebuild()) {
      all_indices_inconsistent_ = false;
      indices_consistent_[index_num] = true;
      block_header_.set_index_consistent(index_num, true);
    } else {
      all_indices_consistent_ = false;
      indices_consistent_[index_num] = false;
      block_header_.set_index_consistent(index_num, false);
      if (short_circuit) {
        return false;
      }
    }
  }
  updateHeader();

  return all_indices_consistent_;
}

TupleIdSequence* StorageBlock::getMatchesForPredicate(const Predicate *predicate,
                                                      const TupleIdSequence *filter) const {
  if (predicate == nullptr) {
    TupleIdSequence *matches = tuple_store_->getExistenceMap();
    if (filter != nullptr) {
      matches->intersectWith(*filter);
    }
    return matches;
  }

  std::unique_ptr<ValueAccessor> value_accessor(tuple_store_->createValueAccessor());
  SubBlocksReference sub_blocks_ref(*tuple_store_,
                                    indices_,
                                    indices_consistent_);

  if (!tuple_store_->isPacked()) {
    std::unique_ptr<TupleIdSequence> existence_map(tuple_store_->getExistenceMap());
    if (filter != nullptr) {
      existence_map->intersectWith(*filter);
    }
    return predicate->getAllMatches(value_accessor.get(),
                                    &sub_blocks_ref,
                                    nullptr,
                                    existence_map.get());
  } else {
    return predicate->getAllMatches(value_accessor.get(),
                                    &sub_blocks_ref,
                                    nullptr,
                                    filter);
  }
}

std::unordered_map<attribute_id, TypedValue>* StorageBlock::generateUpdatedValues(
    const ValueAccessor &accessor,
    const tuple_id tuple,
    const unordered_map<attribute_id, unique_ptr<const Scalar>> &assignments) const {
  std::unordered_map<attribute_id, TypedValue> *update_map
      = new std::unordered_map<attribute_id, TypedValue>();

  for (unordered_map<attribute_id, unique_ptr<const Scalar>>::const_iterator it = assignments.begin();
       it != assignments.end();
       ++it) {
    pair<std::unordered_map<attribute_id, TypedValue>::iterator, bool> insert_result
        = update_map->emplace(it->first,
                              it->second->getValueForSingleTuple(accessor, tuple));
    DCHECK(insert_result.second);
    insert_result.first->second.ensureNotReference();
  }

  return update_map;
}

void StorageBlock::updateHeader() {
  DEBUG_ASSERT(*static_cast<const int*>(block_memory_) == block_header_.ByteSize());

  if (!block_header_.SerializeToArray(static_cast<char*>(block_memory_) + sizeof(int),
                                      block_header_.ByteSize())) {
    FATAL_ERROR("Failed to do binary serialization of StorageBlockHeader in StorageBlock::updateHeader()");
  }
}

void StorageBlock::invalidateAllIndexes() {
  if ((!indices_.empty()) && (!all_indices_inconsistent_)) {
    for (unsigned int index_num = 0;
         index_num < indices_.size();
         ++index_num) {
      indices_consistent_[index_num] = false;
      block_header_.set_index_consistent(index_num, false);
    }
    all_indices_consistent_ = false;
    all_indices_inconsistent_ = true;

    updateHeader();
  }
}

const std::size_t StorageBlock::getNumTuples() const {
  DCHECK(tuple_store_ != nullptr);
  return tuple_store_->numTuples();
}

}  // namespace quickstep