LibSQL: Introduce Serializer as a mediator between Heap and client code

Classes reading and writing to the data heap would communicate directly
with the Heap object, and transfer ByteBuffers back and forth with it.
This makes things like caching and locking hard. Therefore all data
persistence activity will be funneled through a Serializer object which
in turn submits it to the Heap.

Introducing this unfortunately resulted in a huge amount of churn, in
which a number of smaller refactorings got caught up as well.

27 files changed, 926 insertions, 720 deletions

diff --git a/Userland/Libraries/LibSQL/BTree.cpp b/Userland/Libraries/LibSQL/BTree.cpp
index bb0d07d77b..d83a353e8d 100644
--- a/Userland/Libraries/LibSQL/BTree.cpp
+++ b/Userland/Libraries/LibSQL/BTree.cpp
@@ -10,14 +10,14 @@
 
 namespace SQL {
 
-BTree::BTree(Heap& heap, NonnullRefPtr<TupleDescriptor> const& descriptor, bool unique, u32 pointer)
-    : Index(heap, descriptor, unique, pointer)
+BTree::BTree(Serializer& serializer, NonnullRefPtr<TupleDescriptor> const& descriptor, bool unique, u32 pointer)
+    : Index(serializer, descriptor, unique, pointer)
     , m_root(nullptr)
 {
 }
 
-BTree::BTree(Heap& heap, NonnullRefPtr<TupleDescriptor> const& descriptor, u32 pointer)
-    : BTree(heap, descriptor, true, pointer)
+BTree::BTree(Serializer& serializer, NonnullRefPtr<TupleDescriptor> const& descriptor, u32 pointer)
+    : BTree(serializer, descriptor, true, pointer)
 {
 }
 
@@ -37,10 +37,9 @@ BTreeIterator BTree::end()
 void BTree::initialize_root()
 {
     if (pointer()) {
-        if (pointer() < heap().size()) {
-            auto buffer = read_block(pointer());
-            size_t offset = 0;
-            m_root = make<TreeNode>(*this, nullptr, pointer(), buffer, offset);
+        if (serializer().has_block(pointer())) {
+            serializer().get_block(pointer());
+            m_root = serializer().make_and_deserialize<TreeNode>(*this, pointer());
         } else {
             m_root = make<TreeNode>(*this, nullptr, pointer());
         }
@@ -50,13 +49,14 @@ void BTree::initialize_root()
         if (on_new_root)
             on_new_root();
     }
+    m_root->dump_if(0, "initialize_root");
 }
 
 TreeNode* BTree::new_root()
 {
     set_pointer(new_record_pointer());
     m_root = make<TreeNode>(*this, nullptr, m_root.leak_ptr(), pointer());
-    add_to_write_ahead_log(m_root->as_index_node());
+    serializer().serialize_and_write(*m_root.ptr(), m_root->pointer());
     if (on_new_root)
         on_new_root();
     return m_root;
diff --git a/Userland/Libraries/LibSQL/BTree.h b/Userland/Libraries/LibSQL/BTree.h
index 717d5299b0..d02222e38d 100644
--- a/Userland/Libraries/LibSQL/BTree.h
+++ b/Userland/Libraries/LibSQL/BTree.h
@@ -43,7 +43,7 @@ public:
     TreeNode* node();
 
 private:
-    void inflate();
+    void deserialize(Serializer&);
 
     TreeNode* m_owner;
     u32 m_pointer { 0 };
@@ -53,27 +53,27 @@ private:
 
 class TreeNode : public IndexNode {
 public:
+    TreeNode(BTree&, u32 = 0);
     TreeNode(BTree&, TreeNode*, u32 = 0);
     TreeNode(BTree&, TreeNode*, TreeNode*, u32 = 0);
-    TreeNode(BTree&, TreeNode*, u32 pointer, ByteBuffer&, size_t&);
     ~TreeNode() override = default;
 
     [[nodiscard]] BTree& tree() const { return m_tree; }
     [[nodiscard]] TreeNode* up() const { return m_up; }
     [[nodiscard]] size_t size() const { return m_entries.size(); }
+    [[nodiscard]] size_t length() const;
     [[nodiscard]] Vector<Key> entries() const { return m_entries; }
     [[nodiscard]] u32 down_pointer(size_t) const;
     [[nodiscard]] TreeNode* down_node(size_t);
     [[nodiscard]] bool is_leaf() const { return m_is_leaf; }
 
-    [[nodiscard]] size_t max_keys_in_node();
     Key const& operator[](size_t) const;
     bool insert(Key const&);
     bool update_key_pointer(Key const&);
     TreeNode* node_for(Key const&);
     Optional<u32> get(Key&);
-    void serialize(ByteBuffer&) const override;
-    IndexNode* as_index_node() override { return dynamic_cast<IndexNode*>(this); }
+    void deserialize(Serializer&);
+    void serialize(Serializer&) const;
 
 private:
     TreeNode(BTree&, TreeNode*, DownPointer&, u32 = 0);
@@ -111,8 +111,8 @@ public:
     Function<void(void)> on_new_root;
 
 private:
-    BTree(Heap& heap, NonnullRefPtr<TupleDescriptor> const&, bool unique, u32 pointer);
-    BTree(Heap& heap, NonnullRefPtr<TupleDescriptor> const&, u32 pointer);
+    BTree(Serializer&, NonnullRefPtr<TupleDescriptor> const&, bool unique, u32 pointer);
+    BTree(Serializer&, NonnullRefPtr<TupleDescriptor> const&, u32 pointer);
     void initialize_root();
     TreeNode* new_root();
     OwnPtr<TreeNode> m_root { nullptr };
diff --git a/Userland/Libraries/LibSQL/BTreeIterator.cpp b/Userland/Libraries/LibSQL/BTreeIterator.cpp
index a926b49c0a..d062287489 100644
--- a/Userland/Libraries/LibSQL/BTreeIterator.cpp
+++ b/Userland/Libraries/LibSQL/BTreeIterator.cpp
@@ -232,7 +232,7 @@ bool BTreeIterator::update(Key const& new_value)
 
     // We are friend of BTree and TreeNode. Don't know how I feel about that.
     m_current->m_entries[m_index] = new_value;
-    m_current->tree().add_to_write_ahead_log(m_current);
+    m_current->tree().serializer().serialize_and_write(*m_current, m_current->pointer());
     return true;
 }
 
diff --git a/Userland/Libraries/LibSQL/CMakeLists.txt b/Userland/Libraries/LibSQL/CMakeLists.txt
index d9e8b63c3e..744d7980aa 100644
--- a/Userland/Libraries/LibSQL/CMakeLists.txt
+++ b/Userland/Libraries/LibSQL/CMakeLists.txt
@@ -17,6 +17,7 @@ set(SOURCES
     Key.cpp
     Meta.cpp
     Row.cpp
+    Serializer.cpp
     SQLClient.cpp
     TreeNode.cpp
     Tuple.cpp
diff --git a/Userland/Libraries/LibSQL/Database.cpp b/Userland/Libraries/LibSQL/Database.cpp
index ccbe61cbaf..1854dd4cd7 100644
--- a/Userland/Libraries/LibSQL/Database.cpp
+++ b/Userland/Libraries/LibSQL/Database.cpp
@@ -19,9 +19,10 @@ namespace SQL {
 
 Database::Database(String name)
     : m_heap(Heap::construct(name))
-    , m_schemas(BTree::construct(*m_heap, SchemaDef::index_def()->to_tuple_descriptor(), m_heap->schemas_root()))
-    , m_tables(BTree::construct(*m_heap, TableDef::index_def()->to_tuple_descriptor(), m_heap->tables_root()))
-    , m_table_columns(BTree::construct(*m_heap, ColumnDef::index_def()->to_tuple_descriptor(), m_heap->table_columns_root()))
+    , m_serializer(m_heap)
+    , m_schemas(BTree::construct(m_serializer, SchemaDef::index_def()->to_tuple_descriptor(), m_heap->schemas_root()))
+    , m_tables(BTree::construct(m_serializer, TableDef::index_def()->to_tuple_descriptor(), m_heap->tables_root()))
+    , m_table_columns(BTree::construct(m_serializer, ColumnDef::index_def()->to_tuple_descriptor(), m_heap->table_columns_root()))
 {
     m_schemas->on_new_root = [&]() {
         m_heap->set_schemas_root(m_schemas->root());
@@ -118,10 +119,7 @@ Vector<Row> Database::select_all(TableDef const& table)
     VERIFY(m_table_cache.get(table.key().hash()).has_value());
     Vector<Row> ret;
     for (auto pointer = table.pointer(); pointer; pointer = ret.last().next_pointer()) {
-        auto buffer_or_error = m_heap->read_block(pointer);
-        if (buffer_or_error.is_error())
-            VERIFY_NOT_REACHED();
-        ret.empend(table, pointer, buffer_or_error.value());
+        ret.append(m_serializer.deserialize_block<Row>(pointer, table, pointer));
     }
     return ret;
 }
@@ -134,10 +132,7 @@ Vector<Row> Database::match(TableDef const& table, Key const& key)
     // TODO Match key against indexes defined on table. If found,
     // use the index instead of scanning the table.
     for (auto pointer = table.pointer(); pointer;) {
-        auto buffer_or_error = m_heap->read_block(pointer);
-        if (buffer_or_error.is_error())
-            VERIFY_NOT_REACHED();
-        Row row(table, pointer, buffer_or_error.value());
+        auto row = m_serializer.deserialize_block<Row>(pointer, table, pointer);
         if (row.match(key))
             ret.append(row);
         pointer = ret.last().next_pointer();
@@ -164,9 +159,8 @@ bool Database::insert(Row& row)
 bool Database::update(Row& tuple)
 {
     VERIFY(m_table_cache.get(tuple.table()->key().hash()).has_value());
-    ByteBuffer buffer;
-    tuple.serialize(buffer);
-    m_heap->add_to_wal(tuple.pointer(), buffer);
+    m_serializer.reset();
+    return m_serializer.serialize_and_write<Tuple>(tuple, tuple.pointer());
 
     // TODO update indexes defined on table.
     return true;
diff --git a/Userland/Libraries/LibSQL/Database.h b/Userland/Libraries/LibSQL/Database.h
index 5603dcc2e1..b3bfc0d3af 100644
--- a/Userland/Libraries/LibSQL/Database.h
+++ b/Userland/Libraries/LibSQL/Database.h
@@ -11,6 +11,7 @@
 #include <LibCore/Object.h>
 #include <LibSQL/Forward.h>
 #include <LibSQL/Heap.h>
+#include <LibSQL/Meta.h>
 
 namespace SQL {
 
@@ -42,7 +43,8 @@ public:
     bool update(Row&);
 
 private:
-    RefPtr<Heap> m_heap;
+    NonnullRefPtr<Heap> m_heap;
+    Serializer m_serializer;
     RefPtr<BTree> m_schemas;
     RefPtr<BTree> m_tables;
     RefPtr<BTree> m_table_columns;
diff --git a/Userland/Libraries/LibSQL/Forward.h b/Userland/Libraries/LibSQL/Forward.h
index 1ef5e733be..273b813b9a 100644
--- a/Userland/Libraries/LibSQL/Forward.h
+++ b/Userland/Libraries/LibSQL/Forward.h
@@ -21,13 +21,16 @@ class IndexNode;
 class IndexDef;
 class Key;
 class KeyPartDef;
+class Relation;
 class Row;
+class SchemaDef;
 class SQLResult;
+class Serializer;
 class TableDef;
 class TreeNode;
 class Tuple;
 class TupleDescriptor;
-struct TupleElement;
+struct TupleElementDescriptor;
 class Value;
 }
 
diff --git a/Userland/Libraries/LibSQL/HashIndex.cpp b/Userland/Libraries/LibSQL/HashIndex.cpp
index b437daa971..2642ada69a 100644
--- a/Userland/Libraries/LibSQL/HashIndex.cpp
+++ b/Userland/Libraries/LibSQL/HashIndex.cpp
@@ -7,7 +7,7 @@
 #include <LibSQL/HashIndex.h>
 #include <LibSQL/Heap.h>
 #include <LibSQL/Key.h>
-#include <LibSQL/Serialize.h>
+#include <LibSQL/Serializer.h>
 
 namespace SQL {
 
@@ -19,18 +19,19 @@ HashDirectoryNode::HashDirectoryNode(HashIndex& index, u32 node_number, size_t o
 {
 }
 
-HashDirectoryNode::HashDirectoryNode(HashIndex& index, u32 pointer, ByteBuffer& buffer)
+HashDirectoryNode::HashDirectoryNode(HashIndex& index, u32 pointer)
     : IndexNode(pointer)
     , m_hash_index(index)
 {
+}
+
+void HashDirectoryNode::deserialize(Serializer& serializer)
+{
     dbgln_if(SQL_DEBUG, "Deserializing Hash Directory Node");
-    size_t offset = 0;
-    deserialize_from<u32>(buffer, offset, index.m_global_depth);
-    u32 size;
-    deserialize_from<u32>(buffer, offset, size);
-    dbgln_if(SQL_DEBUG, "Global Depth {}, #Bucket pointers {}", index.global_depth(), size);
-    u32 next_node;
-    deserialize_from<u32>(buffer, offset, next_node);
+    m_hash_index.m_global_depth = serializer.deserialize<u32>();
+    auto size = serializer.deserialize<u32>();
+    dbgln_if(SQL_DEBUG, "Global Depth {}, #Bucket pointers {}", m_hash_index.global_depth(), size);
+    auto next_node = serializer.deserialize<u32>();
     if (next_node) {
         dbgln_if(SQL_DEBUG, "Next node {}", next_node);
         m_hash_index.m_nodes.append(next_node);
@@ -39,20 +40,18 @@ HashDirectoryNode::HashDirectoryNode(HashIndex& index, u32 pointer, ByteBuffer&
         m_is_last = true;
     }
     for (auto ix = 0u; ix < size; ix++) {
-        u32 bucket_pointer;
-        deserialize_from(buffer, offset, bucket_pointer);
-        u32 local_depth;
-        deserialize_from(buffer, offset, local_depth);
-        dbgln_if(SQL_DEBUG, "Bucket pointer {} local depth {}", bucket_pointer, local_depth);
-        index.append_bucket(ix, local_depth, bucket_pointer);
+        auto bucket_pointer = serializer.deserialize<u32>();
+        auto local_depth = serializer.deserialize<u32>();
+        dbgln_if(SQL_DEBUG, "--Index {} bucket pointer {} local depth {}", ix, bucket_pointer, local_depth);
+        m_hash_index.append_bucket(ix, local_depth, bucket_pointer);
     }
 }
 
-void HashDirectoryNode::serialize(ByteBuffer& buffer) const
+void HashDirectoryNode::serialize(Serializer& serializer) const
 {
     dbgln_if(SQL_DEBUG, "Serializing directory node #{}. Offset {}", m_node_number, m_offset);
-    serialize_to(buffer, m_hash_index.global_depth());
-    serialize_to(buffer, number_of_pointers());
+    serializer.serialize<u32>((u32)m_hash_index.global_depth());
+    serializer.serialize<u32>(number_of_pointers());
     dbgln_if(SQL_DEBUG, "Global depth {}, #bucket pointers {}", m_hash_index.global_depth(), number_of_pointers());
 
     u32 next_node;
@@ -64,12 +63,12 @@ void HashDirectoryNode::serialize(ByteBuffer& buffer) const
         dbgln_if(SQL_DEBUG, "This is the last directory node");
     }
 
-    serialize_to(buffer, next_node);
+    serializer.serialize<u32>(next_node);
     for (auto ix = 0u; ix < number_of_pointers(); ix++) {
         auto& bucket = m_hash_index.m_buckets[m_offset + ix];
-        dbgln_if(SQL_DEBUG, "Bucket pointer {} local depth {}", bucket->pointer(), bucket->local_depth());
-        serialize_to(buffer, bucket->pointer());
-        serialize_to(buffer, bucket->local_depth());
+        dbgln_if(SQL_DEBUG, "Bucket index #{} pointer {} local depth {} size {}", ix, bucket->pointer(), bucket->local_depth(), bucket->size());
+        serializer.serialize<u32>(bucket->pointer());
+        serializer.serialize<u32>(bucket->local_depth());
     }
 }
 
@@ -81,44 +80,41 @@ HashBucket::HashBucket(HashIndex& hash_index, u32 index, u32 local_depth, u32 po
 {
 }
 
-void HashBucket::serialize(ByteBuffer& buffer) const
+void HashBucket::serialize(Serializer& serializer) const
 {
     dbgln_if(SQL_DEBUG, "Serializing bucket: pointer {}, index #{}, local depth {} size {}",
         pointer(), index(), local_depth(), size());
-    dbgln_if(SQL_DEBUG, "key_length: {} max_entries: {}", m_hash_index.descriptor()->data_length(), max_entries_in_bucket());
-    serialize_to(buffer, local_depth());
-    serialize_to(buffer, size());
-    dbgln_if(SQL_DEBUG, "buffer size after prolog {}", buffer.size());
+    serializer.serialize<u32>(local_depth());
+    serializer.serialize<u32>(size());
     for (auto& key : m_entries) {
-        key.serialize(buffer);
-        dbgln_if(SQL_DEBUG, "Key {} buffer size {}", key.to_string(), buffer.size());
+        serializer.serialize<Key>(key);
     }
 }
 
-void HashBucket::inflate()
+void HashBucket::deserialize(Serializer& serializer)
 {
     if (m_inflated || !pointer())
         return;
     dbgln_if(SQL_DEBUG, "Inflating Hash Bucket {}", pointer());
-    auto buffer = m_hash_index.read_block(pointer());
-    size_t offset = 0;
-    deserialize_from(buffer, offset, m_local_depth);
+    m_local_depth = serializer.deserialize<u32>();
     dbgln_if(SQL_DEBUG, "Bucket Local Depth {}", m_local_depth);
-    u32 size;
-    deserialize_from(buffer, offset, size);
+    auto size = serializer.deserialize<u32>();
     dbgln_if(SQL_DEBUG, "Bucket has {} keys", size);
     for (auto ix = 0u; ix < size; ix++) {
-        Key key(m_hash_index.descriptor(), buffer, offset);
+        auto key = serializer.deserialize<Key>(m_hash_index.descriptor());
         dbgln_if(SQL_DEBUG, "Key {}: {}", ix, key.to_string());
         m_entries.append(key);
     }
     m_inflated = true;
 }
 
-size_t HashBucket::max_entries_in_bucket() const
+size_t HashBucket::length() const
 {
-    auto key_size = m_hash_index.descriptor()->data_length() + sizeof(u32);
-    return (BLOCKSIZE - 2 * sizeof(u32)) / key_size;
+    size_t len = 2 * sizeof(u32);
+    for (auto& key : m_entries) {
+        len += key.length();
+    }
+    return len;
 }
 
 Optional<u32> HashBucket::get(Key& key)
@@ -134,15 +130,17 @@ Optional<u32> HashBucket::get(Key& key)
 
 bool HashBucket::insert(Key const& key)
 {
-    inflate();
+    if (!m_inflated)
+        m_hash_index.serializer().deserialize_block_to(pointer(), *this);
     if (find_key_in_bucket(key).has_value()) {
         return false;
     }
-    if (size() >= max_entries_in_bucket()) {
+    if ((length() + key.length()) > BLOCKSIZE) {
+        dbgln_if(SQL_DEBUG, "Adding key {} would make length exceed block size", key.to_string());
         return false;
     }
     m_entries.append(key);
-    m_hash_index.add_to_write_ahead_log(this);
+    m_hash_index.serializer().serialize_and_write(*this, pointer());
     return true;
 }
 
@@ -161,7 +159,7 @@ HashBucket const* HashBucket::next_bucket()
 {
     for (auto ix = m_index + 1; ix < m_hash_index.size(); ix++) {
         auto bucket = m_hash_index.get_bucket_by_index(ix);
-        bucket->inflate();
+        m_hash_index.serializer().deserialize_block_to<HashBucket>(bucket->pointer(), *bucket);
         if (bucket->size())
             return bucket;
     }
@@ -178,13 +176,27 @@ HashBucket const* HashBucket::previous_bucket()
     return nullptr;
 }
 
+Vector<Key> const& HashBucket::entries()
+{
+    if (!m_inflated)
+        m_hash_index.serializer().deserialize_block_to(pointer(), *this);
+    return m_entries;
+}
+
 Key const& HashBucket::operator[](size_t ix)
 {
-    inflate();
+    if (!m_inflated)
+        m_hash_index.serializer().deserialize_block_to(pointer(), *this);
     VERIFY(ix < size());
     return m_entries[ix];
 }
 
+Key const& HashBucket::operator[](size_t ix) const
+{
+    VERIFY(ix < m_entries.size());
+    return m_entries[ix];
+}
+
 void HashBucket::list_bucket()
 {
     warnln("Bucket #{} size {} local depth {} pointer {}{}",
@@ -194,20 +206,19 @@ void HashBucket::list_bucket()
     }
 }
 
-HashIndex::HashIndex(Heap& heap, NonnullRefPtr<TupleDescriptor> const& descriptor, u32 first_node)
-    : Index(heap, descriptor, true, first_node)
+HashIndex::HashIndex(Serializer& serializer, NonnullRefPtr<TupleDescriptor> const& descriptor, u32 first_node)
+    : Index(serializer, descriptor, true, first_node)
     , m_nodes()
     , m_buckets()
 {
     if (!first_node) {
         set_pointer(new_record_pointer());
     }
-    if (this->heap().has_block(first_node)) {
+    if (serializer.has_block(first_node)) {
         u32 pointer = first_node;
         do {
-            VERIFY(this->heap().has_block(pointer));
-            auto buffer = read_block(pointer);
-            auto node = HashDirectoryNode(*this, pointer, buffer);
+            VERIFY(serializer.has_block(pointer));
+            auto node = serializer.deserialize_block<HashDirectoryNode>(pointer, *this, pointer);
             if (node.is_last())
                 break;
             pointer = m_nodes.last(); // FIXME Ugly
@@ -215,10 +226,10 @@ HashIndex::HashIndex(Heap& heap, NonnullRefPtr<TupleDescriptor> const& descripto
     } else {
         auto bucket = append_bucket(0u, 1u, new_record_pointer());
         bucket->m_inflated = true;
-        add_to_write_ahead_log(bucket);
+        serializer.serialize_and_write(*bucket, bucket->pointer());
         bucket = append_bucket(1u, 1u, new_record_pointer());
         bucket->m_inflated = true;
-        add_to_write_ahead_log(bucket);
+        serializer.serialize_and_write(*bucket, bucket->pointer());
         m_nodes.append(first_node);
         write_directory_to_write_ahead_log();
     }
@@ -242,10 +253,12 @@ HashBucket* HashIndex::get_bucket_for_insert(Key const& key)
     auto key_hash = key.hash();
 
     do {
+        dbgln_if(SQL_DEBUG, "HashIndex::get_bucket_for_insert({}) bucket {} of {}", key.to_string(), key_hash % size(), size());
         auto bucket = get_bucket(key_hash % size());
-        if (bucket->size() < bucket->max_entries_in_bucket()) {
+        if (bucket->length() + key.length() < BLOCKSIZE) {
             return bucket;
         }
+        dbgln_if(SQL_DEBUG, "Bucket is full (bucket size {}/length {} key length {}). Expanding directory", bucket->size(), bucket->length(), key.length());
 
         // We previously doubled the directory but the target bucket is
         // still at an older depth. Create new buckets at the current global
@@ -254,36 +267,47 @@ HashBucket* HashIndex::get_bucket_for_insert(Key const& key)
         while (bucket->local_depth() < global_depth()) {
             auto base_index = bucket->index();
             auto step = 1 << (global_depth() - bucket->local_depth());
+            auto total_moved = 0;
             for (auto ix = base_index + step; ix < size(); ix += step) {
                 auto& sub_bucket = m_buckets[ix];
                 sub_bucket->set_local_depth(bucket->local_depth() + 1);
+                auto moved = 0;
                 for (auto entry_index = (int)bucket->m_entries.size() - 1; entry_index >= 0; entry_index--) {
                     if (bucket->m_entries[entry_index].hash() % size() == ix) {
                         if (!sub_bucket->pointer()) {
                             sub_bucket->set_pointer(new_record_pointer());
                         }
                         sub_bucket->insert(bucket->m_entries.take(entry_index));
+                        moved++;
                     }
                 }
-                if (m_buckets[ix]->pointer())
-                    add_to_write_ahead_log(m_buckets[ix]);
+                if (moved > 0) {
+                    dbgln_if(SQL_DEBUG, "Moved {} entries from bucket #{} to #{}", moved, base_index, ix);
+                    serializer().serialize_and_write(*sub_bucket, sub_bucket->pointer());
+                }
+                total_moved += moved;
             }
+            if (total_moved)
+                dbgln_if(SQL_DEBUG, "Redistributed {} entries from bucket #{}", total_moved, base_index);
+            else
+                dbgln_if(SQL_DEBUG, "Nothing redistributed from bucket #{}", base_index);
             bucket->set_local_depth(bucket->local_depth() + 1);
-            add_to_write_ahead_log(bucket);
+            serializer().serialize_and_write(*bucket, bucket->pointer());
             write_directory_to_write_ahead_log();
 
             auto bucket_after_redistribution = get_bucket(key_hash % size());
-            if (bucket_after_redistribution->size() < bucket_after_redistribution->max_entries_in_bucket()) {
+            if (bucket_after_redistribution->length() + key.length() < BLOCKSIZE)
                 return bucket_after_redistribution;
-            }
         }
         expand();
     } while (true);
+    VERIFY_NOT_REACHED();
 }
 
 void HashIndex::expand()
 {
     auto sz = size();
+    dbgln_if(SQL_DEBUG, "Expanding directory from {} to {} buckets", sz, 2 * sz);
     for (auto i = 0u; i < sz; i++) {
         auto bucket = get_bucket(i);
         bucket = append_bucket(sz + i, bucket->local_depth(), 0u);
@@ -303,7 +327,7 @@ void HashIndex::write_directory_to_write_ahead_log()
     size_t num_node = 0u;
     while (offset < size()) {
         HashDirectoryNode node(*this, num_node, offset);
-        add_to_write_ahead_log(node.as_index_node());
+        serializer().serialize_and_write(node, node.pointer());
         offset += node.number_of_pointers();
     }
 }
@@ -325,14 +349,20 @@ Optional<u32> HashIndex::get(Key& key)
 {
     auto hash = key.hash();
     auto bucket_index = hash % size();
+    dbgln_if(SQL_DEBUG, "HashIndex::get({}) bucket_index {}", key.to_string(), bucket_index);
     auto bucket = get_bucket(bucket_index);
+    if constexpr (SQL_DEBUG)
+        bucket->list_bucket();
     return bucket->get(key);
 }
 
 bool HashIndex::insert(Key const& key)
 {
+    dbgln_if(SQL_DEBUG, "HashIndex::insert({})", key.to_string());
     auto bucket = get_bucket_for_insert(key);
     bucket->insert(key);
+    if constexpr (SQL_DEBUG)
+        bucket->list_bucket();
     return true;
 }
 
@@ -369,7 +399,6 @@ void HashIndex::list_hash()
     bool first_bucket = true;
     for (auto& bucket : m_buckets) {
         if (first_bucket) {
-            warnln("Max. keys in bucket {}", bucket->max_entries_in_bucket());
             first_bucket = false;
         }
         bucket->list_bucket();
diff --git a/Userland/Libraries/LibSQL/HashIndex.h b/Userland/Libraries/LibSQL/HashIndex.h
index 4c56706ee8..b7f55198e8 100644
--- a/Userland/Libraries/LibSQL/HashIndex.h
+++ b/Userland/Libraries/LibSQL/HashIndex.h
@@ -28,23 +28,16 @@ public:
     ~HashBucket() override = default;
     Optional<u32> get(Key&);
     bool insert(Key const&);
-    Vector<Key> const& entries()
-    {
-        inflate();
-        return m_entries;
-    }
+    Vector<Key> const& entries();
     Key const& operator[](size_t);
-    Key const& operator[](size_t ix) const
-    {
-        VERIFY(ix < m_entries.size());
-        return m_entries[ix];
-    }
+    Key const& operator[](size_t ix) const;
     [[nodiscard]] u32 local_depth() const { return m_local_depth; }
     [[nodiscard]] u32 size() { return entries().size(); }
+    [[nodiscard]] size_t length() const;
     [[nodiscard]] u32 size() const { return m_entries.size(); }
     [[nodiscard]] u32 index() const { return m_index; }
-    void serialize(ByteBuffer&) const override;
-    IndexNode* as_index_node() override { return dynamic_cast<IndexNode*>(this); }
+    void serialize(Serializer&) const;
+    void deserialize(Serializer&);
     [[nodiscard]] HashIndex const& hash_index() const { return m_hash_index; }
     [[nodiscard]] HashBucket const* next_bucket();
     [[nodiscard]] HashBucket const* previous_bucket();
@@ -54,8 +47,6 @@ private:
     Optional<size_t> find_key_in_bucket(Key const&);
     void set_index(u32 index) { m_index = index; }
     void set_local_depth(u32 depth) { m_local_depth = depth; }
-    [[nodiscard]] size_t max_entries_in_bucket() const;
-    void inflate();
 
     HashIndex& m_hash_index;
     u32 m_local_depth { 1 };
@@ -88,7 +79,7 @@ public:
     void list_hash();
 
 private:
-    HashIndex(Heap&, NonnullRefPtr<TupleDescriptor> const&, u32);
+    HashIndex(Serializer&, NonnullRefPtr<TupleDescriptor> const&, u32);
 
     void expand();
     void write_directory_to_write_ahead_log();
@@ -107,10 +98,10 @@ private:
 class HashDirectoryNode : public IndexNode {
 public:
     HashDirectoryNode(HashIndex&, u32, size_t);
-    HashDirectoryNode(HashIndex&, u32, ByteBuffer&);
+    HashDirectoryNode(HashIndex&, u32);
     HashDirectoryNode(HashDirectoryNode const& other) = default;
-    void serialize(ByteBuffer&) const override;
-    IndexNode* as_index_node() override { return dynamic_cast<IndexNode*>(this); }
+    void deserialize(Serializer&);
+    void serialize(Serializer&) const;
     [[nodiscard]] u32 number_of_pointers() const { return min(max_pointers_in_node(), m_hash_index.size() - m_offset); }
     [[nodiscard]] bool is_last() const { return m_is_last; }
     static constexpr size_t max_pointers_in_node() { return (BLOCKSIZE - 3 * sizeof(u32)) / (2 * sizeof(u32)); }
diff --git a/Userland/Libraries/LibSQL/Heap.cpp b/Userland/Libraries/LibSQL/Heap.cpp
index cab2e89611..9b76899ff6 100644
--- a/Userland/Libraries/LibSQL/Heap.cpp
+++ b/Userland/Libraries/LibSQL/Heap.cpp
@@ -9,7 +9,7 @@
 #include <AK/String.h>
 #include <LibCore/IODevice.h>
 #include <LibSQL/Heap.h>
-#include <LibSQL/Serialize.h>
+#include <LibSQL/Serializer.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 
@@ -41,6 +41,7 @@ Heap::Heap(String file_name)
         read_zero_block();
     else
         initialize_zero_block();
+    dbgln_if(SQL_DEBUG, "Heap file {} opened. Size = {}", file_name, size());
 }
 
 Result<ByteBuffer, String> Heap::read_block(u32 block)
@@ -56,12 +57,18 @@ Result<ByteBuffer, String> Heap::read_block(u32 block)
     auto ret = m_file->read(BLOCKSIZE);
     if (ret.is_empty())
         return String("Could not read block");
+    dbgln_if(SQL_DEBUG, "{:02x} {:02x} {:02x} {:02x} {:02x} {:02x} {:02x} {:02x}",
+        *ret.offset_pointer(0), *ret.offset_pointer(1),
+        *ret.offset_pointer(2), *ret.offset_pointer(3),
+        *ret.offset_pointer(4), *ret.offset_pointer(5),
+        *ret.offset_pointer(6), *ret.offset_pointer(7));
     return ret;
 }
 
 bool Heap::write_block(u32 block, ByteBuffer& buffer)
 {
-    VERIFY(block < m_next_block);
+    dbgln_if(SQL_DEBUG, "write_block({}): m_next_block {}", block, m_next_block);
+    VERIFY(block <= m_next_block);
     if (!seek_block(block))
         VERIFY_NOT_REACHED();
     dbgln_if(SQL_DEBUG, "Write heap block {} size {}", block, buffer.size());
@@ -71,6 +78,11 @@ bool Heap::write_block(u32 block, ByteBuffer& buffer)
         buffer.resize(BLOCKSIZE);
         memset(buffer.offset_pointer((int)sz), 0, BLOCKSIZE - sz);
     }
+    dbgln_if(SQL_DEBUG, "{:02x} {:02x} {:02x} {:02x} {:02x} {:02x} {:02x} {:02x}",
+        *buffer.offset_pointer(0), *buffer.offset_pointer(1),
+        *buffer.offset_pointer(2), *buffer.offset_pointer(3),
+        *buffer.offset_pointer(4), *buffer.offset_pointer(5),
+        *buffer.offset_pointer(6), *buffer.offset_pointer(7));
     if (m_file->write(buffer.data(), (int)buffer.size())) {
         if (block == m_end_of_file)
             m_end_of_file++;
@@ -110,8 +122,7 @@ u32 Heap::new_record_pointer()
             VERIFY_NOT_REACHED();
         }
         auto new_pointer = m_free_list;
-        size_t offset = 0;
-        deserialize_from<u32>(block_or_error.value(), offset, m_free_list);
+        memcpy(&m_free_list, block_or_error.value().offset_pointer(0), sizeof(u32));
         update_zero_block();
         return new_pointer;
     }
@@ -134,6 +145,7 @@ void Heap::flush()
         write_block(block, buffer_or_empty.value());
     }
     m_write_ahead_log.clear();
+    dbgln_if(SQL_DEBUG, "WAL flushed. Heap size = {}", size());
 }
 
 constexpr static const char* FILE_ID = "SerenitySQL ";
diff --git a/Userland/Libraries/LibSQL/Heap.h b/Userland/Libraries/LibSQL/Heap.h
index 9375106f8f..4f83928795 100644
--- a/Userland/Libraries/LibSQL/Heap.h
+++ b/Userland/Libraries/LibSQL/Heap.h
@@ -12,8 +12,6 @@
 #include <AK/Vector.h>
 #include <LibCore/File.h>
 #include <LibCore/Object.h>
-#include <LibSQL/Meta.h>
-#include <LibSQL/Serialize.h>
 
 namespace SQL {
 
@@ -81,7 +79,17 @@ public:
         update_zero_block();
     }
 
-    void add_to_wal(u32 block, ByteBuffer& buffer) { m_write_ahead_log.set(block, buffer); }
+    void add_to_wal(u32 block, ByteBuffer& buffer)
+    {
+        dbgln_if(SQL_DEBUG, "Adding to WAL: block #{}, size {}", block, buffer.size());
+        dbgln_if(SQL_DEBUG, "{:02x} {:02x} {:02x} {:02x} {:02x} {:02x} {:02x} {:02x}",
+            *buffer.offset_pointer(0), *buffer.offset_pointer(1),
+            *buffer.offset_pointer(2), *buffer.offset_pointer(3),
+            *buffer.offset_pointer(4), *buffer.offset_pointer(5),
+            *buffer.offset_pointer(6), *buffer.offset_pointer(7));
+        m_write_ahead_log.set(block, buffer);
+    }
+
     void flush();
 
 private:
diff --git a/Userland/Libraries/LibSQL/Index.cpp b/Userland/Libraries/LibSQL/Index.cpp
index ecd14d5b25..264eadc897 100644
--- a/Userland/Libraries/LibSQL/Index.cpp
+++ b/Userland/Libraries/LibSQL/Index.cpp
@@ -10,37 +10,19 @@
 
 namespace SQL {
 
-Index::Index(Heap& heap, NonnullRefPtr<TupleDescriptor> const& descriptor, bool unique, u32 pointer)
-    : m_heap(heap)
+Index::Index(Serializer& serializer, NonnullRefPtr<TupleDescriptor> const& descriptor, bool unique, u32 pointer)
+    : m_serializer(serializer)
     , m_descriptor(descriptor)
     , m_unique(unique)
     , m_pointer(pointer)
 {
 }
 
-Index::Index(Heap& heap, NonnullRefPtr<TupleDescriptor> const& descriptor, u32 pointer)
-    : m_heap(heap)
+Index::Index(Serializer& serializer, NonnullRefPtr<TupleDescriptor> const& descriptor, u32 pointer)
+    : m_serializer(serializer)
     , m_descriptor(descriptor)
     , m_pointer(pointer)
 {
 }
 
-ByteBuffer Index::read_block(u32 block)
-{
-    auto ret = m_heap.read_block(block);
-    if (ret.is_error()) {
-        warnln("Error reading block {}: {}", block, ret.error());
-        VERIFY_NOT_REACHED();
-    }
-    return ret.value();
-}
-
-void Index::add_to_write_ahead_log(IndexNode* node)
-{
-    VERIFY(node->pointer());
-    ByteBuffer buffer;
-    node->serialize(buffer);
-    m_heap.add_to_wal(node->pointer(), buffer);
-}
-
 }
diff --git a/Userland/Libraries/LibSQL/Index.h b/Userland/Libraries/LibSQL/Index.h
index 4805737f3c..fad9cfd245 100644
--- a/Userland/Libraries/LibSQL/Index.h
+++ b/Userland/Libraries/LibSQL/Index.h
@@ -16,8 +16,7 @@ class IndexNode {
 public:
     virtual ~IndexNode() = default;
     [[nodiscard]] u32 pointer() const { return m_pointer; }
-    virtual void serialize(ByteBuffer&) const = 0;
-    virtual IndexNode* as_index_node() = 0;
+    IndexNode* as_index_node() { return dynamic_cast<IndexNode*>(this); }
 
 protected:
     explicit IndexNode(u32 pointer)
@@ -43,18 +42,16 @@ public:
     [[nodiscard]] u32 pointer() const { return m_pointer; }
 
 protected:
-    Index(Heap& heap, NonnullRefPtr<TupleDescriptor> const&, bool unique, u32 pointer);
-    Index(Heap& heap, NonnullRefPtr<TupleDescriptor> const&, u32 pointer);
+    Index(Serializer&, NonnullRefPtr<TupleDescriptor> const&, bool unique, u32 pointer);
+    Index(Serializer&, NonnullRefPtr<TupleDescriptor> const&, u32 pointer);
 
-    [[nodiscard]] Heap const& heap() const { return m_heap; }
-    [[nodiscard]] Heap& heap() { return m_heap; }
+    [[nodiscard]] Serializer& serializer() { return m_serializer; }
     void set_pointer(u32 pointer) { m_pointer = pointer; }
-    u32 new_record_pointer() { return m_heap.new_record_pointer(); }
-    ByteBuffer read_block(u32);
-    void add_to_write_ahead_log(IndexNode*);
+    u32 new_record_pointer() { return m_serializer.new_record_pointer(); }
+    //    ByteBuffer read_block(u32);
 
 private:
-    Heap& m_heap;
+    Serializer m_serializer;
     NonnullRefPtr<TupleDescriptor> m_descriptor;
     bool m_unique { false };
     u32 m_pointer { 0 };
diff --git a/Userland/Libraries/LibSQL/Key.cpp b/Userland/Libraries/LibSQL/Key.cpp
index 206462cf19..a31c040403 100644
--- a/Userland/Libraries/LibSQL/Key.cpp
+++ b/Userland/Libraries/LibSQL/Key.cpp
@@ -9,11 +9,6 @@
 
 namespace SQL {
 
-Key::Key()
-    : Tuple()
-{
-}
-
 Key::Key(NonnullRefPtr<TupleDescriptor> const& descriptor)
     : Tuple(descriptor)
 {
@@ -25,15 +20,15 @@ Key::Key(NonnullRefPtr<IndexDef> index)
 {
 }
 
-Key::Key(NonnullRefPtr<TupleDescriptor> const& descriptor, ByteBuffer& buffer, size_t& offset)
-    : Tuple(descriptor, buffer, offset)
+Key::Key(NonnullRefPtr<TupleDescriptor> const& descriptor, Serializer& serializer)
+    : Tuple(descriptor, serializer)
 {
 }
 
-Key::Key(RefPtr<IndexDef> index, ByteBuffer& buffer, size_t& offset)
+Key::Key(RefPtr<IndexDef> index, Serializer& serializer)
     : Key(index->to_tuple_descriptor())
 {
-    deserialize(buffer, offset);
+    Tuple::deserialize(serializer);
 }
 
 }
diff --git a/Userland/Libraries/LibSQL/Key.h b/Userland/Libraries/LibSQL/Key.h
index edc12a40b7..e05b017631 100644
--- a/Userland/Libraries/LibSQL/Key.h
+++ b/Userland/Libraries/LibSQL/Key.h
@@ -14,14 +14,12 @@ namespace SQL {
 
 class Key : public Tuple {
 public:
-    Key();
+    Key() = default;
     explicit Key(NonnullRefPtr<TupleDescriptor> const&);
     explicit Key(NonnullRefPtr<IndexDef>);
-    Key(NonnullRefPtr<TupleDescriptor> const&, ByteBuffer&, size_t& offset);
-    Key(RefPtr<IndexDef>, ByteBuffer&, size_t& offset);
-    Key(Key const&) = default;
+    Key(NonnullRefPtr<TupleDescriptor> const&, Serializer&);
+    Key(RefPtr<IndexDef>, Serializer&);
     RefPtr<IndexDef> index() const { return m_index; }
-    [[nodiscard]] virtual size_t data_length() const override { return Tuple::data_length() + sizeof(u32); }
 
 private:
     RefPtr<IndexDef> m_index { nullptr };
diff --git a/Userland/Libraries/LibSQL/Row.cpp b/Userland/Libraries/LibSQL/Row.cpp
index 9a5c94cc95..bf3b549513 100644
--- a/Userland/Libraries/LibSQL/Row.cpp
+++ b/Userland/Libraries/LibSQL/Row.cpp
@@ -6,7 +6,7 @@
 
 #include <LibSQL/Meta.h>
 #include <LibSQL/Row.h>
-#include <LibSQL/Serialize.h>
+#include <LibSQL/Serializer.h>
 #include <LibSQL/Tuple.h>
 
 namespace SQL {
@@ -21,26 +21,29 @@ Row::Row(TupleDescriptor const& descriptor)
 {
 }
 
-Row::Row(RefPtr<TableDef> table)
+Row::Row(RefPtr<TableDef> table, u32 pointer)
     : Tuple(table->to_tuple_descriptor())
     , m_table(table)
 {
+    set_pointer(pointer);
 }
 
-Row::Row(RefPtr<TableDef> table, u32 pointer, ByteBuffer& buffer)
-    : Tuple(table->to_tuple_descriptor())
+Row::Row(RefPtr<TableDef> table, u32 pointer, Serializer& serializer)
+    : Row(move(table), pointer)
 {
-    // FIXME Sanitize constructor situation in Tuple so this can be better
-    size_t offset = 0;
-    deserialize(buffer, offset);
-    deserialize_from<u32>(buffer, offset, m_next_pointer);
-    set_pointer(pointer);
+    Row::deserialize(serializer);
+}
+
+void Row::deserialize(Serializer& serializer)
+{
+    Tuple::deserialize(serializer);
+    m_next_pointer = serializer.deserialize<u32>();
 }
 
-void Row::serialize(ByteBuffer& buffer) const
+void Row::serialize(Serializer& serializer) const
 {
-    Tuple::serialize(buffer);
-    serialize_to<u32>(buffer, next_pointer());
+    Tuple::serialize(serializer);
+    serializer.serialize<u32>(next_pointer());
 }
 
 void Row::copy_from(Row const& other)
diff --git a/Userland/Libraries/LibSQL/Row.h b/Userland/Libraries/LibSQL/Row.h
index 27c98339f0..25c6a8cab7 100644
--- a/Userland/Libraries/LibSQL/Row.h
+++ b/Userland/Libraries/LibSQL/Row.h
@@ -9,6 +9,7 @@
 #include <AK/ByteBuffer.h>
 #include <AK/RefPtr.h>
 #include <LibSQL/Forward.h>
+#include <LibSQL/Meta.h>
 #include <LibSQL/Value.h>
 
 namespace SQL {
@@ -26,16 +27,17 @@ class Row : public Tuple {
 public:
     Row();
     explicit Row(TupleDescriptor const&);
-    explicit Row(RefPtr<TableDef>);
-    Row(RefPtr<TableDef>, u32, ByteBuffer&);
+    explicit Row(RefPtr<TableDef>, u32 pointer = 0);
+    Row(RefPtr<TableDef>, u32, Serializer&);
     Row(Row const&) = default;
     virtual ~Row() override = default;
 
     [[nodiscard]] u32 next_pointer() const { return m_next_pointer; }
     void next_pointer(u32 ptr) { m_next_pointer = ptr; }
     RefPtr<TableDef> table() const { return m_table; }
-    virtual void serialize(ByteBuffer&) const override;
-    [[nodiscard]] virtual size_t data_length() const override { return Tuple::data_length() + sizeof(u32); }
+    [[nodiscard]] virtual size_t length() const override { return Tuple::length() + sizeof(u32); }
+    virtual void serialize(Serializer&) const override;
+    virtual void deserialize(Serializer&) override;
 
 protected:
     void copy_from(Row const&);
diff --git a/Userland/Libraries/LibSQL/Serialize.h b/Userland/Libraries/LibSQL/Serialize.h
deleted file mode 100644
index bc03590058..0000000000
--- a/Userland/Libraries/LibSQL/Serialize.h
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Copyright (c) 2021, Jan de Visser <jan@de-visser.net>
- *
- * SPDX-License-Identifier: BSD-2-Clause
- */
-
-#pragma once
-
-#include <AK/ByteBuffer.h>
-#include <AK/Debug.h>
-#include <AK/Format.h>
-#include <AK/String.h>
-#include <string.h>
-
-namespace SQL {
-
-inline void dump(u8 const* ptr, size_t sz, String const& prefix)
-{
-    StringBuilder builder;
-    builder.appendff("{0} {1:04x} | ", prefix, sz);
-    Vector<String> bytes;
-    for (auto ix = 0u; ix < sz; ++ix) {
-        bytes.append(String::formatted("{0:02x}", *(ptr + ix)));
-    }
-    StringBuilder bytes_builder;
-    bytes_builder.join(" ", bytes);
-    builder.append(bytes_builder.to_string());
-    dbgln(builder.to_string());
-}
-
-inline void write(ByteBuffer& buffer, u8 const* ptr, size_t sz)
-{
-    if constexpr (SQL_DEBUG)
-        dump(ptr, sz, "->");
-    buffer.append(ptr, sz);
-}
-
-inline u8* read(ByteBuffer& buffer, size_t& at_offset, size_t sz)
-{
-    auto buffer_ptr = buffer.offset_pointer((int)at_offset);
-    if constexpr (SQL_DEBUG)
-        dump(buffer_ptr, sz, "<-");
-    at_offset += sz;
-    return buffer_ptr;
-}
-
-template<typename T>
-void deserialize_from(ByteBuffer& buffer, size_t& at_offset, T& t)
-{
-    memcpy(&t, read(buffer, at_offset, sizeof(T)), sizeof(T));
-}
-
-template<typename T>
-void serialize_to(ByteBuffer& buffer, T const& t)
-{
-    write(buffer, (u8 const*)(&t), sizeof(T));
-}
-
-template<>
-inline void deserialize_from<String>(ByteBuffer& buffer, size_t& at_offset, String& string)
-{
-    u32 length;
-    deserialize_from(buffer, at_offset, length);
-    if (length > 0) {
-        string = String((const char*)read(buffer, at_offset, length), length);
-    } else {
-        string = "";
-    }
-}
-
-template<>
-inline void serialize_to<String>(ByteBuffer& buffer, String const& t)
-{
-    u32 number_of_bytes = min(t.length(), 64);
-    serialize_to(buffer, number_of_bytes);
-    if (t.length() > 0) {
-        write(buffer, (u8 const*)(t.characters()), number_of_bytes);
-    }
-}
-
-}
diff --git a/Userland/Libraries/LibSQL/Serializer.cpp b/Userland/Libraries/LibSQL/Serializer.cpp
new file mode 100644
index 0000000000..e4d9e59449
--- /dev/null
+++ b/Userland/Libraries/LibSQL/Serializer.cpp
@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2021, Jan de Visser <jan@de-visser.net>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#include <LibSQL/Serializer.h>
+
+namespace SQL {
+
+void Serializer::serialize(String const& text)
+{
+    serialize<u32>((u32)text.length());
+    if (!text.is_empty())
+        write((u8 const*)text.characters(), text.length());
+}
+
+void Serializer::deserialize_to(String& text)
+{
+    auto length = deserialize<u32>();
+    if (length > 0) {
+        text = String(reinterpret_cast<char const*>(read(length)), length);
+    } else {
+        text = "";
+    }
+}
+
+}
diff --git a/Userland/Libraries/LibSQL/Serializer.h b/Userland/Libraries/LibSQL/Serializer.h
new file mode 100644
index 0000000000..65e8841e93
--- /dev/null
+++ b/Userland/Libraries/LibSQL/Serializer.h
@@ -0,0 +1,176 @@
+/*
+ * Copyright (c) 2021, Jan de Visser <jan@de-visser.net>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#pragma once
+
+#include <AK/ByteBuffer.h>
+#include <AK/Debug.h>
+#include <AK/Format.h>
+#include <AK/ScopeGuard.h>
+#include <AK/String.h>
+#include <LibSQL/Forward.h>
+#include <LibSQL/Heap.h>
+#include <string.h>
+
+namespace SQL {
+
+class Serializer {
+public:
+    Serializer() = default;
+
+    Serializer(RefPtr<Heap> heap)
+        : m_heap(heap)
+    {
+    }
+
+    void get_block(u32 pointer)
+    {
+        VERIFY(m_heap.ptr() != nullptr);
+        auto buffer_or_error = m_heap->read_block(pointer);
+        if (buffer_or_error.is_error())
+            VERIFY_NOT_REACHED();
+        m_buffer = buffer_or_error.value();
+        m_current_offset = 0;
+    }
+
+    void reset()
+    {
+        m_buffer.clear();
+        m_current_offset = 0;
+    }
+
+    void rewind()
+    {
+        m_current_offset = 0;
+    }
+
+    template<typename T, typename... Args>
+    T deserialize_block(u32 pointer, Args&&... args)
+    {
+        get_block(pointer);
+        return deserialize<T>(forward<Args>(args)...);
+    }
+
+    template<typename T>
+    void deserialize_block_to(u32 pointer, T& t)
+    {
+        get_block(pointer);
+        return deserialize_to<T>(t);
+    }
+
+    template<typename T>
+    void deserialize_to(T& t)
+    {
+        if constexpr (IsArithmetic<T>)
+            memcpy(&t, read(sizeof(T)), sizeof(T));
+        else
+            t.deserialize(*this);
+    }
+
+    void deserialize_to(String& text);
+
+    template<typename T, typename... Args>
+    NonnullOwnPtr<T> make_and_deserialize(Args&&... args)
+    {
+        auto ptr = make<T>(forward<Args>(args)...);
+        ptr->deserialize(*this);
+        return ptr;
+    }
+
+    template<typename T, typename... Args>
+    NonnullRefPtr<T> adopt_and_deserialize(Args&&... args)
+    {
+        auto ptr = adopt_ref(*new T(forward<Args>(args)...));
+        ptr->deserialize(*this);
+        return ptr;
+    }
+
+    template<typename T, typename... Args>
+    T deserialize(Args&&... args)
+    {
+        T t(forward<Args>(args)...);
+        deserialize_to(t);
+        return t;
+    }
+
+    template<typename T>
+    void serialize(T const& t)
+    {
+        if constexpr (IsArithmetic<T>)
+            write((u8 const*)(&t), sizeof(T));
+        else
+            t.serialize(*this);
+    }
+
+    void serialize(String const&);
+
+    template<typename T>
+    bool serialize_and_write(T const& t, u32 pointer)
+    {
+        VERIFY(m_heap.ptr() != nullptr);
+        reset();
+        serialize<T>(t);
+        m_heap->add_to_wal(pointer, m_buffer);
+        return true;
+    }
+
+    [[nodiscard]] size_t offset() const { return m_current_offset; }
+    u32 new_record_pointer()
+    {
+        VERIFY(m_heap.ptr() != nullptr);
+        return m_heap->new_record_pointer();
+    }
+
+    bool has_block(u32 pointer) const
+    {
+        VERIFY(m_heap.ptr() != nullptr);
+        return pointer < m_heap->size();
+    }
+
+    Heap& heap()
+    {
+        VERIFY(m_heap.ptr() != nullptr);
+        return *(m_heap.ptr());
+    }
+
+private:
+    void write(u8 const* ptr, size_t sz)
+    {
+        if constexpr (SQL_DEBUG)
+            dump(ptr, sz, "(out) =>");
+        m_buffer.append(ptr, sz);
+        m_current_offset += sz;
+    }
+
+    u8 const* read(size_t sz)
+    {
+        auto buffer_ptr = m_buffer.offset_pointer((int)m_current_offset);
+        if constexpr (SQL_DEBUG)
+            dump(buffer_ptr, sz, "<= (in)");
+        m_current_offset += sz;
+        return buffer_ptr;
+    }
+
+    static void dump(u8 const* ptr, size_t sz, String const& prefix)
+    {
+        StringBuilder builder;
+        builder.appendff("{0} {1:04x} | ", prefix, sz);
+        Vector<String> bytes;
+        for (auto ix = 0u; ix < sz; ++ix) {
+            bytes.append(String::formatted("{0:02x}", *(ptr + ix)));
+        }
+        StringBuilder bytes_builder;
+        bytes_builder.join(" ", bytes);
+        builder.append(bytes_builder.to_string());
+        dbgln(builder.to_string());
+    }
+
+    ByteBuffer m_buffer {};
+    size_t m_current_offset { 0 };
+    RefPtr<Heap> m_heap { nullptr };
+};
+
+}
diff --git a/Userland/Libraries/LibSQL/TreeNode.cpp b/Userland/Libraries/LibSQL/TreeNode.cpp
index 67cc8b779c..6ae7a1f5fb 100644
--- a/Userland/Libraries/LibSQL/TreeNode.cpp
+++ b/Userland/Libraries/LibSQL/TreeNode.cpp
@@ -9,7 +9,7 @@
 #include <AK/NonnullOwnPtr.h>
 #include <AK/StringBuilder.h>
 #include <LibSQL/BTree.h>
-#include <LibSQL/Serialize.h>
+#include <LibSQL/Serializer.h>
 
 namespace SQL {
 
@@ -53,17 +53,25 @@ DownPointer::DownPointer(DownPointer const& other)
 TreeNode* DownPointer::node()
 {
     if (!m_node)
-        inflate();
+        deserialize(m_owner->tree().serializer());
     return m_node;
 }
 
-void DownPointer::inflate()
+void DownPointer::deserialize(Serializer& serializer)
 {
     if (m_node || !m_pointer)
         return;
-    auto buffer = m_owner->tree().read_block(m_pointer);
-    size_t offset = 0;
-    m_node = make<TreeNode>(m_owner->tree(), m_owner, m_pointer, buffer, offset);
+    serializer.get_block(m_pointer);
+    m_node = serializer.make_and_deserialize<TreeNode>(m_owner->tree(), m_owner, m_pointer);
+}
+
+TreeNode::TreeNode(BTree& tree, u32 pointer)
+    : IndexNode(pointer)
+    , m_tree(tree)
+    , m_up(nullptr)
+    , m_entries()
+    , m_down()
+{
 }
 
 TreeNode::TreeNode(BTree& tree, TreeNode* up, u32 pointer)
@@ -103,36 +111,55 @@ TreeNode::TreeNode(BTree& tree, TreeNode* up, TreeNode* left, u32 pointer)
     m_is_leaf = left->pointer() == 0;
 }
 
-TreeNode::TreeNode(BTree& tree, TreeNode* up, u32 pointer, ByteBuffer& buffer, size_t& at_offset)
-    : IndexNode(pointer)
-    , m_tree(tree)
-    , m_up(up)
-    , m_entries()
-    , m_down()
+void TreeNode::deserialize(Serializer& serializer)
 {
-    u32 nodes;
-    deserialize_from<u32>(buffer, at_offset, nodes);
+    auto nodes = serializer.deserialize<u32>();
     dbgln_if(SQL_DEBUG, "Deserializing node. Size {}", nodes);
     if (nodes > 0) {
         for (u32 i = 0; i < nodes; i++) {
-            u32 left;
-            deserialize_from<u32>(buffer, at_offset, left);
+            auto left = serializer.deserialize<u32>();
             dbgln_if(SQL_DEBUG, "Down[{}] {}", i, left);
             if (!m_down.is_empty())
                 VERIFY((left == 0) == m_is_leaf);
             else
                 m_is_leaf = (left == 0);
-            m_entries.append(Key(m_tree.descriptor(), buffer, at_offset));
+            m_entries.append(serializer.deserialize<Key>(m_tree.descriptor()));
             m_down.empend(this, left);
         }
-        u32 right;
-        deserialize_from<u32>(buffer, at_offset, right);
+        auto right = serializer.deserialize<u32>();
         dbgln_if(SQL_DEBUG, "Right {}", right);
         VERIFY((right == 0) == m_is_leaf);
         m_down.empend(this, right);
     }
 }
 
+void TreeNode::serialize(Serializer& serializer) const
+{
+    u32 sz = size();
+    serializer.serialize<u32>(sz);
+    if (sz > 0) {
+        for (auto ix = 0u; ix < size(); ix++) {
+            auto& entry = m_entries[ix];
+            dbgln_if(SQL_DEBUG, "Serializing Left[{}] = {}", ix, m_down[ix].pointer());
+            serializer.serialize<u32>(is_leaf() ? 0u : m_down[ix].pointer());
+            serializer.serialize<Key>(entry);
+        }
+        dbgln_if(SQL_DEBUG, "Serializing Right = {}", m_down[size()].pointer());
+        serializer.serialize<u32>(is_leaf() ? 0u : m_down[size()].pointer());
+    }
+}
+
+size_t TreeNode::length() const
+{
+    if (!size())
+        return 0;
+    size_t len = sizeof(u32);
+    for (auto& key : m_entries) {
+        len += sizeof(u32) + key.length();
+    }
+    return len;
+}
+
 bool TreeNode::insert(Key const& key)
 {
     dbgln_if(SQL_DEBUG, "[#{}] INSERT({})", pointer(), key.to_string());
@@ -154,7 +181,7 @@ bool TreeNode::update_key_pointer(Key const& key)
             if (m_entries[ix].pointer() != key.pointer()) {
                 m_entries[ix].set_pointer(key.pointer());
                 dump_if(SQL_DEBUG, "To WAL");
-                tree().add_to_write_ahead_log(this);
+                tree().serializer().serialize_and_write<TreeNode>(*this, pointer());
             }
             return true;
         }
@@ -179,16 +206,6 @@ bool TreeNode::insert_in_leaf(Key const& key)
     return true;
 }
 
-size_t TreeNode::max_keys_in_node()
-{
-    auto descriptor = m_tree.descriptor();
-    auto key_size = descriptor->data_length() + sizeof(u32);
-    auto ret = (BLOCKSIZE - 2 * sizeof(u32)) / key_size;
-    if ((ret % 2) == 0)
-        --ret;
-    return ret;
-}
-
 Key const& TreeNode::operator[](size_t ix) const
 {
     VERIFY(ix < size());
@@ -263,22 +280,6 @@ Optional<u32> TreeNode::get(Key& key)
     return down_node(size())->get(key);
 }
 
-void TreeNode::serialize(ByteBuffer& buffer) const
-{
-    u32 sz = size();
-    serialize_to<u32>(buffer, sz);
-    if (sz > 0) {
-        for (auto ix = 0u; ix < size(); ix++) {
-            auto& entry = m_entries[ix];
-            dbgln_if(SQL_DEBUG, "Serializing Left[{}] = {}", ix, m_down[ix].pointer());
-            serialize_to<u32>(buffer, is_leaf() ? 0u : m_down[ix].pointer());
-            entry.serialize(buffer);
-        }
-        dbgln_if(SQL_DEBUG, "Serializing Right = {}", m_down[size()].pointer());
-        serialize_to<u32>(buffer, is_leaf() ? 0u : m_down[size()].pointer());
-    }
-}
-
 void TreeNode::just_insert(Key const& key, TreeNode* right)
 {
     dbgln_if(SQL_DEBUG, "[#{}] just_insert({}, right = {})",
@@ -289,11 +290,11 @@ void TreeNode::just_insert(Key const& key, TreeNode* right)
             m_entries.insert(ix, key);
             VERIFY(is_leaf() == (right == nullptr));
             m_down.insert(ix + 1, DownPointer(this, right));
-            if (size() > max_keys_in_node()) {
+            if (length() > BLOCKSIZE) {
                 split();
             } else {
                 dump_if(SQL_DEBUG, "To WAL");
-                tree().add_to_write_ahead_log(this);
+                tree().serializer().serialize_and_write(*this, pointer());
             }
             return;
         }
@@ -301,11 +302,11 @@ void TreeNode::just_insert(Key const& key, TreeNode* right)
     m_entries.append(key);
     m_down.empend(this, right);
 
-    if (size() > max_keys_in_node()) {
+    if (length() > BLOCKSIZE) {
         split();
     } else {
         dump_if(SQL_DEBUG, "To WAL");
-        tree().add_to_write_ahead_log(this);
+        tree().serializer().serialize_and_write(*this, pointer());
     }
 }
 
@@ -317,15 +318,18 @@ void TreeNode::split()
         m_up = m_tree.new_root();
 
     // Take the left pointer for the new node:
-    DownPointer left = m_down.take(max_keys_in_node() / 2 + 1);
+    auto median_index = size() / 2;
+    if (!(size() % 2))
+        ++median_index;
+    DownPointer left = m_down.take(median_index);
 
     // Create the new right node:
     auto* new_node = new TreeNode(tree(), m_up, left);
 
     // Move the rightmost keys from this node to the new right node:
-    while (m_entries.size() > max_keys_in_node() / 2 + 1) {
-        auto entry = m_entries.take(max_keys_in_node() / 2 + 1);
-        auto down = m_down.take(max_keys_in_node() / 2 + 1);
+    while (m_entries.size() > median_index) {
+        auto entry = m_entries.take(median_index);
+        auto down = m_down.take(median_index);
 
         // Reparent to new right node:
         if (down.m_node != nullptr) {
@@ -340,9 +344,9 @@ void TreeNode::split()
     auto median = m_entries.take_last();
 
     dump_if(SQL_DEBUG, "Split Left To WAL");
-    tree().add_to_write_ahead_log(this);
+    tree().serializer().serialize_and_write(*this, pointer());
     new_node->dump_if(SQL_DEBUG, "Split Right to WAL");
-    tree().add_to_write_ahead_log(new_node);
+    tree().serializer().serialize_and_write(*new_node, pointer());
 
     m_up->just_insert(median, new_node);
 }
diff --git a/Userland/Libraries/LibSQL/Tuple.cpp b/Userland/Libraries/LibSQL/Tuple.cpp
index 8706a07d50..96d98ae906 100644
--- a/Userland/Libraries/LibSQL/Tuple.cpp
+++ b/Userland/Libraries/LibSQL/Tuple.cpp
@@ -8,7 +8,7 @@
 
 #include <AK/String.h>
 #include <AK/StringBuilder.h>
-#include <LibSQL/Serialize.h>
+#include <LibSQL/Serializer.h>
 #include <LibSQL/Tuple.h>
 #include <LibSQL/TupleDescriptor.h>
 #include <LibSQL/Value.h>
@@ -31,44 +31,36 @@ Tuple::Tuple(NonnullRefPtr<TupleDescriptor> const& descriptor, u32 pointer)
     }
 }
 
-Tuple::Tuple(NonnullRefPtr<TupleDescriptor> const& descriptor, ByteBuffer& buffer, size_t& offset)
+Tuple::Tuple(NonnullRefPtr<TupleDescriptor> const& descriptor, Serializer& serializer)
     : Tuple(descriptor)
 {
-    deserialize(buffer, offset);
+    deserialize(serializer);
 }
 
-Tuple::Tuple(NonnullRefPtr<TupleDescriptor> const& descriptor, ByteBuffer& buffer)
-    : Tuple(descriptor)
+void Tuple::deserialize(Serializer& serializer)
 {
-    size_t offset = 0;
-    deserialize(buffer, offset);
-}
-
-void Tuple::deserialize(ByteBuffer& buffer, size_t& offset)
-{
-    dbgln_if(SQL_DEBUG, "deserialize tuple at offset {}", offset);
-    deserialize_from<u32>(buffer, offset, m_pointer);
+    dbgln_if(SQL_DEBUG, "deserialize tuple at offset {}", serializer.offset());
+    serializer.deserialize_to<u32>(m_pointer);
     dbgln_if(SQL_DEBUG, "pointer: {}", m_pointer);
+    auto sz = serializer.deserialize<u32>();
     m_data.clear();
-    for (auto& part : *m_descriptor) {
-        m_data.append(Value::deserialize_from(buffer, offset));
-        dbgln_if(SQL_DEBUG, "Deserialized element {} = {}", part.name, m_data.last().to_string());
+    m_descriptor->clear();
+    for (auto ix = 0u; ix < sz; ++ix) {
+        m_descriptor->append(serializer.deserialize<TupleElementDescriptor>());
+        m_data.append(serializer.deserialize<Value>());
     }
 }
 
-void Tuple::serialize(ByteBuffer& buffer) const
+void Tuple::serialize(Serializer& serializer) const
 {
     VERIFY(m_descriptor->size() == m_data.size());
     dbgln_if(SQL_DEBUG, "Serializing tuple pointer {}", pointer());
-    serialize_to<u32>(buffer, pointer());
+    serializer.serialize<u32>(pointer());
+    serializer.serialize<u32>((u32)m_descriptor->size());
     for (auto ix = 0u; ix < m_descriptor->size(); ix++) {
         auto& key_part = m_data[ix];
-        if constexpr (SQL_DEBUG) {
-            auto key_string = key_part.to_string();
-            auto& key_part_definition = (*m_descriptor)[ix];
-            dbgln("Serialized part {} = {}", key_part_definition.name, key_string);
-        }
-        key_part.serialize_to(buffer);
+        serializer.serialize<TupleElementDescriptor>((*m_descriptor)[ix]);
+        serializer.serialize<Value>(key_part);
     }
 }
 
@@ -158,6 +150,18 @@ bool Tuple::is_compatible(Tuple const& other) const
     return true;
 }
 
+size_t Tuple::length() const
+{
+    size_t len = 2 * sizeof(u32);
+    for (auto ix = 0u; ix < m_descriptor->size(); ix++) {
+        auto& descriptor = (*m_descriptor)[ix];
+        auto& value = m_data[ix];
+        len += descriptor.length();
+        len += value.length();
+    }
+    return len;
+}
+
 String Tuple::to_string() const
 {
     StringBuilder builder;
@@ -186,7 +190,7 @@ void Tuple::copy_from(const Tuple& other)
 {
     if (*m_descriptor != *other.m_descriptor) {
         m_descriptor->clear();
-        for (TupleElement const& part : *other.m_descriptor) {
+        for (TupleElementDescriptor const& part : *other.m_descriptor) {
             m_descriptor->append(part);
         }
     }
diff --git a/Userland/Libraries/LibSQL/Tuple.h b/Userland/Libraries/LibSQL/Tuple.h
index 5447e9bc6d..8831fe88db 100644
--- a/Userland/Libraries/LibSQL/Tuple.h
+++ b/Userland/Libraries/LibSQL/Tuple.h
@@ -29,8 +29,7 @@ class Tuple {
 public:
     Tuple();
     explicit Tuple(NonnullRefPtr<TupleDescriptor> const&, u32 pointer = 0);
-    Tuple(NonnullRefPtr<TupleDescriptor> const&, ByteBuffer&, size_t&);
-    Tuple(NonnullRefPtr<TupleDescriptor> const&, ByteBuffer&);
+    Tuple(NonnullRefPtr<TupleDescriptor> const&, Serializer&);
     Tuple(Tuple const&);
     virtual ~Tuple() = default;
 
@@ -62,23 +61,25 @@ public:
     void set_pointer(u32 ptr) { m_pointer = ptr; }
 
     [[nodiscard]] size_t size() const { return m_data.size(); }
-    [[nodiscard]] size_t length() const { return m_descriptor->size(); }
+    [[nodiscard]] virtual size_t length() const;
+    void clear() { m_descriptor->clear(); }
     [[nodiscard]] NonnullRefPtr<TupleDescriptor> descriptor() const { return m_descriptor; }
     [[nodiscard]] int compare(Tuple const&) const;
     [[nodiscard]] int match(Tuple const&) const;
     [[nodiscard]] u32 hash() const;
-    virtual void serialize(ByteBuffer&) const;
-    [[nodiscard]] virtual size_t data_length() const { return descriptor()->data_length(); }
 
 protected:
     [[nodiscard]] Optional<size_t> index_of(String) const;
     void copy_from(Tuple const&);
-    void deserialize(ByteBuffer&, size_t&);
+    virtual void serialize(Serializer&) const;
+    virtual void deserialize(Serializer&);
 
 private:
     NonnullRefPtr<TupleDescriptor> m_descriptor;
     Vector<Value> m_data;
-    u32 m_pointer { 0 };
+    u32 m_pointer { 2 * sizeof(u32) };
+
+    friend Serializer;
 };
 
 }
diff --git a/Userland/Libraries/LibSQL/TupleDescriptor.h b/Userland/Libraries/LibSQL/TupleDescriptor.h
index afb93a00f7..ddbb68830b 100644
--- a/Userland/Libraries/LibSQL/TupleDescriptor.h
+++ b/Userland/Libraries/LibSQL/TupleDescriptor.h
@@ -7,34 +7,44 @@
 #pragma once
 
 #include <AK/Vector.h>
+#include <LibSQL/Serializer.h>
 #include <LibSQL/Type.h>
 
 namespace SQL {
 
-struct TupleElement {
+struct TupleElementDescriptor {
     String name { "" };
     SQLType type { SQLType::Text };
     Order order { Order::Ascending };
 
-    bool operator==(TupleElement const&) const = default;
+    bool operator==(TupleElementDescriptor const&) const = default;
+
+    void serialize(Serializer& serializer) const
+    {
+        serializer.serialize(name);
+        serializer.serialize<u8>((u8)type);
+        serializer.serialize<u8>((u8)order);
+    }
+    void deserialize(Serializer& serializer)
+    {
+        name = serializer.deserialize<String>();
+        type = (SQLType)serializer.deserialize<u8>();
+        order = (Order)serializer.deserialize<u8>();
+    }
+
+    size_t length() const
+    {
+        return (sizeof(u32) + name.length()) + 2 * sizeof(u8);
+    }
 };
 
 class TupleDescriptor
-    : public Vector<TupleElement>
+    : public Vector<TupleElementDescriptor>
     , public RefCounted<TupleDescriptor> {
 public:
     TupleDescriptor() = default;
     ~TupleDescriptor() = default;
 
-    [[nodiscard]] size_t data_length() const
-    {
-        size_t sz = sizeof(u32);
-        for (auto& part : *this) {
-            sz += size_of(part.type);
-        }
-        return sz;
-    }
-
     [[nodiscard]] int compare_ignoring_names(TupleDescriptor const& other) const
     {
         if (size() != other.size())
@@ -49,7 +59,32 @@ public:
         return 0;
     }
 
-    using Vector<TupleElement>::operator==;
+    void serialize(Serializer& serializer) const
+    {
+        serializer.serialize<u32>(size());
+        for (auto& element : *this) {
+            serializer.serialize<TupleElementDescriptor>(element);
+        }
+    }
+
+    void deserialize(Serializer& serializer)
+    {
+        auto sz = serializer.deserialize<u32>();
+        for (auto ix = 0u; ix < sz; ix++) {
+            append(serializer.deserialize<TupleElementDescriptor>());
+        }
+    }
+
+    size_t length() const
+    {
+        size_t len = sizeof(u32);
+        for (auto& element : *this) {
+            len += element.length();
+        }
+        return len;
+    }
+
+    using Vector<TupleElementDescriptor>::operator==;
 };
 
 }
diff --git a/Userland/Libraries/LibSQL/Value.cpp b/Userland/Libraries/LibSQL/Value.cpp
index c415a130f1..e30e402381 100644
--- a/Userland/Libraries/LibSQL/Value.cpp
+++ b/Userland/Libraries/LibSQL/Value.cpp
@@ -4,7 +4,7 @@
  * SPDX-License-Identifier: BSD-2-Clause
  */
 
-#include <LibSQL/Serialize.h>
+#include <LibSQL/Serializer.h>
 #include <LibSQL/Value.h>
 #include <math.h>
 #include <string.h>
@@ -377,25 +377,19 @@ bool Value::operator>=(Value const& other) const
     return compare(other) >= 0;
 }
 
-void Value::serialize_to(ByteBuffer& buffer) const
+void Value::serialize(Serializer& serializer) const
 {
     u8 type_flags = (u8)type();
     if (is_null())
         type_flags |= (u8)SQLType::Null;
-    SQL::serialize_to(buffer, type_flags);
+    serializer.serialize<u8>(type_flags);
     if (!is_null())
-        m_impl.visit([&](auto& impl) { impl.serialize(buffer); });
+        m_impl.visit([&](auto& impl) { serializer.serialize(impl); });
 }
 
-void Value::deserialize(ByteBuffer& buffer, size_t& offset_at)
+void Value::deserialize(Serializer& serializer)
 {
-    m_impl.visit([&](auto& impl) { impl.deserialize(buffer, offset_at); });
-}
-
-Value Value::deserialize_from(ByteBuffer& buffer, size_t& at_offset)
-{
-    u8 type_flags;
-    SQL::deserialize_from(buffer, at_offset, type_flags);
+    auto type_flags = serializer.deserialize<u8>();
     bool is_null = false;
     if ((type_flags & (u8)SQLType::Null) && (type_flags != (u8)SQLType::Null)) {
         type_flags &= ~((u8)SQLType::Null);
@@ -403,15 +397,12 @@ Value Value::deserialize_from(ByteBuffer& buffer, size_t& at_offset)
     }
     auto type = (SQLType)type_flags;
     VERIFY(!is_null || (type != SQLType::Tuple && type != SQLType::Array));
-    Value ret(type);
+    setup(type);
     if (!is_null) {
-        ret.deserialize(buffer, at_offset);
+        m_impl.visit([&](auto& impl) { impl.deserialize(serializer); });
     }
-    return ret;
 }
 
-// -----------------------------------------------------------------
-
 bool NullImpl::can_cast(Value const& value)
 {
     return value.is_null();
@@ -422,8 +413,6 @@ int NullImpl::compare(Value const& other)
     return other.type() == SQLType::Null;
 }
 
-// -----------------------------------------------------------------
-
 String TextImpl::to_string() const
 {
     return value();
@@ -490,7 +479,7 @@ void TextImpl::assign_bool(bool bool_value)
 
 size_t TextImpl::length() const
 {
-    return (is_null()) ? 0 : sizeof(u32) + min(value().length(), 64) + 1;
+    return (is_null()) ? 0 : sizeof(u32) + value().length();
 }
 
 int TextImpl::compare(Value const& other) const
@@ -777,32 +766,38 @@ bool ContainerValueImpl::append(Value const& value)
 
 bool ContainerValueImpl::append(BaseTypeImpl const& impl)
 {
-    if (m_max_size.has_value() && (size() >= m_max_size.value()))
-        return false;
     if (!validate(impl))
         return false;
     m_value.value().empend(impl);
     return true;
 }
 
-void ContainerValueImpl::serialize_values(ByteBuffer& buffer) const
+void ContainerValueImpl::serialize_values(Serializer& serializer) const
 {
-    serialize_to(buffer, (u32)size());
-    for (auto& value : value()) {
-        Value(value).serialize_to(buffer);
+    serializer.serialize((u32)size());
+    for (auto& impl : value()) {
+        serializer.serialize<Value>(Value(impl));
     }
 }
 
-void ContainerValueImpl::deserialize_values(ByteBuffer& buffer, size_t& at_offset)
+void ContainerValueImpl::deserialize_values(Serializer& serializer)
 {
-    u32 sz;
-    deserialize_from(buffer, at_offset, sz);
+    auto sz = serializer.deserialize<u32>();
     m_value = Vector<BaseTypeImpl>();
     for (auto ix = 0u; ix < sz; ix++) {
-        append(Value::deserialize_from(buffer, at_offset));
+        append(serializer.deserialize<Value>());
     }
 }
 
+size_t ContainerValueImpl::length() const
+{
+    size_t len = sizeof(u32);
+    for (auto& impl : value()) {
+        len += Value(impl).length();
+    }
+    return len;
+}
+
 void TupleImpl::assign(Value const& other)
 {
     if (other.type() != SQLType::Tuple) {
@@ -820,7 +815,7 @@ void TupleImpl::assign(Value const& other)
 
 size_t TupleImpl::length() const
 {
-    return (m_descriptor) ? m_descriptor->data_length() : 0;
+    return m_descriptor->length() + ContainerValueImpl::length();
 }
 
 bool TupleImpl::can_cast(Value const& other_value) const
@@ -853,55 +848,61 @@ int TupleImpl::compare(Value const& other) const
     return 0;
 }
 
-void TupleImpl::serialize(ByteBuffer& buffer) const
+void TupleImpl::serialize(Serializer& serializer) const
 {
-    if (m_descriptor) {
-        serialize_to(buffer, (u32)m_descriptor->size());
-        for (auto& tuple_element : *m_descriptor) {
-            u8 elem_type = (u8)tuple_element.type;
-            serialize_to(buffer, elem_type);
-            u8 elem_order = (u8)tuple_element.order;
-            serialize_to(buffer, elem_order);
-        }
-    } else {
-        serialize_to(buffer, (u32)-1);
+    serializer.serialize<TupleDescriptor>(*m_descriptor);
+    serialize_values(serializer);
+}
+
+void TupleImpl::deserialize(Serializer& serializer)
+{
+    m_descriptor = serializer.adopt_and_deserialize<TupleDescriptor>();
+    deserialize_values(serializer);
+}
+
+void TupleImpl::infer_descriptor()
+{
+    if (!m_descriptor) {
+        m_descriptor = adopt_ref(*new TupleDescriptor);
+        m_descriptor_inferred = true;
     }
-    serialize_values(buffer);
 }
 
-void TupleImpl::deserialize(ByteBuffer& buffer, size_t& at_offset)
+void TupleImpl::extend_descriptor(Value const& value)
 {
-    u32 sz;
-    deserialize_from(buffer, at_offset, sz);
-    if (sz != (u32)-1) {
-        NonnullRefPtr<TupleDescriptor> serialized_descriptor = adopt_ref(*new TupleDescriptor);
-        for (auto ix = 0u; ix < sz; ix++) {
-            u8 elem_type, elem_order;
-            deserialize_from(buffer, at_offset, elem_type);
-            deserialize_from(buffer, at_offset, elem_order);
-            serialized_descriptor->empend("", (SQLType)elem_type, (Order)elem_order);
-        }
-        m_descriptor = serialized_descriptor;
-        m_max_size = m_descriptor->size();
-    } else {
+    VERIFY(m_descriptor_inferred);
+    m_descriptor->empend("", value.type(), Order::Ascending);
+}
+
+bool TupleImpl::validate_before_assignment(Vector<Value> const& values)
+{
+    if (m_descriptor_inferred)
         m_descriptor = nullptr;
-        m_max_size = {};
+    if (!m_descriptor) {
+        infer_descriptor();
+        if (values.size() > m_descriptor->size()) {
+            for (auto ix = m_descriptor->size(); ix < values.size(); ix++) {
+                extend_descriptor(values[ix]);
+            }
+        }
     }
-    deserialize_values(buffer, at_offset);
+    return true;
 }
 
 bool TupleImpl::validate(BaseTypeImpl const& value)
 {
     if (!m_descriptor)
-        return true;
-    auto required_type = (*m_descriptor)[size()].type;
+        infer_descriptor();
+    if (m_descriptor_inferred && (this->value().size() == m_descriptor->size()))
+        extend_descriptor(Value(value));
+    if (m_descriptor->size() == this->value().size())
+        return false;
+    auto required_type = (*m_descriptor)[this->value().size()].type;
     return Value(value).type() == required_type;
 }
 
 bool TupleImpl::validate_after_assignment()
 {
-    if (!m_descriptor)
-        return true;
     for (auto ix = value().size(); ix < m_descriptor->size(); ++ix) {
         auto required_type = (*m_descriptor)[ix].type;
         append(Value(required_type));
@@ -925,11 +926,7 @@ void ArrayImpl::assign(Value const& other)
 
 size_t ArrayImpl::length() const
 {
-    size_t ret = 0;
-    for (auto& value : value()) {
-        ret += Value(value).length();
-    }
-    return ret;
+    return sizeof(u8) + sizeof(u32) + ContainerValueImpl::length();
 }
 
 bool ArrayImpl::can_cast(Value const& other_value) const
@@ -960,32 +957,31 @@ int ArrayImpl::compare(Value const& other) const
     return 0;
 }
 
-void ArrayImpl::serialize(ByteBuffer& buffer) const
+void ArrayImpl::serialize(Serializer& serializer) const
 {
-    serialize_to(buffer, (u8)m_element_type);
+    serializer.serialize((u8)m_element_type);
     if (m_max_size.has_value())
-        serialize_to(buffer, (u32)m_max_size.value());
+        serializer.serialize((u32)m_max_size.value());
     else
-        serialize_to(buffer, (u32)0);
-    serialize_values(buffer);
+        serializer.serialize((u32)0);
+    serialize_values(serializer);
 }
 
-void ArrayImpl::deserialize(ByteBuffer& buffer, size_t& at_offset)
+void ArrayImpl::deserialize(Serializer& serializer)
 {
-    u8 elem_type;
-    deserialize_from(buffer, at_offset, elem_type);
-    m_element_type = (SQLType)elem_type;
-    u32 max_sz;
-    deserialize_from(buffer, at_offset, max_sz);
+    m_element_type = (SQLType)serializer.deserialize<u8>();
+    auto max_sz = serializer.deserialize<u32>();
     if (max_sz)
         m_max_size = max_sz;
     else
         m_max_size = {};
-    deserialize_values(buffer, at_offset);
+    deserialize_values(serializer);
 }
 
 bool ArrayImpl::validate(BaseTypeImpl const& impl)
 {
+    if (m_max_size.has_value() && (size() >= m_max_size.value()))
+        return false;
     return Value(impl).type() == m_element_type;
 }
 
diff --git a/Userland/Libraries/LibSQL/Value.h b/Userland/Libraries/LibSQL/Value.h
index 624e32e4ae..76233c4248 100644
--- a/Userland/Libraries/LibSQL/Value.h
+++ b/Userland/Libraries/LibSQL/Value.h
@@ -11,287 +11,16 @@
 #include <AK/ScopeGuard.h>
 #include <AK/String.h>
 #include <AK/Variant.h>
-#include <LibSQL/Serialize.h>
+#include <LibSQL/Forward.h>
 #include <LibSQL/TupleDescriptor.h>
 #include <LibSQL/Type.h>
+#include <LibSQL/ValueImpl.h>
 #include <string.h>
 
 namespace SQL {
 
-class Value;
-
-class BaseImpl {
-public:
-    explicit BaseImpl(SQLType type = SQLType::Null)
-        : m_type(type)
-    {
-    }
-
-    [[nodiscard]] SQLType type() const { return m_type; }
-    [[nodiscard]] String type_name() const { return SQLType_name(type()); }
-
-private:
-    SQLType m_type { SQLType::Null };
-};
-
-class NullImpl : public BaseImpl {
-public:
-    explicit NullImpl()
-        : BaseImpl(SQLType::Null)
-    {
-    }
-
-    [[nodiscard]] static bool is_null() { return true; }
-    [[nodiscard]] static String to_string() { return "(null)"; }
-    [[nodiscard]] static Optional<int> to_int() { return {}; }
-    [[nodiscard]] static Optional<double> to_double() { return {}; }
-    [[nodiscard]] static Optional<bool> to_bool() { return {}; }
-    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
-    static void assign(Value const&) { }
-    static void assign_string(String const&) { }
-    static void assign_int(int) { }
-    static void assign_double(double) { }
-    static void assign_bool(bool) { }
-    static void assign_vector(Vector<Value> const&) { }
-    [[nodiscard]] static size_t length() { return 0; }
-    [[nodiscard]] static bool can_cast(Value const&);
-    [[nodiscard]] static int compare(Value const&);
-    static void serialize(ByteBuffer&) { }
-    static void deserialize(ByteBuffer&, size_t&) { }
-    [[nodiscard]] static u32 hash() { return 0; }
-};
-
-template<typename T>
-class Impl : public BaseImpl {
-public:
-    [[nodiscard]] bool is_null() const
-    {
-        return !m_value.has_value();
-    }
-
-    [[nodiscard]] T const& value() const
-    {
-        VERIFY(m_value.has_value());
-        return m_value.value();
-    }
-
-    [[nodiscard]] size_t length() const
-    {
-        return sizeof(T);
-    }
-
-    void serialize(ByteBuffer& buffer) const
-    {
-        serialize_to(buffer, value());
-    }
-
-    void deserialize(ByteBuffer& buffer, size_t& at_offset)
-    {
-        T value;
-        deserialize_from(buffer, at_offset, value);
-        m_value = value;
-    }
-
-protected:
-    explicit Impl(SQLType sql_type)
-        : BaseImpl(sql_type)
-    {
-    }
-
-    Optional<T> m_value {};
-};
-
-class TextImpl : public Impl<String> {
-public:
-    explicit TextImpl()
-        : Impl(SQLType::Text)
-    {
-    }
-
-    [[nodiscard]] String to_string() const;
-    [[nodiscard]] Optional<int> to_int() const;
-    [[nodiscard]] Optional<double> to_double() const;
-    [[nodiscard]] Optional<bool> to_bool() const;
-    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
-    void assign(Value const&);
-    void assign_string(String const&);
-    void assign_int(int);
-    void assign_double(double);
-    void assign_bool(bool);
-    void assign_vector(Vector<Value> const&) { m_value = {}; }
-    [[nodiscard]] size_t length() const;
-    [[nodiscard]] static bool can_cast(Value const&) { return true; }
-    [[nodiscard]] int compare(Value const& other) const;
-    [[nodiscard]] u32 hash() const;
-};
-
-class IntegerImpl : public Impl<int> {
-public:
-    IntegerImpl()
-        : Impl(SQLType::Integer)
-    {
-    }
-
-    [[nodiscard]] String to_string() const;
-    [[nodiscard]] Optional<int> to_int() const;
-    [[nodiscard]] Optional<double> to_double() const;
-    [[nodiscard]] Optional<bool> to_bool() const;
-    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
-    void assign(Value const&);
-    void assign_string(String const&);
-    void assign_int(int);
-    void assign_double(double);
-    void assign_bool(bool);
-    void assign_vector(Vector<Value> const&) { m_value = {}; }
-    [[nodiscard]] static bool can_cast(Value const&);
-    [[nodiscard]] int compare(Value const& other) const;
-    [[nodiscard]] u32 hash() const;
-};
-
-class FloatImpl : public Impl<double> {
-public:
-    explicit FloatImpl()
-        : Impl(SQLType::Float)
-    {
-    }
-
-    [[nodiscard]] String to_string() const;
-    [[nodiscard]] Optional<int> to_int() const;
-    [[nodiscard]] Optional<double> to_double() const;
-    [[nodiscard]] static Optional<bool> to_bool() { return {}; }
-    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
-    void assign(Value const&);
-    void assign_string(String const&);
-    void assign_int(int);
-    void assign_double(double);
-    void assign_bool(bool) { m_value = {}; }
-    void assign_vector(Vector<Value> const&) { m_value = {}; }
-    [[nodiscard]] static bool can_cast(Value const&);
-    [[nodiscard]] int compare(Value const& other) const;
-
-    // Using floats in hash functions is a bad idea. Let's disable that for now.
-    [[nodiscard]] static u32 hash() { VERIFY_NOT_REACHED(); }
-};
-
-class BooleanImpl : public Impl<bool> {
-public:
-    explicit BooleanImpl()
-        : Impl(SQLType::Boolean)
-    {
-    }
-
-    [[nodiscard]] String to_string() const;
-    [[nodiscard]] Optional<int> to_int() const;
-    [[nodiscard]] static Optional<double> to_double();
-    [[nodiscard]] Optional<bool> to_bool() const;
-    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
-    void assign(Value const&);
-    void assign_string(String const&);
-    void assign_int(int);
-    void assign_double(double);
-    void assign_bool(bool);
-    void assign_vector(Vector<Value> const&) { m_value = {}; }
-    [[nodiscard]] static bool can_cast(Value const&);
-    [[nodiscard]] int compare(Value const& other) const;
-    [[nodiscard]] u32 hash() const;
-};
-
-using BaseTypeImpl = Variant<NullImpl, TextImpl, IntegerImpl, FloatImpl, BooleanImpl>;
-
-class ContainerValueImpl : public Impl<Vector<BaseTypeImpl>> {
-public:
-    virtual ~ContainerValueImpl() = default;
-
-    [[nodiscard]] String to_string() const;
-    [[nodiscard]] static Optional<int> to_int() { return {}; }
-    [[nodiscard]] static Optional<double> to_double() { return {}; }
-    [[nodiscard]] static Optional<bool> to_bool() { return {}; }
-    [[nodiscard]] bool to_vector(Vector<Value>&) const;
-    void assign_string(String const&) { m_value = {}; }
-    void assign_int(int) { m_value = {}; }
-    void assign_double(double) { m_value = {}; }
-    void assign_bool(bool) { m_value = {}; }
-    void assign_vector(Vector<Value> const&);
-    [[nodiscard]] u32 hash() const;
-
-    virtual bool validate_before_assignment(Vector<Value> const&) { return true; }
-    virtual bool validate(BaseTypeImpl const&) { return true; }
-    virtual bool validate_after_assignment() { return true; }
-    [[nodiscard]] Vector<String> to_string_vector() const;
-    [[nodiscard]] size_t size() const { return is_null() ? 0 : value().size(); }
-    bool append(Value const&);
-    bool append(BaseTypeImpl const& value);
-    void serialize_values(ByteBuffer& buffer) const;
-    void deserialize_values(ByteBuffer&, size_t& at_offset);
-
-protected:
-    explicit ContainerValueImpl(SQLType sql_type, Optional<size_t> const& max_size = {})
-        : Impl(sql_type)
-        , m_max_size(max_size)
-    {
-    }
-
-    Optional<size_t> m_max_size {};
-};
-
-class TupleImpl : public ContainerValueImpl {
-public:
-    explicit TupleImpl(NonnullRefPtr<TupleDescriptor> const& descriptor, bool is_null = true)
-        : ContainerValueImpl(SQLType::Tuple, is_null)
-        , m_descriptor(descriptor)
-    {
-        m_max_size = m_descriptor->size();
-    }
-
-    explicit TupleImpl()
-        : ContainerValueImpl(SQLType::Tuple, {})
-    {
-    }
-
-    void assign(Value const&);
-    [[nodiscard]] size_t length() const;
-    [[nodiscard]] bool can_cast(Value const&) const;
-    [[nodiscard]] int compare(Value const& other) const;
-
-    virtual bool validate(BaseTypeImpl const&) override;
-    virtual bool validate_after_assignment() override;
-    void serialize(ByteBuffer& buffer) const;
-    void deserialize(ByteBuffer& buffer, size_t&);
-
-private:
-    RefPtr<TupleDescriptor> m_descriptor;
-};
-
-class ArrayImpl : public ContainerValueImpl {
-public:
-    explicit ArrayImpl(SQLType element_type, Optional<size_t> const& max_size = {})
-        : ContainerValueImpl(SQLType::Array, max_size)
-        , m_element_type(element_type)
-    {
-    }
-
-    explicit ArrayImpl()
-        : ContainerValueImpl(SQLType::Array, {})
-        , m_element_type(SQLType::Null)
-    {
-    }
-
-    void assign(Value const&);
-    [[nodiscard]] size_t length() const;
-    [[nodiscard]] bool can_cast(Value const&) const;
-    [[nodiscard]] int compare(Value const& other) const;
-    void serialize(ByteBuffer& buffer) const;
-    void deserialize(ByteBuffer& buffer, size_t&);
-    virtual bool validate(BaseTypeImpl const&) override;
-
-private:
-    SQLType m_element_type { SQLType::Text };
-};
-
-using ValueTypeImpl = Variant<NullImpl, TextImpl, IntegerImpl, FloatImpl, BooleanImpl, TupleImpl, ArrayImpl>;
-
 /**
- * A `Value` is an atomic piece of SQL data. A `Value` has a basic type
+ * A `Value` is an atomic piece of SQL data`. A `Value` has a basic type
  * (Text/String, Integer, Float, etc). Richer types are implemented in higher
  * level layers, but the resulting data is stored in these `Value` objects.
  */
@@ -373,8 +102,8 @@ public:
     [[nodiscard]] size_t length() const;
     [[nodiscard]] u32 hash() const;
     [[nodiscard]] bool can_cast(Value const&) const;
-    void serialize_to(ByteBuffer&) const;
-    void deserialize(ByteBuffer&, size_t&);
+    void serialize(Serializer&) const;
+    void deserialize(Serializer&);
 
     [[nodiscard]] int compare(Value const&) const;
     bool operator==(Value const&) const;
@@ -390,12 +119,12 @@ public:
     static Value const& null();
     static Value create_tuple(NonnullRefPtr<TupleDescriptor> const&);
     static Value create_array(SQLType element_type, Optional<size_t> const& max_size = {});
-    static Value deserialize_from(ByteBuffer&, size_t&);
 
 private:
     void setup(SQLType type);
 
-    ValueTypeImpl m_impl {};
+    ValueTypeImpl m_impl { NullImpl() };
+    friend Serializer;
 };
 
 }
diff --git a/Userland/Libraries/LibSQL/ValueImpl.h b/Userland/Libraries/LibSQL/ValueImpl.h
new file mode 100644
index 0000000000..70fe611ed5
--- /dev/null
+++ b/Userland/Libraries/LibSQL/ValueImpl.h
@@ -0,0 +1,297 @@
+/*
+ * Copyright (c) 2021, Jan de Visser <jan@de-visser.net>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#pragma once
+
+#include <AK/Badge.h>
+#include <AK/ByteBuffer.h>
+#include <AK/ScopeGuard.h>
+#include <AK/String.h>
+#include <AK/Variant.h>
+#include <LibSQL/Forward.h>
+#include <LibSQL/Serializer.h>
+#include <LibSQL/TupleDescriptor.h>
+#include <LibSQL/Type.h>
+#include <string.h>
+
+namespace SQL {
+
+class Value;
+
+class BaseImpl {
+public:
+    explicit BaseImpl(SQLType type = SQLType::Null)
+        : m_type(type)
+    {
+    }
+
+    [[nodiscard]] SQLType type() const { return m_type; }
+    [[nodiscard]] String type_name() const { return SQLType_name(type()); }
+
+private:
+    SQLType m_type { SQLType::Null };
+};
+
+class NullImpl : public BaseImpl {
+public:
+    explicit NullImpl()
+        : BaseImpl(SQLType::Null)
+    {
+    }
+
+    [[nodiscard]] static bool is_null() { return true; }
+    [[nodiscard]] static String to_string() { return "(null)"; }
+    [[nodiscard]] static Optional<int> to_int() { return {}; }
+    [[nodiscard]] static Optional<double> to_double() { return {}; }
+    [[nodiscard]] static Optional<bool> to_bool() { return {}; }
+    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
+    static void assign(Value const&) { }
+    static void assign_string(String const&) { }
+    static void assign_int(int) { }
+    static void assign_double(double) { }
+    static void assign_bool(bool) { }
+    static void assign_vector(Vector<Value> const&) { }
+    [[nodiscard]] static size_t length() { return 0; }
+    [[nodiscard]] static bool can_cast(Value const&);
+    [[nodiscard]] static int compare(Value const&);
+    static void serialize(Serializer&) { }
+    static void deserialize(Serializer&) { }
+    [[nodiscard]] static u32 hash() { return 0; }
+};
+
+template<typename T>
+class Impl : public BaseImpl {
+public:
+    [[nodiscard]] bool is_null() const
+    {
+        return !m_value.has_value();
+    }
+
+    [[nodiscard]] T const& value() const
+    {
+        VERIFY(m_value.has_value());
+        return m_value.value();
+    }
+
+    [[nodiscard]] size_t length() const
+    {
+        return sizeof(T);
+    }
+
+    void serialize(Serializer& serializer) const
+    {
+        serializer.serialize(value());
+    }
+
+    void deserialize(Serializer& serializer)
+    {
+        T value;
+        serializer.deserialize_to(value);
+        m_value = value;
+    }
+
+protected:
+    explicit Impl(SQLType sql_type)
+        : BaseImpl(sql_type)
+    {
+    }
+
+    Optional<T> m_value {};
+};
+
+class TextImpl : public Impl<String> {
+public:
+    explicit TextImpl()
+        : Impl(SQLType::Text)
+    {
+    }
+
+    [[nodiscard]] String to_string() const;
+    [[nodiscard]] Optional<int> to_int() const;
+    [[nodiscard]] Optional<double> to_double() const;
+    [[nodiscard]] Optional<bool> to_bool() const;
+    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
+    void assign(Value const&);
+    void assign_string(String const&);
+    void assign_int(int);
+    void assign_double(double);
+    void assign_bool(bool);
+    void assign_vector(Vector<Value> const&) { m_value = {}; }
+    [[nodiscard]] size_t length() const;
+    [[nodiscard]] static bool can_cast(Value const&) { return true; }
+    [[nodiscard]] int compare(Value const& other) const;
+    [[nodiscard]] u32 hash() const;
+};
+
+class IntegerImpl : public Impl<int> {
+public:
+    IntegerImpl()
+        : Impl(SQLType::Integer)
+    {
+    }
+
+    [[nodiscard]] String to_string() const;
+    [[nodiscard]] Optional<int> to_int() const;
+    [[nodiscard]] Optional<double> to_double() const;
+    [[nodiscard]] Optional<bool> to_bool() const;
+    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
+    void assign(Value const&);
+    void assign_string(String const&);
+    void assign_int(int);
+    void assign_double(double);
+    void assign_bool(bool);
+    void assign_vector(Vector<Value> const&) { m_value = {}; }
+    [[nodiscard]] static bool can_cast(Value const&);
+    [[nodiscard]] int compare(Value const& other) const;
+    [[nodiscard]] u32 hash() const;
+};
+
+class FloatImpl : public Impl<double> {
+public:
+    explicit FloatImpl()
+        : Impl(SQLType::Float)
+    {
+    }
+
+    [[nodiscard]] String to_string() const;
+    [[nodiscard]] Optional<int> to_int() const;
+    [[nodiscard]] Optional<double> to_double() const;
+    [[nodiscard]] static Optional<bool> to_bool() { return {}; }
+    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
+    void assign(Value const&);
+    void assign_string(String const&);
+    void assign_int(int);
+    void assign_double(double);
+    void assign_bool(bool) { m_value = {}; }
+    void assign_vector(Vector<Value> const&) { m_value = {}; }
+    [[nodiscard]] static bool can_cast(Value const&);
+    [[nodiscard]] int compare(Value const& other) const;
+
+    // Using floats in hash functions is a bad idea. Let's disable that for now.
+    [[nodiscard]] static u32 hash() { VERIFY_NOT_REACHED(); }
+};
+
+class BooleanImpl : public Impl<bool> {
+public:
+    explicit BooleanImpl()
+        : Impl(SQLType::Boolean)
+    {
+    }
+
+    [[nodiscard]] String to_string() const;
+    [[nodiscard]] Optional<int> to_int() const;
+    [[nodiscard]] static Optional<double> to_double();
+    [[nodiscard]] Optional<bool> to_bool() const;
+    [[nodiscard]] static bool to_vector(Vector<Value>&) { return false; }
+    void assign(Value const&);
+    void assign_string(String const&);
+    void assign_int(int);
+    void assign_double(double);
+    void assign_bool(bool);
+    void assign_vector(Vector<Value> const&) { m_value = {}; }
+    [[nodiscard]] static bool can_cast(Value const&);
+    [[nodiscard]] int compare(Value const& other) const;
+    [[nodiscard]] u32 hash() const;
+};
+
+using BaseTypeImpl = Variant<NullImpl, TextImpl, IntegerImpl, FloatImpl, BooleanImpl>;
+
+class ContainerValueImpl : public Impl<Vector<BaseTypeImpl>> {
+public:
+    virtual ~ContainerValueImpl() = default;
+
+    [[nodiscard]] String to_string() const;
+    [[nodiscard]] static Optional<int> to_int() { return {}; }
+    [[nodiscard]] static Optional<double> to_double() { return {}; }
+    [[nodiscard]] static Optional<bool> to_bool() { return {}; }
+    [[nodiscard]] bool to_vector(Vector<Value>&) const;
+    void assign_string(String const&) { m_value = {}; }
+    void assign_int(int) { m_value = {}; }
+    void assign_double(double) { m_value = {}; }
+    void assign_bool(bool) { m_value = {}; }
+    void assign_vector(Vector<Value> const&);
+    [[nodiscard]] u32 hash() const;
+
+    virtual bool validate_before_assignment(Vector<Value> const&) { return true; }
+    virtual bool validate(BaseTypeImpl const&) { return true; }
+    virtual bool validate_after_assignment() { return true; }
+    [[nodiscard]] Vector<String> to_string_vector() const;
+    [[nodiscard]] size_t length() const;
+    [[nodiscard]] size_t size() const { return is_null() ? 0 : value().size(); }
+    bool append(Value const&);
+    bool append(BaseTypeImpl const& value);
+    void serialize_values(Serializer&) const;
+    void deserialize_values(Serializer&);
+
+protected:
+    explicit ContainerValueImpl(SQLType sql_type)
+        : Impl(sql_type)
+    {
+    }
+};
+
+class TupleImpl : public ContainerValueImpl {
+public:
+    explicit TupleImpl(NonnullRefPtr<TupleDescriptor> const& descriptor)
+        : ContainerValueImpl(SQLType::Tuple)
+        , m_descriptor(descriptor)
+    {
+    }
+
+    explicit TupleImpl()
+        : ContainerValueImpl(SQLType::Tuple)
+    {
+    }
+
+    void assign(Value const&);
+    [[nodiscard]] size_t length() const;
+    [[nodiscard]] bool can_cast(Value const&) const;
+    [[nodiscard]] int compare(Value const& other) const;
+
+    virtual bool validate_before_assignment(Vector<Value> const&) override;
+    virtual bool validate(BaseTypeImpl const&) override;
+    virtual bool validate_after_assignment() override;
+    void serialize(Serializer&) const;
+    void deserialize(Serializer&);
+
+private:
+    void infer_descriptor();
+    void extend_descriptor(Value const&);
+    RefPtr<TupleDescriptor> m_descriptor;
+    bool m_descriptor_inferred { false };
+};
+
+class ArrayImpl : public ContainerValueImpl {
+public:
+    explicit ArrayImpl(SQLType element_type, Optional<size_t> const& max_size = {})
+        : ContainerValueImpl(SQLType::Array)
+        , m_element_type(element_type)
+        , m_max_size(max_size)
+    {
+    }
+
+    explicit ArrayImpl()
+        : ContainerValueImpl(SQLType::Array)
+        , m_element_type(SQLType::Null)
+    {
+    }
+
+    void assign(Value const&);
+    [[nodiscard]] size_t length() const;
+    [[nodiscard]] bool can_cast(Value const&) const;
+    [[nodiscard]] int compare(Value const& other) const;
+    void serialize(Serializer&) const;
+    void deserialize(Serializer&);
+    virtual bool validate(BaseTypeImpl const&) override;
+
+private:
+    SQLType m_element_type { SQLType::Text };
+    Optional<size_t> m_max_size {};
+};
+
+using ValueTypeImpl = Variant<NullImpl, TextImpl, IntegerImpl, FloatImpl, BooleanImpl, TupleImpl, ArrayImpl>;
+
+}