diff --git a/cpp/include/cuvs/core/dataset.hpp b/cpp/include/cuvs/core/dataset.hpp
new file mode 100644
index 0000000000..a5445db6f1
--- /dev/null
+++ b/cpp/include/cuvs/core/dataset.hpp
@@ -0,0 +1,512 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cuvs/core/export.hpp>
+
+#include <raft/core/coo_matrix.hpp>
+#include <raft/core/csr_matrix.hpp>
+#include <raft/core/device_mdarray.hpp>
+#include <raft/core/host_mdarray.hpp>
+#include <raft/core/mdarray.hpp>
+#include <raft/core/mdspan.hpp>
+#include <raft/util/integer_utils.hpp>
+
+#include <atomic>
+#include <concepts>
+#include <memory>
+#include <type_traits>
+#include <utility>
+
+namespace CUVS_EXPORT cuvs {
+
+/************************************************************************************************
+ * The core dataset type
+ ************************************************************************************************/
+
+/**
+ * A spec is compressing if it defines a dictionary: the codebooks, the quantization range, or
+ * whatever else is needed to interpret the data. Plain specs declare `std::monostate` instead, so
+ * "there is no dictionary" and "the dictionary slot is empty" are the same vocabulary type.
+ */
+template <typename SpecT>
+concept compressed_dataset_spec = requires {
+  typename SpecT::dictionary_type;
+  typename SpecT::dictionary_view_type;
+  typename SpecT::code_type;
+} && !std::same_as<typename SpecT::dictionary_type, std::monostate>;
+
+template <typename T, typename IdxT, typename SpecT>
+struct dataset {
+  using spec_type  = typename SpecT::template apply<T, IdxT>;
+  using value_type = typename spec_type::value_type;
+  using index_type = typename spec_type::index_type;
+
+  using data_type       = typename spec_type::data_type;
+  using view_type       = typename spec_type::view_type;
+  using const_view_type = typename spec_type::const_view_type;
+
+  /* The second, optional storage slot; `std::monostate` for an uncompressed dataset. */
+  using dictionary_type = typename spec_type::dictionary_type;
+
+  static constexpr bool is_compressed = compressed_dataset_spec<spec_type>;
+
+  explicit dataset(data_type&& data)
+    requires(!is_compressed)
+    : data_{std::make_shared<data_type>(std::move(data))}
+  {
+  }
+
+  dataset(data_type&& data, std::shared_ptr<const dictionary_type> dictionary)
+    requires(is_compressed)
+    : data_{std::make_shared<data_type>(std::move(data))}, dictionary_{std::move(dictionary)}
+  {
+  }
+
+  /*
+   @return A non-owning view of the data, such as mdspan.
+  */
+  [[nodiscard]] auto data_view() noexcept -> view_type { return spec_type::get_data_view(*data_); }
+  [[nodiscard]] auto data_const_view() const noexcept -> const_view_type
+  {
+    return spec_type::get_data_const_view(*data_);
+  }
+  [[nodiscard]] auto n_rows() const noexcept -> index_type { return spec_type::get_n_rows(*data_); }
+  // NOTE: a compressed dataset may not be able to tell its dimension from the encoded data alone
+  [[nodiscard]] auto dim() const noexcept -> uint32_t
+  {
+    return spec_type::get_dim(*data_, dict_ref());
+  }
+
+  /*
+   * Check if the caller is the only owner of the data.
+   * This is useful to determine if the data can be modified in place without copying.
+   */
+  [[nodiscard]] auto is_data_unique() const noexcept -> bool
+  {
+    /*
+    The count cannot grow behind our back: only copying a `dataset` increments it and `data_` never
+    escapes the class, so no weak_ptr can resurrect a reference. Hence, if it reads one, the sole
+    reference is the one we hold and no other thread has anything to copy from. The converse does
+    not hold - a concurrent release may not be visible yet - but that only costs the optimization.
+
+    The fence must follow the load: releasing a copy decrements with release ordering, yet
+    `use_count()` reads relaxed and shared_ptr pairs the decrement with an acquire only once the
+    counter reaches zero. Without the fence we could miss the writes of a copy that another thread
+    has already destroyed.
+    */
+    if (data_.use_count() != 1) { return false; }
+    std::atomic_thread_fence(std::memory_order_acquire);
+    return true;
+  }
+
+  /*
+   @return A non-owning view of the dictionary, such as the two codebooks of a VPQ dataset.
+  */
+  [[nodiscard]] auto dictionary_view() const noexcept
+    requires(is_compressed)
+  {
+    return spec_type::get_dictionary_view(*dictionary_);
+  }
+  /*
+   A dictionary is immutable once trained, so several datasets may encode against the same one.
+
+   @return A shared handle to the dictionary.
+  */
+  [[nodiscard]] auto share_dictionary() const noexcept -> std::shared_ptr<const dictionary_type>
+    requires(is_compressed)
+  {
+    return dictionary_;
+  }
+
+ private:
+  std::shared_ptr<data_type> data_;
+  /* Either a shared handle to the dictionary, or the empty dictionary itself. */
+  [[no_unique_address]] std::conditional_t<is_compressed,
+                                           std::shared_ptr<const dictionary_type>,
+                                           dictionary_type> dictionary_;
+
+  [[nodiscard]] auto dict_ref() const noexcept -> const dictionary_type&
+  {
+    if constexpr (is_compressed) {
+      return *dictionary_;
+    } else {
+      return dictionary_;
+    }
+  }
+};
+
+/************************************************************************************************
+ * Dataset specifications
+ ************************************************************************************************/
+
+/**
+ * Empty dataset specification: contains no data, but keeps the dimension of the dataset implement
+ * the common interface.
+ */
+struct empty_spec {
+  struct empty_dataset_rep {
+    uint32_t dim;
+  };
+
+  template <typename T, typename IdxT>
+  struct apply {
+    using data_type       = empty_dataset_rep;
+    using view_type       = empty_dataset_rep;
+    using const_view_type = const empty_dataset_rep;
+    using value_type      = std::remove_cv_t<T>;
+    using index_type      = std::remove_cv_t<IdxT>;
+    using dictionary_type = std::monostate;
+
+    [[nodiscard]] static auto get_data_view(data_type& data) noexcept -> view_type { return data; }
+    [[nodiscard]] static auto get_data_const_view(const data_type& data) noexcept -> const_view_type
+    {
+      return data;
+    }
+    [[nodiscard]] static auto get_n_rows(const data_type& data) noexcept -> index_type { return 0; }
+    [[nodiscard]] static auto get_dim(const data_type& data, const dictionary_type&) noexcept
+      -> uint32_t
+    {
+      return data.dim;
+    }
+  };
+};
+
+/**
+ * Dense plain/padded dataset implemented via raft::mdarray.
+ */
+template <typename LayoutPolicy, typename ContainerPolicy>
+struct mdarray_spec {
+  template <typename T, typename IdxT>
+  struct apply {
+    /* NOTE: index type != extents type
+    The index type can vary depending on the use case; we often may want to store indices in 32-bit
+    slots to save memory on large datasets. The extents are always fixed to 64-bit signed integers
+    to simplify the conversion and avoid integer overflow issues.
+    */
+    using data_type = raft::mdarray<T, raft::matrix_extent<int64_t>, LayoutPolicy, ContainerPolicy>;
+    using view_type = typename data_type::view_type;
+    using const_view_type = typename data_type::const_view_type;
+    using value_type      = std::remove_cv_t<T>;
+    using index_type      = std::remove_cv_t<IdxT>;
+    using dictionary_type = std::monostate;
+
+    [[nodiscard]] static auto get_data_view(data_type& data) noexcept -> view_type
+    {
+      return data.view();
+    }
+
+    [[nodiscard]] static auto get_data_const_view(const data_type& data) noexcept -> const_view_type
+    {
+      return data.view();
+    }
+    [[nodiscard]] static auto get_n_rows(const data_type& data) noexcept -> index_type
+    {
+      return static_cast<index_type>(data.extent(0));
+    }
+    [[nodiscard]] static auto get_dim(const data_type& data, const dictionary_type&) noexcept
+      -> uint32_t
+    {
+      return static_cast<uint32_t>(data.extent(1));
+    }
+  };
+};
+
+/**
+ * Sparse dataset specification implemented via the raft sparse matrix hierarchy. `SparseLayoutT`
+ * selects the representation (`csr_layout`, `coo_layout`).
+ *
+ * NOTE: only the memory space of the container policy is used here; raft's sparse types bind the
+ * policy themselves, separately for the values, the offsets and the indices.
+ */
+template <typename SparseLayoutT, typename ContainerPolicy>
+struct sparse_spec {
+  template <typename T, typename IdxT>
+  struct apply {
+    using value_type      = std::remove_cv_t<T>;
+    using index_type      = std::remove_cv_t<IdxT>;
+    using dictionary_type = std::monostate;
+
+    static constexpr raft::memory_type mem_type = ContainerPolicy::mem_type;
+    static_assert(mem_type == raft::memory_type::host || mem_type == raft::memory_type::device,
+                  "raft's sparse hierarchy only provides host and device containers");
+
+    using data_type = std::conditional_t<
+      mem_type == raft::memory_type::device,
+      typename SparseLayoutT::
+        template matrix_type<value_type, index_type, true, raft::device_container_policy>,
+      typename SparseLayoutT::
+        template matrix_type<value_type, index_type, false, raft::host_container_policy>>;
+    using view_type = typename data_type::view_type;
+    /* NOTE: raft's sparse hierarchy provides no read-only view yet (see the TODO in
+    raft/core/sparse_types.hpp); until then, even reading the structure requires a mutable matrix.
+    */
+    using const_view_type = view_type;
+
+    [[nodiscard]] static auto get_data_view(data_type& data) noexcept -> view_type
+    {
+      return data.view();
+    }
+
+    [[nodiscard]] static auto get_data_const_view(const data_type& data) noexcept -> const_view_type
+    {
+      return const_cast<data_type&>(data).view();
+    }
+    [[nodiscard]] static auto get_n_rows(const data_type& data) noexcept -> index_type
+    {
+      return static_cast<index_type>(const_cast<data_type&>(data).structure_view().get_n_rows());
+    }
+    [[nodiscard]] static auto get_dim(const data_type& data, const dictionary_type&) noexcept
+      -> uint32_t
+    {
+      return static_cast<uint32_t>(const_cast<data_type&>(data).structure_view().get_n_cols());
+    }
+  };
+};
+
+/**
+ * CSR representation: one compressed row per vector of the dataset.
+ *
+ * NOTE: the row offsets are 64-bit for the same reason the dense extents are: the number of
+ * non-zeros is not bounded by the index type. The column indices are feature ids, so they are
+ * bounded by the dataset dimension rather than by its size.
+ */
+struct csr_layout {
+  template <typename T, typename IdxT, bool IsDevice, template <typename> typename ContainerPolicy>
+  using matrix_type = raft::csr_matrix<T, int64_t, uint32_t, uint64_t, IsDevice, ContainerPolicy>;
+};
+
+/**
+ * COO representation: one (row, column, value) triple per non-zero.
+ *
+ * NOTE: unlike CSR, the row component is an index into the dataset rather than an offset into the
+ * non-zeros, hence the dataset index type.
+ */
+struct coo_layout {
+  template <typename T, typename IdxT, bool IsDevice, template <typename> typename ContainerPolicy>
+  using matrix_type = raft::coo_matrix<T, IdxT, uint32_t, uint64_t, IsDevice, ContainerPolicy>;
+};
+
+/**
+ * VQ+PQ compressed dataset specification, mirroring the VPQ dataset of cuvs/neighbors/common.hpp:
+ * the two codebooks form the dictionary, while the data slot keeps the encoded rows - the VQ label
+ * in the row prefix, followed by the packed PQ codes.
+ *
+ * `BookPolicy` is the container policy of the codebooks (elements of `MathT`) and `CodePolicy` the
+ * one of the encoded rows (elements of `uint8_t`); a bound policy cannot be rebound to another
+ * element type, hence the two parameters.
+ */
+template <typename MathT, typename BookPolicy, typename CodePolicy>
+struct vpq_spec {
+  /* The encoded rows are stored like the rows of any other dense dataset; only how a row is read
+  and where the dimension comes from are specific to the compression.
+  */
+  template <typename IdxT>
+  using storage_spec =
+    typename mdarray_spec<raft::layout_c_contiguous, CodePolicy>::template apply<uint8_t, IdxT>;
+
+  template <typename T, typename IdxT>
+  struct apply : storage_spec<IdxT> {
+    /* The dataset stores codes, but it still represents vectors of `T`. */
+    using value_type = std::remove_cv_t<T>;
+    /* NOTE: the members of a dependent base are not visible to unqualified lookup, so `data_type`
+    has to be pulled into this scope to be usable in the signatures below.
+    */
+    using typename storage_spec<IdxT>::data_type;
+    /* The type the codebooks are stored in; the rows themselves are always packed codes. */
+    using math_type = MathT;
+    using code_type = uint8_t;
+
+    /* [vq_n_centers, dim] */
+    using vq_book_type =
+      raft::mdarray<math_type, raft::matrix_extent<int64_t>, raft::layout_c_contiguous, BookPolicy>;
+    /* [n_books, pq_n_centers, pq_len], where `n_books` is 1 for a codebook shared by all the
+    subspaces, or `pq_dim` for one codebook per subspace. A row-major 3d array with `n_books == 1`
+    is bit-identical to the flat 2d codebook of the current implementation.
+    */
+    using pq_book_type =
+      raft::mdarray<math_type, raft::extent_3d<int64_t>, raft::layout_c_contiguous, BookPolicy>;
+
+    struct dictionary_type {
+      vq_book_type vq_code_book;
+      pq_book_type pq_code_book;
+    };
+
+    struct dictionary_view_type {
+      typename vq_book_type::const_view_type vq_code_book;
+      typename pq_book_type::const_view_type pq_code_book;
+
+      /* NOTE: the dimension of the dataset is a property of the VQ codebook: the encoded rows tell
+      only how many bytes it took to compress a vector, which the padding makes ambiguous.
+      */
+      [[nodiscard]] auto dim() const noexcept -> uint32_t
+      {
+        return static_cast<uint32_t>(vq_code_book.extent(1));
+      }
+      [[nodiscard]] auto vq_n_centers() const noexcept -> uint32_t
+      {
+        return static_cast<uint32_t>(vq_code_book.extent(0));
+      }
+      [[nodiscard]] auto n_books() const noexcept -> uint32_t
+      {
+        return static_cast<uint32_t>(pq_code_book.extent(0));
+      }
+      [[nodiscard]] auto pq_n_centers() const noexcept -> uint32_t
+      {
+        return static_cast<uint32_t>(pq_code_book.extent(1));
+      }
+      [[nodiscard]] auto pq_len() const noexcept -> uint32_t
+      {
+        return static_cast<uint32_t>(pq_code_book.extent(2));
+      }
+      [[nodiscard]] auto pq_bits() const noexcept -> uint32_t
+      {
+        return static_cast<uint32_t>(std::countr_zero(pq_n_centers()));
+      }
+      [[nodiscard]] auto pq_dim() const noexcept -> uint32_t
+      {
+        return raft::div_rounding_up_safe(dim(), pq_len());
+      }
+      /* Whether every subspace has a codebook of its own, as opposed to sharing a single one. */
+      [[nodiscard]] auto per_subspace() const noexcept -> bool { return n_books() > 1; }
+    };
+
+    /* NOTE: the data accessors are inherited; only `get_dim` differs from a dense dataset. */
+    [[nodiscard]] static auto get_dim(const data_type&, const dictionary_type& dict) noexcept
+      -> uint32_t
+    {
+      return static_cast<uint32_t>(dict.vq_code_book.extent(1));
+    }
+    [[nodiscard]] static auto get_dictionary_view(const dictionary_type& dict) noexcept
+      -> dictionary_view_type
+    {
+      return {dict.vq_code_book.view(), dict.pq_code_book.view()};
+    }
+    /* The length of an encoded row in bytes, including the inlined VQ label. */
+    [[nodiscard]] static auto get_encoded_row_length(const data_type& data) noexcept -> uint32_t
+    {
+      return static_cast<uint32_t>(data.extent(1));
+    }
+  };
+};
+
+/**
+ * Scalar-quantized dataset specification: every component of a vector becomes one code, so the
+ * dictionary is only the range the codes are mapped back onto and the dimension is still a property
+ * of the data.
+ */
+template <typename MathT, typename CodePolicy>
+struct sq_spec {
+  /* One code per component, so the codes are laid out exactly like a dense dataset. */
+  template <typename IdxT>
+  using storage_spec =
+    typename mdarray_spec<raft::layout_c_contiguous, CodePolicy>::template apply<int8_t, IdxT>;
+
+  template <typename T, typename IdxT>
+  struct apply : storage_spec<IdxT> {
+    /* The dataset stores codes, but it still represents vectors of `T`. */
+    using value_type = std::remove_cv_t<T>;
+    /* NOTE: the members of a dependent base are not visible to unqualified lookup, so `data_type`
+    has to be pulled into this scope to be usable in the signature below.
+    */
+    using typename storage_spec<IdxT>::data_type;
+    using math_type = MathT;
+    using code_type = int8_t;
+
+    /* Nothing to allocate: the whole dictionary is the interval the codes are dequantized into.
+    The members are named as in cuvs::preprocessing::quantize::scalar::quantizer.
+    */
+    struct dictionary_type {
+      math_type min_;
+      math_type max_;
+    };
+    using dictionary_view_type = dictionary_type;
+
+    /* NOTE: only the signature differs from the inherited accessor, which cannot be reused because
+    it expects an empty dictionary.
+    */
+    [[nodiscard]] static auto get_dim(const data_type& data, const dictionary_type&) noexcept
+      -> uint32_t
+    {
+      return static_cast<uint32_t>(data.extent(1));
+    }
+    [[nodiscard]] static auto get_dictionary_view(const dictionary_type& dict) noexcept
+      -> dictionary_view_type
+    {
+      return dict;
+    }
+  };
+};
+
+/************************************************************************************************
+ * Convenience aliases
+ ************************************************************************************************/
+
+// Empty
+template <typename T, typename IdxT>
+using empty_dataset = dataset<T, IdxT, empty_spec>;
+
+// Dense
+template <typename T, typename IdxT, typename ContainerPolicy>
+using contiguous_dataset =
+  dataset<T, IdxT, mdarray_spec<raft::layout_c_contiguous, ContainerPolicy>>;
+template <typename T, typename IdxT, typename ContainerPolicy>
+using padded_dataset = dataset<T, IdxT, mdarray_spec<raft::layout_left_padded<T>, ContainerPolicy>>;
+template <typename T, typename IdxT>
+using device_contiguous_dataset =
+  contiguous_dataset<T, IdxT, raft::device_accessor<raft::device_container_policy<T>>>;
+template <typename T, typename IdxT>
+using device_padded_dataset =
+  padded_dataset<T, IdxT, raft::device_accessor<raft::device_container_policy<T>>>;
+template <typename T, typename IdxT>
+using host_contiguous_dataset =
+  contiguous_dataset<T, IdxT, raft::host_accessor<raft::host_container_policy<T>>>;
+template <typename T, typename IdxT>
+using host_padded_dataset =
+  padded_dataset<T, IdxT, raft::host_accessor<raft::host_container_policy<T>>>;
+
+// Sparse
+template <typename T, typename IdxT, typename ContainerPolicy>
+using csr_dataset = dataset<T, IdxT, sparse_spec<csr_layout, ContainerPolicy>>;
+template <typename T, typename IdxT, typename ContainerPolicy>
+using coo_dataset = dataset<T, IdxT, sparse_spec<coo_layout, ContainerPolicy>>;
+template <typename T, typename IdxT>
+using device_csr_dataset =
+  csr_dataset<T, IdxT, raft::device_accessor<raft::device_container_policy<T>>>;
+template <typename T, typename IdxT>
+using device_coo_dataset =
+  coo_dataset<T, IdxT, raft::device_accessor<raft::device_container_policy<T>>>;
+template <typename T, typename IdxT>
+using host_csr_dataset = csr_dataset<T, IdxT, raft::host_accessor<raft::host_container_policy<T>>>;
+template <typename T, typename IdxT>
+using host_coo_dataset = coo_dataset<T, IdxT, raft::host_accessor<raft::host_container_policy<T>>>;
+
+// Compressed
+template <typename T, typename IdxT, typename MathT, typename BookPolicy, typename CodePolicy>
+using vpq_dataset = dataset<T, IdxT, vpq_spec<MathT, BookPolicy, CodePolicy>>;
+template <typename T, typename IdxT, typename MathT, typename CodePolicy>
+using sq_dataset = dataset<T, IdxT, sq_spec<MathT, CodePolicy>>;
+template <typename T, typename IdxT, typename MathT>
+using device_vpq_dataset =
+  vpq_dataset<T,
+              IdxT,
+              MathT,
+              raft::device_accessor<raft::device_container_policy<MathT>>,
+              raft::device_accessor<raft::device_container_policy<uint8_t>>>;
+template <typename T, typename IdxT, typename MathT>
+using host_vpq_dataset = vpq_dataset<T,
+                                     IdxT,
+                                     MathT,
+                                     raft::host_accessor<raft::host_container_policy<MathT>>,
+                                     raft::host_accessor<raft::host_container_policy<uint8_t>>>;
+template <typename T, typename IdxT, typename MathT>
+using device_sq_dataset =
+  sq_dataset<T, IdxT, MathT, raft::device_accessor<raft::device_container_policy<int8_t>>>;
+template <typename T, typename IdxT, typename MathT>
+using host_sq_dataset =
+  sq_dataset<T, IdxT, MathT, raft::host_accessor<raft::host_container_policy<int8_t>>>;
+
+}  // namespace CUVS_EXPORT cuvs
diff --git a/examples/cpp/CMakeLists.txt b/examples/cpp/CMakeLists.txt
index f4dffad3a9..7a4e0f504f 100644
--- a/examples/cpp/CMakeLists.txt
+++ b/examples/cpp/CMakeLists.txt
@@ -44,6 +44,7 @@ add_executable(CAGRA_BLOOM_FILTER_EXAMPLE src/cagra_bloom_filter_example.cu)
 add_executable(CAGRA_HNSW_ACE_BUILD_EXAMPLE src/cagra_hnsw_ace_build.cu)
 add_executable(CAGRA_HNSW_ACE_EXAMPLE src/cagra_hnsw_ace_example.cu)
 add_executable(CAGRA_PERSISTENT_EXAMPLE src/cagra_persistent_example.cu)
