void Table::serializeTo(SerializeOutput &serialOutput) {
    // The table is serialized as:
    // [(int) total size]
    // [(int) header size] [num columns] [column types] [column names]
    // [(int) num tuples] [tuple data]

    /* NOTE:
       VoltDBEngine uses a binary template to create tables of single integers.
       It's called m_templateSingleLongTable and if you are seeing a serialization
       bug in tables of single integers, make sure that's correct.
    */

    // a placeholder for the total table size
    std::size_t pos = serialOutput.position();
    serialOutput.writeInt(-1);

    serializeColumnHeaderTo(serialOutput);

    // active tuple counts
    serialOutput.writeInt(static_cast<int32_t>(m_tupleCount));
    int64_t written_count = 0;
    TableIterator titer = iterator();
    TableTuple tuple(m_schema);
    while (titer.next(tuple)) {
        tuple.serializeTo(serialOutput);
        ++written_count;
    }
    assert(written_count == m_tupleCount);

    // length prefix is non-inclusive
    int32_t sz = static_cast<int32_t>(serialOutput.position() - pos - sizeof(int32_t));
    assert(sz > 0);
    serialOutput.writeIntAt(pos, sz);
}

void VarlenType::SerializeTo(const Value &val, SerializeOutput &out) const {
  uint32_t len = GetLength(val);
  out.WriteInt(len);
  if (len > 0 && len < PELOTON_VALUE_NULL) {
    out.WriteBytes(val.GetData(), len);
  }
}

bool Table::serializeTo(int32_t offset, int32_t limit, SerializeOutput &serialize_io) {
    // The table is serialized as:
    // [(int) total size]
    // [(int) header size] [num columns] [column types] [column names]
    // [(int) num tuples] [tuple data]

    /* NOTE:
       VoltDBEngine uses a binary template to create tables of single integers.
       It's called m_templateSingleLongTable and if you are seeing a serialization
       bug in tables of single integers, make sure that's correct.
    */

    // a placeholder for the total table size
    std::size_t pos = serialize_io.position();
    serialize_io.writeInt(-1);

    if (!serializeColumnHeaderTo(serialize_io))
        return false;

    // active tuple counts
    uint32_t output_size = m_tupleCount;
    if (limit != -1 || offset != -1) {
        if (offset == -1) {
            output_size = (limit < m_tupleCount ? limit : m_tupleCount);
        } else if (offset > m_tupleCount) {
            output_size = 0;
        } else {
            output_size = m_tupleCount - offset;
            if (limit != -1 && limit < output_size) output_size = limit;
        }
    }
    serialize_io.writeInt(static_cast<int32_t>(output_size));
//     fprintf(stderr, "SERIALIZE(output=%d, offset=%d, limit=%d, total=%d)\n", output_size, offset, limit, m_tupleCount);
    
    int64_t written_count = 0;
    int64_t read_count = 0;
    TableIterator titer(this);
    TableTuple tuple(m_schema);
    while (titer.next(tuple)) {
        if (offset == -1 || read_count >= offset) {
            tuple.serializeTo(serialize_io);
            if (limit != -1 && ++written_count == limit) break;
        }
        read_count++;
    }
    // assert(written_count == m_tupleCount);

    // length prefix is non-inclusive
    int32_t sz = static_cast<int32_t>(serialize_io.position() - pos - sizeof(int32_t));
    assert(sz > 0);
    serialize_io.writeIntAt(pos, sz);

    return true;
}

void Tuple::SerializeTo(SerializeOutput &output) {
  PL_ASSERT(tuple_schema_);
  size_t start = output.ReserveBytes(4);
  const int column_count = tuple_schema_->GetColumnCount();

  for (int column_itr = 0; column_itr < column_count; column_itr++) {
    type::Value value(GetValue(column_itr));
    value.SerializeTo(output);
  }

  output.WriteIntAt(
      start, static_cast<int32_t>(output.Position() - start - sizeof(int32_t)));
}

