diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d111120..1226776 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,6 +39,10 @@ jobs: cargo check -p lakekit-iceberg --features write cargo check -p lakekit-iceberg --features arrow cargo check -p lakekit-iceberg --features parquet-stats + cargo check -p lakekit-core --no-default-features + cargo check -p lakekit-core --features tokio + cargo check -p lakekit-core --features getrandom + cargo check -p lakekit --features object-store wasm: name: WebAssembly (browser and WASI targets) diff --git a/AGENTS.md b/AGENTS.md index 34c8930..cccc979 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,7 +19,7 @@ Timelake and lakeglass use it. - **Keep things in sync:** a change in behaviour updates its spec (status, design, history), the tests, and `CHANGELOG.md` (under `Unreleased`) in the same commit. - **Everything builds for WASM.** Every crate builds for `wasm32-unknown-unknown`: no threads, no blocking I/O, no - `std::time`, no C dependencies. All I/O goes through the `Host` trait. CI checks it. + `std::time`, no C dependencies. All I/O goes through the `Store`, `Clock`, and `Random` traits. CI checks it. - **Readers don't trust files.** Bytes from a store can be anything: check lengths, return errors, never panic. - **Interop is tested.** What a format crate writes must read in that format's reference implementation and the other way round; the tests show it. diff --git a/CHANGELOG.md b/CHANGELOG.md index b642f9d..fbec074 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,11 @@ versions follow [Semantic Versioning](https://semver.org/). ### Added -- `lakekit-core`: the `Host` trait for object stores, clocks, and randomness; `MemoryHost`; one `Error` type. +- `lakekit-core`: loko's `Store`, `Clock`, and `Random` services with their test and native implementations + (`TokioClock` with feature `tokio`, `OsRandom` with feature `getrandom`); `Services`, the three borrowed + together for commits; one `Error` type, which a store's error converts into. +- `lakekit` feature `object-store`: a `Store` over an `object_store` client (S3, GCS, Azure, local files, memory), + from a client or a URL (`loko-object-store`). - `lakekit-iceberg`: Iceberg table metadata, schemas, partition specs, manifest lists, and manifests read for format versions 1 to 3, including delete files and deletion vector references; values in Iceberg's single-value encoding. - `lakekit-iceberg` feature `write`: format v2 manifests, manifest lists, and metadata updates, and commits to @@ -23,5 +27,10 @@ versions follow [Semantic Versioning](https://semver.org/). - The project is named lakekit, and its crates `lakekit-core`, `lakekit-iceberg`, and `lakekit`; it was lakebed (`lakebed-core`, …) until 2026-10-21. Nothing was published under the old names; GitHub redirects the old repository URL. +- The I/O boundary is three traits, `Store` (objects), `Clock` (the time and a timer), and `Random` (random numbers), + instead of one `Host`; `MemoryHost` is `MemoryServices`. UUIDs come from the `uuid` crate. +- The three traits are [loko](https://github.com/polythings/loko)'s, and `Clock::sleep` takes a `Duration`. The + `object_store` adapter is `loko-object-store`, re-exported as `lakekit::object_store`; `lakekit-object-store` is + gone (it was never published). The Iceberg code started as Timelake's shared writer (`timelake-format`), made general. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c15535a..79f6e62 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -10,7 +10,8 @@ enforce the mechanical rules (see [Checks](#checks)). This guide covers what the the body. 2. **One crate per format.** A format's crate depends on `lakekit-core` and on nothing of another format's, so adding one never touches the others. Formats are shown as they are: no shared model flattens their differences. -3. **No I/O of our own.** Crates read and write through `lakekit_core::Host`. They never open sockets, files, or +3. **No I/O of our own.** Crates read and write through `lakekit_core::Store` and take the time and random numbers + from `Clock` and `Random`. They never open sockets, files, or threads, so they build for `wasm32-unknown-unknown`. 4. **One definition of each thing.** Before writing a helper, constant, or type, look for an existing one. 5. **Behaviour, spec, and tests change together.** A change users can see updates its spec, adds or updates tests, and @@ -43,13 +44,13 @@ ideas we took. - **Types over strings:** enums and newtypes instead of strings and magic numbers. Parse once at the boundary. - **Don't trust files.** Check lengths and offsets before using them, cap allocations, and return an error instead of panicking on a malformed file. Tests feed readers truncated and garbage input. -- **Async:** host futures aren't `Send` (JavaScript promises aren't). Never block inside async code. +- **Async:** store and clock futures aren't `Send` (JavaScript promises aren't). Never block inside async code. - **Suppressing a lint:** `#[expect(lint, reason = "...")]` with a real reason. Never `#[allow]`. ## Tests - **A bug fix starts with a test that fails**, then the fix. -- **Tests run offline**, against `MemoryHost` or a host over the reference implementation's in-memory storage. +- **Tests run offline**, against a `MemoryStore` with a `TestClock` and a `SeededRandom`, or a store over the reference implementation's in-memory storage. - **Interop:** writers are tested against the format's reference reader, and readers against the reference writer. ## Checks diff --git a/Cargo.lock b/Cargo.lock index 0466307..9a9c3d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1209,6 +1209,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "humantime" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15" + [[package]] name = "hybrid-array" version = "0.4.15" @@ -1316,7 +1322,7 @@ dependencies = [ "flate2", "fnv", "futures", - "itertools", + "itertools 0.13.0", "moka", "murmur3", "once_cell", @@ -1514,6 +1520,15 @@ dependencies = [ "either", ] +[[package]] +name = "itertools" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b4baf93f58d4425749ca49a51c50ebab072c5df6994d08fed93541c331481dc" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -1601,13 +1616,18 @@ version = "0.1.0" dependencies = [ "lakekit-core", "lakekit-iceberg", + "loko-object-store", ] [[package]] name = "lakekit-core" version = "0.1.0" dependencies = [ + "loko-clock", + "loko-random", + "loko-store", "thiserror", + "uuid", ] [[package]] @@ -1622,6 +1642,8 @@ dependencies = [ "futures-util", "iceberg", "lakekit-core", + "loko-object-store", + "object_store", "parquet", "serde_json", "tokio", @@ -1717,6 +1739,39 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "loko-clock" +version = "0.2.0" +source = "git+https://github.com/polythings/loko?rev=0c425d3bf65cdedbadd628843fe68d3e9ddc419b#0c425d3bf65cdedbadd628843fe68d3e9ddc419b" +dependencies = [ + "tokio", +] + +[[package]] +name = "loko-object-store" +version = "0.2.0" +source = "git+https://github.com/polythings/loko?rev=0c425d3bf65cdedbadd628843fe68d3e9ddc419b#0c425d3bf65cdedbadd628843fe68d3e9ddc419b" +dependencies = [ + "futures-util", + "loko-store", + "object_store", + "url", +] + +[[package]] +name = "loko-random" +version = "0.2.0" +source = "git+https://github.com/polythings/loko?rev=0c425d3bf65cdedbadd628843fe68d3e9ddc419b#0c425d3bf65cdedbadd628843fe68d3e9ddc419b" +dependencies = [ + "getrandom 0.4.3", + "rand_core 0.10.1", +] + +[[package]] +name = "loko-store" +version = "0.2.0" +source = "git+https://github.com/polythings/loko?rev=0c425d3bf65cdedbadd628843fe68d3e9ddc419b#0c425d3bf65cdedbadd628843fe68d3e9ddc419b" + [[package]] name = "lz4_flex" version = "0.13.1" @@ -1833,6 +1888,29 @@ dependencies = [ "libm", ] +[[package]] +name = "object_store" +version = "0.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1796bc93603f78c5760a69f2d58badc9618d22adade0a95385bb2adbae4eb94" +dependencies = [ + "async-trait", + "bytes", + "chrono", + "futures-channel", + "futures-core", + "futures-util", + "http", + "humantime", + "itertools 0.15.0", + "parking_lot", + "percent-encoding", + "thiserror", + "url", + "wasm-bindgen-futures", + "web-time", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -3006,6 +3084,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "windows-core" version = "0.62.2" diff --git a/Cargo.toml b/Cargo.toml index b7cdc73..c228fda 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,14 @@ categories = ["database-implementations", "wasm"] [workspace.dependencies] lakekit-core = { path = "crates/lakekit-core", version = "0.1.0" } lakekit-iceberg = { path = "crates/lakekit-iceberg", version = "0.1.0", default-features = false } +# loko provides the Store, Clock, and Random services. From git until loko 0.2.0 is on crates.io +# (polythings/loko#1); publishing uses the version. +loko-store = { git = "https://github.com/polythings/loko", rev = "0c425d3bf65cdedbadd628843fe68d3e9ddc419b", version = "0.2.0" } +loko-clock = { git = "https://github.com/polythings/loko", rev = "0c425d3bf65cdedbadd628843fe68d3e9ddc419b", version = "0.2.0" } +loko-random = { git = "https://github.com/polythings/loko", rev = "0c425d3bf65cdedbadd628843fe68d3e9ddc419b", version = "0.2.0" } +loko-object-store = { git = "https://github.com/polythings/loko", rev = "0c425d3bf65cdedbadd628843fe68d3e9ddc419b", version = "0.2.0" } +object_store = { version = "0.14", default-features = false } +url = "2" thiserror = "2" serde_json = "1" apache-avro = { version = "0.22", default-features = false } diff --git a/PRIOR_ART.md b/PRIOR_ART.md index da7a0db..741bdad 100644 --- a/PRIOR_ART.md +++ b/PRIOR_ART.md @@ -11,12 +11,16 @@ never code: every line of lakekit is our own. How we credit is in [CONTRIBUTING. | [iceberg-rust](https://github.com/apache/iceberg-rust) | Apache-2.0 | The reference the interop tests check against (a test dependency only) | | [Apache Avro](https://avro.apache.org) (`apache-avro`) | Apache-2.0 | Reading and writing manifests and manifest lists | | [arrow-rs](https://github.com/apache/arrow-rs) (`arrow-schema`, `parquet`) | Apache-2.0 | Arrow schemas; Parquet footers for statistics | +| [loko](https://github.com/polythings/loko) | Apache-2.0 | The `Store`, `Clock`, and `Random` services, and the `object_store` adapter | +| [object_store](https://github.com/apache/arrow-rs-object-store) | Apache-2.0 / MIT | The stores `loko-object-store` adapts (re-exported by `lakekit`) | +| [uuid](https://github.com/uuid-rs/uuid) | Apache-2.0 / MIT | UUIDs for file names, from a `Random` and a `Clock` | ## Ideas | Idea | From | License of the source | Where in lakekit | |---|---|---|---| -| A format library with no I/O of its own: the application supplies storage through a trait the library calls | [delta-kernel-rs](https://github.com/delta-io/delta-kernel-rs) (its `Engine` trait) | Apache-2.0 | `lakekit-core`'s `Host` | +| A format library with no I/O of its own: the application supplies storage through a trait the library calls | [delta-kernel-rs](https://github.com/delta-io/delta-kernel-rs) (its `Engine` trait) | Apache-2.0 | `lakekit-core`'s `Store`, `Clock`, and `Random` | | One crate per backend under a small shared core, re-exported by an umbrella crate with a feature each | [arrow-rs](https://github.com/apache/arrow-rs), [Apache OpenDAL](https://github.com/apache/opendal) | Apache-2.0 | The workspace layout | | Commits to tables without a catalog: `v.metadata.json` written only if absent, `version-hint.text` for readers | Iceberg's file-system ("Hadoop") tables | Apache-2.0 | `lakekit_iceberg::commit` | +| The time and randomness as separate, swappable services rather than ambient globals | [ZIO](https://github.com/zio/zio) (`Clock`, `Random`) | Apache-2.0 | `lakekit-core`'s `Clock` and `Random` | | Avro-safe names for partition fields (`_x` and the character's code in hex) | Iceberg's Java implementation | Apache-2.0 | Manifest writing | diff --git a/README.md b/README.md index 13a5876..f76a851 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,10 @@ # lakekit Table formats read and written without I/O of their own, in Rust, for WebAssembly and native targets. Each format's -metadata (Apache Iceberg first) is parsed and written by plain code that asks a `Host` for bytes: the application -implements the host over its object store, so the same crate runs in a browser, a Cloudflare Worker, a CLI, or a -server. +metadata (Apache Iceberg first) is parsed and written by plain code that asks a `Store` for bytes, a `Clock` for the +time, and a `Random` for random numbers ([loko](https://github.com/polythings/loko)'s services): the application +supplies them, so the same crate runs in a browser, a +Cloudflare Worker, a CLI, or a server. lakekit is what [Timelake](https://github.com/polythings/timelake) and [lakeglass](https://github.com/polythings/lakeglass) share. The official Rust crates for these formats build on tokio and HTTP clients, which don't compile to @@ -13,11 +14,11 @@ share. The official Rust crates for these formats build on tokio and HTTP client | Crate | What it does | |---|---| -| [`lakekit-core`](crates/lakekit-core) | The `Host` trait (ranged reads, listing, conditional writes, clock, randomness), errors, an in-memory host | +| [`lakekit-core`](crates/lakekit-core) | loko's `Store`, `Clock`, and `Random` with their test and native implementations, `Services`, and one error type | | [`lakekit-iceberg`](crates/lakekit-iceberg) | Apache Iceberg: metadata, manifests, and commits | -| [`lakekit`](crates/lakekit) | Re-exports the format crates, one feature each | +| [`lakekit`](crates/lakekit) | Re-exports the crates, one feature each, and `loko-object-store` (a `Store` over an `object_store` client) | -More formats get their own crate as they are added (spec 2). +More formats get their own crate as they are added (spec 2). Store adapters are loko's (spec 4). ## Iceberg @@ -33,14 +34,21 @@ More formats get their own crate as they are added (spec 2). What lakekit writes, iceberg-rust reads, and the other way round: the tests commit to one table with both in turn. ```rust +use std::sync::Arc; + use lakekit_iceberg::Table; +use loko_object_store::ObjectStoreAdapter; -async fn files(host: &H) -> lakekit_core::Result { - let mut table = Table::load(host, "s3://bucket/warehouse/db/events").await?; - Ok(table.live_files(host).await?.len()) +async fn files(s3: Arc) -> lakekit_core::Result { + let store = ObjectStoreAdapter::new(s3, "s3://bucket"); + let mut table = Table::load(&store, "s3://bucket/warehouse/db/events").await?; + Ok(table.live_files(&store).await?.len()) } ``` +Commits also need a `Clock` and a `Random`: `Services::new(&store, &TokioClock, &OsRandom)` natively, the application's +own in WebAssembly. + ## Development ```bash diff --git a/crates/lakekit-core/Cargo.toml b/crates/lakekit-core/Cargo.toml index ec137de..6c3ee67 100644 --- a/crates/lakekit-core/Cargo.toml +++ b/crates/lakekit-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lakekit-core" -description = "The I/O boundary shared by lakekit's table format crates: a host trait for object stores, clocks, and randomness" +description = "The I/O boundary shared by lakekit's table format crates: loko's Store, Clock, and Random, and one error type" readme = "README.md" version.workspace = true edition.workspace = true @@ -10,8 +10,22 @@ repository.workspace = true keywords.workspace = true categories.workspace = true +[features] +default = [] +# `TokioClock`: the system clock and tokio's timer. Native targets only. +tokio = ["loko-clock/tokio"] +# `OsRandom`: the operating system's random number generator. Native targets and WASI. +getrandom = ["loko-random/getrandom"] + [dependencies] +loko-store.workspace = true +loko-clock.workspace = true +loko-random.workspace = true thiserror.workspace = true +uuid = { version = "1", default-features = false } + +[package.metadata.docs.rs] +all-features = true [lints] workspace = true diff --git a/crates/lakekit-core/README.md b/crates/lakekit-core/README.md index 734e18c..b3d28ee 100644 --- a/crates/lakekit-core/README.md +++ b/crates/lakekit-core/README.md @@ -1,12 +1,20 @@ # lakekit-core The I/O boundary of [lakekit](https://github.com/polythings/lakekit)'s table format crates. Format crates do no I/O of -their own: they call a `Host` that the application implements over its object store, clock, and randomness, so they -build for `wasm32-unknown-unknown` and native targets alike. +their own: they read and write through a `Store`, take the time from a `Clock`, and take random numbers from a +`Random`, all supplied by the application ([loko](https://github.com/polythings/loko)'s services, re-exported here), so they build for `wasm32-unknown-unknown` and native targets alike. -- `Host`: get, ranged get, exists, put, put-if-absent (the commit primitive), multipart upload, list, delete, the time, - random numbers, and waiting. Its futures aren't `Send`, so a host can wrap JavaScript promises. -- `MemoryHost`: a host in memory with a fixed clock and seeded randomness, for tests. -- `Error`: one error type, whose variant says what a caller can do (`NotFound`, `Conflict`, `CommitUnknown`, …). +- `Store`: get, ranged get, exists, put, put-if-absent (the commit primitive), multipart upload, list, delete. Its + futures aren't `Send`, so a store can wrap JavaScript promises. `loko-object-store` implements it over + `object_store`. +- `Clock`: the time and a timer. `TokioClock` (feature `tokio`, native only) uses the system clock and tokio's timer; + `TestClock` moves only when told. +- `Random`: random numbers. `OsRandom` (feature `getrandom`, native and WASI) uses the operating system's generator; + `SeededRandom` repeats a sequence from a seed. +- `Services`: a store, a clock, and a random source, each borrowed, for APIs that need all three (commits): + `Services::new(&store, &clock, &random)`. +- `uuid_v4`, `uuid_v7`: identifiers from a `Random` (and a `Clock`), with the `uuid` crate. +- `Error`: one error type, whose variant says what a caller can do (`NotFound`, `Conflict`, `CommitUnknown`, …); a + store's error converts into it. Apache License 2.0. diff --git a/crates/lakekit-core/src/error.rs b/crates/lakekit-core/src/error.rs index 662fd96..8d6441a 100644 --- a/crates/lakekit-core/src/error.rs +++ b/crates/lakekit-core/src/error.rs @@ -28,3 +28,31 @@ pub enum Error { /// `Result` with [`Error`]. pub type Result = std::result::Result; + +impl From for Error { + fn from(e: loko_store::Error) -> Self { + match e { + loko_store::Error::NotFound(m) => Error::NotFound(m), + loko_store::Error::Invalid(m) => Error::Invalid(m), + loko_store::Error::Unsupported(m) => Error::Unsupported(m), + other => Error::Store(match other { + loko_store::Error::Store(m) => m, + other => other.to_string(), + }), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn store_errors_keep_their_kind() { + let from = |e| Error::from(e).to_string(); + assert_eq!(from(loko_store::Error::NotFound("a".into())), "not found: a"); + assert_eq!(from(loko_store::Error::Invalid("b".into())), "invalid: b"); + assert_eq!(from(loko_store::Error::Unsupported("c".into())), "unsupported: c"); + assert_eq!(from(loko_store::Error::Store("d".into())), "store: d"); + } +} diff --git a/crates/lakekit-core/src/host.rs b/crates/lakekit-core/src/host.rs deleted file mode 100644 index 4151c36..0000000 --- a/crates/lakekit-core/src/host.rs +++ /dev/null @@ -1,55 +0,0 @@ -use std::future::Future; - -use crate::Result; - -/// An object found by [`Host::list`]. -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct Listed { - /// Absolute URI. - pub uri: String, - /// Size in bytes. - pub size: u64, -} - -/// What a table format needs from where it runs: an object store, a clock, randomness, and a way -/// to wait. Paths are absolute URIs, as table metadata records them. -/// -/// The futures aren't required to be `Send`, so hosts can wrap JavaScript promises. -pub trait Host { - /// A multipart upload in progress. - type Upload: Upload; - - /// The object at `uri`, or `None` if there is none. - fn get(&self, uri: &str) -> impl Future>>>; - /// `len` bytes of the object at `uri` from `start`. - fn get_range(&self, uri: &str, start: u64, len: u64) -> impl Future>>; - /// Whether an object exists at `uri`. - fn exists(&self, uri: &str) -> impl Future>; - /// Writes (or replaces) the object at `uri`. - fn put(&self, uri: &str, bytes: Vec) -> impl Future>; - /// Writes the object at `uri` unless one exists; `false` if one did. The commit primitive. - fn put_if_absent(&self, uri: &str, bytes: Vec) -> impl Future>; - /// Starts a multipart upload to `uri`. - fn upload(&self, uri: &str) -> impl Future>; - /// Objects whose URI starts with `prefix`. - fn list(&self, prefix: &str) -> impl Future>>; - /// Deletes objects; missing ones are not an error. - fn delete(&self, uris: &[String]) -> impl Future>; - /// Milliseconds since the Unix epoch. - fn now_ms(&self) -> i64; - /// A random number. - fn random(&self) -> u64; - /// Waits `ms` milliseconds. - fn sleep(&self, ms: u64) -> impl Future; -} - -/// A multipart upload: parts in order, then [`Upload::complete`]. -pub trait Upload { - /// Sends the next part. Every part but the last must be at least the store's minimum (5 MiB on - /// S3 and R2). - fn put_part(&mut self, bytes: Vec) -> impl Future>; - /// Completes the upload: the object appears. - fn complete(self) -> impl Future>; - /// Drops the parts sent. - fn abort(self) -> impl Future; -} diff --git a/crates/lakekit-core/src/ids.rs b/crates/lakekit-core/src/ids.rs new file mode 100644 index 0000000..ef6dfa5 --- /dev/null +++ b/crates/lakekit-core/src/ids.rs @@ -0,0 +1,42 @@ +use uuid::Builder; + +use crate::{Clock, Random, Rng}; + +/// A random UUID (version 4) from `random`. +pub fn uuid_v4(random: &impl Random) -> String { + let mut bytes = [0u8; 16]; + random.rng().fill_bytes(&mut bytes); + Builder::from_random_bytes(bytes).into_uuid().to_string() +} + +/// A time-ordered UUID (version 7) from `clock` and `random`: names made with it sort by creation +/// time. +pub fn uuid_v7(clock: &impl Clock, random: &impl Random) -> String { + let millis = u64::try_from(clock.now_ms()).unwrap_or(0); + let mut bytes = [0u8; 10]; + random.rng().fill_bytes(&mut bytes); + Builder::from_unix_timestamp_millis(millis, &bytes).into_uuid().to_string() +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + use crate::{SeededRandom, TestClock}; + + #[test] + fn uuids_are_well_formed() { + let (clock, random) = (TestClock::at_ms(1_767_225_600_000), SeededRandom::new(0)); + let v4 = uuid_v4(&random); + let v7 = uuid_v7(&clock, &random); + for (u, version) in [(&v4, '4'), (&v7, '7')] { + assert_eq!(u.len(), 36); + assert_eq!(u.chars().nth(14), Some(version)); + assert!(matches!(u.chars().nth(19), Some('8' | '9' | 'a' | 'b'))); + } + // Version 7 starts with the time, so later ones sort later. + clock.advance(Duration::from_millis(1)); + assert!(uuid_v7(&clock, &random) > v7); + } +} diff --git a/crates/lakekit-core/src/lib.rs b/crates/lakekit-core/src/lib.rs index ddc231a..9b5b22c 100644 --- a/crates/lakekit-core/src/lib.rs +++ b/crates/lakekit-core/src/lib.rs @@ -1,18 +1,26 @@ //! The I/O boundary shared by lakekit's table format crates. Format crates do no I/O of their own: -//! they read and write through a [`Host`] the application implements over its object store (an -//! `object_store` client natively, `fetch` or an R2 binding in WebAssembly), so they build for -//! `wasm32-unknown-unknown` and native targets alike. +//! they read and write through a [`Store`] the application supplies (an `object_store` client +//! natively, `fetch` or an R2 binding in WebAssembly), and take the time from a [`Clock`] and random +//! numbers from a [`Random`], so they build for `wasm32-unknown-unknown` and native targets alike. +//! The three are [loko](https://github.com/polythings/loko)'s services, re-exported here. +//! +//! - [`Store`]: get, ranged get, put, put-if-absent (the commit primitive), multipart upload, list, +//! delete. `loko-object-store` implements it over `object_store`. +//! - [`Clock`]: the time and a timer. `TokioClock` (feature `tokio`) is a native one. +//! - [`Random`]: random numbers. `OsRandom` (feature `getrandom`) is a native one. +//! - [`Services`]: a store, a clock, and a random source, borrowed, for APIs that need all three. +//! - [`Error`]: one error type for the format crates, converted from the store's. //! //! ``` -//! use lakekit_core::{Host, MemoryHost}; +//! use lakekit_core::{MemoryStore, Store}; //! //! # block_on(async { -//! let host = MemoryHost::new(0); -//! host.put("memory://t/a", b"x".to_vec()).await?; -//! assert!(host.exists("memory://t/a").await?); +//! let store = MemoryStore::new(); +//! store.put("memory://t/a", b"x".to_vec()).await?; +//! assert!(store.exists("memory://t/a").await?); //! # Ok::<(), lakekit_core::Error>(()) //! # }).unwrap(); -//! # // MemoryHost's futures are ready at once, so one poll runs them. +//! # // MemoryStore's futures are ready at once, so one poll runs them. //! # fn block_on(f: F) -> F::Output { //! # let mut cx = std::task::Context::from_waker(std::task::Waker::noop()); //! # let mut f = std::pin::pin!(f); @@ -21,11 +29,16 @@ //! ``` mod error; -mod host; -mod memory; -mod uuid; +mod ids; +mod services; pub use error::{Error, Result}; -pub use host::{Host, Listed, Upload}; -pub use memory::{MemoryHost, MemoryUpload}; -pub use uuid::{uuid_v4, uuid_v7}; +pub use ids::{uuid_v4, uuid_v7}; +#[cfg(all(feature = "tokio", not(target_arch = "wasm32")))] +pub use loko_clock::TokioClock; +pub use loko_clock::{Clock, TestClock}; +#[cfg(all(feature = "getrandom", not(all(target_arch = "wasm32", target_os = "unknown"))))] +pub use loko_random::OsRandom; +pub use loko_random::{CryptoRng, Random, Rng, SeededRandom, SeededRng}; +pub use loko_store::{Error as StoreError, Listed, MemoryStore, MemoryUpload, Store, Upload}; +pub use services::Services; diff --git a/crates/lakekit-core/src/memory.rs b/crates/lakekit-core/src/memory.rs deleted file mode 100644 index 919ebce..0000000 --- a/crates/lakekit-core/src/memory.rs +++ /dev/null @@ -1,138 +0,0 @@ -use std::collections::BTreeMap; -use std::future::{Future, ready}; -use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; - -use crate::{Error, Host, Listed, Result, Upload}; - -/// A [`Host`] in memory, with a fixed clock and a seeded random sequence: for tests and examples. -/// Clones share the same objects. -#[derive(Clone, Debug, Default)] -pub struct MemoryHost { - objects: Arc>>>, - clock: Arc>, - seed: Arc>, -} - -impl MemoryHost { - /// An empty store at time `now_ms`. - pub fn new(now_ms: i64) -> Self { - let host = Self::default(); - *lock(&host.clock) = now_ms; - host - } - - /// Moves the clock forward. - pub fn advance(&self, ms: i64) { - *lock(&self.clock) += ms; - } - - /// The URIs stored, sorted. - pub fn uris(&self) -> Vec { - self.objects().keys().cloned().collect() - } - - fn objects(&self) -> MutexGuard<'_, BTreeMap>> { - lock(&self.objects) - } -} - -fn lock(m: &Mutex) -> MutexGuard<'_, T> { - m.lock().unwrap_or_else(PoisonError::into_inner) -} - -/// An upload to a [`MemoryHost`]. -#[derive(Debug)] -pub struct MemoryUpload { - host: MemoryHost, - uri: String, - bytes: Vec, -} - -impl Upload for MemoryUpload { - fn put_part(&mut self, bytes: Vec) -> impl Future> { - self.bytes.extend(bytes); - ready(Ok(())) - } - - fn complete(self) -> impl Future> { - self.host.objects().insert(self.uri, self.bytes); - ready(Ok(())) - } - - fn abort(self) -> impl Future { - ready(()) - } -} - -impl Host for MemoryHost { - type Upload = MemoryUpload; - - fn get(&self, uri: &str) -> impl Future>>> { - ready(Ok(self.objects().get(uri).cloned())) - } - - fn get_range(&self, uri: &str, start: u64, len: u64) -> impl Future>> { - let objects = self.objects(); - ready(objects.get(uri).ok_or_else(|| Error::NotFound(uri.to_string())).map(|bytes| { - let start = usize::try_from(start).unwrap_or(usize::MAX).min(bytes.len()); - let end = start.saturating_add(usize::try_from(len).unwrap_or(usize::MAX)).min(bytes.len()); - bytes[start..end].to_vec() - })) - } - - fn exists(&self, uri: &str) -> impl Future> { - ready(Ok(self.objects().contains_key(uri))) - } - - fn put(&self, uri: &str, bytes: Vec) -> impl Future> { - self.objects().insert(uri.to_string(), bytes); - ready(Ok(())) - } - - fn put_if_absent(&self, uri: &str, bytes: Vec) -> impl Future> { - let mut objects = self.objects(); - let absent = !objects.contains_key(uri); - if absent { - objects.insert(uri.to_string(), bytes); - } - ready(Ok(absent)) - } - - fn upload(&self, uri: &str) -> impl Future> { - ready(Ok(MemoryUpload { host: self.clone(), uri: uri.to_string(), bytes: vec![] })) - } - - fn list(&self, prefix: &str) -> impl Future>> { - let listed = self - .objects() - .range(prefix.to_string()..) - .take_while(|(k, _)| k.starts_with(prefix)) - .map(|(k, v)| Listed { uri: k.clone(), size: v.len() as u64 }) - .collect(); - ready(Ok(listed)) - } - - fn delete(&self, uris: &[String]) -> impl Future> { - let mut objects = self.objects(); - for uri in uris { - objects.remove(uri); - } - ready(Ok(())) - } - - fn now_ms(&self) -> i64 { - *lock(&self.clock) - } - - fn random(&self) -> u64 { - // A linear congruential generator (Knuth's MMIX constants), with the high bits folded in. - let mut seed = lock(&self.seed); - *seed = seed.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1_442_695_040_888_963_407); - *seed ^ (*seed >> 29) - } - - fn sleep(&self, ms: u64) -> impl Future { - self.advance(i64::try_from(ms).unwrap_or(i64::MAX)); - ready(()) - } -} diff --git a/crates/lakekit-core/src/services.rs b/crates/lakekit-core/src/services.rs new file mode 100644 index 0000000..0ab4f62 --- /dev/null +++ b/crates/lakekit-core/src/services.rs @@ -0,0 +1,35 @@ +use std::fmt; + +/// The store, clock, and random source an operation such as a commit uses, each borrowed from +/// wherever the caller keeps it: `Services::new(&s3, &TokioClock, &OsRandom)`. One object that is +/// all three (a JavaScript host) goes in all three places: `Services::new(&host, &host, &host)`. +pub struct Services<'a, S, C, R> { + /// Where objects are read and written. + pub store: &'a S, + /// The time, and waiting between attempts. + pub clock: &'a C, + /// Random numbers: snapshot ids, file names, and jitter. + pub random: &'a R, +} + +impl<'a, S, C, R> Services<'a, S, C, R> { + /// `store`, `clock`, and `random`. + pub fn new(store: &'a S, clock: &'a C, random: &'a R) -> Self { + Self { store, clock, random } + } +} + +// By hand, so `Services` is `Copy` whatever the services are (a derive would ask them to be). +impl Clone for Services<'_, S, C, R> { + fn clone(&self) -> Self { + *self + } +} + +impl Copy for Services<'_, S, C, R> {} + +impl fmt::Debug for Services<'_, S, C, R> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Services").finish_non_exhaustive() + } +} diff --git a/crates/lakekit-core/src/uuid.rs b/crates/lakekit-core/src/uuid.rs deleted file mode 100644 index 320b951..0000000 --- a/crates/lakekit-core/src/uuid.rs +++ /dev/null @@ -1,55 +0,0 @@ -use std::fmt::Write as _; - -use crate::Host; - -/// A random UUID (version 4) from `host`. -pub fn uuid_v4(host: &impl Host) -> String { - let mut b = [0u8; 16]; - b[..8].copy_from_slice(&host.random().to_be_bytes()); - b[8..].copy_from_slice(&host.random().to_be_bytes()); - b[6] = (b[6] & 0x0f) | 0x40; - b[8] = (b[8] & 0x3f) | 0x80; - format_uuid(&b) -} - -/// A time-ordered UUID (version 7) from `host`'s clock and randomness: names made with it sort by -/// creation time. -pub fn uuid_v7(host: &impl Host) -> String { - let mut b = [0u8; 16]; - let ms = u64::try_from(host.now_ms()).unwrap_or(0); - b[..6].copy_from_slice(&ms.to_be_bytes()[2..]); - b[6..14].copy_from_slice(&host.random().to_be_bytes()); - b[14..].copy_from_slice(&host.random().to_be_bytes()[..2]); - b[6] = (b[6] & 0x0f) | 0x70; - b[8] = (b[8] & 0x3f) | 0x80; - format_uuid(&b) -} - -fn format_uuid(b: &[u8; 16]) -> String { - let hex = b.iter().fold(String::with_capacity(32), |mut s, x| { - let _ = write!(s, "{x:02x}"); - s - }); - format!("{}-{}-{}-{}-{}", &hex[..8], &hex[8..12], &hex[12..16], &hex[16..20], &hex[20..]) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::MemoryHost; - - #[test] - fn uuids_are_well_formed() { - let host = MemoryHost::new(1_767_225_600_000); - let v4 = uuid_v4(&host); - let v7 = uuid_v7(&host); - for (u, version) in [(&v4, '4'), (&v7, '7')] { - assert_eq!(u.len(), 36); - assert_eq!(u.chars().nth(14), Some(version)); - assert!(matches!(u.chars().nth(19), Some('8' | '9' | 'a' | 'b'))); - } - // Version 7 starts with the time, so later ones sort later. - host.advance(1); - assert!(uuid_v7(&host) > v7); - } -} diff --git a/crates/lakekit-iceberg/Cargo.toml b/crates/lakekit-iceberg/Cargo.toml index b6f0d96..1f75ec8 100644 --- a/crates/lakekit-iceberg/Cargo.toml +++ b/crates/lakekit-iceberg/Cargo.toml @@ -36,6 +36,9 @@ arrow-array = "58" arrow-schema.workspace = true bytes = "1" futures = "0.3" +lakekit-core = { workspace = true, features = ["tokio", "getrandom"] } +loko-object-store.workspace = true +object_store.workspace = true [package.metadata.docs.rs] all-features = true diff --git a/crates/lakekit-iceberg/README.md b/crates/lakekit-iceberg/README.md index efd7e77..766356f 100644 --- a/crates/lakekit-iceberg/README.md +++ b/crates/lakekit-iceberg/README.md @@ -1,7 +1,8 @@ # lakekit-iceberg [Apache Iceberg](https://iceberg.apache.org) tables read and written without I/O of their own, through a -[`lakekit_core::Host`](../lakekit-core), so they work in WebAssembly (browsers, Cloudflare Workers) as well as natively. +[`lakekit_core::Store`](../lakekit-core) (and, for commits, a `Clock` and a `Random`), so they work in WebAssembly +(browsers, Cloudflare Workers) as well as natively. | Feature | What it adds | |---|---| diff --git a/crates/lakekit-iceberg/src/commit.rs b/crates/lakekit-iceberg/src/commit.rs index 506fc8b..dc31632 100644 --- a/crates/lakekit-iceberg/src/commit.rs +++ b/crates/lakekit-iceberg/src/commit.rs @@ -6,13 +6,14 @@ //! Applications extend a commit through [`CommitHooks`]: files of their own written with the //! snapshot, summary properties, a check that skips a commit already made, and how to reload. +use core::time::Duration; use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt; use std::future::Future; use std::hash::BuildHasher; use std::sync::Arc; -use lakekit_core::{Host, uuid_v4}; +use lakekit_core::{Clock, Random, Rng, Services, Store, uuid_v4}; use crate::manifest::{ DataFile, Entry, ManifestContext, ManifestFile, ManifestWrite, Status, manifest_file, write_manifest, @@ -125,13 +126,17 @@ pub struct Extra { } /// An application's part in a commit. Every method has a default that does nothing extra. -pub trait CommitHooks { +pub trait CommitHooks { /// The error the hooks (and so the commit) return. type Error: From + fmt::Debug; /// The newest version of the table, after another writer won the race. - fn reload(&mut self, host: &H, location: &str) -> impl Future> { - async move { Ok(Table::load(host, location).await?) } + fn reload( + &mut self, + services: Services<'_, S, C, R>, + location: &str, + ) -> impl Future> { + async move { Ok(Table::load(services.store, location).await?) } } /// Whether `base` already holds this change (an idempotent append retried), so nothing is @@ -144,7 +149,7 @@ pub trait CommitHooks { /// per attempt. fn extra( &mut self, - _host: &H, + _services: Services<'_, S, C, R>, _base: &Table, _snapshot_id: i64, ) -> impl Future> { @@ -156,17 +161,17 @@ pub trait CommitHooks { #[derive(Clone, Copy, Debug, Default)] pub struct NoHooks; -impl CommitHooks for NoHooks { +impl CommitHooks for NoHooks { type Error = Error; } /// Commits `change` on top of `base` (or the newest version, after a lost race), with `props` in /// the snapshot summary. The returned table knows its files when `base` did. -pub async fn commit, S: BuildHasher>( - host: &H, +pub async fn commit, P: BuildHasher>( + services: Services<'_, S, C, R>, base: Table, change: &Change, - props: &HashMap, + props: &HashMap, options: &CommitOptions, hooks: &mut K, ) -> Result<(Table, CommitInfo), K::Error> { @@ -181,8 +186,9 @@ pub async fn commit, S: BuildHasher>( for attempt in 1..=options.max_attempts { if attempt > 1 { let (lo, hi) = options.backoff_ms; - host.sleep((lo + host.random() % hi.saturating_sub(lo).max(1)) * attempt).await; - base = hooks.reload(host, &base.location).await?; + let ms = (lo + services.random.rng().next_u64() % hi.saturating_sub(lo).max(1)) * attempt; + services.clock.sleep(Duration::from_millis(ms)).await; + base = hooks.reload(services, &base.location).await?; } if hooks.skip(&base) { let info = CommitInfo { @@ -193,7 +199,7 @@ pub async fn commit, S: BuildHasher>( }; return Ok((base, info)); } - if let Some((table, snapshot_id)) = attempt_commit(host, &base, change, props, options, hooks).await? { + if let Some((table, snapshot_id)) = attempt_commit(services, &base, change, props, options, hooks).await? { let info = CommitInfo { snapshot_id, version: table.version, attempts: attempt, skipped: false }; return Ok((table, info)); } @@ -208,18 +214,18 @@ pub async fn commit, S: BuildHasher>( /// that landed gets "already exists", and a timeout says nothing. The version is then read back: /// our own metadata there is the commit, another writer's means it lost. If that can't be told, /// the error is [`Error::CommitUnknown`], and callers must not delete what the metadata references. -pub async fn put_metadata(host: &H, uri: &str, bytes: Vec) -> Result { +pub async fn put_metadata(store: &S, uri: &str, bytes: Vec) -> Result { let unknown = |e: &dyn fmt::Display| Error::CommitUnknown(format!("{uri}: {e}")); - match host.put_if_absent(uri, bytes.clone()).await { + match store.put_if_absent(uri, bytes.clone()).await { Ok(true) => Ok(true), // Ours (an earlier try landed) or another writer's. Absent now, it was never ours: our // version would still be there. - Ok(false) => match host.get(uri).await { + Ok(false) => match store.get(uri).await { Ok(found) => Ok(found.is_some_and(|found| found == bytes)), Err(e) => Err(unknown(&e)), }, // Absent now, the PUT may still land. - Err(e) => match host.get(uri).await { + Err(e) => match store.get(uri).await { Ok(Some(found)) => Ok(found == bytes), Ok(None) | Err(_) => Err(unknown(&e)), }, @@ -227,9 +233,10 @@ pub async fn put_metadata(host: &H, uri: &str, bytes: Vec) -> Resul } /// A snapshot id not in the table yet. -fn new_snapshot_id(host: &impl Host, base: &Table) -> i64 { +fn new_snapshot_id(random: &impl Random, base: &Table) -> i64 { + let mut rng = random.rng(); loop { - let id = i64::try_from(host.random() >> 1).unwrap_or(1).max(1); + let id = i64::try_from(rng.next_u64() >> 1).unwrap_or(1).max(1); if !base.metadata.has_snapshot(id) { return id; } @@ -239,35 +246,35 @@ fn new_snapshot_id(host: &impl Host, base: &Table) -> i64 { /// One attempt: writes the snapshot's files and publishes it. `None` if another writer committed /// this version first (this attempt's files are deleted). With an unknown outcome, nothing is /// deleted. -async fn attempt_commit, S: BuildHasher>( - host: &H, +async fn attempt_commit, P: BuildHasher>( + services: Services<'_, S, C, R>, base: &Table, change: &Change, - props: &HashMap, + props: &HashMap, options: &CommitOptions, hooks: &mut K, ) -> Result, K::Error> { - let snapshot_id = new_snapshot_id(host, base); + let snapshot_id = new_snapshot_id(services.random, base); let sequence_number = base.metadata.next_sequence_number(); let mut uploads: Vec<(String, Vec)> = vec![]; let mut unknown = false; let result = async { - let manifests = write_manifests(host, base, change, snapshot_id, options, &mut uploads).await?; - let extra = hooks.extra(host, base, snapshot_id).await?; + let manifests = write_manifests(services, base, change, snapshot_id, options, &mut uploads).await?; + let extra = hooks.extra(services, base, snapshot_id).await?; uploads.extend(extra.files); - let list_uri = format!("{}/metadata/snap-{snapshot_id}-{}.avro", base.location, uuid_v4(host)); + let list_uri = format!("{}/metadata/snap-{snapshot_id}-{}.avro", base.location, uuid_v4(services.random)); let parent = base.metadata.current_snapshot_id(); let list = write_manifest_list(&manifests, snapshot_id, parent, sequence_number)?; uploads.push((list_uri.clone(), list)); - let puts = uploads.iter().map(|(uri, bytes)| host.put(uri, bytes.clone())); - futures_util::future::try_join_all(puts).await?; + let puts = uploads.iter().map(|(uri, bytes)| services.store.put(uri, bytes.clone())); + futures_util::future::try_join_all(puts).await.map_err(Error::from)?; let mut summary: HashMap = props.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); summary.extend(extra.summary); summary.extend(counters(change, &base.metadata.summary())); // Snapshot and metadata logs must stay in time order, so a writer whose clock is behind the // last one's dates its commit at the table's last update. - let now = host.now_ms().max(base.metadata.last_updated_ms()); + let now = services.clock.now_ms().max(base.metadata.last_updated_ms()); let snapshot = NewSnapshot { snapshot_id, sequence_number, @@ -278,21 +285,22 @@ async fn attempt_commit, S: BuildHasher>( }; let metadata = base.metadata.with_snapshot(&snapshot, &base.metadata_file, now)?; let version = base.version + 1; - let ours = put_metadata(host, &metadata_uri(&base.location, version), metadata.to_json().into_bytes()) - .await - .inspect_err(|e| unknown = matches!(e, Error::CommitUnknown(_)))?; + let ours = + put_metadata(services.store, &metadata_uri(&base.location, version), metadata.to_json().into_bytes()) + .await + .inspect_err(|e| unknown = matches!(e, Error::CommitUnknown(_)))?; Ok::<_, K::Error>(ours.then_some((metadata, version))) } .await; let (metadata, version) = match result { Ok(Some(committed)) => committed, Ok(None) => { - delete_uploads(host, uploads).await; + delete_uploads(services.store, uploads).await; return Ok(None); } Err(e) => { if !unknown { - delete_uploads(host, uploads).await; + delete_uploads(services.store, uploads).await; } return Err(e); } @@ -317,20 +325,20 @@ async fn attempt_commit, S: BuildHasher>( }); let table = Table { files, ..Table::new(base.location.clone(), version, metadata) }; if options.write_version_hint { - let _ = host.put(&hint_uri(&table.location), version.to_string().into_bytes()).await; + let _ = services.store.put(&hint_uri(&table.location), version.to_string().into_bytes()).await; } Ok(Some((table, snapshot_id))) } -async fn delete_uploads(host: &impl Host, uploads: Vec<(String, Vec)>) { +async fn delete_uploads(store: &impl Store, uploads: Vec<(String, Vec)>) { let written: Vec = uploads.into_iter().map(|(uri, _)| uri).collect(); - let _ = host.delete(&written).await; + let _ = store.delete(&written).await; } /// Writes a manifest of the added files, carries the base snapshot's manifests over, and rewrites /// the ones holding removed files (or all of them once there are too many). -async fn write_manifests( - host: &H, +async fn write_manifests( + services: Services<'_, S, C, R>, base: &Table, change: &Change, snapshot_id: i64, @@ -350,7 +358,7 @@ async fn write_manifests( partition_type: &partition_type, }; let mut write = |kind: &str, entries: &[ManifestWrite]| -> Result { - let uri = format!("{}/metadata/{}-{kind}-m.avro", base.location, uuid_v4(host)); + let uri = format!("{}/metadata/{}-{kind}-m.avro", base.location, uuid_v4(services.random)); let bytes = write_manifest(&ctx, entries)?; let mf = manifest_file(&uri, &bytes, spec_id, snapshot_id, entries, &partition_type); uploads.push((uri, bytes)); @@ -373,14 +381,14 @@ async fn write_manifests( new.push(write("added", &entries)?); } - let current = base.manifests(host).await?; + let current = base.manifests(services.store).await?; let removed: HashSet<&str> = change.removed().iter().map(|e| e.file.path.as_str()).collect(); let merge = removed.is_empty() && current.len() + new.len() > options.merge_manifests_at; if removed.is_empty() && !merge { new.extend(current); return Ok(new); } - let loaded = futures_util::future::try_join_all(current.iter().map(|m| Table::manifest(host, m))).await?; + let loaded = futures_util::future::try_join_all(current.iter().map(|m| Table::manifest(services.store, m))).await?; let (mut kept, mut deleted, mut carried) = (vec![], vec![], vec![]); for (manifest, entries) in current.into_iter().zip(loaded) { let live: Vec<&Entry> = entries.iter().filter(|e| e.status != Status::Deleted).collect(); @@ -469,7 +477,7 @@ pub fn counters(change: &Change, base: &HashMap Table { + /// A store, a clock, and a random source in memory: what a test commit uses. + struct Mem { + store: MemoryStore, + clock: TestClock, + random: SeededRandom, + } + + impl Mem { + fn at_ms(ms: i64) -> Self { + Self { store: MemoryStore::new(), clock: TestClock::at_ms(ms), random: SeededRandom::new(0) } + } + + fn services(&self) -> Services<'_, MemoryStore, TestClock, SeededRandom> { + Services::new(&self.store, &self.clock, &self.random) + } + } + + fn create(store: &Mem, format_version: u8) -> Table { let metadata = json!({ "format-version": format_version, "table-uuid": "u", "location": LOCATION, "last-sequence-number": 0, "last-updated-ms": 0, "last-column-id": 2, "current-schema-id": 0, @@ -491,7 +516,7 @@ mod tests { "properties": {}, "current-snapshot-id": -1, "snapshots": [], "snapshot-log": [], "metadata-log": [] }); let table = Table::new(LOCATION, 1, Metadata::from_json(metadata).unwrap()); - block_on(host.put(&table.metadata_file, table.metadata.to_json().into_bytes())).unwrap(); + block_on(store.store.put(&table.metadata_file, table.metadata.to_json().into_bytes())).unwrap(); Table { files: Some(vec![]), ..table } } @@ -509,22 +534,22 @@ mod tests { Change::Append { files } } - fn try_run>( - host: &MemoryHost, + fn try_run>( + store: &Mem, base: Table, change: &Change, hooks: &mut K, ) -> Result<(Table, CommitInfo), K::Error> { - block_on(commit(host, base, change, &HashMap::new(), &CommitOptions::default(), hooks)) + block_on(commit(store.services(), base, change, &HashMap::new(), &CommitOptions::default(), hooks)) } - fn run>( - host: &MemoryHost, + fn run>( + store: &Mem, base: Table, change: &Change, hooks: &mut K, ) -> (Table, CommitInfo) { - try_run(host, base, change, hooks).unwrap() + try_run(store, base, change, hooks).unwrap() } fn files_named(t: &Table, name: &str) -> Vec { @@ -533,9 +558,9 @@ mod tests { /// The live files of the newest version, read back from the store: (name, data sequence /// number, file sequence number). - fn live(host: &MemoryHost) -> Vec<(String, i64, i64)> { - let mut t = block_on(Table::load(host, LOCATION)).unwrap(); - let mut files: Vec<_> = block_on(t.live_files(host)) + fn live(store: &Mem) -> Vec<(String, i64, i64)> { + let mut t = block_on(Table::load(&store.store, LOCATION)).unwrap(); + let mut files: Vec<_> = block_on(t.live_files(&store.store)) .unwrap() .iter() .map(|e| { @@ -552,11 +577,11 @@ mod tests { #[test] fn append_replace_and_delete() { - let host = MemoryHost::new(1_000); - let t = create(&host, 2); - let (t, info) = run(&host, t, &append(vec![file("a", 1, 10), file("b", 2, 20)]), &mut NoHooks); + let store = Mem::at_ms(1_000); + let t = create(&store, 2); + let (t, info) = run(&store, t, &append(vec![file("a", 1, 10), file("b", 2, 20)]), &mut NoHooks); assert_eq!((info.version, info.attempts, info.skipped), (2, 1, false)); - assert_eq!(live(&host), [("a.parquet".into(), 1, 1), ("b.parquet".into(), 1, 1)]); + assert_eq!(live(&store), [("a.parquet".into(), 1, 1), ("b.parquet".into(), 1, 1)]); // Compaction keeps the data sequence number of what it replaces. let change = Change::Replace { @@ -564,53 +589,58 @@ mod tests { added: vec![file("c", 1, 10)], sequence_numbers: vec![1], }; - let (t, _) = run(&host, t, &change, &mut NoHooks); - assert_eq!(live(&host), [("b.parquet".into(), 1, 1), ("c.parquet".into(), 1, 2)]); + let (t, _) = run(&store, t, &change, &mut NoHooks); + assert_eq!(live(&store), [("b.parquet".into(), 1, 1), ("c.parquet".into(), 1, 2)]); assert_eq!(t.files.as_ref().map(Vec::len), Some(2), "the commit derives the files"); - let (t, _) = run(&host, t.clone(), &Change::Delete { removed: files_named(&t, "b.parquet") }, &mut NoHooks); - assert_eq!(live(&host), [("c.parquet".into(), 1, 2)]); + let (t, _) = run(&store, t.clone(), &Change::Delete { removed: files_named(&t, "b.parquet") }, &mut NoHooks); + assert_eq!(live(&store), [("c.parquet".into(), 1, 2)]); let summary = t.metadata.summary(); assert_eq!((summary["operation"].as_str(), summary[crate::summary::TOTAL_RECORDS].as_str()), ("delete", "10")); assert_eq!(summary[crate::summary::DELETED_DATA_FILES], "1"); assert_eq!(t.metadata.metadata_log_len(), 3); - assert_eq!(block_on(host.get(&hint_uri(LOCATION))).unwrap().as_deref(), Some(&b"4"[..])); + assert_eq!(block_on(store.store.get(&hint_uri(LOCATION))).unwrap().as_deref(), Some(&b"4"[..])); } #[test] fn a_lost_race_retries_on_the_newer_version() { - let host = MemoryHost::new(1_000); - let stale = create(&host, 2); - let (_, first) = run(&host, stale.clone(), &append(vec![file("a", 1, 10)]), &mut NoHooks); - let (_, second) = run(&host, stale, &append(vec![file("b", 1, 10)]), &mut NoHooks); + let store = Mem::at_ms(1_000); + let stale = create(&store, 2); + let (_, first) = run(&store, stale.clone(), &append(vec![file("a", 1, 10)]), &mut NoHooks); + let (_, second) = run(&store, stale, &append(vec![file("b", 1, 10)]), &mut NoHooks); assert_eq!((first.version, second.version, second.attempts), (2, 3, 2)); - assert_eq!(live(&host).len(), 2); + assert_eq!(live(&store).len(), 2); // The lost attempt's manifests and list were deleted: every list is a snapshot's. - let lists = host.uris().iter().filter(|u| u.contains("/snap-")).count(); + let lists = store.store.uris().iter().filter(|u| u.contains("/snap-")).count(); assert_eq!(lists, 2); } #[test] fn removing_files_that_are_gone_is_a_conflict() { - let host = MemoryHost::new(1_000); - let (t, _) = run(&host, create(&host, 2), &append(vec![file("a", 1, 10)]), &mut NoHooks); + let store = Mem::at_ms(1_000); + let (t, _) = run(&store, create(&store, 2), &append(vec![file("a", 1, 10)]), &mut NoHooks); let removed = t.files.clone().unwrap_or_default(); - let (t, _) = run(&host, t, &Change::Delete { removed: removed.clone() }, &mut NoHooks); - let err = try_run(&host, t, &Change::Delete { removed }, &mut NoHooks); + let (t, _) = run(&store, t, &Change::Delete { removed: removed.clone() }, &mut NoHooks); + let err = try_run(&store, t, &Change::Delete { removed }, &mut NoHooks); assert!(matches!(err, Err(Error::Conflict(_))), "{err:?}"); } /// Hooks that skip a batch already committed, and add a file and summary properties. struct Batch(&'static str); - impl CommitHooks for Batch { + impl CommitHooks for Batch { type Error = Error; fn skip(&mut self, base: &Table) -> bool { base.metadata.ancestors().iter().any(|s| s.summary.get("batch").map(String::as_str) == Some(self.0)) } - fn extra(&mut self, _: &MemoryHost, base: &Table, snapshot_id: i64) -> impl Future> { + fn extra( + &mut self, + _: Services<'_, MemoryStore, TestClock, SeededRandom>, + base: &Table, + snapshot_id: i64, + ) -> impl Future> { let uri = format!("{}/metadata/extra-{snapshot_id}", base.location); let summary = HashMap::from([("batch".to_string(), self.0.to_string()), ("extra".to_string(), uri.clone())]); @@ -620,11 +650,11 @@ mod tests { #[test] fn hooks_skip_and_extend_commits() { - let host = MemoryHost::new(1_000); - let (t, info) = run(&host, create(&host, 2), &append(vec![file("a", 1, 10)]), &mut Batch("b1")); + let store = Mem::at_ms(1_000); + let (t, info) = run(&store, create(&store, 2), &append(vec![file("a", 1, 10)]), &mut Batch("b1")); let extra = t.metadata.summary()["extra"].clone(); - assert!(block_on(host.exists(&extra)).unwrap()); - let (t, again) = run(&host, t, &append(vec![file("a", 1, 10)]), &mut Batch("b1")); + assert!(block_on(store.store.exists(&extra)).unwrap()); + let (t, again) = run(&store, t, &append(vec![file("a", 1, 10)]), &mut Batch("b1")); assert!(again.skipped); assert_eq!((again.version, again.snapshot_id), (info.version, info.snapshot_id)); assert_eq!(t.version, 2); @@ -632,22 +662,22 @@ mod tests { #[test] fn many_manifests_merge() { - let host = MemoryHost::new(1_000); - let mut t = create(&host, 2); + let store = Mem::at_ms(1_000); + let mut t = create(&store, 2); for i in 0..9 { - t = run(&host, t, &append(vec![file(&format!("f{i}"), i, 1)]), &mut NoHooks).0; + t = run(&store, t, &append(vec![file(&format!("f{i}"), i, 1)]), &mut NoHooks).0; } - assert_eq!(block_on(t.manifests(&host)).unwrap().len(), 2, "eight merged into one, plus the new one"); - assert_eq!(live(&host).len(), 9); + assert_eq!(block_on(t.manifests(&store.store)).unwrap().len(), 2, "eight merged into one, plus the new one"); + assert_eq!(live(&store).len(), 9); } #[test] fn a_clock_behind_the_table_keeps_the_logs_in_order() { - let host = MemoryHost::new(5_000); - let (t, _) = run(&host, create(&host, 2), &append(vec![file("a", 1, 10)]), &mut NoHooks); - let late = MemoryHost::new(1_000); - for uri in host.uris() { - block_on(late.put(&uri, block_on(host.get(&uri)).unwrap().unwrap())).unwrap(); + let store = Mem::at_ms(5_000); + let (t, _) = run(&store, create(&store, 2), &append(vec![file("a", 1, 10)]), &mut NoHooks); + let late = Mem::at_ms(1_000); + for uri in store.store.uris() { + block_on(late.store.put(&uri, block_on(store.store.get(&uri)).unwrap().unwrap())).unwrap(); } let (t, _) = run(&late, t, &append(vec![file("b", 1, 10)]), &mut NoHooks); assert_eq!((t.metadata.current_snapshot().unwrap().timestamp_ms, t.metadata.last_updated_ms()), (5_000, 5_000)); @@ -655,19 +685,19 @@ mod tests { #[test] fn a_replace_needs_a_sequence_number_per_added_file() { - let host = MemoryHost::new(1_000); - let (t, _) = run(&host, create(&host, 2), &append(vec![file("a", 1, 10)]), &mut NoHooks); + let store = Mem::at_ms(1_000); + let (t, _) = run(&store, create(&store, 2), &append(vec![file("a", 1, 10)]), &mut NoHooks); let removed = files_named(&t, "a.parquet"); let change = Change::Replace { removed, added: vec![file("b", 1, 5), file("c", 1, 5)], sequence_numbers: vec![1] }; - let err = try_run(&host, t, &change, &mut NoHooks); + let err = try_run(&store, t, &change, &mut NoHooks); assert!(matches!(err, Err(Error::Invalid(_))), "{err:?}"); } #[test] fn only_v2_tables_are_written() { - let host = MemoryHost::new(1_000); - let err = try_run(&host, create(&host, 1), &append(vec![]), &mut NoHooks); + let store = Mem::at_ms(1_000); + let err = try_run(&store, create(&store, 1), &append(vec![]), &mut NoHooks); assert!(matches!(err, Err(Error::Unsupported(_)))); } } diff --git a/crates/lakekit-iceberg/src/lib.rs b/crates/lakekit-iceberg/src/lib.rs index da23ca7..f2f19d4 100644 --- a/crates/lakekit-iceberg/src/lib.rs +++ b/crates/lakekit-iceberg/src/lib.rs @@ -1,6 +1,8 @@ //! Apache Iceberg table metadata, manifests and commits, with no I/O of its own: everything is read -//! and written through a [`Host`](lakekit_core::Host) the application implements, so the crate -//! builds for `wasm32-unknown-unknown` (browsers, Cloudflare Workers) as well as native targets. +//! and written through a [`Store`](lakekit_core::Store) the application supplies, and commits take +//! the time from a [`Clock`](lakekit_core::Clock) and random numbers from a +//! [`Random`](lakekit_core::Random), so the crate builds for +//! `wasm32-unknown-unknown` (browsers, Cloudflare Workers) as well as native targets. //! //! - Reading (always): table metadata of format versions 1 to 3 ([`Metadata`]), schemas and //! partition specs, manifest lists and manifests with data and delete files ([`read_manifest`]), @@ -76,7 +78,7 @@ fn corrupt(msg: impl Into) -> Error { Error::Corrupt(msg.into()) } -/// Runs a future whose host calls complete at once ([`lakekit_core::MemoryHost`]'s do). +/// Runs a future whose store calls complete at once ([`lakekit_core::MemoryStore`]'s do). #[cfg(test)] fn block_on(f: F) -> F::Output { let mut cx = std::task::Context::from_waker(std::task::Waker::noop()); diff --git a/crates/lakekit-iceberg/src/table.rs b/crates/lakekit-iceberg/src/table.rs index a2a7df2..2602338 100644 --- a/crates/lakekit-iceberg/src/table.rs +++ b/crates/lakekit-iceberg/src/table.rs @@ -3,7 +3,7 @@ //! them, with `version-hint.text` naming the newest for stores that can't list. Tables a catalog //! names with other file names load with [`Table::from_metadata_file`]. -use lakekit_core::Host; +use lakekit_core::Store; use crate::manifest::{Entry, ManifestFile, Status, read_manifest, read_manifest_list}; use crate::metadata::Metadata; @@ -56,12 +56,12 @@ impl Table { /// The newest metadata version of the table at `location`: the highest listed, or what /// `version-hint.text` says where nothing is listed. - pub async fn latest_version(host: &H, location: &str) -> Result { - let listed = host.list(&format!("{location}/metadata/v")).await?; + pub async fn latest_version(store: &S, location: &str) -> Result { + let listed = store.list(&format!("{location}/metadata/v")).await?; if let Some(newest) = listed.iter().filter_map(|o| metadata_version(&o.uri)).max() { return Ok(newest); } - let hint = host.get(&hint_uri(location)).await?.ok_or_else(|| Error::NotFound(location.to_string()))?; + let hint = store.get(&hint_uri(location)).await?.ok_or_else(|| Error::NotFound(location.to_string()))?; std::str::from_utf8(&hint) .ok() .and_then(|h| h.trim().parse().ok()) @@ -69,22 +69,22 @@ impl Table { } /// The newest version of the table at `location`. - pub async fn load(host: &H, location: &str) -> Result { - let version = Self::latest_version(host, location).await?; - Self::load_version(host, location, version).await + pub async fn load(store: &S, location: &str) -> Result { + let version = Self::latest_version(store, location).await?; + Self::load_version(store, location, version).await } /// Version `version` of the table at `location`. - pub async fn load_version(host: &H, location: &str, version: u64) -> Result { + pub async fn load_version(store: &S, location: &str, version: u64) -> Result { let uri = metadata_uri(location, version); - let bytes = host.get(&uri).await?.ok_or_else(|| Error::NotFound(uri.clone()))?; + let bytes = store.get(&uri).await?.ok_or_else(|| Error::NotFound(uri.clone()))?; Ok(Self::new(location, version, Metadata::parse(&bytes)?)) } /// The table whose metadata file is `uri` (a catalog's pointer), at the location the metadata /// records. - pub async fn from_metadata_file(host: &H, uri: &str) -> Result { - let bytes = host.get(uri).await?.ok_or_else(|| Error::NotFound(uri.to_string()))?; + pub async fn from_metadata_file(store: &S, uri: &str) -> Result { + let bytes = store.get(uri).await?.ok_or_else(|| Error::NotFound(uri.to_string()))?; let metadata = Metadata::parse(&bytes)?; let location = metadata.location().trim_end_matches('/').to_string(); let table = Self::new(location, metadata_version(uri).unwrap_or(0), metadata); @@ -92,28 +92,28 @@ impl Table { } /// The current snapshot's manifests. - pub async fn manifests(&self, host: &H) -> Result> { + pub async fn manifests(&self, store: &S) -> Result> { let Some(list) = self.metadata.manifest_list() else { if self.metadata.current_snapshot().is_some() { return Err(Error::Unsupported("format v1 snapshots without a manifest list".into())); } return Ok(vec![]); }; - let bytes = host.get(&list).await?.ok_or_else(|| Error::NotFound(list.clone()))?; + let bytes = store.get(&list).await?.ok_or_else(|| Error::NotFound(list.clone()))?; read_manifest_list(&bytes) } /// The entries of `manifest`. - pub async fn manifest(host: &H, manifest: &ManifestFile) -> Result> { - let bytes = host.get(&manifest.path).await?.ok_or_else(|| Error::NotFound(manifest.path.clone()))?; + pub async fn manifest(store: &S, manifest: &ManifestFile) -> Result> { + let bytes = store.get(&manifest.path).await?.ok_or_else(|| Error::NotFound(manifest.path.clone()))?; read_manifest(manifest, &bytes) } /// The live files (not deleted by the current snapshot), read from the manifests unless known. - pub async fn live_files(&mut self, host: &H) -> Result<&[Entry]> { + pub async fn live_files(&mut self, store: &S) -> Result<&[Entry]> { if self.files.is_none() { - let manifests = self.manifests(host).await?; - let read = futures_util::future::try_join_all(manifests.iter().map(|m| Self::manifest(host, m))).await?; + let manifests = self.manifests(store).await?; + let read = futures_util::future::try_join_all(manifests.iter().map(|m| Self::manifest(store, m))).await?; let files = read.into_iter().flatten().filter(|e| e.status != Status::Deleted).collect(); self.files = Some(files); } @@ -123,7 +123,7 @@ impl Table { #[cfg(test)] mod tests { - use lakekit_core::MemoryHost; + use lakekit_core::MemoryStore; use super::*; use crate::block_on; @@ -138,22 +138,22 @@ mod tests { #[test] fn loads_the_newest_version() { - let host = MemoryHost::new(0); + let store = MemoryStore::new(); let meta = |n: u8| format!(r#"{{"format-version": 2, "location": "memory://t", "last-sequence-number": {n}}}"#); block_on(async { - assert!(matches!(Table::load(&host, "memory://t").await, Err(Error::NotFound(_)))); + assert!(matches!(Table::load(&store, "memory://t").await, Err(Error::NotFound(_)))); // Nothing listed: the hint names the version. - host.put(&hint_uri("memory://t"), b"2\n".to_vec()).await?; - assert_eq!(Table::latest_version(&host, "memory://t").await?, 2); - host.put(&metadata_uri("memory://t", 2), meta(2).into_bytes()).await?; - assert_eq!(Table::load(&host, "memory://t").await?.version, 2); - host.put(&metadata_uri("memory://t", 10), meta(10).into_bytes()).await?; - let mut t = Table::load(&host, "memory://t").await?; + store.put(&hint_uri("memory://t"), b"2\n".to_vec()).await?; + assert_eq!(Table::latest_version(&store, "memory://t").await?, 2); + store.put(&metadata_uri("memory://t", 2), meta(2).into_bytes()).await?; + assert_eq!(Table::load(&store, "memory://t").await?.version, 2); + store.put(&metadata_uri("memory://t", 10), meta(10).into_bytes()).await?; + let mut t = Table::load(&store, "memory://t").await?; assert_eq!((t.version, t.metadata.next_sequence_number()), (10, 11)); - assert!(t.live_files(&host).await?.is_empty(), "no snapshot, no files"); + assert!(t.live_files(&store).await?.is_empty(), "no snapshot, no files"); let uri = "memory://t/metadata/00002-a.metadata.json"; - host.put(uri, meta(2).into_bytes()).await?; - let t = Table::from_metadata_file(&host, uri).await?; + store.put(uri, meta(2).into_bytes()).await?; + let t = Table::from_metadata_file(&store, uri).await?; assert_eq!((t.location.as_str(), t.version, t.metadata_file.as_str()), ("memory://t", 2, uri)); Ok::<_, Error>(()) }) diff --git a/crates/lakekit-iceberg/tests/iceberg_rust.rs b/crates/lakekit-iceberg/tests/iceberg_rust.rs index 88878ed..35fd18d 100644 --- a/crates/lakekit-iceberg/tests/iceberg_rust.rs +++ b/crates/lakekit-iceberg/tests/iceberg_rust.rs @@ -21,7 +21,7 @@ use iceberg::spec::{ use iceberg::transaction::{ApplyTransactionAction, Transaction}; use iceberg::writer::file_writer::{FileWriter as _, FileWriterBuilder, ParquetWriterBuilder}; use iceberg::{Catalog, CatalogBuilder, MemoryCatalog, NamespaceIdent, TableCreation, TableIdent}; -use lakekit_core::{Error, Host, Listed, MemoryHost, MemoryUpload}; +use lakekit_core::{Listed, MemoryUpload, SeededRandom, Services, Store, StoreError as Error, TestClock}; use lakekit_iceberg::stats::ParquetStats; use lakekit_iceberg::{Change, CommitOptions, DataFile, Datum, Entry, NoHooks, Status, Table, commit}; use parquet::arrow::ArrowWriter; @@ -32,76 +32,66 @@ use parquet::file::properties::WriterProperties; const T0: i64 = 1_767_225_600_000_000; const DAY_US: i64 = 86_400_000_000; -/// A [`Host`] over iceberg-rust's `FileIO`, so both see the same objects. Clock and randomness come -/// from a [`MemoryHost`]; listing isn't needed (tables load from their metadata file). -struct IoHost { +/// A [`Store`] over iceberg-rust's `FileIO`, so both see the same objects. Listing isn't needed +/// (tables load from their metadata file). +struct IoStore { io: FileIO, - memory: MemoryHost, } +/// The store's results. +type Result = std::result::Result; + fn store(e: impl std::fmt::Display) -> Error { Error::Store(e.to_string()) } -impl Host for IoHost { +impl Store for IoStore { type Upload = MemoryUpload; - async fn get(&self, uri: &str) -> lakekit_core::Result>> { + async fn get(&self, uri: &str) -> Result>> { if !self.io.exists(uri).await.map_err(store)? { return Ok(None); } Ok(Some(self.io.new_input(uri).map_err(store)?.read().await.map_err(store)?.to_vec())) } - async fn get_range(&self, uri: &str, start: u64, len: u64) -> lakekit_core::Result> { + async fn get_range(&self, uri: &str, start: u64, len: u64) -> Result> { let bytes = self.get(uri).await?.ok_or_else(|| Error::NotFound(uri.to_string()))?; let start = usize::try_from(start).map_err(store)?.min(bytes.len()); let end = start.saturating_add(usize::try_from(len).map_err(store)?).min(bytes.len()); Ok(bytes[start..end].to_vec()) } - async fn exists(&self, uri: &str) -> lakekit_core::Result { + async fn exists(&self, uri: &str) -> Result { self.io.exists(uri).await.map_err(store) } - async fn put(&self, uri: &str, bytes: Vec) -> lakekit_core::Result<()> { + async fn put(&self, uri: &str, bytes: Vec) -> Result<()> { self.io.new_output(uri).map_err(store)?.write(Bytes::from(bytes)).await.map_err(store) } /// The tests run one writer at a time, so checking first is enough. - async fn put_if_absent(&self, uri: &str, bytes: Vec) -> lakekit_core::Result { + async fn put_if_absent(&self, uri: &str, bytes: Vec) -> Result { if self.exists(uri).await? { return Ok(false); } self.put(uri, bytes).await.map(|()| true) } - fn upload(&self, _: &str) -> impl Future> { + fn upload(&self, _: &str) -> impl Future> { ready(Err(Error::Unsupported("uploads".into()))) } - fn list(&self, _: &str) -> impl Future>> { + fn list(&self, _: &str) -> impl Future>> { ready(Err(Error::Unsupported("listing".into()))) } - async fn delete(&self, uris: &[String]) -> lakekit_core::Result<()> { + async fn delete(&self, uris: &[String]) -> Result<()> { for uri in uris { self.io.delete(uri).await.map_err(store)?; } Ok(()) } - - fn now_ms(&self) -> i64 { - self.memory.now_ms() - } - - fn random(&self) -> u64 { - self.memory.random() - } - - fn sleep(&self, ms: u64) -> impl Future { - self.memory.sleep(ms) - } } fn ice_schema() -> IceSchema { @@ -152,14 +142,14 @@ async fn write_theirs(table: &iceberg::table::Table, batch: &RecordBatch, name: } /// Writes `batch` with arrow-rs's Parquet writer; returns lakekit's data file. -async fn write_ours(host: &IoHost, table: &Table, batch: &RecordBatch, name: &str, day: i32) -> DataFile { +async fn write_ours(store: &IoStore, table: &Table, batch: &RecordBatch, name: &str, day: i32) -> DataFile { let mut bytes = vec![]; let mut writer = ArrowWriter::try_new(&mut bytes, batch.schema(), None).unwrap(); writer.write(batch).unwrap(); let footer = writer.close().unwrap(); let stats = ParquetStats::new(&footer, bytes.len() as u64, nan_counts(batch)); let uri = format!("{}/data/{name}.parquet", table.location); - host.put(&uri, bytes).await.unwrap(); + store.put(&uri, bytes).await.unwrap(); stats.data_file(uri, vec![Some(Datum::Date(day))]) } @@ -213,9 +203,9 @@ async fn planned(io: &FileIO, metadata_file: &str) -> Vec { paths } -async fn live(host: &IoHost, metadata_file: &str) -> (Table, Vec) { - let mut table = Table::from_metadata_file(host, metadata_file).await.unwrap(); - let mut files = table.live_files(host).await.unwrap().to_vec(); +async fn live(store: &IoStore, metadata_file: &str) -> (Table, Vec) { + let mut table = Table::from_metadata_file(store, metadata_file).await.unwrap(); + let mut files = table.live_files(store).await.unwrap().to_vec(); files.sort_by(|a, b| a.file.path.cmp(&b.file.path)); (table, files) } @@ -238,14 +228,15 @@ async fn tables_round_trip_between_lakekit_and_iceberg_rust() { Some(UnboundPartitionSpec::builder().add_partition_field(1, "time_day", IceTransform::Day).unwrap().build()); let table = catalog.create_table(&ns, creation).await.unwrap(); let io = table.file_io().clone(); - let host = IoHost { io: io.clone(), memory: MemoryHost::new(T0 / 1000) }; + let store = IoStore { io: io.clone() }; + let (clock, random) = (TestClock::at_ms(T0 / 1000), SeededRandom::new(0)); let arrow = Arc::new(iceberg::arrow::schema_to_arrow_schema(table.metadata().current_schema()).unwrap()); // iceberg-rust appends; lakekit reads the file as iceberg-rust records it. let first = batch(&arrow, 0, 5_000); let theirs = write_theirs(&table, &first, "theirs-1", 20_454).await; let table = append_theirs(&catalog, &table, theirs.clone()).await; - let (base, files) = live(&host, table.metadata_location().unwrap()).await; + let (base, files) = live(&store, table.metadata_location().unwrap()).await; assert_eq!(files.len(), 1); assert_eq!(*files[0].file, as_ours(&theirs)); assert_eq!((files[0].status, files[0].sequence_number, files[0].file_sequence_number), (Status::Added, 1, 1)); @@ -257,9 +248,9 @@ async fn tables_round_trip_between_lakekit_and_iceberg_rust() { assert_eq!(stats.data_file(theirs.file_path().into(), vec![Some(Datum::Date(20_454))]), as_ours(&theirs)); // lakekit commits on top; iceberg-rust reads the new version and lakekit's manifests. - let ours = write_ours(&host, &base, &batch(&arrow, 1, 3_000), "ours-1", 20_455).await; + let ours = write_ours(&store, &base, &batch(&arrow, 1, 3_000), "ours-1", 20_455).await; let (committed, info) = commit( - &host, + Services::new(&store, &clock, &random), base, &Change::Append { files: vec![ours.clone()] }, &HashMap::new(), @@ -284,11 +275,11 @@ async fn tables_round_trip_between_lakekit_and_iceberg_rust() { let ident = TableIdent::from_strs(["ns", "registered"]).unwrap(); let copy = committed.metadata_file.replace("v2.metadata.json", "00002-1d5e8c1a-0000-4000-8000-000000000000.metadata.json"); - host.put(©, committed.metadata.to_json().into_bytes()).await.unwrap(); + store.put(©, committed.metadata.to_json().into_bytes()).await.unwrap(); let registered = catalog.register_table(&ident, copy).await.unwrap(); let again = write_theirs(®istered, &batch(&arrow, 2, 1_000), "theirs-2", 20_456).await; let registered = append_theirs(&catalog, ®istered, again.clone()).await; - let (_, files) = live(&host, registered.metadata_location().unwrap()).await; + let (_, files) = live(&store, registered.metadata_location().unwrap()).await; let seen: Vec<(&str, i64)> = files.iter().map(|e| (e.file.path.rsplit('/').next().unwrap(), e.sequence_number)).collect(); assert_eq!(seen, [("ours-1.parquet", 2), ("theirs-1.parquet", 1), ("theirs-2.parquet", 3)]); diff --git a/crates/lakekit-iceberg/tests/object_store.rs b/crates/lakekit-iceberg/tests/object_store.rs new file mode 100644 index 0000000..8cb1bf3 --- /dev/null +++ b/crates/lakekit-iceberg/tests/object_store.rs @@ -0,0 +1,51 @@ +//! An Iceberg table committed and read back through `object_store` (`loko-object-store`), with the +//! system clock and random numbers. +#![cfg(feature = "write")] + +use std::collections::HashMap; +use std::sync::Arc; + +use lakekit_core::{OsRandom, Services, Store, TokioClock}; +use lakekit_iceberg::{Change, CommitOptions, DataFile, Datum, Metadata, NoHooks, Table, commit}; +use loko_object_store::ObjectStoreAdapter; +use object_store::memory::InMemory; + +#[tokio::test] +async fn iceberg_commits_through_object_store() { + let store = ObjectStoreAdapter::new(Arc::new(InMemory::new()), "memory://"); + let location = "memory://warehouse/t"; + let metadata = Metadata::parse( + br#"{"format-version": 2, "table-uuid": "u", "location": "memory://warehouse/t", "last-sequence-number": 0, + "last-updated-ms": 0, "last-column-id": 1, "current-schema-id": 0, + "schemas": [{"schema-id": 0, "type": "struct", "fields": [{"id": 1, "name": "x", "required": true, "type": "long"}]}], + "default-spec-id": 0, "partition-specs": [{"spec-id": 0, "fields": []}], "last-partition-id": 999, + "default-sort-order-id": 0, "sort-orders": [{"order-id": 0, "fields": []}], "properties": {}, + "current-snapshot-id": -1, "snapshots": [], "snapshot-log": [], "metadata-log": []}"#, + ) + .unwrap(); + let table = Table::new(location, 1, metadata); + store.put(&table.metadata_file, table.metadata.to_json().into_bytes()).await.unwrap(); + let file = DataFile { + path: format!("{location}/data/a.parquet"), + partition: Vec::>::new(), + record_count: 3, + file_size: 30, + ..DataFile::default() + }; + let change = Change::Append { files: vec![file] }; + let (_, info) = commit( + Services::new(&store, &TokioClock, &OsRandom), + table, + &change, + &HashMap::new(), + &CommitOptions::default(), + &mut NoHooks, + ) + .await + .unwrap(); + assert_eq!(info.version, 2); + + let mut loaded = Table::load(&store, location).await.unwrap(); + assert_eq!(loaded.version, 2); + assert_eq!(loaded.live_files(&store).await.unwrap()[0].file.record_count, 3); +} diff --git a/crates/lakekit/Cargo.toml b/crates/lakekit/Cargo.toml index 31e61f7..686d2fd 100644 --- a/crates/lakekit/Cargo.toml +++ b/crates/lakekit/Cargo.toml @@ -20,10 +20,17 @@ iceberg-write = ["iceberg", "lakekit-iceberg/write"] iceberg-arrow = ["iceberg", "lakekit-iceberg/arrow"] # Iceberg statistics from Parquet footers. iceberg-parquet-stats = ["iceberg", "lakekit-iceberg/parquet-stats"] +# A store over an object_store client (loko-object-store). +object-store = ["dep:loko-object-store"] +# TokioClock, a native clock (lakekit-core/tokio). +tokio = ["lakekit-core/tokio"] +# OsRandom, the operating system's random numbers (lakekit-core/getrandom). +getrandom = ["lakekit-core/getrandom"] [dependencies] lakekit-core.workspace = true lakekit-iceberg = { workspace = true, optional = true } +loko-object-store = { workspace = true, optional = true } [package.metadata.docs.rs] all-features = true diff --git a/crates/lakekit/src/lib.rs b/crates/lakekit/src/lib.rs index 50feaf8..cbc7e9c 100644 --- a/crates/lakekit/src/lib.rs +++ b/crates/lakekit/src/lib.rs @@ -1,11 +1,16 @@ //! Table formats read and written without I/O of their own, for WebAssembly and native: one crate -//! per format, re-exported here behind a feature each. The I/O boundary is [`core::Host`]. +//! per format, re-exported here behind a feature each. The I/O boundary is +//! [`core::Store`], [`core::Clock`], and [`core::Random`]. //! //! | Feature | Crate | //! |---|---| //! | `iceberg` (default) | [`iceberg`]: Apache Iceberg (`lakekit-iceberg`) | //! | `iceberg-write`, `iceberg-arrow`, `iceberg-parquet-stats` | its optional parts | +//! | `object-store` | `object_store`: a store over an `object_store` client (`loko-object-store`) | +//! | `tokio`, `getrandom` | `core::TokioClock` and `core::OsRandom`, native ones | pub use lakekit_core as core; #[cfg(feature = "iceberg")] pub use lakekit_iceberg as iceberg; +#[cfg(feature = "object-store")] +pub use loko_object_store as object_store; diff --git a/spec/0001-host.md b/spec/0001-host.md deleted file mode 100644 index 416ff6d..0000000 --- a/spec/0001-host.md +++ /dev/null @@ -1,50 +0,0 @@ ---- -number: 1 -title: The host boundary -status: done -updated: 2026-10-20 -versions: [unreleased] -related: [2, 3] -issues: [] ---- - -# 1. The host boundary - -**Done** · updated 2026-10-20 · versions unreleased · milestone: [0.1](milestones/v0-1.md) - -## Summary - -Format crates do no I/O of their own. They call a `Host` (`lakekit-core`) that the application implements over its -object store, clock, and randomness, so the same code runs in a browser, a Cloudflare Worker, a CLI, or a server, and -tests run against memory. - -## Design - -- **`Host`:** `get` (the object or `None`), `get_range`, `exists`, `put`, `put_if_absent` (the commit primitive: - `false` when an object is there), `upload` (a multipart upload: parts in order, then complete or abort), `list` (by - prefix), `delete` (missing objects are not an error), `now_ms`, `random`, and `sleep`. Paths are absolute URIs, as - table metadata records them. -- **Futures aren't `Send`.** Methods return `impl Future`, without a `Send` bound, so a host can wrap JavaScript - promises (wasm-bindgen). Native hosts are free to be `Send`. -- **`MemoryHost`:** objects in memory, a clock that only moves when told (or when `sleep` is called), and a seeded - random sequence, so tests are deterministic. -- **`Error`:** one error type for all crates, with the variant saying what a caller can do: `Invalid`, `Unsupported`, - `Conflict` (retry on the newer version), `CommitUnknown` (the commit may have landed: delete nothing it wrote, load - the table again), `NotFound`, `Store`, and `Corrupt`. It is `non_exhaustive`. -- **Identifiers:** `uuid_v4` and `uuid_v7` from the host's randomness and clock, for file names. - -## Decisions - -| ID | Decision | Status | -|---|---|---| -| D1 | Format crates do no I/O: everything goes through a `Host` trait the application implements (the idea of delta-kernel-rs's `Engine`) | Accepted | -| D2 | Host futures aren't `Send`, so JavaScript hosts work; native applications lose nothing | Accepted | -| D3 | One error type across crates, whose variants say what a caller can do | Accepted | - -## Open questions and issues - -None open. - -## History - -- 2026-10-20: built, from Timelake's `timelake_format::host`. diff --git a/spec/0001-store-clock-random.md b/spec/0001-store-clock-random.md new file mode 100644 index 0000000..6237888 --- /dev/null +++ b/spec/0001-store-clock-random.md @@ -0,0 +1,80 @@ +--- +number: 1 +title: Store, Clock, and Random +status: done +updated: 2026-10-21 +versions: [unreleased] +related: [2, 3, 4] +issues: [] +--- + +# 1. Store, Clock, and Random + +**Done** · updated 2026-10-21 · versions unreleased · milestone: [0.1](milestones/v0-1.md) + +## Summary + +Format crates do no I/O of their own. They read and write objects through a `Store`, take the time from a `Clock`, and +take random numbers from a `Random`, all supplied by the application, so the same code runs in a +browser, a Cloudflare Worker, a CLI, or a server, and tests run against memory. + +## Design + +- **`Store`:** `get` (the object or `None`), `get_range`, `exists`, `put`, `put_if_absent` (the commit primitive: + `false` when an object is there), `upload` (a multipart upload: parts in order, then complete or abort), `list` (by + string prefix, not only by directory), and `delete` (missing objects are not an error). Paths are absolute URIs, as + table metadata records them. Adapters implement it over store libraries ([spec 4](0004-store-adapters.md)). +- **Where they come from:** the three are [loko](https://github.com/polythings/loko)'s services (`loko-store`, + `loko-clock`, `loko-random`), shared by the polythings projects; `lakekit-core` re-exports them with their test and + native implementations, and adds what is lakekit's: the error type, the identifiers, and `Services`. +- **`Clock`:** `now` and `now_ms`, and `sleep(Duration)`: commit timestamps, time-ordered file names, and the wait + between commit attempts. +- **`Random`:** `rng()`, a generator of the caller's own: snapshot ids, file names, and the jitter between commit + attempts. +- **Who needs what:** readers take only a `Store`; commits take `Services`, a store, a clock, and a random source. +- **`Services`:** a struct of three references, `store`, `clock`, and `random`, each borrowed from wherever the + caller keeps it: `Services::new(&s3, &TokioClock, &OsRandom)`. One object that is all three (a JavaScript host) goes + in all three places: `Services::new(&host, &host, &host)`. It's `Copy`, and commit hooks receive it too. +- **Native implementations:** `TokioClock` (feature `tokio`: the system clock and tokio's timer, native targets) and + `OsRandom` (feature `getrandom`: the operating system's generator, native targets and WASI). +- **Futures aren't `Send`.** Methods return `impl Future`, without a `Send` bound, so a store or clock can wrap + JavaScript promises (wasm-bindgen). Native ones are free to be `Send`. +- **For tests:** a `MemoryStore`, a `TestClock` that only moves when told (or when `sleep` is called), and a + `SeededRandom`, so tests are deterministic. +- **`Error`:** one error type for all crates, with the variant saying what a caller can do: `Invalid`, `Unsupported`, + `Conflict` (retry on the newer version), `CommitUnknown` (the commit may have landed: delete nothing it wrote, load + the table again), `NotFound`, `Store`, and `Corrupt`. It is `non_exhaustive`. A store's error (`StoreError`) converts + into it, keeping `NotFound`, `Invalid`, and `Unsupported`. +- **Identifiers:** `uuid_v4` (from a `Random`) and `uuid_v7` (from a `Clock + Random`), built by the `uuid` crate (its + builder needs neither its own randomness nor `std::time`, so it works in WebAssembly). + +## Decisions + +| ID | Decision | Status | +|---|---|---| +| D1 | Format crates do no I/O: everything goes through traits the application implements (the idea of delta-kernel-rs's `Engine`) | Accepted | +| D2 | Store and clock futures aren't `Send`, so JavaScript hosts work; native applications lose nothing | Accepted | +| D3 | One error type across crates, whose variants say what a caller can do | Accepted | +| D10 | Objects are a `Store` of their own, apart from the time and randomness: readers need only a store, and store adapters don't decide how time is told | Accepted | +| D13 | The time and randomness are two traits, `Clock` and `Random`, as ZIO's services are, rather than one environment trait: each says what it is, and each can be swapped on its own | Accepted | +| D14 | `Store`, `Clock`, and `Random` are loko's services, which every polythings project shares, rather than lakekit's own; `lakekit-core` re-exports them and keeps the error type | Accepted | +| D16 | Commits take `Services`, three references with named fields, rather than one value bounded by `Store + Clock + Random`: callers pass what they have without gluing it into a struct that implements all three, and the code says which service each call uses | Accepted | + +## Open questions and issues + +None open. + +## History + +- 2026-10-20: built, from Timelake's `timelake_format::host`. +- 2026-10-21: one `Host` trait split into `Store` (D10), `Clock`, and `Random` (D13); `Services`, `TokioClock`, and + `OsRandom` added; UUIDs from the `uuid` crate. The spec was "The host boundary". +- 2026-10-21: the traits come from loko (D14); `Clock::sleep` takes a `Duration`; `MemoryStore` is only a store, and + `MemoryServices` bundles it with a test clock and seeded randomness. +- 2026-10-21: `Services` and `MemoryServices` are `Resources` and `MemoryResources`, as loko calls the things it + passes services; `Services` implements loko's `Has` traits. +- 2026-10-21: loko's `Random` hands out generators (`rng()`); lakekit draws its identifiers and jitter from them. +- 2026-10-21: `Resources` and `MemoryResources` are `Services` and `MemoryServices` again, as loko calls them services + again. +- 2026-10-21: `Services` is three references with named fields, and commits and hooks take it instead of one value that + is all three (D16); `MemoryServices` is gone, and `uuid_v7` takes a clock and a random source. diff --git a/spec/0002-formats.md b/spec/0002-formats.md index 7af34e9..1551192 100644 --- a/spec/0002-formats.md +++ b/spec/0002-formats.md @@ -2,15 +2,15 @@ number: 2 title: One crate per format status: in-progress -updated: 2026-10-20 +updated: 2026-10-21 versions: [unreleased] -related: [1, 3] -issues: [6] +related: [1, 3, 4] +issues: [6, 11] --- # 2. One crate per format -**In progress** · updated 2026-10-20 · versions unreleased · milestone: [0.1](milestones/v0-1.md) +**In progress** · updated 2026-10-21 · versions unreleased · milestone: [0.1](milestones/v0-1.md) ## Summary @@ -32,6 +32,8 @@ lakeglass reads it; the formats lakeglass explores come next. - **Building for WebAssembly:** every crate builds for `wasm32-unknown-unknown` and is checked for `wasm32-wasip2`. - **Next formats,** in the order lakeglass needs them: Delta Lake, then Apache Hudi, Apache Paimon, Lance, and DuckLake. Each is proposed as its own spec before it is built. +- **Engines** get adapters the same way, a crate each, if they are built: a DataFusion table provider per format is + an open question ([#11](https://github.com/polythings/lakekit/issues/11)). ## Decisions @@ -44,6 +46,7 @@ lakeglass reads it; the formats lakeglass explores come next. ## Open questions and issues - [#6](https://github.com/polythings/lakekit/issues/6) Which format comes after Iceberg, and in what order? +- [#11](https://github.com/polythings/lakekit/issues/11) DataFusion table providers (lakekit-datafusion)? ## History diff --git a/spec/0003-iceberg.md b/spec/0003-iceberg.md index 21f4254..5fd15fb 100644 --- a/spec/0003-iceberg.md +++ b/spec/0003-iceberg.md @@ -41,13 +41,13 @@ shared writer, made general. any metadata file a catalog names (`from_metadata_file`); manifests and live files. - **Writing (feature `write`):** format v2 data manifests and manifest lists in the layout iceberg-rust writes, and metadata updates (`with_snapshot`, `without_snapshots`) made as iceberg-rust's builder makes them, with the metadata - log trimmed to `write.metadata.previous-versions-max`; the current snapshot can't be removed. Partition values of type uuid, fixed, and decimal are refused - for now. + log trimmed to `write.metadata.previous-versions-max`; the current snapshot can't be removed. Partition values of + type uuid, fixed, and decimal are refused for now. - **Commits (feature `write`):** upload a snapshot's files, then write `v.metadata.json` only if absent; on a lost race, reload and try again on the newer version, deleting the lost attempt's files; when the outcome is unknown (a conditional put that failed and can't be read back), delete nothing. Appends, replacements that keep data - sequence numbers (compaction, one sequence number per added file), and deletions; manifests merge once there would be more than `merge_manifests_at`. - Iceberg's summary counters are kept. A writer whose clock is behind the table's last update dates its commit at that + sequence numbers (compaction, one sequence number per added file), and deletions; manifests merge once there would be + more than `merge_manifests_at`. Iceberg's summary counters are kept. A writer whose clock is behind the table's last update dates its commit at that update, so the snapshot log stays in order (iceberg-rust refuses tables whose log isn't). - **Options and hooks:** `CommitOptions` (`max_attempts` 10, `backoff_ms` 5–50 times the attempt, `merge_manifests_at` 8, `write_version_hint` on) and `CommitHooks` (`reload`, `skip` for idempotent retries, `extra` for files and diff --git a/spec/0004-store-adapters.md b/spec/0004-store-adapters.md new file mode 100644 index 0000000..8b2c8df --- /dev/null +++ b/spec/0004-store-adapters.md @@ -0,0 +1,60 @@ +--- +number: 4 +title: Store adapters +status: done +updated: 2026-10-21 +versions: [unreleased] +related: [1] +issues: [] +--- + +# 4. Store adapters + +**Done** · updated 2026-10-21 · versions unreleased · milestone: [0.1](milestones/v0-1.md) + +## Summary + +Ready-made `Store`s ([spec 1](0001-store-clock-random.md)) over the store libraries applications already use, so a +native application doesn't write its own. They are [loko](https://github.com/polythings/loko)'s crates, since `Store` +is loko's service; `lakekit` re-exports `loko-object-store` (feature `object-store`). Each adapter is a crate of its +own; the `Store` trait stays the boundary, +because some stores aren't these libraries (a Cloudflare R2 binding reached from JavaScript) and applications wrap +their store (lakeglass counts every request). + +## Design + +**`loko-object-store`** (built, as `lakekit-object-store` first): `ObjectStoreAdapter` over an `Arc` (0.14). + +- **Names:** lakekit uses absolute URIs; the adapter has a root (`s3://bucket/`, `memory://`, `file:///` for a local + file system) and maps `` to the store's path ``. A URI outside the root is `Error::Invalid`. +- **Operations:** `get` (missing is `None`), `get_range`, `exists` (a `head`), `put`, `put_if_absent` as + `PutMode::Create` (`AlreadyExists` or `Precondition` is `false`), multipart uploads, `delete` (missing is fine, + deletes run concurrently). `list` is by string prefix: the adapter lists the prefix's directory and keeps what + starts with the prefix, since `object_store` lists whole path segments. +- **Construction:** `new(store, root)`, or `from_url(url, options)` through `object_store::parse_url_opts` for the + URL's scheme and bucket. +- **Backends are the application's:** the crate enables no `object_store` features. Applications turn on `aws`, + `gcp`, `azure`, `http`, or `fs` (or the `*-base` variants, which build for WebAssembly) on their own dependency, so + the adapter builds for `wasm32-unknown-unknown` and pulls in nothing unasked. +- **Tests:** objects, conditional writes, multipart, prefix listing, and local files through the adapter (in loko), + and an Iceberg commit loaded back through it (here, `lakekit-iceberg/tests/object_store.rs`). + +**OpenDAL** ([loko#6](https://github.com/polythings/loko/issues/6)): the same over an OpenDAL `Operator`, once an +application needs it, as a loko crate. + +## Decisions + +| ID | Decision | Status | +|---|---|---| +| D11 | Store libraries are adapters, one crate each, over the `Store` trait; lakekit's crates never depend on them | Accepted | +| D12 | An adapter enables none of its library's backends; the application chooses them | Accepted | +| D15 | Store adapters are loko's crates, next to the `Store` trait; lakekit uses them | Accepted | + +## Open questions and issues + +None open here. The OpenDAL adapter is [loko#6](https://github.com/polythings/loko/issues/6). + +## History + +- 2026-10-21: proposed, and `lakekit-object-store` built. +- 2026-10-21: the adapter moved to loko as `loko-object-store` (D15); the OpenDAL issue moved with it. diff --git a/spec/README.md b/spec/README.md index b973c5c..0f1a882 100644 --- a/spec/README.md +++ b/spec/README.md @@ -28,13 +28,14 @@ lakekit's design, laid out like Timelake's and lakeglass's. Start here, people a ## Index -3 specs: 2 in progress, 1 done. Decisions D1–D9. +4 specs: 2 in progress, 2 done. Decisions D1–D16. | # | Spec | Status | Updated | |---:|---|---|---| -| 1 | [The host boundary](0001-host.md) | Done | 2026-10-20 | -| 2 | [One crate per format](0002-formats.md) | In progress | 2026-10-20 | +| 1 | [Store, Clock, and Random](0001-store-clock-random.md) | Done | 2026-10-21 | +| 2 | [One crate per format](0002-formats.md) | In progress | 2026-10-21 | | 3 | [Apache Iceberg](0003-iceberg.md) | In progress | 2026-10-20 | +| 4 | [Store adapters](0004-store-adapters.md) | Done | 2026-10-21 | ## Milestones diff --git a/spec/milestones/README.md b/spec/milestones/README.md index 5f6ca11..3a2f70c 100644 --- a/spec/milestones/README.md +++ b/spec/milestones/README.md @@ -4,4 +4,4 @@ The roadmap: goals made of specs. A milestone is done, in progress, planned, or | Milestone | Status | Specs | |---|---|---| -| [0.1: Iceberg for Timelake and lakeglass](v0-1.md) | In progress | 1, 2, 3 | +| [0.1: Iceberg for Timelake and lakeglass](v0-1.md) | In progress | 1, 2, 3, 4 | diff --git a/spec/milestones/v0-1.md b/spec/milestones/v0-1.md index 83e8246..e3e40ba 100644 --- a/spec/milestones/v0-1.md +++ b/spec/milestones/v0-1.md @@ -2,14 +2,16 @@ **In progress** -The first release: the host boundary and the Iceberg crate, enough for Timelake's WASM reader and writer to use -lakekit instead of their own Iceberg code, and for lakeglass to read Iceberg tables in the browser. +The first release: the Iceberg crate on loko's store, clock, and random services, and the `object_store` adapter, enough for +Timelake's WASM reader and writer to use lakekit instead of their own Iceberg code, and for lakeglass to read Iceberg +tables in the browser. | # | Spec | Status | |---:|---|---| -| 1 | [The host boundary](../0001-host.md) | Done | +| 1 | [Store, Clock, and Random](../0001-store-clock-random.md) | Done | | 2 | [One crate per format](../0002-formats.md) | In progress | | 3 | [Apache Iceberg](../0003-iceberg.md) | In progress | +| 4 | [Store adapters](../0004-store-adapters.md) | Done | -**Done when:** the crates are on crates.io ([#7](https://github.com/polythings/lakekit/issues/7)), Timelake depends +**Done when:** loko 0.2.0 ([loko#1](https://github.com/polythings/loko/issues/1)) and these crates are on crates.io ([#7](https://github.com/polythings/lakekit/issues/7)), Timelake depends on them by version, and lakeglass's Iceberg reader is built on them.