stats::QueryMetric::QueryParamBuf QueryMetricsCatalog::GetParamTypes(concurrency::TransactionContext *txn,
                                                                     const std::string &name) {
  std::vector<oid_t> column_ids({ColumnId::PARAM_TYPES});  // param_types
  oid_t index_offset = IndexId::PRIMARY_KEY;               // Primary key index
  std::vector<type::Value> values;
  values.push_back(type::ValueFactory::GetVarcharValue(name, nullptr).Copy());
  values.push_back(type::ValueFactory::GetIntegerValue(database_oid_).Copy());

  auto result_tiles =
      GetResultWithIndexScan(txn,
                             column_ids,
                             index_offset,
                             values);

  stats::QueryMetric::QueryParamBuf param_types;
  PELOTON_ASSERT(result_tiles->size() <= 1);  // unique
  if (result_tiles->size() != 0) {
    PELOTON_ASSERT((*result_tiles)[0]->GetTupleCount() <= 1);
    if ((*result_tiles)[0]->GetTupleCount() != 0) {
      auto param_types_value = (*result_tiles)[0]->GetValue(0, 0);
      param_types.buf = const_cast<uchar *>(
          reinterpret_cast<const uchar *>(param_types_value.GetData()));
      param_types.len = param_types_value.GetLength();
    }
  }

  return param_types;
}

column_map_type Layout::DeserializeColumnMap(oid_t num_columns,
                                             std::string column_map_str) {
  column_map_type column_map;
  std::stringstream ss(column_map_str);

  for (oid_t col_id = 0; col_id < num_columns; col_id++) {
    oid_t str_col_id, tile_id, tile_col_id;
    // Read col_id from column_map_str
    ss >> str_col_id;
    PELOTON_ASSERT(str_col_id == col_id);

    PELOTON_ASSERT(ss.peek() == ':');
    ss.ignore();
    // Read tile_id from column_map_str
    ss >> tile_id;

    PELOTON_ASSERT(ss.peek() == ':');
    ss.ignore();
    // Read tile_col_id from column_map_str
    ss >> tile_col_id;

    // Insert the column info into column_map
    column_map[col_id] = std::make_pair(tile_id, tile_col_id);

    if (ss.peek() == ',') {
      ss.ignore();
    }
  }

  return column_map;
}

/**
 * @brief Basic checks.
 * @return true on success, false otherwise.
 */
bool AppendExecutor::DInit() {
  // should have >= 2 children, otherwise pointless.
  PELOTON_ASSERT(children_.size() >= 2);
  PELOTON_ASSERT(cur_child_id_ == 0);

  return true;
}

void TimestampOrderingTransactionManager::PerformInsert(
    TransactionContext *const current_txn, const ItemPointer &location,
    ItemPointer *index_entry_ptr) {
  PELOTON_ASSERT(!current_txn->IsReadOnly());

  oid_t tile_group_id = location.block;
  oid_t tuple_id = location.offset;

  auto storage_manager = storage::StorageManager::GetInstance();
  auto tile_group_header = storage_manager->GetTileGroup(tile_group_id)->GetHeader();
  auto transaction_id = current_txn->GetTransactionId();

  // check MVCC info
  // the tuple slot must be empty.
  PELOTON_ASSERT(tile_group_header->GetTransactionId(tuple_id) ==
                 INVALID_TXN_ID);
  PELOTON_ASSERT(tile_group_header->GetBeginCommitId(tuple_id) == MAX_CID);
  PELOTON_ASSERT(tile_group_header->GetEndCommitId(tuple_id) == MAX_CID);

  tile_group_header->SetTransactionId(tuple_id, transaction_id);
  tile_group_header->SetLastReaderCommitId(tuple_id,
                                           current_txn->GetCommitId());

  // no need to set next item pointer.

  // Add the new tuple into the insert set
  current_txn->RecordInsert(location);

  // Write down the head pointer's address in tile group header
  tile_group_header->SetIndirection(tuple_id, index_entry_ptr);
}

/**
 * @brief Nothing to init at the moment.
 * @return true on success, false otherwise.
 */
bool InsertExecutor::DInit() {
  PELOTON_ASSERT(children_.size() == 0 || children_.size() == 1);
  PELOTON_ASSERT(executor_context_);

  done_ = false;
  return true;
}