/**
 * Serialized the table, but only includes the tuples specified (columns data and all).
 * Used by the exception stuff Ariel put in.
 */
void Table::serializeTupleTo(SerializeOutput &serialOutput, voltdb::TableTuple *tuples, int numTuples) {
    //assert(m_schema->equals(tuples[0].getSchema()));

    std::size_t pos = serialOutput.position();
    serialOutput.writeInt(-1);

    assert(!tuples[0].isNullTuple());

    serializeColumnHeaderTo(serialOutput);

    serialOutput.writeInt(static_cast<int32_t>(numTuples));
    for (int ii = 0; ii < numTuples; ii++) {
        tuples[ii].serializeTo(serialOutput);
    }

    serialOutput.writeIntAt(pos, static_cast<int32_t>(serialOutput.position() - pos - sizeof(int32_t)));
}

void IntegerValue::SerializeTo(SerializeOutput &out) const {
  switch(GetTypeId()) {
    case Type::TINYINT:
      out.WriteByte(value_.tinyint);
      return;
    case Type::SMALLINT:
      out.WriteShort(value_.smallint);
      return;
    case Type::INTEGER:
    case Type::PARAMETER_OFFSET:
      out.WriteInt(value_.integer);
      return;
    case Type::BIGINT:
      out.WriteLong(value_.bigint);
      return;
    default:
      return;
  }
  throw Exception("type error");
}

void Tuple::SerializeWithHeaderTo(SerializeOutput &output) {
  PL_ASSERT(tuple_schema_);
  PL_ASSERT(tuple_data_);

  size_t start = output.Position();
  output.WriteInt(0);  // reserve first 4 bytes for the total tuple size

  const int column_count = tuple_schema_->GetColumnCount();

  for (int column_itr = 0; column_itr < column_count; column_itr++) {
    type::Value value = GetValue(column_itr);
    value.SerializeTo(output);
  }

  int32_t serialized_size =
      static_cast<int32_t>(output.Position() - start - sizeof(int32_t));

  // write out the length of the tuple at start
  output.WriteIntAt(start, serialized_size);
}

//  Serialized only the tuples specified, along with header.
bool Tile::SerializeTuplesTo(SerializeOutput &output, Tuple *tuples,
                             int num_tuples) {
  std::size_t pos = output.Position();
  output.WriteInt(-1);

  PL_ASSERT(!tuples[0].IsNull());

  // Serialize the header
  if (!SerializeHeaderTo(output)) return false;

  output.WriteInt(static_cast<int32_t>(num_tuples));
  for (int tuple_itr = 0; tuple_itr < num_tuples; tuple_itr++) {
    tuples[tuple_itr].SerializeTo(output);
  }

  // Length prefix is non-inclusive
  output.WriteIntAt(
      pos, static_cast<int32_t>(output.Position() - pos - sizeof(int32_t)));

  return true;
}

bool Tile::SerializeTo(SerializeOutput &output, oid_t num_tuples) {
  /**
   * The table is serialized as:
   *
   * [(int) total size]
   * [(int) header size] [num columns] [column types] [column names]
   * [(int) num tuples] [tuple data]
   *
   */

  // A placeholder for the total table size written at the end
  std::size_t pos = output.Position();
  output.WriteInt(-1);

  // Serialize the header
  if (!SerializeHeaderTo(output)) return false;

  // Active tuple count
  output.WriteInt(static_cast<int>(num_tuples));

  oid_t written_count = 0;
  TupleIterator tile_itr(this);
  Tuple tuple(&schema);

  while (tile_itr.Next(tuple) && written_count < num_tuples) {
    tuple.SerializeTo(output);
    ++written_count;
  }

  tuple.SetNull();

  PL_ASSERT(written_count == num_tuples);

  // Length prefix is non-inclusive
  int32_t sz = static_cast<int32_t>(output.Position() - pos - sizeof(int32_t));
  PL_ASSERT(sz > 0);
  output.WriteIntAt(pos, sz);

  return true;
}

