diff --git a/Android.bp b/Android.bp
index 9d91a4e..7378bad 100644
--- a/Android.bp
+++ b/Android.bp
@@ -8241,7 +8241,6 @@
     srcs: [
         "src/trace_processor/containers/bit_vector.cc",
         "src/trace_processor/containers/bit_vector_iterators.cc",
-        "src/trace_processor/containers/nullable_vector.cc",
         "src/trace_processor/containers/row_map.cc",
         "src/trace_processor/containers/string_pool.cc",
     ],
@@ -8264,6 +8263,7 @@
     name: "perfetto_src_trace_processor_db_db",
     srcs: [
         "src/trace_processor/db/column.cc",
+        "src/trace_processor/db/column_storage.cc",
         "src/trace_processor/db/table.cc",
         "src/trace_processor/db/view.cc",
     ],
diff --git a/BUILD b/BUILD
index 0c50584..7d2bfaf 100644
--- a/BUILD
+++ b/BUILD
@@ -958,7 +958,6 @@
     srcs = [
         "src/trace_processor/containers/bit_vector.cc",
         "src/trace_processor/containers/bit_vector_iterators.cc",
-        "src/trace_processor/containers/nullable_vector.cc",
         "src/trace_processor/containers/row_map.cc",
         "src/trace_processor/containers/string_pool.cc",
     ],
@@ -988,6 +987,8 @@
         "src/trace_processor/db/base_id.h",
         "src/trace_processor/db/column.cc",
         "src/trace_processor/db/column.h",
+        "src/trace_processor/db/column_storage.cc",
+        "src/trace_processor/db/column_storage.h",
         "src/trace_processor/db/compare.h",
         "src/trace_processor/db/table.cc",
         "src/trace_processor/db/table.h",
diff --git a/src/trace_processor/containers/BUILD.gn b/src/trace_processor/containers/BUILD.gn
index 1e92c62..e4750d2 100644
--- a/src/trace_processor/containers/BUILD.gn
+++ b/src/trace_processor/containers/BUILD.gn
@@ -32,7 +32,6 @@
   sources = [
     "bit_vector.cc",
     "bit_vector_iterators.cc",
-    "nullable_vector.cc",
     "row_map.cc",
     "string_pool.cc",
   ]
diff --git a/src/trace_processor/containers/bit_vector.cc b/src/trace_processor/containers/bit_vector.cc
index f2d14e8..ca523d5 100644
--- a/src/trace_processor/containers/bit_vector.cc
+++ b/src/trace_processor/containers/bit_vector.cc
@@ -16,6 +16,8 @@
 
 #include "src/trace_processor/containers/bit_vector.h"
 
+#include <limits>
+
 #include "src/trace_processor/containers/bit_vector_iterators.h"
 
 #if PERFETTO_BUILDFLAG(PERFETTO_X64_CPU_OPT)
