diff --git a/crates/strata/src/bin/loadgen.rs b/crates/strata/src/bin/loadgen.rs
index 91c6f93..22c8615 100644
--- a/crates/strata/src/bin/loadgen.rs
+++ b/crates/strata/src/bin/loadgen.rs
@@ -14,7 +14,7 @@ use schema::{DataType, SchemaBuilder};
 use serde_json::Value;
 use strata::datagen::Generator;
 use strata::harness::{Plan, Stop, run};
-use strata::{Body, Dataset, Method};
+use strata::{Body, DataStream, Method};
 
 #[derive(Parser)]
 #[command(about = "Measure write throughput against a provider endpoint")]
@@ -80,7 +80,7 @@ async fn main() -> Result<()> {
             let start = offset as usize;
             let records = generator.rows(start..start + batch)?;
             let body = Body {
-                data: Some(Dataset::new(schema.clone(), records).into_stream()),
+                data: Some(DataStream::once(schema.clone(), records)),
                 meta: Value::Null,
             };
             let response = provider
diff --git a/crates/strata/src/datagen.rs b/crates/strata/src/datagen.rs
index f9939c3..0afc3de 100644
--- a/crates/strata/src/datagen.rs
+++ b/crates/strata/src/datagen.rs
@@ -11,8 +11,7 @@ use anyhow::Result;
 use schema::{DataType, Field, Schema};
 use serde_json::{Map, Value, json};
 
-use crate::dataset::Dataset;
-use crate::record::{Batch, stringify_text_columns};
+use crate::record::{Batch, DataStream, stringify_text_columns};
 
 pub struct Generator {
     schema: Schema,
@@ -48,9 +47,9 @@ impl Generator {
         Batch::encode(&self.schema, &rows)
     }
 
-    /// The first `n` rows as a [`Dataset`] (schema + Arrow records).
-    pub fn dataset(&self, n: usize) -> Result<Dataset> {
-        Ok(Dataset::new(self.schema.clone(), self.rows(0..n)?))
+    /// The first `n` rows as a single-page [`DataStream`], ready to `put`.
+    pub fn stream(&self, n: usize) -> Result<DataStream> {
+        Ok(DataStream::once(self.schema.clone(), self.rows(0..n)?))
     }
 }
 
diff --git a/crates/strata/src/dataset.rs b/crates/strata/src/dataset.rs
deleted file mode 100644
index 9965e1e..0000000
--- a/crates/strata/src/dataset.rs
+++ /dev/null
@@ -1,106 +0,0 @@
-//! A dataset: the `(schema, rows)` unit that flows from a reader to a writer.
-//!
-//! It's exactly what a read produces — the rows of a `list`/`get` plus that
-//! endpoint's resolved [`DataType`] schema — so a writer can use the schema as
-//! the contract (create a table if absent, validate if present) and load the
-//! rows. This is the framework-level shape behind "pipe readers to writers";
-//! over Arrow Flight the same `(schema, rows)` rides natively in a `DoPut`.
-
-use anyhow::{Result, bail};
-use futures::stream::StreamExt;
-use schema::Schema;
-use serde::Serialize;
-use serde_json::Value;
-
-use crate::{
-    DataStream,
-    record::{Batch, BatchPage},
-};
-
-/// How a sink should apply a written dataset. Rides as metadata on the existing
-/// `put` verb (the reserved `disposition` query param) rather than a new verb, so
-/// every sink shares one write surface.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize)]
-pub enum Disposition {
-    /// Insert every row (the historical behavior). Re-running adds duplicates.
-    #[default]
-    Append,
-    /// Idempotent write-by-key: upsert each row on the dataset's key fields
-    /// (insert, or update the non-key columns on conflict). Requires the schema
-    /// to declare a key; this is what makes a re-fetching pipe dedup itself.
-    Merge,
-}
-
-impl Disposition {
-    /// The reserved `put` query param that selects the disposition.
-    pub const PARAM: &str = "disposition";
-
-    /// Parse the param value (`append` | `merge`/`upsert`); defaults to `Append`
-    /// when absent.
-    pub fn from_param(value: Option<&str>) -> Result<Self> {
-        match value {
-            None | Some("append") => Ok(Disposition::Append),
-            Some("merge") | Some("upsert") => Ok(Disposition::Merge),
-            Some(other) => bail!("unknown write disposition `{other}` (append|merge)"),
-        }
-    }
-
-    /// The param value (inverse of [`from_param`](Self::from_param)).
-    pub fn as_param(self) -> &'static str {
-        match self {
-            Disposition::Append => "append",
-            Disposition::Merge => "merge",
-        }
-    }
-}
-
-/// The reader→writer unit: the rows as Arrow [`Records`] (strata's internal
-/// currency) paired with the **native** `schema` that governs how a sink stores
-/// them. `schema` is carried explicitly rather than derived from the Arrow
-/// schema, so `Decimal`/`Json` fidelity survives — the Arrow mapping folds those
-/// to `Utf8`, and a sink's DDL needs the real types. It's typically a
-/// `List(Struct(..))` (a list endpoint's response) or a bare `Struct`.
-///
-/// Future direction: `records` should become a *stream* of batches following
-/// `schema` (a checkpointed cursor-driven stream) so a writer can consume
-/// unbounded data without materializing it — the materialized batches here are
-/// the stand-in until the streaming list driver lands.
-#[derive(Debug, Clone)]
-pub struct Dataset {
-    pub schema: Schema,
-    pub records: Batch,
-}
-
-impl Dataset {
-    pub fn new(schema: Schema, records: Batch) -> Self {
-        Dataset { schema, records }
-    }
-
-    /// Build a dataset from typed `rows`: the schema is `T`'s [`Schema`] (with any
-    /// `#[schema(key)]`/annotations it declares) and the rows are Arrow-encoded
-    /// against it. The typed-input counterpart of a `put`.
-    pub fn of<T: serde::Serialize + schema::HasSchema>(rows: &[T]) -> Result<Self> {
-        let schema = T::schema();
-        let records = Batch::encode(&schema, rows)?;
-        Ok(Dataset::new(schema, records))
-    }
-
-    /// Interim bridge: decode the Arrow rows to JSON for sinks that aren't yet
-    /// Arrow-native. This is the one labeled Arrow→JSON on the write path; Phase B
-    /// removes it as each sink binds Arrow columns directly.
-    pub fn to_json_rows(&self) -> Result<Vec<Value>> {
-        self.records.to_json_rows(&self.schema)
-    }
-
-    /// Bridge to a single-chunk [`DataStream`], until readers produce batches lazily.
-    pub fn into_stream(self) -> DataStream {
-        let chunk = BatchPage {
-            data: self.records,
-            cursor: None,
-        };
-        DataStream {
-            schema: self.schema,
-            chunks: futures::stream::once(async move { Ok(chunk) }).boxed(),
-        }
-    }
-}
diff --git a/crates/strata/src/graphql/mod.rs b/crates/strata/src/graphql/mod.rs
index 03539e0..db5afbc 100644
--- a/crates/strata/src/graphql/mod.rs
+++ b/crates/strata/src/graphql/mod.rs
@@ -404,7 +404,7 @@ mod tests {
         rows: &[T],
     ) -> Result<()> {
         let body = crate::Body {
-            data: Some(crate::Dataset::of(rows)?.into_stream()),
+            data: Some(crate::DataStream::of(rows)?),
             meta: Value::Null,
         };
         registry
diff --git a/crates/strata/src/lib.rs b/crates/strata/src/lib.rs
index 59c2f06..7087f5d 100644
--- a/crates/strata/src/lib.rs
+++ b/crates/strata/src/lib.rs
@@ -4,7 +4,6 @@
 pub mod catalog;
 pub mod config;
 pub mod datagen;
-pub mod dataset;
 pub mod flight;
 pub mod graphql;
 pub mod harness;
@@ -25,10 +24,9 @@ use anyhow::Result;
 
 pub use catalog::Catalog;
 pub use config::{Config, ProviderConfig};
-pub use dataset::{Dataset, Disposition};
 pub use page::{Cursor, Page};
 pub use provider::{Provider, ProviderObject, Registry};
-pub use record::DataStream;
+pub use record::{DataStream, Disposition};
 pub use router::{Body, EndpointInfo, Method, Params, Response, Router};
 
 /// Config file consulted by [`registry`] when no explicit path is given.
diff --git a/crates/strata/src/providers/clickhouse/mod.rs b/crates/strata/src/providers/clickhouse/mod.rs
index b6c6ed0..0801031 100644
--- a/crates/strata/src/providers/clickhouse/mod.rs
+++ b/crates/strata/src/providers/clickhouse/mod.rs
@@ -5,9 +5,9 @@ use schema::{DataType, Field, Schema};
 use serde::Deserialize;
 use serde_json::Value;
 
-use crate::dataset::{Dataset, Disposition};
 use crate::provider::Provider;
 use crate::providers::sql::{self, Filter, SqlCursor, SqlError, SqlSource, WriteResult, quote_str};
+use crate::record::{Batch, Disposition};
 use crate::router::Router;
 
 #[config]
@@ -197,10 +197,10 @@ impl SqlSource for Clickhouse {
     async fn write_table(
         &self,
         table: &str,
-        data: Dataset,
+        schema: &Schema,
+        data: Batch,
         _disposition: Disposition,
     ) -> Result<WriteResult> {
-        let schema = &data.schema;
         let ident = quote_ident(table);
 
         // Build a single `INSERT … FORMAT JSONEachRow` body: one JSON object per row,
@@ -209,7 +209,7 @@ impl SqlSource for Clickhouse {
         // ClickHouse's HTTP interface speaks JSONEachRow, so decode the Arrow rows to
         // JSON here — this is the one provider where JSON is the wire protocol, not an
         // interim bridge.
-        let rows = data.to_json_rows()?;
+        let rows = data.to_json_rows(schema)?;
         let mut rows_written = 0u64;
         if !rows.is_empty() {
             let mut body = format!("INSERT INTO {ident} FORMAT JSONEachRow\n");
diff --git a/crates/strata/src/providers/iceberg/mod.rs b/crates/strata/src/providers/iceberg/mod.rs
index 8536990..bdbf2ab 100644
--- a/crates/strata/src/providers/iceberg/mod.rs
+++ b/crates/strata/src/providers/iceberg/mod.rs
@@ -29,10 +29,9 @@ mod convert;
 
 use convert::{align_to, iceberg_to_strata_schema, strata_to_iceberg_schema};
 
-use crate::dataset::Disposition;
 use crate::page::{Cursor, ListStrategy, Page};
 use crate::provider::Provider;
-use crate::record::{Batch, BatchPage, DataStream};
+use crate::record::{Batch, BatchPage, DataStream, Disposition};
 use crate::router::{Pages, Params, Route, Router};
 
 /// All strata tables live in one namespace.
@@ -440,11 +439,14 @@ mod tests {
         let c = client(&dir)?;
 
         const ROWS: usize = 100;
-        let generator = crate::datagen::Generator::new(&Event::schema())?;
-        let dataset = generator.dataset(ROWS)?;
-        let expected: Vec<Event> = dataset.records.decode(&dataset.schema)?;
+        let schema = Event::schema();
+        let generator = crate::datagen::Generator::new(&schema)?;
+        let batch = generator.rows(0..ROWS)?;
+        let expected: Vec<Event> = batch.decode(&schema)?;
 
-        let result: WriteResult = c.put("/tables/events", dataset).await?;
+        let result: WriteResult = c
+            .put("/tables/events", DataStream::once(schema, batch))
+            .await?;
         assert!(result.created);
         assert_eq!(result.rows_written, ROWS as u64);
 
diff --git a/crates/strata/src/providers/mysql/mod.rs b/crates/strata/src/providers/mysql/mod.rs
index fda2f5a..1714f92 100644
--- a/crates/strata/src/providers/mysql/mod.rs
+++ b/crates/strata/src/providers/mysql/mod.rs
@@ -5,11 +5,11 @@ use serde_json::Value;
 use sqlx::Row as _;
 use sqlx::mysql::{MySqlArguments, MySqlPool};
 
-use crate::dataset::{Dataset, Disposition};
 use crate::provider::Provider;
 use crate::providers::sql::{
     self, Filter, SqlCursor, SqlError, SqlSource, WriteResult, is_table_not_found, quote_str,
 };
+use crate::record::{Batch, Disposition};
 use crate::router::Router;
 
 /// Placeholders per statement. MySQL's protocol caps them at 65535; stay well
@@ -166,11 +166,7 @@ impl SqlSource for Mysql {
             if !keys.is_empty() {
                 cols.push(format!("PRIMARY KEY ({})", quote_idents(&key_refs)));
             }
-            let ddl = format!(
-                "CREATE TABLE {} ({})",
-                quote_ident(table),
-                cols.join(", ")
-            );
+            let ddl = format!("CREATE TABLE {} ({})", quote_ident(table), cols.join(", "));
             sqlx::query(&ddl)
                 .execute(&pool)
                 .await
@@ -192,11 +188,12 @@ impl SqlSource for Mysql {
     async fn write_table(
         &self,
         table: &str,
-        data: Dataset,
+        schema: &Schema,
+        data: Batch,
         disposition: Disposition,
     ) -> Result<WriteResult> {
-        let fields = &data.schema.fields;
-        let keys = data.schema.get_key_fields();
+        let fields = &schema.fields;
+        let keys = schema.get_key_fields();
         let key_refs: Vec<&str> = keys.iter().map(String::as_str).collect();
         let ident = quote_ident(table);
         let pool = self.connect().await?;
@@ -234,7 +231,7 @@ impl SqlSource for Mysql {
         };
         // Interim: decode the Arrow rows to JSON to bind positionally (Phase B binds
         // Arrow columns directly).
-        let rows = data.to_json_rows()?;
+        let rows = data.to_json_rows(schema)?;
 
         // One statement per chunk of rows rather than one per row, chunked by the
         // placeholder budget a single statement can carry.
diff --git a/crates/strata/src/providers/postgres/mod.rs b/crates/strata/src/providers/postgres/mod.rs
index 9b1c9b2..7a3817c 100644
--- a/crates/strata/src/providers/postgres/mod.rs
+++ b/crates/strata/src/providers/postgres/mod.rs
@@ -3,11 +3,11 @@ use schema::{DataType, Field, Schema};
 use serde_json::Value;
 use tokio_postgres::NoTls;
 
-use crate::dataset::{Dataset, Disposition};
 use crate::provider::Provider;
 use crate::providers::sql::{
     self, Filter, SqlCursor, SqlError, SqlSource, WriteResult, is_table_not_found,
 };
+use crate::record::{Batch, Disposition};
 use crate::router::Router;
 use config_macro::config;
 
@@ -182,11 +182,12 @@ impl SqlSource for Postgres {
     async fn write_table(
         &self,
         table: &str,
-        data: Dataset,
+        schema: &Schema,
+        data: Batch,
         disposition: Disposition,
     ) -> Result<WriteResult> {
-        let fields = &data.schema.fields;
-        let keys = data.schema.get_key_fields();
+        let fields = &schema.fields;
+        let keys = schema.get_key_fields();
         let key_refs: Vec<&str> = keys.iter().map(String::as_str).collect();
         let ident = quote_ident(table);
         let client = self.connect().await?;
@@ -226,7 +227,7 @@ impl SqlSource for Postgres {
         );
         // Interim: decode the Arrow rows to JSON to bind as jsonb (Phase B binds
         // Arrow columns directly).
-        let rows = Value::Array(data.to_json_rows()?);
+        let rows = Value::Array(data.to_json_rows(schema)?);
         let rows_written = client
             .execute(&insert, &[&rows])
             .await
diff --git a/crates/strata/src/providers/sql/mod.rs b/crates/strata/src/providers/sql/mod.rs
index 20b48d6..8bdf6f5 100644
--- a/crates/strata/src/providers/sql/mod.rs
+++ b/crates/strata/src/providers/sql/mod.rs
@@ -22,10 +22,9 @@ use schema::{DataType, HasSchema, Schema};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
-use crate::dataset::{Dataset, Disposition};
 use crate::page::{Cursor, ListStrategy, Page};
 use crate::provider::Provider;
-use crate::record::{Batch, BatchPage, DataStream, stringify_text_columns};
+use crate::record::{Batch, BatchPage, DataStream, Disposition, stringify_text_columns};
 use crate::router::{Pages, Params, Route, Router};
 
 mod filter;
@@ -214,7 +213,8 @@ pub trait SqlSource: Send + Sync + 'static {
     fn write_table(
         &self,
         table: &str,
-        data: Dataset,
+        schema: &Schema,
+        data: Batch,
         disposition: Disposition,
     ) -> impl Future<Output = Result<WriteResult>> + Send;
 
@@ -236,12 +236,8 @@ pub trait SqlSource: Send + Sync + 'static {
             };
             let mut rows_written = 0;
             while let Some(chunk) = chunks.next().await {
-                // TODO: We still create a dataset object, later on we should send
-                // the stream directly to the SQL provider but we would need a way for the provider
-                // to notify which batch has been written.
-                let dataset = Dataset::new(schema.clone(), chunk?.data);
                 rows_written += self
-                    .write_table(table, dataset, disposition)
+                    .write_table(table, &schema, chunk?.data, disposition)
                     .await?
                     .rows_written;
             }
@@ -552,8 +548,8 @@ fn next_cursor(cursor: &SqlCursor, returned: usize) -> Result<Cursor> {
 #[cfg(test)]
 pub mod suite {
     use super::{TableName, WriteResult};
-    use crate::dataset::Dataset;
     use crate::provider::Provider;
+    use crate::record::DataStream;
     use crate::testkit::Client;
     use anyhow::Result;
     use schema::{DataType, HasSchema};
@@ -587,7 +583,9 @@ pub mod suite {
             id: 1,
             name: "a".into(),
         }];
-        let _: WriteResult = client.put("/tables/catalog", Dataset::of(&rows)?).await?;
+        let _: WriteResult = client
+            .put("/tables/catalog", DataStream::of(&rows)?)
+            .await?;
 
         let mut tables = client.list("/tables").await?;
         let names: Vec<TableName> = tables.next().await?;
@@ -613,7 +611,7 @@ pub mod suite {
         client: &Client<S>,
     ) -> Result<()> {
         let generator = crate::datagen::Generator::new(&Event::schema())?;
-        let result: WriteResult = client.put("/tables/events", generator.dataset(20)?).await?;
+        let result: WriteResult = client.put("/tables/events", generator.stream(20)?).await?;
         assert!(result.created);
         assert_eq!(result.rows_written, 20);
 
@@ -631,7 +629,7 @@ pub mod suite {
 
         let generator = crate::datagen::Generator::new(&Event::schema())?;
         let result: WriteResult = client
-            .put("/tables/streamed", generator.dataset(ROWS)?)
+            .put("/tables/streamed", generator.stream(ROWS)?)
             .await?;
         assert_eq!(result.rows_written, ROWS as u64);
 
@@ -673,8 +671,8 @@ pub mod suite {
                 name: "b".into(),
             },
         ];
-        let _: WriteResult = client.put("/tables/dedup", Dataset::of(&rows)?).await?;
-        let _: WriteResult = client.put("/tables/dedup", Dataset::of(&rows)?).await?;
+        let _: WriteResult = client.put("/tables/dedup", DataStream::of(&rows)?).await?;
+        let _: WriteResult = client.put("/tables/dedup", DataStream::of(&rows)?).await?;
 
         let mut stream = client.list("/tables/dedup").await?;
         let found_rows: Vec<Row> = stream.next().await?;
@@ -692,7 +690,7 @@ pub mod suite {
         let _: WriteResult = client
             .put(
                 "/tables/up",
-                Dataset::of(&[Row {
+                DataStream::of(&[Row {
                     id: 1,
                     name: "a".into(),
                 }])?,
@@ -701,7 +699,7 @@ pub mod suite {
         let _: WriteResult = client
             .put(
                 "/tables/up?disposition=merge",
-                Dataset::of(&[Row {
+                DataStream::of(&[Row {
                     id: 1,
                     name: "b".into(),
                 }])?,
@@ -739,7 +737,9 @@ pub mod suite {
                 name: "d".into(),
             },
         ];
-        let _: WriteResult = client.put("/tables/filtered", Dataset::of(&rows)?).await?;
+        let _: WriteResult = client
+            .put("/tables/filtered", DataStream::of(&rows)?)
+            .await?;
 
         let predicate = serde_json::json!({
             "and": [
@@ -781,7 +781,9 @@ pub mod suite {
                 name: "b".into(),
             },
         ];
-        let _: WriteResult = client.put("/tables/projected", Dataset::of(&rows)?).await?;
+        let _: WriteResult = client
+            .put("/tables/projected", DataStream::of(&rows)?)
+            .await?;
 
         let mut stream = client
             .list::<RowProjected>("/tables/projected?fields=name")
@@ -819,7 +821,7 @@ pub mod suite {
                 name: "b".into(),
             },
         ];
-        let _: WriteResult = client.put("/tables/getone", Dataset::of(&rows)?).await?;
+        let _: WriteResult = client.put("/tables/getone", DataStream::of(&rows)?).await?;
 
         let row: Row = client.get("/tables/getone/2").await?;
         assert_eq!(row.id, 2);
diff --git a/crates/strata/src/providers/sqlite/mod.rs b/crates/strata/src/providers/sqlite/mod.rs
index 5928a04..847c215 100644
--- a/crates/strata/src/providers/sqlite/mod.rs
+++ b/crates/strata/src/providers/sqlite/mod.rs
@@ -23,11 +23,11 @@ use serde_json::Value;
 use sqlx::Row as _;
 use sqlx::sqlite::{SqliteArguments, SqliteConnectOptions, SqlitePool};
 
-use crate::dataset::{Dataset, Disposition};
 use crate::provider::Provider;
 use crate::providers::sql::{
     self, Filter, SqlCursor, SqlError, SqlSource, WriteResult, is_table_not_found, quote_str,
 };
+use crate::record::{Batch, Disposition};
 use crate::router::Router;
 
 /// Bound parameters per statement. SQLite's own cap is 999 before 3.32 and higher
@@ -187,11 +187,7 @@ impl SqlSource for Sqlite {
             if !keys.is_empty() {
                 cols.push(format!("PRIMARY KEY ({})", quote_idents(&key_refs)));
             }
-            let ddl = format!(
-                "CREATE TABLE {} ({})",
-                quote_ident(table),
-                cols.join(", ")
-            );
+            let ddl = format!("CREATE TABLE {} ({})", quote_ident(table), cols.join(", "));
             sqlx::query(&ddl)
                 .execute(&pool)
                 .await
@@ -213,11 +209,12 @@ impl SqlSource for Sqlite {
     async fn write_table(
         &self,
         table: &str,
-        data: Dataset,
+        schema: &Schema,
+        data: Batch,
         disposition: Disposition,
     ) -> Result<WriteResult> {
-        let fields = &data.schema.fields;
-        let keys = data.schema.get_key_fields();
+        let fields = &schema.fields;
+        let keys = schema.get_key_fields();
         let key_refs: Vec<&str> = keys.iter().map(String::as_str).collect();
         let ident = quote_ident(table);
         let pool = self.connect().await?;
@@ -255,7 +252,7 @@ impl SqlSource for Sqlite {
         };
         // Interim: decode the Arrow rows to JSON to bind positionally (Phase B binds
         // Arrow columns directly).
-        let rows = data.to_json_rows()?;
+        let rows = data.to_json_rows(schema)?;
 
         // One statement per chunk of rows rather than one per row, all inside a
         // single transaction — SQLite otherwise autocommits (and fsyncs) every insert.
diff --git a/crates/strata/src/record.rs b/crates/strata/src/record.rs
index 5b481eb..b34a88b 100644
--- a/crates/strata/src/record.rs
+++ b/crates/strata/src/record.rs
@@ -1,12 +1,12 @@
 use std::sync::Arc;
 
-use anyhow::{Result, anyhow};
+use anyhow::{Result, anyhow, bail};
 use arrow::datatypes::{DataType as ArrowType, Field, FieldRef, Fields, TimeUnit};
 use arrow::record_batch::RecordBatch;
 use arrow_array::Array;
 use arrow_schema::Schema;
 use futures::stream::{BoxStream, StreamExt};
-use schema::{Annotations, DataType, Schema as StrataSchema};
+use schema::{Annotations, DataType, HasSchema, Schema as StrataSchema};
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use serde_json::Value;
@@ -22,6 +22,24 @@ impl DataStream {
     pub async fn first(mut self) -> Result<Option<BatchPage>> {
         self.chunks.next().await.transpose()
     }
+
+    /// A stream of exactly one page — the shape a caller that already has all the
+    /// rows in hand writes into a sink.
+    pub fn once(schema: StrataSchema, data: Batch) -> DataStream {
+        let page = BatchPage { data, cursor: None };
+        DataStream {
+            schema,
+            chunks: futures::stream::once(async move { Ok(page) }).boxed(),
+        }
+    }
+
+    /// One page built from typed `rows`: the schema is `T`'s, with whatever
+    /// `#[schema(key)]` annotations it declares, and the rows are encoded against it.
+    pub fn of<T: Serialize + HasSchema>(rows: &[T]) -> Result<DataStream> {
+        let schema = T::schema();
+        let data = Batch::encode(&schema, rows)?;
+        Ok(DataStream::once(schema, data))
+    }
 }
 pub struct BatchPage {
     pub data: Batch,
@@ -222,6 +240,43 @@ fn data_type_from_arrow(ty: &ArrowType) -> DataType {
     }
 }
 
+/// How a sink should apply a written dataset. Rides as metadata on the existing
+/// `put` verb (the reserved `disposition` query param) rather than a new verb, so
+/// every sink shares one write surface.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize)]
+pub enum Disposition {
+    /// Insert every row (the historical behavior). Re-running adds duplicates.
+    #[default]
+    Append,
+    /// Idempotent write-by-key: upsert each row on the dataset's key fields
+    /// (insert, or update the non-key columns on conflict). Requires the schema
+    /// to declare a key; this is what makes a re-fetching pipe dedup itself.
+    Merge,
+}
+
+impl Disposition {
+    /// The reserved `put` query param that selects the disposition.
+    pub const PARAM: &str = "disposition";
+
+    /// Parse the param value (`append` | `merge`/`upsert`); defaults to `Append`
+    /// when absent.
+    pub fn from_param(value: Option<&str>) -> Result<Self> {
+        match value {
+            None | Some("append") => Ok(Disposition::Append),
+            Some("merge") | Some("upsert") => Ok(Disposition::Merge),
+            Some(other) => bail!("unknown write disposition `{other}` (append|merge)"),
+        }
+    }
+
+    /// The param value (inverse of [`from_param`](Self::from_param)).
+    pub fn as_param(self) -> &'static str {
+        match self {
+            Disposition::Append => "append",
+            Disposition::Merge => "merge",
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/crates/strata/src/request.rs b/crates/strata/src/request.rs
index 922561a..77df413 100644
--- a/crates/strata/src/request.rs
+++ b/crates/strata/src/request.rs
@@ -2,7 +2,7 @@
 //! resume cursor or a write disposition without hand-encoding `?cursor=` onto a path
 //! string. Each wraps a path and yields the encoded path via `.path()`.
 
-use crate::dataset::Disposition;
+use crate::record::Disposition;
 use crate::router::CURSOR_PARAM;
 
 /// A read path plus an optional resume `cursor`.
@@ -53,7 +53,11 @@ impl WriteRequest {
 
     /// The encoded path, with `?disposition=…` set to match the disposition.
     pub fn path(self) -> String {
-        set_param(&self.path, Disposition::PARAM, self.disposition.map(|d| d.as_param()))
+        set_param(
+            &self.path,
+            Disposition::PARAM,
+            self.disposition.map(|d| d.as_param()),
+        )
     }
 }
 
diff --git a/crates/strata/src/router.rs b/crates/strata/src/router.rs
index 9a06f2f..1a61e19 100644
--- a/crates/strata/src/router.rs
+++ b/crates/strata/src/router.rs
@@ -21,9 +21,8 @@ use futures::StreamExt;
 use futures::stream::BoxStream;
 
 use crate::DataStream;
-use crate::dataset::Disposition;
 use crate::page::{ListStrategy, Page};
-use crate::record::{Batch, BatchPage};
+use crate::record::{Batch, BatchPage, Disposition};
 
 /// A boxed, owned future. `'static` because handlers take owned `Params` and an
 /// `Arc<S>`, so nothing is borrowed across the await.
diff --git a/crates/strata/src/testkit.rs b/crates/strata/src/testkit.rs
index 099651b..46a8496 100644
--- a/crates/strata/src/testkit.rs
+++ b/crates/strata/src/testkit.rs
@@ -14,7 +14,6 @@ use serde::Serialize;
 use serde::de::DeserializeOwned;
 use serde_json::Value;
 
-use crate::dataset::Dataset;
 use crate::provider::Provider;
 use crate::record::DataStream;
 use crate::router::{Body, BoxFuture, Method, Response, Router, SchemaSource};
@@ -73,9 +72,9 @@ impl<S: Provider> Client<S> {
         Ok(serde_json::from_value(r.entity.unwrap_or(Value::Null))?)
     }
 
-    pub async fn put<T: DeserializeOwned>(&self, path: &str, data: Dataset) -> Result<T> {
+    pub async fn put<T: DeserializeOwned>(&self, path: &str, data: DataStream) -> Result<T> {
         let body = Body {
-            data: Some(data.into_stream()),
+            data: Some(data),
             meta: Value::Null,
         };
         let r = self.dispatch(Method::Put, path, Some(body)).await?;