void Table::serializeToWithoutTotalSize(SerializeOutput &serialOutput) {
    serializeColumnHeaderTo(serialOutput);

    // active tuple counts
    serialOutput.writeInt(static_cast<int32_t>(m_tupleCount));
    int64_t written_count = 0;
    TableIterator titer = iterator();
    TableTuple tuple(m_schema);
    while (titer.next(tuple)) {
        tuple.serializeTo(serialOutput);
        ++written_count;
    }
    assert(written_count == m_tupleCount);
}

bool Table::serializeToWithoutTotalSize(SerializeOutput &serialize_io) {
    if (!serializeColumnHeaderTo(serialize_io))
        return false;

    // active tuple counts
    serialize_io.writeInt(static_cast<int32_t>(m_tupleCount));
    int64_t written_count = 0;
    TableIterator titer = iterator();
    TableTuple tuple(m_schema);
    while (titer.next(tuple)) {
        tuple.serializeTo(serialize_io);
        ++written_count;
    }
    assert(written_count == m_tupleCount);

    return true;
}

bool SeqScanPlan::SerializeTo(SerializeOutput &output) {
  // A placeholder for the total size written at the end
  int start = output.Position();
  output.WriteInt(-1);

  // Write the SeqScanPlan type
  PlanNodeType plan_type = GetPlanNodeType();
  output.WriteByte(static_cast<int8_t>(plan_type));

  // Write database id and table id
  if (!GetTable()) {
    // The plan is not completed
    return false;
  }
  oid_t database_id = GetTable()->GetDatabaseOid();
  oid_t table_id = GetTable()->GetOid();

  output.WriteInt(static_cast<int>(database_id));
  output.WriteInt(static_cast<int>(table_id));

  // If column has 0 item, just write the columnid_count with 0
  int columnid_count = GetColumnIds().size();
  output.WriteInt(columnid_count);

  // If column has 0 item, nothing happens here
  for (int it = 0; it < columnid_count; it++) {
    oid_t col_id = GetColumnIds()[it];
    output.WriteInt(static_cast<int>(col_id));
  }

  // Write predicate
  if (GetPredicate() == nullptr) {
    // Write the type
    output.WriteByte(static_cast<int8_t>(EXPRESSION_TYPE_INVALID));
  } else {
    // Write the expression type
    ExpressionType expr_type = GetPredicate()->GetExpressionType();
    output.WriteByte(static_cast<int8_t>(expr_type));

    // Write predicate
    //GetPredicate()->SerializeTo(output);
  }

  // Write parent, but parent seems never be set or used right now
  if (GetParent() == nullptr) {
    // Write the type
    output.WriteByte(static_cast<int8_t>(PLAN_NODE_TYPE_INVALID));
  } else {
    // Write the parent type
    PlanNodeType parent_type = GetParent()->GetPlanNodeType();
    output.WriteByte(static_cast<int8_t>(parent_type));

    // Write parent
    GetParent()->SerializeTo(output);
  }

  // Write the total length
  int32_t sz = static_cast<int32_t>(output.Position() - start - sizeof(int));
  PL_ASSERT(sz > 0);
  output.WriteIntAt(start, sz);

  return true;
}

void TimestampType::SerializeTo(const Value& val, SerializeOutput &out) const {
  out.WriteLong(val.value_.timestamp);
}

void BooleanType::SerializeTo(const Value& val, SerializeOutput& out) const {
  out.WriteByte(val.value_.boolean);
  return;
}