diff --git a/src/trace_processor/containers/nullable_vector.h b/src/trace_processor/containers/nullable_vector.h
index 44bf4f6..cb97492 100644
--- a/src/trace_processor/containers/nullable_vector.h
+++ b/src/trace_processor/containers/nullable_vector.h
@@ -28,20 +28,6 @@
 namespace perfetto {
 namespace trace_processor {
 
-// Base class for NullableVector which allows type erasure to be implemented
-// (e.g. allows for std::unique_ptr<NullableVectorBase>).
-class NullableVectorBase {
- public:
-  NullableVectorBase() = default;
-  virtual ~NullableVectorBase();
-
-  NullableVectorBase(const NullableVectorBase&) = delete;
-  NullableVectorBase& operator=(const NullableVectorBase&) = delete;
-
-  NullableVectorBase(NullableVectorBase&&) = default;
-  NullableVectorBase& operator=(NullableVectorBase&&) noexcept = default;
-};
-
 // A data structure which compactly stores a list of possibly nullable data.
 //
 // Internally, this class is implemented using a combination of a std::deque
@@ -50,7 +36,7 @@
 // BitVector at a slight cost (searching the BitVector to find the index into
 // the std::deque) when looking up the data.
 template <typename T>
-class NullableVector : public NullableVectorBase {
+class NullableVector {
  private:
   enum class Mode {
     // Sparse mode is the default mode and ensures that nulls are stored using
@@ -68,9 +54,8 @@
  public:
   // Creates an empty NullableVector.
   NullableVector() : NullableVector<T>(Mode::kSparse) {}
-  ~NullableVector() override = default;
 
-  explicit NullableVector(const NullableVector&) = delete;
+  NullableVector(const NullableVector&) = delete;
   NullableVector& operator=(const NullableVector&) = delete;
 
   NullableVector(NullableVector&&) = default;
diff --git a/src/trace_processor/db/BUILD.gn b/src/trace_processor/db/BUILD.gn
index f118943..539cde3 100644
--- a/src/trace_processor/db/BUILD.gn
+++ b/src/trace_processor/db/BUILD.gn
@@ -19,6 +19,8 @@
     "base_id.h",
     "column.cc",
     "column.h",
+    "column_storage.cc",
+    "column_storage.h",
     "compare.h",
     "table.cc",
     "table.h",
diff --git a/src/trace_processor/db/column.cc b/src/trace_processor/db/column.cc
index 98602c4..8c8842b 100644
--- a/src/trace_processor/db/column.cc
+++ b/src/trace_processor/db/column.cc
@@ -33,7 +33,7 @@
              table,
              col_idx,
              row_map_idx,
-             column.nullable_vector_) {}
+             column.storage_) {}
 
 Column::Column(const char* name,
                ColumnType type,
@@ -41,9 +41,9 @@
                Table* table,
                uint32_t col_idx_in_table,
                uint32_t row_map_idx,
-               NullableVectorBase* nv)
+               ColumnStorageBase* st)
     : type_(type),
-      nullable_vector_(nv),
+      storage_(st),
       name_(name),
       flags_(flags),
       table_(table),
@@ -51,25 +51,29 @@
       row_map_idx_(row_map_idx),
       string_pool_(table->string_pool_) {
   // Check that the dense-ness of the column and the nullable vector match.
-  switch (type_) {
-    case ColumnType::kInt32:
-      PERFETTO_DCHECK(nullable_vector<int32_t>().IsDense() == IsDense());
-      break;
-    case ColumnType::kUint32:
-      PERFETTO_DCHECK(nullable_vector<uint32_t>().IsDense() == IsDense());
-      break;
-    case ColumnType::kInt64:
-      PERFETTO_DCHECK(nullable_vector<int64_t>().IsDense() == IsDense());
-      break;
-    case ColumnType::kDouble:
-      PERFETTO_DCHECK(nullable_vector<double>().IsDense() == IsDense());
-      break;
-    case ColumnType::kString:
-      PERFETTO_DCHECK(nullable_vector<StringPool::Id>().IsDense() == IsDense());
-      break;
-    case ColumnType::kId:
-    case ColumnType::kDummy:
-      break;
+  if (IsNullable() && !IsDummy()) {
+    bool is_storage_dense;
+    switch (type_) {
+      case ColumnType::kInt32:
+        is_storage_dense = storage<int32_t>().IsDense();
+        break;
+      case ColumnType::kUint32:
+        is_storage_dense = storage<uint32_t>().IsDense();
+        break;
+      case ColumnType::kInt64:
+        is_storage_dense = storage<int64_t>().IsDense();
+        break;
+      case ColumnType::kDouble:
+        is_storage_dense = storage<double>().IsDense();
+        break;
+      case ColumnType::kString:
+        PERFETTO_FATAL("String column should not be nullable");
+      case ColumnType::kId:
+        PERFETTO_FATAL("Id column should not be nullable");
+      case ColumnType::kDummy:
+        PERFETTO_FATAL("Dummy column excluded above");
+    }
+    PERFETTO_DCHECK(is_storage_dense == IsDense());
   }
   PERFETTO_DCHECK(IsFlagsAndTypeValid(flags_, type_));
 }
@@ -154,7 +158,7 @@
     PERFETTO_DCHECK(value.is_null());
     if (is_nullable) {
       row_map().FilterInto(rm, [this](uint32_t row) {
-        return !nullable_vector<T>().Get(row).has_value();
+        return !storage<T>().Get(row).has_value();
       });
     } else {
       rm->Clear();
@@ -164,7 +168,7 @@
     PERFETTO_DCHECK(value.is_null());
     if (is_nullable) {
       row_map().FilterInto(rm, [this](uint32_t row) {
-        return nullable_vector<T>().Get(row).has_value();
+        return storage<T>().Get(row).has_value();
       });
     }
     return;
@@ -222,55 +226,55 @@
     case FilterOp::kLt:
       row_map().FilterInto(rm, [this, &cmp](uint32_t idx) {
         if (is_nullable) {
-          auto opt_value = nullable_vector<T>().Get(idx);
+          auto opt_value = storage<T>().Get(idx);
           return opt_value && cmp(*opt_value) < 0;
         }
-        return cmp(nullable_vector<T>().GetNonNull(idx)) < 0;
+        return cmp(storage<T>().GetNonNull(idx)) < 0;
       });
       break;
     case FilterOp::kEq:
       row_map().FilterInto(rm, [this, &cmp](uint32_t idx) {
         if (is_nullable) {
-          auto opt_value = nullable_vector<T>().Get(idx);
+          auto opt_value = storage<T>().Get(idx);
           return opt_value && cmp(*opt_value) == 0;
         }
-        return cmp(nullable_vector<T>().GetNonNull(idx)) == 0;
+        return cmp(storage<T>().GetNonNull(idx)) == 0;
       });
       break;
     case FilterOp::kGt:
       row_map().FilterInto(rm, [this, &cmp](uint32_t idx) {
         if (is_nullable) {
-          auto opt_value = nullable_vector<T>().Get(idx);
+          auto opt_value = storage<T>().Get(idx);
           return opt_value && cmp(*opt_value) > 0;
         }
-        return cmp(nullable_vector<T>().GetNonNull(idx)) > 0;
+        return cmp(storage<T>().GetNonNull(idx)) > 0;
       });
       break;
     case FilterOp::kNe:
       row_map().FilterInto(rm, [this, &cmp](uint32_t idx) {
         if (is_nullable) {
-          auto opt_value = nullable_vector<T>().Get(idx);
+          auto opt_value = storage<T>().Get(idx);
           return opt_value && cmp(*opt_value) != 0;
         }
-        return cmp(nullable_vector<T>().GetNonNull(idx)) != 0;
+        return cmp(storage<T>().GetNonNull(idx)) != 0;
       });
       break;
     case FilterOp::kLe:
       row_map().FilterInto(rm, [this, &cmp](uint32_t idx) {
         if (is_nullable) {
-          auto opt_value = nullable_vector<T>().Get(idx);
+          auto opt_value = storage<T>().Get(idx);
           return opt_value && cmp(*opt_value) <= 0;
         }
-        return cmp(nullable_vector<T>().GetNonNull(idx)) <= 0;
+        return cmp(storage<T>().GetNonNull(idx)) <= 0;
       });
       break;
     case FilterOp::kGe:
       row_map().FilterInto(rm, [this, &cmp](uint32_t idx) {
         if (is_nullable) {
-          auto opt_value = nullable_vector<T>().Get(idx);
+          auto opt_value = storage<T>().Get(idx);
           return opt_value && cmp(*opt_value) >= 0;
         }
-        return cmp(nullable_vector<T>().GetNonNull(idx)) >= 0;
+        return cmp(storage<T>().GetNonNull(idx)) >= 0;
       });
       break;
     case FilterOp::kIsNull:
@@ -465,20 +469,19 @@
   PERFETTO_DCHECK(IsNullable() == is_nullable);
   PERFETTO_DCHECK(ColumnTypeHelper<T>::ToColumnType() == type_);
 
-  const auto& nv = nullable_vector<T>();
-  row_map().StableSort(out, [&nv](uint32_t a_idx, uint32_t b_idx) {
+  row_map().StableSort(out, [this](uint32_t a_idx, uint32_t b_idx) {
     if (is_nullable) {
-      auto a_val = nv.Get(a_idx);
-      auto b_val = nv.Get(b_idx);
+      auto a_val = storage<T>().Get(a_idx);
+      auto b_val = storage<T>().Get(b_idx);
 
       int res = compare::NullableNumeric(a_val, b_val);
       return desc ? res > 0 : res < 0;
     }
-    auto a_val = nv.GetNonNull(a_idx);
-    auto b_val = nv.GetNonNull(b_idx);
+    auto a_val = storage<T>().GetNonNull(a_idx);
+    auto b_val = storage<T>().GetNonNull(b_idx);
 
-    return desc ? compare::Numeric(a_val, b_val) > 0
-                : compare::Numeric(a_val, b_val) < 0;
+    int res = compare::Numeric(a_val, b_val);
+    return desc ? res > 0 : res < 0;
   });
 }
 
diff --git a/src/trace_processor/db/column.h b/src/trace_processor/db/column.h
index f83e56e..dd4a8f5 100644
--- a/src/trace_processor/db/column.h
+++ b/src/trace_processor/db/column.h
@@ -22,10 +22,11 @@
 #include "perfetto/base/logging.h"
 #include "perfetto/ext/base/optional.h"
 #include "perfetto/trace_processor/basic_types.h"
-#include "src/trace_processor/containers/nullable_vector.h"
 #include "src/trace_processor/containers/row_map.h"
 #include "src/trace_processor/containers/string_pool.h"
+#include "src/trace_processor/db/column_storage.h"
 #include "src/trace_processor/db/compare.h"
+#include "src/trace_processor/db/typed_column_internal.h"
 
 namespace perfetto {
 namespace trace_processor {
@@ -218,7 +219,7 @@
 
   template <typename T>
   Column(const char* name,
-         NullableVector<T>* storage,
+         ColumnStorage<T>* storage,
          /* Flag */ uint32_t flags,
          Table* table,
          uint32_t col_idx_in_table,
@@ -435,26 +436,23 @@
   }
 
  protected:
+  template <typename T>
+  using storage_t = typename tc_internal::TypeHandler<T>::storage_type;
+
   // Returns the backing sparse vector cast to contain data of type T.
   // Should only be called when |type_| == ToColumnType<T>().
   template <typename T>
-  NullableVector<T>* mutable_nullable_vector() {
+  storage_t<T>* mutable_storage() {
     PERFETTO_DCHECK(ColumnTypeHelper<T>::ToColumnType() == type_);
-    return static_cast<NullableVector<T>*>(nullable_vector_);
+    return static_cast<storage_t<T>*>(storage_);
   }
 
   // Returns the backing sparse vector cast to contain data of type T.
   // Should only be called when |type_| == ToColumnType<T>().
   template <typename T>
-  const NullableVector<T>& nullable_vector() const {
+  const storage_t<T>& storage() const {
     PERFETTO_DCHECK(ColumnTypeHelper<T>::ToColumnType() == type_);
-    return *static_cast<const NullableVector<T>*>(nullable_vector_);
-  }
-
-  // Returns the type of this Column in terms of SqlValue::Type.
-  template <typename T>
-  static SqlValue::Type ToSqlValueType() {
-    return ToSqlValueType(ColumnTypeHelper<T>::ToColumnType());
+    return *static_cast<storage_t<T>*>(storage_);
   }
 
   // Returns true if this column is a dense column.
@@ -465,6 +463,20 @@
 
   const StringPool& string_pool() const { return *string_pool_; }
 
+  // Returns the type of this Column in terms of SqlValue::Type.
+  template <typename T>
+  static SqlValue::Type ToSqlValueType() {
+    return ToSqlValueType(ColumnTypeHelper<T>::ToColumnType());
+  }
+
+  static SqlValue ToSqlValue(double value) { return SqlValue::Double(value); }
+  static SqlValue ToSqlValue(int32_t value) { return SqlValue::Long(value); }
+  static SqlValue ToSqlValue(uint32_t value) { return SqlValue::Long(value); }
+  static SqlValue ToSqlValue(int64_t value) { return SqlValue::Long(value); }
+  static SqlValue ToSqlValue(NullTermStringView value) {
+    return SqlValue::String(value.c_str());
+  }
+
  private:
   friend class Table;
   friend class View;
@@ -476,7 +488,7 @@
          Table* table,
          uint32_t col_idx_in_table,
          uint32_t row_map_idx,
-         NullableVectorBase* nullable_vector);
+         ColumnStorageBase* nullable_vector);
 
   Column(const Column&) = delete;
   Column& operator=(const Column&) = delete;
@@ -484,22 +496,14 @@
   // Gets the value of the Column at the given |idx|.
   SqlValue GetAtIdx(uint32_t idx) const {
     switch (type_) {
-      case ColumnType::kInt32: {
-        auto opt_value = nullable_vector<int32_t>().Get(idx);
-        return opt_value ? SqlValue::Long(*opt_value) : SqlValue();
-      }
-      case ColumnType::kUint32: {
-        auto opt_value = nullable_vector<uint32_t>().Get(idx);
-        return opt_value ? SqlValue::Long(*opt_value) : SqlValue();
-      }
-      case ColumnType::kInt64: {
-        auto opt_value = nullable_vector<int64_t>().Get(idx);
-        return opt_value ? SqlValue::Long(*opt_value) : SqlValue();
-      }
-      case ColumnType::kDouble: {
-        auto opt_value = nullable_vector<double>().Get(idx);
-        return opt_value ? SqlValue::Double(*opt_value) : SqlValue();
-      }
+      case ColumnType::kInt32:
+        return GetAtIdxTyped<int32_t>(idx);
+      case ColumnType::kUint32:
+        return GetAtIdxTyped<uint32_t>(idx);
+      case ColumnType::kInt64:
+        return GetAtIdxTyped<int64_t>(idx);
+      case ColumnType::kDouble:
+        return GetAtIdxTyped<double>(idx);
       case ColumnType::kString: {
         auto str = GetStringPoolStringAtIdx(idx).c_str();
         return str == nullptr ? SqlValue() : SqlValue::String(str);
@@ -512,6 +516,15 @@
     PERFETTO_FATAL("For GCC");
   }
 
+  template <typename T>
+  SqlValue GetAtIdxTyped(uint32_t idx) const {
+    if (IsNullable()) {
+      auto opt_value = storage<T>().Get(idx);
+      return opt_value ? ToSqlValue(*opt_value) : SqlValue();
+    }
+    return ToSqlValue(storage<T>().GetNonNull(idx));
+  }
+
   // Optimized filter method for sorted columns.
   // Returns whether the constraint was handled by the method.
   bool FilterIntoSorted(FilterOp op, SqlValue value, RowMap* rm) const {
@@ -565,7 +578,7 @@
     PERFETTO_DCHECK(!IsNullable());
 
     uint32_t filter_set_id = static_cast<uint32_t>(value);
-    const auto& nv = nullable_vector<uint32_t>();
+    const auto& nv = storage<uint32_t>();
     const RowMap& col_rm = row_map();
 
     // If the set id is beyond the end of the column, there's no chance that
@@ -676,12 +689,12 @@
   // Should only be called when |type_| == ColumnType::kString.
   NullTermStringView GetStringPoolStringAtIdx(uint32_t idx) const {
     PERFETTO_DCHECK(type_ == ColumnType::kString);
-    return string_pool_->Get(nullable_vector<StringPool::Id>().GetNonNull(idx));
+    return string_pool_->Get(storage<StringPool::Id>().GetNonNull(idx));
   }
 
   // type_ is used to cast nullable_vector_ to the correct type.
   ColumnType type_ = ColumnType::kInt64;
-  NullableVectorBase* nullable_vector_ = nullptr;
+  ColumnStorageBase* storage_ = nullptr;
 
   const char* name_ = nullptr;
   uint32_t flags_ = Flag::kNoFlag;
diff --git a/src/trace_processor/containers/nullable_vector.cc b/src/trace_processor/db/column_storage.cc
similarity index 85%
rename from src/trace_processor/containers/nullable_vector.cc
rename to src/trace_processor/db/column_storage.cc
index 7c4cd68..d1d348f 100644
--- a/src/trace_processor/containers/nullable_vector.cc
+++ b/src/trace_processor/db/column_storage.cc
@@ -14,12 +14,12 @@
  * limitations under the License.
  */
 
-#include "src/trace_processor/containers/nullable_vector.h"
+#include "src/trace_processor/db/column_storage.h"
 
 namespace perfetto {
 namespace trace_processor {
 
-NullableVectorBase::~NullableVectorBase() = default;
+ColumnStorageBase::~ColumnStorageBase() = default;
 
 }  // namespace trace_processor
 }  // namespace perfetto
diff --git a/src/trace_processor/db/column_storage.h b/src/trace_processor/db/column_storage.h
new file mode 100644
index 0000000..2ba443f
--- /dev/null
+++ b/src/trace_processor/db/column_storage.h
@@ -0,0 +1,78 @@
+/*
+ * Copyright (C) 2022 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SRC_TRACE_PROCESSOR_DB_COLUMN_STORAGE_H_
+#define SRC_TRACE_PROCESSOR_DB_COLUMN_STORAGE_H_
+
+#include "src/trace_processor/containers/nullable_vector.h"
+
+namespace perfetto {
+namespace trace_processor {
+
+// Base class for allowing type erasure when defining plug-in implementations
+// of backing storage for columns.
+class ColumnStorageBase {
+ public:
+  ColumnStorageBase() = default;
+  virtual ~ColumnStorageBase();
+
+  ColumnStorageBase(const ColumnStorageBase&) = delete;
+  ColumnStorageBase& operator=(const ColumnStorageBase&) = delete;
+
+  ColumnStorageBase(ColumnStorageBase&&) = default;
+  ColumnStorageBase& operator=(ColumnStorageBase&&) noexcept = default;
+};
+
+// Class used for implementing storage for columns.
+// TODO(lalitm): split this class up into a nullable and non-null components.
+template <typename T>
+class ColumnStorage : public ColumnStorageBase {
+ public:
+  ColumnStorage() = default;
+
+  explicit ColumnStorage(const ColumnStorage&) = delete;
+  ColumnStorage& operator=(const ColumnStorage&) = delete;
+
+  ColumnStorage(ColumnStorage&&) = default;
+  ColumnStorage& operator=(ColumnStorage&&) noexcept = default;
+
+  base::Optional<T> Get(uint32_t idx) const { return nv_.Get(idx); }
+  T GetNonNull(uint32_t non_null_idx) const {
+    return nv_.GetNonNull(non_null_idx);
+  }
+  void Append(T val) { nv_.Append(val); }
+  void Append(base::Optional<T> val) { nv_.Append(val); }
+  void Set(uint32_t idx, T val) { nv_.Set(idx, val); }
+  uint32_t size() const { return nv_.size(); }
+  bool IsDense() const { return nv_.IsDense(); }
+
+  static ColumnStorage<T> Sparse() {
+    return ColumnStorage<T>(NullableVector<T>::Sparse());
+  }
+  static ColumnStorage<T> Dense() {
+    return ColumnStorage<T>(NullableVector<T>::Dense());
+  }
+
+ private:
+  explicit ColumnStorage(NullableVector<T> nv) : nv_(std::move(nv)) {}
+
+  NullableVector<T> nv_;
+};
+
+}  // namespace trace_processor
+}  // namespace perfetto
+
+#endif  // SRC_TRACE_PROCESSOR_DB_COLUMN_STORAGE_H_
diff --git a/src/trace_processor/db/typed_column.h b/src/trace_processor/db/typed_column.h
index c5aec10..e34d38d 100644
--- a/src/trace_processor/db/typed_column.h
+++ b/src/trace_processor/db/typed_column.h
@@ -89,13 +89,11 @@
   }
 
   // Inserts the value at the end of the column.
-  void Append(T v) {
-    mutable_nullable_vector()->Append(Serializer::Serialize(v));
-  }
+  void Append(T v) { mutable_storage()->Append(Serializer::Serialize(v)); }
 
   // Returns the row containing the given value in the Column.
   base::Optional<uint32_t> IndexOf(sql_value_type v) const {
-    return Column::IndexOf(ToValue(v));
+    return Column::IndexOf(ToSqlValue(v));
   }
 
   std::vector<T> ToVectorForTesting() const {
@@ -106,12 +104,12 @@
   }
 
   // Helper functions to create constraints for the given value.
-  Constraint eq(sql_value_type v) const { return eq_value(ToValue(v)); }
-  Constraint gt(sql_value_type v) const { return gt_value(ToValue(v)); }
-  Constraint lt(sql_value_type v) const { return lt_value(ToValue(v)); }
-  Constraint ne(sql_value_type v) const { return ne_value(ToValue(v)); }
-  Constraint ge(sql_value_type v) const { return ge_value(ToValue(v)); }
-  Constraint le(sql_value_type v) const { return le_value(ToValue(v)); }
+  Constraint eq(sql_value_type v) const { return eq_value(ToSqlValue(v)); }
+  Constraint gt(sql_value_type v) const { return gt_value(ToSqlValue(v)); }
+  Constraint lt(sql_value_type v) const { return lt_value(ToSqlValue(v)); }
+  Constraint ne(sql_value_type v) const { return ne_value(ToSqlValue(v)); }
+  Constraint ge(sql_value_type v) const { return ge_value(ToSqlValue(v)); }
+  Constraint le(sql_value_type v) const { return le_value(ToSqlValue(v)); }
 
   // Implements equality between two items of type |T|.
   static constexpr bool Equals(T a, T b) { return TH::Equals(a, b); }
@@ -144,18 +142,18 @@
   // Public for use by macro tables.
   void SetAtIdx(uint32_t idx, non_optional_type v) {
     auto serialized = Serializer::Serialize(v);
-    mutable_nullable_vector()->Set(idx, serialized);
+    mutable_storage()->Set(idx, serialized);
   }
 
   // Public for use by macro tables.
   T GetAtIdx(uint32_t idx) const {
-    return Serializer::Deserialize(TH::Get(nullable_vector(), idx));
+    return Serializer::Deserialize(TH::Get(storage(), idx));
   }
 
   template <bool is_string = TH::is_string>
   typename std::enable_if<is_string, NullTermStringView>::type GetStringAtIdx(
       uint32_t idx) const {
-    return string_pool().Get(nullable_vector().GetNonNull(idx));
+    return string_pool().Get(storage().GetNonNull(idx));
   }
 
  private:
@@ -178,18 +176,11 @@
     }
   }
 
-  static SqlValue ToValue(double value) { return SqlValue::Double(value); }
-  static SqlValue ToValue(uint32_t value) { return SqlValue::Long(value); }
-  static SqlValue ToValue(int64_t value) { return SqlValue::Long(value); }
-  static SqlValue ToValue(NullTermStringView value) {
-    return SqlValue::String(value.c_str());
+  const storage_type& storage() const {
+    return Column::storage<serialized_type>();
   }
-
-  const NullableVector<serialized_type>& nullable_vector() const {
-    return Column::nullable_vector<serialized_type>();
-  }
-  NullableVector<serialized_type>* mutable_nullable_vector() {
-    return Column::mutable_nullable_vector<serialized_type>();
+  storage_type* mutable_storage() {
+    return Column::mutable_storage<serialized_type>();
   }
 };
 
diff --git a/src/trace_processor/db/typed_column_internal.h b/src/trace_processor/db/typed_column_internal.h
index d3c6061..cc9207b 100644
--- a/src/trace_processor/db/typed_column_internal.h
+++ b/src/trace_processor/db/typed_column_internal.h
@@ -20,6 +20,7 @@
 #include "src/trace_processor/containers/nullable_vector.h"
 #include "src/trace_processor/containers/string_pool.h"
 #include "src/trace_processor/db/base_id.h"
+#include "src/trace_processor/db/column_storage.h"
 
 namespace perfetto {
 namespace trace_processor {
@@ -95,7 +96,7 @@
   using serialized_type =
       typename Serializer<non_optional_type>::serialized_type;
   using sql_value_type = serialized_type;
-  using storage_type = NullableVector<serialized_type>;
+  using storage_type = ColumnStorage<serialized_type>;
 
   static constexpr bool is_optional = false;
   static constexpr bool is_string = false;
@@ -124,7 +125,7 @@
   using serialized_type =
       typename Serializer<non_optional_type>::serialized_type;
   using sql_value_type = serialized_type;
-  using storage_type = NullableVector<serialized_type>;
+  using storage_type = ColumnStorage<serialized_type>;
 
   static constexpr bool is_optional = true;
   static constexpr bool is_string = false;
@@ -156,7 +157,7 @@
   using non_optional_type = StringPool::Id;
   using serialized_type = StringPool::Id;
   using sql_value_type = NullTermStringView;
-  using storage_type = NullableVector<serialized_type>;
+  using storage_type = ColumnStorage<serialized_type>;
 
   static constexpr bool is_optional = false;
   static constexpr bool is_string = true;
@@ -181,7 +182,7 @@
   using non_optional_type = StringPool::Id;
   using serialized_type = StringPool::Id;
   using sql_value_type = NullTermStringView;
-  using storage_type = NullableVector<serialized_type>;
+  using storage_type = ColumnStorage<serialized_type>;
 
   // is_optional is false again because we always unwrap
   // base::Optional<StringPool::Id> into StringPool::Id.
diff --git a/src/trace_processor/dynamic/ancestor_generator.cc b/src/trace_processor/dynamic/ancestor_generator.cc
index e7a3154..722724e 100644
--- a/src/trace_processor/dynamic/ancestor_generator.cc
+++ b/src/trace_processor/dynamic/ancestor_generator.cc
@@ -100,7 +100,7 @@
     ConstraintType constraint_value,
     const ParentTable& table,
     std::vector<typename ParentTable::RowNumber> parent_rows) {
-  NullableVector<ConstraintType> start_ids;
+  ColumnStorage<ConstraintType> start_ids;
   for (uint32_t i = 0; i < parent_rows.size(); ++i)
     start_ids.Append(constraint_value);
   return ChildTable::SelectAndExtendParent(table, std::move(parent_rows),
diff --git a/src/trace_processor/dynamic/connected_flow_generator.cc b/src/trace_processor/dynamic/connected_flow_generator.cc
index bb3cf3d..1565a4c 100644
--- a/src/trace_processor/dynamic/connected_flow_generator.cc
+++ b/src/trace_processor/dynamic/connected_flow_generator.cc
@@ -251,7 +251,7 @@
       std::move(bfs).TakeResultingFlows();
 
   // Aditional column for start_id
-  NullableVector<uint32_t> start_ids;
+  ColumnStorage<uint32_t> start_ids;
   for (size_t i = 0; i < result_rows.size(); i++) {
     start_ids.Append(start_id.value);
   }
diff --git a/src/trace_processor/dynamic/descendant_generator.cc b/src/trace_processor/dynamic/descendant_generator.cc
index a2dd5f5..72d260f 100644
--- a/src/trace_processor/dynamic/descendant_generator.cc
+++ b/src/trace_processor/dynamic/descendant_generator.cc
@@ -53,7 +53,7 @@
     ConstraintType constraint_id,
     const ParentTable& table,
     std::vector<typename ParentTable::RowNumber> parent_rows) {
-  NullableVector<ConstraintType> start_ids;
+  ColumnStorage<ConstraintType> start_ids;
   for (uint32_t i = 0; i < parent_rows.size(); ++i)
     start_ids.Append(constraint_id);
   return ChildTable::SelectAndExtendParent(table, std::move(parent_rows),
diff --git a/src/trace_processor/dynamic/experimental_annotated_stack_generator.cc b/src/trace_processor/dynamic/experimental_annotated_stack_generator.cc
index 1dfa722..3cee91a 100644
--- a/src/trace_processor/dynamic/experimental_annotated_stack_generator.cc
+++ b/src/trace_processor/dynamic/experimental_annotated_stack_generator.cc
@@ -258,14 +258,14 @@
 
   // Build the dynamic table.
   PERFETTO_DCHECK(cs_rows.size() == annotations_reversed.size());
-  NullableVector<StringPool::Id> annotation_vals;
+  ColumnStorage<StringPool::Id> annotation_vals;
   for (auto it = annotations_reversed.rbegin();
        it != annotations_reversed.rend(); ++it) {
     annotation_vals.Append(*it);
   }
 
   // Hidden column - always the input, i.e. the callsite leaf.
-  NullableVector<uint32_t> start_id_vals;
+  ColumnStorage<uint32_t> start_id_vals;
   for (uint32_t i = 0; i < cs_rows.size(); i++)
     start_id_vals.Append(start_id.value);
 
diff --git a/src/trace_processor/dynamic/experimental_counter_dur_generator.cc b/src/trace_processor/dynamic/experimental_counter_dur_generator.cc
index d752d4c..2407e0d 100644
--- a/src/trace_processor/dynamic/experimental_counter_dur_generator.cc
+++ b/src/trace_processor/dynamic/experimental_counter_dur_generator.cc
@@ -69,11 +69,11 @@
 }
 
 // static
-NullableVector<int64_t> ExperimentalCounterDurGenerator::ComputeDurColumn(
+ColumnStorage<int64_t> ExperimentalCounterDurGenerator::ComputeDurColumn(
     const Table& table) {
   // Keep track of the last seen row for each track id.
   std::unordered_map<TrackId, uint32_t> last_row_for_track_id;
-  NullableVector<int64_t> dur;
+  ColumnStorage<int64_t> dur;
 
   const auto* ts_col =
       TypedColumn<int64_t>::FromColumn(table.GetColumnByName("ts"));
@@ -104,11 +104,11 @@
 }
 
 // static
-NullableVector<double> ExperimentalCounterDurGenerator::ComputeDeltaColumn(
+ColumnStorage<double> ExperimentalCounterDurGenerator::ComputeDeltaColumn(
     const Table& table) {
   // Keep track of the last seen row for each track id.
   std::unordered_map<TrackId, uint32_t> last_row_for_track_id;
-  NullableVector<double> delta;
+  ColumnStorage<double> delta;
 
   const auto* value_col =
       TypedColumn<double>::FromColumn(table.GetColumnByName("value"));
diff --git a/src/trace_processor/dynamic/experimental_counter_dur_generator.h b/src/trace_processor/dynamic/experimental_counter_dur_generator.h
index 640cbf1..d513588 100644
--- a/src/trace_processor/dynamic/experimental_counter_dur_generator.h
+++ b/src/trace_processor/dynamic/experimental_counter_dur_generator.h
@@ -38,8 +38,8 @@
                             std::unique_ptr<Table>& table_return) override;
 
   // public + static for testing
-  static NullableVector<int64_t> ComputeDurColumn(const Table& table);
-  static NullableVector<double> ComputeDeltaColumn(const Table& table);
+  static ColumnStorage<int64_t> ComputeDurColumn(const Table& table);
+  static ColumnStorage<double> ComputeDeltaColumn(const Table& table);
 
  private:
   const tables::CounterTable* counter_table_ = nullptr;
diff --git a/src/trace_processor/dynamic/experimental_sched_upid_generator.cc b/src/trace_processor/dynamic/experimental_sched_upid_generator.cc
index 923e963..c2e0dcf 100644
--- a/src/trace_processor/dynamic/experimental_sched_upid_generator.cc
+++ b/src/trace_processor/dynamic/experimental_sched_upid_generator.cc
@@ -65,8 +65,8 @@
   return base::OkStatus();
 }
 
-NullableVector<uint32_t> ExperimentalSchedUpidGenerator::ComputeUpidColumn() {
-  NullableVector<uint32_t> upid;
+ColumnStorage<uint32_t> ExperimentalSchedUpidGenerator::ComputeUpidColumn() {
+  ColumnStorage<uint32_t> upid;
   for (uint32_t i = 0; i < sched_slice_table_->row_count(); ++i) {
     upid.Append(thread_table_->upid()[sched_slice_table_->utid()[i]]);
   }
diff --git a/src/trace_processor/dynamic/experimental_sched_upid_generator.h b/src/trace_processor/dynamic/experimental_sched_upid_generator.h
index 9c0f7ae..d0d8538 100644
--- a/src/trace_processor/dynamic/experimental_sched_upid_generator.h
+++ b/src/trace_processor/dynamic/experimental_sched_upid_generator.h
@@ -41,7 +41,7 @@
                             std::unique_ptr<Table>& table_return) override;
 
  private:
-  NullableVector<uint32_t> ComputeUpidColumn();
+  ColumnStorage<uint32_t> ComputeUpidColumn();
 
   const tables::SchedSliceTable* sched_slice_table_;
   const tables::ThreadTable* thread_table_;
diff --git a/src/trace_processor/dynamic/experimental_slice_layout_generator.cc b/src/trace_processor/dynamic/experimental_slice_layout_generator.cc
index 63fcd8f..ded6fe4 100644
--- a/src/trace_processor/dynamic/experimental_slice_layout_generator.cc
+++ b/src/trace_processor/dynamic/experimental_slice_layout_generator.cc
@@ -265,8 +265,8 @@
   }
 
   // Step 3: Add the two new columns layout_depth and filter_track_ids:
-  NullableVector<uint32_t> layout_depth_column;
-  NullableVector<StringPool::Id> filter_column;
+  ColumnStorage<uint32_t> layout_depth_column;
+  ColumnStorage<StringPool::Id> filter_column;
 
   for (tables::SliceTable::RowNumber i : rows) {
     auto ref = i.ToRowReference(*slice_table_);
diff --git a/src/trace_processor/tables/macros_internal.h b/src/trace_processor/tables/macros_internal.h
index b1a4020..320329b 100644
--- a/src/trace_processor/tables/macros_internal.h
+++ b/src/trace_processor/tables/macros_internal.h
@@ -93,7 +93,7 @@
       row_maps_.emplace_back();
       columns_.emplace_back(Column::IdColumn(this, 0, 0));
       columns_.emplace_back(
-          Column("type", &type_, Column::kNoFlag, this, 1, 0));
+          Column("type", &type_, Column::kNonNull, this, 1, 0));
       return;
     }