bool TransactionContext::RecordDelete(const ItemPointer &location) {
  RWType rw_type;

  if (rw_set_.Find(location, rw_type)) {
    if (rw_type == RWType::READ || rw_type == RWType::READ_OWN) {
      rw_set_.Update(location, RWType::DELETE);
      // record write
      is_written_ = true;
      return false;
    }
    if (rw_type == RWType::UPDATE) {
      rw_set_.Update(location, RWType::DELETE);
      return false;
    }
    if (rw_type == RWType::INSERT) {
      rw_set_.Update(location, RWType::INS_DEL);
      --insert_count_;
      return true;
    }
    if (rw_type == RWType::DELETE) {
      PELOTON_ASSERT(false);
      return false;
    }
    PELOTON_ASSERT(false);
  } else {
    rw_set_.Insert(location, RWType::DELETE);
  }
  return false;
}

Value IntegerType::Modulo(const Value& left, const Value &right) const {
  PELOTON_ASSERT(left.CheckInteger());
  PELOTON_ASSERT(left.CheckComparable(right));
  if (left.IsNull() || right.IsNull())
    return left.OperateNull(right);

  if (right.IsZero()) {
    throw Exception(ExceptionType::DIVIDE_BY_ZERO,
                    "Division by zero on right-hand side");
  }

  switch (right.GetTypeId()) {
  case TypeId::TINYINT:
    return ModuloValue<int32_t, int8_t>(left, right);
  case TypeId::SMALLINT:
    return ModuloValue<int32_t, int16_t>(left, right);
  case TypeId::INTEGER:
  case TypeId::PARAMETER_OFFSET:
    return ModuloValue<int32_t, int32_t>(left, right);
  case TypeId::BIGINT:
    return ModuloValue<int32_t, int64_t>(left, right);
  case TypeId::DECIMAL:
    return ValueFactory::GetDecimalValue(
        ValMod(left.value_.integer, right.GetAs<double>()));
  case TypeId::VARCHAR: {
      auto r_value = right.CastAs(TypeId::INTEGER);
      return ModuloValue<int32_t, int32_t>(left, r_value);
  }
  default:
    break;
  }

  throw Exception("type error");
}

typename Cache<Key, Value>::iterator Cache<Key, Value>::insert(
    const Entry &entry) {
  PELOTON_ASSERT(list_.size() == map_.size());
  PELOTON_ASSERT(list_.size() <= this->capacity_);
  auto map_itr = map_.find(entry.first);
  auto cache_itr = iterator(map_itr);

  if (map_itr == map_.end()) {
    /* new key */
    list_.push_front(entry.first);
    auto ret =
        map_.emplace(entry.first, std::make_pair(entry.second, list_.begin()));
    PELOTON_ASSERT(ret.second); /* should not fail */
    cache_itr = iterator(ret.first);
    while (map_.size() > this->capacity_) {
      auto deleted = list_.back();
      this->map_.erase(deleted);
      list_.erase(std::prev(list_.end()));
    }
  } else {
    list_.splice(list_.begin(), list_, map_itr->second.second);
    map_itr->second = std::make_pair(entry.second, list_.begin());
  }
  PELOTON_ASSERT(list_.size() == map_.size());
  PELOTON_ASSERT(list_.size() <= capacity_);
  return cache_itr;
}

Value IntegerParentType::Max(const Value& left, const Value& right) const {
  PELOTON_ASSERT(left.CheckInteger());
  PELOTON_ASSERT(left.CheckComparable(right));
  if (left.IsNull() || right.IsNull()) return left.OperateNull(right);

  if (left.CompareGreaterThanEquals(right) == CmpBool::CmpTrue) return left.Copy();
  return right.Copy();
}

bool ModelUtil::EarlyStop(vector_t val_losses, size_t patience, float delta) {
  // Check for edge cases
  PELOTON_ASSERT(patience > 1);
  PELOTON_ASSERT(delta > 0);
  if (val_losses.size() < patience) return false;
  float cur_loss = val_losses[val_losses.size() - 1];
  float pat_loss = val_losses[val_losses.size() - patience];
  // Loss should have at least dropped by delta at this point
  return (pat_loss - cur_loss) < delta;
}

type::Value TupleValueExpression::Evaluate(
    const AbstractTuple *tuple1, const AbstractTuple *tuple2,
    UNUSED_ATTRIBUTE executor::ExecutorContext *context) const {
  if (tuple_idx_ == 0) {
    PELOTON_ASSERT(tuple1 != nullptr);
    return (tuple1->GetValue(value_idx_));
  } else {
    PELOTON_ASSERT(tuple2 != nullptr);
    return (tuple2->GetValue(value_idx_));
  }
}