+add_executable(DATASET_EXAMPLE src/dataset_example.cu)
 add_executable(DYNAMIC_BATCHING_EXAMPLE src/dynamic_batching_example.cu)
 add_executable(HNSW_ACE_EXAMPLE src/hnsw_ace_example.cu)
 add_executable(HNSW_OPENAI_EXAMPLE src/hnsw_openai_example.cu)
@@ -73,6 +74,7 @@ target_link_libraries(
 target_link_libraries(
   DYNAMIC_BATCHING_EXAMPLE PRIVATE cuvs::cuvs $<TARGET_NAME_IF_EXISTS:conda_env> Threads::Threads
 )
+target_link_libraries(DATASET_EXAMPLE PRIVATE cuvs::cuvs $<TARGET_NAME_IF_EXISTS:conda_env>)
 target_link_libraries(HNSW_ACE_EXAMPLE PRIVATE cuvs::cuvs $<TARGET_NAME_IF_EXISTS:conda_env>)
 target_link_libraries(HNSW_OPENAI_EXAMPLE PRIVATE cuvs::cuvs $<TARGET_NAME_IF_EXISTS:conda_env>)
 
diff --git a/examples/cpp/src/dataset_example.cu b/examples/cpp/src/dataset_example.cu
new file mode 100644
index 0000000000..af51d6296a
--- /dev/null
+++ b/examples/cpp/src/dataset_example.cu
@@ -0,0 +1,187 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <cuvs/core/dataset.hpp>
+
+#include <raft/core/device_coo_matrix.hpp>
+#include <raft/core/device_csr_matrix.hpp>
+#include <raft/core/device_mdarray.hpp>
+#include <raft/core/host_mdarray.hpp>
+#include <raft/core/math.hpp>
+#include <raft/core/operators.hpp>
+#include <raft/core/resources.hpp>
+#include <raft/linalg/map.cuh>
+#include <raft/linalg/matrix_vector_op.cuh>
+#include <raft/stats/meanvar.cuh>
+
+#include <cstdint>
+#include <iostream>
+#include <memory>
+#include <utility>
+#include <variant>
+
+/** The interface every spec provides, regardless of how the data is stored. */
+template <typename DatasetT>
+void print_dataset(const char* name, const DatasetT& dataset)
+{
+  std::cout << name << ": n_rows = " << dataset.n_rows() << ", dim = " << dataset.dim() << '\n';
+}
+
+/**
+ * A transformation that takes its input by value and hands back the result. Written this way, it
+ * can skip the second buffer whenever the caller passes its only copy of the dataset:
+ * `d = normalize(res, std::move(d))` allocates only the column statistics.
+ */
+template <typename T, typename IdxT>
+auto normalize(const raft::resources& res, cuvs::device_contiguous_dataset<T, IdxT> data)
+  -> cuvs::device_contiguous_dataset<T, IdxT>
+{
+  auto in   = data.data_const_view();
+  auto mean = raft::make_device_vector<T, int64_t>(res, in.extent(1));
+  auto var  = raft::make_device_vector<T, int64_t>(res, in.extent(1));
+  // Both statistics come out of a single sweep over the data.
+  raft::stats::meanvar(res, in, mean.view(), var.view(), false);
+  auto mean_view = raft::make_const_mdspan(mean.view());
+  auto var_view  = raft::make_const_mdspan(var.view());
+  // The linewise op broadcasts the per-column statistics for us; a constant column stays at zero.
+  auto standardize = [] __device__(T x, T mean, T var) -> T {
+    return var > T{0} ? (x - mean) / raft::sqrt(var) : T{0};
+  };
+
+  if (data.is_data_unique()) {
+    // No other copy can observe the buffer, so the transform may write over its own input.
+    raft::linalg::matrix_vector_op<raft::Apply::ALONG_ROWS>(
+      res, in, mean_view, var_view, data.data_view(), standardize);
+    return data;
+  }
+  cuvs::device_contiguous_dataset<T, IdxT> out{
+    raft::make_device_matrix<T, int64_t>(res, in.extent(0), in.extent(1))};
+  raft::linalg::matrix_vector_op<raft::Apply::ALONG_ROWS>(
+    res, in, mean_view, var_view, out.data_view(), standardize);
+  return out;
+}
+
+/* The dictionary slot is free for the datasets that have no dictionary. */
+static_assert(!cuvs::device_contiguous_dataset<float, uint32_t>::is_compressed);
+static_assert(
+  std::same_as<cuvs::device_contiguous_dataset<float, uint32_t>::dictionary_type, std::monostate>);
+static_assert(
+  std::same_as<cuvs::device_csr_dataset<float, uint32_t>::dictionary_type, std::monostate>);
+static_assert(std::same_as<cuvs::empty_dataset<float, uint32_t>::dictionary_type, std::monostate>);
+static_assert(sizeof(cuvs::device_contiguous_dataset<float, uint32_t>) ==
+              sizeof(std::shared_ptr<void>));
+static_assert(cuvs::device_vpq_dataset<float, uint32_t, float>::is_compressed);
+
+int main()
+{
+  raft::resources res;
+
+  int64_t n_rows = 8;
+  int64_t dim    = 4;
+
+  // An empty dataset carries the dimension only; there is no storage behind it.
+  cuvs::empty_dataset<float, uint32_t> empty{{static_cast<uint32_t>(dim)}};
+  print_dataset("empty", empty);
+
+  // A host dataset takes ownership of a raft::mdarray of a matching layout and storage policy.
+  auto host_data = raft::make_host_matrix<float, int64_t>(res, n_rows, dim);
+  for (int64_t i = 0; i < n_rows; i++) {
+    for (int64_t j = 0; j < dim; j++) {
+      host_data(i, j) = static_cast<float>(i * dim + j);
+    }
+  }
+  cuvs::host_contiguous_dataset<float, uint32_t> host_dataset{std::move(host_data)};
+  print_dataset("host contiguous", host_dataset);
+  std::cout << "  last element = " << host_dataset.data_view()(n_rows - 1, dim - 1) << '\n';
+
+  // The device variant differs only in the storage policy baked into the spec.
+  cuvs::device_contiguous_dataset<float, uint32_t> device_dataset{
+    raft::make_device_matrix<float, int64_t>(res, n_rows, dim)};
+  print_dataset("device contiguous", device_dataset);
+
+  // Moving the dataset in leaves `normalize` as its only owner, so the data is scaled in-place.
+  raft::linalg::map_offset(res, device_dataset.data_view(), raft::cast_op<float>{});
+  const auto* original_data = device_dataset.data_const_view().data_handle();
+  device_dataset            = normalize(res, std::move(device_dataset));
+  std::cout << "  reused the input buffer: " << std::boolalpha
+            << (device_dataset.data_const_view().data_handle() == original_data) << '\n';
+
+  // Passing a copy keeps a second owner alive, so the same call has to allocate its output.
+  auto shared_dataset = device_dataset;
+  auto rescaled       = normalize(res, shared_dataset);
+  std::cout << "  reused the input buffer: "
+            << (rescaled.data_const_view().data_handle() == original_data) << '\n';
+
+  // The same dataset over a padded layout: the spec, not the dataset, selects the layout.
+  using padded_dataset_type = cuvs::device_padded_dataset<float, uint32_t>;
+  using padded_data_type    = typename padded_dataset_type::data_type;
+  padded_dataset_type padded_dataset{padded_data_type{
+    res,
+    typename padded_data_type::mapping_type{raft::make_extents<int64_t>(n_rows, dim)},
+    typename padded_data_type::container_policy_type{}}};
+  print_dataset("device padded", padded_dataset);
+
+  // A sparse dataset: one CSR row per vector, so `dim` is the number of features.
+  uint64_t nnz = static_cast<uint64_t>(n_rows) * 2;
+  cuvs::device_csr_dataset<float, uint32_t> csr_dataset{
+    raft::make_device_csr_matrix<float, int64_t, uint32_t, uint64_t>(
+      res, n_rows, static_cast<uint32_t>(dim), nnz)};
+  print_dataset("device csr", csr_dataset);
+  std::cout << "  nnz = " << csr_dataset.data_view().structure_view().get_nnz() << '\n';
+
+  // The same dataset as a list of (row, column, value) triples.
+  cuvs::device_coo_dataset<float, uint32_t> coo_dataset{
+    raft::make_device_coo_matrix<float, uint32_t, uint32_t, uint64_t>(
+      res, static_cast<uint32_t>(n_rows), static_cast<uint32_t>(dim), nnz)};
+  print_dataset("device coo", coo_dataset);
+
+  // A compressed dataset fills the second slot with a dictionary; for VPQ that is the pair of
+  // codebooks. Note the dimension of the dataset comes from the VQ codebook here, since the encoded
+  // rows know only how many bytes it took to compress a vector.
+  using vpq_dataset_type    = cuvs::device_vpq_dataset<float, uint32_t, float>;
+  using vpq_dictionary_type = typename vpq_dataset_type::dictionary_type;
+
+  int64_t vq_n_centers = 4;
+  int64_t pq_n_centers = 256;
+  int64_t pq_len       = 2;
+  // One byte per code at pq_bits == 8, prefixed with the inlined VQ label.
+  int64_t encoded_row_length = sizeof(uint32_t) + dim / pq_len;
+
+  vpq_dictionary_type codebooks{
+    raft::make_device_matrix<float, int64_t>(res, vq_n_centers, dim),
+    // A single codebook shared by all the subspaces; `pq_dim` of them would be per-subspace.
+    raft::make_device_mdarray<float, int64_t>(
+      res, raft::make_extents<int64_t>(1, pq_n_centers, pq_len))};
+  auto dictionary = std::make_shared<const vpq_dictionary_type>(std::move(codebooks));
+
+  vpq_dataset_type vpq_dataset{
+    raft::make_device_matrix<uint8_t, int64_t>(res, n_rows, encoded_row_length), dictionary};
+  print_dataset("device vpq", vpq_dataset);
+  auto books = vpq_dataset.dictionary_view();
+  std::cout << "  vq_n_centers = " << books.vq_n_centers() << ", pq_bits = " << books.pq_bits()
+            << ", pq_dim = " << books.pq_dim() << ", per_subspace = " << books.per_subspace()
+            << '\n';
+  // The data slot is a plain dense matrix of bytes, hence the shared spec.
+  std::cout << "  encoded row length = " << vpq_dataset.data_view().extent(1) << '\n';
+
+  // A dictionary is immutable, so another dataset may be encoded against the very same codebooks.
+  vpq_dataset_type shared_vpq_dataset{
+    raft::make_device_matrix<uint8_t, int64_t>(res, n_rows * 2, encoded_row_length),
+    vpq_dataset.share_dictionary()};
+  print_dataset("device vpq (shared dictionary)", shared_vpq_dataset);
+
+  // Scalar quantization keeps one code per component, so the dimension is still a property of the
+  // data and the dictionary is only the interval to dequantize into.
+  using sq_dataset_type    = cuvs::device_sq_dataset<float, uint32_t, float>;
+  using sq_dictionary_type = typename sq_dataset_type::dictionary_type;
+  sq_dataset_type sq_dataset{
+    raft::make_device_matrix<int8_t, int64_t>(res, n_rows, dim),
+    std::make_shared<const sq_dictionary_type>(sq_dictionary_type{-1.0F, 1.0F})};
+  print_dataset("device sq", sq_dataset);
+  std::cout << "  range = [" << sq_dataset.dictionary_view().min_ << ", "
+            << sq_dataset.dictionary_view().max_ << "]\n";
+
+  return 0;
+}