From 9ac2c53f1202862c65d49d8dc139954ef7f1b640 Mon Sep 17 00:00:00 2001 From: Sergei Turukin Date: Fri, 20 Sep 2024 17:54:40 -0500 Subject: [PATCH] Slatedb based Catalog --- Cargo.lock | 503 +++++++++- Cargo.toml | 2 + crates/iceberg-catalog/Cargo.toml | 5 +- .../src/api/iceberg/v1/tables.rs | 1 + .../src/implementations/mod.rs | 3 + .../src/implementations/slatedb/mod.rs | 884 ++++++++++++++++++ 6 files changed, 1393 insertions(+), 5 deletions(-) create mode 100644 crates/iceberg-catalog/src/implementations/slatedb/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 38fa6023bb..22468cd244 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -210,6 +210,15 @@ version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d62b7694a562cdf5a74227903507c56ab2cc8bdd1f781ed5cb4cf9c9f810bfc" +[[package]] +name = "array-util" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e509844de8f09b90a2c3444684a2b6695f4071360e13d2fda0af9f749cc2ed6" +dependencies = [ + "arrayvec", +] + [[package]] name = "arrayvec" version = "0.7.4" @@ -497,6 +506,28 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "async-stream" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd56dd203fef61ac097dd65721a419ddccb106b2d2b70ba60a6b529f03961a51" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.74", +] + [[package]] name = "async-task" version = "4.7.1" @@ -1217,6 +1248,15 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230c5f1ca6a325a32553f8640d31ac9b49f2411e901e427570154868b46da4f7" +[[package]] +name = "bincode" +version = "1.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -1350,9 +1390,9 @@ dependencies = [ [[package]] name = "bytemuck" -version = "1.16.3" +version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "102087e286b4677862ea56cf8fc58bb2cdfa8725c40ffb80fe3a008eb7f2fc83" +checksum = "94bbb0ad554ad961ddc5da507a12a29b14e4ae5bda06b19f575a3e6079d2e2ae" [[package]] name = "byteorder" @@ -1488,12 +1528,21 @@ dependencies = [ "hostname 0.3.1", "serde", "serde_json", - "snafu", + "snafu 0.6.10", "url", "uuid", "web-sys", ] +[[package]] +name = "cmsketch" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aeccf706e341a5fcdc7f309af21f75eb4dd68fd7474e171bfe1a5570ea48307a" +dependencies = [ + "paste", +] + [[package]] name = "colorchoice" version = "1.0.2" @@ -1602,6 +1651,19 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8" +dependencies = [ + "crossbeam-channel", + "crossbeam-deque", + "crossbeam-epoch", + "crossbeam-queue", + "crossbeam-utils", +] + [[package]] name = "crossbeam-channel" version = "0.5.13" @@ -1611,6 +1673,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613f8cc01fe9cf1a3eb3d7f488fd2fa8388403e97039e2f73692932e291a770d" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-epoch" version = "0.9.18" @@ -1629,6 +1701,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-skiplist" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df29de440c58ca2cc6e587ec3d22347551a32435fbde9d2bff64e78a9ffa151b" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.20" @@ -1673,6 +1755,16 @@ dependencies = [ "typenum", ] +[[package]] +name = "ctor" +version = "0.1.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d2301688392eb071b0bf1a37be05c469d3cc4dbbd95df672fe28ab021e6a096" +dependencies = [ + "quote", + "syn 1.0.109", +] + [[package]] name = "curve25519-dalek" version = "4.1.3" @@ -1960,6 +2052,12 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "downcast-rs" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" + [[package]] name = "dyn-clone" version = "1.0.17" @@ -2092,6 +2190,44 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "fail-parallel" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5666e8ca4ec174d896fb742789c29b1bea9319dcfd623c41bececc0a60c4939d" +dependencies = [ + "log", + "once_cell", + "rand 0.8.5", +] + +[[package]] +name = "fastrace" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25767929385a5128ff5a58c77d5fef43779bb580656801f72d4d32f7a97fea28" +dependencies = [ + "fastrace-macro", + "minstant", + "once_cell", + "parking_lot", + "pin-project", + "rand 0.8.5", + "rtrb", +] + +[[package]] +name = "fastrace-macro" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d11870c0168b3d10fc7a7f0b6f9319fd7244d4caacc669ae86b3ecc5b6ca1b8c" +dependencies = [ + "proc-macro-error", + "proc-macro2", + "quote", + "syn 2.0.74", +] + [[package]] name = "fastrand" version = "1.9.0" @@ -2188,6 +2324,121 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "foyer" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f5636678af1395409e90f7dc14874801c0ef09738688b25655dab2075802dbc" +dependencies = [ + "ahash 0.8.11", + "anyhow", + "fastrace", + "foyer-common", + "foyer-memory", + "foyer-storage", + "futures", + "madsim-tokio", + "pin-project", + "tracing", +] + +[[package]] +name = "foyer-common" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a63e2733a28014366c58f4df1402a11af5ae73c225895a510a137abfced3175" +dependencies = [ + "bytes", + "cfg-if", + "crossbeam", + "fastrace", + "futures", + "hashbrown 0.14.5", + "itertools", + "madsim-tokio", + "metrics", + "parking_lot", + "pin-project", + "serde", +] + +[[package]] +name = "foyer-intrusive" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a836bcc1eddb7a8a0a6bfd1a10d7713176dad006e02c62c3ef8a89cf5119e3" +dependencies = [ + "foyer-common", + "itertools", +] + +[[package]] +name = "foyer-memory" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "374589bc32b3d349a8598c3f542ed682183e089dc1cc927e8dbba99e1945516f" +dependencies = [ + "ahash 0.8.11", + "bitflags 2.6.0", + "cmsketch", + "fastrace", + "foyer-common", + "foyer-intrusive", + "futures", + "hashbrown 0.14.5", + "itertools", + "madsim-tokio", + "parking_lot", + "pin-project", + "serde", + "tracing", +] + +[[package]] +name = "foyer-storage" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "630d1f3617fafc1ccd8aea1ffb7f91a29386a826bd0cb9f976aae27c193892ce" +dependencies = [ + "ahash 0.8.11", + "allocator-api2", + "anyhow", + "array-util", + "async-channel 2.3.1", + "bincode", + "bitflags 2.6.0", + "bytes", + "clap", + "either", + "fastrace", + "foyer-common", + "foyer-memory", + "fs4", + "futures", + "itertools", + "libc", + "lz4", + "madsim-tokio", + "parking_lot", + "pin-project", + "rand 0.8.5", + "serde", + "thiserror", + "tracing", + "twox-hash", + "zstd", +] + +[[package]] +name = "fs4" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8c6b3bd49c37d2aa3f3f2220233b29a7cd23f79d1fe70e5337d25fb390793de" +dependencies = [ + "rustix", + "windows-sys 0.52.0", +] + [[package]] name = "funty" version = "2.0.0" @@ -2635,6 +2886,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "humantime" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4" + [[package]] name = "hyper" version = "0.14.30" @@ -2707,6 +2964,7 @@ dependencies = [ "hyper 1.4.1", "hyper-util", "rustls 0.23.12", + "rustls-native-certs 0.7.1", "rustls-pki-types", "tokio", "tokio-rustls 0.26.0", @@ -2839,11 +3097,13 @@ dependencies = [ "lru", "maplit", "needs_env_var", + "object_store", "rand 0.8.5", "reqwest 0.12.7", "serde", "serde_json", "serde_urlencoded", + "slatedb", "sqlx", "strum", "strum_macros", @@ -3236,6 +3496,25 @@ dependencies = [ "hashbrown 0.14.5", ] +[[package]] +name = "lz4" +version = "1.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a231296ca742e418c43660cb68e082486ff2538e8db432bc818580f3965025ed" +dependencies = [ + "lz4-sys", +] + +[[package]] +name = "lz4-sys" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcb44a01837a858d47e5a630d2ccf304c8efcc4b83b8f9f75b7a9ee4fcc6e57d" +dependencies = [ + "cc", + "libc", +] + [[package]] name = "lz4_flex" version = "0.11.3" @@ -3245,6 +3524,60 @@ dependencies = [ "twox-hash", ] +[[package]] +name = "madsim" +version = "0.2.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f88753ddf8d3cd43b9cf71a93626dd9aad3c24086a04420beb31922e1f856d02" +dependencies = [ + "ahash 0.8.11", + "async-channel 2.3.1", + "async-stream", + "async-task", + "bincode", + "bytes", + "downcast-rs", + "futures-util", + "lazy_static", + "libc", + "madsim-macros", + "naive-timer", + "panic-message", + "rand 0.8.5", + "rand_xoshiro", + "rustversion", + "serde", + "spin", + "tokio", + "tokio-util", + "toml", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "madsim-macros" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3d248e97b1a48826a12c3828d921e8548e714394bf17274dd0a93910dc946e1" +dependencies = [ + "darling 0.14.4", + "proc-macro2", + "quote", + "syn 1.0.109", +] + +[[package]] +name = "madsim-tokio" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53a36782df7996da4863e04526454fbd43067fac1a4dcb8fc663acd538cbf6af" +dependencies = [ + "madsim", + "spin", + "tokio", +] + [[package]] name = "maplit" version = "1.0.2" @@ -3373,6 +3706,16 @@ dependencies = [ "adler2", ] +[[package]] +name = "minstant" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fb9b5c752f145ac5046bccc3c4f62892e3c950c1d1eab80c5949cd68a2078db" +dependencies = [ + "ctor", + "web-time", +] + [[package]] name = "mio" version = "1.0.2" @@ -3415,6 +3758,12 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9252111cf132ba0929b6f8e030cac2a24b507f3a4d6db6fb2896f27b354c714b" +[[package]] +name = "naive-timer" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "034a0ad7deebf0c2abcf2435950a6666c3c15ea9d8fad0c0f48efa8a7f843fed" + [[package]] name = "needs_env_var" version = "1.1.0" @@ -3622,6 +3971,36 @@ dependencies = [ "memchr", ] +[[package]] +name = "object_store" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25a0c4b3a0e31f8b66f71ad8064521efa773910196e2cde791436f13409f3b45" +dependencies = [ + "async-trait", + "base64 0.22.1", + "bytes", + "chrono", + "futures", + "humantime", + "hyper 1.4.1", + "itertools", + "md-5", + "parking_lot", + "percent-encoding", + "quick-xml 0.36.1", + "rand 0.8.5", + "reqwest 0.12.7", + "ring", + "serde", + "serde_json", + "snafu 0.8.4", + "tokio", + "tracing", + "url", + "walkdir", +] + [[package]] name = "once_cell" version = "1.19.0" @@ -3714,6 +4093,12 @@ dependencies = [ "sha2", ] +[[package]] +name = "panic-message" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384e52fd8fbd4cbe3c317e8216260c21a0f9134de108cea8a4dd4e7e152c472d" + [[package]] name = "parking" version = "2.2.0" @@ -4225,6 +4610,24 @@ dependencies = [ "rand_core 0.5.1", ] +[[package]] +name = "rand_xorshift" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d25bf25ec5ae4a3f1b92f929810509a2f53d7dca2f50b794ff57e3face536c8f" +dependencies = [ + "rand_core 0.6.4", +] + +[[package]] +name = "rand_xoshiro" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f97cdb2a36ed4183de61b2f824cc45c9f1037f28afe0a322e9fff4c108b5aaa" +dependencies = [ + "rand_core 0.6.4", +] + [[package]] name = "raw-cpuid" version = "11.1.0" @@ -4394,6 +4797,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", + "h2 0.4.5", "http 1.1.0", "http-body 1.0.1", "http-body-util", @@ -4409,6 +4813,7 @@ dependencies = [ "pin-project-lite", "quinn", "rustls 0.23.12", + "rustls-native-certs 0.7.1", "rustls-pemfile 2.1.3", "rustls-pki-types", "serde", @@ -4510,6 +4915,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rtrb" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3f94e84c073f3b85d4012b44722fa8842b9986d741590d4f2636ad0a5b14143" + [[package]] name = "rust-embed" version = "8.5.0" @@ -5100,6 +5511,12 @@ dependencies = [ "time", ] +[[package]] +name = "siphasher" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56199f7ddabf13fe5074ce809e7d3f42b42ae711800501b5b16ea82ad029c39d" + [[package]] name = "sketches-ddsketch" version = "0.2.2" @@ -5115,6 +5532,41 @@ dependencies = [ "autocfg", ] +[[package]] +name = "slatedb" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c9e4921d27d6b16256acc2146f90d9ac6b1978688cb2f2b5c1737e5066d6f34" +dependencies = [ + "async-channel 2.3.1", + "async-trait", + "atomic", + "bytemuck", + "bytes", + "chrono", + "crc32fast", + "crossbeam-channel", + "crossbeam-skiplist", + "fail-parallel", + "flatbuffers", + "foyer", + "futures", + "log", + "moka", + "object_store", + "once_cell", + "parking_lot", + "rand 0.8.5", + "rand_xorshift", + "serde", + "serde_json", + "siphasher", + "thiserror", + "tokio", + "tracing", + "ulid", +] + [[package]] name = "smallvec" version = "1.13.2" @@ -5131,7 +5583,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eab12d3c261b2308b0d80c26fffb58d17eba81a4be97890101f416b478c79ca7" dependencies = [ "doc-comment", - "snafu-derive", + "snafu-derive 0.6.10", +] + +[[package]] +name = "snafu" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b835cb902660db3415a672d862905e791e54d306c6e8189168c7f3d9ae1c79d" +dependencies = [ + "snafu-derive 0.8.4", ] [[package]] @@ -5145,6 +5606,18 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "snafu-derive" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38d1e02fca405f6280643174a50c942219f0bbf4dbf7d480f1dd864d6f211ae5" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 2.0.74", +] + [[package]] name = "snap" version = "1.1.1" @@ -5947,6 +6420,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "97fee6b57c6a41524a810daee9286c02d7752c4253064d0b05472833a438f675" dependencies = [ "cfg-if", + "rand 0.8.5", "static_assertions", ] @@ -5976,6 +6450,17 @@ version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825" +[[package]] +name = "ulid" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04f903f293d11f31c0c29e4148f6dc0d033a7f80cebc0282bea147611667d289" +dependencies = [ + "getrandom 0.2.15", + "rand 0.8.5", + "web-time", +] + [[package]] name = "uncased" version = "0.9.10" @@ -6336,6 +6821,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 = "webpki-roots" version = "0.25.4" diff --git a/Cargo.toml b/Cargo.toml index 3bb4248efa..20a525aa83 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -72,6 +72,8 @@ jwks_client_rs = { version = "0.5.1", default-features = false, features = [ ] } lazy-regex = { version = "3.2.0", features = ["lite"] } lru = "0.12.4" +slatedb = { version = "*" } +object_store = { version = "0.11.0" } strum = { version = "^0.26", features = ["derive"] } sqlx = { version = "^0.8.0", default-features = false, features = [ "runtime-tokio", diff --git a/crates/iceberg-catalog/Cargo.toml b/crates/iceberg-catalog/Cargo.toml index 51e91b6df8..49323f8709 100644 --- a/crates/iceberg-catalog/Cargo.toml +++ b/crates/iceberg-catalog/Cargo.toml @@ -12,12 +12,13 @@ Implementation of the Iceberg REST Catalog server. """ [features] -all = ["sqlx-postgres", "s3-signer", "router", "nats"] +all = ["sqlx-postgres", "s3-signer", "router", "nats", "slatedb"] sqlx-postgres = ["sqlx"] sqlx = ["dep:sqlx"] s3-signer = ["dep:aws-sigv4", "dep:aws-credential-types"] router = ["dep:tower-http"] nats = ["dep:async-nats"] +slatedb = ["dep:slatedb", "dep:object_store"] default = ["sqlx-postgres", "s3-signer", "router"] [dependencies] @@ -53,10 +54,12 @@ lazy-regex = { workspace = true } lazy_static = { workspace = true } lru = { workspace = true } maplit = { workspace = true } +object_store = { workspace = true, optional = true } rand = "0.8.5" reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true, features = ["preserve_order"] } +slatedb = { workspace = true, optional = true } sqlx = { workspace = true, optional = true, features = ["tls-rustls"] } strum = { workspace = true } strum_macros = { workspace = true } diff --git a/crates/iceberg-catalog/src/api/iceberg/v1/tables.rs b/crates/iceberg-catalog/src/api/iceberg/v1/tables.rs index 3b073ccf51..c5b5da3c68 100644 --- a/crates/iceberg-catalog/src/api/iceberg/v1/tables.rs +++ b/crates/iceberg-catalog/src/api/iceberg/v1/tables.rs @@ -10,6 +10,7 @@ use axum::extract::{Path, Query, State}; use axum::response::IntoResponse; use axum::routing::{get, post}; use axum::{async_trait, Extension, Json, Router}; +use axum_prometheus::lifecycle::service; use http::{HeaderMap, StatusCode}; use iceberg::TableIdent; diff --git a/crates/iceberg-catalog/src/implementations/mod.rs b/crates/iceberg-catalog/src/implementations/mod.rs index 6ab754e64b..1174260302 100644 --- a/crates/iceberg-catalog/src/implementations/mod.rs +++ b/crates/iceberg-catalog/src/implementations/mod.rs @@ -4,6 +4,9 @@ pub const DEFAULT_PROJECT_ID: uuid::Uuid = uuid::uuid!("00000000-0000-0000-0000- #[cfg(feature = "sqlx-postgres")] pub mod postgres; +#[cfg(feature = "slatedb")] +pub mod slatedb; + mod authz; pub mod kv2; diff --git a/crates/iceberg-catalog/src/implementations/slatedb/mod.rs b/crates/iceberg-catalog/src/implementations/slatedb/mod.rs new file mode 100644 index 0000000000..6de0571e44 --- /dev/null +++ b/crates/iceberg-catalog/src/implementations/slatedb/mod.rs @@ -0,0 +1,884 @@ +#![allow(unused_variables)] + +use futures::future::join_all; +use tokio::task::JoinSet; +use http::StatusCode; +use uuid::Uuid; +use std::collections::{HashMap, HashSet}; +use slatedb::error::SlateDBError; +use std::sync::Arc; +use iceberg_ext::catalog::rest::ErrorModel; + +use iceberg::spec::ViewMetadata; +use crate::service::{health::{HealthExt, HealthStatus, Health}, Catalog, Transaction, Result, TableCommit, IcebergErrorResponse}; +use crate::{ + service::{ + storage::StorageProfile, CreateTableResponse, GetNamespaceResponse, + GetTableMetadataResponse, LoadTableResponse, NamespaceIdentUuid, ProjectIdent, + TableIdentUuid, WarehouseIdent, + }, + SecretIdent, +}; +use crate::api::management::v1::warehouse::TabularDeleteProfile; +use crate::api::iceberg::v1::{PaginatedTabulars, PaginationQuery}; +use crate::api::iceberg::types::PageToken; +use crate::service::tabular_idents::{TabularIdentOwned, TabularIdentUuid}; +use crate::service::{ + CreateNamespaceRequest, CreateNamespaceResponse, CreateTableRequest, DeletionDetails, + GetWarehouseResponse, ListFlags, ListNamespacesQuery, ListNamespacesResponse, NamespaceIdent, + TableIdent, WarehouseStatus, +}; + + +#[derive(Clone)] +pub struct SlateDbCatalog {} + +#[derive(Clone)] +pub struct SlateDbTransaction { + state: SlateDbState +} + +#[derive(Clone)] +pub struct SlateDbState { + db: Arc +} + +#[async_trait::async_trait] +impl HealthExt for SlateDbState { + async fn health(&self) -> Vec { + vec![Health::now("slatedb", HealthStatus::Healthy)] + } + async fn update_health(&self) {} +} + +#[async_trait::async_trait] +impl Transaction for SlateDbTransaction { + type Transaction<'a> = SlateDbState; + + async fn begin_write(state: SlateDbState) -> Result { + Ok(Self { state }) + } + + async fn begin_read(state: SlateDbState) -> Result { + Ok(Self { state }) + } + + async fn commit(self) -> Result<()> { + Ok(()) + } + + async fn rollback(self) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + fn transaction(&mut self) -> Self::Transaction<'_> { self.state.clone() } +} + +#[derive(serde::Serialize, serde::Deserialize)] +struct IdList(Vec); + +type Warehouse = crate::api::management::v1::warehouse::GetWarehouseResponse; +#[derive(serde::Serialize, serde::Deserialize)] +struct Namespace { + namespace_id: Uuid, + namespace: NamespaceIdent, + properties: Option>, +} + + + +pub(crate) trait SlateDbErrorHandler +where + Self: ToString + Sized + Send + Sync + std::error::Error + 'static, +{ + fn into_error_model(self, message: String) -> ErrorModel { + ErrorModel::internal(message, "DatabaseError", Some(Box::new(self))) + } +} + +impl SlateDbErrorHandler for SlateDBError { } +impl SlateDbErrorHandler for serde_json::Error { } + +#[async_trait::async_trait] +impl Catalog for SlateDbCatalog { + type Transaction = SlateDbTransaction; + type State = SlateDbState; + + async fn create_warehouse<'a>( + warehouse_name: String, + project_id: ProjectIdent, + storage_profile: StorageProfile, + tabular_delete_profile: TabularDeleteProfile, + storage_secret_id: Option, + transaction: >::Transaction<'a>, + ) -> Result { + let state = transaction; + // Generate UUID + let wid = WarehouseIdent::from(uuid::Uuid::new_v4()); + // Convert payload to binary format (serialize GetWarehouseResponse) + // let payload = Warehouse::from(Warehouse::new(wid, warehouse_name, WarehouseStatus::Active, project_id, storage_profile)); + let payload = Warehouse { + id: *wid, + name: warehouse_name, + status: WarehouseStatus::Active, + project_id: *project_id, + storage_profile, + }; + let value = serde_json::to_vec(&payload).map_err(|e| { + e.into_error_model(String::from("Failed to serialize warehouse")) + })?; + // Put into db + // Update warehouse list + let key = format!("warehouse/{}", wid.to_string()); + state.db.put(key.as_bytes(), &value).await; + + let key = format!("warehouse/_all_"); + let value = state.db.get(key.as_bytes()).await.map_err(|e| { + e.into_error_model(String::from("Failed to read warehouse list")) + })?; + + let mut list: Vec = match value { + Some(bytes) => serde_json::from_slice(&bytes).map_err(|e| { + e.into_error_model(String::from("Failed to deserialize warehouse list")) + })?, + None => Vec::new(), + }; + list.push(wid.0); + + let value = serde_json::to_vec(&list).map_err(|e| { + e.into_error_model(String::from("Failed to serialize warehouse list")) + })?; + state.db.put(key.as_bytes(), &value).await; + + Ok(wid) + } + + async fn get_warehouse<'a>( + warehouse_id: WarehouseIdent, + transaction: >::Transaction<'a>, + ) -> Result { + let state = transaction; + let value = state.db.get(format!("warehouse/{}", warehouse_id.to_string()).as_bytes()).await; + let payload: Warehouse = serde_json::from_slice(&value.unwrap().unwrap()).map_err(|e| { + e.into_error_model(String::from("Failed to deserialize warehouse")) + })?; + Ok(GetWarehouseResponse { + id: WarehouseIdent::from(payload.id), + name: payload.name, + status: payload.status, + project_id: ProjectIdent::from(payload.project_id), + storage_profile: payload.storage_profile, + tabular_delete_profile: TabularDeleteProfile::Soft { + expiration_seconds: chrono::Duration::seconds(5), + }, + storage_secret_id: None, + }) + } + + async fn get_namespace<'a>( + warehouse_id: WarehouseIdent, + namespace: &NamespaceIdent, + transaction: >::Transaction<'a>, + ) -> Result { + let state = transaction; + + let namespace = namespace.to_url_string(); + let key = format!("namespace/{warehouse_id}/{namespace}"); + let value = state.db.get(key.as_bytes()).await.map_err(|e| { + e.into_error_model(String::from("Failed to read namespace")) + })?; + + let payload: Namespace = match value { + Some(bytes) => serde_json::from_slice(&bytes).map_err(|e| { + e.into_error_model(String::from("Failed to deserialize namespace")) + })?, + None => return Err(ErrorModel::builder() + .code(StatusCode::NOT_FOUND.into()) + .message("Namespace not found".to_string()) + .r#type("NotFound".to_string()) + .build() + .into()) + }; + Ok(GetNamespaceResponse { + warehouse_id, + namespace: payload.namespace, + namespace_id: NamespaceIdentUuid::from(payload.namespace_id), + properties: payload.properties, + }) + } + + async fn list_namespaces( + warehouse_id: WarehouseIdent, + query: &ListNamespacesQuery, + catalog_state: Self::State, + ) -> Result { + let state = catalog_state; + + let key = format!("namespace/{warehouse_id}/_all_"); + let value = state.db.get(key.as_bytes()).await.map_err(|e| { + e.into_error_model(String::from("Failed to read namespace list")) + })?; + + let list: Vec = match value { + Some(bytes) => serde_json::from_slice(&bytes).map_err(|e| { + e.into_error_model(String::from("Failed to deserialize namespace list")) + })?, + None => Vec::new(), + }; + + + let mut set = JoinSet::new(); + list.into_iter().map(|ns| { + let state = state.clone(); + set.spawn(async move { + Self::get_namespace(warehouse_id, &ns, state.clone()).await + }) + }).collect::>(); + let results = set.join_all().await.into_iter().map(|r| r.unwrap().namespace); + + Ok(ListNamespacesResponse { next_page_token: None, namespaces: results.collect() }) + } + + async fn create_namespace<'a>( + warehouse_id: WarehouseIdent, + namespace_id: NamespaceIdentUuid, + request: CreateNamespaceRequest, + transaction: >::Transaction<'a>, + ) -> Result { + let state = transaction; + // Convert payload to binary format (serialize GetNamespaceResponse) + let namespace_name = request.namespace.to_url_string(); + let payload = Namespace { + namespace: request.namespace.clone(), + namespace_id: *namespace_id, + properties: request.properties.clone(), + }; + let value = serde_json::to_vec(&payload).map_err(|e| { + e.into_error_model(String::from("Failed to serialize namespace")) + })?; + // Put into db + let key = format!("namespace/{warehouse_id}/{namespace_name}"); + state.db.put(key.as_bytes(), &value).await; + + // Save into list + let key = format!("namespace/{warehouse_id}/_all_"); + let value = state.db.get(key.as_bytes()).await.map_err(|e| { + e.into_error_model(String::from("Failed to read namespace list")) + })?; + let mut list: Vec = match value { + Some(bytes) => serde_json::from_slice(&bytes).map_err(|e| { + e.into_error_model(String::from("Failed to deserialize namespace list")) + })?, + None => Vec::new(), + }; + list.push(request.namespace); + let value = serde_json::to_vec(&list).map_err(|e| { + e.into_error_model(String::from("Failed to serialize namespace list")) + })?; + state.db.put(key.as_bytes(), &value).await; + + Ok(CreateNamespaceResponse { + namespace: payload.namespace, + properties: payload.properties, + }) + } + + async fn namespace_ident_to_id( + warehouse_id: WarehouseIdent, + namespace: &NamespaceIdent, + catalog_state: Self::State, + ) -> Result> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn drop_namespace<'a>( + warehouse_id: WarehouseIdent, + namespace: &NamespaceIdent, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn update_namespace_properties<'a>( + warehouse_id: WarehouseIdent, + namespace: &NamespaceIdent, + properties: HashMap, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn create_table<'a>( + namespace_id: NamespaceIdentUuid, + table: &TableIdent, + table_id: TableIdentUuid, + request: CreateTableRequest, + // Metadata location may be none if stage-create is true + metadata_location: Option<&str>, + transaction: >::Transaction<'a>, + ) -> Result { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn list_tables( + warehouse_id: WarehouseIdent, + namespace: &NamespaceIdent, + list_flags: crate::service::ListFlags, + catalog_state: Self::State, + pagination_query: PaginationQuery, + ) -> Result> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + // Should also load staged tables but not tables of inactive warehouses + async fn load_tables<'a>( + warehouse_id: WarehouseIdent, + tables: impl IntoIterator + Send, + include_deleted: bool, + transaction: >::Transaction<'a>, + ) -> Result> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn get_table_metadata_by_id( + warehouse_id: WarehouseIdent, + table: TableIdentUuid, + list_flags: crate::service::ListFlags, + catalog_state: Self::State, + ) -> Result { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn get_table_metadata_by_s3_location( + warehouse_id: WarehouseIdent, + location: &str, + list_flags: crate::service::ListFlags, + catalog_state: Self::State, + ) -> Result { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn table_ident_to_id( + warehouse_id: WarehouseIdent, + table: &TableIdent, + list_flags: crate::service::ListFlags, + catalog_state: Self::State, + ) -> Result> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn rename_table<'a>( + warehouse_id: WarehouseIdent, + source_id: TableIdentUuid, + source: &TableIdent, + destination: &TableIdent, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn drop_table<'a>( + table_id: TableIdentUuid, + transaction: >::Transaction<'a>, + ) -> Result { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn table_idents_to_ids( + warehouse_id: WarehouseIdent, + tables: HashSet<&TableIdent>, + list_flags: crate::service::ListFlags, + catalog_state: Self::State, + ) -> Result>> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn commit_table_transaction<'a>( + warehouse_id: WarehouseIdent, + commits: impl IntoIterator + Send, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + // ---------------- Management API ---------------- + async fn list_projects(catalog_state: Self::State) -> Result> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn list_warehouses( + project_id: ProjectIdent, + include_inactive: Option>, + warehouse_id_filter: Option<&HashSet>, + catalog_state: Self::State, + ) -> Result> { + let value = catalog_state.db.get("warehouse/_all_".as_bytes()).await.map_err(|e| { + e.into_error_model(String::from("Failed to read warehouse list")) + }); + + let list: Vec = match value { + Ok(Some(bytes)) => serde_json::from_slice(&bytes).map_err(|e| { + e.into_error_model(String::from("Failed to deserialize warehouse list")) + })?, + _ => Vec::new(), + }; + + // Now we need to spawn a task for each warehouse to get the warehouse details + let mut tasks = Vec::new(); + for wid in list { + let state = catalog_state.clone(); + tasks.push(tokio::spawn(async move { + Self::get_warehouse(WarehouseIdent::from(wid), state).await + })); + } + + let res = join_all(tasks).await; + res.into_iter().map(|r| r.unwrap()).collect() + } + + async fn delete_warehouse<'a>( + warehouse_id: WarehouseIdent, + transaction: >::Transaction<'a>, + ) -> Result<()> { + let state = transaction; + // Delete warehouse + state.db.delete(format!("warehouse/{warehouse_id}").as_bytes()).await; + // Update warehouse list + let key = String::from("warehouse/_all_"); + let value = state.db.get(key.as_bytes()).await.map_err(|e| { + e.into_error_model(String::from("Failed to read warehouse list")) + })?; + + let mut list: Vec = match value { + Some(bytes) => serde_json::from_slice(&bytes).map_err(|e| { + e.into_error_model(String::from("Failed to deserialize warehouse list")) + })?, + None => Vec::new(), + }; + list.retain(|&x| x != warehouse_id.0); + + let value = serde_json::to_vec(&list).map_err(|e| { + e.into_error_model(String::from("Failed to serialize warehouse list")) + })?; + state.db.put(key.as_bytes(), &value).await; + + Ok(()) + } + + async fn rename_warehouse<'a>( + warehouse_id: WarehouseIdent, + new_name: &str, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn set_warehouse_status<'a>( + warehouse_id: WarehouseIdent, + status: WarehouseStatus, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn update_storage_profile<'a>( + warehouse_id: WarehouseIdent, + storage_profile: StorageProfile, + storage_secret_id: Option, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn create_view<'a>( + namespace_id: NamespaceIdentUuid, + view: &TableIdent, + request: ViewMetadata, + metadata_location: &str, + transaction: >::Transaction<'a>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn view_ident_to_id( + warehouse_id: WarehouseIdent, + view: &TableIdent, + catalog_state: Self::State, + ) -> Result> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn load_view<'a>( + view_id: TableIdentUuid, + include_deleted: bool, + transaction: >::Transaction<'a>, + ) -> Result { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn list_views( + warehouse_id: WarehouseIdent, + namespace: &NamespaceIdent, + include_deleted: bool, + catalog_state: Self::State, + pagination_query: PaginationQuery, + ) -> Result> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn update_view_metadata( + namespace_id: NamespaceIdentUuid, + view_id: TableIdentUuid, + view: &TableIdent, + metadata_location: &str, + metadata: ViewMetadata, + transaction: >::Transaction<'_>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn rename_view( + warehouse_id: WarehouseIdent, + source_id: TableIdentUuid, + source: &TableIdent, + destination: &TableIdent, + transaction: >::Transaction<'_>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn drop_view<'a>( + table_id: TableIdentUuid, + transaction: >::Transaction<'a>, + ) -> Result { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn list_tabulars( + warehouse_id: WarehouseIdent, + list_flags: ListFlags, + catalog_state: Self::State, + pagination_query: PaginationQuery, + ) -> Result)>> + { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } + + async fn mark_tabular_as_deleted( + table_id: TabularIdentUuid, + transaction: >::Transaction<'_>, + ) -> Result<()> { + Err(ErrorModel::builder() + .code(StatusCode::INTERNAL_SERVER_ERROR.into()) + .message("Not implemented".to_string()) + .r#type("NotImplemented".to_string()) + .build() + .into()) + } +} + +#[cfg(test)] +pub mod tests { + use super::*; + use tokio; + use crate::service::storage::{S3Flavor, S3Profile}; + use object_store::{ObjectStore, memory::InMemory, path::Path}; + use slatedb::db::Db; + use slatedb::config::DbOptions; + + async fn create_state() -> SlateDbState { + let object_store: Arc = Arc::new(InMemory::new()); + let options = DbOptions::default(); + SlateDbState { + db: Arc::new(Db::open_with_opts(Path::from("/tmp/kv_store"), options, object_store).await.unwrap()) + } + } + + async fn create_warehouse(state: SlateDbState, name: Option) -> WarehouseIdent { + + let project_id = ProjectIdent::from(uuid::Uuid::nil()); + let storage_profile = StorageProfile::S3(S3Profile { + bucket: "test_bucket".to_string(), + endpoint: None, + region: "us-east-1".to_string(), + assume_role_arn: None, + path_style_access: None, + key_prefix: None, + sts_role_arn: None, + sts_enabled: false, + flavor: S3Flavor::Minio, + }); + + let mut txn = SlateDbTransaction::begin_write(state.clone()).await.unwrap(); + + SlateDbCatalog::create_warehouse( + name.unwrap_or("test_warehouse".to_string()), + project_id, + storage_profile, + TabularDeleteProfile::Soft { + expiration_seconds: chrono::Duration::seconds(5), + }, + None, + txn.transaction(), + ) + .await + .expect("Failed to create warehouse") + } + + async fn create_namespace(state: SlateDbState, warehouse_id: WarehouseIdent, name: Option) -> CreateNamespaceResponse { + let mut txn = SlateDbTransaction::begin_write(state.clone()).await.unwrap(); + let namespace_id = NamespaceIdentUuid::from(uuid::Uuid::new_v4()); + let request = CreateNamespaceRequest { + namespace: NamespaceIdent::new(name.unwrap_or("test_namespace".to_string())), + properties: None, + }; + SlateDbCatalog::create_namespace( + warehouse_id, + namespace_id, + request, + txn.transaction(), + ) + .await + .expect("Failed to create namespace") + } + + #[tokio::test] + async fn test_namespace_create() { + let state = create_state().await; + let wh = create_warehouse(state.clone(), None).await; + let ns = create_namespace(state.clone(), wh, None).await; + assert_eq!(ns.namespace.to_url_string(), "test_namespace"); + } + + #[tokio::test] + async fn test_namespace_get() { + let state = create_state().await; + let wh = create_warehouse(state.clone(), None).await; + let ns = create_namespace(state.clone(), wh, Some(String::from("ns1"))).await; + let result = SlateDbCatalog::get_namespace(wh, &ns.namespace, state).await.expect("Failed to get namespace"); + assert_eq!(result.namespace.to_url_string(), "ns1"); + } + + #[tokio::test] + async fn test_namespace_list() { + let empty_query = ListNamespacesQuery { + page_token: PageToken::Empty, // Assuming PageToken::Empty is a valid empty state + page_size: None, + parent: None, + }; + let state = create_state().await; + let wh = create_warehouse(state.clone(), None).await; + let ns1 = create_namespace(state.clone(), wh, Some(String::from("ns1"))).await; + let ns2 = create_namespace(state.clone(), wh, Some(String::from("ns2"))).await; + let result = SlateDbCatalog::list_namespaces(wh, &empty_query, state).await.expect("Failed to list namespaces"); + assert_eq!(result.namespaces.len(), 2); + assert_eq!(result.namespaces[0].to_url_string(), "ns1"); + assert_eq!(result.namespaces[1].to_url_string(), "ns2"); + } + + #[tokio::test] + async fn test_namespace_delete() { + let empty_query = ListNamespacesQuery { + page_token: PageToken::Empty, // Assuming PageToken::Empty is a valid empty state + page_size: None, + parent: None, + }; + let state = create_state().await; + let wh = create_warehouse(state.clone(), None).await; + let ns = create_namespace(state.clone(), wh, None).await; + SlateDbCatalog::drop_namespace(wh, &ns.namespace, state.clone()).await.expect("Failed to delete namespace"); + + let result = SlateDbCatalog::list_namespaces(wh, &empty_query, state).await.expect("Failed to list namespaces"); + assert_eq!(result.namespaces.len(), 0); + } + + + #[tokio::test] + async fn test_warehouse_create() { + let state = create_state().await; + let wh = create_warehouse(state, None).await; + assert_ne!(wh.0.to_string(), ""); + } + + #[tokio::test] + async fn test_warehouse_get() { + let state = create_state().await; + let wh = create_warehouse(state.clone(), None).await; + let result = SlateDbCatalog::get_warehouse(wh, state).await.expect("Failed to get warehouse"); + assert_eq!(result.id, wh); + assert_eq!(result.name, "test_warehouse"); + if let StorageProfile::S3(s3_profile) = &result.storage_profile { + assert_eq!(s3_profile.bucket, "test_bucket"); + assert_eq!(s3_profile.region, "us-east-1"); + } else { + panic!("Unexpected storage profile type"); + } + } + + #[tokio::test] + async fn test_warehouse_list() { + let state = create_state().await; + let wh1 = create_warehouse(state.clone(), Some(String::from("wh1"))).await; + let wh2 = create_warehouse(state.clone(), Some(String::from("wh2"))).await; + let result = SlateDbCatalog::list_warehouses( + ProjectIdent::from(uuid::Uuid::nil()), + None, + None, + state, + ) + .await + .expect("Failed to list warehouses"); + assert_eq!(result.len(), 2); + assert_eq!(result[0].id, wh1); + assert_eq!(result[0].name, "wh1"); + assert_eq!(result[1].id, wh2); + assert_eq!(result[1].name, "wh2"); + } + + #[tokio::test] + async fn test_warehouse_delete() { + let state = create_state().await; + let wh = create_warehouse(state.clone(), None).await; + SlateDbCatalog::delete_warehouse(wh, state.clone()).await.expect("Failed to delete warehouse"); + + let result = SlateDbCatalog::list_warehouses( + ProjectIdent::from(uuid::Uuid::nil()), + None, + None, + state, + ) + .await + .expect("Failed to list warehouses"); + assert_eq!(result.len(), 0); + } +} \ No newline at end of file