serve/tests/data_plane_warehouse.rs
//! Functional test: **`skade-katalog serve` fronts a warehouse the skade DATA
//! PLANE wrote** — the round trip Doris (or any Iceberg-REST client) needs.
//!
//! Two halves, matching the two things that were actually broken when a
//! `skade::Warehouse` was handed to `serve`:
//!
//! 1. `serves_a_local_data_plane_warehouse` (always on) — the catalog side.
//! `skade::Warehouse` and `skade-katalog serve` must agree on the redb
//! **catalog name** (both use `skade`) and on the redb keyspace, or the REST
//! front lists nothing. This pins that agreement so a rename on either side
//! goes red instead of quiet.
//!
//! 2. `serves_an_object_store_data_plane_warehouse` (feature `object-store`) —
//! the storage side. A warehouse opened with
//! `Warehouse::open_with_object_store_config` stamps `s3://…` metadata
//! locations, which a `LocalFsStorageFactory` cannot read: the table LISTS
//! but `loadTable` 500s. The test drives the exact production shape —
//! scheme-carrying `s3://bucket/prefix` locations — through a `LocalFs`
//! object store, so it exercises the object-store code path end to end with
//! no MinIO in the loop.
use std::sync::Arc;
use skade::arrow_array::{Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use skade_katalog_serve::ServeConfig;
use tempfile::TempDir;
fn schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("namn", DataType::Utf8, false),
]))
}
fn batch(n: i64) -> RecordBatch {
let ids: Vec<i64> = (0..n).collect();
let names: Vec<String> = (0..n).map(|i| format!("rad-{i}")).collect();
RecordBatch::try_new(
schema(),
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(StringArray::from(names)),
],
)
.expect("batch")
}
/// GET `url`, returning the parsed JSON body and the HTTP status.
async fn get_json(url: &str) -> (u16, serde_json::Value) {
// No HTTP client dep in this crate — one raw GET over the tokio socket the
// axum front is already listening on is enough (and zero extra deps).
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let (addr, path) = url.split_once('/').expect("host/path");
let mut sock = tokio::net::TcpStream::connect(addr).await.expect("connect");
sock.write_all(
format!("GET /{path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n").as_bytes(),
)
.await
.expect("write");
let mut raw = Vec::new();
sock.read_to_end(&mut raw).await.expect("read");
let text = String::from_utf8_lossy(&raw).to_string();
let status: u16 = text
.split_whitespace()
.nth(1)
.and_then(|s| s.parse().ok())
.unwrap_or(0);
let body = text.split("\r\n\r\n").nth(1).unwrap_or("").to_string();
// Responses are `Connection: close` and small, so the body arrives whole;
// it may still be chunk-framed, so fall back to the first `{`.
let json_start = body.find('{').unwrap_or(0);
let value = serde_json::from_str(&body[json_start..])
.or_else(|_| serde_json::from_str(body.trim()))
.unwrap_or(serde_json::Value::Null);
(status, value)
}
/// The catalog side: a plain `skade::Warehouse::open` warehouse, served.
///
/// This is the half that is NOT about S3 — it pins that the data plane and the
/// serve binary key the same redb under the same catalog name.
#[tokio::test]
async fn serves_a_local_data_plane_warehouse() {
let tmp = TempDir::new().unwrap();
let root = tmp.path().join("wh");
// --- write with the DATA PLANE ---
{
let wh = skade::Warehouse::open(&root).await.expect("open warehouse");
let mut t = wh
.table_or_create("personer", &schema())
.await
.expect("create table");
t.append(&[batch(64)]).await.expect("append");
}
// Warehouse dropped → the redb write-lock is released for `serve`.
// --- serve it ---
let mut cfg = ServeConfig::new(&root);
cfg.rest = Some("127.0.0.1:0".parse().unwrap());
let (rest, _) = skade_katalog_serve::spawn(&cfg).await.expect("spawn");
let rest = rest.expect("rest bound");
assert_eq!(
cfg.warehouse_location(),
format!("file://{}", root.display()),
"with no --warehouse-uri the location is the historical file://ROOT"
);
let (code, ns) = get_json(&format!("{rest}/v1/namespaces")).await;
assert_eq!(code, 200, "GET /v1/namespaces");
assert_eq!(
ns["namespaces"],
serde_json::json!([["main"]]),
"the data plane's DEFAULT_NAMESPACE must be visible over REST — if this \
is empty, serve and skade disagree on the redb catalog name"
);
let (code, tables) = get_json(&format!("{rest}/v1/namespaces/main/tables")).await;
assert_eq!(code, 200, "GET /v1/namespaces/main/tables");
assert_eq!(tables["identifiers"][0]["name"], "personer");
let (code, loaded) = get_json(&format!("{rest}/v1/namespaces/main/tables/personer")).await;
assert_eq!(code, 200, "loadTable: {loaded}");
assert!(
loaded["metadata"]["current-snapshot-id"].is_i64(),
"loadTable returns the real Iceberg metadata document: {loaded}"
);
}
/// The storage side: a warehouse whose blobs live in a skade object store under
/// `s3://…` locations. Without the `object-store` wiring the table lists but
/// `loadTable` fails, because `LocalFsStorageFactory` cannot read an `s3://`
/// metadata-location.
#[cfg(feature = "object-store")]
#[tokio::test]
async fn serves_an_object_store_data_plane_warehouse() {
use skade::object_store::ObjectStoreConfig;
use skade_katalog_serve::WarehouseStorage;
let tmp = TempDir::new().unwrap();
// The object store's backing root. In production this is a MinIO bucket;
// here it is a directory, so the test needs no network — but the LOCATIONS
// the catalog records are the real `s3://bucket/prefix` shape either way,
// which is the thing under test.
let store_root = tmp.path().join("objectstore");
let db_path = tmp.path().join("catalog.redb");
let warehouse_uri = "s3://testbucket/skade-warehouse";
let store_cfg = ObjectStoreConfig::LocalFs {
root: store_root.clone(),
};
// --- write with the DATA PLANE, blobs into the object store ---
{
let wh = skade::Warehouse::open_with_object_store_config(
&db_path,
warehouse_uri,
store_cfg.clone(),
)
.await
.expect("open object-store warehouse");
let mut t = wh
.table_or_create("personer", &schema())
.await
.expect("create table");
t.append(&[batch(64)]).await.expect("append");
}
// --- serve it: same redb, same store, same warehouse URI ---
let mut cfg = ServeConfig::new(tmp.path());
cfg.db_path = db_path;
cfg.warehouse_uri = Some(warehouse_uri.to_string());
cfg.storage = WarehouseStorage::ObjectStore(store_cfg);
cfg.rest = Some("127.0.0.1:0".parse().unwrap());
let (rest, _) = skade_katalog_serve::spawn(&cfg).await.expect("spawn");
let rest = rest.expect("rest bound");
assert_eq!(
cfg.warehouse_location(),
warehouse_uri,
"--warehouse-uri overrides the file://ROOT default"
);
let (code, ns) = get_json(&format!("{rest}/v1/namespaces")).await;
assert_eq!(code, 200);
assert_eq!(ns["namespaces"], serde_json::json!([["main"]]));
let (code, tables) = get_json(&format!("{rest}/v1/namespaces/main/tables")).await;
assert_eq!(code, 200);
assert_eq!(tables["identifiers"][0]["name"], "personer");
// The load is the assertion that matters: it can only succeed if the REST
// front read an `s3://…` metadata document through the object store.
let (code, loaded) = get_json(&format!("{rest}/v1/namespaces/main/tables/personer")).await;
assert_eq!(
code, 200,
"loadTable over an object-store warehouse: {loaded}"
);
let loc = loaded["metadata-location"].as_str().unwrap_or_default();
assert!(
loc.starts_with("s3://testbucket/"),
"the served metadata-location is the object-store URI, not a local path: {loc}"
);
assert!(
loaded["metadata"]["current-snapshot-id"].is_i64(),
"the metadata document itself came back: {loaded}"
);
let data_file = loaded["metadata"]["snapshots"][0]["manifest-list"]
.as_str()
.unwrap_or_default();
assert!(
data_file.starts_with("s3://testbucket/"),
"manifests are addressed in the object store too: {data_file}"
);
}