Remove dataset struct

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

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)?))
     }
 }
 

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

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.

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");

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);
 

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.

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