/**
 * Write buffer data to the file. Although fwrite() is not a system call,
 * calling fwrite frequently with small byte is not efficient because fwrite
 * does many sanity checks. We use local buffer to amortize that.
 */
void CopyExecutor::FlushBuffer() {
  PELOTON_ASSERT(buff_ptr < COPY_BUFFER_SIZE);
  PELOTON_ASSERT(buff_size + buff_ptr <= COPY_BUFFER_SIZE);
  while (buff_size > 0) {
    size_t bytes_written =
        fwrite(buff + buff_ptr, sizeof(char), buff_size, file_handle_.file);
    // Book keeping
    buff_ptr += bytes_written;
    buff_size -= bytes_written;
    total_bytes_written += bytes_written;
    LOG_TRACE("fwrite %d bytes", (int)bytes_written);
  }
  buff_ptr = 0;
}

std::shared_ptr<DatabaseCatalogObject> DatabaseCatalog::GetDatabaseObject(
    oid_t database_oid, concurrency::TransactionContext *txn) {
  if (txn == nullptr) {
    throw CatalogException("Transaction is invalid!");
  }
  // try get from cache
  auto database_object = txn->catalog_cache.GetDatabaseObject(database_oid);
  if (database_object) return database_object;

  // cache miss, get from pg_database
  std::vector<oid_t> column_ids(all_column_ids);
  oid_t index_offset = IndexId::PRIMARY_KEY;  // Index of database_oid
  std::vector<type::Value> values;
  values.push_back(type::ValueFactory::GetIntegerValue(database_oid).Copy());

  auto result_tiles =
      GetResultWithIndexScan(column_ids, index_offset, values, txn);

  if (result_tiles->size() == 1 && (*result_tiles)[0]->GetTupleCount() == 1) {
    auto database_object =
        std::make_shared<DatabaseCatalogObject>((*result_tiles)[0].get(), txn);
    // insert into cache
    bool success = txn->catalog_cache.InsertDatabaseObject(database_object);
    PELOTON_ASSERT(success == true);
    (void)success;
    return database_object;
  } else {
    LOG_DEBUG("Found %lu database tiles with oid %u", result_tiles->size(),
              database_oid);
  }

  // return empty object if not found
  return nullptr;
}

GroupExpression *Memo::InsertExpression(std::shared_ptr<GroupExpression> gexpr,
                                        GroupID target_group, bool enforced) {
  // If leaf, then just return
  if (gexpr->Op().GetType() == OpType::Leaf) {
    const LeafOperator *leaf = gexpr->Op().As<LeafOperator>();
    PELOTON_ASSERT(target_group == UNDEFINED_GROUP ||
           target_group == leaf->origin_group);
    gexpr->SetGroupID(leaf->origin_group);
    return nullptr;
  }

  // Lookup in hash table
  auto it = group_expressions_.find(gexpr.get());

  if (it != group_expressions_.end()) {
    gexpr->SetGroupID((*it)->GetGroupID());
    return *it;
  } else {
    group_expressions_.insert(gexpr.get());
    // New expression, so try to insert into an existing group or
    // create a new group if none specified
    GroupID group_id;
    if (target_group == UNDEFINED_GROUP) {
      group_id = AddNewGroup(gexpr);
    } else {
      group_id = target_group;
    }
    Group *group = GetGroupByID(group_id);
    group->AddExpression(gexpr, enforced);
    return gexpr.get();
  }
}

// Update Predicate expression
// this is used in the NLJoin executor
void SeqScanExecutor::UpdatePredicate(const std::vector<oid_t> &column_ids,
                                      const std::vector<type::Value> &values) {
  std::vector<oid_t> predicate_column_ids;

  PELOTON_ASSERT(column_ids.size() <= column_ids_.size());

  // columns_ids is the column id
  // in the join executor, should
  // convert to the column id in the
  // seq scan executor
  for (auto column_id : column_ids) {
    predicate_column_ids.push_back(column_ids_[column_id]);
  }

  expression::AbstractExpression *new_predicate =
      values.size() != 0 ? ColumnsValuesToExpr(predicate_column_ids, values, 0)
                         : nullptr;

  // combine with original predicate
  if (old_predicate_ != nullptr) {
    expression::AbstractExpression *lexpr = new_predicate,
                                   *rexpr = old_predicate_->Copy();

    new_predicate = new expression::ConjunctionExpression(
        ExpressionType::CONJUNCTION_AND, lexpr, rexpr);
  }

  // Currently a hack that prevent memory leak
  // we should eventually make prediate_ a unique_ptr
  new_predicate_.reset(new_predicate);
  predicate_ = new_predicate;
}

