From 1c015854b7cb1748d8a54c40a3b9d8dfb67861f1 Mon Sep 17 00:00:00 2001 From: Beinan Date: Mon, 19 Oct 2026 02:28:06 +0000 Subject: [PATCH] perf: add Rust WAL codec and durable disk experiments --- .github/workflows/rust-test.yml | 4 + .github/workflows/style.yml | 2 + Cargo.lock | 321 +++++++++++++++++-- Cargo.toml | 1 + crates/lance-context-wal/Cargo.toml | 28 ++ crates/lance-context-wal/README.md | 43 +++ crates/lance-context-wal/src/bin/lab.rs | 408 ++++++++++++++++++++++++ crates/lance-context-wal/src/lib.rs | 5 + 8 files changed, 779 insertions(+), 33 deletions(-) create mode 100644 crates/lance-context-wal/Cargo.toml create mode 100644 crates/lance-context-wal/README.md create mode 100644 crates/lance-context-wal/src/bin/lab.rs create mode 100644 crates/lance-context-wal/src/lib.rs diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 5ff351a0..6dbf727f 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -59,6 +59,10 @@ jobs: # regression test was dead weight in CI. `--workspace` covers the crates # that were silently untested (api, server, client, metrics, the facade). run: cargo test --workspace --all-targets + - name: WAL component correctness smoke tests + run: | + cargo run -p lance-context-wal --features lab --bin context-wal-lab -- codec --payload-bytes 1024 --rows 8 --iterations 2 + cargo run -p lance-context-wal --features lab --bin context-wal-lab -- disk --directory "$RUNNER_TEMP/wal-disk-smoke" --payload-bytes 1024 --batch 4 --writers 2 --groups-per-writer 2 --rounds 8 - name: Start etcd for HA scheduler test run: | ETCD_VERSION=3.7.0 diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml index fae651c2..151a2aea 100644 --- a/.github/workflows/style.yml +++ b/.github/workflows/style.yml @@ -55,6 +55,8 @@ jobs: # cfg-gated breakage is caught here instead of at release time. - name: Clippy (lance-context, default features) run: cargo clippy --manifest-path crates/lance-context/Cargo.toml --all-targets -- -D warnings + - name: Clippy (WAL component experiments) + run: cargo clippy -p lance-context-wal --features lab --all-targets -- -D warnings typos: name: Spell Check diff --git a/Cargo.lock b/Cargo.lock index 4686434a..963cc91d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -332,7 +332,7 @@ version = "58.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f633dbfdf39c039ada1bf9e34c694816eb71fbb7dc78f613993b7245e078a1ed" dependencies = [ - "bitflags", + "bitflags 2.11.1", "serde_core", "serde_json", ] @@ -961,7 +961,7 @@ version = "0.72.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895" dependencies = [ - "bitflags", + "bitflags 2.11.1", "cexpr", "clang-sys", "itertools 0.13.0", @@ -973,6 +973,12 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + [[package]] name = "bitflags" version = "2.11.1" @@ -1241,7 +1247,7 @@ version = "4.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.117", @@ -1299,8 +1305,8 @@ version = "7.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e0d05af1e006a2407bedef5af410552494ce5be9090444dbbcb57258c1af3d56" dependencies = [ - "strum", - "strum_macros", + "strum 0.26.3", + "strum_macros 0.26.4", "unicode-width", ] @@ -1438,6 +1444,19 @@ dependencies = [ "cfg-if 1.0.4", ] +[[package]] +name = "crossbeam" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e71406cd8807725f7ac2f999a4cdd32e98f829fdf65f528343cebf945e41df1e" +dependencies = [ + "crossbeam-channel", + "crossbeam-deque", + "crossbeam-epoch", + "crossbeam-queue", + "crossbeam-utils", +] + [[package]] name = "crossbeam-channel" version = "0.5.15" @@ -2289,7 +2308,7 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e0e367e4e7da84520dedcac1901e4da967309406d1e51017ae1abfb97adbd38" dependencies = [ - "bitflags", + "bitflags 2.11.1", "objc2", ] @@ -2411,6 +2430,17 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "fail" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5e43d0f78a42ad591453aedb1d7ae631ce7ee445c7643691055a9ed8d3b01c" +dependencies = [ + "log", + "once_cell", + "rand 0.8.6", +] + [[package]] name = "fast-float2" version = "0.2.3" @@ -2441,7 +2471,7 @@ version = "25.12.19" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "35f6839d7b3b98adde531effaf34f0c2badc6f4735d26fe74709d8e513a96ef3" dependencies = [ - "bitflags", + "bitflags 2.11.1", "rustc_version", ] @@ -2525,7 +2555,7 @@ version = "0.10.0-dev" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e4fee46bea69e0596130e3210e65d3424e0ac1e6df3bde6636304bdf1ca4a3b" dependencies = [ - "memoffset", + "memoffset 0.9.1", ] [[package]] @@ -2536,7 +2566,7 @@ checksum = "ae51f5089f3d9025ae77f17ea66d2489ac0f82fbb1d91462807bea174d486d82" dependencies = [ "anyhow", "asyncband", - "bitflags", + "bitflags 2.11.1", "datasketches", "equivalent", "foyer-common", @@ -2568,6 +2598,16 @@ version = "1.20260821.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e2488a10218a17366ce6bd38b6d081b4387a9548d350029db9feddc9bb44d6d" +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -3009,6 +3049,10 @@ name = "hashbrown" version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", + "allocator-api2", +] [[package]] name = "hashbrown" @@ -3043,6 +3087,20 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hdrhistogram" +version = "7.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f49d1053f4708f0af3cf9fc5bffc7e68a914a3c45becb231c80068c9c3f78bea" +dependencies = [ + "base64 0.22.1", + "byteorder", + "crossbeam-channel", + "flate2", + "nom 8.0.0", + "num-traits", +] + [[package]] name = "heapify" version = "0.2.0" @@ -3059,6 +3117,12 @@ dependencies = [ "stable_deref_trait", ] +[[package]] +name = "heck" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8" + [[package]] name = "heck" version = "0.5.0" @@ -3532,6 +3596,12 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "if_chain" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb" + [[package]] name = "indexmap" version = "1.9.3" @@ -3580,7 +3650,7 @@ version = "0.7.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4d09b98f7eace8982db770e4408e7470b028ce513ac28fecdc6bf4c30fe92b62" dependencies = [ - "bitflags", + "bitflags 2.11.1", "cfg-if 1.0.4", "libc", ] @@ -4097,6 +4167,23 @@ dependencies = [ "uuid", ] +[[package]] +name = "lance-context-wal" +version = "0.6.7" +dependencies = [ + "anyhow", + "arrow-array", + "arrow-buffer", + "arrow-ipc", + "arrow-schema", + "bytes", + "clap", + "hdrhistogram", + "raft-engine", + "raft-proto", + "serde_json", +] + [[package]] name = "lance-core" version = "13.0.0" @@ -4797,6 +4884,15 @@ version = "2.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b947ae49db0d222b1dbc6b113ce7248a3fc3a6ca21b696717bfc000ba4484d8" +[[package]] +name = "memoffset" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5de893c32cde5f383baa4c04c5d6dbdd735cfd4a794b0debdb2bb1b421da5ff4" +dependencies = [ + "autocfg", +] + [[package]] name = "memoffset" version = "0.9.1" @@ -4981,13 +5077,26 @@ dependencies = [ "rawpointer", ] +[[package]] +name = "nix" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b" +dependencies = [ + "bitflags 1.3.2", + "cfg-if 1.0.4", + "libc", + "memoffset 0.7.1", + "pin-utils", +] + [[package]] name = "nix" version = "0.31.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" dependencies = [ - "bitflags", + "bitflags 2.11.1", "cfg-if 1.0.4", "cfg_aliases", "libc", @@ -5071,6 +5180,17 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-derive" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed3955f1a9c7c0c15e092f9c887db08b1fc683305fdf6eb6684f22555355e202" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "num-integer" version = "0.1.46" @@ -5148,7 +5268,7 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" dependencies = [ - "bitflags", + "bitflags 2.11.1", "dispatch2", "objc2", ] @@ -5165,7 +5285,7 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3e0adef53c21f888deb4fa59fc59f7eb17404926ee8a6f59f5df0fd7f9f3272" dependencies = [ - "bitflags", + "bitflags 2.11.1", "objc2", ] @@ -5248,7 +5368,7 @@ dependencies = [ "hyper", "itertools 0.15.0", "md-5", - "nix", + "nix 0.31.3", "parking_lot", "percent-encoding", "quick-xml 0.41.0", @@ -5593,7 +5713,7 @@ version = "0.10.80" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a45fa2aa886c42762255da344f0a0d313e254066c46aad76f300c3d3da62d967" dependencies = [ - "bitflags", + "bitflags 2.11.1", "cfg-if 1.0.4", "foreign-types", "libc", @@ -5958,7 +6078,7 @@ version = "0.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25485360a54d6861439d60facef26de713b1e126bf015ec8f98239467a2b82f7" dependencies = [ - "bitflags", + "bitflags 2.11.1", "procfs-core", "rustix", ] @@ -5969,10 +6089,37 @@ version = "0.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6401bf7b6af22f78b563665d15a22e9aef27775b79b149a66ca022468a4e405" dependencies = [ - "bitflags", + "bitflags 2.11.1", "hex", ] +[[package]] +name = "prometheus" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" +dependencies = [ + "cfg-if 1.0.4", + "fnv", + "lazy_static", + "memchr", + "parking_lot", + "protobuf", + "thiserror 1.0.69", +] + +[[package]] +name = "prometheus-static-metric" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8f30cdb09c39930b8fa5e0f23cbb895ab3f766b187403a0ba0956fc1ef4f0e5" +dependencies = [ + "lazy_static", + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "prost" version = "0.14.3" @@ -5989,7 +6136,7 @@ version = "0.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ - "heck", + "heck 0.5.0", "itertools 0.14.0", "log", "multimap", @@ -6026,13 +6173,43 @@ dependencies = [ "prost", ] +[[package]] +name = "protobuf" +version = "2.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "106dd99e98437432fed6519dedecfade6a06a73bb7b2a1e019fdd2bee5778d94" +dependencies = [ + "bytes", +] + +[[package]] +name = "protobuf-build" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2df9942df2981178a930a72d442de47e2f0df18ad68e50a30f816f1848215ad0" +dependencies = [ + "bitflags 1.3.2", + "protobuf", + "protobuf-codegen", + "regex", +] + +[[package]] +name = "protobuf-codegen" +version = "2.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "033460afb75cf755fcfc16dfaed20b86468082a2ea24e05ac35ab4a099a017d6" +dependencies = [ + "protobuf", +] + [[package]] name = "pulldown-cmark" version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e9f068eba8e7071c5f9511831b44f32c740d5adf574e990f946ddb53db2f314e" dependencies = [ - "bitflags", + "bitflags 2.11.1", "memchr", "unicase", ] @@ -6054,7 +6231,7 @@ checksum = "8970a78afe0628a3e3430376fc5fd76b6b45c4d43360ffd6cdd40bdde72b682a" dependencies = [ "indoc", "libc", - "memoffset", + "memoffset 0.9.1", "once_cell", "portable-atomic", "pyo3-build-config", @@ -6101,7 +6278,7 @@ version = "0.25.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4109984c22491085343c05b0dbc54ddc405c3cf7b4374fc533f5c3313a572ccc" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "pyo3-build-config", "quote", @@ -6238,12 +6415,57 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" +[[package]] +name = "raft-engine" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1213c3a24e3fee8afcc74b2be08c4081adde96f092e0fc1c607abb3e16ae722e" +dependencies = [ + "byteorder", + "crc32fast", + "crossbeam", + "fail", + "fs2", + "hashbrown 0.14.5", + "hex", + "if_chain", + "lazy_static", + "libc", + "log", + "lz4-sys", + "nix 0.26.4", + "num-derive", + "num-traits", + "parking_lot", + "prometheus", + "prometheus-static-metric", + "protobuf", + "rayon", + "scopeguard", + "serde", + "serde_repr", + "strum 0.25.0", + "thiserror 1.0.69", +] + +[[package]] +name = "raft-proto" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb6884896294f553e8d5cfbdb55080b9f5f2f43394afff59c9f077e0f4b46d6b" +dependencies = [ + "bytes", + "protobuf", + "protobuf-build", +] + [[package]] name = "rand" version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" dependencies = [ + "libc", "rand_chacha 0.3.1", "rand_core 0.6.4", ] @@ -6353,7 +6575,7 @@ version = "11.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186" dependencies = [ - "bitflags", + "bitflags 2.11.1", ] [[package]] @@ -6397,7 +6619,7 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags", + "bitflags 2.11.1", ] [[package]] @@ -6826,7 +7048,7 @@ version = "1.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" dependencies = [ - "bitflags", + "bitflags 2.11.1", "errno", "libc", "linux-raw-sys", @@ -7008,7 +7230,7 @@ version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ - "bitflags", + "bitflags 2.11.1", "core-foundation 0.10.1", "core-foundation-sys", "libc", @@ -7309,7 +7531,7 @@ version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f103c50866b8743da9429b8a581d81a27c2d3a9c4ac7df8f8571c1dd7896eda" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.117", @@ -7429,19 +7651,41 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "strum" +version = "0.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290d54ea6f91c969195bdbcd7442c8c2a2ba87da8bf60a7ee86a235d4bc1e125" +dependencies = [ + "strum_macros 0.25.3", +] + [[package]] name = "strum" version = "0.26.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" +[[package]] +name = "strum_macros" +version = "0.25.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23dc1fa9ac9c169a78ba62f0b841814b7abae11bdd047b9c58f893439e309ea0" +dependencies = [ + "heck 0.4.1", + "proc-macro2", + "quote", + "rustversion", + "syn 2.0.117", +] + [[package]] name = "strum_macros" version = "0.26.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "rustversion", @@ -7460,6 +7704,17 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" +[[package]] +name = "syn" +version = "1.0.109" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "2.0.117" @@ -7523,7 +7778,7 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" dependencies = [ - "bitflags", + "bitflags 2.11.1", "core-foundation 0.9.4", "system-configuration-sys", ] @@ -7915,7 +8170,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" dependencies = [ "async-compression", - "bitflags", + "bitflags 2.11.1", "bytes", "futures-core", "futures-util", @@ -8352,7 +8607,7 @@ version = "0.244.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" dependencies = [ - "bitflags", + "bitflags 2.11.1", "hashbrown 0.15.5", "indexmap 2.14.0", "semver", @@ -8748,7 +9003,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc" dependencies = [ "anyhow", - "heck", + "heck 0.5.0", "wit-parser", ] @@ -8759,7 +9014,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" dependencies = [ "anyhow", - "heck", + "heck 0.5.0", "indexmap 2.14.0", "prettyplease", "syn 2.0.117", @@ -8790,7 +9045,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" dependencies = [ "anyhow", - "bitflags", + "bitflags 2.11.1", "indexmap 2.14.0", "log", "serde", diff --git a/Cargo.toml b/Cargo.toml index 2cb7893b..c3017177 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ members = [ "crates/lance-context-merge", "crates/lance-context-core", + "crates/lance-context-wal", "crates/lance-context-metrics", "crates/lance-context", "crates/lance-context-api", diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml new file mode 100644 index 00000000..ee61044d --- /dev/null +++ b/crates/lance-context-wal/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "lance-context-wal" +version = "0.6.7" +edition = "2021" +license = "Apache-2.0" +publish = false +description = "Independent Rust WAL experiments owned by lance-context" + +[features] +lab = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "dep:arrow-schema", "dep:bytes", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "dep:raft-proto", "dep:serde_json"] + +[dependencies] +anyhow = { version = "1", optional = true } +arrow-array = { version = "58", optional = true } +arrow-buffer = { version = "58", optional = true } +arrow-ipc = { version = "58", optional = true } +arrow-schema = { version = "58", optional = true } +bytes = { version = "1", optional = true } +clap = { version = "4", features = ["derive"], optional = true } +hdrhistogram = { version = "7", optional = true } +raft-engine = { version = "=0.4.2", default-features = false, optional = true } +raft-proto = { version = "=0.7.0", optional = true } +serde_json = { version = "1", optional = true } + +[[bin]] +name = "context-wal-lab" +path = "src/bin/lab.rs" +required-features = ["lab"] diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md new file mode 100644 index 00000000..c101ddac --- /dev/null +++ b/crates/lance-context-wal/README.md @@ -0,0 +1,43 @@ +# WAL component experiments + +Feature-gated Rust measurements for choosing a WAL payload path and durable +local log engine. This crate does not change production ingestion or Lance OSS. + +```bash +cargo build --release -p lance-context-wal --features lab --bin context-wal-lab +target/release/context-wal-lab codec --payload-bytes 262144 --rows 8 --iterations 30 +target/release/context-wal-lab disk --directory /tmp/wal-disk-new \ + --payload-bytes 262144 --batch 4 --writers 4 --groups-per-writer 4 --rounds 64 +``` + +`codec` measures encoding, Arrow FileReader, shared-buffer FileDecoder, an +unaligned shared buffer, and decoding followed by per-row payload copies. All +paths use the same IPC bytes; data generation and correctness checks are outside +timing. JSON reports time, allocation count, allocated bytes and logical IPC +MiB/s. Shared-buffer decoding may inspect only metadata and retain payload +buffers: logical IPC MiB/s is not memory-copy bandwidth or a full payload scan. +Unaligned numeric buffers may need repair copies. The allocator counts allocation +requests, including reallocations, rather than peak or retained memory. + +`disk` rotates each writer over separate Raft groups sharing one real raft-engine. +Every nonempty write uses `sync=true`; JSON includes batch latency, payload +throughput, engine bytes, reopen time and separate verification time. It reopens +the engine and compares every persisted payload. Inputs are deterministic bytes, +prepared before timing and reused across rounds. Compression is disabled unless +`--compression-threshold` is supplied. The directory must be new and remains for +inspection. This measures local durable writes, not replicated ACKs, crashes, +power loss or Blob/Lance publication. + +The pinned raft-engine 0.4.2 has a separate `sync-probe` command to inspect empty +sync versus a nonempty synced barrier: + +```bash +strace -f -e trace=fsync,fdatasync target/release/context-wal-lab \ + sync-probe --directory /tmp/wal-sync-new +``` + +Use the printed markers to associate syscalls with each phase. Do not infer a +durability boundary from an empty batch without checking the engine's behavior. +For comparisons, preserve the source commit, binary hash, release/debug mode, +command, filesystem/storage placement and JSON. CI runs tiny correctness smoke +cases; their numbers are not performance evidence. diff --git a/crates/lance-context-wal/src/bin/lab.rs b/crates/lance-context-wal/src/bin/lab.rs new file mode 100644 index 00000000..733282e7 --- /dev/null +++ b/crates/lance-context-wal/src/bin/lab.rs @@ -0,0 +1,408 @@ +//! Component measurements only: these numbers are NOT distributed-WAL throughput. +use std::alloc::{GlobalAlloc, Layout, System}; +use std::hint::black_box; +use std::io::Cursor; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Barrier}; +use std::time::Instant; + +use anyhow::{ensure, Context, Result}; +use arrow_array::{Array, LargeBinaryArray, RecordBatch, StringArray, UInt64Array}; +use arrow_buffer::Buffer; +use arrow_ipc::reader::{read_footer_length, FileDecoder, FileReader}; +use arrow_ipc::writer::FileWriter; +use bytes::Bytes; +use clap::{Parser, Subcommand}; +use hdrhistogram::Histogram; +use raft_engine::{Config, Engine, LogBatch, MessageExt, ReadableSize}; +use raft_proto::eraftpb::Entry; +use serde_json::json; + +struct CountingAllocator; +static ALLOCS: AtomicU64 = AtomicU64::new(0); +static ALLOC_BYTES: AtomicU64 = AtomicU64::new(0); +unsafe impl GlobalAlloc for CountingAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + ALLOCS.fetch_add(1, Ordering::Relaxed); + ALLOC_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed); + System.alloc(layout) + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + System.dealloc(ptr, layout) + } + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, size: usize) -> *mut u8 { + ALLOCS.fetch_add(1, Ordering::Relaxed); + ALLOC_BYTES.fetch_add(size as u64, Ordering::Relaxed); + System.realloc(ptr, layout, size) + } +} +#[global_allocator] +static ALLOCATOR: CountingAllocator = CountingAllocator; + +#[derive(Parser)] +struct Args { + #[command(subcommand)] + mode: Mode, +} +#[derive(Subcommand)] +enum Mode { + /// With strace, demonstrate the pinned engine's empty-sync behavior. + SyncProbe { + #[arg(long)] + directory: PathBuf, + }, + /// Same Arrow IPC bytes through reader vs shared-buffer decoder vs row copies. + Codec { + #[arg(long, default_value_t = 262144)] + payload_bytes: usize, + #[arg(long, default_value_t = 8)] + rows: usize, + #[arg(long, default_value_t = 30)] + iterations: usize, + }, + /// Multi-group fsync baseline, followed by reopen and full payload verification. + Disk { + /// Must not exist; no existing data is deleted or overwritten. + #[arg(long)] + directory: PathBuf, + #[arg(long, default_value_t = 262144)] + payload_bytes: usize, + #[arg(long, default_value_t = 4)] + batch: usize, + #[arg(long, default_value_t = 4)] + writers: usize, + #[arg(long, default_value_t = 4)] + groups_per_writer: usize, + #[arg(long, default_value_t = 64)] + rounds: usize, + #[arg(long, default_value_t = 0)] + compression_threshold: u64, + }, +} + +fn payload(size: usize, seed: u64) -> Bytes { + // Stable, non-zero, incompressible-ish bytes; generation is outside timing. + let mut x = seed + 1; + let mut out = Vec::with_capacity(size); + for _ in 0..size { + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + out.push(x as u8); + } + out.into() +} + +fn make_batch(rows: usize, size: usize) -> Result { + let bodies: Vec<_> = (0..rows).map(|i| payload(size, i as u64)).collect(); + Ok(RecordBatch::try_from_iter([ + ( + "id", + Arc::new(StringArray::from_iter_values( + (0..rows).map(|i| format!("event-{i}")), + )) as Arc, + ), + ( + "rollout_id", + Arc::new(StringArray::from_iter_values( + (0..rows).map(|i| format!("session-{}", i % 4)), + )) as _, + ), + ( + "sequence_order", + Arc::new(UInt64Array::from_iter_values(0..rows as u64)) as _, + ), + ( + "binary_payload", + Arc::new(LargeBinaryArray::from_iter_values( + bodies.iter().map(|b| b.as_ref()), + )) as _, + ), + ])?) +} + +fn encode(batch: &RecordBatch) -> Result> { + let mut data = Vec::new(); + let mut writer = FileWriter::try_new(&mut data, batch.schema().as_ref())?; + writer.write(batch)?; + writer.finish()?; + drop(writer); + Ok(data) +} + +fn decode(buffer: &Buffer, schema: arrow_schema::SchemaRef) -> Result { + let end = buffer.len().checked_sub(10).context("short IPC file")?; + let len = read_footer_length(buffer[end..].try_into()?)?; + let start = end.checked_sub(len).context("bad footer")?; + let footer = arrow_ipc::root_as_footer(&buffer[start..end]) + .map_err(|e| anyhow::anyhow!("invalid IPC footer: {e}"))?; + let decoder = FileDecoder::new(schema, footer.version()); + let blocks = footer.recordBatches().context("no batches")?; + ensure!(blocks.len() == 1, "one batch required"); + let block = blocks.get(0); + let length = block.bodyLength() as usize + block.metaDataLength() as usize; + let bytes = buffer.slice_with_length(block.offset() as usize, length); + decoder + .read_record_batch(block, &bytes)? + .context("missing batch") +} + +fn measured( + name: &str, + iterations: usize, + bytes: usize, + mut work: impl FnMut() -> Result, +) -> Result<()> { + black_box(work()?); // Warm caches and lazy initialization equally for every path. + let count = ALLOCS.load(Ordering::Relaxed); + let allocated = ALLOC_BYTES.load(Ordering::Relaxed); + let start = Instant::now(); + for _ in 0..iterations { + black_box(work()?); + } + let elapsed = start.elapsed(); + let allocations = ALLOCS.load(Ordering::Relaxed) - count; + let allocated_bytes = ALLOC_BYTES.load(Ordering::Relaxed) - allocated; + println!( + "{}", + json!({"kind":"codec", "path":name, "iterations":iterations, + "ipc_bytes":bytes, "ns_per_iteration":elapsed.as_nanos() / iterations as u128, + "logical_mib_per_second": bytes as f64 * iterations as f64 / elapsed.as_secs_f64() / 1048576.0, + "allocations_per_iteration":allocations as f64 / iterations as f64, + "allocated_bytes_per_iteration":allocated_bytes as f64 / iterations as f64}) + ); + Ok(()) +} + +fn codec(size: usize, rows: usize, iterations: usize) -> Result<()> { + ensure!(size > 0 && rows > 0 && iterations > 0); + let batch = make_batch(rows, size)?; + let buffer = Buffer::from_vec(encode(&batch)?); + verify_shared_decode(&buffer, &batch)?; + let reader_batch = FileReader::try_new(Cursor::new(buffer.as_slice()), None)? + .next() + .context("missing reader batch")??; + ensure!(reader_batch == batch, "reader round trip mismatch"); + measured("encode_ipc", iterations, buffer.len(), || encode(&batch))?; + measured("file_reader", iterations, buffer.len(), || { + Ok(FileReader::try_new(Cursor::new(buffer.as_slice()), None)? + .next() + .context("missing batch")??) + })?; + measured("shared_buffer_decoder", iterations, buffer.len(), || { + decode(&buffer, batch.schema()) + })?; + // Numeric offsets may need alignment repair; LargeBinary body need not copy. + let mut unaligned = Vec::with_capacity(buffer.len() + 1); + unaligned.push(0); + unaligned.extend_from_slice(&buffer); + let unaligned = Buffer::from_vec(unaligned).slice(1); + verify_shared_decode(&unaligned, &batch)?; + measured( + "unaligned_shared_buffer_decoder", + iterations, + buffer.len(), + || decode(&unaligned, batch.schema()), + )?; + measured( + "shared_decoder_then_row_vec", + iterations, + buffer.len(), + || { + let decoded = decode(&buffer, batch.schema())?; + let bodies = decoded + .column(3) + .as_any() + .downcast_ref::() + .unwrap(); + Ok((0..rows) + .map(|row| bodies.value(row).to_vec()) + .collect::>()) + }, + )?; + Ok(()) +} + +/// Check equality and actual payload ownership outside the measured interval. +/// Numeric offset buffers can still allocate when fixing alignment. +fn verify_shared_decode(buffer: &Buffer, expected: &RecordBatch) -> Result<()> { + let decoded = decode(buffer, expected.schema())?; + ensure!(&decoded == expected, "full Arrow round trip mismatch"); + let bodies = decoded + .column(3) + .as_any() + .downcast_ref::() + .context("missing binary payload column")?; + let input_start = buffer.as_ptr() as usize; + let output_start = bodies.value_data().as_ptr() as usize; + let offset = output_start + .checked_sub(input_start) + .context("decoded payload does not share IPC storage")?; + ensure!( + offset + .checked_add(bodies.value_data().len()) + .is_some_and(|end| end <= buffer.len()), + "decoded payload does not share IPC storage" + ); + Ok(()) +} + +struct EntryIndex; +impl MessageExt for EntryIndex { + type Entry = Entry; + fn index(entry: &Entry) -> u64 { + entry.index + } +} + +#[allow(clippy::too_many_arguments)] +fn disk( + directory: PathBuf, + size: usize, + batch: usize, + writers: usize, + groups: usize, + rounds: usize, + compression: u64, +) -> Result<()> { + ensure!(size > 0 && batch > 0 && writers > 0 && groups > 0 && rounds > 0); + ensure!( + size.checked_mul(batch).context("batch overflow")? < 512 * 1024 * 1024, + "cap per-writer batch at 512 MiB" + ); + std::fs::create_dir(&directory).context("directory must be new")?; + std::fs::File::open(directory.parent().context("missing parent")?)?.sync_all()?; + let config = Config { + dir: directory.to_string_lossy().into_owned(), + batch_compression_threshold: ReadableSize(compression), + ..Config::default() + }; + let engine = Arc::new(Engine::open(config.clone())?); + let gate = Arc::new(Barrier::new(writers + 1)); + let mut handles = Vec::new(); + for writer in 0..writers { + let engine = engine.clone(); + let gate = gate.clone(); + handles.push(std::thread::spawn(move || -> Result> { + let bodies: Vec<_> = (0..batch) + .map(|i| payload(size, (writer * batch + i) as u64)) + .collect(); + let mut latency = Histogram::::new(3)?; + gate.wait(); + gate.wait(); + for round in 0..rounds { + let before = Instant::now(); + let region = (writer * groups + round % groups + 1) as u64; + let first = (round / groups * batch + 1) as u64; + let entries: Vec<_> = bodies + .iter() + .enumerate() + .map(|(i, body)| Entry { + index: first + i as u64, + data: body.clone(), + ..Default::default() + }) + .collect(); + let mut log = LogBatch::default(); + log.add_entries::(region, &entries)?; + engine.write(&mut log, true)?; // ACK boundary: includes fdatasync. + latency.record(before.elapsed().as_micros().max(1) as u64)?; + } + Ok(latency) + })); + } + let start = Instant::now(); + gate.wait(); + // Timestamp after generation/barrier; includes all write and join completion. + let writes_started = Instant::now(); + gate.wait(); + let mut latency = Histogram::::new(3)?; + for handle in handles { + latency.add( + handle + .join() + .map_err(|_| anyhow::anyhow!("writer panicked"))??, + )?; + } + let elapsed = writes_started.elapsed(); + let disk_bytes = engine.get_used_size(); + drop(engine); + let recovery_started = Instant::now(); + let engine = Engine::open(config)?; + let reopen = recovery_started.elapsed(); + let verify_started = Instant::now(); + for writer in 0..writers { + let bodies: Vec<_> = (0..batch) + .map(|i| payload(size, (writer * batch + i) as u64)) + .collect(); + for round in 0..rounds { + let region = (writer * groups + round % groups + 1) as u64; + let first = (round / groups * batch + 1) as u64; + for (i, body) in bodies.iter().enumerate() { + let entry = engine + .get_entry::(region, first + i as u64)? + .context("lost entry")?; + ensure!(entry.data == *body, "recovered payload mismatch"); + } + } + } + println!( + "{}", + json!({"kind":"disk", "engine":"raft-engine-0.4.2", "sync":true, + "payload_bytes":size, "batch":batch, "writers":writers, "groups_per_writer":groups, + "rounds":rounds, "compression_threshold":compression, "payload_mib": size as f64 * batch as f64 * rounds as f64 * writers as f64 / 1048576.0, + "elapsed_seconds":elapsed.as_secs_f64(), "payload_mib_per_second":size as f64 * batch as f64 * rounds as f64 * writers as f64 / 1048576.0 / elapsed.as_secs_f64(), + "batch_latency_us_p50":latency.value_at_quantile(0.5),"batch_latency_us_p99":latency.value_at_quantile(0.99), + "disk_bytes":disk_bytes,"reopen_seconds":reopen.as_secs_f64(),"verify_seconds":verify_started.elapsed().as_secs_f64(), + "verified_entries":writers * rounds * batch,"preparation_seconds": writes_started.duration_since(start).as_secs_f64(), + "directory":directory}) + ); + Ok(()) +} + +fn main() -> Result<()> { + match Args::parse().mode { + Mode::SyncProbe { directory } => { + std::fs::create_dir(&directory)?; + let engine = Engine::open(Config { + dir: directory.to_string_lossy().into_owned(), + ..Config::default() + })?; + let mut batch = LogBatch::default(); + batch.put(1, b"probe".to_vec(), vec![3; 8192])?; + engine.write(&mut batch, false)?; + eprintln!("BEFORE_EMPTY_SYNC"); + engine.sync()?; + eprintln!("AFTER_EMPTY_SYNC"); + let mut barrier = LogBatch::default(); + barrier.put(1, b"barrier".to_vec(), vec![1])?; + eprintln!("BEFORE_NONEMPTY_SYNC"); + engine.write(&mut barrier, true)?; + eprintln!("AFTER_NONEMPTY_SYNC"); + Ok(()) + } + Mode::Codec { + payload_bytes, + rows, + iterations, + } => codec(payload_bytes, rows, iterations), + Mode::Disk { + directory, + payload_bytes, + batch, + writers, + groups_per_writer, + rounds, + compression_threshold, + } => disk( + directory, + payload_bytes, + batch, + writers, + groups_per_writer, + rounds, + compression_threshold, + ), + } +} diff --git a/crates/lance-context-wal/src/lib.rs b/crates/lance-context-wal/src/lib.rs new file mode 100644 index 00000000..60f4e07c --- /dev/null +++ b/crates/lance-context-wal/src/lib.rs @@ -0,0 +1,5 @@ +//! Independent Rust WAL experiments owned by lance-context. +//! +//! The optional `lab` feature builds Arrow IPC and raft-engine component +//! benchmarks. It is not connected to production ingestion and makes no +//! distributed durability or end-to-end Lance throughput claim.