bool Table::serializeColumnHeaderTo(SerializeOutput &serialize_io) {

    /* NOTE:
       VoltDBEngine uses a binary template to create tables of single integers.
       It's called m_templateSingleLongTable and if you are seeing a serialization
       bug in tables of single integers, make sure that's correct.
    */

    // skip header position
    std::size_t start;

    // use a cache
    if (m_columnHeaderData) {
        assert(m_columnHeaderSize != -1);
        serialize_io.writeBytes(m_columnHeaderData, m_columnHeaderSize);
        return true;
    }
    assert(m_columnHeaderSize == -1);

    start = serialize_io.position();

    // skip header position
    serialize_io.writeInt(-1);

    //status code
    serialize_io.writeByte(-128);

    // column counts as a short
    serialize_io.writeShort(static_cast<int16_t>(m_columnCount));

    // write an array of column types as bytes
    for (int i = 0; i < m_columnCount; ++i) {
        ValueType type = m_schema->columnType(i);
        serialize_io.writeByte(static_cast<int8_t>(type));
    }

    // write the array of column names as voltdb strings
    // NOTE: strings are ASCII only in metadata (UTF-8 in table storage)
    for (int i = 0; i < m_columnCount; ++i) {
        // column name: write (offset, length) for column definition, and string to string table
        const string& name = columnName(i);
        // column names can't be null, so length must be >= 0
        int32_t length = static_cast<int32_t>(name.size());
        assert(length >= 0);

        // this is standard string serialization for voltdb
        serialize_io.writeInt(length);
        serialize_io.writeBytes(name.data(), length);
    }


    // write the header size which is a non-inclusive int
    size_t position = serialize_io.position();
    m_columnHeaderSize = static_cast<int32_t>(position - start);
    int32_t nonInclusiveHeaderSize = static_cast<int32_t>(m_columnHeaderSize - sizeof(int32_t));
    serialize_io.writeIntAt(start, nonInclusiveHeaderSize);

    // cache the results
    m_columnHeaderData = new char[m_columnHeaderSize];
    memcpy(m_columnHeaderData, static_cast<const char*>(serialize_io.data()) + start, m_columnHeaderSize);

    return true;

}

bool Tile::SerializeHeaderTo(SerializeOutput &output) {
  std::size_t start;

  // Use the cache if possible
  if (column_header != NULL) {
    PL_ASSERT(column_header_size != INVALID_OID);
    output.WriteBytes(column_header, column_header_size);
    return true;
  }

  PL_ASSERT(column_header_size == INVALID_OID);

  // Skip header position
  start = output.Position();
  output.WriteInt(-1);

  // Status code
  output.WriteByte(-128);

  // Column counts as a short
  output.WriteShort(static_cast<int16_t>(column_count));

  // Write an array of column types as bytes
  for (oid_t column_itr = 0; column_itr < column_count; ++column_itr) {
    type::Type::TypeId type = schema.GetType(column_itr);
    output.WriteByte(static_cast<int8_t>(type));
  }

  // Write the array of column names as strings
  // NOTE: strings are ASCII only in metadata (UTF-8 in table storage)
  for (oid_t column_itr = 0; column_itr < column_count; ++column_itr) {
    // Column name: Write (offset, length) for column definition, and string to
    // string table
    const std::string &name = GetColumnName(column_itr);

    // Column names can't be null, so length must be >= 0
    int32_t length = static_cast<int32_t>(name.size());
    PL_ASSERT(length >= 0);

    // this is standard string serialization for voltdb
    output.WriteInt(length);
    output.WriteBytes(name.data(), length);
  }

  // Write the header size which is a non-inclusive int
  size_t Position = output.Position();
  column_header_size = static_cast<int32_t>(Position - start);

  int32_t non_inclusive_header_size =
      static_cast<int32_t>(column_header_size - sizeof(int32_t));
  output.WriteIntAt(start, non_inclusive_header_size);

  // Cache the column header
  column_header = new char[column_header_size];
  PL_MEMCPY(column_header, static_cast<const char *>(output.Data()) + start,
            column_header_size);

  return true;
}