// Number of bytes in string
type::Value OldEngineStringFunctions::OctetLength(
    const std::vector<type::Value> &args) {
  PELOTON_ASSERT(args.size() == 1);
  std::string str = args[0].ToString();
  int32_t len = str.length();
  return (type::ValueFactory::GetIntegerValue(len));
}

StringFunctions::StrWithLen StringFunctions::BTrim(
    UNUSED_ATTRIBUTE executor::ExecutorContext &ctx, const char *str,
    uint32_t str_len, const char *from, UNUSED_ATTRIBUTE uint32_t from_len) {
  PELOTON_ASSERT(str != nullptr && from != nullptr);

  // Skip the tailing 0
  str_len--;

  if (str_len == 0) {
    return StringFunctions::StrWithLen{str, 1};
  }

  int head = 0;
  int tail = str_len - 1;

  // Trim tail
  while (tail >= 0 && strchr(from, str[tail]) != nullptr) {
    tail--;
  }

  // Trim head
  while (head < (int)str_len && strchr(from, str[head]) != nullptr) {
    head++;
  }

  // Done
  auto new_len = static_cast<uint32_t>(std::max(tail - head + 1, 0) + 1);
  return StringFunctions::StrWithLen{str + head, new_len};
}

llvm::Value *RowBatch::Row::GetTID(CodeGen &codegen) {
  if (tid_ == nullptr) {
    tid_ = batch_.GetPhysicalPosition(codegen, *this);
  }
  PELOTON_ASSERT(tid_ != nullptr);
  return tid_;
}

void InputColumnDeriver::Visit(const PhysicalOrderBy *) {
  // we need to pass down both required columns and sort columns
  auto prop = properties_->GetPropertyOfType(PropertyType::SORT);
  PELOTON_ASSERT(prop.get() != nullptr);
  ExprSet input_cols_set;
  for (auto expr : required_cols_) {
    if (expression::ExpressionUtil::IsAggregateExpression(expr)) {
      input_cols_set.insert(expr);
    } else {
      expression::ExpressionUtil::GetTupleValueExprs(input_cols_set, expr);
    }
  }
  auto sort_prop = prop->As<PropertySort>();
  size_t sort_col_size = sort_prop->GetSortColumnSize();
  for (size_t idx = 0; idx < sort_col_size; ++idx) {
    input_cols_set.insert(sort_prop->GetSortColumn(idx));
  }
  vector<AbstractExpression *> cols;
  for (auto &expr : input_cols_set) {
    cols.push_back(expr);
  }
  output_input_cols_ =
      pair<vector<AbstractExpression *>, vector<vector<AbstractExpression *>>>{
          cols, {cols}};
}

type::Value ComparisonExpression::Evaluate(
    const AbstractTuple *tuple1, const AbstractTuple *tuple2,
    executor::ExecutorContext *context) const {
  PELOTON_ASSERT(children_.size() == 2);
  auto vl = children_[0]->Evaluate(tuple1, tuple2, context);
  auto vr = children_[1]->Evaluate(tuple1, tuple2, context);
  switch (exp_type_) {
    case (ExpressionType::COMPARE_EQUAL):
      return type::ValueFactory::GetBooleanValue(vl.CompareEquals(vr));
    case (ExpressionType::COMPARE_NOTEQUAL):
      return type::ValueFactory::GetBooleanValue(vl.CompareNotEquals(vr));
    case (ExpressionType::COMPARE_LESSTHAN):
      return type::ValueFactory::GetBooleanValue(vl.CompareLessThan(vr));
    case (ExpressionType::COMPARE_GREATERTHAN):
      return type::ValueFactory::GetBooleanValue(vl.CompareGreaterThan(vr));
    case (ExpressionType::COMPARE_LESSTHANOREQUALTO):
      return type::ValueFactory::GetBooleanValue(vl.CompareLessThanEquals(vr));
    case (ExpressionType::COMPARE_GREATERTHANOREQUALTO):
      return type::ValueFactory::GetBooleanValue(
          vl.CompareGreaterThanEquals(vr));
    case (ExpressionType::COMPARE_DISTINCT_FROM): {
      if (vl.IsNull() && vr.IsNull()) {
        return type::ValueFactory::GetBooleanValue(false);
      } else if (!vl.IsNull() && !vr.IsNull()) {
        return type::ValueFactory::GetBooleanValue(vl.CompareNotEquals(vr));
      }
      return type::ValueFactory::GetBooleanValue(true);
    }
    default:
      throw Exception("Invalid comparison expression type.");
  }
}